[eric] agents: a context overflow now compacts and retries once instead of dying, and never fakes a completed

backend/apps/agents/agent_manager.py

@@ -181,14 +181,17 @@ class AgentManager(SessionLifecycle, SessionPersistence, Messaging, SessionContr
             turn.stream_text_msg_id = None
             turn.stream_text_accum = ""
         except Exception as e:
-            from backend.apps.agents.core.error_classify import is_context_pressure_death
+            from backend.apps.agents.core.error_classify import is_context_overflow_error, is_context_pressure_death
             p_stderr_tail = "\n".join(p_stderr_buffer[-50:])
-            if not context_valve_retry and is_context_pressure_death(e, turn.compact_boundaries, extra_text=p_stderr_tail):
-                # Pressure-release valve: the CLI compacted this turn and still died (its "autocompact is thrashing" giving-up class). Its resume transcript is beyond saving, but ours isn't: rebuild from the local mirror via the proven fresh-session recap path and transparently re-run the turn ONCE.
+            p_overflow = is_context_overflow_error(e, extra_text=p_stderr_tail)
+            if not context_valve_retry and (p_overflow or is_context_pressure_death(e, turn.compact_boundaries, extra_text=p_stderr_tail)):
+                # Pressure-release valve, two entry shapes: the CLI compacted this turn and still died (autocompact thrash), or the provider rejected the query outright as over the context window. Either way the CLI's resume transcript is beyond saving, but ours isn't: rebuild from the local mirror via the proven fresh-session recap path and transparently re-run the turn ONCE.
                 logger.warning(
-                    f"Agent {session_id}: context-pressure death after "
+                    f"Agent {session_id}: {'context overflow' if p_overflow else 'context-pressure death'} after "
                     f"{turn.compact_boundaries} compact boundaries; one fresh-session recap retry"
                 )
+                # The recap rebuild trims at compacted_through_msg_id; an overflow can hit before the proactive threshold ever fired, so force a cutoff or the rebuilt prompt is full history again.
+                self.maybe_compact(session, force=True)
                 session.needs_fresh_session = True
                 if turn.stream_text_msg_id:
                     await ws_manager.send_to_session(session_id, "agent:stream_end", {
@@ -210,9 +213,10 @@ class AgentManager(SessionLifecycle, SessionPersistence, Messaging, SessionContr
                     logger.debug("context_recovered broadcast failed", exc_info=True)
                 try:
                     from backend.apps.service.client import submit_diagnostic
-                    from backend.apps.agents.core.error_classify import redact_for_telemetry
+                    from backend.apps.agents.core.redact_for_telemetry import redact_for_telemetry
                     submit_diagnostic({
                         "kind": "context_pressure_valve",
+                        "trigger": "overflow" if p_overflow else "pressure_death",
                         "session_id": session_id,
                         "model": session.model,
                         "compact_boundaries": turn.compact_boundaries,

backend/apps/agents/core/error_classify.py

@@ -5,28 +5,6 @@ import anthropic
 import httpx
 from typeguard import typechecked
 
-# Secret shapes that must never ride along when we ship a stderr tail or an error string to telemetry. own_key mode means the subprocess stderr can echo the user's OWN provider key, so this scrub is the wall between a diagnostic and a key leak; over-redacting is fine, leaking is not.
-P_TELEMETRY_SECRET_PATTERNS = (
-    re.compile(r"sk-ant-[A-Za-z0-9_\-]{12,}"),
-    re.compile(r"sk-[A-Za-z0-9_\-]{16,}"),
-    re.compile(r"AIza[A-Za-z0-9_\-]{20,}"),
-    re.compile(r"gh[pousr]_[A-Za-z0-9]{20,}"),
-    re.compile(r"(?i)bearer\s+[A-Za-z0-9._\-]{12,}"),
-    re.compile(r"(?i)\b(?:api[_-]?key|access[_-]?token|refresh[_-]?token|secret|password|authorization)\b[\"']?\s*[:=]\s*[\"']?[A-Za-z0-9._\-]{6,}"),
-)
-
-
-def redact_for_telemetry(text: str, *, limit: int = 2000) -> str:
-    """Scrub secret-shaped substrings, then keep the tail (where the real error
-    lands), bounded so a runaway log can't bloat the payload. Every raw
-    error/stderr string goes through here before it leaves the machine."""
-    if not text:
-        return ""
-    for pat in P_TELEMETRY_SECRET_PATTERNS:
-        text = pat.sub("[redacted]", text)
-    return text[-limit:]
-
-
 # Patterns that indicate an upstream transient problem (overload / rate limit / infra blip), safe to silently retry with backoff. Checked against the stringified exception from claude_agent_sdk / Claude CLI.
 TRANSIENT_CAPACITY_PATTERNS = re.compile(
     r"(?:\b(?:429|500|502|503|504|529)\b"
@@ -90,6 +68,31 @@ def is_long_context_error(exc: BaseException, extra_text: str = "") -> bool:
     ))
 
 
+@typechecked
+def is_context_overflow_error(exc: BaseException, extra_text: str = "") -> bool:
+    """The context-window overflow family across providers: Anthropic's 'prompt is too
+    long' 400 and long-context tier gate, OpenAI's 'maximum context length' /
+    'context_length_exceeded' / 'request too large', Gemini's 'input token count exceeds'.
+    Gates the reactive compact-and-retry valve in run_agent_loop; a misfire costs one
+    bounded fresh-session recap retry, a miss means today's terminal error card.
+    """
+    if is_long_context_error(exc, extra_text):
+        return True
+    combined = f"{exc!s}\n{extra_text}".strip()
+    if not combined:
+        return False
+    return bool(re.search(
+        r"prompt\s+is\s+too\s+long"
+        r"|maximum\s+context\s+length"
+        r"|context[_\s-]?length[_\s-]?exceeded"
+        r"|input\s+token\s+count[^.\n]{0,40}exceeds"
+        r"|exceeds?\s+the\s+(?:maximum\s+)?(?:context|token)\s+(?:window|limit)"
+        r"|request\s+too\s+large",
+        combined,
+        re.IGNORECASE,
+    ))
+
+
 @typechecked
 def is_free_trial_exhausted(exc: BaseException, extra_text: str = "") -> bool:
     """True when the cloud says the machine's free runs are spent (a 402 with
@@ -215,6 +218,9 @@ def is_transient_capacity_error(exc: BaseException, extra_text: str = "") -> boo
     combined = f"{exc!s}\n{extra_text}".strip()
     if combined and NON_TRANSIENT_PATTERNS.search(combined):
         return False
+    # An overflow can arrive dressed as a 429 ("request too large"); retrying the identical oversized request is guaranteed futile, the valve owns it.
+    if is_context_overflow_error(exc, extra_text):
+        return False
     # Ahead of the empty-string bail on purpose: what the exception IS doesn't depend on whether it bothered to say anything.
     if isinstance(exc, P_TRANSIENT_EXC_TYPES):
         return True
@@ -284,16 +290,3 @@ def is_context_pressure_death(exc: BaseException, compact_boundaries: int, extra
     return True
 
 
-@typechecked
-def extract_reset_hint(text: str) -> str:
-    """Pull a human reset phrase ('at 7:42 AM', 'in 2h 30m', 'after 1m 59s') out of
-    a provider usage error so we can tell the user when their limit comes back.
-    """
-    if not text:
-        return ""
-    m = re.search(
-        r"(?:try\s+again|resets?|reset)\s+((?:in|at|after)\s+[^.\n)]{1,40})",
-        text,
-        re.IGNORECASE,
-    )
-    return m.group(1).strip() if m else ""

backend/apps/agents/core/extract_reset_hint.py

@@ -0,0 +1,18 @@
+import re
+
+from typeguard import typechecked
+
+
+@typechecked
+def extract_reset_hint(text: str) -> str:
+    """Pull a human reset phrase ('at 7:42 AM', 'in 2h 30m', 'after 1m 59s') out of
+    a provider usage error so we can tell the user when their limit comes back.
+    """
+    if not text:
+        return ""
+    m = re.search(
+        r"(?:try\s+again|resets?|reset)\s+((?:in|at|after)\s+[^.\n)]{1,40})",
+        text,
+        re.IGNORECASE,
+    )
+    return m.group(1).strip() if m else ""

backend/apps/agents/core/redact_for_telemetry.py

@@ -0,0 +1,25 @@
+import re
+
+from typeguard import typechecked
+
+# Secret shapes that must never ride along when we ship a stderr tail or an error string to telemetry. own_key mode means the subprocess stderr can echo the user's OWN provider key, so this scrub is the wall between a diagnostic and a key leak; over-redacting is fine, leaking is not.
+P_TELEMETRY_SECRET_PATTERNS = (
+    re.compile(r"sk-ant-[A-Za-z0-9_\-]{12,}"),
+    re.compile(r"sk-[A-Za-z0-9_\-]{16,}"),
+    re.compile(r"AIza[A-Za-z0-9_\-]{20,}"),
+    re.compile(r"gh[pousr]_[A-Za-z0-9]{20,}"),
+    re.compile(r"(?i)bearer\s+[A-Za-z0-9._\-]{12,}"),
+    re.compile(r"(?i)\b(?:api[_-]?key|access[_-]?token|refresh[_-]?token|secret|password|authorization)\b[\"']?\s*[:=]\s*[\"']?[A-Za-z0-9._\-]{6,}"),
+)
+
+
+@typechecked
+def redact_for_telemetry(text: str, *, limit: int = 2000) -> str:
+    """Scrub secret-shaped substrings, then keep the tail (where the real error
+    lands), bounded so a runaway log can't bloat the payload. Every raw
+    error/stderr string goes through here before it leaves the machine."""
+    if not text:
+        return ""
+    for pat in P_TELEMETRY_SECRET_PATTERNS:
+        text = pat.sub("[redacted]", text)
+    return text[-limit:]

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

@@ -12,17 +12,18 @@ from backend.apps.agents.core.ws_manager import ws_manager
 from backend.apps.settings.settings import load_settings
 from backend.apps.agents.manager.streaming.state import TurnState
 from backend.apps.agents.core.error_classify import (
+    is_context_overflow_error,
     is_long_context_error,
     is_transient_capacity_error,
     is_free_trial_exhausted,
     is_out_of_tokens,
-    extract_reset_hint,
     is_auth_error,
     is_cli_binary_missing,
     is_unknown_model_error,
     parse_retry_after,
-    redact_for_telemetry,
 )
+from backend.apps.agents.core.extract_reset_hint import extract_reset_hint
+from backend.apps.agents.core.redact_for_telemetry import redact_for_telemetry
 
 logger = logging.getLogger(__name__)
 
@@ -37,32 +38,25 @@ async def handle_run_error(e: Exception, session: AgentSession, session_id: str,
         p_stderr_tail = "\n".join(p_stderr_buffer[-50:])
     except Exception:
         p_stderr_tail = ""
-    # If we already streamed a substantive assistant response this turn, the user got their answer; the error fired on a subsequent step (title gen, follow-up tool turn, etc.). Don't blast a "context exceeded" card over a completed reply.
-    p_streamed_substantive = bool(turn.stream_text_msg_id) and turn.current_turn_emitted
-    if p_streamed_substantive and is_long_context_error(e, extra_text=p_stderr_tail):
-        # Mark the session completed (not error), keep the assistant reply visible, and skip the overflow card. The next user turn will properly hit the pre-send guard if the chat is still over cap.
-        session.status = "completed"
-        if turn.stream_text_msg_id:
-            try:
-                await ws_manager.send_to_session(session_id, "agent:stream_end", {
-                    "session_id": session_id,
-                    "message_id": turn.stream_text_msg_id,
-                })
-            except Exception:
-                pass
-        return
-    if is_long_context_error(e, extra_text=p_stderr_tail):
+    # No completed-mask here anymore: current_turn_emitted stays True until a ResultMessage lands, so the old "already answered" early-return fired on every MID-TASK death (models narrate between tool calls) and converted a dead run into a fake "completed". Reaching this handler with an overflow means the valve's compact-and-retry already failed once; the user must see the card.
+    if is_context_overflow_error(e, extra_text=p_stderr_tail):
+        p_tier_gate = is_long_context_error(e, extra_text=p_stderr_tail)
         friendly_msg = (
             "This conversation has grown too large for your account's "
             "standard context window. Long-context requests require an "
             "upgraded tier, switch to Chat mode or start a fresh chat "
             "to continue."
+        ) if p_tier_gate else (
+            "This conversation outgrew the model's context window, and "
+            "automatic compaction couldn't shrink it enough. Start a fresh "
+            "chat (your recent context carries over) or switch to a model "
+            "with a larger window."
         )
         error_msg = Message(role="system", content=friendly_msg, branch_id=session.active_branch_id)
         session.messages.append(error_msg)
         p_ovf_payload = {
             "session_id": session_id,
-            "reason": "long_context_required",
+            "reason": "long_context_required" if p_tier_gate else "context_overflow",
             "message": friendly_msg,
             "model": session.model,
             "provider": session.provider,

backend/apps/nine_router/process.py

@@ -349,7 +349,7 @@ def p_report_start_failure(reason: str, *, detail: str = "", **fields: Any) -> N
     every other telemetry string. Never raises."""
     logger.warning("9Router start failed (%s)", reason)
     try:
-        from backend.apps.agents.core.error_classify import redact_for_telemetry
+        from backend.apps.agents.core.redact_for_telemetry import redact_for_telemetry
         from backend.apps.service.client import submit_diagnostic
         payload: dict[str, Any] = {
             "kind": "9router_start_failed",

backend/tests/test_context_pressure_valve.py

@@ -102,6 +102,54 @@ def test_no_valve_without_compaction_churn(monkeypatch) -> None:
     assert [m for m in session.messages if m.role == "system" and str(m.content).startswith("Error:")]
 
 
+def test_overflow_valve_retries_with_forced_compaction(monkeypatch) -> None:
+    from backend.apps.agents.core.models import Message
+    from backend.apps.agents.manager.streaming.handle_result_message import TurnResultError
+    session = p_seed_session()
+    # Enough history that the forced compact mark has something to cut (keeps last 6).
+    for i in range(10):
+        session.messages.append(Message(role="user" if i % 2 == 0 else "assistant", content=f"m{i}", branch_id=session.active_branch_id))
+    calls: list = []
+
+    async def fake_run_turn(sess, session_id, prompt_content, options, options_kwargs,
+                            turn, thinking, stderr, resolved_model, api_type,
+                            global_settings, force_respawn=False):
+        calls.append({"needs_fresh": sess.needs_fresh_session})
+        # Zero compact boundaries on purpose: an overflow can hit before autocompact ever fired.
+        if len(calls) == 1:
+            raise TurnResultError("Prompt is too long")
+
+    p_install_run_fakes(monkeypatch, fake_run_turn)
+    asyncio.run(agent_manager.run_agent_loop(session.id, "hello"))
+
+    assert len(calls) == 2
+    assert calls[1]["needs_fresh"] is True
+    assert session.compacted_through_msg_id is not None
+    assert session.status == "completed"
+    assert not [m for m in session.messages if m.role == "system" and str(m.content).startswith("Error:")]
+
+
+def test_overflow_on_retry_surfaces_the_card_not_a_fake_completed(monkeypatch) -> None:
+    session = p_seed_session()
+    calls: list = []
+
+    async def fake_run_turn(sess, session_id, prompt_content, options, options_kwargs,
+                            turn, thinking, stderr, resolved_model, api_type,
+                            global_settings, force_respawn=False):
+        calls.append(1)
+        # Fake mid-task streaming: the old handle_run_error early-return keyed on exactly this and marked the dead run "completed".
+        turn.stream_text_msg_id = "msg-1"
+        turn.current_turn_emitted = True
+        raise Exception("Error code: 429 - extra usage is required for long context")
+
+    p_install_run_fakes(monkeypatch, fake_run_turn)
+    asyncio.run(agent_manager.run_agent_loop(session.id, "hello"))
+
+    assert len(calls) == 2
+    assert session.status == "error"
+    assert [m for m in session.messages if m.role == "system" and "context window" in str(m.content)]
+
+
 def test_valve_never_loops(monkeypatch) -> None:
     session = p_seed_session()
     calls: list = []

backend/tests/test_error_classify.py

@@ -13,11 +13,12 @@ from backend.apps.agents.core.error_classify import (
     capacity_retry_wait,
     is_auth_error,
     is_cli_binary_missing,
+    is_context_overflow_error,
     is_free_trial_exhausted,
     is_transient_capacity_error,
     is_unknown_model_error,
-    redact_for_telemetry,
 )
+from backend.apps.agents.core.redact_for_telemetry import redact_for_telemetry
 from backend.apps.agents.core.first_real_exception import first_real_exception
 
 # Verbatim field strings from prod analytics (2026-07): the exact shapes users hit.
@@ -89,6 +90,31 @@ def test_cli_missing_matches_sdk_exception_type():
     assert is_cli_binary_missing(CLINotFoundError("whatever text"))
 
 
+# The overflow family across providers; each of these shapes used to kill the run with either a raw error card or (worse) a fake "completed".
+P_OVERFLOW_SHAPES = (
+    "API Error: 400 {\"type\":\"error\",\"error\":{\"type\":\"invalid_request_error\",\"message\":\"prompt is too long: 214384 tokens > 200000 maximum\"}}",
+    "Error code: 429 - extra usage is required for long context",
+    "This model's maximum context length is 128000 tokens. However, your messages resulted in 131074 tokens.",
+    "Error code: 400 - {'error': {'code': 'context_length_exceeded'}}",
+    "The input token count (1048577) exceeds the maximum number of tokens allowed (1048576).",
+    "Error code: 429 - Request too large for gpt-4o on tokens per min (TPM)",
+)
+
+
+def test_overflow_family_is_claimed_and_never_retried_verbatim():
+    for s in P_OVERFLOW_SHAPES:
+        e = Exception(s)
+        assert is_context_overflow_error(e), s
+        # Retrying the identical oversized request is guaranteed futile; the valve owns it.
+        assert not is_transient_capacity_error(e), s
+        assert capacity_retry_wait(e, 0) is None, s
+
+
+def test_overflow_does_not_claim_ordinary_errors():
+    for s in (P_FIELD_POOL_BUSY, P_FIELD_CLI_MISSING, "529 overloaded, try again shortly", "401 invalid x-api-key"):
+        assert not is_context_overflow_error(Exception(s)), s
+
+
 def test_first_real_exception_unwraps_nested_groups():
     boom = ValueError("boom")
     group = BaseExceptionGroup(