fix(acp): replay native todo plans

acp_adapter/events.py

@@ -25,6 +25,17 @@ from .tools import (
 logger = logging.getLogger(__name__)
 
 
+def _json_loads_maybe_prefix(value: str) -> Any:
+    """Parse a JSON object even when Hermes appended a human hint after it."""
+    text = value.strip()
+    try:
+        return json.loads(text)
+    except Exception:
+        decoder = json.JSONDecoder()
+        data, _ = decoder.raw_decode(text)
+        return data
+
+
 def _build_plan_update_from_todo_result(result: Any) -> AgentPlanUpdate | None:
     """Translate Hermes' todo tool result into ACP's native plan update.
 
@@ -37,13 +48,17 @@ def _build_plan_update_from_todo_result(result: Any) -> AgentPlanUpdate | None:
         return None
 
     try:
-        data = json.loads(result)
+        data = _json_loads_maybe_prefix(result)
     except Exception:
         return None
 
     if not isinstance(data, dict) or not isinstance(data.get("todos"), list):
         return None
 
+    todos = data["todos"]
+    if not todos:
+        return AgentPlanUpdate(session_update="plan", entries=[])
+
     status_map = {
         "pending": "pending",
         "in_progress": "in_progress",
@@ -54,7 +69,7 @@ def _build_plan_update_from_todo_result(result: Any) -> AgentPlanUpdate | None:
         "cancelled": "completed",
     }
     entries: list[PlanEntry] = []
-    for item in data["todos"]:
+    for item in todos:
         if not isinstance(item, dict):
             continue
         content = str(item.get("content") or item.get("id") or "").strip()
@@ -66,8 +81,6 @@ def _build_plan_update_from_todo_result(result: Any) -> AgentPlanUpdate | None:
             content = f"[cancelled] {content}"
         entries.append(PlanEntry(content=content, priority="medium", status=status))
 
-    if not entries:
-        return None
     return AgentPlanUpdate(session_update="plan", entries=entries)
 
 

acp_adapter/server.py

@@ -59,6 +59,7 @@ from acp.schema import (
 
 from acp_adapter.auth import TERMINAL_SETUP_AUTH_METHOD_ID, build_auth_methods, detect_provider
 from acp_adapter.events import (
+    _build_plan_update_from_todo_result,
     make_message_cb,
     make_step_cb,
     make_thinking_cb,
@@ -910,15 +911,20 @@ class HermesACPAgent(acp.Agent):
                 if not tool_call_id or not tool_name:
                     continue
                 result = message.get("content")
+                result_text = result if isinstance(result, str) else None
                 if not await _send(
                     build_tool_complete(
                         tool_call_id,
                         tool_name,
-                        result=result if isinstance(result, str) else None,
+                        result=result_text,
                         function_args=function_args,
                     )
                 ):
                     return
+                if tool_name == "todo":
+                    plan_update = _build_plan_update_from_todo_result(result_text)
+                    if plan_update is not None and not await _send(plan_update):
+                        return
 
     async def new_session(
         self,

tests/acp/test_events.py

@@ -12,6 +12,7 @@ import acp
 from acp.schema import AgentPlanUpdate, ToolCallStart, ToolCallProgress, AgentThoughtChunk, AgentMessageChunk
 
 from acp_adapter.events import (
+    _build_plan_update_from_todo_result,
     _send_update,
     make_message_cb,
     make_step_cb,
@@ -296,7 +297,7 @@ class TestStepCallback:
         }
         mock_send.assert_called_once()
 
-    def test_todo_completion_emits_native_plan_update(self, mock_conn, event_loop_fixture):
+    def test_todo_completion_emits_native_plan_update_after_tool_completion(self, mock_conn, event_loop_fixture):
         from collections import deque
 
         tool_call_ids = {"todo": deque(["tc-todo"])}
@@ -314,9 +315,11 @@ class TestStepCallback:
             cb(1, [{"name": "todo", "result": todo_result}])
 
         updates = [call.args[3] for call in mock_send.call_args_list]
-        plan_updates = [u for u in updates if getattr(u, "session_update", None) == "plan"]
+        assert [getattr(update, "session_update", None) for update in updates] == [
-        assert len(plan_updates) == 1
+            "tool_call_update",
-        plan = plan_updates[0]
+            "plan",
+        ]
+        plan = updates[1]
         assert isinstance(plan, AgentPlanUpdate)
         assert [entry.content for entry in plan.entries] == [
             "Inspect ACP",
@@ -326,6 +329,22 @@ class TestStepCallback:
         assert [entry.status for entry in plan.entries] == ["completed", "in_progress", "completed"]
         assert [entry.priority for entry in plan.entries] == ["medium", "medium", "medium"]
 
+    def test_todo_plan_update_parses_json_with_trailing_hint(self):
+        result = '{"todos":[{"id":"ship","content":"Ship ACP plan","status":"pending"}]}\n\n[Hint: persisted]'
+
+        update = _build_plan_update_from_todo_result(result)
+
+        assert isinstance(update, AgentPlanUpdate)
+        assert [entry.content for entry in update.entries] == ["Ship ACP plan"]
+        assert [entry.status for entry in update.entries] == ["pending"]
+
+    def test_todo_plan_update_with_empty_todos_clears_plan(self):
+        update = _build_plan_update_from_todo_result('{"todos":[],"summary":{"total":0}}')
+
+        assert isinstance(update, AgentPlanUpdate)
+        assert update.session_update == "plan"
+        assert update.entries == []
+
 
 # ---------------------------------------------------------------------------
 # Message callback

tests/acp/test_server.py

@@ -12,6 +12,7 @@ from acp.agent.router import build_agent_router
 from acp.schema import (
     AgentCapabilities,
     AgentMessageChunk,
+    AgentPlanUpdate,
     AuthenticateResponse,
     AvailableCommandsUpdate,
     Implementation,
@@ -391,6 +392,57 @@ class TestSessionOps:
         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_load_session_replays_native_plan_for_persisted_todo_tool(self, agent):
+        """Persisted todo tool results should rebuild Zed's native plan panel."""
+        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",
+                "content": "",
+                "tool_calls": [
+                    {
+                        "id": "call_todo_1",
+                        "type": "function",
+                        "function": {
+                            "name": "todo",
+                            "arguments": '{"todos":[{"id":"ship","content":"Ship it","status":"in_progress"}]}',
+                        },
+                    }
+                ],
+            },
+            {
+                "role": "tool",
+                "tool_call_id": "call_todo_1",
+                "content": '{"todos":[{"id":"ship","content":"Ship it","status":"in_progress"}]}',
+            },
+        ]
+
+        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)
+        relevant_updates = [
+            update for update in (call.kwargs["update"] for call in mock_conn.session_update.await_args_list)
+            if getattr(update, "session_update", None) in {"tool_call", "tool_call_update", "plan"}
+        ]
+        assert [getattr(update, "session_update", None) for update in relevant_updates] == [
+            "tool_call",
+            "tool_call_update",
+            "plan",
+        ]
+        plan = relevant_updates[2]
+        assert isinstance(plan, AgentPlanUpdate)
+        assert [entry.content for entry in plan.entries] == ["Ship it"]
+        assert [entry.status for entry in plan.entries] == ["in_progress"]
+
     @pytest.mark.asyncio
     async def test_resume_session_replays_persisted_history_to_client(self, agent):
         mock_conn = MagicMock(spec=acp.Client)