# NousResearch/hermes-agent · ACP session load history replay fixes

## Open now

- Nothing open on GitHub.

Multiple fixes and follow-ups addressing a bug where conversation history was not being replayed when loading or resuming an ACP session. Work spans text history, reasoning/thoughts, and tool/context rendering during session load.

## TL;DR

- 7 member(s): 0 open issue(s), 1 closed issue(s), 0 open PR(s), 4 merged PR(s), 2 PR(s) closed without merging.
- A fix was merged to main in PR #17874. Verify it addresses your case.
- **Recommended:** read PR #17874 (and any competing PRs below) before filing. Confirm the merged change covers your symptom.

## Maintainer resolution verdict

- Internal-tracker verdict over this cluster's issue members: `resolved` · deciding PR(s): `pr-26943`. This is distinct from the fix status above. Weigh it yourself.

## How this cluster was grouped

- Link evidence (`STRUCTURAL_SPANNING`): every member is connected by structural evidence: shared changed files, or a closing/superseded reference.
- Pairwise link density `0.29`: the share of member pairs that carry a link, counting a structural link in full and a relevance-only link at 0.36. `1.00` means every pair is linked.

## Closed / duplicate issues

- #12285: ACP: session/load does not replay conversation history per spec

## Prior & competing PRs

- **#17874: fix(acp): replay persisted session history on load/resume (salvage of #17652)**. merged to main · fix tier `verified`
  - maintainer direction: category `merged`; implication `superseded`
  - files: `acp_adapter/server.py`, `tests/acp/test_server.py`
  - fix hunks:
  - `acp_adapter/server.py`

    ```diff
    @@ -13,6 +13,7 @@
     import acp
     from acp.schema import (
         AgentCapabilities,
    +    AgentMessageChunk,
         AuthenticateResponse,
         AvailableCommand,
         AvailableCommandsUpdate,
    ```

    ```diff
    @@ -45,6 +46,7 @@
         TextContentBlock,
         UnstructuredCommandInput,
         Usage,
    +    UserMessageChunk,
     )
     
     # AuthMethodAgent was renamed from AuthMethod in agent-client-protocol 0.9.0
    ```

    ```diff
    @@ -377,6 +379,78 @@ async def authenticate(self, method_id: str, **kwargs: Any) -> AuthenticateRespo
     
         # ---- Session management -------------------------------------------------
     
    +    @staticmethod
    +    def _history_message_text(message: dict[str, Any]) -> str:
    +        """Extract displayable text from a persisted OpenAI-style message."""
    +        content = message.get("content")
    +        if isinstance(content, str):
    +            return content.strip()
    +        if isinstance(content, list):
    +            parts: list[str] = []
    +            for item in content:
    +                if isinstance(item, dict):
    +                    text = item.get("text")
    +                    if isinstance(text, str):
    +                        parts.append(text)
    +                    elif item.get("type") == "text" and isinstance(item.get("content"), str):
    +                        parts.append(item["content"])
    +                elif isinstance(item, str):
    +                    parts.append(item)
    +            return "\n".join(part.strip() for part in parts if part and part.strip()).strip()
    +        return ""
    +
    +    @staticmethod
    +    def _history_message_update(
    +        *,
    +        role: str,
    +        text: str,
    +    ) -> UserMessageChunk | AgentMessageChunk | None:
    +        """Build an ACP history replay update for a user/assistant message."""
    +        block = TextContentBlock(type="text", text=text)
    +        if role == "user":
    +            return UserMessageChunk(
    +                session_update="user_message_chunk",
    +                content=block,
    +            )
    +        if role == "assistant":
    +            return AgentMessageChunk(
    +                session_update="agent_message_chunk",
    +                content=block,
    +            )
    +        return None
    +
    +    async def _replay_session_history(self, state: SessionState) -> None:
    +        """Send persisted user/assistant history to clients during session/load.
    +
    +        Zed's ACP history UI calls ``session/load`` after the user picks an item
    +        from the Agents sidebar. The agent must then replay the full conversation
    +        as ``user_message_chunk`` / ``agent_message_chunk`` notifications; merely
    +        restoring server-side state makes Hermes remember context, but leaves the
    +        editor looking like a clean thread.
    +        """
    +        if not self._conn or not state.history:
    +            return
    +
    +        for message in state.history:
    +            role = str(message.get("role") or "")
    +            if role not in {"user", "assistant"}:
    +                continue
    +            text = self._history_message_text(message)
    +            if not text:
    +                continue
    +            update = self._history_message_update(role=role, text=text)
    +            if update is None:
    +                continue
    +            try:
    +                await self._conn.session_update(session_id=state.session_id, update=update)
    +            except Exception:
    +                logger.warning(
    +                    "Failed to replay ACP history for session %s",
    +                    state.session_id,
    +                    exc_info=True,
    +                )
    +                return
    +
         async def new_session(
             self,
             cwd: str,
    ```

    ```diff
    @@ -405,6 +479,7 @@ async def load_session(
                 return None
             await self._register_session_mcp_servers(state, mcp_servers)
             logger.info("Loaded session %s", session_id)
    +        await self._replay_session_history(state)
             self._schedule_available_commands_update(session_id)
             return LoadSessionResponse(models=self._build_model_state(state))
     
    ```

    ```diff
    @@ -421,6 +496,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)
    +        await self._replay_session_history(state)
             self._schedule_available_commands_update(state.session_id)
             return ResumeSessionResponse(models=self._build_model_state(state))
     
    ```
  - `tests/acp/test_server.py`

    ```diff
    @@ -11,6 +11,7 @@
     from acp.agent.router import build_agent_router
     from acp.schema import (
         AgentCapabilities,
    +    AgentMessageChunk,
         AuthenticateResponse,
         AvailableCommandsUpdate,
         Implementation,
    ```

    ```diff
    @@ -27,6 +28,7 @@
         SessionInfo,
         TextContentBlock,
         Usage,
    +    UserMessageChunk,
     )
     from acp_adapter.server import HermesACPAgent, HERMES_VERSION
     from acp_adapter.session import SessionManager
    ```

    ```diff
    @@ -224,6 +226,58 @@ async def test_load_session_not_found_returns_none(self, agent):
             resp = await agent.load_session(cwd="/tmp", session_id="bogus")
             assert resp is None
     
    +    @pytest.mark.asyncio
    +    async def test_load_session_replays_persisted_history_to_client(self, agent):
    +        mock_conn = MagicMock(spec=acp.Client)
    +        mock_conn.session_update = AsyncMock()
    +        agent._conn = mock_conn
    +
    +        new_resp = await agent.new_session(cwd="/tmp")
    +        state = agent.session_manager.get_session(new_resp.session_id)
    +        state.history = [
    +            {"role": "system", "content": "hidden system"},
    +            {"role": "user", "content": "what controls the / slash commands?"},
    +            {"role": "assistant", "content": "HermesACPAgent._ADVERTISED_COMMANDS controls them."},
    +            {"role": "tool", "content": "tool output should not replay"},
    +        ]
    +
    +        mock_conn.session_update.reset_mock()
    +        resp = await agent.load_session(cwd="/tmp", session_id=new_resp.session_id)
    +
    +        assert isinstance(resp, LoadSessionResponse)
    +        calls = mock_conn.session_update.await_args_list
    +        replay_calls = [
    +            call for call in calls
    +            if getattr(call.kwargs.get("update"), "session_update", None)
    +            in {"user_message_chunk", "agent_message_chunk"}
    +        ]
    +        assert len(replay_calls) == 2
    +        assert isinstance(replay_calls[0].kwargs["update"], UserMessageChunk)
    +        assert replay_calls[0].kwargs["update"].content.text == "what controls the / slash commands?"
    +        assert isinstance(replay_calls[1].kwargs["update"], AgentMessageChunk)
    +        assert replay_calls[1].kwargs["update"].content.text.startswith("HermesACPAgent")
    +
    +    @pytest.mark.asyncio
    +    async def test_resume_session_replays_persisted_history_to_client(self, agent):
    +        mock_conn = MagicMock(spec=acp.Client)
    +        mock_conn.session_update = AsyncMock()
    +        agent._conn = mock_conn
    +
    +        new_resp = await agent.new_session(cwd="/tmp")
    +        state = agent.session_manager.get_session(new_resp.session_id)
    +        state.history = [{"role": "user", "content": "So tell me the current state"}]
    +
    +        mock_conn.session_update.reset_mock()
    +        resp = await agent.resume_session(cwd="/tmp", session_id=new_resp.session_id)
    +
    +        assert isinstance(resp, ResumeSessionResponse)
    +        updates = [call.kwargs["update"] for call in mock_conn.session_update.await_args_list]
    +        assert any(
    +            isinstance(update, UserMessageChunk)
    +            and update.content.text == "So tell me the current state"
    +            for update in updates
    +        )
    +
         @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")
    ```
