fix(acp): honour the executor's is_error when closing a tool call

tool.completed carries is_error, but the ACP bridge dropped it and
re-derived status from the result text alone, so a tool cancelled by a
user interrupt (plain-text result) or one returning an error dict closed
as completed. Pass the flag through close_tool_call and OR it into
build_tool_complete's failed predicate; the text heuristic stays as the
fallback for the step-closer path.

acp_adapter/events.py

@@ -77,7 +77,7 @@ def _upgrade_queue(tool_call_ids: Dict[str, Deque[str]], name: str) -> Deque[str
 
 def close_tool_call(
     conn: acp.Client, session_id: str, loop: asyncio.AbstractEventLoop, tool_call_ids: Dict[str, Deque[str]],
-    tool_call_meta: Dict[str, Dict[str, Any]], name: str, result: Any = None,
+    tool_call_meta: Dict[str, Dict[str, Any]], name: str, result: Any = None, is_error: bool = False,
 ) -> str | None:
     """Close the oldest open ACP tool call for ``name``; returns its id, or None when none is open."""
     queue = _upgrade_queue(tool_call_ids, name)
@@ -87,7 +87,7 @@ def close_tool_call(
     meta = tool_call_meta.pop(tc_id, {})
     _send_update(conn, session_id, loop, build_tool_complete(
         tc_id, name, result=str(result) if result is not None else None,
-        function_args=meta.get("args"), snapshot=meta.get("snapshot"),
+        function_args=meta.get("args"), snapshot=meta.get("snapshot"), is_error=is_error,
     ))
     if not queue:
         tool_call_ids.pop(name, None)
@@ -134,7 +134,11 @@ def make_tool_progress_cb(
         if event_type == "tool.completed" and name:
             if turn_state is not None:
                 turn_state["saw_completion"] = True
-            close_tool_call(conn, session_id, loop, tool_call_ids, tool_call_meta, name, kwargs.get("result"))
+            # The executor's verdict: a cancelled/errored tool may return plain text the heuristic misses.
+            close_tool_call(
+                conn, session_id, loop, tool_call_ids, tool_call_meta, name, kwargs.get("result"),
+                is_error=bool(kwargs.get("is_error")),
+            )
             return
         if event_type != "tool.started":
             return

acp_adapter/tools.py

@@ -817,9 +817,11 @@ def _build_tool_start(tool_call_id: str, tool_name: str, arguments: Args, *, edi
 
 def build_tool_complete(
     tool_call_id: str, tool_name: str, result: Optional[str] = None, function_args: Optional[Args] = None,
-    snapshot: Any = None,
+    snapshot: Any = None, is_error: bool = False,
 ) -> ToolCallProgress:
-    """Create a ToolCallUpdate (progress) event for a completed tool call."""
+    """Create a ToolCallUpdate (progress) event for a completed tool call.
+
+    ``is_error`` is the executor's own verdict; the result-text heuristic stays as fallback."""
     if tool_name == "web_extract":  # errors only; success stays compact via the title
         error_text = _format_web_extract_result(tool_name, result, function_args)
         content = [_text(error_text)] if error_text else None
@@ -828,7 +830,7 @@ def build_tool_complete(
     structured = isinstance(_json_loads_maybe(result), (dict, list))
     return acp.update_tool_call(
         tool_call_id, kind=get_tool_kind(tool_name),
-        status="failed" if _tool_result_failed(result, tool_name) else "completed", content=content,
+        status="failed" if is_error or _tool_result_failed(result, tool_name) else "completed", content=content,
         raw_output=None if tool_name in _POLISHED_TOOLS or structured else result,
     )
 

tests/acp_adapter/test_events.py

@@ -332,7 +332,9 @@ class TestToolCallsAlwaysReachATerminalStatus:
             rcts.return_value = MagicMock(spec=Future)
             progress("tool.completed", "read", None, None, result="file body")
             step(2, [{"name": "read", "result": "file body", "arguments": '{"path": "a"}'}])
-        btc.assert_called_once_with("tc-1", "read", result="file body", function_args={"path": "a"}, snapshot=None)
+        btc.assert_called_once_with(
+            "tc-1", "read", result="file body", function_args={"path": "a"}, snapshot=None, is_error=False,
+        )
         assert list(ids["read"]) == ["tc-2"] and "tc-1" not in meta
 
     def test_step_fallback_coerces_wire_arguments_and_turn_end_flush_fails_what_is_still_open(
@@ -357,3 +359,16 @@ class TestToolCallsAlwaysReachATerminalStatus:
         statuses = [c.args[1].status for c in mock_conn.session_update.call_args_list]
         assert statuses == ["completed", "failed"]
         assert ids == {} and meta == {}
+
+    def test_tool_completed_is_error_flag_closes_the_call_as_failed(self, mock_conn, event_loop_fixture):
+        """``tool.completed`` carries the executor's ``is_error``; a cancelled tool's plain-text
+        result trips no heuristic, so dropping the flag showed an interrupted call green."""
+        from collections import deque
+
+        ids = {"terminal": deque(["tc-1"])}
+        progress = make_tool_progress_cb(mock_conn, "s", event_loop_fixture, ids, {})
+        with self._patch() as rcts:
+            rcts.return_value = MagicMock(spec=Future)
+            progress("tool.completed", "terminal", None, None, is_error=True,
+                     result="[Tool execution cancelled — terminal was skipped due to user interrupt]")
+        assert [c.args[1].status for c in mock_conn.session_update.call_args_list] == ["failed"]