diff --git a/backend/apps/agents/agent_manager.py b/backend/apps/agents/agent_manager.py index bc31944e..2f10b257 100644 --- a/backend/apps/agents/agent_manager.py +++ b/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: diff --git a/backend/apps/agents/core/models.py b/backend/apps/agents/core/models.py index 510fbe2f..4a574540 100644 --- a/backend/apps/agents/core/models.py +++ b/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) diff --git a/backend/apps/agents/manager/run/handle_run_error.py b/backend/apps/agents/manager/run/handle_run_error.py index f40e7a5c..b4a1f156 100644 --- a/backend/apps/agents/manager/run/handle_run_error.py +++ b/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"), diff --git a/backend/apps/agents/manager/streaming/auth_retry.py b/backend/apps/agents/manager/streaming/auth_retry.py index db6a3462..9d8797b8 100644 --- a/backend/apps/agents/manager/streaming/auth_retry.py +++ b/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 diff --git a/backend/apps/agents/manager/streaming/handle_assistant_message.py b/backend/apps/agents/manager/streaming/handle_assistant_message.py index 3ab26084..4d876f20 100644 --- a/backend/apps/agents/manager/streaming/handle_assistant_message.py +++ b/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: diff --git a/backend/tests/test_auth_self_heal_all_lanes.py b/backend/tests/test_auth_self_heal_all_lanes.py new file mode 100644 index 00000000..1a66c9e5 --- /dev/null +++ b/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)]