diff --git a/backend/apps/agents/agents.py b/backend/apps/agents/agents.py index 11176d08..b4664e12 100644 --- a/backend/apps/agents/agents.py +++ b/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 diff --git a/backend/apps/agents/core/models.py b/backend/apps/agents/core/models.py index 58420541..a9e41961 100644 --- a/backend/apps/agents/core/models.py +++ b/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. diff --git a/backend/apps/agents/manager/session/SessionPersistence.py b/backend/apps/agents/manager/session/SessionPersistence.py index 2aafc3ea..fdf7f033 100644 --- a/backend/apps/agents/manager/session/SessionPersistence.py +++ b/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 diff --git a/backend/apps/nine_router/subscription_health.py b/backend/apps/nine_router/subscription_health.py index f1d7a6c0..e3908943 100644 --- a/backend/apps/nine_router/subscription_health.py +++ b/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") + diff --git a/backend/tests/test_dead_login_chain.py b/backend/tests/test_dead_login_chain.py index d777c7e9..bb344970 100644 --- a/backend/tests/test_dead_login_chain.py +++ b/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) diff --git a/backend/tests/test_dead_login_reprobe.py b/backend/tests/test_dead_login_reprobe.py new file mode 100644 index 00000000..7a85e4c3 --- /dev/null +++ b/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" diff --git a/backend/tests/test_subscription_health_dedupe.py b/backend/tests/test_subscription_health_dedupe.py index e1b05d9d..4cd270b5 100644 --- a/backend/tests/test_subscription_health_dedupe.py +++ b/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()