From 86e323cbce29244763fb45c985427eb30cdaa893 Mon Sep 17 00:00:00 2001 From: ciregenz Date: Thu, 6 Aug 2026 22:39:30 -0700 Subject: [PATCH] [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 | 116 ++++++++++++++++++ backend/apps/agents/manager/run/TurnRunner.py | 13 ++ .../agents/manager/run/handle_run_error.py | 3 + .../manager/session/SessionLifecycle.py | 2 + backend/tests/test_flight_recorder.py | 63 ++++++++++ 5 files changed, 197 insertions(+) create mode 100644 backend/apps/agents/core/flight_recorder.py create mode 100644 backend/tests/test_flight_recorder.py diff --git a/backend/apps/agents/core/flight_recorder.py b/backend/apps/agents/core/flight_recorder.py new file mode 100644 index 00000000..e42a8694 --- /dev/null +++ b/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 diff --git a/backend/apps/agents/manager/run/TurnRunner.py b/backend/apps/agents/manager/run/TurnRunner.py index 38d59121..87a2b80a 100644 --- a/backend/apps/agents/manager/run/TurnRunner.py +++ b/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} " diff --git a/backend/apps/agents/manager/run/handle_run_error.py b/backend/apps/agents/manager/run/handle_run_error.py index 92b5a4c3..7e0185f9 100644 --- a/backend/apps/agents/manager/run/handle_run_error.py +++ b/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"), diff --git a/backend/apps/agents/manager/session/SessionLifecycle.py b/backend/apps/agents/manager/session/SessionLifecycle.py index 5b958144..ca23310d 100644 --- a/backend/apps/agents/manager/session/SessionLifecycle.py +++ b/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" diff --git a/backend/tests/test_flight_recorder.py b/backend/tests/test_flight_recorder.py new file mode 100644 index 00000000..dca4cc57 --- /dev/null +++ b/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")