[eric] agents: context-pressure valve, one fresh-recap retry when the CLI dies mid-compaction-churn

backend/apps/agents/agent_manager.py

@@ -51,7 +51,7 @@ class AgentManager(SessionLifecycle, SessionPersistence, Messaging, SessionContr
 
 
     @typechecked
-    async def run_agent_loop(self, session_id: str, prompt: str, images: Optional[List] = None, context_paths: Optional[List] = None, forced_tools: Optional[List[str]] = None, attached_skills: Optional[List] = None, fork_session: bool = False, selected_browser_ids: Optional[List[str]] = None, selected_app_output_ids: Optional[List[str]] = None, selected_setting_ids: Optional[List[str]] = None):
+    async def run_agent_loop(self, session_id: str, prompt: str, images: Optional[List] = None, context_paths: Optional[List] = None, forced_tools: Optional[List[str]] = None, attached_skills: Optional[List] = None, fork_session: bool = False, selected_browser_ids: Optional[List[str]] = None, selected_app_output_ids: Optional[List[str]] = None, selected_setting_ids: Optional[List[str]] = None, context_valve_retry: bool = False):
         """Run the Claude Agent SDK query loop for a session."""
         session = self.sessions.get(session_id)
         if not session:
@@ -131,6 +131,45 @@ 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
+            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.
+                logger.warning(
+                    f"Agent {session_id}: context-pressure death after "
+                    f"{turn.compact_boundaries} compact boundaries; one fresh-session recap retry"
+                )
+                session.needs_fresh_session = True
+                if turn.stream_text_msg_id:
+                    await ws_manager.send_to_session(session_id, "agent:stream_end", {
+                        "session_id": session_id,
+                        "message_id": turn.stream_text_msg_id,
+                    })
+                for p_tool_msg_id in turn.stream_tool_msg_ids_ordered:
+                    await ws_manager.send_to_session(session_id, "agent:stream_end", {
+                        "session_id": session_id,
+                        "message_id": p_tool_msg_id,
+                    })
+                self.live_partial.pop(session_id, None)
+                try:
+                    from backend.apps.service.client import submit_diagnostic
+                    from backend.apps.agents.core.error_classify import redact_for_telemetry
+                    submit_diagnostic({
+                        "kind": "context_pressure_valve",
+                        "session_id": session_id,
+                        "model": session.model,
+                        "compact_boundaries": turn.compact_boundaries,
+                        "error_preview": redact_for_telemetry(str(e), limit=300),
+                    })
+                except Exception:
+                    logger.debug("submit_diagnostic context_pressure_valve failed", exc_info=True)
+                await self.run_agent_loop(
+                    session_id, prompt, images, context_paths, forced_tools,
+                    attached_skills, fork_session, selected_browser_ids,
+                    selected_app_output_ids, selected_setting_ids,
+                    context_valve_retry=True,
+                )
+                return
             await handle_run_error(e, session, session_id, turn, p_stderr_buffer)
         except BaseException as e:
             # Catch BaseExceptionGroup from anyio task groups (e.g. concurrent CLI crash + pending approval cancellation) so it doesn't escape and kill the uvicorn process.

backend/apps/agents/core/error_classify.py

@@ -232,6 +232,29 @@ def is_out_of_tokens(exc: BaseException, extra_text: str = "") -> bool:
     ))
 
 
+@typechecked
+def is_context_pressure_death(exc: BaseException, compact_boundaries: int, extra_text: str = "") -> bool:
+    """The CLI autocompact-thrash class: the process compacted during this turn and then
+    died with a bare exit-1 ProcessError (its thrash detector gives up after 3 refill
+    cycles, which can straddle turns on a persistent client, so one boundary in the dying
+    turn is the reliable tell). Only claims deaths no other classifier owns, so auth/
+    capacity/credit errors keep their specific handling; a misfire costs one bounded
+    silent retry, a miss just means today's error card.
+    """
+    if compact_boundaries < 1:
+        return False
+    # Type-name check, not isinstance: the SDK is lazy-imported (mock mode must work without it), mirroring the client-pool dead-client idiom.
+    if "ProcessError" not in type(exc).__name__:
+        return False
+    for p_claimed_by in (
+        is_long_context_error, is_transient_capacity_error, is_free_trial_exhausted,
+        is_out_of_tokens, is_auth_error, is_unknown_model_error,
+    ):
+        if p_claimed_by(exc, extra_text=extra_text):
+            return False
+    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

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

@@ -99,6 +99,8 @@ class TurnRunner(AgentManagerProtocol):
                 if isinstance(message, SystemMessage):
                     raw = message.__dict__ if hasattr(message, '__dict__') else str(message)
                     logger.info(f"[MCP-DEBUG] SystemMessage: {raw}")
