diff --git a/cecli/args.py b/cecli/args.py index 5ee878775f1..88ba7e9d3e3 100644 --- a/cecli/args.py +++ b/cecli/args.py @@ -1313,6 +1313,24 @@ def get_parser(default_config_files, git_root): help="Specify server configuration (WebSocket and headless) as a JSON/YAML string", default=None, ) + group.add_argument( + "--acp", + action="store_true", + default=False, + help=( + "Run as an Agent Client Protocol (ACP) agent over stdio, for editors such as Zed" + " (stdout carries the protocol)" + ), + ) + group.add_argument( + "--acp-models", + action="append", + metavar="PATTERN", + help=( + "Limit the ACP model picker's provider lists to models matching this pattern," + " matched like /models (can be used multiple times)" + ), + ) ########## group = parser.add_argument_group("Deprecated model settings") diff --git a/cecli/coders/agent_coder.py b/cecli/coders/agent_coder.py index 45f067f1edd..47f2e13d5e9 100644 --- a/cecli/coders/agent_coder.py +++ b/cecli/coders/agent_coder.py @@ -842,6 +842,7 @@ async def _execute_local_tools(self, tool_calls_list): for tool_call in tool_calls_list: tool_name = tool_call.function.name result_message = "" + self.io.tool_call_event(tool_call.id, "in_progress") try: if responses.unprefix_tool_name(tool_name)[1].lower() in self.write_tools: used_write_tool = True diff --git a/cecli/coders/base_coder.py b/cecli/coders/base_coder.py index 5372fd2f98a..7c29458398c 100755 --- a/cecli/coders/base_coder.py +++ b/cecli/coders/base_coder.py @@ -3284,6 +3284,7 @@ async def _execute_mcp_tools(self, server, tool_calls): continue tool_id_set.add(tool_call.id) + self.io.tool_call_event(tool_call.id, "in_progress") try: # Arguments can be a stream of JSON objects. @@ -3490,6 +3491,11 @@ async def process_tool_calls(self, tool_call_response): for server, server_responses in tool_responses_by_server.items(): for tool_response in server_responses: tool_responses.append(tool_response) + self.io.tool_call_event( + tool_response["tool_call_id"], + "completed", + result=tool_response.get("content"), + ) ConversationService.get_manager(self).add_message( message_dict=tool_response, @@ -3507,6 +3513,12 @@ def _print_tool_call_info(self, server_tool_calls): for server, tool_calls in server_tool_calls.items(): for tool_call in tool_calls: + self.io.tool_call_event( + tool_call.id, + "pending", + name=tool_call.function.name, + arguments=tool_call.function.arguments, + ) try: if ToolRegistry.get_tool(tool_call.function.name.lower()): ToolRegistry.get_tool(tool_call.function.name.lower()).format_output( @@ -4828,7 +4840,7 @@ def show_usage_report(self): if self.tui and self.tui(): self.tui().update_cost(self.usage_report.replace("\n", " ")) else: - self.io.tool_output(self.usage_report) + self.io.tool_output(self.usage_report, type="usage-report") self.io.rule() self.message_cost = 0.0 diff --git a/cecli/helpers/agents/service.py b/cecli/helpers/agents/service.py index 0dae1b08619..9d31c1e82e7 100644 --- a/cecli/helpers/agents/service.py +++ b/cecli/helpers/agents/service.py @@ -41,6 +41,25 @@ AGENT_INHERIT_PREFIXES = ("ws:", "wf:") +def _subagent_event(coder: Any, agent_uuid: Any, event: str, **kwargs: Any) -> None: + """Report a sub-agent's lifecycle to the IO (see ``InputOutput.subagent_event``).""" + io = getattr(coder, "io", None) + if io is None: + return + try: + io.subagent_event(str(agent_uuid), event, **kwargs) + except Exception: + logger.warning("Failed to report sub-agent event %r", event, exc_info=True) + + +def _was_cancelled(info: "SubAgentInfo") -> bool: + """Whether the current run was cancelled, even if the coder absorbed it as a ^C.""" + if info.cancel_requested: + return True + cancelling = getattr(asyncio.current_task(), "cancelling", None) # Python 3.11+ + return bool(cancelling and cancelling()) + + class SubAgentStatus(Enum): """Status of a sub-agent.""" @@ -66,6 +85,7 @@ class SubAgentInfo: auto_reap: bool = True # If True, agent may be automatically reaped when FINISHED independent: bool = False detached: bool = False + cancel_requested: bool = False # the current run was cancelled (see _was_cancelled) class AgentService: @@ -522,6 +542,8 @@ def _cleanup_sub_agent(self, agent_uuid: str) -> None: except (AttributeError, RuntimeError): logger.warning("Failed to notify TUI to remove sub-agent container", exc_info=True) + _subagent_event(self.coder, info.coder.uuid, "removed") + # Cancel any tracked generate task to avoid floating tasks if info.generate_task is not None and not info.generate_task.done(): info.generate_task.cancel() @@ -733,6 +755,16 @@ async def _create_sub_agent_coder( except Exception: logger.warning("Failed to notify TUI to create sub-agent container", exc_info=True) + _subagent_event( + parent_coder, + new_coder.uuid, + "created", + parent_uuid=str(parent_coder.uuid), + name=name, + description=metadata.get("description") if isinstance(metadata, dict) else None, + root=str(getattr(new_coder, "root", "") or ""), + ) + # Initialize system prompt from config system_prompt = getattr(config, "prompt", "") from cecli.helpers.conversation.service import ConversationService @@ -807,8 +839,14 @@ def start_generate_task(self, info: SubAgentInfo, user_message: str) -> asyncio. async def _run_generate(): info.status = SubAgentStatus.RUNNING + info.cancel_requested = False + _subagent_event(info.coder, info.coder.uuid, "running") + stop_reason, error = "end_turn", None try: await info.coder.generate(user_message=user_message, preproc=True) + if _was_cancelled(info): + stop_reason = "cancelled" + info.summary = info.summary or DEFAULT_SUMMARY_INTERRUPTED await self._inject_sub_agent_result(info) if info.status == SubAgentStatus.RUNNING: info.status = SubAgentStatus.FINISHED @@ -831,12 +869,14 @@ async def _cleanup(): fire_and_forget(_cleanup()) except asyncio.CancelledError: + stop_reason = "cancelled" await self._inject_sub_agent_result(info) info.status = SubAgentStatus.FINISHED info.summary = info.summary or DEFAULT_SUMMARY_INTERRUPTED logger.debug("Sub-agent %s generate cancelled (interrupted)", info.name) raise except Exception as exc: + stop_reason, error = "error", str(exc) or exc.__class__.__name__ await self._inject_sub_agent_result(info) info.status = SubAgentStatus.ERROR info.error = str(exc) @@ -849,6 +889,10 @@ async def _cleanup(): raise except SwitchCoderSignal: raise + finally: + _subagent_event( + info.coder, info.coder.uuid, "idle", stop_reason=stop_reason, error=error + ) # Cancel any previous generate task to prevent duplicate concurrent generates if info.generate_task is not None and not info.generate_task.done(): @@ -986,6 +1030,10 @@ async def _inject_sub_agent_result(self, info: SubAgentInfo) -> None: force=True, ) + _subagent_event( + parent_coder, parent_coder.uuid, "message", sender_uuid=coder_uuid, text=user_content + ) + async def invoke( self, name: str, @@ -1011,8 +1059,20 @@ async def invoke( return None # Blocking: run the sub-agent with the prompt using start_generate_task + sender = parent if parent is not None else self.coder + _subagent_event( + sender, new_coder.uuid, "message", sender_uuid=str(sender.uuid), text=prompt + ) task = self.start_generate_task(info, prompt) - await task + try: + await asyncio.shield(task) + except asyncio.CancelledError: + if not task.done(): + # The caller was cancelled: stop the sub-agent it was waiting for + info.cancel_requested = True + task.cancel() + raise + # Only the sub-agent was cancelled; its summary says it was interrupted return info.summary async def spawn( @@ -1040,6 +1100,10 @@ async def spawn( name, auto_reap=auto_reap, parent=parent, independent=independent ) if prompt: + sender = parent if parent is not None else self.coder + _subagent_event( + sender, new_coder.uuid, "message", sender_uuid=str(sender.uuid), text=prompt + ) self.start_generate_task(info, prompt) return new_coder, info @@ -1079,6 +1143,19 @@ async def wait(self, parent: Any) -> List[str]: return [info.summary for info in children] + def cancel_sub_agent(self, agent_uuid: str) -> bool: + """Stop a sub-agent's current work, leaving its parent running. + + A parent waiting on it (a blocking delegation) gets the interrupted + summary. Returns False if the sub-agent was not working. + """ + info = self.sub_agents.get(str(agent_uuid)) + if info is None or info.generate_task is None or info.generate_task.done(): + return False + info.cancel_requested = True + info.generate_task.cancel() + return True + def get_active_agents(self) -> List[Dict[str, Any]]: """Return list of active sub-agents for display.""" return [ diff --git a/cecli/helpers/io_proxy.py b/cecli/helpers/io_proxy.py index f6963082102..da7a430412f 100644 --- a/cecli/helpers/io_proxy.py +++ b/cecli/helpers/io_proxy.py @@ -126,6 +126,12 @@ def assistant_output(self, message: str = "", pretty: Any = None, **kwargs: Any) kwargs["coder_uuid"] = self._coder_uuid return self._target.assistant_output(message=message, pretty=pretty, **kwargs) + def tool_call_event(self, tool_call_id: str, status: str, **kwargs: Any) -> Any: + """Forward tool_call_event with coder_uuid injected.""" + if "coder_uuid" not in kwargs: + kwargs["coder_uuid"] = self._coder_uuid + return self._target.tool_call_event(tool_call_id, status, **kwargs) + def reset_streaming_response(self, **kwargs) -> Any: """Forward reset_streaming_response with coder_uuid injected.""" if "coder_uuid" not in kwargs: diff --git a/cecli/helpers/onboarding/providers.py b/cecli/helpers/onboarding/providers.py index 5cd27099c5b..b37dea0913d 100644 --- a/cecli/helpers/onboarding/providers.py +++ b/cecli/helpers/onboarding/providers.py @@ -3,7 +3,7 @@ import importlib.resources as importlib_resources import json import os -from typing import Dict, List +from typing import Collection, Dict, List, Optional # Providers that authenticate through mechanisms other than a simple _API_KEY # (e.g. GitHub Copilot tokens or AWS credentials), so onboarding should never @@ -61,12 +61,19 @@ def _load_metadata() -> Dict: return data -def _models_from_metadata(data: Dict, slug: str) -> List[str]: - """Return selectable model names for a provider from the metadata dict.""" +def _models_from_metadata( + data: Dict, slug: str, modes: Optional[Collection[str]] = None +) -> List[str]: + """Return selectable model names for a provider from the metadata dict. + + ``modes`` keeps only models with those metadata modes (e.g. ``("chat",)``). + """ models = [] for name, meta in data.items(): if not isinstance(meta, dict): continue + if modes and meta.get("mode", "chat") not in modes: + continue provider = meta.get("litellm_provider") or "" if provider == slug: models.append(name if name.startswith(slug + "/") else f"{slug}/{name}") @@ -128,9 +135,12 @@ def provider_display(entry: Dict) -> str: return label -def get_models_for_provider(slug: str) -> List[str]: - """Return the selectable model names for a provider.""" - models = _models_from_metadata(_load_metadata(), slug) +def get_models_for_provider(slug: str, modes: Optional[Collection[str]] = None) -> List[str]: + """Return the selectable model names for a provider. + + ``modes`` filters the bundled metadata by mode; a live fetch is not filtered. + """ + models = _models_from_metadata(_load_metadata(), slug, modes) if models: return sorted(set(models)) diff --git a/cecli/helpers/server/acp/agent/__init__.py b/cecli/helpers/server/acp/agent/__init__.py new file mode 100644 index 00000000000..96086b12b20 --- /dev/null +++ b/cecli/helpers/server/acp/agent/__init__.py @@ -0,0 +1,30 @@ +"""``cecli --acp``: cecli as an Agent Client Protocol (ACP) v1 agent. + +ACP clients such as Zed launch cecli as a subprocess and drive it with +newline-delimited JSON-RPC 2.0 on stdin/stdout: +https://agentclientprotocol.com/protocol/v1/schema + +- ``agent``: AcpAgent, which answers the client and runs each prompt as a turn +- ``transport``: the channel to the client (stdio) +- ``io``: AcpInputOutput, which routes cecli's input and output through the agent +- ``updates``: cecli's output as ``session/update`` notifications +- ``config``: the mode, model and effort pickers +- ``mcp``: MCP servers passed by the client + +The WebSocket bridge in the parent package is a separate feature, which streams +cecli's events to dashboards. +""" + +from __future__ import annotations + +from cecli.helpers.server.acp.agent.agent import AcpAgent +from cecli.helpers.server.acp.agent.transport import StdioTransport + +_agent: AcpAgent | None = None + + +def get_agent() -> AcpAgent: + global _agent + if _agent is None: + _agent = AcpAgent(StdioTransport()) + return _agent diff --git a/cecli/helpers/server/acp/agent/agent.py b/cecli/helpers/server/acp/agent/agent.py new file mode 100644 index 00000000000..9fa63382a2c --- /dev/null +++ b/cecli/helpers/server/acp/agent/agent.py @@ -0,0 +1,760 @@ +"""AcpAgent: the agent side of the Agent Client Protocol (ACP) v1. + +One process serves one project. ``bootstrap()`` runs before ``main_async`` so +the first ``session/new`` can pick the working directory the coder is built in; +its response is deferred until the coder first asks for input. From then on the +coder's normal linear run loop drives each turn: a turn starts when +``next_input`` hands the loop a prompt and ends when the loop asks again. +""" + +from __future__ import annotations + +import asyncio +import os +import sys +import uuid +import weakref +from dataclasses import dataclass, field +from typing import Any, Callable +from urllib.parse import unquote, urlparse + +from cecli import __version__ +from cecli.helpers.server.acp.agent.config import SessionConfig, current_value +from cecli.helpers.server.acp.agent.mcp import ClientMcpServers +from cecli.helpers.server.acp.agent.transport import ( + EOF, + INTERNAL_ERROR, + INVALID_PARAMS, + METHOD_NOT_FOUND, + Transport, + error_message, + log, + result_message, +) +from cecli.helpers.server.acp.agent.updates import CONSOLE_FENCE, UpdateEmitter + +PROTOCOL_VERSION = 1 + +# Commands that make no sense when an editor owns the UI +HIDDEN_COMMANDS = { + "edit", + "editor", + "exit", + "hot-reload", + "multiline-mode", + "quit", + "terminal-setup", + "voice", +} + +_DEFERRED = object() + + +class AcpError(Exception): + def __init__(self, code: int, message: str): + super().__init__(message) + self.code = code + self.message = message + + +# ── Prompt conversion ────────────────────────────────────────── + + +def _file_path(uri: str | None) -> str | None: + if not uri or not uri.startswith("file://"): + return None + path = unquote(urlparse(uri).path) + if os.name == "nt" and len(path) > 2 and path[0] == "/" and path[2] == ":": + path = path[1:] + return path + + +def _display_path(path: str, root: str) -> str: + try: + rel = os.path.relpath(path, root) + except ValueError: # different drive + return path + return path if rel.startswith("..") else rel + + +def prompt_to_text(blocks: list, root: str) -> tuple[str, list[str]]: + """Flatten ACP content blocks into cecli input text plus mentioned file paths.""" + parts: list[str] = [] + paths: list[str] = [] + for block in blocks or []: + if not isinstance(block, dict): + continue + kind = block.get("type") + if kind == "text": + parts.append(block.get("text") or "") + elif kind == "resource_link": + path = _file_path(block.get("uri")) + if path: + paths.append(path) + parts.append(_display_path(path, root)) + else: + parts.append(block.get("uri") or block.get("name") or "") + elif kind == "resource": + resource = block.get("resource") or {} + uri = resource.get("uri") or "" + path = _file_path(uri) + if path: + paths.append(path) + label = _display_path(path, root) if path else uri + text = resource.get("text") + if text is not None and (not path or "#" in uri): + # A selection or non-file context: the text itself is the point + parts.append(f"\n{label}:\n```\n{text}\n```\n") + else: + parts.append(label) + return " ".join(p for p in parts if p).strip(), paths + + +# ── Agent ────────────────────────────────────────────────────── + + +@dataclass +class _Turn: + text: str + on_finish: Callable[[str], None] + request_id: Any = None # None for internal commands, whose output is not shown + paths: list[str] = field(default_factory=list) + cancelled: bool = False + + +@dataclass +class _Subagent: + """A cecli sub-agent, shown to the client as a child session.""" + + agent_uuid: str + session_id: str + parent_session_id: str + emitter: UpdateEmitter + running: bool = False + cancelled: bool = False + removed: bool = False + + +class AcpAgent: + def __init__(self, transport: Transport): + self.transport = transport + self.emitter = UpdateEmitter(self._send_turn_update) + self.config = SessionConfig() + self.mcp = ClientMcpServers(log) + self.session_id: str | None = None + self.cwd: str | None = None + self.turn: _Turn | None = None + self._deferred_new: list[Any] = [] + self._ready = False + self._loop: asyncio.AbstractEventLoop | None = None + self._inputs: asyncio.Queue | None = None + self._dispatcher: asyncio.Task | None = None + self._coder_ref = None + self._next_request_id = 0 + # request id -> (future, the session that asked) + self._pending: dict[Any, tuple[asyncio.Future, str | None]] = {} + self._sent_options: list[dict] | None = None + # Sub-agents as child sessions, keyed by coder uuid; only for clients + # that advertise the (unstable) subagents capability + self.subagents_supported = False + self.subagents: dict[str, _Subagent] = {} + self._message_count = 0 + + # ── Lifecycle ─────────────────────────────────────────────── + + def bootstrap(self) -> str | None: + """Open the transport, answer ``initialize`` and wait for the first ``session/new``. + + Returns that session's cwd, or None if the client hung up first. + """ + self.transport.open() + # Commands run in a pseudo-terminal, where git log and friends start a + # pager that waits for a keypress nobody can send + os.environ["PAGER"] = "cat" + os.environ["GIT_PAGER"] = "cat" + return asyncio.run(self._await_first_session()) + + async def _await_first_session(self) -> str | None: + self.transport.attach(asyncio.get_running_loop()) + try: + while True: + msg = await self.transport.receive() + if msg is EOF: + return None + if msg.get("method") == "session/new" and "id" in msg: + try: + cwd = _require_cwd(msg.get("params") or {}) + except AcpError as err: + self.transport.send(error_message(msg["id"], err.code, err.message)) + continue + self.cwd = cwd + self.session_id = _new_session_id() + self.mcp.want((msg.get("params") or {}).get("mcpServers")) + self._deferred_new.append(msg["id"]) + return cwd + self._handle(msg) + finally: + self.transport.detach() + + @property + def launched(self) -> bool: + """True once ``bootstrap`` has a session to serve.""" + return self.session_id is not None + + def start(self) -> None: + """Serve the protocol on the running loop (``main_async``'s).""" + if not self.launched: + raise RuntimeError("--acp must be launched through the cecli command line") + loop = asyncio.get_running_loop() + if self._loop is loop: + return + if self.turn is not None: + # A program reload dropped the loop mid-turn + self._finish_turn() + self._loop = loop + self._inputs = asyncio.Queue() + self.transport.attach(loop) + self._dispatcher = loop.create_task(self._dispatch()) + + def shutdown(self) -> None: + """Answer whatever is still waiting before the process exits.""" + for request_id in self._deferred_new: + self.transport.send( + error_message( + request_id, + INTERNAL_ERROR, + "cecli exited before the session was ready; see its stderr for details", + ) + ) + self._deferred_new.clear() + self._finish_turn() + + async def _dispatch(self) -> None: + while True: + msg = await self.transport.receive() + if msg is EOF: + self._on_eof() + return + if "method" in msg: + self._handle(msg) + else: + self._on_response(msg) + + def _on_eof(self) -> None: + self._cancel_pending_requests() + if self.turn is not None: + self.turn.cancelled = True + self._interrupt_coder() + self._inputs.put_nowait(EOF) + + # ── Inbound ───────────────────────────────────────────────── + + def _handle(self, msg: dict) -> None: + method = msg.get("method") + params = msg.get("params") or {} + is_request = "id" in msg + request_id = msg.get("id") + + handlers = { + "initialize": self._on_initialize, + "authenticate": lambda params, request_id: {}, + "session/new": self._on_session_new, + "session/prompt": self._on_session_prompt, + "session/cancel": self._on_session_cancel, + "session/set_config_option": self._on_set_config_option, + "session/set_mode": self._on_set_mode, + } + handler = handlers.get(method) + if handler is None: + if is_request: + self.transport.send( + error_message(request_id, METHOD_NOT_FOUND, f"Method not found: {method}") + ) + return + + try: + result = handler(params, request_id) + except AcpError as err: + if is_request: + self.transport.send(error_message(request_id, err.code, err.message)) + return + except Exception as err: # never let one bad message kill the dispatcher + log(f"error handling {method}: {err!r}") + if is_request: + self.transport.send(error_message(request_id, INTERNAL_ERROR, str(err))) + return + + if is_request and result is not _DEFERRED: + self.transport.send(result_message(request_id, result)) + + def _on_initialize(self, params: dict, request_id: Any) -> dict: + capabilities = params.get("clientCapabilities") or {} + self.subagents_supported = isinstance(capabilities.get("subagents"), dict) + return { + "protocolVersion": PROTOCOL_VERSION, + "agentCapabilities": { + "loadSession": False, + "promptCapabilities": {"image": False, "audio": False, "embeddedContext": True}, + "mcpCapabilities": {"http": True, "sse": True}, + }, + "authMethods": [], + "agentInfo": {"name": "cecli", "title": "cecli", "version": __version__}, + } + + def _on_session_new(self, params: dict, request_id: Any) -> Any: + cwd = _require_cwd(params) + if self.cwd is None: + raise AcpError(INTERNAL_ERROR, "cecli is not running in ACP mode") + if os.path.realpath(cwd) != os.path.realpath(self.cwd): + raise AcpError( + INVALID_PARAMS, + f"This cecli process serves {self.cwd}; start another agent for {cwd}", + ) + self.mcp.want(params.get("mcpServers")) + if not self._ready: + self._deferred_new.append(request_id) + return _DEFERRED + + # A new thread on the same project: clear the chat, then hand out a new id + session_id = _new_session_id() + + def finish(_stop_reason: str) -> None: + self.session_id = session_id + self.transport.send(result_message(request_id, self._session_state())) + self._announce_commands() + + self._inputs.put_nowait(_Turn("/clear", on_finish=finish)) + return _DEFERRED + + def _on_session_prompt(self, params: dict, request_id: Any) -> Any: + self._require_session(params.get("sessionId")) + text, paths = prompt_to_text(params.get("prompt"), self.cwd or os.getcwd()) + if not text and not paths: + return {"stopReason": "end_turn"} + + def finish(stop_reason: str) -> None: + self.transport.send(result_message(request_id, {"stopReason": stop_reason})) + + self._inputs.put_nowait(_Turn(text, on_finish=finish, request_id=request_id, paths=paths)) + return _DEFERRED + + def _on_session_cancel(self, params: dict, request_id: Any) -> None: + if params.get("sessionId") != self.session_id: + self._cancel_subagent(params.get("sessionId")) + return None + + # Prompts queued behind the running one are dropped + kept = [] + while not self._inputs.empty(): + queued = self._inputs.get_nowait() + if isinstance(queued, _Turn) and queued.request_id is not None: + queued.on_finish("cancelled") + else: + kept.append(queued) + for queued in kept: + self._inputs.put_nowait(queued) + + if self.turn is not None and self.turn.request_id is not None: + self.turn.cancelled = True + # Sub-agents the turn waits for are cancelled with it and report that + # themselves; ones working in the background keep going + self._cancel_pending_requests(self.session_id) + self._interrupt_coder() + return None + + def _on_set_config_option(self, params: dict, request_id: Any) -> Any: + self._require_session(params.get("sessionId")) + command = self._config_command(params.get("configId"), params.get("value")) + if command is None: + return {"configOptions": self._current_options()} + + def finish(_stop_reason: str) -> None: + self.transport.send( + result_message(request_id, {"configOptions": self._current_options()}) + ) + + self._inputs.put_nowait(_Turn(command, on_finish=finish)) + return _DEFERRED + + def _on_set_mode(self, params: dict, request_id: Any) -> Any: + # Legacy modes, for clients without config options + self._require_session(params.get("sessionId")) + command = self._config_command("mode", params.get("modeId")) + if command is None: + return {} + + def finish(_stop_reason: str) -> None: + self._sync_config() + self.transport.send(result_message(request_id, {})) + + self._inputs.put_nowait(_Turn(command, on_finish=finish)) + return _DEFERRED + + def _config_command(self, config_id: Any, value: Any) -> str | None: + try: + return self.config.command(self._coder(), config_id, value) + except ValueError as err: + raise AcpError(INVALID_PARAMS, str(err)) + + def _on_response(self, msg: dict) -> None: + future, _session_id = self._pending.get(msg.get("id"), (None, None)) + if future is None or future.done(): + return + if "error" in msg: + log(f"client error: {msg['error']}") + future.set_result(None) + else: + future.set_result(msg.get("result")) + + def _require_session(self, session_id: Any) -> None: + if not self._ready or session_id != self.session_id: + raise AcpError(INVALID_PARAMS, f"Unknown session: {session_id}") + + # ── Turns (called from the coder loop via AcpInputOutput) ─── + + async def next_input(self, coder: Any = None) -> str: + """End the current turn, then wait for the next prompt.""" + # A mode or model switch replaces the coder, so track it before the + # finished turn reports the new configuration + if coder is not None: + self._coder_ref = weakref.ref(coder) + if self.mcp.pending: + # Before session/new is answered, so the first prompt has the tools + await self._sync_mcp(coder) + self._finish_turn() + + if not self._ready: + self._ready = True + for request_id in self._deferred_new: + self.transport.send(result_message(request_id, self._session_state())) + self._deferred_new.clear() + self._announce_commands() + + turn = await self._inputs.get() + if turn is EOF: + self._inputs.put_nowait(EOF) + raise EOFError + self.turn = turn + if turn.paths and coder is not None: + _add_mentions(coder, turn.paths) + return turn.text + + async def _sync_mcp(self, coder: Any) -> None: + try: + await self.mcp.sync( + getattr(coder, "mcp_manager", None), coder.io, getattr(coder, "verbose", False) + ) + except Exception as err: # a broken MCP server must not break the session + log(f"could not connect the client's MCP servers: {err!r}") + + def in_prompt(self) -> bool: + """True while a client prompt (not an internal command) is running.""" + return self.turn is not None and self.turn.request_id is not None + + def _finish_turn(self) -> None: + turn = self.turn + if turn is not None and turn.request_id is not None: + # Flush while the turn is still current so the updates go out + self.emitter.finish_turn(cancelled=turn.cancelled) + self._sync_config() + else: + self.emitter.reset() + self.turn = None + if turn is not None: + turn.on_finish("cancelled" if turn.cancelled else "end_turn") + + def _coder(self) -> Any: + return self._coder_ref() if self._coder_ref is not None else None + + def _interrupt_coder(self) -> None: + coder = self._coder() + if coder is None: + return + task = getattr(coder.io, "output_task", None) + if task is not None and not task.done(): + task.cancel() + event = getattr(coder, "interrupt_event", None) + if event is not None: + event.set() + + # ── Outbound ──────────────────────────────────────────────── + + def _send_turn_update(self, update: dict) -> None: + if self.in_prompt(): + self.send_update(update) + elif update.get("sessionUpdate") == "agent_message_chunk": + # Output outside a prompt (startup, internal commands) is only logged, + # so startup failures still explain themselves in the client's logs + text = update["content"]["text"].replace(CONSOLE_FENCE + "\n", "") + sys.stderr.write(text) + sys.stderr.flush() + + def send_update(self, update: dict, session_id: str | None = None) -> None: + self.transport.send( + { + "jsonrpc": "2.0", + "method": "session/update", + "params": {"sessionId": session_id or self.session_id, "update": update}, + } + ) + + def _current_options(self) -> list[dict]: + """The config options as they now stand, remembered as what the client has.""" + try: + options = self.config.options(self._coder()) + except Exception as err: # never let a model quirk break the session + log(f"could not build config options: {err!r}") + options = [] + self._sent_options = options + return options + + def _session_state(self) -> dict: + state: dict[str, Any] = {"sessionId": self.session_id} + options = self._current_options() + if options: + state["configOptions"] = options + state["modes"] = self.config.modes(options) + return state + + def _sync_config(self) -> None: + """Tell the client when a typed command (/model, /ask, ...) changed the options.""" + previous = self._sent_options + options = self._current_options() + if not options or options == previous: + return + self.send_update({"sessionUpdate": "config_option_update", "configOptions": options}) + mode = current_value(options, "mode") + if mode is not None and mode != current_value(previous or [], "mode"): + self.send_update({"sessionUpdate": "current_mode_update", "currentModeId": mode}) + + def _announce_commands(self) -> None: + from cecli.commands.utils.registry import CommandRegistry + + commands = [] + for name in sorted(CommandRegistry.list_commands()): + if name in HIDDEN_COMMANDS: + continue + command = CommandRegistry.get_command(name) + description = getattr(command, "DESCRIPTION", "") or name + commands.append({"name": name, "description": description}) + self.send_update( + {"sessionUpdate": "available_commands_update", "availableCommands": commands} + ) + + async def request(self, method: str, params: dict) -> Any: + """Send a request to the client and wait for its result (None on error).""" + self._next_request_id += 1 + request_id = self._next_request_id + future = asyncio.get_running_loop().create_future() + self._pending[request_id] = (future, params.get("sessionId")) + self.transport.send( + {"jsonrpc": "2.0", "id": request_id, "method": method, "params": params} + ) + try: + return await future + finally: + self._pending.pop(request_id, None) + + def _cancel_pending_requests(self, session_id: str | None = None) -> None: + """Answer outstanding requests as cancelled: all of them, or one session's.""" + for future, owner in self._pending.values(): + if not future.done() and session_id in (None, owner): + future.set_result({"outcome": {"outcome": "cancelled"}}) + + async def ask_permission( + self, question: str, subject: str | None, options: list[dict], agent_uuid: str | None = None + ) -> str | None: + """Ask the client to approve something; returns the chosen optionId or None.""" + subagent = self.subagents.get(agent_uuid) if agent_uuid else None + if subagent is not None: + # A sub-agent may ask while the parent's turn is over + if not subagent.running or subagent.cancelled: + return None + emitter, session_id = subagent.emitter, subagent.session_id + self._send_subagent_state(subagent, {"state": "requires_action"}) + elif not self.in_prompt() or self.turn.cancelled: + return None + else: + emitter, session_id = self.emitter, self.session_id + + tool_call_id, synthetic = emitter.permission_target(question, subject) + result = await self.request( + "session/request_permission", + { + "sessionId": session_id, + "toolCall": emitter.permission_prompt(tool_call_id, question), + "options": options, + }, + ) + if subagent is not None and subagent.running and not subagent.cancelled: + self._send_subagent_state(subagent, {"state": "running"}) + outcome = (result or {}).get("outcome") or {} + choice = outcome.get("optionId") if outcome.get("outcome") == "selected" else None + + allowed = any(o["optionId"] == choice and o["kind"].startswith("allow") for o in options) + emitter.permission_resolved(tool_call_id, synthetic, allowed) + return choice + + # ── Sub-agents (the unstable ACP subagents RFD) ───────────── + + def emitter_for(self, agent_uuid: str | None) -> UpdateEmitter: + """The emitter for output from the coder with this uuid.""" + subagent = self.subagents.get(agent_uuid) if agent_uuid else None + return subagent.emitter if subagent is not None else self.emitter + + def subagent_event(self, agent_uuid: str, event: str, **details: Any) -> None: + """Show a cecli sub-agent to the client as a child session. + + Clients that don't advertise the subagents capability see sub-agent + output in the parent session instead, as before. + """ + if not self.subagents_supported or self.session_id is None: + return + if event == "created": + self._announce_subagent(agent_uuid, **details) + return + if event == "message": + self._report_message(details.get("sender_uuid"), agent_uuid, details.get("text")) + return + + subagent = self.subagents.get(agent_uuid) + if subagent is None: + return + if event == "running": + subagent.running, subagent.cancelled = True, False + self._send_subagent_state(subagent, {"state": "running"}) + elif event == "idle": + stop_reason = details.get("stop_reason") or "end_turn" + subagent.emitter.finish_turn(cancelled=stop_reason == "cancelled") + subagent.running = False + self._cancel_pending_requests(subagent.session_id) + state = {"state": "idle", "stopReason": stop_reason} + if stop_reason == "error": + state["error"] = { + "code": INTERNAL_ERROR, + "message": details.get("error") or "The sub-agent failed", + } + self._send_subagent_state(subagent, state) + elif event == "removed": + # Reaped: nothing left to cancel. The child's history stays with the client + subagent.removed = True + self.send_update( + { + "sessionUpdate": "subagent_update", + "sessionId": subagent.session_id, + "capabilities": None, + }, + subagent.parent_session_id, + ) + + def _announce_subagent( + self, + agent_uuid: str, + parent_uuid: str | None = None, + name: str | None = None, + description: str | None = None, + root: str | None = None, + **_details: Any, + ) -> None: + if agent_uuid in self.subagents: + return + session_id = _new_session_id() + parent_session_id = self._session_of(parent_uuid) + emitter = UpdateEmitter(lambda update: self.send_update(update, session_id)) + if root: + emitter.root = root + + update: dict[str, Any] = { + "sessionUpdate": "subagent_update", + "sessionId": session_id, + "capabilities": {"cancel": {}}, + } + if name: + update["title"] = name + if description: + update["description"] = str(description) + # The announcement goes out before anything that names the child + self.send_update(update, parent_session_id) + self.subagents[agent_uuid] = _Subagent(agent_uuid, session_id, parent_session_id, emitter) + + def _session_of(self, agent_uuid: str | None) -> str: + """The session showing this coder: its own if it's a sub-agent, else the parent's.""" + subagent = self.subagents.get(agent_uuid) if agent_uuid else None + return subagent.session_id if subagent is not None else self.session_id + + def _report_message(self, sender_uuid: str | None, recipient_uuid: str, text: Any) -> None: + """A message between agents, as the sender sent it and as the recipient got it.""" + if not text: + return + sender, recipient = self._session_of(sender_uuid), self._session_of(recipient_uuid) + if sender == recipient: + return + content = [{"type": "text", "text": str(text)}] + for session_id in (sender, recipient): + self._message_count += 1 + self.send_update( + { + "sessionUpdate": "session_message", + "messageId": f"cecli-message-{self._message_count}", + "senderSessionId": sender, + "recipientSessionId": recipient, + "content": content, + }, + session_id, + ) + + def _send_subagent_state(self, subagent: _Subagent, state: dict) -> None: + self.send_update( + {"sessionUpdate": "subagent_update", "sessionId": subagent.session_id, "state": state}, + subagent.parent_session_id, + ) + + def _cancel_subagent(self, session_id: Any) -> None: + """``session/cancel`` on a child: stop that sub-agent's current work only.""" + subagent = next((s for s in self.subagents.values() if s.session_id == session_id), None) + if subagent is None or subagent.removed or not subagent.running: + return + subagent.cancelled = True + self._cancel_pending_requests(subagent.session_id) + coder = self._coder() + if coder is None: + return + from cecli.helpers.agents.service import AgentService + + # It reports the cancelled idle state itself once it has stopped + AgentService.get_instance(coder).cancel_sub_agent(subagent.agent_uuid) + + +def _require_cwd(params: dict) -> str: + cwd = params.get("cwd") + if not isinstance(cwd, str) or not os.path.isabs(cwd): + raise AcpError(INVALID_PARAMS, "session/new requires an absolute cwd") + if not os.path.isdir(cwd): + raise AcpError(INVALID_PARAMS, f"Not a directory: {cwd}") + return cwd + + +def _new_session_id() -> str: + return uuid.uuid4().hex + + +def _add_mentions(coder: Any, paths: list[str]) -> None: + """Put @-mentioned files in the chat, as cecli would for /add.""" + root = os.path.realpath(coder.root) + for path in paths: + if not os.path.isfile(path): + continue + real = os.path.realpath(path) + if real in coder.abs_fnames or real in coder.abs_read_only_fnames: + continue + try: + inside = os.path.commonpath([root, real]) == root + except ValueError: # different drives + inside = False + if inside: + rel = coder.get_rel_fname(real) + coder.add_rel_fname(rel) + coder.io.tool_output(f"Added {rel} to the chat") + else: + coder.abs_read_only_fnames.add(real) + coder.io.tool_output(f"Added {real} to the chat (read-only)") diff --git a/cecli/helpers/server/acp/agent/config.py b/cecli/helpers/server/acp/agent/config.py new file mode 100644 index 00000000000..82e3fe54e68 --- /dev/null +++ b/cecli/helpers/server/acp/agent/config.py @@ -0,0 +1,327 @@ +"""Session config options for ``cecli --acp``: chat mode, model and thinking effort. + +ACP clients render these as pickers next to the prompt box: +https://agentclientprotocol.com/protocol/session-config-options + +Picking a value runs the matching cecli command (``/ask``, ``/model ...``, +``/reasoning-effort ...``), so a switch behaves exactly as if it were typed. +""" + +from __future__ import annotations + +import fnmatch +from typing import Any + +import yaml + +from cecli.helpers.server.acp.agent.transport import log + +MODES = [ + ("code", "Code", "Make the requested changes to the code"), + ("ask", "Ask", "Answer questions without editing files"), + ("architect", "Architect", "Plan with the main model, then edit with the editor model"), + ("context", "Context", "Work out which files a change needs"), + ("agent", "Agent", "Work autonomously, finding files and running tools"), + ("agent-yolo", "Agent (YOLO)", "Agent mode that runs tools and commands without asking"), +] +# Agent mode with what --yolo turns on: every confirmation answered yes +YOLO_MODE = "agent-yolo" +REASONING_EFFORTS = ["low", "medium", "high"] +THINKING_BUDGETS = ["4k", "8k", "16k", "32k"] +# Model metadata modes cecli can chat with (not image generation and the like) +CHAT_MODES = ("chat", "responses") + + +def _value(value: str, name: str | None = None, description: str | None = None) -> dict: + entry = {"value": value, "name": name or value} + if description: + entry["description"] = description + return entry + + +def _select(config_id: str, name: str, category: str, current: str, values: list[dict]) -> dict: + return { + "id": config_id, + "name": name, + "category": category, + "type": "select", + "currentValue": current, + "options": values, + } + + +def _token_count(budget: str) -> float: + """``"8k"`` -> 8192, as cecli formats thinking budgets.""" + scale = {"k": 1024, "m": 1024 * 1024}.get(budget[-1:].lower()) + return float(budget[:-1]) * scale if scale else float(budget) + + +def _flat_values(option: dict) -> list[dict]: + """A select option's values, with any groups flattened.""" + values = [] + for entry in option["options"]: + values.extend(entry["options"] if "group" in entry else [entry]) + return values + + +def current_value(options: list[dict], config_id: str) -> str | None: + option = next((o for o in options if o["id"] == config_id), None) + return option["currentValue"] if option else None + + +def resolve_model(value: str) -> tuple[str, bool]: + """The model name cecli would run for ``value``, and whether a tag override applies.""" + from cecli import models + + name = models.MODEL_ALIASES.get(value, value) + base, overrides = models.ModelOverrides.apply(name) + return (base, True) if overrides else (name, False) + + +def settings_file_models(coder: Any) -> list[str]: + """Models defined in the user's .cecli.model.settings.yml files.""" + from cecli.helpers.file_searcher import generate_search_path_list + + args = getattr(coder, "args", None) + repo = getattr(coder, "repo", None) + fnames = generate_search_path_list( + ".cecli.model.settings.yml", + getattr(repo, "root", None), + getattr(args, "model_settings_file", None), + ) + names = [] + for fname in fnames: + try: + with open(fname, encoding="utf-8") as f: + entries = yaml.safe_load(f) or [] + except (OSError, yaml.YAMLError): + continue + for entry in entries if isinstance(entries, list) else []: + name = entry.get("name") if isinstance(entry, dict) else None + # cecli/extra_params holds settings for every model, not a model + if isinstance(name, str) and not name.endswith("/extra_params"): + names.append(name) + return names + + +def keyed_provider_models() -> list[tuple[str, str, list[str]]]: + """``(slug, name, models)`` for each provider whose API key is set.""" + from cecli.helpers.onboarding.providers import ( + get_models_for_provider, + iter_providers, + provider_has_key, + ) + + groups = [] + for provider in iter_providers(): + if provider_has_key(provider): + models = get_models_for_provider(provider["slug"], modes=CHAT_MODES) + if models: + groups.append((provider["slug"], provider["display_name"], models)) + return groups + + +def matches_filter(name: str, patterns: list[str]) -> bool: + """Whether ``name`` matches any of ``patterns``, as ``/models`` matches its search.""" + name = name.lower() + for pattern in patterns: + pattern = pattern.lower() + if any(char in pattern for char in "*?["): + if fnmatch.fnmatchcase(name, f"*{pattern}*"): + return True + elif pattern in name: + return True + return False + + +class SessionConfig: + """Builds the config options from a coder and maps choices to cecli commands.""" + + def __init__(self): + self.model_choice: str | None = None # how the user last named the model + # (group id, group name, [(value, description, model it runs)]) + self._model_groups: list[tuple[str, str, list[tuple[str, str | None, str]]]] | None = None + self._before_yolo: tuple[Any, Any] | None = None + + def options(self, coder: Any) -> list[dict]: + if coder is None: + return [] + options = [self._mode_option(coder), self._model_option(coder)] + effort = self._effort_option(coder) + if effort is not None: + options.append(effort) + return options + + def command(self, coder: Any, config_id: Any, value: Any) -> str | None: + """The cecli command that applies ``value``, or None if it is already current. + + Raises ValueError for an unknown option or value. + """ + option = next((o for o in self.options(coder) if o["id"] == config_id), None) + if option is None: + raise ValueError(f"Unknown config option: {config_id}") + if value not in {entry["value"] for entry in _flat_values(option)}: + raise ValueError(f"Unknown value for {config_id}: {value}") + if value == option["currentValue"]: + return None + + if config_id == "mode": + yolo = value == YOLO_MODE + # Leaving YOLO, or picking plain Agent, brings the confirmations back + if yolo or value == "agent" or option["currentValue"] == YOLO_MODE: + self._set_yolo(coder, yolo) + target = "agent" if yolo else value + return None if coder.edit_format == target else f"/{target}" + if config_id == "model": + self.model_choice = value + # Agent mode answers with the agent model, which /model leaves alone + command = "/agent-model" if coder.edit_format == "agent" else "/model" + return f"{command} {value}" + if config_id == "reasoning_effort": + return f"/reasoning-effort {'none' if value == 'default' else value}" + return f"/think-tokens {value}" + + @staticmethod + def modes(options: list[dict]) -> dict | None: + """The same choice as the legacy ``modes`` state, for clients without config options.""" + mode = next((o for o in options if o["id"] == "mode"), None) + if mode is None: + return None + return { + "currentModeId": mode["currentValue"], + "availableModes": [ + {"id": v["value"], "name": v["name"], "description": v.get("description")} + for v in mode["options"] + ], + } + + # ── Options ───────────────────────────────────────────────── + + def _set_yolo(self, coder: Any, on: bool) -> None: + """Turn on (or back off) the two settings ``--yolo`` sets at startup.""" + args = coder.args + if on: + if self._before_yolo is None: + self._before_yolo = (coder.io.yes, getattr(args, "yes_always_commands", False)) + coder.io.yes = True + args.yes_always_commands = True + else: + # Launched with --yolo: nothing to restore, so ask again (None asks) + coder.io.yes, args.yes_always_commands = self._before_yolo or (None, False) + self._before_yolo = None + + def _mode_option(self, coder: Any) -> dict: + ids = [mode_id for mode_id, _, _ in MODES] + # Every other edit format (diff, whole, udiff, ...) is code mode + current = coder.edit_format if coder.edit_format in ids else "code" + if self._before_yolo is not None and current != "agent": + # Left agent mode with a typed command; YOLO doesn't outlive it + self._set_yolo(coder, False) + if current == "agent" and getattr(coder.args, "yes_always_commands", False): + current = YOLO_MODE + values = [_value(*mode) for mode in MODES] + return _select("mode", "Mode", "mode", current, values) + + def _model_option(self, coder: Any) -> dict: + # The model that answers in this mode (the agent model in agent mode) + running = coder.get_active_model().name + groups = [] + matches = [] + for group_id, name, entries in self._models(coder): + values = [] + for value, description, model in entries: + values.append(_value(value, description=description)) + if model == running: + matches.append(value) + groups.append((group_id, name, values)) + + if self.model_choice in matches: + current = self.model_choice + elif matches: + current = matches[0] + else: + # Switched with /model to something not listed + current = running + groups[0][2].insert(0, _value(running)) + options = [ + {"group": group_id, "name": name, "options": values} + for group_id, name, values in groups + if values + ] + return _select("model", "Model", "model", current, options) + + def _models(self, coder: Any) -> list[tuple[str, str, list[tuple[str, str | None, str]]]]: + """The models you configured, then those of every provider with an API key set.""" + if self._model_groups is not None: + return self._model_groups + + from cecli import models + + args = getattr(coder, "args", None) + candidates: list[tuple[str, str | None]] = [] + if getattr(args, "model", None): + candidates.append((args.model, None)) + for alias_def in getattr(args, "alias", None) or []: + alias, _, target = alias_def.partition(":") + if alias.strip() and target.strip(): + candidates.append((alias.strip(), target.strip())) + for base, tags in getattr(models.ModelOverrides, "_overrides", {}).items(): + for tag in tags if isinstance(tags, dict) else []: + candidates.append((f"{base}:{tag}", None)) + try: + candidates.extend((name, None) for name in settings_file_models(coder)) + except Exception: + pass + + seen = set() + configured = [] + for value, description in candidates: + name, tagged = resolve_model(value) + key = value if tagged else name + if key in seen: + continue + seen.add(key) + if description is None and name != value and not tagged: + description = name + configured.append((value, description, name)) + self._model_groups = [("configured", "Configured", configured)] + + try: + providers = keyed_provider_models() + except Exception as err: # the configured models still work + log(f"could not list provider models: {err!r}") + providers = [] + patterns = getattr(args, "acp_models", None) or [] + for slug, provider_name, names in providers: + entries = [] + for value in names: + if value in seen or (patterns and not matches_filter(value, patterns)): + continue + seen.add(value) + entries.append((value, None, resolve_model(value)[0])) + if entries: + self._model_groups.append((slug, provider_name, entries)) + return self._model_groups + + def _effort_option(self, coder: Any) -> dict | None: + model = coder.get_active_model() + accepts = getattr(model, "accepts_settings", None) or [] + + effort = model.get_reasoning_effort() + if "reasoning_effort" in accepts or effort is not None: + current = "default" if effort is None else str(effort) + choices = ["default", *REASONING_EFFORTS] + if current not in choices: + choices.append(current) + values = [_value(v, "Default" if v == "default" else v.capitalize()) for v in choices] + return _select("reasoning_effort", "Reasoning effort", "thought_level", current, values) + + budget = model.get_raw_thinking_tokens() + if "thinking_tokens" in accepts or budget is not None: + current = model.get_thinking_tokens() if budget else "0" + choices = ["0", *THINKING_BUDGETS] + if current not in choices: + choices = sorted([*choices, current], key=_token_count) + values = [_value(v, "Off" if v == "0" else f"{v} tokens") for v in choices] + return _select("thinking_tokens", "Thinking budget", "thought_level", current, values) + return None diff --git a/cecli/helpers/server/acp/agent/io.py b/cecli/helpers/server/acp/agent/io.py new file mode 100644 index 00000000000..074b45e2421 --- /dev/null +++ b/cecli/helpers/server/acp/agent/io.py @@ -0,0 +1,201 @@ +"""AcpInputOutput - InputOutput adapter for ``cecli --acp``. + +Prompts come from the ACP client instead of a terminal, confirmations become +``session/request_permission`` requests, and output is translated into +``session/update`` notifications by the agent's UpdateEmitter. +""" + +from io import StringIO + +from rich.console import Console +from rich.text import Text + +from cecli.helpers.server.acp.agent import get_agent +from cecli.helpers.server.acp.agent.transport import log +from cecli.helpers.server.acp.agent.updates import strip_ansi +from cecli.io import InputOutput + + +def _plain(*messages) -> str: + return strip_ansi(" ".join(m.plain if isinstance(m, Text) else str(m) for m in messages)) + + +class AcpInputOutput(InputOutput): + def __init__(self, pretty=False, *args, agent=None, **kwargs): + # Set before InputOutput.__init__, which may already emit warnings + self.agent = agent or get_agent() + # Renders the odd Rich object (tables, panels) that reaches stream_print + self._plain_console = Console( + file=StringIO(), force_terminal=False, no_color=True, highlight=False, width=200 + ) + kwargs["fancy_input"] = False + kwargs["notifications"] = False + kwargs["show_spinner"] = False + super().__init__(False, *args, **kwargs) + self.fallback_spinner_enabled = False + + # ── Input ─────────────────────────────────────────────────── + + async def get_input( + self, + root, + rel_fnames, + addable_rel_fnames, + commands, + abs_read_only_fnames=None, + abs_read_only_stubs_fnames=None, + edit_format=None, + ): + self.agent.emitter.root = root + text = await self.agent.next_input(getattr(commands, "coder", None)) + self.user_input(text) + return text + + async def _confirm_ask( + self, + question, + default="y", + subject=None, + explicit_yes_required=False, + group=None, + group_response=None, + allow_never=False, + allow_tweak=False, + acknowledge=False, + **kwargs, + ): + self.num_user_asks += 1 + question_id = (question, subject) + + if question_id in self.never_prompts: + return False + if group and not group.show_group: + group = None + if group: + allow_never = True + + if self.yes is True and not explicit_yes_required: + res = "y" + elif group_response and group_response in self.group_responses: + return self.group_responses[group_response] + elif group and group.preference: + res = group.preference + else: + options = [{"optionId": "y", "name": "Yes", "kind": "allow_once"}] + if group or group_response: + if not explicit_yes_required or group_response: + options.append({"optionId": "a", "name": "Yes to all", "kind": "allow_always"}) + options.append({"optionId": "s", "name": "Skip all", "kind": "reject_always"}) + options.append({"optionId": "n", "name": "No", "kind": "reject_once"}) + if allow_never and not (group or group_response): + options.append( + {"optionId": "d", "name": "Don't ask again", "kind": "reject_always"} + ) + + choice = await self.agent.ask_permission( + question, subject, options, agent_uuid=kwargs.get("coder_uuid") + ) + if choice is None: + log(f"declined: {question}") + choice = "n" + res = choice + + res = res.lower()[0] + + if res == "d" and allow_never: + self.never_prompts.add(question_id) + self.append_chat_history(f"{question.strip()} {res}", linebreak=True, blockquote=True) + return False + + if explicit_yes_required and not group_response: + is_yes = res == "y" + else: + is_yes = res in ("y", "a") + + is_all = res == "a" and ( + (group is not None and not explicit_yes_required) or group_response + ) + is_skip = res == "s" and (group is not None or group_response) + + if group: + if is_all and not explicit_yes_required: + group.preference = "all" + elif is_skip: + group.preference = "skip" + + self.append_chat_history(f"{question.strip()} {res}", linebreak=True, blockquote=True) + + if group_response and (is_all or is_skip): + self.group_responses[group_response] = is_yes + + return is_yes + + def prompt_ask(self, question, default="", subject=None): + # Free-form answers have no ACP equivalent; take the default + log(f"using default for prompt: {question} -> {default!r}") + self.append_chat_history(f"{question.strip()} {default}", linebreak=True, blockquote=True) + return default + + # ── Output ────────────────────────────────────────────────── + + def tool_output(self, *messages, log_only=False, bold=False, type=None, **kwargs): + super().tool_output(*messages, log_only=True) + # The per-request token/cost line stays in the chat history only + if log_only or type == "usage-report": + return + + emitter = self.agent.emitter_for(kwargs.get("coder_uuid")) + if type == "Tool Call": + emitter.tool_header(_plain(*messages)) + elif type == "tool-footer": + emitter.tool_footer() + elif messages: + emitter.block(_plain(*messages)) + + def _tool_message(self, message="", strip=True, color=None, **kwargs): + if message.strip(): + if "\n" in message: + for line in message.splitlines(): + self.append_chat_history(line, linebreak=True, blockquote=True, strip=strip) + else: + hist = message.strip() if strip else message + self.append_chat_history(hist, linebreak=True, blockquote=True) + self.agent.emitter_for(kwargs.get("coder_uuid")).block(_plain(message)) + + def assistant_output(self, message, pretty=None, **kwargs): + emitter = self.agent.emitter_for(kwargs.get("coder_uuid")) + emitter.stream(_plain(message or ""), final=True) + + def stream_output(self, text, final=False, **kwargs): + emitter = self.agent.emitter_for(kwargs.get("coder_uuid")) + emitter.stream(strip_ansi(text or ""), final=final) + + def reset_streaming_response(self, **kwargs): + pass + + def stream_print(self, *messages, **kwargs): + console = self._plain_console + with console.capture() as capture: + console.print(*messages, markup=False, soft_wrap=True) + self.agent.emitter_for(kwargs.get("coder_uuid")).block(strip_ansi(capture.get())) + + def tool_call_event(self, tool_call_id, status, coder_uuid=None, **kwargs): + self.agent.emitter_for(coder_uuid).tool_event(tool_call_id, status, **kwargs) + + def subagent_event(self, agent_uuid, event, **kwargs): + self.agent.subagent_event(agent_uuid, event, **kwargs) + + def display_user_input(self, inp): + # The client already shows what the user sent + pass + + # ── Terminal-only behavior ────────────────────────────────── + + def rule(self): + pass + + def ring_bell(self): + pass + + def notify_user_input_required(self): + pass diff --git a/cecli/helpers/server/acp/agent/mcp.py b/cecli/helpers/server/acp/agent/mcp.py new file mode 100644 index 00000000000..8210cabc434 --- /dev/null +++ b/cecli/helpers/server/acp/agent/mcp.py @@ -0,0 +1,94 @@ +"""MCP servers passed by the ACP client in ``session/new``. + +Editors hand the agent the MCP servers they have configured (Zed's context +servers) so it can use the same tools. They join cecli's own servers in the +shared McpServerManager, as if they were listed in an MCP config file. +""" + +from __future__ import annotations + +from typing import Any, Callable + + +def _pairs(items: Any) -> dict[str, str]: + """ACP ``[{name, value}]`` lists (env vars, headers) as a dict.""" + pairs = {} + for item in items if isinstance(items, list) else []: + if isinstance(item, dict) and isinstance(item.get("name"), str): + pairs[item["name"]] = str(item.get("value", "")) + return pairs + + +def to_cecli_configs(mcp_servers: Any) -> dict[str, dict]: + """ACP McpServer entries as cecli MCP server configs, keyed by lowercased name.""" + configs = {} + for entry in mcp_servers if isinstance(mcp_servers, list) else []: + if not isinstance(entry, dict) or not entry.get("name"): + continue + transport = entry.get("type", "stdio") + if transport == "stdio" and entry.get("command"): + config = { + "transport": "stdio", + "command": entry["command"], + "args": [str(arg) for arg in entry.get("args") or []], + "env": _pairs(entry.get("env")), + } + elif transport in ("http", "sse") and entry.get("url"): + config = { + "transport": transport, + "url": entry["url"], + "headers": _pairs(entry.get("headers")), + } + else: + continue + config["name"] = str(entry["name"]) + configs[config["name"].lower()] = config + return configs + + +def _make_server(config: dict, io: Any, verbose: bool) -> Any: + from cecli.mcp.server import HttpStreamingServer, McpServer, SseServer + + server_class = {"stdio": McpServer, "http": HttpStreamingServer, "sse": SseServer} + return server_class[config["transport"]](dict(config), io=io, verbose=verbose) + + +class ClientMcpServers: + """Keeps the manager's client-provided servers in line with the latest session.""" + + def __init__(self, log: Callable[[str], None]): + self._log = log + self._wanted: dict[str, dict] | None = None + self._ours: dict[str, dict] = {} # servers this class added, by name + + def want(self, mcp_servers: Any) -> None: + """Record a session's servers; ``sync`` applies them on the coder's loop.""" + self._wanted = to_cecli_configs(mcp_servers) + + @property + def pending(self) -> bool: + return self._wanted is not None + + async def sync(self, manager: Any, io: Any = None, verbose: bool = False) -> None: + wanted, self._wanted = self._wanted, None + if wanted is None or manager is None: + return + + for name in set(self._ours) - set(wanted): + # No longer offered by the client (a later session dropped it) + await manager.disconnect_server(name) + + for name, config in wanted.items(): + server = manager.get_server(name) + if server is None: + self._ours[name] = config + if await manager.add_server(_make_server(config, io, verbose), connect=True): + self._log(f"connected MCP server from the client: {config['name']}") + continue + if name not in self._ours: + self._log(f"skipped the client's MCP server {name!r}: cecli has one by that name") + continue + if config != self._ours[name]: + self._log(f"the client changed MCP server {name!r}; restart the agent to apply it") + if server not in manager.connected_servers: + await manager.connect_server(name) diff --git a/cecli/helpers/server/acp/agent/transport.py b/cecli/helpers/server/acp/agent/transport.py new file mode 100644 index 00000000000..daa559b412b --- /dev/null +++ b/cecli/helpers/server/acp/agent/transport.py @@ -0,0 +1,158 @@ +"""How AcpAgent exchanges JSON-RPC messages with the client. + +The agent only needs the small ``Transport`` interface below. ``StdioTransport`` +implements it for clients that launch cecli as a subprocess (``cecli --acp``); +another transport, such as a WebSocket connection, can drive the same agent. +""" + +from __future__ import annotations + +import asyncio +import json +import os +import sys +import threading +from typing import Any, Protocol + +PARSE_ERROR = -32700 +INVALID_REQUEST = -32600 +METHOD_NOT_FOUND = -32601 +INVALID_PARAMS = -32602 +INTERNAL_ERROR = -32603 + +# Returned by ``Transport.receive`` once the client has gone away +EOF = object() + + +def log(message: str) -> None: + """Diagnostics go to stderr; stdout belongs to the protocol.""" + print(f"[cecli acp] {message}", file=sys.stderr, flush=True) + + +def result_message(request_id: Any, result: Any) -> dict: + return {"jsonrpc": "2.0", "id": request_id, "result": result} + + +def error_message(request_id: Any, code: int, message: str) -> dict: + return {"jsonrpc": "2.0", "id": request_id, "error": {"code": code, "message": message}} + + +class Transport(Protocol): + """A connection to one ACP client, carrying JSON-RPC messages as dicts.""" + + def open(self) -> None: + """Start receiving messages from the client.""" + + def attach(self, loop: asyncio.AbstractEventLoop) -> None: + """Deliver incoming messages to ``loop`` (called from that loop).""" + + def detach(self) -> None: + """Stop delivering to the current loop, keeping unread messages for the next.""" + + async def receive(self) -> Any: + """The next message from the client, or ``EOF`` once it has gone away.""" + + def send(self, msg: dict) -> None: + """Send a message to the client (safe to call from any thread).""" + + +class StdioTransport: + """Newline-delimited JSON-RPC on the process's original stdin/stdout. + + ``open()`` moves the protocol onto private duplicates of fds 0 and 1, + then points fd 0 at /dev/null and fd 1 at stderr, so stray prints, Rich + consoles and child processes can neither corrupt nor consume the stream. + + A reader thread hands messages to whichever event loop is attached; + messages that arrive while no loop is attached wait in a backlog. + """ + + def __init__(self): + self._lock = threading.Lock() + self._write_lock = threading.Lock() + self._out = None + self._loop: asyncio.AbstractEventLoop | None = None + self._inbox: asyncio.Queue | None = None + self._backlog: list[Any] = [] + + def open(self) -> None: + sys.stdout.flush() + in_fd = os.dup(0) + out_fd = os.dup(1) + devnull = os.open(os.devnull, os.O_RDONLY) + os.dup2(devnull, 0) + os.close(devnull) + os.dup2(2, 1) + self.start(os.fdopen(in_fd, "rb"), os.fdopen(out_fd, "wb")) + + def start(self, reader, writer) -> None: + self._out = writer + thread = threading.Thread( + target=self._read_loop, args=(reader,), name="cecli-acp-stdin", daemon=True + ) + thread.start() + + def _read_loop(self, reader) -> None: + for raw in reader: + line = raw.strip() + if not line: + continue + try: + msg = json.loads(line) + except ValueError as err: + self.send(error_message(None, PARSE_ERROR, f"Parse error: {err}")) + continue + if not isinstance(msg, dict): + self.send(error_message(None, INVALID_REQUEST, "Invalid Request")) + continue + self._deliver(msg) + self._deliver(EOF) + + def _deliver(self, msg: Any) -> None: + with self._lock: + loop = self._loop + if loop is not None: + try: + loop.call_soon_threadsafe(self._enqueue, msg) + return + except RuntimeError: # loop closed under us + self._loop = None + self._inbox = None + self._backlog.append(msg) + + def _enqueue(self, msg: Any) -> None: + with self._lock: + if self._inbox is not None: + self._inbox.put_nowait(msg) + else: + self._backlog.append(msg) + + def attach(self, loop: asyncio.AbstractEventLoop) -> None: + with self._lock: + self._loop = loop + self._inbox = asyncio.Queue() + for msg in self._backlog: + self._inbox.put_nowait(msg) + self._backlog.clear() + + def detach(self) -> None: + with self._lock: + if self._inbox is not None: + while not self._inbox.empty(): + self._backlog.append(self._inbox.get_nowait()) + self._loop = None + self._inbox = None + + async def receive(self) -> Any: + return await self._inbox.get() + + def send(self, msg: dict) -> None: + data = json.dumps(msg, ensure_ascii=False, separators=(",", ":")).encode("utf-8") + with self._write_lock: + if self._out is None: + return + try: + self._out.write(data + b"\n") + self._out.flush() + except (OSError, ValueError) as err: + log(f"failed to write to client: {err}") diff --git a/cecli/helpers/server/acp/agent/updates.py b/cecli/helpers/server/acp/agent/updates.py new file mode 100644 index 00000000000..052a6a55c5e --- /dev/null +++ b/cecli/helpers/server/acp/agent/updates.py @@ -0,0 +1,458 @@ +"""Translate cecli's console-oriented output into ACP v1 ``session/update`` payloads. + +cecli reports progress as a stream of text: model tokens (with reasoning wrapped +in REASONING_START/REASONING_END markers), ``tool_output`` lines, and tool call +panels printed as header/body/footer. ``UpdateEmitter`` keeps just enough state +to turn that stream into the structured updates an ACP client renders: +message and thought chunks, plus ``tool_call``/``tool_call_update`` cards driven +by ``InputOutput.tool_call_event``. +""" + +from __future__ import annotations + +import json +import os +import re +from dataclasses import dataclass, field +from typing import Any, Callable + +from cecli.helpers.responses import unprefix_tool_name +from cecli.reasoning_tags import REASONING_END, REASONING_START + +# Longest tool output (in characters) mirrored into a tool call card +MAX_TOOL_TEXT = 20_000 + +# Wraps cecli's own console lines; longer than a typical ``` so code in them survives +CONSOLE_FENCE = "````" + +_ANSI_RE = re.compile(r"\x1b\[[0-9;?]*[ -/]*[@-~]") + +# Normalized cecli tool name -> ACP ToolKind +_TOOL_KINDS = { + "readfile": "read", + "ls": "read", + "gitstatus": "read", + "gitlog": "read", + "gitshow": "read", + "gitdiff": "read", + "gitbranch": "read", + "gitremote": "read", + "grep": "search", + "explorecode": "search", + "searchfacts": "search", + "editfile": "edit", + "undochange": "edit", + "replacefacts": "edit", + # Not "execute": Zed treats execute calls as live terminals and hides its + # progress spinner while one runs, but cecli runs commands itself + "command": "other", + "thinking": "think", +} + +# Argument keys worth showing in a tool call's title, in priority order +_TITLE_KEYS = ("command", "file_path", "path", "pattern", "query", "url", "symbol", "name") +_PATH_KEYS = ("file_path", "path") + + +def strip_ansi(text: str) -> str: + return _ANSI_RE.sub("", text) + + +def tool_kind(name: str) -> str: + norm = re.sub(r"[^a-z0-9]", "", name.lower()) + if norm in _TOOL_KINDS: + return _TOOL_KINDS[norm] + if "fetch" in norm or "web" in norm or "scrape" in norm: + return "fetch" + if "search" in norm or "grep" in norm or "find" in norm: + return "search" + if "read" in norm or "view" in norm or norm.startswith("list"): + return "read" + if "delete" in norm or "remove" in norm: + return "delete" + if "edit" in norm or "write" in norm or "replace" in norm or "create" in norm: + return "edit" + return "other" + + +def parse_arguments(arguments: Any) -> dict: + if isinstance(arguments, dict): + return arguments + if not arguments: + return {} + try: + parsed = json.loads(arguments) + except (TypeError, ValueError): + return {} + return parsed if isinstance(parsed, dict) else {} + + +def tool_title(name: str, args: dict) -> str: + for key in _TITLE_KEYS: + value = args.get(key) + if isinstance(value, str) and value.strip(): + first = value.strip().splitlines()[0] + if len(first) > 120: + first = first[:117] + "..." + return f"{name}: {first}" + return name + + +def _fenced(text: str) -> str: + if len(text) > MAX_TOOL_TEXT: + text = "...(truncated)\n" + text[-MAX_TOOL_TEXT:] + fence = "```" + while fence in text: + fence += "`" + return f"{fence}\n{text.rstrip()}\n{fence}" + + +def _partial_suffix(text: str, marker: str) -> int: + """Length of the longest suffix of ``text`` that is a proper prefix of ``marker``.""" + for size in range(min(len(text), len(marker) - 1), 0, -1): + if marker.startswith(text[-size:]): + return size + return 0 + + +def plan_entries(tasks: Any) -> list[dict]: + """UpdateTodoList's ``tasks`` argument as ACP plan entries.""" + entries = [] + for task in tasks if isinstance(tasks, list) else []: + if isinstance(task, dict): + content = str(task.get("task", "")).strip() + if task.get("done"): + status = "completed" + elif task.get("current"): + status = "in_progress" + else: + status = "pending" + else: + content, status = str(task).strip(), "pending" + if content: + entries.append({"content": content, "priority": "medium", "status": status}) + return entries + + +@dataclass +class _Tool: + id: str + title: str + kind: str + status: str = "pending" + output: list[str] = field(default_factory=list) + synthetic: bool = False + # Shown as an agent message once the tool runs (Yield's final answer) + message: str | None = None + + +class UpdateEmitter: + """Stateful translator from cecli output to ``session/update`` payloads. + + ``send`` receives each update dict (the value of the notification's + ``update`` field). State is per turn; call ``finish_turn`` when it ends. + """ + + def __init__(self, send: Callable[[dict], None]): + self._send = send + self.root = os.getcwd() + self._synthetic_count = 0 + self.reset() + + def reset(self) -> None: + self._thinking = False + self._held = "" + self._pending_ws = "" + self._msg_tail = "" + self._console_open = False + self._tools: dict[str, _Tool] = {} + self._active: str | None = None + self._in_panel = False + + # ── Model text ────────────────────────────────────────────── + + def stream(self, text: str, final: bool = False) -> None: + """Model output; reasoning between the markers becomes thought chunks.""" + if text: + self._close_console() + self._settle_tools() + + buf = self._held + (text or "") + self._held = "" + while buf: + marker = REASONING_END if self._thinking else REASONING_START + idx = buf.find(marker) + if idx >= 0: + self._emit_text(buf[:idx]) + buf = buf[idx + len(marker) :] + self._thinking = not self._thinking + continue + keep = 0 if final else _partial_suffix(buf, marker) + self._emit_text(buf[: len(buf) - keep]) + self._held = buf[len(buf) - keep :] + break + + if final: + self._thinking = False + + def block(self, text: str) -> None: + """A line of cecli's own output (tool_output, warnings, errors). + + Consecutive lines stream into one monospace block, so tables and + aligned output survive markdown rendering. + """ + if self._in_panel: + # Tool panel body: the arguments already travel as rawInput + return + if not text or not text.strip(): + return + if self._active is not None: + self._tool_output(self._active, text) + return + + self._flush_held() + prefix = "" + if not self._console_open: + tail = self._msg_tail + if not tail or tail.endswith("\n\n"): + prefix = CONSOLE_FENCE + "\n" + elif tail.endswith("\n"): + prefix = "\n" + CONSOLE_FENCE + "\n" + else: + prefix = "\n\n" + CONSOLE_FENCE + "\n" + self._console_open = True + self._emit_message(prefix + text.strip("\n") + "\n") + + # ── Tool calls ────────────────────────────────────────────── + + def tool_event(self, tool_call_id: str, status: str, **kwargs: Any) -> None: + if not tool_call_id: + return + self._close_console() + if status == "pending": + self._announce(tool_call_id, kwargs.get("name") or "tool", kwargs.get("arguments")) + elif status == "in_progress": + if self._active not in (None, tool_call_id): + self._set_status(self._active, "completed") + if tool_call_id not in self._tools: + self._announce(tool_call_id, kwargs.get("name") or "tool", kwargs.get("arguments")) + self._active = tool_call_id + self._set_status(tool_call_id, "in_progress") + elif status in ("completed", "failed"): + tool = self._tools.get(tool_call_id) + if tool is None: + return + result = kwargs.get("result") + if result and not tool.output: + # Nothing was printed while it ran; show what the model will see + tool.output.append(str(result)) + self._update(tool, content=self._content(tool)) + if self._active == tool_call_id: + self._active = None + if tool.status not in ("completed", "failed"): + self._set_status(tool_call_id, status) + + def tool_header(self, text: str) -> None: + """Start of a printed tool panel (``type="Tool Call"``).""" + self._in_panel = True + + def tool_footer(self) -> None: + self._in_panel = False + + def permission_target(self, question: str, subject: str | None) -> tuple[str, bool]: + """Pick the tool call a confirmation is about, creating one if needed. + + Returns ``(tool_call_id, synthetic)``. + """ + if self._active is not None: + return self._active, False + + pending = [t for t in self._tools.values() if t.status == "pending" and not t.synthetic] + if len(pending) == 1: + return pending[0].id, False + + self._close_console() + self._synthetic_count += 1 + tool_id = f"cecli-confirm-{self._synthetic_count}" + lowered = question.lower() + if "command" in lowered or lowered.startswith("run"): + kind = "execute" + elif "edit" in lowered or "create" in lowered or "apply" in lowered: + kind = "edit" + else: + kind = "other" + + tool = _Tool(id=tool_id, title=question.strip(), kind=kind, synthetic=True) + if subject: + tool.output.append(str(subject)) + elif pending: + tool.output.append("\n".join(t.title for t in pending)) + self._tools[tool_id] = tool + + update = { + "sessionUpdate": "tool_call", + "toolCallId": tool_id, + "title": tool.title, + "kind": kind, + "status": "pending", + } + if tool.output: + update["content"] = self._content(tool) + self._send(update) + return tool_id, True + + def permission_prompt(self, tool_call_id: str, question: str) -> dict: + """The ``toolCall`` of a permission request, showing the question above the options. + + A tool can be asked about more than once ("Run tools?", then "Allow + execution of this command?"), so the question goes in the card's content + until ``permission_resolved`` puts the tool's own content back. + """ + tool = self._tools.get(tool_call_id) + if tool is None or tool.synthetic: + return {"toolCallId": tool_call_id} # its title is already the question + question_text = {"type": "text", "text": question.strip()} + return { + "toolCallId": tool_call_id, + "content": [{"type": "content", "content": question_text}], + } + + def permission_resolved(self, tool_call_id: str, synthetic: bool, allowed: bool) -> None: + tool = self._tools.get(tool_call_id) + if tool is not None and not synthetic: + self._update(tool, content=self._content(tool) if tool.output else []) + if allowed: + if synthetic: + self._set_status(tool_call_id, "completed") + return + + # Rejected: the tool (or every tool waiting behind a group prompt) won't run + targets = [tool_call_id] + if synthetic: + targets += [ + t.id for t in self._tools.values() if t.status == "pending" and not t.synthetic + ] + for target in targets: + self._set_status(target, "failed") + + # ── Turn boundaries ───────────────────────────────────────── + + def finish_turn(self, cancelled: bool = False) -> None: + self._flush_held() + self._close_console() + final_status = "failed" if cancelled else "completed" + for tool in self._tools.values(): + if tool.status not in ("completed", "failed"): + self._set_status(tool.id, final_status) + self.reset() + + # ── Internals ─────────────────────────────────────────────── + + def _announce(self, tool_call_id: str, name: str, arguments: Any) -> None: + _server, short_name = unprefix_tool_name(name) + args = parse_arguments(arguments) + tool = _Tool( + id=tool_call_id, title=tool_title(short_name, args), kind=tool_kind(short_name) + ) + self._tools[tool_call_id] = tool + + norm = re.sub(r"[^a-z0-9]", "", short_name.lower()) + if norm == "updatetodolist": + tool.title, tool.kind = "Update plan", "think" + entries = plan_entries(args.get("tasks")) + if entries: + self._send({"sessionUpdate": "plan", "entries": entries}) + elif norm == "yield" and isinstance(args.get("summary"), str): + tool.title = "Finish" + tool.message = args["summary"] + + update = { + "sessionUpdate": "tool_call", + "toolCallId": tool_call_id, + "title": tool.title, + "kind": tool.kind, + "status": "pending", + "rawInput": args, + } + locations = self._locations(args) + if locations: + update["locations"] = locations + self._send(update) + + def _locations(self, args: dict) -> list[dict]: + for key in _PATH_KEYS: + value = args.get(key) + if isinstance(value, str) and value.strip(): + path = value.strip() + if not os.path.isabs(path): + path = os.path.join(self.root, path) + return [{"path": os.path.normpath(path)}] + return [] + + def _tool_output(self, tool_call_id: str, text: str) -> None: + tool = self._tools.get(tool_call_id) + if tool is None: + return + tool.output.append(text.rstrip("\n")) + self._update(tool, content=self._content(tool)) + + def _content(self, tool: _Tool) -> list[dict]: + text = "\n".join(tool.output) + return [{"type": "content", "content": {"type": "text", "text": _fenced(text)}}] + + def _set_status(self, tool_call_id: str, status: str) -> None: + tool = self._tools.get(tool_call_id) + if tool is None or tool.status == status: + return + tool.status = status + self._update(tool, status=status) + if status == "completed" and tool.message: + message, tool.message = tool.message, None + self._close_console() + tail = self._msg_tail + prefix = "" if not tail or tail.endswith("\n\n") else "\n\n" + self._emit_message(prefix + message.strip() + "\n\n") + + def _update(self, tool: _Tool, **fields: Any) -> None: + self._send({"sessionUpdate": "tool_call_update", "toolCallId": tool.id, **fields}) + + def _settle_tools(self) -> None: + """The model is talking again, so tools that already ran are done.""" + if self._active is not None: + self._set_status(self._active, "completed") + self._active = None + for tool in self._tools.values(): + if tool.status == "in_progress": + self._set_status(tool.id, "completed") + + def _close_console(self) -> None: + if self._console_open: + self._console_open = False + self._emit_message(CONSOLE_FENCE + "\n\n") + + def _flush_held(self) -> None: + if self._held: + held, self._held = self._held, "" + self._emit_text(held) + + def _emit_text(self, text: str) -> None: + if not text: + return + if self._thinking: + # Whitespace before a thought (e.g. the newline ahead of a reasoning + # marker) would otherwise surface as an empty message + self._pending_ws = "" + self._send( + {"sessionUpdate": "agent_thought_chunk", "content": {"type": "text", "text": text}} + ) + elif not text.strip(): + self._pending_ws += text + else: + self._emit_message(text) + + def _emit_message(self, text: str) -> None: + text, self._pending_ws = self._pending_ws + text, "" + self._msg_tail = (self._msg_tail + text)[-2:] + self._send( + {"sessionUpdate": "agent_message_chunk", "content": {"type": "text", "text": text}} + ) diff --git a/cecli/io.py b/cecli/io.py index c696e4e1c33..f4e13e88f9e 100644 --- a/cecli/io.py +++ b/cecli/io.py @@ -1569,6 +1569,29 @@ def tool_output(self, *messages, log_only=False, bold=False, type=None, **kwargs self.stream_print(*messages, style=style, **kwargs) + def tool_call_event(self, tool_call_id, status, **kwargs): + """Structured tool call progress, for front ends that render tool calls natively. + + ``status`` is "pending" when a call is announced (with ``name`` and + ``arguments``), "in_progress" when it starts running, and "completed" + once its ``result`` is known. Terminal output already shows tool panels, + so this is a no-op here; the ACP adapter overrides it. + """ + pass + + def subagent_event(self, agent_uuid, event, **kwargs): + """Sub-agent lifecycle, for front ends that show each sub-agent separately. + + ``event`` is "created" (with ``parent_uuid``, ``name``, ``description`` + and ``root``), "running" when the sub-agent starts working, "idle" when + it stops (with ``stop_reason``: "end_turn", "cancelled" or "error", and + ``error``), "message" when an agent sends it a message (with + ``sender_uuid`` and ``text``), and "removed" once it is reaped. Output + already carries each agent's ``coder_uuid``, so this is a no-op here; + the ACP adapter overrides it. + """ + pass + def escape(self, text): """Formats valid Rich tags and prints invalid ones as literal text using a single regex pass.""" diff --git a/cecli/main.py b/cecli/main.py index 4c93fbbd40e..42554098805 100644 --- a/cecli/main.py +++ b/cecli/main.py @@ -582,6 +582,21 @@ def main(argv=None, input=None, output=None, force_git_root=None, return_coder=F attach_debug_thread_monitor(argv) + if "--acp" in (sys.argv[1:] if argv is None else argv): + # From here on stdout carries the protocol. The client's first + # session/new picks the project, so move there before any setup runs. + import atexit + + from cecli.helpers.server.acp.agent import get_agent + + acp_agent = get_agent() + cwd = acp_agent.bootstrap() + if cwd is None: + return 0 + os.chdir(cwd) + # Answer anything still waiting (e.g. a failed startup) on the way out + atexit.register(acp_agent.shutdown) + # Tracks the coder instance from a ReloadProgramSignal so the new # main_async() can pass it as from_coder to Coder.create(), preserving # UUID, edit_format, and other state across the reload cycle. @@ -723,6 +738,22 @@ async def main_async( args, unknown = parser.parse_known_args(argv) os.unlink(_tmp_cfg) + if args.acp: + from cecli.helpers.server.acp.agent import get_agent + + if not get_agent().launched: + # Set by a config file or CECLI_ACP, too late: the client's first + # session/new has to pick the project before any config is read + print("Ignoring the acp setting: pass --acp on the command line", file=sys.stderr) + args.acp = False + + if args.acp: + # The ACP client owns stdin/stdout: plain linear output, no TUI or prompts + args.tui = False + args.linear_output = True + args.fancy_input = False + args.pretty = False + if args.workspaces or args.workspace_name: from cecli.helpers.workspaces.config import ( find_active_workspace_name, @@ -868,8 +899,14 @@ async def main_async( args.yes_always = True editing_mode = "VI" if args.vim else "EMACS" + io_class = InputOutput + if args.acp: + from cecli.helpers.server.acp.agent.io import AcpInputOutput + + io_class = AcpInputOutput + def get_io(pretty): - return InputOutput( + return io_class( pretty, args.yes_always, args.input_history_file, @@ -937,6 +974,10 @@ def get_io(pretty): raise err io = get_io(False) io.tool_warning("Terminal does not support pretty output (UnicodeDecodeError)") + if args.acp: + from cecli.helpers.server.acp.agent import get_agent + + get_agent().start() if args.set_env: for env_setting in args.set_env: try: diff --git a/cecli/tools/_yield.py b/cecli/tools/_yield.py index 2bfa2472c99..feaf905962a 100644 --- a/cecli/tools/_yield.py +++ b/cecli/tools/_yield.py @@ -153,7 +153,8 @@ async def execute(cls, coder, **kwargs): # Retrieve exceptions from completed sub-agent tasks so they # are not silently lost. for t in done: - if t is not interrupt_task: + # A cancelled sub-agent has no exception, and asking raises + if t is not interrupt_task and not t.cancelled(): exc = t.exception() if exc: logger.warning("Sub-agent task raised an exception: %s", exc) diff --git a/cecli/tools/broadcast.py b/cecli/tools/broadcast.py index 8792d420728..e886af5a299 100644 --- a/cecli/tools/broadcast.py +++ b/cecli/tools/broadcast.py @@ -225,6 +225,10 @@ def _deliver(info, message, agent_service, sender_coder): sender_name = agent_service.get_agent_name(sender_coder) or "primary" sender_uuid = str(sender_coder.uuid) + target_uuid = info.coder.uuid if isinstance(info, SubAgentInfo) else info.uuid + sender_coder.io.subagent_event( + str(target_uuid), "message", sender_uuid=sender_uuid, text=message + ) message = ( f"\n" f"{message}\n\n" diff --git a/cecli/website/docs/config/acp.md b/cecli/website/docs/config/acp.md new file mode 100644 index 00000000000..866d6b8a626 --- /dev/null +++ b/cecli/website/docs/config/acp.md @@ -0,0 +1,117 @@ +--- +parent: Configuration +nav_order: 41 +description: Use cecli as an agent inside editors that speak the Agent Client Protocol, such as Zed. +--- + +# Editor integration (ACP) + +`cecli --acp` runs cecli as an [Agent Client Protocol](https://agentclientprotocol.com) agent. +The editor launches cecli as a subprocess and talks to it with JSON-RPC over stdin/stdout +(ACP v1), so cecli's chat, tool calls and confirmations show up in the editor's agent panel. + +This is separate from the WebSocket server described in [ACP over WebSocket](api.html), which +streams events to custom dashboards. + +## Zed + +Add cecli to `agent_servers` in Zed's `settings.json`: + +```json +{ + "agent_servers": { + "cecli": { + "type": "custom", + "command": "cecli", + "args": ["--acp"], + "env": {} + } + } +} +``` + +Then start a new thread with cecli from the agent panel. cecli reads your usual configuration +(`~/.cecli/conf.yml`, `.cecli.conf.yml`, `.env`), so the model and API keys are the same as on +the command line. Any other flags can go in `args`, for example `["--acp", "--model", "sonnet"]`. +`--acp` itself has to be on the command line: it is ignored in config files and `CECLI_ACP`. + +Use `dev: open acp logs` from the command palette to see the messages exchanged with cecli. +cecli's own diagnostics go to stderr. + +## How cecli maps onto ACP + +- **Prompts** are handled exactly like terminal input, so slash commands work (`/add`, `/ask`, + `/model`, ...) and are offered as completions in the editor. +- **@-mentioned files** are added to the chat. Files outside the project are added read-only. +- **Model output** streams as messages; reasoning streams as thoughts. +- **Tool calls** in agent mode appear as tool cards with their output. The todo list is shown as + the editor's plan, and the final `Yield` summary is shown as a message. +- **Confirmations** ("Run tools?", "Add file to the chat?", shell commands) become permission + requests. "Yes to all" lasts for the rest of the session, just as in the terminal. + `--yes-always` skips them. +- **Cancel** interrupts the running request. +- **MCP servers** configured in the editor (Zed's context servers) are passed to cecli, which + connects them alongside its own MCP servers before the session starts. stdio, HTTP and SSE + servers are supported. If cecli already has a server with the same name, cecli's wins. +- **Shell commands** run without a pager (`PAGER` and `GIT_PAGER` are set to `cat`), since no + one can press a key to page through their output. +- cecli's own status lines (commits, lint results) appear in monospace blocks. The per-request + token and cost line is left out, but it is still written to the chat history file. + +## Mode, model and effort pickers + +cecli offers three pickers next to the editor's prompt box. Choosing a value runs the matching +command, so it behaves exactly as if you had typed it, and typing the command updates the picker. + +- **Mode**: code, ask, architect, context or agent (`/code`, `/ask`, ...), plus **Agent (YOLO)**: + agent mode with what `--yolo` turns on, so tools and shell commands run without asking (the + cost limit still asks). Choosing any other mode, or typing a command that leaves agent mode, + brings the confirmations back. Launched with `--yolo`, agent mode shows as Agent (YOLO). +- **Model**: the model that answers in the current mode. In agent mode that is the agent model + (`/agent-model`); otherwise it is the main model (`/model`). The list starts with the models + you configured (the startup `--model`, your `alias` entries, model tag overrides, and models in + `.cecli.model.settings.yml`), followed by the chat models of every provider whose API key is + set, grouped by provider. Any other model can still be chosen by typing `/model`. +- **Effort**: reasoning effort (`/reasoning-effort`) or thinking budget (`/think-tokens`), + whichever the current model accepts. It is hidden for models that accept neither. + +A key for a large provider such as OpenRouter adds hundreds of models. To list only some of +them, set `acp-models` to patterns, matched the way `/models` matches its search: a plain word +matches anywhere in the name, and `*` and `?` work as wildcards. Your configured models are +always listed. Aliases give your favorite models short names at the top of the list. For +example, in `.cecli.conf.yml`: + +```yaml +acp-models: + - "claude-sonnet" + - "gpt-5*" +alias: + - "fast:gemini/gemini-2.5-flash" + - "smart:anthropic/claude-sonnet-4-5" +``` + +## Sub-agents + +When cecli delegates work to sub-agents (the `Delegate` tool, workspace and workflow agents), +editors that support ACP subagent sessions show each sub-agent as its own session under the +thread. Its messages, tool calls and permission requests appear there instead of in the main +thread, along with whether it is working, waiting for you or idle. The task it was given and the +result it sent back show as messages between the two sessions. + +Each sub-agent can be stopped on its own, and the agent that delegated carries on with the +interrupted result. Cancelling the main thread also stops the sub-agents it is waiting for; +sub-agents working in the background keep going. + +Subagent sessions are an unstable part of ACP, so this only happens with editors that advertise +the `subagents` client capability. With other editors, sub-agent output appears in the main +thread, and a sub-agent still working after the main thread's turn has ended can't ask for +permission: those confirmations are declined. + +## Limitations + +- One process serves one project: the directory of the first session. New threads on the same + project reuse the process and start with a cleared chat; earlier threads can't be continued. +- Sessions can't be reloaded from the editor's history. +- A change to an editor MCP server's settings takes effect after restarting the agent. +- Questions that need free-form answers take their default, and confirmations asked during + startup (before any prompt) are declined. diff --git a/tests/basic/test_main.py b/tests/basic/test_main.py index fe26a786fab..cdf4f7f96fc 100644 --- a/tests/basic/test_main.py +++ b/tests/basic/test_main.py @@ -101,6 +101,14 @@ def test_main_with_empty_dir_no_files_on_command(dummy_io): main(["--no-git", "--exit", "--yes-always"], **dummy_io) +def test_main_ignores_acp_outside_command_line(dummy_io, monkeypatch, capsys): + # ACP mode starts before config is read, so only the --acp flag can turn it on + monkeypatch.setenv("CECLI_ACP", "1") + coder = main(["--no-git", "--exit", "--yes-always"], **dummy_io, return_coder=True) + assert type(coder.io._target) is InputOutput + assert "pass --acp on the command line" in capsys.readouterr().err + + def test_main_with_empty_dir_new_file(dummy_io): main(["foo.txt", "--yes-always", "--no-git", "--exit"], **dummy_io) assert os.path.exists("foo.txt") diff --git a/tests/helpers/server/acp/test_agent.py b/tests/helpers/server/acp/test_agent.py new file mode 100644 index 00000000000..e72767c7ef2 --- /dev/null +++ b/tests/helpers/server/acp/test_agent.py @@ -0,0 +1,998 @@ +"""Tests for the ACP v1 agent (``cecli --acp``).""" + +from __future__ import annotations + +import asyncio +import os +import threading +from types import SimpleNamespace + +import pytest + +from cecli import models +from cecli.helpers.server.acp.agent import config as acp_config +from cecli.helpers.server.acp.agent.agent import AcpAgent, prompt_to_text +from cecli.helpers.server.acp.agent.config import SessionConfig, current_value +from cecli.helpers.server.acp.agent.io import AcpInputOutput +from cecli.helpers.server.acp.agent.mcp import ClientMcpServers, to_cecli_configs +from cecli.helpers.server.acp.agent.transport import ( + EOF, + INTERNAL_ERROR, + INVALID_PARAMS, + METHOD_NOT_FOUND, +) +from cecli.helpers.server.acp.agent.updates import UpdateEmitter +from cecli.reasoning_tags import REASONING_END, REASONING_START + + +@pytest.fixture(autouse=True) +def configured_models(monkeypatch): + """User aliases and model files as main() would register them, without reading ~.""" + monkeypatch.setitem(models.MODEL_ALIASES, "fast", "gpt-x") + monkeypatch.setitem(models.MODEL_ALIASES, "smart", "big-model") + monkeypatch.setattr( + models.ModelOverrides, "_overrides", {"gpt-y": {"high": {"reasoning_effort": "high"}}} + ) + monkeypatch.setattr(acp_config, "settings_file_models", lambda coder: ["local/qwen"]) + # No provider API keys, whatever the environment running the tests has + monkeypatch.setattr(acp_config, "keyed_provider_models", lambda: []) + + +# ── UpdateEmitter ────────────────────────────────────────────── + + +def make_emitter(): + updates = [] + return UpdateEmitter(updates.append), updates + + +def texts(updates, kind): + return "".join(u["content"]["text"] for u in updates if u["sessionUpdate"] == kind) + + +def test_reasoning_markers_split_across_chunks(): + emitter, updates = make_emitter() + stream = f"\n{REASONING_START}pondering{REASONING_END}\nThe answer." + for i in range(0, len(stream), 3): + emitter.stream(stream[i : i + 3]) + emitter.stream("", final=True) + + assert texts(updates, "agent_thought_chunk").strip() == "pondering" + assert texts(updates, "agent_message_chunk").strip() == "The answer." + # The newline ahead of the marker must not become its own message + assert updates[0]["sessionUpdate"] == "agent_thought_chunk" + + +def test_console_lines_share_one_fence_closed_by_model_text(): + emitter, updates = make_emitter() + emitter.stream("Intro.") + emitter.block("line one") + emitter.block("line two") + emitter.stream("More prose.") + + text = texts(updates, "agent_message_chunk") + assert text == "Intro.\n\n````\nline one\nline two\n````\n\nMore prose." + + +def test_tool_lifecycle_and_output(): + emitter, updates = make_emitter() + emitter.root = "/proj" + emitter.tool_event("t1", "pending", name="Local--ReadFile", arguments='{"file_path": "a.py"}') + emitter.tool_header("Tool Call: Local • ReadFile") + emitter.block("Arguments: {...}") # panel body is dropped + emitter.tool_footer() + emitter.tool_event("t1", "in_progress") + emitter.block("Viewed a.py") + emitter.tool_event("t1", "completed", result="contents") + + call = updates[0] + assert call["sessionUpdate"] == "tool_call" + assert call["title"] == "ReadFile: a.py" + assert call["kind"] == "read" + assert call["locations"] == [{"path": "/proj/a.py"}] + assert call["rawInput"] == {"file_path": "a.py"} + + statuses = [u.get("status") for u in updates if u["sessionUpdate"] == "tool_call_update"] + assert statuses == ["in_progress", None, "completed"] + content = updates[2]["content"][0]["content"]["text"] + assert "Viewed a.py" in content and "Arguments" not in content + assert texts(updates, "agent_message_chunk") == "" + + +def test_result_shown_when_tool_printed_nothing(): + emitter, updates = make_emitter() + emitter.tool_event("t1", "pending", name="Grep", arguments="{}") + emitter.tool_event("t1", "in_progress") + emitter.tool_event("t1", "completed", result="match.py:3") + + assert "match.py:3" in updates[-2]["content"][0]["content"]["text"] + assert updates[-1]["status"] == "completed" + + +def test_model_text_settles_running_tools(): + emitter, updates = make_emitter() + emitter.tool_event("t1", "pending", name="Command", arguments='{"command": "ls"}') + emitter.tool_event("t1", "in_progress") + emitter.stream("Done.") + + # Not "execute", which would hide Zed's progress spinner while it runs + assert updates[0]["kind"] == "other" + assert { + "sessionUpdate": "tool_call_update", + "toolCallId": "t1", + "status": "completed", + } in updates + + +def test_finish_turn_fails_open_tools_when_cancelled(): + emitter, updates = make_emitter() + emitter.tool_event("t1", "pending", name="Command", arguments="{}") + emitter.tool_event("t1", "in_progress") + emitter.finish_turn(cancelled=True) + + assert updates[-1] == { + "sessionUpdate": "tool_call_update", + "toolCallId": "t1", + "status": "failed", + } + + +def test_todo_list_becomes_plan_and_yield_summary_a_message(): + emitter, updates = make_emitter() + tasks = [ + {"task": "Read", "done": True}, + {"task": "Write", "done": False, "current": True}, + {"task": "Test", "done": False}, + ] + emitter.tool_event("t1", "pending", name="UpdateTodoList", arguments={"tasks": tasks}) + plan = next(u for u in updates if u["sessionUpdate"] == "plan") + assert [e["status"] for e in plan["entries"]] == ["completed", "in_progress", "pending"] + + emitter.tool_event("t2", "pending", name="Yield", arguments={"summary": "All done."}) + emitter.tool_event("t2", "in_progress") + emitter.tool_event("t2", "completed", result="Yielded.") + assert texts(updates, "agent_message_chunk").strip() == "All done." + + +def test_permission_target_prefers_the_obvious_tool(): + emitter, updates = make_emitter() + emitter.tool_event("t1", "pending", name="Command", arguments="{}") + assert emitter.permission_target("Run tools?", None) == ("t1", False) + + emitter.tool_event("t2", "pending", name="Grep", arguments="{}") + tool_id, synthetic = emitter.permission_target("Run tools?", None) + assert synthetic and tool_id.startswith("cecli-confirm-") + + # Rejecting a group prompt fails every tool waiting behind it + emitter.permission_resolved(tool_id, True, allowed=False) + failed = {u["toolCallId"] for u in updates if u.get("status") == "failed"} + assert failed == {tool_id, "t1", "t2"} + + +def test_permission_question_shows_until_answered(): + emitter, updates = make_emitter() + emitter.tool_event("t1", "pending", name="Command", arguments='{"command": "ls"}') + emitter.tool_event("t1", "in_progress") + + # A tool can be asked about twice; each request carries its own question + for question in ("Allow execution of this command?", "Allow this command for the session?"): + prompt = emitter.permission_prompt("t1", question) + assert prompt["content"][0]["content"]["text"] == question + emitter.permission_resolved("t1", False, allowed=True) + assert updates[-1] == { + "sessionUpdate": "tool_call_update", + "toolCallId": "t1", + "content": [], + } + + # Output printed later replaces the content as usual + emitter.block("a.py") + assert "a.py" in updates[-1]["content"][0]["content"]["text"] + assert emitter.permission_prompt("cecli-confirm-9", "Run?") == {"toolCallId": "cecli-confirm-9"} + + +# ── Prompt conversion ────────────────────────────────────────── + + +def test_prompt_to_text(tmp_path): + root = str(tmp_path) + blocks = [ + {"type": "text", "text": "Explain"}, + {"type": "resource_link", "name": "a.py", "uri": (tmp_path / "a.py").as_uri()}, + { + "type": "resource", + "resource": {"uri": (tmp_path / "b.py").as_uri() + "#L2:3", "text": "x = 1"}, + }, + {"type": "resource", "resource": {"uri": "zed:///rules", "text": "Be terse."}}, + ] + text, paths = prompt_to_text(blocks, root) + + assert text.startswith("Explain a.py") + assert "b.py:\n```\nx = 1\n```" in text + assert "Be terse." in text + assert [os.path.normpath(p) for p in paths] == [str(tmp_path / "a.py"), str(tmp_path / "b.py")] + + +# ── AcpAgent ─────────────────────────────────────────────────── + + +class FakeTransport: + def __init__(self): + self.sent = [] + self._inbox = None + + def attach(self, loop): + self._inbox = asyncio.Queue() + + def detach(self): + pass + + async def receive(self): + return await self._inbox.get() + + def feed(self, msg): + self._inbox.put_nowait(msg) + + def send(self, msg): + self.sent.append(msg) + + def results(self): + return {m["id"]: m for m in self.sent if "method" not in m} + + def updates(self): + return [m["params"]["update"] for m in self.sent if m.get("method") == "session/update"] + + +class StubModel: + def __init__(self, name, accepts=("reasoning_effort",), effort=None, budget=None): + self.name = name + self.accepts_settings = list(accepts) + self.effort = effort + self.budget = budget + + def get_reasoning_effort(self): + return self.effort + + def get_raw_thinking_tokens(self): + return self.budget + + def get_thinking_tokens(self): + return None if self.budget is None else f"{self.budget // 1024}k" + + +class StubCoder: + def __init__(self, root, model=None, edit_format="diff"): + self.root = str(root) + self.abs_fnames = set() + self.abs_read_only_fnames = set() + self.io = SimpleNamespace(output_task=None, yes=None, tool_output=lambda *a, **k: None) + self.interrupt_event = threading.Event() + self.main_model = model or StubModel("gpt-x") + self.agent_model = None + self.edit_format = edit_format + self.args = SimpleNamespace(model="fast", alias=["fast:gpt-x", "smart:big-model"]) + self.repo = None + + def get_active_model(self): + if self.edit_format == "agent" and self.agent_model: + return self.agent_model + return self.main_model + + async def forever(self): + await asyncio.sleep(3600) + + +def make_coder(root): + return StubCoder(root) + + +async def ready_agent(tmp_path): + """An agent past bootstrap, with session/new (id 2) deferred.""" + transport = FakeTransport() + agent = AcpAgent(transport) + agent.cwd = str(tmp_path) + agent.session_id = "s1" + agent._deferred_new.append(2) + agent.start() + return agent, transport + + +async def settle(): + for _ in range(5): + await asyncio.sleep(0) + + +def prompt(request_id, text, session_id="s1"): + return { + "jsonrpc": "2.0", + "id": request_id, + "method": "session/prompt", + "params": {"sessionId": session_id, "prompt": [{"type": "text", "text": text}]}, + } + + +async def test_initialize_and_unknown_method(tmp_path): + agent, transport = await ready_agent(tmp_path) + transport.feed({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}}) + transport.feed({"jsonrpc": "2.0", "id": 9, "method": "session/load", "params": {}}) + await settle() + + result = transport.results()[1]["result"] + assert result["protocolVersion"] == 1 + assert result["agentCapabilities"]["loadSession"] is False + assert transport.results()[9]["error"]["code"] == METHOD_NOT_FOUND + + +async def test_turn_lifecycle(tmp_path): + agent, transport = await ready_agent(tmp_path) + coder = make_coder(tmp_path) + + # Prompts are rejected until the coder is ready (the client has no id yet) + transport.feed(prompt(3, "too early")) + await settle() + assert "error" in transport.results()[3] + + waiting = asyncio.create_task(agent.next_input(coder)) + await settle() + session = transport.results()[2]["result"] + assert session["sessionId"] == "s1" + assert [o["id"] for o in session["configOptions"]] == ["mode", "model", "reasoning_effort"] + assert session["modes"]["currentModeId"] == "code" + assert transport.updates()[0]["sessionUpdate"] == "available_commands_update" + names = {c["name"] for c in transport.updates()[0]["availableCommands"]} + assert "add" in names and "exit" not in names + + transport.feed(prompt(4, "hello")) + assert await waiting == "hello" + + agent.emitter.stream("Hi there.", final=True) + assert 4 not in transport.results() + + # The loop asking for input again ends the turn + waiting = asyncio.create_task(agent.next_input(coder)) + await settle() + assert transport.results()[4]["result"] == {"stopReason": "end_turn"} + # The turn's updates all precede its response + kinds = [m.get("method") or m.get("id") for m in transport.sent] + assert kinds[-2:] == ["session/update", 4] + assert transport.updates()[-1]["content"]["text"] == "Hi there." + waiting.cancel() + + +async def test_cancel_interrupts_coder(tmp_path): + agent, transport = await ready_agent(tmp_path) + coder = make_coder(tmp_path) + waiting = asyncio.create_task(agent.next_input(coder)) + await settle() + transport.feed(prompt(4, "long job")) + await waiting + + coder.io.output_task = asyncio.create_task(coder.forever()) + transport.feed({"jsonrpc": "2.0", "method": "session/cancel", "params": {"sessionId": "s1"}}) + await settle() + assert coder.io.output_task.cancelled() + assert coder.interrupt_event.is_set() + + waiting = asyncio.create_task(agent.next_input(coder)) + await settle() + assert transport.results()[4]["result"] == {"stopReason": "cancelled"} + waiting.cancel() + + +async def test_permission_roundtrip_through_confirm_ask(tmp_path): + agent, transport = await ready_agent(tmp_path) + io = AcpInputOutput(agent=agent) + waiting = asyncio.create_task(agent.next_input(make_coder(tmp_path))) + await settle() + transport.feed(prompt(4, "do it")) + await waiting + + asking = asyncio.create_task(io.confirm_ask("Run tools?", group_response="Run MCP Tools")) + await settle() + request = next(m for m in transport.sent if m.get("method") == "session/request_permission") + assert {o["optionId"] for o in request["params"]["options"]} == {"y", "a", "s", "n"} + + transport.feed( + { + "jsonrpc": "2.0", + "id": request["id"], + "result": {"outcome": {"outcome": "selected", "optionId": "a"}}, + } + ) + assert await asking is True + # "Yes to all" sticks, so the next group prompt is not sent to the client + assert await io.confirm_ask("Run tools?", group_response="Run MCP Tools") is True + assert sum(m.get("method") == "session/request_permission" for m in transport.sent) == 1 + + +async def test_confirm_outside_a_prompt_declines(tmp_path): + agent, transport = await ready_agent(tmp_path) + io = AcpInputOutput(agent=agent) + assert await io.confirm_ask("Add .cecli* to .gitignore?") is False + assert not any(m.get("method") == "session/request_permission" for m in transport.sent) + + +async def test_usage_report_is_not_shown(tmp_path): + agent, transport = await ready_agent(tmp_path) + io = AcpInputOutput(agent=agent) + waiting = asyncio.create_task(agent.next_input(make_coder(tmp_path))) + await settle() + transport.feed(prompt(4, "hi")) + await waiting + + io.tool_output("18k ◇ 95.4% ↑ 139 ↓ $0.00015", type="usage-report") + io.tool_output("Added a.py to the chat") + agent.emitter.stream("Done.", final=True) + shown = "".join(u["content"]["text"] for u in transport.updates() if "content" in u) + assert "Added a.py" in shown + assert "$0.00015" not in shown + + +async def test_eof_ends_the_input_loop(tmp_path): + agent, transport = await ready_agent(tmp_path) + waiting = asyncio.create_task(agent.next_input(make_coder(tmp_path))) + await settle() + transport.feed(EOF) + with pytest.raises(EOFError): + await waiting + + +async def test_second_session_clears_chat(tmp_path): + agent, transport = await ready_agent(tmp_path) + coder = make_coder(tmp_path) + waiting = asyncio.create_task(agent.next_input(coder)) + await settle() + + other = tmp_path / "other" + other.mkdir() + transport.feed( + {"jsonrpc": "2.0", "id": 5, "method": "session/new", "params": {"cwd": str(other)}} + ) + transport.feed( + {"jsonrpc": "2.0", "id": 6, "method": "session/new", "params": {"cwd": str(tmp_path)}} + ) + await settle() + assert transport.results()[5]["error"]["code"] == INVALID_PARAMS + + # The new session runs /clear as an internal turn before it is answered + assert await waiting == "/clear" + assert 6 not in transport.results() + waiting = asyncio.create_task(agent.next_input(coder)) + await settle() + new_id = transport.results()[6]["result"]["sessionId"] + assert new_id != "s1" and agent.session_id == new_id + + transport.feed(prompt(7, "stale", session_id="s1")) + await settle() + assert "error" in transport.results()[7] + waiting.cancel() + + +# ── Sub-agents ───────────────────────────────────────────────── + + +def notifications(transport): + """``(sessionId, update)`` for each session/update sent.""" + return [ + (m["params"]["sessionId"], m["params"]["update"]) + for m in transport.sent + if m.get("method") == "session/update" + ] + + +def subagent_states(transport): + return [ + (session_id, update["sessionId"], update["state"]) + for session_id, update in notifications(transport) + if update["sessionUpdate"] == "subagent_update" and "state" in update + ] + + +async def delegating_agent(tmp_path, subagents=True): + """An agent in the middle of a prompt, for a client with or without subagent support.""" + agent, transport = await ready_agent(tmp_path) + capabilities = {"subagents": {}} if subagents else {} + transport.feed( + { + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": {"protocolVersion": 1, "clientCapabilities": capabilities}, + } + ) + coder = make_coder(tmp_path) + coder.uuid = "primary" + waiting = asyncio.create_task(agent.next_input(coder)) + await settle() + transport.feed(prompt(4, "delegate this")) + await waiting + transport.sent.clear() + # The agent holds the coder weakly; the caller keeps it alive, as cecli's loop would + return agent, transport, AcpInputOutput(agent=agent), coder + + +async def test_subagent_is_a_child_session(tmp_path): + agent, transport, io, _coder = await delegating_agent(tmp_path) + io.subagent_event( + "w1", "created", parent_uuid="primary", name="worker", description="Does work", root="/w" + ) + io.subagent_event("w1", "message", sender_uuid="primary", text="Fix the tests") + io.subagent_event("w1", "running") + io.stream_output("On it.", final=True, coder_uuid="w1") + io.tool_call_event("t1", "pending", coder_uuid="w1", name="ReadFile", arguments='{"path": "a"}') + io.stream_output("Meanwhile, the parent.", final=True, coder_uuid="primary") + io.subagent_event("w1", "idle", stop_reason="end_turn", error=None) + + sent = notifications(transport) + child = agent.subagents["w1"].session_id + # Announced on the parent's session before anything names the child + assert sent[0] == ( + "s1", + { + "sessionUpdate": "subagent_update", + "sessionId": child, + "capabilities": {"cancel": {}}, + "title": "worker", + "description": "Does work", + }, + ) + # The delegation as the parent sent it, then as the child got it + (out_session, outgoing), (in_session, incoming) = sent[1], sent[2] + assert (out_session, in_session) == ("s1", child) + for message in (outgoing, incoming): + assert message["sessionUpdate"] == "session_message" + assert (message["senderSessionId"], message["recipientSessionId"]) == ("s1", child) + assert message["content"] == [{"type": "text", "text": "Fix the tests"}] + assert outgoing["messageId"] != incoming["messageId"] + + # Each agent's output lands in its own session + child_updates = [update for session_id, update in sent if session_id == child] + assert texts(child_updates, "agent_message_chunk") == "On it." + call = next(u for u in child_updates if u["sessionUpdate"] == "tool_call") + assert call["locations"] == [{"path": os.path.normpath("/w/a")}] + parent_updates = [update for session_id, update in sent if session_id == "s1"] + assert texts(parent_updates, "agent_message_chunk") == "Meanwhile, the parent." + + # Going idle settles the child's tools, and the parent's roster shows its state + assert child_updates[-1]["status"] == "completed" + assert subagent_states(transport) == [ + ("s1", child, {"state": "running"}), + ("s1", child, {"state": "idle", "stopReason": "end_turn"}), + ] + + +async def test_subagents_stay_in_the_parent_session_without_client_support(tmp_path): + agent, transport, io, _coder = await delegating_agent(tmp_path, subagents=False) + io.subagent_event("w1", "created", parent_uuid="primary", name="worker") + io.subagent_event("w1", "message", sender_uuid="primary", text="Fix the tests") + io.subagent_event("w1", "running") + io.stream_output("On it.", final=True, coder_uuid="w1") + + sent = notifications(transport) + assert [(session_id, u["sessionUpdate"]) for session_id, u in sent] == [ + ("s1", "agent_message_chunk") + ] + + +async def test_nested_subagent_reports_to_its_parent(tmp_path): + agent, transport, io, _coder = await delegating_agent(tmp_path) + io.subagent_event("w1", "created", parent_uuid="primary", name="worker") + io.subagent_event("h1", "created", parent_uuid="w1", name="helper") + io.subagent_event("h1", "running") + io.subagent_event("h1", "idle", stop_reason="error", error="boom") + io.subagent_event("w1", "message", sender_uuid="h1", text="The helper agent failed") + io.subagent_event("h1", "removed") + + child, grandchild = agent.subagents["w1"].session_id, agent.subagents["h1"].session_id + sent = notifications(transport) + assert sent[1][0] == child and sent[1][1]["sessionId"] == grandchild + assert subagent_states(transport) == [ + (child, grandchild, {"state": "running"}), + ( + child, + grandchild, + { + "state": "idle", + "stopReason": "error", + "error": {"code": INTERNAL_ERROR, "message": "boom"}, + }, + ), + ] + reply = [(s, u) for s, u in sent if u["sessionUpdate"] == "session_message"] + assert [s for s, _ in reply] == [grandchild, child] + assert all(u["senderSessionId"] == grandchild for _, u in reply) + # Once reaped, it can no longer be cancelled + assert sent[-1] == ( + child, + {"sessionUpdate": "subagent_update", "sessionId": grandchild, "capabilities": None}, + ) + + +def answer(transport, request, option_id): + transport.feed( + { + "jsonrpc": "2.0", + "id": request["id"], + "result": {"outcome": {"outcome": "selected", "optionId": option_id}}, + } + ) + + +async def test_subagent_asks_permission_in_its_own_session(tmp_path): + agent, transport, io, _coder = await delegating_agent(tmp_path) + io.subagent_event("w1", "created", parent_uuid="primary", name="worker") + io.subagent_event("w1", "running") + child = agent.subagents["w1"].session_id + # The parent's turn is over, but the sub-agent works on in the background + agent._finish_turn() + + asking = asyncio.create_task(io.confirm_ask("Run tools?", coder_uuid="w1")) + await settle() + request = next(m for m in transport.sent if m.get("method") == "session/request_permission") + assert request["params"]["sessionId"] == child + answer(transport, request, "y") + assert await asking is True + assert [state["state"] for _, _, state in subagent_states(transport)] == [ + "running", + "requires_action", + "running", + ] + + +async def test_cancelling_a_subagent_leaves_the_rest_running(tmp_path, monkeypatch): + from cecli.helpers.agents.service import AgentService + + cancelled = [] + service = SimpleNamespace(cancel_sub_agent=cancelled.append) + monkeypatch.setattr(AgentService, "get_instance", staticmethod(lambda coder: service)) + + agent, transport, io, _coder = await delegating_agent(tmp_path) + io.subagent_event("w1", "created", parent_uuid="primary", name="worker") + io.subagent_event("w1", "running") + child = agent.subagents["w1"].session_id + asking = asyncio.create_task(io.confirm_ask("Run tools?", coder_uuid="w1")) + await settle() + + # Cancelling the parent's turn leaves a background sub-agent's question open + transport.feed({"jsonrpc": "2.0", "method": "session/cancel", "params": {"sessionId": "s1"}}) + await settle() + assert not asking.done() and cancelled == [] + + transport.feed({"jsonrpc": "2.0", "method": "session/cancel", "params": {"sessionId": child}}) + await settle() + assert cancelled == ["w1"] + assert await asking is False + + +# ── Config options ───────────────────────────────────────────── + + +def model_groups(option): + """``{group name: [values]}`` of the grouped model picker.""" + return {group["name"]: [v["value"] for v in group["options"]] for group in option["options"]} + + +def test_config_options_list_configured_models(tmp_path): + options = SessionConfig().options(StubCoder(tmp_path)) + mode, model, effort = options + + assert mode["currentValue"] == "code" # any edit format other than a chat mode + assert [v["value"] for v in mode["options"]] == [ + "code", + "ask", + "architect", + "context", + "agent", + "agent-yolo", + ] + + # Startup model and its alias collapse into one entry; tag overrides stay distinct + assert model_groups(model) == {"Configured": ["fast", "smart", "gpt-y:high", "local/qwen"]} + assert model["currentValue"] == "fast" + assert model["options"][0]["options"][1]["description"] == "big-model" + + assert effort["category"] == "thought_level" + assert effort["currentValue"] == "default" + assert [v["value"] for v in effort["options"]] == ["default", "low", "medium", "high"] + + +def test_config_options_follow_the_running_model(tmp_path): + config = SessionConfig() + coder = StubCoder(tmp_path, model=StubModel("other-model", accepts=(), effort="xhigh")) + model = current_value(config.options(coder), "model") + assert model == "other-model" # switched with a typed /model: listed first + + options = config.options(coder) + assert model_groups(options[1])["Configured"][0] == "other-model" + # An effort set elsewhere is shown even though it is not a standard choice + assert options[2]["currentValue"] == "xhigh" + + coder.main_model = StubModel("claude-x", accepts=("thinking_tokens",), budget=2048) + effort = config.options(coder)[2] + assert effort["id"] == "thinking_tokens" + assert effort["currentValue"] == "2k" + # A model's own default budget slots in by size + assert [v["value"] for v in effort["options"]] == ["0", "2k", "4k", "8k", "16k", "32k"] + + coder.main_model = StubModel("plain", accepts=()) + assert [o["id"] for o in config.options(coder)] == ["mode", "model"] + + +def test_model_picker_lists_providers_with_a_key(tmp_path, monkeypatch): + monkeypatch.setattr( + acp_config, + "keyed_provider_models", + lambda: [ + ("openai", "OpenAI", ["openai/gpt-5", "openai/gpt-5-mini"]), + ("anthropic", "Anthropic", ["anthropic/claude-sonnet-4-5", "big-model"]), + ], + ) + coder = StubCoder(tmp_path) + config = SessionConfig() + + model = config.options(coder)[1] + # "big-model" is already listed as the "smart" alias + assert model_groups(model) == { + "Configured": ["fast", "smart", "gpt-y:high", "local/qwen"], + "OpenAI": ["openai/gpt-5", "openai/gpt-5-mini"], + "Anthropic": ["anthropic/claude-sonnet-4-5"], + } + assert config.command(coder, "model", "openai/gpt-5-mini") == "/model openai/gpt-5-mini" + + # The filter narrows the provider lists; configured models always stay + coder.args.acp_models = ["mini", "claude-*-4"] + model = SessionConfig().options(coder)[1] + assert model_groups(model) == { + "Configured": ["fast", "smart", "gpt-y:high", "local/qwen"], + "OpenAI": ["openai/gpt-5-mini"], + "Anthropic": ["anthropic/claude-sonnet-4-5"], + } + + +def test_model_filter_matches_like_models_command(): + assert acp_config.matches_filter("anthropic/claude-sonnet-4-5", ["SONNET"]) + assert acp_config.matches_filter("openrouter/anthropic/claude-opus-4", ["anthropic/*"]) + assert not acp_config.matches_filter("openai/gpt-5", ["claude", "gemini-*"]) + + +def test_config_choices_map_to_commands(tmp_path): + config = SessionConfig() + coder = StubCoder(tmp_path, model=StubModel("gpt-x", effort="high")) + + assert config.command(coder, "mode", "ask") == "/ask" + assert config.command(coder, "mode", "code") is None # already current + assert config.command(coder, "model", "smart") == "/model smart" + assert config.model_choice == "smart" + assert config.command(coder, "reasoning_effort", "default") == "/reasoning-effort none" + assert config.command(coder, "reasoning_effort", "low") == "/reasoning-effort low" + with pytest.raises(ValueError): + config.command(coder, "model", "not-configured") + with pytest.raises(ValueError): + config.command(coder, "thinking_tokens", "8k") # not offered for this model + + coder.main_model = StubModel("claude-x", accepts=("thinking_tokens",), budget=8192) + assert config.command(coder, "thinking_tokens", "0") == "/think-tokens 0" + + +def test_agent_mode_picks_the_agent_model(tmp_path): + config = SessionConfig() + coder = StubCoder(tmp_path, edit_format="agent") + coder.agent_model = StubModel("big-model", effort="low") + + options = config.options(coder) + assert current_value(options, "model") == "smart" + assert current_value(options, "reasoning_effort") == "low" + assert config.command(coder, "model", "fast") == "/agent-model fast" + + +def config_request(request_id, config_id, value, session_id="s1"): + return { + "jsonrpc": "2.0", + "id": request_id, + "method": "session/set_config_option", + "params": {"sessionId": session_id, "configId": config_id, "value": value}, + } + + +async def test_set_config_option_runs_the_cecli_command(tmp_path): + agent, transport = await ready_agent(tmp_path) + waiting = asyncio.create_task(agent.next_input(StubCoder(tmp_path))) + await settle() + + # Unchanged values answer at once; unknown ones are rejected + transport.feed(config_request(3, "mode", "code")) + transport.feed(config_request(4, "mode", "yolo")) + await settle() + assert current_value(transport.results()[3]["result"]["configOptions"], "mode") == "code" + assert transport.results()[4]["error"]["code"] == INVALID_PARAMS + + transport.feed(config_request(5, "mode", "ask")) + assert await waiting == "/ask" + assert 5 not in transport.results() + + # /ask replaces the coder; the answer reflects the new one + waiting = asyncio.create_task(agent.next_input(StubCoder(tmp_path, edit_format="ask"))) + await settle() + options = transport.results()[5]["result"]["configOptions"] + assert current_value(options, "mode") == "ask" + # The client already has these options from the answer + assert not any(u["sessionUpdate"] == "config_option_update" for u in transport.updates()) + waiting.cancel() + + +async def test_typed_commands_update_the_pickers(tmp_path): + agent, transport = await ready_agent(tmp_path) + coder = StubCoder(tmp_path) + waiting = asyncio.create_task(agent.next_input(coder)) + await settle() + transport.feed(prompt(3, "/ask")) + await waiting + + waiting = asyncio.create_task(agent.next_input(StubCoder(tmp_path, edit_format="ask"))) + await settle() + kinds = [u["sessionUpdate"] for u in transport.updates()] + assert kinds[-2:] == ["config_option_update", "current_mode_update"] + assert transport.updates()[-1]["currentModeId"] == "ask" + # The pickers change before the prompt's response arrives + assert transport.sent[-1]["id"] == 3 + + # Legacy session/set_mode takes the same path + transport.feed( + { + "jsonrpc": "2.0", + "id": 4, + "method": "session/set_mode", + "params": {"sessionId": "s1", "modeId": "agent"}, + } + ) + assert await waiting == "/agent" + waiting = asyncio.create_task(agent.next_input(StubCoder(tmp_path, edit_format="agent"))) + await settle() + assert transport.results()[4]["result"] == {} + assert transport.updates()[-1] == { + "sessionUpdate": "current_mode_update", + "currentModeId": "agent", + } + waiting.cancel() + + +def test_yolo_mode_toggles_what_yolo_sets(tmp_path): + config = SessionConfig() + coder = StubCoder(tmp_path, edit_format="agent") + coder.io.yes = None + + # From Agent, YOLO is only a settings change: no coder switch needed + assert config.command(coder, "mode", "agent-yolo") is None + assert coder.io.yes is True and coder.args.yes_always_commands is True + assert current_value(config.options(coder), "mode") == "agent-yolo" + + # Back to Agent restores the confirmations + assert config.command(coder, "mode", "agent") is None + assert coder.io.yes is None and coder.args.yes_always_commands is False + + # From another mode, YOLO switches to agent mode too; leaving it turns it off + coder.edit_format = "ask" + assert config.command(coder, "mode", "agent-yolo") == "/agent" + coder.edit_format = "agent" + assert config.command(coder, "mode", "code") == "/code" + assert coder.io.yes is None and coder.args.yes_always_commands is False + + # Typing /ask while in YOLO ends it too, so the picker never hides it + assert config.command(coder, "mode", "agent-yolo") is None + coder.edit_format = "ask" + assert current_value(config.options(coder), "mode") == "ask" + assert coder.io.yes is None and coder.args.yes_always_commands is False + coder.edit_format = "agent" + + # Launched with --yolo, agent mode already shows as YOLO + coder.args.yes_always_commands, coder.io.yes = True, True + assert current_value(config.options(coder), "mode") == "agent-yolo" + assert config.command(coder, "mode", "agent") is None + assert coder.io.yes is None and coder.args.yes_always_commands is False + + +# ── Client MCP servers ───────────────────────────────────────── + +ZED_SERVERS = [ + { + "name": "GitHub", + "command": "/usr/local/bin/github-mcp", + "args": ["stdio"], + "env": [{"name": "TOKEN", "value": "t"}], + }, + {"type": "http", "name": "docs", "url": "https://example.com/mcp", "headers": []}, + {"type": "sse", "name": "events", "url": "https://example.com/sse"}, + {"name": "broken"}, # no command: ignored +] + + +def test_client_mcp_servers_become_cecli_configs(): + configs = to_cecli_configs(ZED_SERVERS) + assert set(configs) == {"github", "docs", "events"} + assert configs["github"] == { + "transport": "stdio", + "command": "/usr/local/bin/github-mcp", + "args": ["stdio"], + "env": {"TOKEN": "t"}, + "name": "GitHub", + } + assert configs["docs"]["transport"] == "http" + assert configs["events"] == { + "transport": "sse", + "url": "https://example.com/sse", + "headers": {}, + "name": "events", + } + + +class FakeMcpManager: + def __init__(self, *names): + self.servers = {n: SimpleNamespace(name=n) for n in names} + self.connected = set(names) + self.calls = [] + + def get_server(self, name): + return self.servers.get(name.lower()) + + @property + def connected_servers(self): + return [self.servers[n] for n in self.connected] + + async def add_server(self, server, connect=False): + self.servers[server.name] = server + self.calls.append(("add", server.name, type(server).__name__)) + if connect: + self.connected.add(server.name) + return True + + async def connect_server(self, name): + self.connected.add(name) + self.calls.append(("connect", name)) + return True + + async def disconnect_server(self, name): + self.connected.discard(name) + self.calls.append(("disconnect", name)) + return True + + +async def test_client_mcp_servers_sync_with_the_manager(): + logged = [] + servers = ClientMcpServers(logged.append) + manager = FakeMcpManager("docs") # cecli's own config already has "docs" + + servers.want(ZED_SERVERS) + await servers.sync(manager) + assert manager.calls == [ + ("add", "github", "McpServer"), + ("add", "events", "SseServer"), + ] + assert any("cecli has one by that name" in line for line in logged) + + # A later session drops one server and keeps the other + manager.calls.clear() + servers.want(ZED_SERVERS[:1]) + await servers.sync(manager) + assert manager.calls == [("disconnect", "events")] + + # Nothing pending: nothing happens + manager.calls.clear() + await servers.sync(manager) + assert manager.calls == [] + + +async def test_session_new_connects_client_mcp_servers_first(tmp_path): + agent, transport = await ready_agent(tmp_path) + transport.feed({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}}) + await settle() + capabilities = transport.results()[1]["result"]["agentCapabilities"] + assert capabilities["mcpCapabilities"] == {"http": True, "sse": True} + + agent.mcp.want(ZED_SERVERS[:1]) # as bootstrap() records the first session/new + coder = StubCoder(tmp_path) + coder.mcp_manager = FakeMcpManager() + waiting = asyncio.create_task(agent.next_input(coder)) + await settle() + assert coder.mcp_manager.calls == [("add", "github", "McpServer")] + assert transport.results()[2]["result"]["sessionId"] == "s1" + waiting.cancel() diff --git a/tests/helpers/test_onboarding.py b/tests/helpers/test_onboarding.py index f409dcb34f3..e2fdf5b7278 100644 --- a/tests/helpers/test_onboarding.py +++ b/tests/helpers/test_onboarding.py @@ -62,6 +62,14 @@ def test_get_models_for_provider_openai(): assert "openai/gpt-4o" in models +def test_get_models_for_provider_filters_by_mode(): + every_model = get_models_for_provider("gemini") + chat_models = get_models_for_provider("gemini", modes=("chat",)) + + assert chat_models + assert set(chat_models) < set(every_model) # image generation models are left out + + def test_get_models_for_provider_anthropic_prefixed(): models = get_models_for_provider("anthropic") diff --git a/tests/subagents/test_service.py b/tests/subagents/test_service.py index 974af318a9c..f6ee8469a5b 100644 --- a/tests/subagents/test_service.py +++ b/tests/subagents/test_service.py @@ -10,6 +10,7 @@ import pytest from cecli.helpers.agents.service import ( + DEFAULT_SUMMARY_INTERRUPTED, AgentService, SubAgentInfo, SubAgentStatus, @@ -465,6 +466,95 @@ async def test_invoke_tui_notification(self, service, registry): assert call_args[2] == "reviewer" # name +def make_sub_coder(service, generate): + """A sub-agent coder sharing the parent's io, as IOProxy instances share their target.""" + sub_coder = MagicMock() + sub_coder.tui = None + sub_coder.uuid = "child-uuid" + sub_coder.root = "/proj" + sub_coder.io = service.coder.io + sub_coder.generate = AsyncMock(side_effect=generate) + return sub_coder + + +class TestSubAgentEvents: + """Sub-agent lifecycle reported through ``io.subagent_event``, and cancelling one.""" + + @pytest.fixture(autouse=True) + def patched_coders(self, service, registry): + # After ``service``, which empties the registry + with patch("cecli.coders.Coder") as MockCoder: + with patch("cecli.helpers.conversation.service.ConversationService"): + self.MockCoder = MockCoder + yield + + def events(self, service): + return [ + (call.args[0], call.args[1], call.kwargs) + for call in service.coder.io.subagent_event.call_args_list + ] + + async def test_invoke_reports_the_lifecycle(self, service): + async def finish(user_message, **kwargs): + info = service.sub_agents["child-uuid"] + info.status, info.summary = SubAgentStatus.FINISHED, "all good" + + self.MockCoder.create = AsyncMock(return_value=make_sub_coder(service, finish)) + assert await service.invoke("reviewer", "review this") == "all good" + + events = self.events(service) + assert [(uuid, event) for uuid, event, _ in events] == [ + ("child-uuid", "created"), + ("child-uuid", "message"), + ("child-uuid", "running"), + ("parent-uuid", "message"), + ("child-uuid", "idle"), + ] + created, delegation, _, result, idle = (kwargs for _, _, kwargs in events) + assert created["parent_uuid"] == "parent-uuid" + assert created["name"] == "reviewer" and created["root"] == "/proj" + assert delegation == {"sender_uuid": "parent-uuid", "text": "review this"} + assert result["sender_uuid"] == "child-uuid" and "all good" in result["text"] + assert idle == {"stop_reason": "end_turn", "error": None} + + def start_work(self, service, absorb): + """Invoke a sub-agent that works until cancelled; returns (invoking task, started).""" + started = asyncio.Event() + + async def work(user_message, **kwargs): + started.set() + try: + await asyncio.sleep(3600) + except asyncio.CancelledError: + if not absorb: + raise + # Streaming turns a cancellation into a ^C, and generate() returns + + self.MockCoder.create = AsyncMock(return_value=make_sub_coder(service, work)) + return asyncio.create_task(service.invoke("reviewer", "review this")), started + + @pytest.mark.parametrize("absorb", [False, True]) + async def test_cancelling_the_sub_agent_leaves_its_parent_running(self, service, absorb): + invoking, started = self.start_work(service, absorb) + await started.wait() + + assert service.cancel_sub_agent("child-uuid") is True + assert await invoking == DEFAULT_SUMMARY_INTERRUPTED + assert self.events(service)[-1][1:] == ("idle", {"stop_reason": "cancelled", "error": None}) + assert service.cancel_sub_agent("child-uuid") is False + + @pytest.mark.parametrize("absorb", [False, True]) + async def test_cancelling_the_parent_cancels_the_sub_agent_it_waits_for(self, service, absorb): + invoking, started = self.start_work(service, absorb) + await started.wait() + + invoking.cancel() + with pytest.raises(asyncio.CancelledError): + await invoking + await asyncio.gather(service.sub_agents["child-uuid"].generate_task, return_exceptions=True) + assert self.events(service)[-1][1:] == ("idle", {"stop_reason": "cancelled", "error": None}) + + # ================================================================== # # Public API: spawn # ================================================================== #