[eric] agents: every sub lane self-heals a mid-run token expiry before any card; codex retries wait out the rotation window instead of burning the one shot inside it

backend/apps/agents/agent_manager.py

@@ -43,6 +43,26 @@ MAX_CONCURRENT_TURNS = int(os.environ.get("OSW_MAX_CONCURRENT_TURNS", "8") or "0
 
 
 class AgentManager(SessionLifecycle, SessionPersistence, Messaging, SessionControl, AgentLaunch, SpawnAgentRun, MockAgent, TurnRunner, RunOptions, RunSupport):
+    @typechecked
+    async def dispatch_hidden_continuation(self, session_id: str, prompt: str, delay_s: int) -> None:
+        """Send the self-heal continuation after delay_s (codex rotation windows need ~75s; an
+        instant retry lands inside the same window). A user message during the wait wins: it
+        already resumes the work, so the stale continuation quietly stands down."""
+        if delay_s > 0:
+            p_before = len(getattr(self.sessions.get(session_id), "messages", []) or [])
+            await asyncio.sleep(delay_s)
+            p_session = self.sessions.get(session_id)
+            if p_session is None:
+                return
+            p_tail = [m for m in p_session.messages[p_before:] if getattr(m, "role", "") == "user"]
+            if p_tail:
+                logger.info(f"continuation for {session_id} superseded by a user message during the {delay_s}s wait")
+                return
+        try:
+            await self.send_message(session_id, prompt, hidden=True)
+        except Exception:
+            logger.exception(f"delayed continuation send failed for {session_id}")
+
     @typechecked
     def __init__(self):
         self.sessions: Dict[str, AgentSession] = {}
@@ -223,12 +243,10 @@ class AgentManager(SessionLifecycle, SessionPersistence, Messaging, SessionContr
                     p_continuation_prompt = session.pending_continuation_prompt or "Continue."
                     session.pending_continuation = False
                     session.pending_continuation_prompt = None
-                    asyncio.create_task(self.send_message(
-                        session_id,
-                        p_continuation_prompt,
-                        hidden=True,
-                    ))
-                    logger.info(f"Auto-continuing session {session_id} with hidden prompt")
+                    p_cont_delay = int(getattr(session, "pending_continuation_delay_s", 0) or 0)
+                    session.pending_continuation_delay_s = 0
+                    asyncio.create_task(self.dispatch_hidden_continuation(session_id, p_continuation_prompt, p_cont_delay))
+                    logger.info(f"Auto-continuing session {session_id} with hidden prompt (delay={p_cont_delay}s)")
             except Exception:
                 logger.exception("auto-continuation dispatch failed")
         except asyncio.CancelledError:
@@ -304,8 +322,10 @@ class AgentManager(SessionLifecycle, SessionPersistence, Messaging, SessionContr
                     session.pending_continuation = False
                     session.pending_continuation_prompt = None
                     session.status = "completed"
-                    asyncio.create_task(self.send_message(session_id, p_cont, hidden=True))
-                    logger.info(f"Auto-continuing session {session_id} after a self-healing error path")
+                    p_cont_delay = int(getattr(session, "pending_continuation_delay_s", 0) or 0)
+                    session.pending_continuation_delay_s = 0
+                    asyncio.create_task(self.dispatch_hidden_continuation(session_id, p_cont, p_cont_delay))
+                    logger.info(f"Auto-continuing session {session_id} after a self-healing error path (delay={p_cont_delay}s)")
             except Exception:
                 logger.exception("error-path continuation dispatch failed")
         except BaseException as e:

backend/apps/agents/core/models.py

@@ -152,6 +152,8 @@ class AgentSession(BaseModel):
     suppress_recap_once: bool = False
     # Consecutive dirty deaths this session was MID-TURN for; the crash auto-resume breaker (hermes #30719 pairing: auto-resume must never outrun its circuit breaker).
     crash_interrupt_count: int = 0
+    # Seconds the auto-continuation dispatcher sleeps before sending (codex rotation windows last 1-2 min; an instant retry lands inside the same window and burns the one-shot budget).
+    pending_continuation_delay_s: int = 0
     # Memory prompt block frozen at first compose (prefix-cache discipline: mid-chat fact writes must
     # not shift the prompt bytes). Excluded from persistence so a resumed session re-snapshots fresh.
     memory_snapshot: Optional[str] = Field(default=None, exclude=True)

backend/apps/agents/manager/run/handle_run_error.py

@@ -32,6 +32,21 @@ from backend.apps.agents.core import flight_recorder
 logger = logging.getLogger(__name__)
 
 
+
+@typechecked
+def p_absorb_repeat_card(session: AgentSession, error_msg: Message) -> None:
+    """Append the error card, unless the branch tail is already the IDENTICAL card with nothing
+    after it: a retry ladder re-failing the same way then bumps the existing card instead of
+    stacking a wall of clones (field screenshot 2026-08-19). A user message in between always
+    yields a fresh card, each ask deserves its own honest answer."""
+    p_tail = [m for m in session.messages if getattr(m, "branch_id", None) in (None, session.active_branch_id)]
+    if p_tail and p_tail[-1].role == "system" and p_tail[-1].content == error_msg.content:
+        error_msg.id = p_tail[-1].id
+        p_tail[-1].timestamp = error_msg.timestamp
+        return
+    session.messages.append(error_msg)
+
+
 @typechecked
 def p_report_model_error(subkind: str, session_id: str, session: AgentSession, turn: TurnState,
                          e: BaseException, stderr_tail: str) -> None:
@@ -81,7 +96,7 @@ async def handle_run_error(e: Exception, session: AgentSession, session_id: str,
             "with a larger window."
         )
         error_msg = Message(role="system", content=friendly_msg, branch_id=session.active_branch_id)
-        session.messages.append(error_msg)
+        p_absorb_repeat_card(session, error_msg)
         p_ovf_payload = {
             "session_id": session_id,
             "reason": "long_context_required" if p_tier_gate else "context_overflow",
@@ -128,7 +143,7 @@ async def handle_run_error(e: Exception, session: AgentSession, session_id: str,
             "api.anthropic.com; retrying won't help until one of those changes."
         )
         error_msg = Message(role="system", content=friendly_msg, branch_id=session.active_branch_id)
-        session.messages.append(error_msg)
+        p_absorb_repeat_card(session, error_msg)
         await ws_manager.send_to_session(session_id, "agent:message", {
             "session_id": session_id,
             "message": error_msg.model_dump(mode="json"),
@@ -144,7 +159,7 @@ async def handle_run_error(e: Exception, session: AgentSession, session_id: str,
             "are kept either way."
         )
         error_msg = Message(role="system", content=friendly_msg, branch_id=session.active_branch_id)
-        session.messages.append(error_msg)
+        p_absorb_repeat_card(session, error_msg)
         await ws_manager.send_to_session(session_id, "agent:message", {
             "session_id": session_id,
             "message": error_msg.model_dump(mode="json"),
@@ -189,7 +204,7 @@ async def handle_run_error(e: Exception, session: AgentSession, session_id: str,
             "OpenSwarm Pro."
         )
         error_msg = Message(role="system", content=friendly_msg, branch_id=session.active_branch_id)
-        session.messages.append(error_msg)
+        p_absorb_repeat_card(session, error_msg)
         await ws_manager.send_to_session(session_id, "agent:free_trial_exhausted", {
             "session_id": session_id,
             "message": friendly_msg,
@@ -221,7 +236,7 @@ async def handle_run_error(e: Exception, session: AgentSession, session_id: str,
             "or starting a fresh chat about this topic, usually clears it."
         )
         error_msg = Message(role="system", content=friendly_msg, branch_id=session.active_branch_id)
-        session.messages.append(error_msg)
+        p_absorb_repeat_card(session, error_msg)
         await ws_manager.send_to_session(session_id, "agent:message", {
             "session_id": session_id,
             "message": error_msg.model_dump(mode="json"),
@@ -236,7 +251,7 @@ async def handle_run_error(e: Exception, session: AgentSession, session_id: str,
             "another option in Settings → Models."
         )
         error_msg = Message(role="system", content=friendly_msg, branch_id=session.active_branch_id)
-        session.messages.append(error_msg)
+        p_absorb_repeat_card(session, error_msg)
         await ws_manager.send_to_session(session_id, "agent:out_of_credits", {
             "session_id": session_id,
             "message": friendly_msg,
@@ -251,6 +266,27 @@ async def handle_run_error(e: Exception, session: AgentSession, session_id: str,
         # Three sub-cases the user can hit, with distinct fixes: 1. "No credentials for provider: claude", user picked a -cc route but doesn't have Claude Pro/Max connected via 9Router. Tell them to either connect Claude Pro/Max OR pick a non--cc model. 2. OpenSwarm Pro 401, bearer expired. Reconnect. 3. Anthropic API key 401, wrong key. Re-enter.
         p_model = (session.model or "").lower()
         p_combined = f"{e!s}\n{p_stderr_tail}".lower()
+        p_codex_rotation = (
+            ("codex/" in p_combined or "[codex/" in p_combined or p_model.startswith(("cx/", "gpt-")))
+            and ("authentication token is expired" in p_combined or "authentication token has expired" in p_combined or "401" in p_combined)
+        )
+        # Every sub lane gets ONE silent self-heal before any card; only a missing credential (config problem, retry fails identically) goes straight to the card. Codex waits out its rotation window first.
+        if "no credentials for provider" not in p_combined:
+            from backend.apps.agents.manager.streaming.auth_retry import try_auth_self_heal
+            if try_auth_self_heal(session, delay_s=75 if p_codex_rotation else 5):
+                if p_codex_rotation:
+                    p_notice = Message(
+                        role="system",
+                        content="GPT subscription token just rotated (automatic, every couple minutes). Retrying your request automatically in about a minute, no action needed.",
+                        branch_id=session.active_branch_id,
+                    )
+                    session.messages.append(p_notice)
+                    await ws_manager.send_to_session(session_id, "agent:message", {
+                        "session_id": session_id,
+                        "message": p_notice.model_dump(mode="json"),
+                    })
+                logger.info(f"auth self-heal armed for {session_id} (codex_rotation={p_codex_rotation})")
+                return
         # Codex/OpenAI subscription tokens rotate every ~2-3 minutes, the user sees the rotation window as a 401 with "reset after 1m 59s" or similar. Don't ask them to reconnect; just tell them to wait it out and retry.
         if (
             ("codex/" in p_combined or "[codex/" in p_combined or p_model.startswith(("cx/", "gpt-")))
@@ -293,7 +329,7 @@ async def handle_run_error(e: Exception, session: AgentSession, session_id: str,
             )
             reason = "anthropic_auth_invalid"
         error_msg = Message(role="system", content=friendly_msg, branch_id=session.active_branch_id)
-        session.messages.append(error_msg)
+        p_absorb_repeat_card(session, error_msg)
         try:
             from backend.apps.service.client import submit_diagnostic
             submit_diagnostic({
@@ -320,7 +356,7 @@ async def handle_run_error(e: Exception, session: AgentSession, session_id: str,
         # Upstream rejected the model code itself (e.g. Codex 1211 on a ChatGPT plan that lacks our GPT ids). Track it; the friendly "add an API key / pick another model" card is rendered frontend-side.
         p_report_model_error("unknown_model", session_id, session, turn, e, p_stderr_tail)
         error_msg = Message(role="system", content=f"Error: {str(e)}", branch_id=session.active_branch_id)
-        session.messages.append(error_msg)
+        p_absorb_repeat_card(session, error_msg)
         await ws_manager.send_to_session(session_id, "agent:message", {
             "session_id": session_id,
             "message": error_msg.model_dump(mode="json"),
@@ -330,7 +366,7 @@ async def handle_run_error(e: Exception, session: AgentSession, session_id: str,
         # where the fix is entirely on our side of the wire.
         p_report_model_error("router_unavailable", session_id, session, turn, e, p_stderr_tail)
         error_msg = Message(role="system", content=f"Error: {str(e)}", branch_id=session.active_branch_id)
-        session.messages.append(error_msg)
+        p_absorb_repeat_card(session, error_msg)
         await ws_manager.send_to_session(session_id, "agent:message", {
             "session_id": session_id,
             "message": error_msg.model_dump(mode="json"),
@@ -352,7 +388,7 @@ async def handle_run_error(e: Exception, session: AgentSession, session_id: str,
         if p_cause and "check stderr" in str(e).lower():
             p_card_text += f"\n\nRuntime log tail:\n{p_cause}"
         error_msg = Message(role="system", content=p_card_text, branch_id=session.active_branch_id)
-        session.messages.append(error_msg)
+        p_absorb_repeat_card(session, error_msg)
         await ws_manager.send_to_session(session_id, "agent:message", {
             "session_id": session_id,
             "message": error_msg.model_dump(mode="json"),

backend/apps/agents/manager/streaming/auth_retry.py

@@ -20,13 +20,19 @@ AUTH_RETRY_PROMPT = (
 
 
 @typechecked
-def try_auth_self_heal(session: AgentSession) -> bool:
+def try_auth_self_heal(session: AgentSession, delay_s: int = 0) -> bool:
     """Queue the one hidden retry on a fresh CLI. False = budget spent or a continuation is
-    already pending, and the caller should show the honest banner instead."""
+    already pending, and the caller should show the honest banner instead.
+
+    delay_s: codex tokens ROTATE on a 1-2 minute cadence; an instant retry lands inside the same
+    rotation window, burns the one-shot budget, and the user then gets a banner for a condition
+    that would have healed itself (field screenshot, 2026-08-19). Callers pass ~75s for
+    rotation-shaped failures so the retry fires after the window closes."""
     if session.auth_retry_used or session.pending_continuation:
         return False
     session.auth_retry_used = True
     session.needs_fresh_session = True
     session.pending_continuation = True
     session.pending_continuation_prompt = AUTH_RETRY_PROMPT
+    session.pending_continuation_delay_s = max(0, delay_s)
     return True

backend/apps/agents/manager/streaming/handle_assistant_message.py

@@ -119,14 +119,31 @@ async def handle_assistant_message(
         )
         if looks_like_router_auth_error:
             from backend.apps.agents.manager.streaming.auth_retry import try_auth_self_heal
+            p_is_codex = "codex/" in lower_text or "[codex" in lower_text
             # First expiry in this ask heals silently (fresh CLI + hidden retry); the banner is
             # reserved for the second failure, when the credential is genuinely dead (ENG-294).
-            if not try_auth_self_heal(session):
-                if "codex/" in lower_text or "[codex" in lower_text:
+            # Codex retries wait ~75s so they land AFTER the 1-2 minute rotation window instead of
+            # inside it (an instant retry re-fails and burns the one-shot budget).
+            p_healed = try_auth_self_heal(session, delay_s=75 if p_is_codex else 5)
+            if p_healed and p_is_codex:
+                p_notice = Message(
+                    id=uuid4().hex,
+                    role="system",
+                    content="GPT subscription token just rotated (automatic, every couple minutes). Retrying your request automatically in about a minute, no action needed.",
+                    branch_id=session.active_branch_id,
+                )
+                session.messages.append(p_notice)
+                await ws_manager.send_to_session(session_id, "agent:message", {
+                    "session_id": session_id,
+                    "message": p_notice.model_dump(mode="json"),
+                })
+            if not p_healed:
+                if p_is_codex:
                     friendly = (
-                        "GPT subscription token expired. Open Settings → Models and click "
-                        "Reconnect on the OpenAI / GPT row to refresh, should take ~10s, "
-                        "then send your message again."
+                        "GPT subscription token is still refreshing. This usually clears on "
+                        "its own; wait a minute and send your message again. If it keeps "
+                        "happening, open Settings → Models and click Reconnect on the "
+                        "OpenAI / GPT row."
                     )
                     reason = "codex_token_expired"
                 elif "gemini-cli/" in lower_text or "[gemini" in lower_text:

backend/tests/test_auth_self_heal_all_lanes.py

@@ -0,0 +1,106 @@
+"""Every subscription lane (Claude sub, OpenSwarm Pro, GPT/codex, Gemini) self-heals a mid-run
+token expiry before any card, and the codex retry waits out the 1-2 minute rotation window instead
+of burning its one shot inside it (field screenshot 2026-08-19: a manual-reconnect card while the
+turn was still alive). Only a missing credential, a config problem a retry cannot fix, goes
+straight to the card."""
+
+import asyncio
+
+from backend.apps.agents.core.models import AgentSession, Message
+from backend.apps.agents.manager.run.handle_run_error import handle_run_error
+from backend.apps.agents.manager.streaming.auth_retry import try_auth_self_heal
+from backend.apps.agents.manager.streaming.state import TurnState
+
+# No reset hint here on purpose: a 401 that names its recovery window is claimed by the transient ladder instead (is_auth_error excludes it).
+CODEX_401 = Exception("API Error: 401 [codex/gpt-5.6] authentication token is expired")
+CLAUDE_401 = Exception("API Error: 401 authentication token is expired")
+NO_CREDS = Exception("401 No credentials for provider: claude")
+
+
+def p_session(model: str = "sonnet") -> AgentSession:
+    s = AgentSession(name="t", model=model)
+    s.messages.append(Message(role="user", content="do the thing"))
+    return s
+
+
+def test_delay_lands_on_the_session():
+    s = p_session()
+    assert try_auth_self_heal(s, delay_s=75) is True
+    assert s.pending_continuation_delay_s == 75
+    assert s.pending_continuation is True and s.needs_fresh_session is True
+
+
+def test_budget_is_one_per_ask():
+    s = p_session()
+    assert try_auth_self_heal(s) is True
+    assert try_auth_self_heal(s) is False
+
+
+def test_codex_run_error_heals_with_rotation_wait():
+    s = p_session(model="cx/gpt-5.6")
+    asyncio.run(handle_run_error(CODEX_401, s, "sid-cx", TurnState(), []))
+    assert s.pending_continuation is True
+    assert s.pending_continuation_delay_s == 75
+    cards = [m for m in s.messages if m.role == "system"]
+    assert len(cards) == 1 and "no action needed" in str(cards[0].content)
+    assert "Reconnect" not in str(cards[0].content), "rotation must never demand manual action"
+
+
+def test_claude_run_error_heals_quickly_and_silently():
+    s = p_session(model="sonnet")
+    asyncio.run(handle_run_error(CLAUDE_401, s, "sid-cl", TurnState(), []))
+    assert s.pending_continuation is True
+    assert s.pending_continuation_delay_s == 5
+    assert not any(m.role == "system" for m in s.messages), "short retries stay silent"
+
+
+def test_missing_credential_goes_straight_to_the_card():
+    s = p_session(model="sonnet-cc")
+    asyncio.run(handle_run_error(NO_CREDS, s, "sid-nc", TurnState(), []))
+    assert s.pending_continuation is False, "a retry fails identically; never heal a config problem"
+    cards = [m for m in s.messages if m.role == "system"]
+    assert len(cards) == 1 and "connect" in str(cards[0].content).lower()
+
+
+def test_spent_budget_renders_exactly_one_card():
+    s = p_session(model="cx/gpt-5.6")
+    s.auth_retry_used = True
+    asyncio.run(handle_run_error(CODEX_401, s, "sid-cx2", TurnState(), []))
+    assert s.pending_continuation is False
+    assert len([m for m in s.messages if m.role == "system"]) == 1
+
+
+def test_dispatcher_stands_down_when_the_user_beats_the_delay(monkeypatch):
+    from backend.apps.agents import agent_manager as p_am
+    mgr = p_am.AgentManager.__new__(p_am.AgentManager)
+    s = p_session()
+    mgr.sessions = {"sid": s}
+    sent = []
+
+    async def p_fake_send(sid, prompt, **kw):
+        sent.append(sid)
+    mgr.send_message = p_fake_send
+
+    async def p_fast_sleep(_):
+        s.messages.append(Message(role="user", content="user got here first"))
+    monkeypatch.setattr(p_am.asyncio, "sleep", p_fast_sleep)
+    asyncio.run(mgr.dispatch_hidden_continuation("sid", "redo it", 75))
+    assert sent == [], "a user message during the wait already resumes the work"
+
+
+def test_dispatcher_sends_after_a_quiet_wait(monkeypatch):
+    from backend.apps.agents import agent_manager as p_am
+    mgr = p_am.AgentManager.__new__(p_am.AgentManager)
+    s = p_session()
+    mgr.sessions = {"sid": s}
+    sent = []
+
+    async def p_fake_send(sid, prompt, **kw):
+        sent.append((sid, kw.get("hidden")))
+    mgr.send_message = p_fake_send
+
+    async def p_fast_sleep(_):
+        return None
+    monkeypatch.setattr(p_am.asyncio, "sleep", p_fast_sleep)
+    asyncio.run(mgr.dispatch_hidden_continuation("sid", "redo it", 75))
+    assert sent == [("sid", True)]