fix(acp): handle stale Zed responses

coding-agent/src/coding_agent/acp/server.py

@@ -459,6 +459,7 @@ async def _session_prompt(self, params: JSONObject) -> JSONObject:
                         await _await_cancelled_prompt_task(task)
                         return {"stopReason": "cancelled"}
                     await task
+                    await self._drain_pending_session_updates(session_id, session)
                     return {"stopReason": acp_stop_reason(message)}
 
                 if isinstance(message, ApprovalRequest):
@@ -507,6 +508,36 @@ def _consume_prompt_cancel(self, session_id: str) -> bool:
         self._cancelled_prompt_sessions.discard(session_id)
         return True
 
+    async def _drain_pending_session_updates(
+        self,
+        session_id: str,
+        session: object,
+    ) -> None:
+        wire = getattr(session, "wire", None)
+        consume_outgoing = getattr(wire, "consume_outgoing", None)
+        if not callable(consume_outgoing):
+            return
+
+        while True:
+            message = consume_outgoing()
+            if message is None:
+                return
+            if isinstance(message, ApprovalRequest):
+                continue
+            update = wire_message_to_session_update(message)
+            if update is None:
+                continue
+            await self._emit_notification(
+                {
+                    "jsonrpc": "2.0",
+                    "method": "session/update",
+                    "params": {
+                        "sessionId": session_id,
+                        "update": update,
+                    },
+                }
+            )
+
     async def handle_approval_request(
         self,
         session_id: str,
@@ -534,10 +565,7 @@ async def handle_approval_request(
             scope=scope,
         )
         if submitted is None:
-            raise JsonRpcError(
-                -32603,
-                f"Approval request not found: {request.request_id}",
-            )
+            return
 
     async def _emit_notification(self, message: JSONObject) -> None:
         if self._emit is None:

coding-agent/src/coding_agent/tools/shell.py

@@ -309,8 +309,8 @@ def bash_run(
     env: dict[str, str] | None = None,
     __pipeline_ctx__: object | None = None,
 ) -> str | dict[str, str | int]:
+    timeout_seconds = _positive_int(timeout, name="timeout")
     try:
-        timeout_seconds = _positive_int(timeout, name="timeout")
         shell_config = _pipeline_shell_config(__pipeline_ctx__)
         mode_value = _sandbox_mode(shell_config)
         execution_cwd = _validated_execution_cwd(

coding-agent/tests/acp/test_server.py

@@ -1265,6 +1265,63 @@ async def test_session_prompt_streams_agent_message_chunk_and_returns_end_turn()
     assert response == {"jsonrpc": "2.0", "id": 3, "result": {"stopReason": "end_turn"}}
 
 
+@pytest.mark.asyncio
+async def test_session_prompt_drains_error_chunks_before_error_response() -> None:
+    class ErrorFirstManager(FakeManager):
+        async def run_agent(self, session_id: str, prompt: str) -> None:
+            self.calls.append(
+                ("run_agent", {"session_id": session_id, "prompt": prompt})
+            )
+            await self.wire.send(
+                TurnEnd(
+                    session_id=session_id,
+                    turn_id="turn-1",
+                    completion_status=CompletionStatus.ERROR,
+                )
+            )
+            await self.wire.send(
+                StreamDelta(session_id=session_id, content="Error: provider failed")
+            )
+            await self.wire.send(
+                TurnEnd(
+                    session_id=session_id,
+                    turn_id="turn-1",
+                    completion_status=CompletionStatus.ERROR,
+                )
+            )
+
+    manager = ErrorFirstManager()
+    notifications: list[dict[str, Any]] = []
+    server = AcpServer(manager, emit=notifications.append)
+
+    response = await server.handle_message(
+        {
+            "jsonrpc": "2.0",
+            "id": 13,
+            "method": "session/prompt",
+            "params": {
+                "sessionId": "sess-1",
+                "prompt": [{"type": "text", "text": "hello"}],
+            },
+        }
+    )
+
+    assert notifications == [
+        {
+            "jsonrpc": "2.0",
+            "method": "session/update",
+            "params": {
+                "sessionId": "sess-1",
+                "update": {
+                    "sessionUpdate": "agent_message_chunk",
+                    "content": {"type": "text", "text": "Error: provider failed"},
+                },
+            },
+        }
+    ]
+    assert response == {"jsonrpc": "2.0", "id": 13, "result": {"stopReason": "refusal"}}
+
+
 @pytest.mark.asyncio
 async def test_session_prompt_rejects_active_turn() -> None:
     manager = FakeManager()
@@ -1469,6 +1526,64 @@ async def call_client(_method: str, _params: dict[str, Any]) -> dict[str, Any]:
     ) in manager.calls
 
 
+@pytest.mark.asyncio
+async def test_permission_request_ignores_stale_approval_response() -> None:
+    class StaleApprovalManager(FakeManager):
+        async def submit_approval_response(
+            self,
+            session_id: str,
+            request_id: str,
+            approved: bool,
+            feedback: str | None = None,
+            scope: str = "once",
+        ) -> Any:
+            self.calls.append(
+                (
+                    "submit_approval_response",
+                    {
+                        "session_id": session_id,
+                        "request_id": request_id,
+                        "approved": approved,
+                        "feedback": feedback,
+                        "scope": scope,
+                    },
+                )
+            )
+            return None
+
+    manager = StaleApprovalManager()
+
+    async def call_client(_method: str, _params: dict[str, Any]) -> dict[str, Any]:
+        return {"outcome": {"outcome": "selected", "optionId": "allow-once"}}
+
+    server = AcpServer(manager, call_client=call_client)
+
+    await server.handle_approval_request(
+        "sess-1",
+        ApprovalRequest(
+            session_id="sess-1",
+            request_id="approval-stale",
+            tool_call=ToolCallDelta(
+                session_id="sess-1",
+                tool_name="bash_run",
+                arguments={},
+                call_id="tool-1",
+            ),
+        ),
+    )
+
+    assert (
+        "submit_approval_response",
+        {
+            "session_id": "sess-1",
+            "request_id": "approval-stale",
+            "approved": True,
+            "feedback": None,
+            "scope": "once",
+        },
+    ) in manager.calls
+
+
 @pytest.mark.asyncio
 async def test_stdio_server_writes_jsonrpc_to_stdout_only() -> None:
     manager = FakeManager()

coding-agent/tests/coding_agent/tools/test_shell.py

@@ -19,3 +19,7 @@ def test_rejects_shell_metacharacters(self):
         result = bash_run(command="echo hello && rm -rf /tmp/nope")
         assert isinstance(result, str)
         assert "unsupported shell syntax" in result.lower()
+
+    def test_invalid_timeout_raises_tool_validation_error(self):
+        with pytest.raises(ValueError, match="timeout must be a positive integer"):
+            bash_run(command="echo hello", timeout="10s")  # type: ignore[arg-type]