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