[eric] health: a dead-login verdict owns its second look; the lane answering again closes the pill and resumes its chats, capped at two resumes per chat so a flapping login cannot loop

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

backend/apps/agents/agents.py

@@ -34,6 +34,9 @@ async def agents_lifespan():
     # 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())
+    # A lane that answers again after a dead verdict resumes its chats the way a reconnect does.
+    from backend.apps.nine_router import subscription_health as p_sub_health
+    p_sub_health.p_healed_hooks.append(agent_manager.resume_auth_dead_sessions)
     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

backend/apps/agents/core/models.py

@@ -178,6 +178,8 @@ class AgentSession(BaseModel):
     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
+    # A login that flaps (probe answers, the turn 401s) would otherwise resume and die every re-probe forever.
+    auth_resumes: int = 0
     # 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.

backend/apps/agents/manager/session/SessionPersistence.py

@@ -20,6 +20,8 @@ from backend.apps.agents.manager.session.apply_context_window import apply_conte
 
 logger = logging.getLogger(__name__)
 
+AUTH_RESUME_CAP = 2
+
 
 def auto_resume_held_because() -> Optional[str]:
     """Why auto-resume must not fire this boot, in words, or None to proceed.
@@ -144,6 +146,10 @@ class SessionPersistence(AgentManagerProtocol):
                 continue
             if session.status in ("running", "waiting_approval"):
                 continue
+            if session.auth_resumes >= AUTH_RESUME_CAP:
+                logger.warning(f"reconnect-resume: session {sid} has already been resumed {session.auth_resumes} times on a dead login; leaving it to the user")
+                continue
+            session.auth_resumes += 1
             session.auth_dead_provider = None
             session.lane_credential_dead = False
             session.auth_retry_used = False

backend/apps/nine_router/subscription_health.py

@@ -8,7 +8,7 @@ import asyncio
 import logging
 import os
 import time
-from typing import Dict, List, Optional
+from typing import Awaitable, Callable, Dict, List, Optional
 
 import httpx
 from typeguard import typechecked
@@ -57,6 +57,9 @@ def invalidate_health_cache() -> None:
     for t in p_rechecks.values():
         t.cancel()
     p_rechecks.clear()
+    for t in p_reprobes.values():
+        t.cancel()
+    p_reprobes.clear()
 
 
 @typechecked
@@ -203,6 +206,7 @@ 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})
+    schedule_reprobe(provider, model)
 
 
 @typechecked
@@ -235,5 +239,56 @@ async def report_dead_now(provider: str) -> bool:
     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})
+    schedule_reprobe(provider, None)
     return True
 
+
+# A dead verdict owns its second look too. Once the pill was up nothing ever asked again, so a ChatGPT
+# rotation slower than the turn's 75s retry told the user to reconnect a login that healed by itself.
+P_REPROBE_S = 240.0
+P_REPROBE_MAX_S = 1800.0
+p_reprobes: Dict[str, "asyncio.Task[None]"] = {}
+p_healed_hooks: List[Callable[[str], Awaitable[object]]] = []
+
+
+@typechecked
+def schedule_reprobe(provider: str, model: Optional[str], delay: float = P_REPROBE_S) -> bool:
+    if provider not in PREFIX_BY_PROVIDER:
+        return False
+    live = p_reprobes.get(provider)
+    if live is not None and not live.done() and live is not asyncio.current_task():
+        live.cancel()
+    p_reprobes[provider] = asyncio.create_task(p_reprobe(provider, model, delay))
+    return True
+
+
+async def p_reprobe(provider: str, model: Optional[str], delay: float) -> None:
+    await asyncio.sleep(delay)
+    async with httpx.AsyncClient(timeout=P_PROBE_TIMEOUT_S) as client:
+        probe_model = model or await p_pick_probe_model(client, PREFIX_BY_PROVIDER[provider])
+        verdict = await p_probe_one(client, probe_model) if probe_model else "unknown"
+    if verdict == "healthy":
+        await mark_healed(provider)
+        return
+    again = min(delay * 2, P_REPROBE_MAX_S)
+    logger.info(f"[sub-health] {provider}: still {verdict} on the re-probe; asking again in {int(again)}s")
+    schedule_reprobe(provider, probe_model, again)
+
+
+@typechecked
+async def mark_healed(provider: str) -> None:
+    """The lane answers again: the pill closes this second and every chat that died on it resumes."""
+    global p_cached_result, p_cached_at
+    p_refreshing_since.pop(provider, None)
+    dead = [d for d in (p_cached_result or []) if d.get("provider") != provider]
+    p_cached_result = dead
+    p_cached_at = time.monotonic()
+    logger.info(f"[sub-health] {provider}: answered on the re-probe; closing the pill and resuming its chats")
+    from backend.apps.agents.core.ws_manager import ws_manager
+    await ws_manager.broadcast_global("subscriptions:health", {"dead": dead})
+    for hook in list(p_healed_hooks):
+        try:
+            await hook(provider)
+        except Exception:
+            logger.exception(f"[sub-health] {provider}: a healed hook failed")
+

backend/tests/test_dead_login_chain.py

@@ -51,10 +51,13 @@ async def test_a_second_401_reports_the_login_dead_this_second(monkeypatch):
     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)
+    armed = []
+    monkeypatch.setattr(sh, "schedule_reprobe", lambda provider, model, delay=sh.P_REPROBE_S: armed.append((provider, delay)) or True)
     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
+    assert armed == [("codex", sh.P_REPROBE_S)], "the dead-now verdict arms its own re-probe"
 
 
 def p_session(sid, provider, status="error", ended=False):
@@ -99,7 +102,7 @@ def test_the_refresh_loop_asks_only_inside_the_margin():
     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)], now, lent={"c1"}) == [], "a lent login (no refresh token in the persisted file) is the cloud's to rotate"
     assert orf.connections_due([conn(10 * 60, authType="api_key")], now) == []
 
 
@@ -117,6 +120,10 @@ async def test_a_login_the_router_cannot_renew_is_reported_through_the_one_door(
     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)
+    async def live():
+        return conns
+
+    monkeypatch.setattr(orf.process, "get_providers", live)
     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)
@@ -133,6 +140,10 @@ async def test_a_login_the_router_cannot_renew_is_reported_through_the_one_door(
 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()}]
+    async def live():
+        return conns
+
+    monkeypatch.setattr(orf.process, "get_providers", live)
     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)

backend/tests/test_dead_login_reprobe.py

@@ -0,0 +1,111 @@
+"""A dead-login verdict owns its second look: the pill closes and the chats resume when the lane answers again."""
+import asyncio
+import pathlib
+
+import pytest
+
+from backend.apps.nine_router import subscription_health as sh
+
+
+@pytest.fixture(autouse=True)
+def clean(monkeypatch):
+    sh.invalidate_health_cache()
+    sh.p_healed_hooks.clear()
+    sent = []
+
+    async def fake_broadcast(kind, payload):
+        sent.append((kind, payload))
+
+    from backend.apps.agents.core.ws_manager import ws_manager
+    monkeypatch.setattr(ws_manager, "broadcast_global", fake_broadcast)
+    yield sent
+    sh.invalidate_health_cache()
+    sh.p_healed_hooks.clear()
+
+
+@pytest.mark.asyncio
+async def test_a_dead_verdict_arms_its_own_second_look(clean):
+    await sh.report_dead_now("codex")
+    task = sh.p_reprobes.get("codex")
+    assert task is not None and not task.done()
+    task.cancel()
+
+
+@pytest.mark.asyncio
+async def test_a_healthy_reprobe_closes_the_pill_and_resumes_the_chats(clean, monkeypatch):
+    sh.p_cached_result = [{"provider": "claude", "label": "Claude"}, {"provider": "codex", "label": "ChatGPT"}]
+    sh.p_refreshing_since["claude"] = 1.0
+    healed = []
+
+    async def probe(client, model):
+        return "healthy"
+
+    async def hook(provider):
+        healed.append(provider)
+
+    monkeypatch.setattr(sh, "p_probe_one", probe)
+    sh.p_healed_hooks.append(hook)
+    await sh.p_reprobe("claude", "cc/claude-sonnet-5", 0)
+    assert clean[-1] == ("subscriptions:health", {"dead": [{"provider": "codex", "label": "ChatGPT"}]})
+    assert healed == ["claude"]
+    assert "claude" not in sh.p_refreshing_since
+
+
+@pytest.mark.asyncio
+async def test_a_still_dead_reprobe_asks_again_with_backoff_up_to_the_cap(clean, monkeypatch):
+    slept, armed = [], []
+
+    async def nosleep(d):
+        slept.append(d)
+
+    async def probe(client, model):
+        return "dead"
+
+    monkeypatch.setattr(sh.asyncio, "sleep", nosleep)
+    monkeypatch.setattr(sh, "p_probe_one", probe)
+    monkeypatch.setattr(sh, "schedule_reprobe", lambda provider, model, delay=sh.P_REPROBE_S: armed.append(delay) or True)
+    await sh.p_reprobe("codex", "cx/gpt-5.5", 240)
+    await sh.p_reprobe("codex", "cx/gpt-5.5", 1000)
+    await sh.p_reprobe("codex", "cx/gpt-5.5", 1800)
+    assert slept == [240, 1000, 1800]
+    assert armed == [480, 1800, 1800]
+    assert clean == []
+
+
+@pytest.mark.asyncio
+async def test_a_reconnect_cancels_the_reprobe(clean):
+    sh.schedule_reprobe("claude", None, 999)
+    task = sh.p_reprobes["claude"]
+    sh.invalidate_health_cache()
+    await asyncio.sleep(0)
+    assert task.cancelled()
+    assert sh.p_reprobes == {}
+
+
+def test_the_resume_hook_is_registered_at_boot():
+    src = pathlib.Path("backend/apps/agents/agents.py").read_text(encoding="utf-8")
+    assert "p_healed_hooks.append(agent_manager.resume_auth_dead_sessions)" in src
+
+
+@pytest.mark.asyncio
+async def test_a_flapping_login_resumes_a_chat_at_most_twice():
+    from backend.apps.agents.core.models import AgentSession
+    from backend.apps.agents.manager.session.SessionPersistence import AUTH_RESUME_CAP, SessionPersistence
+
+    class Mgr(SessionPersistence):
+        def __init__(self):
+            s = AgentSession(id="flap", name="flap", prompt="x", status="error")
+            s.auth_dead_provider = "claude"
+            self.sessions = {"flap": s}
+            self.sent = 0
+
+        async def send_message(self, sid, text, hidden=False):
+            self.sent += 1
+
+    m = Mgr()
+    for _ in range(AUTH_RESUME_CAP + 2):
+        await m.resume_auth_dead_sessions("claude")
+        m.sessions["flap"].auth_dead_provider = "claude"
+        m.sessions["flap"].status = "error"
+    assert m.sent == AUTH_RESUME_CAP
+    assert m.sessions["flap"].auth_dead_provider == "claude", "the marker stays so the card still says which login died"

backend/tests/test_subscription_health_dedupe.py

@@ -130,6 +130,8 @@ async def test_the_first_mid_refresh_sighting_gets_its_own_second_look(monkeypat
         sent.append((event, data))
 
     monkeypatch.setattr(ws_manager, "broadcast_global", fake_broadcast)
+    armed = []
+    monkeypatch.setattr(sh, "schedule_reprobe", lambda provider, model, delay=sh.P_REPROBE_S: armed.append((provider, delay)) or True)
     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"
@@ -137,6 +139,7 @@ async def test_the_first_mid_refresh_sighting_gets_its_own_second_look(monkeypat
     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 armed == [("codex", sh.P_REPROBE_S)], "a dead verdict arms its own re-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()