feat(platform): hire an expert and hand off the task in one click (#16101)

Linear: [SECRT-3138](https://linear.app/autogpt/issue/SECRT-3138)

## What

When Otto proposes a hire for a task the user already described, **one
click now hires the expert and starts that task** in the expert's own
chat. Before, the user approved the hire, then had to type "sure, hand
him the task".

## How

- `hire_expert` takes the task (`handoff_title`, `handoff_task`). The
card asks one yes/no question: **Hire and start** (always the primary)
or **Not now**. The plain hire id stays on the backend for a user who
asks in chat to hire without starting the task.
- Confirming hires and starts the task as a real delegation, with the
same thread, permissions and limits as `delegate_to_expert`. One answer
can never hire twice.
- If the task can't start, the hire stands and the card says so. **Try
again** shows only when the backend marks the failure retryable
(`handoff.retry`: nothing reached the expert), and only while the new
hire has no chat yet. Otto is told to resend a retryable task himself,
so if he already did, the card links to that chat instead and the task
can't start twice. Refusals (paused, out of budget or depth, no free
slot) never offer it.
- The receipt records the choice ("You chose to hire James to take on
X."), since the confirm runs after the send and can still fail. If the
expert's thread can't be read, the card says only what is known ("Sent X
to James. Open their chat to see where it stands."). Two hires confirmed
together keep their pager.
- A confirm held by Ask First names the task its hand-off id would start
("James to take on X", "New hire with a first task").
- The typed dispatch errors keep the original error in their message, so
the AutoPilot block's `error` output still shows the reason.
- The card shows only what the user needs: one question to answer, then
one line of status. The brief and the expert's profile sit behind **What
James will get**.
- Prompt: adds one line to Otto's team-building supplement (pass the
task to `hire_expert`; never ask in text whether to hand a task to a new
hire from `tool:hire_expert`). Langfuse untouched.
- Otto's wording, after the prompt hold was lifted: `hire_expert` tells
Otto the card has two buttons (Hire and start, Not now) and to confirm
`handoff_confirmation_id` on an approval. A sent hand-off no longer says
"Their result comes back here": Otto is told to read the result with
`get_sub_session_result`. A retryable failure has one retry path, the
card's Try again, and Otto is told not to hand it over himself. The
prompt rule against asking in text now names only `tool:hire_expert`
hires, so `raise_expert` hires stay under "Never delegate silently".

## For product review: the flow

**1. Otto proposes the hire**
![Otto
proposes](https://github.com/user-attachments/assets/42bae703-e924-428b-8c61-bbf78a560454)

**2. Details, only if the user wants them**
![Details
open](https://github.com/user-attachments/assets/771defb4-9a21-46c4-b5f6-6a8c18a80f2b)

**3. After one click, the card shrinks to a record of the choice**

![Receipt](https://github.com/user-attachments/assets/4d814760-5462-460f-ad01-7dc99ed5ec4b)

**4. Waiting for a free slot**

![Waiting](https://github.com/user-attachments/assets/32c3dc9e-7ad6-4f8e-8034-ab21884ad1fe)

**5. James is working**

![Working](https://github.com/user-attachments/assets/ff1ade98-741b-4c47-8b64-a09e8cb6e889)

**6. James stopped (the result, or a question, is in his chat)**

![Stopped](https://github.com/user-attachments/assets/1e1a6ca5-2c27-4ea3-a724-d083ec1d552c)

**7. The task didn't start** (Try again appears only for a retryable
failure while the new hire has no chat)
![Didn't
start](https://github.com/user-attachments/assets/ed27400f-302a-4173-9c8d-f29564c6da39)

**Mobile (390 px)**

| Proposal | Working | Stopped |
|---|---|---|
| ![Mobile
proposal](https://github.com/user-attachments/assets/cd3393bf-3cf9-4b4e-a8cf-1a1b992e3dbf)
| ![Mobile
working](https://github.com/user-attachments/assets/03763797-d03a-4824-bfcd-b0f0ca296f94)
| ![Mobile
stopped](https://github.com/user-attachments/assets/e25d3856-221d-40dc-8586-a401ab5ffbae)
|

## Tests

- Frontend: `ExpertHandoff.test.tsx` covers every card state, the
two-button answer, retry shown only when retryable and hidden once Otto
sent it again, the unknown-status sentence, and paging between two
confirms. All copilot and approval-field tests pass (329 files, 4,445
tests), as do lint, types and the React Compiler check.
- Backend: `hire_handoff_test.py` (18 tests, including `retry` on every
outcome), `gate/references_test.py` (the held hand-off confirm names the
task), and `session_waiter_test.py` / `executor/utils_test.py` (the
error message keeps its reason). Each new assertion fails without its
fix. The related tool, gate and prompt suites pass (638 tests).
- Screenshots 1, 2, 3, 5 and the mobile proposal were retaken on the
current head in the real app, with a real `hire_expert` proposal and a
real confirm. This sandbox has no LLM key or copilot executor, so the
turn and the expert's thread stay running.

## Checklist

- [x] Changes are clearly listed above
- [x] Test plan made and run (unit tests above, every card state
screenshotted at desktop and 390 px)
- [x] No configuration changes

<!-- conductor-workspace-link -->

---

[Open workspace in
Conductor](https://app.conductor.build/pro-user-8d964eae-256d-46a9-b469-fd3c36fa6673/w/162)

---------

Co-authored-by: Claude Opus 5.5 (1M context) <[email protected]>
Co-authored-by: Cursor <[email protected]>
This commit is contained in:
authored and GitHub committed 2026-10-27 12:20:58 +00:00
1 parent 2ba900a09b
commit 7ca3671f6a
35 files changed
+3396 -248

No files matched your search

@@ -225,6 +225,17 @@ async def acquire_turn_slot(
try:
yield handle
finally:
except BaseException:
# Cleanup on a failure path: a release that fails too must not replace
# the error the caller acts on (a publish that may have reached the
# broker would otherwise read as a turn that was never sent).
if handle.admitted and not handle._kept:
await release_turn_slot(user_id, session_id)
try:
await release_turn_slot(user_id, session_id)
except Exception:
logger.exception(
f"acquire_turn_slot: release failed for session={session_id}"
)
raise
if handle.admitted and not handle._kept:
await release_turn_slot(user_id, session_id)
@@ -199,3 +199,28 @@ def test_schema_default_concurrent_turn_limit_is_15() -> None:
from backend.util.settings import Config
assert Config.model_fields["max_inflight_copilot_turns_per_user"].default == 15
@pytest.mark.asyncio
async def test_a_failed_release_never_hides_the_error_that_caused_it() -> None:
"""On a failure path the release is cleanup: if it fails too, the error
the caller must act on is still the one raised in the block (a publish
that may have reached the broker must not turn into a Redis error that
reads as "nothing was sent")."""
db = _mock_db()
db.update_chat_session_status = AsyncMock(side_effect=ConnectionError("db blip"))
with patch.object(active_turns, "chat_db", return_value=db):
with pytest.raises(RuntimeError, match="publish may have landed"):
async with acquire_turn_slot("user-1", "session-a"):
raise RuntimeError("publish may have landed")
db.update_chat_session_status.assert_awaited_once()
@pytest.mark.asyncio
async def test_a_failed_release_on_a_clean_exit_still_raises() -> None:
db = _mock_db()
db.update_chat_session_status = AsyncMock(side_effect=ConnectionError("db blip"))
with patch.object(active_turns, "chat_db", return_value=db):
with pytest.raises(ConnectionError, match="db blip"):
async with acquire_turn_slot("user-1", "session-a"):
pass
@@ -393,6 +393,16 @@ class CancelCoPilotEvent(BaseModel):
# ============ Queue Publishing Helpers ============ #
class TurnPublishUnconfirmed(Exception):
"""Publishing a turn raised, so the broker may hold it or not.
A publish can fail after the broker took the message (a lost confirm),
so unlike any failure before it, this one does not prove the turn was
never sent. A caller deciding whether to send a task again must treat it
as possibly running. The original error is the ``__cause__``.
"""
async def enqueue_copilot_turn(
session_id: str,
user_id: str | None,
@@ -477,11 +487,16 @@ async def enqueue_copilot_turn(
)
queue_client = await get_async_copilot_queue()
await queue_client.publish_message(
routing_key=COPILOT_EXECUTION_ROUTING_KEY,
message=entry.model_dump_json(),
exchange=COPILOT_EXECUTION_EXCHANGE,
)
try:
await queue_client.publish_message(
routing_key=COPILOT_EXECUTION_ROUTING_KEY,
message=entry.model_dump_json(),
exchange=COPILOT_EXECUTION_EXCHANGE,
)
except Exception as e:
raise TurnPublishUnconfirmed(
f"The turn's publish raised; it may or may not be queued: {e}"
) from e
async def schedule_turn(
@@ -711,7 +726,14 @@ async def dispatch_turn(
finally:
if not committed:
if admitted_here:
await release_turn(envelope)
try:
await release_turn(envelope)
except BaseException:
# Same rule as the meta cleanup below: never mask the
# original error, and never skip the rest of the cleanup.
logger.exception(
f"dispatch_turn: tree release failed for session={session_id}"
)
try:
await stream_registry.delete_session_meta(session_id)
except BaseException:
@@ -628,3 +628,134 @@ async def test_queue_spawned_turn_stores_impersonation_with_the_queued_row() ->
usage_snapshot = usage_activation_fixtures.usage_snapshot
@pytest.mark.asyncio
async def test_a_failed_publish_is_unconfirmed_not_unsent() -> None:
"""The broker may hold a message whose publish raised (a lost confirm),
so that failure must not read as "nothing was sent" to a caller that
would then send the same task again (SECRT-3138 review)."""
queue = MagicMock(publish_message=AsyncMock(side_effect=OSError("confirm lost")))
slot = MagicMock()
with (
patch.object(
utils,
"_admitted_turn_envelope",
new=AsyncMock(return_value=root_envelope("t1")),
),
patch.object(utils, "release_turn", new=AsyncMock()),
patch("backend.copilot.stream_registry.create_session", new=AsyncMock()),
patch("backend.copilot.stream_registry.delete_session_meta", new=AsyncMock()),
patch(
"backend.util.clients.get_async_copilot_queue",
new=AsyncMock(return_value=queue),
),
pytest.raises(utils.TurnPublishUnconfirmed) as raised,
):
await utils.dispatch_turn(
slot, session_id="s1", user_id="u1", turn_id="t1", message="hi"
)
assert isinstance(raised.value.__cause__, OSError)
# Callers that report str(exc) (the AutoPilot block) keep the reason.
assert "confirm lost" in str(raised.value)
slot.keep.assert_not_called()
@pytest.mark.asyncio
async def test_a_failure_before_the_publish_keeps_its_own_type() -> None:
"""Before the publish nothing can have reached the broker, so the error
stays what it was and callers may treat the turn as never sent."""
queue = MagicMock(publish_message=AsyncMock())
with (
patch.object(
utils,
"_admitted_turn_envelope",
new=AsyncMock(return_value=root_envelope("t1")),
),
patch.object(utils, "release_turn", new=AsyncMock()),
patch(
"backend.copilot.stream_registry.create_session",
new=AsyncMock(side_effect=ConnectionError("redis down")),
),
patch("backend.copilot.stream_registry.delete_session_meta", new=AsyncMock()),
patch(
"backend.util.clients.get_async_copilot_queue",
new=AsyncMock(return_value=queue),
),
pytest.raises(ConnectionError),
):
await utils.dispatch_turn(
MagicMock(), session_id="s1", user_id="u1", turn_id="t1", message="hi"
)
queue.publish_message.assert_not_awaited()
@pytest.mark.asyncio
async def test_a_queue_client_that_cannot_connect_is_not_a_publish() -> None:
"""Getting the queue client happens before anything is published, so its
failure proves nothing was sent (Sentry, CodeRabbit on #16101)."""
with (
patch.object(
utils,
"_admitted_turn_envelope",
new=AsyncMock(return_value=root_envelope("t1")),
),
patch.object(utils, "release_turn", new=AsyncMock()),
patch("backend.copilot.stream_registry.create_session", new=AsyncMock()),
patch("backend.copilot.stream_registry.delete_session_meta", new=AsyncMock()),
patch(
"backend.util.clients.get_async_copilot_queue",
new=AsyncMock(side_effect=ConnectionError("broker unreachable")),
),
pytest.raises(ConnectionError),
):
await utils.dispatch_turn(
MagicMock(), session_id="s1", user_id="u1", turn_id="t1", message="hi"
)
@pytest.mark.asyncio
async def test_a_failed_publish_stays_unconfirmed_when_cleanup_fails_too() -> None:
"""Releasing the tree node and the slot is cleanup on the way out; if
those fail as well, the caller must still see TurnPublishUnconfirmed,
or a turn the broker accepted is treated as never sent (Astra review)."""
queue = MagicMock(publish_message=AsyncMock(side_effect=OSError("confirm lost")))
db = MagicMock()
db.admit_chat_session_turn = AsyncMock(return_value="admitted")
db.update_chat_session_status = AsyncMock(side_effect=ConnectionError("db blip"))
delete_meta = AsyncMock()
with (
patch(
"backend.copilot.turn_queue.count_inflight_turns",
new=AsyncMock(return_value=0),
),
patch("backend.copilot.active_turns.chat_db", return_value=db),
patch.object(
utils,
"_admitted_turn_envelope",
new=AsyncMock(return_value=root_envelope("t1")),
),
patch.object(
utils, "release_turn", new=AsyncMock(side_effect=ConnectionError("x"))
),
patch("backend.copilot.stream_registry.create_session", new=AsyncMock()),
patch("backend.copilot.stream_registry.delete_session_meta", new=delete_meta),
patch(
"backend.util.clients.get_async_copilot_queue",
new=AsyncMock(return_value=queue),
),
pytest.raises(utils.TurnPublishUnconfirmed),
):
await utils.schedule_turn(
session_id="s1",
user_id="u1",
turn_id="t1",
message="hi",
tool_call_id="delegate:parent",
tool_name="delegate_to_expert",
)
# The node release failing must not skip the rest of the cleanup.
delete_meta.assert_awaited_once()
@@ -389,9 +389,12 @@ async def _team_change(confirmation_id: str, call: _Call) -> _Found | None:
if preview.template_id
else None
)
# Confirming a hire's hand-off id also starts a task, so a held confirm
# names it: otherwise its card reads like the plain hire.
task = proposal.handoff if proposal.hand_off else None
return _Found(
kind=_TEAM_CHANGE_KINDS[preview.kind],
name=preview.name,
kind="New hire with a first task" if task else _TEAM_CHANGE_KINDS[preview.kind],
name=f"{preview.name} to take on {task.title}" if task else preview.name,
href=_team_href(proposal.expert_id) if proposal.expert_id else None,
description=preview.tagline or preview.about,
meta=[preview.job_title or preview.role, _area(template) if template else None],
@@ -246,6 +246,45 @@ async def test_a_pending_team_change_is_named_only_in_its_own_chat(
redis.delete.assert_not_called()
@pytest.mark.parametrize(
"hand_off, name, kind",
[
(
True,
"Ada to take on the Printify order pipeline",
"New hire with a first task",
),
(False, "Ada", "New hire"),
],
)
async def test_a_held_hire_names_the_task_its_confirm_would_start(hand_off, name, kind):
"""Confirming a hire's hand-off id also starts the task, so a held confirm
must say so; the hire-only id of the same hire reads as the plain hire
(#16101 review)."""
from backend.copilot.tools.expert_proposal import ExpertChangeProposal
from backend.copilot.tools.models import ExpertChangePreview, ExpertHandoffTask
session = _session()
proposal = ExpertChangeProposal(
user_id="user-1",
session_id=session.session_id,
preview=ExpertChangePreview(kind="hire", name="Ada"),
handoff=ExpertHandoffTask(
title="the Printify order pipeline", brief="Move Etsy orders."
),
hand_off=hand_off,
sibling_confirmation_id="c-2",
)
redis = MagicMock(get=AsyncMock(return_value=proposal.model_dump_json()))
with patch.object(references, "get_redis_async", AsyncMock(return_value=redis)):
[ref] = await resolve_references(
"confirm_expert_change", {"confirmation_id": "c-1"}, "user-1", session
)
assert ref.name == name
assert ref.kind == kind
async def test_a_held_hire_is_named_however_slow_the_whole_roster_is():
"""Prod timed out listing every template inside the budget and showed the
raw id instead of the name (2026-09-30); the template is read by id now."""
@@ -427,7 +427,8 @@ def get_team_building_supplement(
### Building the team
- You are the user's Head of AI. When the user wants help from a specialist, or recurring work has no owner, look at `<team_context>` first. Recommend one hired expert or roster template with a reason (`tool:hire_expert` for a template). Offer a custom expert (`tool:raise_expert`) only after that, or when nothing fits.
- One proposal at a time.
- Never hire silently. Both tools show the user an approval card; don't restate it. One short line, then wait. """
- Never hire silently. Both tools show the user an approval card; don't restate it. One short line, then wait.
- When the hire is for a task the user already described, pass it to `tool:hire_expert` as `handoff_title` and `handoff_task`: the card then hires and starts the task in one click. Never ask in text whether to hand a task to a new hire from `tool:hire_expert`. """
def get_expert_oversight_supplement(
@@ -207,6 +207,29 @@ class TestTeamBuildingSupplement:
assert "One proposal at a time" in result
assert "Never hire silently" in result
def test_a_hire_for_a_known_task_hands_it_off_on_the_card(self):
"""SECRT-3138: the approval is the hand-off, never a question in text."""
result = prompting.get_team_building_supplement(
experts_enabled=True, expert_id=None
)
assert "handoff_title" in result
assert "handoff_task" in result
assert (
"Never ask in text whether to hand a task to a new hire from "
"`tool:hire_expert`." in result
)
def test_a_raised_expert_is_not_barred_from_being_asked_about_a_task(self):
"""A raise_expert hire has no hand-off card, so Otto may still ask
before delegating to it ("Never delegate silently"). The rule that
forbids asking in text names only tool:hire_expert hires (#16101)."""
result = prompting.get_team_building_supplement(
experts_enabled=True, expert_id=None
)
assert "Never ask in text whether to hand a task to a new hire." not in result
def test_an_expert_session_is_not_told_to_hire(self):
result = prompting.get_team_building_supplement(
experts_enabled=True, expert_id="expert-a"
@@ -28,7 +28,11 @@ from typing import TYPE_CHECKING, Any, Literal
from backend.copilot import stream_registry
from backend.copilot.active_turns import ConcurrentTurnLimitError
from backend.copilot.config import CopilotLlmAuthProvider
from backend.copilot.executor.utils import queue_spawned_turn, schedule_turn
from backend.copilot.executor.utils import (
TurnPublishUnconfirmed,
queue_spawned_turn,
schedule_turn,
)
from backend.copilot.llm_attribution import is_turn_impersonated
from backend.copilot.model import (
CHAT_STATUS_QUEUED,
@@ -72,6 +76,20 @@ _ALREADY_WAITING = (
)
class TurnNotDispatched(Exception):
"""The turn provably never reached the executor queue.
Raised for a failure before the publish (the session lookup, the status
checks, the slot or tree admission, the stream registry write), each of
which leaves nothing for a worker to pick up, and for a wait for a task
slot that raised with nothing stored (``_nothing_queued``). Only this lets
a caller remove the thread it opened or send the task again; a publish
that raised is :class:`TurnPublishUnconfirmed` instead, and any other
error from this module leaves the turn's state unknown. The original
error is the ``__cause__``.
"""
@dataclass
class SessionResult:
"""Aggregated result from a copilot session turn observed via
@@ -243,15 +261,20 @@ async def run_copilot_turn_via_queue(
what happened next"; no separate deferred-state branch needed in
``run_sub_session`` / ``AutoPilotBlock``.
"""
session = await get_chat_session(session_id, user_id)
try:
session = await get_chat_session(session_id, user_id)
already_waiting = not allow_queue and (
await chat_db().get_chat_session_status(session_id) == CHAT_STATUS_QUEUED
)
in_flight = not already_waiting and await is_turn_in_flight(session_id)
except Exception as e:
raise TurnNotDispatched(f"The checks before the dispatch failed: {e}") from e
if session is None:
raise RuntimeError("copilot_session_not_found")
raise TurnNotDispatched("copilot_session_not_found")
if not allow_queue and (
await chat_db().get_chat_session_status(session_id) == CHAT_STATUS_QUEUED
):
if already_waiting:
return "refused", SessionResult(refusal=_ALREADY_WAITING)
if await is_turn_in_flight(session_id):
if in_flight:
if not allow_queue:
return "refused", SessionResult(
refusal=(
@@ -316,6 +339,8 @@ async def run_copilot_turn_via_queue(
)
except TreeRefusal as refused:
return "refused", SessionResult(refusal=refused.message)
except TurnPublishUnconfirmed:
raise
except ConcurrentTurnLimitError:
if delegated:
return await _queue_for_slot(
@@ -344,6 +369,14 @@ async def run_copilot_turn_via_queue(
tool_name,
)
return "rejected_concurrent_turn_cap", SessionResult()
except Exception as e:
raise TurnNotDispatched(f"The turn could not be dispatched: {e}") from e
if timeout <= 0:
# The turn is queued and the caller waits for nothing. Subscribing
# anyway could only fail, and a failure here, after the dispatch,
# would read as "not started" to a caller that then runs the same
# task a second time.
return "running", SessionResult()
return await wait_for_session_result(
session_id=session_id,
user_id=user_id,
@@ -351,6 +384,23 @@ async def run_copilot_turn_via_queue(
)
async def _nothing_queued(session_id: str) -> bool:
"""Whether a queue attempt that raised left nothing that can run. Unlike
the broker, the queue is these rows: a turn waits only as the session's
pending message with the session ``queued``, so a thread with no message
that is not queued holds nothing. A thread with history can't be read
this way, and a read that fails proves nothing."""
try:
db = chat_db()
return (
await db.get_chat_session_status(session_id) != CHAT_STATUS_QUEUED
and await db.get_next_sequence(session_id) == 0
)
except Exception:
logger.warning("Could not check a failed queue attempt", exc_info=True)
return False
async def wait_for_queued_session(
*, session_id: str, user_id: str, timeout: float
) -> tuple[SessionOutcome, SessionResult]:
@@ -416,6 +466,14 @@ async def _queue_for_slot(
f"inflight cap (tool={tool_name})"
)
return "rejected_concurrent_turn_cap", SessionResult()
except Exception as e:
if await _nothing_queued(session_id):
raise TurnNotDispatched(f"The turn could not be queued: {e}") from e
raise
if timeout <= 0:
# Stored, so it runs when a slot frees; as after a dispatch, a read
# with nothing to wait for could only fail and read as "not queued".
return "queued_for_slot", SessionResult(queued=True)
return await wait_for_queued_session(
session_id=session_id, user_id=user_id, timeout=timeout
)
@@ -18,6 +18,8 @@ from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from backend.copilot import active_turns
from backend.copilot.active_turns import ConcurrentTurnLimitError
from backend.copilot.executor.utils import TurnPublishUnconfirmed
from backend.copilot.response_model import (
StreamFinish,
StreamStart,
@@ -26,6 +28,7 @@ from backend.copilot.response_model import (
)
from backend.copilot.sdk.session_waiter import (
SessionResult,
TurnNotDispatched,
run_copilot_turn_via_queue,
wait_for_session_result,
)
@@ -619,3 +622,215 @@ async def test_a_queued_child_turn_carries_its_parents_impersonation(
)
assert queued.await_args.kwargs["impersonated"] is True
@pytest.mark.asyncio
async def test_a_dispatched_turn_with_no_wait_returns_without_reading_its_stream():
"""``timeout == 0`` after a dispatch has nothing to wait for. Reading the
stream anyway is pure risk: the turn is already queued, so a Redis error
there would read as "not started" and a caller retrying on it would run
the same task twice (SECRT-3138 review)."""
scheduled = AsyncMock()
stream_read = AsyncMock(side_effect=RuntimeError("redis subscribe failed"))
with (
patch(
"backend.copilot.sdk.session_waiter.is_turn_in_flight",
new=AsyncMock(return_value=False),
),
patch("backend.copilot.sdk.session_waiter.schedule_turn", new=scheduled),
patch(
"backend.copilot.sdk.session_waiter.wait_for_session_result",
new=stream_read,
),
):
outcome, result = await run_copilot_turn_via_queue(
session_id="sess-fire-and-forget",
user_id="u1",
message="own the Printify pipeline",
timeout=0,
tool_call_id="delegate:parent",
tool_name="delegate_to_expert",
)
assert outcome == "running"
assert result == SessionResult()
scheduled.assert_awaited_once()
stream_read.assert_not_awaited()
@pytest.mark.asyncio
async def test_a_turn_queued_for_a_slot_with_no_wait_returns_without_reading_it(
mock_session_lookup,
):
"""Once queue_spawned_turn has stored the turn it will run when a slot
frees, so with nothing to wait for a status read can only fail, and a
failure there would read as "not queued" (SECRT-3138 review)."""
mock_session_lookup.return_value.metadata.delegated_by_session_id = "parent"
queued = AsyncMock()
status_read = AsyncMock(side_effect=RuntimeError("db blip"))
with (
patch(
"backend.copilot.sdk.session_waiter.is_turn_in_flight",
new=AsyncMock(return_value=False),
),
patch(
"backend.copilot.sdk.session_waiter.schedule_turn",
new=AsyncMock(side_effect=ConcurrentTurnLimitError("full")),
),
patch("backend.copilot.sdk.session_waiter.queue_spawned_turn", new=queued),
patch(
"backend.copilot.sdk.session_waiter.wait_for_queued_session",
new=status_read,
),
):
outcome, result = await run_copilot_turn_via_queue(
session_id="sess-slot",
user_id="u1",
message="own the Printify pipeline",
timeout=0,
tool_call_id="delegate:parent",
tool_name="delegate_to_expert",
)
assert outcome == "queued_for_slot"
assert result.queued is True
queued.assert_awaited_once()
status_read.assert_not_awaited()
@pytest.mark.asyncio
async def test_a_failure_before_the_dispatch_is_typed_as_not_dispatched():
"""Callers may remove a thread or send again only when the turn provably
never reached the queue, so that case is typed; a publish that raised
stays TurnPublishUnconfirmed (Sentry on #16101)."""
with (
patch(
"backend.copilot.sdk.session_waiter.is_turn_in_flight",
new=AsyncMock(return_value=False),
),
patch(
"backend.copilot.sdk.session_waiter.schedule_turn",
new=AsyncMock(side_effect=ConnectionError("redis down")),
),
pytest.raises(TurnNotDispatched) as raised,
):
await run_copilot_turn_via_queue(
session_id="s",
user_id="u1",
message="m",
timeout=0,
tool_call_id="delegate:p",
tool_name="delegate_to_expert",
)
assert isinstance(raised.value.__cause__, ConnectionError)
# Callers that report str(exc) (the AutoPilot block) keep the reason.
assert "redis down" in str(raised.value)
@pytest.mark.asyncio
async def test_a_failed_status_check_is_typed_as_not_dispatched():
with (
patch(
"backend.copilot.sdk.session_waiter.is_turn_in_flight",
new=AsyncMock(side_effect=ConnectionError("redis down")),
),
pytest.raises(TurnNotDispatched, match="redis down"),
):
await run_copilot_turn_via_queue(
session_id="s",
user_id="u1",
message="m",
timeout=0,
tool_call_id="delegate:p",
tool_name="delegate_to_expert",
)
@pytest.mark.asyncio
async def test_a_publish_that_raised_is_not_called_undispatched():
with (
patch(
"backend.copilot.sdk.session_waiter.is_turn_in_flight",
new=AsyncMock(return_value=False),
),
patch(
"backend.copilot.sdk.session_waiter.schedule_turn",
new=AsyncMock(side_effect=TurnPublishUnconfirmed("confirm lost")),
),
pytest.raises(TurnPublishUnconfirmed),
):
await run_copilot_turn_via_queue(
session_id="s",
user_id="u1",
message="m",
timeout=0,
tool_call_id="delegate:p",
tool_name="delegate_to_expert",
)
async def _queue_that_raised(chat):
with (
patch(
"backend.copilot.sdk.session_waiter.is_turn_in_flight",
new=AsyncMock(return_value=False),
),
patch(
"backend.copilot.sdk.session_waiter.schedule_turn",
new=AsyncMock(side_effect=ConcurrentTurnLimitError("full")),
),
patch(
"backend.copilot.sdk.session_waiter.queue_spawned_turn",
new=AsyncMock(side_effect=ConnectionError("redis down")),
),
patch("backend.copilot.sdk.session_waiter.chat_db", return_value=chat),
):
await run_copilot_turn_via_queue(
session_id="sess-slot",
user_id="u1",
message="own the Printify pipeline",
timeout=0,
tool_call_id="delegate:parent",
tool_name="delegate_to_expert",
)
def _chat(*, status="idle", next_sequence=0, read_error=None):
chat = MagicMock()
chat.get_chat_session_status = AsyncMock(return_value=status)
chat.get_next_sequence = AsyncMock(
return_value=next_sequence, side_effect=read_error
)
return chat
@pytest.mark.asyncio
async def test_a_wait_for_a_slot_that_stored_nothing_is_not_dispatched(
mock_session_lookup,
):
"""The queue is the session's rows, so a queue attempt that raised and
left a fresh thread with no message, not queued, holds nothing that can
run (Sentry on #16101)."""
mock_session_lookup.return_value.metadata.delegated_by_session_id = "parent"
with pytest.raises(TurnNotDispatched) as raised:
await _queue_that_raised(_chat())
assert isinstance(raised.value.__cause__, ConnectionError)
# Callers that report str(exc) (the AutoPilot block) keep the reason.
assert "redis down" in str(raised.value)
@pytest.mark.asyncio
@pytest.mark.parametrize(
"chat",
[
pytest.param(_chat(next_sequence=1), id="a-message-was-stored"),
pytest.param(_chat(status="queued"), id="the-session-is-queued"),
pytest.param(_chat(read_error=RuntimeError("db")), id="the-check-failed"),
],
)
async def test_a_wait_for_a_slot_that_may_have_stored_the_turn_stays_untyped(
mock_session_lookup, chat
):
mock_session_lookup.return_value.metadata.delegated_by_session_id = "parent"
with pytest.raises(ConnectionError):
await _queue_that_raised(chat)
@@ -3,7 +3,8 @@
Step 2 of the confirm-gated team-change flow. One confirm tool serves both
kinds: it takes nothing but the id, so the applied change is exactly what the
user saw. The id is single-use and bound to the Otto session that
produced it.
produced it. A hire offered with its task also starts that task when the id
confirmed is the "hire and start" one (see ``hire_handoff``).
"""
from typing import Any
@@ -17,7 +18,8 @@ from .expert_proposal import (
autopilot_session_guard,
load_bound_proposal,
)
from .models import ErrorResponse, ToolResponseBase
from .hire_handoff import hand_off_to_new_hire, with_hire_only
from .models import ErrorResponse, ExpertChangeAppliedResponse, ToolResponseBase
_PROPOSAL_FIELDS = (
"template_id",
@@ -102,4 +104,10 @@ class ConfirmExpertChangeTool(BaseTool):
)
if isinstance(proposal, ErrorResponse):
return proposal
return await apply_proposal(user_id, session_id, proposal)
applied = await apply_proposal(user_id, session_id, proposal)
task = proposal.handoff
if task is None or not isinstance(applied, ExpertChangeAppliedResponse):
return applied
if not proposal.hand_off:
return with_hire_only(applied, task)
return await hand_off_to_new_hire(user_id, session, applied, task)
@@ -41,9 +41,13 @@ from backend.copilot.model import (
ChatSession,
child_session_origin,
create_chat_session,
delete_chat_session,
get_chat_session,
)
from backend.copilot.sdk.session_waiter import run_copilot_turn_via_queue
from backend.copilot.sdk.session_waiter import (
TurnNotDispatched,
run_copilot_turn_via_queue,
)
from backend.copilot.tree import SpawnRequest
from backend.data.db_accessors import experts_db
@@ -185,78 +189,18 @@ class DelegateToExpertTool(BaseTool):
if refusal is not None:
return self._error(refusal, session)
inner_session_id = await self._resolve_session(
return await start_delegation(
user_id=user_id,
session=session,
target=target,
prompt=prompt,
system_context=system_context,
delegated_session_id=delegated_session_id.strip(),
)
if isinstance(inner_session_id, ErrorResponse):
return inner_session_id
caller = await self._caller_name(user_id, session.expert_id)
started_at = time.monotonic()
outcome, result = await run_copilot_turn_via_queue(
session_id=inner_session_id,
user_id=user_id,
message=_handoff_message(caller, system_context, prompt),
timeout=max(0, min(wait_for_result, MAX_SUB_SESSION_WAIT_SECONDS)),
permissions=get_current_permissions(),
tool_call_id=(
f"delegate:{session.session_id}" if session.session_id else "delegate"
),
tool_name="delegate_to_expert",
# A teammate keeps their own teammates; depth still bounds the chain.
spawn=SpawnRequest(may_spawn=True),
allow_queue=False,
message_metadata=sent_from_metadata(session, caller),
)
elapsed = time.monotonic() - started_at
discarded = (
not delegated_session_id.strip()
and await discard_unused_sub_session(inner_session_id, user_id, outcome)
)
if discarded:
# The thread is gone; handing back its id would send the model to
# poll a delegation that no longer exists.
return self._error(
(
spawn_limit_message()
if outcome == "rejected_concurrent_turn_cap"
else result.refusal or f"{target.name} could not start this task."
),
session,
)
workspace_files = (
await list_sub_workspace_files(user_id, inner_session_id)
if outcome == "completed"
else None
)
delegated = response_from_outcome(
outcome=outcome,
result=result,
inner_session_id=inner_session_id,
parent_session_id=session.session_id,
elapsed=elapsed,
workspace_files=workspace_files,
actor=target.name,
)
delegated.message += await build_spawn_state_note()
return apply_delegated_expert(
delegated,
DelegatedExpertInfo(
id=target.id,
name=target.name,
role=target.role,
avatar_url=target.avatar_url,
color=target.color,
),
wait_for_result=wait_for_result,
)
def _error(self, message: str, session: ChatSession) -> ErrorResponse:
return ErrorResponse(
message=message, session_id=session.session_id
).from_outside()
return _error(message, session)
async def _load_delegate_target(
self, user_id: str, target_id: str, session: ChatSession
@@ -274,85 +218,246 @@ class DelegateToExpertTool(BaseTool):
await unknown_target_message(user_id, target_id, session.expert_id),
session,
)
if target.schedules_paused_at is not None:
return self._error(
f"{target.name} is paused (budget guardrail or archive) and "
"cannot take delegated work until the user resumes them.",
session,
)
if (paused := paused_refusal(target)) is not None:
return self._error(paused, session)
return target
async def _resolve_session(
self,
*,
user_id: str,
session: ChatSession,
target: Expert,
delegated_session_id: str,
) -> str | ErrorResponse:
"""Reuse a prior delegation thread with this teammate, or open one.
Resuming is restricted to threads this session itself delegated, so a
session can never read or steer another scope's conversation by
guessing an id.
"""
if not delegated_session_id:
new_session = await create_chat_session(
user_id,
dry_run=session.dry_run,
llm_auth_provider=session.metadata.llm_auth_provider,
llm_credential_id=session.metadata.llm_credential_id,
expert_id=target.id,
delegated_by_expert_id=session.expert_id,
delegated_by_session_id=session.session_id,
origin=child_session_origin(session.metadata),
# Decided here, from the expert row already in hand, so the
# thread cannot pick up a looser default written in between.
inherited_autopilot_mode=child_autopilot_mode(
session, target.autopilot_mode
),
)
return new_session.session_id
def paused_refusal(target: Expert) -> str | None:
"""Why a paused teammate can't take delegated work, or ``None``.
prior = await get_chat_session(delegated_session_id)
if (
prior is None
or prior.user_id != user_id
or prior.expert_id != target.id
or prior.metadata.delegated_by_session_id != session.session_id
or prior.metadata.handed_off_from_expert_id is not None
):
return self._error(
f"delegated_session_id {delegated_session_id} is not a "
f"delegation you started with {target.name}. Leave it empty "
"to open a fresh one.",
session,
)
if (
prior.metadata.llm_auth_provider != session.metadata.llm_auth_provider
or prior.metadata.llm_credential_id != session.metadata.llm_credential_id
):
return self._error(
f"That delegation thread with {target.name} runs on a "
"different model connection than this chat does, so it "
"cannot be resumed from here. Leave delegated_session_id "
"empty to open a fresh one.",
session,
)
return delegated_session_id
Shared with ``hire_handoff``: re-hiring a template returns the existing
hire unchanged, so a "new" hire can be one the budget guardrail paused.
"""
if target.schedules_paused_at is None:
return None
return (
f"{target.name} is paused (budget guardrail or archive) and "
"cannot take delegated work until the user resumes them."
)
async def _caller_name(self, user_id: str, caller_expert_id: str | None) -> str:
"""Who to introduce the hand-off as. Plain sessions are Otto."""
if caller_expert_id is None:
return "Otto"
try:
caller = await experts_db().get_expert(
user_id, caller_expert_id, include_workflows=False
)
except Exception as e:
logger.warning(f"Delegating expert lookup failed: {e}")
return "a teammate"
return caller.name if caller else "a teammate"
async def start_delegation(
*,
user_id: str,
session: ChatSession,
target: Expert,
prompt: str,
system_context: str = "",
delegated_session_id: str = "",
wait_for_result: int = 60,
) -> ToolResponseBase:
"""Run *prompt* as *target* in a thread delegated from *session*.
The part of ``delegate_to_expert`` after its checks. The caller has
already resolved *target* and cleared ``chain_refusal``.
"""
inner_session_id = await _resolve_session(
user_id=user_id,
session=session,
target=target,
delegated_session_id=delegated_session_id,
)
if isinstance(inner_session_id, ErrorResponse):
return inner_session_id
return await run_delegated_turn(
user_id=user_id,
session=session,
target=target,
inner_session_id=inner_session_id,
prompt=prompt,
system_context=system_context,
opened_here=not delegated_session_id,
wait_for_result=wait_for_result,
)
async def run_delegated_turn(
*,
user_id: str,
session: ChatSession,
target: Expert,
inner_session_id: str,
prompt: str,
system_context: str = "",
opened_here: bool,
wait_for_result: int,
) -> ToolResponseBase:
"""Queue *prompt* on a delegation thread and report what happened.
Shared with ``hire_handoff``, which opens the thread itself first
(``open_delegation_thread``) so it always knows where a task went, so a
task handed to a new hire runs exactly like any other delegation: same
provenance, permissions, mode and result contract. ``opened_here``
drops a thread this call opened when its turn never reached it
(``TurnNotDispatched`` is re-raised after).
"""
caller = await _caller_name(user_id, session.expert_id)
started_at = time.monotonic()
try:
outcome, result = await run_copilot_turn_via_queue(
session_id=inner_session_id,
user_id=user_id,
message=_handoff_message(caller, system_context, prompt),
timeout=max(0, min(wait_for_result, MAX_SUB_SESSION_WAIT_SECONDS)),
permissions=get_current_permissions(),
tool_call_id=(
f"delegate:{session.session_id}" if session.session_id else "delegate"
),
tool_name="delegate_to_expert",
# A teammate keeps their own teammates; depth still bounds the chain.
spawn=SpawnRequest(may_spawn=True),
allow_queue=False,
message_metadata=sent_from_metadata(session, caller),
)
except TurnNotDispatched:
# Nothing reached the queue, so a thread opened for this task is
# empty. Any other error, or a cancellation (which can land inside
# the publish), may have left the task in it, so the thread stays.
if opened_here:
await _discard_thread(inner_session_id, user_id)
raise
elapsed = time.monotonic() - started_at
discarded = opened_here and await discard_unused_sub_session(
inner_session_id, user_id, outcome
)
if discarded:
# The thread is gone; handing back its id would send the model to
# poll a delegation that no longer exists.
return _error(
(
spawn_limit_message()
if outcome == "rejected_concurrent_turn_cap"
else result.refusal or f"{target.name} could not start this task."
),
session,
)
workspace_files = (
await list_sub_workspace_files(user_id, inner_session_id)
if outcome == "completed"
else None
)
delegated = response_from_outcome(
outcome=outcome,
result=result,
inner_session_id=inner_session_id,
parent_session_id=session.session_id,
elapsed=elapsed,
workspace_files=workspace_files,
actor=target.name,
)
delegated.message += await _spawn_note()
return apply_delegated_expert(
delegated,
DelegatedExpertInfo(
id=target.id,
name=target.name,
role=target.role,
avatar_url=target.avatar_url,
color=target.color,
),
)
async def _spawn_note() -> str:
"""The tree's state for the result, or nothing: it is read after the task
is queued, so failing here must not report that task as failed."""
try:
return await build_spawn_state_note()
except Exception:
logger.warning("Could not read the spawn state note", exc_info=True)
return ""
async def _discard_thread(inner_session_id: str, user_id: str) -> None:
"""Remove a delegation thread nothing was sent to. Best-effort, like
``discard_unused_sub_session``: an empty thread is the lesser harm."""
try:
await delete_chat_session(inner_session_id, user_id)
except Exception:
logger.warning("Failed to remove an unused delegation thread", exc_info=True)
def _error(message: str, session: ChatSession) -> ErrorResponse:
return ErrorResponse(message=message, session_id=session.session_id).from_outside()
async def _resolve_session(
*,
user_id: str,
session: ChatSession,
target: Expert,
delegated_session_id: str,
) -> str | ErrorResponse:
"""Reuse a prior delegation thread with this teammate, or open one.
Resuming is restricted to threads this session itself delegated, so a
session can never read or steer another scope's conversation by
guessing an id.
"""
if not delegated_session_id:
return await open_delegation_thread(user_id, session, target)
prior = await get_chat_session(delegated_session_id)
if (
prior is None
or prior.user_id != user_id
or prior.expert_id != target.id
or prior.metadata.delegated_by_session_id != session.session_id
or prior.metadata.handed_off_from_expert_id is not None
):
return _error(
f"delegated_session_id {delegated_session_id} is not a "
f"delegation you started with {target.name}. Leave it empty "
"to open a fresh one.",
session,
)
if (
prior.metadata.llm_auth_provider != session.metadata.llm_auth_provider
or prior.metadata.llm_credential_id != session.metadata.llm_credential_id
):
return _error(
f"That delegation thread with {target.name} runs on a "
"different model connection than this chat does, so it "
"cannot be resumed from here. Leave delegated_session_id "
"empty to open a fresh one.",
session,
)
return delegated_session_id
async def open_delegation_thread(
user_id: str, session: ChatSession, target: Expert
) -> str:
"""Open a fresh thread for *target*, delegated from *session*."""
new_session = await create_chat_session(
user_id,
dry_run=session.dry_run,
llm_auth_provider=session.metadata.llm_auth_provider,
llm_credential_id=session.metadata.llm_credential_id,
expert_id=target.id,
delegated_by_expert_id=session.expert_id,
delegated_by_session_id=session.session_id,
origin=child_session_origin(session.metadata),
# Decided here, from the expert row already in hand, so the
# thread cannot pick up a looser default written in between.
inherited_autopilot_mode=child_autopilot_mode(session, target.autopilot_mode),
)
return new_session.session_id
async def _caller_name(user_id: str, caller_expert_id: str | None) -> str:
"""Who to introduce the hand-off as. Plain sessions are Otto."""
if caller_expert_id is None:
return "Otto"
try:
caller = await experts_db().get_expert(
user_id, caller_expert_id, include_workflows=False
)
except Exception as e:
logger.warning(f"Delegating expert lookup failed: {e}")
return "a teammate"
return caller.name if caller else "a teammate"
def _handoff_message(caller: str, system_context: str, prompt: str) -> str:
@@ -9,12 +9,14 @@ sub is created in, and who may poll it afterwards.
from __future__ import annotations
import asyncio
from unittest.mock import AsyncMock, MagicMock
import pytest
from backend.copilot.active_turns import spawn_limit_message
from backend.copilot.sdk.session_waiter import SessionResult
from backend.copilot.executor.utils import TurnPublishUnconfirmed
from backend.copilot.sdk.session_waiter import SessionResult, TurnNotDispatched
from .delegate_to_expert import DelegateToExpertTool
from .expert_delegation import CALLER_NAME_LIMIT
@@ -1042,3 +1044,97 @@ class TestRefusedDelegationCleanup:
)
deleted.assert_not_awaited()
class TestAfterTheDispatch:
@pytest.mark.asyncio
async def test_a_failed_spawn_note_does_not_undo_a_dispatched_task(
self, roster, mock_turn, mock_sessions, monkeypatch
):
"""The tree note is decoration on a task already queued. Raising there
would report a running delegation as a failure, and a caller that
retries on it would run the task twice (SECRT-3138 review)."""
monkeypatch.setattr(
"backend.copilot.tools.delegate_to_expert.build_spawn_state_note",
AsyncMock(side_effect=RuntimeError("redis blip")),
)
result = await DelegateToExpertTool()._execute(
user_id="alice",
session=_session(),
expert_id="expert-b",
prompt="hi",
wait_for_result=0,
)
assert isinstance(result, SubSessionStatusResponse)
assert result.status == "running"
assert result.sub_session_id == mock_sessions[0].session_id
class TestASendThatFailed:
"""A fresh delegation thread is removed only when the turn provably never
reached the queue (``TurnNotDispatched``). A cancellation, the user
pressing Stop, can land inside the publish, so like a publish that raised
it may have delivered the task, and that thread stays (Sentry)."""
def _failed_send(self, monkeypatch, error: BaseException):
monkeypatch.setattr(
"backend.copilot.tools.delegate_to_expert.run_copilot_turn_via_queue",
AsyncMock(side_effect=error),
)
deleted = AsyncMock()
monkeypatch.setattr(
"backend.copilot.tools.delegate_to_expert.delete_chat_session", deleted
)
return deleted
async def _delegate(self, **kwargs):
await DelegateToExpertTool()._execute(
user_id="alice",
session=_session(session_id="s1"),
expert_id="expert-b",
prompt="hi",
wait_for_result=0,
**kwargs,
)
@pytest.mark.asyncio
async def test_a_thread_nothing_reached_is_removed(
self, roster, mock_sessions, monkeypatch
):
deleted = self._failed_send(monkeypatch, TurnNotDispatched("redis down"))
with pytest.raises(TurnNotDispatched):
await self._delegate()
deleted.assert_awaited_once_with(mock_sessions[0].session_id, "alice")
@pytest.mark.asyncio
@pytest.mark.parametrize(
"error",
[
pytest.param(asyncio.CancelledError(), id="stopped-mid-send"),
pytest.param(TurnPublishUnconfirmed("confirm lost"), id="publish"),
pytest.param(RuntimeError("unknown"), id="unknown"),
],
)
async def test_a_thread_the_task_may_have_reached_is_kept(
self, roster, mock_sessions, monkeypatch, error
):
deleted = self._failed_send(monkeypatch, error)
with pytest.raises(type(error)):
await self._delegate()
assert len(mock_sessions) == 1
deleted.assert_not_awaited()
@pytest.mark.asyncio
async def test_a_thread_it_did_not_open_is_kept(
self, roster, mock_turn, mock_sessions, monkeypatch
):
await self._delegate()
mock_turn.side_effect = TurnNotDispatched("redis down")
deleted = AsyncMock()
monkeypatch.setattr(
"backend.copilot.tools.delegate_to_expert.delete_chat_session", deleted
)
with pytest.raises(TurnNotDispatched):
await self._delegate(delegated_session_id="inner-1")
deleted.assert_not_awaited()
@@ -154,7 +154,12 @@ async def test_describe_expert_asks_only_for_what_the_id_does_not_carry(team):
assert isinstance(result, CapabilityDetailsResponse)
assert "calls hire_expert" in result.message
assert set(result.parameters["properties"]) == {"name"}
# The task a hire is for is not carried by the id either (SECRT-3138).
assert set(result.parameters["properties"]) == {
"name",
"handoff_title",
"handoff_task",
}
assert result.parameters["required"] == []
@@ -39,6 +39,7 @@ from .models import (
ExpertChangeAppliedResponse,
ExpertChangeKind,
ExpertChangePreview,
ExpertHandoffTask,
ExpertSummary,
ToolResponseBase,
)
@@ -52,6 +53,9 @@ _PROPOSAL_KEY_PREFIX = "copilot:expert_change_proposal:"
# answered with "already done" instead of the same message an expired
# preview gets.
_CONSUMED_KEY_PREFIX = "copilot:expert_change_consumed:"
# Claimed by whichever of a hire's two ids (the plain hire / hire and start)
# is confirmed first, so one approval can never hire twice.
_CLAIM_KEY_PREFIX = "copilot:expert_change_claim:"
_LOG_ID_PREFIX_LENGTH = 12
_PREVIEW_TOOLS = "tool:hire_expert, tool:raise_expert or tool:update_expert"
@@ -87,6 +91,12 @@ class ExpertChangeProposal(BaseModel):
# gate refuses it rather than reading it as "answered at -1", which any
# session with one sequenced user message would clear.
user_turn_watermark: int | None = None
# Only for a hire proposed for a task already in the chat (see
# ``hire_expert``): the task, whether confirming this id hands it over,
# and the id of the other choice for the same hire.
handoff: ExpertHandoffTask | None = None
hand_off: bool = False
sibling_confirmation_id: str | None = None
def proposal_key(confirmation_id: str) -> str:
@@ -97,6 +107,10 @@ def _consumed_key(confirmation_id: str) -> str:
return f"{_CONSUMED_KEY_PREFIX}{confirmation_id}"
def _claim_key(confirmation_id: str, sibling_confirmation_id: str) -> str:
return f"{_CLAIM_KEY_PREFIX}{min(confirmation_id, sibling_confirmation_id)}"
def _stale_preview_error(session_id: str) -> ErrorResponse:
return ErrorResponse(
message=(
@@ -292,15 +306,51 @@ async def load_bound_proposal(
if user_turn_watermark(session) <= proposal.user_turn_watermark:
return _unapproved_preview_error(session.session_id)
sibling = proposal.sibling_confirmation_id
if sibling is None:
return await _consume(redis, confirmation_id, session.session_id, proposal)
# One hire parked two ways (the plain hire / "hire and start"): a SET NX
# on a key both ids share decides which answer takes it. The other id's
# proposal is left in place and refused by this claim, so nothing is
# destroyed before this confirm is sure to complete; a failure releases
# the claim and the user's answer stays answerable. One shared key rather
# than a multi-key DEL, because the two proposal keys can sit on
# different cluster slots.
claim = _claim_key(confirmation_id, sibling)
if not await redis.set(claim, confirmation_id, nx=True, ex=PROPOSAL_TTL_SECONDS):
return _already_confirmed_error(session.session_id)
try:
consumed = await _consume(redis, confirmation_id, session.session_id, proposal)
except BaseException:
await _release_claim(redis, claim)
raise
if isinstance(consumed, ErrorResponse):
await _release_claim(redis, claim)
return consumed
async def _consume(
redis: AsyncRedisClient,
confirmation_id: str,
session_id: str,
proposal: ExpertChangeProposal,
) -> ExpertChangeProposal | ErrorResponse:
# GET and DEL are intentionally separate: binding must be checked before
# consumption so a mismatched caller cannot invalidate a legitimate
# proposal.
if await redis.delete(key) == 0:
return _stale_preview_error(session.session_id)
if await redis.delete(proposal_key(confirmation_id)) == 0:
return _stale_preview_error(session_id)
await redis.setex(_consumed_key(confirmation_id), PROPOSAL_TTL_SECONDS, "1")
return proposal
async def _release_claim(redis: AsyncRedisClient, claim: str) -> None:
try:
await redis.delete(claim)
except Exception:
logger.warning("Could not release a hire's confirm claim", exc_info=True)
async def apply_proposal(
user_id: str,
session_id: str,
@@ -344,14 +394,14 @@ async def _apply_hire(
except Exception as e:
return _hire_failure_response(e, session_id)
return ExpertChangeAppliedResponse(
message=_hire_message(result.expert.name),
message=hire_message(result.expert.name),
session_id=session_id,
kind="hire",
expert=_summary(result.expert),
)
def _hire_message(name: str) -> str:
def hire_message(name: str) -> str:
return (
f"{name} is hired and on the team; their workflows and skills finish "
"installing in the background. Tell the user who joined and what they own."
@@ -25,7 +25,12 @@ from typing import Any
from backend.copilot import stream_registry
from backend.copilot.executor.utils import enqueue_cancel_task
from backend.copilot.legacy_first_turn_memory import without_stored_first_turn_memory
from backend.copilot.model import CHAT_STATUS_QUEUED, ChatSession, get_chat_session
from backend.copilot.model import (
CHAT_STATUS_IDLE,
CHAT_STATUS_QUEUED,
ChatSession,
get_chat_session,
)
from backend.copilot.sdk.session_waiter import (
SessionOutcome,
SessionResult,
@@ -214,6 +219,13 @@ class GetSubSessionResultTool(BaseTool):
refusal = None if turn_in_flight else queued_turn_refusal(sub)
outcome: SessionOutcome
result: SessionResult
if registry_session is None and _nothing_reached(sub):
return apply_delegated_expert(
_nothing_reached_response(
sub, inner_session_id, session.session_id, actor
),
delegate,
)
if queued:
outcome, result = await wait_for_queued_session(
session_id=inner_session_id, user_id=user_id, timeout=effective_wait
@@ -280,6 +292,37 @@ class GetSubSessionResultTool(BaseTool):
)
def _nothing_reached(sub: ChatSession) -> bool:
"""An idle thread with no messages and no stream meta: no turn was ever
dispatched to it, or the dispatch failed at the publish (the meta is
removed then). A dispatched turn always has meta until it ends, and an
ended one has messages."""
return sub.chat_status == CHAT_STATUS_IDLE and not sub.messages
def _nothing_reached_response(
sub: ChatSession, inner_session_id: str, parent_session_id: str | None, actor: str
) -> SubSessionStatusResponse:
"""Polling a thread nothing reached would otherwise read "running"
forever. From here a task the broker never got cannot be told from one it
still holds (a lost confirm, a stopped executor), so it is "unconfirmed"
at any age and never advice to send again, which could run it twice."""
return SubSessionStatusResponse(
message=(
f"Nothing has reached {actor}'s thread, and from here it can't be "
"confirmed whether the task was delivered: it may still arrive. "
"Don't send it again on this basis. Tell the user it may not have "
"started and point them to the thread."
),
session_id=parent_session_id,
status="unconfirmed",
sub_session_id=inner_session_id,
sub_autopilot_session_id=inner_session_id,
sub_autopilot_session_link=_sub_session_link(inner_session_id),
elapsed_seconds=0.0,
).from_outside()
def _in_caller_scope(sub: ChatSession, session: ChatSession) -> bool:
"""Whether *session* is allowed to poll *sub*.
@@ -3,6 +3,13 @@
Step 1 of the confirm-gated hire flow: resolve the template, check the team
has room, and park the exact proposal under a one-time ``confirmation_id``.
``confirm_expert_change`` applies it once the user approves.
A hire proposed for a task the user already described carries that task, and
is parked twice: the plain hire under ``confirmation_id`` and "hire and
start" under ``handoff_confirmation_id``. The card asks one question, Hire
and start or Not now, so the user's one click both hires and starts the work
(see ``hire_handoff``); the plain id stays for a user who asks in chat to
hire without starting the task.
"""
import uuid
@@ -27,14 +34,28 @@ from .models import (
ErrorResponse,
ExpertChangePreview,
ExpertChangeProposedResponse,
ExpertHandoffTask,
ToolResponseBase,
)
# Long enough to name a task the way the user would, short enough to sit in
# a button beside the expert's name.
HANDOFF_TITLE_MAX_LENGTH = 80
HANDOFF_TASK_MAX_LENGTH = 4000
class _HireParams(BaseModel):
"""The rename field, bounded the same way as raise_expert's ``name``."""
"""The rename field, bounded the same way as raise_expert's ``name``,
and the task the hire is for."""
name: str = Field(default="", max_length=EXPERT_NAME_MAX_LENGTH)
handoff_title: str = Field(default="", max_length=HANDOFF_TITLE_MAX_LENGTH)
handoff_task: str = Field(default="", max_length=HANDOFF_TASK_MAX_LENGTH)
def handoff(self) -> ExpertHandoffTask | None:
if not self.handoff_task:
return None
return ExpertHandoffTask(title=self.handoff_title, brief=self.handoff_task)
class HireExpertTool(BaseTool):
@@ -53,11 +74,11 @@ class HireExpertTool(BaseTool):
return (
"Propose hiring a ready-made expert from the roster. Never "
"writes: returns who would join plus a one-time confirmation_id. "
"The user sees who would join on a card with Approve and Decline "
"buttons, so do not describe them in text — one short line at "
"most, then wait. Only after they approve, call "
"tool:confirm_expert_change with that id. Use tool:raise_expert when no "
"template fits."
"The user answers on a card, so do not describe it in text — one "
"short line at most, then wait. Only after they approve, call "
"tool:confirm_expert_change with the id they approved. For a task "
"already in the chat, pass handoff_title + handoff_task. Use "
"tool:raise_expert when no template fits."
)
@property
@@ -73,6 +94,14 @@ class HireExpertTool(BaseTool):
"type": "string",
"description": "Rename the hire; omit to keep the template name.",
},
"handoff_title": {
"type": "string",
"description": "That task in a few words, for the card.",
},
"handoff_task": {
"type": "string",
"description": "The task, written for the expert.",
},
},
"required": ["template_id"],
}
@@ -84,6 +113,8 @@ class HireExpertTool(BaseTool):
*,
template_id: str = "",
name: str = "",
handoff_title: str = "",
handoff_task: str = "",
**kwargs,
) -> ToolResponseBase:
session_id = session.session_id
@@ -96,13 +127,9 @@ class HireExpertTool(BaseTool):
message="template_id is required.", session_id=session_id
)
try:
params = _HireParams(name=" ".join(name.split()))
except ValidationError as e:
return ErrorResponse(
message=f"Invalid expert name — {_validation_detail(e)}",
session_id=session_id,
)
params = _hire_params(name, handoff_title, handoff_task)
if isinstance(params, str):
return ErrorResponse(message=params, session_id=session_id)
template = await _find_template(template_id.strip())
if template is None:
@@ -116,43 +143,102 @@ class HireExpertTool(BaseTool):
if error := await capacity_error(user_id, session_id, "hire"):
return error
preview = ExpertChangePreview(
kind="hire",
name=params.name or template.name,
role=template.role,
tagline=template.tagline or "",
about=template.identity,
boundaries=template.boundaries,
voice_preferences=template.voice_preferences,
template_id=template.id,
avatar_url=template.avatar_url,
color=template.color,
)
confirmation_id = str(uuid.uuid4())
await store_proposal(
await get_redis_async(),
confirmation_id,
preview = _preview(template, params.name)
handoff = params.handoff()
confirmation_id, handoff_confirmation_id = await _park(
ExpertChangeProposal(
user_id=user_id,
session_id=session_id,
preview=preview,
user_turn_watermark=user_turn_watermark(session),
),
handoff=handoff,
)
)
return ExpertChangeProposedResponse(
message=(
"Nothing hired yet. The user is looking at this expert on a "
"card with Approve and Decline buttons — do not describe "
"them in text. Reply with one short line at most and wait. "
"Only after they explicitly approve, call "
"tool:confirm_expert_change with this confirmation_id."
),
message=_proposed_message(handoff is not None),
session_id=session_id,
preview=preview,
confirmation_id=confirmation_id,
handoff=handoff,
handoff_confirmation_id=handoff_confirmation_id,
)
def _hire_params(name: str, handoff_title: str, handoff_task: str) -> _HireParams | str:
"""The validated hire arguments, or why they are refused."""
try:
params = _HireParams(
name=" ".join(name.split()),
handoff_title=" ".join(handoff_title.split()),
handoff_task=handoff_task.strip(),
)
except ValidationError as e:
return f"Invalid hire — {_validation_detail(e)}"
if bool(params.handoff_title) != bool(params.handoff_task):
return "Pass handoff_title and handoff_task together, or neither."
return params
def _preview(template: Expert, name: str) -> ExpertChangePreview:
return ExpertChangePreview(
kind="hire",
name=name or template.name,
role=template.role,
tagline=template.tagline or "",
about=template.identity,
boundaries=template.boundaries,
voice_preferences=template.voice_preferences,
template_id=template.id,
avatar_url=template.avatar_url,
color=template.color,
)
async def _park(proposal: ExpertChangeProposal) -> tuple[str, str | None]:
"""Park *proposal* under a fresh id and, when it carries a task, its
"hire and start" twin under a second; returns both ids."""
redis = await get_redis_async()
confirmation_id = str(uuid.uuid4())
if proposal.handoff is None:
await store_proposal(redis, confirmation_id, proposal)
return confirmation_id, None
handoff_confirmation_id = str(uuid.uuid4())
await store_proposal(
redis,
confirmation_id,
proposal.model_copy(
update={"sibling_confirmation_id": handoff_confirmation_id}
),
)
await store_proposal(
redis,
handoff_confirmation_id,
proposal.model_copy(
update={"hand_off": True, "sibling_confirmation_id": confirmation_id}
),
)
return confirmation_id, handoff_confirmation_id
def _proposed_message(with_handoff: bool) -> str:
if not with_handoff:
return (
"Nothing hired yet. The user is looking at this expert on a "
"card with Approve and Decline buttons — do not describe "
"them in text. Reply with one short line at most and wait. "
"Only after they explicitly approve, call "
"tool:confirm_expert_change with this confirmation_id."
)
return (
"Nothing hired yet. The user is looking at this expert on a card "
"with two buttons: hire and start the task, or not now — do not "
"describe them in text. Reply with one short line at most and wait. "
"If they approve, call tool:confirm_expert_change with "
"handoff_confirmation_id. Use confirmation_id only if they ask to "
"hire without starting the task."
)
async def _find_template(template_id: str) -> Expert | None:
templates = await experts_db().list_templates()
return next((t for t in templates if t.id == template_id), None)
@@ -0,0 +1,210 @@
"""Start the task a new hire was approved for, in the same confirm.
A hire Otto proposes for a task the user already described carries that task
(``hire_expert``'s ``handoff_title`` / ``handoff_task``), and its card asks
Hire and start or Not now. Without this the approval only hired:
the model then asked in prose whether to hand the task over, and the user had
to type a second answer to something they had just approved (SECRT-3138).
Confirming the hand-off id hires and then runs the task as a normal
``delegate_to_expert`` from this chat (``open_delegation_thread`` and
``run_delegated_turn``), so the thread, its permissions, mode and result
contract are exactly a delegation's, and a paused teammate is refused the
same way. The user's click on the card is the approval for both; the hand-off
is not put to them again.
The hire stands whatever happens to the task, and what is reported is only
what is known: started, queued for a task slot, refused (no retry: the
refusal says why), or, when the send raised, whether it provably never left.
``TurnNotDispatched`` means nothing reached the queue: ``run_delegated_turn``
has removed the empty thread and a retry is safe, through one path only: the
card's Try again (``handoff.retry``), never Otto re-sending it as well. Any other error, a publish
that raised included, may have delivered the task, so it is reported
unconfirmed with its thread and never sent a second time.
"""
import logging
from typing import Literal
from backend.api.features.experts.models import Expert
from backend.copilot.model import ChatSession
from backend.copilot.sdk.session_waiter import TurnNotDispatched
from .delegate_to_expert import (
open_delegation_thread,
paused_refusal,
run_delegated_turn,
)
from .expert_delegation import chain_refusal, resolve_target_expert
from .expert_proposal import hire_message
from .models import (
ExpertChangeAppliedResponse,
ExpertHandoffOutcome,
ExpertHandoffTask,
SubSessionStatusResponse,
)
from .run_sub_session import _sub_session_link
logger = logging.getLogger(__name__)
# With nothing waited for, "running" only means the turn is on the queue: a
# worker can still refuse it at pickup, so the hand-off claims "sent", never
# "started".
_SENT = frozenset({"running", "completed"})
class _NotStarted(Exception):
"""The task did not reach the new hire; carries the model-facing reason
and whether trying the delegation again can help."""
def __init__(self, reason: str, *, retry: bool) -> None:
super().__init__(reason)
self.reason = reason
self.retry = retry
async def hand_off_to_new_hire(
user_id: str,
session: ChatSession,
applied: ExpertChangeAppliedResponse,
task: ExpertHandoffTask,
) -> ExpertChangeAppliedResponse:
"""Start *task* with the expert *applied* just hired."""
expert = applied.expert
try:
target = await _target(user_id, session, expert.id)
thread = await open_delegation_thread(user_id, session, target)
except _NotStarted as e:
return _report(applied, task, "failed", None, e.reason, e.retry)
except Exception:
logger.warning("Hand-off to a new hire could not open", exc_info=True)
return _report(applied, task, "failed", None, "It could not be sent.", True)
try:
sent = await run_delegated_turn(
user_id=user_id,
session=session,
target=target,
inner_session_id=thread,
prompt=task.brief,
opened_here=True,
wait_for_result=0,
)
except TurnNotDispatched:
logger.warning("Hand-off to a new hire was not sent", exc_info=True)
return _report(applied, task, "failed", None, "It could not be sent.", True)
except Exception:
logger.warning("Hand-off to a new hire is unconfirmed", exc_info=True)
return _report(applied, task, "unconfirmed", thread)
if not isinstance(sent, SubSessionStatusResponse) or sent.status == "error":
# A refusal (tree budget or depth, a full set of task slots) says
# itself what has to change first.
return _report(applied, task, "failed", None, sent.message, False)
status = "sent" if sent.status in _SENT else "queued"
return _report(applied, task, status, sent.sub_session_id)
def with_hire_only(
applied: ExpertChangeAppliedResponse, task: ExpertHandoffTask
) -> ExpertChangeAppliedResponse:
"""The plain hire id was confirmed for a hire proposed with a task: the
user asked in chat to hire without starting it (the card itself offers
only Hire and start or Not now)."""
return applied.model_copy(
update={
"message": (
f"{hire_message(applied.expert.name)} The user chose hire only, "
f"without handing over {task.title}: don't hand it over, and "
"don't ask about it unless they bring it up."
)
}
)
async def _target(user_id: str, session: ChatSession, expert_id: str) -> Expert:
"""The hire, if it can take delegated work; ``delegate_to_expert``'s rules."""
target = await resolve_target_expert(user_id, expert_id)
if target is None or target.is_archived:
raise _NotStarted("They are not on the team right now.", retry=False)
if (paused := paused_refusal(target)) is not None:
raise _NotStarted(paused, retry=False)
if (refusal := await chain_refusal(user_id, session, target)) is not None:
raise _NotStarted(refusal, retry=False)
return target
def _report(
applied: ExpertChangeAppliedResponse,
task: ExpertHandoffTask,
status: Literal["sent", "queued", "unconfirmed", "failed"],
thread: str | None,
reason: str = "",
retry: bool = False,
) -> ExpertChangeAppliedResponse:
expert = applied.expert
link = _sub_session_link(thread)
if status == "failed":
message = _failed_message(expert.name, task, reason, retry)
else:
message = _sent_message(status, expert.name, task, thread or "", link)
return applied.model_copy(
update={
"message": message,
"handoff": ExpertHandoffOutcome(
title=task.title,
status=status,
sub_session_id=thread,
link=link,
retry=status == "failed" and retry,
),
}
)
def _sent_message(
status: Literal["sent", "queued", "unconfirmed"],
name: str,
task: ExpertHandoffTask,
thread: str,
link: str | None,
) -> str:
where = f"their own chat at {link}"
if status == "unconfirmed":
return (
f"{name} is hired. {task.title} was sent to {where}, but whether it "
"started could not be confirmed. Check it with "
f"tool:get_sub_session_result (sub_session_id: {thread}) before "
"anything else, and don't hand it over again: it may already be "
"running."
)
state = (
f"{task.title} was sent to {where}"
if status == "sent"
else f"{task.title} is waiting in {where} for one of the user's task "
"slots to free"
)
return (
f"{name} is hired and {state}. Tell the user in one short line that "
f"it is with {name} and where to follow along. Don't ask whether to hand "
"it over: they chose that on the card. Nothing sends their result back "
"to this chat: read it with tool:get_sub_session_result "
f"(sub_session_id: {thread}) when you need it. Their workflows and "
"skills finish installing in the background."
)
def _failed_message(
name: str, task: ExpertHandoffTask, reason: str, retry: bool
) -> str:
lead = f"{name} is hired and on the team, but {task.title} did not reach them: "
if not retry:
return (
f"{lead}{reason} Tell the user why in one short line, and don't ask "
"them again whether to hand it over."
)
# One retry path: the card's Try again. Otto re-sending it too could
# start the task twice.
return (
f"{lead}{reason} Nothing reached them, so it can be sent again: tell "
"the user in one short line that it didn't start and that Try again on "
"the card sends it. Don't hand it over yourself, so it can't start twice."
)
@@ -0,0 +1,534 @@
"""A hire proposed for a task the user already described hires AND hands it
off on one approval (SECRT-3138).
The principle under test, end to end through the real tools: ``hire_expert``
with a task parks two one-time ids, "hire and start" (what the card's Hire
and start sends) and the plain hire (for a user who asks in chat to hire
without starting the task); confirming the first hires exactly the previewed
template and starts the task in a thread run by the new expert, and says
where; confirming the second only hires. Either one burns the other, so the
user's single answer can never hire twice. Only the queue, the session store
and the roster are faked.
"""
import json
from contextlib import contextmanager
from datetime import UTC, datetime
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from backend.copilot.executor.utils import TurnPublishUnconfirmed
from backend.copilot.model import ChatMessage
from backend.copilot.sdk.session_waiter import SessionResult, TurnNotDispatched
from ._test_data import make_session
from .confirm_expert_change import ConfirmExpertChangeTool
from .expert_proposal import ExpertChangeProposal, proposal_key
from .hire_expert import HireExpertTool
from .models import (
ErrorResponse,
ExpertChangeAppliedResponse,
ExpertChangePreview,
ExpertChangeProposedResponse,
)
_USER = "user-hire-handoff"
_TITLE = "the Printify order pipeline"
_BRIEF = (
"Own the Printify order pipeline: move new Etsy orders into Printify and "
"sync tracking numbers back to Etsy every morning."
)
class _FakeRedis:
"""The setex/get/delete/set-NX surface the proposal store uses."""
def __init__(self):
self.store: dict[str, str] = {}
self.fail_next_setex = False
async def setex(self, key, ttl, value):
if self.fail_next_setex:
self.fail_next_setex = False
raise ConnectionError("redis blip")
self.store[key] = value
async def set(self, key, value, *, nx=False, ex=None):
if nx and key in self.store:
return None
self.store[key] = value
return True
async def get(self, key):
value = self.store.get(key)
return value.encode() if value is not None else None
async def delete(self, *keys):
return sum(1 for key in keys if self.store.pop(key, None) is not None)
def _template():
return SimpleNamespace(
id="tpl-james",
name="James",
role="Operations",
tagline="Keeps orders, stock and suppliers moving.",
identity="You run operations.",
boundaries="You never pay a supplier yourself.",
voice_preferences="Plain and brief.",
avatar_url=None,
color="amber-300",
)
def _hired(*, paused: bool = False):
return SimpleNamespace(
id="exp-james",
name="James",
role="Operations",
tagline="Keeps orders, stock and suppliers moving.",
identity="You run operations.",
boundaries="You never pay a supplier yourself.",
voice_preferences="Plain and brief.",
weekly_budget=None,
avatar_url=None,
color="amber-300",
is_archived=False,
schedules_paused_at=datetime(2026, 10, 1, tzinfo=UTC) if paused else None,
autopilot_mode=None,
)
@contextmanager
def _env(
*,
turn_outcome: str = "running",
turn_error: Exception | None = None,
paused: bool = False,
refusal: str = "",
):
hired = _hired(paused=paused)
db = MagicMock()
db.list_templates = AsyncMock(return_value=[_template()])
db.count_active_experts = AsyncMock(return_value=0)
db.hire_expert = AsyncMock(return_value=SimpleNamespace(expert=hired))
db.get_expert = AsyncMock(return_value=hired)
db.list_experts = AsyncMock(return_value=[hired])
fake_redis = _FakeRedis()
redis = AsyncMock(return_value=fake_redis)
turn = AsyncMock(
side_effect=turn_error,
return_value=(turn_outcome, SessionResult(refusal=refusal)),
)
created = []
deleted = AsyncMock()
async def create_session(user_id, **kwargs):
created.append(kwargs)
return SimpleNamespace(session_id=f"thread-{len(created)}")
with (
patch("backend.copilot.tools.hire_expert.experts_db", return_value=db),
patch("backend.copilot.tools.expert_proposal.experts_db", return_value=db),
patch("backend.copilot.tools.expert_delegation.experts_db", return_value=db),
patch("backend.copilot.tools.hire_expert.get_redis_async", redis),
patch("backend.copilot.tools.confirm_expert_change.get_redis_async", redis),
patch(
"backend.copilot.tools.delegate_to_expert.create_chat_session",
create_session,
),
patch(
"backend.copilot.tools.delegate_to_expert.run_copilot_turn_via_queue", turn
),
patch(
"backend.copilot.tools.delegate_to_expert.discard_unused_sub_session",
AsyncMock(
return_value=turn_outcome in ("rejected_concurrent_turn_cap", "refused")
),
),
patch(
"backend.copilot.tools.delegate_to_expert.build_spawn_state_note",
AsyncMock(return_value=""),
),
patch("backend.copilot.tools.delegate_to_expert.delete_chat_session", deleted),
):
yield SimpleNamespace(
db=db, turn=turn, created=created, deleted=deleted, redis=fake_redis
)
def _approve(session):
session.messages.append(
ChatMessage(role="user", content="approved", sequence=len(session.messages))
)
return session
async def _hire(session, **kwargs):
return await HireExpertTool()._execute(user_id=_USER, session=session, **kwargs)
async def _confirm(session, confirmation_id):
return await ConfirmExpertChangeTool()._execute(
user_id=_USER, session=session, confirmation_id=confirmation_id
)
async def _propose_with_task(session) -> ExpertChangeProposedResponse:
preview = await _hire(
session,
template_id="tpl-james",
handoff_title=_TITLE,
handoff_task=_BRIEF,
)
assert isinstance(preview, ExpertChangeProposedResponse)
return preview
class TestHireAndHandOffInOneApproval:
@pytest.mark.asyncio(loop_scope="session")
async def test_a_hire_for_a_task_parks_the_hand_off_beside_the_plain_hire(self):
with _env() as env:
preview = await _propose_with_task(make_session(_USER))
assert preview.handoff is not None
assert preview.handoff.title == _TITLE
assert preview.handoff.brief == _BRIEF
assert preview.handoff_confirmation_id
assert preview.handoff_confirmation_id != preview.confirmation_id
env.db.hire_expert.assert_not_called()
env.turn.assert_not_called()
@pytest.mark.asyncio(loop_scope="session")
async def test_otto_is_told_the_card_has_two_buttons(self):
"""The card asks Hire and start or Not now. Telling Otto about a
Hire only button it doesn't have could make him confirm the plain id
on a free-text yes, and the task would never start (Sentry, #16101)."""
with _env():
preview = await _propose_with_task(make_session(_USER))
message = preview.message.lower()
assert "two buttons: hire and start the task, or not now" in message
assert "hire only" not in message
assert "three buttons" not in message
# An approval starts the task; the plain id only on an explicit ask.
assert "if they approve, call tool:confirm_expert_change with " in message
assert "handoff_confirmation_id" in message
assert "only if they ask to hire without starting the task" in message
@pytest.mark.asyncio(loop_scope="session")
async def test_approving_the_hand_off_hires_and_starts_the_task(self):
with _env() as env:
session = make_session(_USER)
preview = await _propose_with_task(session)
assert preview.handoff_confirmation_id
resp = await _confirm(_approve(session), preview.handoff_confirmation_id)
assert isinstance(resp, ExpertChangeAppliedResponse)
env.db.hire_expert.assert_awaited_once_with(
_USER, "tpl-james", "James", "copilot"
)
# The task went to the new hire, in a thread this chat delegated.
assert env.created == [
{
"dry_run": False,
"llm_auth_provider": "platform",
"llm_credential_id": None,
"expert_id": "exp-james",
"delegated_by_expert_id": None,
"delegated_by_session_id": session.session_id,
"origin": "interactive",
"inherited_autopilot_mode": env.created[0]["inherited_autopilot_mode"],
}
]
turn = env.turn.await_args.kwargs
assert turn["session_id"] == "thread-1"
assert _BRIEF in turn["message"]
assert turn["timeout"] == 0
# The card and Otto both learn where the work is happening.
assert resp.handoff is not None
# Only queued so far: a worker can still refuse it at pickup, so
# nothing may claim it started (Sentry).
assert resp.handoff.status == "sent"
assert resp.handoff.retry is False
assert "started" not in resp.message
assert resp.handoff.title == _TITLE
assert resp.handoff.sub_session_id == "thread-1"
assert resp.handoff.link == "/copilot?sessionId=thread-1"
assert "/copilot?sessionId=thread-1" in resp.message
assert "don't ask whether to hand it over" in resp.message.lower()
# Nothing pushes a delegated result back to this chat: Otto reads it
# with get_sub_session_result, and must not wait for it (Pwuts).
assert "comes back here" not in resp.message
assert "Nothing sends their result back to this chat" in resp.message
assert "tool:get_sub_session_result (sub_session_id: thread-1)" in (
resp.message
)
@pytest.mark.asyncio(loop_scope="session")
async def test_hire_only_hires_without_handing_anything_off(self):
with _env() as env:
session = make_session(_USER)
preview = await _propose_with_task(session)
resp = await _confirm(_approve(session), preview.confirmation_id)
assert isinstance(resp, ExpertChangeAppliedResponse)
env.db.hire_expert.assert_awaited_once()
env.turn.assert_not_called()
assert resp.handoff is None
assert "hire only" in resp.message.lower()
assert _TITLE in resp.message
@pytest.mark.asyncio(loop_scope="session")
async def test_a_confirm_that_broke_midway_leaves_the_hire_answerable(self):
"""A Redis error after the shared claim must not leave that claim
behind, or the user's real answer is refused as already done when
nothing was hired (Sentry)."""
with _env() as env:
session = make_session(_USER)
preview = await _propose_with_task(session)
assert preview.handoff_confirmation_id
env.redis.fail_next_setex = True
with pytest.raises(ConnectionError):
await _confirm(_approve(session), preview.confirmation_id)
env.db.hire_expert.assert_not_called()
resp = await _confirm(_approve(session), preview.handoff_confirmation_id)
assert isinstance(resp, ExpertChangeAppliedResponse)
env.db.hire_expert.assert_awaited_once()
@pytest.mark.asyncio(loop_scope="session")
async def test_one_answer_never_hires_twice(self):
"""Both ids name the same hire, so confirming one burns the other."""
with _env() as env:
session = make_session(_USER)
preview = await _propose_with_task(session)
assert preview.handoff_confirmation_id
first = await _confirm(_approve(session), preview.confirmation_id)
second = await _confirm(_approve(session), preview.handoff_confirmation_id)
assert isinstance(first, ExpertChangeAppliedResponse)
assert isinstance(second, ErrorResponse)
assert "already confirmed" in second.message
assert env.db.hire_expert.await_count == 1
env.turn.assert_not_called()
@pytest.mark.asyncio(loop_scope="session")
async def test_a_hand_off_that_cannot_start_still_reports_the_hire(self):
with _env(turn_outcome="rejected_concurrent_turn_cap") as env:
session = make_session(_USER)
preview = await _propose_with_task(session)
assert preview.handoff_confirmation_id
resp = await _confirm(_approve(session), preview.handoff_confirmation_id)
assert isinstance(resp, ExpertChangeAppliedResponse)
env.db.hire_expert.assert_awaited_once()
assert resp.handoff is not None
assert resp.handoff.status == "failed"
assert resp.handoff.link is None
# A full set of task slots is no reason to ask again, and no reason
# to retry at once either (Sentry, third review).
assert "don't ask them again" in resp.message
assert "delegate_to_expert" not in resp.message
# The card offers Try again only when the backend says it can help.
assert resp.handoff.retry is False
class TestHandOffOnlyClaimsWhatHappened:
"""Review findings on the first cut (Capy, Sentry): the card and Otto
must never be told "started" or "retry" when that is not what happened."""
@pytest.mark.asyncio(loop_scope="session")
async def test_a_paused_expert_is_not_handed_the_task(self):
"""Re-hiring a template returns the existing hire unchanged, paused
or not; a paused teammate takes no delegated work, here as in
delegate_to_expert."""
with _env(paused=True) as env:
session = make_session(_USER)
preview = await _propose_with_task(session)
assert preview.handoff_confirmation_id
resp = await _confirm(_approve(session), preview.handoff_confirmation_id)
assert isinstance(resp, ExpertChangeAppliedResponse)
env.turn.assert_not_called()
assert env.created == []
assert resp.handoff is not None
assert resp.handoff.status == "failed"
assert "paused" in resp.message
# Nothing to retry until the user resumes them.
assert "delegate_to_expert" not in resp.message
assert resp.handoff.retry is False
@pytest.mark.asyncio(loop_scope="session")
async def test_a_task_waiting_for_a_slot_is_queued_not_started(self):
with _env(turn_outcome="queued_for_slot") as env:
session = make_session(_USER)
preview = await _propose_with_task(session)
assert preview.handoff_confirmation_id
resp = await _confirm(_approve(session), preview.handoff_confirmation_id)
assert isinstance(resp, ExpertChangeAppliedResponse)
env.turn.assert_awaited_once()
assert resp.handoff is not None
assert resp.handoff.status == "queued"
assert resp.handoff.link == "/copilot?sessionId=thread-1"
assert "started" not in resp.message
@pytest.mark.asyncio(loop_scope="session")
async def test_a_tree_refusal_is_not_retried(self):
with _env(turn_outcome="refused", refusal="This tree is out of budget.") as env:
session = make_session(_USER)
preview = await _propose_with_task(session)
assert preview.handoff_confirmation_id
resp = await _confirm(_approve(session), preview.handoff_confirmation_id)
assert isinstance(resp, ExpertChangeAppliedResponse)
assert resp.handoff is not None
assert resp.handoff.status == "failed"
assert "out of budget" in resp.message
assert "delegate_to_expert" not in resp.message
assert resp.handoff.retry is False
env.turn.assert_awaited_once()
@pytest.mark.asyncio(loop_scope="session")
@pytest.mark.parametrize(
"error",
[
pytest.param(TurnPublishUnconfirmed("confirm lost"), id="publish"),
pytest.param(RuntimeError("unknown"), id="unknown"),
],
)
async def test_a_send_that_may_have_landed_is_reported_unconfirmed(self, error):
"""The broker may hold a turn whose publish raised, and an error not
typed as before the dispatch proves nothing either, so the task may be
running: point at the thread, never ask for a second delegation that
would run the task twice (Sentry, CodeRabbit)."""
with _env(turn_error=error) as env:
session = make_session(_USER)
preview = await _propose_with_task(session)
assert preview.handoff_confirmation_id
resp = await _confirm(_approve(session), preview.handoff_confirmation_id)
assert isinstance(resp, ExpertChangeAppliedResponse)
assert len(env.created) == 1
assert resp.handoff is not None
assert resp.handoff.status == "unconfirmed"
# It may be running: sending it again could run it twice.
assert resp.handoff.retry is False
assert resp.handoff.sub_session_id == "thread-1"
assert resp.handoff.link == "/copilot?sessionId=thread-1"
assert "thread-1" in resp.message
assert "delegate_to_expert" not in resp.message
env.deleted.assert_not_called()
@pytest.mark.asyncio(loop_scope="session")
async def test_an_error_before_the_dispatch_is_a_retryable_failure(self):
"""A failure typed as before the dispatch proves nothing reached the
queue: the empty thread goes, and a retry is safe."""
with _env(turn_error=TurnNotDispatched("redis blip")) as env:
session = make_session(_USER)
preview = await _propose_with_task(session)
assert preview.handoff_confirmation_id
resp = await _confirm(_approve(session), preview.handoff_confirmation_id)
assert isinstance(resp, ExpertChangeAppliedResponse)
assert resp.handoff is not None
assert resp.handoff.status == "failed"
assert resp.handoff.link is None
assert resp.handoff.retry is True
# One retry path, the card's Try again: Otto re-sending it as well
# could start the task twice (kcze).
assert "Try again on the card sends it" in resp.message
assert "Don't hand it over yourself" in resp.message
assert "delegate_to_expert" not in resp.message
env.deleted.assert_awaited_once_with("thread-1", _USER)
class TestHandOffTaskValidation:
@pytest.mark.asyncio(loop_scope="session")
async def test_a_task_without_a_title_is_refused(self):
with _env():
resp = await _hire(
make_session(_USER), template_id="tpl-james", handoff_task=_BRIEF
)
assert isinstance(resp, ErrorResponse)
assert "handoff_title" in resp.message
@pytest.mark.asyncio(loop_scope="session")
async def test_a_title_without_a_task_is_refused(self):
with _env():
resp = await _hire(
make_session(_USER), template_id="tpl-james", handoff_title=_TITLE
)
assert isinstance(resp, ErrorResponse)
assert "handoff_task" in resp.message
@pytest.mark.asyncio(loop_scope="session")
async def test_the_title_is_one_line_so_it_fits_the_button(self):
with _env():
resp = await _hire(
make_session(_USER),
template_id="tpl-james",
handoff_title=" the Printify\n\norder pipeline ",
handoff_task=_BRIEF,
)
assert isinstance(resp, ExpertChangeProposedResponse)
assert resp.handoff is not None
assert resp.handoff.title == "the Printify order pipeline"
@pytest.mark.asyncio(loop_scope="session")
async def test_an_overlong_title_is_refused(self):
with _env():
resp = await _hire(
make_session(_USER),
template_id="tpl-james",
handoff_title="x" * 200,
handoff_task=_BRIEF,
)
assert isinstance(resp, ErrorResponse)
@pytest.mark.asyncio(loop_scope="session")
async def test_a_hire_without_a_task_offers_one_id_as_before(self):
with _env():
resp = await _hire(make_session(_USER), template_id="tpl-james")
assert isinstance(resp, ExpertChangeProposedResponse)
assert resp.handoff is None
assert resp.handoff_confirmation_id is None
class TestProposalsParkedBeforeTheDeploy:
@pytest.mark.asyncio(loop_scope="session")
async def test_an_old_hire_proposal_still_confirms_as_a_plain_hire(self):
"""A proposal parked by the previous build has none of the hand-off
fields; it must still confirm, as the hire it always was."""
redis = _FakeRedis()
session = make_session(_USER)
redis.store[proposal_key("old-id")] = json.dumps(
{
"user_id": _USER,
"session_id": session.session_id,
"preview": {
"kind": "hire",
"name": "James",
"template_id": "tpl-james",
},
"expert_id": None,
"expected_soul": None,
"user_turn_watermark": -1,
}
)
assert ExpertChangeProposal.model_validate_json(
redis.store[proposal_key("old-id")]
).preview == ExpertChangePreview(
kind="hire", name="James", template_id="tpl-james"
)
with _env() as env, patch(
"backend.copilot.tools.confirm_expert_change.get_redis_async",
AsyncMock(return_value=redis),
):
resp = await _confirm(_approve(session), "old-id")
assert isinstance(resp, ExpertChangeAppliedResponse)
assert resp.handoff is None
env.db.hire_expert.assert_awaited_once()
env.turn.assert_not_called()
@@ -471,7 +471,13 @@ class SubSessionStatusResponse(ToolResponseBase):
type: ResponseType = ResponseType.MCP_TOOL_OUTPUT
status: Literal[
"running", "completed", "cancelled", "error", "queued", "transferred"
"running",
"completed",
"cancelled",
"error",
"queued",
"transferred",
"unconfirmed",
] = Field(
description=(
"Current state of the child session. ``queued`` means the "
@@ -480,7 +486,9 @@ class SubSessionStatusResponse(ToolResponseBase):
"existing turn on its next drain. ``transferred`` is terminal "
"for the caller: ``handoff_to_expert`` gave the task away, so no "
"result is coming back and there is nothing to poll — the "
"receiving expert now owns it and reports to the user directly."
"receiving expert now owns it and reports to the user directly. "
"``unconfirmed``: nothing has reached the thread and delivery "
"can't be confirmed; it must not be sent again on that basis."
),
)
sub_session_id: str = Field(
@@ -652,6 +660,38 @@ class ExpertChangePreview(BaseModel):
color: str = ""
class ExpertHandoffTask(BaseModel):
"""The task a hire was proposed for, handed over when the user approves.
``title`` is how the user would name it and labels the card's primary
button; ``brief`` is the task written for the expert, who cannot see the
chat it came from.
"""
title: str
brief: str
class ExpertHandoffOutcome(BaseModel):
"""Whether the task a hire was approved for reached the new expert.
``link`` opens the expert's thread, where the work is happening; it is
unset when the task did not start.
"""
title: str
# ``sent``: on the queue for the expert's thread (a worker picks it up).
# ``queued``: waiting for one of the user's task slots. ``unconfirmed``:
# the send raised after it may have reached the broker.
status: Literal["sent", "queued", "unconfirmed", "failed"]
sub_session_id: str | None = None
link: str | None = None
# Only for ``failed``: nothing reached the expert and sending it again can
# help. A refusal (paused, out of budget or depth, no free task slot)
# says itself what has to change first, so it is never retryable.
retry: bool = False
class ExpertSummary(BaseModel):
"""The expert ``confirm_expert_change`` created — same charter fields as
the preview, so the card can show the whole thing after the fact."""
@@ -811,6 +851,12 @@ class ExpertChangeProposedResponse(ToolResponseBase):
applied: bool = False
preview: ExpertChangePreview
confirmation_id: str
# Set when the hire was proposed for a task already in the chat: the card
# then asks Hire and start (``handoff_confirmation_id``) or Not now; the
# plain ``confirmation_id`` is for a user who asks in chat to hire without
# starting the task.
handoff: ExpertHandoffTask | None = None
handoff_confirmation_id: str | None = None
class ExpertChangeAppliedResponse(ToolResponseBase):
@@ -821,6 +867,8 @@ class ExpertChangeAppliedResponse(ToolResponseBase):
kind: ExpertChangeKind
expert: ExpertSummary
failed_workflows: list[str] = Field(default_factory=list)
# Only for a hire approved together with its task.
handoff: ExpertHandoffOutcome | None = None
# Agent generation models
@@ -11,6 +11,7 @@ tests patch the three integration seams — ``enqueue_copilot_turn``,
from __future__ import annotations
import json
from datetime import UTC, datetime, timedelta
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
@@ -1419,3 +1420,55 @@ class TestRefusedSpawnCleanup:
)
deleted.assert_not_awaited()
class TestAThreadNothingReached:
"""A send that raised at the publish leaves an empty, idle thread with no
stream meta. Polling it must say so, not report "running" forever: the
hand-off tells Otto to check this thread before sending again (Sentry)."""
def _empty_thread(self, monkeypatch, age: timedelta):
sub = MagicMock(user_id="alice", expert_id=None, messages=[])
sub.chat_status = "idle"
sub.started_at = datetime.now(UTC) - age
sub.metadata.delegated_by_session_id = None
async def fake_get(_sid):
return sub
async def no_stream(_sid):
return None
monkeypatch.setattr(
"backend.copilot.tools.get_sub_session_result.get_chat_session",
fake_get,
)
monkeypatch.setattr(
"backend.copilot.tools.get_sub_session_result.stream_registry.get_session",
no_stream,
)
@pytest.mark.parametrize(
"age", [timedelta(seconds=20), timedelta(minutes=10), timedelta(days=2)]
)
@pytest.mark.asyncio
async def test_an_empty_thread_is_unconfirmed_and_never_resent(
self, monkeypatch, mock_waiter, age
):
"""From an empty thread alone a task the broker never got cannot be
told from one it still holds (a lost confirm, a stopped executor), so
whatever its age it is "unconfirmed", never "safe to send again", and
never "queued", whose live view turns an idle session into
"completed" (CodeRabbit)."""
self._empty_thread(monkeypatch, age)
r = await GetSubSessionResultTool()._execute(
user_id="alice",
session=_session("alice"),
sub_session_id="inner-7",
wait_if_running=30,
)
assert isinstance(r, SubSessionStatusResponse)
assert r.status == "unconfirmed"
assert "Don't send it again" in r.message
assert "safe to send" not in r.message
mock_waiter.result_mock.assert_not_awaited()
@@ -279,10 +279,22 @@ from ._test_data import make_session
# re-measured on that merged tree (89 tools), with the stack's margin:
# merged tree 77,301 (89 tools)
# + headroom +300 77,601
# Bumped 77_601 -> 77_720 for hire_expert's `handoff_title` and `handoff_task`
# (SECRT-3138): a hire proposed for a task already in the chat carries it, so
# the card's Approve hires and hands it off in one click. The rule itself
# lives in the team-building prompt, so the schema only names the fields and
# the rest of hire_expert's description was cut to pay for them; no tool was
# added. Measured on this branch, which is dev f0ae7dfd4 plus this PR:
# dev f0ae7dfd4 77,501 (89 tools)
# + handoff_title / handoff_task +218 77,719 (89 tools)
# Plus one.
# Re-measured on the v2 External API root merged with dev 8df80022bc, which
# brings the MCP tools' external schemas and the preset tools back:
# merged tree 77,510 (89 tools)
# + headroom +300 77,810
# Both met merging dev into #16101 (SECRT-3138); re-measured on that merged
# tree, which is under dev's ceiling, so the higher value stands:
# merged tree 77,728 (89 tools)
_CHAR_BUDGET = 77_810
@@ -157,8 +157,11 @@ async def test_retry_uses_persisted_trial_snapshot_after_publish_failure(fifo):
{"cost_context": snapshot.model_dump(mode="json")}
)
publisher.publish_message.side_effect = [RuntimeError("RabbitMQ blip"), None]
with pytest.raises(RuntimeError, match="RabbitMQ blip"):
# A publish that raised may still have reached the broker, so it surfaces
# as TurnPublishUnconfirmed with the broker error as its cause.
with pytest.raises(utils.TurnPublishUnconfirmed) as raised:
await turn_queue.dispatch_next_for_user("user-1")
assert str(raised.value.__cause__) == "RabbitMQ blip"
db.update_chat_session_status.assert_any_await(
session_id="session-1", expect_status="running", status="queued"
)
@@ -113,7 +113,6 @@ test("a Soul edit is put to the user on its own card, whose Approve the gate tak
expect(screen.getByText("Was: Short and factual.")).toBeDefined();
await user.click(screen.getByRole("button", { name: "Approve" }));
await user.click(screen.getByRole("button", { name: "Send decisions" }));
// The line gate/card_approval.py takes for confirm_expert_soul_update.
expect(onSend).toHaveBeenCalledWith(
@@ -25,6 +25,13 @@ import { CopilotChatActionsContext } from "../CopilotChatActionsProvider/useCopi
import { SOUL_CHANGE_TOOLS } from "./helpers";
import { useLiveTipMoment } from "@/app/(platform)/components/ContextualTips/useTipMoment";
import { asObject, str } from "./resultHelpers";
import {
handoffOutcomeOf,
handoffSubject,
handoffTaskOf,
} from "./expertHandoff";
import { HandoffProposal, HandoffReceipt } from "./ExpertHandoffParts";
import { HandoffStatus } from "./HandoffStatus";
interface Props {
output: Record<string, unknown>;
@@ -40,12 +47,18 @@ interface Props {
onDecide?: (decision: Decision) => void;
/** Takes the decision back so Approve/Decline can be chosen again. */
onUndo?: () => void;
/** The turn is still streaming: answers wait until the proposals are all
* in, so a click always does what its button says. */
waiting?: boolean;
/** Footer control the group adds once every proposal is decided, or
* the sent note after. */
action?: ReactNode;
/** Which way the pager just moved — the identity block slides in from
* that side while the shell (Details, pager, decision) stays put. */
stepDirection?: StepDirection;
/** Sends a chat message (a hand-off that didn't start offers Try again);
* absent on a read-only transcript. */
onSend?: (message: string) => void;
}
export type StepDirection = "forward" | "back";
@@ -100,6 +113,8 @@ interface Proposal {
kind: ExpertKind | "soul" | null;
name: string;
confirmationId: string;
/** The "hire and start" choice, when the hire is for a task. */
handoff: { confirmationId: string; subject: string; title: string } | null;
}
function isOpenExpertProposal(part: ToolUIPart) {
@@ -118,18 +133,48 @@ function proposalOf(part: ToolUIPart): Proposal | null {
kind: "soul",
name: "your Soul",
confirmationId,
handoff: null,
};
}
const preview = asObject(output.preview);
if (!confirmationId || !preview) return null;
const name = str(preview, "name") ?? "this expert";
const task = handoffTaskOf(output);
const handoffConfirmationId = str(output, "handoff_confirmation_id");
return {
toolCallId: part.toolCallId,
kind: kindOf(output),
name: str(preview, "name") ?? "this expert",
name,
confirmationId,
handoff:
task && handoffConfirmationId
? {
confirmationId: handoffConfirmationId,
subject: handoffSubject(name, task),
title: task.title,
}
: null,
};
}
/** The line for one answer. A hire proposed for a task asks one question,
* Hire and start or Not now, so its approval is always of the proposal's
* hand-off id, which confirm_expert_change reads as "also start the task".
* Not now declines the hire itself. */
function answerLine(proposal: Proposal, decision: Decision): string {
if (decision === "approved" && proposal.handoff) {
return decisionLine(
{
...proposal,
name: proposal.handoff.subject,
confirmationId: proposal.handoff.confirmationId,
},
"approved",
);
}
return decisionLine(proposal, decision);
}
// gate/card_approval.py reads the Approved line as the answer to the confirm.
function decisionLine(proposal: Proposal, decision: Decision): string {
const verb =
@@ -198,6 +243,8 @@ export function ExpertChangeCard({
onUndo,
action,
stepDirection,
waiting = false,
onSend,
}: Props) {
const [showCharter, setShowCharter] = useState(false);
const openArtifact = useCopilotUIStore((s) => s.openArtifact);
@@ -224,7 +271,82 @@ export function ExpertChangeCard({
const hasCharter = (!!tagline && !!about) || !!boundaries;
const showDetails = panelAvailable || hasCharter;
const showFooter = !!pager || !!onDecide || !!decision || !!action;
const header = headerFor(kindOf(output), output.applied === true);
const applied = output.applied === true;
const header = headerFor(kindOf(output), applied);
const task = applied ? null : handoffTaskOf(output);
const outcome = applied ? handoffOutcomeOf(output) : null;
const stepClassName = cn(
stepDirection &&
"duration-300 animate-in fade-in fill-mode-both motion-reduce:animate-none",
stepDirection === "forward" && "slide-in-from-right-4",
stepDirection === "back" && "slide-in-from-left-4",
);
if (outcome) {
// Two confirms can share a group (two hires in one send): the pager
// stays under the status so every expert's outcome can be reached.
return (
<div className="flex flex-col gap-2">
<HandoffStatus
key={artifactId ?? name}
outcome={outcome}
name={name}
expertId={str(expert, "id")}
avatarUrl={str(expert, "avatar_url")}
color={str(expert, "color")}
onSend={onSend}
/>
{(!!pager || !!action) && (
<div className="flex items-center justify-between gap-2">
<div className="min-w-0">{pager}</div>
{action}
</div>
)}
</div>
);
}
if (task) {
return (
<div className={SHELL}>
<HandoffProposal
name={name}
role={role}
avatarUrl={str(expert, "avatar_url")}
color={str(expert, "color")}
about={summary}
task={task}
actions={
!decision && onDecide
? {
onStart: () => onDecide("approved"),
onNotNow: () => onDecide("declined"),
disabled: waiting,
}
: null
}
onShowProfile={panelAvailable ? handleShowMore : undefined}
stepClassName={stepClassName}
stepKey={artifactId ?? name}
/>
{(!!pager || !!decision || !!action) && (
<DecisionFooter
pager={pager}
decision={decision}
onUndo={onUndo}
action={action}
chosen={
decision
? decision === "approved"
? "Hire and start chosen"
: "Not now chosen"
: undefined
}
/>
)}
</div>
);
}
return (
<div className={SHELL}>
@@ -306,6 +428,7 @@ export function ExpertChangeCard({
onDecide={onDecide}
onUndo={onUndo}
action={action}
waiting={waiting}
/>
)}
</div>
@@ -328,6 +451,7 @@ function SoulChangeCard({
onDecide,
onUndo,
action,
waiting = false,
}: Props) {
const showFooter = !!pager || !!onDecide || !!decision || !!action;
return (
@@ -366,6 +490,7 @@ function SoulChangeCard({
onDecide={onDecide}
onUndo={onUndo}
action={action}
waiting={waiting}
/>
)}
</div>
@@ -374,8 +499,11 @@ function SoulChangeCard({
type FooterProps = Pick<
Props,
"pager" | "decision" | "onDecide" | "onUndo" | "action"
>;
"pager" | "decision" | "onDecide" | "onUndo" | "action" | "waiting"
> & {
/** What a stored approval will send, for cards offering two approvals. */
chosen?: string;
};
function DecisionFooter({
pager,
@@ -383,11 +511,16 @@ function DecisionFooter({
onDecide,
onUndo,
action,
waiting = false,
chosen,
}: FooterProps) {
return (
<div className={FOOTER}>
<div className="min-w-0">{pager}</div>
<div className="flex shrink-0 items-center gap-1.5">
{chosen && (
<span className="text-xs text-foreground-tertiary">{chosen}</span>
)}
{decision && onUndo && (
<Button
variant="secondary"
@@ -396,7 +529,11 @@ function DecisionFooter({
leftIcon={<Icon icon={UndoIcon} size={14} />}
className="h-8 !min-w-0 px-3"
>
{decision === "approved" ? "Unapprove" : "Undo decline"}
{chosen
? "Change"
: decision === "approved"
? "Unapprove"
: "Undo decline"}
</Button>
)}
{!decision && onDecide && (
@@ -404,6 +541,7 @@ function DecisionFooter({
<Button
variant="ghost"
size="small"
disabled={waiting}
onClick={() => onDecide("declined")}
className="h-8 !min-w-0 px-3"
>
@@ -412,6 +550,7 @@ function DecisionFooter({
<Button
variant="primary"
size="small"
disabled={waiting}
onClick={() => onDecide("approved")}
className="h-8 !min-w-0 px-3"
>
@@ -484,6 +623,8 @@ interface PartProps {
onUndo?: () => void;
action?: ReactNode;
stepDirection?: StepDirection;
waiting?: boolean;
onSend?: (message: string) => void;
}
/** The expert is still being written while the row has no output — hold
@@ -499,6 +640,8 @@ export function ExpertChangePart({
onUndo,
action,
stepDirection,
waiting,
onSend,
}: PartProps) {
if (part.state === "output-error") return <GenericTool part={part} />;
const output = asObject(part.output);
@@ -514,6 +657,8 @@ export function ExpertChangePart({
onUndo={onUndo}
action={action}
stepDirection={stepDirection}
waiting={waiting}
onSend={onSend}
/>
);
}
@@ -577,30 +722,23 @@ export function ExpertChangeGroup({
proposals.length > 0 &&
proposals.every((proposal) => decisions[proposal.toolCallId]);
const currentProposal = readOnly ? null : proposalOf(part);
// While the turn is still streaming another proposal part can land in
// this group, so the send waits for the stream to settle.
const sendable = !readOnly && !!actions && !isCurrentlyStreaming;
// A lone proposal has nothing to review before the send, so its answer
// is the send: one click hires (and hands off), not Approve then Send.
const answerSends =
sendable && visible.length === 1 && proposals.length === 1;
function goTo(next: number) {
setDirection(next > current ? "forward" : "back");
setStep(next);
}
function decide(decision: Decision) {
if (!currentProposal) return;
setDecisions({ ...decisions, [currentProposal.toolCallId]: decision });
if (!isLast) goTo(current + 1);
}
function undo() {
if (!currentProposal) return;
setDecisions(withoutDecision(decisions, currentProposal.toolCallId));
}
function send() {
function sendMessage(message: string) {
if (!actions) return;
const message = proposals
.map((proposal) => decisionLine(proposal, decisions[proposal.toolCallId]))
.join("\n");
// The card stays marked sent because the message is already in the
// thread for Retry — the failure only needs saying.
// The message is already in the thread for Retry — the failure only
// needs saying.
void Promise.resolve(actions.onSend(message)).catch((error: unknown) =>
toastSendFailure({
title: "Couldn't send message",
@@ -610,9 +748,36 @@ export function ExpertChangeGroup({
"it is still in the thread, use Retry once you have a plan",
}),
);
}
function send(answered: Record<string, Decision>) {
sendMessage(
proposals
.map((proposal) => answerLine(proposal, answered[proposal.toolCallId]))
.join("\n"),
);
setSent(true);
}
function decide(decision: Decision) {
if (!currentProposal || isCurrentlyStreaming) return;
const nextDecisions = {
...decisions,
[currentProposal.toolCallId]: decision,
};
setDecisions(nextDecisions);
if (answerSends) {
send(nextDecisions);
return;
}
if (!isLast) goTo(current + 1);
}
function undo() {
if (!currentProposal) return;
setDecisions(withoutDecision(decisions, currentProposal.toolCallId));
}
const pager = visible.length > 1 && (
<div className="flex items-center gap-2">
<button
@@ -664,9 +829,7 @@ export function ExpertChangeGroup({
</div>
);
// While the turn is still streaming another proposal part can land in
// this group, so the send waits for the stream to settle.
const canSend = !readOnly && !!actions && allDecided && !isCurrentlyStreaming;
const canSend = sendable && allDecided;
const action = sent ? (
<span className="flex items-center gap-1 text-xs text-foreground-subtle">
<Icon icon={Tick02Icon} size={14} />
@@ -677,13 +840,26 @@ export function ExpertChangeGroup({
variant="primary"
size="icon"
aria-label="Send decisions"
onClick={send}
onClick={() => send(decisions)}
className="size-8 p-0"
>
<Icon icon={SentIcon} size={15} />
</Button>
) : null;
// A lone hand-off card, once answered, is only the record of the choice:
// the expert's own card below says where the work went.
const lone = proposals.length === 1 && visible.length === 1;
if (sent && lone && proposals[0].handoff) {
return (
<div className="my-2">
<HandoffReceipt>
{receiptFor(proposals[0], decisions[proposals[0].toolCallId])}
</HandoffReceipt>
</div>
);
}
// The shell (Details, pager, decision buttons) is the same from one
// expert to the next, so it stays mounted; only the identity block
// inside the card slides — see ExpertChangeCard.
@@ -698,7 +874,19 @@ export function ExpertChangeGroup({
onUndo={currentProposal && !sent ? undo : undefined}
action={action || undefined}
stepDirection={direction}
waiting={isCurrentlyStreaming}
onSend={sendable ? sendMessage : undefined}
/>
</div>
);
}
/** The record of the choice, worded as the choice: the confirm runs after
* the send and can still fail, and the expert's own card says what came of
* it. Only a hire proposed for a task shrinks to a receipt. */
function receiptFor(proposal: Proposal, decision: Decision | undefined) {
if (decision === "approved" && proposal.handoff) {
return `You chose to hire ${proposal.name} to take on ${proposal.handoff.title}.`;
}
return `You chose not to hire ${proposal.name} for now.`;
}
@@ -0,0 +1,172 @@
"use client";
import { ArrowDown01Icon, Tick02Icon } from "@hugeicons/core-free-icons";
import { type ReactNode, useId, useState } from "react";
import { Button } from "@/components/atoms/Button/Button";
import { Icon } from "@/components/atoms/Icon/Icon";
import { ExpertAvatar } from "@/components/molecules/ExpertAvatar/ExpertAvatar";
import { cn } from "@/lib/utils";
import type { HandoffTask } from "./expertHandoff";
interface ProposalProps {
name: string;
role: string | null;
avatarUrl: string | null;
color: string | null;
/** One line about the expert, written for the user. */
about: string | null;
task: HandoffTask;
/** Absent once answered, or on a read-only transcript. */
actions: {
onStart: () => void;
onNotNow: () => void;
disabled: boolean;
} | null;
/** Opens the full charter in the side panel, where there is one. */
onShowProfile?: () => void;
/** Classes that slide the identity in when a pager moves. */
stepClassName?: string;
stepKey: string;
}
/** A hire proposed for a task the user already described, asked as one
* yes/no question: Hire and start, or Not now. The brief and the expert's
* charter stay behind one disclosure: the user wrote the task a moment ago
* and only needs who, what and the choice to answer. */
export function HandoffProposal({
name,
role,
avatarUrl,
color,
about,
task,
actions,
onShowProfile,
stepClassName,
stepKey,
}: ProposalProps) {
return (
<div className="flex gap-3 p-4">
<ExpertAvatar
key={`${stepKey}-avatar`}
name={name}
avatarUrl={avatarUrl}
color={color}
size={40}
/>
<div className="min-w-0 flex-1">
<div key={stepKey} className={stepClassName}>
<p className="text-body font-medium leading-snug text-foreground-emphasis [overflow-wrap:anywhere]">
Hire {name} to take on {task.title}?
</p>
<p className="mt-0.5 text-sm text-foreground-tertiary [overflow-wrap:anywhere]">
{role
? `${role} · works on it in their own chat`
: "Works on it in their own chat"}
</p>
</div>
{actions && (
<div className="mt-3 flex flex-wrap items-center gap-2">
<Button
variant="primary"
size="small"
disabled={actions.disabled}
onClick={actions.onStart}
className="h-8 !min-w-0 px-3.5"
>
Hire and start
</Button>
<Button
variant="ghost"
size="small"
disabled={actions.disabled}
onClick={actions.onNotNow}
className="h-8 !min-w-0 px-3"
>
Not now
</Button>
</div>
)}
<HandoffDetails
name={name}
about={about}
task={task}
onShowProfile={onShowProfile}
/>
</div>
</div>
);
}
interface DetailsProps {
name: string;
about: string | null;
task: HandoffTask;
onShowProfile?: () => void;
}
function HandoffDetails({ name, about, task, onShowProfile }: DetailsProps) {
const [open, setOpen] = useState(false);
const panelId = useId();
return (
<div className="mt-3">
<button
type="button"
aria-expanded={open}
aria-controls={panelId}
onClick={() => setOpen(!open)}
className="inline-flex items-center gap-1 rounded text-xs font-medium text-foreground-secondary underline-offset-2 hover:underline focus-visible:outline-none focus-visible:ring-1 focus-visible:ring-focus"
>
What {name} will get
<Icon
icon={ArrowDown01Icon}
size={12}
aria-hidden
className={cn(
"transition-transform motion-reduce:transition-none",
open && "rotate-180",
)}
/>
</button>
<div
id={panelId}
hidden={!open}
className={cn("mt-2 flex-col gap-2", open ? "flex" : "hidden")}
>
<p className="whitespace-pre-line rounded-2xl bg-surface-muted px-3 py-2.5 text-sm text-foreground-tertiary [overflow-wrap:anywhere]">
{task.brief}
</p>
{about && (
<p className="text-xs text-foreground-subtle [overflow-wrap:anywhere]">
About {name}: {about}
</p>
)}
{onShowProfile && (
<button
type="button"
onClick={onShowProfile}
className="self-start rounded text-xs font-medium text-foreground-secondary underline-offset-2 hover:underline focus-visible:outline-none focus-visible:ring-1 focus-visible:ring-focus"
>
See {name}&apos;s full profile
</button>
)}
</div>
</div>
);
}
interface ReceiptProps {
children: ReactNode;
}
/** What an answered hand-off card shrinks to: the record of the choice. */
export function HandoffReceipt({ children }: ReceiptProps) {
return (
<p className="flex items-center gap-2 text-sm text-foreground-tertiary">
<span className="flex size-5 shrink-0 items-center justify-center rounded-full bg-surface-muted">
<Icon icon={Tick02Icon} size={12} aria-hidden />
</span>
<span className="min-w-0 [overflow-wrap:anywhere]">{children}</span>
</p>
);
}
@@ -0,0 +1,59 @@
"use client";
import { ArrowUpRight01Icon } from "@hugeicons/core-free-icons";
import NextLink from "next/link";
import { Button } from "@/components/atoms/Button/Button";
import { Icon } from "@/components/atoms/Icon/Icon";
import { possessive } from "./expertHandoff";
import { useHandoffRetry } from "./useHandoffRetry";
interface Props {
title: string;
name: string;
expertId: string;
onSend: (message: string) => void;
}
/** Try again, for a hand-off the backend says can be sent again, and only
* while nothing has reached the expert since: one copy of the task, never
* two. */
export function HandoffRetry({ title, name, expertId, onSend }: Props) {
const { asked, markAsked, known, threadId } = useHandoffRetry(expertId);
if (asked) {
return (
<span className="text-sm text-foreground-tertiary">
Asked Otto to try again.
</span>
);
}
if (!known) return null;
if (threadId) {
return (
<NextLink
href={`/copilot?sessionId=${encodeURIComponent(threadId)}`}
className="inline-flex shrink-0 items-center gap-1 rounded text-sm font-medium text-foreground-emphasis underline-offset-2 hover:underline focus-visible:outline-none focus-visible:ring-1 focus-visible:ring-focus"
>
Open {possessive(name)} chat
<Icon
icon={ArrowUpRight01Icon}
size={14}
aria-hidden
className="rtl:-scale-x-100"
/>
</NextLink>
);
}
return (
<Button
variant="secondary"
size="small"
onClick={() => {
markAsked();
onSend(`Try handing ${title} to ${name} again.`);
}}
className="h-8 !min-w-0 px-3"
>
Try again
</Button>
);
}
@@ -0,0 +1,183 @@
"use client";
import { ArrowUpRight01Icon } from "@hugeicons/core-free-icons";
import NextLink from "next/link";
import { Icon } from "@/components/atoms/Icon/Icon";
import {
frozenStatusOf,
type HandoffOutcome,
type HandoffState,
handoffStateOf,
possessive,
} from "./expertHandoff";
import { HandoffRetry } from "./HandoffRetry";
import { StatusAvatar } from "./HandoffStatusAvatar";
import { useSubSessionEffectiveStatus } from "./SubSessionLive";
interface Props {
outcome: HandoffOutcome;
name: string;
/** The hire; absent on outputs that don't carry it. */
expertId: string | null;
avatarUrl: string | null;
color: string | null;
/** Sends a chat message; absent on a read-only transcript. */
onSend?: (message: string) => void;
}
/** Where the task went once the hire and hand-off landed. The confirm only
* queued it, so this follows the expert's thread from there. While the work
* is in flight it is a card; once it is over it recedes to one line. The
* hire itself needs no card of its own: the answered proposal above already
* records it. */
export function HandoffStatus({
outcome,
name,
expertId,
avatarUrl,
color,
onSend,
}: Props) {
const live = useSubSessionEffectiveStatus(
outcome.subSessionId,
frozenStatusOf(outcome),
);
const state = handoffStateOf(outcome, live);
const unknown = state === "unknown";
const avatar = (size: number) => (
<StatusAvatar
name={name}
avatarUrl={avatarUrl}
color={color}
state={state}
size={size}
/>
);
if (state === "waiting" || state === "working" || unknown) {
return (
<div className="w-full overflow-hidden rounded-3xl border border-separator bg-surface">
<div className="flex flex-wrap items-center gap-x-3 gap-y-2 px-4 py-3">
{avatar(32)}
<p className="min-w-[min(100%,14rem)] flex-1 text-sm text-foreground-secondary [overflow-wrap:anywhere]">
<HandoffSentence outcome={outcome} name={name} state={state} />
</p>
{unknown && (
<span className="inline-flex shrink-0 items-center rounded-full bg-surface-hover px-2 py-0.5 text-caption-sm font-medium leading-inherit text-foreground-tertiary">
Status unknown
</span>
)}
{outcome.link && (
<NextLink
href={outcome.link}
className="inline-flex shrink-0 items-center gap-1.5 rounded-full border border-border-base bg-surface px-3 py-1.5 text-sm font-medium text-foreground-emphasis shadow-sm transition-colors hover:border-border-strong hover:bg-surface-muted focus-visible:outline-none focus-visible:ring-1 focus-visible:ring-focus"
>
Open {possessive(name)} chat
<Icon
icon={ArrowUpRight01Icon}
size={14}
aria-hidden
className="rtl:-scale-x-100"
/>
</NextLink>
)}
</div>
</div>
);
}
return (
<div className="flex items-start gap-2.5">
{avatar(24)}
<div className="flex min-w-0 flex-1 flex-wrap items-center gap-x-2.5 gap-y-1 pt-0.5">
<p
role={state === "failed" ? "status" : undefined}
className="min-w-0 text-sm text-foreground-secondary [overflow-wrap:anywhere]"
>
<HandoffSentence outcome={outcome} name={name} state={state} />
</p>
{state === "failed"
? outcome.retry &&
onSend &&
expertId && (
<HandoffRetry
title={outcome.title}
name={name}
expertId={expertId}
onSend={onSend}
/>
)
: outcome.link && (
<NextLink
href={outcome.link}
className="inline-flex shrink-0 items-center gap-1 rounded text-sm font-medium text-foreground-emphasis underline-offset-2 hover:underline focus-visible:outline-none focus-visible:ring-1 focus-visible:ring-focus"
>
Open {possessive(name)} chat
<Icon
icon={ArrowUpRight01Icon}
size={14}
aria-hidden
className="rtl:-scale-x-100"
/>
</NextLink>
)}
</div>
</div>
);
}
interface SentenceProps {
outcome: HandoffOutcome;
name: string;
state: HandoffState;
}
function HandoffSentence({ outcome, name, state }: SentenceProps) {
const who = (
<span className="font-medium text-foreground-emphasis">{name}</span>
);
switch (state) {
case "failed":
return (
<>
{who} is hired, but {outcome.title} didn&apos;t start.
</>
);
case "unconfirmed":
return (
<>
{outcome.title} may have started in {possessive(name)} chat. Open it
to check.
</>
);
// An idle thread holds a result, a question for the user or an error;
// the poll can't tell which, so the line doesn't guess.
case "stopped":
return (
<>
{who} stopped working on {outcome.title}. Their chat has the result,
or a question for you.
</>
);
case "waiting":
return (
<>
{who} starts on {outcome.title} as soon as one of your other tasks
finishes.
</>
);
case "working":
return (
<>
{who} is working on {outcome.title}.
</>
);
// The live status can't be read: say only what is known.
case "unknown":
return (
<>
Sent {outcome.title} to {who}. Open their chat to see where it stands.
</>
);
}
}
@@ -0,0 +1,82 @@
"use client";
import {
Alert02Icon,
Clock01Icon,
InformationCircleIcon,
} from "@hugeicons/core-free-icons";
import { Icon } from "@/components/atoms/Icon/Icon";
import { ExpertAvatar } from "@/components/molecules/ExpertAvatar/ExpertAvatar";
import { cn } from "@/lib/utils";
import type { HandoffState } from "./expertHandoff";
const BADGES: Record<
Exclude<HandoffState, "unconfirmed" | "unknown">,
{ className: string; icon: typeof Clock01Icon | null }
> = {
working: { className: "bg-info-surface text-info", icon: null },
waiting: {
className: "bg-warning-surface text-warning-foreground",
icon: Clock01Icon,
},
// Neutral, not success: an idle thread may hold a question or an error.
stopped: {
className: "bg-surface-hover text-foreground-secondary",
icon: InformationCircleIcon,
},
failed: {
className: "bg-warning-surface text-warning-foreground",
icon: Alert02Icon,
},
};
interface AvatarProps {
name: string;
avatarUrl: string | null;
color: string | null;
state: HandoffState | null;
size: number;
}
/** The expert with what their thread is doing on the corner. The badge is
* decoration: the sentence beside it always says the same in words. */
export function StatusAvatar({
name,
avatarUrl,
color,
state,
size,
}: AvatarProps) {
const badge =
state && state !== "unconfirmed" && state !== "unknown"
? BADGES[state]
: null;
const badgeSize = Math.round(size * 0.5);
return (
<span className="relative inline-flex shrink-0">
<ExpertAvatar
name={name}
avatarUrl={avatarUrl}
color={color}
size={size}
/>
{badge && (
<span
aria-hidden
data-testid="handoff-badge"
style={{ width: badgeSize, height: badgeSize }}
className={cn(
"absolute -bottom-0.5 -end-0.5 flex items-center justify-center rounded-full ring-2 ring-surface",
badge.className,
)}
>
{badge.icon ? (
<Icon icon={badge.icon} size={Math.round(badgeSize * 0.7)} />
) : (
<span className="size-[60%] animate-spin rounded-full border-[1.5px] border-current border-t-transparent motion-reduce:animate-none" />
)}
</span>
)}
</span>
);
}
@@ -364,8 +364,10 @@ function useDelegatedSessionId(expertId: string | null) {
/** A sub-session tool output freezes the status it had when the tool
* returned — a "running" card would say running forever after the work
* lands. While the frozen status is running/queued, read the truth off the
* polled session and flip to completed once it goes idle.
* lands, and a "queued" one would say queued after it started. While the
* frozen status is running/queued, read the status off the polled session
* instead: queued while it waits for a slot, running once it runs, and
* completed once it goes idle.
*
* Owns the poll through `useLiveSubSession` rather than piggybacking on a
* mounted live view: a minimal delegate card renders no live view, so this
@@ -386,7 +388,14 @@ export function useSubSessionEffectiveStatus(
// dead poll has to show up in the pill or it reads as fact.
if (isError || isPaused) return "unknown";
if (!session) return status;
return isSessionLive(session) ? status : "completed";
return liveStatusOf(session);
}
function liveStatusOf(session: SessionDetailResponse) {
if (session.active_stream) return "running";
const status = session.chat_status?.toLowerCase();
if (status === "running" || status === "queued") return status;
return "completed";
}
function isSessionLive(session: SessionDetailResponse): boolean {
@@ -417,7 +417,6 @@ describe("expert approval", () => {
renderGroup([expertPart("output-available", gatedProposal)]);
await user.click(screen.getByRole("button", { name: "Approve" }));
await user.click(screen.getByRole("button", { name: "Send decisions" }));
// The line gate/card_approval.py takes as the answer to the confirm.
expect(onSend).toHaveBeenCalledWith(
@@ -456,14 +455,22 @@ describe("expert approval", () => {
expect(screen.queryByRole("button", { name: "Undo decline" })).toBeNull();
});
it("holds the send until the stream has settled", async () => {
it("takes no answer until the stream has settled, then sends on one click", async () => {
const user = userEvent.setup();
const { rerender } = renderGroup([proposal("Otto", "a")], {
isCurrentlyStreaming: true,
});
await user.click(screen.getByRole("button", { name: "Approve" }));
expect(screen.queryByRole("button", { name: "Send decisions" })).toBeNull();
// Another proposal can still land and turn this into a pager, so an
// answer now could not do what the button says.
const approve = screen.getByRole("button", { name: "Approve" });
expect(approve).toHaveProperty("disabled", true);
expect(screen.getByRole("button", { name: "Decline" })).toHaveProperty(
"disabled",
true,
);
await user.click(approve);
expect(onSend).not.toHaveBeenCalled();
rerender(
<CopilotChatActionsProvider onSend={onSend}>
@@ -474,9 +481,10 @@ describe("expert approval", () => {
</CopilotChatActionsProvider>,
);
expect(
screen.getByRole("button", { name: "Send decisions" }),
).toBeDefined();
await user.click(screen.getByRole("button", { name: "Approve" }));
expect(onSend).toHaveBeenCalledWith(
"Approved: create Otto (confirmation_id: c-a).",
);
});
it("offers no send without a chat actions provider", async () => {
@@ -509,8 +517,9 @@ describe("expert approval", () => {
const user = userEvent.setup();
renderGroup([proposal("Otto", "a")]);
// A lone proposal's answer is the send: no separate Send click.
await user.click(screen.getByRole("button", { name: "Approve" }));
await user.click(screen.getByRole("button", { name: "Send decisions" }));
expect(screen.queryByRole("button", { name: "Send decisions" })).toBeNull();
expect(onSend).toHaveBeenCalledWith(
"Approved: create Otto (confirmation_id: c-a).",
@@ -523,7 +532,6 @@ describe("expert approval", () => {
renderGroup([proposal("Otto", "a")]);
await user.click(screen.getByRole("button", { name: "Approve" }));
await user.click(screen.getByRole("button", { name: "Send decisions" }));
await waitFor(() =>
expect(toast).toHaveBeenCalledWith(
@@ -554,7 +562,6 @@ describe("expert approval", () => {
]);
await user.click(screen.getByRole("button", { name: "Approve" }));
await user.click(screen.getByRole("button", { name: "Send decisions" }));
expect(onSend).toHaveBeenCalledWith(
"Approved: update Otto (confirmation_id: c-u).",
@@ -0,0 +1,532 @@
import { getGetV2GetSessionMockHandler200 } from "@/app/api/__generated__/endpoints/chat/chat.msw";
import { server } from "@/mocks/mock-server";
import { render, screen } from "@/tests/integrations/test-utils";
import userEvent from "@testing-library/user-event";
import type { ToolUIPart } from "ai";
import { http, HttpResponse } from "msw";
import { beforeEach, describe, expect, it, vi } from "vitest";
import { CopilotChatActionsProvider } from "../../CopilotChatActionsProvider/CopilotChatActionsProvider";
import { ExpertChangeCard, ExpertChangeGroup } from "../ExpertCards";
vi.mock("@/components/molecules/Toast/use-toast", () => ({
toast: vi.fn(),
useToast: () => ({ toast: vi.fn(), dismiss: vi.fn() }),
}));
const onSend = vi.fn();
const TITLE = "the Printify order pipeline";
const BRIEF =
"Own the Printify order pipeline: move new Etsy orders into Printify and sync tracking numbers back to Etsy every morning.";
const TAGLINE = "Runs your operating rhythm, SOPs and vendors.";
const HAND_OFF = "Hire and start";
const NOT_NOW = "Not now";
function hire(id: string, name = "James", withTask = true): ToolUIPart {
return {
type: "tool-hire_expert",
state: "output-available",
toolCallId: `call-${id}`,
input: {},
output: {
type: "expert_change_proposed",
applied: false,
confirmation_id: `c-${id}`,
...(withTask
? {
handoff: { title: TITLE, brief: BRIEF },
handoff_confirmation_id: `h-${id}`,
}
: {}),
preview: { kind: "hire", name, role: "Operations", tagline: TAGLINE },
},
} as ToolUIPart;
}
function renderGroup(parts: ToolUIPart[]) {
return render(
<CopilotChatActionsProvider onSend={onSend}>
<ExpertChangeGroup parts={parts} />
</CopilotChatActionsProvider>,
);
}
function isHidden(text: string) {
return !!screen.getByText(text, { exact: false }).closest("[hidden]");
}
describe("hiring an expert for a task already in the chat", () => {
beforeEach(() => onSend.mockClear());
it("asks one yes/no question: hire and start, or not now", () => {
renderGroup([hire("a")]);
expect(screen.getByText(`Hire James to take on ${TITLE}?`)).toBeDefined();
const answers = screen
.getAllByRole("button")
.filter((b) => b.textContent === HAND_OFF || b.textContent === NOT_NOW);
expect(answers.map((b) => b.textContent)).toEqual([HAND_OFF, NOT_NOW]);
expect(screen.queryByRole("button", { name: "Hire only" })).toBeNull();
expect(screen.queryByRole("button", { name: "Decline" })).toBeNull();
expect(screen.queryByRole("button", { name: "Approve" })).toBeNull();
});
it("keeps the brief and the charter behind one disclosure", async () => {
const user = userEvent.setup();
renderGroup([hire("a")]);
const toggle = screen.getByRole("button", { name: "What James will get" });
expect(toggle.getAttribute("aria-expanded")).toBe("false");
expect(isHidden(BRIEF)).toBe(true);
expect(isHidden(TAGLINE)).toBe(true);
await user.click(toggle);
expect(toggle.getAttribute("aria-expanded")).toBe("true");
expect(isHidden(BRIEF)).toBe(false);
expect(isHidden(TAGLINE)).toBe(false);
});
it("hires and starts in one click, then records the choice", async () => {
const user = userEvent.setup();
renderGroup([hire("a")]);
await user.click(screen.getByRole("button", { name: HAND_OFF }));
// The hand-off id: confirm_expert_change hires AND starts the task.
expect(onSend).toHaveBeenCalledTimes(1);
expect(onSend).toHaveBeenCalledWith(
`Approved: hire James to take on ${TITLE} (confirmation_id: h-a).`,
);
// Worded as the choice: the confirm runs after the send and can fail.
expect(
screen.getByText(`You chose to hire James to take on ${TITLE}.`),
).toBeDefined();
expect(screen.queryByText(/You hired/)).toBeNull();
expect(screen.queryByText(`Hire James to take on ${TITLE}?`)).toBeNull();
expect(screen.queryByRole("button")).toBeNull();
});
it("waits for the stream to settle before the one-click hand-off", async () => {
const user = userEvent.setup();
const { rerender } = render(
<CopilotChatActionsProvider onSend={onSend}>
<ExpertChangeGroup parts={[hire("a")]} isCurrentlyStreaming />
</CopilotChatActionsProvider>,
);
for (const label of [HAND_OFF, NOT_NOW]) {
expect(screen.getByRole("button", { name: label })).toHaveProperty(
"disabled",
true,
);
}
await user.click(screen.getByRole("button", { name: HAND_OFF }));
expect(onSend).not.toHaveBeenCalled();
rerender(
<CopilotChatActionsProvider onSend={onSend}>
<ExpertChangeGroup parts={[hire("a")]} isCurrentlyStreaming={false} />
</CopilotChatActionsProvider>,
);
await user.click(screen.getByRole("button", { name: HAND_OFF }));
expect(onSend).toHaveBeenCalledTimes(1);
expect(onSend).toHaveBeenCalledWith(
`Approved: hire James to take on ${TITLE} (confirmation_id: h-a).`,
);
});
it("says not now to the hire, not just the task", async () => {
const user = userEvent.setup();
renderGroup([hire("a")]);
await user.click(screen.getByRole("button", { name: NOT_NOW }));
expect(onSend).toHaveBeenCalledWith(
"Not approved: do not hire James, discard that proposal (confirmation_id: c-a).",
);
expect(
screen.getByText("You chose not to hire James for now."),
).toBeDefined();
});
it("keeps each expert's choice when several are sent together", async () => {
const user = userEvent.setup();
renderGroup([hire("a"), hire("b", "Nadia")]);
await user.click(screen.getByRole("button", { name: HAND_OFF }));
await user.click(screen.getByRole("button", { name: NOT_NOW }));
expect(onSend).not.toHaveBeenCalled();
await user.click(screen.getByRole("button", { name: "Send decisions" }));
expect(onSend).toHaveBeenCalledWith(
`Approved: hire James to take on ${TITLE} (confirmation_id: h-a).\n` +
"Not approved: do not hire Nadia, discard that proposal (confirmation_id: c-b).",
);
// Several experts keep their pager after the send, not one receipt.
expect(screen.getByText("Sent")).toBeDefined();
});
it("shows which choice was made when paging back before the send", async () => {
const user = userEvent.setup();
renderGroup([hire("a"), hire("b", "Nadia")]);
await user.click(screen.getByRole("button", { name: HAND_OFF }));
await user.click(screen.getByRole("button", { name: NOT_NOW }));
expect(screen.getByText("Not now chosen")).toBeDefined();
await user.click(screen.getByRole("button", { name: "Previous expert" }));
// The label names the button that was pressed.
expect(screen.getByText(`${HAND_OFF} chosen`)).toBeDefined();
expect(screen.getByRole("button", { name: "Change" })).toBeDefined();
expect(screen.queryByRole("button", { name: HAND_OFF })).toBeNull();
expect(onSend).not.toHaveBeenCalled();
});
it("keeps the plain Approve card for a hire with no task", async () => {
const user = userEvent.setup();
renderGroup([hire("a", "James", false)]);
expect(screen.queryByText(/What James will get/)).toBeNull();
expect(screen.queryByRole("button", { name: "Hire only" })).toBeNull();
await user.click(screen.getByRole("button", { name: "Approve" }));
expect(onSend).toHaveBeenCalledWith(
"Approved: hire James (confirmation_id: c-a).",
);
expect(screen.queryByText(/You chose/)).toBeNull();
});
});
function thread(chatStatus: string) {
return getGetV2GetSessionMockHandler200({
id: "thread-1",
created_at: "2026-10-26T00:00:00Z",
updated_at: "2026-10-26T00:00:00Z",
user_id: "u-1",
messages: [],
chat_status: chatStatus,
active_stream: null,
});
}
const expert = { id: "exp-james", name: "James", role: "Operations" };
function appliedOutput(
status: string,
link: string | null = "/copilot?sessionId=thread-1",
{ retry = false, who = expert } = {},
) {
return {
type: "expert_change_applied",
applied: true,
kind: "hire",
expert: who,
handoff: {
title: TITLE,
status,
...(link ? { sub_session_id: "thread-1" } : {}),
link,
retry,
},
};
}
function confirmPart(id: string, output: Record<string, unknown>) {
return {
type: "tool-confirm_expert_change",
state: "output-available",
toolCallId: `call-${id}`,
input: {},
output,
} as ToolUIPart;
}
/** The hire's own chats, as the session list returns them. */
function expertThreads(ids: string[]) {
return http.get("/api/proxy/api/chat/sessions", () =>
HttpResponse.json({
sessions: ids.map((id) => ({
id,
created_at: "2026-10-26T00:00:00Z",
updated_at: "2026-10-26T00:00:00Z",
title: id,
chat_status: "running",
is_processing: true,
is_pinned: false,
expert_id: "exp-james",
})),
total: ids.length,
}),
);
}
function sentCard(status: "sent" | "queued" | "unconfirmed") {
return <ExpertChangeCard output={appliedOutput(status)} />;
}
function chatLink() {
return screen
.getByRole("link", { name: "Open James's chat" })
.getAttribute("href");
}
describe("after the hire and hand-off", () => {
beforeEach(() => onSend.mockClear());
it("says who is working on what and links to their chat, nothing more", async () => {
server.use(thread("running"));
render(sentCard("sent"));
const line = await screen.findByText(/is working on/);
expect(line.textContent).toBe(`James is working on ${TITLE}.`);
expect(chatLink()).toBe("/copilot?sessionId=thread-1");
// The proposal above already records the hire.
expect(screen.queryByText("Expert hired")).toBeNull();
expect(screen.queryByRole("button", { name: "Details" })).toBeNull();
});
it("says the task waits for a slot rather than that it started", async () => {
server.use(thread("queued"));
render(sentCard("queued"));
expect(
screen.getByText(
/starts on the Printify order pipeline as soon as one of your other tasks finishes/,
),
).toBeDefined();
expect(screen.queryByText(/is working on/)).toBeNull();
expect(chatLink()).toBe("/copilot?sessionId=thread-1");
});
it("points at the thread when it is not known whether the task started", () => {
render(sentCard("unconfirmed"));
expect(
screen.getByText(/may have started in James's chat\. Open it to check\./),
).toBeDefined();
expect(screen.queryByText(/is working on/)).toBeNull();
expect(chatLink()).toBe("/copilot?sessionId=thread-1");
});
it("shows a queued task start working once the expert picks it up", async () => {
// Mirrors POLL_MS in SubSessionLive.tsx: the thread is re-read this often.
const POLL_MS = 3000;
let chatStatus = "queued";
server.use(
http.get("/api/proxy/api/chat/sessions/:sessionId", () =>
HttpResponse.json({
id: "thread-1",
created_at: "2026-10-26T00:00:00Z",
updated_at: "2026-10-26T00:00:00Z",
user_id: "u-1",
messages: [],
chat_status: chatStatus,
active_stream: null,
}),
),
);
vi.useFakeTimers({ shouldAdvanceTime: true });
try {
render(sentCard("queued"));
expect(await screen.findByText(/starts on/)).toBeDefined();
chatStatus = "running";
await vi.advanceTimersByTimeAsync(POLL_MS + 100);
expect(await screen.findByText(/is working on/)).toBeDefined();
expect(screen.queryByText(/starts on/)).toBeNull();
} finally {
vi.useRealTimers();
}
});
it("follows the expert's thread and says when they stop working", async () => {
server.use(thread("idle"));
render(sentCard("sent"));
const line = await screen.findByText(/stopped working on/);
expect(line.textContent).toBe(
`James stopped working on ${TITLE}. Their chat has the result, or a question for you.`,
);
expect(screen.queryByText(/is working on/)).toBeNull();
expect(chatLink()).toBe("/copilot?sessionId=thread-1");
});
it("does not claim the task is running when the thread can't be read", async () => {
server.use(
http.get(
"/api/proxy/api/chat/sessions/:sessionId",
() => new HttpResponse(null, { status: 500 }),
),
);
render(sentCard("sent"));
expect(await screen.findByText("Status unknown")).toBeDefined();
expect(screen.queryByTestId("handoff-badge")).toBeNull();
// The sentence claims no more than the chip: it was sent, nothing else.
expect(screen.queryByText(/is working on/)).toBeNull();
expect(screen.getByText(/Sent the Printify/).textContent).toBe(
`Sent ${TITLE} to James. Open their chat to see where it stands.`,
);
});
it("never reads an idle thread as a reply when the start is unconfirmed", async () => {
let polled = false;
server.use(
http.get("/api/proxy/api/chat/sessions/:sessionId", () => {
polled = true;
return HttpResponse.json({});
}),
);
render(sentCard("unconfirmed"));
expect(screen.getByText(/may have started in James's chat/)).toBeDefined();
await new Promise((resolve) => setTimeout(resolve, 50));
expect(polled).toBe(false);
expect(screen.queryByText(/stopped working on/)).toBeNull();
});
it("never calls a sent task failed just because it has no link", () => {
render(<ExpertChangeCard output={appliedOutput("sent", null)} />);
expect(screen.getByText(/is working on/)).toBeDefined();
expect(screen.queryByRole("status")).toBeNull();
expect(screen.queryByRole("link")).toBeNull();
});
it("says plainly when the task did not start", () => {
render(<ExpertChangeCard output={appliedOutput("failed", null)} />);
expect(screen.getByRole("status").textContent).toBe(
`James is hired, but ${TITLE} didn't start.`,
);
expect(screen.queryByRole("link")).toBeNull();
// Read-only: nothing to send from.
expect(screen.queryByRole("button", { name: "Try again" })).toBeNull();
});
it("offers to try again when the backend says it can help", async () => {
server.use(expertThreads([]));
const user = userEvent.setup();
renderGroup([
confirmPart("confirm", appliedOutput("failed", null, { retry: true })),
]);
await user.click(await screen.findByRole("button", { name: "Try again" }));
expect(onSend).toHaveBeenCalledWith(`Try handing ${TITLE} to James again.`);
expect(screen.getByText("Asked Otto to try again.")).toBeDefined();
expect(screen.queryByRole("button", { name: "Try again" })).toBeNull();
});
it("never offers to try again after a refusal a retry can't fix", async () => {
server.use(expertThreads([]));
renderGroup([confirmPart("confirm", appliedOutput("failed", null))]);
expect((await screen.findByRole("status")).textContent).toBe(
`James is hired, but ${TITLE} didn't start.`,
);
await new Promise((resolve) => setTimeout(resolve, 50));
expect(screen.queryByRole("button", { name: "Try again" })).toBeNull();
});
it("points at the expert's chat instead of a second copy once Otto sent it again", async () => {
server.use(expertThreads(["thread-2"]));
renderGroup([
confirmPart("confirm", appliedOutput("failed", null, { retry: true })),
]);
const link = await screen.findByRole("link", { name: "Open James's chat" });
expect(link.getAttribute("href")).toBe("/copilot?sessionId=thread-2");
expect(screen.queryByRole("button", { name: "Try again" })).toBeNull();
});
it("keeps every expert's outcome reachable when two hires were confirmed together", async () => {
server.use(thread("running"), expertThreads([]));
const user = userEvent.setup();
renderGroup([
confirmPart("one", appliedOutput("sent")),
confirmPart(
"two",
appliedOutput("failed", null, {
who: { id: "exp-nadia", name: "Nadia", role: "Finance" },
}),
),
]);
expect((await screen.findByText(/is working on/)).textContent).toBe(
`James is working on ${TITLE}.`,
);
await user.click(screen.getByRole("button", { name: "Next expert" }));
expect(screen.getByRole("status").textContent).toBe(
`Nadia is hired, but ${TITLE} didn't start.`,
);
expect(screen.getByText("2 of 2")).toBeDefined();
});
it("keeps the full card for a hire saved without a hand-off", () => {
render(
<ExpertChangeCard
output={{
type: "expert_change_applied",
applied: true,
kind: "hire",
expert,
}}
/>,
);
expect(screen.getByText("Expert hired")).toBeDefined();
expect(screen.queryByRole("link")).toBeNull();
expect(screen.queryByRole("status")).toBeNull();
});
});
describe("hand-off parts follow the theme", () => {
// Raw palette classes don't follow the dark theme; physical margins don't
// flip in RTL. Every hand-off state must use tokens and logical sides.
const RAW =
/\b(?:bg|text|border|ring|divide)-(?:zinc|neutral|gray|white|black|amber|slate)\b|\bml-|\bmr-|\bpl-|\bpr-/;
// The Button atom's and the avatar's own classes belong to the design
// system, not here.
function classesIn(container: HTMLElement) {
return [container, ...container.querySelectorAll<HTMLElement>("[class]")]
.filter(
(el) =>
el.tagName !== "BUTTON" &&
!el.closest("[data-slot='expert-avatar'], [data-slot='avatar']"),
)
.flatMap((el) => el.getAttribute("class")?.split(/\s+/) ?? [])
.filter((cls) => RAW.test(cls.replace(/^[a-z-]+:/, "")));
}
it("uses only design tokens on the proposal card", () => {
const { container } = renderGroup([hire("a")]);
expect(classesIn(container)).toEqual([]);
});
it("uses only design tokens on the receipt", async () => {
const user = userEvent.setup();
const { container } = renderGroup([hire("a")]);
await user.click(screen.getByRole("button", { name: HAND_OFF }));
expect(classesIn(container)).toEqual([]);
});
it.each(["sent", "queued", "unconfirmed", "failed"])(
"uses only design tokens on the %s status",
(status) => {
const { container } = render(
<ExpertChangeCard
output={appliedOutput(
status,
status === "failed" ? null : "/copilot?sessionId=t",
)}
/>,
);
expect(classesIn(container)).toEqual([]);
},
);
});
@@ -431,6 +431,17 @@ describe("SubSessionLive", () => {
expect(screen.queryByText("Looking for the right agent now.")).toBeNull();
});
it("shows a queued delegation as running once its session runs", async () => {
server.use(subSession([], { chat_status: "running", active_stream: null }));
render(
<SubSessionCard output={{ status: "queued", sub_session_id: "sub-1" }} />,
);
expect(await screen.findByText("running")).toBeDefined();
expect(screen.queryByText("queued")).toBeNull();
});
it("says live updates failed instead of spinning on running forever", async () => {
server.use(failingSubSession());
@@ -0,0 +1,97 @@
import { asObject, str } from "./resultHelpers";
export interface HandoffTask {
title: string;
brief: string;
}
type HandoffStatus = "sent" | "queued" | "unconfirmed" | "failed";
export interface HandoffOutcome {
title: string;
status: HandoffStatus;
subSessionId: string | null;
link: string | null;
/** Only for a failed hand-off: the backend says sending it again can help.
* A refusal (paused, out of budget, no free slot) is never retryable. */
retry: boolean;
}
/** The task a hire was proposed for, when the proposal offers the hand-off. */
export function handoffTaskOf(
output: Record<string, unknown>,
): HandoffTask | null {
const handoff = asObject(output.handoff);
if (!handoff || !str(output, "handoff_confirmation_id")) return null;
const title = str(handoff, "title");
const brief = str(handoff, "brief");
return title && brief ? { title, brief } : null;
}
/** Whether the task a hire was approved with reached the new expert. */
export function handoffOutcomeOf(
output: Record<string, unknown>,
): HandoffOutcome | null {
const handoff = asObject(output.handoff);
const title = handoff && str(handoff, "title");
if (!handoff || !title) return null;
return {
title,
status: statusOf(handoff),
subSessionId: str(handoff, "sub_session_id") ?? null,
link: str(handoff, "link") ?? null,
retry: handoff.retry === true,
};
}
function statusOf(handoff: Record<string, unknown>): HandoffStatus {
const status = str(handoff, "status");
return status === "sent" || status === "queued" || status === "unconfirmed"
? status
: "failed";
}
/** The status the confirm froze for a task that reached the thread, so the
* card can follow the thread live from there. An unconfirmed task may never
* have started, so an idle thread proves nothing about it. */
export function frozenStatusOf(outcome: HandoffOutcome): string | null {
if (outcome.status === "sent") return "running";
if (outcome.status === "queued") return "queued";
return null;
}
export type HandoffState =
| "failed"
| "unconfirmed"
| "waiting"
| "working"
| "stopped"
| "unknown";
/** What the card says now: the thread's live status once the poll has read
* it, and the outcome frozen at hire time only until the first read. A poll
* that died ("unknown") claims nothing: the task was sent, and whether it
* is running is not known. */
export function handoffStateOf(
outcome: HandoffOutcome,
live: string | null,
): HandoffState {
if (outcome.status === "failed") return "failed";
if (outcome.status === "unconfirmed") return "unconfirmed";
if (live === "unknown") return "unknown";
if (live === "completed") return "stopped";
if (live === "queued") return "waiting";
if (live === "running") return "working";
return outcome.status === "queued" ? "waiting" : "working";
}
/** Subject of the decision line for "hire and start". It reads as one
* phrase after the verb, so the gate's regex and the decision note both
* keep working: "hire James to take on the Printify order pipeline". */
export function handoffSubject(name: string, task: HandoffTask): string {
return `${name} to take on ${task.title}`;
}
export function possessive(name: string): string {
return `${name}'s`;
}
@@ -0,0 +1,28 @@
import { useState } from "react";
import { useGetV2ListSessions } from "@/app/api/__generated__/endpoints/chat/chat";
/** Try again is the one retry path: Otto is told not to resend a failed
* hand-off himself. Someone may still have started work with the hire since
* (the user, by asking Otto in text). A hire is brand new and has no chat of
* its own until someone starts one, so any thread with them means the task
* may already be there: the card then points at it instead of offering a
* second copy. Read fresh once the turn is over (this mounts only then),
* never from a cache filled mid-turn. */
export function useHandoffRetry(expertId: string) {
const [asked, setAsked] = useState(false);
const { data, isFetchedAfterMount, isError } = useGetV2ListSessions(
{ expert_id: expertId, limit: 1, pinned_first: false },
{
query: {
select: (res) => (res.status === 200 ? res.data.sessions : null),
},
},
);
const known = isFetchedAfterMount && !isError && !!data;
return {
asked,
markAsked: () => setAsked(true),
known,
threadId: known ? (data[0]?.id ?? null) : null,
};
}