diff --git a/backend/apps/agents/agents.py b/backend/apps/agents/agents.py index c6c83a0c..11176d08 100644 --- a/backend/apps/agents/agents.py +++ b/backend/apps/agents/agents.py @@ -31,10 +31,14 @@ async def agents_lifespan(): await agent_manager.restore_all_sessions() # Off the critical path: crash-cut turns resume themselves once everything is hydrated. asyncio.create_task(agent_manager.auto_resume_crashed_turns()) + # Subscription logins are renewed ahead of expiry while the app runs; one the router cannot renew is reported at once. + from backend.apps.nine_router.oauth_refresh import oauth_refresh_loop + p_oauth_refresh = asyncio.create_task(oauth_refresh_loop()) from backend.apps.agents.manager.run.client_pool import start_pool_sweeper, stop_pool_sweeper, dispose_all_clients pool_sweeper = start_pool_sweeper(agent_manager.client_pool) yield logger.info("Agents sub-app shutting down") + p_oauth_refresh.cancel() # Stamp before stopping: once stop_agent has run, a live chat is indistinguishable from one the user stopped. agent_manager.note_shutdown_stops() for session_id in list(agent_manager.tasks.keys()): @@ -567,6 +571,24 @@ async def subscriptions_connect(body: dict, request: Request): raise HTTPException(status_code=500, detail=str(e)) +async def p_after_reconnect(provider: str, bounce) -> None: + """The reconnect chokepoint: the router restarts once so its memory of the old token goes, then every chat + that died on this login resumes by itself. Nothing here may raise into the OAuth response.""" + try: + await bounce(provider) + except Exception: + logger.debug("post-reconnect bounce failed", exc_info=True) + # The old verdict dies with the old token: the pill closes and the next ask probes the new login. + from backend.apps.nine_router.subscription_health import invalidate_health_cache + invalidate_health_cache() + try: + n = await agent_manager.resume_auth_dead_sessions(provider) + if n: + logger.info(f"reconnect: {n} chat(s) that died on {provider} resumed") + except Exception: + logger.warning("reconnect-resume failed", exc_info=True) + + @agents.router.post("/subscriptions/poll") async def subscriptions_poll(body: dict): """Poll for OAuth completion.""" @@ -590,7 +612,7 @@ async def subscriptions_poll(body: dict): await clear_free_trial_on_connect() # Background so the UI's "Connected" lands instantly; the bounce takes ~5-10s (ENG-315). from backend.apps.nine_router.bounce_after_connect import bounce_router_after_connect - asyncio.create_task(bounce_router_after_connect(provider)) + asyncio.create_task(p_after_reconnect(provider, bounce_router_after_connect)) return result except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @@ -627,6 +649,8 @@ async def subscriptions_exchange(body: dict): # A connected subscription takes precedence over the free trial right away. from backend.apps.subscription.free_trial import clear_free_trial_on_connect await clear_free_trial_on_connect() + from backend.apps.nine_router.bounce_after_connect import bounce_router_after_connect + asyncio.create_task(p_after_reconnect(provider, bounce_router_after_connect)) return result except Exception as e: if state and state in completed_oauth: diff --git a/backend/apps/agents/core/models.py b/backend/apps/agents/core/models.py index 8eeaa38e..58420541 100644 --- a/backend/apps/agents/core/models.py +++ b/backend/apps/agents/core/models.py @@ -176,6 +176,10 @@ class AgentSession(BaseModel): # Input-token level history must regrow past before another proactive prune may commit; a rebuild busts the prompt cache, so one per runway, never one per turn. proactive_prune_rearm_tokens: int = 0 lane_credential_dead: bool = False + # The router login this chat died on (a definitive auth failure), so a reconnect of that login can pick the chat back up by itself; cleared on resume. + auth_dead_provider: Optional[str] = None + # The router login this chat dispatches through (claude, codex, gemini-cli, antigravity), written by the preflight on every turn; None on a direct API key. The error handler reads it instead of guessing from the vendor. + lane_provider: Optional[str] = None # Transient provider errors arrive as assistant TEXT, so no upstream retry sees them; budgeted apart from auth_retry_used so a rate limit cannot spend the expired-token retry. transient_retry_count: int = 0 # Set once the provider gives a verdict waiting cannot change (a spent plan, a dead credential). diff --git a/backend/apps/agents/manager/run/TurnRunner.py b/backend/apps/agents/manager/run/TurnRunner.py index 3d409794..d20e97db 100644 --- a/backend/apps/agents/manager/run/TurnRunner.py +++ b/backend/apps/agents/manager/run/TurnRunner.py @@ -298,10 +298,13 @@ class TurnRunner(AgentManagerProtocol): f"resuming once in {p_auth_wait}s. err={p_result_err!s}" ) try: - from backend.apps.nine_router.subscription_health import invalidate_health_cache - invalidate_health_cache() + from backend.apps.nine_router.subscription_health import note_auth_failure + from backend.apps.agents.manager.run.lane_preflight import provider_for_model + p_lane = provider_for_model(resolved_model) + if p_lane: + note_auth_failure(p_lane) except Exception: - logger.debug("health-cache invalidate before auth resume failed", exc_info=True) + logger.debug("health sighting before auth resume failed", exc_info=True) await p_finalize_interrupted_stream() await asyncio.sleep(p_auth_wait) p_stderr_buffer.clear() @@ -338,10 +341,13 @@ class TurnRunner(AgentManagerProtocol): flight_recorder.crumb(session_id, "auth-resume", wait_s=p_auth_wait2, err=str(e)[:160]) logger.warning(f"Auth-shaped exception on session {session_id}; refreshing and resuming once in {p_auth_wait2}s. exc={e!r}") try: - from backend.apps.nine_router.subscription_health import invalidate_health_cache - invalidate_health_cache() + from backend.apps.nine_router.subscription_health import note_auth_failure + from backend.apps.agents.manager.run.lane_preflight import provider_for_model + p_lane = provider_for_model(resolved_model) + if p_lane: + note_auth_failure(p_lane) except Exception: - logger.debug("health-cache invalidate before auth resume failed", exc_info=True) + logger.debug("health sighting before auth resume failed", exc_info=True) await p_finalize_interrupted_stream() await asyncio.sleep(p_auth_wait2) p_stderr_buffer.clear() diff --git a/backend/apps/agents/manager/run/handle_run_error.py b/backend/apps/agents/manager/run/handle_run_error.py index fa91baab..87e1ca26 100644 --- a/backend/apps/agents/manager/run/handle_run_error.py +++ b/backend/apps/agents/manager/run/handle_run_error.py @@ -162,6 +162,20 @@ async def p_try_runtime_repair(session, session_id: str) -> bool: return False +async def p_mark_login_dead(session: AgentSession) -> None: + """The turn and its one retry both failed auth on a router login: the chat remembers which login, so a + reconnect can pick it back up by itself, and the reconnect pill goes up this second (one door, one story).""" + lane = getattr(session, "lane_provider", None) + if not lane: + return + session.auth_dead_provider = lane + try: + from backend.apps.nine_router.subscription_health import report_dead_now + await report_dead_now(lane) + except Exception: + logger.debug("report_dead_now failed", exc_info=True) + + async def handle_run_error(e: Exception, session: AgentSession, session_id: str, turn: TurnState, p_stderr_buffer: List[str]) -> None: logger.exception(f"Agent {session_id} error: {e}") session.status = "error" @@ -522,6 +536,7 @@ async def handle_run_error(e: Exception, session: AgentSession, session_id: str, # rotation story is false and the wait is doomed; say the true thing straight away. if getattr(session, "lane_credential_dead", False): from backend.apps.agents.manager.run.lane_preflight import RECONNECT_COPY + await p_mark_login_dead(session) p_prov = (session.provider or "").lower() friendly_msg = RECONNECT_COPY.get( p_prov, @@ -594,6 +609,8 @@ async def handle_run_error(e: Exception, session: AgentSession, session_id: str, "reconnect Claude Pro / Max." ) reason = "anthropic_auth_invalid" + if reason in ("codex_token_rotating", "anthropic_auth_invalid", "openswarm_pro_auth_expired"): + await p_mark_login_dead(session) error_msg = Message(role="system", content=friendly_msg, branch_id=session.active_branch_id) absorb_repeat_card(session, error_msg) try: diff --git a/backend/apps/agents/manager/run/lane_preflight.py b/backend/apps/agents/manager/run/lane_preflight.py index a44c3141..6c18460d 100644 --- a/backend/apps/agents/manager/run/lane_preflight.py +++ b/backend/apps/agents/manager/run/lane_preflight.py @@ -1,42 +1,28 @@ -"""Do not spend a user's turn on a lane we already know is dead. +"""Do not spend a user's turn on a lane we already know is dead... and never restart the router for it. Live drill, 2026-08-20: a codex credential that expired 89 HOURS earlier produced "GPT subscription token just rotated (automatic, every couple minutes), retrying automatically, no action needed", a -75 second wait, a doomed retry, and then five identical cards. Zero files read. Every word of that -was wrong, and the evidence to know better was already sitting in 9Router's own provider list: -`testStatus: "unavailable"` with `errorCode: 401`, published before we spend anything. +75 second wait, a doomed retry, and then five identical cards. Zero files read. The evidence to know +better was already in 9Router's own provider list: `testStatus: "unavailable"` with `errorCode: 401`. -So this looks first, and it tries to fix it before it complains: +What this does with it: - healthy -> say nothing, cost nothing, dispatch as normal - sticky-dead, first -> bounce the router ONCE (re-reads db.json, clears the in-process - `unavailable` stamp and modelLock cooldowns), then DISPATCH ANYWAY and let - the turn itself be the verdict. If it goes through, the user never learns - anything happened. - sticky-dead, bounce - already throttled -> DISPATCH ANYWAY too, and flag the session so that if the turn really does - 401, handle_run_error can say the accurate sentence with no rotation story. + healthy -> say nothing, cost nothing, dispatch as normal + dead -> DISPATCH ANYWAY, and flag the session so that if the turn really does 401, + handle_run_error says the accurate sentence with no rotation story, and the + dead-login recheck pushes the reconnect pill. + +What it deliberately does NOT do any more (2026-09-06): restart the router. A restart cannot revive +a dead token; it is a dead port for every chat on every lane for 1 to 30 s (ENG-394's shape); and the +one thing it did fix, stale router memory after the user reconnected, is bounce_after_connect's job on +the reconnect path. The router's "unavailable" is a timed cooldown it clears on its own. This file NEVER tells the user a credential is dead, because it has not dispatched and therefore -cannot know. It used to, off the bounce cooldown, and that cooldown is a GLOBAL router-restart -throttle: the branch that meant "permanently dead" actually meant "another chat restarted the -router in the last five minutes". It killed a live build on a working credential while telling the -user "waiting will not clear this one", which was backwards, since waiting out the throttle is -exactly what cleared it (ENG-414). The death verdict lives in ONE place now, downstream of a real -failed dispatch. - -The bounce is NOT allowed to declare success on its own, and that mistake is worth recording: the -first version re-read the health flag afterwards and called a cleared stamp a recovery. But a fresh -router starts with no stamp, so the check passed for a credential that was still dead, and the turn -hit the same 401 seconds later. A restart clears the accusation, not the cause. Only a real -dispatch can tell you whether a credential works, so that is what decides it now. - -The bounce is the same one ENG-315 already runs at the connect chokepoint, and is documented safe -mid-session; the in-flight kill drill on 2026-08-20 confirmed a live turn survives one. +cannot know (ENG-414: a throttle read as death grounded a working lane). Only a real failed dispatch +may say that. """ import logging -import time from typing import Dict, Optional, TYPE_CHECKING from typeguard import typechecked @@ -49,10 +35,6 @@ logger = logging.getLogger(__name__) # Router prefix -> the provider name its connection is filed under. P_PREFIX_PROVIDER = {"cc/": "claude", "cx/": "codex", "gc/": "antigravity", "ag/": "antigravity"} -# A bounce restarts a process every other session shares, so one per lane per window, never per turn. -BOUNCE_COOLDOWN_S = 300 - -LAST_BOUNCE: Dict[str, float] = {} RECONNECT_COPY = { "codex": ("Your ChatGPT subscription needs reconnecting: the saved sign-in expired and could " @@ -128,6 +110,11 @@ async def preflight_lane(resolved_model: str, tried themselves. """ provider = provider_for_model(resolved_model) + if session is not None: + try: + session.lane_provider = provider + except Exception: + pass if provider is None: return None @@ -141,37 +128,14 @@ async def preflight_lane(resolved_model: str, pass if dead is None: return None - - now = time.time() - if now - LAST_BOUNCE.get(provider, 0.0) >= BOUNCE_COOLDOWN_S: - LAST_BOUNCE[provider] = now - logger.warning( - f"lane preflight: {provider} is {dead.get('testStatus')} (errorCode={dead.get('errorCode')}); " - "bouncing the router once, then letting the turn decide" - ) - p_back_up = False - try: - from backend.apps.nine_router.bounce_after_connect import bounce_router_after_connect - p_back_up = await bounce_router_after_connect(provider) - except Exception: - logger.debug("lane preflight bounce failed", exc_info=True) - if not p_back_up: - # Dispatching into a router that has not come back is a guaranteed connection error, and - # the user would read that as the model failing rather than us restarting something. - logger.warning("lane preflight: the router did not come back after the bounce; not dispatching into it") - return ("The local AI connection is restarting. This clears itself in a few seconds; " - "send your message again.") - # Deliberately no post-bounce health re-read: see the module docstring. Dispatch is the test. - return None - - # The bounce was throttled, and that says NOTHING about this credential: LAST_BOUNCE is global, - # so the timer belongs to whichever OTHER chat restarted the router last. Carding here declared - # a working lane dead and killed a live build (ENG-414), and it broke this file's own rule that - # only a real dispatch can decide. So dispatch. If the credential really is gone, the turn 401s - # and handle_run_error shows the accurate card immediately off `lane_credential_dead`, which is - # the same sentence this used to return, minus the guessing. - logger.info( - f"lane preflight: {provider} looks dead but the router bounce is throttled; dispatching " - "anyway and letting the turn decide" + # No restart here. A router bounce cannot revive a dead token, and it is a dead port for 1 to 30 s for + # EVERY chat on every lane (the ENG-394 shape) while this lane's own failure was going to be reported + # anyway. The one case a restart fixes, stale router memory after the user reconnected, has its own + # restart on the reconnect path (bounce_after_connect), and the router's "unavailable" is a timed + # cooldown it clears itself. Dispatch is the test: a dead credential 401s, the flag above makes that + # card honest at once, and the health recheck pushes the reconnect pill. + logger.warning( + f"lane preflight: {provider} reads dead in the router (testStatus={dead.get('testStatus')}, " + f"errorCode={dead.get('errorCode')}); dispatching so the real request decides, no restart" ) return None diff --git a/backend/apps/agents/manager/session/SessionPersistence.py b/backend/apps/agents/manager/session/SessionPersistence.py index 9fc15f9d..2aafc3ea 100644 --- a/backend/apps/agents/manager/session/SessionPersistence.py +++ b/backend/apps/agents/manager/session/SessionPersistence.py @@ -134,6 +134,34 @@ class SessionPersistence(AgentManagerProtocol): logger.warning(f"crash-resume: session {sid} failed to auto-resume; amber chip remains", exc_info=True) self.crash_resume_queue = [] + async def resume_auth_dead_sessions(self, provider: str) -> int: + """The user just reconnected this login: every chat that died on it picks itself back up with one hidden + continuation, the way crash-resume does at boot. Only chats marked by a DEFINITIVE auth failure on this + provider resume; a human's Stop always wins. Returns how many were sent.""" + resumed = 0 + for sid, session in list(self.sessions.items()): + if session.auth_dead_provider != provider or session.ended_by_user: + continue + if session.status in ("running", "waiting_approval"): + continue + session.auth_dead_provider = None + session.lane_credential_dead = False + session.auth_retry_used = False + session.provider_verdict_final = False + try: + p_send = getattr(self, "send_message") + await p_send( + sid, + "Your login for this model was reconnected; the earlier failure is cleared. Continue exactly " + "where you left off; do not redo completed steps.", + hidden=True, + ) + resumed += 1 + logger.info(f"reconnect-resume: session {sid} resumed on {provider}") + except Exception: + logger.warning(f"reconnect-resume: session {sid} failed to resume", exc_info=True) + return resumed + @typechecked def note_shutdown_stops(self) -> int: """Stamp every chat with a live turn BEFORE the shutdown stops it. The lifespan stops the tasks diff --git a/backend/apps/nine_router/oauth_refresh.py b/backend/apps/nine_router/oauth_refresh.py new file mode 100644 index 00000000..eb5ead05 --- /dev/null +++ b/backend/apps/nine_router/oauth_refresh.py @@ -0,0 +1,131 @@ +"""Keep every subscription login alive while the app runs, and say the truth the moment one cannot be. + +A login dies quietly: the short-lived token lapses while the app sits idle, nobody refreshes it, and +the next chat on it 401s. 9Router refreshes a token only when something asks it to test the +connection, and it refreshes only inside the last five minutes before expiry. So this asks, on a +clock, for every OAuth connection that is inside the margin. Two outcomes matter: the router +refreshed (nothing to say), or the router reports "refresh failed" / "expired", which is a verdict +waiting cannot change, so the dead-login pill goes up now instead of after the next failed chat. + +Deliberately narrow: only the router's own test route is used (it owns the tokens and db.json), a +lent connection (no refresh token, the cloud rotates it) is left to lent_credential_refresh, and a +router that is down or busy is retried on the next tick, never reported. +""" +from __future__ import annotations + +import asyncio +import logging +import time +from datetime import datetime +from typing import Dict, List, Optional + +import httpx +from typeguard import typechecked + +from backend.apps.nine_router import process +from backend.apps.nine_router.process import NINE_ROUTER_API, is_running + +logger = logging.getLogger(__name__) + +# The router refreshes inside its own 5-minute window; ask a little earlier so a slow refresh still lands before the deadline. +REFRESH_MARGIN_S = 20 * 60 +CHECK_INTERVAL_S = 5 * 60 +TEST_TIMEOUT_S = 40.0 + +P_DEAD_MARKERS = ("refresh failed", "token expired", "invalid or revoked", "sign in") + + +@typechecked +def p_seconds_left(expires_at: Optional[str], now: Optional[float] = None) -> Optional[float]: + if not expires_at: + return None + try: + t = datetime.fromisoformat(expires_at.replace("Z", "+00:00")).timestamp() + except ValueError: + return None + return t - (time.time() if now is None else now) + + +@typechecked +def connections_due(conns: List[Dict], now: Optional[float] = None) -> List[Dict]: + """OAuth connections that hold their own refresh token and are inside the margin (or already past it).""" + due = [] + for c in conns: + if c.get("authType") != "oauth" or not c.get("isActive", True): + continue + if not isinstance(c.get("refreshToken"), str) or not c.get("refreshToken"): + continue + left = p_seconds_left(c.get("expiresAt"), now) + if left is None or left > REFRESH_MARGIN_S: + continue + due.append(c) + return due + + +@typechecked +def verdict_from_test(result: Dict) -> str: + """"refreshed" (the router renewed it), "healthy" (valid, nothing to do), "dead" (the router itself says + waiting cannot help), or "unknown" (anything else, including a router hiccup).""" + if result.get("valid") is True: + return "refreshed" if result.get("refreshed") else "healthy" + err = str(result.get("error") or "").lower() + if result.get("valid") is False and any(m in err for m in P_DEAD_MARKERS): + return "dead" + return "unknown" + + +async def test_connection(client: httpx.AsyncClient, connection_id: str) -> Dict: + try: + r = await client.get(f"{NINE_ROUTER_API}/providers/{connection_id}/test") + return r.json() if r.status_code == 200 else {"error": f"HTTP {r.status_code}"} + except Exception as e: + return {"error": str(e)} + + +async def refresh_pass(now: Optional[float] = None) -> Dict[str, str]: + """One tick: test every due connection; returns {provider: verdict}. Reports a dead login through the + same door the boot probe and the turn use, so there is one pill and one story.""" + verdicts: Dict[str, str] = {} + if not is_running(): + return verdicts + due = connections_due(process.read_persisted_connections(), now) + if not due: + return verdicts + async with httpx.AsyncClient(timeout=TEST_TIMEOUT_S) as client: + for c in due: + provider = str(c.get("provider") or "") + v = verdict_from_test(await test_connection(client, str(c.get("id") or ""))) + verdicts[provider] = v + if v == "refreshed": + logger.info(f"[oauth-refresh] {provider}: token renewed ahead of expiry") + elif v == "dead": + logger.warning(f"[oauth-refresh] {provider}: the router could not renew this login; reporting it dead") + from backend.apps.nine_router.subscription_health import report_dead_now + await report_dead_now(provider) + return verdicts + + +@typechecked +def refresh_held_because() -> Optional[str]: + """A declared off switch (a drill beside the user's router must not renew the user's real logins), never an + incidental one; when it holds, the log says which protection just stood down.""" + import os + from backend.apps.agents.manager.session.SessionPersistence import running_under_test + if os.environ.get("OSW_DISABLE_OAUTH_REFRESH") == "1": + return "OSW_DISABLE_OAUTH_REFRESH=1" + if running_under_test(): + return "the test harness" + return None + + +async def oauth_refresh_loop() -> None: + held = refresh_held_because() + if held: + logger.warning(f"[oauth-refresh] NOT renewing subscription logins because {held}; a login that expires while this process runs will only be reported after its first failed chat") + return + while True: + try: + await refresh_pass() + except Exception: + logger.exception("oauth refresh pass failed") + await asyncio.sleep(CHECK_INTERVAL_S) diff --git a/backend/apps/nine_router/subscription_health.py b/backend/apps/nine_router/subscription_health.py index fffa563c..f1d7a6c0 100644 --- a/backend/apps/nine_router/subscription_health.py +++ b/backend/apps/nine_router/subscription_health.py @@ -203,3 +203,37 @@ async def p_recheck(provider: str, model: str) -> None: p_cached_result = dead from backend.apps.agents.core.ws_manager import ws_manager await ws_manager.broadcast_global("subscriptions:health", {"dead": dead}) + + +@typechecked +def note_auth_failure(provider: str) -> None: + """A real request on this lane just failed auth. That is stronger evidence than a probe, so it ADVANCES + the sighting (the clock starts now if it had not) and drops only the cached answer, so the next ask + re-probes; it never resets the clock or cancels a scheduled recheck, which is what invalidating the + whole cache on every 401 did (a lane agents kept hitting could restart its own window forever).""" + global p_cached_result, p_cached_at + if provider not in PREFIX_BY_PROVIDER: + return + p_refreshing_since.setdefault(provider, time.monotonic()) + p_cached_result = None + p_cached_at = 0.0 + + +@typechecked +async def report_dead_now(provider: str) -> bool: + """The turn's retry failed auth as well: waiting cannot fix this login, so say so this second instead of + at the next boot-time ask. Idempotent; returns False for a lane the probe does not cover.""" + global p_cached_result + if provider not in PREFIX_BY_PROVIDER: + return False + for t in p_rechecks.values(): + t.cancel() + p_rechecks.clear() + entry = {"provider": provider, "label": LABEL_BY_PROVIDER[provider]} + dead = [d for d in (p_cached_result or []) if d.get("provider") != provider] + [entry] + p_cached_result = dead + logger.warning(f"[sub-health] {provider}: a turn and its retry both failed auth; reporting it dead now") + from backend.apps.agents.core.ws_manager import ws_manager + await ws_manager.broadcast_global("subscriptions:health", {"dead": dead}) + return True + diff --git a/backend/tests/test_auth_resume.py b/backend/tests/test_auth_resume.py index 268e09e2..4dfbbc85 100644 --- a/backend/tests/test_auth_resume.py +++ b/backend/tests/test_auth_resume.py @@ -83,8 +83,13 @@ def test_the_exception_path_consults_auth_resume_too(): def test_the_resume_actively_refreshes_credentials(): + # Both auth paths feed the credential health verdict before they wait. Since 2026-09-06 that is note_auth_failure + # (the sighting ADVANCES; the cached answer is dropped so the next ask re-probes) and never invalidate_health_cache, + # which also wiped the first-seen clock and cancelled the scheduled recheck, so a lane agents kept hitting could + # restart its own grace window forever. src = inspect.getsource(TurnRunner) - assert src.count("invalidate_health_cache") >= 2, "both paths must poke the credential health cache, not just wait" + assert src.count("note_auth_failure(p_lane)") >= 2, "both paths must feed the credential health sighting, not just wait" + assert "invalidate_health_cache" not in src, "a turn's 401 must advance the verdict, never reset it" def test_the_recovery_ledger_counts_auth_resumes(): diff --git a/backend/tests/test_dead_login_chain.py b/backend/tests/test_dead_login_chain.py new file mode 100644 index 00000000..d777c7e9 --- /dev/null +++ b/backend/tests/test_dead_login_chain.py @@ -0,0 +1,211 @@ +"""The dead-login chain after 2026-09-06: a turn's 401 advances the sighting, a second 401 reports at once, a +reconnect resumes the chats that died, and the refresh loop renews a login before it expires or reports one the +router cannot renew. Each guard is proven to FIRE, and the innocent case for each is beside it.""" +from datetime import datetime, timezone + +import pytest + +from backend.apps.agents.core.models import AgentSession +from backend.apps.nine_router import oauth_refresh as orf +from backend.apps.nine_router import subscription_health as sh + + +@pytest.fixture(autouse=True) +def p_clean(): + sh.invalidate_health_cache() + yield + sh.invalidate_health_cache() + + +def test_a_turns_401_advances_the_sighting_and_never_resets_it(monkeypatch): + clock = [1000.0] + monkeypatch.setattr(sh.time, "monotonic", lambda: clock[0]) + sh.note_auth_failure("codex") + first = sh.p_refreshing_since["codex"] + clock[0] += 200 + sh.note_auth_failure("codex") + assert sh.p_refreshing_since["codex"] == first, "a later 401 must not restart the clock" + assert sh.p_cached_result is None and sh.p_cached_at == 0.0, "only the cached answer is dropped, so the next ask re-probes" + + +def test_a_turns_401_keeps_a_scheduled_recheck_alive(monkeypatch): + class FakeTask: + cancelled = False + def done(self): return False + def cancel(self): self.cancelled = True + t = FakeTask() + sh.p_rechecks["codex"] = t # type: ignore[assignment] + sh.note_auth_failure("codex") + assert t.cancelled is False, "invalidating the whole cache on a 401 used to cancel the recheck" + sh.p_rechecks.clear() + + +def test_note_auth_failure_ignores_lanes_the_probe_does_not_cover(): + sh.note_auth_failure("openrouter") + assert "openrouter" not in sh.p_refreshing_since + + +@pytest.mark.asyncio +async def test_a_second_401_reports_the_login_dead_this_second(monkeypatch): + sent = [] + from backend.apps.agents.core.ws_manager import ws_manager + async def fake_broadcast(event, data): sent.append((event, data)) + monkeypatch.setattr(ws_manager, "broadcast_global", fake_broadcast) + assert await sh.report_dead_now("codex") is True + assert sent == [("subscriptions:health", {"dead": [{"provider": "codex", "label": "ChatGPT"}]})] + assert sh.p_cached_result == [{"provider": "codex", "label": "ChatGPT"}], "a boot-time ask inside the TTL reads the same verdict" + assert await sh.report_dead_now("openrouter") is False and len(sent) == 1 + + +def p_session(sid, provider, status="error", ended=False): + s = AgentSession(id=sid, name=sid, prompt="x", status=status) + s.auth_dead_provider = provider + s.ended_by_user = ended + s.lane_credential_dead = True + s.auth_retry_used = True + return s + + +@pytest.mark.asyncio +async def test_a_reconnect_resumes_only_the_chats_that_died_on_that_login(): + from backend.apps.agents.manager.session.SessionPersistence import SessionPersistence + + class Mgr(SessionPersistence): + def __init__(self): + self.sessions = { + "dead-codex": p_session("dead-codex", "codex"), + "dead-claude": p_session("dead-claude", "claude"), + "stopped-by-human": p_session("stopped-by-human", "codex", ended=True), + "still-running": p_session("still-running", "codex", status="running"), + } + self.sent = [] + async def send_message(self, sid, text, hidden=False): + self.sent.append((sid, hidden)) + m = Mgr() + assert await m.resume_auth_dead_sessions("codex") == 1 + assert m.sent == [("dead-codex", True)] + s = m.sessions["dead-codex"] + assert s.auth_dead_provider is None and s.lane_credential_dead is False and s.auth_retry_used is False, "the marker and the spent retry are cleared so the resumed chat can heal again next time" + assert m.sessions["dead-claude"].auth_dead_provider == "claude", "another login's chats are untouched" + assert m.sessions["stopped-by-human"].auth_dead_provider == "codex", "a human's Stop always wins" + + +def test_the_refresh_loop_asks_only_inside_the_margin(): + now = 1_000_000.0 + def conn(expires_in_s, **kw): + base = {"id": "c1", "provider": "codex", "authType": "oauth", "isActive": True, "refreshToken": "r", "expiresAt": datetime.fromtimestamp(now + expires_in_s, timezone.utc).isoformat()} + base.update(kw) + return base + assert orf.connections_due([conn(3 * 3600)], now) == [] + assert len(orf.connections_due([conn(10 * 60)], now)) == 1 + assert len(orf.connections_due([conn(-5 * 86400)], now)) == 1, "an already-expired login is asked about too" + assert orf.connections_due([conn(10 * 60, refreshToken="")], now) == [], "a lent login (no refresh token) is the cloud's to rotate" + assert orf.connections_due([conn(10 * 60, authType="api_key")], now) == [] + + +def test_the_router_s_own_words_decide_the_verdict(): + assert orf.verdict_from_test({"valid": True, "refreshed": True}) == "refreshed" + assert orf.verdict_from_test({"valid": True, "refreshed": False}) == "healthy" + assert orf.verdict_from_test({"valid": False, "error": "Token expired and refresh failed", "refreshed": False}) == "dead" + assert orf.verdict_from_test({"valid": False, "error": "Token invalid or revoked"}) == "dead" + assert orf.verdict_from_test({"error": "HTTP 500"}) == "unknown", "a router hiccup is never a death" + assert orf.verdict_from_test({"valid": False, "error": "rate limited"}) == "unknown" + + +@pytest.mark.asyncio +async def test_a_login_the_router_cannot_renew_is_reported_through_the_one_door(monkeypatch): + now = 1_000_000.0 + conns = [{"id": "c1", "provider": "codex", "authType": "oauth", "isActive": True, "refreshToken": "r", "expiresAt": datetime.fromtimestamp(now - 60, timezone.utc).isoformat()}] + monkeypatch.setattr(orf, "is_running", lambda: True) + monkeypatch.setattr(orf.process, "read_persisted_connections", lambda: conns) + async def fake_test(client, cid): return {"valid": False, "error": "Token expired and refresh failed", "refreshed": False} + monkeypatch.setattr(orf, "test_connection", fake_test) + reported = [] + async def fake_report(provider): + reported.append(provider) + return True + monkeypatch.setattr(sh, "report_dead_now", fake_report) + assert await orf.refresh_pass(now) == {"codex": "dead"} + assert reported == ["codex"] + + +@pytest.mark.asyncio +async def test_a_renewed_login_says_nothing_and_a_down_router_asks_nobody(monkeypatch): + now = 1_000_000.0 + conns = [{"id": "c1", "provider": "codex", "authType": "oauth", "isActive": True, "refreshToken": "r", "expiresAt": datetime.fromtimestamp(now + 60, timezone.utc).isoformat()}] + monkeypatch.setattr(orf.process, "read_persisted_connections", lambda: conns) + async def fake_test(client, cid): return {"valid": True, "refreshed": True} + monkeypatch.setattr(orf, "test_connection", fake_test) + reported = [] + async def fake_report(provider): + reported.append(provider) + return True + monkeypatch.setattr(sh, "report_dead_now", fake_report) + monkeypatch.setattr(orf, "is_running", lambda: True) + assert await orf.refresh_pass(now) == {"codex": "refreshed"} and reported == [] + monkeypatch.setattr(orf, "is_running", lambda: False) + assert await orf.refresh_pass(now) == {} + + +def test_the_refresh_loop_s_off_switch_is_declared_and_loud(monkeypatch): + # caplog sees nothing from backend.* once the app is imported (PROJECT.md trap); a handler on the module's own logger does. + import asyncio + import logging + monkeypatch.setenv("OSW_DISABLE_OAUTH_REFRESH", "1") + assert orf.refresh_held_because() == "OSW_DISABLE_OAUTH_REFRESH=1" + seen = [] + + class Grab(logging.Handler): + def emit(self, record): + seen.append(record.getMessage()) + h = Grab(level=logging.WARNING) + orf.logger.addHandler(h) + try: + asyncio.run(orf.oauth_refresh_loop()) + finally: + orf.logger.removeHandler(h) + assert any("NOT renewing subscription logins" in m for m in seen), "a guard that stands down must say so" + + +@pytest.mark.asyncio +async def test_the_preflight_writes_the_lane_the_error_handler_reads(monkeypatch): + import backend.apps.agents.manager.run.lane_preflight as lp + async def p_none(provider): + return None + monkeypatch.setattr(lp, "dead_connection", p_none) + s = AgentSession(id="x", name="x", prompt="x") + await lp.preflight_lane("cc/claude-sonnet-5", s) + assert s.lane_provider == "claude" + await lp.preflight_lane("claude-sonnet-4-6", s) + assert s.lane_provider is None, "a direct API key has no router lane" + + +@pytest.mark.asyncio +async def test_a_definitive_auth_death_marks_the_chat_and_pushes_the_pill(monkeypatch): + from backend.apps.agents.manager.run import handle_run_error as hre + reported = [] + async def fake_report(provider): + reported.append(provider) + return True + monkeypatch.setattr(sh, "report_dead_now", fake_report) + s = AgentSession(id="x", name="x", prompt="x") + s.lane_provider = "codex" + await hre.p_mark_login_dead(s) + assert s.auth_dead_provider == "codex" and reported == ["codex"] + t = AgentSession(id="y", name="y", prompt="y") + await hre.p_mark_login_dead(t) + assert t.auth_dead_provider is None and reported == ["codex"], "a direct API key lane marks nothing" + + +def test_the_error_handler_marks_the_death_on_both_definitive_branches_and_before_the_card(): + import os + src = open(os.path.join(os.path.dirname(__file__), "..", "apps", "agents", "manager", "run", "handle_run_error.py")).read() + dead_branch = src.index('if getattr(session, "lane_credential_dead", False):') + first_mark = src.index("await p_mark_login_dead(session)", dead_branch) + dead_card = src.index("absorb_repeat_card(session, error_msg)", dead_branch) + assert dead_branch < first_mark < dead_card, "the already-dead lane branch marks before it cards" + second_mark = src.index("await p_mark_login_dead(session)", first_mark + 1) + gate = src.index('if reason in ("codex_token_rotating", "anthropic_auth_invalid", "openswarm_pro_auth_expired"):') + final_card = src.index("absorb_repeat_card(session, error_msg)", second_mark) + assert gate < second_mark < final_card, "the definitive auth card marks only on the auth-shaped reasons, before the card" + diff --git a/backend/tests/test_lane_preflight.py b/backend/tests/test_lane_preflight.py index 5ea7ced7..9405355f 100644 --- a/backend/tests/test_lane_preflight.py +++ b/backend/tests/test_lane_preflight.py @@ -8,16 +8,10 @@ testStatus="unavailable" and errorCode=401 the whole time. import asyncio -import pytest import backend.apps.agents.manager.run.lane_preflight as lp -@pytest.fixture(autouse=True) -def p_clear_cooldown(): - lp.LAST_BOUNCE.clear() - yield - lp.LAST_BOUNCE.clear() def p_providers(monkeypatch, conns, bounce_result=None): @@ -50,14 +44,6 @@ def test_a_healthy_lane_costs_nothing_and_says_nothing(monkeypatch): assert st["bounced"] == 0, "a working lane must never trigger a router restart" -def test_the_first_dead_encounter_bounces_and_lets_the_turn_decide(monkeypatch): - """The bounce is an attempt, not a verdict. It must not block the turn, and it must not claim - a recovery it cannot see.""" - st = p_providers(monkeypatch, P_DEAD, bounce_result=P_DEAD) - assert asyncio.run(lp.preflight_lane("cx/gpt-5.6")) is None, "dispatch is the real test" - assert st["bounced"] == 1 - - def test_a_cleared_stamp_is_never_mistaken_for_a_working_credential(monkeypatch): """The bug this test exists for shipped for ten minutes on 2026-08-20. The first version re-read health after the bounce and returned "recovered" because a fresh router has no @@ -78,21 +64,6 @@ def test_a_cleared_stamp_is_never_mistaken_for_a_working_credential(monkeypatch) ) -def test_a_lane_still_dead_inside_the_cooldown_DISPATCHES(monkeypatch): - """CORRECTED 2026-08-27 (ENG-414). This used to assert the opposite, and the assumption it - encoded is the bug: "second encounter inside the cooldown" was read as "we already spent a - bounce and a turn on THIS session". `LAST_BOUNCE` is module-global, so in production it meant - "some other chat bounced recently" and it hard-stopped a live build on a working credential. - - Preflight has not dispatched, so it cannot know. It dispatches and flags the session; the - accurate sentence now comes from handle_run_error after a real 401. The cooldown still holds.""" - st = p_providers(monkeypatch, P_DEAD, bounce_result=P_DEAD) - assert asyncio.run(lp.preflight_lane("cx/gpt-5.6")) is None - assert asyncio.run(lp.preflight_lane("cx/gpt-5.6")) is None, \ - "a throttled bounce is not evidence about the credential" - assert st["bounced"] == 1, "the cooldown holds; one restart, not one per ask" - - def test_the_downstream_card_still_refuses_the_rotation_story(): """What the old assertion above was really protecting: when the card DOES fire, it must not invent a rotation window or claim no action is needed. That copy moved, it did not soften.""" @@ -102,14 +73,6 @@ def test_the_downstream_card_still_refuses_the_rotation_story(): assert "reconnect" in msg.lower(), msg -def test_the_bounce_is_rate_limited(monkeypatch): - """A bounce restarts a process every other session shares, so it is once per lane per window, never once per turn.""" - st = p_providers(monkeypatch, P_DEAD, bounce_result=P_DEAD) - for _ in range(4): - asyncio.run(lp.preflight_lane("cx/gpt-5.6")) - assert st["bounced"] == 1, f"expected a single bounce, got {st['bounced']}" - - def test_direct_api_lanes_are_left_alone(monkeypatch): """Negative control: a direct API key never dispatches through the router, so the router's health says nothing about it and must not ground it.""" st = p_providers(monkeypatch, P_DEAD) @@ -140,26 +103,6 @@ def test_only_terminal_states_count_as_dead(): assert lp.connection_is_dead({}) is False -def test_never_dispatches_into_a_router_that_did_not_come_back(monkeypatch): - """A bounce that fails to restart leaves nothing listening. Dispatching into that is a - guaranteed connection error the user would read as the model failing, rather than as us - restarting something underneath them.""" - async def fake_get_providers(): - return P_DEAD - - async def failed_bounce(provider): - return False - - import backend.apps.nine_router as nr - import backend.apps.nine_router.bounce_after_connect as ba - monkeypatch.setattr(nr, "get_providers", fake_get_providers, raising=True) - monkeypatch.setattr(ba, "bounce_router_after_connect", failed_bounce, raising=True) - - msg = asyncio.run(lp.preflight_lane("cx/gpt-5.6")) - assert msg and "restarting" in msg.lower() - assert "reconnect" not in msg.lower(), "this is our restart, not the user's credential" - - def test_a_rate_limited_lane_is_not_a_dead_credential(): """Live 2026-08-20: antigravity sat at testStatus=unavailable with errorCode=429 and a credential valid for another 30 minutes. Telling that user to reconnect is the same lie as @@ -192,3 +135,16 @@ def test_a_throttle_is_still_not_death(): assert lp.connection_is_dead({"provider": "gemini-cli", "testStatus": "unavailable", "errorCode": 429}) is False assert lp.connection_is_dead({"provider": "codex", "testStatus": "unavailable", "errorCode": 401}) is True + +def test_a_dead_lane_never_restarts_the_router(monkeypatch): + """2026-09-06: the preflight bounce was a dead port for every chat on every lane, and a restart cannot revive + a dead token. A dead lane dispatches (the real request is the test) and restarts nothing.""" + st = p_providers(monkeypatch, P_DEAD, bounce_result=P_DEAD) + for _ in range(4): + assert asyncio.run(lp.preflight_lane("cx/gpt-5.6")) is None + assert st["bounced"] == 0, f"preflight must never restart the router, got {st['bounced']}" + + +def test_the_module_no_longer_owns_a_bounce_throttle(): + assert not hasattr(lp, "LAST_BOUNCE") and not hasattr(lp, "BOUNCE_COOLDOWN_S") + diff --git a/backend/tests/test_lane_preflight_never_declares_death.py b/backend/tests/test_lane_preflight_never_declares_death.py index ed0d551e..42539812 100644 --- a/backend/tests/test_lane_preflight_never_declares_death.py +++ b/backend/tests/test_lane_preflight_never_declares_death.py @@ -7,7 +7,6 @@ the router in the last five minutes". The user was told "Waiting will not clear proved it wrong by typing "continue" a minute later and watching the run finish. """ -import time import pytest @@ -23,20 +22,7 @@ def p_session() -> AgentSession: @pytest.fixture(autouse=True) def p_clean(): - p_pf.LAST_BOUNCE.clear() yield - p_pf.LAST_BOUNCE.clear() - - -@pytest.mark.asyncio -async def test_a_throttled_bounce_dispatches_instead_of_carding(monkeypatch): - """The exact shape that killed the build: another chat bounced 10s ago, this lane reads dead.""" - monkeypatch.setattr(p_pf, "dead_connection", - lambda provider: p_async({"testStatus": "unavailable", "errorCode": 401})) - p_pf.LAST_BOUNCE["claude"] = time.time() - 10 # another chat, well inside the 300s window - s = p_session() - assert await p_pf.preflight_lane("cc/claude-opus-5", s) is None, \ - "a throttled bounce says nothing about the credential; the turn must still be spent" @pytest.mark.asyncio @@ -44,7 +30,6 @@ async def test_the_session_is_flagged_so_a_REAL_401_can_still_be_honest(monkeypa """Dispatching anyway must not lose the accuracy the old card had.""" monkeypatch.setattr(p_pf, "dead_connection", lambda provider: p_async({"testStatus": "unavailable", "errorCode": 401})) - p_pf.LAST_BOUNCE["claude"] = time.time() - 10 s = p_session() await p_pf.preflight_lane("cc/claude-opus-5", s) assert s.lane_credential_dead is True, "handle_run_error keys its accurate card on this" @@ -75,45 +60,6 @@ async def test_the_death_verdict_exists_in_exactly_one_place(monkeypatch): "and it must still be gated on the router having given up BEFORE the turn" -@pytest.mark.asyncio -async def test_the_bounce_itself_is_still_throttled(monkeypatch): - """A bounce restarts a process every session shares. Dispatching anyway must NOT turn into - bouncing on every turn.""" - p_calls = [] - - async def p_fake_bounce(provider): - p_calls.append(provider) - return True - - monkeypatch.setattr(p_pf, "dead_connection", - lambda provider: p_async({"testStatus": "unavailable", "errorCode": 401})) - import backend.apps.nine_router.bounce_after_connect as p_b - monkeypatch.setattr(p_b, "bounce_router_after_connect", p_fake_bounce) - s = p_session() - await p_pf.preflight_lane("cc/claude-opus-5", s) # first: allowed to bounce - await p_pf.preflight_lane("cc/claude-opus-5", s) # second: throttled - await p_pf.preflight_lane("cc/claude-opus-5", s) # third: throttled - assert len(p_calls) == 1, f"one bounce per {p_pf.BOUNCE_COOLDOWN_S}s window, got {len(p_calls)}" - - -@pytest.mark.asyncio -async def test_a_router_that_does_not_come_back_still_stops_the_turn(monkeypatch): - """The one thing preflight CAN know without dispatching: it just restarted the router and the - router is not listening. Dispatching into that is a guaranteed connection error the user would - read as the model failing.""" - monkeypatch.setattr(p_pf, "dead_connection", - lambda provider: p_async({"testStatus": "unavailable", "errorCode": 401})) - import backend.apps.nine_router.bounce_after_connect as p_b - - async def p_dead_bounce(provider): - return False - - monkeypatch.setattr(p_b, "bounce_router_after_connect", p_dead_bounce) - msg = await p_pf.preflight_lane("cc/claude-opus-5", p_session()) - assert msg and "restarting" in msg - assert "Waiting will not clear" not in msg, "a restarting router is not a dead credential" - - async def p_async(value): return value