[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 <noreply@anthropic.com>

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})

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()

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 }))'));
+});

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[] = [];

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) {