diff --git a/backend/apps/agents/core/models.py b/backend/apps/agents/core/models.py index 65b95fc4..a4c390a0 100644 --- a/backend/apps/agents/core/models.py +++ b/backend/apps/agents/core/models.py @@ -159,6 +159,12 @@ class AgentSession(BaseModel): lane_credential_dead: bool = False # 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. transient_retry_count: int = 0 + # Set once the provider gives a verdict waiting cannot change (a spent plan, a dead credential). + # Further recovery retries after that only produce cards contradicting the one we already showed. + provider_verdict_final: bool = False + # The last provider-error KIND surfaced this ask. Cards alternated (spent/rate-limit/spent) so + # the identical-string dedup never engaged and the user got a wall of contradictions. + last_provider_error_kind: str = "" reconnect_attempts: int = 0 # True while a turn is parked waiting for the connection back; persisted so a quit DURING the wait is still an owed turn at next boot. awaiting_reconnect: bool = False diff --git a/backend/apps/agents/manager/run/empty_finish.py b/backend/apps/agents/manager/run/empty_finish.py index 98fdb005..c87f124b 100644 --- a/backend/apps/agents/manager/run/empty_finish.py +++ b/backend/apps/agents/manager/run/empty_finish.py @@ -28,8 +28,16 @@ FINAL_NUDGE_PROMPT = ( # Post-cap honesty: the machinery is out of nudges and the turn STILL ended silent, so say so in # the transcript instead of leaving a Done pill over a wall of tool rows. EXHAUSTED_NOTE = ( - "The agent stopped working without a final report. Ask it to summarize, or check the tool " - "results above for where it got to." + "The agent stopped before reporting back. Its work so far is above; send a message to carry on " + "from there." +) + +# Same situation, no work to point at. The old single string told a user with an empty transcript to +# "check the tool results above" when there were none, and to "ask it to summarize" when there was +# nothing to summarize: two instructions that cannot be followed, on the turn they are most alarmed. +EXHAUSTED_NOTE_NO_PROGRESS = ( + "The agent could not get started on this one. Send your message again, or switch this agent to " + "another model." ) logger = logging.getLogger(__name__) @@ -52,17 +60,17 @@ def maybe_nudge_empty_finish(session: AgentSession, session_id: str) -> bool: # tail is still the user's own message. Nudging it would re-send a prompt the model just # refused, so say so instead of ending mute. if p_turn_produced_nothing(session): - p_surface_exhausted(session, session_id) + surface_exhausted(session, session_id) return False if session.empty_finish_nudges >= NUDGE_HARD_CAP: - p_surface_exhausted(session, session_id) + surface_exhausted(session, session_id) return False p_tool_calls = p_count_tool_calls(session) if session.empty_finish_nudges >= 1 and p_tool_calls <= session.empty_finish_progress_mark: # The nudge bought no new work, so re-nudging would ping-pong a model with nothing left. # Refusing is right; ending the ask in SILENCE is not, and that is what the user actually # reports as "the agent just stopped" (Haik's poke storms). - p_surface_exhausted(session, session_id) + surface_exhausted(session, session_id) return False session.empty_finish_progress_mark = p_tool_calls # At high context the silent quit is usually the model choking on the prompt itself, so @@ -111,17 +119,53 @@ def maybe_nudge_empty_finish(session: AgentSession, session_id: str) -> bool: return True @typechecked -def p_surface_exhausted(session: AgentSession, session_id: str) -> None: +def p_recovery_retry_pending(session: AgentSession) -> bool: + """True when the queued continuation is a retry for a PROVIDER failure, not a nudge of ours.""" + if not getattr(session, "pending_continuation", False): + return False + from backend.apps.agents.manager.streaming.auth_retry import ( + AUTH_RETRY_PROMPT, + TRANSIENT_RETRY_PROMPT, + ) + from backend.apps.agents.manager.run.reconnect_resume import RECONNECT_PROMPT + return getattr(session, "pending_continuation_prompt", "") in ( + AUTH_RETRY_PROMPT, TRANSIENT_RETRY_PROMPT, RECONNECT_PROMPT, + ) + + +@typechecked +def turn_showed_work(session: AgentSession) -> bool: + """Whether anything the note could point at actually exists since the user last spoke.""" + from backend.apps.agents.manager.session.history_compaction import get_branch_messages + msgs = [m for m in get_branch_messages(session) if not getattr(m, "hidden", False)] + p_last_user = -1 + for i, m in enumerate(msgs): + if getattr(m, "role", "") == "user": + p_last_user = i + return any(getattr(m, "role", "") in ("tool_call", "tool_result") + for m in msgs[p_last_user + 1:]) + + +@typechecked +def surface_exhausted(session: AgentSession, session_id: str) -> None: """All nudges spent and the turn still ended mute: put one honest system line in the transcript, once per exhaustion (the flag resets with the counters on a real user message).""" if getattr(session, "empty_finish_surfaced", False): return + # A RECOVERY retry means the turn is still going, so saying it stopped is a lie told at the + # worst possible moment. The nudge ladder also rides pending_continuation, and its whole purpose + # is to end in this very message, so the flag alone is the wrong test: key on whose continuation + # it is. Leave empty_finish_surfaced unset either way, so the honest line still fires if the + # recovery itself ends mute. + if getattr(session, "awaiting_reconnect", False) or p_recovery_retry_pending(session): + return session.empty_finish_surfaced = True try: import asyncio from backend.apps.agents.core.models import Message from backend.apps.agents.core.ws_manager import ws_manager - p_msg = Message(role="system", content=EXHAUSTED_NOTE, branch_id=session.active_branch_id) + p_note = EXHAUSTED_NOTE if turn_showed_work(session) else EXHAUSTED_NOTE_NO_PROGRESS + p_msg = Message(role="system", content=p_note, branch_id=session.active_branch_id) session.messages.append(p_msg) asyncio.get_running_loop().create_task(ws_manager.send_to_session(session_id, "agent:message", { "session_id": session_id, diff --git a/backend/apps/agents/manager/run/turn_spoke.py b/backend/apps/agents/manager/run/turn_spoke.py index a2fb759c..29d92a89 100644 --- a/backend/apps/agents/manager/run/turn_spoke.py +++ b/backend/apps/agents/manager/run/turn_spoke.py @@ -24,7 +24,12 @@ from typing import List from typeguard import typechecked from backend.apps.agents.core.models import AgentSession, Message -from backend.apps.agents.manager.run.empty_finish import EXHAUSTED_NOTE, P_ANSWER_TOOL_MARKERS +from backend.apps.agents.manager.run.empty_finish import ( + EXHAUSTED_NOTE, + EXHAUSTED_NOTE_NO_PROGRESS, + P_ANSWER_TOOL_MARKERS, + turn_showed_work, +) from backend.apps.agents.manager.session.history_compaction import get_branch_messages logger = logging.getLogger(__name__) @@ -81,8 +86,11 @@ def ensure_turn_spoke(session: AgentSession, session_id: str) -> bool: if not turn_left_the_user_with_nothing(session): return False + # Point at work only when there is work to point at; a turn that did nothing gets the copy + # that names an action the user can actually take. + p_note = EXHAUSTED_NOTE if turn_showed_work(session) else EXHAUSTED_NOTE_NO_PROGRESS session.messages.append( - Message(role="system", content=EXHAUSTED_NOTE, branch_id=session.active_branch_id) + Message(role="system", content=p_note, branch_id=session.active_branch_id) ) logger.warning( f"Agent {session_id}: turn ended with nothing readable and no detector claimed it; " diff --git a/backend/apps/agents/manager/streaming/handle_assistant_message.py b/backend/apps/agents/manager/streaming/handle_assistant_message.py index aa4ec238..ce54e68b 100644 --- a/backend/apps/agents/manager/streaming/handle_assistant_message.py +++ b/backend/apps/agents/manager/streaming/handle_assistant_message.py @@ -198,7 +198,13 @@ async def handle_assistant_message( p_delay = 0 p_healed = False - if is_transient(p_provider_error): + # A verdict waiting cannot change ends the ask's recovery ladder. Without this, the + # retries kept running after we had already told the user to switch models, and each + # one added a card contradicting that advice (packaged drill: seven cards, three + # mutually exclusive instructions, twice). + if not is_transient(p_provider_error): + session.provider_verdict_final = True + elif not session.provider_verdict_final: p_delay = min(int(p_provider_error.reset_seconds or 0), 900) p_healed = try_transient_self_heal(session, delay_s=p_delay) @@ -220,10 +226,25 @@ async def handle_assistant_message( content=p_copy, branch_id=session.active_branch_id, ) - # A retry ladder re-failing the same way must bump one card, not stack clones; the live - # drill produced three before this line existed. - from backend.apps.agents.manager.run.handle_run_error import absorb_repeat_card - absorb_repeat_card(session, p_card) + # Dedup by KIND, not by exact string. The identical-card absorber never engaged here + # because consecutive cards alternated wording (spent plan / rate limit / spent plan), + # so the same underlying failure stacked a wall. One card per kind per ask; a repeat + # rewrites it in place so the newest wording wins without adding a row. + p_prev_kind = getattr(session, "last_provider_error_kind", "") + p_existing = None + if p_prev_kind == p_provider_error.kind: + for m in reversed(session.messages): + if m.role == "system" and not getattr(m, "hidden", False): + p_existing = m + break + if p_existing is not None: + p_existing.content = p_copy + p_existing.timestamp = p_card.timestamp + p_card = p_existing + else: + from backend.apps.agents.manager.run.handle_run_error import absorb_repeat_card + absorb_repeat_card(session, p_card) + session.last_provider_error_kind = p_provider_error.kind logger.warning( f"Agent {session_id}: provider returned {p_provider_error.kind} " f"(status={p_provider_error.status}, lane={p_provider_error.lane}) as assistant " diff --git a/backend/apps/agents/manager/streaming/provider_error_speech.py b/backend/apps/agents/manager/streaming/provider_error_speech.py index e4be2bfa..cede340e 100644 --- a/backend/apps/agents/manager/streaming/provider_error_speech.py +++ b/backend/apps/agents/manager/streaming/provider_error_speech.py @@ -41,11 +41,19 @@ P_ROUTER_STAMP = re.compile(r"\[[a-z0-9._-]+/[a-z0-9._-]+\]\s*\[(\d{3})\]", re.I # A bare status inside the envelope, e.g. "Request rejected (429)". P_STATUS = re.compile(r"\((\d{3})\)|\b(4\d{2}|5\d{2})\b") -# "Resets in 125h40m51s" / "(reset after 2m)" / "(reset after 1m 56s)". +# "Resets in 125h40m51s" / "(reset after 2m)" / "(reset after 1m 56s)". Gemini sends BOTH in one +# envelope, and they mean different things: the first is when the subscription quota actually +# resets, the second is only the router's own retry hint. Reading whichever came first made the +# same 429 say "5 days, switch models" one turn and "2 minutes, do nothing" the next (packaged +# drill 2026-08-20, seven contradictory cards in one ask), so collect them all and trust the longest. P_RESET = re.compile( r"reset(?:s)?\s+(?:in|after)\s+((?:\d+\s*[hms]\s*)+)", re.I ) +# Wording that means the plan itself is spent, not that we are going too fast. Waiting cannot fix it. +P_SUBSCRIPTION_SPENT = ("quota reached", "upgrade your subscription", "exceeded your quota", + "subscription limit", "plan limit") + AUTH = "auth" QUOTA = "quota" CONNECTION = "connection" @@ -59,6 +67,9 @@ class ProviderError(BaseModel): model_config = ConfigDict(validate_assignment=True) kind: str + # True when the provider said the PLAN is spent rather than that we are going too fast; the + # reset window is unreliable for this (same condition, sometimes 5 days, sometimes 2 minutes). + subscription_spent: bool status: Optional[int] lane: Optional[str] reset_seconds: Optional[int] @@ -117,10 +128,9 @@ def classify_provider_error(text: str) -> Optional[ProviderError]: if m: status = int(m.group(1) or m.group(2)) - reset_seconds = None - reset_m = P_RESET.search(stripped) - if reset_m: - reset_seconds = parse_duration(reset_m.group(1)) + p_windows = [parse_duration(m.group(1)) for m in P_RESET.finditer(stripped)] + p_windows = [w for w in p_windows if w] + reset_seconds = max(p_windows) if p_windows else None # Status first: it is the provider's own verdict. Words only for shapes carrying no status at all. if status in (401, 403): @@ -139,7 +149,8 @@ def classify_provider_error(text: str) -> Optional[ProviderError]: kind = UNKNOWN return ProviderError( - kind=kind, status=status, lane=lane, reset_seconds=reset_seconds, raw=stripped + kind=kind, status=status, lane=lane, reset_seconds=reset_seconds, raw=stripped, + subscription_spent=any(w in low for w in P_SUBSCRIPTION_SPENT), ) @@ -173,6 +184,13 @@ def user_facing_sentence(err: ProviderError, model: str) -> str: who = model if err.kind == QUOTA: + if err.subscription_spent: + # Only quote a window big enough to BE a quota reset. A couple of minutes is the + # router's retry hint, and repeating it as the plan's reset time is just a wrong fact. + p_when = (f" It resets in {p_humanize_window(err.reset_seconds)}." + if err.reset_seconds and err.reset_seconds > 6 * 3600 else "") + return (f"{who} has used up its subscription allowance.{p_when} Switch this agent to " + "another model to keep going.") if err.reset_seconds and err.reset_seconds > 6 * 3600: return ( f"{who} has hit its subscription limit and will not reset for " @@ -218,5 +236,8 @@ def is_transient(err: ProviderError) -> bool: if err.kind in (CONNECTION, OVERLOADED): return True if err.kind == QUOTA: + # An exhausted plan does not heal by waiting; parking on it is a silent stop in disguise. + if err.subscription_spent: + return False return err.reset_seconds is not None and err.reset_seconds <= 6 * 3600 return False diff --git a/backend/tests/test_assistant_message.py b/backend/tests/test_assistant_message.py index 5892c1c2..56d1ad1c 100644 --- a/backend/tests/test_assistant_message.py +++ b/backend/tests/test_assistant_message.py @@ -138,6 +138,11 @@ REAL_GEMINI_LONG_QUOTA = ( "(reset after 2m)" ) REAL_CLAUDE_CONNECTION = "API Error: Unable to connect. Is the computer able to access the url?" +# A genuine "slow down", with no claim that the plan is spent. This one IS worth waiting out. +TRUE_RATE_LIMIT = ( + "API Error: Request rejected (429) ยท [antigravity/gemini-3-flash] [429]: slow down. " + "(reset after 2m)" +) @pytest.mark.asyncio @@ -161,10 +166,10 @@ async def test_a_short_rate_limit_resumes_itself_rather_than_asking_the_user(): session, turn, thinking = p_fixt() with patch.object(assistant_message.ws_manager, "send_to_session", new=AsyncMock()): await assistant_message.handle_assistant_message( - p_asst([TextBlock(text=REAL_GEMINI_SHORT_QUOTA)]), session, session.id, turn, + p_asst([TextBlock(text=TRUE_RATE_LIMIT)]), session, session.id, turn, thinking, {}, {}) assert session.pending_continuation is True, "a 2-minute wait is ours to do, not the user's" - assert session.pending_continuation_delay_s == 116, "wait the window the provider named" + assert session.pending_continuation_delay_s == 120, "wait the window the provider named" assert session.auth_retry_used is False, "a rate limit must not spend the expired-token retry" @@ -201,7 +206,7 @@ async def test_the_transient_budget_is_two_then_it_says_so(): session.pending_continuation = False # dispatcher consumes it between turns with patch.object(assistant_message.ws_manager, "send_to_session", new=AsyncMock()): await assistant_message.handle_assistant_message( - p_asst([TextBlock(text=REAL_GEMINI_SHORT_QUOTA)]), session, session.id, turn, + p_asst([TextBlock(text=TRUE_RATE_LIMIT)]), session, session.id, turn, thinking, {}, {}) assert session.transient_retry_count == 2, "budget is two, not unbounded" assert session.pending_continuation is False, "the third failure stops retrying" @@ -225,3 +230,64 @@ async def test_the_agent_can_still_talk_about_an_error_it_saw(): "the agent's own analysis must reach the user intact" assert not any(m.role == "system" for m in session.messages) assert session.pending_continuation is False + + +@pytest.mark.asyncio +async def test_a_spent_plan_is_never_retried_however_short_its_window(): + """Packaged drill 2026-08-20: the same 'quota reached' envelope carries a 5-day reset one turn + and a 2-minute one the next, so a window-only reading told the user to switch models and then, + seconds later, to sit tight. A spent plan is spent whatever number rides along.""" + session, turn, thinking = p_fixt() + with patch.object(assistant_message.ws_manager, "send_to_session", new=AsyncMock()): + await assistant_message.handle_assistant_message( + p_asst([TextBlock(text=REAL_GEMINI_SHORT_QUOTA)]), session, session.id, turn, + thinking, {}, {}) + assert session.pending_continuation is False, "waiting cannot refill a plan" + assert session.provider_verdict_final is True + body = [m for m in session.messages if m.role == "system"][0].content + assert "automatically" not in body.lower(), "never promise a resume that cannot happen" + assert "switch" in body.lower() + + +@pytest.mark.asyncio +async def test_a_final_verdict_stops_the_ladder_and_the_cards(): + """The wall: after a terminal verdict the ladder kept retrying and every retry added a card + contradicting the one before it. Seven cards, three mutually exclusive instructions.""" + session, turn, thinking = p_fixt() + for text in (REAL_GEMINI_SHORT_QUOTA, TRUE_RATE_LIMIT, REAL_GEMINI_SHORT_QUOTA): + session.pending_continuation = False + with patch.object(assistant_message.ws_manager, "send_to_session", new=AsyncMock()): + await assistant_message.handle_assistant_message( + p_asst([TextBlock(text=text)]), session, session.id, turn, thinking, {}, {}) + p_cards = [m for m in session.messages if m.role == "system"] + assert len(p_cards) == 1, f"one verdict per ask, got {len(p_cards)}" + assert session.transient_retry_count == 0, "no retries after a terminal verdict" + + +@pytest.mark.asyncio +async def test_the_same_kind_twice_rewrites_one_card_instead_of_stacking(): + session, turn, thinking = p_fixt() + for _ in range(3): + session.pending_continuation = False + with patch.object(assistant_message.ws_manager, "send_to_session", new=AsyncMock()): + await assistant_message.handle_assistant_message( + p_asst([TextBlock(text=REAL_CLAUDE_CONNECTION)]), session, session.id, turn, + thinking, {}, {}) + assert len([m for m in session.messages if m.role == "system"]) == 1 + + +@pytest.mark.asyncio +async def test_a_different_kind_still_earns_its_own_card(): + """NEGATIVE CONTROL. Dedup by kind must not swallow a genuinely different failure, or the user + stops being told when the problem changes underneath them.""" + session, turn, thinking = p_fixt() + with patch.object(assistant_message.ws_manager, "send_to_session", new=AsyncMock()): + await assistant_message.handle_assistant_message( + p_asst([TextBlock(text=REAL_CLAUDE_CONNECTION)]), session, session.id, turn, + thinking, {}, {}) + session.pending_continuation = False + await assistant_message.handle_assistant_message( + p_asst([TextBlock(text=REAL_GEMINI_SHORT_QUOTA)]), session, session.id, turn, + thinking, {}, {}) + p_cards = [m for m in session.messages if m.role == "system"] + assert len(p_cards) == 2, "connection and quota are different problems" diff --git a/backend/tests/test_empty_finish.py b/backend/tests/test_empty_finish.py index 43d6b6ab..a28cf60f 100644 --- a/backend/tests/test_empty_finish.py +++ b/backend/tests/test_empty_finish.py @@ -14,6 +14,8 @@ from backend.apps.agents.manager.run.empty_finish import ( NUDGE_PROMPT, maybe_nudge_empty_finish, turn_finished_empty, + EXHAUSTED_NOTE, + EXHAUSTED_NOTE_NO_PROGRESS, ) @@ -110,11 +112,11 @@ def test_loop_renudges_while_progressing_then_caps(monkeypatch) -> None: asyncio.run(main()) assert continues == [] assert session.empty_finish_nudges == NUDGE_HARD_CAP - p_sys = [m for m in session.messages if m.role == "system" and "without a final report" in str(m.content)] + p_sys = [m for m in session.messages if m.role == "system" and str(m.content) in (EXHAUSTED_NOTE, EXHAUSTED_NOTE_NO_PROGRESS)] assert len(p_sys) == 1, "exhaustion surfaces exactly once" # A second exhausted quit in the same ask must NOT stack another line. asyncio.run(main()) - p_sys = [m for m in session.messages if m.role == "system" and "without a final report" in str(m.content)] + p_sys = [m for m in session.messages if m.role == "system" and str(m.content) in (EXHAUSTED_NOTE, EXHAUSTED_NOTE_NO_PROGRESS)] assert len(p_sys) == 1 diff --git a/backend/tests/test_exhausted_note.py b/backend/tests/test_exhausted_note.py new file mode 100644 index 00000000..f6c29108 --- /dev/null +++ b/backend/tests/test_exhausted_note.py @@ -0,0 +1,64 @@ +"""What the packaged drill caught that no unit test had: the honest "it stopped" line fired while a +recovery retry was already in flight, and it pointed at tool results that did not exist. + +Split out of test_empty_finish.py because it is a different concern (what the user READS when the +machinery gives up) and because that file had reached its line ceiling. +""" + +from backend.apps.agents.core.models import AgentSession, Message +from backend.apps.agents.manager.run.empty_finish import ( + EXHAUSTED_NOTE, + EXHAUSTED_NOTE_NO_PROGRESS, + NUDGE_PROMPT, + surface_exhausted, +) + +# --- found by the packaged drill, 2026-08-20 ------------------------------------------------------ +# +# A Gemini 401 armed a silent auth retry, and the user was still told "the agent stopped working +# without a final report. Ask it to summarize, or check the tool results above" -- on a turn with +# ZERO tool results and a fix already in flight. Three defects in one card: it contradicted the +# retry, it pointed at work that did not exist, and it handed the user the job. + +def test_a_recovery_retry_in_flight_is_not_announced_as_a_stop(): + from backend.apps.agents.manager.streaming.auth_retry import try_auth_self_heal + s = AgentSession(name="t", model="sonnet") + s.messages.append(Message(role="user", content="do the thing")) + assert try_auth_self_heal(s) is True, "precondition: a retry really was armed" + surface_exhausted(s, s.id) + assert not [m for m in s.messages if m.role == "system"], \ + "a queued recovery retry means the turn is still going; do not claim it stopped" + assert s.empty_finish_surfaced is False, \ + "the flag must stay unset so the honest line can still fire if the retry ends mute" + + +def test_the_nudge_ladder_still_reaches_its_own_ending(): + """NEGATIVE CONTROL. The ladder rides pending_continuation too, and its whole point is to end + in this message, so a guard keyed on the flag alone would silently delete the ladder's ending.""" + s = AgentSession(name="t", model="sonnet") + s.messages.append(Message(role="user", content="do the thing")) + s.pending_continuation = True + s.pending_continuation_prompt = NUDGE_PROMPT + surface_exhausted(s, s.id) + assert len([m for m in s.messages if m.role == "system"]) == 1, \ + "our own nudge must not be mistaken for a provider recovery" + + +def test_a_turn_with_no_work_is_not_told_to_check_work(): + s = AgentSession(name="t", model="sonnet") + s.messages.append(Message(role="user", content="do the thing")) + surface_exhausted(s, s.id) + body = [m for m in s.messages if m.role == "system"][0].content + assert body == EXHAUSTED_NOTE_NO_PROGRESS + assert "above" not in body, "there is nothing above to point at" + assert "summarize" not in body.lower(), "never hand the user the agent's job" + + +def test_a_turn_that_did_work_points_at_it(): + s = AgentSession(name="t", model="sonnet") + s.messages.append(Message(role="user", content="do the thing")) + s.messages.append(Message(role="tool_call", content={"tool": "Read"})) + surface_exhausted(s, s.id) + body = [m for m in s.messages if m.role == "system"][0].content + assert body == EXHAUSTED_NOTE + assert "summarize" not in body.lower(), "even with work, do not make the user extract it"