[eric] telemetry: per-session flight recorder, error envelopes name cause+lane+phase+crumbs, silent recoveries hit the ledger

backend/apps/agents/core/flight_recorder.py

@@ -0,0 +1,116 @@
+"""Per-session flight recorder: a fixed breadcrumb ring appended at points that already log,
+flushed into a diagnostic envelope ONLY when an error surfaces (or a silent recovery is counted).
+The happy path pays one O(1) deque append per event and nothing else; nothing here touches disk
+or network on its own."""
+
+import threading
+import time
+from collections import deque
+from typing import Dict, List, Optional
+
+from typeguard import typechecked
+
+P_RING_SIZE = 64
+p_lock = threading.Lock()
+p_rings: Dict[str, deque] = {}
+
+
+@typechecked
+def crumb(session_id: str, label: str, **meta: object) -> None:
+    """Append one breadcrumb; cheap enough for every retry decision and phase stamp."""
+    entry = {"l": label, "t": round(time.time(), 3)}
+    for k, v in meta.items():
+        if v is not None:
+            entry[k] = v if isinstance(v, (int, float, bool)) else str(v)[:200]
+    with p_lock:
+        ring = p_rings.get(session_id)
+        if ring is None:
+            ring = deque(maxlen=P_RING_SIZE)
+            p_rings[session_id] = ring
+        ring.append(entry)
+
+
+@typechecked
+def drop_session(session_id: str) -> None:
+    """Sessions are deleted often; their crumbs must not accumulate forever."""
+    with p_lock:
+        p_rings.pop(session_id, None)
+
+
+@typechecked
+def breadcrumbs(session_id: str, last: int = 20) -> List[dict]:
+    with p_lock:
+        ring = p_rings.get(session_id)
+        return list(ring)[-last:] if ring else []
+
+
+@typechecked
+def lane_for_model(model: Optional[str]) -> str:
+    """The routing lane is a first-class confounder: cc/cx/gc ride the local router, api goes direct."""
+    m = model or ""
+    if m.endswith("-cc") or m.startswith("cc/"):
+        return "cc"
+    if m.endswith("-cx") or m.startswith("cx/"):
+        return "cx"
+    if m.startswith(("gc/", "gemini", "ag/")):
+        return "gc"
+    if m.startswith(("openrouter/", "cp-")):
+        return "openrouter" if m.startswith("openrouter/") else "custom"
+    return "api"
+
+
+@typechecked
+def concurrency_snapshot(sessions: Dict[str, object]) -> dict:
+    """The tandem set at event time, read from state already in memory; assembled only when an
+    envelope is being built, never on the hot path."""
+    try:
+        statuses = [getattr(s, "status", None) for s in sessions.values()]
+        return {
+            "sessions_total": len(statuses),
+            "turns_running": sum(1 for s in statuses if s == "running"),
+        }
+    except Exception:
+        return {"sessions_total": -1, "turns_running": -1}
+
+
+@typechecked
+def build_envelope(
+    session_id: str,
+    family: str,
+    subkind: str,
+    model: Optional[str],
+    phase: str,
+    attempts: int,
+    sessions: Optional[Dict[str, object]] = None,
+) -> dict:
+    """Everything a stranger needs to diagnose the failure without the machine in front of them."""
+    return {
+        "family": family,
+        "subkind": subkind,
+        "lane": lane_for_model(model),
+        "model": model,
+        "phase": phase,
+        "attempts": attempts,
+        "breadcrumbs": breadcrumbs(session_id),
+        "concurrency": concurrency_snapshot(sessions or {}),
+    }
+
+
+@typechecked
+def record_recovery(session_id: str, net: str, model: Optional[str], attempts: int, sessions: Optional[Dict[str, object]] = None) -> None:
+    """The near-miss ledger: a silent recovery the user never saw still counts in analytics, so
+    'how often do the nets fire' has a denominator. Fire-and-forget; failures never block the turn."""
+    crumb(session_id, "recovered", net=net, attempts=attempts)
+    try:
+        from backend.apps.service.client import submit_diagnostic
+        submit_diagnostic({
+            "kind": "recovered",
+            "subkind": net,
+            "session_id": session_id[:8],
+            "lane": lane_for_model(model),
+            "model": model,
+            "attempts": attempts,
+            "concurrency": concurrency_snapshot(sessions or {}),
+        })
+    except Exception:
+        pass

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

@@ -12,6 +12,7 @@ from typeguard import typechecked
 from backend.apps.agents.core.models import AgentSession
 from backend.apps.agents.core.ws_manager import ws_manager
 from backend.apps.agents.core.error_classify import CAPACITY_BACKOFFS, capacity_retry_wait, is_router_unreachable_error
+from backend.apps.agents.core import flight_recorder
 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
@@ -182,6 +183,16 @@ class TurnRunner(AgentManagerProtocol):
                     await p_run_streaming_turn_persistent()
                 else:
                     await p_run_streaming_turn()
