diff --git a/backend/app/api/v1/generation_sessions.py b/backend/app/api/v1/generation_sessions.py index a85159b..88f1a6f 100644 --- a/backend/app/api/v1/generation_sessions.py +++ b/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) diff --git a/backend/app/core/telemetry_context.py b/backend/app/core/telemetry_context.py index 6e0cb9e..6aa524a 100644 --- a/backend/app/core/telemetry_context.py +++ b/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) diff --git a/backend/app/schemas/agent.py b/backend/app/schemas/agent.py index 89ec08a..a08e8c0 100644 --- a/backend/app/schemas/agent.py +++ b/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 diff --git a/backend/app/schemas/agent_error_events.py b/backend/app/schemas/agent_error_events.py new file mode 100644 index 0000000..3fe9bba --- /dev/null +++ b/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 diff --git a/backend/app/services/agent_error_event_recorder.py b/backend/app/services/agent_error_event_recorder.py new file mode 100644 index 0000000..50e00ec --- /dev/null +++ b/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 + ) diff --git a/backend/app/services/agent_metrics.py b/backend/app/services/agent_metrics.py index 569d3e6..bfcc0ee 100644 --- a/backend/app/services/agent_metrics.py +++ b/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. diff --git a/backend/app/services/agent_stream_broker.py b/backend/app/services/agent_stream_broker.py index ef1612a..2a326f1 100644 --- a/backend/app/services/agent_stream_broker.py +++ b/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``. diff --git a/backend/app/services/agent_stream_events.py b/backend/app/services/agent_stream_events.py index 60dc2c7..810d5d7 100644 --- a/backend/app/services/agent_stream_events.py +++ b/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, *, diff --git a/backend/app/services/claude_code.py b/backend/app/services/claude_code.py index 6545e86..9145013 100644 --- a/backend/app/services/claude_code.py +++ b/backend/app/services/claude_code.py @@ -1,7 +1,9 @@ import asyncio from collections.abc import AsyncIterator import contextlib -from datetime import datetime, timezone +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from enum import Enum import logging import os from pathlib import Path @@ -43,7 +45,18 @@ resume_prompt, todo_validator_prompt, ) -from app.schemas.agent import AgentErrorType, AgentResult +from app.schemas.agent import ( + AgentErrorType, + AgentQueryError, + AgentQueryProgress, + AgentResult, +) +from app.schemas.agent_error_events import AgentErrorEventKind, WorkspaceAgentState +from app.services.agent_error_event_recorder import ( + record_agent_error_event_safe, + set_workspace_retry_state_safe, +) +from app.services.git_utils import run_git from app.services.model_routing import ( apply_model_fallback_if_routing_failure, classify_error, @@ -94,8 +107,13 @@ class WorkspaceAbortedError(Exception): """ Raised by execute_all_phases when a workspace should be abandoned. Caught by ParallelAgentExecutor per-workspace; other workspaces continue. + ``phase`` carries the phase number the abort happened in, when known. """ + def __init__(self, message: str, phase: Optional[int] = None): + super().__init__(message) + self.phase = phase + def get_system_context_allowed_tools(workspace_path: str) -> List[str]: """ Get system context allowed tools for agents with standards directory. @@ -363,13 +381,101 @@ async def simple_query(): # There can be still TODOs not finalized by the agent, but still it can exit. We need to at least once resume with another call and add a message on top to check TODOs, and continue or decide to end the session. NO_RESUMING_RESULT = "Original run only, no resuming happened" +# Uniform wait schedule before every crash retry (the first retry always waits a +# minute). The zero-progress budget IS the schedule: each consecutive zero-progress +# crash consumes the next wait; once all six are spent the phase gives up +# (~1h49m of waits rides out a real outage). An attempt with saved progress +# resets the schedule to the start. +_CRASH_BACKOFF_MINUTES = (1, 3, 5, 10, 30, 60) + +# An attempt "saved progress" when it advanced git HEAD (committed work) or used +# at least this many tools — long legitimate pre-commit stretches (reading, +# planning, running tests) must not be misclassified as dead attempts. +RESUME_PROGRESS_MIN_TOOL_USES = 20 + + +class ResumeStopReason(str, Enum): + COMPLETED = "completed" + NO_PROGRESS_BUDGET = "no_progress_budget" + VALIDATOR_BUDGET = "validator_budget" + TOOL_CALL_FAILURE = "tool_call_failure" + +class IterationOutcome(str, Enum): + COMPLETE = "complete" + INCOMPLETE = "incomplete" + VALIDATOR_UNAVAILABLE = "validator_unavailable" + + +@dataclass +class ResumeGeneration: + """Retry state for one phase's agent attempts within a generation workspace. + + Single source of truth for the loop condition, the log lines, the durable + agent_error_events, and the TUI RETRYING badge — they must never disagree. + + Two independent budgets: + - crash schedule: consecutive zero-progress crashes each consume the next + `_CRASH_BACKOFF_MINUTES` wait; saved progress resets the schedule; no + total cap — a phase that keeps saving work may retry indefinitely. + - validator budget: clean-but-incomplete attempts (the anti-stuck contract), + unchanged at 2 resumes. + """ + + max_validator_resumes: int = 2 + backoff_minutes: tuple[int, ...] = _CRASH_BACKOFF_MINUTES + validator_resumes: int = 0 + crash_streak: int = 0 # consecutive zero-progress crashes + + def saved_progress(self, tool_uses: int, head_advanced: bool) -> bool: + return head_advanced or tool_uses >= RESUME_PROGRESS_MIN_TOOL_USES + + def record_crash(self, tool_uses: int, head_advanced: bool) -> Optional[timedelta]: + """Count one crashed attempt; return the wait before the next try. + + None means the zero-progress schedule is exhausted — give up. + """ + if self.saved_progress(tool_uses, head_advanced): + self.crash_streak = 0 + return timedelta(minutes=self.backoff_minutes[0]) + self.crash_streak += 1 + if self.crash_streak > len(self.backoff_minutes): + return None + return timedelta(minutes=self.backoff_minutes[self.crash_streak - 1]) + + def record_incomplete(self) -> bool: + """Count one clean-but-incomplete attempt; True while a resume is allowed.""" + self.validator_resumes += 1 + self.crash_streak = 0 # a clean run proved the connection is alive + return self.validator_resumes <= self.max_validator_resumes + + def crash_backoff_label(self) -> str: + """The "n/6" shown in events and the log line for the current streak.""" + return f"{min(self.crash_streak, len(self.backoff_minutes))}/{len(self.backoff_minutes)}" + + def describe(self) -> str: + return ( + f"crash backoff {self.crash_backoff_label()}, " + f"validator resumes {self.validator_resumes}/{self.max_validator_resumes}" + ) -async def _resume_iteration( - resume_count: int, - max_resume_attempts: int, - current_session_id: Optional[str], +def _format_wait(wait: timedelta) -> str: + minutes = int(wait.total_seconds() // 60) + return f"{minutes}m" if minutes else f"{int(wait.total_seconds())}s" + + +async def _read_workspace_head(workspace_path: str, logger: logging.Logger) -> Optional[str]: + """Current git HEAD of the workspace, or None when unreadable (never raises).""" + try: + return await run_git(Path(workspace_path), ["rev-parse", "HEAD"]) + except Exception: + logger.debug("Could not read workspace HEAD for progress detection", exc_info=True) + return None + + +async def _code_and_validate( + is_first: bool, outputs_dir: str, spec_path: str, system_prompt: str, @@ -382,63 +488,37 @@ async def _resume_iteration( logger: logging.Logger, max_buffer_size: int, provider: Optional[BaseProvider], - qa_results: str, + session_id: Optional[str], phase_number: Optional[int] = None, mcp_servers: Optional[Dict] = None, phase_kind: PhaseKind = PhaseKind.GENERATION, phase_num: int = 0, extra_env: Optional[Dict[str, str]] = None, -) -> tuple[Optional[AgentResult], str, bool]: - """ - Execute a single resume iteration. - - Returns: - Tuple of (resumed_result, qa_results, should_break) - - resumed_result: The result from the resume attempt, or None if failed - - qa_results: DISABLED - Updated QA results string - - should_break: True if the loop should break (completion or failure) +) -> tuple[AgentResult, IterationOutcome]: + """One attempt = coding query + validator query. + + The first attempt runs the phase prompt (optionally resuming session_id); + later attempts run the resume prompt with a deliberately fresh session — + this is a very long horizon task and agents tend to cheat and finish early. """ - # DISABLING QA AGENT DUE TO VERBOSITY AND TOKEN CONSUMPTION - qa_results = "" - - # logger.info(f"Running QA agent with session_id: {current_session_id}") - # qa_result: AgentResult = await agent_query( - # system_prompt=qa_agent_prompt(outputs_dir=outputs_dir), - # workspace_path=workspace_path, - # # Deliberately not resuming session, we want fresh context - # session_id=None, - # max_turns=max_turns, - # allowed_tools=allowed_tools, - # disallowed_tools=disallowed_tools, - # model=model, - # logger=logger, - # max_buffer_size=max_buffer_size, - # provider=provider, - # ) - - # if qa_result.result is None or qa_result.session_id is None: - # logger.warning(f"QA attempt {resume_count} returned None result or session_id, ignoring") - # else: - # qa_results = qa_result.result - - # logger.info(f"QA attempt {resume_count} returned session_id: {qa_result.session_id}") - # current_session_id = qa_result.session_id if qa_result.session_id else current_session_id - - logger.info(f"Attempting resume {resume_count}/{max_resume_attempts}") - TelemetryContext.set_workflow(TelemetryWorkflowLabel.phase(phase_kind, phase_num, "resume")) - resumed_result: AgentResult = await agent_query( - system_prompt=resume_prompt( + if is_first: + TelemetryContext.set_workflow(TelemetryWorkflowLabel.phase(phase_kind, phase_num, "coding")) + coding_prompt = system_prompt + else: + TelemetryContext.set_workflow(TelemetryWorkflowLabel.phase(phase_kind, phase_num, "resume")) + coding_prompt = resume_prompt( dev_system_prompt=system_prompt, workspace_root=workspace_path, outputs_dir=outputs_dir, spec_path=spec_path, - qa_results=qa_results, + qa_results="", phase_number=phase_number, - ), + ) + result: AgentResult = await agent_query( + system_prompt=coding_prompt, workspace_path=workspace_path, subagents=subagents, - # SUPER IMPORTANT: we want fresh context as this is very long horizon task and agents tend to cheat and finish early - session_id=None, + session_id=session_id if is_first else None, max_turns=max_turns, allowed_tools=allowed_tools, disallowed_tools=disallowed_tools, @@ -449,12 +529,9 @@ async def _resume_iteration( mcp_servers=mcp_servers, extra_env=extra_env, ) - - if resumed_result.result is None or resumed_result.session_id is None: - logger.error(f"Resume attempt {resume_count} returned None result or session_id, stopping") + if result.result is None or result.session_id is None: + logger.error("Coding query returned None result or session_id: %s", result) - # Check if the agent indicates completion - # Look for indicators that work is complete TelemetryContext.set_workflow(TelemetryWorkflowLabel.phase(phase_kind, phase_num, "validator")) validator_result: AgentResult = await agent_query( system_prompt=todo_validator_prompt( @@ -475,26 +552,136 @@ async def _resume_iteration( provider=provider, extra_env=extra_env, ) - if validator_result.result is None or validator_result.session_id is None: - logger.warning(f"Validator attempt {resume_count} returned None result or session_id, exiting resume loop") - return AgentResult(result=None, session_id=None), qa_results, True - - # Log validator result for debugging - validator_result_preview = (validator_result.result[:200] + "...") if validator_result.result and len(validator_result.result) > 200 else validator_result.result - logger.info(f"Validator result (attempt {resume_count}): {validator_result_preview}") + logger.warning("Validator returned None result — completion cannot be verified") + return result, IterationOutcome.VALIDATOR_UNAVAILABLE + preview = ( + (validator_result.result[:200] + "...") + if validator_result.result and len(validator_result.result) > 200 + else validator_result.result + ) + logger.info(f"Validator result: {preview}") if is_agent_complete(validator_result): - logger.info(f"Agent indicates completion after {resume_count} resume(s)") - return resumed_result, qa_results, True - + return result, IterationOutcome.COMPLETE + logger.warning(f"Validator indicates work is still incomplete. Validator said: {preview}.") + return result, IterationOutcome.INCOMPLETE + + +async def _crash_backoff_and_wait( + resume: ResumeGeneration, + error_message: str, + error_type: Optional[AgentErrorType], + tool_uses: int, + head_before: Optional[str], + workspace_path: str, + publisher, + logger: logging.Logger, + phase_number: Optional[int], + active_model: str, +) -> bool: + """Record one crashed/unverifiable attempt, surface it, and wait out the backoff. + + True → wait done, the loop should retry; False → the zero-progress schedule + is exhausted, give up. The event, the log line, the live-stream narrative and + the RETRYING badge all derive from the same ResumeGeneration state. + """ + head_after = await _read_workspace_head(workspace_path, logger) + head_advanced = bool(head_before and head_after and head_before != head_after) + made_progress = resume.saved_progress(tool_uses, head_advanced) + wait = resume.record_crash(tool_uses=tool_uses, head_advanced=head_advanced) + if wait is None: + return False + + next_attempt_at = datetime.now(timezone.utc) + wait + if made_progress: + saved = "a new commit" if head_advanced else f"{tool_uses} tool uses" + human = ( + f"Phase agent stopped after saved work ({saved}): {error_message[:150]} — " + f"retrying in {_format_wait(wait)} (backoff reset)." + ) + else: + human = ( + f"Phase agent failed before saving any work: {error_message[:150]} — " + f"retry {resume.crash_backoff_label()} in {_format_wait(wait)}." + ) logger.warning( - f"Validator indicates work is still incomplete after resume attempt {resume_count}. " - f"Validator said: {validator_result_preview}. " - f"This may cause additional resume iterations." + f"{human} ({resume.describe()}, next retry at {next_attempt_at:%H:%M:%S} UTC)" + ) + + await record_agent_error_event_safe( + AgentErrorEventKind.AGENT_CRASH, + human, + phase=phase_number, + error_type=error_type, + crash_backoff=resume.crash_backoff_label(), + next_attempt_at=next_attempt_at.isoformat(), + model=active_model, + ) + if publisher is not None: + publisher.publish_line_nowait("error", error_message) + publisher.publish_line_nowait( + "system", + f"waiting {_format_wait(wait)} before retry ({resume.describe()}), " + f"next attempt at {next_attempt_at:%H:%M} UTC", + ) + await set_workspace_retry_state_safe(WorkspaceAgentState.RETRYING) + await asyncio.sleep(wait.total_seconds()) + await set_workspace_retry_state_safe(None) + if publisher is not None: + publisher.publish_line_nowait("system", f"retrying now ({resume.describe()})") + return True + + +async def _report_gave_up( + stop_reason: ResumeStopReason, + resume: ResumeGeneration, + attempt: int, + phase_number: Optional[int], + last_error: Optional[str], + active_model: str, + publisher, + logger: logging.Logger, +) -> None: + """Log + record the give-up; the message derives from the stop reason so the + warning can never contradict the outcome (it no longer fires on success).""" + phase_label = f"Phase {phase_number}" if phase_number is not None else "The phase" + error_type = classify_error(last_error) if last_error else None + if stop_reason is ResumeStopReason.VALIDATOR_BUDGET: + kind = AgentErrorEventKind.PHASE_INCOMPLETE + human = ( + f"{phase_label} used all {resume.max_validator_resumes} resume attempts but the " + f"completion check is still unsatisfied (the agent may be stuck or the check too " + f"strict). The run continues; check PROGRESS.md and TODOs.md for what remains." + ) + elif stop_reason is ResumeStopReason.TOOL_CALL_FAILURE: + kind = AgentErrorEventKind.PHASE_GAVE_UP + human = f"{phase_label} stopped on a tool failure: {(last_error or 'unknown error')[:150]}" + else: # NO_PROGRESS_BUDGET + kind = AgentErrorEventKind.PHASE_GAVE_UP + cause = (last_error or "no result")[:150] + if error_type is AgentErrorType.MODEL_ROUTING_FAILURE: + remedy = ( + f"Model {active_model} keeps returning unusable responses — " + f"change the model in Settings, then retry." + ) + else: + remedy = "The workspace is paused; already-generated code is preserved." + human = ( + f"{phase_label} stopped after {len(resume.backoff_minutes)} zero-progress " + f"retries ({cause}). {remedy}" + ) + logger.warning(f"{human} [{stop_reason.value}, attempts={attempt}, {resume.describe()}]") + await record_agent_error_event_safe( + kind, + human, + phase=phase_number, + error_type=error_type, + crash_backoff=resume.crash_backoff_label(), + model=active_model, ) - logger.info(f"Resume attempt {resume_count} completed, returned session_id: {resumed_result.session_id}") - return resumed_result, qa_results, False + if publisher is not None: + publisher.publish_line_nowait("error", human) async def agent_query_with_resume( system_prompt: str, @@ -549,97 +736,25 @@ async def agent_query_with_resume( logger = logging.getLogger(__name__) phase_num = phase_number if phase_number is not None else 0 - qa_results = "" - result: Optional[AgentResult] = None - active_model = model _fallback_candidate = get_fallback_model(active_model) + publisher = build_stream_publisher_from_context() + resume = ResumeGeneration(max_validator_resumes=max_resume_attempts) - # Initial query - logger.info("Starting initial agent query with validator follow-up") - try: - TelemetryContext.set_workflow(TelemetryWorkflowLabel.phase(phase_kind, phase_num, "coding")) - result = await agent_query( - system_prompt=system_prompt, - workspace_path=workspace_path, - subagents=subagents, - session_id=session_id if session_id else None, - max_turns=max_turns, - allowed_tools=allowed_tools, - disallowed_tools=disallowed_tools, - model=active_model, - logger=logger, - max_buffer_size=max_buffer_size, - provider=provider, - mcp_servers=mcp_servers, - extra_env=extra_env, - ) - - if result.result is None or result.session_id is None: - logger.error(f"Initial agent query returned None or session_id is None: {result}") - - TelemetryContext.set_workflow(TelemetryWorkflowLabel.phase(phase_kind, phase_num, "validator")) - validator_result: AgentResult = await agent_query( - system_prompt=todo_validator_prompt( - workspace_root=workspace_path, - outputs_dir=outputs_dir, - spec_path=spec_path, - phase_number=phase_number, - ), - workspace_path=workspace_path, - # Deliberately not resuming session, we want fresh context - session_id=None, - max_turns=max_turns, - allowed_tools=allowed_tools, - disallowed_tools=disallowed_tools, - model=active_model, - logger=logger, - max_buffer_size=max_buffer_size, - provider=provider, - extra_env=extra_env, - ) - - if validator_result.result is None or validator_result.session_id is None: - logger.warning("Initial validator attempt returned None result, we have to continue to at least one resume iteration") - else: - # Log validator result for debugging - validator_result_preview = (validator_result.result[:200] + "...") if validator_result.result and len(validator_result.result) > 200 else validator_result.result - logger.info(f"Initial validator result: {validator_result_preview}") - - if is_agent_complete(validator_result): - logger.info("Agent indicates completion after initial query, no need to resume") - return result - else: - logger.warning( - f"Agent indicates incomplete after initial query. Validator said: {validator_result_preview}. " - f"This will trigger resume loop (max {max_resume_attempts} attempts)." - ) - - except Exception as e: - logger.error(f"[AGENT CRASH] Initial agent query failed with exception: {e}", exc_info=True) - err_str = str(e) - result = AgentResult(result=err_str, session_id=None, is_error=True) - active_model, _fallback_candidate = apply_model_fallback_if_routing_failure( - err_str, active_model, _fallback_candidate, logger, active_model - ) - - resume_count = 0 - current_session_id = result.session_id if result else None - logger.info(f"Initial agent query returned session_id: {current_session_id}") - - # Initialize with the initial result; will be updated in the loop - resumed_result: Optional[AgentResult] = result - - # Attempt to resume if needed - while resume_count < max_resume_attempts: - resume_count += 1 - logger.info(f"Starting resume iteration {resume_count}/{max_resume_attempts} for phase {phase_number}") + attempt = 0 + last_returned: Optional[AgentResult] = None + last_error: Optional[str] = None + stop_reason = ResumeStopReason.COMPLETED + while True: + attempt += 1 + logger.info( + f"Starting agent attempt {attempt} for phase {phase_number} ({resume.describe()})" + ) + head_before = await _read_workspace_head(workspace_path, logger) try: - resumed_result, qa_results, should_break = await _resume_iteration( - resume_count=resume_count, - max_resume_attempts=max_resume_attempts, - current_session_id=current_session_id, + result, outcome = await _code_and_validate( + is_first=attempt == 1, outputs_dir=outputs_dir, spec_path=spec_path, system_prompt=system_prompt, @@ -652,32 +767,119 @@ async def agent_query_with_resume( logger=logger, max_buffer_size=max_buffer_size, provider=provider, - qa_results=qa_results, + session_id=session_id, phase_number=phase_number, mcp_servers=mcp_servers, phase_kind=phase_kind, phase_num=phase_num, extra_env=extra_env, ) - - if should_break: - logger.info(f"Resume loop breaking after {resume_count} iteration(s) - work marked as complete") + last_returned = result + + if outcome is IterationOutcome.COMPLETE: + logger.info(f"Agent indicates completion after {attempt} attempt(s)") + break + if outcome is IterationOutcome.INCOMPLETE: + if resume.record_incomplete(): + logger.warning( + f"Work incomplete after attempt {attempt}; resuming ({resume.describe()})." + ) + continue + stop_reason = ResumeStopReason.VALIDATOR_BUDGET + break + # VALIDATOR_UNAVAILABLE: completion cannot be verified — treated as a + # crashed attempt, but progress from the coding query (commits) counts. + if not await _crash_backoff_and_wait( + resume=resume, + error_message="Completion check did not return a result; retrying the phase attempt.", + error_type=None, + tool_uses=0, + head_before=head_before, + workspace_path=workspace_path, + publisher=publisher, + logger=logger, + phase_number=phase_number, + active_model=active_model, + ): + stop_reason = ResumeStopReason.NO_PROGRESS_BUDGET + last_error = "Completion check repeatedly returned no result" break except Exception as e: - logger.error(f"[AGENT CRASH] Resume iteration {resume_count} failed with exception: {e}", exc_info=True) err_str = str(e) + error_type = classify_error(err_str) + progress: Optional[AgentQueryProgress] = getattr(e, "progress", None) + logger.error( + f"[AGENT CRASH] Attempt {attempt} failed with exception: {e}", exc_info=True + ) + if last_returned is None: + # Nothing ever returned — carry the error text so execute_all_phases + # can classify it (connection errors abort the workspace, #47). + last_returned = AgentResult(result=err_str, session_id=None, is_error=True) + last_error = err_str + + previous_model = active_model active_model, _fallback_candidate = apply_model_fallback_if_routing_failure( - err_str, active_model, _fallback_candidate, logger, f"resume {resume_count}" + err_str, active_model, _fallback_candidate, logger, f"attempt {attempt}" ) + if active_model != previous_model: + fallback_msg = ( + f"Model {previous_model} returned an unusable response " + f"({err_str[:120]}); switching this workspace to {active_model}." + ) + await record_agent_error_event_safe( + AgentErrorEventKind.MODEL_FALLBACK, + fallback_msg, + phase=phase_number, + error_type=error_type, + model=active_model, + previous_model=previous_model, + ) + if publisher is not None: + publisher.publish_line_nowait("system", fallback_msg) - if resume_count >= max_resume_attempts: - logger.warning( - f"Reached maximum resume attempts ({max_resume_attempts}) for phase {phase_number}. " - f"This may indicate the agent is stuck or the validator is too strict. " - f"Check PROGRESS.md and TODOs.md to verify actual completion status." + if error_type is AgentErrorType.TOOL_CALL_FAILURE: + # Not retryable in-loop — preserve the phase-level abort machinery. + stop_reason = ResumeStopReason.TOOL_CALL_FAILURE + break + + if not await _crash_backoff_and_wait( + resume=resume, + error_message=err_str, + error_type=error_type, + tool_uses=progress.num_tool_uses if progress else 0, + head_before=head_before, + workspace_path=workspace_path, + publisher=publisher, + logger=logger, + phase_number=phase_number, + active_model=active_model, + ): + stop_reason = ResumeStopReason.NO_PROGRESS_BUDGET + break + + if stop_reason is not ResumeStopReason.COMPLETED: + await _report_gave_up( + stop_reason=stop_reason, + resume=resume, + attempt=attempt, + phase_number=phase_number, + last_error=last_error, + active_model=active_model, + publisher=publisher, + logger=logger, ) - - final = resumed_result if resumed_result else AgentResult(result=None, session_id=None, is_error=True) + + # A crash-based give-up MUST surface is_error=True so execute_all_phases can + # classify it (connection errors abort the workspace, #47) and never checkpoint + # the phase as complete. Otherwise a single clean-but-incomplete attempt before + # an outage leaves last_returned holding that stale success (is_error=False), + # masking the crash: the phase is checkpointed and silently skipped on retry. + # Only COMPLETED and VALIDATOR_BUDGET (clean-but-incomplete, the anti-stuck + # contract) return the last coding result as-is. + if stop_reason in (ResumeStopReason.NO_PROGRESS_BUDGET, ResumeStopReason.TOOL_CALL_FAILURE): + final = AgentResult(result=last_error, session_id=None, is_error=True) + else: + final = last_returned or AgentResult(result=last_error, session_id=None, is_error=True) if active_model != model: final.active_model = active_model return final @@ -827,7 +1029,9 @@ async def agent_query( query_stream: AsyncIterator[Message] = query(prompt=system_prompt, options=options) - stream_metrics = ClaudeCodeSdkAgentMetrics() if telemetry.is_enabled() else None + # Always constructed (cheap in-process counters): progress_snapshot() must be + # available on the crash path even when telemetry capture is disabled. + stream_metrics = ClaudeCodeSdkAgentMetrics() # Best-effort live message tap for the TUI workspace drill-in. Derived from # TelemetryContext (generation_id + workspace name + workflow); None when that @@ -892,11 +1096,19 @@ async def agent_query( is_error=True, error_type="timeout", tool_usage_breakdown=stream_metrics.get_metrics() if stream_metrics is not None else None, ) + await record_agent_error_event_safe( + AgentErrorEventKind.AGENT_CRASH, + f"An agent step hit the {_AGENT_CALL_TIMEOUT_SECONDS // 3600}-hour time limit " + f"(likely a hanging subprocess); checking progress and retrying.", + model=final_model, + ) return AgentResult(result=None, session_id=None) - except Exception: + except Exception as e: lf_tracer.finalize_pending() lf_tracer.end_generation_on_error("exception") - raise + # Carry the observed progress so the resume loop can tell a crash after + # hours of saved work apart from one that never got started. + raise AgentQueryError(str(e), progress=stream_metrics.progress_snapshot()) from e finally: # Guarantee the SDK query generator — and its Claude Code CLI subprocess — is # torn down on every exit path, including task.cancel() (CancelledError) during a @@ -1284,13 +1496,15 @@ async def execute_all_phases( raise WorkspaceAbortedError( f"[{workspace_name}] Phase {phase_num} failed with tool-call incompatibility " f"(model={workspace.model}). Aborting workspace — other workspaces continue. " - f"Error: {phase_result.result}" + f"Error: {phase_result.result}", + phase=phase_num, ) if error_type == AgentErrorType.CONNECTION_ERROR: raise WorkspaceAbortedError( f"[{workspace_name}] Phase {phase_num} lost connection to the API " f"(model={workspace.model}). Aborting workspace — other workspaces continue. " - f"Error: {phase_result.result}" + f"Error: {phase_result.result}", + phase=phase_num, ) consecutive_errors += 1 logger.error( @@ -1301,7 +1515,8 @@ async def execute_all_phases( if consecutive_errors >= _MAX_CONSECUTIVE_PHASE_ERRORS: raise WorkspaceAbortedError( f"[{workspace_name}] Aborting after {consecutive_errors} consecutive phase errors " - f"(last at phase {phase_num}, model={workspace.model})." + f"(last at phase {phase_num}, model={workspace.model}).", + phase=phase_num, ) else: consecutive_errors = 0 @@ -1332,7 +1547,8 @@ async def execute_all_phases( except Exception as e: raise WorkspaceAbortedError( f"[{workspace_name}] Failed to write {phase_kind} {phase_num} checkpoint to Firestore: {e}. " - f"Aborting workspace — cannot track progress, retry would resume from wrong phase." + f"Aborting workspace — cannot track progress, retry would resume from wrong phase.", + phase=phase_num, ) from e else: logger.warning( diff --git a/backend/app/services/generation_session.py b/backend/app/services/generation_session.py index 613470e..3cdf27a 100644 --- a/backend/app/services/generation_session.py +++ b/backend/app/services/generation_session.py @@ -28,6 +28,7 @@ parse_checkpoint, parse_status, ) +from app.schemas.agent_error_events import AgentErrorEvent, WorkspaceAgentState from app.schemas.model_token_usage import ModelTokenUsage from app.schemas.workflow_usage_metrics import ( WORKFLOW_USAGE_METRICS_FIELD, @@ -941,6 +942,28 @@ async def update_workspace_phase( f"workspace {workspace_id}: phase {completed_phase}" ) + async def record_agent_error_event(self, generation_id: str, event: AgentErrorEvent) -> None: + """Append one durable agent error/warning event (bounded; see state machine).""" + await self._esm.record_agent_error_event( + generation_id, + event.model_dump(mode="json", exclude_none=True), + triggered_by=TriggeredBy.orchestrator_step("record_agent_error_event"), + ) + + async def set_workspace_agent_state( + self, + generation_id: str, + workspace_id: str, + state: Optional[WorkspaceAgentState], + ) -> None: + """Set or clear the per-workspace agent-state badge (retrying/aborted).""" + await self._esm.set_workspace_agent_state( + generation_id, + workspace_id=workspace_id, + state=state.value if state is not None else None, + triggered_by=TriggeredBy.orchestrator_step("set_workspace_agent_state"), + ) + async def save_e2e_plan( self, generation_id: str, diff --git a/backend/app/services/generation_workflow_runner.py b/backend/app/services/generation_workflow_runner.py index 23a9222..ae7c8f7 100644 --- a/backend/app/services/generation_workflow_runner.py +++ b/backend/app/services/generation_workflow_runner.py @@ -313,3 +313,5 @@ async def rerun_generation_session( 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) diff --git a/backend/app/services/parallel_executor.py b/backend/app/services/parallel_executor.py index f04070d..b539513 100644 --- a/backend/app/services/parallel_executor.py +++ b/backend/app/services/parallel_executor.py @@ -8,7 +8,12 @@ from typing import Awaitable, Callable, List, Optional from app.schemas.agent import AgentErrorType, AgentResult +from app.schemas.agent_error_events import AgentErrorEventKind, WorkspaceAgentState from app.schemas.workspace import WorkspaceSettings +from app.services.agent_error_event_recorder import ( + record_agent_error_event_safe, + set_workspace_retry_state_safe, +) from app.services.claude_code import agent_query from app.services.model_routing import classify_error from app.services.workspace_manager import WorkspaceManager @@ -178,6 +183,24 @@ async def _execute_single( exc_info=True, ) + # The single recording site for workspace aborts: every abort path + # (classified aborts, checkpoint-write failure, deploy failures) + # funnels through this except-branch exactly once. Use the settings + # name (the workspace_phases key, e.g. "ws-03-1"), not the executor's + # positional logging label ("workspace-1"). + event_workspace_id = workspace.name or name + await record_agent_error_event_safe( + AgentErrorEventKind.WORKSPACE_ABORTED, + error_msg, + workspace_id=event_workspace_id, + phase=getattr(e, "phase", None), + error_type=error_type, + model=workspace.model, + ) + await set_workspace_retry_state_safe( + WorkspaceAgentState.ABORTED, workspace_id=event_workspace_id + ) + return ParallelAgentResult( workspace_name=name, workspace_settings=workspace, diff --git a/backend/app/services/workflow_steps.py b/backend/app/services/workflow_steps.py index fa5b824..fcbf469 100644 --- a/backend/app/services/workflow_steps.py +++ b/backend/app/services/workflow_steps.py @@ -68,7 +68,9 @@ get_common_allowed_tools, ) from app.services.generation_session_output_protection import protect_workspace_after_generation -from app.services.parallel_executor import ParallelAgentExecutor +from app.schemas.agent import AgentErrorType +from app.services.model_routing import classify_error +from app.services.parallel_executor import ParallelAgentExecutor, ParallelAgentResult from app.services.planning_parser import ( construct_planning_result, parse_applicable_agent_mcps, @@ -408,6 +410,12 @@ async def build_workflow_context( TelemetryContext.set_agent_query_totals_handler( generation_session_service.add_agent_query_token_usage ) + TelemetryContext.set_agent_error_event_handler( + generation_session_service.record_agent_error_event + ) + TelemetryContext.set_workspace_agent_state_handler( + generation_session_service.set_workspace_agent_state + ) return WorkflowContext( request=request, @@ -968,6 +976,87 @@ async def run_contract_validator(ctx: WorkflowContext) -> None: await _update_plan_firestore(ctx, plan_type, new_plan) +# Variance-reduced estimates need at least two samples — a run that ends with a +# single surviving workspace is red + retry-able, not silently degraded. +MIN_WORKSPACES_FOR_VARIANCE = 2 + + +class InsufficientWorkspacesError(RuntimeError): + """Too few workspaces survived a fan-out step for the run to be useful.""" + + +def _friendly_workspace_failure(error: Optional[str]) -> str: + """One human phrase per workspace failure, derived from classify_error (SSOT).""" + error_type = classify_error(error or "") + if error_type == AgentErrorType.CONNECTION_ERROR: + return "connection to the model API was lost" + if error_type == AgentErrorType.MODEL_ROUTING_FAILURE: + return "the model kept returning unusable responses" + if error_type == AgentErrorType.TOOL_CALL_FAILURE: + return "the model does not support the required tools" + return (error or "unknown error")[:160] + + +def _raise_if_insufficient_workspaces( + results: list, + *, + step_label: str, + logger: logging.Logger, + min_survivors: int, + requirement_reason: str, +) -> None: + """Fail the run when fewer than ``min(min_survivors, total)`` workspaces survived. + + ``ParallelAgentExecutor`` converts every per-workspace error into + ``success=False`` and never raises, so without this check a total (or + near-total) failure lets the orchestrator advance the checkpoint and report + success with an empty estimate. ``min(..., total)`` keeps legitimate + single-workspace runs green; partial failure above the bar continues with + the survivors (PR #47 intent: never fail a run that still has usable + samples). Raising here never touches workspaces (they stay ALLOCATED with + code intact — Commandment II) and the generation checkpoint was not + advanced, so retry_generation resumes each failed workspace mid-plan. + """ + parallel = [r for r in (results or []) if isinstance(r, ParallelAgentResult)] + if not parallel: + return + failures = [r for r in parallel if not r.success] + if not failures: + return + survivors = len(parallel) - len(failures) + required = min(min_survivors, len(parallel)) + detail = "; ".join( + f"{r.workspace_name}: {_friendly_workspace_failure(r.error)}" for r in failures + ) + if survivors >= required: + logger.warning( + "%s: %d/%d workspaces failed but %d survived — continuing with survivors. Failed: %s", + step_label, len(failures), len(parallel), survivors, detail, + ) + return + + failure_types = {classify_error(r.error or "") for r in failures} + if failure_types == {AgentErrorType.CONNECTION_ERROR}: + hint = ( + "Lost connection to the model API — likely a network interruption " + "(e.g. the machine went offline). " + ) + elif failure_types == {AgentErrorType.MODEL_ROUTING_FAILURE}: + hint = ( + "The model kept returning unusable responses — change the model in " + "Settings before retrying. " + ) + else: + hint = "" + raise InsufficientWorkspacesError( + f"Only {survivors} of {len(parallel)} workspaces completed {step_label} — " + f"at least {required} {'is' if required == 1 else 'are'} required " + f"{requirement_reason}. Failed: {detail}. {hint}" + f"Your generated code is saved — run retry_generation to resume the failed " + f"workspaces from their last completed phase." + ) + + async def run_generation_phases(ctx: WorkflowContext) -> None: """ Run parallel code generation phases across all workspaces. @@ -1060,6 +1149,14 @@ async def agent_fn(workspace: WorkspaceSettings, manager: WorkspaceManager, logg exc_info=True, ) + _raise_if_insufficient_workspaces( + ctx.generation_result, + step_label="code generation", + logger=ctx.logger, + min_survivors=MIN_WORKSPACES_FOR_VARIANCE, + requirement_reason="to compute variance-reduced estimates", + ) + async def _build_deploy_github_context(generation_session_service, workspace_id: str) -> DeployGithubContext: """Return deploy context for the agent prompt. @@ -1221,7 +1318,19 @@ async def agent_fn( enabled_mcps=ctx.enabled_mcps, ) - await executor.execute_parallel(workspaces=ctx.workspaces, agent_fn=agent_fn) + deploy_results = await executor.execute_parallel( + workspaces=ctx.workspaces, agent_fn=agent_fn + ) + # Deploy keeps the all-failed rule (min 1 survivor): failed deploys don't + # invalidate estimation of already-committed code, but zero successful + # deploys when E2E was requested is still a failed run. + _raise_if_insufficient_workspaces( + deploy_results, + step_label="deploy & E2E", + logger=ctx.logger, + min_survivors=1, + requirement_reason="for a usable deployment", + ) async def run_archive_outputs_step(ctx: WorkflowContext) -> None: diff --git a/backend/app/state/generation_session_state_machine.py b/backend/app/state/generation_session_state_machine.py index 45d127e..06c4d4a 100644 --- a/backend/app/state/generation_session_state_machine.py +++ b/backend/app/state/generation_session_state_machine.py @@ -27,6 +27,10 @@ logger = logging.getLogger(__name__) +# Bounded per-session log of agent error/warning events (newest kept). +# ~400 bytes/event worst case -> well under the 1 MB Firestore doc limit. +_MAX_AGENT_ERROR_EVENTS = 200 + class GenerationSessionStateMachine: """ @@ -336,6 +340,79 @@ async def record_deployment_phase_progress( field_name="workspace_phases_deployment", phase_label="deploy phase", ) + async def record_agent_error_event( + self, + generation_id: str, + event: dict, + triggered_by: str, + ) -> dict: + """ + RUNNING → RUNNING: append one agent error/warning event (bounded log). + + Side-effects: + - Appends to agent_error_events, keeping the newest _MAX_AGENT_ERROR_EVENTS + - Updates last_activity_at (a crash-retrying workspace is alive — keeps + stuck_running_detector honest during backoff waits) + + Not a state transition: status, checkpoint, and state_history are untouched + (Commandment IX governs transitions; events carry their own bounded audit). + This is the ONLY correct way to write agent_error_events (Commandment VII). + + Concurrency: the append is read-modify-write in the same consistency + envelope as state_history appends in _record_phase_progress_for_field — + a concurrent append may be lost under contention; acceptable for + advisory data. + """ + doc = await self._db.get_generation_session(generation_id) + self._require_running(doc, "record_agent_error_event", generation_id) + + events = list(doc.get("agent_error_events") or []) + events.append(event) + update = { + "agent_error_events": events[-_MAX_AGENT_ERROR_EVENTS:], + "last_activity_at": datetime.now(timezone.utc), + } + await self._db.update_generation_session(generation_id, update) + logger.info( + "generation %s agent event %s recorded for workspace %s (by %s)", + generation_id, event.get("kind"), event.get("workspace_id"), triggered_by, + ) + return update + + async def set_workspace_agent_state( + self, + generation_id: str, + workspace_id: str, + state: Optional[str], + triggered_by: str, + ) -> dict: + """ + RUNNING → RUNNING: set or clear the transient per-workspace agent badge. + + ``state`` is a WorkspaceAgentState value ("retrying"/"aborted"); None + removes the key (workspace running normally). Stored on + workspace_phases[workspace_id]["agent_state"] so the /status per-workspace + view ships it without a new top-level field. Only writer: this method + (Commandment VII). + """ + doc = await self._db.get_generation_session(generation_id) + self._require_running(doc, "set_workspace_agent_state", generation_id) + + phases_dict = dict(doc.get("workspace_phases") or {}) + entry = dict(phases_dict.get(workspace_id) or {}) + if state is None: + entry.pop("agent_state", None) + else: + entry["agent_state"] = state + phases_dict[workspace_id] = entry + update = {"workspace_phases": phases_dict} + await self._db.update_generation_session(generation_id, update) + logger.info( + "generation %s workspace %s agent_state=%s (by %s)", + generation_id, workspace_id, state, triggered_by, + ) + return update + async def save_generation_plan( self, generation_id: str, @@ -524,10 +601,23 @@ async def reset_for_retry( if retry_count >= max_retries: raise MaxRetriesExceededError(generation_id, retry_count, max_retries) + # A retry restarts every allocated workspace, so wipe the transient + # per-workspace agent_state badge (retrying/aborted) left by the previous + # run — otherwise stale ABORTED/RETRYING markers linger on any workspace + # that doesn't happen to re-enter the crash/backoff path this run. All + # other workspace_phases fields (last_completed_phase, planning_data) are + # preserved so the retry still resumes from the right phase. + phases = doc.get("workspace_phases") or {} + cleaned_phases = { + ws: {k: v for k, v in (entry or {}).items() if k != "agent_state"} + for ws, entry in phases.items() + } + extra_fields = { "retry_count": retry_count + 1, "error": None, "shutdown_interrupted": False, + "workspace_phases": cleaned_phases, } return await self._transition( generation_id, "reset_for_retry", diff --git a/backend/test/api/test_generation_sessions.py b/backend/test/api/test_generation_sessions.py index 28921e8..445cd33 100644 --- a/backend/test/api/test_generation_sessions.py +++ b/backend/test/api/test_generation_sessions.py @@ -1987,3 +1987,81 @@ async def test_unknown_api_key_returns_404(self): with pytest.raises(HTTPException) as exc: await list_generation_sessions(request=self._request("nope"), db=InMemoryDatabase(), limit=50) assert exc.value.status_code == 404 + + +class TestStatusAgentErrorEvents: + """The /status payload carries the durable agent error/warning events.""" + + def _get_status(self, client, mock_generation_session_service, doc): + from app.schemas.generation_workflow_enums import GenerationCheckpoint + + mock_generation_session_service.get_generation_session_status = AsyncMock(return_value=doc) + mock_generation_session_service.get_checkpoint = Mock( + return_value=GenerationCheckpoint.CONTRACT_VALIDATED + ) + mock_generation_session_service.get_current_phase_name = Mock(return_value="x") + response = client.get( + f"/api/v1/generation-sessions/{doc['generation_id']}/status", + headers={"X-API-Key": "test-key"}, + ) + assert response.status_code == 200 + return response.json() + + def _doc(self, **extra): + return { + "generation_id": "est-ev-1", + "status": "running", + "user_email": "u@example.com", + "workspace_ids": ["ws-1"], + "parameters": {"workspace_count": 1}, + "progress": {}, + "error": None, + **extra, + } + + def _event(self, n): + return { + "at": f"2026-07-23T10:00:{n:02d}+00:00", + "workspace_id": "ws-1", + "kind": "agent_crash", + "message": f"crash {n}", + } + + def test_status_includes_recent_events_newest_kept( + self, client, mock_generation_session_service + ): + doc = self._doc(agent_error_events=[self._event(n) for n in range(3)]) + data = self._get_status(client, mock_generation_session_service, doc) + assert [e["message"] for e in data["agent_error_events"]] == [ + "crash 0", "crash 1", "crash 2", + ] + + def test_status_caps_events_to_limit(self, client, mock_generation_session_service): + from app.api.v1.generation_sessions import _AGENT_ERROR_EVENTS_STATUS_LIMIT + + doc = self._doc( + agent_error_events=[self._event(n % 60) for n in range(_AGENT_ERROR_EVENTS_STATUS_LIMIT + 10)] + ) + data = self._get_status(client, mock_generation_session_service, doc) + assert len(data["agent_error_events"]) == _AGENT_ERROR_EVENTS_STATUS_LIMIT + + def test_status_omits_key_for_docs_without_events( + self, client, mock_generation_session_service + ): + """Old docs keep their exact response shape (compat).""" + data = self._get_status(client, mock_generation_session_service, self._doc()) + assert "agent_error_events" not in data + + def test_workspace_view_passes_through_agent_state( + self, client, mock_generation_session_service + ): + doc = self._doc( + workspace_phases={ + "ws-1": {"last_completed_phase": 3, "total_phases": 5, "agent_state": "retrying"}, + "ws-2": {"last_completed_phase": 2, "total_phases": 5}, + }, + workspace_ids=["ws-1", "ws-2"], + ) + data = self._get_status(client, mock_generation_session_service, doc) + assert data["workspace_phases"]["ws-1"]["agent_state"] == "retrying" + assert "agent_state" not in data["workspace_phases"]["ws-2"] diff --git a/backend/test/services/test_agent_error_event_recorder.py b/backend/test/services/test_agent_error_event_recorder.py new file mode 100644 index 0000000..34ed4a0 --- /dev/null +++ b/backend/test/services/test_agent_error_event_recorder.py @@ -0,0 +1,130 @@ +"""Tests for the best-effort agent error event recorder. + +Contract: record via TelemetryContext handlers when a workflow registered +them; silent no-op without handler/identity; never raise into the caller. +""" + +from unittest.mock import AsyncMock + +import pytest + +from app.core.telemetry_context import TelemetryContext +from app.schemas.agent import AgentErrorType +from app.schemas.agent_error_events import ( + AgentErrorEventKind, + WorkspaceAgentState, +) +from app.services.agent_error_event_recorder import ( + record_agent_error_event_safe, + set_workspace_retry_state_safe, +) + + +@pytest.fixture(autouse=True) +def _clean_context(): + TelemetryContext.clear_context() + yield + TelemetryContext.clear_context() + + +class TestRecordAgentErrorEventSafe: + @pytest.mark.asyncio + async def test_records_via_context_handler(self): + handler = AsyncMock() + TelemetryContext.set_agent_error_event_handler(handler) + TelemetryContext.set_user_context(generation_id="est-1", workspace_name="ws-1") + + await record_agent_error_event_safe( + AgentErrorEventKind.AGENT_CRASH, + "Connection to the model API was lost.", + phase=12, + error_type=AgentErrorType.CONNECTION_ERROR, + crash_backoff="1/6", + ) + + handler.assert_awaited_once() + generation_id, event = handler.await_args.args + assert generation_id == "est-1" + assert event.workspace_id == "ws-1" + assert event.kind == AgentErrorEventKind.AGENT_CRASH + assert event.phase == 12 + assert event.error_type == AgentErrorType.CONNECTION_ERROR + assert event.crash_backoff == "1/6" + assert event.at # stamped + + @pytest.mark.asyncio + async def test_explicit_workspace_id_wins_over_context(self): + handler = AsyncMock() + TelemetryContext.set_agent_error_event_handler(handler) + TelemetryContext.set_user_context(generation_id="est-1", workspace_name="ws-ctx") + + await record_agent_error_event_safe( + AgentErrorEventKind.WORKSPACE_ABORTED, "gone", workspace_id="ws-explicit" + ) + assert handler.await_args.args[1].workspace_id == "ws-explicit" + + @pytest.mark.asyncio + async def test_noop_without_handler_or_identity(self): + # No handler at all + TelemetryContext.set_user_context(generation_id="est-1", workspace_name="ws-1") + await record_agent_error_event_safe(AgentErrorEventKind.AGENT_CRASH, "x") + + # Handler but no generation_id + TelemetryContext.clear_context() + handler = AsyncMock() + TelemetryContext.set_agent_error_event_handler(handler) + TelemetryContext.set_user_context(workspace_name="ws-1") + await record_agent_error_event_safe(AgentErrorEventKind.AGENT_CRASH, "x") + + # Handler + generation_id but no workspace + TelemetryContext.set_user_context(generation_id="est-1") + TelemetryContext.set_workspace_name("") + await record_agent_error_event_safe(AgentErrorEventKind.AGENT_CRASH, "x") + + handler.assert_not_awaited() + + @pytest.mark.asyncio + async def test_swallows_handler_exception(self): + handler = AsyncMock(side_effect=RuntimeError("firestore down")) + TelemetryContext.set_agent_error_event_handler(handler) + TelemetryContext.set_user_context(generation_id="est-1", workspace_name="ws-1") + + await record_agent_error_event_safe(AgentErrorEventKind.AGENT_CRASH, "x") + handler.assert_awaited_once() # attempted, exception swallowed + + @pytest.mark.asyncio + async def test_truncates_long_messages(self): + handler = AsyncMock() + TelemetryContext.set_agent_error_event_handler(handler) + TelemetryContext.set_user_context(generation_id="est-1", workspace_name="ws-1") + + await record_agent_error_event_safe(AgentErrorEventKind.AGENT_CRASH, "y" * 1000) + assert len(handler.await_args.args[1].message) == 300 + + +class TestSetWorkspaceRetryStateSafe: + @pytest.mark.asyncio + async def test_sets_and_clears_via_handler(self): + handler = AsyncMock() + TelemetryContext.set_workspace_agent_state_handler(handler) + TelemetryContext.set_user_context(generation_id="est-1", workspace_name="ws-1") + + await set_workspace_retry_state_safe(WorkspaceAgentState.RETRYING) + assert handler.await_args.args == ("est-1", "ws-1", WorkspaceAgentState.RETRYING) + + await set_workspace_retry_state_safe(None) + assert handler.await_args.args == ("est-1", "ws-1", None) + + @pytest.mark.asyncio + async def test_noop_without_handler(self): + TelemetryContext.set_user_context(generation_id="est-1", workspace_name="ws-1") + await set_workspace_retry_state_safe(WorkspaceAgentState.RETRYING) # no raise + + @pytest.mark.asyncio + async def test_swallows_handler_exception(self): + handler = AsyncMock(side_effect=RuntimeError("boom")) + TelemetryContext.set_workspace_agent_state_handler(handler) + TelemetryContext.set_user_context(generation_id="est-1", workspace_name="ws-1") + + await set_workspace_retry_state_safe(WorkspaceAgentState.ABORTED) + handler.assert_awaited_once() diff --git a/backend/test/services/test_generation_session.py b/backend/test/services/test_generation_session.py index e573966..7cab466 100644 --- a/backend/test/services/test_generation_session.py +++ b/backend/test/services/test_generation_session.py @@ -18,6 +18,11 @@ import pytest from app.database.memory import InMemoryDatabase +from app.schemas.agent_error_events import ( + AgentErrorEvent, + AgentErrorEventKind, + WorkspaceAgentState, +) from app.schemas.generation_workflow_enums import ( GenerationCheckpoint, GenerationStatus, @@ -1449,3 +1454,45 @@ async def test_skips_noop_deltas(self, generation_session_service: GenerationSes doc = db.get(COL_GENERATION_SESSIONS, est_id) assert "model_usage" not in doc assert WORKFLOW_USAGE_METRICS_FIELD not in doc + + +class TestAgentErrorEventDelegation: + """Service methods are thin delegates to the state machine (Commandment VII).""" + + @pytest.mark.asyncio + async def test_record_agent_error_event_delegates_to_state_machine( + self, generation_session_service + ): + generation_session_service._esm.record_agent_error_event = AsyncMock() + event = AgentErrorEvent( + at="2026-07-23T10:00:00+00:00", + workspace_id="ws-1", + kind=AgentErrorEventKind.AGENT_CRASH, + message="boom", + ) + await generation_session_service.record_agent_error_event("est-1", event) + + call = generation_session_service._esm.record_agent_error_event.await_args + assert call.args[0] == "est-1" + payload = call.args[1] + assert payload["kind"] == "agent_crash" + assert payload["workspace_id"] == "ws-1" + # exclude_none keeps the stored doc compact + assert "phase" not in payload + + @pytest.mark.asyncio + async def test_set_workspace_agent_state_delegates_and_serializes( + self, generation_session_service + ): + generation_session_service._esm.set_workspace_agent_state = AsyncMock() + + await generation_session_service.set_workspace_agent_state( + "est-1", "ws-1", WorkspaceAgentState.RETRYING + ) + kwargs = generation_session_service._esm.set_workspace_agent_state.await_args.kwargs + assert kwargs["workspace_id"] == "ws-1" + assert kwargs["state"] == "retrying" + + await generation_session_service.set_workspace_agent_state("est-1", "ws-1", None) + kwargs = generation_session_service._esm.set_workspace_agent_state.await_args.kwargs + assert kwargs["state"] is None diff --git a/backend/test/state/test_generation_session_state_machine.py b/backend/test/state/test_generation_session_state_machine.py index effea28..3807846 100644 --- a/backend/test/state/test_generation_session_state_machine.py +++ b/backend/test/state/test_generation_session_state_machine.py @@ -476,6 +476,34 @@ async def test_reset_clears_shutdown_interrupted_marker(self, db): await esm.reset_for_retry("est-1", triggered_by="system:server_boot_recovery") assert db.get_generation_session_data("est-1")["shutdown_interrupted"] is False + @pytest.mark.asyncio + async def test_reset_clears_agent_state_on_all_workspaces_preserving_phase(self, db): + """A retry must wipe the transient RETRYING/ABORTED badge from EVERY workspace + (regression: stale badges lingered on workspaces that didn't re-enter the + crash/backoff path), while preserving last_completed_phase so the retry + resumes from the right place.""" + esm = make_esm(db) + db.seed_generation_session("est-1", { + "status": GenerationStatus.FAILED, + "retry_count": 0, + "max_retries": 3, + "workspace_ids": ["ws-01-1", "ws-01-2", "ws-01-3"], + "workspace_phases": { + "ws-01-1": {"last_completed_phase": 2, "total_phases": 24, "agent_state": "aborted"}, + "ws-01-2": {"last_completed_phase": 0, "total_phases": 24, "agent_state": "aborted"}, + "ws-01-3": {"last_completed_phase": 1, "total_phases": 24}, + }, + "state_history": [], + }) + await esm.reset_for_retry("est-1", triggered_by="api:manual_retry") + phases = db.get_generation_session_data("est-1")["workspace_phases"] + # No workspace keeps a stale agent_state badge... + assert all("agent_state" not in entry for entry in phases.values()) + # ...and phase progress is untouched. + assert phases["ws-01-1"]["last_completed_phase"] == 2 + assert phases["ws-01-2"]["last_completed_phase"] == 0 + assert phases["ws-01-3"]["total_phases"] == 24 + class TestComplete: @pytest.mark.asyncio @@ -1201,3 +1229,136 @@ async def test_raises_when_not_running(self, db): await set_workspace_model_override( "gen-1", "ws-a", "haiku", triggered_by="test", db=db ) + + +class TestRecordAgentErrorEvent: + def _event(self, n=0): + return { + "at": f"2026-07-23T10:00:{n:02d}+00:00", + "workspace_id": "ws-1", + "kind": "agent_crash", + "message": f"crash {n}", + } + + @pytest.mark.asyncio + async def test_appends_event(self, db): + esm = make_esm(db) + db.seed_generation_session("est-1", { + "status": GenerationStatus.RUNNING, + "state_history": [], + }) + await esm.record_agent_error_event("est-1", self._event(), triggered_by="test") + data = db.get_generation_session_data("est-1") + assert len(data["agent_error_events"]) == 1 + assert data["agent_error_events"][0]["kind"] == "agent_crash" + + @pytest.mark.asyncio + async def test_bounds_to_max_events_keeping_newest(self, db): + from app.state.generation_session_state_machine import _MAX_AGENT_ERROR_EVENTS + + esm = make_esm(db) + db.seed_generation_session("est-1", { + "status": GenerationStatus.RUNNING, + "agent_error_events": [self._event(n) for n in range(_MAX_AGENT_ERROR_EVENTS)], + "state_history": [], + }) + await esm.record_agent_error_event( + "est-1", {**self._event(), "message": "newest"}, triggered_by="test" + ) + data = db.get_generation_session_data("est-1") + assert len(data["agent_error_events"]) == _MAX_AGENT_ERROR_EVENTS + assert data["agent_error_events"][-1]["message"] == "newest" + # Oldest entry (n=0) fell off the front. + assert data["agent_error_events"][0]["message"] == "crash 1" + + @pytest.mark.asyncio + async def test_updates_last_activity_at(self, db): + esm = make_esm(db) + db.seed_generation_session("est-1", { + "status": GenerationStatus.RUNNING, + "state_history": [], + }) + await esm.record_agent_error_event("est-1", self._event(), triggered_by="test") + data = db.get_generation_session_data("est-1") + assert "last_activity_at" in data + + @pytest.mark.asyncio + async def test_requires_running(self, db): + esm = make_esm(db) + db.seed_generation_session("est-1", { + "status": GenerationStatus.FAILED, + "state_history": [], + }) + with pytest.raises(InvalidGenerationSessionStateError): + await esm.record_agent_error_event("est-1", self._event(), triggered_by="test") + + @pytest.mark.asyncio + async def test_does_not_touch_status_checkpoint_or_state_history(self, db): + """Events are not state transitions (Commandment IX applies to transitions).""" + esm = make_esm(db) + db.seed_generation_session("est-1", { + "status": GenerationStatus.RUNNING, + "checkpoint": "generating_code", + "state_history": [{"status": "running"}], + }) + await esm.record_agent_error_event("est-1", self._event(), triggered_by="test") + data = db.get_generation_session_data("est-1") + assert data["status"] == GenerationStatus.RUNNING + assert data["checkpoint"] == "generating_code" + assert data["state_history"] == [{"status": "running"}] + + +class TestSetWorkspaceAgentState: + @pytest.mark.asyncio + async def test_sets_and_clears_agent_state(self, db): + esm = make_esm(db) + db.seed_generation_session("est-1", { + "status": GenerationStatus.RUNNING, + "workspace_phases": {"ws-1": {"last_completed_phase": 3, "total_phases": 5}}, + "state_history": [], + }) + await esm.set_workspace_agent_state("est-1", "ws-1", "retrying", triggered_by="test") + data = db.get_generation_session_data("est-1") + assert data["workspace_phases"]["ws-1"]["agent_state"] == "retrying" + + await esm.set_workspace_agent_state("est-1", "ws-1", None, triggered_by="test") + data = db.get_generation_session_data("est-1") + assert "agent_state" not in data["workspace_phases"]["ws-1"] + + @pytest.mark.asyncio + async def test_preserves_other_workspace_phase_fields(self, db): + esm = make_esm(db) + db.seed_generation_session("est-1", { + "status": GenerationStatus.RUNNING, + "workspace_phases": { + "ws-1": {"last_completed_phase": 3, "total_phases": 5}, + "ws-2": {"last_completed_phase": 1, "total_phases": 5}, + }, + "state_history": [], + }) + await esm.set_workspace_agent_state("est-1", "ws-1", "aborted", triggered_by="test") + data = db.get_generation_session_data("est-1") + assert data["workspace_phases"]["ws-1"]["last_completed_phase"] == 3 + assert data["workspace_phases"]["ws-2"] == {"last_completed_phase": 1, "total_phases": 5} + + @pytest.mark.asyncio + async def test_creates_entry_for_unknown_workspace(self, db): + esm = make_esm(db) + db.seed_generation_session("est-1", { + "status": GenerationStatus.RUNNING, + "workspace_phases": {}, + "state_history": [], + }) + await esm.set_workspace_agent_state("est-1", "ws-9", "retrying", triggered_by="test") + data = db.get_generation_session_data("est-1") + assert data["workspace_phases"]["ws-9"]["agent_state"] == "retrying" + + @pytest.mark.asyncio + async def test_requires_running(self, db): + esm = make_esm(db) + db.seed_generation_session("est-1", { + "status": GenerationStatus.FAILED, + "state_history": [], + }) + with pytest.raises(InvalidGenerationSessionStateError): + await esm.set_workspace_agent_state("est-1", "ws-1", "retrying", triggered_by="test") diff --git a/backend/test/test_agent_query_with_resume.py b/backend/test/test_agent_query_with_resume.py new file mode 100644 index 0000000..01b9f94 --- /dev/null +++ b/backend/test/test_agent_query_with_resume.py @@ -0,0 +1,350 @@ +"""agent_query_with_resume: the unified attempt loop. + +Contract under test (single source: ResumeGeneration): +- crashes with saved progress keep retrying (incident regression — the old code + died after 2 resume attempts no matter how much work each one saved); +- zero-progress crashes wait out the 1/3/5/10/30/60-minute schedule then give up; +- validator-driven resumes keep their old budget of 2; +- a dead validator retries instead of silently marking the phase complete; +- completion on the final allowed attempt emits NO give-up warning/event; +- crash/give-up/model-fallback events are recorded; RETRYING is set around waits; +- everything is silent (but functional) when no recorder is configured. +""" + +from unittest.mock import AsyncMock, Mock, patch + +import pytest + +from app.core.telemetry_context import TelemetryContext +from app.schemas.agent import AgentQueryError, AgentQueryProgress, AgentResult +from app.schemas.agent_error_events import AgentErrorEventKind, WorkspaceAgentState +from app.services.claude_code import agent_query_with_resume + +_SOCKET_ERR = "API Error: The socket connection was closed unexpectedly." +_COMPLETE = AgentResult(result="All subtasks done.\nWORK_COMPLETE", session_id="v1") +_INCOMPLETE = AgentResult(result="Remaining: wire the API route.", session_id="v1") +_CODING_OK = AgentResult(result="did some work", session_id="s1") + + +def _crash(tool_uses: int) -> AgentQueryError: + return AgentQueryError(_SOCKET_ERR, progress=AgentQueryProgress(num_messages=tool_uses, num_tool_uses=tool_uses)) + + +@pytest.fixture(autouse=True) +def _clean_context(): + TelemetryContext.clear_context() + yield + TelemetryContext.clear_context() + + +@pytest.fixture +def recorders(): + """Wire event/state handlers + identity so events are observable.""" + events = AsyncMock() + states = AsyncMock() + TelemetryContext.set_agent_error_event_handler(events) + TelemetryContext.set_workspace_agent_state_handler(states) + TelemetryContext.set_user_context(generation_id="est-1", workspace_name="ws-1") + return events, states + + +def _recorded_kinds(events: AsyncMock) -> list[AgentErrorEventKind]: + return [call.args[1].kind for call in events.await_args_list] + + +async def _run(**kwargs): + defaults = dict( + system_prompt="do phase work", + workspace_path="/tmp/ws", + outputs_dir="docs", + spec_path="specs", + model="m1", + logger=Mock(), + phase_number=12, + ) + defaults.update(kwargs) + return await agent_query_with_resume(**defaults) + + +def _patch_loop(agent_side_effects): + """Patch agent_query (side-effect sequence), sleep, and the HEAD reader.""" + sleep = AsyncMock() + agent_query = AsyncMock(side_effect=agent_side_effects) + head = AsyncMock(return_value="same-sha") + patches = [ + patch("app.services.claude_code.agent_query", agent_query), + patch("app.services.claude_code.asyncio.sleep", sleep), + patch("app.services.claude_code._read_workspace_head", head), + ] + return patches, agent_query, sleep, head + + +@pytest.mark.asyncio +async def test_incident_regression_progress_crashes_keep_retrying(recorders): + """Four productive crashes (old cap: 2) then success — the phase completes.""" + events, states = recorders + effects = [] + for _ in range(4): + effects.append(_crash(tool_uses=50)) # coding query crashes after real work + effects += [_CODING_OK, _COMPLETE] # attempt 5 succeeds and validates complete + patches, agent_query, sleep, _ = _patch_loop(effects) + with patches[0], patches[1], patches[2]: + result = await _run() + + assert result.result == "did some work" + assert result.is_error is False + assert agent_query.await_count == 6 # 5 coding + 1 validator + # Every productive crash waits exactly the first (1-minute) step — schedule keeps resetting. + assert [c.args[0] for c in sleep.await_args_list] == [60.0] * 4 + kinds = _recorded_kinds(events) + assert kinds.count(AgentErrorEventKind.AGENT_CRASH) == 4 + assert AgentErrorEventKind.PHASE_GAVE_UP not in kinds + assert AgentErrorEventKind.PHASE_INCOMPLETE not in kinds + + +@pytest.mark.asyncio +async def test_zero_progress_crashes_sleep_schedule_then_give_up(recorders): + """Seven consecutive zero-progress crashes: six scheduled waits, then stop.""" + events, states = recorders + patches, agent_query, sleep, _ = _patch_loop([_crash(tool_uses=0)] * 7) + with patches[0], patches[1], patches[2]: + result = await _run() + + assert result.is_error is True + assert result.result == _SOCKET_ERR # classifiable by execute_all_phases (#47) + assert agent_query.await_count == 7 # all coding attempts, no validator ran + assert [c.args[0] for c in sleep.await_args_list] == [60.0, 180.0, 300.0, 600.0, 1800.0, 3600.0] + kinds = _recorded_kinds(events) + assert kinds.count(AgentErrorEventKind.AGENT_CRASH) == 6 # exhausting crash isn't a retry + assert kinds[-1] == AgentErrorEventKind.PHASE_GAVE_UP + gave_up = events.await_args_list[-1].args[1] + assert "zero-progress" in gave_up.message + assert gave_up.crash_backoff == "6/6" + + +@pytest.mark.asyncio +async def test_progress_reset_then_re_exhaust(recorders): + """A productive crash resets the streak; the follow-up outage still exhausts.""" + events, _ = recorders + effects = [_crash(tool_uses=30)] + [_crash(tool_uses=0)] * 7 + patches, agent_query, sleep, _ = _patch_loop(effects) + with patches[0], patches[1], patches[2]: + result = await _run() + + assert result.is_error is True + assert agent_query.await_count == 8 + assert [c.args[0] for c in sleep.await_args_list] == [60.0, 60.0, 180.0, 300.0, 600.0, 1800.0, 3600.0] + + +@pytest.mark.asyncio +async def test_validator_budget_unchanged_at_two_resumes(recorders): + """Clean-but-incomplete attempts: initial + 2 resumes, then phase_incomplete.""" + events, _ = recorders + patches, agent_query, sleep, _ = _patch_loop([_CODING_OK, _INCOMPLETE] * 3) + with patches[0], patches[1], patches[2]: + result = await _run() + + assert agent_query.await_count == 6 # 3 attempts x (coding + validator) + assert result.result == "did some work" + assert result.is_error is False # checkpoint semantics preserved + sleep.assert_not_awaited() # validator-driven resumes retry immediately + kinds = _recorded_kinds(events) + assert kinds == [AgentErrorEventKind.PHASE_INCOMPLETE] + assert "completion check is still unsatisfied" in events.await_args.args[1].message + + +@pytest.mark.asyncio +async def test_completion_on_final_allowed_attempt_emits_no_warning(recorders): + """The old code logged 'Reached maximum resume attempts' even on success.""" + events, _ = recorders + effects = [_CODING_OK, _INCOMPLETE, _CODING_OK, _INCOMPLETE, _CODING_OK, _COMPLETE] + patches, agent_query, sleep, _ = _patch_loop(effects) + with patches[0], patches[1], patches[2]: + result = await _run() + + assert result.is_error is False + assert _recorded_kinds(events) == [] # no give-up, no incomplete — nothing to report + + +@pytest.mark.asyncio +async def test_dead_validator_retries_instead_of_silently_passing(recorders): + """Old behavior: validator None -> phase marked complete. New: crash-equivalent retry.""" + events, _ = recorders + effects = [ + _CODING_OK, AgentResult(result=None, session_id=None), # validator unusable + _CODING_OK, _COMPLETE, # retry verifies properly + ] + patches, agent_query, sleep, _ = _patch_loop(effects) + with patches[0], patches[1], patches[2]: + result = await _run() + + assert result.is_error is False + assert agent_query.await_count == 4 + assert [c.args[0] for c in sleep.await_args_list] == [60.0] + kinds = _recorded_kinds(events) + assert kinds == [AgentErrorEventKind.AGENT_CRASH] + assert "Completion check" in events.await_args_list[0].args[1].message + + +@pytest.mark.asyncio +async def test_validator_crash_counts_coding_progress_via_head(recorders): + """A validator exception after a committing coding run resets the schedule.""" + events, _ = recorders + effects = [ + _CODING_OK, _crash(tool_uses=0), # validator crashed with no tool uses... + _CODING_OK, _COMPLETE, + ] + patches, agent_query, sleep, head = _patch_loop(effects) + head.side_effect = ["sha-before", "sha-after", "sha-after", "sha-after"] + with patches[0], patches[1], patches[2]: + result = await _run() + + assert result.is_error is False + # ...but the coding query committed (HEAD advanced), so the crash kept the 1m wait. + assert [c.args[0] for c in sleep.await_args_list] == [60.0] + crash_event = events.await_args_list[0].args[1] + assert "saved work" in crash_event.message + + +@pytest.mark.asyncio +async def test_model_routing_crash_triggers_fallback_event(recorders): + events, _ = recorders + routing_crash = AgentQueryError( + "API returned an empty response", progress=AgentQueryProgress(0, 0) + ) + effects = [routing_crash, _CODING_OK, _COMPLETE] + patches, agent_query, sleep, _ = _patch_loop(effects) + with patches[0], patches[1], patches[2], patch( + "app.services.claude_code.apply_model_fallback_if_routing_failure", + Mock(return_value=("fallback-model", None)), + ): + result = await _run() + + assert result.is_error is False + assert result.active_model == "fallback-model" + kinds = _recorded_kinds(events) + assert kinds[0] == AgentErrorEventKind.MODEL_FALLBACK + fallback = events.await_args_list[0].args[1] + assert fallback.previous_model == "m1" + assert fallback.model == "fallback-model" + + +@pytest.mark.asyncio +async def test_tool_call_failure_stops_immediately(recorders): + events, _ = recorders + effects = [AgentQueryError("model does not support tools", progress=AgentQueryProgress(0, 0))] + patches, agent_query, sleep, _ = _patch_loop(effects) + with patches[0], patches[1], patches[2]: + result = await _run() + + assert result.is_error is True + assert agent_query.await_count == 1 + sleep.assert_not_awaited() + kinds = _recorded_kinds(events) + assert kinds == [AgentErrorEventKind.PHASE_GAVE_UP] + assert "tool failure" in events.await_args.args[1].message + + +@pytest.mark.asyncio +async def test_retrying_badge_set_and_cleared_around_waits(recorders): + _, states = recorders + effects = [_crash(tool_uses=0), _CODING_OK, _COMPLETE] + patches, agent_query, sleep, _ = _patch_loop(effects) + with patches[0], patches[1], patches[2]: + await _run() + + badge_calls = [call.args[2] for call in states.await_args_list] + assert badge_calls == [WorkspaceAgentState.RETRYING, None] + + +@pytest.mark.asyncio +async def test_silent_without_recorder_configured(): + """No handlers, no identity: the loop still works and raises nothing.""" + effects = [_crash(tool_uses=0), _CODING_OK, _COMPLETE] + patches, agent_query, sleep, _ = _patch_loop(effects) + with patches[0], patches[1], patches[2]: + result = await _run() + assert result.is_error is False + + +@pytest.mark.asyncio +async def test_model_routing_give_up_recommends_model_change(recorders): + """Part F: a model-caused give-up names the remedy (Settings -> retry).""" + events, _ = recorders + routing_crash = AgentQueryError( + "API returned an empty response", progress=AgentQueryProgress(0, 0) + ) + patches, agent_query, sleep, _ = _patch_loop([routing_crash] * 7) + with patches[0], patches[1], patches[2], patch( + "app.services.claude_code.apply_model_fallback_if_routing_failure", + Mock(side_effect=lambda err, active, cand, log, label: (active, cand)), + ): + result = await _run() + + assert result.is_error is True + gave_up = events.await_args_list[-1].args[1] + assert gave_up.kind == AgentErrorEventKind.PHASE_GAVE_UP + assert "change the model in Settings" in gave_up.message + + +@pytest.mark.asyncio +async def test_connection_give_up_does_not_recommend_model_change(recorders): + events, _ = recorders + patches, agent_query, sleep, _ = _patch_loop([_crash(tool_uses=0)] * 7) + with patches[0], patches[1], patches[2]: + await _run() + + gave_up = events.await_args_list[-1].args[1] + assert gave_up.kind == AgentErrorEventKind.PHASE_GAVE_UP + assert "change the model" not in gave_up.message + assert "code is preserved" in gave_up.message + + +@pytest.mark.asyncio +async def test_clean_incomplete_then_outage_surfaces_error_not_stale_success(recorders): + """Regression: a clean-but-incomplete attempt must not let its success mask a + later sustained outage. + + Before the fix, ``last_returned`` held attempt 1's non-error coding result and + ``final`` returned it verbatim on a NO_PROGRESS_BUDGET give-up (is_error=False). + execute_all_phases then checkpointed the incomplete phase as complete and the + connection-abort path (#47) never fired — the phase was silently skipped on + retry. The give-up must instead surface is_error=True carrying the crash text. + """ + events, _ = recorders + # attempt 1: clean coding + validator INCOMPLETE (resets the crash streak); + # attempts 2-8: sustained zero-progress outage that exhausts the schedule. + effects = [_CODING_OK, _INCOMPLETE] + [_crash(tool_uses=0)] * 7 + patches, agent_query, sleep, _ = _patch_loop(effects) + with patches[0], patches[1], patches[2]: + result = await _run() + + assert result.is_error is True + assert result.result == _SOCKET_ERR # crash text, NOT the stale "did some work" + assert agent_query.await_count == 9 # 1 coding + 1 validator + 7 coding crashes + # The clean-incomplete run reset the streak, so the full schedule still runs. + assert [c.args[0] for c in sleep.await_args_list] == [60.0, 180.0, 300.0, 600.0, 1800.0, 3600.0] + kinds = _recorded_kinds(events) + assert kinds.count(AgentErrorEventKind.AGENT_CRASH) == 6 + assert kinds[-1] == AgentErrorEventKind.PHASE_GAVE_UP + + +@pytest.mark.asyncio +async def test_tool_call_failure_after_incomplete_still_surfaces_error(recorders): + """A tool-call give-up after a prior clean-incomplete attempt still returns + is_error=True so the workspace aborts (model incompatible), rather than + returning the stale success from the earlier attempt.""" + events, _ = recorders + effects = [ + _CODING_OK, _INCOMPLETE, + AgentQueryError("model does not support tools", progress=AgentQueryProgress(0, 0)), + ] + patches, agent_query, sleep, _ = _patch_loop(effects) + with patches[0], patches[1], patches[2]: + result = await _run() + + assert result.is_error is True + assert "model does not support tools" in (result.result or "") + assert agent_query.await_count == 3 # coding + validator + one crashing coding call + sleep.assert_not_awaited() # tool-call failure is not retried in-loop + assert _recorded_kinds(events)[-1] == AgentErrorEventKind.PHASE_GAVE_UP diff --git a/backend/test/test_claude_code_sdk_agent_metrics.py b/backend/test/test_claude_code_sdk_agent_metrics.py index 850c699..e4f8613 100644 --- a/backend/test/test_claude_code_sdk_agent_metrics.py +++ b/backend/test/test_claude_code_sdk_agent_metrics.py @@ -793,3 +793,54 @@ def test_capture_agent_query_event_none_breakdown_no_tool_props(mock_ctx: Mock) props = t._client.capture.call_args.kwargs["properties"] assert "main_agent_tool_tokens" not in props assert "main_agent_tool_usage_json" not in props + + +# --------------------------------------------------------------------------- +# progress_snapshot (resume-loop progress signal) +# --------------------------------------------------------------------------- + +class TestProgressSnapshot: + def test_empty_stream_reports_zero(self): + m = ClaudeCodeSdkAgentMetrics() + snap = m.progress_snapshot() + assert snap.num_messages == 0 + assert snap.num_tool_uses == 0 + + def test_counts_messages_and_matched_tool_uses(self): + m = ClaudeCodeSdkAgentMetrics() + m.push(_assistant([_tool_use("t1", "Bash")])) + m.push(_user([_tool_result("t1")])) + snap = m.progress_snapshot() + assert snap.num_messages == 2 + assert snap.num_tool_uses == 1 + + def test_counts_pending_unmatched_tool_uses(self): + """A crash mid-tool (no ToolResultBlock yet) is still observed work.""" + m = ClaudeCodeSdkAgentMetrics() + m.push(_assistant([_tool_use("t1", "Bash"), _tool_use("t2", "Edit")])) + snap = m.progress_snapshot() + assert snap.num_tool_uses == 2 + + def test_snapshot_is_side_effect_free(self): + """Unlike get_metrics(), snapshot must not flush the pending registry.""" + m = ClaudeCodeSdkAgentMetrics() + m.push(_assistant([_tool_use("t1", "Bash")])) + m.progress_snapshot() + m.push(_user([_tool_result("t1")])) + # The late result still matches: exactly one Bash use counted, not two. + snap = m.progress_snapshot() + assert snap.num_tool_uses == 1 + metrics = m.get_metrics() + bash = metrics["main_agent_tool_usage"]["tools_usage_count"] + assert bash == [{"name": "Bash", "count": 1, "tokens": None}] + + def test_non_tool_messages_count_as_messages_only(self): + m = ClaudeCodeSdkAgentMetrics() + m.push(_assistant([TextBlock(text="thinking about it")])) + m.push(ResultMessage( + subtype="success", duration_ms=1, duration_api_ms=1, is_error=False, + num_turns=1, session_id="s", total_cost_usd=0.0, + )) + snap = m.progress_snapshot() + assert snap.num_messages == 2 + assert snap.num_tool_uses == 0 diff --git a/backend/test/test_execute_all_phases_connection_error.py b/backend/test/test_execute_all_phases_connection_error.py index 3916e52..0ae0b5c 100644 --- a/backend/test/test_execute_all_phases_connection_error.py +++ b/backend/test/test_execute_all_phases_connection_error.py @@ -51,6 +51,7 @@ async def test_connection_error_aborts_without_checkpoint(tmp_path: Path) -> Non request = GenerateAppRequest(spec_path="specs", outputs_dir="specflow", generation_id="e1") svc = Mock() + svc.db_adapter = None # DB-less unit test: raise_if_cancelled no-ops without an adapter svc.update_workspace_phase = AsyncMock() svc.update_deployment_workspace_phase = AsyncMock() @@ -75,6 +76,8 @@ async def fake_phase_fn(**kwargs): ) assert "lost connection" in str(exc_info.value).lower() + # The abort carries the phase it happened in (surfaced in workspace_aborted events). + assert exc_info.value.phase == 1 # The linchpin assertion: the errored phase was NOT checkpointed as completed. svc.update_workspace_phase.assert_not_called() svc.update_deployment_workspace_phase.assert_not_called() diff --git a/backend/test/test_parallel_execution.py b/backend/test/test_parallel_execution.py index 57f3589..6ac05e2 100644 --- a/backend/test/test_parallel_execution.py +++ b/backend/test/test_parallel_execution.py @@ -620,3 +620,86 @@ async def mock_generation_fn(workspace, **kwargs): if __name__ == "__main__": pytest.main([__file__, "-v"]) + + +class TestWorkspaceAbortEvents: + """The executor's except-branch is the single workspace_aborted recording site.""" + + @pytest.fixture + def mock_settings(self): + settings = Mock(spec=Settings) + settings.AGENT_BASE_PATH = "/agent" + settings.ANTHROPIC_API_KEY = "test-key" + settings.OPENROUTER_API_KEY = "test-or-key" + settings.OPENROUTER_BASE_URL = "https://openrouter.ai/api" + return settings + + @pytest.fixture + def workspace(self): + return [ + WorkspaceSettings( + name="ws-1", + workspace_path="/agent/workspace1", + provider="anthropic", + model="claude-sonnet-4-0", + ) + ] + + @pytest.fixture + def recorders(self): + from unittest.mock import AsyncMock + + from app.core.telemetry_context import TelemetryContext + + TelemetryContext.clear_context() + events = AsyncMock() + states = AsyncMock() + TelemetryContext.set_agent_error_event_handler(events) + TelemetryContext.set_workspace_agent_state_handler(states) + TelemetryContext.set_user_context(generation_id="est-1") + yield events, states + TelemetryContext.clear_context() + + @pytest.mark.asyncio + async def test_failure_records_aborted_event_with_error_type_and_phase( + self, mock_settings, workspace, recorders + ): + from app.schemas.agent_error_events import AgentErrorEventKind, WorkspaceAgentState + from app.services.claude_code import WorkspaceAbortedError + + events, states = recorders + executor = ParallelAgentExecutor(WorkspaceManager(mock_settings, Mock()), Mock()) + + async def aborting_agent(ws, manager, log): + raise WorkspaceAbortedError( + "[ws-1] Phase 12 lost connection to the API. Error: socket connection was closed", + phase=12, + ) + + results = await executor.execute_parallel(workspaces=workspace, agent_fn=aborting_agent) + + assert results[0].success is False + events.assert_awaited_once() + _, event = events.await_args.args + assert event.kind == AgentErrorEventKind.WORKSPACE_ABORTED + assert event.workspace_id == "ws-1" + assert event.phase == 12 + assert event.error_type is not None # classified connection error + states.assert_awaited_once() + assert states.await_args.args[1:] == ("ws-1", WorkspaceAgentState.ABORTED) + + @pytest.mark.asyncio + async def test_cancellation_records_no_event(self, mock_settings, workspace, recorders): + from app.state.exceptions import GenerationCancelledError + + events, states = recorders + executor = ParallelAgentExecutor(WorkspaceManager(mock_settings, Mock()), Mock()) + + async def cancelled_agent(ws, manager, log): + raise GenerationCancelledError("est-1") + + with pytest.raises(GenerationCancelledError): + await executor.execute_parallel(workspaces=workspace, agent_fn=cancelled_agent) + + events.assert_not_awaited() + states.assert_not_awaited() diff --git a/backend/test/test_resume_generation.py b/backend/test/test_resume_generation.py new file mode 100644 index 0000000..fd1a34c --- /dev/null +++ b/backend/test/test_resume_generation.py @@ -0,0 +1,87 @@ +"""ResumeGeneration semantics — the resume-loop budget contract, verbatim. + +| Attempt outcome | Effect | +|------------------------------|-------------------------------------------------| +| validator complete | stop (loop-level; not a budget concern) | +| clean but validator incomplete | validator_resumes += 1; crash schedule reset | +| crash WITH saved progress | schedule reset to start -> next wait = 1 min | +| crash WITHOUT progress | schedule advances -> waits 1/3/5/10/30/60 min | + +No total cap: progress keeps resetting the schedule indefinitely. +""" + +from datetime import timedelta + +from app.services.claude_code import ( + _CRASH_BACKOFF_MINUTES, + RESUME_PROGRESS_MIN_TOOL_USES, + ResumeGeneration, +) + + +class TestCrashSchedule: + def test_every_crash_waits_at_least_a_minute(self): + r = ResumeGeneration() + wait = r.record_crash(tool_uses=999, head_advanced=True) + assert wait == timedelta(minutes=1) + + def test_zero_progress_crashes_follow_the_exact_schedule_then_exhaust(self): + r = ResumeGeneration() + waits = [r.record_crash(tool_uses=0, head_advanced=False) for _ in _CRASH_BACKOFF_MINUTES] + assert waits == [timedelta(minutes=m) for m in _CRASH_BACKOFF_MINUTES] + # The next consecutive zero-progress crash exhausts the schedule. + assert r.record_crash(tool_uses=0, head_advanced=False) is None + + def test_progress_resets_the_schedule(self): + r = ResumeGeneration() + r.record_crash(tool_uses=0, head_advanced=False) # 1m + r.record_crash(tool_uses=0, head_advanced=False) # 3m + # Saved progress (commit) -> streak reset, wait back to 1 minute. + assert r.record_crash(tool_uses=0, head_advanced=True) == timedelta(minutes=1) + # Next zero-progress crash starts the schedule over. + assert r.record_crash(tool_uses=0, head_advanced=False) == timedelta(minutes=1) + + def test_no_total_cap_progress_crashes_never_exhaust(self): + r = ResumeGeneration() + for _ in range(50): + assert r.record_crash(tool_uses=RESUME_PROGRESS_MIN_TOOL_USES, head_advanced=False) is not None + + def test_progress_definition_commit_or_tool_use_floor(self): + r = ResumeGeneration() + assert r.saved_progress(tool_uses=0, head_advanced=True) is True + assert r.saved_progress(tool_uses=RESUME_PROGRESS_MIN_TOOL_USES, head_advanced=False) is True + assert r.saved_progress(tool_uses=RESUME_PROGRESS_MIN_TOOL_USES - 1, head_advanced=False) is False + + +class TestValidatorBudget: + def test_two_resumes_allowed_third_denied(self): + r = ResumeGeneration() + assert r.record_incomplete() is True # schedule resume 1 + assert r.record_incomplete() is True # schedule resume 2 + assert r.record_incomplete() is False # a third resume is over budget + + def test_clean_incomplete_run_resets_crash_streak(self): + r = ResumeGeneration() + r.record_crash(tool_uses=0, head_advanced=False) + r.record_crash(tool_uses=0, head_advanced=False) + r.record_incomplete() # a clean run proves the connection is alive + assert r.crash_streak == 0 + assert r.record_crash(tool_uses=0, head_advanced=False) == timedelta(minutes=1) + + +class TestDescribe: + def test_describe_matches_counters(self): + r = ResumeGeneration() + r.record_crash(tool_uses=0, head_advanced=False) + r.record_crash(tool_uses=0, head_advanced=False) + r.record_incomplete() + assert r.describe() == "crash backoff 0/6, validator resumes 1/2" + + def test_backoff_label_tracks_streak(self): + r = ResumeGeneration() + assert r.crash_backoff_label() == "0/6" + r.record_crash(tool_uses=0, head_advanced=False) + assert r.crash_backoff_label() == "1/6" + for _ in range(10): + r.record_crash(tool_uses=0, head_advanced=False) + assert r.crash_backoff_label() == "6/6" diff --git a/backend/test/test_workflow_steps_survivors_gate.py b/backend/test/test_workflow_steps_survivors_gate.py new file mode 100644 index 0000000..f2fdd29 --- /dev/null +++ b/backend/test/test_workflow_steps_survivors_gate.py @@ -0,0 +1,133 @@ +"""Minimum-survivors gate (un-drops what PR #47 deferred, strengthened). + +Codegen: fail when fewer than min(2, total) workspaces survived — variance +scores need >= 2 samples; single-workspace runs still pass with their 1. +Deploy: min 1 survivor (all-failed rule). Partial failure above the bar +continues with survivors. +""" + +from unittest.mock import Mock + +import pytest + +from app.schemas.workspace import WorkspaceSettings +from app.services.parallel_executor import ParallelAgentResult +from app.services.workflow_steps import ( + MIN_WORKSPACES_FOR_VARIANCE, + InsufficientWorkspacesError, + _raise_if_insufficient_workspaces, +) + + +def _ws(name: str) -> WorkspaceSettings: + return WorkspaceSettings(name=name, workspace_path=f"/agent/{name}", provider="anthropic", model="m") + + +def _result(name: str, success: bool, error: str | None = None) -> ParallelAgentResult: + return ParallelAgentResult( + workspace_name=name, + workspace_settings=_ws(name), + result=None, + success=success, + error=error, + ) + + +def _gate(results, min_survivors=MIN_WORKSPACES_FOR_VARIANCE): + _raise_if_insufficient_workspaces( + results, + step_label="code generation", + logger=Mock(), + min_survivors=min_survivors, + requirement_reason="to compute variance-reduced estimates", + ) + + +class TestCodegenGate: + def test_one_of_three_survivors_raises_with_variance_message(self): + results = [ + _result("ws-1", False, "API Error: The socket connection was closed unexpectedly"), + _result("ws-2", False, "API Error: Unable to connect to API (ConnectionRefused)"), + _result("ws-3", True), + ] + with pytest.raises(InsufficientWorkspacesError) as exc: + _gate(results) + msg = str(exc.value) + assert "Only 1 of 3 workspaces completed code generation" in msg + assert "at least 2 are required to compute variance-reduced estimates" in msg + assert "retry_generation" in msg + # All-connection failures get the friendly network hint. + assert "network interruption" in msg + # No internal jargon in the user-facing message. + assert "checkpoint" not in msg.lower() + assert "firestore" not in msg.lower() + + def test_two_of_three_survivors_continues_with_warning(self): + logger = Mock() + results = [ + _result("ws-1", False, "boom"), + _result("ws-2", True), + _result("ws-3", True), + ] + _raise_if_insufficient_workspaces( + results, + step_label="code generation", + logger=logger, + min_survivors=MIN_WORKSPACES_FOR_VARIANCE, + requirement_reason="to compute variance-reduced estimates", + ) + logger.warning.assert_called_once() + + def test_zero_of_three_raises(self): + results = [_result(f"ws-{i}", False, "err") for i in range(3)] + with pytest.raises(InsufficientWorkspacesError): + _gate(results) + + def test_single_workspace_run_success_passes(self): + _gate([_result("ws-1", True)]) # min(2, 1) = 1 -> satisfied + + def test_single_workspace_run_failure_raises(self): + with pytest.raises(InsufficientWorkspacesError) as exc: + _gate([_result("ws-1", False, "err")]) + assert "at least 1 is required" in str(exc.value) + + def test_all_model_routing_failures_recommend_model_change(self): + results = [ + _result("ws-1", False, "API returned an empty response"), + _result("ws-2", False, "malformed response from model"), + _result("ws-3", False, "empty response"), + ] + with pytest.raises(InsufficientWorkspacesError) as exc: + _gate(results) + assert "change the model in Settings" in str(exc.value) + + def test_connection_type_failure_does_not_recommend_model_change(self): + results = [ + _result("ws-1", False, "connection refused"), + _result("ws-2", False, "connection refused"), + _result("ws-3", False, "connection refused"), + ] + with pytest.raises(InsufficientWorkspacesError) as exc: + _gate(results) + assert "change the model" not in str(exc.value) + + def test_all_success_and_non_parallel_items_are_noops(self): + _gate([_result("ws-1", True), _result("ws-2", True)]) + _gate(["not-a-result", None]) + _gate([]) + _gate(None) + + +class TestDeployGate: + def test_one_survivor_of_three_continues(self): + results = [ + _result("ws-1", False, "deploy failed"), + _result("ws-2", False, "deploy failed"), + _result("ws-3", True), + ] + _gate(results, min_survivors=1) + + def test_zero_survivors_raises(self): + results = [_result(f"ws-{i}", False, "deploy failed") for i in range(3)] + with pytest.raises(InsufficientWorkspacesError): + _gate(results, min_survivors=1) diff --git a/mcp_server/server.py b/mcp_server/server.py index 1688fe1..22a3ea8 100644 --- a/mcp_server/server.py +++ b/mcp_server/server.py @@ -129,6 +129,21 @@ def _phase_label(status: str, checkpoint: str | None) -> str: return f"Status: {status}" +def _agent_warnings_clause(response_data: dict) -> str: + """Short clause summarizing agent error/warning events for a running generation.""" + events = response_data.get("agent_error_events") or [] + if not events: + return "" + latest = events[-1] if isinstance(events[-1], dict) else {} + where = latest.get("workspace_id") or "a workspace" + phase = latest.get("phase") + where_full = f"{where}, phase {phase}" if phase is not None else where + return ( + f" {len(events)} agent warning(s) so far (latest: {where_full}) — " + f"the run is retrying automatically." + ) + + def _status_chat_message(response_data: dict) -> str: """One or two sentences for check_status chat display.""" status = (response_data.get("status") or "").lower() @@ -152,7 +167,10 @@ def _status_chat_message(response_data: dict) -> str: ) return "Generation failed. Use `retry_generation` if you want to resume from the last checkpoint." if status in (GenerationStatus.RUNNING, GenerationStatus.INITIALIZING): - return brief_sentences(f"Generation is in progress ({phase}). You'll get an email when it finishes.") + return brief_sentences( + f"Generation is in progress ({phase}).{_agent_warnings_clause(response_data)} " + f"You'll get an email when it finishes." + ) return brief_sentences(f"Status is {status or 'unknown'} ({phase}).") diff --git a/mcp_server/tests/test_server_phase_label.py b/mcp_server/tests/test_server_phase_label.py index ea3b100..b0c4a9a 100644 --- a/mcp_server/tests/test_server_phase_label.py +++ b/mcp_server/tests/test_server_phase_label.py @@ -69,3 +69,39 @@ def test_pending_fallback(self): def test_unknown_status_fallback(self): assert _phase_label("some_future_status", "some_checkpoint") == "Status: some_future_status" + + +class TestStatusChatMessageAgentWarnings: + """RUNNING message mentions agent warnings when the payload carries events.""" + + def _event(self, ws="ws-01-1", phase=12): + return {"workspace_id": ws, "phase": phase, "kind": "agent_crash", "message": "x"} + + def test_running_with_events_mentions_count_and_latest(self): + from server import _status_chat_message + + msg = _status_chat_message( + { + "status": "running", + "checkpoint": "generation_started", + "agent_error_events": [self._event(), self._event(ws="ws-01-2", phase=3)], + } + ) + assert "2 agent warning(s)" in msg + assert "ws-01-2, phase 3" in msg + assert "retrying automatically" in msg + + def test_running_without_events_unchanged(self): + from server import _status_chat_message + + msg = _status_chat_message({"status": "running", "checkpoint": "generation_started"}) + assert "warning" not in msg.lower() + assert "in progress" in msg + + def test_failed_message_does_not_add_warning_clause(self): + from server import _status_chat_message + + msg = _status_chat_message( + {"status": "failed", "error": "boom", "agent_error_events": [self._event()]} + ) + assert "warning" not in msg.lower() diff --git a/mcp_server/tests/test_tui_activity.py b/mcp_server/tests/test_tui_activity.py index a1eb07e..7b84d5b 100644 --- a/mcp_server/tests/test_tui_activity.py +++ b/mcp_server/tests/test_tui_activity.py @@ -53,3 +53,47 @@ def test_reads_log_when_present(self, tmp_path): d.mkdir() (d / "ws-01-1.log").write_text("line1\nline2\n") assert activity.recent_activity(tmp_path, "ws-01-1") == ["line1", "line2"] + + +class TestNestedLogDiscovery: + """Logs live nested (agent_logs//-phase/.log) — see backend logging.""" + + def test_finds_nested_workspace_log(self, tmp_path): + d = tmp_path / "agent_logs" + nested = d / "est-abc123" / "ws-01-1-phase12" + nested.mkdir(parents=True) + target = nested / "2026-07-23_10-00-00.log" + target.write_text("phase 12 log line") + assert activity.find_log_for_workspace(d, "ws-01-1") == target + + def test_ignores_non_log_files_like_ds_store(self, tmp_path): + """The regression from the dashboard screenshot: binary .DS_Store rendered as garbage.""" + import os + + d = tmp_path / "agent_logs" + nested = d / "est-abc123" / "ws-01-1-phase12" + nested.mkdir(parents=True) + log = nested / "run.log" + log.write_text("real log") + junk = d / ".DS_Store" + junk.write_bytes(b"\x00\x01Bud1\x00") + os.utime(log, (1, 1)) + os.utime(junk, (10**9, 10**9)) # junk is newest — must still be ignored + assert activity.find_log_for_workspace(d, None) == log + assert activity.find_log_for_workspace(d, "ws-01-1") == log + + def test_picks_newest_log_across_generations(self, tmp_path): + import os + + d = tmp_path / "agent_logs" + old_dir = d / "est-old" / "ws-01-1-phase3" + new_dir = d / "est-new" / "ws-01-1-phase5" + old_dir.mkdir(parents=True) + new_dir.mkdir(parents=True) + old = old_dir / "old.log" + old.write_text("old") + new = new_dir / "new.log" + new.write_text("new") + os.utime(old, (1, 1)) + os.utime(new, (10**9, 10**9)) + assert activity.find_log_for_workspace(d, "ws-01-1") == new diff --git a/mcp_server/tests/test_tui_app.py b/mcp_server/tests/test_tui_app.py index 1859c3d..f606f6f 100644 --- a/mcp_server/tests/test_tui_app.py +++ b/mcp_server/tests/test_tui_app.py @@ -2214,3 +2214,127 @@ async def test_settings_tier_marker_stays_on_screen(self, tmp_path): region = marker.region assert region.width > 0 assert region.right <= width # fully within the viewport + + +def _events_payload(events=None, agent_state=None) -> dict: + payload = _running_payload() + if events is not None: + payload["agent_error_events"] = events + if agent_state is not None: + payload["progress"]["workspace_phases"]["ws-01-1"]["agent_state"] = agent_state + return payload + + +def _crash_event(ws="ws-01-1", kind="agent_crash", message="Connection to the model API was lost"): + return { + "at": "2026-07-23T18:28:40+00:00", + "workspace_id": ws, + "kind": kind, + "message": message, + "phase": 12, + } + + +class TestDashboardAgentSurfacing: + """The session view stays minimal: it must NOT list agent warnings. Per-workspace + state is conveyed only by bare row markers; detail lives in the drill-in / Events.""" + + def _render(self, renderable) -> str: + console = Console(width=110, record=True) + console.print(renderable) + return console.export_text() + + def test_events_do_not_render_a_warnings_panel_on_the_dashboard(self): + payload = _events_payload(events=[_crash_event()]) + out = self._render( + tui_app.build_dashboard(payload, Path("/tmp/acme"), "gen_8f3abc21") + ) + assert "Agent warnings" not in out + # The full crash message belongs in the drill-in / Events screen, not here. + assert "Connection to the model API was lost" not in out + assert "press e for all" not in out + + def test_workspace_row_shows_retrying_badge(self): + payload = _events_payload(agent_state="retrying") + out = self._render( + tui_app.build_dashboard(payload, Path("/tmp/acme"), "gen_8f3abc21") + ) + assert "RETRYING" in out + + def test_workspace_row_shows_aborted_badge(self): + payload = _events_payload(agent_state="aborted") + out = self._render( + tui_app.build_dashboard(payload, Path("/tmp/acme"), "gen_8f3abc21") + ) + assert "ABORTED" in out + + def test_workspace_row_shows_warning_marker_for_flagged_workspace(self): + payload = _events_payload(events=[_crash_event(ws="ws-01-1")]) + out = self._render( + tui_app.build_dashboard(payload, Path("/tmp/acme"), "gen_8f3abc21") + ) + assert "⚠" in out + + +class TestWorkspaceWarningsBlock: + def _render(self, renderable) -> str: + console = Console(width=110, record=True) + console.print(renderable) + return console.export_text() + + def test_hidden_without_events(self): + assert tui_app.build_workspace_warnings(_running_payload(), "ws-01-1") is None + + def test_lists_only_this_workspace_with_model_fallback_pinned(self): + payload = _events_payload( + events=[ + _crash_event(ws="ws-01-1", message="crash first"), + _crash_event(ws="ws-01-2", message="other workspace crash"), + _crash_event( + ws="ws-01-1", + kind="model_fallback", + message="Model opus kept failing; switching to sonnet", + ), + ] + ) + panel = tui_app.build_workspace_warnings(payload, "ws-01-1") + out = self._render(panel) + assert "Warnings · ws-01-1" in out + assert "other workspace crash" not in out + # Model switch pinned before the earlier crash. + assert out.index("switching to sonnet") < out.index("crash first") + + +class TestEventsScreen: + @pytest.mark.asyncio + async def test_e_opens_events_screen_and_renders_rows(self): + a, b, c = _gate_ready() + payload = _events_payload(events=[_crash_event(message="socket closed at 18:28")]) + with a, b, c, patch("tui.app.poll_once", new=AsyncMock(return_value=payload)): + app = tui_app.SpecFlowTUI(root=Path("/tmp/x"), generation_id="gen_x", poll_interval=999) + async with app.run_test() as pilot: + await pilot.pause() + assert isinstance(app.screen, tui_app.DashboardScreen) + await pilot.press("e") + await pilot.pause() + assert isinstance(app.screen, tui_app.EventsScreen) + log = app.screen.query_one("#events-log") + text = "\n".join(str(line) for line in log.lines) + assert "socket closed at 18:28" in text + await pilot.press("escape") + await pilot.pause() + assert isinstance(app.screen, tui_app.DashboardScreen) + + @pytest.mark.asyncio + async def test_events_screen_empty_state(self): + a, b, c = _gate_ready() + with a, b, c, patch("tui.app.poll_once", new=AsyncMock(return_value=_running_payload())): + app = tui_app.SpecFlowTUI(root=Path("/tmp/x"), generation_id="gen_x", poll_interval=999) + async with app.run_test() as pilot: + await pilot.pause() + await pilot.press("e") + await pilot.pause() + assert isinstance(app.screen, tui_app.EventsScreen) + log = app.screen.query_one("#events-log") + text = "\n".join(str(line) for line in log.lines) + assert "No agent errors or warnings recorded." in text diff --git a/mcp_server/tests/test_tui_render.py b/mcp_server/tests/test_tui_render.py index 34eeb47..82e1f4d 100644 --- a/mcp_server/tests/test_tui_render.py +++ b/mcp_server/tests/test_tui_render.py @@ -390,3 +390,75 @@ def test_ineligible_message_running(self): def test_ineligible_message_terminal(self): msg = render.clear_ws_ineligible_message({"status": "completed"}) assert "Nothing to clear" in msg + + +def _event(ws="ws-01-1", kind="agent_crash", message="connection lost", phase=12, at="2026-07-23T18:28:40+00:00"): + return {"at": at, "workspace_id": ws, "kind": kind, "message": message, "phase": phase} + + +class TestAgentErrorEventRows: + def test_parses_events_into_rows(self): + payload = {"agent_error_events": [_event()]} + rows = render.agent_error_event_rows(payload) + assert len(rows) == 1 + row = rows[0] + assert row.time == "18:28:40" + assert row.workspace_id == "ws-01-1" + assert row.phase_label == "phase 12" + assert row.kind == "agent_crash" + assert row.message == "connection lost" + + def test_missing_phase_gives_empty_label(self): + payload = {"agent_error_events": [{**_event(), "phase": None}]} + assert render.agent_error_event_rows(payload)[0].phase_label == "" + + def test_empty_or_missing_key_returns_empty(self): + assert render.agent_error_event_rows({}) == [] + assert render.agent_error_event_rows({"agent_error_events": []}) == [] + assert render.agent_error_event_rows(None) == [] + + def test_non_dict_entries_are_skipped(self): + payload = {"agent_error_events": ["garbage", _event()]} + assert len(render.agent_error_event_rows(payload)) == 1 + + def test_workspaces_with_events(self): + payload = {"agent_error_events": [_event(ws="ws-01-1"), _event(ws="ws-01-3")]} + assert render.workspaces_with_events(payload) == {"ws-01-1", "ws-01-3"} + + def test_workspace_warning_rows_filters_and_pins_model_fallback_first(self): + payload = { + "agent_error_events": [ + _event(ws="ws-01-1", kind="agent_crash", message="crash 1"), + _event(ws="ws-01-2", kind="agent_crash", message="other ws"), + _event(ws="ws-01-1", kind="model_fallback", message="switched to sonnet"), + ] + } + rows = render.workspace_warning_rows(payload, "ws-01-1") + assert [r.message for r in rows] == ["switched to sonnet", "crash 1"] + + +class TestAgentStateBadge: + def test_known_states(self): + assert render.agent_state_badge("retrying") == ("RETRYING", "bold yellow") + assert render.agent_state_badge("aborted") == ("ABORTED", "bold red") + + def test_none_and_unknown(self): + assert render.agent_state_badge(None) is None + assert render.agent_state_badge("") is None + assert render.agent_state_badge("weird") is None + + def test_workspace_bars_carry_agent_state(self): + payload = { + "workspace_phases": { + "ws-01-1": {"last_completed_phase": 3, "total_phases": 9, "agent_state": "retrying"}, + "ws-01-2": {"last_completed_phase": 4, "total_phases": 9}, + } + } + bars = render.workspace_bars(payload) + assert bars[0].agent_state == "retrying" + assert bars[1].agent_state is None + + +class TestErrorKindStyle: + def test_error_stream_kind_has_style(self): + assert render.kind_style("error") == "bold red" diff --git a/mcp_server/tui/activity.py b/mcp_server/tui/activity.py index 6781619..6c11e37 100644 --- a/mcp_server/tui/activity.py +++ b/mcp_server/tui/activity.py @@ -26,18 +26,21 @@ def logs_dir(root: Path) -> Path: def find_log_for_workspace(directory: Path, workspace_id: str | None) -> Path | None: """Pick the log file for a workspace, or the most recently modified one. - Prefers a file whose name contains ``workspace_id``; falls back to the - newest file in the directory. Returns None when the directory is missing or + Logs are written NESTED (``agent_logs//-…phase/.log``), + so the search is recursive over ``*.log`` only — a stray top-level binary file + (e.g. macOS ``.DS_Store``) must never be tailed into the activity panel. + Prefers files whose path (relative to the logs dir) contains ``workspace_id``; + falls back to the newest log. Returns None when the directory is missing or empty, or on any access error. """ try: if not directory.is_dir(): return None - files = [p for p in directory.iterdir() if p.is_file()] + files = [p for p in directory.rglob("*.log") if p.is_file()] if not files: return None if workspace_id: - matches = [p for p in files if workspace_id in p.name] + matches = [p for p in files if workspace_id in str(p.relative_to(directory))] if matches: return max(matches, key=lambda p: p.stat().st_mtime) return max(files, key=lambda p: p.stat().st_mtime) diff --git a/mcp_server/tui/app.py b/mcp_server/tui/app.py index bf808f6..7f1b36d 100644 --- a/mcp_server/tui/app.py +++ b/mcp_server/tui/app.py @@ -131,6 +131,17 @@ def _pipeline_panel(payload: dict[str, Any]) -> Panel: return Panel(body, title="Pipeline", border_style="blue") +def _workspace_badge(bar: render.WorkspaceBar, flagged: set[str]) -> Text: + """Bare state word for the workspace list; details live in the drill-in.""" + badge = render.agent_state_badge(bar.agent_state) + if badge is not None: + text, style = badge + return Text(text, style=style) + if bar.workspace_id in flagged: + return Text("⚠", style="yellow") + return Text("") + + def _workspaces_panel(payload: dict[str, Any], selected_ws_id: str | None = None) -> Panel: bars = render.workspace_bars(payload) if not bars: @@ -139,6 +150,7 @@ def _workspaces_panel(payload: dict[str, Any], selected_ws_id: str | None = None title="Workspaces", border_style="blue", ) + flagged = render.workspaces_with_events(payload) table = Table.grid(padding=(0, 1)) table.add_column() # selection marker table.add_column() # ws id @@ -146,6 +158,7 @@ def _workspaces_panel(payload: dict[str, Any], selected_ws_id: str | None = None table.add_column() # phase name table.add_column() # bar table.add_column(justify="right") # pct + table.add_column() # agent-state badge / warning marker for bar in bars: selected = bar.workspace_id == selected_ws_id marker = "▶" if selected else " " @@ -157,6 +170,7 @@ def _workspaces_panel(payload: dict[str, Any], selected_ws_id: str | None = None Text(bar.phase_name[:32], style="dim"), Text(render.progress_bar(bar.fraction), style="green"), Text(f"{bar.percent}%"), + _workspace_badge(bar, flagged), ) count = payload.get("workspace_count") title = f"Workspaces · {count} variants" if count else "Workspaces" @@ -220,6 +234,20 @@ def _activity_panel(root: Path, payload: dict[str, Any]) -> Panel | None: return Panel(body, title=title, border_style="grey50") +def _event_row_text(row: render.AgentErrorEventRow) -> Text: + """One agent error/warning event as a Rich line — used by the workspace + drill-in warnings block and the Events screen only (never the dashboard).""" + line = Text() + if row.time: + line.append(f"{row.time} ", style="dim") + line.append(row.workspace_id, style="bold") + if row.phase_label: + line.append(f" {row.phase_label}", style="cyan") + line.append(" — ") + line.append(row.message, style="yellow") + return line + + def build_dashboard( payload: dict[str, Any] | None, root: Path, @@ -241,9 +269,18 @@ def build_dashboard( estimate = _estimate_panel(payload) if estimate is not None: panels.append(estimate) - act = _activity_panel(root, payload) - if act is not None: - panels.append(act) + # "Recent activity" is hidden for now: a raw tail of the DEBUG agent log is too + # noisy to be useful in the dashboard. `_activity_panel` / `tui.activity` are + # kept (and now point at the correct nested log dir) so this can be revived with + # a proper filtered/prettified view later — just re-append here. + # act = _activity_panel(root, payload) + # if act is not None: + # panels.append(act) + # Per-workspace agent warnings are deliberately NOT listed on the session view. + # The Workspaces panel already carries bare RETRYING/ABORTED/⚠ markers at a + # glance; full detail lives in the workspace drill-in + the Events screen (`e`). + # The dashboard stays minimal: pipeline, workspaces, and only the end result + # (estimate) or a fatal Error. if payload.get("error"): panels.append( Panel(Text(str(payload["error"]), style="red"), title="Error", border_style="red") @@ -268,6 +305,20 @@ def stream_row_text(row: render.StreamRow) -> Text: return line +def build_workspace_warnings(payload: dict[str, Any] | None, workspace_id: str) -> Panel | None: + """Yellow per-workspace warnings block for the drill-in screen. + + Lists THIS workspace's agent error/warning events with model switches + pinned first (a silently swapped model is the one thing the user must not + miss). None when the workspace has no events — the widget stays hidden. + """ + rows = render.workspace_warning_rows(payload or {}, workspace_id) + if not rows: + return None + body = Text("\n").join(_event_row_text(row) for row in rows) + return Panel(body, title=f"Warnings · {workspace_id}", border_style="yellow") + + def build_workspace_stats(payload: dict[str, Any] | None, workspace_id: str) -> Panel: """Build the per-workspace stats panel for the drill-in screen.""" stats = render.workspace_stats(payload or {}, workspace_id) @@ -474,6 +525,7 @@ class DashboardScreen(_SpecFlowScreen): Binding("c", "connect_client", "Add MCP to AI tool"), Binding("o", "open_workspace", "open ws"), Binding("enter", "open_workspace", "open ws", show=False), + Binding("e", "open_events", "events"), Binding("h", "open_report", "open report"), # Priority so the workspace selection wins over the scroll container's # own up/down handling (otherwise arrows just scroll the dashboard). @@ -577,6 +629,9 @@ def action_open_workspace(self) -> None: return self.app.push_screen(WorkspaceMessagesScreen(self._generation_id, ws_id)) + def action_open_events(self) -> None: + self.app.push_screen(EventsScreen(self._generation_id)) + def action_open_report(self) -> None: if render.estimate_panel(self._payload) is None: self.notify("No HTML report available yet.", severity="information") @@ -767,6 +822,7 @@ def compose(self) -> ComposeResult: yield Header(show_clock=True) with Vertical(): yield RichLog(id="ws-stream", wrap=True, markup=False, highlight=False, max_lines=2000) + yield Static(id="ws-warnings") yield Static(id="ws-stats") yield Footer() @@ -774,6 +830,7 @@ def on_mount(self) -> None: log = self.query_one("#ws-stream", RichLog) log.border_title = f"Live messages · {self._workspace_id}" log.write(Text("Waiting for live messages…", style="dim")) + self.query_one("#ws-warnings", Static).display = False self.call_later(self.refresh_stats) self.set_interval(self.app.poll_interval, self.refresh_stats) self.run_worker(self._stream_worker(), exclusive=True, group="ws-stream") @@ -782,6 +839,11 @@ async def refresh_stats(self) -> None: payload = await poll_once(self._generation_id) if isinstance(self.app, SpecFlowTUI): self.app.process_payload(self._generation_id, payload) + warnings_widget = self.query_one("#ws-warnings", Static) + warnings_panel = build_workspace_warnings(payload, self._workspace_id) + warnings_widget.display = warnings_panel is not None + if warnings_panel is not None: + warnings_widget.update(warnings_panel) self.query_one("#ws-stats", Static).update( build_workspace_stats(payload, self._workspace_id) ) @@ -797,6 +859,63 @@ def action_back(self) -> None: self.app.pop_screen() +class EventsScreen(_SpecFlowScreen): + """Full scrollable log of agent error/warning events for one generation. + + The dashboard's "Agent warnings" panel shows only the newest few; this + screen lists everything the /status payload carries, refreshed on the same + poll cadence as the dashboard. + """ + + BINDINGS = [ + Binding("escape", "back", "back"), + Binding("b", "back", "back"), + ] + + def __init__(self, generation_id: str) -> None: + super().__init__() + self._generation_id = generation_id + self._rendered_count = 0 + + @property + def generation_id(self) -> str: + return self._generation_id + + def compose(self) -> ComposeResult: + yield Header(show_clock=True) + yield RichLog(id="events-log", wrap=True, markup=False, highlight=False, max_lines=2000) + yield Footer() + + def on_mount(self) -> None: + log = self.query_one("#events-log", RichLog) + log.border_title = f"Agent events · {self._generation_id[:18]}…" + log.write(Text("Loading events…", style="dim")) + self.call_later(self.refresh_events) + self.set_interval(self.app.poll_interval, self.refresh_events) + + async def refresh_events(self) -> None: + payload = await poll_once(self._generation_id) + if isinstance(self.app, SpecFlowTUI): + self.app.process_payload(self._generation_id, payload) + rows = render.agent_error_event_rows(payload) + log = self.query_one("#events-log", RichLog) + if not rows: + if self._rendered_count == 0: + log.clear() + log.write(Text("No agent errors or warnings recorded.", style="dim")) + self._rendered_count = -1 # placeholder written; skip rewrites + return + if len(rows) == self._rendered_count: + return # nothing new — avoid churning the scroll position + log.clear() + for row in rows: + log.write(_event_row_text(row)) + self._rendered_count = len(rows) + + def action_back(self) -> None: + self.app.pop_screen() + + class _SessionItem(ListItem): """A sessions-list row carrying its generation id.""" diff --git a/mcp_server/tui/render.py b/mcp_server/tui/render.py index 3c31b1c..60fd3ce 100644 --- a/mcp_server/tui/render.py +++ b/mcp_server/tui/render.py @@ -48,6 +48,8 @@ class WorkspaceBar: phase_name: str last_completed_phase: int total_phases: int | None + # "retrying"/"aborted" badge from the backend; None = running normally. + agent_state: str | None = None @property def fraction(self) -> float: @@ -169,6 +171,7 @@ def workspace_bars(payload: dict[str, Any]) -> list[WorkspaceBar]: phase_name=data.get("phase_name") or "", last_completed_phase=int(data.get("last_completed_phase") or 0), total_phases=data.get("total_phases"), + agent_state=data.get("agent_state"), ) ) return bars @@ -300,6 +303,7 @@ def estimate_panel(payload: dict[str, Any] | None) -> EstimatePanel | None: "tool_result": "green", "result": "bold green", "system": "dim", + "error": "bold red", "unknown": "dim", } @@ -445,3 +449,71 @@ def format_tokens(value: int | None) -> str: if value < 1_000_000: return f"{value / 1000:.1f}K" return f"{value / 1_000_000:.1f}M" + + +# --------------------------------------------------------------------------- +# Agent error/warning events (durable `agent_error_events` from /status) +# --------------------------------------------------------------------------- + +# Event kind whose entries are pinned first in per-workspace warning views — +# a silent mid-run model switch is the one thing users must not miss. +MODEL_FALLBACK_KIND = "model_fallback" + +# Rich style per agent_state badge shown on the workspace list. +AGENT_STATE_BADGES: dict[str, tuple[str, str]] = { + "retrying": ("RETRYING", "bold yellow"), + "aborted": ("ABORTED", "bold red"), +} + + +@dataclass(frozen=True) +class AgentErrorEventRow: + """One agent error/warning event, display-ready.""" + + time: str + workspace_id: str + phase_label: str # "phase 12" or "" + kind: str + message: str + + +def agent_error_event_rows(payload: dict[str, Any] | None) -> list[AgentErrorEventRow]: + """Flatten ``agent_error_events`` into display rows (oldest first, as sent).""" + events = (payload or {}).get("agent_error_events") or [] + rows: list[AgentErrorEventRow] = [] + for event in events: + if not isinstance(event, dict): + continue + phase = event.get("phase") + rows.append( + AgentErrorEventRow( + time=_hhmmss(event.get("at")), + workspace_id=str(event.get("workspace_id") or ""), + phase_label=f"phase {phase}" if phase is not None else "", + kind=str(event.get("kind") or ""), + message=str(event.get("message") or ""), + ) + ) + return rows + + +def workspaces_with_events(payload: dict[str, Any] | None) -> set[str]: + """Workspace ids that reported at least one error/warning event.""" + return {row.workspace_id for row in agent_error_event_rows(payload) if row.workspace_id} + + +def workspace_warning_rows( + payload: dict[str, Any] | None, workspace_id: str +) -> list[AgentErrorEventRow]: + """This workspace's events for the drill-in block, model switches pinned first.""" + rows = [r for r in agent_error_event_rows(payload) if r.workspace_id == workspace_id] + pinned = [r for r in rows if r.kind == MODEL_FALLBACK_KIND] + others = [r for r in rows if r.kind != MODEL_FALLBACK_KIND] + return pinned + others + + +def agent_state_badge(agent_state: str | None) -> tuple[str, str] | None: + """(text, style) badge for a workspace agent_state; None when running normally.""" + if not agent_state: + return None + return AGENT_STATE_BADGES.get(str(agent_state).lower())