From 88f61d42cacdf7cca9f14b21e539da27f30a5e11 Mon Sep 17 00:00:00 2001 From: kina0630 Date: Fri, 5 Jun 2026 09:53:04 +0800 Subject: [PATCH] fix(acp): handle stale zed responses --- coding-agent/src/coding_agent/acp/server.py | 36 +++++- coding-agent/src/coding_agent/tools/shell.py | 2 +- coding-agent/tests/acp/test_server.py | 115 ++++++++++++++++++ .../tests/coding_agent/tools/test_shell.py | 4 + 4 files changed, 152 insertions(+), 5 deletions(-) diff --git a/coding-agent/src/coding_agent/acp/server.py b/coding-agent/src/coding_agent/acp/server.py index c16c0863c..557222e4f 100644 --- a/coding-agent/src/coding_agent/acp/server.py +++ b/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: diff --git a/coding-agent/src/coding_agent/tools/shell.py b/coding-agent/src/coding_agent/tools/shell.py index 90d47288f..4c5f28997 100644 --- a/coding-agent/src/coding_agent/tools/shell.py +++ b/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( diff --git a/coding-agent/tests/acp/test_server.py b/coding-agent/tests/acp/test_server.py index 0960a7842..4dea9bb63 100644 --- a/coding-agent/tests/acp/test_server.py +++ b/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() diff --git a/coding-agent/tests/coding_agent/tools/test_shell.py b/coding-agent/tests/coding_agent/tools/test_shell.py index 806dfe286..79f83c03e 100644 --- a/coding-agent/tests/coding_agent/tools/test_shell.py +++ b/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]