From 05e8310691f307ef3bb252d4190dd354fbda2e4b Mon Sep 17 00:00:00 2001 From: ciregenz Date: Sat, 5 Sep 2026 09:28:59 -0700 Subject: [PATCH] [eric] health: a mid-refresh 401 that outlives the token rotation window is a dead login; recheck at 240s and push the pill Co-Authored-By: Claude Fable 5.1 --- .../apps/nine_router/subscription_health.py | 84 +++++++++++- .../tests/test_subscription_health_dedupe.py | 125 +++++++++++++++++- .../src/shared/state/healthReported.test.ts | 29 ++++ .../src/shared/state/subscriptionsSlice.ts | 7 +- frontend/src/shared/ws/WebSocketManager.ts | 6 + 5 files changed, 243 insertions(+), 8 deletions(-) create mode 100644 frontend/src/shared/state/healthReported.test.ts diff --git a/backend/apps/nine_router/subscription_health.py b/backend/apps/nine_router/subscription_health.py index 62648c77..fffa563c 100644 --- a/backend/apps/nine_router/subscription_health.py +++ b/backend/apps/nine_router/subscription_health.py @@ -32,10 +32,14 @@ LABEL_BY_PROVIDER: Dict[str, str] = { P_AUTH_DEAD_MARKERS = ("authentication", "expired", "sign in", "signing in", "invalid_grant", "unauthorized", "invalid authentication") P_PROBE_TIMEOUT_S = 25.0 P_CACHE_TTL_S = 300.0 +# A codex refresh lands within 1-2 minutes (ENG-361); a "mid-refresh" 401 still standing at the next probe is a dead login. +P_ROTATION_WINDOW_S = 240.0 p_probe_lock = asyncio.Lock() p_cached_result: Optional[List[Dict[str, str]]] = None p_cached_at: float = 0.0 +p_refreshing_since: Dict[str, float] = {} +p_rechecks: Dict[str, "asyncio.Task[None]"] = {} @typechecked @@ -49,6 +53,10 @@ def invalidate_health_cache() -> None: global p_cached_result, p_cached_at p_cached_result = None p_cached_at = 0.0 + p_refreshing_since.clear() + for t in p_rechecks.values(): + t.cancel() + p_rechecks.clear() @typechecked @@ -64,6 +72,17 @@ def classify_auth_dead(status_code: int, body_text: str) -> bool: return any(m in low for m in P_AUTH_DEAD_MARKERS) +@typechecked +def classify_refreshing(status_code: int, body_text: str) -> bool: + """An auth-shaped 401/403 that classify_auth_dead excused for naming a reset window. The router appends + "(reset after Ns)" to EVERY error, so this text is also what a login that died days ago answers; the + verdict has to come from time, not wording (a codex token expired on 08-30 wore it for six days).""" + if status_code not in (401, 403): + return False + low = body_text.lower() + return not classify_auth_dead(status_code, body_text) and any(m in low for m in P_AUTH_DEAD_MARKERS) + + @typechecked async def p_pick_probe_model(client: httpx.AsyncClient, prefix: str) -> Optional[str]: try: @@ -80,8 +99,9 @@ async def p_pick_probe_model(client: httpx.AsyncClient, prefix: str) -> Optional @typechecked -async def p_probe_one(client: httpx.AsyncClient, model: str) -> Optional[bool]: - """True = auth dead, False = healthy, None = inconclusive (never reported).""" +async def p_probe_one(client: httpx.AsyncClient, model: str) -> str: + """"dead" = definitive auth failure, "healthy" = answered, "refreshing" = auth failure naming a reset + window, "unknown" = inconclusive (never reported).""" try: r = await client.post( f"{NINE_ROUTER_URL}/v1/messages", @@ -89,10 +109,24 @@ async def p_probe_one(client: httpx.AsyncClient, model: str) -> Optional[bool]: headers={"x-api-key": "9router", "anthropic-version": "2023-06-01"}, ) if r.status_code < 400: - return False - return True if classify_auth_dead(r.status_code, r.text or "") else None + return "healthy" + body = r.text or "" + if classify_auth_dead(r.status_code, body): + return "dead" + return "refreshing" if classify_refreshing(r.status_code, body) else "unknown" except Exception: - return None + return "unknown" + + +@typechecked +def refreshing_verdict(provider: str, now: float) -> bool: + """True once a provider has answered "refreshing" across more than the rotation window.""" + first = p_refreshing_since.setdefault(provider, now) + if now - first >= P_ROTATION_WINDOW_S: + logger.warning(f"[sub-health] {provider}: the mid-refresh 401 has stood for {int(now - first)}s, past the rotation window; reporting it dead") + return True + logger.info(f"[sub-health] {provider}: 401 names a reset window; waiting one rotation window before calling it dead") + return False @typechecked @@ -125,9 +159,47 @@ async def probe_subscription_health(connections: List[Dict]) -> List[Dict[str, s if not model: continue verdict = await p_probe_one(client, model) - if verdict is True: + if verdict == "healthy": + p_refreshing_since.pop(provider, None) + elif verdict == "refreshing": + if refreshing_verdict(provider, time.monotonic()): + verdict = "dead" + else: + schedule_recheck(provider, model) + if verdict == "dead": dead.append({"provider": provider, "label": LABEL_BY_PROVIDER[provider]}) logger.info(f"[sub-health] {provider}: auth dead (reconnect needed)") p_cached_result = dead p_cached_at = time.monotonic() return dead + + +@typechecked +def schedule_recheck(provider: str, model: str) -> bool: + """The app asks the health question once per boot, so a "wait and see" verdict needs its own second look + or it never gets one: re-probe after the rotation window and push the answer to every open dashboard.""" + live = p_rechecks.get(provider) + if live is not None and not live.done(): + return False + p_rechecks[provider] = asyncio.create_task(p_recheck(provider, model)) + return True + + +async def p_recheck(provider: str, model: str) -> None: + global p_cached_result + await asyncio.sleep(P_ROTATION_WINDOW_S) + async with httpx.AsyncClient(timeout=P_PROBE_TIMEOUT_S) as client: + verdict = await p_probe_one(client, model) + if verdict == "healthy": + p_refreshing_since.pop(provider, None) + logger.info(f"[sub-health] {provider}: the token rotated; healthy on the recheck") + return + if verdict == "unknown": + logger.info(f"[sub-health] {provider}: recheck inconclusive; not reported") + return + logger.warning(f"[sub-health] {provider}: still refusing auth {int(P_ROTATION_WINDOW_S)}s after the first 401; reporting it dead") + 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 + from backend.apps.agents.core.ws_manager import ws_manager + await ws_manager.broadcast_global("subscriptions:health", {"dead": dead}) diff --git a/backend/tests/test_subscription_health_dedupe.py b/backend/tests/test_subscription_health_dedupe.py index c2a83639..e1b05d9d 100644 --- a/backend/tests/test_subscription_health_dedupe.py +++ b/backend/tests/test_subscription_health_dedupe.py @@ -1,9 +1,12 @@ """One banner line per provider: two active db.json rows for one provider (stale + fresh connect) used to probe twice and render "Your ChatGPT and ChatGPT logins have expired".""" +import asyncio + import pytest from backend.apps.nine_router import subscription_health as sh +from backend.tests.test_subscription_health import FakeResponse @pytest.mark.asyncio @@ -17,7 +20,7 @@ async def test_duplicate_provider_rows_probe_and_report_once(monkeypatch): async def fake_probe(client, model): probed.append(model) - return True + return "dead" monkeypatch.setattr(sh, "p_pick_probe_model", fake_pick) monkeypatch.setattr(sh, "p_probe_one", fake_probe) @@ -41,3 +44,123 @@ def test_self_healing_401_with_reset_window_is_not_dead(): def test_genuine_rotation_death_still_reports(): assert sh.classify_auth_dead(401, "invalid_grant: refresh token rotated") assert sh.classify_auth_dead(403, "Unauthorized: expired credentials, please sign in") + + +# Verbatim live body (2026-09-04): the same "mid-refresh" text, answered by a codex login whose token had +# expired on 08-30; the app said nothing for six days because the reset-window excuse never expired. +LIVE_STALE_401 = '{"error":{"message":"[codex/gpt-5.4] [401]: Provided authentication token is expired. Please try signing in again. (reset after 26s)"}}' + + +class RefreshingClient: + def __init__(self, answers, **kw) -> None: + self.answers = answers + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc) -> None: + return None + + async def get(self, url: str, **kw): + return FakeResponse(200, {"data": [{"id": "cx/gpt-5.4"}]}) + + async def post(self, url: str, **kw): + status, body = self.answers.pop(0) + return FakeResponse(status, body) + + +def p_run(monkeypatch, answers, clock): + monkeypatch.setattr(sh, "is_running", lambda: True) + monkeypatch.setattr(sh.httpx, "AsyncClient", lambda **kw: RefreshingClient(answers, **kw)) + monkeypatch.setattr(sh.time, "monotonic", lambda: clock[0]) + return asyncio.run(sh.probe_subscription_health([{"provider": "codex", "isActive": True}])) + + +def test_a_mid_refresh_401_that_outlives_the_rotation_window_is_dead(monkeypatch): + sh.invalidate_health_cache() + clock = [1000.0] + answers = [(401, LIVE_STALE_401), (401, LIVE_STALE_401)] + assert p_run(monkeypatch, answers, clock) == [], "first sighting: a rotation is still possible" + clock[0] += sh.P_CACHE_TTL_S + 1 + assert p_run(monkeypatch, answers, clock) == [{"provider": "codex", "label": "ChatGPT"}] + assert answers == [], "both probes were spent" + sh.invalidate_health_cache() + + +def test_a_401_that_heals_within_the_window_clears_the_sighting(monkeypatch): + sh.invalidate_health_cache() + clock = [1000.0] + answers = [(401, LIVE_STALE_401), (200, {"content": []}), (401, LIVE_STALE_401)] + assert p_run(monkeypatch, answers, clock) == [] + clock[0] += sh.P_CACHE_TTL_S + 1 + assert p_run(monkeypatch, answers, clock) == [], "healed: the sighting is dropped" + clock[0] += sh.P_CACHE_TTL_S + 1 + assert p_run(monkeypatch, answers, clock) == [], "a fresh 401 starts a new window rather than inheriting the old one" + sh.invalidate_health_cache() + + +def test_reconnect_forgets_the_sighting(monkeypatch): + sh.invalidate_health_cache() + clock = [1000.0] + answers = [(401, LIVE_STALE_401), (401, LIVE_STALE_401)] + assert p_run(monkeypatch, answers, clock) == [] + sh.invalidate_health_cache() + clock[0] += sh.P_CACHE_TTL_S + 1 + assert p_run(monkeypatch, answers, clock) == [], "a deliberate reconnect restarts the window" + sh.invalidate_health_cache() + + +@pytest.mark.asyncio +async def test_the_first_mid_refresh_sighting_gets_its_own_second_look(monkeypatch): + # The dashboard asks once per boot; without a scheduled recheck the "wait and see" verdict is never revisited. + sh.invalidate_health_cache() + answers = [(401, LIVE_STALE_401), (401, LIVE_STALE_401)] + monkeypatch.setattr(sh, "is_running", lambda: True) + monkeypatch.setattr(sh.httpx, "AsyncClient", lambda **kw: RefreshingClient(answers, **kw)) + slept = [] + + async def fake_sleep(s): + slept.append(s) + + monkeypatch.setattr(sh.asyncio, "sleep", fake_sleep) + 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.probe_subscription_health([{"provider": "codex", "isActive": True}]) == [] + task = sh.p_rechecks.get("codex") + assert task is not None, "the first sighting must schedule a recheck" + await task + assert slept == [sh.P_ROTATION_WINDOW_S] + assert sent == [("subscriptions:health", {"dead": [{"provider": "codex", "label": "ChatGPT"}]})] + assert answers == [], "the recheck spent the second probe" + assert sh.p_cached_result == [{"provider": "codex", "label": "ChatGPT"}], "a later boot-time fetch inside the TTL reads the same verdict" + sh.invalidate_health_cache() + + +@pytest.mark.asyncio +async def test_a_recheck_that_finds_the_token_rotated_says_nothing(monkeypatch): + sh.invalidate_health_cache() + answers = [(401, LIVE_STALE_401), (200, {"content": []})] + monkeypatch.setattr(sh, "is_running", lambda: True) + monkeypatch.setattr(sh.httpx, "AsyncClient", lambda **kw: RefreshingClient(answers, **kw)) + + async def fake_sleep(s): + return None + + monkeypatch.setattr(sh.asyncio, "sleep", fake_sleep) + 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.probe_subscription_health([{"provider": "codex", "isActive": True}]) == [] + await sh.p_rechecks["codex"] + assert sent == [] and answers == [] + assert "codex" not in sh.p_refreshing_since, "a healthy recheck forgets the sighting" + sh.invalidate_health_cache() diff --git a/frontend/src/shared/state/healthReported.test.ts b/frontend/src/shared/state/healthReported.test.ts new file mode 100644 index 00000000..8c9b67b7 --- /dev/null +++ b/frontend/src/shared/state/healthReported.test.ts @@ -0,0 +1,29 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import fs from 'node:fs'; +import path from 'node:path'; +import reducer, { healthReported, hideProviderHealthToast } from './subscriptionsSlice'; + +// A ChatGPT login whose token expired on 08-30 answered every probe with the router's "(reset after Ns)" 401 for six +// days; the boot-time verdict excused it as mid-refresh and nothing ever looked again. The backend now re-probes after +// the rotation window and pushes the verdict; the slice has to open the same pill the boot fetch would have. + +test('a pushed verdict opens the reconnect pill', () => { + const s = reducer(undefined, healthReported({ dead: [{ provider: 'codex', label: 'ChatGPT' }] })); + assert.equal(s.healthToastOpen, true); + assert.deepEqual(s.healthDead, [{ provider: 'codex', label: 'ChatGPT' }]); +}); + +test('an empty verdict closes nothing the user already dismissed and opens nothing', () => { + let s = reducer(undefined, healthReported({ dead: [{ provider: 'codex', label: 'ChatGPT' }] })); + s = reducer(s, hideProviderHealthToast()); + s = reducer(s, healthReported({ dead: [] })); + assert.equal(s.healthToastOpen, false); + assert.deepEqual(s.healthDead, []); +}); + +test('the socket routes subscriptions:health into the slice', () => { + const src = fs.readFileSync(path.join(process.cwd(), 'src/shared/ws/WebSocketManager.ts'), 'utf8'); + assert.ok(src.includes("case 'subscriptions:health':")); + assert.ok(src.includes('store.dispatch(healthReported({ dead: data.dead }))')); +}); diff --git a/frontend/src/shared/state/subscriptionsSlice.ts b/frontend/src/shared/state/subscriptionsSlice.ts index 83d2f02d..497beb94 100644 --- a/frontend/src/shared/state/subscriptionsSlice.ts +++ b/frontend/src/shared/state/subscriptionsSlice.ts @@ -72,6 +72,11 @@ const subscriptionsSlice = createSlice({ state.status = action.payload; }, // Optimistic: 9Router /providers lags /exchange, so refetching right after would clobber the just-connected state with stale data. The 30s poller reconciles. + // The backend re-probes a "mid-refresh" 401 on its own after the rotation window and pushes the verdict here; the boot-time fetch is long gone by then. + healthReported(state, action: PayloadAction<{ dead: DeadProvider[] }>) { + state.healthDead = action.payload.dead ?? []; + state.healthToastOpen = state.healthDead.length > 0 || state.healthCliMissing; + }, markSubscriptionConnected(state, action: PayloadAction<{ provider: string }>) { if (!state.status) return; const { provider } = action.payload; @@ -113,7 +118,7 @@ const subscriptionsSlice = createSlice({ }, }); -export const { setSubscriptionStatus, markSubscriptionConnected, hideProviderHealthToast } = subscriptionsSlice.actions; +export const { setSubscriptionStatus, markSubscriptionConnected, hideProviderHealthToast, healthReported } = subscriptionsSlice.actions; // Stable empty ref so the selector doesn't hand back a fresh [] each call (forces needless rerenders). const EMPTY_CONNECTIONS: SubscriptionConnection[] = []; diff --git a/frontend/src/shared/ws/WebSocketManager.ts b/frontend/src/shared/ws/WebSocketManager.ts index b7c38890..c46408e3 100644 --- a/frontend/src/shared/ws/WebSocketManager.ts +++ b/frontend/src/shared/ws/WebSocketManager.ts @@ -34,6 +34,7 @@ import { } from '../state/agentsSlice'; import { streamStart, streamSnapshot, streamDelta, streamEnd, clearStreamingForSession } from '../state/streamingSlice'; import { fetchToolStatus } from '../state/toolsSlice'; +import { healthReported } from '../state/subscriptionsSlice'; import { remountAppPreview } from '../state/outputsSlice'; import { BackgroundDeltaBuffer } from './BackgroundDeltaBuffer'; import { interactionActive, installInteractionListeners } from '../interactionPriority'; @@ -657,6 +658,11 @@ class WebSocketManager { } break; + case 'subscriptions:health': + if (Array.isArray(data.dead)) { + store.dispatch(healthReported({ dead: data.dead })); + } + break; case 'tools:updated': // A connector's auth state changed on the backend (an OAuth claim landed, a disconnect); refetch it and tell the Tools page. if (data.tool_id) {