+                    if getattr(message, "subtype", "") == "compact_boundary":
+                        turn.compact_boundaries += 1
 
                 if isinstance(message, StreamEvent):
                     await handle_stream_event(

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

@@ -54,3 +54,5 @@ class TurnState(BaseModel):
     baseline_children_in: int = 0
     baseline_children_out: int = 0
     baseline_captured: bool = False
+    # CLI compact_boundary events seen this turn; one plus a ProcessError = the autocompact-thrash death the context-pressure valve retries.
+    compact_boundaries: int = 0

backend/tests/test_context_pressure_valve.py

@@ -0,0 +1,121 @@
+"""Context-pressure valve invariant.
+
+The bug class (1.5.4 field reports): an oversized/incompressible context makes
+the CLI's autocompact churn until its own thrash detector gives up and the
+process dies with a bare exit-1 ProcessError; the user got a cryptic error card
+and had to type "continue".
+
+The seal: run_agent_loop detects that death shape structurally (2+ CLI
+compact_boundary events this turn + a ProcessError no other classifier claims)
+and transparently re-runs the turn ONCE through the proven fresh-session recap
+path. Anything else keeps today's error handling, and the retry can never loop.
+"""
+
+import asyncio
+
+from backend.apps.agents.agent_manager import agent_manager
+import backend.apps.agents.agent_manager as agent_manager_module
+from backend.apps.agents.core.error_classify import is_context_pressure_death
+from backend.apps.agents.core.models import AgentSession
+
+
+class ProcessError(Exception):
+    pass
+
+
+def test_predicate_claims_thrash_death() -> None:
+    e = ProcessError("Command failed with exit code 1 (exit code: 1)\nError output: Check stderr output for details")
+    assert is_context_pressure_death(e, 1) is True
+    assert is_context_pressure_death(e, 3) is True
+
+
+def test_predicate_needs_compaction_this_turn() -> None:
+    e = ProcessError("Command failed with exit code 1")
+    assert is_context_pressure_death(e, 0) is False
+
+
+def test_predicate_needs_a_process_death() -> None:
+    assert is_context_pressure_death(ValueError("Command failed with exit code 1"), 3) is False
+
+
+def test_predicate_defers_to_specific_classifiers() -> None:
+    assert is_context_pressure_death(ProcessError("529 overloaded, try again shortly"), 3) is False
+    assert is_context_pressure_death(ProcessError("credit balance is too low"), 3) is False
+    assert is_context_pressure_death(ProcessError("Command failed with exit code 1"), 3, extra_text="401 authentication_error: invalid x-api-key") is False
+
+
+def p_seed_session() -> AgentSession:
+    session = AgentSession(name="t", model="sonnet", dashboard_id="d")
+    agent_manager.sessions[session.id] = session
+    return session
+
+
+def p_install_run_fakes(monkeypatch, run_turn_fake) -> None:
+    async def fake_build(session, session_id, prompt, prompt_content, builtin_perms,
+                         selected_browser_ids, selected_app_output_ids, selected_setting_ids,
+                         fork_session, router_model_id, api_type):
+        from backend.apps.settings.settings import load_settings
+        return object(), {}, prompt_content, [], load_settings()
+
+    monkeypatch.setattr(agent_manager, "build_agent_options", fake_build)
+    monkeypatch.setattr(agent_manager, "run_turn_with_retry", run_turn_fake)
+    monkeypatch.setattr(agent_manager_module, "save_session", lambda sid, data: None)
+
+
+def test_valve_retries_once_through_fresh_path(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({"force_respawn": force_respawn, "needs_fresh": sess.needs_fresh_session})
+        if len(calls) == 1:
+            turn.compact_boundaries = 3
+            raise ProcessError("Command failed with exit code 1 (exit code: 1)")
+
+    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]["force_respawn"] is True
+    assert calls[1]["needs_fresh"] is True
+    assert session.status == "completed"
+    assert not [m for m in session.messages if m.role == "system" and str(m.content).startswith("Error:")]
+
+
+def test_no_valve_without_compaction_churn(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)
+        raise ProcessError("Command failed with exit code 1 (exit code: 1)")
+
+    p_install_run_fakes(monkeypatch, fake_run_turn)
+    asyncio.run(agent_manager.run_agent_loop(session.id, "hello"))
+
+    assert len(calls) == 1
+    assert session.status == "error"
+    assert [m for m in session.messages if m.role == "system" and str(m.content).startswith("Error:")]
+
+
+def test_valve_never_loops(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)
+        turn.compact_boundaries = 3
+        raise ProcessError("Command failed with exit code 1 (exit code: 1)")
+
+    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 str(m.content).startswith("Error:")]