mirror of
https://github.com/openswarm-ai/openswarm.git
synced 2026-08-17 18:25:42 +02:00
201 lines
8.8 KiB
Python
201 lines
8.8 KiB
Python
"""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_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 = []
|
|
|
|
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:")]
|
|
|
|
|
|
def test_midturn_break_completes_and_fires_the_hidden_continuation(monkeypatch) -> None:
|
|
from backend.apps.agents.manager.context_budget import CONTINUATION_PROMPT
|
|
session = p_seed_session()
|
|
continues: 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):
|
|
# Simulate maybe_break_midturn firing inside the stream loop: flags set, turn returns at the boundary.
|
|
turn.context_break_fired = True
|
|
sess.needs_fresh_session = True
|
|
sess.pending_continuation = True
|
|
sess.pending_continuation_prompt = CONTINUATION_PROMPT
|
|
|
|
async def fake_send_message(session_id, prompt, hidden=False, **kwargs):
|
|
continues.append({"prompt": prompt, "hidden": hidden})
|
|
|
|
p_install_run_fakes(monkeypatch, fake_run_turn)
|
|
monkeypatch.setattr(agent_manager, "send_message", fake_send_message)
|
|
|
|
async def main():
|
|
await agent_manager.run_agent_loop(session.id, "audit everything")
|
|
await asyncio.sleep(0)
|
|
|
|
asyncio.run(main())
|
|
assert session.status == "completed"
|
|
assert session.needs_fresh_session is True
|
|
assert session.pending_continuation is False
|
|
assert continues == [{"prompt": CONTINUATION_PROMPT, "hidden": True}]
|