From 993e873eba03940a4cce56eb83b19410f9bd7eb9 Mon Sep 17 00:00:00 2001 From: Mateusz Konopelski Date: Mon, 13 Jul 2026 17:45:31 +0200 Subject: [PATCH 1/2] fix(generation): handle transient API connection errors gracefully A lost API connection (laptop asleep / no internet) during generation was silently swallowed: connection errors were unclassified, a single blip could falsely mark a phase completed, and a total failure still reported "complete". - classify CONNECTION_ERROR in model_routing via a narrow pattern list - abort a connection-lost phase before the checkpoint write so a resume re-runs it instead of skipping it - surface all-workspace failures (generation + deploy) so the run fails cleanly with a "run retry_generation once you're back online" message; partial success still continues (fail only when every workspace fails) Connection retries remain the Claude SDK default (no custom wait-loop). Backend: 2016 passed / 36 skipped. --- agents/IMPLEMENTATION.md | 2 + backend/app/schemas/agent.py | 1 + backend/app/services/claude_code.py | 10 ++ backend/app/services/model_routing.py | 15 +++ backend/app/services/workflow_steps.py | 48 +++++++- backend/test/test_all_workspaces_failed.py | 75 ++++++++++++ backend/test/test_classify_error.py | 38 +++++- ...est_execute_all_phases_connection_error.py | 111 ++++++++++++++++++ 8 files changed, 297 insertions(+), 3 deletions(-) create mode 100644 backend/test/test_all_workspaces_failed.py create mode 100644 backend/test/test_execute_all_phases_connection_error.py diff --git a/agents/IMPLEMENTATION.md b/agents/IMPLEMENTATION.md index 2f75b3e..0c3e204 100644 --- a/agents/IMPLEMENTATION.md +++ b/agents/IMPLEMENTATION.md @@ -30,6 +30,8 @@ ## Recently Completed (June 2026) +- **Graceful transient connection-error handling (Jul 13)**: A lost API connection (laptop asleep / no internet) during generation used to be swallowed — a single blip could even mark a phase `completed` and a total failure still emailed "complete". Now: (1) `model_routing.classify_error` recognizes `AgentErrorType.CONNECTION_ERROR` via a narrow pattern list (`unable to connect to api`, `socket connection was closed`, `connection refused/reset`, `server disconnected`, …), checked before the 4xx/5xx guard. (2) `execute_all_phases` raises `WorkspaceAbortedError` on a `CONNECTION_ERROR` phase **before** the checkpoint write, so a lost phase is never falsely recorded (a later `retry_generation` re-runs it). (3) New `workflow_steps._raise_if_all_workspaces_failed` inspects `execute_parallel` results for both generation and deploy: fails the run only when **every** workspace failed (partial success continues), with a friendly "run retry_generation once you're back online" message when all failures are connection losses — so the orchestrator reaches `fail()` (correct FAILED, workspaces preserved per Commandment II, resumable). **Retry itself is left to the Claude SDK default** (no custom wait-loop / config knobs). Failure-notification was intentionally out of scope. Tests: `test_classify_error.py::TestConnectionError`, `test_execute_all_phases_connection_error.py`, `test_all_workspaces_failed.py` (backend 2016 passed / 36 skipped). Pre-existing, untouched mypy errors remain in `app/state/workspace_models.py:73`, `app/services/openrouter_api.py:21`, `app/services/model_catalog.py:80`. + - **Quickstart P10Y repository lookup pagination (Jul 8)**: Fixed Quickstart initialization for P10Y organizations with more than one repository page. `create_generation_session_repos.py` now searches generated repos with the qualified GitHub owner/prefix (for example `org/specflow-workspace`) and uses a shared paginated P10Y repository listing helper for ID resolution, re-fetch connection discovery, status checks, and status polling. Tests: `uv run pytest test/scripts/test_create_generation_session_repos.py -q` from `backend` (25 passed). - **TUI workspace notification dedupe (Jul 1)**: Fixed workspace drill-in causing repeated desktop "Workspace phase progressed" notifications every poll during KB init. `MilestoneTracker` now emits workspace progress only when `last_completed_phase` increases, so label-only differences between session-list and `/status` payloads stay silent. `WorkspaceMessagesScreen` feeds its `/status` poll into the shared per-run tracker, and the app-wide active-session watcher excludes the workspace screen's generation to avoid mixed payload sources. Tests: `test_tui_poller.py::TestMilestoneTracker::test_kb_init_phase_label_change_without_progress_is_silent`, `test_tui_app.py::TestWorkspaceDrillIn::test_workspace_screen_excludes_app_level_session_watcher`; broader TUI check: `uv run pytest tests/test_tui_app.py tests/test_tui_poller.py -v` from `mcp_server` (78 passed). diff --git a/backend/app/schemas/agent.py b/backend/app/schemas/agent.py index 7d08310..89ec08a 100644 --- a/backend/app/schemas/agent.py +++ b/backend/app/schemas/agent.py @@ -6,6 +6,7 @@ class AgentErrorType(str, Enum): TOOL_CALL_FAILURE = "tool_call_failure" MODEL_ROUTING_FAILURE = "model_routing_failure" + CONNECTION_ERROR = "connection_error" @dataclass diff --git a/backend/app/services/claude_code.py b/backend/app/services/claude_code.py index 845eccb..7d39e63 100644 --- a/backend/app/services/claude_code.py +++ b/backend/app/services/claude_code.py @@ -1270,6 +1270,16 @@ async def execute_all_phases( f"(model={workspace.model}). Aborting workspace — other workspaces continue. " f"Error: {phase_result.result}" ) + if error_type == AgentErrorType.CONNECTION_ERROR: + # The Claude SDK's own retries were already exhausted, so the connection is down + # for more than a transient blip. Abort BEFORE the checkpoint write below so the + # phase is never falsely marked completed — a later retry_generation must re-run + # it, not skip it. Workspaces stay ALLOCATED; the run fails cleanly and resumes. + 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}" + ) consecutive_errors += 1 logger.error( f"[{workspace_name}] Phase {phase_num} reported is_error=True " diff --git a/backend/app/services/model_routing.py b/backend/app/services/model_routing.py index 04899e1..0d68c9f 100644 --- a/backend/app/services/model_routing.py +++ b/backend/app/services/model_routing.py @@ -38,11 +38,26 @@ "empty response", ] +# Transient network/connection failures (laptop asleep, lost internet) — retryable, no HTTP status. +_CONNECTION_ERROR_PATTERNS = [ + "unable to connect to api", + "connectionrefused", + "connection refused", + "socket connection was closed", + "connection error", + "connection reset", + "server disconnected", + "network is unreachable", +] + def classify_error(error_msg: str, api_error_status: Optional[int] = None) -> Optional[AgentErrorType]: """Classify an agent error for targeted logging and telemetry.""" msg_lower = error_msg.lower() if any(p.lower() in msg_lower for p in _TOOL_CALL_ERROR_PATTERNS): return AgentErrorType.TOOL_CALL_FAILURE + # Connection failure checked before the 4xx/5xx guard (there is no HTTP status) and before routing. + if any(p in msg_lower for p in _CONNECTION_ERROR_PATTERNS): + return AgentErrorType.CONNECTION_ERROR # 4xx/5xx = definitive HTTP error, not a routing failure. Routing failures return HTTP 200 # with a malformed body, so api_error_status=200 (or None) must NOT be excluded here. if api_error_status is not None and api_error_status >= 400: diff --git a/backend/app/services/workflow_steps.py b/backend/app/services/workflow_steps.py index fa5b824..236e953 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.services.parallel_executor import ParallelAgentExecutor, ParallelAgentResult +from app.services.model_routing import classify_error +from app.schemas.agent import AgentErrorType from app.services.planning_parser import ( construct_planning_result, parse_applicable_agent_mcps, @@ -968,6 +970,41 @@ async def run_contract_validator(ctx: WorkflowContext) -> None: await _update_plan_firestore(ctx, plan_type, new_plan) +def _raise_if_all_workspaces_failed( + results: list, + *, + phase_label: str, + logger: logging.Logger, +) -> None: + """Surface a swallowed all-workspace failure so the orchestrator marks the run FAILED. + + ``ParallelAgentExecutor`` converts every per-workspace error into ``success=False`` and never + raises, so without this a total failure would let the orchestrator advance the checkpoint and + report success. Policy (D2): fail only when *every* workspace failed — if at least one + succeeded, log and continue with the survivors. When every failure is a lost API connection, + the message is friendly and actionable (the run resumes via ``retry_generation``). + """ + parallel = [r for r in (results or []) if isinstance(r, ParallelAgentResult)] + failures = [r for r in parallel if not r.success] + if not failures: + return + detail = "; ".join(f"{r.workspace_name}: {r.error}" for r in failures) + if len(failures) < len(parallel): + logger.warning( + "%s: %d/%d workspaces failed but the rest succeeded — continuing with survivors. " + "Failed: %s", + phase_label, len(failures), len(parallel), detail, + ) + return + if all(classify_error(r.error or "") == AgentErrorType.CONNECTION_ERROR for r in failures): + raise RuntimeError( + "Lost connection to the API — likely a network interruption (e.g. the machine went " + "offline). Your workspaces are preserved; run retry_generation once you're back online " + "to resume from the last completed phase." + ) + raise RuntimeError(f"All workspaces failed during {phase_label}: {detail}") + + async def run_generation_phases(ctx: WorkflowContext) -> None: """ Run parallel code generation phases across all workspaces. @@ -1060,6 +1097,10 @@ async def agent_fn(workspace: WorkspaceSettings, manager: WorkspaceManager, logg exc_info=True, ) + _raise_if_all_workspaces_failed( + ctx.generation_result, phase_label="code generation", logger=ctx.logger + ) + async def _build_deploy_github_context(generation_session_service, workspace_id: str) -> DeployGithubContext: """Return deploy context for the agent prompt. @@ -1221,7 +1262,10 @@ 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) + _raise_if_all_workspaces_failed( + deploy_results, phase_label="deploy & E2E", logger=ctx.logger + ) async def run_archive_outputs_step(ctx: WorkflowContext) -> None: diff --git a/backend/test/test_all_workspaces_failed.py b/backend/test/test_all_workspaces_failed.py new file mode 100644 index 0000000..953c557 --- /dev/null +++ b/backend/test/test_all_workspaces_failed.py @@ -0,0 +1,75 @@ +""" +_raise_if_all_workspaces_failed: turns a swallowed all-workspace failure into a real +exception so the orchestrator marks the run FAILED (instead of advancing the checkpoint +and emailing "complete"). Partial success survives (D2); a total connection loss gets a +friendly, actionable message. +""" + +from __future__ import annotations + +import logging +from unittest.mock import Mock + +import pytest + +from app.services.parallel_executor import ParallelAgentResult +from app.services.workflow_steps import _raise_if_all_workspaces_failed + +_LOG = logging.getLogger("test") + + +def _result(name: str, *, success: bool, error: str | None = None) -> ParallelAgentResult: + return ParallelAgentResult( + workspace_name=name, + workspace_settings=Mock(), + result=None, + success=success, + error=error, + ) + + +def test_all_failed_connection_raises_friendly() -> None: + results = [ + _result("ws-1", success=False, + error="[ws-1] Phase 2 lost connection to the API. Error: API Error: Unable to connect to API (ConnectionRefused)"), + _result("ws-2", success=False, + error="[ws-2] Phase 1 lost connection to the API. Error: The socket connection was closed unexpectedly."), + ] + with pytest.raises(RuntimeError) as exc: + _raise_if_all_workspaces_failed(results, phase_label="code generation", logger=_LOG) + msg = str(exc.value).lower() + assert "lost connection" in msg + assert "retry_generation" in msg + + +def test_all_failed_generic_raises_with_details() -> None: + results = [ + _result("ws-1", success=False, error="Aborting after 2 consecutive phase errors"), + _result("ws-2", success=False, error="tool-call incompatibility"), + ] + with pytest.raises(RuntimeError) as exc: + _raise_if_all_workspaces_failed(results, phase_label="code generation", logger=_LOG) + msg = str(exc.value) + assert "All workspaces failed during code generation" in msg + assert "ws-1" in msg and "ws-2" in msg + # Not a connection failure → must NOT show the connection guidance. + assert "retry_generation" not in msg + + +def test_partial_success_does_not_raise() -> None: + results = [ + _result("ws-1", success=True), + _result("ws-2", success=False, error="Unable to connect to API (ConnectionRefused)"), + ] + # One workspace succeeded → survivors continue (D2), no raise. + _raise_if_all_workspaces_failed(results, phase_label="code generation", logger=_LOG) + + +def test_all_success_does_not_raise() -> None: + results = [_result("ws-1", success=True), _result("ws-2", success=True)] + _raise_if_all_workspaces_failed(results, phase_label="code generation", logger=_LOG) + + +def test_empty_results_does_not_raise() -> None: + _raise_if_all_workspaces_failed([], phase_label="code generation", logger=_LOG) + _raise_if_all_workspaces_failed(None, phase_label="code generation", logger=_LOG) diff --git a/backend/test/test_classify_error.py b/backend/test/test_classify_error.py index 0b5788a..0593029 100644 --- a/backend/test/test_classify_error.py +++ b/backend/test/test_classify_error.py @@ -48,7 +48,9 @@ def test_empty_response_case_insensitive(self): assert classify_error("EMPTY RESPONSE received") == AgentErrorType.MODEL_ROUTING_FAILURE def test_does_not_match_unrelated_errors(self): - assert classify_error("connection refused") is None + # "rate limit" and generic "timeout" are intentionally NOT connection errors (out of + # scope for the transient-connection retry). "connection refused" now classifies as + # CONNECTION_ERROR — see TestConnectionError. assert classify_error("rate limit exceeded") is None assert classify_error("timeout after 30s") is None @@ -58,6 +60,40 @@ def test_tool_call_takes_priority_over_routing(self): assert classify_error(msg) == AgentErrorType.TOOL_CALL_FAILURE +class TestConnectionError: + """Transient network/connection failures — retryable, HTTP-status-independent.""" + + def test_unable_to_connect_to_api(self): + # The exact string the SDK emits when the laptop has no internet. + assert classify_error("API Error: Unable to connect to API (ConnectionRefused)") == AgentErrorType.CONNECTION_ERROR + + def test_socket_connection_closed(self): + assert classify_error("API Error: The socket connection was closed unexpectedly.") == AgentErrorType.CONNECTION_ERROR + + def test_connection_refused(self): + assert classify_error("connection refused") == AgentErrorType.CONNECTION_ERROR + + def test_connection_reset(self): + assert classify_error("Connection reset by peer") == AgentErrorType.CONNECTION_ERROR + + def test_server_disconnected(self): + assert classify_error("Server disconnected without sending a response.") == AgentErrorType.CONNECTION_ERROR + + def test_case_insensitive(self): + assert classify_error("UNABLE TO CONNECT TO API") == AgentErrorType.CONNECTION_ERROR + + def test_connection_wins_over_spurious_status_in_message(self): + # A stray number must not demote a real connection error to unclassified. + msg = "Unable to connect to API (ConnectionRefused) after 500ms" + inferred = extract_http_status_from_message(msg) + assert inferred == 500 # extractor would pick this up... + assert classify_error(msg, api_error_status=inferred) == AgentErrorType.CONNECTION_ERROR # ...but connection still wins + + def test_tool_call_takes_priority_over_connection(self): + # tool-call incompatibility must abort immediately, never be retried as transient. + assert classify_error("tool_use failed: connection error") == AgentErrorType.TOOL_CALL_FAILURE + + class TestApiErrorStatusExclusion: def test_known_http_error_not_classified_as_routing_failure(self): msg = "API Error: upstream overloaded" diff --git a/backend/test/test_execute_all_phases_connection_error.py b/backend/test/test_execute_all_phases_connection_error.py new file mode 100644 index 0000000..3916e52 --- /dev/null +++ b/backend/test/test_execute_all_phases_connection_error.py @@ -0,0 +1,111 @@ +""" +execute_all_phases: a lost API connection (surfaced after the SDK's own retries are +exhausted) must abort the workspace BEFORE the checkpoint write, so the phase is never +falsely recorded as completed — a later retry_generation must re-run it, not skip it. +""" + +from __future__ import annotations + +from pathlib import Path +from unittest.mock import AsyncMock, Mock, patch + +import pytest + +from app.schemas.agent import AgentResult +from app.schemas.planning import PhaseInfo, PlanningResult +from app.schemas.specification import GenerateAppRequest +from app.services.claude_code import WorkspaceAbortedError, execute_all_phases + + +def _mock_workspace(tmp_path: Path) -> Mock: + ws = Mock() + ws.workspace_path = tmp_path + ws.get_isolated_root = Mock(return_value=str(tmp_path)) + ws.model = "m" + ws.provider = "openrouter" + return ws + + +def _mock_manager() -> Mock: + mgr = Mock() + mgr.settings = Mock() + mgr.settings.ROSETTA_MCP_ENABLED = False + mgr.commit_and_push_outstanding = AsyncMock(return_value=True) + return mgr + + +def _planning_one_phase() -> PlanningResult: + return PlanningResult( + phase_count=1, + phases=[PhaseInfo(number=1, name="Backend", description="API", estimated_commits=1)], + plan_file_path="specflow/plan.md", + ) + + +@pytest.mark.asyncio +async def test_connection_error_aborts_without_checkpoint(tmp_path: Path) -> None: + """A connection-lost phase raises WorkspaceAbortedError and never writes the checkpoint.""" + planning = _planning_one_phase() + mock_ws = _mock_workspace(tmp_path) + mock_mgr = _mock_manager() + request = GenerateAppRequest(spec_path="specs", outputs_dir="specflow", generation_id="e1") + + svc = Mock() + svc.update_workspace_phase = AsyncMock() + svc.update_deployment_workspace_phase = AsyncMock() + + async def fake_phase_fn(**kwargs): + return AgentResult( + result="API Error: Unable to connect to API (ConnectionRefused)", + session_id=None, + is_error=True, + ) + + with patch("app.services.claude_code.phase_agent_fn", side_effect=fake_phase_fn), \ + patch("app.services.claude_code.is_skip_mode_enabled", return_value=False): + with pytest.raises(WorkspaceAbortedError) as exc_info: + await execute_all_phases( + planning_data=planning, + workspace=mock_ws, + manager=mock_mgr, + request=request, + logger=Mock(), + generation_session_service=svc, + workspace_id="ws-1", + ) + + assert "lost connection" in str(exc_info.value).lower() + # 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() + + +@pytest.mark.asyncio +async def test_connection_error_aborts_on_first_occurrence(tmp_path: Path) -> None: + """Unlike a generic error (tolerated up to 2), a single connection loss aborts immediately.""" + planning = _planning_one_phase() + request = GenerateAppRequest(spec_path="specs", outputs_dir="specflow", generation_id="e1") + + calls = {"n": 0} + + async def fake_phase_fn(**kwargs): + calls["n"] += 1 + return AgentResult( + result="API Error: The socket connection was closed unexpectedly.", + session_id=None, + is_error=True, + ) + + with patch("app.services.claude_code.phase_agent_fn", side_effect=fake_phase_fn), \ + patch("app.services.claude_code.is_skip_mode_enabled", return_value=False): + with pytest.raises(WorkspaceAbortedError): + await execute_all_phases( + planning_data=planning, + workspace=_mock_workspace(tmp_path), + manager=_mock_manager(), + request=request, + logger=Mock(), + ) + + # Aborted after the very first errored phase — no second attempt. + assert calls["n"] == 1 From 3e35205a68b7279209dc26650c05e82895905281 Mon Sep 17 00:00:00 2001 From: Mateusz Konopelski Date: Mon, 13 Jul 2026 17:51:01 +0200 Subject: [PATCH 2/2] fix(generation): trim comment, drop all-workspace-failure surfacing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Remove the multi-line comment on the CONNECTION_ERROR abort branch to match the terser style of the sibling TOOL_CALL_FAILURE case, and drop _raise_if_all_workspaces_failed (and its test) — surfacing an all-workspace failure through the orchestrator is out of scope for this PR. --- agents/IMPLEMENTATION.md | 2 +- backend/app/services/claude_code.py | 4 -- backend/app/services/workflow_steps.py | 48 +------------- backend/test/test_all_workspaces_failed.py | 75 ---------------------- 4 files changed, 3 insertions(+), 126 deletions(-) delete mode 100644 backend/test/test_all_workspaces_failed.py diff --git a/agents/IMPLEMENTATION.md b/agents/IMPLEMENTATION.md index 0c3e204..fd8ae9b 100644 --- a/agents/IMPLEMENTATION.md +++ b/agents/IMPLEMENTATION.md @@ -30,7 +30,7 @@ ## Recently Completed (June 2026) -- **Graceful transient connection-error handling (Jul 13)**: A lost API connection (laptop asleep / no internet) during generation used to be swallowed — a single blip could even mark a phase `completed` and a total failure still emailed "complete". Now: (1) `model_routing.classify_error` recognizes `AgentErrorType.CONNECTION_ERROR` via a narrow pattern list (`unable to connect to api`, `socket connection was closed`, `connection refused/reset`, `server disconnected`, …), checked before the 4xx/5xx guard. (2) `execute_all_phases` raises `WorkspaceAbortedError` on a `CONNECTION_ERROR` phase **before** the checkpoint write, so a lost phase is never falsely recorded (a later `retry_generation` re-runs it). (3) New `workflow_steps._raise_if_all_workspaces_failed` inspects `execute_parallel` results for both generation and deploy: fails the run only when **every** workspace failed (partial success continues), with a friendly "run retry_generation once you're back online" message when all failures are connection losses — so the orchestrator reaches `fail()` (correct FAILED, workspaces preserved per Commandment II, resumable). **Retry itself is left to the Claude SDK default** (no custom wait-loop / config knobs). Failure-notification was intentionally out of scope. Tests: `test_classify_error.py::TestConnectionError`, `test_execute_all_phases_connection_error.py`, `test_all_workspaces_failed.py` (backend 2016 passed / 36 skipped). Pre-existing, untouched mypy errors remain in `app/state/workspace_models.py:73`, `app/services/openrouter_api.py:21`, `app/services/model_catalog.py:80`. +- **Graceful transient connection-error handling (Jul 13)**: A lost API connection (laptop asleep / no internet) during generation used to be swallowed — a single blip could mark a phase `completed` that never actually ran. Now: (1) `model_routing.classify_error` recognizes `AgentErrorType.CONNECTION_ERROR` via a narrow pattern list (`unable to connect to api`, `socket connection was closed`, `connection refused/reset`, `server disconnected`, …), checked before the 4xx/5xx guard. (2) `execute_all_phases` raises `WorkspaceAbortedError` on a `CONNECTION_ERROR` phase **before** the checkpoint write, so a lost phase is never falsely recorded (a later `retry_generation` re-runs it). **Retry itself is left to the Claude SDK default** (no custom wait-loop / config knobs). Surfacing an all-workspace failure up through the orchestrator (so a total connection loss ends in FAILED rather than "complete") and failure-notification were both intentionally descoped from this PR. Tests: `test_classify_error.py::TestConnectionError`, `test_execute_all_phases_connection_error.py` (backend 2011 passed / 36 skipped). Pre-existing, untouched mypy errors remain in `app/state/workspace_models.py:73`, `app/services/openrouter_api.py:21`, `app/services/model_catalog.py:80`. - **Quickstart P10Y repository lookup pagination (Jul 8)**: Fixed Quickstart initialization for P10Y organizations with more than one repository page. `create_generation_session_repos.py` now searches generated repos with the qualified GitHub owner/prefix (for example `org/specflow-workspace`) and uses a shared paginated P10Y repository listing helper for ID resolution, re-fetch connection discovery, status checks, and status polling. Tests: `uv run pytest test/scripts/test_create_generation_session_repos.py -q` from `backend` (25 passed). diff --git a/backend/app/services/claude_code.py b/backend/app/services/claude_code.py index 7d39e63..3fc7e9d 100644 --- a/backend/app/services/claude_code.py +++ b/backend/app/services/claude_code.py @@ -1271,10 +1271,6 @@ async def execute_all_phases( f"Error: {phase_result.result}" ) if error_type == AgentErrorType.CONNECTION_ERROR: - # The Claude SDK's own retries were already exhausted, so the connection is down - # for more than a transient blip. Abort BEFORE the checkpoint write below so the - # phase is never falsely marked completed — a later retry_generation must re-run - # it, not skip it. Workspaces stay ALLOCATED; the run fails cleanly and resumes. raise WorkspaceAbortedError( f"[{workspace_name}] Phase {phase_num} lost connection to the API " f"(model={workspace.model}). Aborting workspace — other workspaces continue. " diff --git a/backend/app/services/workflow_steps.py b/backend/app/services/workflow_steps.py index 236e953..fa5b824 100644 --- a/backend/app/services/workflow_steps.py +++ b/backend/app/services/workflow_steps.py @@ -68,9 +68,7 @@ get_common_allowed_tools, ) from app.services.generation_session_output_protection import protect_workspace_after_generation -from app.services.parallel_executor import ParallelAgentExecutor, ParallelAgentResult -from app.services.model_routing import classify_error -from app.schemas.agent import AgentErrorType +from app.services.parallel_executor import ParallelAgentExecutor from app.services.planning_parser import ( construct_planning_result, parse_applicable_agent_mcps, @@ -970,41 +968,6 @@ async def run_contract_validator(ctx: WorkflowContext) -> None: await _update_plan_firestore(ctx, plan_type, new_plan) -def _raise_if_all_workspaces_failed( - results: list, - *, - phase_label: str, - logger: logging.Logger, -) -> None: - """Surface a swallowed all-workspace failure so the orchestrator marks the run FAILED. - - ``ParallelAgentExecutor`` converts every per-workspace error into ``success=False`` and never - raises, so without this a total failure would let the orchestrator advance the checkpoint and - report success. Policy (D2): fail only when *every* workspace failed — if at least one - succeeded, log and continue with the survivors. When every failure is a lost API connection, - the message is friendly and actionable (the run resumes via ``retry_generation``). - """ - parallel = [r for r in (results or []) if isinstance(r, ParallelAgentResult)] - failures = [r for r in parallel if not r.success] - if not failures: - return - detail = "; ".join(f"{r.workspace_name}: {r.error}" for r in failures) - if len(failures) < len(parallel): - logger.warning( - "%s: %d/%d workspaces failed but the rest succeeded — continuing with survivors. " - "Failed: %s", - phase_label, len(failures), len(parallel), detail, - ) - return - if all(classify_error(r.error or "") == AgentErrorType.CONNECTION_ERROR for r in failures): - raise RuntimeError( - "Lost connection to the API — likely a network interruption (e.g. the machine went " - "offline). Your workspaces are preserved; run retry_generation once you're back online " - "to resume from the last completed phase." - ) - raise RuntimeError(f"All workspaces failed during {phase_label}: {detail}") - - async def run_generation_phases(ctx: WorkflowContext) -> None: """ Run parallel code generation phases across all workspaces. @@ -1097,10 +1060,6 @@ async def agent_fn(workspace: WorkspaceSettings, manager: WorkspaceManager, logg exc_info=True, ) - _raise_if_all_workspaces_failed( - ctx.generation_result, phase_label="code generation", logger=ctx.logger - ) - async def _build_deploy_github_context(generation_session_service, workspace_id: str) -> DeployGithubContext: """Return deploy context for the agent prompt. @@ -1262,10 +1221,7 @@ async def agent_fn( enabled_mcps=ctx.enabled_mcps, ) - deploy_results = await executor.execute_parallel(workspaces=ctx.workspaces, agent_fn=agent_fn) - _raise_if_all_workspaces_failed( - deploy_results, phase_label="deploy & E2E", logger=ctx.logger - ) + await executor.execute_parallel(workspaces=ctx.workspaces, agent_fn=agent_fn) async def run_archive_outputs_step(ctx: WorkflowContext) -> None: diff --git a/backend/test/test_all_workspaces_failed.py b/backend/test/test_all_workspaces_failed.py deleted file mode 100644 index 953c557..0000000 --- a/backend/test/test_all_workspaces_failed.py +++ /dev/null @@ -1,75 +0,0 @@ -""" -_raise_if_all_workspaces_failed: turns a swallowed all-workspace failure into a real -exception so the orchestrator marks the run FAILED (instead of advancing the checkpoint -and emailing "complete"). Partial success survives (D2); a total connection loss gets a -friendly, actionable message. -""" - -from __future__ import annotations - -import logging -from unittest.mock import Mock - -import pytest - -from app.services.parallel_executor import ParallelAgentResult -from app.services.workflow_steps import _raise_if_all_workspaces_failed - -_LOG = logging.getLogger("test") - - -def _result(name: str, *, success: bool, error: str | None = None) -> ParallelAgentResult: - return ParallelAgentResult( - workspace_name=name, - workspace_settings=Mock(), - result=None, - success=success, - error=error, - ) - - -def test_all_failed_connection_raises_friendly() -> None: - results = [ - _result("ws-1", success=False, - error="[ws-1] Phase 2 lost connection to the API. Error: API Error: Unable to connect to API (ConnectionRefused)"), - _result("ws-2", success=False, - error="[ws-2] Phase 1 lost connection to the API. Error: The socket connection was closed unexpectedly."), - ] - with pytest.raises(RuntimeError) as exc: - _raise_if_all_workspaces_failed(results, phase_label="code generation", logger=_LOG) - msg = str(exc.value).lower() - assert "lost connection" in msg - assert "retry_generation" in msg - - -def test_all_failed_generic_raises_with_details() -> None: - results = [ - _result("ws-1", success=False, error="Aborting after 2 consecutive phase errors"), - _result("ws-2", success=False, error="tool-call incompatibility"), - ] - with pytest.raises(RuntimeError) as exc: - _raise_if_all_workspaces_failed(results, phase_label="code generation", logger=_LOG) - msg = str(exc.value) - assert "All workspaces failed during code generation" in msg - assert "ws-1" in msg and "ws-2" in msg - # Not a connection failure → must NOT show the connection guidance. - assert "retry_generation" not in msg - - -def test_partial_success_does_not_raise() -> None: - results = [ - _result("ws-1", success=True), - _result("ws-2", success=False, error="Unable to connect to API (ConnectionRefused)"), - ] - # One workspace succeeded → survivors continue (D2), no raise. - _raise_if_all_workspaces_failed(results, phase_label="code generation", logger=_LOG) - - -def test_all_success_does_not_raise() -> None: - results = [_result("ws-1", success=True), _result("ws-2", success=True)] - _raise_if_all_workspaces_failed(results, phase_label="code generation", logger=_LOG) - - -def test_empty_results_does_not_raise() -> None: - _raise_if_all_workspaces_failed([], phase_label="code generation", logger=_LOG) - _raise_if_all_workspaces_failed(None, phase_label="code generation", logger=_LOG)