From 799b15b663b8ea4cded8ec54283291fe7ff00879 Mon Sep 17 00:00:00 2001 From: Frost Ming Date: Thu, 1 Oct 2026 14:19:49 +0800 Subject: [PATCH] feat(acp): support Lody subagent events and automatic session titles Implement two Lody extension contracts from acp-extension-core, both gated on the client identifying itself as Lody in `InitializeRequest.clientInfo`. Subagent events - Add `bub_acp_server.subagents` with the `_lody/subagents/event` wire contract: capability negotiation, a per-execution `SubagentEventEmitter`, JSON-safe payloads, and best-effort delivery that never fails the run. - Bub's `subagent` tool consumes the child stream internally, so replace the runtime agent's tool with a streaming variant when the client negotiates the capability. The replacement keeps the original name, description, parameters, renderer and output schema, and still returns Bub's result, so the model sees no difference and the global registry stays untouched. - Map child stream events onto snapshot, sparse progress and output payloads. `outputRead` is `none` and `cancel` is `false`, since runs are not addressable through the separate list/cancel/output request surface. - Negotiation stays bilateral: a client opt-in is honored even when the agent did not advertise, as the contract requires. Session titles - Advertise `sessionTitle` and push a tagged `session_info_update` after a session's first turn, then persist the title for `session/list`. - Generate with one lean completion: a throwaway `temp/` session with no tools and no skills. Best effort, bounded by a timeout, at most once per session, skipped for sessions that already have a title. Capabilities - Report `agentCapabilities._meta.lody` only to Lody clients; other clients get no Lody-namespaced capability and no Lody-specific behavior. - Keep the non-Lody `{"steering": {"supported": true}}` field unconditional so clients using the original Codex steering route can still detect it. Tests cover the wire contract, negotiation, failure and cancellation paths, client identification, and a real stdio end-to-end run. --- packages/bub-acp-server/README.md | 131 +++++- .../src/bub_acp_server/agent.py | 435 +++++++++++++++++- .../src/bub_acp_server/subagents.py | 199 ++++++++ packages/bub-acp-server/tests/test_e2e.py | 8 +- packages/bub-acp-server/tests/test_mcp.py | 3 + packages/bub-acp-server/tests/test_plugin.py | 13 +- .../tests/test_session_title.py | 399 ++++++++++++++++ .../bub-acp-server/tests/test_subagents.py | 429 +++++++++++++++++ 8 files changed, 1599 insertions(+), 18 deletions(-) create mode 100644 packages/bub-acp-server/src/bub_acp_server/subagents.py create mode 100644 packages/bub-acp-server/tests/test_session_title.py create mode 100644 packages/bub-acp-server/tests/test_subagents.py diff --git a/packages/bub-acp-server/README.md b/packages/bub-acp-server/README.md index d25be4d..0441a66 100644 --- a/packages/bub-acp-server/README.md +++ b/packages/bub-acp-server/README.md @@ -15,6 +15,8 @@ Expose Bub as an Agent Client Protocol agent. - Session MCP servers over stdio, Streamable HTTP, and SSE via `bub-mcp` - ACP context-compaction notifications when `tape.handoff` runs - Mid-turn steering through the `_lody/session/steer` ACP extension +- Streamed subagent execution events through `_lody/subagents/event` +- Automatic session titles pushed as `session_info_update` ## Installation @@ -245,6 +247,48 @@ transport = await create_websocket_stream("ws://127.0.0.1:28200/acp") Initialization, session loading, prompts and tool callbacks work over the socket; `connection.close()` closes it. WebSocket does not require HTTP/2. Like HTTP, it does not advertise the SDK's disabled unstable resume/close routes and adds no authentication. +## Lody extension capabilities + +Everything Lody adds lives under `agentCapabilities._meta.lody`, and that block +is reported only to clients that identify themselves as Lody in +`InitializeRequest.clientInfo`: + +```json +{ + "clientInfo": { "name": "lody", "title": "Lody", "version": "1" } +} +``` + +The `name` is matched case-insensitively as a prefix, so variants such as +`lody-cli` are recognized; an absent or unknown identity is not Lody. Other +clients receive no Lody-namespaced capability and no Lody-specific behavior. +The generic `{"steering": {"supported": true}}` field on the initialize response +is not Lody-namespaced and stays unconditional, so clients using the original +Codex steering route can still detect it. + +A Lody client sees the complete set: + +```json +{ + "agentCapabilities": { + "_meta": { + "lody": { + "steering": { + "version": 1, + "transport": "request", + "upstreamTurn": "same", + "configPolicy": "active" + }, + "subagentEvents": { "version": 1 }, + "sessionTitle": { "version": 1 } + } + } + } +} +``` + +Each section below describes one entry. + ## Steering Clients can detect steering support in the initialize response: @@ -260,7 +304,8 @@ Clients can detect steering support in the initialize response: ``` Clients that require acknowledged steering can also negotiate the Lody -extension under `agentCapabilities._meta`: +extension under `agentCapabilities._meta`, which Lody clients receive in the +capability block above: ```json { @@ -307,6 +352,90 @@ For compatibility with clients using the original Codex steering extension, Bub also accepts `_session/steering`. Its optional `steerId` uses the matching `_session/steering_applied` notification. +## Subagent events + +Bub's `subagent` tool consumes the child agent's stream internally, so an ACP +client would otherwise see only the outer tool call. Clients that want the child +run can negotiate the Lody `subagentEvents` contract. Both sides opt in +independently, so the agent advertises the capability: + +```json +{ + "agentCapabilities": { + "_meta": { + "lody": { + "subagentEvents": { + "version": 1 + } + } + } + } +} +``` + +and a client opts in by sending the same version in +`InitializeRequest.clientCapabilities._meta.lody`. Only then does Bub replace +that session's `subagent` tool with a streaming variant and publish +`_lody/subagents/event` notifications. The advertisement is part of the Lody +capability block, so only Lody clients see it; the client's answer remains the +switch, and an opt-in is honored even from a client that never saw the +advertisement. The replacement keeps the tool's name, +description, parameters, renderer and output schema, and still returns Bub's +`{session_id, output, errors}` result to the model, so delegation behaves +exactly as before. + +Each event carries the ACP `sessionId`, an opaque per-execution `runId`, and one +of three payloads: + +| `type` | Payload | +| ---------- | ------------------------------------------------------------------------------------------ | +| `snapshot` | Task state (`running`, `completed`, `failed`, `cancelled`), description, model, summary | +| `progress` | Sparse absolute observations such as `lastToolName`, `toolCallCount`, `turnCount`, `durationMs` | +| `output` | One child stream chunk as an ACP session update: text, thought, or tool call/update | + +Snapshots replace previously known task metadata, and output follows the child's +connection order. There are no sequence numbers, no replay guarantees and no +cross-reconnect deduplication, and progress counters are display observations +rather than usage accounting. A client that cannot receive an event never fails +the tool. + +`outputRead` is advertised as `none` and `cancel` as `false`, because runs are +not addressable through the separate `_lody/subagents/list|cancel|output` +request surface. Publishing is best effort: a client that cannot receive an +event never fails the run, and the tool still returns its result. + +The capability is always advertised, and client negotiation alone decides +whether events are published. Clients that never opt in keep Bub's stock +`subagent` tool unchanged. + +## Session titles + +This contract has no client-side capability to negotiate, so it rides on the +Lody client identification described above; other clients keep their own title +process. + +After the first turn of a session, Bub asks the session's model for a short +title with a single completion that carries no tools and no skills, then pushes +the result on the standard ACP callback: + +```json +{ + "sessionId": "session-id", + "update": { + "sessionUpdate": "session_info_update", + "title": "Fix login redirect", + "_meta": { "lody": { "titleSource": "generated" } } + } +} +``` + +The title is stored in `acp-sessions.json` and returned by `session/list` +afterwards. Advertising the capability tells a Lody client that it can skip its +own title generator for the session. Generation happens at most once per +session, is skipped for sessions that already have a title, and is best effort: +a failure or a timeout leaves the client's draft title in place and no second +generator is requested. + ## Use In Zed Zed supports external terminal agents through ACP. Custom agents are configured in Zed's `settings.json` under `agent_servers`. diff --git a/packages/bub-acp-server/src/bub_acp_server/agent.py b/packages/bub-acp-server/src/bub_acp_server/agent.py index ea85e4c..d4c1003 100644 --- a/packages/bub-acp-server/src/bub_acp_server/agent.py +++ b/packages/bub-acp-server/src/bub_acp_server/agent.py @@ -6,10 +6,11 @@ import json import logging import re +import time from collections import deque from collections.abc import AsyncIterable, AsyncIterator, Mapping from contextvars import ContextVar -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace from datetime import UTC, datetime from pathlib import Path from typing import TYPE_CHECKING, Any @@ -45,6 +46,7 @@ SessionConfigOptionSelect, SessionConfigSelectOption, SessionInfo, + SessionInfoUpdate, SessionListCapabilities, SessionResumeCapabilities, SetSessionConfigOptionResponse, @@ -61,6 +63,7 @@ from bub.model_selection import ModelChoice, ModelOptions from bub.sidecars import sidecar_tape_name from bub.streaming import StreamEvent +from bub.tools import ToolContext from bub.tape import ( AsyncTapeStoreAdapter, Tape, @@ -73,12 +76,19 @@ from bub_acp_server.client_tools import ACPClientToolRuntime, build_client_tools from bub_acp_server.config import ACPServerSettings from bub_acp_server.mcp import ACPMcpServer, connect_session_mcp, server_configs +from bub_acp_server.subagents import ( + LODY_SUBAGENT_EVENTS_CAPABILITY, + SubagentEventEmitter, + client_supports_subagent_events, + json_safe, +) from bub_mcp.plugin import MCPChannel from bub_acp_server.steering import ACPSteeringInbox if TYPE_CHECKING: - from bub.framework import BubFramework from bub.builtin.agent import Agent + from bub.builtin.tools import SubAgentInput, SubAgentResult + from bub.framework import BubFramework type ACPPromptBlock = ( TextContentBlock @@ -123,6 +133,24 @@ "configPolicy": "active", } +_LODY_EXTENSION_CAPABILITIES: dict[str, object] = { + "steering": _LODY_STEERING_CAPABILITY, + "subagentEvents": dict(LODY_SUBAGENT_EVENTS_CAPABILITY), + "sessionTitle": {"version": 1}, +} + +#: How long a generated session title may take before it is abandoned. A title is +#: best effort: the client keeps its own draft when generation fails. +SESSION_TITLE_TIMEOUT_SECONDS = 30.0 +SESSION_TITLE_MAX_CHARS = 80 +SESSION_TITLE_INPUT_CHARS = 2000 +_SESSION_TITLE_INSTRUCTION = ( + "Write a title for the coding session that starts with the user request below.\n" + "Reply with the title only: at most 8 words, no quotes, no surrounding text, " + "no markdown and no trailing punctuation.\n\n" + "User request:\n{request}" +) + _BUB_PROMPT_CONTEXT = re.compile( r"^(?=[^\n]*channel=\$)(?=[^\n]*chat_id=)[^\n]+\n" r"---Date: [^\n]+---\n", @@ -445,6 +473,10 @@ def __init__( self._prompt_runs = prompt_runs if prompt_runs is not None else {} self._steering_locks: dict[str, asyncio.Lock] = {} self._background_tasks: set[asyncio.Task[PromptResponse]] = set() + self._subagent_events_enabled = False + self._lody_client = False + self._title_tasks: dict[str, asyncio.Task[None]] = {} + self._title_sessions: set[str] = set() self._closing_sessions = ( closing_sessions if closing_sessions is not None else set() ) @@ -473,8 +505,18 @@ async def initialize( bub_version = importlib.metadata.version("bub") except importlib.metadata.PackageNotFoundError: bub_version = "0.0.0" - del client_info, kwargs + del kwargs self.client_tools.set_capabilities(client_capabilities) + # Subagent events are opt-in on both sides: publishing needs the client to + # advertise the same version, whoever the client is. + self._subagent_events_enabled = client_supports_subagent_events( + client_capabilities + ) + # `_meta.lody` is the Lody extension surface, so it is advertised only to + # clients that identify themselves as Lody. Other clients see nothing + # Lody-namespaced and never receive Lody-specific behavior such as + # automatic session titles. + self._lody_client = is_lody_client(client_info) return InitializeResponse( protocol_version=protocol_version, agent_info=Implementation(name="bub", title="Bub", version=bub_version), @@ -482,9 +524,11 @@ async def initialize( agent_capabilities=AgentCapabilities( load_session=True, mcp_capabilities=McpCapabilities(http=True, sse=True), - field_meta={ - "lody": {"steering": _LODY_STEERING_CAPABILITY}, - }, + field_meta=( + {"lody": dict(_LODY_EXTENSION_CAPABILITIES)} + if self._lody_client + else None + ), session_capabilities=SessionCapabilities( delete=SessionDeleteCapabilities(), close=SessionCloseCapabilities() @@ -638,6 +682,7 @@ async def _end_session(self, session_id: str, *, delete_tape: bool) -> None: raise finally: self._steering_locks.pop(session_id, None) + self._title_sessions.discard(session_id) self._closing_sessions.discard(session_id) async def cancel(self, session_id: str, **kwargs: Any) -> None: @@ -837,6 +882,7 @@ async def _execute_prompt( return PromptResponse(stop_reason="end_turn") finally: self._complete_prompt_run(run) + self._schedule_session_title(prompt, session) def _build_inbound( self, prompt: list[ACPPromptBlock], session: ACPSession @@ -864,12 +910,293 @@ def _build_inbound( runtime_agent = Agent(self.framework) runtime_agent.tools.update(build_client_tools(self.client_tools)) + if self._subagent_events_enabled: + self._install_streaming_subagent(runtime_agent, session) self._runtime_agents[session.session_id] = runtime_agent # Bub's builtin load_state uses this instance for recovery and execution. inbound._runtime_agent = runtime_agent inbound._mcp_channel = self._mcp_channels.get(session.session_id) return inbound + def _install_streaming_subagent( + self, runtime_agent: Agent, session: ACPSession + ) -> None: + """Replace this session's `subagent` tool with a streaming variant. + + The replacement keeps the original tool's name, description, parameters, + renderer, output schema and code-mode flags, so the model sees no + difference. Only the handler changes: it republishes the child run + through `_lody/subagents/event` while still returning Bub's result. + """ + + tool = runtime_agent.tools.get("subagent") + if tool is None: + return + + async def handler(*args: Any, **kwargs: Any) -> Any: + from bub.builtin.tools import SubAgentInput + + context = kwargs.pop("context", None) + param = SubAgentInput(*args, **kwargs) + return await self._run_streaming_subagent(param, context, session) + + runtime_agent.tools["subagent"] = replace(tool, handler=handler) + + async def _run_streaming_subagent( + self, + param: SubAgentInput, + context: ToolContext | None, + session: ACPSession, + ) -> SubAgentResult: + from bub.builtin.tools import resolve_tool_names + + if context is None: + raise RuntimeError("subagent tool requires a tool context") + agent = context.state.get("_runtime_agent") + if agent is None: + raise RuntimeError("no runtime agent found in tool context") + + emitter = SubagentEventEmitter(self._require_client(), session.session_id) + child_session = _subagent_session_id( + param.session, str(context.state.get("session_id") or "") + ) + child_state = {**context.state, "session_id": child_session} + allowed_tools = resolve_tool_names( + param.allowed_tools or None, exclude={"subagent"}, all_names=agent.tools + ) + await emitter.started( + description=_subagent_description(param.prompt), model_id=param.model + ) + + started_at = time.time() + output: list[str] = [] + errors: list[str] = [] + pending: list[tuple[str, str]] = [] + next_index = 0 + turn_count = 0 + tool_call_count = 0 + last_tool_name: str | None = None + reported_usage: object = None + + def elapsed_ms() -> int: + return int((time.time() - started_at) * 1000) + + try: + stream = await agent.run_stream( + session_id=child_session, + prompt=param.prompt, + state=child_state, + model=param.model, + allowed_tools=allowed_tools, + allowed_skills=param.allowed_skills, + ) + async with contextlib.aclosing(stream): + async for event in stream: + if event.kind in ("text", "reasoning"): + delta = str(event.data.get("delta", "")) + if not delta: + continue + if event.kind == "text": + output.append(delta) + await emitter.output( + _text_update( + "agent_message_chunk" + if event.kind == "text" + else "agent_thought_chunk", + delta, + ) + ) + elif event.kind == "tool_call": + for call in _list_payload(event.data.get("tool_calls")): + tool_id = _tool_call_id(next_index, call) + name = _tool_name(call) + raw_input = _tool_raw_input(call) + next_index += 1 + tool_call_count += 1 + last_tool_name = name + title = _tool_title(name, raw_input) + pending.append((tool_id, title)) + await emitter.output( + { + "sessionUpdate": "tool_call", + "toolCallId": tool_id, + "title": title, + "kind": _tool_kind(name), + "status": "in_progress", + "rawInput": json_safe(raw_input), + } + ) + await emitter.progress( + lastToolName=last_tool_name, + turnCount=turn_count, + toolCallCount=tool_call_count, + durationMs=elapsed_ms(), + ) + elif event.kind == "tool_result": + results = _list_payload(event.data.get("tool_results")) + for position, result in enumerate(results): + if position < len(pending): + tool_id, title = pending[position] + else: + tool_id, title = f"tool-{next_index}", "tool" + next_index += 1 + await emitter.output( + { + "sessionUpdate": "tool_call_update", + "toolCallId": tool_id, + "title": title, + "status": "completed", + "content": [ + { + "type": "content", + "content": { + "type": "text", + "text": _stringify(result), + }, + } + ], + } + ) + pending = [] + elif event.kind == "error": + errors.append( + str( + event.data.get("message") + or event.data.get("error") + or "unknown error" + ) + ) + elif event.kind == "usage": + reported_usage = _event_usage(event) + elif event.kind == "final": + turn_count += 1 + except asyncio.CancelledError: + await emitter.snapshot("cancelled", reason_code="cancelled") + raise + except Exception as error: + await emitter.snapshot( + "failed", reason_code="error", reason_message=str(error) + ) + raise + + text = "".join(output) + # Progress counters are display observations, never usage accounting input. + await emitter.progress( + lastToolName=last_tool_name, + turnCount=turn_count, + toolCallCount=tool_call_count, + durationMs=elapsed_ms(), + totalTokens=_subagent_total_tokens( + getattr(stream, "usage", None) or reported_usage + ), + ) + await emitter.snapshot("completed", summary=_summary(text)) + return {"session_id": child_session, "output": text, "errors": errors} + + def _schedule_session_title( + self, prompt: list[ACPPromptBlock], session: ACPSession + ) -> None: + """Start best-effort title generation for a session's first turn. + + Only Lody clients receive titles: the capability is a promise that the + client can drop its own title generator for this session. + """ + + if not self._lody_client: + return + if session.session_id not in self._sessions: + return + if session.title is not None or session.session_id in self._title_sessions: + return + request = _prompt_text(prompt) + if not request: + return + + self._title_sessions.add(session.session_id) + task = asyncio.create_task( + self._publish_session_title(session, request), + name=f"acp-session-title-{session.session_id}", + ) + self._title_tasks[session.session_id] = task + task.add_done_callback( + lambda _task, session_id=session.session_id: self._title_tasks.pop( + session_id, None + ) + ) + + async def _publish_session_title(self, session: ACPSession, request: str) -> None: + """Generate a title, persist it and push it to the client.""" + + try: + generated = await asyncio.wait_for( + self._generate_session_title(session, request), + SESSION_TITLE_TIMEOUT_SECONDS, + ) + except asyncio.CancelledError: + raise + except Exception as error: + # A title is best effort: the client keeps its own draft. + logger.warning( + "Session title generation failed for {}: {}", + session.session_id, + error, + ) + return + + title = _clean_title(generated) + if not title or session.session_id not in self._sessions: + return + session.title = title + self._save_sessions() + client = self._client + if client is None: + return + try: + await client.session_update( + session.session_id, + SessionInfoUpdate( + title=title, field_meta={"lody": {"titleSource": "generated"}} + ), + ) + except Exception as error: + # The title is already persisted; a failed push must not strand the task. + logger.warning( + "Failed to push the session title for {}: {}", + session.session_id, + error, + ) + + async def _generate_session_title(self, session: ACPSession, request: str) -> str: + """Ask the session's model for a short title using a throwaway tape.""" + + agent = self._runtime_agents.get(session.session_id) + if agent is None: + return "" + title_session = f"temp/{uuid4().hex[:8]}" + state: dict[str, Any] = { + "_runtime_agent": agent, + "_runtime_workspace": str(session.cwd), + "session_id": title_session, + } + prompt = _SESSION_TITLE_INSTRUCTION.format( + request=request[:SESSION_TITLE_INPUT_CHARS] + ) + parts: list[str] = [] + # A title needs one completion, not the session's tool surface or skills. + stream = await agent.run_stream( + session_id=title_session, + prompt=prompt, + state=state, + model=session.runtime.get("model"), + allowed_tools=set(), + allowed_skills=set(), + ) + async with contextlib.aclosing(stream): + async for event in stream: + if event.kind == "text": + parts.append(str(event.data.get("delta", ""))) + return "".join(parts) + async def _replace_session_mcp( self, session: ACPSession, servers: list[ACPMcpServer] | None ) -> None: @@ -1419,3 +1746,99 @@ def _stringify(value: object) -> str: if isinstance(value, str): return value return repr(value) + + +#: Client identity the Lody app sends in `InitializeRequest.clientInfo`: +#: `{name: "lody", title: "Lody", version: "1"}`. Matching is prefix-based on +#: `name` so variants such as `lody-cli` are still recognized. +_LODY_CLIENT_NAME = "lody" + + +def is_lody_client(client_info: Implementation | None) -> bool: + """Whether this connection comes from the Lody client. + + Session titles are the only contract here that has no client-side capability + to negotiate, so the client has to be recognized from `clientInfo` instead. + An unknown or absent identity is not Lody. + """ + + name = getattr(client_info, "name", None) + if isinstance(name, str) and name.strip().casefold().startswith(_LODY_CLIENT_NAME): + return True + title = getattr(client_info, "title", None) + return isinstance(title, str) and title.strip().casefold() == _LODY_CLIENT_NAME + + +#: Quotation pairs a model may wrap a title in. +_QUOTE_PAIRS = {'"': '"', "'": "'", "\u201c": "\u201d", "\u2018": "\u2019", "`": "`"} + + +def _prompt_text(prompt: list[ACPPromptBlock]) -> str: + """Return the concatenated text blocks of one prompt.""" + + parts = [ + str(_block_value(block, "text", "")) + for block in prompt + if _block_type(block) == "text" + ] + return "\n".join(part for part in parts if part).strip() + + +def _text_update(session_update: str, text: str) -> dict[str, object]: + return { + "sessionUpdate": session_update, + "content": {"type": "text", "text": text}, + } + + +def _tool_title(name: str, raw_input: object) -> str: + """Human-readable tool title, matching the live stream's bash handling.""" + + if name == "bash": + command = _block_value(raw_input, "command") + if isinstance(command, str) and command: + return command + return name + + +def _subagent_session_id(strategy: str, parent_session_id: str) -> str: + """Mirror Bub's `subagent` session strategy.""" + + if strategy == "inherit": + return parent_session_id or f"temp/{uuid4().hex[:8]}" + if strategy == "temp": + return f"temp/{uuid4().hex[:8]}" + return strategy + + +def _subagent_description(prompt: object) -> str | None: + if not isinstance(prompt, str): + return None + text = prompt.strip() + if not text: + return None + return text.splitlines()[0].strip()[:200] or None + + +def _subagent_total_tokens(usage: object) -> int | None: + if not isinstance(usage, Mapping): + return None + return _usage_total_tokens(usage) + + +def _summary(text: str) -> str | None: + line = text.strip().splitlines()[0].strip() if text.strip() else "" + return line[:200] or None + + +def _clean_title(value: str) -> str: + """Normalize a generated title into a single short line.""" + + title = " ".join(str(value).split()).strip() + if not title: + return "" + if title.casefold().startswith("title:"): + title = title[len("title:") :].strip() + if len(title) > 1 and _QUOTE_PAIRS.get(title[0]) == title[-1]: + title = title[1:-1].strip() + return title[:SESSION_TITLE_MAX_CHARS].strip() diff --git a/packages/bub-acp-server/src/bub_acp_server/subagents.py b/packages/bub-acp-server/src/bub_acp_server/subagents.py new file mode 100644 index 0000000..3913ff4 --- /dev/null +++ b/packages/bub-acp-server/src/bub_acp_server/subagents.py @@ -0,0 +1,199 @@ +"""Lody subagent execution events (``_lody/subagents/event``). + +Bub's built-in ``subagent`` tool consumes the child agent's stream internally, so +an ACP client only ever sees the outer tool call. When the client negotiates +``_meta.lody.subagentEvents``, the ACP server replaces the runtime agent's +``subagent`` tool with a streaming variant that republishes the child run through +the Lody contract: one ``snapshot`` before the run, sparse ``progress`` +observations, one ``output`` event per child stream chunk, and a terminal +``snapshot``. + +The contract is deliberately lossy: there are no sequence numbers, no replay +guarantees and no cross-reconnect deduplication, and a dropped event never fails +the tool. Publishing is best effort because the tool result, not the event +stream, is what the model depends on. +""" + +from __future__ import annotations + +import json +import logging +import time +from collections.abc import Mapping +from typing import Any +from uuid import uuid4 + +from acp.interfaces import Client +from acp.schema import ClientCapabilities + +logger = logging.getLogger(__name__) + +LODY_SUBAGENT_EVENT_METHOD = "lody/subagents/event" +LODY_SUBAGENT_EVENTS_CAPABILITY: dict[str, int] = {"version": 1} + +#: Stream kinds this adapter republishes from a child run. Bub's agent loop has no +#: plan updates, so ``plan`` is not advertised. +STREAMED_OUTPUT_KINDS = ("text", "thought", "tool") + +TERMINAL_SUBAGENT_STATES = ("completed", "failed", "cancelled") + +_SUBAGENT_SUPPORT: dict[str, object] = { + "stream": list(STREAMED_OUTPUT_KINDS), + "progress": True, + # Output is pushed live; runs are not addressable through the separate + # `_lody/subagents/list|cancel|output` request surface. + "outputRead": "none", + "cancel": False, +} + + +def client_supports_subagent_events( + capabilities: ClientCapabilities | Mapping[str, object] | None, +) -> bool: + """Whether the client negotiated ``_meta.lody.subagentEvents`` version 1. + + Both sides opt in independently, so the agent may only publish events after + the client advertised the same version. + """ + + events = _lody_capability(capabilities, "subagentEvents") + return events is not None and events.get("version") == 1 + + +def _lody_capability( + capabilities: ClientCapabilities | Mapping[str, object] | None, + name: str, +) -> Mapping[str, object] | None: + meta = _field(capabilities, "field_meta") or _field(capabilities, "_meta") + if not isinstance(meta, Mapping): + return None + lody = meta.get("lody") + if not isinstance(lody, Mapping): + return None + capability = lody.get(name) + return capability if isinstance(capability, Mapping) else None + + +def _field(value: object, name: str) -> object: + if isinstance(value, Mapping): + return value.get(name) + return getattr(value, name, None) + + +def json_safe(value: object) -> object: + """Return a JSON-serializable view of a tool payload for the wire. + + Child tool arguments normally arrive as parsed JSON, but a provider may hand + back a live object instead. Event payloads must survive serialization, so fall + back to ``repr`` rather than dropping the event. + """ + + try: + json.dumps(value) + except (TypeError, ValueError): + return repr(value) + return value + + +class SubagentEventEmitter: + """Publishes one subagent execution through ``_lody/subagents/event``. + + One instance owns one opaque ``runId``. Snapshots replace previously known + task metadata, so the emitter only has to track its own lifetime. + """ + + def __init__(self, client: Client, session_id: str) -> None: + self._client = client + self._session_id = session_id + self._started_at = time.time() + self._terminated = False + self.run_id = uuid4().hex + + async def started( + self, *, description: str | None = None, model_id: str | None = None + ) -> None: + await self.snapshot( + "running", + description=description, + model_id=model_id, + started_at=self._started_at, + ) + + async def snapshot( + self, + state: str, + *, + description: str | None = None, + model_id: str | None = None, + summary: str | None = None, + started_at: float | None = None, + reason_code: str | None = None, + reason_message: str | None = None, + ) -> None: + """Publish a task snapshot; a terminal state ends this emitter.""" + + if self._terminated: + return + if state in TERMINAL_SUBAGENT_STATES: + self._terminated = True + reason = ( + {"code": reason_code, "message": reason_message} if reason_code else None + ) + await self._publish( + { + "type": "snapshot", + "snapshot": _prune( + { + "state": state, + "description": description, + "modelId": model_id, + "summary": summary, + "startedAtEpochSeconds": _epoch_seconds(started_at), + "endedAtEpochSeconds": ( + int(time.time()) + if state in TERMINAL_SUBAGENT_STATES + else None + ), + "reason": reason, + "support": dict(_SUBAGENT_SUPPORT), + } + ), + } + ) + + async def output(self, update: Mapping[str, object]) -> None: + """Publish one child stream chunk as an ACP session-update variant.""" + + if self._terminated: + return + await self._publish({"type": "output", "update": dict(update)}) + + async def progress(self, **fields: Any) -> None: + """Publish a sparse absolute progress observation.""" + + if self._terminated: + return + await self._publish({"type": "progress", "progress": _prune(fields)}) + + async def _publish(self, event: Mapping[str, object]) -> None: + payload = _prune( + { + "version": 1, + "sessionId": self._session_id, + "runId": self.run_id, + **event, + } + ) + try: + await self._client.ext_notification(LODY_SUBAGENT_EVENT_METHOD, payload) + except Exception: + # Observing a run must never break the run itself. + logger.warning("Failed to publish a subagent event", exc_info=True) + + +def _prune(values: Mapping[str, Any]) -> dict[str, Any]: + return {key: value for key, value in values.items() if value is not None} + + +def _epoch_seconds(value: float | None) -> int | None: + return None if value is None else int(value) diff --git a/packages/bub-acp-server/tests/test_e2e.py b/packages/bub-acp-server/tests/test_e2e.py index ceb97e0..9a28c18 100644 --- a/packages/bub-acp-server/tests/test_e2e.py +++ b/packages/bub-acp-server/tests/test_e2e.py @@ -15,6 +15,7 @@ CreateTerminalResponse, DeclineElicitationResponse, ElicitationFormSessionMode, + Implementation, ReadTextFileResponse, TerminalOutputResponse, TextContentBlock, @@ -274,6 +275,7 @@ async def test_acknowledged_steering_starts_turn_over_extension_route( fs={"readTextFile": True, "writeTextFile": True}, terminal=True, ), + client_info=Implementation(name="lody", title="Lody", version="1"), ) session = await connection.new_session(cwd=str(workspace)) response = await connection.ext_method( @@ -293,6 +295,8 @@ async def test_acknowledged_steering_starts_turn_over_extension_route( assert process.returncode is None assert initialized.field_meta == {"steering": {"supported": True}} assert initialized.agent_capabilities is not None + # A Lody client also receives the session-title promise, which has no + # client-side capability to negotiate. assert initialized.agent_capabilities.field_meta == { "lody": { "steering": { @@ -300,7 +304,9 @@ async def test_acknowledged_steering_starts_turn_over_extension_route( "transport": "request", "upstreamTurn": "same", "configPolicy": "active", - } + }, + "subagentEvents": {"version": 1}, + "sessionTitle": {"version": 1}, } } assert response == {"outcome": "injected"} diff --git a/packages/bub-acp-server/tests/test_mcp.py b/packages/bub-acp-server/tests/test_mcp.py index e98b147..8b8390f 100644 --- a/packages/bub-acp-server/tests/test_mcp.py +++ b/packages/bub-acp-server/tests/test_mcp.py @@ -213,6 +213,9 @@ async def events(): monkeypatch.setattr(ModelRunner, "run", run) agent = make_agent(framework) + # Automatic session titles issue their own model call; this test asserts on + # the tool surface of prompt turns only. + monkeypatch.setattr(agent, "_schedule_session_title", lambda *a, **k: None) first = await agent.new_session( cwd=str(tmp_path), mcp_servers=[stdio("first", "first")] ) diff --git a/packages/bub-acp-server/tests/test_plugin.py b/packages/bub-acp-server/tests/test_plugin.py index 0637436..33e963d 100644 --- a/packages/bub-acp-server/tests/test_plugin.py +++ b/packages/bub-acp-server/tests/test_plugin.py @@ -269,16 +269,9 @@ async def test_initialize_advertises_session_capabilities(unstable: bool) -> Non response.agent_capabilities.session_capabilities.resume is not None ) is unstable assert response.agent_capabilities.load_session is True - assert response.agent_capabilities.field_meta == { - "lody": { - "steering": { - "version": 1, - "transport": "request", - "upstreamTurn": "same", - "configPolicy": "active", - } - } - } + # This connection did not identify itself as Lody, so no Lody-namespaced + # capability is reported at all. + assert response.agent_capabilities.field_meta is None assert response.field_meta == {"steering": {"supported": True}} diff --git a/packages/bub-acp-server/tests/test_session_title.py b/packages/bub-acp-server/tests/test_session_title.py new file mode 100644 index 0000000..c90330c --- /dev/null +++ b/packages/bub-acp-server/tests/test_session_title.py @@ -0,0 +1,399 @@ +"""Tests for automatic session titles (`_meta.lody.sessionTitle`).""" + +from __future__ import annotations + +import asyncio +from collections.abc import AsyncIterator +from pathlib import Path +from typing import Any + +import pytest +from acp.schema import Implementation, TextContentBlock +from bub.model_selection import ModelOptions +from bub.streaming import AsyncStreamEvents, StreamEvent, StreamState +from bub.turn import TurnResult + +from bub_acp_server.agent import BubACPAgent, _clean_title, is_lody_client + + +@pytest.fixture(autouse=True) +def isolated_bub_home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("BUB_HOME", str(tmp_path / ".bub")) + + +class FakeClient: + def __init__(self) -> None: + self.updates: list[tuple[str, object]] = [] + + async def session_update( + self, session_id: str, update: object, **kwargs: Any + ) -> None: + del kwargs + self.updates.append((session_id, update)) + + +class FakeFramework: + def get_agent_hooks(self): + return None + + def __init__(self) -> None: + self.workspace = Path.cwd() + self.messages: list[Any] = [] + + def bind_channel_router(self, router: object) -> None: + del router + + def get_tape_store(self) -> None: + return None + + async def quit_via_channel_router(self, session_id: str) -> None: + del session_id + + async def get_model_options( + self, *, session_id: str, workspace: Path + ) -> ModelOptions: + del session_id, workspace + return ModelOptions() + + async def process_inbound( + self, inbound: Any, stream_output: bool = False + ) -> TurnResult: + del stream_output + self.messages.append(inbound) + return TurnResult( + session_id=inbound.session_id, + prompt=inbound.content, + model_output="done", + ) + + +class TitleStreamAgent: + """Stand-in runtime agent that answers title prompts with fixed text.""" + + def __init__(self, text: str = "Fix login redirect") -> None: + self.text = text + self.calls: list[dict[str, Any]] = [] + + async def run_stream(self, **kwargs: Any) -> AsyncStreamEvents: + self.calls.append(kwargs) + + async def stream() -> AsyncIterator[StreamEvent]: + yield StreamEvent("text", {"delta": self.text}) + + return AsyncStreamEvents(stream(), state=StreamState()) + + +async def make_session(agent: BubACPAgent) -> Any: + """Create one ACP session and return its server-side record.""" + + response = await agent.new_session(cwd=str(Path.cwd())) + return agent._sessions[response.session_id] + + +async def wait_for_updates(client: FakeClient, name: str) -> list[Any]: + async with asyncio.timeout(1): + while not any(update.session_update == name for _, update in client.updates): + await asyncio.sleep(0) + return [update for _, update in client.updates if update.session_update == name] + + +def lody_info(**overrides: Any) -> Implementation: + fields: dict[str, Any] = { + "name": "lody", + "title": "Lody", + "version": "1", + } + return Implementation(**(fields | overrides)) + + +@pytest.mark.parametrize( + ("client_info", "expected"), + [ + (Implementation(name="lody", title="Lody", version="1"), True), + (Implementation(name="Lody", title="Lody", version="1"), True), + (Implementation(name="lody-cli", version="0.5.0"), True), + (Implementation(name="LODY-desktop", version="1"), True), + (Implementation(name="lody", version="1"), True), + (Implementation(name="zed", version="0.200.0"), False), + (Implementation(name="Zed", title="Zed", version="1"), False), + ( + Implementation(name="claude-code", title="Claude Code", version="2.0.0"), + False, + ), + (Implementation(name="not-lody", version="1"), False), + (Implementation(name="mylody", version="1"), False), + (None, False), + ], +) +def test_is_lody_client(client_info: Implementation | None, expected: bool) -> None: + assert is_lody_client(client_info) is expected + + +@pytest.mark.asyncio +async def test_session_titles_are_advertised_only_to_lody_clients() -> None: + """The capability is a promise: it only goes to clients that can keep it.""" + + for client_info, expected in ( + (lody_info(), True), + (Implementation(name="zed", title="Zed", version="1"), False), + (None, False), + ): + agent = BubACPAgent(FakeFramework()) + response = await agent.initialize(protocol_version=1, client_info=client_info) + assert response.agent_capabilities is not None + assert agent._lody_client is expected + if not expected: + # Nothing Lody-namespaced goes to a non-Lody client, including + # capabilities that are otherwise negotiated per-connection. + assert response.agent_capabilities.field_meta is None + continue + assert response.agent_capabilities.field_meta is not None + assert response.agent_capabilities.field_meta == { + "lody": { + "steering": { + "version": 1, + "transport": "request", + "upstreamTurn": "same", + "configPolicy": "active", + }, + "subagentEvents": {"version": 1}, + "sessionTitle": {"version": 1}, + } + } + + +@pytest.mark.asyncio +async def test_non_lody_clients_never_generate_titles( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls: list[str] = [] + + async def generate(self: BubACPAgent, session: Any, request: str) -> str: + del self, session + calls.append(request) + return "Unexpected" + + monkeypatch.setattr(BubACPAgent, "_generate_session_title", generate) + client = FakeClient() + agent = BubACPAgent(FakeFramework()) + agent.on_connect(client) + await agent.initialize( + protocol_version=1, + client_info=Implementation(name="zed", title="Zed", version="1"), + ) + session = await make_session(agent) + + await agent.prompt( + [TextContentBlock(type="text", text="please fix the login redirect")], + session.session_id, + ) + await asyncio.sleep(0) + + assert calls == [] + assert session.title is None + assert [ + update + for _, update in client.updates + if update.session_update == "session_info_update" + ] == [] + + +@pytest.mark.asyncio +async def test_first_prompt_publishes_and_persists_a_generated_title( + monkeypatch: pytest.MonkeyPatch, +) -> None: + seen: list[str] = [] + + async def generate(self: BubACPAgent, session: Any, request: str) -> str: + del self, session + seen.append(request) + return '"Fix the login redirect."' + + monkeypatch.setattr(BubACPAgent, "_generate_session_title", generate) + framework = FakeFramework() + client = FakeClient() + agent = BubACPAgent(framework) + agent.on_connect(client) + await agent.initialize(protocol_version=1, client_info=lody_info()) + session = await make_session(agent) + + await agent.prompt( + [TextContentBlock(type="text", text="please fix the login redirect")], + session.session_id, + ) + + pushes = await wait_for_updates(client, "session_info_update") + assert len(pushes) == 1 + assert pushes[0].title == "Fix the login redirect." + assert pushes[0].field_meta == {"lody": {"titleSource": "generated"}} + assert seen == ["please fix the login redirect"] + + fresh = BubACPAgent(framework) + listed = (await fresh.list_sessions()).sessions + assert [item.title for item in listed] == ["Fix the login redirect."] + + +@pytest.mark.asyncio +async def test_titles_are_generated_once_per_session( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls: list[str] = [] + + async def generate(self: BubACPAgent, session: Any, request: str) -> str: + del self, session + calls.append(request) + return "One title" + + monkeypatch.setattr(BubACPAgent, "_generate_session_title", generate) + client = FakeClient() + agent = BubACPAgent(FakeFramework()) + agent.on_connect(client) + await agent.initialize(protocol_version=1, client_info=lody_info()) + session = await make_session(agent) + + for text in ("first prompt", "second prompt"): + await agent.prompt( + [TextContentBlock(type="text", text=text)], session.session_id + ) + + await wait_for_updates(client, "session_info_update") + assert calls == ["first prompt"] + assert session.title == "One title" + + +@pytest.mark.asyncio +async def test_loaded_sessions_keep_their_title( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls: list[str] = [] + + async def generate(self: BubACPAgent, session: Any, request: str) -> str: + del self, session + calls.append(request) + return "Regenerated" + + monkeypatch.setattr(BubACPAgent, "_generate_session_title", generate) + client = FakeClient() + agent = BubACPAgent(FakeFramework()) + agent.on_connect(client) + await agent.initialize(protocol_version=1, client_info=lody_info()) + session = await make_session(agent) + session.title = "Existing title" + agent._save_sessions() + + await agent.prompt( + [TextContentBlock(type="text", text="hello")], session.session_id + ) + + assert calls == [] + assert [ + update + for _, update in client.updates + if update.session_update == "session_info_update" + ] == [] + assert session.title == "Existing title" + + +@pytest.mark.asyncio +async def test_title_generation_failure_pushes_nothing( + monkeypatch: pytest.MonkeyPatch, +) -> None: + async def generate(self: BubACPAgent, session: Any, request: str) -> str: + del self, session, request + raise RuntimeError("model unavailable") + + monkeypatch.setattr(BubACPAgent, "_generate_session_title", generate) + client = FakeClient() + agent = BubACPAgent(FakeFramework()) + agent.on_connect(client) + await agent.initialize(protocol_version=1, client_info=lody_info()) + session = await make_session(agent) + + await agent.prompt( + [TextContentBlock(type="text", text="hello")], session.session_id + ) + await asyncio.sleep(0) + + assert [ + update + for _, update in client.updates + if update.session_update == "session_info_update" + ] == [] + assert session.title is None + + +@pytest.mark.asyncio +async def test_empty_generation_pushes_nothing( + monkeypatch: pytest.MonkeyPatch, +) -> None: + async def generate(self: BubACPAgent, session: Any, request: str) -> str: + del self, session, request + return " " + + monkeypatch.setattr(BubACPAgent, "_generate_session_title", generate) + client = FakeClient() + agent = BubACPAgent(FakeFramework()) + agent.on_connect(client) + await agent.initialize(protocol_version=1, client_info=lody_info()) + session = await make_session(agent) + + await agent.prompt( + [TextContentBlock(type="text", text="hello")], session.session_id + ) + await asyncio.sleep(0) + + assert session.title is None + assert [ + update + for _, update in client.updates + if update.session_update == "session_info_update" + ] == [] + + +@pytest.mark.asyncio +async def test_generate_session_title_uses_a_lean_throwaway_run( + monkeypatch: pytest.MonkeyPatch, +) -> None: + agent = BubACPAgent(FakeFramework()) + session = await make_session(agent) + session.runtime["model"] = "provider:model" + stub = TitleStreamAgent() + agent._runtime_agents[session.session_id] = stub + + title = await agent._generate_session_title(session, "fix the login redirect") + + assert title == "Fix login redirect" + call = stub.calls[0] + assert call["session_id"].startswith("temp/") + assert call["session_id"] != session.session_id + assert call["model"] == "provider:model" + assert call["allowed_tools"] == set() + assert call["allowed_skills"] == set() + assert "fix the login redirect" in call["prompt"] + assert call["state"]["session_id"] == call["session_id"] + assert call["state"]["_runtime_workspace"] == str(session.cwd) + + +@pytest.mark.asyncio +async def test_missing_runtime_agent_skips_generation() -> None: + agent = BubACPAgent(FakeFramework()) + session = await make_session(agent) + + assert await agent._generate_session_title(session, "hello") == "" + + +@pytest.mark.parametrize( + ("raw", "expected"), + [ + ("Fix login redirect", "Fix login redirect"), + ('"Fix login redirect"', "Fix login redirect"), + ("“Fix login redirect”", "Fix login redirect"), + ("Title: Fix login redirect", "Fix login redirect"), + (" Fix login\nredirect ", "Fix login redirect"), + ("", ""), + (" ", ""), + ("x" * 200, "x" * 80), + ], +) +def test_clean_title(raw: str, expected: str) -> None: + assert _clean_title(raw) == expected diff --git a/packages/bub-acp-server/tests/test_subagents.py b/packages/bub-acp-server/tests/test_subagents.py new file mode 100644 index 0000000..1e1f264 --- /dev/null +++ b/packages/bub-acp-server/tests/test_subagents.py @@ -0,0 +1,429 @@ +"""Tests for the Lody subagent event extension (`_lody/subagents/event`).""" + +from __future__ import annotations + +from collections.abc import AsyncIterator, Iterator +from pathlib import Path +from typing import Any + +import bub.builtin.tools # noqa: F401 - importing the builtin tools fills the registry +import pytest +from acp.schema import ClientCapabilities, TextContentBlock +from bub.model_selection import ModelOptions +from bub.streaming import AsyncStreamEvents, StreamEvent, StreamState +from bub.tools import REGISTRY, ToolContext + +from bub_acp_server.agent import BubACPAgent +from bub_acp_server.subagents import ( + LODY_SUBAGENT_EVENT_METHOD, + client_supports_subagent_events, +) + + +@pytest.fixture(autouse=True) +def isolated_bub_home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("BUB_HOME", str(tmp_path / ".bub")) + + +class FakeClient: + def __init__(self) -> None: + self.updates: list[tuple[str, object]] = [] + self.notifications: list[tuple[str, dict[str, Any]]] = [] + + async def session_update( + self, session_id: str, update: object, **kwargs: Any + ) -> None: + del kwargs + self.updates.append((session_id, update)) + + async def ext_notification(self, method: str, params: dict[str, Any]) -> None: + self.notifications.append((method, params)) + + +class BrokenClient(FakeClient): + async def ext_notification(self, method: str, params: dict[str, Any]) -> None: + raise RuntimeError(f"cannot deliver {method}") + + +class FakeFramework: + def get_agent_hooks(self): + return None + + def __init__(self) -> None: + self.workspace = Path.cwd() + + def bind_channel_router(self, router: object) -> None: + del router + + def get_tape_store(self) -> None: + return None + + async def get_model_options( + self, *, session_id: str, workspace: Path + ) -> ModelOptions: + del session_id, workspace + return ModelOptions() + + +class StubAgent: + """Stand-in for Bub's agent that replays a canned child stream.""" + + def __init__(self, events: list[StreamEvent], *, error: Exception | None = None): + self.events = events + self.error = error + self.calls: list[dict[str, Any]] = [] + self.tools = {"subagent": REGISTRY["subagent"], "bash": REGISTRY["bash"]} + + async def run_stream(self, **kwargs: Any) -> AsyncStreamEvents: + self.calls.append(kwargs) + if self.error is not None: + raise self.error + + async def stream() -> AsyncIterator[StreamEvent]: + for event in self.events: + yield event + + return AsyncStreamEvents(stream(), state=StreamState()) + + +def child_events() -> list[StreamEvent]: + return [ + StreamEvent("text", {"delta": "work"}), + StreamEvent("reasoning", {"delta": "thinking"}), + StreamEvent( + "tool_call", + { + "tool_calls": [ + { + "id": "call-1", + "type": "function", + "function": { + "name": "bash", + "arguments": '{"command":"ls"}', + }, + } + ] + }, + ), + StreamEvent("tool_result", {"tool_results": ["file.txt"]}), + StreamEvent("final", {"text": "work", "ok": True}), + StreamEvent("text", {"delta": " done"}), + StreamEvent("usage", {"usage": {"total_tokens": 42}}), + ] + + +async def make_tool_session( + client: FakeClient, + events: list[StreamEvent], + *, + error: Exception | None = None, + negotiate: bool = True, + session: str = "temp", +) -> tuple[BubACPAgent, StubAgent, Any]: + """Install the streaming subagent tool for one session and return it.""" + + framework = FakeFramework() + agent = BubACPAgent(framework) + agent.on_connect(client) + capabilities = ( + ClientCapabilities(field_meta={"lody": {"subagentEvents": {"version": 1}}}) + if negotiate + else ClientCapabilities() + ) + await agent.initialize(protocol_version=1, client_capabilities=capabilities) + created = await agent.new_session(cwd=str(Path.cwd())) + version = {"session": session, "prompt": "delegate this", "model": None} + + runtime_agent = StubAgent(events, error=error) + acp_session = agent._sessions[created.session_id] + if negotiate: + agent._install_streaming_subagent(runtime_agent, acp_session) + context = ToolContext( + tape=None, + state={ + "_runtime_agent": runtime_agent, + "session_id": f"acp-server:{acp_session.session_id}", + }, + ) + return agent, runtime_agent, (runtime_agent.tools["subagent"], context, version) + + +def payloads(client: FakeClient) -> Iterator[dict[str, Any]]: + return (params for _, params in client.notifications) + + +def test_client_negotiation_requires_matching_version() -> None: + assert client_supports_subagent_events(None) is False + assert client_supports_subagent_events(ClientCapabilities()) is False + assert ( + client_supports_subagent_events( + ClientCapabilities(field_meta={"lody": {"subagentEvents": {}}}) + ) + is False + ) + assert ( + client_supports_subagent_events( + ClientCapabilities(field_meta={"lody": {"subagentEvents": {"version": 2}}}) + ) + is False + ) + assert ( + client_supports_subagent_events( + ClientCapabilities(field_meta={"lody": {"subagentEvents": {"version": 1}}}) + ) + is True + ) + # Unnegotiated clients never reach the tool, but a raw mapping is equivalent. + assert ( + client_supports_subagent_events( + {"_meta": {"lody": {"subagentEvents": {"version": 1}}}} + ) + is True + ) + + +def assert_valid_event(payload: dict[str, Any]) -> None: + """Enforce the wire rules of `isLodySubagentEvent` from acp-extension-core.""" + + _assert_no_nulls(payload) + assert payload["version"] == 1 + assert isinstance(payload["sessionId"], str) and payload["sessionId"] + assert isinstance(payload["runId"], str) and payload["runId"] + kind = payload["type"] + if kind == "snapshot": + snapshot = payload["snapshot"] + assert snapshot["state"] in { + "pending", + "running", + "completed", + "failed", + "cancelled", + "unknown", + } + support = snapshot["support"] + assert set(support["stream"]) <= {"text", "thought", "tool", "plan"} + assert support["progress"] is True + assert support["cancel"] is False + assert support["outputRead"] in {"none", "live_tail", "final_tail"} + elif kind == "progress": + assert isinstance(payload["progress"], dict) + elif kind == "output": + update = payload["update"] + variant = update["sessionUpdate"] + assert variant in { + "agent_message_chunk", + "agent_thought_chunk", + "tool_call", + "tool_call_update", + "plan", + } + if variant == "tool_call": + assert update["toolCallId"] + assert update["title"] + assert update["status"] in {"pending", "in_progress", "completed", "failed"} + if variant in {"agent_message_chunk", "agent_thought_chunk"}: + assert update["content"]["type"] == "text" + else: # pragma: no cover - guards the test helper itself + raise AssertionError(f"unknown event type {kind}") + + +def _assert_no_nulls(value: object, path: str = "$") -> None: + if value is None: + raise AssertionError(f"explicit null at {path}") + if isinstance(value, dict): + for key, item in value.items(): + _assert_no_nulls(item, f"{path}.{key}") + elif isinstance(value, list): + for index, item in enumerate(value): + _assert_no_nulls(item, f"{path}[{index}]") + + +@pytest.mark.asyncio +async def test_subagent_run_publishes_snapshot_output_and_progress() -> None: + client = FakeClient() + _, runtime_agent, (tool, context, version) = await make_tool_session( + client, child_events() + ) + + result = await tool.run(context=context, **version) + + assert client.notifications + assert {method for method, _ in client.notifications} == { + LODY_SUBAGENT_EVENT_METHOD + } + events = list(payloads(client)) + for payload in events: + assert_valid_event(payload) + + assert events[0]["sessionId"] == context.state["session_id"].split(":", 1)[1] + assert len({payload["runId"] for payload in events}) == 1 + + assert events[0]["type"] == "snapshot" + assert events[0]["snapshot"]["state"] == "running" + assert events[0]["snapshot"]["description"] == "delegate this" + assert events[0]["snapshot"]["support"]["stream"] == ["text", "thought", "tool"] + + assert [event["type"] for event in events] == [ + "snapshot", + "output", + "output", + "output", + "progress", + "output", + "output", + "progress", + "snapshot", + ] + assert events[1]["update"] == { + "sessionUpdate": "agent_message_chunk", + "content": {"type": "text", "text": "work"}, + } + assert events[2]["update"]["sessionUpdate"] == "agent_thought_chunk" + assert events[3]["update"]["sessionUpdate"] == "tool_call" + assert events[3]["update"]["toolCallId"] == "call-1" + assert events[3]["update"]["title"] == "ls" + assert events[3]["update"]["kind"] == "execute" + assert events[3]["update"]["rawInput"] == {"command": "ls"} + assert events[4]["progress"]["lastToolName"] == "bash" + assert events[4]["progress"]["toolCallCount"] == 1 + assert events[5]["update"]["sessionUpdate"] == "tool_call_update" + assert events[5]["update"]["toolCallId"] == "call-1" + assert events[5]["update"]["status"] == "completed" + assert events[5]["update"]["content"][0]["content"]["text"] == "file.txt" + assert events[6]["update"]["content"]["text"] == " done" + assert events[7]["progress"] == { + "lastToolName": "bash", + "turnCount": 1, + "toolCallCount": 1, + "durationMs": events[7]["progress"]["durationMs"], + "totalTokens": 42, + } + assert events[8]["snapshot"]["state"] == "completed" + assert events[8]["snapshot"]["summary"] == "work done" + + # The tool still returns Bub's structured result to the model. + assert result["output"] == "work done" + assert result["errors"] == [] + assert result["session_id"].startswith("temp/") + assert runtime_agent.calls[0]["session_id"] == result["session_id"] + assert runtime_agent.calls[0]["prompt"] == "delegate this" + + +@pytest.mark.asyncio +async def test_subagent_inherit_session_reuses_parent_session() -> None: + client = FakeClient() + _, runtime_agent, (tool, context, version) = await make_tool_session( + client, [StreamEvent("text", {"delta": "ok"})], session="inherit" + ) + + result = await tool.run(context=context, **version) + + assert result["session_id"] == context.state["session_id"] + assert runtime_agent.calls[0]["session_id"] == context.state["session_id"] + + +@pytest.mark.asyncio +async def test_subagent_failure_reports_terminal_snapshot_and_raises() -> None: + client = FakeClient() + _, _, (tool, context, version) = await make_tool_session( + client, [], error=RuntimeError("child exploded") + ) + + with pytest.raises(RuntimeError, match="child exploded"): + await tool.run(context=context, **version) + + events = list(payloads(client)) + for payload in events: + assert_valid_event(payload) + assert [event["type"] for event in events] == ["snapshot", "snapshot"] + assert events[-1]["snapshot"]["state"] == "failed" + assert events[-1]["snapshot"]["reason"] == { + "code": "error", + "message": "child exploded", + } + + +@pytest.mark.asyncio +async def test_subagent_stream_errors_are_collected_without_failing_the_run() -> None: + client = FakeClient() + events = [ + StreamEvent("error", {"message": "tool blew up"}), + StreamEvent("text", {"delta": "partial"}), + ] + _, _, (tool, context, version) = await make_tool_session(client, events) + + result = await tool.run(context=context, **version) + + assert result["errors"] == ["tool blew up"] + assert result["output"] == "partial" + assert list(payloads(client))[-1]["snapshot"]["state"] == "completed" + + +@pytest.mark.asyncio +async def test_broken_client_does_not_fail_the_tool() -> None: + client = BrokenClient() + _, _, (tool, context, version) = await make_tool_session( + client, [StreamEvent("text", {"delta": "ok"})] + ) + + result = await tool.run(context=context, **version) + + assert result["output"] == "ok" + + +@pytest.mark.asyncio +async def test_subagent_tool_is_untouched_without_negotiation() -> None: + client = FakeClient() + _, _, (tool, context, version) = await make_tool_session( + client, child_events(), negotiate=False + ) + + assert tool.handler is REGISTRY["subagent"].handler + assert client.notifications == [] + assert context.state["_runtime_agent"].calls == [] + + +@pytest.mark.asyncio +async def test_negotiated_session_replaces_only_this_session_tool() -> None: + client = FakeClient() + framework = FakeFramework() + agent = BubACPAgent(framework) + agent.on_connect(client) + negotiated = await agent.initialize( + protocol_version=1, + client_capabilities=ClientCapabilities( + field_meta={"lody": {"subagentEvents": {"version": 1}}} + ), + ) + assert negotiated.agent_capabilities is not None + # This connection negotiated subagent events but never identified itself as + # Lody, so nothing Lody-namespaced is reported. The client's own opt-in is + # still honored: the contract is bilateral and does not require the agent to + # have advertised first. + assert negotiated.agent_capabilities.field_meta is None + + session = await agent.new_session(cwd=str(Path.cwd())) + inbound = agent._build_inbound( + [TextContentBlock(type="text", text="hi")], agent._sessions[session.session_id] + ) + installed = inbound._runtime_agent.tools["subagent"] + assert installed.name == REGISTRY["subagent"].name + assert installed.description == REGISTRY["subagent"].description + assert installed.parameters == REGISTRY["subagent"].parameters + assert installed.renderer is REGISTRY["subagent"].renderer + assert installed.preserve == REGISTRY["subagent"].preserve + assert installed.output_schema == REGISTRY["subagent"].output_schema + assert installed.handler is not REGISTRY["subagent"].handler + assert REGISTRY["subagent"].handler is not installed.handler + + plain = BubACPAgent(FakeFramework()) + plain.on_connect(FakeClient()) + await plain.initialize(protocol_version=1, client_capabilities=ClientCapabilities()) + other = await plain.new_session(cwd=str(Path.cwd())) + other_inbound = plain._build_inbound( + [TextContentBlock(type="text", text="hi")], plain._sessions[other.session_id] + ) + assert ( + other_inbound._runtime_agent.tools["subagent"].handler + is REGISTRY["subagent"].handler + )