@@ -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")
 
@@ -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
 
@@ -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
 
@@ -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 [
 
@@ -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:
 
@@ -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))
 
 
@@ -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