[eric] agents: the CLI's silent provider-500 retries become breadcrumbs and near-miss ledger entries

backend/apps/agents/agent_manager.py

@@ -127,7 +127,7 @@ class AgentManager(SessionLifecycle, SessionPersistence, Messaging, SessionContr
             # off), so an empty prewarm prompt would boot a different thinking config than a typical
             # first message and fingerprint-miss into a respawn. 50+ chars matches the common case.
             p_representative = "prewarm placeholder prompt of representative length for boot"
-            (options, options_kwargs, _pc, _stderr, _gs) = await self.build_agent_options(
+            (options, options_kwargs, _, _, _) = await self.build_agent_options(
                 session, session_id, p_representative, "", builtin_perms,
                 None, None, None, False, p_router_model_id, p_api_type)
             from claude_agent_sdk import ClaudeSDKClient

backend/apps/agents/manager/run/TurnRunner.py

@@ -17,6 +17,7 @@ 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 TurnResultError, handle_result_message
+from backend.apps.agents.manager.streaming.note_provider_retry import note_provider_retry, settle_provider_retries
 from backend.apps.agents.manager.run.client_pool import (
     SdkClientLike,
     acquire_client,
@@ -110,8 +111,11 @@ class TurnRunner(AgentManagerProtocol):
                 if isinstance(message, SystemMessage):
                     raw = message.__dict__ if hasattr(message, '__dict__') else str(message)
                     logger.info(f"[MCP-DEBUG] SystemMessage: {raw}")
-                    if getattr(message, "subtype", "") == "compact_boundary":
+                    p_subtype = getattr(message, "subtype", "")
+                    if p_subtype == "compact_boundary":
                         turn.compact_boundaries += 1
+                    elif p_subtype == "api_retry":
+                        note_provider_retry(session_id, raw, turn)
 
                 if isinstance(message, StreamEvent):
                     await handle_stream_event(
@@ -201,6 +205,7 @@ class TurnRunner(AgentManagerProtocol):
                         attempts=p_router_retry_attempt + capacity_retry_attempt,
                         sessions=self.sessions,
                     )
+                settle_provider_retries(session_id, turn, resolved_model, self.sessions)
                 break
             except TurnResultError as p_result_err:
                 # "Unable to connect" in a turn result is the CLI failing to reach our own localhost

backend/apps/agents/manager/streaming/note_provider_retry.py

@@ -0,0 +1,45 @@
+"""The CLI retries provider 500s/429s by itself, up to 10 attempts with backoffs measured in tens
+of seconds, and tells nobody. To the user the card just sits there; to us the turn looks clean.
+
+This turns each of those `api_retry` system events into a breadcrumb, so a turn that eventually
+dies carries "the provider 500'd four times first" in its envelope instead of an unexplained
+timeout. Counting it as a RECOVERED near-miss happens later, at turn end, because a retry that is
+still in flight has not recovered anything yet."""
+
+from typing import Optional
+
+from typeguard import typechecked
+
+from backend.apps.agents.core import flight_recorder
+from backend.apps.agents.manager.streaming.state import TurnState
+
+
+@typechecked
+def note_provider_retry(session_id: str, raw: object, turn: TurnState) -> None:
+    """Record one CLI-internal provider retry. Never raises; diagnostics must not break a turn."""
+    try:
+        data = raw.get("data", {}) if isinstance(raw, dict) else {}
+        if not isinstance(data, dict):
+            data = {}
+        turn.provider_retries += 1
+        delay_ms = data.get("retry_delay_ms")
+        turn.provider_retry_wait_ms += int(delay_ms) if isinstance(delay_ms, int) else 0
+        flight_recorder.crumb(
+            session_id,
+            "provider-retry",
+            status=data.get("error_status"),
+            error=str(data.get("error", ""))[:40],
+            attempt=data.get("attempt"),
+            delay_ms=delay_ms,
+        )
+    except Exception:
+        pass
+
+
+@typechecked
+def settle_provider_retries(session_id: str, turn: TurnState, model: Optional[str], sessions: Optional[dict] = None) -> None:
+    """Called when a turn finishes cleanly: any retries it survived were a silent save, so they get
+    a denominator in the near-miss ledger."""
+    if turn.provider_retries <= 0:
+        return
+    flight_recorder.record_recovery(session_id, "provider-retry", model, turn.provider_retries, sessions)

backend/apps/agents/manager/streaming/state.py

@@ -56,6 +56,9 @@ class TurnState(BaseModel):
     baseline_captured: bool = False
     # CLI compact_boundary events seen this turn; one plus a ProcessError = the autocompact-thrash death the context-pressure valve retries.
     compact_boundaries: int = 0
+    # Provider 500s/429s the CLI retried on its own; the user sees only a long silence, so these are counted rather than lost.
+    provider_retries: int = 0
+    provider_retry_wait_ms: int = 0
     # Mid-turn context breaker: fires once per turn, and only after a below-trigger reading (a turn that STARTS over the trigger must run, or a failed shrink would break-loop forever).
     context_break_fired: bool = False
     saw_input_below_trigger: bool = False

backend/apps/agents/schedule_mcp_server.py

@@ -533,8 +533,8 @@ def handle_delete_step(args: dict) -> dict:
 
 # How long a synchronous test may hold the turn. Long enough for a real multi-step workflow, short
 # enough that a wedged test returns an honest "still running" instead of hanging the conversation.
-P_TEST_WAIT_S = 240
-P_TEST_POLL_S = 3
+TEST_WAIT_S = 240
+TEST_POLL_S = 3
 
 
 def handle_test_workflow(args: dict) -> dict:
@@ -549,10 +549,10 @@ def handle_test_workflow(args: dict) -> dict:
     # to "call ReadTestTranscript once it finishes", but a model has no way to know when that is, so
     # it ended its turn and the HUMAN had to keep re-pinging it. A test whose result the caller
     # cannot observe is not a tool, it is homework for the user.
-    deadline = time.time() + P_TEST_WAIT_S
+    deadline = time.time() + TEST_WAIT_S
     last_status = "running"
     while time.time() < deadline:
-        time.sleep(P_TEST_POLL_S)
+        time.sleep(TEST_POLL_S)
         t = _call("GET", f"/{wid}/test-transcript")
         if "_error" in t:
             continue
@@ -562,7 +562,7 @@ def handle_test_workflow(args: dict) -> dict:
         transcript = t.get("transcript") or "(empty transcript)"
         return _ok(f"Test finished (status: {last_status}). Transcript:\n\n{transcript}")
     return _ok(
-        f"Test Agent (session {sid[:8]}) is still running after {P_TEST_WAIT_S}s, so it is a long one. "
+        f"Test Agent (session {sid[:8]}) is still running after {TEST_WAIT_S}s, so it is a long one. "
         f"Last status: {last_status}. Call ReadTestTranscript to pick up the result."
     )
 

backend/tests/test_flight_recorder.py

@@ -1,7 +1,6 @@
 """The flight recorder's contract: cheap crumbs, bounded rings, envelopes that name everything,
 and a near-miss ledger that counts silent recoveries."""
 
-import time
 
 from backend.apps.agents.core import flight_recorder as fr
 

backend/tests/test_provider_retry_ledger.py

@@ -0,0 +1,95 @@
+"""The silent provider stall (Eric's "the ones that never get picked up").
+
+The CLI retries provider 500s itself, up to 10 attempts, backing off in tens of seconds. Nothing
+reached our telemetry and nothing reached the user: the card just sat there. Both payloads below
+are verbatim from live traffic on 2026-08-07 05:20-05:21, where two turns took 15.7s and >50s for
+exactly this reason and the near-miss ledger recorded zero.
+"""
+
+import inspect
+from typing import Any, Dict, List
+
+from backend.apps.agents.core import flight_recorder
+from backend.apps.agents.manager.run import TurnRunner
+from backend.apps.agents.manager.streaming.note_provider_retry import note_provider_retry, settle_provider_retries
+from backend.apps.agents.manager.streaming.state import TurnState
+
+# Verbatim SystemMessage.__dict__ from the live 500s.
+LIVE_RETRY: Dict[str, Any] = {
+    "subtype": "api_retry",
+    "data": {
+        "type": "system",
+        "subtype": "api_retry",
+        "attempt": 1,
+        "max_retries": 10,
+        "retry_delay_ms": 30000,
+        "error_status": 500,
+        "error": "server_error",
+        "session_id": "2b0f3e16-8a38-4ae4-b1e9-70da2353b5d4",
+        "uuid": "0cc6ec4a-6ead-4f3a-a29b-a5ceac2a4d2e",
+    },
+}
+
+
+def test_the_live_500_becomes_a_breadcrumb_with_its_status_and_backoff():
+    sid = "provretry01"
+    flight_recorder.drop_session(sid)
+    turn = TurnState()
+    note_provider_retry(sid, LIVE_RETRY, turn)
+    crumbs = [c for c in flight_recorder.breadcrumbs(sid) if c.get("l") == "provider-retry"]
+    assert len(crumbs) == 1, "the retry must leave a trace a stranger can read"
+    assert crumbs[0]["status"] == 500
+    assert crumbs[0]["delay_ms"] == 30000, "the backoff is the whole reason the user saw a long silence"
+    assert turn.provider_retries == 1
+    assert turn.provider_retry_wait_ms == 30000
+    flight_recorder.drop_session(sid)
+
+
+def test_a_turn_that_survives_retries_lands_in_the_near_miss_ledger():
+    sid = "provretry02"
+    flight_recorder.drop_session(sid)
+    sent: List[Dict[str, Any]] = []
+    import backend.apps.service.client as service_client
+    original = service_client.submit_diagnostic
+    service_client.submit_diagnostic = lambda payload: sent.append(payload)
+    try:
+        turn = TurnState()
+        note_provider_retry(sid, LIVE_RETRY, turn)
+        note_provider_retry(sid, LIVE_RETRY, turn)
+        settle_provider_retries(sid, turn, "sonnet-cc", {})
+    finally:
+        service_client.submit_diagnostic = original
+    assert len(sent) == 1, "one settle per turn, not one per retry"
+    assert sent[0]["kind"] == "recovered"
+    assert sent[0]["subkind"] == "provider-retry"
+    assert sent[0]["attempts"] == 2, "the denominator has to count every retry the turn rode out"
+    flight_recorder.drop_session(sid)
+
+
+def test_a_turn_with_no_retries_stays_out_of_the_ledger():
+    sid = "provretry03"
+    sent: List[Dict[str, Any]] = []
+    import backend.apps.service.client as service_client
+    original = service_client.submit_diagnostic
+    service_client.submit_diagnostic = lambda payload: sent.append(payload)
+    try:
+        settle_provider_retries(sid, TurnState(), "sonnet-cc", {})
+    finally:
+        service_client.submit_diagnostic = original
+    assert sent == [], "a clean turn must not inflate the near-miss count"
+
+
+def test_a_malformed_retry_event_never_breaks_the_turn():
+    sid = "provretry04"
+    flight_recorder.drop_session(sid)
+    turn = TurnState()
+    for junk in ("not a dict", {"subtype": "api_retry"}, {"subtype": "api_retry", "data": None}):
+        note_provider_retry(sid, junk, turn)
+    flight_recorder.drop_session(sid)
+
+
+def test_the_turn_loop_actually_dispatches_api_retry():
+    src = inspect.getsource(TurnRunner)
+    assert 'p_subtype == "api_retry"' in src, "the SystemMessage branch must recognise the retry subtype"
+    assert "note_provider_retry(session_id, raw, turn)" in src
+    assert "settle_provider_retries(session_id, turn, resolved_model, self.sessions)" in src

backend/tests/test_sync_workflow_test.py

@@ -17,8 +17,8 @@ def test_the_handler_waits_for_the_result_instead_of_returning_a_promise():
 def test_a_long_test_returns_honestly_instead_of_hanging_the_turn():
     src = inspect.getsource(srv.handle_test_workflow)
     assert "still running after" in src
-    assert srv.P_TEST_WAIT_S <= 300, "a bounded wait; a wedged test must never hold a turn forever"
-    assert srv.P_TEST_POLL_S >= 1
+    assert srv.TEST_WAIT_S <= 300, "a bounded wait; a wedged test must never hold a turn forever"
+    assert srv.TEST_POLL_S >= 1
 
 
 def test_the_description_tells_the_model_to_keep_going():