Structured SSE run events and API streaming

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #5292 feat(acp): aggregate ACP improvements — auth compat, protocol fixes, command ads, delegation, SSE events

  merged to main

  files: acp_adapter/events.py, acp_adapter/server.py, acp_adapter/session.py, cli.py, gateway/platforms/api_server.py, gateway/run.py, pyproject.toml, run_agent.py, tests/acp/test_events.py, tests/acp/test_mcp_e2e.py, tests/acp/test_server.py, tests/acp/test_session.py, tests/agent/test_subagent_progress.py, tests/gateway/test_run_progress_topics.py, tests/test_run_agent.py, tests/tools/test_delegate.py, tools/delegate_tool.py

  Show the diff

  • acp_adapter/events.py

    @@ -54,14 +54,18 @@ def make_tool_progress_cb(
     
         Signature expected by AIAgent::
     
    -        tool_progress_callback(name: str, preview: str, args: dict)
    +        tool_progress_callback(event_type: str, name: str, preview: str, args: dict, **kwargs)
     
    -    Emits ``ToolCallStart`` for each tool invocation and tracks IDs in a FIFO
    +    Emits ``ToolCallStart`` for ``tool.started`` events and tracks IDs in a FIFO
         queue per tool name so duplicate/parallel same-name calls still complete
    -    against the correct ACP tool call.
    +    against the correct ACP tool call.  Other event types (``tool.completed``,
    +    ``reasoning.available``) are silently ignored.
         """
     
    -    def _tool_progress(name: str, preview: str, args: Any = None) -> None:
    +    def _tool_progress(event_type: str, name: str = None, preview: str = None, args: Any = None, **kwargs) -> None:
    +        # Only emit ACP ToolCallStart for tool.started; ignore other event types
    +        if event_type != "tool.started":
    +            return
             if isinstance(args, str):
                 try:
                     args = json.loads(args)

  • acp_adapter/server.py

    @@ -12,7 +12,8 @@
     from acp.schema import (
         AgentCapabilities,
         AuthenticateResponse,
    -    AuthMethod,
    +    AvailableCommand,
    +    AvailableCommandsUpdate,
         ClientCapabilities,
         EmbeddedResourceContentBlock,
         ForkSessionResponse,
    @@ -37,9 +38,16 @@
         SessionListCapabilities,
         SessionInfo,
         TextContentBlock,
    +    UnstructuredCommandInput,
         Usage,
     )
     
    +# AuthMethodAgent was renamed from AuthMethod in agent-client-protocol 0.9.0
    +try:
    +    from acp.schema import AuthMethodAgent
    +except ImportError:
    +    from acp.schema import AuthMethod as AuthMethodAgent  # type: ignore[attr-defined]
    +
     from acp_adapter.auth import detect_provider, has_provider
     from acp_adapter.events import (
         make_message_cb,
    @@ -84,6 +92,48 @@ def _extract_text(
     class HermesACPAgent(acp.Agent):
         """ACP Agent implementation wrapping Hermes AIAgent."""
     
    +    _SLASH_COMMANDS = {
    +        "help": "Show available commands",
    +        "model": "Show or change current model",
    +        "tools": "List available tools",
    +        "context": "Show conversation context info",
    +        "reset": "Clear conversation history",
    +        "compact": "Compress conversation context",
    +        "version": "Show Hermes version",
    +    }
    +
    +    _ADVERTISED_COMMANDS = (
    +        {
    +            "name": "help",
    +            "description": "List available commands",
    +        },
    +        {
    +            "name": "model",
    +            "description": "Show current model and provider, or switch models",
    +            "input_hint": "model name to switch to",
    +        },
    +        {
    +            "name": "tools",
    +            "description": "List available tools with descriptions",
    +        },
    +        {
    +            "name": "context",
    +            "description": "Show conversation message counts by role",
    +        },
    +        {
    +            "name": "reset",
    +            "description": "Clear conversation history",
    +        },
    +        {
    +            "name": "compact",
    +            "description": "Compress conversation context",
    +        },
    +        {
    +            "name": "version",
    +            "description": "Show Hermes version",
    +        },
    +    )
    +
         def __init__(self, session_manager: SessionManager | None = None):
             super().__init__()
             self.session_manager = session_manager or SessionManager()
    @@ -177,7 +227,7 @@ async def initialize(
             auth_methods = None
             if provider:
                 auth_methods = [
    -                AuthMethod(
    +                AuthMethodAgent(
                         id=provider,
                         name=f"{provider} runtime credentials",
                         description=f"Authenticate Hermes using the currently configured {provider} runtime credentials.",
    @@ -219,6 +269,7 @@ async def new_session(
             state = self.session_manager.create_session(cwd=cwd)
             await self._register_session_mcp_servers(state, mcp_servers)
             logger.info("New session %s (cwd=%s)", state.session_id, cwd)
    +        self._schedule_available_commands_update(state.session_id)
             return NewSessionResponse(session_id=state.session_id)
     
         async def load_session(
    @@ -234,6 +285,7 @@ async def load_session(
                 return None
             await self._register_session_mcp_servers(state, mcp_servers)
             logger.info("Loaded session %s", session_id)
    +        self._schedule_available_commands_update(session_id)
             return LoadSessionResponse()
     
         async def resume_session(
    @@ -249,6 +301,7 @@ async def resume_session(
                 state = self.session_manager.create_session(cwd=cwd)
             await self._register_session_mcp_servers(state, mcp_servers)
             logger.info("Resumed session %s", state.session_id)
    +        self._schedule_available_commands_update(state.session_id)
             return ResumeSessionResponse()
     
         async def cancel(self, session_id: str, **kwargs: Any) -> None:
    @@ -274,6 +327,8 @@ async def fork_session(
             if state is not None:
                 await self._register_session_mcp_servers(state, mcp_servers)
             logger.info("Forked session %s -> %s", session_id, new_id)
    +        if new_id:
    +            self._schedule_available_commands_update(new_id)
             return ForkSessionResponse(session_id=new_id)
     
         async def list_sessions(
    @@ -411,15 +466,50 @@ def _run_agent() -> dict:
     
         # ---- Slash commands (headless) -------------------------------------------
     
    -    _SLASH_COMMANDS = {
    -        "help": "Show available commands",
    -        "model": "Show or change current model",
    -        "tools": "List available tools",
    -        "context": "Show conversation context info",
    -        "reset": "Clear conversation history",
    -        "compact": "Compress conversation context",
    -        "version": "Show Hermes version",
    -    }
    +    @classmethod
    +    def _available_commands(cls) -> list[AvailableCommand]:
    +        commands: list[AvailableCommand] = []
    +        for spec in cls._ADVERTISED_COMMANDS:
    +            input_hint = spec.get("input_hint")
    +            commands.append(
    +                AvailableCommand(
    +                    name=spec["name"],
    +                    description=spec["description"],
    +                    input=UnstructuredCommandInput(hint=input_hint)
    +                    if input_hint
    +                    else None,
    +                )
    +            )
    +        return commands
    +
    +    async def _send_available_commands_update(self, session_id: str) -> None:
    +        """Advertise supported slash commands to the connected ACP client."""
    +        if not self._conn:
    +            return
    +
    +        try:
    +            await self._conn.session_update(
    +                session_id=session_id,
    +                update=AvailableCommandsUpdate(
    +                    sessionUpdate="available_commands_update",
    +                    availableCommands=self._available_commands(),
    +                ),
    +            )
    +        except Exception:
    +            logger.warning(
    +                "Failed to advertise ACP slash commands for session %s",
    +                session_id,
    +                exc_info=True,
    +            )
    +
    +    def _schedule_available_commands_update(self, session_id: str) -> None:
    +        """Send the command advertisement after the session response is queued."""
    +        if not self._conn:
    +            return
    +        loop = asyncio.get_running_loop()
    +        loop.call_soon(
    +            asyncio.create_task, self._send_available_commands_update(session_id)
    +        )
     
         def _handle_slash_command(self, text: str, state: SessionState) -> str | None:
             """Dispatch a slash command and return the response text.
    @@ -539,11 +629,39 @@ def _cmd_compact(self, args: str, state: SessionState) -> str:
                 return "Nothing to compress — conversation is empty."
             try:
                 agent = state.agent
    -            if hasattr(agent, "compress_context"):
    -                agent.compress_context(state.history)
    -                self.session_manager.save_session(state.session_id)
    -                return f"Context compressed. Messages: {len(state.history)}"
    -            return "Context compression not available for this agent."
    +            if not getattr(agent, "compression_enabled", True):
    +                return "Context compression is disabled for this agent."
    +            if not hasattr(agent, "_compress_context"):
    +                return "Context compression not available for this agent."
    +
    +            from agent.model_metadata import estimate_messages_tokens_rough
    +
    +            original_count = len(state.history)
    +            approx_tokens = [redacted:secret-assignment]
    +            original_session_db = getattr(agent, "_session_db", None)
    +
    +            try:
    +                # ACP sessions must keep a stable session id, so avoid the
    +                # SQLite session-splitting side effect inside _compress_context.
    +                agent._session_db = None
    +                compressed, _ = agent._compress_context(
    +                    state.history,
    +                    getattr(agent, "_cached_system_prompt", "") or "",
    +                    approx_tokens=[redacted:secret-assignment]
    +                    task_id=state.session_id,
    +                )
    +            finally:
    +                agent._session_db = original_session_db
    +
    +            state.history = compressed
    +            self.session_manager.save_session(state.session_id)
    +
    +            new_count = len(state.history)
    +            new_tokens = [redacted:secret-assignment]
    +            return (
    +                f"Context compressed: {original_count} -> {new_count} messages\n"
    +                f"~{approx_tokens:,} -> ~{new_tokens:,} tokens"
    +            )
             except Exception as e:
                 return f"Compression failed: {e}"
     

  • acp_adapter/session.py

    @@ -13,6 +13,7 @@
     import copy
     import json
     import logging
    +import sys
     import uuid
     from dataclasses import dataclass, field
     from threading import Lock
    @@ -21,6 +22,17 @@
     logger = logging.getLogger(__name__)
     
     
    +def _acp_stderr_print(*args, **kwargs) -> None:
    +    """Best-effort human-readable output sink for ACP stdio sessions.
    +
    +    ACP reserves stdout for JSON-RPC frames, so any incidental CLI/status output
    +    from AIAgent must be redirected away from stdout. Route it to stderr instead.
    +    """
    +    kwargs = dict(kwargs)
    +    kwargs.setdefault("file", sys.stderr)
    +    print(*args, **kwargs)
    +
    +
     def _register_task_cwd(task_id: str, cwd: str) -> None:
         """Bind a task/session id to the editor's working directory for tools."""
         if not task_id:
    @@ -458,4 +470,8 @@ def _make_agent(
                 logger.debug("ACP session falling back to default provider resolution", exc_info=True)
     
             _register_task_cwd(session_id, cwd)
    -        return AIAgent(**kwargs)
    +        agent = AIAgent(**kwargs)
    +        # ACP stdio transport requires stdout to remain protocol-only JSON-RPC.
    +        # Route any incidental human-readable agent output to stderr instead.
    +        agent._print_fn = _acp_stderr_print
    +        return agent

  • cli.py

    @@ -5454,14 +5454,17 @@ def _on_tool_gen_start(self, tool_name: str) -> None:
         # Tool progress callback (audio cues for voice mode)
         # ====================================================================
     
    -    def _on_tool_progress(self, function_name: str, preview: str, function_args: dict):
    -        """Called when a tool starts executing.
    +    def _on_tool_progress(self, event_type: str, function_name: str = None, preview: str = None, function_args: dict = None, **kwargs):
    +        """Called on tool lifecycle events (tool.started, tool.completed, reasoning.available, etc.).
     
             Updates the TUI spinner widget so the user can see what the agent
             is doing during tool execution (fills the gap between thinking
             spinner and next response).  Also plays audio cue in voice mode.
             """
    -        if not function_name.startswith("_"):
    +        # Only act on tool.started; ignore tool.completed, reasoning.available, etc.
    +        if event_type != "tool.started":
    +            return
    +        if function_name and not function_name.startswith("_"):
                 from agent.display import get_tool_emoji
                 emoji = get_tool_emoji(function_name)
                 label = preview or function_name
    @@ -5474,7 +5477,7 @@ def _on_tool_progress(self, function_name: str, preview: str, function_args: dic
     
             if not self._voice_mode:
                 return
    -        if function_name.startswith("_"):
    +        if not function_name or function_name.startswith("_"):
                 return
             try:
                 from tools.voice_mode import play_beep

  • gateway/platforms/api_server.py

    @@ -7,6 +7,8 @@
     - GET  /v1/responses/{response_id} — Retrieve a stored response
     - DELETE /v1/responses/{response_id} — Delete a stored response
     - GET  /v1/models                  — lists hermes-agent as an available model
    +- POST /v1/runs                    — start a run, returns run_id immediately (202)
    +- GET  /v1/runs/{run_id}/events    — SSE stream of structured lifecycle events
     - GET  /health                     — health check
     
     Any OpenAI-compatible frontend (Open WebUI, LobeChat, LibreChat,
    @@ -300,6 +302,10 @@ def __init__(self, config: PlatformConfig):
             self._runner: Optional["web.AppRunner"] = None
             self._site: Optional["web.TCPSite"] = None
             self._response_store = ResponseStore()
    +        # Active run streams: run_id -> asyncio.Queue of SSE event dicts
    +        self._run_streams: Dict[str, "asyncio.Queue[Optional[Dict]]"] = {}
    +        # Creation timestamps for orphaned-run TTL sweep
    +        self._run_streams_created: Dict[str, float] = {}
             self._session_db: Optional[Any] = None  # Lazy-init SessionDB for session continuity
     
         @staticmethod
    @@ -1281,6 +1287,236 @@ def _run():
     
             return await loop.run_in_executor(None, _run)
     
    +    # ------------------------------------------------------------------
    +    # /v1/runs — structured event streaming
    +    # ------------------------------------------------------------------
    +
    +    _MAX_CONCURRENT_RUNS = 10  # Prevent unbounded resource allocation
    +    _RUN_STREAM_TTL = 300  # seconds before orphaned runs are swept
    +
    +    def _make_run_event_callback(self, run_id: str, loop: "asyncio.AbstractEventLoop"):
    +        """Return a tool_progress_callback that pushes structured events to the run's SSE queue."""
    +        def _push(event: Dict[str, Any]) -> None:
    +            q = self._run_streams.get(run_id)
    +            if q is None:
    +                return
    +            try:
    +                loop.call_soon_threadsafe(q.put_nowait, event)
    +            except Exception:
    +                pass
    +
    +        def _callback(event_type: str, tool_name: str = None, preview: str = None, args=None, **kwargs):
    +            ts = time.time()
    +            if event_type == "tool.started":
    +                _push({
    +                    "event": "tool.started",
    +                    "run_id": run_id,
    +                    "timestamp": ts,
    +                    "tool": tool_name,
    +                    "preview": preview,
    +                })
    +            elif event_type == "tool.completed":
    +                _push({
    +                    "event": "tool.completed",
    +                    "run_id": run_id,
    +                    "timestamp": ts,
    +                    "tool": tool_name,
    +                    "duration": round(kwargs.get("duration", 0), 3),
    +                    "error": kwargs.get("is_error", False),
    +                })
    +            elif event_type == "reasoning.available":
    +                _push({
    +                    "event": "reasoning.available",
    +                    "run_id": run_id,
    +                    "timestamp": ts,
    +                    "text": preview or "",
    +                })
    +            # _thinking and subagent_progress are intentionally not forwarded
    +
    +        return _callback
    +
    +    async def _handle_runs(self, request: "web.Request") -> "web.Response":
    +        """POST /v1/runs — start an agent run, return run_id immediately."""
    +        auth_err = self._check_auth(request)
    +        if auth_err:
    +            return auth_err
    +
    +        # Enforce concurrency limit
    +        if len(self._run_streams) >= self._MAX_CONCURRENT_RUNS:
    +            return web.json_response(
    +                _openai_error(f"Too many concurrent runs (max {self._MAX_CONCURRENT_RUNS})", code="rate_limit_exceeded"),
    +                status=429,
    +            )
    +
    +        try:
    +            body = await request.json()
    +        except Exception:
    +            return web.json_response(_openai_error("Invalid JSON"), status=400)
    +
    +        raw_input = body.get("input")
    +        if not raw_input:
    +            return web.json_response(_openai_error("Missing 'input' field"), status=400)
    +
    +        user_message = raw_input if isinstance(raw_input, str) else (raw_input[-1].get("content", "") if isinstance(raw_input, list) else "")
    +        if not user_message:
    +            return web.json_response(_openai_error("No user message found in input"), status=400)
    +
    +        run_id = f"run_{uuid.uuid4().hex}"
    +        loop = asyncio.get_running_loop()
    +        q: "asyncio.Queue[Optional[Dict]]" = asyncio.Queue()
    +        self._run_streams[run_id] = q
    +        self._run_streams_created[run_id] = time.time()
    +
    +        event_cb = self._make_run_event_callback(run_id, loop)
    +
    +        # Also wire stream_delta_callback so message.delta events flow through
    +        def _text_cb(delta: Optional[str]) -> None:
    +            if delta is None:
    +                return
    +            try:
    +                loop.call_soon_threadsafe(q.put_nowait, {
    +                    "event": "message.delta",
    +                    "run_id": run_id,
    +                    "timestamp": time.time(),
    +                    "delta": delta,
    +                })
    +            except Exception:
    +                pass
    +
    +        instructions = body.get("instructions")
    +        previous_response_id = body.get("previous_response_id")
    +        conversation_history: List[Dict[str, str]] = []
    +        if previous_response_id:
    +            stored = self._response_store.get(previous_response_id)
    +            if stored:
    +                conversation_history = list(stored.get("conversation_history", []))
    +                if instructions is None:
    +                    instructions = stored.get("instructions")
    +
    +        session_id = body.get("session_id") or run_id
    +        ephemeral_system_prompt = instructions
    +
    +        async def _run_and_close():
    +            try:
    +                agent = self._create_agent(
    +                    ephemeral_system_prompt=ephemeral_system_prompt,
    +                    session_id=session_id,
    +                    stream_delta_callback=_text_cb,
    +                    tool_progress_callback=event_cb,
    +                )
    +                def _run_sync():
    +                    r = agent.run_conversation(
    +                        user_message=user_message,
    +                        conversation_history=conversation_history,
    +                    )
    +                    u = {
    +                        "input_tokens": [redacted:secret-assignment] "session_prompt_tokens", 0) or 0,
    +                        "output_tokens": [redacted:secret-assignment] "session_completion_tokens", 0) or 0,
    +                        "total_tokens": [redacted:secret-assignment] "session_total_tokens", 0) or 0,
    +                    }
    +                    return r, u
    +
    +                result, usage = await asyncio.get_running_loop().run_in_executor(None, _run_sync)
    +                final_response = result.get("final_response", "") if isinstance(result, dict) else ""
    +                q.put_nowait({
    +                    "event": "run.completed",
    +                    "run_id": run_id,
    +                    "timestamp": time.time(),
    +                    "output": final_response,
    +                    "usage": usage,
    +                })
    +            except Exception as exc:
    +                logger.exception("[api_server] run %s failed", run_id)
    +                try:
    +                    q.put_nowait({
    +                        "event": "run.failed",
    +                        "run_id": run_id,
    +                        "timestamp": time.time(),
    +                        "error": str(exc),
    +                    })
    +                except Exception:
    +                    pass
    +            finally:
    +                # Sentinel: signal SSE stream to close
    +                try:
    +                    q.put_nowait(None)
    +                except Exception:
    +                    pass
    +
    +        task = asyncio.create_task(_run_and_close())
    +        try:
    +            self._background_tasks.add(task)
    +        except TypeError:
    +            pass
    +        if hasattr(task, "add_done_callback"):
    +            task.add_done_callback(self._background_tasks.discard)
    +
    +        return web.json_response({"run_id": run_id, "status": "started"}, status=202)
    +
    +    async def _handle_run_events(self, request: "web.Request") -> "web.StreamResponse":
    +        """GET /v1/runs/{run_id}/events — SSE stream of structured agent lifecycle events."""
    +        auth_err = self._check_auth(request)
    +        if auth_err:
    +            return auth_err
    +
    +        run_id = request.match_info["run_id"]
    +
    +        # Allow subscribing slightly before the run is registered (race condition window)
    +        for _ in range(20):
    +            if run_id in self._run_streams:
    +                break
    +            await asyncio.sleep(0.05)
    +        else:
    +            return web.json_response(_openai_error(f"Run not found: {run_id}", code="run_not_found"), status=404)
    +
    +        q = self._run_streams[run_id]
    +
    +        response = web.StreamResponse(
    +            status=200,
    +            headers={
    +                "Content-Type": "text/event-stream",
    +                "Cache-Control": "no-cache",
    +                "X-Accel-Buffering": "no",
    +            },
    +        )
    +        await response.prepare(request)
    +
    +        try:
    +            while True:
    +                try:
    +                    event = await asyncio.wait_for(q.get(), timeout=30.0)
    +                except asyncio.TimeoutError:
    +                    await response.write(b": keepalive\n\n")
    +                    continue
    +                if event is None:
    +                    # Run finished — send final SSE comment and close
    +                    await response.write(b": stream closed\n\n")
    +                    break
    +                payload = f"data: {json.dumps(event)}\n\n"
    +                await response.write(payload.encode())
    +        except Exception as exc:
    +            logger.debug("[api_server] SSE stream error for run %s: %s", run_id, exc)
    +        finally:
    +            self._run_streams.pop(run_id, None)
    +            self._run_streams_created.pop(run_id, None)
    +
    +        return response
    +
    +    async def _sweep_orphaned_runs(self) -> None:
    +        """Periodically clean up run streams that were never consumed."""
    +        while True:
    +            await asyncio.sleep(60)
    +            now = time.time()
    +            stale = [
    +                run_id
    +                for run_id, created_at in list(self._run_streams_created.items())
    +                if now - created_at > self._RUN_STREAM_TTL
    +            ]
    +            for run_id in stale:
    +                logger.debug("[api_server] sweeping orphaned run %s", run_id)
    +                self._run_streams.pop(run_id, None)
    +                self._run_streams_created.pop(run_id, None)
    +
         # ------------------------------------------------------------------
         # BasePlatformAdapter interface
         # ------------------------------------------------------------------
    @@ -1311,6 +1547,17 @@ async def connect(self) -> bool:
                 self._app.router.add_post("/api/jobs/{job_id}/pause", self._handle_pause_job)
                 self._app.router.add_post("/api/jobs/{job_id}/resume", self._handle_resume_job)
                 self._app.router.add_post("/api/jobs/{job_id}/run", self._handle_run_job)
    +            # Structured event streaming
    +            self._app.router.add_post("/v1/runs", self._handle_runs)
    +            self._app.router.add_get("/v1/runs/{run_id}/events", self._handle_run_events)
    +            # Start background sweep to clean up orphaned (unconsumed) run streams
    +            sweep_task = asyncio.create_task(self._sweep_orphaned_runs())
    +            try:
    +                self._background_tasks.add(sweep_task)
    +            except TypeError:
    +                pass
    +            if hasattr(sweep_task, "add_done_callback"):
    +                sweep_task.add_done_callback(self._background_tasks.discard)
     
                 # Port conflict detection — fail fast if port is already in use
                 import socket as _socket

  • gateway/run.py

    @@ -5950,11 +5950,15 @@ async def _run_agent(
             last_progress_msg = [None]  # Track last message for dedup
             repeat_count = [0]  # How many times the same message repeated
             
    -        def progress_callback(tool_name: str, preview: str = None, args: dict = None):
    -            """Callback invoked by agent when a tool is called."""
    +        def progress_callback(event_type: str, tool_name: str = None, preview: str = None, args: dict = None, **kwargs):
    +            """Callback invoked by agent on tool lifecycle events."""
                 if not progress_queue:
                     return
    -            
    +
    +            # Only act on tool.started events (ignore tool.completed, reasoning.available, etc.)
    +            if event_type not in ("tool.started",):
    +                return
    +
                 # "new" mode: only report when tool changes
                 if progress_mode == "new" and tool_name == last_tool[0]:
                     return

  • pyproject.toml

    @@ -61,7 +61,7 @@ honcho = ["honcho-ai>=2.0.1,<3"]
     mcp = ["mcp>=1.2.0,<2"]
     homeassistant = ["aiohttp>=3.9.0,<4"]
     sms = ["aiohttp>=3.9.0,<4"]
    -acp = ["agent-client-protocol>=0.8.1,<0.9"]
    +acp = ["agent-client-protocol>=0.9.0,<1.0"]
     dingtalk = ["dingtalk-stream>=0.1.0,<1"]
     feishu = ["lark-oapi>=1.5.3,<2"]
     rl = [

  • run_agent.py

    @@ -1491,6 +1491,25 @@ def _vprint(self, *args, force: bool = False, **kwargs):
                 return
             self._safe_print(*args, **kwargs)
     
    +    def _should_start_quiet_spinner(self) -> bool:
    +        """Return True when quiet-mode spinner output has a safe sink.
    +
    +        In headless/stdio-protocol environments, a raw spinner with no custom
    +        ``_print_fn`` falls back to ``sys.stdout`` and can corrupt protocol
    +        streams such as ACP JSON-RPC. Allow quiet spinners only when either:
    +        - output is explicitly rerouted via ``_print_fn``; or
    +        - stdout is a real TTY.
    +        """
    +        if self._print_fn is not None:
    +            return True
    +        stream = getattr(sys, "stdout", None)
    +        if stream is None:
    +            return False
    +        try:
    +            return bool(stream.isatty())
    +        except (AttributeError, ValueError, OSError):
    +            return False
    +
         def _emit_status(self, message: str) -> None:
             """Emit a lifecycle status message to both CLI and gateway channels.
     
    @@ -6037,7 +6056,7 @@ def _execute_tool_calls_concurrent(self, assistant_message, messages: list, effe
                 if self.tool_progress_callback:
                     try:
                         preview = _build_tool_preview(name, args)
    -                    self.tool_progress_callback(name, preview, args)
    +                    self.tool_progress_callback("tool.started", name, preview, args)
                     except Exception as cb_err:
                         logging.debug(f"Tool progress callback error: {cb_err}")
     
    @@ -6066,7 +6085,7 @@ def _run_tool(index, tool_call, function_name, function_args):
     
             # Start spinner for CLI mode (skip when TUI handles tool progress)
             spinner = None
    -        if self.quiet_mode and not self.tool_progress_callback:
    +        if self.quiet_mode and not self.tool_progress_callback and self._should_start_quiet_spinner():
                 face = random.choice(KawaiiSpinner.KAWAII_WAITING)
                 spinner = KawaiiSpinner(f"{face} ⚡ running {num_tools} tools concurrently", spinner_type='dots', print_fn=self._print_fn)
                 spinner.start()
    @@ -6102,6 +6121,15 @@ def _run_tool(index, tool_call, function_name, function_args):
                         result_preview = function_result[:200] if len(function_result) > 200 else function_result
                         logger.warning("Tool %s returned error (%.2fs): %s", function_name, tool_duration, result_preview)
     
    +                if self.tool_progress_callback:
    +                    try:
    +                        self.tool_progress_callback(
    +                            "tool.completed", function_name, None, None,
    +                            duration=tool_duration, is_error=is_error,
    +                        )
    +                    except Exception as cb_err:
    +                        logging.debug(f"Tool progress callback error: {cb_err}")
    +
                     if self.verbose_logging:
                         logging.debug(f"Tool {function_name} completed in {tool_duration:.2f}s")
                         logging.debug(f"Tool result ({len(function_result)} chars): {function_result}")
    @@ -6201,7 +6229,7 @@ def _execute_tool_calls_sequential(self, assistant_message, messages: list, effe
                 if self.tool_progress_callback:
                     try:
                         preview = _build_tool_preview(function_name, function_args)
    -                    self.tool_progress_callback(function_name, preview, function_args)
    +                    self.tool_progress_callback("tool.started", function_name, preview, function_args)
                     except Exception as cb_err:
                         logging.debug(f"Tool progress callback error: {cb_err}")
     
    @@ -6294,7 +6322,7 @@ def _execute_tool_calls_sequential(self, assistant_message, messages: list, effe
                         goal_preview = (function_args.get("goal") or "")[:30]
                         spinner_label = f"🔀 {goal_preview}" if goal_preview else "🔀 delegating"
                     spinner = None
    -                if self.quiet_mode and not self.tool_progress_callback:
    +                if self.quiet_mode and not self.tool_progress_callback and self._should_start_quiet_spinner():
                         face = random.choice(KawaiiSpinner.KAWAII_WAITING)
                         spinner = KawaiiSpinner(f"{face} {spinner_label}", spinner_type='dots', print_fn=self._print_fn)
                         spinner.start()
    @@ -6388,6 +6416,15 @@ def _execute_tool_calls_sequential(self, assistant_message, messages: list, effe
                 if _is_error_result:
                     logger.warning("Tool %s returned error (%.2fs): %s", function_name, tool_duration, result_preview)
     
    +            if self.tool_progress_callback:
    +                try:
    +                    self.tool_progress_callback(
    +                        "tool.completed", function_name, None, None,
    +                        duration=tool_duration, is_error=_is_error_result,
    +                    )
    +                except Exception as cb_err:
    +                    logging.debug(f"Tool progress callback error: {cb_err}")
    +
                 if self.verbose_logging:
                     logging.debug(f"Tool {function_name} completed in {tool_duration:.2f}s")
                     logging.debug(f"Tool result ({len(function_result)} chars): {function_result}")
    @@ -7120,9 +7157,9 @@ def run_conversation(
                         # CLI TUI mode: use prompt_toolkit widget instead of raw spinner
                         # (works in both streaming and non-streaming modes)
                         self.thinking_callback(f"{face} {verb}...")
    -                elif not self._has_stream_consumers():
    -                    # Raw KawaiiSpinner only when no streaming consumers
    -                    # (would conflict with streamed token output)
    +                elif not self._has_stream_consumers() and self._should_start_quiet_spinner():
    +                    # Raw KawaiiSpinner only when no streaming consumers and the
    +                    # spinner output has a safe sink.
                         spinner_type = random.choice(['brain', 'sparkle', 'pulse', 'moon', 'star'])
                         thinking_spinner = KawaiiSpinner(f"{face} {verb}...", spinner_type=spinner_type, print_fn=self._print_fn)
                         thinking_spinner.start()
    @@ -8244,21 +8281,25 @@ def _stop_spinner():
     
                     # Notify progress callback of model's thinking (used by subagent
                     # delegation to relay the child's reasoning to the parent display).
    -                # Guard: only fire for subagents (_delegate_depth >= 1) to avoid
    -                # spamming gateway platforms with the main agent's every thought.
    -                if (assistant_message.content and self.tool_progress_callback
    -                        and getattr(self, '_delegate_depth', 0) > 0):
    +                if (assistant_message.content and self.tool_progress_callback):
                         _think_text = assistant_message.content.strip()
                         # Strip reasoning XML tags that shouldn't leak to parent display
                         _think_text = re.sub(
                             r'</?(?:REASONING_SCRATCHPAD|think|reasoning)>', '', _think_text
                         ).strip()
    +                    # For subagents: relay first line to parent display (existing behaviour).
    +                    # For all agents with a structured callback: emit reasoning.available event.
                         first_line = _think_text.split('\n')[0][:80] if _think_text else ""
    -                    if first_line:
    +                    if first_line and getattr(self, '_delegate_depth', 0) > 0:
                             try:
                                 self.tool_progress_callback("_thinking", first_line)
                             except Exception:
                                 pass
    +                    elif _think_text:
    +                        try:
    +                            self.tool_progress_callback("reasoning.available", "_thinking", _think_text[:500], None)
    +                        except Exception:
    +                            pass
                     
                     # Check for incomplete <REASONING_SCRATCHPAD> (opened but never closed)
                     # This means the model ran out of output tokens mid-reasoning — retry up to 2 times

  • tests/acp/test_events.py

    @@ -52,7 +52,7 @@ def test_emits_tool_call_start(self, mock_conn, event_loop_fixture):
                 future.result.return_value = None
                 mock_rcts.return_value = future
     
    -            cb("terminal", "$ ls -la", {"command": "ls -la"})
    +            cb("tool.started", "terminal", "$ ls -la", {"command": "ls -la"})
     
             # Should have tracked the tool call ID
             assert "terminal" in tool_call_ids
    @@ -75,7 +75,7 @@ def test_handles_string_args(self, mock_conn, event_loop_fixture):
                 future.result.return_value = None
                 mock_rcts.return_value = future
     
    -            cb("read_file", "Reading /etc/hosts", '{"path": "/etc/hosts"}')
    +            cb("tool.started", "read_file", "Reading /etc/hosts", '{"path": "/etc/hosts"}')
     
             assert "read_file" in tool_call_ids
     
    @@ -91,7 +91,7 @@ def test_handles_non_dict_args(self, mock_conn, event_loop_fixture):
                 future.result.return_value = None
                 mock_rcts.return_value = future
     
    -            cb("terminal", "$ echo hi", None)
    +            cb("tool.started", "terminal", "$ echo hi", None)
     
             assert "terminal" in tool_call_ids
     
    @@ -108,8 +108,8 @@ def test_duplicate_same_name_tool_calls_use_fifo_ids(self, mock_conn, event_loop
                 future.result.return_value = None
                 mock_rcts.return_value = future
     
    -            progress_cb("terminal", "$ ls", {"command": "ls"})
    -            progress_cb("terminal", "$ pwd", {"command": "pwd"})
    +            progress_cb("tool.started", "terminal", "$ ls", {"command": "ls"})
    +            progress_cb("tool.started", "terminal", "$ pwd", {"command": "pwd"})
                 assert len(tool_call_ids["terminal"]) == 2
     
                 step_cb(1, [{"name": "terminal", "result": "ok-1"}])

  • tests/acp/test_mcp_e2e.py

    @@ -130,7 +130,7 @@ def mock_run_conversation(user_message, conversation_history=None, task_id=None)
                 # 1) Agent fires tool_progress_callback (ToolCallStart)
                 if agent.tool_progress_callback:
                     agent.tool_progress_callback(
    -                    "terminal", "$ echo hello", {"command": "echo hello"}
    +                    "tool.started", "terminal", "$ echo hello", {"command": "echo hello"}
                     )
     
                 # 2) Agent fires step_callback with tool results (ToolCallUpdate)
    @@ -197,8 +197,8 @@ def mock_run(user_message, conversation_history=None, task_id=None):
                 agent = state.agent
                 # Fire two tool calls
                 if agent.tool_progress_callback:
    -                agent.tool_progress_callback("read_file", "read: /etc/hosts", {"path": "/etc/hosts"})
    -                agent.tool_progress_callback("web_search", "web search: test", {"query": "test"})
    +                agent.tool_progress_callback("tool.started", "read_file", "read: /etc/hosts", {"path": "/etc/hosts"})
    +                agent.tool_progress_callback("tool.started", "web_search", "web search: test", {"query": "test"})
     
                 if agent.step_callback:
                     agent.step_callback(1, [

  • tests/acp/test_server.py

    @@ -12,6 +12,7 @@
     from acp.schema import (
         AgentCapabilities,
         AuthenticateResponse,
    +    AvailableCommandsUpdate,
         Implementation,
         InitializeResponse,
         ListSessionsResponse,
    @@ -113,6 +114,53 @@ async def test_new_session_creates_session(self, agent):
             assert state is not None
             assert state.cwd == "/home/user/project"
     
    +    @pytest.mark.asyncio
    +    async def test_available_commands_include_help(self, agent):
    +        help_cmd = next(
    +            (cmd for cmd in agent._available_commands() if cmd.name == "help"),
    +            None,
    +        )
    +
    +        assert help_cmd is not None
    +        assert help_cmd.description == "List available commands"
    +        assert help_cmd.input is None
    +
    +    @pytest.mark.asyncio
    +    async def test_send_available_commands_update(self, agent):
    +        mock_conn = MagicMock(spec=acp.Client)
    +        mock_conn.session_update = AsyncMock()
    +        agent._conn = mock_conn
    +
    +        await agent._send_available_commands_update("session-123")
    +
    +        mock_conn.session_update.assert_awaited_once()
    +        call = mock_conn.session_update.await_args
    +        assert call.kwargs["session_id"] == "session-123"
    +        update = call.kwargs["update"]
    +        assert isinstance(update, AvailableCommandsUpdate)
    +        assert update.session_update == "available_commands_update"
    +        assert [cmd.name for cmd in update.available_commands] == [
    +            "help",
    +            "model",
    +            "tools",
    +            "context",
    +            "reset",
    +            "compact",
    +            "version",
    +        ]
    +        model_cmd = next(
    +            cmd for cmd in update.available_commands if cmd.name == "model"
    +        )
    +        assert model_cmd.input is not None
    +        assert model_cmd.input.root.hint == "model name to switch to"
    +
    +    @pytest.mark.asyncio
    +    async def test_new_session_schedules_available_commands_update(self, agent):
    +        with patch.object(agent, "_schedule_available_commands_update") as mock_schedule:
    +            resp = await agent.new_session(cwd="/home/user/project")
    +
    +        mock_schedule.assert_called_once_with(resp.session_id)
    +
         @pytest.mark.asyncio
         async def test_cancel_sets_event(self, agent):
             resp = await agent.new_session(cwd=".")
    @@ -132,6 +180,15 @@ async def test_load_session_returns_response(self, agent):
             load_resp = await agent.load_session(cwd="/tmp", session_id=resp.session_id)
             assert isinstance(load_resp, LoadSessionResponse)
     
    +    @pytest.mark.asyncio
    +    async def test_load_session_schedules_available_commands_update(self, agent):
    +        resp = await agent.new_session(cwd="/tmp")
    +        with patch.object(agent, "_schedule_available_commands_update") as mock_schedule:
    +            load_resp = await agent.load_session(cwd="/tmp", session_id=resp.session_id)
    +
    +        assert isinstance(load_resp, LoadSessionResponse)
    +        mock_schedule.assert_called_once_with(resp.session_id)
    +
         @pytest.mark.asyncio
         async def test_load_session_not_found_returns_none(self, agent):
             resp = await agent.load_session(cwd="/tmp", session_id="bogus")
    @@ -143,6 +200,15 @@ async def test_resume_session_returns_response(self, agent):
             resume_resp = await agent.resume_session(cwd="/tmp", session_id=resp.session_id)
             assert isinstance(resume_resp, ResumeSessionResponse)
     
    +    @pytest.mark.asyncio
    +    async def test_resume_session_schedules_available_commands_update(self, agent):
    +        resp = await agent.new_session(cwd="/tmp")
    +        with patch.object(agent, "_schedule_available_commands_update") as mock_schedule:
    +            resume_resp = await agent.resume_session(cwd="/tmp", session_id=resp.session_id)
    +
    +        assert isinstance(resume_resp, ResumeSessionResponse)
    +        mock_schedule.assert_called_once_with(resp.session_id)
    +
         @pytest.mark.asyncio
         async def test_resume_session_creates_new_if_missing(self, agent):
             resume_resp = await agent.resume_session(cwd="/tmp", session_id="nonexistent")
    @@ -170,6 +236,15 @@ async def test_fork_session(self, agent):
             assert fork_resp.session_id
             assert fork_resp.session_id != new_resp.session_id
     
    +    @pytest.mark.asyncio
    +    async def test_fork_session_schedules_available_commands_update(self, agent):
    +        new_resp = await agent.new_session(cwd="/original")
    +        with patch.object(agent, "_schedule_available_commands_update") as mock_schedule:
    +            fork_resp = await agent.fork_session(cwd="/forked", session_id=new_resp.session_id)
    +
    +        assert fork_resp.session_id
    +        mock_schedule.assert_called_once_with(fork_resp.session_id)
    +
     
     # ---------------------------------------------------------------------------
     # session configuration / model routing
    @@ -427,6 +502,55 @@ def test_version(self, agent, mock_manager):
             result = agent._handle_slash_command("/version", state)
             assert HERMES_VERSION in result
     
    +    def test_compact_compresses_context(self, agent, mock_manager):
    +        state = self._make_state(mock_manager)
    +        state.history = [
    +            {"role": "user", "content": "one"},
    +            {"role": "assistant", "content": "two"},
    +            {"role": "user", "content": "three"},
    +            {"role": "assistant", "content": "four"},
    +        ]
    +        state.agent.compression_enabled = True
    +        state.agent._cached_system_prompt = "system"
    +        original_session_db = object()
    +        state.agent._session_db = original_session_db
    +
    +        def _compress_context(messages, system_prompt, *, approx_tokens, task_id):
    +            assert state.agent._session_db is None
    +            assert messages == state.history
    +            assert system_prompt == "system"
    +            assert approx_tokens == 40
    +            assert task_id == state.session_id
    +            return [{"role": "user", "content": "summary"}], "new-system"
    +
    +        state.agent._compress_context = MagicMock(side_effect=_compress_context)
    +
    +        with (
    +            patch.object(agent.session_manager, "save_session") as mock_save,
    +            patch(
    +                "agent.model_metadata.estimate_messages_tokens_rough",
    +                side_effect=[40, 12],
    +            ),
    +        ):
    +            result = agent._handle_slash_command("/compact", state)
    +
    +        assert "Context compressed: 4 -> 1 messages" in result
    +        assert "~40 -> ~12 tokens" in result
    +        assert state.history == [{"role": "user", "content": "summary"}]
    +        assert state.agent._session_db is original_session_db
    +        state.agent._compress_context.assert_called_once_with(
    +            [
    +                {"role": "user", "content": "one"},
    +                {"role": "assistant", "content": "two"},
    +                {"role": "user", "content": "three"},
    +                {"role": "assistant", "content": "four"},
    +            ],
    +            "system",
    +            approx_tokens=40,
    +            task_id=state.session_id,
    +        )
    +        mock_save.assert_called_once_with(state.session_id)
    +
         def test_unknown_command_returns_none(self, agent, mock_manager):
             state = self._make_state(mock_manager)
             result = agent._handle_slash_command("/nonexistent", state)
    @@ -436,7 +560,8 @@ def test_unknown_command_returns_none(self, agent, mock_manager):
         async def test_slash_command_intercepted_in_prompt(self, agent, mock_manager):
             """Slash commands should be handled without calling the LLM."""
             new_resp = await agent.new_session(cwd="/tmp")
    -        mock_conn = AsyncMock(spec=acp.Client)
    +        mock_conn = MagicMock(spec=acp.Client)
    +        mock_conn.session_update = AsyncMock()
             agent._conn = mock_conn
     
             prompt = [TextContentBlock(type="text", text="/help")]
    @@ -449,7 +574,9 @@ async def test_slash_command_intercepted_in_prompt(self, agent, mock_manager):
         async def test_unknown_slash_falls_through_to_llm(self, agent, mock_manager):
             """Unknown /commands should be sent to the LLM, not intercepted."""
             new_resp = await agent.new_session(cwd="/tmp")
    -        mock_conn = AsyncMock(spec=acp.Client)
    +        mock_conn = MagicMock(spec=acp.Client)
    +        mock_conn.session_update = AsyncMock()
    +        mock_conn.request_permission = AsyncMock(return_value=None)
             agent._conn = mock_conn
     
             # Mock run_in_executor to avoid actually running the agent

  • tests/acp/test_session.py

    @@ -1,5 +1,7 @@
     """Tests for acp_adapter.session — SessionManager and SessionState."""
     
    +import contextlib
    +import io
     import json
     from types import SimpleNamespace
     import pytest
    @@ -329,3 +331,40 @@ def fake_agent(**kwargs):
             assert restored is not None
             assert restored.agent.provider == "anthropic"
             assert restored.agent.base_url == "https://anthropic.example/v1"
    +
    +    def test_acp_agents_route_human_output_to_stderr(self, tmp_path, monkeypatch):
    +        """ACP agents must keep stdout clean for JSON-RPC stdio transport."""
    +
    +        def fake_resolve_runtime_provider(requested=None, **kwargs):
    +            return {
    +                "provider": "openrouter",
    +                "api_mode": "chat_completions",
    +                "base_url": "https://openrouter.example/v1",
    +                "api_key": "[redacted:secret-assignment]",
    +                "command": None,
    +                "args": [],
    +            }
    +
    +        def fake_agent(**kwargs):
    +            return SimpleNamespace(model=kwargs.get("model"), _print_fn=None)
    +
    +        monkeypatch.setattr("hermes_cli.config.load_config", lambda: {
    +            "model": {"provider": "openrouter", "default": "test-model"}
    +        })
    +        monkeypatch.setattr(
    +            "hermes_cli.runtime_provider.resolve_runtime_provider",
    +            fake_resolve_runtime_provider,
    +        )
    +        db = SessionDB(tmp_path / "state.db")
    +
    +        with patch("run_agent.AIAgent", side_effect=fake_agent):
    +            manager = SessionManager(db=db)
    +            state = manager.create_session(cwd="/work")
    +
    +        stdout_buf = io.StringIO()
    +        stderr_buf = io.StringIO()
    +        with contextlib.redirect_stdout(stdout_buf), contextlib.redirect_stderr(stderr_buf):
    +            state.agent._print_fn("ACP noise")
    +
    +        assert stdout_buf.getvalue() == ""
    +        assert stderr_buf.getvalue() == "ACP noise\n"

  • tests/agent/test_subagent_progress.py

    @@ -96,7 +96,7 @@ def test_cli_spinner_tool_event(self):
             cb = _build_child_progress_callback(0, parent)
             assert cb is not None
             
    -        cb("web_search", "quantum computing")
    +        cb("tool.started", "web_search", "quantum computing", {})
             output = buf.getvalue()
             assert "web_search" in output
             assert "quantum computing" in output
    @@ -131,11 +131,11 @@ def test_gateway_batched_progress(self):
             
             # Send 4 tool calls — shouldn't flush yet (BATCH_SIZE = 5)
             for i in range(4):
    -            cb(f"tool_{i}", f"arg_{i}")
    +            cb("tool.started", f"tool_{i}", f"arg_{i}", {})
             parent_cb.assert_not_called()
             
             # 5th call should trigger flush
    -        cb("tool_4", "arg_4")
    +        cb("tool.started", "tool_4", "arg_4", {})
             parent_cb.assert_called_once()
             call_args = parent_cb.call_args
             assert "tool_0" in call_args[0][1]
    @@ -207,7 +207,7 @@ def test_single_task_no_prefix(self):
             parent.tool_progress_callback = None
             
             cb = _build_child_progress_callback(0, parent, task_count=1)
    -        cb("web_search", "test")
    +        cb("tool.started", "web_search", "test", {})
             
             output = buf.getvalue()
             assert "[" not in output
    @@ -330,9 +330,9 @@ def test_flush_sends_remaining_batch(self):
             cb = _build_child_progress_callback(0, parent)
     
             # Send 3 tools (below batch size of 5)
    -        cb("web_search", "query1")
    -        cb("read_file", "file.txt")
    -        cb("write_file", "out.txt")
    +        cb("tool.started", "web_search", "query1", {})
    +        cb("tool.started", "read_file", "file.txt", {})
    +        cb("tool.started", "write_file", "out.txt", {})
             parent_cb.assert_not_called()
     
             # Flush should send the remaining 3
    @@ -365,7 +365,7 @@ def test_flush_noop_when_no_parent_callback(self):
             parent.tool_progress_callback = None
     
             cb = _build_child_progress_callback(0, parent)
    -        cb("web_search", "test")
    +        cb("tool.started", "web_search", "test", {})
             cb._flush()  # Should not crash
     
     

  • tests/gateway/test_run_progress_topics.py

    @@ -60,9 +60,9 @@ def __init__(self, **kwargs):
             self.tools = []
     
         def run_conversation(self, message, conversation_history=None, task_id=None):
    -        self.tool_progress_callback("terminal", "pwd")
    +        self.tool_progress_callback("tool.started", "terminal", "pwd", {})
             time.sleep(0.35)
    -        self.tool_progress_callback("browser_navigate", "https://example.com")
    +        self.tool_progress_callback("tool.started", "browser_navigate", "https://example.com", {})
             time.sleep(0.35)
             return {
                 "final_response": "done",

  • tests/test_run_agent.py

    @@ -2612,6 +2612,24 @@ def test_aiagent_uses_copilot_acp_client():
         assert mock_acp_client.call_args.kwargs["args"] == ["--acp", "--stdio"]
     
     
    +def test_quiet_spinner_allowed_with_explicit_print_fn(agent):
    +    agent._print_fn = lambda *_a, **_kw: None
    +    with patch.object(run_agent.sys.stdout, "isatty", return_value=False):
    +        assert agent._should_start_quiet_spinner() is True
    +
    +
    +def test_quiet_spinner_allowed_on_real_tty(agent):
    +    agent._print_fn = None
    +    with patch.object(run_agent.sys.stdout, "isatty", return_value=True):
    +        assert agent._should_start_quiet_spinner() is True
    +
    +
    +def test_quiet_spinner_suppressed_on_non_tty_without_print_fn(agent):
    +    agent._print_fn = None
    +    with patch.object(run_agent.sys.stdout, "isatty", return_value=False):
    +        assert agent._should_start_quiet_spinner() is False
    +
    +
     def test_is_openai_client_closed_honors_custom_client_flag():
         assert AIAgent._is_openai_client_closed(SimpleNamespace(is_closed=True)) is True
         assert AIAgent._is_openai_client_closed(SimpleNamespace(is_closed=False)) is False

  • tests/tools/test_delegate.py

    @@ -34,7 +34,7 @@ def _make_mock_parent(depth=0):
         """Create a mock parent agent with the fields delegate_task expects."""
         parent = MagicMock()
         parent.base_url = "https://openrouter.ai/api/v1"
    -    parent.api_key = "[redacted:secret-assignment]"
    +    parent.api_key="***"
         parent.provider = "openrouter"
         parent.api_mode = "chat_completions"
         parent.model = "anthropic/claude-sonnet-4"
    @@ -47,6 +47,9 @@ def _make_mock_parent(depth=0):
         parent._delegate_depth = depth
         parent._active_children = []
         parent._active_children_lock = threading.Lock()
    +    parent._print_fn = None
    +    parent.tool_progress_callback = None
    +    parent.thinking_callback = None
         return parent
     
     
    @@ -228,7 +231,7 @@ def test_active_children_tracking(self):
         def test_child_inherits_runtime_credentials(self):
             parent = _make_mock_parent(depth=0)
             parent.base_url = "https://chatgpt.com/backend-api/codex"
    -        parent.api_key = "[redacted:secret-assignment]"
    +        parent.api_key="***"
             parent.provider = "openai-codex"
             parent.api_mode = "codex_responses"
     
    @@ -249,6 +252,49 @@ def test_child_inherits_runtime_credentials(self):
                 self.assertEqual(kwargs["provider"], parent.provider)
                 self.assertEqual(kwargs["api_mode"], parent.api_mode)
     
    +    def test_child_inherits_parent_print_fn(self):
    +        parent = _make_mock_parent(depth=0)
    +        sink = MagicMock()
    +        parent._print_fn = sink
    +
    +        with patch("run_agent.AIAgent") as MockAgent:
    +            mock_child = MagicMock()
    +            MockAgent.return_value = mock_child
    +
    +            _build_child_agent(
    +                task_index=0,
    +                goal="Keep stdout clean",
    +                context=None,
    +                toolsets=None,
    +                model=None,
    +                max_iterations=10,
    +                parent_agent=parent,
    +            )
    +
    +        self.assertIs(mock_child._print_fn, sink)
    +
    +    def test_child_uses_thinking_callback_when_progress_callback_available(self):
    +        parent = _make_mock_parent(depth=0)
    +        parent.tool_progress_callback = MagicMock()
    +
    +        with patch("run_agent.AIAgent") as MockAgent:
    +            mock_child = MagicMock()
    +            MockAgent.return_value = mock_child
    +
    +            _build_child_agent(
    +                task_index=0,
    +                goal="Avoid raw child spinners",
    +                context=None,
    +                toolsets=None,
    +                model=None,
    +                max_iterations=10,
    +                parent_agent=parent,
    +            )
    +
    +        self.assertTrue(callable(mock_child.thinking_callback))
    +        mock_child.thinking_callback("deliberating...")
    +        parent.tool_progress_callback.assert_not_called()
    +
     
     class TestToolNamePreservation(unittest.TestCase):
         """Verify _last_resolved_tool_names is restored after subagent runs."""

  • tools/delegate_tool.py

    @@ -98,11 +98,15 @@ def _build_child_progress_callback(task_index: int, parent_agent, task_count: in
         _BATCH_SIZE = 5
         _batch: List[str] = []
     
    -    def _callback(tool_name: str, preview: str = None):
    -        # Special "_thinking" event: model produced text content (reasoning)
    -        if tool_name == "_thinking":
    +    def _callback(event_type: str, tool_name: str = None, preview: str = None, args=None, **kwargs):
    +        # event_type is one of: "tool.started", "tool.completed",
    +        # "reasoning.available", "_thinking", "subagent_progress"
    +
    +        # "_thinking" / reasoning events
    +        if event_type in ("_thinking", "reasoning.available"):
    +            text = preview or tool_name or ""
                 if spinner:
    -                short = (preview[:55] + "...") if preview and len(preview) > 55 else (preview or "")
    +                short = (text[:55] + "...") if len(text) > 55 else text
                     try:
                         spinner.print_above(f" {prefix}├─ 💭 \"{short}\"")
                     except Exception as e:
    @@ -110,11 +114,15 @@ def _callback(tool_name: str, preview: str = None):
                 # Don't relay thinking to gateway (too noisy for chat)
                 return
     
    -        # Regular tool call event
    +        # tool.completed — no display needed here (spinner shows on started)
    +        if event_type == "tool.completed":
    +            return
    +
    +        # tool.started — display and batch for parent relay
             if spinner:
                 short = (preview[:35] + "...") if preview and len(preview) > 35 else (preview or "")
                 from agent.display import get_tool_emoji
    -            emoji = get_tool_emoji(tool_name)
    +            emoji = get_tool_emoji(tool_name or "")
                 line = f" {prefix}├─ {emoji} {tool_name}"
                 if short:
                     line += f"  \"{short}\""
    @@ -124,7 +132,7 @@ def _callback(tool_name: str, preview: str = None):
                     logger.debug("Spinner print_above failed: %s", e)
     
             if parent_cb:
    -            _batch.append(tool_name)
    +            _batch.append(tool_name or "")
                 if len(_batch) >= _BATCH_SIZE:
                     summary = ", ".join(_batch)
                     try:
    @@ -160,6 +168,9 @@ def _build_child_agent(
         override_base_url: Optional[str] = None,
         override_api_key: [redacted:secret-assignment] = None,
         override_api_mode: Optional[str] = None,
    +    # ACP transport overrides — lets a non-ACP parent spawn ACP child agents
    +    override_acp_command: Optional[str] = None,
    +    override_acp_args: Optional[List[str]] = None,
     ):
         """
         Build a child AIAgent on the main thread (thread-safe construction).
    @@ -197,14 +208,26 @@ def _build_child_agent(
         # total iterations across parent + subagents can exceed the parent's
         # max_iterations.  The user controls the per-subagent cap in config.yaml.
     
    +    child_thinking_cb = None
    +    if child_progress_cb:
    +        def _child_thinking(text: str) -> None:
    +            if not text:
    +                return
    +            try:
    +                child_progress_cb("_thinking", text)
    +            except Exception as e:
    +                logger.debug("Child thinking callback relay failed: %s", e)
    +
    +        child_thinking_cb = _child_thinking
    +
         # Resolve effective credentials: config override > parent inherit
         effective_model = model or parent_agent.model
         effective_provider = override_provider or getattr(parent_agent, "provider", None)
         effective_base_url = override_base_url or parent_agent.base_url
         effective_api_key = [redacted:secret-assignment] or parent_api_key
         effective_api_mode = override_api_mode or getattr(parent_agent, "api_mode", None)
    -    effective_acp_command = getattr(parent_agent, "acp_command", None)
    -    effective_acp_args = list(getattr(parent_agent, "acp_args", []) or [])
    +    effective_acp_command = override_acp_command or getattr(parent_agent, "acp_command", None)
    +    effective_acp_args = list(override_acp_args if override_acp_args is not None else (getattr(parent_agent, "acp_args", []) or []))
     
         child = AIAgent(
             base_url=effective_base_url,
    @@ -226,6 +249,7 @@ def _build_child_agent(
             skip_context_files=True,
             skip_memory=True,
             clarify_callback=None,
    +        thinking_callback=child_thinking_cb,
             session_db=getattr(parent_agent, '_session_db', None),
             providers_allowed=parent_agent.providers_allowed,
             providers_ignored=parent_agent.providers_ignored,
    @@ -234,6 +258,7 @@ def _build_child_agent(
             tool_progress_callback=child_progress_cb,
             iteration_budget=None,  # fresh budget per subagent
         )
    +    child._print_fn = getattr(parent_agent, '_print_fn', None)
         # Set delegation depth so children can't spawn grandchildren
         child._delegate_depth = getattr(parent_agent, '_delegate_depth', 0) + 1
     
    @@ -406,6 +431,8 @@ def delegate_task(
         toolsets: Optional[List[str]] = None,
         tasks: Optional[List[Dict[str, Any]]] = None,
         max_iterations: Optional[int] = None,
    +    acp_command: Optional[str] = None,
    +    acp_args: Optional[List[str]] = None,
         parent_agent=None,
     ) -> str:
         """
    @@ -487,6 +514,8 @@ def delegate_task(
                     override_provider=creds["provider"], override_base_url=creds["base_url"],
                     override_api_key=[redacted:secret-assignment]"api_key"],
                     override_api_mode=creds["api_mode"],
    +                override_acp_command=t.get("acp_command") or acp_command,
    +                override_acp_args=t.get("acp_args") or acp_args,
                 )
                 # Override with correct parent tool names (before child construction mutated global)
                 child._delegate_saved_tool_names = _parent_tool_names
    @@ -763,7 +792,16 @@ def _load_config() -> dict:
                             "toolsets": {
                                 "type": "array",
                                 "items": {"type": "string"},
    -                            "description": "Toolsets for this specific task",
    +                            "description": "Toolsets for this specific task. Use 'web' for network access, 'terminal' for shell.",
    +                        },
    +                        "acp_command": {
    +                            "type": "string",
    +                            "description": "Per-task ACP command override (e.g. 'claude'). Overrides the top-level acp_command for this task only.",
    +                        },
    +                        "acp_args": {
    +                            "type": "array",
    +                            "items": {"type": "string"},
    +                            "description": "Per-task ACP args override.",
                             },
                         },
                         "required": ["goal"],
    @@ -782,6 +820,23 @@ def _load_config() -> dict:
                         "Only set lower for simple tasks."
                     ),
                 },
    +            "acp_command": {
    +                "type": "string",
    +                "description": (
    +                    "Override ACP command for child agents (e.g. 'claude', 'copilot'). "
    +                    "When set, children use ACP subprocess transport instead of inheriting "
    +                    "the parent's transport. Enables spawning Claude Code (claude --acp --stdio) "
    +                    "or other ACP-capable agents from any parent, including Discord/Telegram/CLI."
    +                ),
    +            },
    +            "acp_args": {
    +                "type": "array",
    +                "items": {"type": "string"},
    +                "description": (
    +                    "Arguments for the ACP command (default: ['--acp', '--stdio']). "
    +                    "Only used when acp_command is set. Example: ['--acp', '--stdio', '--model', 'claude-opus-4-6']"
    +                ),
    +            },
             },
             "required": [],
         },
    @@ -801,6 +856,8 @@ def _load_config() -> dict:
             toolsets=args.get("toolsets"),
             tasks=args.get("tasks"),
             max_iterations=args.get("max_iterations"),
    +        acp_command=args.get("acp_command"),
    +        acp_args=args.get("acp_args"),
             parent_agent=kw.get("parent_agent")),
         check_fn=check_delegate_requirements,
         emoji="🔀",