WEBDIENSTE/openswarm
1
0
Fork 0
mirror of https://github.com/openswarm-ai/openswarm.git synced 2026-09-07 18:27:45 +02:00
Code Issues Packages Projects Releases Wiki Activity

[eric] auth: a dead login heals itself: renewed before expiry, a turn's 401 advances the verdict, the retry pushes the pill, a reconnect resumes the chats; preflight never restarts the router

Browse Source 
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
ciregenz
2026-09-06 15:55:26 -07:00
co-authored by Claude Fable 5.1
parent 063fb6863c
commit da56b2cd1a
12 changed files with 510 additions and 184 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/apps/agents/agents.py
+25 -1
View File
@@ -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:
backend/apps/agents/core/models.py
+4
View File
@@ -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).
backend/apps/agents/manager/run/TurnRunner.py
+12 -6
View File
@@ -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()
backend/apps/agents/manager/run/handle_run_error.py
+17
View File
@@ -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:
backend/apps/agents/manager/run/lane_preflight.py
+29 -65
View File
@@ -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
backend/apps/agents/manager/session/SessionPersistence.py
+28
View File
@@ -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
backend/apps/nine_router/oauth_refresh.py
+131
View File
@@ -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)
backend/apps/nine_router/subscription_health.py
+34
View File
@@ -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
backend/tests/test_auth_resume.py
+6 -1
View File
@@ -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():
backend/tests/test_dead_login_chain.py
+211
View File
@@ -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"
backend/tests/test_lane_preflight.py
+13 -57
View File
@@ -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")
backend/tests/test_lane_preflight_never_declares_death.py
-54
View File
@@ -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