Schedule ACP history replay and fence file output

acp_adapter/server.py

@@ -658,6 +658,18 @@ class HermesACPAgent(acp.Agent):
             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,
@@ -671,7 +683,7 @@ 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))
@@ -689,7 +701,7 @@ 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))

acp_adapter/tools.py

@@ -208,6 +208,13 @@ def _truncate_text(text: str, limit: int = 5000) -> str:
     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):
@@ -261,7 +268,10 @@ def _format_read_file_result(result: Optional[str], args: Optional[Dict[str, Any
     header = f"Read {path}{suffix}"
     if data.get("total_lines") is not None:
         header += f" — {data.get('total_lines')} total lines"
-    return _truncate_text(f"{header}\n\n{content}")
+    # 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]:

tests/acp/test_server.py

@@ -306,6 +306,8 @@ class TestSessionOps:
 
         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
@@ -347,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]
@@ -356,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")

tests/acp/test_tools.py

@@ -344,7 +344,7 @@ class TestBuildToolComplete:
         )
         text = result.content[0].content.text
         assert "Read README.md" in text
-        assert "1|hello" 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):