feat: surface agent errors to the TUI + progress-aware resume backoff

backend/app/api/v1/generation_sessions.py

@@ -396,6 +396,10 @@ def _ws_usage_view(usage: Optional[ModelTokenUsage]) -> Optional[Dict[str, int]]
 # here would mislabel the KB-init step as "Phase 1".
 KB_INIT_PHASE_NAME = "Knowledge Base Initialization with Rosetta"
 
+# The /status payload ships only the newest agent error/warning events; the doc
+# itself retains more (see _MAX_AGENT_ERROR_EVENTS in the state machine).
+_AGENT_ERROR_EVENTS_STATUS_LIMIT = 50
+
 
 def _ws_phase_view_entry(
     ws_data: dict,
@@ -426,6 +430,9 @@ def _ws_phase_view_entry(
         entry["usage"] = usage_view
     if models:
         entry["models"] = models
+    agent_state = ws_data.get("agent_state")
+    if agent_state:
+        entry["agent_state"] = agent_state
     return entry
 
 
@@ -678,6 +685,12 @@ async def get_generation_session_status(
             "workspace_phases": workspace_phases_view,
         }
 
+        # Agent error/warning events: newest tail only, key omitted when empty
+        # (old docs without the field keep their exact response shape).
+        agent_error_events = session_doc.get("agent_error_events") or []
+        if agent_error_events:
+            response["agent_error_events"] = agent_error_events[-_AGENT_ERROR_EVENTS_STATUS_LIMIT:]
+
         # 2.5a: include result fields when COMPLETED (Gap 1 fix)
         if session_doc.get("status") == GenerationStatus.COMPLETED.value:
             response["result"] = session_doc.get("result")
@@ -1138,6 +1151,8 @@ async def _run_generation_session_workflow(
     finally:
         task_registry.deregister_task(generation_id)
         TelemetryContext.set_agent_query_totals_handler(None)
+        TelemetryContext.set_agent_error_event_handler(None)
+        TelemetryContext.set_workspace_agent_state_handler(None)
 
 
 @router.post("/{generation_id}/resend-email", response_model=ResendEmailResponse)

backend/app/core/telemetry_context.py

@@ -16,6 +16,7 @@
 from typing import Any, Awaitable, Callable, Dict, Optional, Union
 
 from app.core.mcp_config import EnabledMcpsResolution, enabled_mcps_to_parameter_string
+from app.schemas.agent_error_events import AgentErrorEvent, WorkspaceAgentState
 from app.schemas.model_token_usage import ModelTokenUsage
 from app.schemas.telemetry_workflow import TelemetryWorkflowLabel
 
@@ -32,6 +33,19 @@
     contextvars.ContextVar("agent_query_totals_handler", default=None)
 )
 
+AgentErrorEventHandler = Callable[[str, AgentErrorEvent], Awaitable[None]]
+_agent_error_event_handler: contextvars.ContextVar[Optional[AgentErrorEventHandler]] = (
+    contextvars.ContextVar("agent_error_event_handler", default=None)
+)
+
+WorkspaceAgentStateHandler = Callable[
+    [str, str, Optional[WorkspaceAgentState]],
+    Awaitable[None],
+]
+_workspace_agent_state_handler: contextvars.ContextVar[Optional[WorkspaceAgentStateHandler]] = (
+    contextvars.ContextVar("workspace_agent_state_handler", default=None)
+)
+
 
 class TelemetryContext:
     """
@@ -176,6 +190,30 @@ def set_agent_query_totals_handler(handler: Optional[AgentQueryTotalsHandler]) -
     def get_agent_query_totals_handler() -> Optional[AgentQueryTotalsHandler]:
         return _agent_query_totals_handler.get()
 
+    @staticmethod
+    def set_agent_error_event_handler(handler: Optional[AgentErrorEventHandler]) -> None:
+        """Register async callback for durable agent error/warning events.
+
+        Signature: ``(generation_id, event: AgentErrorEvent) -> Awaitable[None]``
+        """
+        _agent_error_event_handler.set(handler)
+
+    @staticmethod
+    def get_agent_error_event_handler() -> Optional[AgentErrorEventHandler]:
+        return _agent_error_event_handler.get()
+
+    @staticmethod
+    def set_workspace_agent_state_handler(handler: Optional[WorkspaceAgentStateHandler]) -> None:
+        """Register async callback for the per-workspace agent-state badge.
+
+        Signature: ``(generation_id, workspace_id, state: WorkspaceAgentState | None) -> Awaitable[None]``
+        """
+        _workspace_agent_state_handler.set(handler)
+
+    @staticmethod
+    def get_workspace_agent_state_handler() -> Optional[WorkspaceAgentStateHandler]:
+        return _workspace_agent_state_handler.get()
+
     @staticmethod
     def set_workspace_name(workspace_name: str) -> None:
         """
@@ -293,3 +331,5 @@ def clear_context() -> None:
         """Clear user context (called after request completes)."""
         _user_context.set(None)
         _agent_query_totals_handler.set(None)
+        _agent_error_event_handler.set(None)
+        _workspace_agent_state_handler.set(None)

backend/app/schemas/agent.py

@@ -15,3 +15,27 @@ class AgentResult:
     session_id: Optional[str] = None
     is_error: bool = False
     active_model: Optional[str] = None  # set when agent_query_with_resume switched to fallback
+
+
+@dataclass(frozen=True)
+class AgentQueryProgress:
+    """Observable work one agent_query performed before it ended.
+
+    Captured from the message stream as it flows, so the numbers survive a
+    mid-stream crash (the exception path reads the same accumulator).
+    """
+    num_messages: int = 0
+    num_tool_uses: int = 0
+
+
+class AgentQueryError(Exception):
+    """agent_query failure carrying the progress observed before the crash.
+
+    Subclasses Exception so every existing broad handler keeps working; the
+    resume loop reads ``progress`` to decide whether the failed attempt still
+    counts as saved work.
+    """
+
+    def __init__(self, message: str, progress: Optional[AgentQueryProgress] = None):
+        super().__init__(message)
+        self.progress = progress

backend/app/schemas/agent_error_events.py

@@ -0,0 +1,46 @@
+"""Durable agent error/warning events surfaced to /status and the TUI.
+
+Issue-focused by design: these are the states a user must know about
+(crashes, retries, model switches, give-ups, workspace aborts) — not the
+full agent lifecycle. Stored on the generation session doc under
+``agent_error_events``, written ONLY via
+``GenerationSessionStateMachine.record_agent_error_event`` (Commandment VII).
+"""
+
+from enum import Enum
+from typing import Optional
+
+from pydantic import BaseModel
+
+from app.schemas.agent import AgentErrorType
+
+
+class AgentErrorEventKind(str, Enum):
+    AGENT_CRASH = "agent_crash"            # SDK/API crash or timeout in a phase query
+    MODEL_FALLBACK = "model_fallback"      # mid-generation model switch (old -> new + reason)
+    PHASE_GAVE_UP = "phase_gave_up"        # resume budget exhausted
+    PHASE_INCOMPLETE = "phase_incomplete"  # phase checkpointed but validator unsatisfied
+    WORKSPACE_ABORTED = "workspace_aborted"
+
+
+class WorkspaceAgentState(str, Enum):
+    """Transient per-workspace badge shown in the TUI workspace list.
+
+    Absence of ``workspace_phases[ws]["agent_state"]`` means running normally.
+    """
+
+    RETRYING = "retrying"
+    ABORTED = "aborted"
+
+
+class AgentErrorEvent(BaseModel):
+    at: str  # UTC ISO timestamp
+    workspace_id: str
+    kind: AgentErrorEventKind
+    message: str  # human-readable, no internal jargon
+    phase: Optional[int] = None
+    error_type: Optional[AgentErrorType] = None
+    crash_backoff: Optional[str] = None  # "n/6" at the time of the event
+    next_attempt_at: Optional[str] = None  # UTC ISO, set when a backoff wait was scheduled
+    model: Optional[str] = None  # active model; for model_fallback: the NEW model
+    previous_model: Optional[str] = None  # for model_fallback

backend/app/services/agent_error_event_recorder.py

@@ -0,0 +1,91 @@
+"""Best-effort recording of agent error events from deep inside agent code.
+
+Mirrors the ``agent_query_totals_handler`` pattern: the workflow registers
+bound GenerationSessionService methods on TelemetryContext
+(``build_workflow_context``), and agent-level code calls the module functions
+below without threading services through call signatures. Outside a workflow
+(unit tests, one-off queries) both functions are silent no-ops.
+
+They NEVER raise — error reporting must not break the agent loop it reports
+on (and events fired while the session is no longer RUNNING are dropped by
+the state machine guard and swallowed here).
+"""
+
+import logging
+from datetime import datetime, timezone
+from typing import Optional
+
+from app.core.telemetry_context import TelemetryContext
+from app.schemas.agent import AgentErrorType
+from app.schemas.agent_error_events import (
+    AgentErrorEvent,
+    AgentErrorEventKind,
+    WorkspaceAgentState,
+)
+
+logger = logging.getLogger(__name__)
+
+# Keep durable event messages compact; full detail stays in the agent logs.
+_MAX_EVENT_MESSAGE_CHARS = 300
+
+
+def _resolve_identity(workspace_id: Optional[str]) -> tuple[Optional[str], Optional[str]]:
+    """(generation_id, workspace_id) from args + TelemetryContext fallback."""
+    generation_id = TelemetryContext.get_generation_id()
+    ws = workspace_id or TelemetryContext.get_workspace_name()
+    return generation_id, ws
+
+
+async def record_agent_error_event_safe(
+    kind: AgentErrorEventKind,
+    message: str,
+    *,
+    workspace_id: Optional[str] = None,
+    phase: Optional[int] = None,
+    error_type: Optional[AgentErrorType] = None,
+    crash_backoff: Optional[str] = None,
+    next_attempt_at: Optional[str] = None,
+    model: Optional[str] = None,
+    previous_model: Optional[str] = None,
+) -> None:
+    """Record one durable agent error/warning event; silent no-op outside a workflow."""
+    handler = TelemetryContext.get_agent_error_event_handler()
+    generation_id, ws = _resolve_identity(workspace_id)
+    if handler is None or not generation_id or not ws:
+        return
+    event = AgentErrorEvent(
+        at=datetime.now(timezone.utc).isoformat(),
+        workspace_id=ws,
+        kind=kind,
+        message=(message or "")[:_MAX_EVENT_MESSAGE_CHARS],
+        phase=phase,
+        error_type=error_type,
+        crash_backoff=crash_backoff,
+        next_attempt_at=next_attempt_at,
+        model=model,
+        previous_model=previous_model,
+    )
+    try:
+        await handler(generation_id, event)
+    except Exception:
+        logger.warning(
+            "Failed to record agent error event (kind=%s ws=%s)", kind, ws, exc_info=True
+        )
+
+
+async def set_workspace_retry_state_safe(
+    state: Optional[WorkspaceAgentState],
+    *,
+    workspace_id: Optional[str] = None,
+) -> None:
+    """Set/clear the per-workspace agent badge; silent no-op outside a workflow."""
+    handler = TelemetryContext.get_workspace_agent_state_handler()
+    generation_id, ws = _resolve_identity(workspace_id)
+    if handler is None or not generation_id or not ws:
+        return
+    try:
+        await handler(generation_id, ws, state)
+    except Exception:
+        logger.warning(
+            "Failed to set workspace agent state (%s ws=%s)", state, ws, exc_info=True
+        )

backend/app/services/agent_metrics.py

@@ -17,6 +17,7 @@
 )
 
 from app.core.tool_usage import MCP_TOOL_PREFIX
+from app.schemas.agent import AgentQueryProgress
 from app.schemas.agent_metrics import AgentUsage, ToolUsage
 
 logger = logging.getLogger(__name__)
@@ -179,8 +180,10 @@ def __init__(self) -> None:
             _BUCKET_SUB_MCP: {},
         }
         self._unknown_result_ids: set[str] = set()
