From dbade1b25e00da5f42e73e75aa1233d8dbaf2169 Mon Sep 17 00:00:00 2001 From: ciregenz Date: Thu, 6 Aug 2026 20:50:13 -0700 Subject: [PATCH] [eric] agents: a turn whose router died mid-flight re-ensures it and resumes instead of a terminal snag --- backend/apps/agents/core/error_classify.py | 19 ++++++ backend/apps/agents/manager/run/TurnRunner.py | 67 +++++++++++++------ backend/tests/test_capacity_retry.py | 31 +++++++++ backend/tests/test_router_respawn_retry.py | 41 ++++++++++++ 4 files changed, 138 insertions(+), 20 deletions(-) create mode 100644 backend/tests/test_router_respawn_retry.py diff --git a/backend/apps/agents/core/error_classify.py b/backend/apps/agents/core/error_classify.py index 27e915d3..2d2b382c 100644 --- a/backend/apps/agents/core/error_classify.py +++ b/backend/apps/agents/core/error_classify.py @@ -49,6 +49,25 @@ NON_TRANSIENT_PATTERNS = re.compile( ) +@typechecked +def is_router_unreachable_error(text: str) -> bool: + """True when a turn-result error is the CLI failing to REACH its endpoint (our localhost + 9Router, which every provider call goes through). A dev reload kills and respawns the router, + so this is a seconds-long outage: the caller re-ensures the router and resumes the turn + instead of surfacing a terminal error card.""" + if not text.strip(): + return False + return bool(re.search( + r"unable\s+to\s+connect" + r"|econnrefused" + r"|connection\s+refused" + r"|fetch\s+failed" + r"|connection\s+error", + text, + re.IGNORECASE, + )) + + @typechecked def is_long_context_error(exc: BaseException, extra_text: str = "") -> bool: """True when the upstream error is the 'long context tier required' 429. diff --git a/backend/apps/agents/manager/run/TurnRunner.py b/backend/apps/agents/manager/run/TurnRunner.py index c39f3464..38d59121 100644 --- a/backend/apps/agents/manager/run/TurnRunner.py +++ b/backend/apps/agents/manager/run/TurnRunner.py @@ -11,7 +11,7 @@ from typeguard import typechecked from backend.apps.agents.core.models import AgentSession from backend.apps.agents.core.ws_manager import ws_manager -from backend.apps.agents.core.error_classify import CAPACITY_BACKOFFS, capacity_retry_wait +from backend.apps.agents.core.error_classify import CAPACITY_BACKOFFS, capacity_retry_wait, is_router_unreachable_error from backend.apps.agents.manager.streaming.state import ThinkingState, TurnState from backend.apps.agents.manager.streaming.handle_stream_event import handle_stream_event from backend.apps.agents.manager.streaming.handle_assistant_message import handle_assistant_message @@ -154,8 +154,28 @@ class TurnRunner(AgentManagerProtocol): await dispose_client(self.client_pool, session_id) raise + async def p_finalize_interrupted_stream(): + # Finalize any in-flight stream messages so the UI doesn't leave them pinned as "still streaming" while we wait and restart. On resume the CLI re-runs the last turn from scratch (Anthropic doesn't persist in-progress responses), so the partial assistant text / tool call we emitted is now orphaned, cap it with stream_end and start the fresh turn under a new message id. + if turn.stream_text_msg_id: + await ws_manager.send_to_session(session_id, "agent:stream_end", { + "session_id": session_id, + "message_id": turn.stream_text_msg_id, + }) + turn.stream_text_msg_id = None + turn.stream_text_accum = "" + self.live_partial.pop(session_id, None) + for p_tool_msg_id in turn.stream_tool_msg_ids_ordered: + await ws_manager.send_to_session(session_id, "agent:stream_end", { + "session_id": session_id, + "message_id": p_tool_msg_id, + }) + turn.stream_tool_msg_ids_ordered = [] + turn.stream_block_index_map = {} + turn.current_turn_emitted = False + p_use_persistent = persistent_client_enabled() capacity_retry_attempt = 0 + p_router_retry_attempt = 0 while True: try: if p_use_persistent: @@ -163,8 +183,31 @@ class TurnRunner(AgentManagerProtocol): else: await p_run_streaming_turn() break - except TurnResultError: - # The CLI already ran the whole turn (tools executed) and then reported failure; a resume-retry would re-execute side effects, so this goes straight to the error card. + except TurnResultError as p_result_err: + # "Unable to connect" in a turn result is the CLI failing to reach our own localhost + # router, which a dev reload kills and the watchdog revives within seconds. The CLI + # transcript keeps the tools that already ran, so a resume continues the SAME + # conversation without re-executing side effects: re-ensure the router, resume, go. + if p_router_retry_attempt < 2 and is_router_unreachable_error(str(p_result_err)): + p_router_retry_attempt += 1 + logger.warning( + f"Router unreachable mid-turn on session {session_id} " + f"(attempt {p_router_retry_attempt}/2); re-ensuring router and resuming. " + f"err={p_result_err!s}" + ) + try: + from backend.apps.nine_router.process import ensure_running + await ensure_running() + except Exception: + logger.exception("Router re-ensure failed; resuming anyway after the wait") + await p_finalize_interrupted_stream() + await asyncio.sleep(2.0 if p_router_retry_attempt == 1 else 5.0) + p_stderr_buffer.clear() + if session.sdk_session_id: + options_kwargs["resume"] = session.sdk_session_id + options = ClaudeAgentOptions(**options_kwargs) + continue + # Any other error-shaped result: the CLI already ran the whole turn (tools executed) and then reported failure; a resume-retry would re-execute side effects, so this goes straight to the error card. raise except Exception as e: # Make sure the consolidated-thinking ticker doesn't outlive the turn on error/retry. Without this, an exception mid-stream leaves a dangling task that keeps re-emitting against a stale msg id. @@ -192,23 +235,7 @@ class TurnRunner(AgentManagerProtocol): f"mid_stream={mid_stream}); sleeping {wait}s before retry. " f"exc={e!r} stderr_tail={stderr_snapshot[-400:]!r}" ) - # Finalize any in-flight stream messages so the UI doesn't leave them pinned as "still streaming" while we wait and restart. On resume the CLI re-runs the last turn from scratch (Anthropic doesn't persist in-progress responses), so the partial assistant text / tool call we emitted is now orphaned, cap it with stream_end and start the fresh turn under a new message id. - if turn.stream_text_msg_id: - await ws_manager.send_to_session(session_id, "agent:stream_end", { - "session_id": session_id, - "message_id": turn.stream_text_msg_id, - }) - turn.stream_text_msg_id = None - turn.stream_text_accum = "" - self.live_partial.pop(session_id, None) - for p_tool_msg_id in turn.stream_tool_msg_ids_ordered: - await ws_manager.send_to_session(session_id, "agent:stream_end", { - "session_id": session_id, - "message_id": p_tool_msg_id, - }) - turn.stream_tool_msg_ids_ordered = [] - turn.stream_block_index_map = {} - turn.current_turn_emitted = False + await p_finalize_interrupted_stream() await asyncio.sleep(wait) p_stderr_buffer.clear() if session.sdk_session_id: diff --git a/backend/tests/test_capacity_retry.py b/backend/tests/test_capacity_retry.py index 4fac75c3..5ad29db6 100644 --- a/backend/tests/test_capacity_retry.py +++ b/backend/tests/test_capacity_retry.py @@ -77,3 +77,34 @@ def test_an_auth_failure_stays_non_transient_even_when_it_is_a_transport_type(): def test_a_transport_error_that_says_nothing_at_all_still_retries(): # An exception stringifying to "" used to bail out before it was ever classified. assert capacity_retry_wait(httpx.ConnectError(""), 0) == 5 + + +# --- the router-respawn family: turn-RESULT errors, which bypass capacity_retry_wait entirely --- +# The CLI reports "API Error: Unable to connect" as an error-shaped ResultMessage when our +# localhost 9Router is mid-respawn (a dev reload kills it, the watchdog revives it in seconds). +# TurnRunner consults is_router_unreachable_error on the TurnResultError text and resumes the turn +# instead of surfacing a terminal card; these pin exactly which texts qualify. + +from backend.apps.agents.core.error_classify import is_router_unreachable_error + + +def test_the_cli_unable_to_connect_text_is_router_unreachable(): + live = ("The agent runtime reported this turn failed (error_during_execution). " + "API Error: Unable to connect. Is the computer able to access the url?") + assert is_router_unreachable_error(live) + + +def test_connection_refused_variants_are_router_unreachable(): + for text in ("ECONNREFUSED 127.0.0.1:20128", "connect: Connection refused", "fetch failed", "Connection error."): + assert is_router_unreachable_error(text), text + + +def test_ordinary_turn_failures_are_not_router_unreachable(): + for text in ( + "The model hit its maximum output length before finishing (max_tokens).", + "The model refused to continue this turn (refusal).", + "invalid_request_error: tool schema rejected", + "denied tools: Bash", + "", + ): + assert not is_router_unreachable_error(text), text diff --git a/backend/tests/test_router_respawn_retry.py b/backend/tests/test_router_respawn_retry.py new file mode 100644 index 00000000..4dbb8dfb --- /dev/null +++ b/backend/tests/test_router_respawn_retry.py @@ -0,0 +1,41 @@ +"""The router-respawn retry seam (task: a turn must survive the localhost router dying). + +Live-proven separately: a SIGKILLed router revives in ~1s and the CLI itself rides out 12-58s +outages. This file pins the LAST wall, the TurnRunner branch that catches the CLI's give-up shape +("API Error: Unable to connect" arriving as an error ResultMessage) and resumes instead of raising +straight to the error card, the way session 345a05eb died on 2026-08-06.""" + +import inspect + +from backend.apps.agents.core.error_classify import is_router_unreachable_error +from backend.apps.agents.manager.run import TurnRunner + + +def test_turn_result_error_consults_the_router_classifier_before_raising(): + src = inspect.getsource(TurnRunner) + handler = src.split("except TurnResultError", 1)[1] + body = handler.split("except Exception as e", 1)[0] + assert "is_router_unreachable_error" in body, "the router check must live on the TurnResultError path" + assert body.index("is_router_unreachable_error") < body.index("raise"), "classify BEFORE the unconditional raise" + + +def test_the_retry_re_ensures_the_router_and_resumes_the_same_conversation(): + src = inspect.getsource(TurnRunner) + body = src.split("except TurnResultError", 1)[1].split("except Exception as e", 1)[0] + assert "ensure_running" in body, "the retry must actively revive the router, not just wait" + assert 'options_kwargs["resume"]' in body, "the retry must resume the CLI conversation" + assert "continue" in body + + +def test_the_retry_is_capped_so_a_dead_router_still_surfaces(): + src = inspect.getsource(TurnRunner) + body = src.split("except TurnResultError", 1)[1].split("except Exception as e", 1)[0] + assert "p_router_retry_attempt < 2" in body, "two attempts, then the honest error card" + + +def test_the_exact_live_incident_text_qualifies(): + # Verbatim shape from session 345a05eb7ca5470d9585b98618e81002 (2026-08-06 17:38:05). + assert is_router_unreachable_error( + "The agent runtime reported this turn failed (error_during_execution). " + "API Error: Unable to connect. Is the computer able to access the url?" + )