- **#19139: fix(acp): route Zed thoughts to reasoning + polish tool/context rendering**. merged to main · fix tier `verified`
  - maintainer direction: category `merged`; implication `superseded`
  - files: `acp_adapter/server.py`, `acp_adapter/tools.py`, `tests/acp/test_mcp_e2e.py`, `tests/acp/test_server.py`, `tests/acp/test_tools.py`
  - fix hunks:
  - `acp_adapter/server.py`

    ```diff
    @@ -4,6 +4,7 @@
     
     import asyncio
     import contextvars
    +import json
     import logging
     import os
     from collections import defaultdict, deque
    ```

    ```diff
    @@ -47,6 +48,7 @@
         TextContentBlock,
         UnstructuredCommandInput,
         Usage,
    +    UsageUpdate,
         UserMessageChunk,
     )
     
    ```

    ```diff
    @@ -65,6 +67,7 @@
     )
     from acp_adapter.permissions import make_approval_callback
     from acp_adapter.session import SessionManager, SessionState, _expand_acp_enabled_toolsets
    +from acp_adapter.tools import build_tool_complete, build_tool_start
     
     logger = logging.getLogger(__name__)
     
    ```

    ```diff
    @@ -315,6 +318,66 @@ def _resolve_model_selection(raw_model: str, current_provider: str) -> tuple[str
     
             return target_provider, new_model
     
    +    @staticmethod
    +    def _build_usage_update(state: SessionState) -> UsageUpdate | None:
    +        """Build ACP native context-usage data for clients like Zed.
    +
    +        Zed's circular context indicator is driven by ACP ``usage_update``
    +        session updates: ``size`` is the model context window and ``used`` is
    +        the current request pressure.  Hermes estimates ``used`` from the same
    +        buckets it sends to providers: system prompt, conversation history, and
    +        tool schemas.
    +        """
    +        agent = state.agent
    +        compressor = getattr(agent, "context_compressor", None)
    +        size = int(getattr(compressor, "context_length", 0) or 0)
    +        if size <= 0:
    +            return None
    +
    +        try:
    +            from agent.model_metadata import estimate_request_tokens_rough
    +
    +            used = estimate_request_tokens_rough(
    +                state.history,
    +                system_prompt=getattr(agent, "_cached_system_prompt", "") or "",
    +                tools=getattr(agent, "tools", None) or None,
    +            )
    +        except Exception:
    +            logger.debug("Could not estimate ACP native context usage", exc_info=True)
    +            used = int(getattr(compressor, "last_prompt_tokens", 0) or 0)
    +
    +        return UsageUpdate(
    +            session_update="usage_update",
    +            size=max(size, 0),
    +            used=max(used, 0),
    +        )
    +
    +    async def _send_usage_update(self, state: SessionState) -> None:
    +        """Send ACP native context usage to the connected client."""
    +        if not self._conn:
    +            return
    +        update = self._build_usage_update(state)
    +        if update is None:
    +            return
    +        try:
    +            await self._conn.session_update(
    +                session_id=state.session_id,
    +                update=update,
    +            )
    +        except Exception:
    +            logger.warning(
    +                "Failed to send ACP usage update for session %s",
    +                state.session_id,
    +                exc_info=True,
    +            )
    +
    +    def _schedule_usage_update(self, state: SessionState) -> None:
    +        """Schedule native context indicator refresh after ACP responses."""
    +        if not self._conn:
    +            return
    +        loop = asyncio.get_running_loop()
    +        loop.call_soon(asyncio.create_task, self._send_usage_update(state))
    +
         async def _register_session_mcp_servers(
             self,
             state: SessionState,
    ```

    ```diff
    @@ -485,37 +548,99 @@ def _history_message_update(
                 )
             return None
     
    +    @staticmethod
    +    def _history_tool_call_name_args(tool_call: dict[str, Any]) -> tuple[str, dict[str, Any]]:
    +        """Extract function name/arguments from an OpenAI-style tool_call."""
    +        function = tool_call.get("function") if isinstance(tool_call.get("function"), dict) else {}
    +        name = str(function.get("name") or tool_call.get("name") or "unknown_tool")
    +        raw_args = function.get("arguments") or tool_call.get("arguments") or tool_call.get("args") or {}
    +        if isinstance(raw_args, str):
    +            try:
    +                parsed = json.loads(raw_args)
    +            except Exception:
    +                parsed = {"raw": raw_args}
    +            raw_args = parsed
    +        if not isinstance(raw_args, dict):
    +            raw_args = {}
    +        return name, raw_args
    +
    +    @staticmethod
    +    def _history_tool_call_id(tool_call: dict[str, Any]) -> str:
    +        """Return the stable provider tool call id for ACP history replay."""
    +        return str(
    +            tool_call.get("id")
    +            or tool_call.get("call_id")
    +            or tool_call.get("tool_call_id")
    +            or ""
    +        ).strip()
    +
         async def _replay_session_history(self, state: SessionState) -> None:
             """Send persisted user/assistant history to clients during session/load.
     
             Zed's ACP history UI calls ``session/load`` after the user picks an item
             from the Agents sidebar. The agent must then replay the full conversation
    -        as ``user_message_chunk`` / ``agent_message_chunk`` notifications; merely
    -        restoring server-side state makes Hermes remember context, but leaves the
    -        editor looking like a clean thread.
    +        as user/assistant chunks plus reconstructed tool-call start/completion
    +        notifications; merely restoring server-side state makes Hermes remember
    +        context, but leaves the editor looking like a clean thread.
             """
             if not self._conn or not state.history:
                 return
     
    -        for message in state.history:
    -            role = str(message.get("role") or "")
    -            if role not in {"user", "assistant"}:
    -                continue
    -            text = self._history_message_text(message)
    -            if not text:
    -                continue
    -            update = self._history_message_update(role=role, text=text)
    -            if update is None:
    -                continue
    +        active_tool_calls: dict[str, tuple[str, dict[str, Any]]] = {}
    +
    +        async def _send(update: Any) -> bool:
                 try:
                     await self._conn.session_update(session_id=state.session_id, update=update)
    +                return True
                 except Exception:
                     logger.warning(
                         "Failed to replay ACP history for session %s",
                         state.session_id,
                         exc_info=True,
                     )
    -                return
    +                return False
    +
    +        for message in state.history:
    +            role = str(message.get("role") or "")
    +
    +            if role in {"user", "assistant"}:
    +                text = self._history_message_text(message)
    +                if text:
    +                    update = self._history_message_update(role=role, text=text)
    +                    if update is not None and not await _send(update):
    +                        return
    +
    +            if role == "assistant" and isinstance(message.get("tool_calls"), list):
    +                for tool_call in message["tool_calls"]:
    +                    if not isinstance(tool_call, dict):
    +                        continue
    +                    tool_call_id = self._history_tool_call_id(tool_call)
    +                    if not tool_call_id:
    +                        continue
    +                    tool_name, args = self._history_tool_call_name_args(tool_call)
    +                    active_tool_calls[tool_call_id] = (tool_name, args)
    +                    if not await _send(build_tool_start(tool_call_id, tool_name, args)):
    +                        return
    +                continue
    +
    +            if role == "tool":
    +                tool_call_id = str(message.get("tool_call_id") or "").strip()
    +                tool_name = str(message.get("tool_name") or "").strip()
    +                function_args: dict[str, Any] | None = None
    +                if tool_call_id in active_tool_calls:
    +                    tool_name, function_args = active_tool_calls.pop(tool_call_id)
    +                if not tool_call_id or not tool_name:
    +                    continue
    +                result = message.get("content")
    +                if not await _send(
    +                    build_tool_complete(
    +                        tool_call_id,
    +                        tool_name,
    +                        result=result if isinstance(result, str) else None,
    +                        function_args=function_args,
    +                    )
    +                ):
    +                    return
     
         async def new_session(
             self,
    ```

    ```diff
    @@ -527,11 +652,24 @@ async def new_session(
             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)
    +        self._schedule_usage_update(state)
             return NewSessionResponse(
                 session_id=state.session_id,
                 models=self._build_model_state(state),
             )
     
    +    def _schedule_history_replay(self, state: SessionState) -> None:
    +        """Replay persisted history after session/load or session/resume returns.
    +
    +        Zed only attaches streamed transcript/tool updates once the load/resume
    +        response has completed. Sending replay notifications while the request is
    +        still in-flight can make the server look correct in logs while the editor
    +        drops or fails to attach the tool-call history.
    +        """
    +        loop = asyncio.get_running_loop()
    +        replay_coro = self._replay_session_history(state)
    +        loop.call_soon(asyncio.create_task, replay_coro)
    +
         async def load_session(
             self,
             cwd: str,
    ```

    ```diff
    @@ -545,8 +683,9 @@ async def load_session(
                 return None
             await self._register_session_mcp_servers(state, mcp_servers)
             logger.info("Loaded session %s", session_id)
    -        await self._replay_session_history(state)
    +        self._schedule_history_replay(state)
             self._schedule_available_commands_update(session_id)
    +        self._schedule_usage_update(state)
             return LoadSessionResponse(models=self._build_model_state(state))
     
         async def resume_session(
    ```

    ```diff
    @@ -562,8 +701,9 @@ 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)
    -        await self._replay_session_history(state)
    +        self._schedule_history_replay(state)
             self._schedule_available_commands_update(state.session_id)
    +        self._schedule_usage_update(state)
             return ResumeSessionResponse(models=self._build_model_state(state))
     
         async def cancel(self, session_id: str, **kwargs: Any) -> None:
    ```

    ```diff
    @@ -712,6 +852,7 @@ async def prompt(
                     if self._conn:
                         update = acp.update_agent_message_text(response_text)
                         await self._conn.session_update(session_id, update)
    +                    await self._send_usage_update(state)
                     return PromptResponse(stop_reason="end_turn")
     
             # If Zed sends another regular prompt while the same ACP session is
    ```

    ```diff
    @@ -744,24 +885,37 @@ async def prompt(
             tool_call_meta: dict[str, dict[str, Any]] = {}
             previous_approval_cb = None
     
    +        streamed_message = False
    +
             if conn:
                 tool_progress_cb = make_tool_progress_cb(conn, session_id, loop, tool_call_ids, tool_call_meta)
    -            thinking_cb = make_thinking_cb(conn, session_id, loop)
    +            reasoning_cb = make_thinking_cb(conn, session_id, loop)
                 step_cb = make_step_cb(conn, session_id, loop, tool_call_ids, tool_call_meta)
                 message_cb = make_message_cb(conn, session_id, loop)
    +
    +            def stream_delta_cb(text: str) -> None:
    +                nonlocal streamed_message
    +                if text:
    +                    streamed_message = True
    +                message_cb(text)
    +
                 approval_cb = make_approval_callback(conn.request_permission, loop, session_id)
             else:
                 tool_progress_cb = None
    -            thinking_cb = None
    +            reasoning_cb = None
                 step_cb = None
    -            message_cb = None
    +            stream_delta_cb = None
                 approval_cb = None
     
             agent = state.agent
             agent.tool_progress_callback = tool_progress_cb
    -        agent.thinking_callback = thinking_cb
    +        # ACP thought panes should not receive Hermes' local kawaii waiting/status
    +        # updates. Route provider/model reasoning deltas instead; if the provider
    +        # emits no reasoning, Zed should not get a fake "thinking" accordion.
    +        agent.thinking_callback = None
    +        agent.reasoning_callback = reasoning_cb
             agent.step_callback = step_cb
    -        agent.message_callback = message_cb
    +        agent.stream_delta_callback = stream_delta_cb
     
             # Approval callback is per-thread (thread-local, GHSA-qg5c-hvr5-hjgr).
             # Set it INSIDE _run_agent so the TLS write happens in the executor
    ```

    ```diff
    @@ -867,7 +1021,7 @@ def _run_agent() -> dict:
                     )
                 except Exception:
                     logger.debug("Failed to auto-title ACP session %s", session_id, exc_info=True)
    -        if final_response and conn:
    +        if final_response and conn and not streamed_message:
                 update = acp.update_agent_message_text(final_response)
                 await conn.session_update(session_id, update)
     
    ```

    ```diff
    @@ -903,6 +1057,8 @@ def _run_agent() -> dict:
                     cached_read_tokens=[redacted:secret-assignment]"cache_read_tokens"),
                 )
     
    +        await self._send_usage_update(state)
    +
             stop_reason = "cancelled" if state.cancel_event and state.cancel_event.is_set() else "end_turn"
             return PromptResponse(stop_reason=stop_reason, usage=usage)
     
    ```

    ```diff
    @@ -1035,22 +1191,84 @@ def _cmd_tools(self, args: str, state: SessionState) -> str:
                 return f"Could not list tools: {e}"
     
         def _cmd_context(self, args: str, state: SessionState) -> str:
    +        """Show ACP session context pressure and compression guidance."""
             n_messages = len(state.history)
    -        if n_messages == 0:
    -            return "Conversation is empty (no messages yet)."
    -        # Count by role
    +
    +        # Count by role.
             roles: dict[str, int] = {}
             for msg in state.history:
                 role = msg.get("role", "unknown")
                 roles[role] = roles.get(role, 0) + 1
    +
    +        agent = state.agent
    +        model = state.model or getattr(agent, "model", "")
    +        provider = getattr(agent, "provider", None) or "auto"
    +        compressor = getattr(agent, "context_compressor", None)
    +        context_length = int(getattr(compressor, "context_length", 0) or 0)
    +        threshold_tokens = [redacted:secret-assignment] "threshold_tokens", 0) or 0)
    +
    +        try:
    +            from agent.model_metadata import estimate_request_tokens_rough
    +
    +            system_prompt = getattr(agent, "_cached_system_prompt", "") or ""
    +            tools = getattr(agent, "tools", None) or None
    +            approx_tokens = [redacted:secret-assignment]
    +                state.history,
    +                system_prompt=system_prompt,
    +                tools=tools,
    +            )
    +        except Exception:
    +            logger.debug("Could not estimate ACP context usage", exc_info=True)
    +            approx_tokens = 0
    +
    +        if threshold_tokens <= 0 and context_length > 0:
    +            threshold_tokens = [redacted:secret-assignment] * 0.80)
    +
             lines = [
    -            f"Conversation: {n_messages} messages",
    +            f"Conversation: {n_messages} messages"
    +            if n_messages
    +            else "Conversation is empty (no messages yet).",
                 f"  user: {roles.get('user', 0)}, assistant: {roles.get('assistant', 0)}, "
                 f"tool: {roles.get('tool', 0)}, system: {roles.get('system', 0)}",
             ]
    -        model = state.model or getattr(state.agent, "model", "")
             if model:
                 lines.append(f"Model: {model}")
    +        lines.append(f"Provider: {provider}")
    +
    +        if approx_tokens > 0:
    +            if context_length > 0:
    +                usage_pct = (approx_tokens / context_length) * 100
    +                lines.append(
    +                    f"Context usage: ~{approx_tokens:,} / {context_length:,} tokens ({usage_pct:.1f}%)"
    +                )
    +            else:
    +                lines.append(f"Context usage: ~{approx_tokens:,} tokens")
    +
    +        if threshold_tokens > 0:
    +            if approx_tokens > 0:
    +                threshold_pct = (threshold_tokens / context_length) * 100 if context_length > 0 else 0
    +                remaining = max(threshold_tokens - approx_tokens, 0)
    +                if approx_tokens >= threshold_tokens:
    +                    lines.append(
    +                        f"Compression: due now (threshold ~{threshold_tokens:,}"
    +                        + (f", {threshold_pct:.0f}%" if threshold_pct else "")
    +                        + "). Run /compact."
    +                    )
    +                else:
    +                    lines.append(
    +                        f"Compression: ~{remaining:,} tokens until threshold "
    +                        f"(~{threshold_tokens:,}"
    +                        + (f", {threshold_pct:.0f}%" if threshold_pct else "")
    +                        + ")."
    +                    )
    +            else:
    +                lines.append(f"Compression threshold: ~{threshold_tokens:,} tokens")
    +
    +        if getattr(agent, "compression_enabled", True) is False:
    +            lines.append("Compression is disabled for this agent.")
    +        else:
    +            lines.append("Tip: run /compact to compress manually before the threshold.")
    +
             return "\n".join(lines)
     
         def _cmd_reset(self, args: str, state: SessionState) -> str:
    ```
  - `acp_adapter/tools.py`

    ```diff
    @@ -28,6 +28,11 @@
         "terminal": "execute",
         "process": "execute",
         "execute_code": "execute",
    +    # Session/meta tools
    +    "todo": "other",
    +    "skill_view": "read",
    +    "skills_list": "read",
    +    "skill_manage": "edit",
         # Web / fetch
         "web_search": "fetch",
         "web_extract": "fetch",
    ```

    ```diff
    @@ -51,6 +56,28 @@
     }
     
     
    +_POLISHED_TOOLS = {
    +    # Core operator loop
    +    "todo", "memory", "session_search", "delegate_task",
    +    # Files / execution
    +    "read_file", "write_file", "patch", "search_files", "terminal", "process", "execute_code",
    +    # Skills / web / browser / media
    +    "skill_view", "skills_list", "skill_manage", "web_search", "web_extract",
    +    "browser_navigate", "browser_click", "browser_type", "browser_press", "browser_scroll",
    +    "browser_back", "browser_snapshot", "browser_console", "browser_get_images", "browser_vision",
    +    "vision_analyze", "image_generate", "text_to_speech",
    +    # Schedulers / platform integrations
    +    "cronjob", "send_message", "clarify", "discord", "discord_admin",
    +    "ha_list_entities", "ha_get_state", "ha_list_services", "ha_call_service",
    +    "feishu_doc_read", "feishu_drive_list_comments", "feishu_drive_list_comment_replies",
    +    "feishu_drive_reply_comment", "feishu_drive_add_comment",
    +    "kanban_create", "kanban_show", "kanban_comment", "kanban_complete",
    +    "kanban_block", "kanban_link", "kanban_heartbeat",
    +    "yb_query_group_info", "yb_query_group_members", "yb_search_sticker",
    +    "yb_send_dm", "yb_send_sticker", "mixture_of_agents",
    +}
    +
    +
     def get_tool_kind(tool_name: str) -> ToolKind:
         """Return the ACP ToolKind for a hermes tool, defaulting to 'other'."""
         return TOOL_KIND_MAP.get(tool_name, "other")
    ```

    ```diff
    @@ -85,18 +112,645 @@ def build_tool_title(tool_name: str, args: Dict[str, Any]) -> str:
             if urls:
                 return f"extract: {urls[0]}" + (f" (+{len(urls)-1})" if len(urls) > 1 else "")
             return "web extract"
    +    if tool_name == "process":
    +        action = str(args.get("action") or "").strip() or "manage"
    +        sid = str(args.get("session_id") or "").strip()
    +        return f"process {action}: {sid}" if sid else f"process {action}"
         if tool_name == "delegate_task":
    +        tasks = args.get("tasks")
    +        if isinstance(tasks, list) and tasks:
    +            return f"delegate batch ({len(tasks)} tasks)"
             goal = args.get("goal", "")
             if goal and len(goal) > 60:
                 goal = goal[:57] + "..."
             return f"delegate: {goal}" if goal else "delegate task"
    +    if tool_name == "session_search":
    +        query = str(args.get("query") or "").strip()
    +        return f"session search: {query}" if query else "recent sessions"
    +    if tool_name == "memory":
    +        action = str(args.get("action") or "manage").strip() or "manage"
    +        target = str(args.get("target") or "memory").strip() or "memory"
    +        return f"memory {action}: {target}"
         if tool_name == "execute_code":
    -        return "execute code"
    +        code = str(args.get("code") or "").strip()
    +        first_line = next((line.strip() for line in code.splitlines() if line.strip()), "")
    +        if first_line:
    +            if len(first_line) > 70:
    +                first_line = first_line[:67] + "..."
    +            return f"python: {first_line}"
    +        return "python code"
    +    if tool_name == "todo":
    +        items = args.get("todos")
    +        if isinstance(items, list):
    +            return f"todo ({len(items)} item{'s' if len(items) != 1 else ''})"
    +        return "todo"
    +    if tool_name == "skill_view":
    +        name = str(args.get("name") or "?").strip() or "?"
    +        file_path = str(args.get("file_path") or "").strip()
    +        suffix = f"/{file_path}" if file_path else ""
    +        return f"skill view ({name}{suffix})"
    +    if tool_name == "skills_list":
    +        category = str(args.get("category") or "").strip()
    +        return f"skills list ({category})" if category else "skills list"
    +    if tool_name == "skill_manage":
    +        action = str(args.get("action") or "manage").strip() or "manage"
    +        name = str(args.get("name") or "?").strip() or "?"
    +        file_path = str(args.get("file_path") or "").strip()
    +        target = f"{name}/{file_path}" if file_path else name
    +        if len(target) > 64:
    +            target = target[:61] + "..."
    +        return f"skill {action}: {target}"
    +    if tool_name == "browser_navigate":
    +        return f"navigate: {args.get('url', '?')}"
    +    if tool_name == "browser_snapshot":
    +        return "browser snapshot"
    +    if tool_name == "browser_vision":
    +        return f"browser vision: {str(args.get('question', '?'))[:50]}"
    +    if tool_name == "browser_get_images":
    +        return "browser images"
         if tool_name == "vision_analyze":
    -        return f"analyze image: {args.get('question', '?')[:50]}"
    +        return f"analyze image: {str(args.get('question', '?'))[:50]}"
    +    if tool_name == "image_generate":
    +        prompt = str(args.get("prompt") or args.get("description") or "").strip()
    +        return f"generate image: {prompt[:50]}" if prompt else "generate image"
    +    if tool_name == "cronjob":
    +        action = str(args.get("action") or "manage").strip() or "manage"
    +        job_id = str(args.get("job_id") or args.get("id") or "").strip()
    +        return f"cron {action}: {job_id}" if job_id else f"cron {action}"
         return tool_name
     
     
    +def _text(content: str) -> Any:
    +    return acp.tool_content(acp.text_block(content))
    +
    +
    +def _json_loads_maybe(value: Optional[str]) -> Any:
    +    if not isinstance(value, str):
    +        return value
    +    try:
    +        return json.loads(value)
    +    except Exception:
    +        pass
    +
    +    # Some Hermes tools append a human hint after a JSON payload, e.g.
    +    # ``{...}\n\n[Hint: Results truncated...]``. Keep the structured rendering path
    +    # by decoding the first JSON value instead of falling back to raw text.
    +    try:
    +        decoded, _ = json.JSONDecoder().raw_decode(value.lstrip())
    +        return decoded
    +    except Exception:
    +        return None
    +
    +
    +def _truncate_text(text: str, limit: int = 5000) -> str:
    +    if len(text) <= limit:
    +        return text
    +    return text[: max(0, limit - 100)] + f"\n... ({len(text)} chars total, truncated)"
    +
    +
    +def _fenced_text(text: str, language: str = "") -> str:
    +    """Return a Markdown fence that cannot be broken by backticks in text."""
    +    longest = max((len(run) for run in text.split("`")[1::2]), default=0)
    +    fence = "`" * max(3, longest + 1)
    +    return f"{fence}{language}\n{text}\n{fence}"
    +
    +
    +def _format_todo_result(result: Optional[str]) -> Optional[str]:
    +    data = _json_loads_maybe(result)
    +    if not isinstance(data, dict) or not isinstance(data.get("todos"), list):
    +        return None
    +    summary = data.get("summary") if isinstance(data.get("summary"), dict) else {}
    +    icon = {
    +        "completed": "✅",
    +        "in_progress": "🔄",
    +        "pending": "⏳",
    +        "cancelled": "✗",
    +    }
    +    lines = ["**Todo list**", ""]
    +    for item in data["todos"]:
    +        if not isinstance(item, dict):
    +            continue
    +        status = str(item.get("status") or "pending")
    +        content = str(item.get("content") or item.get("id") or "").strip()
    +        if content:
    +            lines.append(f"- {icon.get(status, '•')} {content}")
    +    if summary:
    +        cancelled = summary.get("cancelled", 0)
    +        lines.extend([
    +            "",
    +            "**Progress:** "
    +            f"{summary.get('completed', 0)} completed, "
    +            f"{summary.get('in_progress', 0)} in progress, "
    +            f"{summary.get('pending', 0)} pending"
    +            + (f", {cancelled} cancelled" if cancelled else ""),
    +        ])
    +    return "\n".join(lines)
    +
    +
    +def _format_read_file_result(result: Optional[str], args: Optional[Dict[str, Any]]) -> Optional[str]:
    +    data = _json_loads_maybe(result)
    +    if not isinstance(data, dict):
    +        return None
    +    if data.get("error") and not data.get("content"):
    +        return f"Read failed: {data.get('error')}"
    +    content = data.get("content")
    +    if not isinstance(content, str):
    +        return None
    +    path = str((args or {}).get("path") or data.get("path") or "file").strip()
    +    offset = (args or {}).get("offset")
    +    limit = (args or {}).get("limit")
    +    range_bits = []
    +    if offset:
    +        range_bits.append(f"from line {offset}")
    +    if limit:
    +        range_bits.append(f"limit {limit}")
    +    suffix = f" ({', '.join(range_bits)})" if range_bits else ""
    +    header = f"Read {path}{suffix}"
    +    if data.get("total_lines") is not None:
    +        header += f" — {data.get('total_lines')} total lines"
    +    # Hermes read_file output is line-numbered with `|`. If we send it as raw
    +    # Markdown, Zed can interpret pipes as tables and collapse the layout.
    +    # Fence the payload so file lines stay readable and literal.
    +    return _truncate_text(f"{header}\n\n{_fenced_text(content)}")
    +
    +
    +def _format_search_files_result(result: Optional[str]) -> Optional[str]:
    +    data = _json_loads_maybe(result)
    +    if not isinstance(data, dict):
    +        return None
    +    matches = data.get("matches")
    +    if not isinstance(matches, list):
    +        return None
    +
    +    total = data.get("total_count", len(matches))
    +    shown = min(len(matches), 12)
    +    truncated = bool(data.get("truncated")) or len(matches) > shown
    +    lines = [
    +        "Search results",
    +        f"Found {total} match{'es' if total != 1 else ''}; showing {shown}.",
    +        "",
    +    ]
    +
    +    for match in matches[:shown]:
    +        if not isinstance(match, dict):
    +            lines.append(f"- {match}")
    +            continue
    +
    +        path = str(match.get("path") or match.get("file") or match.get("filename") or "?")
    +        line = match.get("line") or match.get("line_number")
    +        content = str(match.get("content") or match.get("text") or "").strip()
    +        loc = f"{path}:{line}" if line else path
    +        lines.append(f"- {loc}")
    +        if content:
    +            snippet = _truncate_text(" ".join(content.split()), 300)
    +            lines.append(f"  {snippet}")
    +
    +    if truncated:
    +        lines.extend([
    +            "",
    +            "Results truncated. Narrow the search, add file_glob, or use offset to page.",
    +        ])
    +    return _truncate_text("\n".join(lines), limit=7000)
    +
    +
    +def _format_execute_code_result(result: Optional[str]) -> Optional[str]:
    +    data = _json_loads_maybe(result)
    +    if not isinstance(data, dict):
    +        return result if isinstance(result, str) and result.strip() else None
    +    output = str(data.get("output") or "")
    +    error = str(data.get("error") or "")
    +    exit_code = data.get("exit_code")
    +    parts = [f"Exit code: {exit_code}" if exit_code is not None else "Execution complete"]
    +    if output:
    +        parts.extend(["", "Output:", output])
    +    if error:
    +        parts.extend(["", "Error:", error])
    +    return _truncate_text("\n".join(parts))
    +
    +
    +def _extract_markdown_headings(content: str, limit: int = 8) -> list[str]:
    +    headings: list[str] = []
    +    for line in content.splitlines():
    +        stripped = line.strip()
    +        if stripped.startswith("#"):
    +            heading = stripped.lstrip("#").strip()
    +            if heading:
    +                headings.append(heading)
    +        if len(headings) >= limit:
    +            break
    +    return headings
    +
    +
    +def _format_skill_view_result(result: Optional[str]) -> Optional[str]:
    +    data = _json_loads_maybe(result)
    +    if not isinstance(data, dict):
    +        return None
    +    if data.get("success") is False:
    +        return f"Skill view failed: {data.get('error', 'unknown error')}"
    +    name = str(data.get("name") or "skill")
    +    file_path = str(data.get("file") or data.get("path") or "SKILL.md")
    +    description = str(data.get("description") or "").strip()
    +    content = str(data.get("content") or "")
    +    linked = data.get("linked_files") if isinstance(data.get("linked_files"), dict) else None
    +
    +    lines = ["**Skill loaded**", "", f"- **Name:** `{name}`", f"- **File:** `{file_path}`"]
    +    if description:
    +        lines.append(f"- **Description:** {description}")
    +    if content:
    +        lines.append(f"- **Content:** {len(content):,} chars loaded into agent context")
    +    if linked:
    +        linked_count = sum(len(v) for v in linked.values() if isinstance(v, list))
    +        lines.append(f"- **Linked files:** {linked_count}")
    +
    +    headings = _extract_markdown_headings(content)
    +    if headings:
    +        lines.extend(["", "**Sections**"])
    +        lines.extend(f"- {heading}" for heading in headings)
    +
    +    lines.extend([
    +        "",
    +        "_Full skill content is available to the agent but hidden here to keep ACP readable._",
    +    ])
    +    return "\n".join(lines)
    +
    +
    +def _format_skill_manage_result(result: Optional[str], args: Optional[Dict[str, Any]]) -> Optional[str]:
    +    data = _json_loads_maybe(result)
    +    if not isinstance(data, dict):
    +        return None
    +
    +    action = str((args or {}).get("action") or "manage").strip() or "manage"
    +    name = str((args or {}).get("name") or data.get("name") or "skill").strip() or "skill"
    +    file_path = str((args or {}).get("file_path") or data.get("file_path") or "SKILL.md").strip() or "SKILL.md"
    +    success = data.get("success")
    +    status = "✅ Skill updated" if success is not False else "✗ Skill update failed"
    +
    +    lines = [f"**{status}**", "", f"- **Action:** `{action}`", f"- **Skill:** `{name}`"]
    +    if action not in {"delete"}:
    +        lines.append(f"- **File:** `{file_path}`")
    +
    +    message = str(data.get("message") or data.get("error") or "").strip()
    +    if message:
    +        lines.append(f"- **Result:** {message}")
    +
    +    replacements = data.get("replacements") or data.get("replacement_count")
    +    if replacements is not None:
    +        lines.append(f"- **Replacements:** {replacements}")
    +
    +    path = str(data.get("path") or "").strip()
    +    if path:
    +        lines.append(f"- **Path:** `{path}`")
    +
    +    return "\n".join(lines)
    +
    +
    +def _format_web_search_result(result: Optional[str]) -> Optional[str]:
    +    data = _json_loads_maybe(result)
    +    if not isinstance(data, dict):
    +        return None
    +    web = data.get("data", {}).get("web") if isinstance(data.get("data"), dict) else data.get("web")
    +    if not isinstance(web, list):
    +        return None
    +    lines = [f"Web results: {len(web)}"]
    +    for item in web[:10]:
    +        if not isinstance(item, dict):
    +            continue
    +        title = str(item.get("title") or item.get("url") or "result").strip()
    +        url = str(item.get("url") or "").strip()
    +        desc = str(item.get("description") or "").strip()
    +        lines.append(f"• {title}" + (f" — {url}" if url else ""))
    +        if desc:
    +            lines.append(f"  {desc}")
    +    return _truncate_text("\n".join(lines))
    +
    +
    +def _format_web_extract_result(result: Optional[str]) -> Optional[str]:
    +    """Return only web_extract errors for ACP; success stays compact via title."""
    +    data = _json_loads_maybe(result)
    +    if not isinstance(data, dict):
    +        return None
    +    if data.get("success") is False and data.get("error"):
    +        return f"Web extract failed: {data.get('error')}"
    +    results = data.get("results")
    +    if not isinstance(results, list):
    +        return None
    +
    +    failures: list[str] = []
    +    for item in results[:10]:
    +        if not isinstance(item, dict):
    +            continue
    +        error = str(item.get("error") or "").strip()
    +        if not error or error in {"None", "null"}:
    +            continue
    +        url = str(item.get("url") or "").strip()
    +        title = str(item.get("title") or url or "Untitled").strip()
    +        failures.append(
    +            f"- {title}" + (f" — {url}" if url and url != title else "") + f"\n  Error: {_truncate_text(error, limit=500)}"
    +        )
    +
    +    if not failures:
    +        return None
    +    lines = [f"Web extract failed for {len(failures)} URL{'s' if len(failures) != 1 else ''}"]
    +    lines.extend(failures)
    +    return "\n".join(lines)
    +
    +
    +def _format_process_result(result: Optional[str], args: Optional[Dict[str, Any]]) -> Optional[str]:
    +    data = _json_loads_maybe(result)
    +    if not isinstance(data, dict):
    +        return result if isinstance(result, str) and result.strip() else None
    +    if data.get("success") is False and data.get("error"):
    +        return f"Process error: {data.get('error')}"
    +    action = str((args or {}).get("action") or "process").strip() or "process"
    +    if isinstance(data.get("processes"), list):
    +        processes = data["processes"]
    +        lines = [f"Processes: {len(processes)}"]
    +        for proc in processes[:20]:
    +            if not isinstance(proc, dict):
    +                lines.append(f"- {proc}")
    +                continue
    +            sid = str(proc.get("session_id") or proc.get("id") or "?")
    +            status = str(proc.get("status") or ("exited" if proc.get("exited") else "running"))
    +            cmd = str(proc.get("command") or "").strip()
    +            pid = proc.get("pid")
    +            code = proc.get("exit_code")
    +            bits = [status]
    +            if pid is not None:
    +                bits.append(f"pid {pid}")
    +            if code is not None:
    +                bits.append(f"exit {code}")
    +            lines.append(f"- `{sid}` — {', '.join(bits)}" + (f" — {cmd[:120]}" if cmd else ""))
    +        if len(processes) > 20:
    +            lines.append(f"... {len(processes) - 20} more process(es)")
    +        return "\n".join(lines)
    +
    +    status = str(data.get("status") or data.get("state") or action).strip()
    +    sid = str(data.get("session_id") or (args or {}).get("session_id") or "").strip()
    +    lines = [f"Process {action}: {status}" + (f" (`{sid}`)" if sid else "")]
    +    for key, label in (("command", "Command"), ("pid", "PID"), ("exit_code", "Exit code"), ("returncode", "Exit code"), ("lines", "Lines")):
    +        if data.get(key) is not None:
    +            lines.append(f"- **{label}:** {data.get(key)}")
    +    output = data.get("output") or data.get("new_output") or data.get("log") or data.get("stdout")
    +    error = data.get("error") or data.get("stderr")
    +    if output:
    +        lines.extend(["", "Output:", _truncate_text(str(output), limit=5000)])
    +    if error:
    +        lines.extend(["", "Error:", _truncate_text(str(error), limit=2000)])
    +    msg = data.get("message")
    +    if msg and not output and not error:
    +        lines.append(str(msg))
    +    return _truncate_text("\n".join(lines), limit=7000)
    +
    +
    +def _format_delegate_result(result: Optional[str]) -> Optional[str]:
    +    data = _json_loads_maybe(result)
    +    if not isinstance(data, dict):
    +        return None
    +    if data.get("error") and not isinstance(data.get("results"), list):
    +        return f"Delegation failed: {data.get('error')}"
    +    results = data.get("results")
    +    if not isinstance(results, list):
    +        return None
    +    total = data.get("total_duration_seconds")
    +    lines = [f"Delegation results: {len(results)} task{'s' if len(results) != 1 else ''}" + (f" in {total}s" if total is not None else "")]
    +    icon = {"completed": "✅", "failed": "✗", "error": "✗", "timeout": "⏱", "interrupted": "⚠"}
    +    for item in results:
    +        if not isinstance(item, dict):
    +            lines.append(f"- {item}")
    +            continue
    +        idx = item.get("task_index")
    +        status = str(item.get("status") or "unknown")
    +        model = item.get("model")
    +        dur = item.get("duration_seconds")
    +        role = item.get("_child_role")
    +        header = f"{icon.get(status, '•')} Task {idx + 1 if isinstance(idx, int) else '?'}: {status}"
    +        bits = []
    +        if model:
    +            bits.append(str(model))
    +        if role:
    +            bits.append(f"role={role}")
    +        if dur is not None:
    +            bits.append(f"{dur}s")
    +        if bits:
    +            header += " (" + ", ".join(bits) + ")"
    +        lines.extend(["", header])
    +        summary = str(item.get("summary") or "").strip()
    +        error = str(item.get("error") or "").strip()
    +        if summary:
    +            lines.append(_truncate_text(summary, limit=1200))
    +        if error:
    +            lines.append("Error: " + _truncate_text(error, limit=800))
    +        trace = item.get("tool_trace")
    +        if isinstance(trace, list) and trace:
    +            names = [str(t.get("tool") or "?") for t in trace if isinstance(t, dict)]
    +            if names:
    +                lines.append("Tools: " + ", ".join(names[:12]) + (f" (+{len(names)-12})" if len(names) > 12 else ""))
    +    return _truncate_text("\n".join(lines), limit=8000)
    +
    +
    +def _format_session_search_result(result: Optional[str]) -> Optional[str]:
    +    data = _json_loads_maybe(result)
    +    if not isinstance(data, dict):
    +        return None
    +    if data.get("success") is False:
    +        return f"Session search failed: {data.get('error', 'unknown error')}"
    +    results = data.get("results")
    +    if not isinstance(results, list):
    +        return None
    +    mode = data.get("mode") or "search"
    +    query = data.get("query")
    +    lines = ["Recent sessions" if mode == "recent" else f"Session search results" + (f" for `{query}`" if query else "")]
    +    if not results:
    +        lines.append(str(data.get("message") or "No matching sessions found."))
    +        return "\n".join(lines)
    +    for item in results:
    +        if not isinstance(item, dict):
    +            continue
    +        sid = str(item.get("session_id") or "?")
    +        title = str(item.get("title") or item.get("when") or "Untitled session").strip()
    +        when = str(item.get("last_active") or item.get("started_at") or item.get("when") or "").strip()
    +        count = item.get("message_count")
    +        source = str(item.get("source") or "").strip()
    +        meta = ", ".join(str(x) for x in [when, source, f"{count} msgs" if count is not None else ""] if x)
    +        lines.append(f"- **{title}** (`{sid}`)" + (f" — {meta}" if meta else ""))
    +        summary = str(item.get("summary") or item.get("preview") or "").strip()
    +        if summary:
    +            lines.append("  " + _truncate_text(" ".join(summary.split()), limit=500))
    +    return _truncate_text("\n".join(lines), limit=7000)
    +
    +
    +def _format_memory_result(result: Optional[str], args: Optional[Dict[str, Any]]) -> Optional[str]:
    +    data = _json_loads_maybe(result)
    +    if not isinstance(data, dict):
    +        return None
    +    action = str((args or {}).get("action") or "memory").strip() or "memory"
    +    target = str(data.get("target") or (args or {}).get("target") or "memory")
    +    if data.get("success") is False:
    +        lines = [f"✗ Memory {action} failed ({target})", str(data.get("error") or "unknown error")]
    +        matches = data.get("matches")
    +        if isinstance(matches, list) and matches:
    +            lines.append("Matches:")
    +            lines.extend(f"- {_truncate_text(str(m), 160)}" for m in matches[:5])
    +        return "\n".join(lines)
    +    lines = [f"✅ Memory {action} saved ({target})"]
    +    if data.get("message"):
    +        lines.append(str(data.get("message")))
    +    if data.get("entry_count") is not None:
    +        lines.append(f"Entries: {data.get('entry_count')}")
    +    if data.get("usage"):
    +        lines.append(f"Usage: {data.get('usage')}")
    +    # Avoid dumping all memory entries into ACP UI; show only the explicit new value preview.
    +    preview = str((args or {}).get("content") or (args or {}).get("old_text") or "").strip()
    +    if preview:
    +        lines.append("Preview: " + _truncate_text(preview, limit=300))
    +    return "\n".join(lines)
    +
    +
    +def _format_edit_result(tool_name: str, result: Optional[str], args: Optional[Dict[str, Any]]) -> Optional[str]:
    +    data = _json_loads_maybe(result)
    +    path = str((args or {}).get("path") or "file").strip()
    +    if isinstance(data, dict):
    +        if data.get("success") is False or data.get("error"):
    +            return f"{tool_name} failed for {path}: {data.get('error', 'unknown error')}"
    +        message = str(data.get("message") or "").strip()
    +        replacements = data.get("replacements") or data.get("replacement_count")
    +        lines = [f"✅ {tool_name} completed" + (f" for `{path}`" if path else "")]
    +        if message:
    +            lines.append(message)
    +        if replacements is not None:
    +            lines.append(f"Replacements: {replacements}")
    +        if data.get("files_modified"):
    +            files = data.get("files_modified")
    +            if isinstance(files, list):
    +                lines.append("Files: " + ", ".join(f"`{f}`" for f in files[:8]))
    +        return "\n".join(lines)
    +    if isinstance(result, str) and result.strip():
    +        return _truncate_text(result, limit=3000)
    +    return f"✅ {tool_name} completed" + (f" for `{path}`" if path else "")
    +
    +
    +def _format_browser_result(tool_name: str, result: Optional[str], args: Optional[Dict[str, Any]]) -> Optional[str]:
    +    data = _json_loads_maybe(result)
    +    if not isinstance(data, dict):
    +        return result if isinstance(result, str) and result.strip() else None
    +    if data.get("success") is False or data.get("error"):
    +        return f"{tool_name} failed: {data.get('error', 'unknown error')}"
    +    if tool_name == "browser_get_images":
    +        images = data.get("images") or data.get("data")
    +        if isinstance(images, list):
    +            lines = [f"Images found: {len(images)}"]
    +            for img in images[:12]:
    +                if isinstance(img, dict):
    +                    alt = str(img.get("alt") or "").strip()
    +                    url = str(img.get("url") or img.get("src") or "").strip()
    +                    lines.append(f"- {alt or 'image'}" + (f" — {url}" if url else ""))
    +            return _truncate_text("\n".join(lines), limit=5000)
    +    title = str(data.get("title") or data.get("url") or data.get("status") or tool_name)
    +    text = str(data.get("text") or data.get("content") or data.get("snapshot") or data.get("analysis") or data.get("message") or "").strip()
    +    lines = [title]
    +    if data.get("url") and data.get("url") != title:
    +        lines.append(str(data.get("url")))
    +    if text:
    +        lines.extend(["", _truncate_text(text, limit=5000)])
    +    return _truncate_text("\n".join(lines), limit=7000)
    +
    +
    +def _format_media_or_cron_result(tool_name: str, result: Optional[str]) -> Optional[str]:
    +    data = _json_loads_maybe(result)
    +    if not isinstance(data, dict):
    +        return result if isinstance(result, str) and result.strip() else None
    +    if data.get("success") is False or data.get("error"):
    +        return f"{tool_name} failed: {data.get('error', 'unknown error')}"
    +    lines = [f"✅ {tool_name} completed"]
    +    for key in ("file_path", "path", "url", "image_url", "job_id", "id", "status", "message", "next_run"):
    +        if data.get(key):
    +            lines.append(f"- **{key}:** {data.get(key)}")
    +    return "\n".join(lines)
    +
    +
    +def _format_generic_structured_result(tool_name: str, result: Optional[str]) -> Optional[str]:
    +    data = _json_loads_maybe(result)
    +    if not isinstance(data, (dict, list)):
    +        return result if isinstance(result, str) and result.strip() else None
    +    if isinstance(data, list):
    +        lines = [f"{tool_name}: {len(data)} item{'s' if len(data) != 1 else ''}"]
    +        for item in data[:12]:
    +            lines.append(f"- {_truncate_text(str(item), limit=240)}")
    +        return _truncate_text("\n".join(lines), limit=5000)
    +
    +    if data.get("success") is False or data.get("error"):
    +        return f"{tool_name} failed: {data.get('error', 'unknown error')}"
    +
    +    lines = [f"✅ {tool_name} completed" if data.get("success") is True else f"{tool_name} result"]
    +    priority_keys = (
    +        "message", "status", "id", "task_id", "issue_id", "title", "name", "entity_id",
    +        "state", "service", "url", "path", "file_path", "count", "total", "next_run",
    +    )
    +    seen = set()
    +    for key in priority_keys:
    +        value = data.get(key)
    +        if value in (None, "", [], {}):
    +            continue
    +        seen.add(key)
    +        lines.append(f"- **{key}:** {_truncate_text(str(value), limit=500)}")
    +
    +    for key, value in data.items():
    +        if key in seen or key in {"success", "raw", "content", "entries"}:
    +            continue
    +        if value in (None, "", [], {}):
    +            continue
    +        if isinstance(value, (dict, list)):
    +            preview = json.dumps(value, ensure_ascii=False, default=str)
    +        else:
    +            preview = str(value)
    +        lines.append(f"- **{key}:** {_truncate_text(preview, limit=500)}")
    +        if len(lines) >= 14:
    +            break
    +
    +    content = data.get("content")
    +    if isinstance(content, str) and content.strip():
    +        lines.extend(["", _truncate_text(content.strip(), limit=1500)])
    +    return _truncate_text("\n".join(lines), limit=7000)
    +
    +
    +def _build_polished_completion_content(
    +    tool_name: str,
    +    result: Optional[str],
    +    function_args: Optional[Dict[str, Any]],
    +) -> Optional[List[Any]]:
    +    formatter = {
    +        "todo": lambda: _format_todo_result(result),
    +        "read_file": lambda: _format_read_file_result(result, function_args),
    +        "write_file": lambda: _format_edit_result(tool_name, result, function_args),
    +        "patch": lambda: _format_edit_result(tool_name, result, function_args),
    +        "search_files": lambda: _format_search_files_result(result),
    +        "execute_code": lambda: _format_execute_code_result(result),
    +        "process": lambda: _format_process_result(result, function_args),
    +        "delegate_task": lambda: _format_delegate_result(result),
    +        "session_search": lambda: _format_session_search_result(result),
    +        "memory": lambda: _format_memory_result(result, function_args),
    +        "skill_view": lambda: _format_skill_view_result(result),
    +        "skill_manage": lambda: _format_skill_manage_result(result, function_args),
    +        "web_search": lambda: _format_web_search_result(result),
    +        "web_extract": lambda: _format_web_extract_result(result),
    +        "browser_navigate": lambda: _format_browser_result(tool_name, result, function_args),
    +        "browser_snapshot": lambda: _format_browser_result(tool_name, result, function_args),
    +        "browser_vision": lambda: _format_browser_result(tool_name, result, function_args),
    +        "browser_get_images": lambda: _format_browser_result(tool_name, result, function_args),
    +        "vision_analyze": lambda: _format_media_or_cron_result(tool_name, result),
    +        "image_generate": lambda: _format_media_or_cron_result(tool_name, result),
    +        "cronjob": lambda: _format_media_or_cron_result(tool_name, result),
    +    }.get(tool_name)
    +    if formatter is None and tool_name in _POLISHED_TOOLS:
    +        formatter = lambda: _format_generic_structured_result(tool_name, result)
    +    if formatter is None:
    +        return None
    +    text = formatter()
    +    if not text:
    +        return None
    +    return [_text(text)]
    +
    +
     def _build_patch_mode_content(patch_text: str) -> List[Any]:
         """Parse V4A patch mode input into ACP diff blocks when possible."""
         if not patch_text:
    ```

    ```diff
    @@ -258,7 +912,11 @@ def _build_tool_complete_content(
             except Exception:
                 pass
     
    -    return [acp.tool_content(acp.text_block(display_result))]
    +    polished_content = _build_polished_completion_content(tool_name, result, function_args)
    +    if polished_content:
    +        return polished_content
    +
    +    return [_text(display_result)]
     
     
     # ---------------------------------------------------------------------------
    ```

    ```diff
    @@ -288,7 +946,6 @@ def build_tool_start(
                 content = _build_patch_mode_content(patch_text)
             return acp.start_tool_call(
                 tool_call_id, title, kind=kind, content=content, locations=locations,
    -            raw_input=arguments,
             )
     
         if tool_name == "write_file":
    ```

    ```diff
    @@ -297,32 +954,172 @@ def build_tool_start(
             content = [acp.tool_diff_content(path=path, new_text=file_content)]
             return acp.start_tool_call(
                 tool_call_id, title, kind=kind, content=content, locations=locations,
    -            raw_input=arguments,
             )
     
         if tool_name == "terminal":
             command = arguments.get("command", "")
    -        content = [acp.tool_content(acp.text_block(f"$ {command}"))]
    +        content = [_text(f"$ {command}")]
             return acp.start_tool_call(
                 tool_call_id, title, kind=kind, content=content, locations=locations,
    -            raw_input=arguments,
             )
     
         if tool_name == "read_file":
    -        path = arguments.get("path", "")
    -        content = [acp.tool_content(acp.text_block(f"Reading {path}"))]
    +        # The title and location already identify the file. Sending a synthetic
    +        # "Reading ..." content block makes Zed render an unhelpful Output
    +        # section before the real file contents arrive on completion.
             return acp.start_tool_call(
    -            tool_call_id, title, kind=kind, content=content, locations=locations,
    -            raw_input=arguments,
    +            tool_call_id, title, kind=kind, content=None, locations=locations,
             )
     
         if tool_name == "search_files":
             pattern = arguments.get("pattern", "")
             target = arguments.get("target", "content")
    -        content = [acp.tool_content(acp.text_block(f"Searching for '{pattern}' ({target})"))]
    +        search_path = arguments.get("path")
    +        where = f" in {search_path}" if search_path else ""
    +        content = [_text(f"Searching for '{pattern}' ({target}){where}")]
    +        return acp.start_tool_call(
    +            tool_call_id, title, kind=kind, content=content, locations=locations,
    +        )
    +
    +    if tool_name == "todo":
    +        items = arguments.get("todos")
    +        if isinstance(items, list):
    +            preview_lines = ["Updating todo list", ""]
    +            for item in items[:8]:
    +                if isinstance(item, dict):
    +                    preview_lines.append(f"- {item.get('status', 'pending')}: {item.get('content', item.get('id', ''))}")
    +            if len(items) > 8:
    +                preview_lines.append(f"... {len(items) - 8} more")
    +            content = [_text("\n".join(preview_lines))]
    +        else:
    +            content = [_text("Reading todo list")]
    +        return acp.start_tool_call(
    +            tool_call_id, title, kind=kind, content=content, locations=locations,
    +        )
    +
    +    if tool_name == "skill_view":
    +        name = str(arguments.get("name") or "?").strip() or "?"
    +        file_path = str(arguments.get("file_path") or "SKILL.md").strip() or "SKILL.md"
    +        content = [_text(f"Loading skill '{name}' ({file_path})")]
    +        return acp.start_tool_call(
    +            tool_call_id, title, kind=kind, content=content, locations=locations,
    +        )
    +
    +    if tool_name == "skill_manage":
    +        action = str(arguments.get("action") or "manage").strip() or "manage"
    +        name = str(arguments.get("name") or "?").strip() or "?"
    +        file_path = str(arguments.get("file_path") or "SKILL.md").strip() or "SKILL.md"
    +        path = f"skills/{name}/{file_path}" if file_path else f"skills/{name}"
    +
    +        if action == "patch":
    +            old = str(arguments.get("old_string") or "")
    +            new = str(arguments.get("new_string") or "")
    +            content = [acp.tool_diff_content(path=path, old_text=old or None, new_text=new)]
    +        elif action in {"edit", "create"}:
    +            content = [
    +                acp.tool_diff_content(
    +                    path=path,
    +                    new_text=str(arguments.get("content") or ""),
    +                )
    +            ]
    +        elif action == "write_file":
    +            target = str(arguments.get("file_path") or "file")
    +            content = [
    +                acp.tool_diff_content(
    +                    path=f"skills/{name}/{target}",
    +                    new_text=str(arguments.get("file_content") or ""),
    +                )
    +            ]
    +        elif action in {"delete", "remove_file"}:
    +            target = str(arguments.get("file_path") or file_path or name)
    +            content = [_text(f"Removing {target} from skill '{name}'")]
    +        else:
    +            content = [_text(f"Running skill_manage action '{action}' on skill '{name}' ({file_path})")]
    +
    +        return acp.start_tool_call(
    +            tool_call_id, title, kind=kind, content=content, locations=locations,
    +        )
    +
    +    if tool_name == "execute_code":
    +        code = str(arguments.get("code") or "").strip()
    +        preview = code[:1200] + (f"\n... ({len(code)} chars total, truncated)" if len(code) > 1200 else "")
    +        content = [_text(f"Running Python helper script:\n\n```python\n{preview}\n```" if preview else "Running Python helper script")]
    +        return acp.start_tool_call(
    +            tool_call_id, title, kind=kind, content=content, locations=locations,
    +        )
    +
    +    if tool_name == "web_search":
    +        query = str(arguments.get("query") or "").strip()
    +        content = [_text(f"Searching the web for: {query}" if query else "Searching the web")]
    +        return acp.start_tool_call(
    +            tool_call_id, title, kind=kind, content=content, locations=locations,
    +        )
    +
    +    if tool_name == "web_extract":
    +        # The title identifies the URL(s). Avoid a duplicate content block so
    +        # Zed renders this like read_file: compact start, concise completion.
    +        return acp.start_tool_call(
    +            tool_call_id, title, kind=kind, content=None, locations=locations,
    +        )
    +
    +    if tool_name == "process":
    +        action = str(arguments.get("action") or "").strip() or "manage"
    +        sid = str(arguments.get("session_id") or "").strip()
    +        data_preview = str(arguments.get("data") or "").strip()
    +        text = f"Process action: {action}" + (f"\nSession: {sid}" if sid else "")
    +        if data_preview:
    +            text += "\nInput: " + _truncate_text(data_preview, limit=500)
    +        content = [_text(text)]
    +        return acp.start_tool_call(
    +            tool_call_id, title, kind=kind, content=content, locations=locations,
    +        )
    +
    +    if tool_name == "delegate_task":
    +        tasks = arguments.get("tasks")
    +        if isinstance(tasks, list) and tasks:
    +            lines = [f"Delegating {len(tasks)} tasks", ""]
    +            for i, task in enumerate(tasks[:8], 1):
    +                if isinstance(task, dict):
    +                    goal = str(task.get("goal") or "").strip()
    +                    role = str(task.get("role") or "").strip()
    +                    lines.append(f"{i}. " + _truncate_text(goal, limit=160) + (f" ({role})" if role else ""))
    +            if len(tasks) > 8:
    +                lines.append(f"... {len(tasks) - 8} more")
    +            content = [_text("\n".join(lines))]
    +        else:
    +            goal = str(arguments.get("goal") or "").strip()
    +            content = [_text("Delegating task" + (f":\n{_truncate_text(goal, limit=800)}" if goal else ""))]
    +        return acp.start_tool_call(
    +            tool_call_id, title, kind=kind, content=content, locations=locations,
    +        )
    +
    +    if tool_name == "session_search":
    +        query = str(arguments.get("query") or "").strip()
    +        content = [_text(f"Searching past sessions for: {query}" if query else "Loading recent sessions")]
    +        return acp.start_tool_call(
    +            tool_call_id, title, kind=kind, content=content, locations=locations,
    +        )
    +
    +    if tool_name == "memory":
    +        action = str(arguments.get("action") or "manage").strip() or "manage"
    +        target = str(arguments.get("target") or "memory").strip() or "memory"
    +        preview = str(arguments.get("content") or arguments.get("old_text") or "").strip()
    +        text = f"Memory {action} ({target})"
    +        if preview:
    +            text += "\nPreview: " + _truncate_text(preview, limit=500)
    +        content = [_text(text)]
    +        return acp.start_tool_call(
    +            tool_call_id, title, kind=kind, content=content, locations=locations,
    +        )
    +
    +    if tool_name in _POLISHED_TOOLS:
    +        try:
    +            args_text = json.dumps(arguments, indent=2, default=str)
    +        except (TypeError, ValueError):
    +            args_text = str(arguments)
    +        content = [_text(_truncate_text(args_text, limit=1200))]
             return acp.start_tool_call(
                 tool_call_id, title, kind=kind, content=content, locations=locations,
    -            raw_input=arguments,
             )
     
         # Generic fallback
    ```

    ```diff
    @@ -334,7 +1131,7 @@ def build_tool_start(
         content = [acp.tool_content(acp.text_block(args_text))]
         return acp.start_tool_call(
             tool_call_id, title, kind=kind, content=content, locations=locations,
    -        raw_input=arguments,
    +        raw_input=None if tool_name in _POLISHED_TOOLS else arguments,
         )
     
     
    ```

    ```diff
    @@ -347,18 +1144,22 @@ def build_tool_complete(
     ) -> ToolCallProgress:
         """Create a ToolCallUpdate (progress) event for a completed tool call."""
         kind = get_tool_kind(tool_name)
    -    content = _build_tool_complete_content(
    -        tool_name,
    -        result,
    -        function_args=function_args,
    -        snapshot=snapshot,
    -    )
    +    if tool_name == "web_extract":
    +        error_text = _format_web_extract_result(result)
    +        content = [_text(error_text)] if error_text else None
    +    else:
    +        content = _build_tool_complete_content(
    +            tool_name,
    +            result,
    +            function_args=function_args,
    +            snapshot=snapshot,
    +        )
         return acp.update_tool_call(
             tool_call_id,
             kind=kind,
             status="completed",
             content=content,
    -        raw_output=result,
    +        raw_output=None if tool_name in _POLISHED_TOOLS else result,
         )
     
     
    ```
  - `tests/acp/test_mcp_e2e.py`

    ```diff
    @@ -178,9 +178,10 @@ def mock_run_conversation(user_message, conversation_history=None, task_id=None,
             complete_event = completions[0]
             assert isinstance(complete_event, ToolCallProgress)
             assert complete_event.status == "completed"
    -        # rawOutput should contain the tool result string
    -        assert complete_event.raw_output is not None
    -        assert "hello" in str(complete_event.raw_output)
    +        # Completion should contain human-readable output rather than forcing raw JSON panes.
    +        assert complete_event.content
    +        assert "hello" in complete_event.content[0].content.text
    +        assert complete_event.raw_output is None
     
         def test_patch_mode_tool_start_emits_diff_blocks_for_v4a_patch(self):
             update = build_tool_start(
    ```
  - `tests/acp/test_server.py`

    ```diff
    @@ -27,7 +27,10 @@
         SetSessionModeResponse,
         SessionInfo,
         TextContentBlock,
    +    ToolCallProgress,
    +    ToolCallStart,
         Usage,
    +    UsageUpdate,
         UserMessageChunk,
     )
     from acp_adapter.server import HermesACPAgent, HERMES_VERSION
    ```

    ```diff
    @@ -200,6 +203,8 @@ async def test_send_available_commands_update(self, agent):
                 "context",
                 "reset",
                 "compact",
    +            "steer",
    +            "queue",
                 "version",
             ]
             model_cmd = next(
    ```

    ```diff
    @@ -208,6 +213,46 @@ async def test_send_available_commands_update(self, agent):
             assert model_cmd.input is not None
             assert model_cmd.input.root.hint == "model name to switch to"
     
    +    def test_build_usage_update_for_zed_context_indicator(self, agent, mock_manager):
    +        state = mock_manager.create_session(cwd="/tmp")
    +        state.history = [{"role": "user", "content": "hello"}]
    +        state.agent.context_compressor = MagicMock(context_length=100_000)
    +        state.agent._cached_system_prompt = "system"
    +        state.agent.tools = [{"type": "function", "function": {"name": "demo"}}]
    +
    +        with patch(
    +            "agent.model_metadata.estimate_request_tokens_rough",
    +            return_value=25_000,
    +        ):
    +            update = agent._build_usage_update(state)
    +
    +        assert isinstance(update, UsageUpdate)
    +        assert update.session_update == "usage_update"
    +        assert update.size == 100_000
    +        assert update.used == 25_000
    +
    +    @pytest.mark.asyncio
    +    async def test_send_usage_update_to_client(self, agent, mock_manager):
    +        state = mock_manager.create_session(cwd="/tmp")
    +        state.agent.context_compressor = MagicMock(context_length=100_000)
    +        mock_conn = MagicMock(spec=acp.Client)
    +        mock_conn.session_update = AsyncMock()
    +        agent._conn = mock_conn
    +
    +        with patch(
    +            "agent.model_metadata.estimate_request_tokens_rough",
    +            return_value=25_000,
    +        ):
    +            await agent._send_usage_update(state)
    +
    +        mock_conn.session_update.assert_awaited_once()
    +        call = mock_conn.session_update.await_args
    +        assert call.kwargs["session_id"] == state.session_id
    +        update = call.kwargs["update"]
    +        assert isinstance(update, UsageUpdate)
    +        assert update.size == 100_000
    +        assert update.used == 25_000
    +
         @pytest.mark.asyncio
         async def test_cancel_sets_event(self, agent):
             resp = await agent.new_session(cwd=".")
    ```

    ```diff
    @@ -238,11 +283,31 @@ async def test_load_session_replays_persisted_history_to_client(self, agent):
                 {"role": "system", "content": "hidden system"},
                 {"role": "user", "content": "what controls the / slash commands?"},
                 {"role": "assistant", "content": "HermesACPAgent._ADVERTISED_COMMANDS controls them."},
    -            {"role": "tool", "content": "tool output should not replay"},
    +            {
    +                "role": "assistant",
    +                "content": "",
    +                "tool_calls": [
    +                    {
    +                        "id": "call_search_1",
    +                        "type": "function",
    +                        "function": {
    +                            "name": "search_files",
    +                            "arguments": '{"pattern":"slash commands","path":"."}',
    +                        },
    +                    }
    +                ],
    +            },
    +            {
    +                "role": "tool",
    +                "tool_call_id": "call_search_1",
    +                "content": '{"total_count":1,"matches":[{"path":"cli.py","line":42,"content":"slash commands"}]}',
    +            },
             ]
     
             mock_conn.session_update.reset_mock()
             resp = await agent.load_session(cwd="/tmp", session_id=new_resp.session_id)
    +        await asyncio.sleep(0)
    +        await asyncio.sleep(0)
     
             assert isinstance(resp, LoadSessionResponse)
             calls = mock_conn.session_update.await_args_list
    ```

    ```diff
    @@ -257,6 +322,21 @@ async def test_load_session_replays_persisted_history_to_client(self, agent):
             assert isinstance(replay_calls[1].kwargs["update"], AgentMessageChunk)
             assert replay_calls[1].kwargs["update"].content.text.startswith("HermesACPAgent")
     
    +        tool_updates = [
    +            call.kwargs["update"]
    +            for call in calls
    +            if getattr(call.kwargs.get("update"), "session_update", None)
    +            in {"tool_call", "tool_call_update"}
    +        ]
    +        assert len(tool_updates) == 2
    +        assert isinstance(tool_updates[0], ToolCallStart)
    +        assert tool_updates[0].tool_call_id == "call_search_1"
    +        assert tool_updates[0].title == "search: slash commands"
    +        assert isinstance(tool_updates[1], ToolCallProgress)
    +        assert tool_updates[1].tool_call_id == "call_search_1"
    +        assert "Search results" in tool_updates[1].content[0].content.text
    +        assert "cli.py:42" in tool_updates[1].content[0].content.text
    +
         @pytest.mark.asyncio
         async def test_resume_session_replays_persisted_history_to_client(self, agent):
             mock_conn = MagicMock(spec=acp.Client)
    ```

    ```diff
    @@ -269,6 +349,8 @@ async def test_resume_session_replays_persisted_history_to_client(self, agent):
     
             mock_conn.session_update.reset_mock()
             resp = await agent.resume_session(cwd="/tmp", session_id=new_resp.session_id)
    +        await asyncio.sleep(0)
    +        await asyncio.sleep(0)
     
             assert isinstance(resp, ResumeSessionResponse)
             updates = [call.kwargs["update"] for call in mock_conn.session_update.await_args_list]
    ```

    ```diff
    @@ -278,6 +360,27 @@ async def test_resume_session_replays_persisted_history_to_client(self, agent):
                 for update in updates
             )
     
    +    @pytest.mark.asyncio
    +    async def test_load_session_schedules_history_replay_after_response(self, agent):
    +        """Zed only attaches replayed updates after session/load has completed."""
    +        new_resp = await agent.new_session(cwd="/tmp")
    +        state = agent.session_manager.get_session(new_resp.session_id)
    +        state.history = [{"role": "user", "content": "hello from history"}]
    +        events = []
    +
    +        async def replay_after_response(_state):
    +            events.append("replay")
    +
    +        with patch.object(agent, "_replay_session_history", side_effect=replay_after_response):
    +            resp = await agent.load_session(cwd="/tmp", session_id=new_resp.session_id)
    +            events.append("returned")
    +
    +        assert isinstance(resp, LoadSessionResponse)
    +        assert events == ["returned"]
    +        await asyncio.sleep(0)
    +        await asyncio.sleep(0)
    +        assert events == ["returned", "replay"]
    +
         @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")
    ```

    ```diff
    @@ -522,6 +625,11 @@ async def test_prompt_runs_agent(self, agent):
             assert isinstance(resp, PromptResponse)
             assert resp.stop_reason == "end_turn"
             state.agent.run_conversation.assert_called_once()
    +        assert state.agent.tool_progress_callback is not None
    +        assert state.agent.step_callback is not None
    +        assert state.agent.stream_delta_callback is not None
    +        assert state.agent.reasoning_callback is not None
    +        assert state.agent.thinking_callback is None
     
         @pytest.mark.asyncio
         async def test_prompt_updates_history(self, agent):
    ```

    ```diff
    @@ -565,12 +673,40 @@ async def test_prompt_sends_final_message_update(self, agent):
             prompt = [TextContentBlock(type="text", text="help me")]
             await agent.prompt(prompt=prompt, session_id=new_resp.session_id)
     
    -        # session_update should have been called with the final message
    +        # session_update should include the final message (usage_update may follow it)
             mock_conn.session_update.assert_called()
    -        # Get the last call's update argument
    -        last_call = mock_conn.session_update.call_args_list[-1]
    -        update = last_call[1].get("update") or last_call[0][1]
    -        assert update.session_update == "agent_message_chunk"
    +        updates = [
    +            call.kwargs.get("update") or call.args[1]
    +            for call in mock_conn.session_update.call_args_list
    +        ]
    +        assert any(update.session_update == "agent_message_chunk" for update in updates)
    +
    +    @pytest.mark.asyncio
    +    async def test_prompt_does_not_duplicate_streamed_final_message(self, agent):
    +        """If ACP already streamed response chunks, final_response should not be sent again."""
    +        new_resp = await agent.new_session(cwd=".")
    +        state = agent.session_manager.get_session(new_resp.session_id)
    +
    +        def mock_run(*args, **kwargs):
    +            state.agent.stream_delta_callback("streamed answer")
    +            return {"final_response": "streamed answer", "messages": []}
    +
    +        state.agent.run_conversation = mock_run
    +
    +        mock_conn = MagicMock(spec=acp.Client)
    +        mock_conn.session_update = AsyncMock()
    +        agent._conn = mock_conn
    +
    +        prompt = [TextContentBlock(type="text", text="hello")]
    +        await agent.prompt(prompt=prompt, session_id=new_resp.session_id)
    +
    +        updates = [
    +            call.kwargs.get("update") or call.args[1]
    +            for call in mock_conn.session_update.call_args_list
    +        ]
    +        agent_chunks = [update for update in updates if update.session_update == "agent_message_chunk"]
    +        assert len(agent_chunks) == 1
    +        assert agent_chunks[0].content.text == "streamed answer"
     
         @pytest.mark.asyncio
         async def test_prompt_auto_titles_session(self, agent):
    ```

    ```diff
    @@ -708,6 +844,43 @@ def test_context_with_messages(self, agent, mock_manager):
             assert "2 messages" in result
             assert "user: 1" in result
     
    +    def test_context_shows_usage_and_compression_threshold(self, agent, mock_manager):
    +        state = self._make_state(mock_manager)
    +        state.history = [{"role": "user", "content": "hello"}]
    +        state.agent.context_compressor = MagicMock(
    +            context_length=100_000,
    +            threshold_tokens=[redacted:secret-assignment]
    +        )
    +        state.agent._cached_system_prompt = "system"
    +        state.agent.tools = [{"type": "function", "function": {"name": "demo"}}]
    +
    +        with patch(
    +            "agent.model_metadata.estimate_request_tokens_rough",
    +            return_value=25_000,
    +        ):
    +            result = agent._handle_slash_command("/context", state)
    +
    +        assert "Context usage: ~25,000 / 100,000 tokens (25.0%)" in result
    +        assert "Compression: ~55,000 tokens until threshold (~80,000, 80%)" in result
    +        assert "Tip: run /compact" in result
    +
    +    def test_context_says_compression_due_when_past_threshold(self, agent, mock_manager):
    +        state = self._make_state(mock_manager)
    +        state.history = [{"role": "user", "content": "hello"}]
    +        state.agent.context_compressor = MagicMock(
    +            context_length=100_000,
    +            threshold_tokens=[redacted:secret-assignment]
    +        )
    +
    +        with patch(
    +            "agent.model_metadata.estimate_request_tokens_rough",
    +            return_value=82_000,
    +        ):
    +            result = agent._handle_slash_command("/context", state)
    +
    +        assert "Context usage: ~82,000 / 100,000 tokens (82.0%)" in result
    +        assert "Compression: due now (threshold ~80,000, 80%). Run /compact." in result
    +
         def test_reset_clears_history(self, agent, mock_manager):
             state = self._make_state(mock_manager)
             state.history = [{"role": "user", "content": "hello"}]
    ```

    ```diff
    @@ -787,7 +960,12 @@ async def test_slash_command_intercepted_in_prompt(self, agent, mock_manager):
             resp = await agent.prompt(prompt=prompt, session_id=new_resp.session_id)
     
             assert resp.stop_reason == "end_turn"
    -        mock_conn.session_update.assert_called_once()
    +        updates = [
    +            call.kwargs.get("update") or call.args[1]
    +            for call in mock_conn.session_update.call_args_list
    +        ]
    +        assert any(update.session_update == "agent_message_chunk" for update in updates)
    +        assert any(update.session_update == "usage_update" for update in updates)
     
         @pytest.mark.asyncio
         async def test_unknown_slash_falls_through_to_llm(self, agent, mock_manager):
    ```
  - `tests/acp/test_tools.py`

    ```diff
    @@ -52,6 +52,12 @@ def test_tool_kind_web_search(self):
         def test_tool_kind_execute_code(self):
             assert get_tool_kind("execute_code") == "execute"
     
    +    def test_tool_kind_todo(self):
    +        assert get_tool_kind("todo") == "other"
    +
    +    def test_tool_kind_skill_view(self):
    +        assert get_tool_kind("skill_view") == "read"
    +
         def test_tool_kind_browser_navigate(self):
             assert get_tool_kind("browser_navigate") == "fetch"
     
    ```

    ```diff
    @@ -110,6 +116,25 @@ def test_web_search_title(self):
             title = build_tool_title("web_search", {"query": "python asyncio"})
             assert "python asyncio" in title
     
    +    def test_skill_view_title_includes_skill_name(self):
    +        title = build_tool_title("skill_view", {"name": "github-pitfalls"})
    +        assert title == "skill view (github-pitfalls)"
    +
    +    def test_skill_view_title_includes_linked_file(self):
    +        title = build_tool_title("skill_view", {"name": "github-pitfalls", "file_path": "references/api.md"})
    +        assert title == "skill view (github-pitfalls/references/api.md)"
    +
    +    def test_execute_code_title_includes_first_code_line(self):
    +        title = build_tool_title("execute_code", {"code": "\nfrom hermes_tools import terminal\nprint('done')"})
    +        assert title == "python: from hermes_tools import terminal"
    +
    +    def test_skill_manage_title_includes_action_and_target(self):
    +        title = build_tool_title(
    +            "skill_manage",
    +            {"action": "patch", "name": "hermes-agent-operations", "file_path": "references/acp.md"},
    +        )
    +        assert title == "skill patch: hermes-agent-operations/references/acp.md"
    +
         def test_unknown_tool_uses_name(self):
             title = build_tool_title("some_new_tool", {"foo": "bar"})
             assert title == "some_new_tool"
    ```

    ```diff
    @@ -164,15 +189,23 @@ def test_build_tool_start_for_terminal(self):
             assert "ls -la /tmp" in text
     
         def test_build_tool_start_for_read_file(self):
    -        """read_file should include the path in content."""
    +        """read_file start should stay compact; completion carries file contents."""
             args = {"path": "/etc/hosts", "offset": 1, "limit": 50}
             result = build_tool_start("tc-3", "read_file", args)
             assert isinstance(result, ToolCallStart)
             assert result.kind == "read"
    -        assert len(result.content) >= 1
    -        content_item = result.content[0]
    -        assert isinstance(content_item, ContentToolCallContent)
    -        assert "/etc/hosts" in content_item.content.text
    +        assert result.content is None
    +        assert result.raw_input is None
    +
    +    def test_build_tool_start_for_web_extract_is_compact(self):
    +        """web_extract start should stay compact; title identifies URLs."""
    +        args = {"urls": ["https://example.com/docs"]}
    +        result = build_tool_start("tc-web-start", "web_extract", args)
    +        assert isinstance(result, ToolCallStart)
    +        assert result.title == "extract: https://example.com/docs"
    +        assert result.kind == "fetch"
    +        assert result.content is None
    +        assert result.raw_input is None
     
         def test_build_tool_start_for_search(self):
             """search_files should include pattern in content."""
    ```

    ```diff
    @@ -181,6 +214,48 @@ def test_build_tool_start_for_search(self):
             assert isinstance(result, ToolCallStart)
             assert result.kind == "search"
             assert "TODO" in result.content[0].content.text
    +        assert result.raw_input is None
    +
    +    def test_build_tool_start_for_todo_is_human_readable(self):
    +        args = {"todos": [{"id": "one", "content": "Fix ACP rendering", "status": "in_progress"}]}
    +        result = build_tool_start("tc-todo", "todo", args)
    +        assert result.title == "todo (1 item)"
    +        assert "Fix ACP rendering" in result.content[0].content.text
    +        assert result.raw_input is None
    +
    +    def test_build_tool_start_for_skill_view_is_human_readable(self):
    +        result = build_tool_start("tc-skill", "skill_view", {"name": "github-pitfalls"})
    +        assert result.title == "skill view (github-pitfalls)"
    +        assert "github-pitfalls" in result.content[0].content.text
    +        assert result.raw_input is None
    +
    +    def test_build_tool_start_for_execute_code_shows_code_preview(self):
    +        result = build_tool_start("tc-code", "execute_code", {"code": "print('hello')"})
    +        assert result.kind == "execute"
    +        assert result.title == "python: print('hello')"
    +        assert "```python" in result.content[0].content.text
    +        assert "print('hello')" in result.content[0].content.text
    +        assert result.raw_input is None
    +
    +    def test_build_tool_start_for_skill_manage_patch_shows_diff(self):
    +        result = build_tool_start(
    +            "tc-skill-manage",
    +            "skill_manage",
    +            {
    +                "action": "patch",
    +                "name": "hermes-agent-operations",
    +                "file_path": "references/acp.md",
    +                "old_string": "old advice",
    +                "new_string": "new advice",
    +            },
    +        )
    +        assert result.kind == "edit"
    +        assert result.title == "skill patch: hermes-agent-operations/references/acp.md"
    +        assert isinstance(result.content[0], FileEditToolCallContent)
    +        assert result.content[0].path == "skills/hermes-agent-operations/references/acp.md"
    +        assert result.content[0].old_text == "old advice"
    +        assert result.content[0].new_text == "new advice"
    +        assert result.raw_input is None
     
         def test_build_tool_start_generic_fallback(self):
             """Unknown tools should get a generic text representation."""
    ```

    ```diff
    @@ -205,6 +280,158 @@ def test_build_tool_complete_for_terminal(self):
             content_item = result.content[0]
             assert isinstance(content_item, ContentToolCallContent)
             assert "total 42" in content_item.content.text
    +        assert result.raw_output is None
    +
    +    def test_build_tool_complete_for_todo_is_checklist(self):
    +        result = build_tool_complete(
    +            "tc-todo",
    +            "todo",
    +            '{"todos":[{"id":"a","content":"Inspect ACP","status":"completed"},{"id":"b","content":"Patch renderers","status":"in_progress"}],"summary":{"total":2,"pending":0,"in_progress":1,"completed":1,"cancelled":0}}',
    +        )
    +        text = result.content[0].content.text
    +        assert "✅ Inspect ACP" in text
    +        assert "- 🔄 Patch renderers" in text
    +        assert "**Progress:** 1 completed, 1 in progress, 0 pending" in text
    +        assert result.raw_output is None
    +
    +    def test_build_tool_complete_for_skill_view_summarizes_content_without_raw_json(self):
    +        result = build_tool_complete(
    +            "tc-skill",
    +            "skill_view",
    +            '{"success":true,"name":"github-pitfalls","description":"GitHub gotchas","content":"# GitHub Pitfalls\\nUse gh carefully.","path":"github/github-pitfalls/SKILL.md"}',
    +        )
    +        text = result.content[0].content.text
    +        assert "**Skill loaded**" in text
    +        assert "`github-pitfalls`" in text
    +        assert "GitHub gotchas" in text
    +        assert "GitHub Pitfalls" in text
    +        assert "Use gh carefully" not in text
    +        assert "Full skill content is available to the agent" in text
    +        assert result.raw_output is None
    +
    +    def test_build_tool_complete_for_execute_code_formats_output(self):
    +        result = build_tool_complete("tc-code", "execute_code", '{"output":"hello\\n","exit_code":0}')
    +        text = result.content[0].content.text
    +        assert "Exit code: 0" in text
    +        assert "hello" in text
    +        assert result.raw_output is None
    +
    +    def test_build_tool_complete_for_skill_manage_summarizes_without_raw_json(self):
    +        result = build_tool_complete(
    +            "tc-skill-manage",
    +            "skill_manage",
    +            '{"success":true,"message":"Patched references/hermes-acp-zed-rendering.md in skill \'hermes-agent-operations\' (1 replacement)."}',
    +            function_args={
    +                "action": "patch",
    +                "name": "hermes-agent-operations",
    +                "file_path": "references/hermes-acp-zed-rendering.md",
    +            },
    +        )
    +        text = result.content[0].content.text
    +        assert "**✅ Skill updated**" in text
    +        assert "`patch`" in text
    +        assert "`hermes-agent-operations`" in text
    +        assert "references/hermes-acp-zed-rendering.md" in text
    +        assert "{\"success\"" not in text
    +        assert result.raw_output is None
    +
    +    def test_build_tool_complete_for_read_file_formats_content(self):
    +        result = build_tool_complete(
    +            "tc-read",
    +            "read_file",
    +            '{"content":"1|hello\\n2|world","total_lines":2}',
    +            function_args={"path":"README.md","offset":1,"limit":20},
    +        )
    +        text = result.content[0].content.text
    +        assert "Read README.md" in text
    +        assert "```\n1|hello\n2|world\n```" in text
    +        assert result.raw_output is None
    +
    +    def test_build_tool_complete_for_search_files_formats_matches(self):
    +        result = build_tool_complete(
    +            "tc-search",
    +            "search_files",
    +            '{"total_count":2,"matches":[{"path":"README.md","line":3,"content":"TODO: fix this"},{"path":"src/app.py","line":9,"content":"needle"}],"truncated":true}\n\n[Hint: Results truncated. Use offset=12 to see more.]',
    +        )
    +        text = result.content[0].content.text
    +        assert "Search results" in text
    +        assert "Found 2 matches" in text
    +        assert "README.md:3" in text
    +        assert "TODO: fix this" in text
    +        assert "Results truncated" in text
    +        assert result.raw_output is None
    +
    +    def test_build_tool_complete_for_process_list_formats_table(self):
    +        result = build_tool_complete(
    +            "tc-process",
    +            "process",
    +            '{"processes":[{"session_id":"p1","status":"running","pid":123,"command":"npm run dev"}]}',
    +            function_args={"action":"list"},
    +        )
    +        text = result.content[0].content.text
    +        assert "Processes: 1" in text
    +        assert "`p1`" in text
    +        assert "npm run dev" in text
    +        assert result.raw_output is None
    +
    +    def test_build_tool_complete_for_delegate_task_summarizes_children(self):
    +        result = build_tool_complete(
    +            "tc-delegate",
    +            "delegate_task",
    +            '{"results":[{"task_index":0,"status":"completed","summary":"Reviewed ACP rendering.","model":"gpt-5.5","duration_seconds":3.2,"tool_trace":[{"tool":"read_file"}]}],"total_duration_seconds":3.4}',
    +        )
    +        text = result.content[0].content.text
    +        assert "Delegation results: 1 task" in text
    +        assert "Reviewed ACP rendering" in text
    +        assert "gpt-5.5" in text
    +        assert "Tools: read_file" in text
    +        assert result.raw_output is None
    +
    +    def test_build_tool_complete_for_session_search_recent(self):
    +        result = build_tool_complete(
    +            "tc-session",
    +            "session_search",
    +            '{"success":true,"mode":"recent","results":[{"session_id":"s1","title":"ACP work","last_active":"2026-05-02","message_count":12,"preview":"Polished tool rendering."}],"count":1}',
    +        )
    +        text = result.content[0].content.text
    +        assert "Recent sessions" in text
    +        assert "ACP work" in text
    +        assert "Polished tool rendering" in text
    +        assert result.raw_output is None
    +
    +    def test_build_tool_complete_for_memory_avoids_dumping_entries(self):
    +        result = build_tool_complete(
    +            "tc-memory",
    +            "memory",
    +            '{"success":true,"target":"user","entries":["private long memory"],"usage":"1% — 19/2000 chars","entry_count":1,"message":"Entry added."}',
    +            function_args={"action":"add","target":"user","content":"User likes concise ACP rendering."},
    +        )
    +        text = result.content[0].content.text
    +        assert "Memory add saved" in text
    +        assert "User likes concise ACP rendering" in text
    +        assert "private long memory" not in text
    +        assert result.raw_output is None
    +
    +    def test_build_tool_complete_for_web_extract_success_stays_compact(self):
    +        result = build_tool_complete(
    +            "tc-web-extract",
    +            "web_extract",
    +            '{"results":[{"url":"https://example.com","title":"Example","content":"# Intro\\nThis is extracted content."}]}',
    +        )
    +        assert result.content is None
    +        assert result.raw_output is None
    +
    +    def test_build_tool_complete_for_web_extract_error_shows_error(self):
    +        result = build_tool_complete(
    +            "tc-web-extract-error",
    +            "web_extract",
    +            '{"results":[{"url":"https://example.com","title":"Example","error":"timeout"}]}',
    +        )
    +        text = result.content[0].content.text
    +        assert "Web extract failed" in text
    +        assert "https://example.com" in text
    +        assert "timeout" in text
    +        assert result.raw_output is None
     
         def test_build_tool_complete_truncates_large_output(self):
             """Very large outputs should be truncated."""
    ```
- **#26943: fix(acp): replay assistant reasoning as agent_thought_chunk on session/load (#12285)**. merged to main · fix tier `verified`
  - maintainer direction: category `merged`; implication `superseded`
  - files: `acp_adapter/server.py`, `tests/acp/test_server.py`
  - fix hunks:
  - `acp_adapter/server.py`

    ```diff
    @@ -18,6 +18,7 @@
     from acp.schema import (
         AgentCapabilities,
         AgentMessageChunk,
    +    AgentThoughtChunk,
         AuthenticateResponse,
         AvailableCommand,
         AvailableCommandsUpdate,
    ```

    ```diff
    @@ -788,14 +789,20 @@ async def authenticate(self, method_id: str, **kwargs: Any) -> AuthenticateRespo
         # ---- Session management -------------------------------------------------
     
         @staticmethod
    -    def _history_message_text(message: dict[str, Any]) -> str:
    -        """Extract displayable text from a persisted OpenAI-style message."""
    -        content = message.get("content")
    -        if isinstance(content, str):
    -            return content.strip()
    -        if isinstance(content, list):
    +    def _flatten_history_text(value: Any) -> str:
    +        """Normalize a persisted text-or-text-parts value into a single string.
    +
    +        OpenAI-style assistant content (and provider reasoning fields) can arrive
    +        as either a scalar string or a list of ``{"text": ...}`` /
    +        ``{"type": "text", "content": ...}`` parts. Whitespace-only inputs
    +        collapse to an empty string so callers can treat ``""`` as "nothing to
    +        emit".
    +        """
    +        if isinstance(value, str):
    +            return value.strip()
    +        if isinstance(value, list):
                 parts: list[str] = []
    -            for item in content:
    +            for item in value:
                     if isinstance(item, dict):
                         text = item.get("text")
                         if isinstance(text, str):
    ```

    ```diff
    @@ -807,6 +814,29 @@ def _history_message_text(message: dict[str, Any]) -> str:
                 return "\n".join(part.strip() for part in parts if part and part.strip()).strip()
             return ""
     
    +    @classmethod
    +    def _history_message_text(cls, message: dict[str, Any]) -> str:
    +        """Extract displayable text from a persisted OpenAI-style message."""
    +        return cls._flatten_history_text(message.get("content"))
    +
    +    @classmethod
    +    def _history_reasoning_text(cls, message: dict[str, Any]) -> str:
    +        """Extract displayable reasoning/thought text from a persisted assistant message.
    +
    +        Returns the first non-empty value among ``reasoning_content`` (the
    +        canonical field used by DeepSeek / Moonshot and the post-#16892
    +        chat-completions normalizer) and ``reasoning`` (used by the codex
    +        event projector and several other transports). Both keys are
    +        actively written by live code paths, so neither branch is
    +        deprecated — they cover different transports rather than old vs.
    +        new sessions.
    +        """
    +        for key in ("reasoning_content", "reasoning"):
    +            text = cls._flatten_history_text(message.get(key))
    +            if text:
    +                return text
    +        return ""
    +
         @staticmethod
         def _history_message_update(
             *,
    ```

    ```diff
    @@ -827,6 +857,11 @@ def _history_message_update(
                 )
             return None
     
    +    @staticmethod
    +    def _history_thought_update(text: str) -> AgentThoughtChunk:
    +        """Build an ACP history replay update for an assistant thought."""
    +        return acp.update_agent_thought_text(text)
    +
         @staticmethod
         def _history_tool_call_name_args(tool_call: dict[str, Any]) -> tuple[str, dict[str, Any]]:
             """Extract function name/arguments from an OpenAI-style tool_call."""
    ```

    ```diff
    @@ -858,9 +893,10 @@ async def _replay_session_history(self, state: SessionState) -> None:
     
             Zed's ACP history UI calls ``session/load`` after the user picks an item
             from the Agents sidebar. The agent must then replay the full conversation
    -        as user/assistant chunks plus reconstructed tool-call start/completion
    -        notifications; merely restoring server-side state makes Hermes remember
    -        context, but leaves the editor looking like a clean thread.
    +        as user/assistant chunks, thinking-mode thought chunks, plus reconstructed
    +        tool-call start/completion notifications; merely restoring server-side
    +        state makes Hermes remember context, but leaves the editor looking like a
    +        clean thread.
             """
             if not self._conn or not state.history:
                 return
    ```

    ```diff
    @@ -882,24 +918,37 @@ async def _send(update: Any) -> bool:
             for message in state.history:
                 role = str(message.get("role") or "")
     
    -            if role in {"user", "assistant"}:
    +            if role == "user":
                     text = self._history_message_text(message)
                     if text:
                         update = self._history_message_update(role=role, text=text)
                         if update is not None and not await _send(update):
                             return
    +                continue
     
    -            if role == "assistant" and isinstance(message.get("tool_calls"), list):
    -                for tool_call in message["tool_calls"]:
    -                    if not isinstance(tool_call, dict):
    -                        continue
    -                    tool_call_id = self._history_tool_call_id(tool_call)
    -                    if not tool_call_id:
    -                        continue
    -                    tool_name, args = self._history_tool_call_name_args(tool_call)
    -                    active_tool_calls[tool_call_id] = (tool_name, args)
    -                    if not await _send(build_tool_start(tool_call_id, tool_name, args)):
    +            if role == "assistant":
    +                thought = self._history_reasoning_text(message)
    +                if thought and not await _send(self._history_thought_update(thought)):
    +                    return
    +
    +                text = self._history_message_text(message)
    +                if text:
    +                    update = self._history_message_update(role=role, text=text)
    +                    if update is not None and not await _send(update):
                             return
    +
    +                tool_calls = message.get("tool_calls")
    +                if isinstance(tool_calls, list):
    +                    for tool_call in tool_calls:
    +                        if not isinstance(tool_call, dict):
    +                            continue
    +                        tool_call_id = self._history_tool_call_id(tool_call)
    +                        if not tool_call_id:
    +                            continue
    +                        tool_name, args = self._history_tool_call_name_args(tool_call)
    +                        active_tool_calls[tool_call_id] = (tool_name, args)
    +                        if not await _send(build_tool_start(tool_call_id, tool_name, args)):
    +                            return
                     continue
     
                 if role == "tool":
    ```
  - `tests/acp/test_server.py`

    ```diff
    @@ -13,6 +13,7 @@
         AgentCapabilities,
         AgentMessageChunk,
         AgentPlanUpdate,
    +    AgentThoughtChunk,
         AuthenticateResponse,
         AvailableCommandsUpdate,
         Implementation,
    ```

    ```diff
    @@ -466,6 +467,211 @@ async def test_resume_session_replays_persisted_history_to_client(self, agent):
                 for update in updates
             )
     
    +    @pytest.mark.asyncio
    +    async def test_load_session_replays_reasoning_thought_before_message(self, agent):
    +        """Thinking-model thoughts must be replayed via ``agent_thought_chunk``.
    +
    +        Regression for #12285 — when a session is loaded, persisted assistant
    +        ``reasoning_content`` / ``reasoning`` fields must surface as ACP
    +        ``AgentThoughtChunk`` notifications in the same relative position they
    +        had live (thought streams before the assistant message text), so Zed's
    +        collapsed Thinking pane rebuilds instead of vanishing on reconnect.
    +        """
    +        mock_conn = MagicMock(spec=acp.Client)
    +        mock_conn.session_update = AsyncMock()
    +        agent._conn = mock_conn
    +
    +        new_resp = await agent.new_session(cwd="/tmp")
    +        state = agent.session_manager.get_session(new_resp.session_id)
    +        state.history = [
    +            {"role": "user", "content": "Walk me through it."},
    +            {
    +                "role": "assistant",
    +                "reasoning_content": "Let me think step by step about the request.",
    +                "content": "Here is the plan.",
    +            },
    +            {"role": "user", "content": "And the legacy case?"},
    +            {
    +                "role": "assistant",
    +                # No reasoning_content — exercise the legacy "reasoning" fallback
    +                # path so sessions persisted before #16892 still replay thoughts.
    +                "reasoning": "Older sessions stored the trace under the internal key.",
    +                "content": "Same idea, older field name.",
    +            },
    +        ]
    +
    +        mock_conn.session_update.reset_mock()
    +        resp = await agent.load_session(cwd="/tmp", session_id=new_resp.session_id)
    +        await asyncio.sleep(0)
    +        await asyncio.sleep(0)
    +
    +        assert isinstance(resp, LoadSessionResponse)
    +
    +        replay_kinds = [
    +            getattr(call.kwargs.get("update"), "session_update", None)
    +            for call in mock_conn.session_update.await_args_list
    +            if getattr(call.kwargs.get("update"), "session_update", None)
    +            in {"user_message_chunk", "agent_message_chunk", "agent_thought_chunk"}
    +        ]
    +        assert replay_kinds == [
    +            "user_message_chunk",
    +            "agent_thought_chunk",
    +            "agent_message_chunk",
    +            "user_message_chunk",
    +            "agent_thought_chunk",
    +            "agent_message_chunk",
    +        ]
    +
    +        thought_updates = [
    +            call.kwargs["update"]
    +            for call in mock_conn.session_update.await_args_list
    +            if isinstance(call.kwargs.get("update"), AgentThoughtChunk)
    +        ]
    +        assert len(thought_updates) == 2
    +        assert thought_updates[0].content.text == "Let me think step by step about the request."
    +        assert thought_updates[1].content.text == "Older sessions stored the trace under the internal key."
    +
    +    @pytest.mark.asyncio
    +    async def test_load_session_replays_reasoning_only_turn(self, agent):
    +        """Assistant turns with reasoning but no content should still emit a thought.
    +
    +        Pure reasoning-only assistant entries (e.g. a thinking step before a
    +        tool-call turn) commonly carry ``reasoning_content`` with empty
    +        ``content``. The replay must still surface the thought so the editor's
    +        Thinking pane rebuilds, even when there is no message text to follow.
    +        """
    +        mock_conn = MagicMock(spec=acp.Client)
    +        mock_conn.session_update = AsyncMock()
    +        agent._conn = mock_conn
    +
    +        new_resp = await agent.new_session(cwd="/tmp")
    +        state = agent.session_manager.get_session(new_resp.session_id)
    +        state.history = [
    +            {
    +                "role": "assistant",
    +                "reasoning_content": "I should call the search tool next.",
    +                "content": "",
    +            },
    +        ]
    +
    +        mock_conn.session_update.reset_mock()
    +        await agent.load_session(cwd="/tmp", session_id=new_resp.session_id)
    +        await asyncio.sleep(0)
    +        await asyncio.sleep(0)
    +
    +        thought_updates = [
    +            call.kwargs["update"]
    +            for call in mock_conn.session_update.await_args_list
    +            if isinstance(call.kwargs.get("update"), AgentThoughtChunk)
    +        ]
    +        message_updates = [
    +            call.kwargs["update"]
    +            for call in mock_conn.session_update.await_args_list
    +            if isinstance(call.kwargs.get("update"), AgentMessageChunk)
    +        ]
    +        assert len(thought_updates) == 1
    +        assert thought_updates[0].content.text == "I should call the search tool next."
    +        assert message_updates == []
    +
    +    @pytest.mark.asyncio
    +    async def test_load_session_skips_empty_reasoning_fields(self, agent):
    +        """Empty/whitespace reasoning fields must not produce notifications."""
    +        mock_conn = MagicMock(spec=acp.Client)
    +        mock_conn.session_update = AsyncMock()
    +        agent._conn = mock_conn
    +
    +        new_resp = await agent.new_session(cwd="/tmp")
    +        state = agent.session_manager.get_session(new_resp.session_id)
    +        state.history = [
    +            {
    +                "role": "assistant",
    +                "reasoning_content": "",
    +                "reasoning": "   \n\t",
    +                "content": "Just a regular answer.",
    +            },
    +        ]
    +
    +        mock_conn.session_update.reset_mock()
    +        await agent.load_session(cwd="/tmp", session_id=new_resp.session_id)
    +        await asyncio.sleep(0)
    +        await asyncio.sleep(0)
    +
    +        thought_updates = [
    +            call.kwargs["update"]
    +            for call in mock_conn.session_update.await_args_list
    +            if isinstance(call.kwargs.get("update"), AgentThoughtChunk)
    +        ]
    +        assert thought_updates == []
    +
    +    @pytest.mark.asyncio
    +    async def test_load_session_replays_thought_then_tool_call_without_message(self, agent):
    +        """Canonical thinking-model shape: reasoning + tool_call + no body text.
    +
    +        Thinking models commonly emit a pre-tool thought followed by a
    +        tool_calls turn with empty ``content``. Replay must emit:
    +        ``agent_thought_chunk`` then ``tool_call`` then ``tool_call_update``
    +        for the matching tool result — and crucially, NO ``agent_message_chunk``
    +        for the empty-text assistant body. Regression for the canonical
    +        thinking-then-tool flow on #12285.
    +        """
    +        mock_conn = MagicMock(spec=acp.Client)
    +        mock_conn.session_update = AsyncMock()
    +        agent._conn = mock_conn
    +
    +        new_resp = await agent.new_session(cwd="/tmp")
    +        state = agent.session_manager.get_session(new_resp.session_id)
    +        state.history = [
    +            {"role": "user", "content": "Find the bug."},
    +            {
    +                "role": "assistant",
    +                "reasoning_content": "I should grep for the function name first.",
    +                "content": "",
    +                "tool_calls": [
    +                    {
    +                        "id": "call_grep_1",
    +                        "type": "function",
    +                        "function": {
    +                            "name": "search_files",
    +                            "arguments": '{"pattern":"foo","path":"."}',
    +                        },
    +                    }
    +                ],
    +            },
    +            {
    +                "role": "tool",
    +                "tool_call_id": "call_grep_1",
    +                "content": '{"total_count":1,"matches":[{"path":"x.py","line":1,"content":"foo"}]}',
    +            },
    +        ]
    +
    +        mock_conn.session_update.reset_mock()
    +        await agent.load_session(cwd="/tmp", session_id=new_resp.session_id)
    +        await asyncio.sleep(0)
    +        await asyncio.sleep(0)
    +
    +        kinds = [
    +            getattr(call.kwargs.get("update"), "session_update", None)
    +            for call in mock_conn.session_update.await_args_list
    +            if getattr(call.kwargs.get("update"), "session_update", None)
    +            in {
    +                "user_message_chunk",
    +                "agent_thought_chunk",
    +                "agent_message_chunk",
    +                "tool_call",
    +                "tool_call_update",
    +            }
    +        ]
    +        # No agent_message_chunk for the empty-content assistant turn.
    +        assert "agent_message_chunk" not in kinds
    +        # Thought must precede the tool_call_start within the assistant turn,
    +        # and the tool result follows.
    +        assert kinds == [
    +            "user_message_chunk",
    +            "agent_thought_chunk",
    +            "tool_call",
    +            "tool_call_update",
    +        ]
    +
         @pytest.mark.asyncio
         async def test_load_session_schedules_history_replay_after_response(self, agent):
             """Zed only attaches replayed updates after session/load has completed."""
    ```
- **#26957: fix(acp): replay session history before responding to session/load (#12285 follow-up)**. merged to main · fix tier `verified`
  - maintainer direction: category `merged`; implication `superseded`
  - files: `acp_adapter/server.py`, `tests/acp/test_server.py`
  - fix hunks:
  - `acp_adapter/server.py`

    ```diff
    @@ -889,14 +889,17 @@ def _history_tool_call_id(tool_call: dict[str, Any]) -> str:
             ).strip()
     
         async def _replay_session_history(self, state: SessionState) -> None:
    -        """Send persisted user/assistant history to clients during session/load.
    -
    -        Zed's ACP history UI calls ``session/load`` after the user picks an item
    -        from the Agents sidebar. The agent must then replay the full conversation
    -        as user/assistant chunks, thinking-mode thought chunks, plus reconstructed
    -        tool-call start/completion notifications; merely restoring server-side
    -        state makes Hermes remember context, but leaves the editor looking like a
    -        clean thread.
    +        """Replay persisted user/assistant history during session/load or session/resume.
    +
    +        Invoked inline (``await``) from both ``load_session`` and
    +        ``resume_session`` so that spec-compliant ACP clients receive the
    +        full transcript within the request's lifetime — see the comment at
    +        the call sites for the rationale and prior-art citations.
    +
    +        Replays the conversation as user/assistant chunks, thinking-mode
    +        thought chunks, plus reconstructed tool-call start/completion
    +        notifications. Merely restoring server-side state makes Hermes
    +        remember context, but leaves the editor looking like a clean thread.
             """
             if not self._conn or not state.history:
                 return
    ```

    ```diff
    @@ -991,18 +994,6 @@ async def new_session(
                 models=self._build_model_state(state),
             )
     
    -    def _schedule_history_replay(self, state: SessionState) -> None:
    -        """Replay persisted history after session/load or session/resume returns.
    -
    -        Zed only attaches streamed transcript/tool updates once the load/resume
    -        response has completed. Sending replay notifications while the request is
    -        still in-flight can make the server look correct in logs while the editor
    -        drops or fails to attach the tool-call history.
    -        """
    -        loop = asyncio.get_running_loop()
    -        replay_coro = self._replay_session_history(state)
    -        loop.call_soon(asyncio.create_task, replay_coro)
    -
         async def load_session(
             self,
             cwd: str,
    ```

    ```diff
    @@ -1016,7 +1007,30 @@ async def load_session(
                 return None
             await self._register_session_mcp_servers(state, mcp_servers)
             logger.info("Loaded session %s", session_id)
    -        self._schedule_history_replay(state)
    +        # Per ACP spec, `session/load` must stream the prior conversation back
    +        # to the client via `session/update` notifications BEFORE responding,
    +        # so the client receives the full transcript within the load request's
    +        # lifetime. Awaiting the replay here matches Codex / Claude Code /
    +        # OpenCode / Pi and the Zed client (which registers the session-update
    +        # routing entry before awaiting the loadSession RPC specifically so
    +        # in-call history replay updates can find the thread). Deferring this
    +        # via `loop.call_soon` (as we did briefly in May 2026) broke every
    +        # spec-compliant ACP client that measures notifications synchronously
    +        # against the load response — see #12285 follow-up.
    +        try:
    +            await self._replay_session_history(state)
    +        except Exception:
    +            # Replay is best-effort — a corrupted or unexpected message shape
    +            # must not turn a successful session/load into a JSON-RPC error
    +            # response. Per-notification failures are already caught inside
    +            # ``_replay_session_history``; this outer guard covers anything
    +            # raised by the helpers themselves before reaching ``_send``.
    +            logger.warning(
    +                "ACP history replay raised during session/load for %s — "
    +                "load will still succeed, partial transcript may be missing",
    +                session_id,
    +                exc_info=True,
    +            )
             self._schedule_available_commands_update(session_id)
             self._schedule_usage_update(state)
             return LoadSessionResponse(models=self._build_model_state(state))
    ```

    ```diff
    @@ -1034,7 +1048,18 @@ 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_history_replay(state)
    +        # See `load_session` above for the spec rationale — replay must
    +        # complete before the response so clients receive the full transcript
    +        # within the request's lifetime.
    +        try:
    +            await self._replay_session_history(state)
    +        except Exception:
    +            logger.warning(
    +                "ACP history replay raised during session/resume for %s — "
    +                "resume will still succeed, partial transcript may be missing",
    +                state.session_id,
    +                exc_info=True,
    +            )
             self._schedule_available_commands_update(state.session_id)
             self._schedule_usage_update(state)
             return ResumeSessionResponse(models=self._build_model_state(state))
    ```
  - `tests/acp/test_server.py`

    ```diff
    @@ -673,25 +673,91 @@ async def test_load_session_replays_thought_then_tool_call_without_message(self,
             ]
     
         @pytest.mark.asyncio
    -    async def test_load_session_schedules_history_replay_after_response(self, agent):
    -        """Zed only attaches replayed updates after session/load has completed."""
    +    async def test_load_session_replays_history_before_returning_response(self, agent):
    +        """Per ACP spec, replay must complete BEFORE load_session returns.
    +
    +        Spec-compliant ACP clients (Codex, Claude Code, OpenCode, Pi, Zed)
    +        attach their ``session/update`` listeners before awaiting the
    +        ``loadSession`` RPC and rely on receiving the full transcript within
    +        the request's lifetime. Deferring replay via ``loop.call_soon`` (the
    +        prior behavior in May 2026) broke clients that read notification
    +        counts synchronously against the load response — see #12285 follow-up.
    +        """
             new_resp = await agent.new_session(cwd="/tmp")
             state = agent.session_manager.get_session(new_resp.session_id)
             state.history = [{"role": "user", "content": "hello from history"}]
    -        events = []
    +        events: list[str] = []
     
    -        async def replay_after_response(_state):
    +        async def replay_records(_state):
                 events.append("replay")
     
    -        with patch.object(agent, "_replay_session_history", side_effect=replay_after_response):
    +        with patch.object(agent, "_replay_session_history", side_effect=replay_records):
                 resp = await agent.load_session(cwd="/tmp", session_id=new_resp.session_id)
                 events.append("returned")
     
             assert isinstance(resp, LoadSessionResponse)
    -        assert events == ["returned"]
    -        await asyncio.sleep(0)
    -        await asyncio.sleep(0)
    -        assert events == ["returned", "replay"]
    +        # Replay must have happened BEFORE the response was constructed —
    +        # i.e. before the `events.append("returned")` after the await resolves.
    +        assert events == ["replay", "returned"]
    +
    +    @pytest.mark.asyncio
    +    async def test_resume_session_replays_history_before_returning_response(self, agent):
    +        """Same spec rationale as ``load_session`` — replay before responding."""
    +        new_resp = await agent.new_session(cwd="/tmp")
    +        state = agent.session_manager.get_session(new_resp.session_id)
    +        state.history = [{"role": "user", "content": "hello from history"}]
    +        events: list[str] = []
    +
    +        async def replay_records(_state):
    +            events.append("replay")
    +
    +        with patch.object(agent, "_replay_session_history", side_effect=replay_records):
    +            resp = await agent.resume_session(cwd="/tmp", session_id=new_resp.session_id)
    +            events.append("returned")
    +
    +        assert isinstance(resp, ResumeSessionResponse)
    +        assert events == ["replay", "returned"]
    +
    +    @pytest.mark.asyncio
    +    async def test_load_session_survives_replay_helper_exception(self, agent, caplog):
    +        """A replay helper raising must not turn load_session into an error.
    +
    +        With awaited replay, an exception in ``_replay_session_history`` now
    +        propagates into the ``load_session`` handler. The defensive try/except
    +        guard at the call site must catch and log it so the JSON-RPC client
    +        still receives a ``LoadSessionResponse`` — partial transcripts are
    +        acceptable, total load failure is not.
    +        """
    +        new_resp = await agent.new_session(cwd="/tmp")
    +        state = agent.session_manager.get_session(new_resp.session_id)
    +        state.history = [{"role": "user", "content": "hi"}]
    +
    +        async def boom(_state):
    +            raise RuntimeError("simulated replay helper crash")
    +
    +        with caplog.at_level("WARNING", logger="acp_adapter.server"):
    +            with patch.object(agent, "_replay_session_history", side_effect=boom):
    +                resp = await agent.load_session(cwd="/tmp", session_id=new_resp.session_id)
    +
    +        assert isinstance(resp, LoadSessionResponse)
    +        assert "history replay raised during session/load" in caplog.text
    +
    +    @pytest.mark.asyncio
    +    async def test_resume_session_survives_replay_helper_exception(self, agent, caplog):
    +        """Same guarantee as ``load_session`` for the resume path."""
    +        new_resp = await agent.new_session(cwd="/tmp")
    +        state = agent.session_manager.get_session(new_resp.session_id)
    +        state.history = [{"role": "user", "content": "hi"}]
    +
    +        async def boom(_state):
    +            raise RuntimeError("simulated replay helper crash")
    +
    +        with caplog.at_level("WARNING", logger="acp_adapter.server"):
    +            with patch.object(agent, "_replay_session_history", side_effect=boom):
    +                resp = await agent.resume_session(cwd="/tmp", session_id=new_resp.session_id)
    +
    +        assert isinstance(resp, ResumeSessionResponse)
    +        assert "history replay raised during session/resume" in caplog.text
     
         @pytest.mark.asyncio
         async def test_resume_session_creates_new_if_missing(self, agent):
    ```
- **#14691: fix(acp): replay conversation history via session_update on load_session**. closed without merge
  - maintainer direction: category `merged_via`; implication `superseded`; refs `f3a4af9cf`, `#26943`, `#19139`
- **#16531: fix(acp): replay text history on session load**. closed without merge
  - maintainer direction: category `merged_via`; implication `superseded`; refs `f3a4af9cf`, `#26943`, `#17874`, `#19139`, `#26943`, `#26943`

## Fix status

- Merged to main in PR #17874 (cited above). No release data is tracked. Verify downstream.
- Fix last verified (merged) `2026-05-16`.