feat(runtimes): strands subagent delegation demo with standard SUBAGENT_* emission

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 <chat-tool-calls> renders the
+// delegation inline as a <chat-subagent-card> 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);
+  });
+});

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"
+            }
+          }
+        ]
+      }
     }
   ]
 }

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"

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,18 +217,32 @@ 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.",
     config=StrandsAgentConfig(
         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,
+            ),
         },
     ),
 )

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),
+        )