fix(claude-agent-sdk): dedup errored-turn assistant messages, surface RunErrorEvent (#2145)#2152
Conversation
… RunErrorEvent Fixes ag-ui-protocol#2145. ResultMessage.is_error now gates the non-streaming fallback and drops ids minted by the AssistantMessage fallback before the snapshot, then emits RunErrorEvent -- reusing the same pattern already used on the three exception paths.
BenTaylorDev
left a comment
BenTaylorDev
left a comment
There was a problem hiding this comment.
The dedup half of this is in good shape — the not is_error gate plus the unstreamed_fallback_ids drop collapse the errored turn to a single assistant message, _get_msg_id handles both dict/object rows, and both duplicate sources in _stream_claude_sdk are covered. The one thing holding it back is the RunErrorEvent emission point, which I think we can fix with a small move.
🔴 Blocker — errored turns emit two terminal events (RUN_ERROR then RUN_FINISHED).
The new RunErrorEvent is yielded from inside _stream_claude_sdk (adapter.py ~L1127), and the generator then returns normally. Back in run(), the async for … yield event loop completes without an exception, so control falls through to the unconditional yield RunFinishedEvent at L402. On an errored turn that produces:
RUN_STARTED → TEXT_* → MESSAGES_SNAPSHOT → RUN_ERROR → RUN_FINISHED.
AG-UI's verifyEvents rejects anything after RUN_ERROR ("Cannot send event type 'RUN_FINISHED': The run has already errored…"), so verifying consumers (incl. CopilotKit) throw on every errored turn.
This file already encodes the opposite invariant for the exception paths: the three existing RunErrorEvents live in run()'s except blocks (mutually exclusive with RUN_FINISHED), and TestRunErrorPath::test_run_emits_run_error_on_worker_failure asserts RUN_ERROR in types and RUN_FINISHED not in types.
The smallest fix that keeps your dedup work intact: don't emit the terminal from the stream. You're already recording is_error in _per_run_result — so let run() read that after the stream drains and emit RUN_ERROR in place of RUN_FINISHED, matching the existing except-path pattern. That keeps terminal ownership in one function and reuses the contract the codebase already tests for.
🟠 The current tests can't catch this one. _drive() (test_adapter.py L28-35) drives _stream_claude_sdk directly and bypasses run(), so RUN_FINISHED never appears in the captured events, and test_errored_turn_emits_one_assistant_message_and_run_error asserts one RUN_ERROR without asserting RUN_FINISHED is absent. Once the emission moves into run(), a regression test that goes through adapter.run() (mirroring TestRunErrorPath) and asserts RUN_FINISHED absent on an errored turn would lock the ordering.
🟡 Minor — code/api_error_status pin floor. Your read on the field is right: ResultMessage.api_error_status is present in current SDKs (confirmed on 0.2.116). It just isn't there at the declared floor >=0.1.12, so on that older end getattr(…, None) quietly yields code=None. No crash either way — just worth bumping the pin floor to the version that introduced it (or a quick note that code is best-effort) so the signal isn't silently inert for older-pinned installs.
🟡 Nit — at L1131 result_text relies on being bound in the ResultMessage branch (L1044); safe today but worth initializing result_text = None at the top of the function.
Happy to look again as soon as the terminal emission moves into run().
Python Preview PackagesVersion
Install with uvAdd the TestPyPI index to your [[tool.uv.index]]
name = "testpypi"
url = "https://test.pypi.org/simple/"
explicit = trueThen install the packages you need: # Core SDK
uv add 'ag-ui-protocol==0.0.0.dev1784111109' --index testpypi
# Integrations (each already depends on the matching ag-ui-protocol preview)
uv add 'ag-ui-langgraph==0.0.0.dev1784111109' --index testpypi
uv add 'ag-ui-crewai==0.0.0.dev1784111109' --index testpypi
# NOTE: ag-ui-agent-spec depends on pyagentspec (git-only, not on PyPI).
# You will need to install pyagentspec separately from its git repo.
uv add 'ag-ui-agent-spec==0.0.0.dev1784111109' --index testpypi
uv add 'ag_ui_adk==0.0.0.dev1784111109' --index testpypi
uv add 'ag_ui_strands==0.0.0.dev1784111109' --index testpypiInstall with pippip install \
--index-url https://test.pypi.org/simple/ \
--extra-index-url https://pypi.org/simple/ \
ag-ui-protocol==0.0.0.dev1784111109
Commit: 6856d80 |
@ag-ui/a2a-middleware
@ag-ui/a2ui-middleware
@ag-ui/event-throttle-middleware
@ag-ui/mcp-apps-middleware
@ag-ui/mcp-middleware
@ag-ui/a2a
@ag-ui/adk
@ag-ui/ag2
@ag-ui/agno
@ag-ui/aws-strands
@ag-ui/claude-agent-sdk
@ag-ui/crewai
@ag-ui/langchain
@ag-ui/langgraph
@ag-ui/llamaindex
@ag-ui/mastra
@ag-ui/pydantic-ai
@ag-ui/vercel-ai-sdk
@ag-ui/watsonx
@ag-ui/a2ui-toolkit
create-ag-ui-app
@ag-ui/client
@ag-ui/core
@ag-ui/encoder
@ag-ui/proto
commit: |
…SHED on errored turns Review follow-up: yielding RunErrorEvent from _stream_claude_sdk let run() fall through to its unconditional RunFinishedEvent, producing two terminal events per errored run — verifyEvents rejects anything after RUN_ERROR. Terminal ownership now lives in run(): after the stream drains it reads the per-run result slot and emits RUN_ERROR instead of RUN_FINISHED, matching the existing except-path contract. The failure text is threaded through _per_run_result only on errored turns, so the success-path RunFinishedEvent.result payload is unchanged. The worker is not evicted on this path — the stream completed cleanly, unlike the exception paths. New run()-level regression test (mirroring TestRunErrorPath) locks in RUN_ERROR-as-final-event and RUN_FINISHED absence; api_error_status is documented as best-effort (None on claude-agent-sdk versions predating the field). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
|
Thanks for the thorough review — you were right on all four points. Fixed in
Full suite: 109 passed. |
BenTaylorDev
left a comment
BenTaylorDev
left a comment
There was a problem hiding this comment.
Thanks for turning this around so cleanly — you addressed all four points exactly as suggested, and the extra rigor is appreciated. Verified against the diff:
- Terminal emission now lives in
run(): readsis_erroroff the per-run result slot after the stream drains and emitsRUN_ERRORin place ofRUN_FINISHED. Terminal ownership is back in one function alongside the except paths, and the stream no longer yields a terminal. 👍 test_run_replaces_run_finished_with_run_error_on_api_errorgoes throughadapter.run()and locks the contract: exactly oneRUN_ERROR,RUN_FINISHEDabsent,RUN_ERRORlast. That's the regression guard the old_drive-based tests couldn't provide.api_error_statusbest-effort comment +setattrinjection in the test is a fine resolution — thecode=Nonedegradation on older-pinned SDKs is now documented rather than silently inert.result_textnit is genuinely moot now that the stream's terminal block is gone.
Two things I especially liked: not evicting the worker on this path (correct — the stream completed cleanly, unlike the exception paths) with an assertion pinning it, and test_successful_turn_unaffected guaranteeing the success-path RunFinishedEvent.result payload stays untouched.
LGTM — approving. 🚀
…ate-error-messages # Conflicts: # integrations/claude-agent-sdk/python/ag_ui_claude_sdk/adapter.py
2 participants
Fixes #2145.
Changes
ResultMessagefallback withnot is_errorAssistantMessagefallback(
unstreamed_fallback_ids) and drop them from the snapshot when theturn errored
RunErrorEventfromrun()in place ofRUN_FINISHEDonerrored turns — terminal ownership stays in one function, matching the
pattern already used on the three exception paths (review follow-up in
a0019cec; the stream itself no longer emits terminals)Why combined
The first two dedup the duplicate text; the third makes the errored turn
identifiable at all instead of leaving one raw-text assistant message
indistinguishable from a normal reply. Sending together since dedup alone
still leaves the original bug's second symptom ("no indication anything
went wrong") unresolved -- happy to split into two PRs if you'd rather
review them separately.
Testing
test_adapter.py(TestResultMessageErrorHandling);confirmed they fail on the pre-fix code
test_run_replaces_run_finished_with_run_error_on_api_errorgoes through
adapter.run()(mirroringTestRunErrorPath) and assertsexactly one
RUN_ERROR,RUN_FINISHEDabsent,RUN_ERRORas the finalevent — confirmed it fails on the pre-review commit with the
double-terminal
against the patched package