From e9eb7a1b811e63e0b0cb9366e038b85e781bc8ea Mon Sep 17 00:00:00 2001 From: ciregenz Date: Fri, 31 Jul 2026 14:17:17 -0700 Subject: [PATCH] [eric] backend: surface error-shaped ResultMessages as failed turns instead of silent successes --- backend/apps/agents/manager/run/TurnRunner.py | 5 ++- .../streaming/handle_result_message.py | 41 ++++++++++++++++- backend/tests/test_result_message.py | 44 +++++++++++++++++++ backend/tests/test_streaming_harness.py | 34 ++++++++++++++ 4 files changed, 122 insertions(+), 2 deletions(-) diff --git a/backend/apps/agents/manager/run/TurnRunner.py b/backend/apps/agents/manager/run/TurnRunner.py index b749a32a..3a650d50 100644 --- a/backend/apps/agents/manager/run/TurnRunner.py +++ b/backend/apps/agents/manager/run/TurnRunner.py @@ -15,7 +15,7 @@ from backend.apps.agents.core.error_classify import CAPACITY_BACKOFFS, capacity_ from backend.apps.agents.manager.streaming.state import ThinkingState, TurnState from backend.apps.agents.manager.streaming.handle_stream_event import handle_stream_event from backend.apps.agents.manager.streaming.handle_assistant_message import handle_assistant_message -from backend.apps.agents.manager.streaming.handle_result_message import handle_result_message +from backend.apps.agents.manager.streaming.handle_result_message import TurnResultError, handle_result_message from backend.apps.agents.manager.run.client_pool import ( SdkClientLike, acquire_client, @@ -152,6 +152,9 @@ class TurnRunner(AgentManagerProtocol): else: await p_run_streaming_turn() break + except TurnResultError: + # The CLI already ran the whole turn (tools executed) and then reported failure; a resume-retry would re-execute side effects, so this goes straight to the error card. + raise except Exception as e: # Make sure the consolidated-thinking ticker doesn't outlive the turn on error/retry. Without this, an exception mid-stream leaves a dangling task that keeps re-emitting against a stale msg id. if thinking.ticker_task is not None and not thinking.ticker_task.done(): diff --git a/backend/apps/agents/manager/streaming/handle_result_message.py b/backend/apps/agents/manager/streaming/handle_result_message.py index b9f72259..e69b5d43 100644 --- a/backend/apps/agents/manager/streaming/handle_result_message.py +++ b/backend/apps/agents/manager/streaming/handle_result_message.py @@ -6,7 +6,7 @@ inline. resolved_model / api_type / global_settings are the loop's per-run confi import asyncio import logging -from typing import Dict, Optional +from typing import Dict, List, Optional from typeguard import typechecked @@ -23,6 +23,35 @@ except ImportError: # the SDK is optional at runtime (mock mode); keep this mod logger = logging.getLogger(__name__) +class TurnResultError(Exception): + """The CLI's ResultMessage reported the turn ended in an error state (is_error, an + error_* subtype, or a max_tokens/refusal stop). Raised after the turn's token/cost + accounting so the run loop's existing error-card path owns the failure instead of the + turn being consumed as a silent success.""" + + +@typechecked +def p_turn_result_error_text(message: ResultMessage, subtype: str, stop_reason: Optional[str]) -> str: + parts: List[str] = [str(x).strip() for x in (getattr(message, "errors", None) or []) if str(x).strip()] + if not parts: + result_text = getattr(message, "result", None) + if isinstance(result_text, str) and result_text.strip(): + parts.append(result_text.strip()) + denials = getattr(message, "permission_denials", None) or [] + denied_tools = [str(d.get("tool_name")) for d in denials if isinstance(d, dict) and d.get("tool_name")] + if denied_tools: + parts.append("denied tools: " + ", ".join(denied_tools)) + if stop_reason == "max_tokens": + headline = "The model hit its maximum output length before finishing" + elif stop_reason == "refusal": + headline = "The model refused to continue this turn" + else: + headline = "The agent runtime reported this turn failed" + label = subtype if subtype and subtype != "success" else (stop_reason or "unknown") + detail = "; ".join(parts) + return f"{headline} ({label})." + (f" {detail}" if detail else "") + + @typechecked async def handle_result_message( message: ResultMessage, @@ -189,3 +218,13 @@ async def handle_result_message( }) except Exception: logger.exception("Failed to emit agent:context_update") + + # An error-shaped result used to be consumed as a normal end-of-turn: the user got preamble, then silence. Raise AFTER the accounting above so the failure surfaces as a real error card. + p_subtype = str(getattr(message, "subtype", "") or "") + p_stop_reason = getattr(message, "stop_reason", None) + if ( + bool(getattr(message, "is_error", False)) + or p_subtype.startswith("error") + or p_stop_reason in ("max_tokens", "refusal") + ): + raise TurnResultError(p_turn_result_error_text(message, p_subtype, p_stop_reason)) diff --git a/backend/tests/test_result_message.py b/backend/tests/test_result_message.py index 916ab2a4..44ecad4d 100644 --- a/backend/tests/test_result_message.py +++ b/backend/tests/test_result_message.py @@ -57,6 +57,50 @@ async def test_free_route_zeroes_cost(): assert session.cost_usd == 0.0 # cc/ is a subscription (server-funded) route, never billed per-token +@pytest.mark.asyncio +async def test_error_shaped_result_raises_after_accounting(): + # is_error / error_* subtype used to be consumed as a normal end-of-turn (silent success). + session, turn, thinking = p_fixt() + m = ResultMessage(subtype="error_during_execution", duration_ms=100, duration_api_ms=80, + is_error=True, num_turns=1, session_id="sdk-1", + usage={"input_tokens": 100, "output_tokens": 50}, + errors=["tool crashed hard"]) + with patch.object(result_message.ws_manager, "send_to_session", new=AsyncMock()): + with pytest.raises(result_message.TurnResultError) as exc: + await result_message.handle_result_message( + m, session, session.id, turn, thinking, {}, "sonnet", "anthropic", load_settings()) + assert "tool crashed hard" in str(exc.value) + assert session.tokens["output"] == 50 # token accounting still lands before the raise + + +@pytest.mark.asyncio +async def test_max_tokens_and_refusal_stops_raise_even_with_success_subtype(): + for stop_reason, phrase in (("max_tokens", "maximum output length"), ("refusal", "refused")): + session, turn, thinking = p_fixt() + m = ResultMessage(subtype="success", duration_ms=100, duration_api_ms=80, + is_error=False, num_turns=1, session_id="sdk-1", + usage={"input_tokens": 10, "output_tokens": 5}, stop_reason=stop_reason) + with patch.object(result_message.ws_manager, "send_to_session", new=AsyncMock()): + with pytest.raises(result_message.TurnResultError) as exc: + await result_message.handle_result_message( + m, session, session.id, turn, thinking, {}, "sonnet", "anthropic", load_settings()) + assert phrase in str(exc.value) + + +@pytest.mark.asyncio +async def test_success_result_never_raises(): + # The mutation pair for the error detection: the happy path must stay a normal completion. + session, turn, thinking = p_fixt() + m = ResultMessage(subtype="success", duration_ms=100, duration_api_ms=80, + is_error=False, num_turns=1, session_id="sdk-1", + usage={"input_tokens": 10, "output_tokens": 5}, stop_reason="end_turn", + permission_denials=[{"tool_name": "Bash"}]) + with patch.object(result_message.ws_manager, "send_to_session", new=AsyncMock()): + await result_message.handle_result_message( + m, session, session.id, turn, thinking, {}, "sonnet", "anthropic", load_settings()) + assert session.tokens["output"] == 5 + + @pytest.mark.asyncio async def test_resets_per_turn_state_at_completion(): session, turn, thinking = p_fixt() diff --git a/backend/tests/test_streaming_harness.py b/backend/tests/test_streaming_harness.py index da327b39..01bb3779 100644 --- a/backend/tests/test_streaming_harness.py +++ b/backend/tests/test_streaming_harness.py @@ -345,6 +345,40 @@ def test_thinking_block_before_text_is_handled(monkeypatch): assert session.status == "completed" +def test_error_result_surfaces_error_card_not_silent_success(monkeypatch): + # An error-shaped ResultMessage used to be consumed as a normal end-of-turn: status "completed", no card, the user saw preamble then silence. + session, events = p_drive(monkeypatch, [ + p_assistant([TextBlock(text="Let me work on that.")]), + p_result(subtype="error_during_execution", is_error=True, + errors=["upstream exploded mid-run"]), + ]) + assert session.status == "error" + p_cards = [m for m in session.messages if m.role == "system"] + assert any("upstream exploded mid-run" in str(m.content) for m in p_cards) + # the card is broadcast so the UI renders it, not just stored + assert any(e == "agent:message" and (d.get("message") or {}).get("role") == "system" + for e, d in events) + + +def test_max_tokens_stop_reason_is_a_failed_turn(monkeypatch): + session, events = p_drive(monkeypatch, [ + p_assistant([TextBlock(text="half an answ")]), + p_result(stop_reason="max_tokens"), + ]) + assert session.status == "error" + assert any(m.role == "system" and "maximum output length" in str(m.content) for m in session.messages) + + +def test_success_result_still_completes_normally(monkeypatch): + # the mutation pair for the error-result detection: success + is_error False must never be flagged + session, events = p_drive(monkeypatch, [ + p_assistant([TextBlock(text="all done")]), + p_result(subtype="success", is_error=False, stop_reason="end_turn"), + ]) + assert session.status == "completed" + assert not any(m.role == "system" for m in session.messages) + + def test_transient_capacity_error_is_retried_then_succeeds(monkeypatch): # the capacity-retry while-loop: first query() raises a transient error, the loop backs off (sleep mocked to no-op) and re-queries, which succeeds. This is the exact behavior the streaming restructuring must preserve. real_sleep = asyncio.sleep # capture before patching to avoid self-recursion