diff --git a/cockpit/runtimes/microsoft-agent-framework/angular/e2e/fixtures/microsoft-agent-framework.json b/cockpit/runtimes/microsoft-agent-framework/angular/e2e/fixtures/microsoft-agent-framework.json index 1c1cf6aff..a441c5ab7 100644 --- a/cockpit/runtimes/microsoft-agent-framework/angular/e2e/fixtures/microsoft-agent-framework.json +++ b/cockpit/runtimes/microsoft-agent-framework/angular/e2e/fixtures/microsoft-agent-framework.json @@ -34,6 +34,32 @@ } ] } + }, + { + "match": { "userMessage": "conference travel", "hasToolResult": true }, + "response": { + "content": "Yes — submit the $900 conference travel expense. It is under the $1,500 travel cap; attach itemized receipts since it exceeds the $75 receipts threshold." + } + }, + { + "match": { "systemMessage": "expense-policy researcher" }, + "response": { + "content": "- Travel expenses up to $1,500 per trip are reimbursable with manager approval.\n- Any expense over the $75 receipts threshold requires itemized receipts.\n- Conference travel must be filed within 30 days of the trip end date." + } + }, + { + "match": { "userMessage": "conference travel" }, + "response": { + "toolCalls": [ + { + "name": "research_policy", + "arguments": { + "category": "travel", + "amount": 900 + } + } + ] + } } ] } diff --git a/cockpit/runtimes/microsoft-agent-framework/angular/e2e/manual/subagent-card-live.png b/cockpit/runtimes/microsoft-agent-framework/angular/e2e/manual/subagent-card-live.png new file mode 100644 index 000000000..e46f41cf5 Binary files /dev/null and b/cockpit/runtimes/microsoft-agent-framework/angular/e2e/manual/subagent-card-live.png differ diff --git a/cockpit/runtimes/microsoft-agent-framework/angular/e2e/microsoft-agent-framework.spec.ts b/cockpit/runtimes/microsoft-agent-framework/angular/e2e/microsoft-agent-framework.spec.ts index 02716871b..5ac29eeef 100644 --- a/cockpit/runtimes/microsoft-agent-framework/angular/e2e/microsoft-agent-framework.spec.ts +++ b/cockpit/runtimes/microsoft-agent-framework/angular/e2e/microsoft-agent-framework.spec.ts @@ -1,5 +1,6 @@ // SPDX-License-Identifier: MIT import { test, expect } from '@playwright/test'; +import { submitAndWaitForResponse } from '@threadplane-internal/e2e-harness'; // First proof outside unit tests that the neutral Agent contract's interrupt // path works against a genuinely non-LangGraph AG-UI backend: the Microsoft @@ -34,3 +35,22 @@ test.describe('cockpit runtimes/microsoft-agent-framework: expense approval', () await expect(page.getByText(/has been submitted for reimbursement/i)).toBeVisible({ timeout: 30_000 }); }); }); + +// Delegation over the same non-LangGraph bridge: the orchestrator's +// `research_policy` tool streams the tool-less `policy_researcher` +// specialist, and the queue-merge emitter (src/subagent_emitter.py) +// translates its deltas 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/microsoft-agent-framework: subagent delegation', () => { + test('rt-maf: delegated policy research renders a streaming subagent card', async ({ page }) => { + const bubble = await submitAndWaitForResponse( + page, + 'Should I submit a $900 conference travel expense? Research the policy first', + ); + await expect(page.locator('chat-subagent-card')).toHaveCount(1); + await expect(page.locator('chat-subagent-card')).toContainText('policy_researcher'); + await expect(bubble).toContainText(/policy|expense/i); + }); +}); diff --git a/cockpit/runtimes/microsoft-agent-framework/python/docs/wire-capture-subagents.md b/cockpit/runtimes/microsoft-agent-framework/python/docs/wire-capture-subagents.md new file mode 100644 index 000000000..93207619b --- /dev/null +++ b/cockpit/runtimes/microsoft-agent-framework/python/docs/wire-capture-subagents.md @@ -0,0 +1,269 @@ +# MAF delegation wire capture — subagent pattern decision (Task 0 spike) + +Date: 2026-09-02. Live capture against `src/server.py` (uvicorn, port 5330) with the +plain-OpenAI client path (`build_chat_client`, `gpt-4o-mini`). The scratch delegation +code described below was NOT committed; only this document is. + +Installed bridge inspected end to end: `agent-framework-ag-ui` 1.2.x in +`.venv/lib/python3.14/site-packages/agent_framework_ag_ui/` (all venv line numbers below +refer to that tree). + +## Verdict: Candidate A (agents-as-tools), emitter injects via a run-wrapper merge queue + +- **Candidate A works and is observable.** A specialist `Agent` invoked from an async + function tool streams its updates INTO the tool body in real time (101 streamed + updates observed in-tool for a ~550-char answer). That is everything the emitter + needs to synthesize the SUBAGENT_* sequence with streaming child deltas. +- **Candidate B (two-executor workflow) is not needed.** The endpoint does mount + workflows (`_endpoint.py:137-138` wraps a `Workflow` in `AgentFrameworkWorkflow`), + so B remains a fallback, but A is simpler and keeps the demo's existing + approval/predictive-state surfaces untouched. + +## Seam analysis (venv file:line) + +### (a) Where MAF run events become AG-UI events + +- Single entry point: `run_agent_stream` (`agent_framework_ag_ui/_agent_run.py:2259`), + reached from `AgentFrameworkAgent.run` (`_agent.py:147-166`). +- The wrapped agent is invoked at `_agent_run.py:2723` + (`response_stream = (a2ui_runner or agent).run(messages, stream=True, **run_kwargs)`); + updates are pulled at `_agent_run.py:2726` and each content item is converted to + AG-UI events by `_emit_content` (`_run_common.py:1166`, dispatched from + `_agent_run.py:2828`). `_emit_content` handles `text`, `function_call`, + `function_result`, `function_approval_request`, `usage`, reasoning, and MCP content + types (`_run_common.py:1174-1200`); anything else is dropped with a debug log + (`_run_common.py:1200`). +- The FastAPI endpoint consumes `protocol_runner.run(input_data)` and encodes each + yielded event generically (`_endpoint.py:212-242`). + +### (b) Can a function tool reach an event emitter/queue/context? + +**No.** There is no ContextVar, queue, writer, or middleware hook anywhere in +`agent_framework_ag_ui/*.py` or in `agent_framework/_tools.py` / `_middleware.py` / +`_agents.py` that a tool body could use to inject AG-UI events +(`grep -rn ContextVar` over those modules returns nothing). The event pipeline is a +pure pull-driven async generator; tools execute deep inside the framework's function +invocation loop within `agent.run(stream=True)` and only their return value surfaces +(as `function_result` content → `TOOL_CALL_RESULT`). + +**Injection seam (named):** wrap `AgentFrameworkAgent.run` — the exact method the +endpoint calls at `_endpoint.py:212`. Our emitter will be a small subclass (or +compositional wrapper) in the demo: + +1. `run()` creates an `asyncio.Queue` and sets a module-level `ContextVar` to it + before delegating to the inner `run_agent_stream` generator. Because the tool body + executes on the same async call chain (endpoint → wrapper → `run_agent_stream` → + `agent.run` → function invocation), the ContextVar value propagates into the tool. +2. The wrapper pumps the inner generator into the same queue from an + `asyncio.create_task` and yields from the merged queue. This is required for LIVE + interleaving: while the tool runs, the bridge generator is suspended awaiting the + next provider update, so a naive "drain queue between inner yields" design would + batch all child deltas until the tool returns. With the pump-task merge, a + `queue.put_nowait` from the tool body wakes the outer consumer immediately. +3. The tool body reads the ContextVar and enqueues + `SubagentStartedEvent {subagentRunId: -sub, name: "policy_researcher", + parentToolCallId: }` → attributed `TextMessageStart/Content×N/End` + (one delta per specialist update) → `SubagentFinishedEvent success` + (`SubagentErrorEvent` on exception). The tool's own `toolCallId` is available to + the body via the framework's function-call content on the update stream; the + emitter wrapper can also correlate it by observing the preceding + `TOOL_CALL_START` for the delegation tool on the bridge stream. + +This is the same "emit from inside the tool body" shape the Strands PR proved, with +the writer supplied by our own wrapper instead of the runtime (MAF's bridge provides +none). Reference translator: `cockpit/ag-ui/subagents/python/src/streaming/activity_emitting_agent.py`. + +### (c) Does the encoder accept ag_ui.core pydantic events generally? + +**Yes.** `EventEncoder.encode` takes any `BaseEvent` and does a generic +model-dump → SSE `data:` frame (`ag_ui/encoder/encoder.py`, `encode`/`_encode_sse`); +the endpoint applies it uniformly with no per-type allowlist (`_endpoint.py:224`). +`Subagent*` events are `BaseEvent` subclasses, so they pass through untouched. + +SDK check (in this venv): + +``` +$ 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'] +``` + +### (d) What does the bridge do with nested-agent activity inside a tool? + +**Nothing is observable.** The specialist's `run(stream=True)` updates are consumed +entirely inside the tool body; the bridge sees only the tool's `function_call` +(streamed as `TOOL_CALL_START/ARGS/END`) and its string return value +(`TOOL_CALL_RESULT`). No ACTIVITY_*, no per-child events, no specialist name on the +wire beyond the delegation tool's own name. This matches the "measured red upstream" +note in `src/agent.py` and is confirmed by the capture below. + +## Scratch setup (uncommitted, reverted after capture) + +Added to `src/agent.py`: a `policy_researcher` `Agent` (same `build_chat_client()`, +instructions: expense-policy researcher, 3 short bullets) plus an async +`@tool research_policy(category: str, amount: float) -> str` that ran +`specialist.run(prompt, stream=True)`, accumulated `update.text`, logged each update +to stderr, and returned the joined text; registered on the primary agent with one +instruction sentence about delegating policy research. + +## In-tool streaming datum + +The specialist's deltas DID stream into the tool body, token by token: + +``` +[spike] specialist update #2: '-' +[spike] specialist update #3: ' **' +[spike] specialist update #4: 'Pre' +... +[spike] specialist DONE: 101 streamed updates, 548 chars (attempt 2; attempt 1: 106 updates, 582 chars) +``` + +So the emitter can produce **streaming child deltas** (preferred contract), not just a +single final chunk. + +## Live wire capture (attempt 2 of 2; attempt 1 also delegated but was truncated client-side) + +Request: POST `/agent` with `threadId: spike-thread-2`, `runId: spike-run-2`, single +user message "Should I submit a $900 conference travel expense? Research the policy +first". The model delegated on the first turn in both attempts. Full event-type +census of the complete stream: + +``` +1 RUN_STARTED 1 STATE_SNAPSHOT 3 CUSTOM (PredictState, usage×2) +2 TEXT_MESSAGE_START 35 TEXT_MESSAGE_CONTENT 2 TEXT_MESSAGE_END +1 TOOL_CALL_START 9 TOOL_CALL_ARGS 1 TOOL_CALL_END 1 TOOL_CALL_RESULT +1 MESSAGES_SNAPSHOT 1 RUN_FINISHED +0 SUBAGENT_* / ACTIVITY_* / anything child-related +``` + +Abridged stream (ids as captured; no secrets present): + +``` +data: {"type":"RUN_STARTED","threadId":"spike-thread-2","runId":"spike-run-2"} +data: {"type":"CUSTOM","name":"PredictState","value":[{"state_key":"expense","tool":"submit_expense","tool_argument":"expense"}]} +data: {"type":"STATE_SNAPSHOT","snapshot":{"expense":{}}} +data: {"type":"TEXT_MESSAGE_START","messageId":"5cff954e-...","role":"assistant"} +data: {"type":"TOOL_CALL_START","toolCallId":"call_7sxPY1sC236nPyHRTWAZMJB9","toolCallName":"research_policy","parentMessageId":"5cff954e-..."} +data: {"type":"TOOL_CALL_ARGS","toolCallId":"call_7sxP...","delta":"{\""} +... (9 ARGS deltas spelling {"category":"travel","amount":900}) +data: {"type":"TOOL_CALL_END","toolCallId":"call_7sxP..."} + <-- specialist runs HERE; 101 updates streamed in-tool; NOTHING on the wire --> +data: {"type":"TOOL_CALL_RESULT","messageId":"c2c72fd2-...","toolCallId":"call_7sxP...","content":"1. **Pre-Approval Required**: Travel expenses exceeding $500 ...","role":"tool"} +data: {"type":"TEXT_MESSAGE_END","messageId":"5cff954e-..."} +data: {"type":"TEXT_MESSAGE_START","messageId":"534fe3b3-...","role":"assistant"} +data: {"type":"TEXT_MESSAGE_CONTENT","messageId":"534fe3b3-...","delta":"The"} +... (35 deltas: policy summary + ask to confirm approval) +data: {"type":"TEXT_MESSAGE_END","messageId":"534fe3b3-..."} +data: {"type":"MESSAGES_SNAPSHOT","messages":[...user, assistant toolCalls(research_policy), tool result, assistant text...]} +data: {"type":"RUN_FINISHED","threadId":"spike-thread-2","runId":"spike-run-2"} +``` + +### Explicit statements + +- **Native delegation on the wire:** an ordinary function tool call — + `TOOL_CALL_START(research_policy)` → streamed `TOOL_CALL_ARGS` → `TOOL_CALL_END` → + a single `TOOL_CALL_RESULT` carrying the specialist's complete final text. The + wall-clock gap between `TOOL_CALL_END` and `TOOL_CALL_RESULT` is where the + specialist runs, silently. +- **Child updates streamed in-tool:** YES — 101 streamed updates (attempt 2; 106 in + attempt 1), token-granular. +- **Anything child-related on the wire:** NO — zero events; the specialist is + invisible except as the tool's result string. + +## Emitter plan (for the implementation PR) + +Target sequence, injected by the wrapper-queue seam around the existing bridge stream +for tool call id ``: + +`SUBAGENT_STARTED {subagentRunId: "-sub", name: "policy_researcher", parentToolCallId: ""}` +→ `TEXT_MESSAGE_START/CONTENT×N/END` attributed to the subagent run (one CONTENT per +specialist update; live-interleaved via the pump-task merge) → `SUBAGENT_FINISHED +{outcome: success}` (or `SUBAGENT_ERROR` on tool-body exception), all before the +bridge's own `TOOL_CALL_RESULT` for `` reaches the client. + +## After the emitter + +Date: 2026-09-02, post-implementation. Live capture against the committed +`src/server.py` (uvicorn, port 5330; plain-OpenAI path, `gpt-4o-mini`), request +identical in shape to the spike: single user message "Should I submit a $900 +conference travel expense? Research the policy first" (`threadId: +smoke-thread-1`, `runId: smoke-run-1`). The model called +`lookup_expense_policy` first and then delegated via `research_policy` on the +same turn. Full event-type census of the complete stream: + +``` +1 RUN_STARTED 1 STATE_SNAPSHOT 4 CUSTOM (PredictState, usage×3) +4 TEXT_MESSAGE_START 128 TEXT_MESSAGE_CONTENT 4 TEXT_MESSAGE_END +2 TOOL_CALL_START 14 TOOL_CALL_ARGS 2 TOOL_CALL_END 2 TOOL_CALL_RESULT +1 SUBAGENT_STARTED 1 SUBAGENT_FINISHED 0 SUBAGENT_ERROR +1 MESSAGES_SNAPSHOT 1 RUN_FINISHED +``` + +**Child delta count: 100 streamed TEXT_MESSAGE_CONTENT events** attributed to +the subagent run (`messageId: -sub-m1`, token-granular — same order of +magnitude as the spike's 101/106 in-tool updates), live-interleaved between the +bridge's own events: the bridge yields `TOOL_CALL_END` for `research_policy` +only after the tool returns, and the entire SUBAGENT_* sequence lands between +`TOOL_CALL_ARGS` and that `TOOL_CALL_END` — proof the pump-task merge queue +delivered the deltas while the bridge generator was suspended inside the tool. + +Abridged stream around the delegation (ids as captured; no secrets present): + +``` +data: {"type":"TOOL_CALL_START","toolCallId":"call_drwxfbPJnrzSBBLnrcqehqfx","toolCallName":"research_policy","parentMessageId":"c1a737b4-..."} +... (11 ARGS deltas spelling {"category":"travel","amount":900}) +data: {"type":"SUBAGENT_STARTED","subagentRunId":"call_drwxfbPJnrzSBBLnrcqehqfx-sub","name":"policy_researcher","parentToolCallId":"call_drwxfbPJnrzSBBLnrcqehqfx"} +data: {"type":"TEXT_MESSAGE_START","messageId":"call_drwxfbPJnrzSBBLnrcqehqfx-sub-m1","role":"assistant","subagentRunId":"call_drwxfbPJnrzSBBLnrcqehqfx-sub"} +data: {"type":"TEXT_MESSAGE_CONTENT","messageId":"call_drwxfbPJnrzSBBLnrcqehqfx-sub-m1","delta":"-","subagentRunId":"call_drwxfbPJnrzSBBLnrcqehqfx-sub"} +data: {"type":"TEXT_MESSAGE_CONTENT","messageId":"call_drwxfbPJnrzSBBLnrcqehqfx-sub-m1","delta":" **","subagentRunId":"call_drwxfbPJnrzSBBLnrcqehqfx-sub"} +data: {"type":"TEXT_MESSAGE_CONTENT","messageId":"call_drwxfbPJnrzSBBLnrcqehqfx-sub-m1","delta":"Pre","subagentRunId":"call_drwxfbPJnrzSBBLnrcqehqfx-sub"} +... (100 CONTENT deltas total, token-granular) +data: {"type":"TEXT_MESSAGE_END","messageId":"call_drwxfbPJnrzSBBLnrcqehqfx-sub-m1","subagentRunId":"call_drwxfbPJnrzSBBLnrcqehqfx-sub"} +data: {"type":"SUBAGENT_FINISHED","subagentRunId":"call_drwxfbPJnrzSBBLnrcqehqfx-sub","outcome":{"type":"success"}} +data: {"type":"TOOL_CALL_END","toolCallId":"call_drwxfbPJnrzSBBLnrcqehqfx"} +data: {"type":"TOOL_CALL_RESULT","messageId":"7dec6623-...","toolCallId":"call_drwxfbPJnrzSBBLnrcqehqfx","content":"- **Pre-Approval Required**: Travel expenses exceeding $500 ...","role":"tool"} +``` + +Tid derivation on the live run: the pump-recorded `TOOL_CALL_START` path +(`current_tool_call_id("research_policy")`) — the real wire toolCallId keys +the whole sequence (`subagentRunId = -sub`, `parentToolCallId = `); +the generated `sub-` fallback was not needed. Existing surfaces are +untouched: PredictState CUSTOM, STATE_SNAPSHOT, both TOOL_CALL_RESULTs, and +MESSAGES_SNAPSHOT all present as before. + +## Browser verification + +2026-09-02, live backend (real `OPENAI_API_KEY`, uvicorn on :5330) + `nx +serve cockpit-runtimes-microsoft-agent-framework-angular` on :4330, driven +headlessly with Playwright. Screenshot: +`cockpit/runtimes/microsoft-agent-framework/angular/e2e/manual/subagent-card-live.png`. + +What rendered: the delegation prompt (*"Should I submit a $900 conference +travel expense? Research the policy first"*) produced an inline +`` anchored to the `research_policy` tool call — header +`policy_researcher` + wire `toolCallId` + status badge — with the +specialist's 3-bullet policy transcript inside it, preceded by the +orchestrator's own `lookup_expense_policy` chip and followed by its 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 card mounting at 66 chars (header only) the moment +SUBAGENT_STARTED landed, then the specialist's message growing +monotonically while the run was live — one run sampled 66 → 163 → 277 → +398 → 422 → 553 → 622 chars between t≈1.5s and t≈3.0s; a second run +sampled 66 → 105 → 207 → 314 → 430 → 519 → 614 chars — before the badge +flipped to `complete` and the card collapsed to its 67-char summary row. +This confirms the queue-merge emitter's attributed `TEXT_MESSAGE_CONTENT` +deltas render progressively in the card during the run, not as one +post-hoc paste. + +Same `libs/chat` anchoring dependency as the Strands verification: the wire +and the `@threadplane/ag-ui` reducer were correct, but `chat-tool-calls` +anchored subagent cards by the adapter's `subagents()` map key (for native +SUBAGENT_* that is the `subagentRunId`, `-sub`) instead of the +contract field `Subagent.toolCallId`, so the card never mounted. The fix +(re-index on `Subagent.toolCallId`) plus its pinning spec are cherry-picked +onto this branch. diff --git a/cockpit/runtimes/microsoft-agent-framework/python/pyproject.toml b/cockpit/runtimes/microsoft-agent-framework/python/pyproject.toml index 8f718a771..dbf6404e4 100644 --- a/cockpit/runtimes/microsoft-agent-framework/python/pyproject.toml +++ b/cockpit/runtimes/microsoft-agent-framework/python/pyproject.toml @@ -10,9 +10,18 @@ dependencies = [ "uvicorn[standard]>=0.29", ] +[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/microsoft-agent-framework/python/src/agent.py b/cockpit/runtimes/microsoft-agent-framework/python/src/agent.py index 9414d37f9..4d28e8c45 100644 --- a/cockpit/runtimes/microsoft-agent-framework/python/src/agent.py +++ b/cockpit/runtimes/microsoft-agent-framework/python/src/agent.py @@ -14,8 +14,13 @@ protocol-standard ``RUN_FINISHED.outcome = {type: 'interrupt', ...}`` and resumes from the client's top-level ``resume`` entries. -No subagents surface: the MAF bridge emits no per-subagent -ACTIVITY_SNAPSHOT/ACTIVITY_DELTA stream (measured red upstream). +- subagents: ``research_policy`` delegates to the tool-less + ``policy_researcher`` specialist Agent and streams its deltas through the + ``delegation_*`` helpers in src/subagent_emitter.py, which merge standard + ``SUBAGENT_*`` + attributed child ``TEXT_MESSAGE_*`` events into the run + stream at the run-wrapper seam (the MAF bridge natively emits NOTHING for + nested-agent activity inside a tool — measured red upstream and in + docs/wire-capture-subagents.md). Model client: Azure OpenAI is the DEFAULT path — when ``AZURE_OPENAI_ENDPOINT`` is set the client routes to Azure (key auth via @@ -32,6 +37,8 @@ from agent_framework.openai import OpenAIChatCompletionClient from pydantic import BaseModel, Field +from . import subagent_emitter + _POLICIES = { "meals": {"limit_usd": 300, "receipt_required_over_usd": 25, "notes": "Team meals require attendee count in the memo."}, "travel": {"limit_usd": 1500, "receipt_required_over_usd": 0, "notes": "Book through the travel portal when possible."}, @@ -97,6 +104,10 @@ def submit_expense(expense: Expense) -> str: _INSTRUCTIONS = """You are an expense approval copilot. +Before recommending whether to submit an expense, delegate the policy +research to the specialist by calling `research_policy` with the category +and amount. + When the user asks to file an expense: 1. FIRST call `lookup_expense_policy` with the expense category. 2. THEN call `submit_expense` with the complete structured expense @@ -138,12 +149,63 @@ def build_chat_client() -> OpenAIChatCompletionClient: ) +policy_researcher = Agent( + name="policy_researcher", + instructions=( + "You are an expense-policy researcher. Given an expense category and " + "amount, summarize the applicable policy rules in 3 short bullets." + ), + client=build_chat_client(), +) + + +@tool( + name="research_policy", + description="Delegate policy research for this expense to a specialist.", +) +async def research_policy(category: str, amount: float) -> str: + """Delegate policy research for this expense to a specialist. + + Streams the ``policy_researcher`` specialist and mirrors each text delta + onto the AG-UI wire as attributed SUBAGENT_* / TEXT_MESSAGE_* events via + src/subagent_emitter.py (no-ops outside the wrapped run). + + Args: + category: Expense category, e.g. 'meals' or 'travel'. + amount: Expense amount in USD. + + Returns: + The specialist's complete policy summary. + """ + # Deterministically recorded by the run wrapper's pump before this body + # runs (the bridge streams TOOL_CALL_START/ARGS/END first); None when + # invoked outside a wrapped run. + tid = subagent_emitter.current_tool_call_id("research_policy") + subagent_emitter.delegation_started(tid, policy_researcher.name) + parts: list[str] = [] + try: + prompt = ( + f"Expense category: {category}. Amount: ${amount:.2f}. " + "Summarize the applicable policy rules." + ) + async for update in policy_researcher.run(prompt, stream=True): + text = update.text + if text: + parts.append(text) + subagent_emitter.delegation_delta(tid, text) + except Exception as exc: + subagent_emitter.delegation_error(tid, str(exc)) + raise + subagent_emitter.delegation_finished(tid) + return "".join(parts) + + agent = AgentFrameworkAgent( agent=Agent( name="expense_approval_copilot", instructions=_INSTRUCTIONS, client=build_chat_client(), - tools=[lookup_expense_policy, submit_expense], + tools=[lookup_expense_policy, research_policy, submit_expense], ), name="ExpenseApprovalCopilot", description="Files expense reports with policy lookup, shared state, and human approval.", diff --git a/cockpit/runtimes/microsoft-agent-framework/python/src/server.py b/cockpit/runtimes/microsoft-agent-framework/python/src/server.py index 7b192ca0b..e26a54f02 100644 --- a/cockpit/runtimes/microsoft-agent-framework/python/src/server.py +++ b/cockpit/runtimes/microsoft-agent-framework/python/src/server.py @@ -3,9 +3,14 @@ from agent_framework_ag_ui import add_agent_framework_fastapi_endpoint from .agent import agent +from .subagent_emitter import wrap_agent_run app = FastAPI(title="cockpit-runtimes-microsoft-agent-framework") -add_agent_framework_fastapi_endpoint(app, agent, path="/agent") +# The wrapper is the SUBAGENT_* injection seam: the endpoint consumes +# protocol_runner.run, and wrap_agent_run merges the delegation tool's +# enqueued child events into that stream (src/subagent_emitter.py). +wrapped_agent = wrap_agent_run(agent) +add_agent_framework_fastapi_endpoint(app, wrapped_agent, path="/agent") @app.get("/ok") diff --git a/cockpit/runtimes/microsoft-agent-framework/python/src/subagent_emitter.py b/cockpit/runtimes/microsoft-agent-framework/python/src/subagent_emitter.py new file mode 100644 index 000000000..0b231ab4c --- /dev/null +++ b/cockpit/runtimes/microsoft-agent-framework/python/src/subagent_emitter.py @@ -0,0 +1,309 @@ +# SPDX-License-Identifier: MIT +"""SUBAGENT_* emitter for the `research_policy` delegation tool. + +The MAF AG-UI bridge is a pure pull-driven async generator with no writer a +tool body could reach: the specialist's streamed updates are consumed +entirely inside the tool and only the return string surfaces (as +TOOL_CALL_RESULT). Measured in docs/wire-capture-subagents.md, which also +names the injection seam implemented here: wrap ``AgentFrameworkAgent.run`` +— the exact method the FastAPI endpoint consumes (`_endpoint.py:212`) — +with a queue-merge generator. + +How the seam works (``SubagentEmittingAgent`` / ``wrap_agent_run``): + +1. ``run()`` creates an ``asyncio.Queue`` and publishes it (plus a small + correlation map) through a module-level ``ContextVar``. The delegation + tool executes on the same async call chain, so the value propagates + into the tool body. +2. A pump task drains the inner bridge generator into that queue. This is + required for LIVE interleaving: while the tool runs, the bridge + generator is suspended awaiting the next provider update, so a naive + "drain between inner yields" design would batch every child delta until + the tool returned. With the pump-task merge, a ``put_nowait`` from the + tool body wakes the outer consumer immediately. +3. The tool body calls the ``delegation_*`` helpers below, which build the + typed ``ag_ui.core`` events and enqueue them: + + SUBAGENT_STARTED {subagentRunId: -sub, parentToolCallId: } + TEXT_MESSAGE_START/CONTENT.../END (streamed specialist deltas) + SUBAGENT_FINISHED outcome=success (or SUBAGENT_ERROR on failure) + +Correlation: the pump appends every TOOL_CALL_START's ``toolCallId`` to a +per-tool-name FIFO as it passes through the queue, and each tool body pops +the oldest via ``current_tool_call_id`` — so a multi-tool batch calling the +same tool twice (MAF runs batches concurrently via ``asyncio.gather``, and +the bridge streams all TOOL_CALL_STARTs first) gives each invocation its +own tid and its own delegation. The bridge yields TOOL_CALL_START (and the +ARGS deltas) for the delegation call BEFORE the framework invokes the tool +on the same driving chain, so the FIFO is deterministically populated by +the time the tool body runs; TOOL_CALL_END arrives only AFTER the tool +returns (measured wire order — the SUBAGENT_* sequence lands between ARGS +and END), which is why correlation relies on START alone. If a caller ever +invokes the tool outside +the wrapped run, the helpers fall back to a generated ``sub-`` run id +with ``parentToolCallId`` omitted — and with no queue at all they are pure +no-ops, which is what keeps unit tests and direct agent runs side-effect +free. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import uuid +from contextvars import ContextVar +from dataclasses import dataclass, field +from typing import Any, AsyncGenerator + +from ag_ui.core import ( + BaseEvent, + EventType, + SubagentErrorEvent, + SubagentFinishedEvent, + SubagentFinishedSuccessOutcome, + SubagentStartedEvent, + TextMessageContentEvent, + TextMessageEndEvent, + TextMessageStartEvent, +) +from agent_framework.ag_ui import AgentFrameworkAgent + +DELEGATION_TOOL_NAME = "research_policy" + + +@dataclass +class _Delegation: + """Lifecycle of one delegation call, keyed by parent tool-call id.""" + + run_id: str + message_id: str + message_open: bool = False + finished: bool = False + + +@dataclass +class _EmitterSession: + """Per-run channel shared between the run wrapper and the tool body.""" + + queue: asyncio.Queue[Any] + # Per-tool-name FIFO of not-yet-claimed TOOL_CALL_START toolCallIds: + # the pump appends, each tool body pops the oldest — one tid per call + # even when a batch invokes the same tool twice. + tool_call_ids: dict[str, list[str]] = field(default_factory=dict) + runs: dict[str | None, _Delegation] = field(default_factory=dict) + + +_event_queue: ContextVar[_EmitterSession | None] = ContextVar( + "maf_subagent_emitter_session", default=None +) + + +def current_tool_call_id(tool_name: str) -> str | None: + """Claim the oldest unclaimed TOOL_CALL_START toolCallId for a tool. + + Pops from the per-name FIFO the pump fills, so each concurrent + invocation of the same tool gets its own tid. Deterministically + populated before the tool body runs (see module docstring); ``None`` + outside a wrapped run. + """ + session = _event_queue.get() + if session is None: + return None + pending = session.tool_call_ids.get(tool_name) + if not pending: + return None + return pending.pop(0) + + +def emit(event: BaseEvent) -> None: + """Enqueue one AG-UI event onto the live run stream; no-op unwrapped.""" + session = _event_queue.get() + if session is not None: + session.queue.put_nowait(event) + + +def delegation_started(tid: str | None, name: str) -> None: + """Announce the specialist run. Ids derive from the delegation tool-call + id; without one (unwrapped fallback) a ``sub-`` run id is generated + and ``parentToolCallId`` omitted.""" + session = _event_queue.get() + run_id = f"{tid}-sub" if tid else f"sub-{uuid.uuid4().hex[:8]}" + if session is not None: + session.runs[tid] = _Delegation(run_id=run_id, message_id=f"{run_id}-m1") + emit( + SubagentStartedEvent( + type=EventType.SUBAGENT_STARTED, + subagent_run_id=run_id, + name=name, + parent_tool_call_id=tid, + ) + ) + + +def _active(tid: str | None) -> _Delegation | None: + session = _event_queue.get() + if session is None: + return None + delegation = session.runs.get(tid) + if delegation is None or delegation.finished: + return None + return delegation + + +def delegation_delta(tid: str | None, text: str) -> None: + """Stream one specialist text delta, lazily opening the attributed + message (so a zero-delta run emits no empty message).""" + delegation = _active(tid) + if delegation is None or not text: + return + if not delegation.message_open: + delegation.message_open = True + emit( + TextMessageStartEvent( + type=EventType.TEXT_MESSAGE_START, + message_id=delegation.message_id, + role="assistant", + subagent_run_id=delegation.run_id, + ) + ) + emit( + TextMessageContentEvent( + type=EventType.TEXT_MESSAGE_CONTENT, + message_id=delegation.message_id, + delta=text, + subagent_run_id=delegation.run_id, + ) + ) + + +def _close_message(delegation: _Delegation) -> None: + if delegation.message_open: + delegation.message_open = False + emit( + TextMessageEndEvent( + type=EventType.TEXT_MESSAGE_END, + message_id=delegation.message_id, + subagent_run_id=delegation.run_id, + ) + ) + + +def delegation_finished(tid: str | None) -> None: + """Close the open child message and finish the subagent with success.""" + delegation = _active(tid) + if delegation is None: + return + delegation.finished = True + _close_message(delegation) + emit( + SubagentFinishedEvent( + type=EventType.SUBAGENT_FINISHED, + subagent_run_id=delegation.run_id, + outcome=SubagentFinishedSuccessOutcome(), + ) + ) + + +def delegation_error(tid: str | None, message: str) -> None: + """Close the open child message and report the specialist failure. The + tool re-raises afterwards, so the bridge's own tool-error path still + runs normally.""" + delegation = _active(tid) + if delegation is None: + return + delegation.finished = True + _close_message(delegation) + emit( + SubagentErrorEvent( + type=EventType.SUBAGENT_ERROR, + subagent_run_id=delegation.run_id, + message=message, + ) + ) + + +class _PumpFailure: + """Sentinel carrying an inner-generator exception across the queue.""" + + def __init__(self, exc: BaseException) -> None: + self.exc = exc + + +_DONE = object() + + +def _record_tool_call(session: _EmitterSession, event: Any) -> None: + if getattr(event, "type", None) == EventType.TOOL_CALL_START: + session.tool_call_ids.setdefault(event.tool_call_name, []).append( + event.tool_call_id + ) + + +class SubagentEmittingAgent(AgentFrameworkAgent): + """AgentFrameworkAgent whose ``run`` merges tool-enqueued SUBAGENT_* + events into the bridge stream via the pump-task queue. + + Constructed from an already-configured ``AgentFrameworkAgent`` (shares + its config and approval-state store rather than re-running ``__init__``), + so the endpoint's ``isinstance(agent, AgentFrameworkAgent)`` dispatch + and approval resume flow are untouched. + """ + + def __init__(self, inner: AgentFrameworkAgent) -> None: + self._inner = inner + self.agent = inner.agent + self.name = inner.name + self.description = inner.description + self.config = inner.config + self._approval_state_store = inner._approval_state_store + + async def run( + self, input_data: dict[str, Any] + ) -> AsyncGenerator[BaseEvent, None]: + queue: asyncio.Queue[Any] = asyncio.Queue() + session = _EmitterSession(queue=queue) + token = _event_queue.set(session) + inner_gen = self._inner.run(input_data) + + async def _pump() -> None: + try: + async for event in inner_gen: + _record_tool_call(session, event) + queue.put_nowait(event) + except asyncio.CancelledError: + raise + except BaseException as exc: # propagate to the consumer, never swallow + queue.put_nowait(_PumpFailure(exc)) + else: + queue.put_nowait(_DONE) + + # create_task copies the current context AFTER the ContextVar set, + # so the tool body (which executes on the pump's driving chain) + # sees this session. + pump = asyncio.create_task(_pump()) + try: + while True: + item = await queue.get() + if item is _DONE: + break + if isinstance(item, _PumpFailure): + raise item.exc + yield item + finally: + # Consumer break / client disconnect (GeneratorExit) or pump + # failure: cancel and await the pump so no task is orphaned, + # then close the inner generator. + pump.cancel() + with contextlib.suppress(asyncio.CancelledError): + await pump + with contextlib.suppress(Exception): + await inner_gen.aclose() + # reset raises ValueError if a GC-driven aclose runs the finally + # in a different context than the one that set the var. + with contextlib.suppress(ValueError): + _event_queue.reset(token) + + +def wrap_agent_run(agent: AgentFrameworkAgent) -> SubagentEmittingAgent: + """Wrap an AgentFrameworkAgent so its run stream carries SUBAGENT_*.""" + return SubagentEmittingAgent(agent) diff --git a/cockpit/runtimes/microsoft-agent-framework/python/tests/test_delegation.py b/cockpit/runtimes/microsoft-agent-framework/python/tests/test_delegation.py new file mode 100644 index 000000000..16e465e26 --- /dev/null +++ b/cockpit/runtimes/microsoft-agent-framework/python/tests/test_delegation.py @@ -0,0 +1,58 @@ +# SPDX-License-Identifier: MIT +"""Tests for the `research_policy` delegation scenario — the tool is a +registered async `@tool` that hands expense-policy research to the tool-less +`policy_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, policy_researcher, research_policy + + +def _tool_names() -> list[str]: + return [t.name for t in agent.agent.default_options["tools"]] + + +def test_research_policy_is_registered_on_the_agent(): + assert "research_policy" in _tool_names() + # Existing tools stay registered untouched. + assert "lookup_expense_policy" in _tool_names() + assert "submit_expense" in _tool_names() + + +def test_tool_name_and_docstring(): + assert research_policy.name == "research_policy" + assert research_policy.description.startswith( + "Delegate policy research for this expense to a specialist." + ) + schema = research_policy.parameters() + assert set(schema["required"]) == {"category", "amount"} + + +def test_tool_is_async(): + # The seam depends on it: the tool must be able to async-iterate the + # specialist's streamed updates and enqueue deltas as they arrive. + assert inspect.iscoroutinefunction(research_policy.func) + + +def test_specialist_is_toolless_researcher(): + assert policy_researcher.name == "policy_researcher" + assert policy_researcher.default_options.get("tools") == [] + assert "expense-policy researcher" in policy_researcher.default_options["instructions"] + + +def test_instructions_mention_delegation(): + instructions = agent.agent.default_options["instructions"] + assert "research_policy" in instructions + + +def test_untouched_surfaces_still_configured(): + # The subagent scenario must not disturb the existing shared-state and + # approval surfaces. + assert agent.config.state_schema == { + "expense": {"type": "object", "description": "The expense entry being drafted."}, + } + assert agent.config.predict_state_config == { + "expense": {"tool": "submit_expense", "tool_argument": "expense"}, + } diff --git a/cockpit/runtimes/microsoft-agent-framework/python/tests/test_subagent_emitter.py b/cockpit/runtimes/microsoft-agent-framework/python/tests/test_subagent_emitter.py new file mode 100644 index 000000000..a9a28d0fa --- /dev/null +++ b/cockpit/runtimes/microsoft-agent-framework/python/tests/test_subagent_emitter.py @@ -0,0 +1,410 @@ +# SPDX-License-Identifier: MIT +"""Tests for the SUBAGENT_* emitter — drives the `delegation_*` helpers and +the queue-merge run wrapper with a fake inner bridge generator plus a +scripted tool enqueue (the fake generator calls the helpers between its own +yields, exactly where the framework invokes the real tool on the pump's +driving chain) and asserts the exact merged sequence field-for-field.""" + +import asyncio + +import pytest + +from ag_ui.core import ( + EventType, + RunFinishedEvent, + RunStartedEvent, + ToolCallArgsEvent, + ToolCallEndEvent, + ToolCallResultEvent, + ToolCallStartEvent, +) + +from src import subagent_emitter +from src.subagent_emitter import ( + SubagentEmittingAgent, + current_tool_call_id, + delegation_delta, + delegation_error, + delegation_finished, + delegation_started, + wrap_agent_run, +) + +TID = "call_7sxPY1sC236nPyHRTWAZMJB9" +RUN_ID = f"{TID}-sub" +MESSAGE_ID = f"{TID}-sub-m1" + + +# --------------------------------------------------------------------------- +# Helper-level tests: install a session directly and inspect the queue. +# --------------------------------------------------------------------------- + + +@pytest.fixture +def session(): + s = subagent_emitter._EmitterSession(queue=asyncio.Queue()) + token = subagent_emitter._event_queue.set(s) + yield s + subagent_emitter._event_queue.reset(token) + + +def _drain(session) -> list: + out = [] + while not session.queue.empty(): + out.append(session.queue.get_nowait()) + return out + + +def test_success_sequence_field_for_field(session): + delegation_started(TID, "policy_researcher") + delegation_delta(TID, "- Pre") + delegation_delta(TID, "-approval") + delegation_delta(TID, " required") + delegation_finished(TID) + + out = _drain(session) + assert [ev.type for ev in out] == [ + 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 == "policy_researcher" + assert started.parent_tool_call_id == TID + + 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] == ["- Pre", "-approval", " required"] + 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" + + +def test_no_deltas_still_brackets_with_started_and_finished(session): + delegation_started(TID, "policy_researcher") + delegation_finished(TID) + assert [ev.type for ev in _drain(session)] == [ + EventType.SUBAGENT_STARTED, + EventType.SUBAGENT_FINISHED, + ] + + +def test_error_closes_open_message_then_reports(session): + delegation_started(TID, "policy_researcher") + delegation_delta(TID, "- Pre") + delegation_error(TID, "specialist exploded") + + out = _drain(session) + 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" + + +def test_events_after_terminal_are_ignored(session): + delegation_started(TID, "policy_researcher") + delegation_finished(TID) + delegation_delta(TID, "late straggler") + delegation_finished(TID) + assert [ev.type for ev in _drain(session)] == [ + EventType.SUBAGENT_STARTED, + EventType.SUBAGENT_FINISHED, + ] + + +def test_empty_delta_is_dropped(session): + delegation_started(TID, "policy_researcher") + delegation_delta(TID, "") + delegation_finished(TID) + assert [ev.type for ev in _drain(session)] == [ + EventType.SUBAGENT_STARTED, + EventType.SUBAGENT_FINISHED, + ] + + +def test_none_tid_falls_back_to_generated_run_id(session): + delegation_started(None, "policy_researcher") + delegation_delta(None, "- x") + delegation_finished(None) + + out = _drain(session) + started = out[0] + assert started.parent_tool_call_id is None + assert started.subagent_run_id.startswith("sub-") + assert len(started.subagent_run_id) == len("sub-") + 8 + # All subsequent events carry the same generated run id. + assert {ev.subagent_run_id for ev in out} == {started.subagent_run_id} + assert out[1].message_id == f"{started.subagent_run_id}-m1" + + +def test_helpers_are_noops_without_a_session(): + # Unit tests / direct agent runs: no wrapper, no queue — nothing raises. + assert subagent_emitter._event_queue.get() is None + assert current_tool_call_id("research_policy") is None + delegation_started(TID, "policy_researcher") + delegation_delta(TID, "- x") + delegation_finished(TID) + delegation_error(TID, "boom") + + +# --------------------------------------------------------------------------- +# Wrapper-level tests: queue-merge around a fake inner bridge generator. +# --------------------------------------------------------------------------- + + +class _FakeInner: + """Duck-typed AgentFrameworkAgent carrying the attributes the wrapper + copies plus a scripted `run` generator.""" + + def __init__(self, gen_fn): + self.agent = object() + self.name = "fake" + self.description = "" + self.config = object() + self._approval_state_store = object() + self._gen_fn = gen_fn + + def run(self, input_data): + return self._gen_fn(input_data) + + +def _run_started(): + return RunStartedEvent(type=EventType.RUN_STARTED, thread_id="t", run_id="r") + + +def _run_finished(): + return RunFinishedEvent(type=EventType.RUN_FINISHED, thread_id="t", run_id="r") + + +async def _collect(agent) -> list: + return [ev async for ev in agent.run({"messages": []})] + + +async def test_wrapper_merges_tool_enqueued_events_mid_stream(): + async def inner(_input): + # Measured wire order: the bridge streams TOOL_CALL_START + ARGS + # before invoking the tool; TOOL_CALL_END arrives only AFTER the + # tool returns (docs/wire-capture-subagents.md, "After the emitter"). + yield _run_started() + yield ToolCallStartEvent( + type=EventType.TOOL_CALL_START, tool_call_id=TID, tool_call_name="research_policy" + ) + yield ToolCallArgsEvent( + type=EventType.TOOL_CALL_ARGS, tool_call_id=TID, delta='{"category":"travel","amount":900}' + ) + # The framework invokes the tool HERE, on the pump's driving chain, + # while the outer consumer is awaiting the queue. + tid = current_tool_call_id("research_policy") + assert tid == TID # recorded by the pump from TOOL_CALL_START + delegation_started(tid, "policy_researcher") + delegation_delta(tid, "- Pre-approval") + delegation_delta(tid, " required") + delegation_finished(tid) + yield ToolCallEndEvent(type=EventType.TOOL_CALL_END, tool_call_id=TID) + yield ToolCallResultEvent( + type=EventType.TOOL_CALL_RESULT, message_id="m", tool_call_id=TID, content="- Pre-approval required" + ) + yield _run_finished() + + out = await _collect(wrap_agent_run(_FakeInner(inner))) + assert [ev.type for ev in out] == [ + EventType.RUN_STARTED, + EventType.TOOL_CALL_START, + EventType.TOOL_CALL_ARGS, + # Child events land BETWEEN inner generator items — before the + # bridge's own TOOL_CALL_END/RESULT reach the client. + EventType.SUBAGENT_STARTED, + EventType.TEXT_MESSAGE_START, + EventType.TEXT_MESSAGE_CONTENT, + EventType.TEXT_MESSAGE_CONTENT, + EventType.TEXT_MESSAGE_END, + EventType.SUBAGENT_FINISHED, + EventType.TOOL_CALL_END, + EventType.TOOL_CALL_RESULT, + EventType.RUN_FINISHED, + ] + started = out[3] + assert started.subagent_run_id == RUN_ID + assert started.parent_tool_call_id == TID + assert [ev.delta for ev in out[5:7]] == ["- Pre-approval", " required"] + + +async def test_same_tool_double_call_gets_distinct_tids_and_delegations(): + # MAF runs multi-tool batches concurrently (asyncio.gather) and the + # bridge streams every TOOL_CALL_START before the tools execute — so + # two research_policy calls must each claim their OWN tid from the + # FIFO and drive their own delegation, never sharing one message. + tid2 = "call_secondResearchPolicyCall00" + + async def inner(_input): + yield ToolCallStartEvent( + type=EventType.TOOL_CALL_START, tool_call_id=TID, tool_call_name="research_policy" + ) + yield ToolCallStartEvent( + type=EventType.TOOL_CALL_START, tool_call_id=tid2, tool_call_name="research_policy" + ) + # Both tool bodies run concurrently; their deltas interleave. + tid_a = current_tool_call_id("research_policy") + tid_b = current_tool_call_id("research_policy") + assert (tid_a, tid_b) == (TID, tid2) # FIFO: oldest first + delegation_started(tid_a, "policy_researcher") + delegation_started(tid_b, "policy_researcher") + delegation_delta(tid_a, "- travel rules") + delegation_delta(tid_b, "- meal rules") + delegation_delta(tid_a, " apply") + delegation_finished(tid_b) + delegation_finished(tid_a) + yield ToolCallEndEvent(type=EventType.TOOL_CALL_END, tool_call_id=TID) + yield ToolCallEndEvent(type=EventType.TOOL_CALL_END, tool_call_id=tid2) + + out = await _collect(wrap_agent_run(_FakeInner(inner))) + + started = [ev for ev in out if ev.type == EventType.SUBAGENT_STARTED] + assert [ev.subagent_run_id for ev in started] == [f"{TID}-sub", f"{tid2}-sub"] + assert [ev.parent_tool_call_id for ev in started] == [TID, tid2] + + # Identity separation: every child event carries its own delegation's + # ids — interleaved ORDER between the two runs is fine. + for ev in out: + if ev.type == EventType.TEXT_MESSAGE_CONTENT and ev.delta.startswith("- travel"): + assert ev.message_id == f"{TID}-sub-m1" + if ev.type == EventType.TEXT_MESSAGE_CONTENT and ev.delta.startswith("- meal"): + assert ev.message_id == f"{tid2}-sub-m1" + a_events = [ev for ev in out if getattr(ev, "subagent_run_id", None) == f"{TID}-sub"] + b_events = [ev for ev in out if getattr(ev, "subagent_run_id", None) == f"{tid2}-sub"] + assert [ev.type for ev in a_events] == [ + EventType.SUBAGENT_STARTED, + EventType.TEXT_MESSAGE_START, + EventType.TEXT_MESSAGE_CONTENT, + EventType.TEXT_MESSAGE_CONTENT, + EventType.TEXT_MESSAGE_END, + EventType.SUBAGENT_FINISHED, + ] + assert [ev.type for ev in b_events] == [ + EventType.SUBAGENT_STARTED, + EventType.TEXT_MESSAGE_START, + EventType.TEXT_MESSAGE_CONTENT, + EventType.TEXT_MESSAGE_END, + EventType.SUBAGENT_FINISHED, + ] + assert {ev.message_id for ev in a_events if hasattr(ev, "message_id")} == {f"{TID}-sub-m1"} + assert {ev.message_id for ev in b_events if hasattr(ev, "message_id")} == {f"{tid2}-sub-m1"} + + +async def test_wrapper_error_path_emits_subagent_error_then_propagates(): + class _Boom(RuntimeError): + pass + + async def inner(_input): + yield _run_started() + yield ToolCallStartEvent( + type=EventType.TOOL_CALL_START, tool_call_id=TID, tool_call_name="research_policy" + ) + tid = current_tool_call_id("research_policy") + delegation_started(tid, "policy_researcher") + delegation_delta(tid, "- Pre") + delegation_error(tid, "specialist exploded") + raise _Boom("specialist exploded") + + agent = wrap_agent_run(_FakeInner(inner)) + out = [] + with pytest.raises(_Boom): + async for ev in agent.run({"messages": []}): + out.append(ev) + # Everything enqueued before the failure was delivered, ending in the + # SUBAGENT_ERROR (with the open child message closed first). + assert [ev.type for ev in out] == [ + EventType.RUN_STARTED, + EventType.TOOL_CALL_START, + EventType.SUBAGENT_STARTED, + EventType.TEXT_MESSAGE_START, + EventType.TEXT_MESSAGE_CONTENT, + EventType.TEXT_MESSAGE_END, + EventType.SUBAGENT_ERROR, + ] + assert out[-1].message == "specialist exploded" + + +async def test_wrapper_clean_shutdown_on_consumer_break(): + closed = asyncio.Event() + + async def inner(_input): + try: + yield _run_started() + while True: # endless stream: only a cancel/close ends it + await asyncio.sleep(0.01) + yield _run_started() + finally: + closed.set() + + agent = wrap_agent_run(_FakeInner(inner)) + gen = agent.run({"messages": []}) + first = await gen.__anext__() + assert first.type == EventType.RUN_STARTED + await gen.aclose() # consumer break / client disconnect + + await asyncio.wait_for(closed.wait(), timeout=1) + # No orphaned tasks: everything spawned by the wrapper is done. + pending = [ + t + for t in asyncio.all_tasks() + if t is not asyncio.current_task() and not t.done() + ] + assert pending == [] + # The ContextVar session was uninstalled. + assert subagent_emitter._event_queue.get() is None + + +async def test_wrapper_resets_contextvar_after_normal_completion(): + async def inner(_input): + yield _run_started() + yield _run_finished() + + out = await _collect(wrap_agent_run(_FakeInner(inner))) + assert [ev.type for ev in out] == [EventType.RUN_STARTED, EventType.RUN_FINISHED] + assert subagent_emitter._event_queue.get() is None + + +def test_wrap_agent_run_returns_agentframeworkagent_for_endpoint_dispatch(): + fake = _FakeInner(None) + wrapped = wrap_agent_run(fake) + assert isinstance(wrapped, SubagentEmittingAgent) + # The endpoint dispatches on isinstance(agent, AgentFrameworkAgent) and + # shares the config / approval-state store. + from agent_framework.ag_ui import AgentFrameworkAgent + + assert isinstance(wrapped, AgentFrameworkAgent) + assert wrapped.config is fake.config + assert wrapped._approval_state_store is fake._approval_state_store + + +def test_server_mounts_the_wrapped_agent(): + from src import server + + # The FastAPI endpoint consumes the wrapped run (protocol_runner is the + # SubagentEmittingAgent), so SUBAGENT_* events reach the wire. + assert isinstance(server.wrapped_agent, SubagentEmittingAgent) diff --git a/cockpit/runtimes/microsoft-agent-framework/python/uv.lock b/cockpit/runtimes/microsoft-agent-framework/python/uv.lock index d164f91fb..aafd57d70 100644 --- a/cockpit/runtimes/microsoft-agent-framework/python/uv.lock +++ b/cockpit/runtimes/microsoft-agent-framework/python/uv.lock @@ -121,6 +121,12 @@ dependencies = [ { name = "uvicorn", extra = ["standard"] }, ] +[package.dev-dependencies] +dev = [ + { name = "pytest" }, + { name = "pytest-asyncio" }, +] + [package.metadata] requires-dist = [ { name = "agent-framework-ag-ui", specifier = ">=1.2.1" }, @@ -130,6 +136,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" @@ -246,6 +258,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" @@ -385,6 +406,24 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/ca/6f/a04e900f465ff3221ccc395522503e2d10e79fa21f2723c8e177aae1e0d1/opentelemetry_api-1.44.0-py3-none-any.whl", hash = "sha256:94b98c893a91b88657eaac1e3ba89618cdb85be6918196705354f34728b2cdef", size = 60018, upload-time = "2026-07-16T15:25:11.657Z" }, ] +[[package]] +name = "packaging" +version = "26.3" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/7d/fa/3944b40b07da9ce895c0e6303a5ab7d53da063554f534556b134a54d6093/packaging-26.3.tar.gz", hash = "sha256:94edc256424af38762eb31306eed28beb9f0efc50a8837492c9d6fd6004aed79", size = 313412, upload-time = "2026-08-04T18:15:28.737Z" } +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 = "pydantic" version = "2.13.5" @@ -475,6 +514,44 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/fa/04/c81d4841331c2178b6fb09ae225425e110ed72d990c9fe556c4ec03d1013/pydantic_core-2.46.5-graalpy312-graalpy250_312_native-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:8e24d8f05fa2d28513d94e877e9c75ad66175376209b3977f916e240e623193c", size = 2111034, upload-time = "2026-08-28T10:01:07.345Z" }, ] +[[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 = "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-dotenv" version = "1.2.3" diff --git a/deployments/ag-ui-dev/deps/microsoft_agent_framework/docs/wire-capture-subagents.md b/deployments/ag-ui-dev/deps/microsoft_agent_framework/docs/wire-capture-subagents.md new file mode 100644 index 000000000..93207619b --- /dev/null +++ b/deployments/ag-ui-dev/deps/microsoft_agent_framework/docs/wire-capture-subagents.md @@ -0,0 +1,269 @@ +# MAF delegation wire capture — subagent pattern decision (Task 0 spike) + +Date: 2026-09-02. Live capture against `src/server.py` (uvicorn, port 5330) with the +plain-OpenAI client path (`build_chat_client`, `gpt-4o-mini`). The scratch delegation +code described below was NOT committed; only this document is. + +Installed bridge inspected end to end: `agent-framework-ag-ui` 1.2.x in +`.venv/lib/python3.14/site-packages/agent_framework_ag_ui/` (all venv line numbers below +refer to that tree). + +## Verdict: Candidate A (agents-as-tools), emitter injects via a run-wrapper merge queue + +- **Candidate A works and is observable.** A specialist `Agent` invoked from an async + function tool streams its updates INTO the tool body in real time (101 streamed + updates observed in-tool for a ~550-char answer). That is everything the emitter + needs to synthesize the SUBAGENT_* sequence with streaming child deltas. +- **Candidate B (two-executor workflow) is not needed.** The endpoint does mount + workflows (`_endpoint.py:137-138` wraps a `Workflow` in `AgentFrameworkWorkflow`), + so B remains a fallback, but A is simpler and keeps the demo's existing + approval/predictive-state surfaces untouched. + +## Seam analysis (venv file:line) + +### (a) Where MAF run events become AG-UI events + +- Single entry point: `run_agent_stream` (`agent_framework_ag_ui/_agent_run.py:2259`), + reached from `AgentFrameworkAgent.run` (`_agent.py:147-166`). +- The wrapped agent is invoked at `_agent_run.py:2723` + (`response_stream = (a2ui_runner or agent).run(messages, stream=True, **run_kwargs)`); + updates are pulled at `_agent_run.py:2726` and each content item is converted to + AG-UI events by `_emit_content` (`_run_common.py:1166`, dispatched from + `_agent_run.py:2828`). `_emit_content` handles `text`, `function_call`, + `function_result`, `function_approval_request`, `usage`, reasoning, and MCP content + types (`_run_common.py:1174-1200`); anything else is dropped with a debug log + (`_run_common.py:1200`). +- The FastAPI endpoint consumes `protocol_runner.run(input_data)` and encodes each + yielded event generically (`_endpoint.py:212-242`). + +### (b) Can a function tool reach an event emitter/queue/context? + +**No.** There is no ContextVar, queue, writer, or middleware hook anywhere in +`agent_framework_ag_ui/*.py` or in `agent_framework/_tools.py` / `_middleware.py` / +`_agents.py` that a tool body could use to inject AG-UI events +(`grep -rn ContextVar` over those modules returns nothing). The event pipeline is a +pure pull-driven async generator; tools execute deep inside the framework's function +invocation loop within `agent.run(stream=True)` and only their return value surfaces +(as `function_result` content → `TOOL_CALL_RESULT`). + +**Injection seam (named):** wrap `AgentFrameworkAgent.run` — the exact method the +endpoint calls at `_endpoint.py:212`. Our emitter will be a small subclass (or +compositional wrapper) in the demo: + +1. `run()` creates an `asyncio.Queue` and sets a module-level `ContextVar` to it + before delegating to the inner `run_agent_stream` generator. Because the tool body + executes on the same async call chain (endpoint → wrapper → `run_agent_stream` → + `agent.run` → function invocation), the ContextVar value propagates into the tool. +2. The wrapper pumps the inner generator into the same queue from an + `asyncio.create_task` and yields from the merged queue. This is required for LIVE + interleaving: while the tool runs, the bridge generator is suspended awaiting the + next provider update, so a naive "drain queue between inner yields" design would + batch all child deltas until the tool returns. With the pump-task merge, a + `queue.put_nowait` from the tool body wakes the outer consumer immediately. +3. The tool body reads the ContextVar and enqueues + `SubagentStartedEvent {subagentRunId: -sub, name: "policy_researcher", + parentToolCallId: }` → attributed `TextMessageStart/Content×N/End` + (one delta per specialist update) → `SubagentFinishedEvent success` + (`SubagentErrorEvent` on exception). The tool's own `toolCallId` is available to + the body via the framework's function-call content on the update stream; the + emitter wrapper can also correlate it by observing the preceding + `TOOL_CALL_START` for the delegation tool on the bridge stream. + +This is the same "emit from inside the tool body" shape the Strands PR proved, with +the writer supplied by our own wrapper instead of the runtime (MAF's bridge provides +none). Reference translator: `cockpit/ag-ui/subagents/python/src/streaming/activity_emitting_agent.py`. + +### (c) Does the encoder accept ag_ui.core pydantic events generally? + +**Yes.** `EventEncoder.encode` takes any `BaseEvent` and does a generic +model-dump → SSE `data:` frame (`ag_ui/encoder/encoder.py`, `encode`/`_encode_sse`); +the endpoint applies it uniformly with no per-type allowlist (`_endpoint.py:224`). +`Subagent*` events are `BaseEvent` subclasses, so they pass through untouched. + +SDK check (in this venv): + +``` +$ 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'] +``` + +### (d) What does the bridge do with nested-agent activity inside a tool? + +**Nothing is observable.** The specialist's `run(stream=True)` updates are consumed +entirely inside the tool body; the bridge sees only the tool's `function_call` +(streamed as `TOOL_CALL_START/ARGS/END`) and its string return value +(`TOOL_CALL_RESULT`). No ACTIVITY_*, no per-child events, no specialist name on the +wire beyond the delegation tool's own name. This matches the "measured red upstream" +note in `src/agent.py` and is confirmed by the capture below. + +## Scratch setup (uncommitted, reverted after capture) + +Added to `src/agent.py`: a `policy_researcher` `Agent` (same `build_chat_client()`, +instructions: expense-policy researcher, 3 short bullets) plus an async +`@tool research_policy(category: str, amount: float) -> str` that ran +`specialist.run(prompt, stream=True)`, accumulated `update.text`, logged each update +to stderr, and returned the joined text; registered on the primary agent with one +instruction sentence about delegating policy research. + +## In-tool streaming datum + +The specialist's deltas DID stream into the tool body, token by token: + +``` +[spike] specialist update #2: '-' +[spike] specialist update #3: ' **' +[spike] specialist update #4: 'Pre' +... +[spike] specialist DONE: 101 streamed updates, 548 chars (attempt 2; attempt 1: 106 updates, 582 chars) +``` + +So the emitter can produce **streaming child deltas** (preferred contract), not just a +single final chunk. + +## Live wire capture (attempt 2 of 2; attempt 1 also delegated but was truncated client-side) + +Request: POST `/agent` with `threadId: spike-thread-2`, `runId: spike-run-2`, single +user message "Should I submit a $900 conference travel expense? Research the policy +first". The model delegated on the first turn in both attempts. Full event-type +census of the complete stream: + +``` +1 RUN_STARTED 1 STATE_SNAPSHOT 3 CUSTOM (PredictState, usage×2) +2 TEXT_MESSAGE_START 35 TEXT_MESSAGE_CONTENT 2 TEXT_MESSAGE_END +1 TOOL_CALL_START 9 TOOL_CALL_ARGS 1 TOOL_CALL_END 1 TOOL_CALL_RESULT +1 MESSAGES_SNAPSHOT 1 RUN_FINISHED +0 SUBAGENT_* / ACTIVITY_* / anything child-related +``` + +Abridged stream (ids as captured; no secrets present): + +``` +data: {"type":"RUN_STARTED","threadId":"spike-thread-2","runId":"spike-run-2"} +data: {"type":"CUSTOM","name":"PredictState","value":[{"state_key":"expense","tool":"submit_expense","tool_argument":"expense"}]} +data: {"type":"STATE_SNAPSHOT","snapshot":{"expense":{}}} +data: {"type":"TEXT_MESSAGE_START","messageId":"5cff954e-...","role":"assistant"} +data: {"type":"TOOL_CALL_START","toolCallId":"call_7sxPY1sC236nPyHRTWAZMJB9","toolCallName":"research_policy","parentMessageId":"5cff954e-..."} +data: {"type":"TOOL_CALL_ARGS","toolCallId":"call_7sxP...","delta":"{\""} +... (9 ARGS deltas spelling {"category":"travel","amount":900}) +data: {"type":"TOOL_CALL_END","toolCallId":"call_7sxP..."} + <-- specialist runs HERE; 101 updates streamed in-tool; NOTHING on the wire --> +data: {"type":"TOOL_CALL_RESULT","messageId":"c2c72fd2-...","toolCallId":"call_7sxP...","content":"1. **Pre-Approval Required**: Travel expenses exceeding $500 ...","role":"tool"} +data: {"type":"TEXT_MESSAGE_END","messageId":"5cff954e-..."} +data: {"type":"TEXT_MESSAGE_START","messageId":"534fe3b3-...","role":"assistant"} +data: {"type":"TEXT_MESSAGE_CONTENT","messageId":"534fe3b3-...","delta":"The"} +... (35 deltas: policy summary + ask to confirm approval) +data: {"type":"TEXT_MESSAGE_END","messageId":"534fe3b3-..."} +data: {"type":"MESSAGES_SNAPSHOT","messages":[...user, assistant toolCalls(research_policy), tool result, assistant text...]} +data: {"type":"RUN_FINISHED","threadId":"spike-thread-2","runId":"spike-run-2"} +``` + +### Explicit statements + +- **Native delegation on the wire:** an ordinary function tool call — + `TOOL_CALL_START(research_policy)` → streamed `TOOL_CALL_ARGS` → `TOOL_CALL_END` → + a single `TOOL_CALL_RESULT` carrying the specialist's complete final text. The + wall-clock gap between `TOOL_CALL_END` and `TOOL_CALL_RESULT` is where the + specialist runs, silently. +- **Child updates streamed in-tool:** YES — 101 streamed updates (attempt 2; 106 in + attempt 1), token-granular. +- **Anything child-related on the wire:** NO — zero events; the specialist is + invisible except as the tool's result string. + +## Emitter plan (for the implementation PR) + +Target sequence, injected by the wrapper-queue seam around the existing bridge stream +for tool call id ``: + +`SUBAGENT_STARTED {subagentRunId: "-sub", name: "policy_researcher", parentToolCallId: ""}` +→ `TEXT_MESSAGE_START/CONTENT×N/END` attributed to the subagent run (one CONTENT per +specialist update; live-interleaved via the pump-task merge) → `SUBAGENT_FINISHED +{outcome: success}` (or `SUBAGENT_ERROR` on tool-body exception), all before the +bridge's own `TOOL_CALL_RESULT` for `` reaches the client. + +## After the emitter + +Date: 2026-09-02, post-implementation. Live capture against the committed +`src/server.py` (uvicorn, port 5330; plain-OpenAI path, `gpt-4o-mini`), request +identical in shape to the spike: single user message "Should I submit a $900 +conference travel expense? Research the policy first" (`threadId: +smoke-thread-1`, `runId: smoke-run-1`). The model called +`lookup_expense_policy` first and then delegated via `research_policy` on the +same turn. Full event-type census of the complete stream: + +``` +1 RUN_STARTED 1 STATE_SNAPSHOT 4 CUSTOM (PredictState, usage×3) +4 TEXT_MESSAGE_START 128 TEXT_MESSAGE_CONTENT 4 TEXT_MESSAGE_END +2 TOOL_CALL_START 14 TOOL_CALL_ARGS 2 TOOL_CALL_END 2 TOOL_CALL_RESULT +1 SUBAGENT_STARTED 1 SUBAGENT_FINISHED 0 SUBAGENT_ERROR +1 MESSAGES_SNAPSHOT 1 RUN_FINISHED +``` + +**Child delta count: 100 streamed TEXT_MESSAGE_CONTENT events** attributed to +the subagent run (`messageId: -sub-m1`, token-granular — same order of +magnitude as the spike's 101/106 in-tool updates), live-interleaved between the +bridge's own events: the bridge yields `TOOL_CALL_END` for `research_policy` +only after the tool returns, and the entire SUBAGENT_* sequence lands between +`TOOL_CALL_ARGS` and that `TOOL_CALL_END` — proof the pump-task merge queue +delivered the deltas while the bridge generator was suspended inside the tool. + +Abridged stream around the delegation (ids as captured; no secrets present): + +``` +data: {"type":"TOOL_CALL_START","toolCallId":"call_drwxfbPJnrzSBBLnrcqehqfx","toolCallName":"research_policy","parentMessageId":"c1a737b4-..."} +... (11 ARGS deltas spelling {"category":"travel","amount":900}) +data: {"type":"SUBAGENT_STARTED","subagentRunId":"call_drwxfbPJnrzSBBLnrcqehqfx-sub","name":"policy_researcher","parentToolCallId":"call_drwxfbPJnrzSBBLnrcqehqfx"} +data: {"type":"TEXT_MESSAGE_START","messageId":"call_drwxfbPJnrzSBBLnrcqehqfx-sub-m1","role":"assistant","subagentRunId":"call_drwxfbPJnrzSBBLnrcqehqfx-sub"} +data: {"type":"TEXT_MESSAGE_CONTENT","messageId":"call_drwxfbPJnrzSBBLnrcqehqfx-sub-m1","delta":"-","subagentRunId":"call_drwxfbPJnrzSBBLnrcqehqfx-sub"} +data: {"type":"TEXT_MESSAGE_CONTENT","messageId":"call_drwxfbPJnrzSBBLnrcqehqfx-sub-m1","delta":" **","subagentRunId":"call_drwxfbPJnrzSBBLnrcqehqfx-sub"} +data: {"type":"TEXT_MESSAGE_CONTENT","messageId":"call_drwxfbPJnrzSBBLnrcqehqfx-sub-m1","delta":"Pre","subagentRunId":"call_drwxfbPJnrzSBBLnrcqehqfx-sub"} +... (100 CONTENT deltas total, token-granular) +data: {"type":"TEXT_MESSAGE_END","messageId":"call_drwxfbPJnrzSBBLnrcqehqfx-sub-m1","subagentRunId":"call_drwxfbPJnrzSBBLnrcqehqfx-sub"} +data: {"type":"SUBAGENT_FINISHED","subagentRunId":"call_drwxfbPJnrzSBBLnrcqehqfx-sub","outcome":{"type":"success"}} +data: {"type":"TOOL_CALL_END","toolCallId":"call_drwxfbPJnrzSBBLnrcqehqfx"} +data: {"type":"TOOL_CALL_RESULT","messageId":"7dec6623-...","toolCallId":"call_drwxfbPJnrzSBBLnrcqehqfx","content":"- **Pre-Approval Required**: Travel expenses exceeding $500 ...","role":"tool"} +``` + +Tid derivation on the live run: the pump-recorded `TOOL_CALL_START` path +(`current_tool_call_id("research_policy")`) — the real wire toolCallId keys +the whole sequence (`subagentRunId = -sub`, `parentToolCallId = `); +the generated `sub-` fallback was not needed. Existing surfaces are +untouched: PredictState CUSTOM, STATE_SNAPSHOT, both TOOL_CALL_RESULTs, and +MESSAGES_SNAPSHOT all present as before. + +## Browser verification + +2026-09-02, live backend (real `OPENAI_API_KEY`, uvicorn on :5330) + `nx +serve cockpit-runtimes-microsoft-agent-framework-angular` on :4330, driven +headlessly with Playwright. Screenshot: +`cockpit/runtimes/microsoft-agent-framework/angular/e2e/manual/subagent-card-live.png`. + +What rendered: the delegation prompt (*"Should I submit a $900 conference +travel expense? Research the policy first"*) produced an inline +`` anchored to the `research_policy` tool call — header +`policy_researcher` + wire `toolCallId` + status badge — with the +specialist's 3-bullet policy transcript inside it, preceded by the +orchestrator's own `lookup_expense_policy` chip and followed by its 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 card mounting at 66 chars (header only) the moment +SUBAGENT_STARTED landed, then the specialist's message growing +monotonically while the run was live — one run sampled 66 → 163 → 277 → +398 → 422 → 553 → 622 chars between t≈1.5s and t≈3.0s; a second run +sampled 66 → 105 → 207 → 314 → 430 → 519 → 614 chars — before the badge +flipped to `complete` and the card collapsed to its 67-char summary row. +This confirms the queue-merge emitter's attributed `TEXT_MESSAGE_CONTENT` +deltas render progressively in the card during the run, not as one +post-hoc paste. + +Same `libs/chat` anchoring dependency as the Strands verification: the wire +and the `@threadplane/ag-ui` reducer were correct, but `chat-tool-calls` +anchored subagent cards by the adapter's `subagents()` map key (for native +SUBAGENT_* that is the `subagentRunId`, `-sub`) instead of the +contract field `Subagent.toolCallId`, so the card never mounted. The fix +(re-index on `Subagent.toolCallId`) plus its pinning spec are cherry-picked +onto this branch. diff --git a/deployments/ag-ui-dev/deps/microsoft_agent_framework/pyproject.toml b/deployments/ag-ui-dev/deps/microsoft_agent_framework/pyproject.toml index 8f718a771..dbf6404e4 100644 --- a/deployments/ag-ui-dev/deps/microsoft_agent_framework/pyproject.toml +++ b/deployments/ag-ui-dev/deps/microsoft_agent_framework/pyproject.toml @@ -10,9 +10,18 @@ dependencies = [ "uvicorn[standard]>=0.29", ] +[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/microsoft_agent_framework/src/agent.py b/deployments/ag-ui-dev/deps/microsoft_agent_framework/src/agent.py index 9414d37f9..4d28e8c45 100644 --- a/deployments/ag-ui-dev/deps/microsoft_agent_framework/src/agent.py +++ b/deployments/ag-ui-dev/deps/microsoft_agent_framework/src/agent.py @@ -14,8 +14,13 @@ protocol-standard ``RUN_FINISHED.outcome = {type: 'interrupt', ...}`` and resumes from the client's top-level ``resume`` entries. -No subagents surface: the MAF bridge emits no per-subagent -ACTIVITY_SNAPSHOT/ACTIVITY_DELTA stream (measured red upstream). +- subagents: ``research_policy`` delegates to the tool-less + ``policy_researcher`` specialist Agent and streams its deltas through the + ``delegation_*`` helpers in src/subagent_emitter.py, which merge standard + ``SUBAGENT_*`` + attributed child ``TEXT_MESSAGE_*`` events into the run + stream at the run-wrapper seam (the MAF bridge natively emits NOTHING for + nested-agent activity inside a tool — measured red upstream and in + docs/wire-capture-subagents.md). Model client: Azure OpenAI is the DEFAULT path — when ``AZURE_OPENAI_ENDPOINT`` is set the client routes to Azure (key auth via @@ -32,6 +37,8 @@ from agent_framework.openai import OpenAIChatCompletionClient from pydantic import BaseModel, Field +from . import subagent_emitter + _POLICIES = { "meals": {"limit_usd": 300, "receipt_required_over_usd": 25, "notes": "Team meals require attendee count in the memo."}, "travel": {"limit_usd": 1500, "receipt_required_over_usd": 0, "notes": "Book through the travel portal when possible."}, @@ -97,6 +104,10 @@ def submit_expense(expense: Expense) -> str: _INSTRUCTIONS = """You are an expense approval copilot. +Before recommending whether to submit an expense, delegate the policy +research to the specialist by calling `research_policy` with the category +and amount. + When the user asks to file an expense: 1. FIRST call `lookup_expense_policy` with the expense category. 2. THEN call `submit_expense` with the complete structured expense @@ -138,12 +149,63 @@ def build_chat_client() -> OpenAIChatCompletionClient: ) +policy_researcher = Agent( + name="policy_researcher", + instructions=( + "You are an expense-policy researcher. Given an expense category and " + "amount, summarize the applicable policy rules in 3 short bullets." + ), + client=build_chat_client(), +) + + +@tool( + name="research_policy", + description="Delegate policy research for this expense to a specialist.", +) +async def research_policy(category: str, amount: float) -> str: + """Delegate policy research for this expense to a specialist. + + Streams the ``policy_researcher`` specialist and mirrors each text delta + onto the AG-UI wire as attributed SUBAGENT_* / TEXT_MESSAGE_* events via + src/subagent_emitter.py (no-ops outside the wrapped run). + + Args: + category: Expense category, e.g. 'meals' or 'travel'. + amount: Expense amount in USD. + + Returns: + The specialist's complete policy summary. + """ + # Deterministically recorded by the run wrapper's pump before this body + # runs (the bridge streams TOOL_CALL_START/ARGS/END first); None when + # invoked outside a wrapped run. + tid = subagent_emitter.current_tool_call_id("research_policy") + subagent_emitter.delegation_started(tid, policy_researcher.name) + parts: list[str] = [] + try: + prompt = ( + f"Expense category: {category}. Amount: ${amount:.2f}. " + "Summarize the applicable policy rules." + ) + async for update in policy_researcher.run(prompt, stream=True): + text = update.text + if text: + parts.append(text) + subagent_emitter.delegation_delta(tid, text) + except Exception as exc: + subagent_emitter.delegation_error(tid, str(exc)) + raise + subagent_emitter.delegation_finished(tid) + return "".join(parts) + + agent = AgentFrameworkAgent( agent=Agent( name="expense_approval_copilot", instructions=_INSTRUCTIONS, client=build_chat_client(), - tools=[lookup_expense_policy, submit_expense], + tools=[lookup_expense_policy, research_policy, submit_expense], ), name="ExpenseApprovalCopilot", description="Files expense reports with policy lookup, shared state, and human approval.", diff --git a/deployments/ag-ui-dev/deps/microsoft_agent_framework/src/server.py b/deployments/ag-ui-dev/deps/microsoft_agent_framework/src/server.py index 7b192ca0b..e26a54f02 100644 --- a/deployments/ag-ui-dev/deps/microsoft_agent_framework/src/server.py +++ b/deployments/ag-ui-dev/deps/microsoft_agent_framework/src/server.py @@ -3,9 +3,14 @@ from agent_framework_ag_ui import add_agent_framework_fastapi_endpoint from .agent import agent +from .subagent_emitter import wrap_agent_run app = FastAPI(title="cockpit-runtimes-microsoft-agent-framework") -add_agent_framework_fastapi_endpoint(app, agent, path="/agent") +# The wrapper is the SUBAGENT_* injection seam: the endpoint consumes +# protocol_runner.run, and wrap_agent_run merges the delegation tool's +# enqueued child events into that stream (src/subagent_emitter.py). +wrapped_agent = wrap_agent_run(agent) +add_agent_framework_fastapi_endpoint(app, wrapped_agent, path="/agent") @app.get("/ok") diff --git a/deployments/ag-ui-dev/deps/microsoft_agent_framework/src/subagent_emitter.py b/deployments/ag-ui-dev/deps/microsoft_agent_framework/src/subagent_emitter.py new file mode 100644 index 000000000..0b231ab4c --- /dev/null +++ b/deployments/ag-ui-dev/deps/microsoft_agent_framework/src/subagent_emitter.py @@ -0,0 +1,309 @@ +# SPDX-License-Identifier: MIT +"""SUBAGENT_* emitter for the `research_policy` delegation tool. + +The MAF AG-UI bridge is a pure pull-driven async generator with no writer a +tool body could reach: the specialist's streamed updates are consumed +entirely inside the tool and only the return string surfaces (as +TOOL_CALL_RESULT). Measured in docs/wire-capture-subagents.md, which also +names the injection seam implemented here: wrap ``AgentFrameworkAgent.run`` +— the exact method the FastAPI endpoint consumes (`_endpoint.py:212`) — +with a queue-merge generator. + +How the seam works (``SubagentEmittingAgent`` / ``wrap_agent_run``): + +1. ``run()`` creates an ``asyncio.Queue`` and publishes it (plus a small + correlation map) through a module-level ``ContextVar``. The delegation + tool executes on the same async call chain, so the value propagates + into the tool body. +2. A pump task drains the inner bridge generator into that queue. This is + required for LIVE interleaving: while the tool runs, the bridge + generator is suspended awaiting the next provider update, so a naive + "drain between inner yields" design would batch every child delta until + the tool returned. With the pump-task merge, a ``put_nowait`` from the + tool body wakes the outer consumer immediately. +3. The tool body calls the ``delegation_*`` helpers below, which build the + typed ``ag_ui.core`` events and enqueue them: + + SUBAGENT_STARTED {subagentRunId: -sub, parentToolCallId: } + TEXT_MESSAGE_START/CONTENT.../END (streamed specialist deltas) + SUBAGENT_FINISHED outcome=success (or SUBAGENT_ERROR on failure) + +Correlation: the pump appends every TOOL_CALL_START's ``toolCallId`` to a +per-tool-name FIFO as it passes through the queue, and each tool body pops +the oldest via ``current_tool_call_id`` — so a multi-tool batch calling the +same tool twice (MAF runs batches concurrently via ``asyncio.gather``, and +the bridge streams all TOOL_CALL_STARTs first) gives each invocation its +own tid and its own delegation. The bridge yields TOOL_CALL_START (and the +ARGS deltas) for the delegation call BEFORE the framework invokes the tool +on the same driving chain, so the FIFO is deterministically populated by +the time the tool body runs; TOOL_CALL_END arrives only AFTER the tool +returns (measured wire order — the SUBAGENT_* sequence lands between ARGS +and END), which is why correlation relies on START alone. If a caller ever +invokes the tool outside +the wrapped run, the helpers fall back to a generated ``sub-`` run id +with ``parentToolCallId`` omitted — and with no queue at all they are pure +no-ops, which is what keeps unit tests and direct agent runs side-effect +free. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import uuid +from contextvars import ContextVar +from dataclasses import dataclass, field +from typing import Any, AsyncGenerator + +from ag_ui.core import ( + BaseEvent, + EventType, + SubagentErrorEvent, + SubagentFinishedEvent, + SubagentFinishedSuccessOutcome, + SubagentStartedEvent, + TextMessageContentEvent, + TextMessageEndEvent, + TextMessageStartEvent, +) +from agent_framework.ag_ui import AgentFrameworkAgent + +DELEGATION_TOOL_NAME = "research_policy" + + +@dataclass +class _Delegation: + """Lifecycle of one delegation call, keyed by parent tool-call id.""" + + run_id: str + message_id: str + message_open: bool = False + finished: bool = False + + +@dataclass +class _EmitterSession: + """Per-run channel shared between the run wrapper and the tool body.""" + + queue: asyncio.Queue[Any] + # Per-tool-name FIFO of not-yet-claimed TOOL_CALL_START toolCallIds: + # the pump appends, each tool body pops the oldest — one tid per call + # even when a batch invokes the same tool twice. + tool_call_ids: dict[str, list[str]] = field(default_factory=dict) + runs: dict[str | None, _Delegation] = field(default_factory=dict) + + +_event_queue: ContextVar[_EmitterSession | None] = ContextVar( + "maf_subagent_emitter_session", default=None +) + + +def current_tool_call_id(tool_name: str) -> str | None: + """Claim the oldest unclaimed TOOL_CALL_START toolCallId for a tool. + + Pops from the per-name FIFO the pump fills, so each concurrent + invocation of the same tool gets its own tid. Deterministically + populated before the tool body runs (see module docstring); ``None`` + outside a wrapped run. + """ + session = _event_queue.get() + if session is None: + return None + pending = session.tool_call_ids.get(tool_name) + if not pending: + return None + return pending.pop(0) + + +def emit(event: BaseEvent) -> None: + """Enqueue one AG-UI event onto the live run stream; no-op unwrapped.""" + session = _event_queue.get() + if session is not None: + session.queue.put_nowait(event) + + +def delegation_started(tid: str | None, name: str) -> None: + """Announce the specialist run. Ids derive from the delegation tool-call + id; without one (unwrapped fallback) a ``sub-`` run id is generated + and ``parentToolCallId`` omitted.""" + session = _event_queue.get() + run_id = f"{tid}-sub" if tid else f"sub-{uuid.uuid4().hex[:8]}" + if session is not None: + session.runs[tid] = _Delegation(run_id=run_id, message_id=f"{run_id}-m1") + emit( + SubagentStartedEvent( + type=EventType.SUBAGENT_STARTED, + subagent_run_id=run_id, + name=name, + parent_tool_call_id=tid, + ) + ) + + +def _active(tid: str | None) -> _Delegation | None: + session = _event_queue.get() + if session is None: + return None + delegation = session.runs.get(tid) + if delegation is None or delegation.finished: + return None + return delegation + + +def delegation_delta(tid: str | None, text: str) -> None: + """Stream one specialist text delta, lazily opening the attributed + message (so a zero-delta run emits no empty message).""" + delegation = _active(tid) + if delegation is None or not text: + return + if not delegation.message_open: + delegation.message_open = True + emit( + TextMessageStartEvent( + type=EventType.TEXT_MESSAGE_START, + message_id=delegation.message_id, + role="assistant", + subagent_run_id=delegation.run_id, + ) + ) + emit( + TextMessageContentEvent( + type=EventType.TEXT_MESSAGE_CONTENT, + message_id=delegation.message_id, + delta=text, + subagent_run_id=delegation.run_id, + ) + ) + + +def _close_message(delegation: _Delegation) -> None: + if delegation.message_open: + delegation.message_open = False + emit( + TextMessageEndEvent( + type=EventType.TEXT_MESSAGE_END, + message_id=delegation.message_id, + subagent_run_id=delegation.run_id, + ) + ) + + +def delegation_finished(tid: str | None) -> None: + """Close the open child message and finish the subagent with success.""" + delegation = _active(tid) + if delegation is None: + return + delegation.finished = True + _close_message(delegation) + emit( + SubagentFinishedEvent( + type=EventType.SUBAGENT_FINISHED, + subagent_run_id=delegation.run_id, + outcome=SubagentFinishedSuccessOutcome(), + ) + ) + + +def delegation_error(tid: str | None, message: str) -> None: + """Close the open child message and report the specialist failure. The + tool re-raises afterwards, so the bridge's own tool-error path still + runs normally.""" + delegation = _active(tid) + if delegation is None: + return + delegation.finished = True + _close_message(delegation) + emit( + SubagentErrorEvent( + type=EventType.SUBAGENT_ERROR, + subagent_run_id=delegation.run_id, + message=message, + ) + ) + + +class _PumpFailure: + """Sentinel carrying an inner-generator exception across the queue.""" + + def __init__(self, exc: BaseException) -> None: + self.exc = exc + + +_DONE = object() + + +def _record_tool_call(session: _EmitterSession, event: Any) -> None: + if getattr(event, "type", None) == EventType.TOOL_CALL_START: + session.tool_call_ids.setdefault(event.tool_call_name, []).append( + event.tool_call_id + ) + + +class SubagentEmittingAgent(AgentFrameworkAgent): + """AgentFrameworkAgent whose ``run`` merges tool-enqueued SUBAGENT_* + events into the bridge stream via the pump-task queue. + + Constructed from an already-configured ``AgentFrameworkAgent`` (shares + its config and approval-state store rather than re-running ``__init__``), + so the endpoint's ``isinstance(agent, AgentFrameworkAgent)`` dispatch + and approval resume flow are untouched. + """ + + def __init__(self, inner: AgentFrameworkAgent) -> None: + self._inner = inner + self.agent = inner.agent + self.name = inner.name + self.description = inner.description + self.config = inner.config + self._approval_state_store = inner._approval_state_store + + async def run( + self, input_data: dict[str, Any] + ) -> AsyncGenerator[BaseEvent, None]: + queue: asyncio.Queue[Any] = asyncio.Queue() + session = _EmitterSession(queue=queue) + token = _event_queue.set(session) + inner_gen = self._inner.run(input_data) + + async def _pump() -> None: + try: + async for event in inner_gen: + _record_tool_call(session, event) + queue.put_nowait(event) + except asyncio.CancelledError: + raise + except BaseException as exc: # propagate to the consumer, never swallow + queue.put_nowait(_PumpFailure(exc)) + else: + queue.put_nowait(_DONE) + + # create_task copies the current context AFTER the ContextVar set, + # so the tool body (which executes on the pump's driving chain) + # sees this session. + pump = asyncio.create_task(_pump()) + try: + while True: + item = await queue.get() + if item is _DONE: + break + if isinstance(item, _PumpFailure): + raise item.exc + yield item + finally: + # Consumer break / client disconnect (GeneratorExit) or pump + # failure: cancel and await the pump so no task is orphaned, + # then close the inner generator. + pump.cancel() + with contextlib.suppress(asyncio.CancelledError): + await pump + with contextlib.suppress(Exception): + await inner_gen.aclose() + # reset raises ValueError if a GC-driven aclose runs the finally + # in a different context than the one that set the var. + with contextlib.suppress(ValueError): + _event_queue.reset(token) + + +def wrap_agent_run(agent: AgentFrameworkAgent) -> SubagentEmittingAgent: + """Wrap an AgentFrameworkAgent so its run stream carries SUBAGENT_*.""" + return SubagentEmittingAgent(agent) diff --git a/deployments/ag-ui-dev/deps/microsoft_agent_framework/tests/test_delegation.py b/deployments/ag-ui-dev/deps/microsoft_agent_framework/tests/test_delegation.py new file mode 100644 index 000000000..16e465e26 --- /dev/null +++ b/deployments/ag-ui-dev/deps/microsoft_agent_framework/tests/test_delegation.py @@ -0,0 +1,58 @@ +# SPDX-License-Identifier: MIT +"""Tests for the `research_policy` delegation scenario — the tool is a +registered async `@tool` that hands expense-policy research to the tool-less +`policy_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, policy_researcher, research_policy + + +def _tool_names() -> list[str]: + return [t.name for t in agent.agent.default_options["tools"]] + + +def test_research_policy_is_registered_on_the_agent(): + assert "research_policy" in _tool_names() + # Existing tools stay registered untouched. + assert "lookup_expense_policy" in _tool_names() + assert "submit_expense" in _tool_names() + + +def test_tool_name_and_docstring(): + assert research_policy.name == "research_policy" + assert research_policy.description.startswith( + "Delegate policy research for this expense to a specialist." + ) + schema = research_policy.parameters() + assert set(schema["required"]) == {"category", "amount"} + + +def test_tool_is_async(): + # The seam depends on it: the tool must be able to async-iterate the + # specialist's streamed updates and enqueue deltas as they arrive. + assert inspect.iscoroutinefunction(research_policy.func) + + +def test_specialist_is_toolless_researcher(): + assert policy_researcher.name == "policy_researcher" + assert policy_researcher.default_options.get("tools") == [] + assert "expense-policy researcher" in policy_researcher.default_options["instructions"] + + +def test_instructions_mention_delegation(): + instructions = agent.agent.default_options["instructions"] + assert "research_policy" in instructions + + +def test_untouched_surfaces_still_configured(): + # The subagent scenario must not disturb the existing shared-state and + # approval surfaces. + assert agent.config.state_schema == { + "expense": {"type": "object", "description": "The expense entry being drafted."}, + } + assert agent.config.predict_state_config == { + "expense": {"tool": "submit_expense", "tool_argument": "expense"}, + } diff --git a/deployments/ag-ui-dev/deps/microsoft_agent_framework/tests/test_subagent_emitter.py b/deployments/ag-ui-dev/deps/microsoft_agent_framework/tests/test_subagent_emitter.py new file mode 100644 index 000000000..a9a28d0fa --- /dev/null +++ b/deployments/ag-ui-dev/deps/microsoft_agent_framework/tests/test_subagent_emitter.py @@ -0,0 +1,410 @@ +# SPDX-License-Identifier: MIT +"""Tests for the SUBAGENT_* emitter — drives the `delegation_*` helpers and +the queue-merge run wrapper with a fake inner bridge generator plus a +scripted tool enqueue (the fake generator calls the helpers between its own +yields, exactly where the framework invokes the real tool on the pump's +driving chain) and asserts the exact merged sequence field-for-field.""" + +import asyncio + +import pytest + +from ag_ui.core import ( + EventType, + RunFinishedEvent, + RunStartedEvent, + ToolCallArgsEvent, + ToolCallEndEvent, + ToolCallResultEvent, + ToolCallStartEvent, +) + +from src import subagent_emitter +from src.subagent_emitter import ( + SubagentEmittingAgent, + current_tool_call_id, + delegation_delta, + delegation_error, + delegation_finished, + delegation_started, + wrap_agent_run, +) + +TID = "call_7sxPY1sC236nPyHRTWAZMJB9" +RUN_ID = f"{TID}-sub" +MESSAGE_ID = f"{TID}-sub-m1" + + +# --------------------------------------------------------------------------- +# Helper-level tests: install a session directly and inspect the queue. +# --------------------------------------------------------------------------- + + +@pytest.fixture +def session(): + s = subagent_emitter._EmitterSession(queue=asyncio.Queue()) + token = subagent_emitter._event_queue.set(s) + yield s + subagent_emitter._event_queue.reset(token) + + +def _drain(session) -> list: + out = [] + while not session.queue.empty(): + out.append(session.queue.get_nowait()) + return out + + +def test_success_sequence_field_for_field(session): + delegation_started(TID, "policy_researcher") + delegation_delta(TID, "- Pre") + delegation_delta(TID, "-approval") + delegation_delta(TID, " required") + delegation_finished(TID) + + out = _drain(session) + assert [ev.type for ev in out] == [ + 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 == "policy_researcher" + assert started.parent_tool_call_id == TID + + 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] == ["- Pre", "-approval", " required"] + 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" + + +def test_no_deltas_still_brackets_with_started_and_finished(session): + delegation_started(TID, "policy_researcher") + delegation_finished(TID) + assert [ev.type for ev in _drain(session)] == [ + EventType.SUBAGENT_STARTED, + EventType.SUBAGENT_FINISHED, + ] + + +def test_error_closes_open_message_then_reports(session): + delegation_started(TID, "policy_researcher") + delegation_delta(TID, "- Pre") + delegation_error(TID, "specialist exploded") + + out = _drain(session) + 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" + + +def test_events_after_terminal_are_ignored(session): + delegation_started(TID, "policy_researcher") + delegation_finished(TID) + delegation_delta(TID, "late straggler") + delegation_finished(TID) + assert [ev.type for ev in _drain(session)] == [ + EventType.SUBAGENT_STARTED, + EventType.SUBAGENT_FINISHED, + ] + + +def test_empty_delta_is_dropped(session): + delegation_started(TID, "policy_researcher") + delegation_delta(TID, "") + delegation_finished(TID) + assert [ev.type for ev in _drain(session)] == [ + EventType.SUBAGENT_STARTED, + EventType.SUBAGENT_FINISHED, + ] + + +def test_none_tid_falls_back_to_generated_run_id(session): + delegation_started(None, "policy_researcher") + delegation_delta(None, "- x") + delegation_finished(None) + + out = _drain(session) + started = out[0] + assert started.parent_tool_call_id is None + assert started.subagent_run_id.startswith("sub-") + assert len(started.subagent_run_id) == len("sub-") + 8 + # All subsequent events carry the same generated run id. + assert {ev.subagent_run_id for ev in out} == {started.subagent_run_id} + assert out[1].message_id == f"{started.subagent_run_id}-m1" + + +def test_helpers_are_noops_without_a_session(): + # Unit tests / direct agent runs: no wrapper, no queue — nothing raises. + assert subagent_emitter._event_queue.get() is None + assert current_tool_call_id("research_policy") is None + delegation_started(TID, "policy_researcher") + delegation_delta(TID, "- x") + delegation_finished(TID) + delegation_error(TID, "boom") + + +# --------------------------------------------------------------------------- +# Wrapper-level tests: queue-merge around a fake inner bridge generator. +# --------------------------------------------------------------------------- + + +class _FakeInner: + """Duck-typed AgentFrameworkAgent carrying the attributes the wrapper + copies plus a scripted `run` generator.""" + + def __init__(self, gen_fn): + self.agent = object() + self.name = "fake" + self.description = "" + self.config = object() + self._approval_state_store = object() + self._gen_fn = gen_fn + + def run(self, input_data): + return self._gen_fn(input_data) + + +def _run_started(): + return RunStartedEvent(type=EventType.RUN_STARTED, thread_id="t", run_id="r") + + +def _run_finished(): + return RunFinishedEvent(type=EventType.RUN_FINISHED, thread_id="t", run_id="r") + + +async def _collect(agent) -> list: + return [ev async for ev in agent.run({"messages": []})] + + +async def test_wrapper_merges_tool_enqueued_events_mid_stream(): + async def inner(_input): + # Measured wire order: the bridge streams TOOL_CALL_START + ARGS + # before invoking the tool; TOOL_CALL_END arrives only AFTER the + # tool returns (docs/wire-capture-subagents.md, "After the emitter"). + yield _run_started() + yield ToolCallStartEvent( + type=EventType.TOOL_CALL_START, tool_call_id=TID, tool_call_name="research_policy" + ) + yield ToolCallArgsEvent( + type=EventType.TOOL_CALL_ARGS, tool_call_id=TID, delta='{"category":"travel","amount":900}' + ) + # The framework invokes the tool HERE, on the pump's driving chain, + # while the outer consumer is awaiting the queue. + tid = current_tool_call_id("research_policy") + assert tid == TID # recorded by the pump from TOOL_CALL_START + delegation_started(tid, "policy_researcher") + delegation_delta(tid, "- Pre-approval") + delegation_delta(tid, " required") + delegation_finished(tid) + yield ToolCallEndEvent(type=EventType.TOOL_CALL_END, tool_call_id=TID) + yield ToolCallResultEvent( + type=EventType.TOOL_CALL_RESULT, message_id="m", tool_call_id=TID, content="- Pre-approval required" + ) + yield _run_finished() + + out = await _collect(wrap_agent_run(_FakeInner(inner))) + assert [ev.type for ev in out] == [ + EventType.RUN_STARTED, + EventType.TOOL_CALL_START, + EventType.TOOL_CALL_ARGS, + # Child events land BETWEEN inner generator items — before the + # bridge's own TOOL_CALL_END/RESULT reach the client. + EventType.SUBAGENT_STARTED, + EventType.TEXT_MESSAGE_START, + EventType.TEXT_MESSAGE_CONTENT, + EventType.TEXT_MESSAGE_CONTENT, + EventType.TEXT_MESSAGE_END, + EventType.SUBAGENT_FINISHED, + EventType.TOOL_CALL_END, + EventType.TOOL_CALL_RESULT, + EventType.RUN_FINISHED, + ] + started = out[3] + assert started.subagent_run_id == RUN_ID + assert started.parent_tool_call_id == TID + assert [ev.delta for ev in out[5:7]] == ["- Pre-approval", " required"] + + +async def test_same_tool_double_call_gets_distinct_tids_and_delegations(): + # MAF runs multi-tool batches concurrently (asyncio.gather) and the + # bridge streams every TOOL_CALL_START before the tools execute — so + # two research_policy calls must each claim their OWN tid from the + # FIFO and drive their own delegation, never sharing one message. + tid2 = "call_secondResearchPolicyCall00" + + async def inner(_input): + yield ToolCallStartEvent( + type=EventType.TOOL_CALL_START, tool_call_id=TID, tool_call_name="research_policy" + ) + yield ToolCallStartEvent( + type=EventType.TOOL_CALL_START, tool_call_id=tid2, tool_call_name="research_policy" + ) + # Both tool bodies run concurrently; their deltas interleave. + tid_a = current_tool_call_id("research_policy") + tid_b = current_tool_call_id("research_policy") + assert (tid_a, tid_b) == (TID, tid2) # FIFO: oldest first + delegation_started(tid_a, "policy_researcher") + delegation_started(tid_b, "policy_researcher") + delegation_delta(tid_a, "- travel rules") + delegation_delta(tid_b, "- meal rules") + delegation_delta(tid_a, " apply") + delegation_finished(tid_b) + delegation_finished(tid_a) + yield ToolCallEndEvent(type=EventType.TOOL_CALL_END, tool_call_id=TID) + yield ToolCallEndEvent(type=EventType.TOOL_CALL_END, tool_call_id=tid2) + + out = await _collect(wrap_agent_run(_FakeInner(inner))) + + started = [ev for ev in out if ev.type == EventType.SUBAGENT_STARTED] + assert [ev.subagent_run_id for ev in started] == [f"{TID}-sub", f"{tid2}-sub"] + assert [ev.parent_tool_call_id for ev in started] == [TID, tid2] + + # Identity separation: every child event carries its own delegation's + # ids — interleaved ORDER between the two runs is fine. + for ev in out: + if ev.type == EventType.TEXT_MESSAGE_CONTENT and ev.delta.startswith("- travel"): + assert ev.message_id == f"{TID}-sub-m1" + if ev.type == EventType.TEXT_MESSAGE_CONTENT and ev.delta.startswith("- meal"): + assert ev.message_id == f"{tid2}-sub-m1" + a_events = [ev for ev in out if getattr(ev, "subagent_run_id", None) == f"{TID}-sub"] + b_events = [ev for ev in out if getattr(ev, "subagent_run_id", None) == f"{tid2}-sub"] + assert [ev.type for ev in a_events] == [ + EventType.SUBAGENT_STARTED, + EventType.TEXT_MESSAGE_START, + EventType.TEXT_MESSAGE_CONTENT, + EventType.TEXT_MESSAGE_CONTENT, + EventType.TEXT_MESSAGE_END, + EventType.SUBAGENT_FINISHED, + ] + assert [ev.type for ev in b_events] == [ + EventType.SUBAGENT_STARTED, + EventType.TEXT_MESSAGE_START, + EventType.TEXT_MESSAGE_CONTENT, + EventType.TEXT_MESSAGE_END, + EventType.SUBAGENT_FINISHED, + ] + assert {ev.message_id for ev in a_events if hasattr(ev, "message_id")} == {f"{TID}-sub-m1"} + assert {ev.message_id for ev in b_events if hasattr(ev, "message_id")} == {f"{tid2}-sub-m1"} + + +async def test_wrapper_error_path_emits_subagent_error_then_propagates(): + class _Boom(RuntimeError): + pass + + async def inner(_input): + yield _run_started() + yield ToolCallStartEvent( + type=EventType.TOOL_CALL_START, tool_call_id=TID, tool_call_name="research_policy" + ) + tid = current_tool_call_id("research_policy") + delegation_started(tid, "policy_researcher") + delegation_delta(tid, "- Pre") + delegation_error(tid, "specialist exploded") + raise _Boom("specialist exploded") + + agent = wrap_agent_run(_FakeInner(inner)) + out = [] + with pytest.raises(_Boom): + async for ev in agent.run({"messages": []}): + out.append(ev) + # Everything enqueued before the failure was delivered, ending in the + # SUBAGENT_ERROR (with the open child message closed first). + assert [ev.type for ev in out] == [ + EventType.RUN_STARTED, + EventType.TOOL_CALL_START, + EventType.SUBAGENT_STARTED, + EventType.TEXT_MESSAGE_START, + EventType.TEXT_MESSAGE_CONTENT, + EventType.TEXT_MESSAGE_END, + EventType.SUBAGENT_ERROR, + ] + assert out[-1].message == "specialist exploded" + + +async def test_wrapper_clean_shutdown_on_consumer_break(): + closed = asyncio.Event() + + async def inner(_input): + try: + yield _run_started() + while True: # endless stream: only a cancel/close ends it + await asyncio.sleep(0.01) + yield _run_started() + finally: + closed.set() + + agent = wrap_agent_run(_FakeInner(inner)) + gen = agent.run({"messages": []}) + first = await gen.__anext__() + assert first.type == EventType.RUN_STARTED + await gen.aclose() # consumer break / client disconnect + + await asyncio.wait_for(closed.wait(), timeout=1) + # No orphaned tasks: everything spawned by the wrapper is done. + pending = [ + t + for t in asyncio.all_tasks() + if t is not asyncio.current_task() and not t.done() + ] + assert pending == [] + # The ContextVar session was uninstalled. + assert subagent_emitter._event_queue.get() is None + + +async def test_wrapper_resets_contextvar_after_normal_completion(): + async def inner(_input): + yield _run_started() + yield _run_finished() + + out = await _collect(wrap_agent_run(_FakeInner(inner))) + assert [ev.type for ev in out] == [EventType.RUN_STARTED, EventType.RUN_FINISHED] + assert subagent_emitter._event_queue.get() is None + + +def test_wrap_agent_run_returns_agentframeworkagent_for_endpoint_dispatch(): + fake = _FakeInner(None) + wrapped = wrap_agent_run(fake) + assert isinstance(wrapped, SubagentEmittingAgent) + # The endpoint dispatches on isinstance(agent, AgentFrameworkAgent) and + # shares the config / approval-state store. + from agent_framework.ag_ui import AgentFrameworkAgent + + assert isinstance(wrapped, AgentFrameworkAgent) + assert wrapped.config is fake.config + assert wrapped._approval_state_store is fake._approval_state_store + + +def test_server_mounts_the_wrapped_agent(): + from src import server + + # The FastAPI endpoint consumes the wrapped run (protocol_runner is the + # SubagentEmittingAgent), so SUBAGENT_* events reach the wire. + assert isinstance(server.wrapped_agent, SubagentEmittingAgent) diff --git a/deployments/ag-ui-dev/deps/microsoft_agent_framework/uv.lock b/deployments/ag-ui-dev/deps/microsoft_agent_framework/uv.lock index d164f91fb..aafd57d70 100644 --- a/deployments/ag-ui-dev/deps/microsoft_agent_framework/uv.lock +++ b/deployments/ag-ui-dev/deps/microsoft_agent_framework/uv.lock @@ -121,6 +121,12 @@ dependencies = [ { name = "uvicorn", extra = ["standard"] }, ] +[package.dev-dependencies] +dev = [ + { name = "pytest" }, + { name = "pytest-asyncio" }, +] + [package.metadata] requires-dist = [ { name = "agent-framework-ag-ui", specifier = ">=1.2.1" }, @@ -130,6 +136,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" @@ -246,6 +258,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" @@ -385,6 +406,24 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/ca/6f/a04e900f465ff3221ccc395522503e2d10e79fa21f2723c8e177aae1e0d1/opentelemetry_api-1.44.0-py3-none-any.whl", hash = "sha256:94b98c893a91b88657eaac1e3ba89618cdb85be6918196705354f34728b2cdef", size = 60018, upload-time = "2026-07-16T15:25:11.657Z" }, ] +[[package]] +name = "packaging" +version = "26.3" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/7d/fa/3944b40b07da9ce895c0e6303a5ab7d53da063554f534556b134a54d6093/packaging-26.3.tar.gz", hash = "sha256:94edc256424af38762eb31306eed28beb9f0efc50a8837492c9d6fd6004aed79", size = 313412, upload-time = "2026-08-04T18:15:28.737Z" } +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 = "pydantic" version = "2.13.5" @@ -475,6 +514,44 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/fa/04/c81d4841331c2178b6fb09ae225425e110ed72d990c9fe556c4ec03d1013/pydantic_core-2.46.5-graalpy312-graalpy250_312_native-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:8e24d8f05fa2d28513d94e877e9c75ad66175376209b3977f916e240e623193c", size = 2111034, upload-time = "2026-08-28T10:01:07.345Z" }, ] +[[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 = "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-dotenv" version = "1.2.3"