[eric] agents: one provider verdict per ask, and a spent plan is never sold as a short wait

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014wtspwSFzZmjCx9UNPAorQ

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

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,

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; "

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 "

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

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"

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
 
 

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"