diff --git a/cockpit/runtimes/aws-strands/angular/e2e/aws-strands.spec.ts b/cockpit/runtimes/aws-strands/angular/e2e/aws-strands.spec.ts index 42266309c..38ebee249 100644 --- a/cockpit/runtimes/aws-strands/angular/e2e/aws-strands.spec.ts +++ b/cockpit/runtimes/aws-strands/angular/e2e/aws-strands.spec.ts @@ -1,5 +1,6 @@ // SPDX-License-Identifier: MIT import { test, expect } from '@playwright/test'; +import { submitAndWaitForResponse } from '@threadplane-internal/e2e-harness'; // Second proof outside unit tests that the neutral Agent contract's // interrupt path works against a genuinely non-LangGraph AG-UI backend: the @@ -37,3 +38,22 @@ test.describe('cockpit runtimes/aws-strands: meeting booking approval', () => { await expect(page.getByText(/is booked for Tuesday 10:00/i)).toBeVisible({ timeout: 30_000 }); }); }); + +// Delegation over the same non-LangGraph bridge: the orchestrator's +// `research_availability` async-generator tool re-yields the specialist's +// stream, and the per-tool ToolBehavior handler (src/subagent_emitter.py) +// translates it into SUBAGENT_STARTED / attributed TEXT_MESSAGE_* / +// SUBAGENT_FINISHED wire events. The @threadplane/ag-ui reducer keys the +// subagent to its spawning toolCallId, so renders the +// delegation inline as a instead of a tool-call chip. +test.describe('cockpit runtimes/aws-strands: subagent delegation', () => { + test('rt-strands: delegated availability research renders a streaming subagent card', async ({ page }) => { + const bubble = await submitAndWaitForResponse( + page, + 'Find a slot for Ada and Grace next week — research their availability first', + ); + await expect(page.locator('chat-subagent-card')).toHaveCount(1); + await expect(page.locator('chat-subagent-card')).toContainText('availability_researcher'); + await expect(bubble).toContainText(/slot|available/i); + }); +}); diff --git a/cockpit/runtimes/aws-strands/angular/e2e/fixtures/aws-strands.json b/cockpit/runtimes/aws-strands/angular/e2e/fixtures/aws-strands.json index 5f5ca4389..5c93c8bc5 100644 --- a/cockpit/runtimes/aws-strands/angular/e2e/fixtures/aws-strands.json +++ b/cockpit/runtimes/aws-strands/angular/e2e/fixtures/aws-strands.json @@ -30,6 +30,32 @@ } ] } + }, + { + "match": { "userMessage": "Ada and Grace", "hasToolResult": true }, + "response": { + "content": "Both are available Tuesday 10:00 next week — that slot works for Ada and Grace." + } + }, + { + "match": { "systemMessage": "availability researcher" }, + "response": { + "content": "- Ada: free Tuesday 10:00–12:00 and Thursday afternoon next week.\n- Grace: free Tuesday 10:00–11:30 and Friday morning next week.\n- Overlap: Tuesday 10:00 works for both." + } + }, + { + "match": { "userMessage": "Ada and Grace" }, + "response": { + "toolCalls": [ + { + "name": "research_availability", + "arguments": { + "attendees": "Ada, Grace", + "date_range": "next week" + } + } + ] + } } ] } diff --git a/cockpit/runtimes/aws-strands/angular/e2e/manual/subagent-card-live.png b/cockpit/runtimes/aws-strands/angular/e2e/manual/subagent-card-live.png new file mode 100644 index 000000000..72c0c6cfa Binary files /dev/null and b/cockpit/runtimes/aws-strands/angular/e2e/manual/subagent-card-live.png differ diff --git a/cockpit/runtimes/aws-strands/python/docs/wire-capture-subagents.md b/cockpit/runtimes/aws-strands/python/docs/wire-capture-subagents.md new file mode 100644 index 000000000..520222b19 --- /dev/null +++ b/cockpit/runtimes/aws-strands/python/docs/wire-capture-subagents.md @@ -0,0 +1,289 @@ +# Strands agents-as-tools: wire capture + emitter-seam decision + +Task 0 spike evidence for translating Strands agents-as-tools delegation into +AG-UI `SUBAGENT_*` events. Captured 2026-09-02 against the live meeting-scheduler +backend (`src/agent.py` + `src/server.py`, bridge pinned to git rev +`363d3878e30887e88c1fd5ca1916ec3a5962b6be`), scratch delegation tool +`research_availability` wrapping a tool-less specialist `Agent(name="availability_researcher")`. +The scratch edit was reverted after capture; only this doc lands. + +All bridge citations are into the installed venv source: +`.venv/lib/python3.14/site-packages/ag_ui_strands/` (referred to as `agent.py` +etc. below). + +## 1. Seam decision: `ToolBehavior.tool_stream_event_handler` (config-level), not a subclass + +**Decision: emit `SUBAGENT_*` from a per-tool `tool_stream_event_handler` +registered in `StrandsAgentConfig.tool_behaviors["research_availability"]`, +with the delegation tool written as an async-generator `@tool` that re-yields +the specialist's `stream_async` events.** + +The exact hook point: the bridge's run loop dispatches every +`tool_stream_event` to the tool's registered handler in +`StrandsAgent.run`, `agent.py:4644-4663` — the handler is an async generator +called with a `ToolStreamEventContext` (`config.py:56-91`) and *"may yield zero +or more AG-UI Event objects which are forwarded directly into the top-level +event stream"* (`config.py:76-91`). Registering a handler suppresses the +default routing for that tool (state snapshots at `agent.py:4664-4669`, +agent-as-tool lifecycle forwarding at `agent.py:4670-4682`), so the handler +owns the whole child stream. + +Why not the LangGraph lane's dispatch-hook subclass +(`cockpit/ag-ui/subagents/python/src/streaming/activity_emitting_agent.py` +overriding `_dispatch_event`): + +- (1b) **A `@tool` body cannot reach an emitter directly, but it does not need + one.** Strands wraps every value an async-generator tool yields as a + `tool_stream_event` in the parent stream (`agent.py:4596-4598`), and the + bridge hands that payload to the per-tool handler with `tool_use_id` and + `tool_name` attached (`agent.py:4648-4652`). That IS the sanctioned + tool-body-to-wire channel — no writer/contextvar/queue plumbing, no fork of + the 5,771-line bridge module. +- The Strands bridge has no `_dispatch_event` seam at all: `StrandsAgent.run` + (`agent.py:2956` onward) is one ~2,800-line async generator with dozens of + inline `yield` sites. A subclass would have to wrap the entire generator and + pattern-match already-serialized events to find the delegation window — + strictly worse information than the handler gets (raw inner Strands events, + pre-translation). +- (1a) Strands→AG-UI translation happens inline in that same `run` generator + (text deltas, `current_tool_use`, `contentBlockStop`, tool results — e.g. + the `tool_stream_event` branch at `agent.py:4596-4682`, tool results at + `agent.py:4684+`), then each pydantic event is SSE-serialized by + `EventEncoder.encode` → `event.model_dump_json(by_alias=True)` + (`ag_ui/encoder/encoder.py:22-36`), called from the endpoint's + `event_generator` (`endpoint.py:290-343`). +- (1c) **Unknown raw dicts do NOT pass through.** The encoder requires pydantic + `BaseEvent` instances (`model_dump_json` call, `encoder.py:36`); a plain dict + would crash the stream. Unmapped *Strands* events are forwarded only as + sanitized `RawEvent` payloads (`_sanitize_raw_event`, `agent.py:1089-1125`) + — and inner-agent payload keys (`data`, `current_tool_use`, ...) are in + `_RAW_SUPPRESSED_KEYS` (`agent.py:1078-1086`), so nothing from the child + leaks via RAW either. The handler must therefore yield real + `ag_ui.core` event objects — which exist, see §2. +- (1d) **`tool_stream_event` from a nested agent-as-tool is forwarded today + only for the inner TOOL-CALL lifecycle, never for inner text.** + `_forward_inner_agent_events` (`agent.py:1195-1313`, invoked at + `agent.py:4677-4682`) translates inner `current_tool_use` / + `contentBlockStop` / `toolResult` into namespaced `TOOL_CALL_*` events and + explicitly nothing else ("Only the tool-call lifecycle is forwarded", + `agent.py:1209`). Inner `{"data": ...}` text deltas fall through every + branch and are dropped. The live capture in §3 confirms this on the wire. + +### Emitter shape (next task) + +- Delegation tool: async-generator `@tool` that does + `async for event in specialist.stream_async(prompt): yield event`, then + yields the accumulated text as its final value (Strands takes the last + yielded value as the tool result — confirmed in §3, the `TOOL_CALL_RESULT` + content is exactly the joined child text). +- Handler on that tool: lazily emits `SubagentStartedEvent` + (`subagent_run_id=f"{ctx.tool_use_id}-sub"`, `name="availability_researcher"`, + `parent_tool_call_id=ctx.tool_use_id`) on the first inner event, translates + inner `{"data": }` into `TEXT_MESSAGE_START/CONTENT/END` carrying + `subagent_run_id`, and emits `SubagentFinishedEvent(outcome=success)` when it + sees the inner terminal `{"result": AgentResult}` event; + `SubagentErrorEvent` when the inner stream surfaces an error / + `forceStop`. `ctx.tool_use_id` equals the wire `toolCallId` (the model's + `call_...` id — §3 line 4 vs. the handler context), so `parentToolCallId` + lines up with the bridge-native `TOOL_CALL_START` with zero bookkeeping. +- Wire-order nuance: the bridge emits `TOOL_CALL_END` when the *args* finish + streaming, before the tool executes (§3 line 19). So the shipped order will + be `TOOL_CALL_START → TOOL_CALL_ARGS → TOOL_CALL_END → SUBAGENT_STARTED → + TEXT_MESSAGE_* → SUBAGENT_FINISHED → TOOL_CALL_RESULT` — the `SUBAGENT_*` + block nests inside the tool call's start/result span, not inside + start/end. + +## 2. SDK check + +``` +$ uv run python -c "import ag_ui.core as c; print([n for n in dir(c) if 'Subagent' in n])" +['SubagentErrorEvent', 'SubagentFinishedEvent', 'SubagentFinishedOutcome', + 'SubagentFinishedSuccessOutcome', 'SubagentFinishedSuspendedOutcome', + 'SubagentStartedEvent'] +``` + +The pinned SDK has first-class subagent events (`ag_ui/core/events.py:455-512`) +with exactly the target fields (`subagent_run_id`, `name`, +`parent_tool_call_id`, `outcome` discriminated union), and every +`TextMessage*` / `ToolCall*` event carries an optional `subagent_run_id` +(`events.py:127-235`). **No raw-dict fallback is needed** — the handler +constructs typed events and the stock encoder serializes them. + +## 3. Live captures (scrubbed) + +Prompt: *"Find a slot for Ada and Grace next week — research their availability +first"*. The model called `research_availability` on the first attempt in both +runs. No API keys or org ids appeared in either stream; nothing was scrubbed — +only long `MESSAGES_SNAPSHOT` lines and repetitive delta runs are elided, each +marked with a `# [elided: ...]` comment. + +### 3a. Async-generator delegation tool (the seam-relevant variant) + +The tool re-yielded the specialist's entire `stream_async` output, so every +child event crossed the bridge as `tool_stream_event` — and the wire between +`TOOL_CALL_END` (line 19) and `TOOL_CALL_RESULT` (line 21) still carries +**zero child events**: `_forward_inner_agent_events` dropped every inner text +delta (the child called no tools, so nothing was forwardable). Line numbers +refer to non-blank SSE lines. + +``` +data: {"type":"RUN_STARTED","threadId":"spike-thread-2","runId":"spike-run-2"} +data: {"type":"STATE_SNAPSHOT","snapshot":{}} +data: {"type":"MESSAGES_SNAPSHOT","messages":[{"id":"u1","role":"user","content":"Find a slot for Ada and Grace next week — research their availability first"}]} +data: {"type":"TOOL_CALL_START","toolCallId":"call_vF6Vc6Wzl40vz9pBZOOrDxS7","toolCallName":"research_availability","parentMessageId":"0130e374-95eb-4a08-aed3-2b6f877331c6"} +data: {"type":"TOOL_CALL_ARGS","toolCallId":"call_vF6Vc6Wzl40vz9pBZOOrDxS7","delta":"{\""} +# [elided: 12 more TOOL_CALL_ARGS deltas spelling {"attendees": "Ada, Grace", "date_range": "next week"}] +data: {"type":"TOOL_CALL_ARGS","toolCallId":"call_vF6Vc6Wzl40vz9pBZOOrDxS7","delta":"\"}"} +data: {"type":"TOOL_CALL_END","toolCallId":"call_vF6Vc6Wzl40vz9pBZOOrDxS7"} +# [elided: MESSAGES_SNAPSHOT mirroring the assistant tool-call message] +# <-- the specialist ran HERE; its interim + final text produced tool_stream_events, none reached the wire +data: {"type":"TOOL_CALL_RESULT","messageId":"2d0594ca-dc44-4c24-841a-1461d55759fc","toolCallId":"call_vF6Vc6Wzl40vz9pBZOOrDxS7","content":"\"To provide a summary of likely availability windows for Ada and Grace for the next week, I will check their schedules. Please hold on for a moment.\\nI actually do not have access to the scheduling information for the attendees. Please provide their typical availability or any specific constraints you might know about them, and I can help you summarize likely availability windows accordingly.\""} +# [elided: MESSAGES_SNAPSHOT adding the tool-result message] +data: {"type":"TEXT_MESSAGE_START","messageId":"51b44583-454d-4252-803a-1f91fa681f5e","role":"assistant"} +data: {"type":"TEXT_MESSAGE_CONTENT","messageId":"51b44583-454d-4252-803a-1f91fa681f5e","delta":"I"} +# [elided: 27 more TEXT_MESSAGE_CONTENT deltas — the ORCHESTRATOR's own summary, after the tool result] +data: {"type":"TEXT_MESSAGE_CONTENT","messageId":"51b44583-454d-4252-803a-1f91fa681f5e","delta":"."} +data: {"type":"TEXT_MESSAGE_END","messageId":"51b44583-454d-4252-803a-1f91fa681f5e"} +# [elided: final MESSAGES_SNAPSHOT] +data: {"type":"STATE_SNAPSHOT","snapshot":{}} +data: {"type":"RUN_FINISHED","threadId":"spike-thread-2","runId":"spike-run-2","outcome":{"type":"success"}} +``` + +Event tally: 1 RUN_STARTED, 2 STATE_SNAPSHOT, 4 MESSAGES_SNAPSHOT, +1 TOOL_CALL_START, 14 TOOL_CALL_ARGS, 1 TOOL_CALL_END, 1 TOOL_CALL_RESULT, +1 TEXT_MESSAGE_START, 29 TEXT_MESSAGE_CONTENT, 1 TEXT_MESSAGE_END, +1 RUN_FINISHED. No SUBAGENT_*, no RAW, no CUSTOM, no STEP_*. + +### 3b. Plain sync delegation tool (the naive idiom) + +Same prompt, tool body `result = availability_researcher(...); return str(result)`. +Identical wire shape: `TOOL_CALL_START/ARGS×14/END → MESSAGES_SNAPSHOT → +TOOL_CALL_RESULT` (child's full multi-paragraph answer as one string) `→ +MESSAGES_SNAPSHOT → TEXT_MESSAGE_*` (orchestrator summary) `→ RUN_FINISHED +success`. A sync tool yields nothing mid-flight, so no `tool_stream_event` +fires at all — the child is a black box by construction, and this variant can +never feed a subagent emitter. (Capture withheld here as it adds nothing over +3a; tally: 24 TEXT_MESSAGE_CONTENT, otherwise identical event mix.) + +## 4. Did child tokens appear on the wire? + +**No — in neither variant.** During the delegation call the stream goes +straight from `TOOL_CALL_END` (§3a line 19) to `TOOL_CALL_RESULT` (§3a line +21) with only a `MESSAGES_SNAPSHOT` between. The specialist's interim sentence +("To provide a summary ... Please hold on for a moment.") exists in the run — +it surfaces verbatim *inside* the final `TOOL_CALL_RESULT` content — proving +the child streamed internally and the bridge dropped the deltas +(`_forward_inner_agent_events` forwards tool-call lifecycle only, +`agent.py:1195-1313`). Natively, delegation is: parent tool-call args stream → +silence → one opaque result string. This is the matrix cell the emitter fixes: +child tokens must be re-emitted by our `tool_stream_event_handler` as +`TEXT_MESSAGE_*` events carrying `subagentRunId`. + +## 5. Other observations + +- Restreaming the child through the generator tool logs repeated + `ValueError: was created + in a different Context` server-side (OTel context tokens crossing task + boundaries; cosmetic with OTel disabled, but worth watching once the real + emitter lands). +- `RunAgentInput` requires `threadId`, `runId`, `messages`, `tools`, + `context`, `forwardedProps` (camelCase; `ag_ui/core/types.py:396-412`); + the endpoint validates with `model_validate` and 422s otherwise + (`endpoint.py:55-86`). +- `emit_messages_snapshot` (on by default) interleaves full + `MESSAGES_SNAPSHOT`s after every tool END/RESULT and TEXT_MESSAGE_END — + the future emitter must not splice subagent messages into those snapshots + (mirroring `_forward_inner_agent_events`' deliberate choice, + `agent.py:1209-1211`). + +## After the emitter + +Captured 2026-09-02 against the shipped scenario (`research_availability` +registered with `ToolBehavior(tool_stream_event_handler=emit_subagent_events)`, +`src/subagent_emitter.py`), same `RunAgentInput` as §3 (prompt *"Find a slot +for Ada and Grace next week — research their availability first"*, empty +`tools`/`context`/`state`/`forwardedProps`). No keys or org ids appeared in +the stream; only repetitive delta runs and `MESSAGES_SNAPSHOT`s are elided, +marked with `# [elided: ...]`. + +The model delegated TWICE this run (first to ask for concrete dates, then to +research them) — both rounds carried the full `SUBAGENT_*` block, each nested +between its own `TOOL_CALL_END` and `TOOL_CALL_RESULT` exactly as §1 +predicted. First round shown; the second is shape-identical with +`toolCallId=call_ur6NyfucZhR4FlDTxrDt0hy9` and 61 content deltas. + +``` +data: {"type":"RUN_STARTED","threadId":"smoke-thread-1","runId":"smoke-run-1"} +data: {"type":"STATE_SNAPSHOT","snapshot":{}} +# [elided: MESSAGES_SNAPSHOT] +data: {"type":"TOOL_CALL_START","toolCallId":"call_YW6LslLJbLNZul9qYlYfeySg","toolCallName":"research_availability","parentMessageId":"6228ef46-8396-4044-903c-ef9e3f3abc3d"} +# [elided: 14 TOOL_CALL_ARGS deltas spelling {"attendees": "Ada, Grace", "date_range": "next week"}] +data: {"type":"TOOL_CALL_END","toolCallId":"call_YW6LslLJbLNZul9qYlYfeySg"} +# [elided: MESSAGES_SNAPSHOT] +data: {"type":"SUBAGENT_STARTED","subagentRunId":"call_YW6LslLJbLNZul9qYlYfeySg-sub","name":"availability_researcher","parentToolCallId":"call_YW6LslLJbLNZul9qYlYfeySg"} +data: {"type":"TEXT_MESSAGE_START","messageId":"call_YW6LslLJbLNZul9qYlYfeySg-sub-m1","role":"assistant","subagentRunId":"call_YW6LslLJbLNZul9qYlYfeySg-sub"} +data: {"type":"TEXT_MESSAGE_CONTENT","messageId":"call_YW6LslLJbLNZul9qYlYfeySg-sub-m1","delta":"Please","subagentRunId":"call_YW6LslLJbLNZul9qYlYfeySg-sub"} +data: {"type":"TEXT_MESSAGE_CONTENT","messageId":"call_YW6LslLJbLNZul9qYlYfeySg-sub-m1","delta":" provide","subagentRunId":"call_YW6LslLJbLNZul9qYlYfeySg-sub"} +# [elided: 25 more TEXT_MESSAGE_CONTENT deltas — the CHILD's text, token by token] +data: {"type":"TEXT_MESSAGE_CONTENT","messageId":"call_YW6LslLJbLNZul9qYlYfeySg-sub-m1","delta":".","subagentRunId":"call_YW6LslLJbLNZul9qYlYfeySg-sub"} +data: {"type":"TEXT_MESSAGE_END","messageId":"call_YW6LslLJbLNZul9qYlYfeySg-sub-m1","subagentRunId":"call_YW6LslLJbLNZul9qYlYfeySg-sub"} +data: {"type":"SUBAGENT_FINISHED","subagentRunId":"call_YW6LslLJbLNZul9qYlYfeySg-sub","outcome":{"type":"success"}} +data: {"type":"TOOL_CALL_RESULT","messageId":"b4c51965-4781-4683-8808-05fddd85135f","toolCallId":"call_YW6LslLJbLNZul9qYlYfeySg","content":"\"Please provide the specific dates for next week (e.g., from October 16 to October 20) so I can summarize the availability accordingly.\""} +# [elided: MESSAGES_SNAPSHOT, then the second delegation round — TOOL_CALL_START/ARGS×19/END, +# SUBAGENT_STARTED, TEXT_MESSAGE_START, 61 TEXT_MESSAGE_CONTENT deltas, TEXT_MESSAGE_END, +# SUBAGENT_FINISHED success, TOOL_CALL_RESULT — ids derived from call_ur6NyfucZhR4FlDTxrDt0hy9] +data: {"type":"TEXT_MESSAGE_START","messageId":"77635fb9-a418-443a-af7e-c70a29d079f6","role":"assistant"} +# [elided: 26 TEXT_MESSAGE_CONTENT deltas — the ORCHESTRATOR's own summary, no subagentRunId] +data: {"type":"TEXT_MESSAGE_END","messageId":"77635fb9-a418-443a-af7e-c70a29d079f6"} +# [elided: final MESSAGES_SNAPSHOT] +data: {"type":"STATE_SNAPSHOT","snapshot":{}} +data: {"type":"RUN_FINISHED","threadId":"smoke-thread-1","runId":"smoke-run-1","outcome":{"type":"success"}} +``` + +Event tally (174 SSE lines): 1 RUN_STARTED, 2 STATE_SNAPSHOT, +6 MESSAGES_SNAPSHOT, 2 TOOL_CALL_START, 33 TOOL_CALL_ARGS, 2 TOOL_CALL_END, +2 SUBAGENT_STARTED, 2 TEXT_MESSAGE_START(sub), 89 TEXT_MESSAGE_CONTENT(sub), +2 TEXT_MESSAGE_END(sub), 2 SUBAGENT_FINISHED, 2 TOOL_CALL_RESULT, +1 TEXT_MESSAGE_START, 26 TEXT_MESSAGE_CONTENT, 1 TEXT_MESSAGE_END, +1 RUN_FINISHED. No RAW, no CUSTOM, no STEP_*, no SUBAGENT_ERROR. + +**Child deltas: streaming (89 content events across the two delegation +rounds: 28 + 61)** — §4's black box is gone. Every child event carries +`subagentRunId` derived from the wire `toolCallId` +(`-sub` / `-sub-m1`), `SUBAGENT_STARTED.parentToolCallId` +matches the bridge-native `TOOL_CALL_START.toolCallId` verbatim, and the +`TOOL_CALL_RESULT` content equals the joined child deltas (the tool's final +yield), confirming the last-yield-as-result contract survived the handler +registration. The §5 OTel `ContextVar` warnings still log server-side and +remain cosmetic. + +## Browser verification + +2026-09-02, live backend (real `OPENAI_API_KEY`, uvicorn on :5331) + `nx +serve cockpit-runtimes-aws-strands-angular` on :4331, driven headlessly +with Playwright. Screenshot: +`cockpit/runtimes/aws-strands/angular/e2e/manual/subagent-card-live.png`. + +What rendered: the delegation prompt (*"Find a slot for Ada and Grace next +week — research their availability first"*) produced an inline +`` anchored to the `research_availability` tool call — +header `availability_researcher` + wire `toolCallId` + status badge — with +the specialist's transcript inside it, followed by the orchestrator's own +summary bubble. The child text never leaked into the parent bubble, and the +card persists (collapsed to `complete`) after the run. + +Did the card text stream mid-run: **yes** (matrix cell: expected +streaming = yes). Polling the card's `innerText` every 150ms during the +run showed the badge at `running` while the specialist's message grew +monotonically across successive samples (one run: lengths 144 → 213 → 288 +→ 403 → 420 → 438 → 501 → 643 chars between t≈2.0s and t≈3.5s), then +flipped to `complete` and collapsed. This confirms the attributed +`TEXT_MESSAGE_CONTENT` deltas render progressively in the card, not as one +post-hoc paste. + +One rendering defect surfaced and was fixed on this branch: the wire and +the `@threadplane/ag-ui` reducer were both correct, but `chat-tool-calls` +anchored subagent cards by the adapter's `subagents()` **map key** (here +the `subagentRunId`, `-sub`) instead of the contract field +`Subagent.toolCallId`, so the card never mounted for Strands delegations. +Fixed in `libs/chat` by re-indexing on `Subagent.toolCallId`. diff --git a/cockpit/runtimes/aws-strands/python/pyproject.toml b/cockpit/runtimes/aws-strands/python/pyproject.toml index a95c6f092..d93b5f47a 100644 --- a/cockpit/runtimes/aws-strands/python/pyproject.toml +++ b/cockpit/runtimes/aws-strands/python/pyproject.toml @@ -24,9 +24,18 @@ dependencies = [ [tool.uv.sources] ag-ui-strands = { git = "https://github.com/ag-ui-protocol/ag-ui.git", rev = "363d3878e30887e88c1fd5ca1916ec3a5962b6be", subdirectory = "integrations/aws-strands/python" } +[dependency-groups] +dev = [ + "pytest>=8.0", + "pytest-asyncio>=0.23", +] + [build-system] requires = ["hatchling"] build-backend = "hatchling.build" [tool.hatch.build.targets.wheel] packages = ["src"] + +[tool.pytest.ini_options] +asyncio_mode = "auto" diff --git a/cockpit/runtimes/aws-strands/python/src/agent.py b/cockpit/runtimes/aws-strands/python/src/agent.py index a74c3e5ea..0a709ee30 100644 --- a/cockpit/runtimes/aws-strands/python/src/agent.py +++ b/cockpit/runtimes/aws-strands/python/src/agent.py @@ -19,11 +19,16 @@ resumes from the client's top-level ``resume`` entries keyed by ``interruptId`` (never the LangGraph bridge's CUSTOM ``on_interrupt``). -No subagents surface and no multi-agent route: the Strands bridge routes -delegation through CUSTOM MultiAgentHandoff + STEP_* with zero ACTIVITY -events (measured red upstream), and multi-agent routes crash the stale -PyPI ``ag-ui-strands`` 0.3.0 wheel — which is why this example pins the -bridge to a git ref (see pyproject.toml). +- subagents: ``research_availability`` delegates to a tool-less specialist + ``availability_researcher`` Agent via an async-generator ``@tool`` that + re-yields the specialist's ``stream_async`` events; a per-tool + ``ToolBehavior.tool_stream_event_handler`` (src/subagent_emitter.py) + translates them into standard ``SUBAGENT_*`` + child ``TEXT_MESSAGE_*`` + wire events. (The bridge natively drops inner text deltas and would + otherwise route delegation through CUSTOM MultiAgentHandoff + STEP_*; + multi-agent routes also crash the stale PyPI ``ag-ui-strands`` 0.3.0 + wheel — which is why this example pins the bridge to a git ref, see + pyproject.toml and docs/wire-capture-subagents.md.) Model: Strands' native OpenAI provider on plain ``OPENAI_API_KEY`` — no AWS credentials involved. ``OPENAI_BASE_URL`` is honored, which is how the @@ -45,6 +50,8 @@ from ag_ui_strands import StrandsAgent, StrandsAgentConfig, ToolBehavior +from .subagent_emitter import emit_subagent_events + _SLOTS = { "monday": ["09:00", "13:30"], "tuesday": ["10:00", "15:00"], @@ -139,6 +146,41 @@ async def booking_state(context) -> dict | None: return _complete_state() +_RESEARCHER_INSTRUCTIONS = ( + "You are an availability researcher. Given attendee names and a date " + "range, produce a short bullet summary of likely availability windows. " + "Be concise: 3 bullets max." +) + + +@tool +async def research_availability(attendees: str, date_range: str): + """Delegate availability research for the given attendees to a specialist. + + Args: + attendees: Comma-separated attendee names, e.g. 'Ada, Grace'. + date_range: The window to research, e.g. 'next week'. + + Returns: + The specialist's bullet summary of likely availability windows. + """ + chunks: list[str] = [] + try: + async for event in availability_researcher.stream_async( + f"Attendees: {attendees}\nDate range: {date_range}" + ): + if isinstance(event, dict) and isinstance(event.get("data"), str): + chunks.append(event["data"]) + yield event + except Exception as exc: # pragma: no cover - not reachable without a live model failure + # Surface the failure to the emitter (which owns the SUBAGENT_ERROR + # wire event), then let the tool error propagate to Strands normally. + yield {"delegation_error": str(exc)} + raise + # Strands takes the LAST yielded value as the tool result. + yield "".join(chunks) + + _INSTRUCTIONS = """You are a meeting scheduling copilot. When the user asks to book a meeting: @@ -155,6 +197,9 @@ async def booking_state(context) -> dict | None: Keep every response brief and factual. Never invent availability — use the tool. + +When the user asks about attendees' availability, delegate that research to +the `research_availability` tool before proposing meeting slots. """ @@ -172,11 +217,22 @@ def build_model() -> OpenAIModel: return OpenAIModel(client_args=client_args, model_id=os.environ.get("OPENAI_CHAT_MODEL", "gpt-4o-mini")) +# Tool-less specialist the orchestrator delegates availability research to +# via the `research_availability` async-generator tool above. Its streamed +# events cross the bridge as tool_stream_events and are translated into +# SUBAGENT_* wire events by the emitter registered in ToolBehavior below. +availability_researcher = Agent( + model=build_model(), + system_prompt=_RESEARCHER_INSTRUCTIONS, + name="availability_researcher", + tools=[], +) + agent = StrandsAgent( agent=Agent( model=build_model(), system_prompt=_INSTRUCTIONS, - tools=[check_availability, book_meeting], + tools=[check_availability, book_meeting, research_availability], ), name="aws-strands", description="Books meetings with availability lookup, shared state, and human approval.", @@ -184,6 +240,9 @@ def build_model() -> OpenAIModel: tool_behaviors={ "check_availability": ToolBehavior(state_from_result=availability_state), "book_meeting": ToolBehavior(state_from_args=booking_state), + "research_availability": ToolBehavior( + tool_stream_event_handler=emit_subagent_events, + ), }, ), ) diff --git a/cockpit/runtimes/aws-strands/python/src/subagent_emitter.py b/cockpit/runtimes/aws-strands/python/src/subagent_emitter.py new file mode 100644 index 000000000..e858babb1 --- /dev/null +++ b/cockpit/runtimes/aws-strands/python/src/subagent_emitter.py @@ -0,0 +1,189 @@ +# SPDX-License-Identifier: MIT +"""SUBAGENT_* emitter for the `research_availability` delegation tool. + +The Strands bridge natively drops a child agent's text deltas: an +async-generator `@tool` re-yielding a specialist's ``stream_async`` events +produces ``tool_stream_event``s, but `_forward_inner_agent_events` forwards +only the inner TOOL-CALL lifecycle and never inner text (measured in +docs/wire-capture-subagents.md). Registering this +``ToolBehavior.tool_stream_event_handler`` claims the whole child stream and +re-emits it as standard AG-UI subagent wire events: + + SUBAGENT_STARTED (first inner event) + TEXT_MESSAGE_START/CONTENT.../END (inner ``data`` deltas, streamed) + SUBAGENT_FINISHED outcome=success (inner terminal ``result`` event) + +or ``SUBAGENT_ERROR`` when the delegation fails (the tool yields a +``delegation_error`` sentinel before re-raising, or the inner stream +force-stops). Ids derive from the wire ``toolCallId`` (``ctx.tool_use_id``) +so the client can key the subagent card on ``parentToolCallId`` with zero +bookkeeping. + +The bridge instantiates this handler ONCE PER EVENT (a fresh async generator +per ``tool_stream_event``), so per-invocation lifecycle state lives in a +module-level dict keyed by ``tool_use_id``. The encoder requires pydantic +``BaseEvent`` instances — raw dicts crash the stream — so only typed +``ag_ui.core`` events are yielded. +""" + +from dataclasses import dataclass + +from ag_ui.core import ( + EventType, + SubagentErrorEvent, + SubagentFinishedEvent, + SubagentFinishedSuccessOutcome, + SubagentStartedEvent, + TextMessageContentEvent, + TextMessageEndEvent, + TextMessageStartEvent, +) + +from ag_ui_strands import ToolStreamEventContext + +SPECIALIST_NAME = "availability_researcher" + + +@dataclass +class _DelegationState: + """Lifecycle of one delegation call, keyed by tool_use_id.""" + + started: bool = False + message_open: bool = False + finished: bool = False + generation: int = 1 + """Bumped when a reused tool_use_id starts a fresh inner stream, so the + re-run's message id (-m2, -m3, ...) never collides with an already-emitted + one.""" + + +# Finished entries are kept (not popped) so the tool's trailing result-string +# yield and any stragglers stay suppressed. Growth is capped at _MAX_SESSIONS +# (dict insertion order = age; each entry is a short key string plus a +# 4-field dataclass, ~414 bytes measured with sys.getsizeof — the cap bounds +# the dict at ~200 KiB). +_sessions: dict[str, _DelegationState] = {} +_MAX_SESSIONS = 512 + + +def _subagent_run_id(tool_use_id: str) -> str: + return f"{tool_use_id}-sub" + + +def _message_id(tool_use_id: str, generation: int) -> str: + return f"{tool_use_id}-sub-m{generation}" + + +async def emit_subagent_events(ctx: ToolStreamEventContext): + """tool_stream_event_handler translating child events to SUBAGENT_* wire + events. Async generator, called once per inner event.""" + state = _sessions.get(ctx.tool_use_id) + if state is None: + state = _DelegationState() + _sessions[ctx.tool_use_id] = state + while len(_sessions) > _MAX_SESSIONS: + del _sessions[next(k for k in _sessions if k != ctx.tool_use_id)] + run_id = _subagent_run_id(ctx.tool_use_id) + data = ctx.stream_data + if state.finished and isinstance(data, dict) and "init_event_loop" in data: + # Reused tool_use_id (real for some Strands providers — see the + # bridge's _reused_frontend_tool_identity_error): a fresh inner + # stream always opens with init_event_loop, so reset for a second + # full SUBAGENT_* sequence under the same subagent_run_id (the + # adapter treats an identity-unchanged re-announce as content-only). + # Non-init stragglers after the terminal stay swallowed below. + state = _DelegationState(generation=state.generation + 1) + _sessions[ctx.tool_use_id] = state + message_id = _message_id(ctx.tool_use_id, state.generation) + try: + if state.finished: + # The tool's final yield (the result string) and any stragglers + # arrive after the inner terminal event — nothing left to emit. + return + + if not state.started: + state.started = True + yield SubagentStartedEvent( + type=EventType.SUBAGENT_STARTED, + subagent_run_id=run_id, + name=SPECIALIST_NAME, + parent_tool_call_id=ctx.tool_use_id, + ) + + if isinstance(data, str): + # Terminal-success fallback: the tool's final result-string yield + # arriving on an UNFINISHED session means the inner stream ended + # without a {"result": ...} event — close out rather than leaving + # the subagent card open forever. + state.finished = True + if state.message_open: + state.message_open = False + yield TextMessageEndEvent( + type=EventType.TEXT_MESSAGE_END, + message_id=message_id, + subagent_run_id=run_id, + ) + yield SubagentFinishedEvent( + type=EventType.SUBAGENT_FINISHED, + subagent_run_id=run_id, + outcome=SubagentFinishedSuccessOutcome(), + ) + return + if not isinstance(data, dict): + return + + if "delegation_error" in data or data.get("force_stop"): + message = str( + data.get("delegation_error") + or data.get("force_stop_reason") + or "subagent stream force-stopped" + ) + state.finished = True + if state.message_open: + state.message_open = False + yield TextMessageEndEvent( + type=EventType.TEXT_MESSAGE_END, + message_id=message_id, + subagent_run_id=run_id, + ) + yield SubagentErrorEvent( + type=EventType.SUBAGENT_ERROR, + subagent_run_id=run_id, + message=message, + ) + elif isinstance(data.get("data"), str) and data["data"]: + if not state.message_open: + state.message_open = True + yield TextMessageStartEvent( + type=EventType.TEXT_MESSAGE_START, + message_id=message_id, + role="assistant", + subagent_run_id=run_id, + ) + yield TextMessageContentEvent( + type=EventType.TEXT_MESSAGE_CONTENT, + message_id=message_id, + delta=data["data"], + subagent_run_id=run_id, + ) + elif "result" in data: + state.finished = True + if state.message_open: + state.message_open = False + yield TextMessageEndEvent( + type=EventType.TEXT_MESSAGE_END, + message_id=message_id, + subagent_run_id=run_id, + ) + yield SubagentFinishedEvent( + type=EventType.SUBAGENT_FINISHED, + subagent_run_id=run_id, + outcome=SubagentFinishedSuccessOutcome(), + ) + except Exception as exc: # pragma: no cover - defensive: never crash the run + state.finished = True + yield SubagentErrorEvent( + type=EventType.SUBAGENT_ERROR, + subagent_run_id=run_id, + message=str(exc), + ) diff --git a/cockpit/runtimes/aws-strands/python/tests/test_delegation.py b/cockpit/runtimes/aws-strands/python/tests/test_delegation.py new file mode 100644 index 000000000..293b3cc95 --- /dev/null +++ b/cockpit/runtimes/aws-strands/python/tests/test_delegation.py @@ -0,0 +1,45 @@ +# SPDX-License-Identifier: MIT +"""Tests for the `research_availability` delegation scenario — the tool is a +registered async-generator `@tool` that hands availability research to the +tool-less `availability_researcher` specialist. No live model calls: these +tests only inspect registration metadata (the module builds its OpenAI +clients with a placeholder key that would 401 at request time).""" + +import inspect + +from src.agent import agent, availability_researcher, research_availability + + +def _tool_names() -> list[str]: + # Private-attr coupling (StrandsAgent._tools) is frozen by the git-ref + # pin on ag-ui-strands in pyproject.toml. + return [t.tool_name for t in agent._tools] + + +def test_research_availability_is_registered_on_the_agent(): + assert "research_availability" in _tool_names() + # Existing tools stay registered untouched. + assert "check_availability" in _tool_names() + assert "book_meeting" in _tool_names() + + +def test_tool_name_and_docstring(): + assert research_availability.tool_name == "research_availability" + description = research_availability.tool_spec["description"] + assert description.startswith( + "Delegate availability research for the given attendees to a specialist." + ) + schema = research_availability.tool_spec["inputSchema"]["json"] + assert set(schema["required"]) == {"attendees", "date_range"} + + +def test_tool_is_an_async_generator(): + # The seam depends on it: only an async-generator tool produces + # tool_stream_events for the bridge to hand to the subagent emitter. + assert inspect.isasyncgenfunction(research_availability._tool_func) + + +def test_specialist_is_toolless_researcher(): + assert availability_researcher.name == "availability_researcher" + assert availability_researcher.tool_registry.registry == {} + assert "availability researcher" in availability_researcher.system_prompt diff --git a/cockpit/runtimes/aws-strands/python/tests/test_subagent_emitter.py b/cockpit/runtimes/aws-strands/python/tests/test_subagent_emitter.py new file mode 100644 index 000000000..0f2c7a073 --- /dev/null +++ b/cockpit/runtimes/aws-strands/python/tests/test_subagent_emitter.py @@ -0,0 +1,222 @@ +# SPDX-License-Identifier: MIT +"""Tests for the SUBAGENT_* emitter — drives the tool_stream_event_handler +with synthetic Strands events (shapes copied from the Task-0 wire capture in +docs/wire-capture-subagents.md) and asserts the exact emitted AG-UI sequence +field-for-field. The bridge calls the handler once per event with a fresh +generator, which is exactly how these tests drive it.""" + +import pytest + +from ag_ui.core import EventType +from ag_ui_strands import ToolStreamEventContext + +from src import subagent_emitter +from src.subagent_emitter import emit_subagent_events + +TOOL_USE_ID = "call_vF6Vc6Wzl40vz9pBZOOrDxS7" +RUN_ID = f"{TOOL_USE_ID}-sub" +MESSAGE_ID = f"{TOOL_USE_ID}-sub-m1" + + +@pytest.fixture(autouse=True) +def _clean_sessions(): + subagent_emitter._sessions.clear() + yield + subagent_emitter._sessions.clear() + + +async def _drive(events: list, tool_use_id: str = TOOL_USE_ID) -> list: + """Feed synthetic stream payloads one at a time, the way the bridge + dispatches tool_stream_events, collecting everything yielded.""" + out = [] + for data in events: + ctx = ToolStreamEventContext( + tool_use_id=tool_use_id, + tool_name="research_availability", + stream_data=data, + ) + async for ev in emit_subagent_events(ctx): + out.append(ev) + return out + + +async def test_success_sequence_field_for_field(): + out = await _drive([ + {"init_event_loop": True}, # specialist loop init + {"start": True}, + {"data": "- Ada"}, # streamed text deltas + {"data": " is free"}, + {"data": " Tuesday"}, + {"result": object()}, # terminal AgentResult event + "- Ada is free Tuesday", # the tool's final yield (result string) + ]) + + types = [ev.type for ev in out] + assert types == [ + EventType.SUBAGENT_STARTED, + EventType.TEXT_MESSAGE_START, + EventType.TEXT_MESSAGE_CONTENT, + EventType.TEXT_MESSAGE_CONTENT, + EventType.TEXT_MESSAGE_CONTENT, + EventType.TEXT_MESSAGE_END, + EventType.SUBAGENT_FINISHED, + ] + + started = out[0] + assert started.subagent_run_id == RUN_ID + assert started.name == "availability_researcher" + assert started.parent_tool_call_id == TOOL_USE_ID + + start = out[1] + assert start.message_id == MESSAGE_ID + assert start.role == "assistant" + assert start.subagent_run_id == RUN_ID + + deltas = out[2:5] + assert [ev.delta for ev in deltas] == ["- Ada", " is free", " Tuesday"] + for ev in deltas: + assert ev.message_id == MESSAGE_ID + assert ev.subagent_run_id == RUN_ID + + end = out[5] + assert end.message_id == MESSAGE_ID + assert end.subagent_run_id == RUN_ID + + finished = out[6] + assert finished.subagent_run_id == RUN_ID + assert finished.outcome.type == "success" + + # The session stays, marked finished, so stragglers stay suppressed. + assert subagent_emitter._sessions[TOOL_USE_ID].finished is True + + +async def test_no_deltas_still_brackets_with_started_and_finished(): + out = await _drive([{"init_event_loop": True}, {"result": object()}]) + assert [ev.type for ev in out] == [ + EventType.SUBAGENT_STARTED, + EventType.SUBAGENT_FINISHED, + ] + + +async def test_delegation_error_yields_subagent_error(): + out = await _drive([ + {"init_event_loop": True}, + {"data": "- Ada"}, + {"delegation_error": "specialist exploded"}, + ]) + assert [ev.type for ev in out] == [ + EventType.SUBAGENT_STARTED, + EventType.TEXT_MESSAGE_START, + EventType.TEXT_MESSAGE_CONTENT, + EventType.TEXT_MESSAGE_END, # open message is closed before the error + EventType.SUBAGENT_ERROR, + ] + err = out[-1] + assert err.subagent_run_id == RUN_ID + assert err.message == "specialist exploded" + assert subagent_emitter._sessions[TOOL_USE_ID].finished is True + + +async def test_force_stop_yields_subagent_error(): + out = await _drive([ + {"init_event_loop": True}, + {"force_stop": True, "force_stop_reason": "max tokens"}, + ]) + assert [ev.type for ev in out] == [ + EventType.SUBAGENT_STARTED, + EventType.SUBAGENT_ERROR, + ] + assert out[-1].message == "max tokens" + + +async def test_events_after_terminal_are_ignored(): + out = await _drive([ + {"result": object()}, + {"data": "late straggler"}, + "final string", + ]) + assert [ev.type for ev in out] == [ + EventType.SUBAGENT_STARTED, + EventType.SUBAGENT_FINISHED, + ] + + +async def test_reused_tool_use_id_resets_on_init_event_loop(): + # Some Strands providers reuse tool_use_ids across calls (see the + # bridge's _reused_frontend_tool_identity_error). A fresh inner stream + # always opens with init_event_loop, so a full second sequence must be + # emitted — with a bumped message-id generation so -m1 is not reused. + first = await _drive([ + {"init_event_loop": True}, + {"data": "- Ada"}, + {"result": object()}, + "- Ada", + ]) + assert [ev.type for ev in first] == [ + EventType.SUBAGENT_STARTED, + EventType.TEXT_MESSAGE_START, + EventType.TEXT_MESSAGE_CONTENT, + EventType.TEXT_MESSAGE_END, + EventType.SUBAGENT_FINISHED, + ] + assert first[1].message_id == f"{TOOL_USE_ID}-sub-m1" + + second = await _drive([ + {"init_event_loop": True}, + {"data": "- Grace"}, + {"result": object()}, + "- Grace", + ]) + assert [ev.type for ev in second] == [ + EventType.SUBAGENT_STARTED, + EventType.TEXT_MESSAGE_START, + EventType.TEXT_MESSAGE_CONTENT, + EventType.TEXT_MESSAGE_END, + EventType.SUBAGENT_FINISHED, + ] + # Same subagent_run_id (identity-unchanged re-announce), fresh message id. + assert second[0].subagent_run_id == RUN_ID + assert second[1].message_id == f"{TOOL_USE_ID}-sub-m2" + assert second[2].delta == "- Grace" + + +async def test_str_payload_on_unfinished_session_is_terminal_success(): + # A stream that ends without a {"result": ...} event still terminates: + # the tool's final result-string yield closes the message and finishes + # the subagent instead of leaving the card open forever. + out = await _drive([ + {"init_event_loop": True}, + {"data": "- Ada is free Tuesday"}, + "- Ada is free Tuesday", + ]) + assert [ev.type for ev in out] == [ + EventType.SUBAGENT_STARTED, + EventType.TEXT_MESSAGE_START, + EventType.TEXT_MESSAGE_CONTENT, + EventType.TEXT_MESSAGE_END, + EventType.SUBAGENT_FINISHED, + ] + assert out[-1].outcome.type == "success" + assert subagent_emitter._sessions[TOOL_USE_ID].finished is True + + +async def test_sessions_growth_is_capped(): + for i in range(subagent_emitter._MAX_SESSIONS + 5): + await _drive([{"result": object()}], tool_use_id=f"call_{i}") + assert len(subagent_emitter._sessions) == subagent_emitter._MAX_SESSIONS + # Oldest entries were evicted, newest kept. + assert "call_0" not in subagent_emitter._sessions + assert f"call_{subagent_emitter._MAX_SESSIONS + 4}" in subagent_emitter._sessions + + +async def test_ids_derive_from_tool_use_id(): + out = await _drive([{"data": "x"}, {"result": object()}], tool_use_id="call_other") + assert out[0].subagent_run_id == "call_other-sub" + assert out[1].message_id == "call_other-sub-m1" + + +def test_handler_is_registered_on_the_agent_config(): + from src.agent import agent + + behavior = agent.config.tool_behaviors["research_availability"] + assert behavior.tool_stream_event_handler is emit_subagent_events diff --git a/cockpit/runtimes/aws-strands/python/uv.lock b/cockpit/runtimes/aws-strands/python/uv.lock index f38a5b2da..26bd24a2b 100644 --- a/cockpit/runtimes/aws-strands/python/uv.lock +++ b/cockpit/runtimes/aws-strands/python/uv.lock @@ -236,6 +236,12 @@ dependencies = [ { name = "uvicorn", extra = ["standard"] }, ] +[package.dev-dependencies] +dev = [ + { name = "pytest" }, + { name = "pytest-asyncio" }, +] + [package.metadata] requires-dist = [ { name = "ag-ui-strands", git = "https://github.com/ag-ui-protocol/ag-ui.git?subdirectory=integrations%2Faws-strands%2Fpython&rev=363d3878e30887e88c1fd5ca1916ec3a5962b6be" }, @@ -245,6 +251,12 @@ requires-dist = [ { name = "uvicorn", extras = ["standard"], specifier = ">=0.29" }, ] +[package.metadata.requires-dev] +dev = [ + { name = "pytest", specifier = ">=8.0" }, + { name = "pytest-asyncio", specifier = ">=0.23" }, +] + [[package]] name = "colorama" version = "0.4.6" @@ -429,6 +441,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/57/b0/0e52c878c53f245edd3a11020f20979b3f490f245af532c7cae3027754b5/idna-3.19-py3-none-any.whl", hash = "sha256:815e7be7a7806d54abb586dc943addc79e8b2ee16915059658cbeff4b1b43bf4", size = 68550, upload-time = "2026-08-18T05:14:22.343Z" }, ] +[[package]] +name = "iniconfig" +version = "2.3.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/72/34/14ca021ce8e5dfedc35312d08ba8bf51fdd999c576889fc2c24cb97f4f10/iniconfig-2.3.0.tar.gz", hash = "sha256:c76315c77db068650d49c5b56314774a7804df16fee4402c1f19d6d15d8c4730", size = 20503, upload-time = "2025-10-18T21:55:43.219Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/cb/b1/3846dd7f199d53cb17f49cba7e651e9ce294d8497c8c150530ed11865bb8/iniconfig-2.3.0-py3-none-any.whl", hash = "sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12", size = 7484, upload-time = "2025-10-18T21:55:41.639Z" }, +] + [[package]] name = "jiter" version = "0.16.0" @@ -654,6 +675,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/63/34/ba1c580383c9eada3711951fef0795c80b829a078d72188184bcab9dd527/packaging-26.3-py3-none-any.whl", hash = "sha256:d7193f7c8e4e93f444fde0262bf90af30e16fa0ad0ad44cb553c87339b23cd1c", size = 129956, upload-time = "2026-08-04T18:15:27.159Z" }, ] +[[package]] +name = "pluggy" +version = "1.6.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/f9/e2/3e91f31a7d2b083fe6ef3fa267035b518369d9511ffab804f839851d2779/pluggy-1.6.0.tar.gz", hash = "sha256:7dcc130b76258d33b90f61b658791dede3486c3e6bfb003ee5c9bfb396dd22f3", size = 69412, upload-time = "2025-05-15T12:30:07.975Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/54/20/4d324d65cc6d9205fabedc306948156824eb9f0ee1633355a8f7ec5c66bf/pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746", size = 20538, upload-time = "2025-05-15T12:30:06.134Z" }, +] + [[package]] name = "pycparser" version = "3.0" @@ -767,6 +797,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/30/a4/2bffa9f8e804325a09867f0e9d30795c80ea9f8d62560bd1b6ad6220eb2f/pydantic_settings-2.15.0-py3-none-any.whl", hash = "sha256:0ba092c291c94baceb5eff768aa0d56400a457585bc0175925a5a5510303da42", size = 69413, upload-time = "2026-08-07T09:24:55.839Z" }, ] +[[package]] +name = "pygments" +version = "2.21.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/49/2e/ced460408999b33da6b31b0021b0f37d329e202d4169aeb164493778f25b/pygments-2.21.0.tar.gz", hash = "sha256:610ca751c9bc2492b38eb9a38a7fbc93edbbb2d7182edaf34e66ae493dee5c8c", size = 5005329, upload-time = "2026-08-17T08:02:48.824Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/71/46/17f022dd3e953bf20a04a028a21ec746d942f8d2af30fa0f124fa0e6a684/pygments-2.21.0-py3-none-any.whl", hash = "sha256:2363c69b61c4a97c838da3b130dcd6468f4848992b21a82f2a63ec34377137d9", size = 1250147, upload-time = "2026-08-17T08:02:44.912Z" }, +] + [[package]] name = "pyjwt" version = "2.13.0" @@ -781,6 +820,35 @@ crypto = [ { name = "cryptography" }, ] +[[package]] +name = "pytest" +version = "9.1.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "colorama", marker = "sys_platform == 'win32'" }, + { name = "iniconfig" }, + { name = "packaging" }, + { name = "pluggy" }, + { name = "pygments" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/e4/47/b9efed96c114afcfa3c9d3fe98a76a1d14c74a9e266d397cf6eb64be5e01/pytest-9.1.1.tar.gz", hash = "sha256:1088fbde8f2b49d95a549a195707afa7a76a3ce9bcadc26b6d71f0ffda5fe313", size = 1636369, upload-time = "2026-06-19T10:58:32.857Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/24/25/1de2678b631f5a49215c6c96fff41ba892b0a34df68d6d80292b1b48aa7f/pytest-9.1.1-py3-none-any.whl", hash = "sha256:37a86b45efb9a47a61a36449063e8e18d0cab3161329fc099eb21783169c4f0c", size = 386536, upload-time = "2026-06-19T10:58:31.347Z" }, +] + +[[package]] +name = "pytest-asyncio" +version = "1.4.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "pytest" }, + { name = "typing-extensions", marker = "python_full_version < '3.13'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/43/7c/d36d04db312ecf4298932ef77e6e4a9e8ad017906e24e34f0b0c361a2473/pytest_asyncio-1.4.0.tar.gz", hash = "sha256:c6c0d2259945122819f171a32ecea2c349ead889ee28176caaf492143424be42", size = 58514, upload-time = "2026-05-26T09:56:04.083Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/03/e2/08a497ef684b88559c9cc5f4ad53a37e7b99e727094a86d6ea32536d5d3c/pytest_asyncio-1.4.0-py3-none-any.whl", hash = "sha256:933ca923a23075a87fb7070c0ec272a6848489824d887c85c812670932835aa1", size = 16930, upload-time = "2026-05-26T09:56:02.576Z" }, +] + [[package]] name = "python-dateutil" version = "2.9.0.post0" diff --git a/deployments/ag-ui-dev/deps/aws_strands/docs/wire-capture-subagents.md b/deployments/ag-ui-dev/deps/aws_strands/docs/wire-capture-subagents.md new file mode 100644 index 000000000..520222b19 --- /dev/null +++ b/deployments/ag-ui-dev/deps/aws_strands/docs/wire-capture-subagents.md @@ -0,0 +1,289 @@ +# Strands agents-as-tools: wire capture + emitter-seam decision + +Task 0 spike evidence for translating Strands agents-as-tools delegation into +AG-UI `SUBAGENT_*` events. Captured 2026-09-02 against the live meeting-scheduler +backend (`src/agent.py` + `src/server.py`, bridge pinned to git rev +`363d3878e30887e88c1fd5ca1916ec3a5962b6be`), scratch delegation tool +`research_availability` wrapping a tool-less specialist `Agent(name="availability_researcher")`. +The scratch edit was reverted after capture; only this doc lands. + +All bridge citations are into the installed venv source: +`.venv/lib/python3.14/site-packages/ag_ui_strands/` (referred to as `agent.py` +etc. below). + +## 1. Seam decision: `ToolBehavior.tool_stream_event_handler` (config-level), not a subclass + +**Decision: emit `SUBAGENT_*` from a per-tool `tool_stream_event_handler` +registered in `StrandsAgentConfig.tool_behaviors["research_availability"]`, +with the delegation tool written as an async-generator `@tool` that re-yields +the specialist's `stream_async` events.** + +The exact hook point: the bridge's run loop dispatches every +`tool_stream_event` to the tool's registered handler in +`StrandsAgent.run`, `agent.py:4644-4663` — the handler is an async generator +called with a `ToolStreamEventContext` (`config.py:56-91`) and *"may yield zero +or more AG-UI Event objects which are forwarded directly into the top-level +event stream"* (`config.py:76-91`). Registering a handler suppresses the +default routing for that tool (state snapshots at `agent.py:4664-4669`, +agent-as-tool lifecycle forwarding at `agent.py:4670-4682`), so the handler +owns the whole child stream. + +Why not the LangGraph lane's dispatch-hook subclass +(`cockpit/ag-ui/subagents/python/src/streaming/activity_emitting_agent.py` +overriding `_dispatch_event`): + +- (1b) **A `@tool` body cannot reach an emitter directly, but it does not need + one.** Strands wraps every value an async-generator tool yields as a + `tool_stream_event` in the parent stream (`agent.py:4596-4598`), and the + bridge hands that payload to the per-tool handler with `tool_use_id` and + `tool_name` attached (`agent.py:4648-4652`). That IS the sanctioned + tool-body-to-wire channel — no writer/contextvar/queue plumbing, no fork of + the 5,771-line bridge module. +- The Strands bridge has no `_dispatch_event` seam at all: `StrandsAgent.run` + (`agent.py:2956` onward) is one ~2,800-line async generator with dozens of + inline `yield` sites. A subclass would have to wrap the entire generator and + pattern-match already-serialized events to find the delegation window — + strictly worse information than the handler gets (raw inner Strands events, + pre-translation). +- (1a) Strands→AG-UI translation happens inline in that same `run` generator + (text deltas, `current_tool_use`, `contentBlockStop`, tool results — e.g. + the `tool_stream_event` branch at `agent.py:4596-4682`, tool results at + `agent.py:4684+`), then each pydantic event is SSE-serialized by + `EventEncoder.encode` → `event.model_dump_json(by_alias=True)` + (`ag_ui/encoder/encoder.py:22-36`), called from the endpoint's + `event_generator` (`endpoint.py:290-343`). +- (1c) **Unknown raw dicts do NOT pass through.** The encoder requires pydantic + `BaseEvent` instances (`model_dump_json` call, `encoder.py:36`); a plain dict + would crash the stream. Unmapped *Strands* events are forwarded only as + sanitized `RawEvent` payloads (`_sanitize_raw_event`, `agent.py:1089-1125`) + — and inner-agent payload keys (`data`, `current_tool_use`, ...) are in + `_RAW_SUPPRESSED_KEYS` (`agent.py:1078-1086`), so nothing from the child + leaks via RAW either. The handler must therefore yield real + `ag_ui.core` event objects — which exist, see §2. +- (1d) **`tool_stream_event` from a nested agent-as-tool is forwarded today + only for the inner TOOL-CALL lifecycle, never for inner text.** + `_forward_inner_agent_events` (`agent.py:1195-1313`, invoked at + `agent.py:4677-4682`) translates inner `current_tool_use` / + `contentBlockStop` / `toolResult` into namespaced `TOOL_CALL_*` events and + explicitly nothing else ("Only the tool-call lifecycle is forwarded", + `agent.py:1209`). Inner `{"data": ...}` text deltas fall through every + branch and are dropped. The live capture in §3 confirms this on the wire. + +### Emitter shape (next task) + +- Delegation tool: async-generator `@tool` that does + `async for event in specialist.stream_async(prompt): yield event`, then + yields the accumulated text as its final value (Strands takes the last + yielded value as the tool result — confirmed in §3, the `TOOL_CALL_RESULT` + content is exactly the joined child text). +- Handler on that tool: lazily emits `SubagentStartedEvent` + (`subagent_run_id=f"{ctx.tool_use_id}-sub"`, `name="availability_researcher"`, + `parent_tool_call_id=ctx.tool_use_id`) on the first inner event, translates + inner `{"data": }` into `TEXT_MESSAGE_START/CONTENT/END` carrying + `subagent_run_id`, and emits `SubagentFinishedEvent(outcome=success)` when it + sees the inner terminal `{"result": AgentResult}` event; + `SubagentErrorEvent` when the inner stream surfaces an error / + `forceStop`. `ctx.tool_use_id` equals the wire `toolCallId` (the model's + `call_...` id — §3 line 4 vs. the handler context), so `parentToolCallId` + lines up with the bridge-native `TOOL_CALL_START` with zero bookkeeping. +- Wire-order nuance: the bridge emits `TOOL_CALL_END` when the *args* finish + streaming, before the tool executes (§3 line 19). So the shipped order will + be `TOOL_CALL_START → TOOL_CALL_ARGS → TOOL_CALL_END → SUBAGENT_STARTED → + TEXT_MESSAGE_* → SUBAGENT_FINISHED → TOOL_CALL_RESULT` — the `SUBAGENT_*` + block nests inside the tool call's start/result span, not inside + start/end. + +## 2. SDK check + +``` +$ uv run python -c "import ag_ui.core as c; print([n for n in dir(c) if 'Subagent' in n])" +['SubagentErrorEvent', 'SubagentFinishedEvent', 'SubagentFinishedOutcome', + 'SubagentFinishedSuccessOutcome', 'SubagentFinishedSuspendedOutcome', + 'SubagentStartedEvent'] +``` + +The pinned SDK has first-class subagent events (`ag_ui/core/events.py:455-512`) +with exactly the target fields (`subagent_run_id`, `name`, +`parent_tool_call_id`, `outcome` discriminated union), and every +`TextMessage*` / `ToolCall*` event carries an optional `subagent_run_id` +(`events.py:127-235`). **No raw-dict fallback is needed** — the handler +constructs typed events and the stock encoder serializes them. + +## 3. Live captures (scrubbed) + +Prompt: *"Find a slot for Ada and Grace next week — research their availability +first"*. The model called `research_availability` on the first attempt in both +runs. No API keys or org ids appeared in either stream; nothing was scrubbed — +only long `MESSAGES_SNAPSHOT` lines and repetitive delta runs are elided, each +marked with a `# [elided: ...]` comment. + +### 3a. Async-generator delegation tool (the seam-relevant variant) + +The tool re-yielded the specialist's entire `stream_async` output, so every +child event crossed the bridge as `tool_stream_event` — and the wire between +`TOOL_CALL_END` (line 19) and `TOOL_CALL_RESULT` (line 21) still carries +**zero child events**: `_forward_inner_agent_events` dropped every inner text +delta (the child called no tools, so nothing was forwardable). Line numbers +refer to non-blank SSE lines. + +``` +data: {"type":"RUN_STARTED","threadId":"spike-thread-2","runId":"spike-run-2"} +data: {"type":"STATE_SNAPSHOT","snapshot":{}} +data: {"type":"MESSAGES_SNAPSHOT","messages":[{"id":"u1","role":"user","content":"Find a slot for Ada and Grace next week — research their availability first"}]} +data: {"type":"TOOL_CALL_START","toolCallId":"call_vF6Vc6Wzl40vz9pBZOOrDxS7","toolCallName":"research_availability","parentMessageId":"0130e374-95eb-4a08-aed3-2b6f877331c6"} +data: {"type":"TOOL_CALL_ARGS","toolCallId":"call_vF6Vc6Wzl40vz9pBZOOrDxS7","delta":"{\""} +# [elided: 12 more TOOL_CALL_ARGS deltas spelling {"attendees": "Ada, Grace", "date_range": "next week"}] +data: {"type":"TOOL_CALL_ARGS","toolCallId":"call_vF6Vc6Wzl40vz9pBZOOrDxS7","delta":"\"}"} +data: {"type":"TOOL_CALL_END","toolCallId":"call_vF6Vc6Wzl40vz9pBZOOrDxS7"} +# [elided: MESSAGES_SNAPSHOT mirroring the assistant tool-call message] +# <-- the specialist ran HERE; its interim + final text produced tool_stream_events, none reached the wire +data: {"type":"TOOL_CALL_RESULT","messageId":"2d0594ca-dc44-4c24-841a-1461d55759fc","toolCallId":"call_vF6Vc6Wzl40vz9pBZOOrDxS7","content":"\"To provide a summary of likely availability windows for Ada and Grace for the next week, I will check their schedules. Please hold on for a moment.\\nI actually do not have access to the scheduling information for the attendees. Please provide their typical availability or any specific constraints you might know about them, and I can help you summarize likely availability windows accordingly.\""} +# [elided: MESSAGES_SNAPSHOT adding the tool-result message] +data: {"type":"TEXT_MESSAGE_START","messageId":"51b44583-454d-4252-803a-1f91fa681f5e","role":"assistant"} +data: {"type":"TEXT_MESSAGE_CONTENT","messageId":"51b44583-454d-4252-803a-1f91fa681f5e","delta":"I"} +# [elided: 27 more TEXT_MESSAGE_CONTENT deltas — the ORCHESTRATOR's own summary, after the tool result] +data: {"type":"TEXT_MESSAGE_CONTENT","messageId":"51b44583-454d-4252-803a-1f91fa681f5e","delta":"."} +data: {"type":"TEXT_MESSAGE_END","messageId":"51b44583-454d-4252-803a-1f91fa681f5e"} +# [elided: final MESSAGES_SNAPSHOT] +data: {"type":"STATE_SNAPSHOT","snapshot":{}} +data: {"type":"RUN_FINISHED","threadId":"spike-thread-2","runId":"spike-run-2","outcome":{"type":"success"}} +``` + +Event tally: 1 RUN_STARTED, 2 STATE_SNAPSHOT, 4 MESSAGES_SNAPSHOT, +1 TOOL_CALL_START, 14 TOOL_CALL_ARGS, 1 TOOL_CALL_END, 1 TOOL_CALL_RESULT, +1 TEXT_MESSAGE_START, 29 TEXT_MESSAGE_CONTENT, 1 TEXT_MESSAGE_END, +1 RUN_FINISHED. No SUBAGENT_*, no RAW, no CUSTOM, no STEP_*. + +### 3b. Plain sync delegation tool (the naive idiom) + +Same prompt, tool body `result = availability_researcher(...); return str(result)`. +Identical wire shape: `TOOL_CALL_START/ARGS×14/END → MESSAGES_SNAPSHOT → +TOOL_CALL_RESULT` (child's full multi-paragraph answer as one string) `→ +MESSAGES_SNAPSHOT → TEXT_MESSAGE_*` (orchestrator summary) `→ RUN_FINISHED +success`. A sync tool yields nothing mid-flight, so no `tool_stream_event` +fires at all — the child is a black box by construction, and this variant can +never feed a subagent emitter. (Capture withheld here as it adds nothing over +3a; tally: 24 TEXT_MESSAGE_CONTENT, otherwise identical event mix.) + +## 4. Did child tokens appear on the wire? + +**No — in neither variant.** During the delegation call the stream goes +straight from `TOOL_CALL_END` (§3a line 19) to `TOOL_CALL_RESULT` (§3a line +21) with only a `MESSAGES_SNAPSHOT` between. The specialist's interim sentence +("To provide a summary ... Please hold on for a moment.") exists in the run — +it surfaces verbatim *inside* the final `TOOL_CALL_RESULT` content — proving +the child streamed internally and the bridge dropped the deltas +(`_forward_inner_agent_events` forwards tool-call lifecycle only, +`agent.py:1195-1313`). Natively, delegation is: parent tool-call args stream → +silence → one opaque result string. This is the matrix cell the emitter fixes: +child tokens must be re-emitted by our `tool_stream_event_handler` as +`TEXT_MESSAGE_*` events carrying `subagentRunId`. + +## 5. Other observations + +- Restreaming the child through the generator tool logs repeated + `ValueError: was created + in a different Context` server-side (OTel context tokens crossing task + boundaries; cosmetic with OTel disabled, but worth watching once the real + emitter lands). +- `RunAgentInput` requires `threadId`, `runId`, `messages`, `tools`, + `context`, `forwardedProps` (camelCase; `ag_ui/core/types.py:396-412`); + the endpoint validates with `model_validate` and 422s otherwise + (`endpoint.py:55-86`). +- `emit_messages_snapshot` (on by default) interleaves full + `MESSAGES_SNAPSHOT`s after every tool END/RESULT and TEXT_MESSAGE_END — + the future emitter must not splice subagent messages into those snapshots + (mirroring `_forward_inner_agent_events`' deliberate choice, + `agent.py:1209-1211`). + +## After the emitter + +Captured 2026-09-02 against the shipped scenario (`research_availability` +registered with `ToolBehavior(tool_stream_event_handler=emit_subagent_events)`, +`src/subagent_emitter.py`), same `RunAgentInput` as §3 (prompt *"Find a slot +for Ada and Grace next week — research their availability first"*, empty +`tools`/`context`/`state`/`forwardedProps`). No keys or org ids appeared in +the stream; only repetitive delta runs and `MESSAGES_SNAPSHOT`s are elided, +marked with `# [elided: ...]`. + +The model delegated TWICE this run (first to ask for concrete dates, then to +research them) — both rounds carried the full `SUBAGENT_*` block, each nested +between its own `TOOL_CALL_END` and `TOOL_CALL_RESULT` exactly as §1 +predicted. First round shown; the second is shape-identical with +`toolCallId=call_ur6NyfucZhR4FlDTxrDt0hy9` and 61 content deltas. + +``` +data: {"type":"RUN_STARTED","threadId":"smoke-thread-1","runId":"smoke-run-1"} +data: {"type":"STATE_SNAPSHOT","snapshot":{}} +# [elided: MESSAGES_SNAPSHOT] +data: {"type":"TOOL_CALL_START","toolCallId":"call_YW6LslLJbLNZul9qYlYfeySg","toolCallName":"research_availability","parentMessageId":"6228ef46-8396-4044-903c-ef9e3f3abc3d"} +# [elided: 14 TOOL_CALL_ARGS deltas spelling {"attendees": "Ada, Grace", "date_range": "next week"}] +data: {"type":"TOOL_CALL_END","toolCallId":"call_YW6LslLJbLNZul9qYlYfeySg"} +# [elided: MESSAGES_SNAPSHOT] +data: {"type":"SUBAGENT_STARTED","subagentRunId":"call_YW6LslLJbLNZul9qYlYfeySg-sub","name":"availability_researcher","parentToolCallId":"call_YW6LslLJbLNZul9qYlYfeySg"} +data: {"type":"TEXT_MESSAGE_START","messageId":"call_YW6LslLJbLNZul9qYlYfeySg-sub-m1","role":"assistant","subagentRunId":"call_YW6LslLJbLNZul9qYlYfeySg-sub"} +data: {"type":"TEXT_MESSAGE_CONTENT","messageId":"call_YW6LslLJbLNZul9qYlYfeySg-sub-m1","delta":"Please","subagentRunId":"call_YW6LslLJbLNZul9qYlYfeySg-sub"} +data: {"type":"TEXT_MESSAGE_CONTENT","messageId":"call_YW6LslLJbLNZul9qYlYfeySg-sub-m1","delta":" provide","subagentRunId":"call_YW6LslLJbLNZul9qYlYfeySg-sub"} +# [elided: 25 more TEXT_MESSAGE_CONTENT deltas — the CHILD's text, token by token] +data: {"type":"TEXT_MESSAGE_CONTENT","messageId":"call_YW6LslLJbLNZul9qYlYfeySg-sub-m1","delta":".","subagentRunId":"call_YW6LslLJbLNZul9qYlYfeySg-sub"} +data: {"type":"TEXT_MESSAGE_END","messageId":"call_YW6LslLJbLNZul9qYlYfeySg-sub-m1","subagentRunId":"call_YW6LslLJbLNZul9qYlYfeySg-sub"} +data: {"type":"SUBAGENT_FINISHED","subagentRunId":"call_YW6LslLJbLNZul9qYlYfeySg-sub","outcome":{"type":"success"}} +data: {"type":"TOOL_CALL_RESULT","messageId":"b4c51965-4781-4683-8808-05fddd85135f","toolCallId":"call_YW6LslLJbLNZul9qYlYfeySg","content":"\"Please provide the specific dates for next week (e.g., from October 16 to October 20) so I can summarize the availability accordingly.\""} +# [elided: MESSAGES_SNAPSHOT, then the second delegation round — TOOL_CALL_START/ARGS×19/END, +# SUBAGENT_STARTED, TEXT_MESSAGE_START, 61 TEXT_MESSAGE_CONTENT deltas, TEXT_MESSAGE_END, +# SUBAGENT_FINISHED success, TOOL_CALL_RESULT — ids derived from call_ur6NyfucZhR4FlDTxrDt0hy9] +data: {"type":"TEXT_MESSAGE_START","messageId":"77635fb9-a418-443a-af7e-c70a29d079f6","role":"assistant"} +# [elided: 26 TEXT_MESSAGE_CONTENT deltas — the ORCHESTRATOR's own summary, no subagentRunId] +data: {"type":"TEXT_MESSAGE_END","messageId":"77635fb9-a418-443a-af7e-c70a29d079f6"} +# [elided: final MESSAGES_SNAPSHOT] +data: {"type":"STATE_SNAPSHOT","snapshot":{}} +data: {"type":"RUN_FINISHED","threadId":"smoke-thread-1","runId":"smoke-run-1","outcome":{"type":"success"}} +``` + +Event tally (174 SSE lines): 1 RUN_STARTED, 2 STATE_SNAPSHOT, +6 MESSAGES_SNAPSHOT, 2 TOOL_CALL_START, 33 TOOL_CALL_ARGS, 2 TOOL_CALL_END, +2 SUBAGENT_STARTED, 2 TEXT_MESSAGE_START(sub), 89 TEXT_MESSAGE_CONTENT(sub), +2 TEXT_MESSAGE_END(sub), 2 SUBAGENT_FINISHED, 2 TOOL_CALL_RESULT, +1 TEXT_MESSAGE_START, 26 TEXT_MESSAGE_CONTENT, 1 TEXT_MESSAGE_END, +1 RUN_FINISHED. No RAW, no CUSTOM, no STEP_*, no SUBAGENT_ERROR. + +**Child deltas: streaming (89 content events across the two delegation +rounds: 28 + 61)** — §4's black box is gone. Every child event carries +`subagentRunId` derived from the wire `toolCallId` +(`-sub` / `-sub-m1`), `SUBAGENT_STARTED.parentToolCallId` +matches the bridge-native `TOOL_CALL_START.toolCallId` verbatim, and the +`TOOL_CALL_RESULT` content equals the joined child deltas (the tool's final +yield), confirming the last-yield-as-result contract survived the handler +registration. The §5 OTel `ContextVar` warnings still log server-side and +remain cosmetic. + +## Browser verification + +2026-09-02, live backend (real `OPENAI_API_KEY`, uvicorn on :5331) + `nx +serve cockpit-runtimes-aws-strands-angular` on :4331, driven headlessly +with Playwright. Screenshot: +`cockpit/runtimes/aws-strands/angular/e2e/manual/subagent-card-live.png`. + +What rendered: the delegation prompt (*"Find a slot for Ada and Grace next +week — research their availability first"*) produced an inline +`` anchored to the `research_availability` tool call — +header `availability_researcher` + wire `toolCallId` + status badge — with +the specialist's transcript inside it, followed by the orchestrator's own +summary bubble. The child text never leaked into the parent bubble, and the +card persists (collapsed to `complete`) after the run. + +Did the card text stream mid-run: **yes** (matrix cell: expected +streaming = yes). Polling the card's `innerText` every 150ms during the +run showed the badge at `running` while the specialist's message grew +monotonically across successive samples (one run: lengths 144 → 213 → 288 +→ 403 → 420 → 438 → 501 → 643 chars between t≈2.0s and t≈3.5s), then +flipped to `complete` and collapsed. This confirms the attributed +`TEXT_MESSAGE_CONTENT` deltas render progressively in the card, not as one +post-hoc paste. + +One rendering defect surfaced and was fixed on this branch: the wire and +the `@threadplane/ag-ui` reducer were both correct, but `chat-tool-calls` +anchored subagent cards by the adapter's `subagents()` **map key** (here +the `subagentRunId`, `-sub`) instead of the contract field +`Subagent.toolCallId`, so the card never mounted for Strands delegations. +Fixed in `libs/chat` by re-indexing on `Subagent.toolCallId`. diff --git a/deployments/ag-ui-dev/deps/aws_strands/pyproject.toml b/deployments/ag-ui-dev/deps/aws_strands/pyproject.toml index a95c6f092..d93b5f47a 100644 --- a/deployments/ag-ui-dev/deps/aws_strands/pyproject.toml +++ b/deployments/ag-ui-dev/deps/aws_strands/pyproject.toml @@ -24,9 +24,18 @@ dependencies = [ [tool.uv.sources] ag-ui-strands = { git = "https://github.com/ag-ui-protocol/ag-ui.git", rev = "363d3878e30887e88c1fd5ca1916ec3a5962b6be", subdirectory = "integrations/aws-strands/python" } +[dependency-groups] +dev = [ + "pytest>=8.0", + "pytest-asyncio>=0.23", +] + [build-system] requires = ["hatchling"] build-backend = "hatchling.build" [tool.hatch.build.targets.wheel] packages = ["src"] + +[tool.pytest.ini_options] +asyncio_mode = "auto" diff --git a/deployments/ag-ui-dev/deps/aws_strands/src/agent.py b/deployments/ag-ui-dev/deps/aws_strands/src/agent.py index a74c3e5ea..0a709ee30 100644 --- a/deployments/ag-ui-dev/deps/aws_strands/src/agent.py +++ b/deployments/ag-ui-dev/deps/aws_strands/src/agent.py @@ -19,11 +19,16 @@ resumes from the client's top-level ``resume`` entries keyed by ``interruptId`` (never the LangGraph bridge's CUSTOM ``on_interrupt``). -No subagents surface and no multi-agent route: the Strands bridge routes -delegation through CUSTOM MultiAgentHandoff + STEP_* with zero ACTIVITY -events (measured red upstream), and multi-agent routes crash the stale -PyPI ``ag-ui-strands`` 0.3.0 wheel — which is why this example pins the -bridge to a git ref (see pyproject.toml). +- subagents: ``research_availability`` delegates to a tool-less specialist + ``availability_researcher`` Agent via an async-generator ``@tool`` that + re-yields the specialist's ``stream_async`` events; a per-tool + ``ToolBehavior.tool_stream_event_handler`` (src/subagent_emitter.py) + translates them into standard ``SUBAGENT_*`` + child ``TEXT_MESSAGE_*`` + wire events. (The bridge natively drops inner text deltas and would + otherwise route delegation through CUSTOM MultiAgentHandoff + STEP_*; + multi-agent routes also crash the stale PyPI ``ag-ui-strands`` 0.3.0 + wheel — which is why this example pins the bridge to a git ref, see + pyproject.toml and docs/wire-capture-subagents.md.) Model: Strands' native OpenAI provider on plain ``OPENAI_API_KEY`` — no AWS credentials involved. ``OPENAI_BASE_URL`` is honored, which is how the @@ -45,6 +50,8 @@ from ag_ui_strands import StrandsAgent, StrandsAgentConfig, ToolBehavior +from .subagent_emitter import emit_subagent_events + _SLOTS = { "monday": ["09:00", "13:30"], "tuesday": ["10:00", "15:00"], @@ -139,6 +146,41 @@ async def booking_state(context) -> dict | None: return _complete_state() +_RESEARCHER_INSTRUCTIONS = ( + "You are an availability researcher. Given attendee names and a date " + "range, produce a short bullet summary of likely availability windows. " + "Be concise: 3 bullets max." +) + + +@tool +async def research_availability(attendees: str, date_range: str): + """Delegate availability research for the given attendees to a specialist. + + Args: + attendees: Comma-separated attendee names, e.g. 'Ada, Grace'. + date_range: The window to research, e.g. 'next week'. + + Returns: + The specialist's bullet summary of likely availability windows. + """ + chunks: list[str] = [] + try: + async for event in availability_researcher.stream_async( + f"Attendees: {attendees}\nDate range: {date_range}" + ): + if isinstance(event, dict) and isinstance(event.get("data"), str): + chunks.append(event["data"]) + yield event + except Exception as exc: # pragma: no cover - not reachable without a live model failure + # Surface the failure to the emitter (which owns the SUBAGENT_ERROR + # wire event), then let the tool error propagate to Strands normally. + yield {"delegation_error": str(exc)} + raise + # Strands takes the LAST yielded value as the tool result. + yield "".join(chunks) + + _INSTRUCTIONS = """You are a meeting scheduling copilot. When the user asks to book a meeting: @@ -155,6 +197,9 @@ async def booking_state(context) -> dict | None: Keep every response brief and factual. Never invent availability — use the tool. + +When the user asks about attendees' availability, delegate that research to +the `research_availability` tool before proposing meeting slots. """ @@ -172,11 +217,22 @@ def build_model() -> OpenAIModel: return OpenAIModel(client_args=client_args, model_id=os.environ.get("OPENAI_CHAT_MODEL", "gpt-4o-mini")) +# Tool-less specialist the orchestrator delegates availability research to +# via the `research_availability` async-generator tool above. Its streamed +# events cross the bridge as tool_stream_events and are translated into +# SUBAGENT_* wire events by the emitter registered in ToolBehavior below. +availability_researcher = Agent( + model=build_model(), + system_prompt=_RESEARCHER_INSTRUCTIONS, + name="availability_researcher", + tools=[], +) + agent = StrandsAgent( agent=Agent( model=build_model(), system_prompt=_INSTRUCTIONS, - tools=[check_availability, book_meeting], + tools=[check_availability, book_meeting, research_availability], ), name="aws-strands", description="Books meetings with availability lookup, shared state, and human approval.", @@ -184,6 +240,9 @@ def build_model() -> OpenAIModel: tool_behaviors={ "check_availability": ToolBehavior(state_from_result=availability_state), "book_meeting": ToolBehavior(state_from_args=booking_state), + "research_availability": ToolBehavior( + tool_stream_event_handler=emit_subagent_events, + ), }, ), ) diff --git a/deployments/ag-ui-dev/deps/aws_strands/src/subagent_emitter.py b/deployments/ag-ui-dev/deps/aws_strands/src/subagent_emitter.py new file mode 100644 index 000000000..e858babb1 --- /dev/null +++ b/deployments/ag-ui-dev/deps/aws_strands/src/subagent_emitter.py @@ -0,0 +1,189 @@ +# SPDX-License-Identifier: MIT +"""SUBAGENT_* emitter for the `research_availability` delegation tool. + +The Strands bridge natively drops a child agent's text deltas: an +async-generator `@tool` re-yielding a specialist's ``stream_async`` events +produces ``tool_stream_event``s, but `_forward_inner_agent_events` forwards +only the inner TOOL-CALL lifecycle and never inner text (measured in +docs/wire-capture-subagents.md). Registering this +``ToolBehavior.tool_stream_event_handler`` claims the whole child stream and +re-emits it as standard AG-UI subagent wire events: + + SUBAGENT_STARTED (first inner event) + TEXT_MESSAGE_START/CONTENT.../END (inner ``data`` deltas, streamed) + SUBAGENT_FINISHED outcome=success (inner terminal ``result`` event) + +or ``SUBAGENT_ERROR`` when the delegation fails (the tool yields a +``delegation_error`` sentinel before re-raising, or the inner stream +force-stops). Ids derive from the wire ``toolCallId`` (``ctx.tool_use_id``) +so the client can key the subagent card on ``parentToolCallId`` with zero +bookkeeping. + +The bridge instantiates this handler ONCE PER EVENT (a fresh async generator +per ``tool_stream_event``), so per-invocation lifecycle state lives in a +module-level dict keyed by ``tool_use_id``. The encoder requires pydantic +``BaseEvent`` instances — raw dicts crash the stream — so only typed +``ag_ui.core`` events are yielded. +""" + +from dataclasses import dataclass + +from ag_ui.core import ( + EventType, + SubagentErrorEvent, + SubagentFinishedEvent, + SubagentFinishedSuccessOutcome, + SubagentStartedEvent, + TextMessageContentEvent, + TextMessageEndEvent, + TextMessageStartEvent, +) + +from ag_ui_strands import ToolStreamEventContext + +SPECIALIST_NAME = "availability_researcher" + + +@dataclass +class _DelegationState: + """Lifecycle of one delegation call, keyed by tool_use_id.""" + + started: bool = False + message_open: bool = False + finished: bool = False + generation: int = 1 + """Bumped when a reused tool_use_id starts a fresh inner stream, so the + re-run's message id (-m2, -m3, ...) never collides with an already-emitted + one.""" + + +# Finished entries are kept (not popped) so the tool's trailing result-string +# yield and any stragglers stay suppressed. Growth is capped at _MAX_SESSIONS +# (dict insertion order = age; each entry is a short key string plus a +# 4-field dataclass, ~414 bytes measured with sys.getsizeof — the cap bounds +# the dict at ~200 KiB). +_sessions: dict[str, _DelegationState] = {} +_MAX_SESSIONS = 512 + + +def _subagent_run_id(tool_use_id: str) -> str: + return f"{tool_use_id}-sub" + + +def _message_id(tool_use_id: str, generation: int) -> str: + return f"{tool_use_id}-sub-m{generation}" + + +async def emit_subagent_events(ctx: ToolStreamEventContext): + """tool_stream_event_handler translating child events to SUBAGENT_* wire + events. Async generator, called once per inner event.""" + state = _sessions.get(ctx.tool_use_id) + if state is None: + state = _DelegationState() + _sessions[ctx.tool_use_id] = state + while len(_sessions) > _MAX_SESSIONS: + del _sessions[next(k for k in _sessions if k != ctx.tool_use_id)] + run_id = _subagent_run_id(ctx.tool_use_id) + data = ctx.stream_data + if state.finished and isinstance(data, dict) and "init_event_loop" in data: + # Reused tool_use_id (real for some Strands providers — see the + # bridge's _reused_frontend_tool_identity_error): a fresh inner + # stream always opens with init_event_loop, so reset for a second + # full SUBAGENT_* sequence under the same subagent_run_id (the + # adapter treats an identity-unchanged re-announce as content-only). + # Non-init stragglers after the terminal stay swallowed below. + state = _DelegationState(generation=state.generation + 1) + _sessions[ctx.tool_use_id] = state + message_id = _message_id(ctx.tool_use_id, state.generation) + try: + if state.finished: + # The tool's final yield (the result string) and any stragglers + # arrive after the inner terminal event — nothing left to emit. + return + + if not state.started: + state.started = True + yield SubagentStartedEvent( + type=EventType.SUBAGENT_STARTED, + subagent_run_id=run_id, + name=SPECIALIST_NAME, + parent_tool_call_id=ctx.tool_use_id, + ) + + if isinstance(data, str): + # Terminal-success fallback: the tool's final result-string yield + # arriving on an UNFINISHED session means the inner stream ended + # without a {"result": ...} event — close out rather than leaving + # the subagent card open forever. + state.finished = True + if state.message_open: + state.message_open = False + yield TextMessageEndEvent( + type=EventType.TEXT_MESSAGE_END, + message_id=message_id, + subagent_run_id=run_id, + ) + yield SubagentFinishedEvent( + type=EventType.SUBAGENT_FINISHED, + subagent_run_id=run_id, + outcome=SubagentFinishedSuccessOutcome(), + ) + return + if not isinstance(data, dict): + return + + if "delegation_error" in data or data.get("force_stop"): + message = str( + data.get("delegation_error") + or data.get("force_stop_reason") + or "subagent stream force-stopped" + ) + state.finished = True + if state.message_open: + state.message_open = False + yield TextMessageEndEvent( + type=EventType.TEXT_MESSAGE_END, + message_id=message_id, + subagent_run_id=run_id, + ) + yield SubagentErrorEvent( + type=EventType.SUBAGENT_ERROR, + subagent_run_id=run_id, + message=message, + ) + elif isinstance(data.get("data"), str) and data["data"]: + if not state.message_open: + state.message_open = True + yield TextMessageStartEvent( + type=EventType.TEXT_MESSAGE_START, + message_id=message_id, + role="assistant", + subagent_run_id=run_id, + ) + yield TextMessageContentEvent( + type=EventType.TEXT_MESSAGE_CONTENT, + message_id=message_id, + delta=data["data"], + subagent_run_id=run_id, + ) + elif "result" in data: + state.finished = True + if state.message_open: + state.message_open = False + yield TextMessageEndEvent( + type=EventType.TEXT_MESSAGE_END, + message_id=message_id, + subagent_run_id=run_id, + ) + yield SubagentFinishedEvent( + type=EventType.SUBAGENT_FINISHED, + subagent_run_id=run_id, + outcome=SubagentFinishedSuccessOutcome(), + ) + except Exception as exc: # pragma: no cover - defensive: never crash the run + state.finished = True + yield SubagentErrorEvent( + type=EventType.SUBAGENT_ERROR, + subagent_run_id=run_id, + message=str(exc), + ) diff --git a/deployments/ag-ui-dev/deps/aws_strands/tests/test_delegation.py b/deployments/ag-ui-dev/deps/aws_strands/tests/test_delegation.py new file mode 100644 index 000000000..293b3cc95 --- /dev/null +++ b/deployments/ag-ui-dev/deps/aws_strands/tests/test_delegation.py @@ -0,0 +1,45 @@ +# SPDX-License-Identifier: MIT +"""Tests for the `research_availability` delegation scenario — the tool is a +registered async-generator `@tool` that hands availability research to the +tool-less `availability_researcher` specialist. No live model calls: these +tests only inspect registration metadata (the module builds its OpenAI +clients with a placeholder key that would 401 at request time).""" + +import inspect + +from src.agent import agent, availability_researcher, research_availability + + +def _tool_names() -> list[str]: + # Private-attr coupling (StrandsAgent._tools) is frozen by the git-ref + # pin on ag-ui-strands in pyproject.toml. + return [t.tool_name for t in agent._tools] + + +def test_research_availability_is_registered_on_the_agent(): + assert "research_availability" in _tool_names() + # Existing tools stay registered untouched. + assert "check_availability" in _tool_names() + assert "book_meeting" in _tool_names() + + +def test_tool_name_and_docstring(): + assert research_availability.tool_name == "research_availability" + description = research_availability.tool_spec["description"] + assert description.startswith( + "Delegate availability research for the given attendees to a specialist." + ) + schema = research_availability.tool_spec["inputSchema"]["json"] + assert set(schema["required"]) == {"attendees", "date_range"} + + +def test_tool_is_an_async_generator(): + # The seam depends on it: only an async-generator tool produces + # tool_stream_events for the bridge to hand to the subagent emitter. + assert inspect.isasyncgenfunction(research_availability._tool_func) + + +def test_specialist_is_toolless_researcher(): + assert availability_researcher.name == "availability_researcher" + assert availability_researcher.tool_registry.registry == {} + assert "availability researcher" in availability_researcher.system_prompt diff --git a/deployments/ag-ui-dev/deps/aws_strands/tests/test_subagent_emitter.py b/deployments/ag-ui-dev/deps/aws_strands/tests/test_subagent_emitter.py new file mode 100644 index 000000000..0f2c7a073 --- /dev/null +++ b/deployments/ag-ui-dev/deps/aws_strands/tests/test_subagent_emitter.py @@ -0,0 +1,222 @@ +# SPDX-License-Identifier: MIT +"""Tests for the SUBAGENT_* emitter — drives the tool_stream_event_handler +with synthetic Strands events (shapes copied from the Task-0 wire capture in +docs/wire-capture-subagents.md) and asserts the exact emitted AG-UI sequence +field-for-field. The bridge calls the handler once per event with a fresh +generator, which is exactly how these tests drive it.""" + +import pytest + +from ag_ui.core import EventType +from ag_ui_strands import ToolStreamEventContext + +from src import subagent_emitter +from src.subagent_emitter import emit_subagent_events + +TOOL_USE_ID = "call_vF6Vc6Wzl40vz9pBZOOrDxS7" +RUN_ID = f"{TOOL_USE_ID}-sub" +MESSAGE_ID = f"{TOOL_USE_ID}-sub-m1" + + +@pytest.fixture(autouse=True) +def _clean_sessions(): + subagent_emitter._sessions.clear() + yield + subagent_emitter._sessions.clear() + + +async def _drive(events: list, tool_use_id: str = TOOL_USE_ID) -> list: + """Feed synthetic stream payloads one at a time, the way the bridge + dispatches tool_stream_events, collecting everything yielded.""" + out = [] + for data in events: + ctx = ToolStreamEventContext( + tool_use_id=tool_use_id, + tool_name="research_availability", + stream_data=data, + ) + async for ev in emit_subagent_events(ctx): + out.append(ev) + return out + + +async def test_success_sequence_field_for_field(): + out = await _drive([ + {"init_event_loop": True}, # specialist loop init + {"start": True}, + {"data": "- Ada"}, # streamed text deltas + {"data": " is free"}, + {"data": " Tuesday"}, + {"result": object()}, # terminal AgentResult event + "- Ada is free Tuesday", # the tool's final yield (result string) + ]) + + types = [ev.type for ev in out] + assert types == [ + EventType.SUBAGENT_STARTED, + EventType.TEXT_MESSAGE_START, + EventType.TEXT_MESSAGE_CONTENT, + EventType.TEXT_MESSAGE_CONTENT, + EventType.TEXT_MESSAGE_CONTENT, + EventType.TEXT_MESSAGE_END, + EventType.SUBAGENT_FINISHED, + ] + + started = out[0] + assert started.subagent_run_id == RUN_ID + assert started.name == "availability_researcher" + assert started.parent_tool_call_id == TOOL_USE_ID + + start = out[1] + assert start.message_id == MESSAGE_ID + assert start.role == "assistant" + assert start.subagent_run_id == RUN_ID + + deltas = out[2:5] + assert [ev.delta for ev in deltas] == ["- Ada", " is free", " Tuesday"] + for ev in deltas: + assert ev.message_id == MESSAGE_ID + assert ev.subagent_run_id == RUN_ID + + end = out[5] + assert end.message_id == MESSAGE_ID + assert end.subagent_run_id == RUN_ID + + finished = out[6] + assert finished.subagent_run_id == RUN_ID + assert finished.outcome.type == "success" + + # The session stays, marked finished, so stragglers stay suppressed. + assert subagent_emitter._sessions[TOOL_USE_ID].finished is True + + +async def test_no_deltas_still_brackets_with_started_and_finished(): + out = await _drive([{"init_event_loop": True}, {"result": object()}]) + assert [ev.type for ev in out] == [ + EventType.SUBAGENT_STARTED, + EventType.SUBAGENT_FINISHED, + ] + + +async def test_delegation_error_yields_subagent_error(): + out = await _drive([ + {"init_event_loop": True}, + {"data": "- Ada"}, + {"delegation_error": "specialist exploded"}, + ]) + assert [ev.type for ev in out] == [ + EventType.SUBAGENT_STARTED, + EventType.TEXT_MESSAGE_START, + EventType.TEXT_MESSAGE_CONTENT, + EventType.TEXT_MESSAGE_END, # open message is closed before the error + EventType.SUBAGENT_ERROR, + ] + err = out[-1] + assert err.subagent_run_id == RUN_ID + assert err.message == "specialist exploded" + assert subagent_emitter._sessions[TOOL_USE_ID].finished is True + + +async def test_force_stop_yields_subagent_error(): + out = await _drive([ + {"init_event_loop": True}, + {"force_stop": True, "force_stop_reason": "max tokens"}, + ]) + assert [ev.type for ev in out] == [ + EventType.SUBAGENT_STARTED, + EventType.SUBAGENT_ERROR, + ] + assert out[-1].message == "max tokens" + + +async def test_events_after_terminal_are_ignored(): + out = await _drive([ + {"result": object()}, + {"data": "late straggler"}, + "final string", + ]) + assert [ev.type for ev in out] == [ + EventType.SUBAGENT_STARTED, + EventType.SUBAGENT_FINISHED, + ] + + +async def test_reused_tool_use_id_resets_on_init_event_loop(): + # Some Strands providers reuse tool_use_ids across calls (see the + # bridge's _reused_frontend_tool_identity_error). A fresh inner stream + # always opens with init_event_loop, so a full second sequence must be + # emitted — with a bumped message-id generation so -m1 is not reused. + first = await _drive([ + {"init_event_loop": True}, + {"data": "- Ada"}, + {"result": object()}, + "- Ada", + ]) + assert [ev.type for ev in first] == [ + EventType.SUBAGENT_STARTED, + EventType.TEXT_MESSAGE_START, + EventType.TEXT_MESSAGE_CONTENT, + EventType.TEXT_MESSAGE_END, + EventType.SUBAGENT_FINISHED, + ] + assert first[1].message_id == f"{TOOL_USE_ID}-sub-m1" + + second = await _drive([ + {"init_event_loop": True}, + {"data": "- Grace"}, + {"result": object()}, + "- Grace", + ]) + assert [ev.type for ev in second] == [ + EventType.SUBAGENT_STARTED, + EventType.TEXT_MESSAGE_START, + EventType.TEXT_MESSAGE_CONTENT, + EventType.TEXT_MESSAGE_END, + EventType.SUBAGENT_FINISHED, + ] + # Same subagent_run_id (identity-unchanged re-announce), fresh message id. + assert second[0].subagent_run_id == RUN_ID + assert second[1].message_id == f"{TOOL_USE_ID}-sub-m2" + assert second[2].delta == "- Grace" + + +async def test_str_payload_on_unfinished_session_is_terminal_success(): + # A stream that ends without a {"result": ...} event still terminates: + # the tool's final result-string yield closes the message and finishes + # the subagent instead of leaving the card open forever. + out = await _drive([ + {"init_event_loop": True}, + {"data": "- Ada is free Tuesday"}, + "- Ada is free Tuesday", + ]) + assert [ev.type for ev in out] == [ + EventType.SUBAGENT_STARTED, + EventType.TEXT_MESSAGE_START, + EventType.TEXT_MESSAGE_CONTENT, + EventType.TEXT_MESSAGE_END, + EventType.SUBAGENT_FINISHED, + ] + assert out[-1].outcome.type == "success" + assert subagent_emitter._sessions[TOOL_USE_ID].finished is True + + +async def test_sessions_growth_is_capped(): + for i in range(subagent_emitter._MAX_SESSIONS + 5): + await _drive([{"result": object()}], tool_use_id=f"call_{i}") + assert len(subagent_emitter._sessions) == subagent_emitter._MAX_SESSIONS + # Oldest entries were evicted, newest kept. + assert "call_0" not in subagent_emitter._sessions + assert f"call_{subagent_emitter._MAX_SESSIONS + 4}" in subagent_emitter._sessions + + +async def test_ids_derive_from_tool_use_id(): + out = await _drive([{"data": "x"}, {"result": object()}], tool_use_id="call_other") + assert out[0].subagent_run_id == "call_other-sub" + assert out[1].message_id == "call_other-sub-m1" + + +def test_handler_is_registered_on_the_agent_config(): + from src.agent import agent + + behavior = agent.config.tool_behaviors["research_availability"] + assert behavior.tool_stream_event_handler is emit_subagent_events diff --git a/deployments/ag-ui-dev/deps/aws_strands/uv.lock b/deployments/ag-ui-dev/deps/aws_strands/uv.lock index f38a5b2da..26bd24a2b 100644 --- a/deployments/ag-ui-dev/deps/aws_strands/uv.lock +++ b/deployments/ag-ui-dev/deps/aws_strands/uv.lock @@ -236,6 +236,12 @@ dependencies = [ { name = "uvicorn", extra = ["standard"] }, ] +[package.dev-dependencies] +dev = [ + { name = "pytest" }, + { name = "pytest-asyncio" }, +] + [package.metadata] requires-dist = [ { name = "ag-ui-strands", git = "https://github.com/ag-ui-protocol/ag-ui.git?subdirectory=integrations%2Faws-strands%2Fpython&rev=363d3878e30887e88c1fd5ca1916ec3a5962b6be" }, @@ -245,6 +251,12 @@ requires-dist = [ { name = "uvicorn", extras = ["standard"], specifier = ">=0.29" }, ] +[package.metadata.requires-dev] +dev = [ + { name = "pytest", specifier = ">=8.0" }, + { name = "pytest-asyncio", specifier = ">=0.23" }, +] + [[package]] name = "colorama" version = "0.4.6" @@ -429,6 +441,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/57/b0/0e52c878c53f245edd3a11020f20979b3f490f245af532c7cae3027754b5/idna-3.19-py3-none-any.whl", hash = "sha256:815e7be7a7806d54abb586dc943addc79e8b2ee16915059658cbeff4b1b43bf4", size = 68550, upload-time = "2026-08-18T05:14:22.343Z" }, ] +[[package]] +name = "iniconfig" +version = "2.3.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/72/34/14ca021ce8e5dfedc35312d08ba8bf51fdd999c576889fc2c24cb97f4f10/iniconfig-2.3.0.tar.gz", hash = "sha256:c76315c77db068650d49c5b56314774a7804df16fee4402c1f19d6d15d8c4730", size = 20503, upload-time = "2025-10-18T21:55:43.219Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/cb/b1/3846dd7f199d53cb17f49cba7e651e9ce294d8497c8c150530ed11865bb8/iniconfig-2.3.0-py3-none-any.whl", hash = "sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12", size = 7484, upload-time = "2025-10-18T21:55:41.639Z" }, +] + [[package]] name = "jiter" version = "0.16.0" @@ -654,6 +675,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/63/34/ba1c580383c9eada3711951fef0795c80b829a078d72188184bcab9dd527/packaging-26.3-py3-none-any.whl", hash = "sha256:d7193f7c8e4e93f444fde0262bf90af30e16fa0ad0ad44cb553c87339b23cd1c", size = 129956, upload-time = "2026-08-04T18:15:27.159Z" }, ] +[[package]] +name = "pluggy" +version = "1.6.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/f9/e2/3e91f31a7d2b083fe6ef3fa267035b518369d9511ffab804f839851d2779/pluggy-1.6.0.tar.gz", hash = "sha256:7dcc130b76258d33b90f61b658791dede3486c3e6bfb003ee5c9bfb396dd22f3", size = 69412, upload-time = "2025-05-15T12:30:07.975Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/54/20/4d324d65cc6d9205fabedc306948156824eb9f0ee1633355a8f7ec5c66bf/pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746", size = 20538, upload-time = "2025-05-15T12:30:06.134Z" }, +] + [[package]] name = "pycparser" version = "3.0" @@ -767,6 +797,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/30/a4/2bffa9f8e804325a09867f0e9d30795c80ea9f8d62560bd1b6ad6220eb2f/pydantic_settings-2.15.0-py3-none-any.whl", hash = "sha256:0ba092c291c94baceb5eff768aa0d56400a457585bc0175925a5a5510303da42", size = 69413, upload-time = "2026-08-07T09:24:55.839Z" }, ] +[[package]] +name = "pygments" +version = "2.21.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/49/2e/ced460408999b33da6b31b0021b0f37d329e202d4169aeb164493778f25b/pygments-2.21.0.tar.gz", hash = "sha256:610ca751c9bc2492b38eb9a38a7fbc93edbbb2d7182edaf34e66ae493dee5c8c", size = 5005329, upload-time = "2026-08-17T08:02:48.824Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/71/46/17f022dd3e953bf20a04a028a21ec746d942f8d2af30fa0f124fa0e6a684/pygments-2.21.0-py3-none-any.whl", hash = "sha256:2363c69b61c4a97c838da3b130dcd6468f4848992b21a82f2a63ec34377137d9", size = 1250147, upload-time = "2026-08-17T08:02:44.912Z" }, +] + [[package]] name = "pyjwt" version = "2.13.0" @@ -781,6 +820,35 @@ crypto = [ { name = "cryptography" }, ] +[[package]] +name = "pytest" +version = "9.1.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "colorama", marker = "sys_platform == 'win32'" }, + { name = "iniconfig" }, + { name = "packaging" }, + { name = "pluggy" }, + { name = "pygments" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/e4/47/b9efed96c114afcfa3c9d3fe98a76a1d14c74a9e266d397cf6eb64be5e01/pytest-9.1.1.tar.gz", hash = "sha256:1088fbde8f2b49d95a549a195707afa7a76a3ce9bcadc26b6d71f0ffda5fe313", size = 1636369, upload-time = "2026-06-19T10:58:32.857Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/24/25/1de2678b631f5a49215c6c96fff41ba892b0a34df68d6d80292b1b48aa7f/pytest-9.1.1-py3-none-any.whl", hash = "sha256:37a86b45efb9a47a61a36449063e8e18d0cab3161329fc099eb21783169c4f0c", size = 386536, upload-time = "2026-06-19T10:58:31.347Z" }, +] + +[[package]] +name = "pytest-asyncio" +version = "1.4.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "pytest" }, + { name = "typing-extensions", marker = "python_full_version < '3.13'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/43/7c/d36d04db312ecf4298932ef77e6e4a9e8ad017906e24e34f0b0c361a2473/pytest_asyncio-1.4.0.tar.gz", hash = "sha256:c6c0d2259945122819f171a32ecea2c349ead889ee28176caaf492143424be42", size = 58514, upload-time = "2026-05-26T09:56:04.083Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/03/e2/08a497ef684b88559c9cc5f4ad53a37e7b99e727094a86d6ea32536d5d3c/pytest_asyncio-1.4.0-py3-none-any.whl", hash = "sha256:933ca923a23075a87fb7070c0ec272a6848489824d887c85c812670932835aa1", size = 16930, upload-time = "2026-05-26T09:56:02.576Z" }, +] + [[package]] name = "python-dateutil" version = "2.9.0.post0" diff --git a/libs/chat/src/lib/primitives/chat-tool-calls/chat-tool-calls.component.spec.ts b/libs/chat/src/lib/primitives/chat-tool-calls/chat-tool-calls.component.spec.ts index 48a519f39..85cf95848 100644 --- a/libs/chat/src/lib/primitives/chat-tool-calls/chat-tool-calls.component.spec.ts +++ b/libs/chat/src/lib/primitives/chat-tool-calls/chat-tool-calls.component.spec.ts @@ -398,6 +398,48 @@ describe('ChatToolCallsComponent — subagent cards anchored to spawning task ca expect(fullText).toContain('search'); }); + it('anchors the card by the Subagent.toolCallId FIELD even when the map key differs', () => { + // The AG-UI adapter keys subagents() by subagentRunId (e.g. `-sub` + // for Strands' native SUBAGENT_* events) — only the wrapper's toolCallId + // field is guaranteed to match the spawning call. + const agent = mockAgent({ + withSubagents: true, + toolCalls: [ + { id: 'call_t', name: 'research_availability', args: {}, status: 'success' as never }, + ], + }); + const sub: Subagent = { + toolCallId: 'call_t', + name: 'availability_researcher', + status: signal('running'), + messages: signal([{ + id: 'm1', + role: 'assistant', + content: 'checking calendars', + delivery: staticDelivery('m1'), + }]), + state: signal({}), + }; + agent.subagents!.set(new Map([['call_t-sub', sub]])); + + const fixture = TestBed.createComponent(SubagentHost); + fixture.componentInstance.agent = agent; + fixture.componentInstance.message = { + id: 'a1', + role: 'assistant', + content: '', + toolCallIds: ['call_t'], + delivery: staticDelivery('a1'), + }; + fixture.detectChanges(); + + const cards = fixture.nativeElement.querySelectorAll('chat-subagent-card'); + expect(cards.length).toBe(1); + expect(cards[0].textContent).toContain('availability_researcher'); + // The spawning call renders AS the card, not as a generic tool-call chip. + expect(fixture.nativeElement.querySelectorAll('chat-tool-call-card').length).toBe(0); + }); + it('renders two separate subagent cards for two task calls (not collapsed into one strip)', () => { const agent = mockAgent({ withSubagents: true, diff --git a/libs/chat/src/lib/primitives/chat-tool-calls/chat-tool-calls.component.ts b/libs/chat/src/lib/primitives/chat-tool-calls/chat-tool-calls.component.ts index bd3e14c90..02cb2ea77 100644 --- a/libs/chat/src/lib/primitives/chat-tool-calls/chat-tool-calls.component.ts +++ b/libs/chat/src/lib/primitives/chat-tool-calls/chat-tool-calls.component.ts @@ -124,7 +124,15 @@ export class ChatToolCallsComponent { readonly groups = computed((): Group[] => { const excludeSet = new Set(this.excludeToolNames()); const calls = this.toolCalls().filter(tc => !excludeSet.has(tc.name)); - const subs = this.agent().subagents?.() ?? new Map(); + // Anchor subagents by the contract field `Subagent.toolCallId` (the tool + // call that spawned them), NOT by the adapter's map key. Adapters key the + // map differently — LangGraph by toolCallId, AG-UI activities by + // messageId, AG-UI native SUBAGENT_* by subagentRunId (e.g. + // `-sub`) — and only the wrapper's toolCallId is guaranteed + // to match the spawning call. + const rawSubs = this.agent().subagents?.() ?? new Map(); + const subs = new Map(); + rawSubs.forEach((sa) => subs.set(sa.toolCallId, sa)); const groupingMode = this.grouping(); const registry = this.templateRegistry(); const wildcard = registry.get('*');