fix(acp): compact Zed tool replay rendering

acp_adapter/server.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import asyncio
 import contextvars
+import json
 import logging
 import os
 from collections import defaultdict, deque
@@ -47,6 +48,7 @@ from acp.schema import (
     TextContentBlock,
     UnstructuredCommandInput,
     Usage,
+    UsageUpdate,
     UserMessageChunk,
 )
 
@@ -65,6 +67,7 @@ from acp_adapter.events import (
 )
 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__)
 
@@ -315,6 +318,66 @@ class HermesACPAgent(acp.Agent):
 
         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,
@@ -485,37 +548,99 @@ class HermesACPAgent(acp.Agent):
             )
         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,
@@ -527,11 +652,24 @@ class HermesACPAgent(acp.Agent):
         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,
@@ -545,8 +683,9 @@ class HermesACPAgent(acp.Agent):
             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(
@@ -562,8 +701,9 @@ class HermesACPAgent(acp.Agent):
             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:
@@ -712,6 +852,7 @@ class HermesACPAgent(acp.Agent):
                 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
@@ -744,24 +885,37 @@ class HermesACPAgent(acp.Agent):
         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
@@ -867,7 +1021,7 @@ class HermesACPAgent(acp.Agent):
                 )
             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)
 
@@ -903,6 +1057,8 @@ class HermesACPAgent(acp.Agent):
                 cached_read_tokens=result.get("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)
 
@@ -1035,22 +1191,84 @@ class HermesACPAgent(acp.Agent):
             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 = int(getattr(compressor, "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 = estimate_request_tokens_rough(
+                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 = int(context_length * 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

@@ -28,6 +28,11 @@ TOOL_KIND_MAP: Dict[str, ToolKind] = {
     "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",
@@ -51,6 +56,28 @@ TOOL_KIND_MAP: Dict[str, ToolKind] = {
 }
 
 
+_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")
@@ -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:
@@ -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)]
 
 
 # ---------------------------------------------------------------------------
@@ -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":
@@ -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
@@ -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,
     )
 
 
@@ -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

@@ -178,9 +178,10 @@ class TestMcpRegistrationE2E:
         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

@@ -27,7 +27,10 @@ from acp.schema import (
     SetSessionModeResponse,
     SessionInfo,
     TextContentBlock,
+    ToolCallProgress,
+    ToolCallStart,
     Usage,
+    UsageUpdate,
     UserMessageChunk,
 )
 from acp_adapter.server import HermesACPAgent, HERMES_VERSION
@@ -200,6 +203,8 @@ class TestSessionOps:
             "context",
             "reset",
             "compact",
+            "steer",
+            "queue",
             "version",
         ]
         model_cmd = next(
@@ -208,6 +213,46 @@ class TestSessionOps:
         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=".")
@@ -238,11 +283,31 @@ class TestSessionOps:
             {"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
@@ -257,6 +322,21 @@ class TestSessionOps:
         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)
@@ -269,6 +349,8 @@ class TestSessionOps:
 
         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]
@@ -278,6 +360,27 @@ class TestSessionOps:
             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")
@@ -522,6 +625,11 @@ class TestPrompt:
         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):
@@ -565,12 +673,40 @@ class TestPrompt:
         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):
@@ -708,6 +844,43 @@ class TestSlashCommands:
         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=80_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,
+        ):
+            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=80_000,
+        )
+
+        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"}]
@@ -787,7 +960,12 @@ class TestSlashCommands:
         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

@@ -52,6 +52,12 @@ class TestToolKindMap:
     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"
 
@@ -110,6 +116,25 @@ class TestBuildToolTitle:
         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"
@@ -164,15 +189,23 @@ class TestBuildToolStart:
         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."""
@@ -181,6 +214,48 @@ class TestBuildToolStart:
         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."""
@@ -205,6 +280,158 @@ class TestBuildToolComplete:
         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."""