[eric] backend: surface error-shaped ResultMessages as failed turns instead of silent successes

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():

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))

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()

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