fix(claude-agent-sdk): dedup errored-turn assistant messages, surface RunErrorEvent (#2145)

integrations/claude-agent-sdk/python/ag_ui_claude_sdk/adapter.py

@@ -397,14 +397,37 @@ async def run(self, input_data: RunAgentInput) -> AsyncIterator[BaseEvent]:
                 ):
                     yield event
 
-            # Emit RUN_FINISHED — read THIS run's own result (keyed per-run, so a
-            # serialized peer on the same thread cannot have clobbered it). (Fix 4)
-            yield RunFinishedEvent(
-                type=EventType.RUN_FINISHED,
-                thread_id=thread_id,
-                run_id=run_id,
-                result=self._per_run_result.get(result_key, None),
-            )
+            # Terminal event — owned by run(), one per run. On a turn whose
+            # ResultMessage carried is_error (API failure, #2145) RUN_ERROR
+            # REPLACES RUN_FINISHED: AG-UI's verifyEvents rejects any event
+            # after RUN_ERROR, so the two terminals are mutually exclusive,
+            # matching the except paths below. The worker is NOT evicted here —
+            # the stream completed cleanly, so the CLI session stays reusable
+            # (unlike the exception paths).
+            run_result = self._per_run_result.get(result_key) or {}
+            if run_result.get("is_error"):
+                api_error_status = run_result.get("api_error_status")
+                logger.error(
+                    f"Run {run_id} on thread={thread_id} ended with an API error"
+                    + (f" (status {api_error_status})" if api_error_status is not None else "")
+                )
+                yield RunErrorEvent(
+                    type=EventType.RUN_ERROR,
+                    thread_id=thread_id,
+                    run_id=run_id,
+                    message=run_result.get("result") or "The run ended with an API error.",
+                    code=str(api_error_status) if api_error_status is not None else None,
+                )
+            else:
+                # Emit RUN_FINISHED — read THIS run's own result (keyed per-run,
+                # so a serialized peer on the same thread cannot have clobbered
+                # it). (Fix 4)
+                yield RunFinishedEvent(
+                    type=EventType.RUN_FINISHED,
+                    thread_id=thread_id,
+                    run_id=run_id,
+                    result=self._per_run_result.get(result_key, None),
+                )
 
         except asyncio.TimeoutError as e:
             logger.error(f"Query timeout in run for thread={thread_id}: {e}")