+        self._messages_seen = 0
 
     def push(self, message: Any) -> None:
+        self._messages_seen += 1
         try:
             if isinstance(message, AssistantMessage):
                 self._push_assistant(message)
@@ -310,6 +313,21 @@ def _flush_pending_registry(self) -> None:
             bucket_key = self._bucket_key(name, is_subagent_context)
             self._increment_bucket(bucket_key, name, tokens=None)
 
+    def progress_snapshot(self) -> AgentQueryProgress:
+        """Side-effect-free snapshot of observed progress (messages + tool uses).
+
+        Unlike get_metrics(), does NOT flush the pending registry — safe to call
+        mid-stream or from an exception handler. Pending (unmatched) tool calls
+        count as tool uses: a crash mid-tool is still observed work.
+        """
+        completed = sum(
+            entry.count for bucket in self._buckets.values() for entry in bucket.values()
+        )
+        return AgentQueryProgress(
+            num_messages=self._messages_seen,
+            num_tool_uses=completed + len(self._registry),
+        )
+
     def get_metrics(self) -> dict[str, Any]:
         """Return JSON-friendly metrics for logs and telemetry.
 

backend/app/services/agent_stream_broker.py

@@ -25,7 +25,12 @@
 from typing import Dict, Optional, Set, Tuple
 
 from app.core.telemetry_context import TelemetryContext
-from app.services.agent_stream_events import AgentStreamEvent, message_to_ui_events
+from app.services.agent_stream_events import (
+    AgentStreamEvent,
+    EventKind,
+    message_to_ui_events,
+    synthetic_event,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -143,6 +148,22 @@ def publish_nowait(self, sdk_message: object) -> None:
             # event can never escape into the agent stream loop.
             logger.debug("StreamPublisher: publish_nowait failed", exc_info=True)
 
+    def publish_line_nowait(self, kind: EventKind, message: str) -> None:
+        """Publish one backend-authored line (crash/retry narrative). Never raises."""
+        try:
+            if not self._broker.has_subscribers(self._generation_id, self._workspace_id):
+                return
+            event = synthetic_event(
+                kind,
+                message,
+                generation_id=self._generation_id,
+                workspace_id=self._workspace_id,
+                workflow=self._workflow,
+            )
+            self._broker.publish(self._generation_id, self._workspace_id, event)
+        except Exception:
+            logger.debug("StreamPublisher: publish_line_nowait failed", exc_info=True)
+
 
 def build_stream_publisher_from_context() -> Optional[StreamPublisher]:
     """Build a publisher from the current ``TelemetryContext``, or ``None``.

backend/app/services/agent_stream_events.py

@@ -62,6 +62,7 @@
     "tool_result",
     "result",
     "system",
+    "error",
     "unknown",
 ]
 
@@ -240,6 +241,29 @@ def _user_events(
     return events
 
 
+def synthetic_event(
+    kind: EventKind,
+    message: str,
+    *,
+    generation_id: str,
+    workspace_id: str,
+    workflow: Optional[str] = None,
+) -> AgentStreamEvent:
+    """Build a backend-authored event (crash/retry narrative), not from an SDK message.
+
+    Used by the resume loop to surface errors and retry waits inline in the
+    live message feed; same truncation contract as converted SDK messages.
+    """
+    return AgentStreamEvent(
+        timestamp=_now_iso(),
+        generation_id=generation_id,
+        workspace_id=workspace_id,
+        workflow=workflow,
+        kind=kind,
+        message=_truncate(_collapse_whitespace(message)),
+    )
+
+
 def message_to_ui_events(
     message: Any,
     *,