From f787414fc6a66c768f18cd6a5e2b359d465b611b Mon Sep 17 00:00:00 2001 From: ciregenz Date: Sun, 5 Jul 2026 21:40:16 -0700 Subject: [PATCH] [eric] agents: context-pressure valve, one fresh-recap retry when the CLI dies mid-compaction-churn --- backend/apps/agents/agent_manager.py | 41 +++++- backend/apps/agents/core/error_classify.py | 23 ++++ backend/apps/agents/manager/run/TurnRunner.py | 2 + .../apps/agents/manager/streaming/state.py | 2 + backend/tests/test_context_pressure_valve.py | 121 ++++++++++++++++++ 5 files changed, 188 insertions(+), 1 deletion(-) create mode 100644 backend/tests/test_context_pressure_valve.py diff --git a/backend/apps/agents/agent_manager.py b/backend/apps/agents/agent_manager.py index 2277b7a9..6f642ac5 100644 --- a/backend/apps/agents/agent_manager.py +++ b/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. diff --git a/backend/apps/agents/core/error_classify.py b/backend/apps/agents/core/error_classify.py index 2e262e68..2f683d21 100644 --- a/backend/apps/agents/core/error_classify.py +++ b/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 diff --git a/backend/apps/agents/manager/run/TurnRunner.py b/backend/apps/agents/manager/run/TurnRunner.py index 4af66a69..6acecd96 100644 --- a/backend/apps/agents/manager/run/TurnRunner.py +++ b/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( diff --git a/backend/apps/agents/manager/streaming/state.py b/backend/apps/agents/manager/streaming/state.py index d2545985..f7537cc0 100644 --- a/backend/apps/agents/manager/streaming/state.py +++ b/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 diff --git a/backend/tests/test_context_pressure_valve.py b/backend/tests/test_context_pressure_valve.py new file mode 100644 index 00000000..8f80d2b9 --- /dev/null +++ b/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:")]