+                # The near-miss ledger: a turn that needed retries and still finished is a net that
+                # FIRED, and "how often do the nets fire" needs a denominator in analytics.
+                if p_router_retry_attempt or capacity_retry_attempt:
+                    flight_recorder.record_recovery(
+                        session_id,
+                        net="router-resume" if p_router_retry_attempt else "transient-backoff",
+                        model=resolved_model,
+                        attempts=p_router_retry_attempt + capacity_retry_attempt,
+                        sessions=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
@@ -190,6 +201,7 @@ class TurnRunner(AgentManagerProtocol):
                 # conversation without re-executing side effects: re-ensure the router, resume, go.
                 if p_router_retry_attempt < 2 and is_router_unreachable_error(str(p_result_err)):
                     p_router_retry_attempt += 1
+                    flight_recorder.crumb(session_id, "router-retry", attempt=p_router_retry_attempt, err=str(p_result_err)[:160])
                     logger.warning(
                         f"Router unreachable mid-turn on session {session_id} "
                         f"(attempt {p_router_retry_attempt}/2); re-ensuring router and resuming. "
@@ -228,6 +240,7 @@ class TurnRunner(AgentManagerProtocol):
                         wait = 0.0
                 if wait is not None:
                     capacity_retry_attempt += 1
+                    flight_recorder.crumb(session_id, "transient-retry", attempt=capacity_retry_attempt, wait=wait, err=str(e)[:160])
                     mid_stream = turn.current_turn_emitted
                     logger.warning(
                         f"Transient upstream error on session {session_id} "

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

@@ -24,6 +24,7 @@ from backend.apps.agents.core.error_classify import (
 )
 from backend.apps.agents.core.extract_reset_hint import extract_reset_hint
 from backend.apps.agents.core.redact_for_telemetry import redact_for_telemetry
+from backend.apps.agents.core import flight_recorder
 
 logger = logging.getLogger(__name__)
 
@@ -237,6 +238,7 @@ async def handle_run_error(e: Exception, session: AgentSession, session_id: str,
             submit_diagnostic({
                 "kind": "model_error",
                 "subkind": "unknown_model",
+                "flight": flight_recorder.build_envelope(session_id, "model_error", "unknown_model", session.model, "stream" if turn.current_turn_emitted else "spawn", -1),
                 "model": session.model,
                 "provider": session.provider,
                 "connection_mode": getattr(load_settings(), "connection_mode", "own_key"),
@@ -258,6 +260,7 @@ async def handle_run_error(e: Exception, session: AgentSession, session_id: str,
             submit_diagnostic({
                 "kind": "model_error",
                 "subkind": "unclassified",
+                "flight": flight_recorder.build_envelope(session_id, "model_error", "unclassified", session.model, "stream" if turn.current_turn_emitted else "spawn", -1),
                 "model": session.model,
                 "provider": session.provider,
                 "connection_mode": getattr(load_settings(), "connection_mode", "own_key"),

backend/apps/agents/manager/session/SessionLifecycle.py

@@ -116,6 +116,8 @@ class SessionLifecycle(AgentManagerProtocol):
     async def delete_session(self, session_id: str) -> None:
         """Permanently delete a session: remove from memory and JSON file.
         Also stops browser-agent children first."""
+        from backend.apps.agents.core.flight_recorder import drop_session
+        drop_session(session_id)
         children = [
             s for s in self.sessions.values()
             if s.parent_session_id == session_id and s.mode == "browser-agent"

backend/tests/test_flight_recorder.py

@@ -0,0 +1,63 @@
+"""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
+
+
+def test_ring_is_bounded_and_ordered():
+    sid = "t-ring"
+    fr.drop_session(sid)
+    for i in range(100):
+        fr.crumb(sid, "step", n=i)
+    crumbs = fr.breadcrumbs(sid, last=100)
+    assert len(crumbs) == 64, "ring must cap at 64"
+    assert crumbs[-1]["n"] == 99 and crumbs[0]["n"] == 36, "oldest drop first"
+    fr.drop_session(sid)
+    assert fr.breadcrumbs(sid) == []
+
+
+def test_meta_values_are_truncated_and_typed():
+    sid = "t-meta"
+    fr.drop_session(sid)
+    fr.crumb(sid, "err", msg="x" * 999, count=3, flag=True, skipped=None)
+    c = fr.breadcrumbs(sid)[0]
+    assert len(c["msg"]) == 200 and c["count"] == 3 and c["flag"] is True and "skipped" not in c
+    fr.drop_session(sid)
+
+
+def test_lane_classification_matches_the_routing_reality():
+    assert fr.lane_for_model("sonnet-cc") == "cc"
+    assert fr.lane_for_model("cx/gpt-5.2") == "cx"
+    assert fr.lane_for_model("gemini-2.5-pro") == "gc"
+    assert fr.lane_for_model("openrouter/meta/llama") == "openrouter"
+    assert fr.lane_for_model("cp-openai/local") == "custom"
+    assert fr.lane_for_model("sonnet") == "api"
+    assert fr.lane_for_model(None) == "api"
+
+
+def test_envelope_carries_cause_context_and_crumbs():
+    sid = "t-env"
+    fr.drop_session(sid)
+    fr.crumb(sid, "router-retry", attempt=1)
+
+    class FakeSession:
+        status = "running"
+
+    env = fr.build_envelope(sid, "model_error", "unclassified", "sonnet-cc", "stream", 2, {"a": FakeSession(), "b": FakeSession()})
+    assert env["family"] == "model_error" and env["lane"] == "cc" and env["phase"] == "stream"
+    assert env["attempts"] == 2 and env["breadcrumbs"][0]["l"] == "router-retry"
+    assert env["concurrency"] == {"sessions_total": 2, "turns_running": 2}
+    fr.drop_session(sid)
+
+
+def test_recovery_ledger_emits_a_countable_diagnostic(monkeypatch):
+    sent = []
+    import backend.apps.service.client as svc
+    monkeypatch.setattr(svc, "submit_diagnostic", lambda d: sent.append(d))
+    fr.record_recovery("t-rec-12345678", "router-resume", "sonnet-cc", 1, None)
+    assert len(sent) == 1
+    d = sent[0]
+    assert d["kind"] == "recovered" and d["subkind"] == "router-resume" and d["lane"] == "cc" and d["attempts"] == 1
+    fr.drop_session("t-rec-12345678")