@@ -636,7 +659,8 @@ async def _stream_claude_sdk(
         in_reasoning_block: bool = False
         reasoning_message_id: Optional[str] = None
         has_streamed_text: bool = False
-
+        unstreamed_fallback_ids: set[str] = set()
+
         # Tool call streaming state
         current_tool_call_id: Optional[str] = None
         current_tool_call_name: Optional[str] = None
@@ -982,6 +1006,8 @@ def flush_pending_msg():
                 # streaming loop and is already None here.
                 msg_id = current_message_id or str(uuid.uuid4())
                 if isinstance(message, AssistantMessage):
+                    if current_message_id is None:
+                        unstreamed_fallback_ids.add(msg_id)
                     agui_msg = build_agui_assistant_message(message, msg_id)
                     if agui_msg:
                         upsert_message(agui_msg)
@@ -1043,20 +1069,32 @@ def flush_pending_msg():
                 is_error = getattr(message, 'is_error', None)
                 result_text = getattr(message, 'result', None)
 
-                # Capture metadata for RunFinished event. Key per-run
+                # Capture metadata for the terminal event. Key per-run
                 # (thread_id, run_id) so a serialized peer on the same thread
                 # cannot clobber this run's result. (Fix 4)
-                self._per_run_result[(thread_id, run_id)] = {
+                run_result = {
                     "is_error": is_error,
                     "duration_ms": getattr(message, 'duration_ms', None),
                     "duration_api_ms": getattr(message, 'duration_api_ms', None),
                     "num_turns": getattr(message, 'num_turns', None),
                     "total_cost_usd": getattr(message, 'total_cost_usd', None),
                     "usage": getattr(message, 'usage', None),
                     "structured_output": getattr(message, 'structured_output', None),
+                    # Best-effort: absent from ResultMessage on claude-agent-sdk
+                    # versions predating the field (including the current pin
+                    # floor), in which case RUN_ERROR.code is None.
+                    "api_error_status": getattr(message, 'api_error_status', None),
                 }
-
-                if not has_streamed_text and result_text:
+                if is_error:
+                    # Thread the failure text to run(), which owns terminal
+                    # events and emits RUN_ERROR in place of RUN_FINISHED
+                    # (#2145). Added only on errored turns so the success-path
+                    # RunFinishedEvent.result payload (emitted verbatim) is
+                    # unchanged.
+                    run_result["result"] = result_text
+                self._per_run_result[(thread_id, run_id)] = run_result
+
+                if not has_streamed_text and result_text and not is_error:
                     result_msg_id = str(uuid.uuid4())
                     yield TextMessageStartEvent(type=EventType.TEXT_MESSAGE_START, thread_id=thread_id, run_id=run_id, message_id=result_msg_id, role="assistant")
                     yield TextMessageContentEvent(type=EventType.TEXT_MESSAGE_CONTENT, thread_id=thread_id, run_id=run_id, message_id=result_msg_id, delta=result_text)
@@ -1106,6 +1144,10 @@ def flush_pending_msg():
 
         flush_pending_msg()
 
+        run_result = self._per_run_result.get((thread_id, run_id), {}) or {}
+        if run_result.get("is_error") and unstreamed_fallback_ids:
+            run_messages[:] = [m for m in run_messages if _get_msg_id(m) not in unstreamed_fallback_ids]
+
         # Emit MESSAGES_SNAPSHOT with input messages + new messages from this run
         if run_messages:
             all_messages = list(input_data.messages or []) + run_messages

integrations/claude-agent-sdk/python/tests/test_adapter.py

@@ -80,6 +80,183 @@ async def test_messages_snapshot_emitted_at_end(self, make_input):
         assert any(getattr(m, "content", None) == "Hi" for m in snapshots[0].messages)
 
 
+class TestResultMessageErrorHandling:
+    """Regression tests for ag-ui-protocol/ag-ui#2145.
+
+    An errored turn used to fold its failure text into MESSAGES_SNAPSHOT
+    twice (once via the streamed text, once via the non-streaming
+    AssistantMessage fallback minting a fresh, never-streamed id) with no
+    RUN_ERROR signal at all.
+    """
+
+    @pytest.mark.asyncio
+    async def test_errored_turn_emits_one_assistant_message(self, make_input):
+        from claude_agent_sdk import AssistantMessage, ResultMessage
+        from claude_agent_sdk.types import TextBlock
+
+        adapter = ClaudeAgentAdapter(name="t")
+        error_text = "API Error: 400 You have reached your specified API usage limits."
+        stream = [
+            # Streamed text: gets a real message id, upserted at message_stop.
+            stream_event({"type": "message_start"}),
+            stream_event(
+                {"type": "content_block_delta", "delta": {"type": "text_delta", "text": error_text}}
+            ),
+            stream_event({"type": "message_stop"}),
+            # SDK redelivers the same content as a complete AssistantMessage
+            # AFTER message_stop reset current_message_id to None — this used
+            # to mint a second, never-streamed id for identical text.
+            AssistantMessage(content=[TextBlock(text=error_text)], model="claude-x"),
+            ResultMessage(
+                subtype="error_during_execution",
+                duration_ms=1,
+                duration_api_ms=1,
+                is_error=True,
+                num_turns=1,
+                session_id="thread-1",
+                result=error_text,
+            ),
+        ]
+        events = await _drive(adapter, stream, make_input)
+
+        snapshots = [e for e in events if e.type == EventType.MESSAGES_SNAPSHOT]
+        assert len(snapshots) == 1
+        assistant_msgs = [
+            m for m in snapshots[0].messages if getattr(m, "role", None) == "assistant"
+        ]
+        assert len(assistant_msgs) == 1, (
+            f"expected exactly 1 assistant message on an errored turn, got "
+            f"{len(assistant_msgs)}"
+        )
+
+        # Terminal events are owned by run(): the stream itself must NOT emit
+        # RUN_ERROR (run() emits it in place of RUN_FINISHED).
+        assert not any(e.type == EventType.RUN_ERROR for e in events)
+        # The failure text is threaded to run() via the per-run result slot
+        # (_drive never pops it, unlike run()'s finally).
+        stored = adapter._per_run_result[("thread-1", "run-1")]
+        assert stored["is_error"] is True
+        assert stored["result"] == error_text
+
+    @pytest.mark.asyncio
+    async def test_successful_turn_unaffected(self, make_input):
+        """The is_error gating must not touch normal, non-errored turns."""
+        from claude_agent_sdk import ResultMessage
+
+        adapter = ClaudeAgentAdapter(name="t")
+        stream = [
+            stream_event({"type": "message_start"}),
+            stream_event(
+                {"type": "content_block_delta", "delta": {"type": "text_delta", "text": "Hi there"}}
+            ),
+            stream_event({"type": "message_stop"}),
+            ResultMessage(
+                subtype="success",
+                duration_ms=1,
+                duration_api_ms=1,
+                is_error=False,
+                num_turns=1,
+                session_id="thread-1",
+                result="Hi there",
+            ),
+        ]
+        events = await _drive(adapter, stream, make_input)
+
+        snapshots = [e for e in events if e.type == EventType.MESSAGES_SNAPSHOT]
+        assert len(snapshots) == 1
+        assistant_msgs = [
+            m for m in snapshots[0].messages if getattr(m, "role", None) == "assistant"
+        ]
+        assert len(assistant_msgs) == 1
+        assert assistant_msgs[0].content == "Hi there"
+        assert not any(e.type == EventType.RUN_ERROR for e in events)
+        # The failure-text slot is error-only: a success turn must not grow a
+        # "result" key, or it would leak into RunFinishedEvent.result.
+        assert "result" not in adapter._per_run_result[("thread-1", "run-1")]
+
+    @pytest.mark.asyncio
+    async def test_run_replaces_run_finished_with_run_error_on_api_error(
+        self, make_input, monkeypatch
+    ):
+        """Terminal events are owned by run(): an errored turn must end in
+        exactly one RUN_ERROR *in place of* RUN_FINISHED (verifyEvents rejects
+        anything after RUN_ERROR), mirroring TestRunErrorPath."""
+        from claude_agent_sdk import AssistantMessage, ResultMessage
+        from claude_agent_sdk.types import TextBlock
+
+        error_text = "API Error: 400 You have reached your specified API usage limits."
+        result_msg = ResultMessage(
+            subtype="error_during_execution",
+            duration_ms=1,
+            duration_api_ms=1,
+            is_error=True,
+            num_turns=1,
+            session_id="thread-1",
+            result=error_text,
+        )
+        # Newer SDKs add api_error_status to ResultMessage; the installed
+        # version predates it, so attach dynamically (plain dataclass, no
+        # slots) to exercise the best-effort code threading.
+        result_msg.api_error_status = 400
+
+        stream = [
+            stream_event({"type": "message_start"}),
+            stream_event(
+                {"type": "content_block_delta", "delta": {"type": "text_delta", "text": error_text}}
+            ),
+            stream_event({"type": "message_stop"}),
+            AssistantMessage(content=[TextBlock(text=error_text)], model="claude-x"),
+            result_msg,
+        ]
+
+        class _FakeStreamingWorker:
+            """SessionWorker stand-in that streams the canned errored turn."""
+
+            def __init__(self, *args, **kwargs):
+                pass
+
+            async def start(self):
+                pass
+
+            def is_alive(self):
+                return True
+
+            def query(self, prompt, session_id="default"):
+                async def _gen():
+                    for m in stream:
+                        yield m
+
+                return _gen()
+
+            async def stop(self):
+                pass
+
+        adapter = ClaudeAgentAdapter(name="t")
+        monkeypatch.setattr("ag_ui_claude_sdk.adapter.SessionWorker", _FakeStreamingWorker)
+        inp = make_input(messages=[{"id": "1", "role": "user", "content": "hi"}])
+        events = [e async for e in adapter.run(inp)]
+        types = _types(events)
+
+        assert EventType.RUN_STARTED in types
+        assert types.count(EventType.RUN_ERROR) == 1
+        assert EventType.RUN_FINISHED not in types
+        assert types[-1] == EventType.RUN_ERROR  # nothing may follow RUN_ERROR
+        err = events[-1]
+        assert err.message == error_text
+        assert err.code == "400"
+
+        snapshots = [e for e in events if e.type == EventType.MESSAGES_SNAPSHOT]
+        assert len(snapshots) == 1
+        assistant_msgs = [
+            m for m in snapshots[0].messages if getattr(m, "role", None) == "assistant"
+        ]
+        assert len(assistant_msgs) == 1
+
+        # The stream completed cleanly (unlike the exception paths), so the
+        # healthy worker/session must NOT be evicted.
+        assert "thread-1" in adapter._workers
+
+
 class TestStreamToolCall:
     @pytest.mark.asyncio
     async def test_backend_tool_call_sequence(self, make_input):