from __future__ import annotations import asyncio import json import os import subprocess from collections.abc import AsyncIterator from types import SimpleNamespace from typing import Any import pytest from opensquilla.engine import ( Agent, AgentConfig, AgentState, DoneEvent, ErrorEvent, RunHeartbeatEvent, SubagentSpec, ToolCall, ToolResult, WarningEvent, ) from opensquilla.engine.agent import _progress_watchdog_guidance_message from opensquilla.engine.runtime import TurnRunner from opensquilla.engine.session_sanitize import session_payload_chars from opensquilla.provider import ( ChatConfig, ContentBlockToolResult, ContentBlockToolUse, Message, ProviderHeartbeatEvent, ToolDefinition, ToolInputSchema, ) from opensquilla.provider import DoneEvent as ProviderDone from opensquilla.provider import ErrorEvent as ProviderError from opensquilla.provider import TextDeltaEvent as ProviderText from opensquilla.provider import ToolUseDeltaEvent as ProviderToolUseDelta from opensquilla.provider import ToolUseEndEvent as ProviderToolUseEnd from opensquilla.provider import ToolUseStartEvent as ProviderToolUseStart from opensquilla.provider.request_proof import ( ProviderRequestBudgetExceeded, prove_provider_payload, ) from opensquilla.session.compaction import CompactionResult from opensquilla.tools.mutation_receipts import ( fingerprint_path, record_semantic_mutation_receipt, ) from opensquilla.tools.types import ToolContext RAW_CURRENT_TURN_OVERFLOW_MESSAGE = ( "Context overflow is in the current turn's recent tool calls or " "reasoning tail; history compaction cannot reduce it." ) class _StallingProvider: provider_name = "fake" def __init__(self) -> None: self.calls: list[list[Message]] = [] self.stream_closed = False def chat( self, messages: list[Message], tools: list[Any] | None = None, config: ChatConfig | None = None, ) -> AsyncIterator[Any]: self.calls.append(messages) return self._stream() async def _stream(self) -> AsyncIterator[Any]: try: await asyncio.sleep(60.0) yield ProviderText(text="late") finally: self.stream_closed = True async def list_models(self) -> list[Any]: return [] class _ActiveLongToolArgumentProvider: provider_name = "fake" def __init__( self, *, fragment_delay: float = 0.02, content: str = "alpha\\nbeta\\ngamma\\n", ) -> None: self.fragment_delay = fragment_delay self.content = content self.calls: list[list[Message]] = [] def chat( self, messages: list[Message], tools: list[Any] | None = None, config: ChatConfig | None = None, ) -> AsyncIterator[Any]: self.calls.append(messages) return self._stream(len(self.calls)) async def _stream(self, call_number: int) -> AsyncIterator[Any]: if call_number > 1: yield ProviderText(text="done") yield ProviderDone(stop_reason="stop", input_tokens=1, output_tokens=1) return tool_use_id = "tool-1" yield ProviderToolUseStart(tool_use_id=tool_use_id, tool_name="write_file") fragments = [ '{"path":"deck.py","content":"', self.content, '"}', ] for fragment in fragments: await asyncio.sleep(self.fragment_delay) yield ProviderToolUseDelta(tool_use_id=tool_use_id, json_fragment=fragment) yield ProviderToolUseEnd( tool_use_id=tool_use_id, tool_name="write_file", arguments={}, ) yield ProviderDone(stop_reason="tool_calls", input_tokens=1, output_tokens=100) async def list_models(self) -> list[Any]: return [] class _ContextOverflowProvider: provider_name = "fake" def __init__(self, *, success_after: int | None = None) -> None: self.success_after = success_after self.calls: list[list[Message]] = [] def chat( self, messages: list[Message], tools: list[Any] | None = None, config: ChatConfig | None = None, ) -> AsyncIterator[Any]: self.calls.append(messages) return self._stream(len(self.calls)) async def _stream(self, call_number: int) -> AsyncIterator[Any]: if self.success_after is not None and call_number > self.success_after: yield ProviderText(text="ok") yield ProviderDone(stop_reason="stop", input_tokens=1, output_tokens=1) return yield ProviderError(message="context length exceeded", code="400") async def list_models(self) -> list[Any]: return [] class _ProviderRequestBudgetExceededProvider: provider_name = "openrouter" def __init__( self, *, success_after: int | None = None, proof: dict[str, Any] | None = None, ) -> None: self.success_after = success_after self.proof = proof self.calls: list[list[Message]] = [] def chat( self, messages: list[Message], tools: list[Any] | None = None, config: ChatConfig | None = None, ) -> AsyncIterator[Any]: self.calls.append(messages) return self._stream(len(self.calls)) async def _stream(self, call_number: int) -> AsyncIterator[Any]: if self.success_after is not None and call_number > self.success_after: yield ProviderText(text="ok") yield ProviderDone(stop_reason="stop", input_tokens=1, output_tokens=1) return message = ( '{"fallback_reason":"provider_request_budget_exhausted"}' if self.proof is None else json.dumps(self.proof) ) yield ProviderError(message=message, code="provider_request_budget_exhausted") async def list_models(self) -> list[Any]: return [] class _RepeatedToolFailureThenDoneProvider: provider_name = "fake" def __init__(self, *, tool_retries: int = 3) -> None: self.tool_retries = tool_retries self.calls: list[list[Message]] = [] def chat( self, messages: list[Message], tools: list[Any] | None = None, config: ChatConfig | None = None, ) -> AsyncIterator[Any]: self.calls.append(messages) return self._stream(len(self.calls)) async def _stream(self, call_number: int) -> AsyncIterator[Any]: if call_number > self.tool_retries: yield ProviderText(text="handled") yield ProviderDone(stop_reason="stop", input_tokens=1, output_tokens=1) return tool_use_id = f"cmd-{call_number}" yield ProviderToolUseStart(tool_use_id=tool_use_id, tool_name="exec_command") yield ProviderToolUseEnd( tool_use_id=tool_use_id, tool_name="exec_command", arguments={"command": "python build_pptx.py", "timeout": 30}, ) yield ProviderDone(stop_reason="tool_calls", input_tokens=1, output_tokens=1) async def list_models(self) -> list[Any]: return [] class _RepeatedSuccessfulToolThenDoneProvider: provider_name = "fake" def __init__( self, *, tool_retries: int = 4, tool_name: str = "grep_search", arguments: dict[str, Any] | None = None, ) -> None: self.tool_retries = tool_retries self.tool_name = tool_name self.calls: list[list[Message]] = [] self.arguments = arguments or { "path": "/testbed/crates/regex/src/matcher.rs", "pattern": 'impl.*Matcher"', } def chat( self, messages: list[Message], tools: list[Any] | None = None, config: ChatConfig | None = None, ) -> AsyncIterator[Any]: self.calls.append(messages) return self._stream(len(self.calls)) async def _stream(self, call_number: int) -> AsyncIterator[Any]: if call_number > self.tool_retries: yield ProviderText(text="handled") yield ProviderDone(stop_reason="stop", input_tokens=1, output_tokens=1) return tool_use_id = f"{self.tool_name}-{call_number}" yield ProviderToolUseStart(tool_use_id=tool_use_id, tool_name=self.tool_name) yield ProviderToolUseEnd( tool_use_id=tool_use_id, tool_name=self.tool_name, arguments=dict(self.arguments), ) yield ProviderDone(stop_reason="tool_calls", input_tokens=1, output_tokens=1) async def list_models(self) -> list[Any]: return [] class _FinalThenDoneProvider: provider_name = "fake" def __init__(self) -> None: self.calls: list[list[Message]] = [] def chat( self, messages: list[Message], tools: list[Any] | None = None, config: ChatConfig | None = None, ) -> AsyncIterator[Any]: self.calls.append(messages) return self._stream(len(self.calls)) async def _stream(self, call_number: int) -> AsyncIterator[Any]: if call_number == 1: yield ProviderText(text="Implemented the fix.") else: yield ProviderText(text="No code change is required.") yield ProviderDone(stop_reason="stop", input_tokens=1, output_tokens=1) async def list_models(self) -> list[Any]: return [] class _FailedToolThenFinalProvider: provider_name = "fake" def __init__(self) -> None: self.calls: list[list[Message]] = [] def chat( self, messages: list[Message], tools: list[Any] | None = None, config: ChatConfig | None = None, ) -> AsyncIterator[Any]: self.calls.append(messages) return self._stream(len(self.calls)) async def _stream(self, call_number: int) -> AsyncIterator[Any]: if call_number == 1: tool_use_id = "cmd-1" yield ProviderToolUseStart(tool_use_id=tool_use_id, tool_name="exec_command") yield ProviderToolUseEnd( tool_use_id=tool_use_id, tool_name="exec_command", arguments={"command": "cargo build 2>&1 | tail -30"}, ) yield ProviderDone(stop_reason="tool_calls", input_tokens=1, output_tokens=1) return yield ProviderText(text=f"final attempt {call_number}") yield ProviderDone(stop_reason="stop", input_tokens=1, output_tokens=1) async def list_models(self) -> list[Any]: return [] class _PostWriteFailedVerificationThenSourceProvider: provider_name = "fake" def __init__(self) -> None: self.calls: list[list[Message]] = [] def chat( self, messages: list[Message], tools: list[Any] | None = None, config: ChatConfig | None = None, ) -> AsyncIterator[Any]: self.calls.append(messages) return self._stream(len(self.calls)) async def _stream(self, call_number: int) -> AsyncIterator[Any]: if call_number == 1: tool_use_id = "edit-1" yield ProviderToolUseStart(tool_use_id=tool_use_id, tool_name="edit_file") yield ProviderToolUseEnd( tool_use_id=tool_use_id, tool_name="edit_file", arguments={"path": "src.py", "old_text": "old", "new_text": "new"}, ) yield ProviderDone(stop_reason="tool_calls", input_tokens=1, output_tokens=1) return if call_number == 2: tool_use_id = "cmd-1" yield ProviderToolUseStart(tool_use_id=tool_use_id, tool_name="exec_command") yield ProviderToolUseEnd( tool_use_id=tool_use_id, tool_name="exec_command", arguments={"command": "cargo build --release --bin ruff 2>&1 | tail -30"}, ) yield ProviderDone(stop_reason="tool_calls", input_tokens=1, output_tokens=1) return if 3 <= call_number <= 5: tool_use_id = f"read-{call_number}" yield ProviderToolUseStart(tool_use_id=tool_use_id, tool_name="read_file") yield ProviderToolUseEnd( tool_use_id=tool_use_id, tool_name="read_file", arguments={"path": "src.py"}, ) yield ProviderDone(stop_reason="tool_calls", input_tokens=1, output_tokens=1) return yield ProviderText(text="done") yield ProviderDone(stop_reason="stop", input_tokens=1, output_tokens=1) async def list_models(self) -> list[Any]: return [] class _StableVerifiedDiffThenSourceProvider: provider_name = "fake" def __init__(self) -> None: self.calls: list[list[Message]] = [] self.tool_lists: list[list[Any] | None] = [] def chat( self, messages: list[Message], tools: list[Any] | None = None, config: ChatConfig | None = None, ) -> AsyncIterator[Any]: self.calls.append(messages) self.tool_lists.append(tools) return self._stream(len(self.calls)) async def _stream(self, call_number: int) -> AsyncIterator[Any]: if call_number == 1: tool_use_id = "edit-1" yield ProviderToolUseStart(tool_use_id=tool_use_id, tool_name="edit_file") yield ProviderToolUseEnd( tool_use_id=tool_use_id, tool_name="edit_file", arguments={"path": "src.py", "old_text": "old", "new_text": "new"}, ) yield ProviderDone(stop_reason="tool_calls", input_tokens=1, output_tokens=1) return if call_number == 2: tool_use_id = "cmd-1" yield ProviderToolUseStart(tool_use_id=tool_use_id, tool_name="exec_command") yield ProviderToolUseEnd( tool_use_id=tool_use_id, tool_name="exec_command", arguments={"command": "pytest tests/test_src.py"}, ) yield ProviderDone(stop_reason="tool_calls", input_tokens=1, output_tokens=1) return if 3 <= call_number <= 8: tool_use_id = f"read-{call_number}" yield ProviderToolUseStart(tool_use_id=tool_use_id, tool_name="read_file") yield ProviderToolUseEnd( tool_use_id=tool_use_id, tool_name="read_file", arguments={"path": "src.py"}, ) yield ProviderDone(stop_reason="tool_calls", input_tokens=1, output_tokens=1) return yield ProviderText(text=f"final after convergence {call_number}") yield ProviderDone(stop_reason="stop", input_tokens=1, output_tokens=1) async def list_models(self) -> list[Any]: return [] class _RepeatedFailedVerificationFinalProvider: provider_name = "fake" def __init__(self) -> None: self.calls: list[list[Message]] = [] def chat( self, messages: list[Message], tools: list[Any] | None = None, config: ChatConfig | None = None, ) -> AsyncIterator[Any]: self.calls.append(messages) return self._stream(len(self.calls)) async def _stream(self, call_number: int) -> AsyncIterator[Any]: if call_number == 1: tool_use_id = "edit-1" yield ProviderToolUseStart(tool_use_id=tool_use_id, tool_name="edit_file") yield ProviderToolUseEnd( tool_use_id=tool_use_id, tool_name="edit_file", arguments={"path": "src.py", "old_text": "old", "new_text": "new"}, ) yield ProviderDone(stop_reason="tool_calls", input_tokens=1, output_tokens=1) return if call_number in {2, 4}: tool_use_id = f"cmd-{call_number}" yield ProviderToolUseStart(tool_use_id=tool_use_id, tool_name="exec_command") yield ProviderToolUseEnd( tool_use_id=tool_use_id, tool_name="exec_command", arguments={"command": "make check"}, ) yield ProviderDone(stop_reason="tool_calls", input_tokens=1, output_tokens=1) return yield ProviderText(text=f"final attempt {call_number}") yield ProviderDone(stop_reason="stop", input_tokens=1, output_tokens=1) async def list_models(self) -> list[Any]: return [] class _PostWriteCleanMavenVerificationThenFinalProvider: provider_name = "fake" def __init__(self) -> None: self.calls: list[list[Message]] = [] def chat( self, messages: list[Message], tools: list[Any] | None = None, config: ChatConfig | None = None, ) -> AsyncIterator[Any]: self.calls.append(messages) return self._stream(len(self.calls)) async def _stream(self, call_number: int) -> AsyncIterator[Any]: if call_number == 1: tool_use_id = "edit-1" yield ProviderToolUseStart(tool_use_id=tool_use_id, tool_name="edit_file") yield ProviderToolUseEnd( tool_use_id=tool_use_id, tool_name="edit_file", arguments={"path": "src.py", "old_text": "old", "new_text": "new"}, ) yield ProviderDone(stop_reason="tool_calls", input_tokens=1, output_tokens=1) return if call_number == 2: tool_use_id = "cmd-2" yield ProviderToolUseStart(tool_use_id=tool_use_id, tool_name="exec_command") yield ProviderToolUseEnd( tool_use_id=tool_use_id, tool_name="exec_command", arguments={"command": "mvn test -Dtest=ParserTest 2>&1 | tail -20"}, ) yield ProviderDone(stop_reason="tool_calls", input_tokens=1, output_tokens=1) return yield ProviderText(text=f"final attempt {call_number}") yield ProviderDone(stop_reason="stop", input_tokens=1, output_tokens=1) async def list_models(self) -> list[Any]: return [] class _EditThenFinalProvider: provider_name = "fake" def __init__(self) -> None: self.calls: list[list[Message]] = [] def chat( self, messages: list[Message], tools: list[Any] | None = None, config: ChatConfig | None = None, ) -> AsyncIterator[Any]: self.calls.append(messages) return self._stream(len(self.calls)) async def _stream(self, call_number: int) -> AsyncIterator[Any]: if call_number == 1: tool_use_id = "edit-1" yield ProviderToolUseStart(tool_use_id=tool_use_id, tool_name="edit_file") yield ProviderToolUseEnd( tool_use_id=tool_use_id, tool_name="edit_file", arguments={"path": "src.py", "old_text": "old", "new_text": "new"}, ) yield ProviderDone(stop_reason="tool_calls", input_tokens=1, output_tokens=1) return yield ProviderText(text=f"final attempt {call_number}") yield ProviderDone(stop_reason="stop", input_tokens=1, output_tokens=1) async def list_models(self) -> list[Any]: return [] class _NoWorkspaceWriteThenPatchProvider: provider_name = "fake" def __init__(self) -> None: self.calls: list[list[Message]] = [] self.tools_by_call: list[list[Any] | None] = [] self.configs: list[ChatConfig | None] = [] def chat( self, messages: list[Message], tools: list[Any] | None = None, config: ChatConfig | None = None, ) -> AsyncIterator[Any]: self.calls.append(messages) self.tools_by_call.append(tools) self.configs.append(config) return self._stream(len(self.calls)) async def _stream(self, call_number: int) -> AsyncIterator[Any]: if call_number <= 17: tool_use_id = f"read-{call_number}" yield ProviderToolUseStart(tool_use_id=tool_use_id, tool_name="read_file") yield ProviderToolUseEnd( tool_use_id=tool_use_id, tool_name="read_file", arguments={"path": "src.py"}, ) yield ProviderDone(stop_reason="tool_calls", input_tokens=1, output_tokens=1) return if call_number == 18: tool_use_id = "patch-1" yield ProviderToolUseStart(tool_use_id=tool_use_id, tool_name="apply_patch") yield ProviderToolUseEnd( tool_use_id=tool_use_id, tool_name="apply_patch", arguments={ "patch": ( "*** Begin Patch\n" "*** Update File: src.py\n" "@@\n" "-old\n" "+new\n" "*** End Patch\n" ) }, ) yield ProviderDone(stop_reason="tool_calls", input_tokens=1, output_tokens=1) return yield ProviderText(text="done") yield ProviderDone(stop_reason="stop", input_tokens=1, output_tokens=1) async def list_models(self) -> list[Any]: return [] class _PatchFailureRecoveryProvider(_NoWorkspaceWriteThenPatchProvider): async def _stream(self, call_number: int) -> AsyncIterator[Any]: if call_number <= 17: tool_use_id = f"read-{call_number}" yield ProviderToolUseStart(tool_use_id=tool_use_id, tool_name="read_file") yield ProviderToolUseEnd( tool_use_id=tool_use_id, tool_name="read_file", arguments={"path": "src.py"}, ) yield ProviderDone(stop_reason="tool_calls", input_tokens=1, output_tokens=1) return if call_number in {18, 20}: tool_use_id = f"patch-{call_number}" yield ProviderToolUseStart(tool_use_id=tool_use_id, tool_name="apply_patch") yield ProviderToolUseEnd( tool_use_id=tool_use_id, tool_name="apply_patch", arguments={ "patch": ( "*** Begin Patch\n" "*** Update File: src.py\n" "@@\n" "-old\n" "+new\n" "*** End Patch\n" ) }, ) yield ProviderDone(stop_reason="tool_calls", input_tokens=1, output_tokens=1) return if call_number == 19: tool_use_id = "read-recovery" yield ProviderToolUseStart(tool_use_id=tool_use_id, tool_name="read_file") yield ProviderToolUseEnd( tool_use_id=tool_use_id, tool_name="read_file", arguments={"path": "src.py"}, ) yield ProviderDone(stop_reason="tool_calls", input_tokens=1, output_tokens=1) return yield ProviderText(text="done") yield ProviderDone(stop_reason="stop", input_tokens=1, output_tokens=1) class _EditFailureRecoveryProvider(_NoWorkspaceWriteThenPatchProvider): async def _stream(self, call_number: int) -> AsyncIterator[Any]: if call_number <= 17: tool_use_id = f"read-{call_number}" yield ProviderToolUseStart(tool_use_id=tool_use_id, tool_name="read_file") yield ProviderToolUseEnd( tool_use_id=tool_use_id, tool_name="read_file", arguments={"path": "src.py"}, ) yield ProviderDone(stop_reason="tool_calls", input_tokens=1, output_tokens=1) return if call_number == 18: tool_use_id = "edit-missing-context" yield ProviderToolUseStart(tool_use_id=tool_use_id, tool_name="edit_file") yield ProviderToolUseEnd( tool_use_id=tool_use_id, tool_name="edit_file", arguments={"path": "src.py", "old_text": "missing", "new_text": "new"}, ) yield ProviderDone(stop_reason="tool_calls", input_tokens=1, output_tokens=1) return if call_number == 19: tool_use_id = "read-recovery" yield ProviderToolUseStart(tool_use_id=tool_use_id, tool_name="read_file") yield ProviderToolUseEnd( tool_use_id=tool_use_id, tool_name="read_file", arguments={"path": "src.py"}, ) yield ProviderDone(stop_reason="tool_calls", input_tokens=1, output_tokens=1) return if call_number == 20: tool_use_id = "patch-after-read" yield ProviderToolUseStart(tool_use_id=tool_use_id, tool_name="apply_patch") yield ProviderToolUseEnd( tool_use_id=tool_use_id, tool_name="apply_patch", arguments={ "patch": ( "*** Begin Patch\n" "*** Update File: src.py\n" "@@\n" "-old\n" "+new\n" "*** End Patch\n" ) }, ) yield ProviderDone(stop_reason="tool_calls", input_tokens=1, output_tokens=1) return yield ProviderText(text="done") yield ProviderDone(stop_reason="stop", input_tokens=1, output_tokens=1) class _HighUsageToolLoopProvider: provider_name = "fake" def __init__(self, *, tool_rounds: int = 3, input_tokens_per_call: int = 4000) -> None: self.tool_rounds = tool_rounds self.input_tokens_per_call = input_tokens_per_call self.calls: list[list[Message]] = [] def chat( self, messages: list[Message], tools: list[Any] | None = None, config: ChatConfig | None = None, ) -> AsyncIterator[Any]: self.calls.append(messages) return self._stream(len(self.calls)) async def _stream(self, call_number: int) -> AsyncIterator[Any]: if call_number > self.tool_rounds: yield ProviderText(text="done") yield ProviderDone( stop_reason="stop", input_tokens=self.input_tokens_per_call, output_tokens=0, ) return tool_use_id = f"read-{call_number}" yield ProviderToolUseStart(tool_use_id=tool_use_id, tool_name="exec_command") yield ProviderToolUseEnd( tool_use_id=tool_use_id, tool_name="exec_command", arguments={"command": f"printf round-{call_number}"}, ) yield ProviderDone( stop_reason="tool_calls", input_tokens=self.input_tokens_per_call, output_tokens=0, ) async def list_models(self) -> list[Any]: return [] class _ConfigCapturingProvider: provider_name = "fake" def __init__(self) -> None: self.configs: list[ChatConfig | None] = [] def chat( self, messages: list[Message], tools: list[Any] | None = None, config: ChatConfig | None = None, ) -> AsyncIterator[Any]: self.configs.append(config) return self._stream() async def _stream(self) -> AsyncIterator[Any]: yield ProviderText(text="ok") yield ProviderDone(stop_reason="stop", input_tokens=1, output_tokens=1) async def list_models(self) -> list[Any]: return [] class _NoBilledCostUsageProvider: provider_name = "fake" def __init__(self, *, input_tokens: int = 1, output_tokens: int = 1) -> None: self.input_tokens = input_tokens self.output_tokens = output_tokens self.calls: list[list[Message]] = [] def chat( self, messages: list[Message], tools: list[Any] | None = None, config: ChatConfig | None = None, ) -> AsyncIterator[Any]: self.calls.append(messages) return self._stream() async def _stream(self) -> AsyncIterator[Any]: yield ProviderText(text="done") yield ProviderDone( stop_reason="stop", input_tokens=self.input_tokens, output_tokens=self.output_tokens, billed_cost=0.0, ) async def list_models(self) -> list[Any]: return [] class _MixedBilledAndEstimatedCostProvider: provider_name = "fake" def __init__(self) -> None: self.calls: list[list[Message]] = [] def chat( self, messages: list[Message], tools: list[Any] | None = None, config: ChatConfig | None = None, ) -> AsyncIterator[Any]: self.calls.append(messages) return self._stream(len(self.calls)) async def _stream(self, call_number: int) -> AsyncIterator[Any]: if call_number == 1: # First call reports a real billed cost, small enough to stay # under budget on its own. tool_use_id = "tool-1" yield ProviderToolUseStart(tool_use_id=tool_use_id, tool_name="exec_command") yield ProviderToolUseEnd( tool_use_id=tool_use_id, tool_name="exec_command", arguments={"command": "echo hi"}, ) yield ProviderDone( stop_reason="tool_calls", input_tokens=1, output_tokens=1, billed_cost=0.0005, ) return # Second call is cost-blind (billed_cost=0.0), forcing the estimator # to supply the remaining component that tips the turn over budget. yield ProviderText(text="done") yield ProviderDone( stop_reason="stop", input_tokens=1000, output_tokens=1000, billed_cost=0.0, ) async def list_models(self) -> list[Any]: return [] class _CompactingErrorSessionManager: def __init__(self, *, compact_raises: bool = False) -> None: self.calls: list[tuple[str, str]] = [] self.compact_raises = compact_raises async def compact(self, session_key: str, budget: int, config: Any | None = None) -> str: self.calls.append(("compact", session_key)) assert budget > 0 if self.compact_raises: raise RuntimeError("compact failed") return "[summary]" async def append_message(self, session_key: str, **kwargs: Any) -> None: self.calls.append(("append", session_key)) assert kwargs["role"] == "system" assert kwargs["content"].startswith("Error: ") @pytest.mark.asyncio async def test_turn_error_persist_records_current_turn_exhaustion_without_compacting() -> None: session_manager = _CompactingErrorSessionManager() runner = TurnRunner( provider_selector=None, session_manager=session_manager, config=SimpleNamespace(context_budget_tokens=96_000), ) await runner._persist_turn_error( "agent:main:webchat:test", ErrorEvent( message="Context overflow is in the current turn's recent tool calls.", code="current_turn_context_exhausted", ), ) assert session_manager.calls == [("append", "agent:main:webchat:test")] @pytest.mark.asyncio async def test_turn_error_persist_skips_error_time_compaction_for_exhaustion() -> None: session_manager = _CompactingErrorSessionManager(compact_raises=True) runner = TurnRunner( provider_selector=None, session_manager=session_manager, config=SimpleNamespace(context_budget_tokens=96_000), ) await runner._persist_turn_error( "agent:main:webchat:test", ErrorEvent( message="Context overflow is in the current turn's recent tool calls.", code="current_turn_context_exhausted", ), ) assert session_manager.calls == [("append", "agent:main:webchat:test")] @pytest.mark.asyncio async def test_agent_blocks_repeated_identical_tool_failures_before_tail_growth() -> None: calls = 0 async def _failing_tool(call: Any) -> ToolResult: nonlocal calls calls += 1 return ToolResult( tool_use_id=call.tool_use_id, tool_name=call.tool_name, content="write failed: " + ("permission denied " * 200), is_error=True, ) agent = Agent( provider=_ContextOverflowProvider(success_after=1), config=AgentConfig( tool_failure_loop_block_threshold=3, ), tool_handler=_failing_tool, ) tool_call = ToolCall( tool_use_id="write-1", tool_name="write_file", arguments={"path": "index.html", "content": "bad"}, ) first = await agent._execute_tool(tool_call) second = await agent._execute_tool(tool_call) third = await agent._execute_tool(tool_call) assert first.is_error is True assert second.is_error is True assert third.is_error is True assert calls == 2 assert "tool_failure_loop_exhausted" not in third.content assert "Do not retry this exact call unchanged" in third.content assert len(third.content) < len(second.content) assert third.execution_status is not None assert third.execution_status.get("reason") == "tool_failure_loop_exhausted" @pytest.mark.asyncio async def test_agent_tool_failure_loop_allows_changed_arguments() -> None: calls = 0 async def _failing_tool(call: Any) -> ToolResult: nonlocal calls calls += 1 return ToolResult( tool_use_id=call.tool_use_id, tool_name=call.tool_name, content="write failed", is_error=True, ) agent = Agent( provider=_ContextOverflowProvider(success_after=1), config=AgentConfig(tool_failure_loop_block_threshold=3), tool_handler=_failing_tool, ) await agent._execute_tool( ToolCall( tool_use_id="write-1", tool_name="write_file", arguments={"path": "index.html", "content": "first"}, ) ) await agent._execute_tool( ToolCall( tool_use_id="write-2", tool_name="write_file", arguments={"path": "index.html", "content": "first"}, ) ) changed = await agent._execute_tool( ToolCall( tool_use_id="write-3", tool_name="write_file", arguments={"path": "index.html", "content": "changed"}, ) ) assert calls == 3 assert changed.content == "write failed" @pytest.mark.asyncio async def test_agent_tool_failure_loop_result_returns_to_model_instead_of_terminal_error() -> None: provider = _RepeatedToolFailureThenDoneProvider(tool_retries=3) handler_calls = 0 async def _failing_tool(call: Any) -> ToolResult: nonlocal handler_calls handler_calls += 1 return ToolResult( tool_use_id=call.tool_use_id, tool_name=call.tool_name, content="syntax error", is_error=True, ) agent = Agent( provider=provider, config=AgentConfig( tool_failure_loop_block_threshold=3, max_iterations=5, flush_enabled=False, ), tool_handler=_failing_tool, ) events = [event async for event in agent.run_turn("build the deck")] assert handler_calls == 2 assert len(provider.calls) == 4 assert any(isinstance(event, DoneEvent) for event in events) assert not any( isinstance(event, ErrorEvent) and getattr(event, "code", None) == "tool_failure_loop_exhausted" for event in events ) assert any( getattr(event, "kind", None) == "tool_result" and (getattr(event, "execution_status", None) or {}).get("reason") == "tool_failure_loop_exhausted" for event in events ) assert not any( isinstance(event, WarningEvent) and event.code == "repeated_tool_call_recovery" for event in events ) @pytest.mark.asyncio async def test_agent_recovers_repeated_successful_identical_tool_calls( tmp_path, ) -> None: provider = _RepeatedSuccessfulToolThenDoneProvider(tool_retries=4) runtime_events_path = tmp_path / "runtime_events.jsonl" handler_calls = 0 async def _tool(call: Any) -> ToolResult: nonlocal handler_calls handler_calls += 1 return ToolResult( tool_use_id=call.tool_use_id, tool_name=call.tool_name, content="No matches", is_error=False, ) def _matching_tool_use_count(messages: list[Message]) -> int: count = 0 for message in messages: if not isinstance(message.content, list): continue for block in message.content: if ( getattr(block, "type", None) == "tool_use" and getattr(block, "name", None) == "grep_search" and getattr(block, "input", None) == provider.arguments ): count += 1 return count agent = Agent( provider=provider, config=AgentConfig( repeated_tool_call_recovery_threshold=3, max_iterations=8, flush_enabled=False, runtime_events_path=str(runtime_events_path), ), tool_handler=_tool, ) events = [event async for event in agent.run_turn("find the matcher impl")] assert any(isinstance(event, DoneEvent) for event in events) assert handler_calls == 2 assert len(provider.calls) == 5 assert _matching_tool_use_count(provider.calls[-1]) == 2 assert any( isinstance(event, WarningEvent) and event.code == "repeated_tool_call_recovery" for event in events ) logged = [json.loads(line) for line in runtime_events_path.read_text().splitlines()] recovery_events = [ event for event in logged if event.get("mechanism") == "repeated_tool_call_recovery" ] assert len(recovery_events) == 2 assert recovery_events[0]["evidence"]["repeat_count"] == 3 assert recovery_events[1]["evidence"]["repeat_count"] == 4 @pytest.mark.asyncio async def test_agent_recovers_repeated_successful_identical_exec_commands() -> None: provider = _RepeatedSuccessfulToolThenDoneProvider( tool_retries=4, tool_name="exec_command", arguments={ "command": ( "cd /testbed && printf 'some.domain.com/x\\n' | " "./target/release/rg --no-config -w domain 2>&1 || true" ) }, ) handler_calls = 0 async def _tool(call: Any) -> ToolResult: nonlocal handler_calls handler_calls += 1 return ToolResult( tool_use_id=call.tool_use_id, tool_name=call.tool_name, content="", is_error=False, ) def _matching_tool_use_count(messages: list[Message]) -> int: count = 0 for message in messages: if not isinstance(message.content, list): continue for block in message.content: if ( getattr(block, "type", None) == "tool_use" and getattr(block, "name", None) == "exec_command" and getattr(block, "input", None) == provider.arguments ): count += 1 return count agent = Agent( provider=provider, config=AgentConfig( repeated_tool_call_recovery_threshold=3, max_iterations=8, flush_enabled=False, ), tool_handler=_tool, ) events = [event async for event in agent.run_turn("verify the regex behavior")] assert any(isinstance(event, DoneEvent) for event in events) assert handler_calls == 2 assert len(provider.calls) == 5 assert _matching_tool_use_count(provider.calls[-1]) == 2 assert any( isinstance(event, WarningEvent) and event.code == "repeated_tool_call_recovery" for event in events ) @pytest.mark.asyncio async def test_agent_repeated_git_diff_not_covered_by_default() -> None: provider = _RepeatedSuccessfulToolThenDoneProvider( tool_retries=4, tool_name="git_diff", arguments={"path": "/testbed"}, ) handler_calls = 0 async def _tool(call: Any) -> ToolResult: nonlocal handler_calls handler_calls += 1 return ToolResult( tool_use_id=call.tool_use_id, tool_name=call.tool_name, content="diff --git a/f b/f", is_error=False, ) agent = Agent( provider=provider, config=AgentConfig( repeated_tool_call_recovery_threshold=3, max_iterations=8, flush_enabled=False, ), tool_handler=_tool, ) events = [event async for event in agent.run_turn("show the current diff")] assert any(isinstance(event, DoneEvent) for event in events) assert handler_calls == 4 assert not any( isinstance(event, WarningEvent) and event.code == "repeated_tool_call_recovery" for event in events ) @pytest.mark.asyncio async def test_agent_repeated_extra_tool_recovery_covers_git_diff() -> None: provider = _RepeatedSuccessfulToolThenDoneProvider( tool_retries=4, tool_name="git_diff", arguments={"path": "/testbed"}, ) handler_calls = 0 async def _tool(call: Any) -> ToolResult: nonlocal handler_calls handler_calls += 1 return ToolResult( tool_use_id=call.tool_use_id, tool_name=call.tool_name, content="diff --git a/f b/f", is_error=False, ) agent = Agent( provider=provider, config=AgentConfig( repeated_tool_call_recovery_threshold=3, repeated_tool_call_recovery_extra_tools=("git_diff",), max_iterations=8, flush_enabled=False, ), tool_handler=_tool, ) events = [event async for event in agent.run_turn("show the current diff")] assert any(isinstance(event, DoneEvent) for event in events) assert handler_calls == 2 assert any( isinstance(event, WarningEvent) and event.code == "repeated_tool_call_recovery" for event in events ) @pytest.mark.asyncio async def test_agent_tool_failure_loop_resets_after_successful_state_change() -> None: calls: list[str] = [] async def _tool(call: Any) -> ToolResult: calls.append(call.tool_name) return ToolResult( tool_use_id=call.tool_use_id, tool_name=call.tool_name, content="ok" if call.tool_name == "edit_file" else "syntax error", is_error=call.tool_name != "edit_file", ) agent = Agent( provider=_ContextOverflowProvider(success_after=1), config=AgentConfig(tool_failure_loop_block_threshold=3), tool_handler=_tool, ) command_call = ToolCall( tool_use_id="cmd-1", tool_name="exec_command", arguments={"command": "python build_pptx.py", "timeout": 30}, ) await agent._execute_tool(command_call) await agent._execute_tool( ToolCall( tool_use_id="cmd-2", tool_name="exec_command", arguments=command_call.arguments, ) ) await agent._execute_tool( ToolCall( tool_use_id="edit-1", tool_name="edit_file", arguments={"path": "build_pptx.py", "old_text": "bad", "new_text": "good"}, ) ) retry_after_edit = await agent._execute_tool( ToolCall( tool_use_id="cmd-3", tool_name="exec_command", arguments=command_call.arguments, ) ) assert calls == ["exec_command", "exec_command", "edit_file", "exec_command"] assert retry_after_edit.content == "syntax error" assert retry_after_edit.execution_status is None @pytest.mark.asyncio async def test_agent_progress_watchdog_log_mode_suppresses_model_warning() -> None: provider = _RepeatedToolFailureThenDoneProvider(tool_retries=2) async def _failing_tool(call: Any) -> ToolResult: return ToolResult( tool_use_id=call.tool_use_id, tool_name=call.tool_name, content="syntax error", is_error=True, ) agent = Agent( provider=provider, config=AgentConfig( max_iterations=5, flush_enabled=False, progress_watchdog_mode="log", progress_watchdog_repeated_tool_error_threshold=2, tool_failure_loop_block_threshold=0, ), tool_handler=_failing_tool, ) events = [event async for event in agent.run_turn("build the deck")] assert any(isinstance(event, DoneEvent) for event in events) assert len(provider.calls) == 3 assert not any( isinstance(message.content, str) and "[Runtime progress warning]" in message.content for message in provider.calls[2] ) @pytest.mark.asyncio async def test_agent_progress_watchdog_can_warn_model_after_repeated_tool_errors() -> None: provider = _RepeatedToolFailureThenDoneProvider(tool_retries=2) async def _failing_tool(call: Any) -> ToolResult: return ToolResult( tool_use_id=call.tool_use_id, tool_name=call.tool_name, content="syntax error", is_error=True, ) agent = Agent( provider=provider, config=AgentConfig( max_iterations=5, flush_enabled=False, progress_watchdog_mode="warn_model", progress_watchdog_repeated_tool_error_threshold=2, tool_failure_loop_block_threshold=0, ), tool_handler=_failing_tool, ) events = [event async for event in agent.run_turn("build the deck")] assert any(isinstance(event, DoneEvent) for event in events) assert len(provider.calls) == 3 assert any( isinstance(message.content, str) and "[Runtime progress warning]" in message.content and "Do not repeat the same action unchanged" in message.content for message in provider.calls[2] ) @pytest.mark.asyncio async def test_agent_warn_model_recovers_once_before_empty_workspace_diff_final( tmp_path, ) -> None: subprocess.run(["git", "init"], cwd=tmp_path, check=True, capture_output=True) provider = _FinalThenDoneProvider() agent = Agent( provider=provider, config=AgentConfig( max_iterations=3, flush_enabled=False, progress_watchdog_mode="warn_model", ), tool_context=ToolContext(workspace_dir=str(tmp_path)), ) events = [event async for event in agent.run_turn("Fix the failing parser test")] assert any(isinstance(event, DoneEvent) for event in events) assert len(provider.calls) == 2 assert any( isinstance(message.content, str) and "[Runtime progress warning]" in message.content and "no visible workspace diff" in message.content for message in provider.calls[1] ) done_events = [event for event in events if isinstance(event, DoneEvent)] assert done_events[-1].text == "No code change is required." def _init_git_repo_with_source(tmp_path) -> None: subprocess.run(["git", "init"], cwd=tmp_path, check=True, capture_output=True) subprocess.run( ["git", "config", "user.email", "test@example.com"], cwd=tmp_path, check=True, capture_output=True, ) subprocess.run( ["git", "config", "user.name", "Test"], cwd=tmp_path, check=True, capture_output=True, ) source = tmp_path / "src" / "parser.py" source.parent.mkdir(parents=True) source.write_text("old\n", encoding="utf-8") subprocess.run(["git", "add", "src/parser.py"], cwd=tmp_path, check=True) subprocess.run(["git", "commit", "-m", "init"], cwd=tmp_path, check=True) source.write_text("new\n", encoding="utf-8") @pytest.mark.asyncio async def test_agent_warns_once_for_suspicious_final_diff_contract(tmp_path) -> None: _init_git_repo_with_source(tmp_path) (tmp_path / "debug_case.py").write_text("print('repro')\n", encoding="utf-8") runtime_events_path = tmp_path / "runtime_events.jsonl" provider = _FinalThenDoneProvider() agent = Agent( provider=provider, config=AgentConfig( max_iterations=3, flush_enabled=False, progress_watchdog_mode="log", final_diff_contract_mode="warn_model", runtime_events_path=str(runtime_events_path), ), tool_context=ToolContext( workspace_dir=str(tmp_path), workspace_file_writes=[ {"relative_path": "src/parser.py", "path": str(tmp_path / "src/parser.py")} ], workspace_mutation_receipts=[ {"relative_path": "src/parser.py", "changed": True, "partial": False}, {"relative_path": "src/parser.py", "changed": False, "partial": False}, {"relative_path": "debug_case.py", "changed": True, "partial": True}, ], workspace_mutation_records=[ { "tool": "exec_command", "paths": [{"relative_path": "debug_case.py", "classification": "scratch"}], } ], ), ) events = [event async for event in agent.run_turn("Fix the failing parser test")] assert any(isinstance(event, DoneEvent) for event in events) assert len(provider.calls) == 2 assert any( isinstance(message.content, str) and "[Runtime final-diff check]" in message.content and "debug_case.py" in message.content for message in provider.calls[1] ) assert any( isinstance(event, WarningEvent) and event.code == "final_diff_contract_recovery" for event in events ) logged = [ json.loads(line) for line in runtime_events_path.read_text(encoding="utf-8").splitlines() ] assert any( event.get("feature") == "final_diff_contract" and event.get("injected_to_model") is True and event.get("reason") == "scratch_artifact_in_final_diff" for event in logged ) final_diff_events = [ event for event in logged if event.get("feature") == "final_diff_contract" ] assert final_diff_events assert all( "runtime_events.jsonl" not in (event.get("diff_paths") or []) for event in final_diff_events ) final_diff_event = final_diff_events[0] expected_receipt_summary = { "workspace_mutation_receipt_count": 3, "changed_receipt_count": 2, "noop_receipt_count": 1, "partial_receipt_count": 1, } for key, value in expected_receipt_summary.items(): assert final_diff_event["details"][key] == value assert final_diff_event["evidence"][key] == value @pytest.mark.asyncio async def test_agent_final_diff_contract_log_mode_does_not_prompt_model(tmp_path) -> None: _init_git_repo_with_source(tmp_path) (tmp_path / "debug_case.py").write_text("print('repro')\n", encoding="utf-8") runtime_events_path = tmp_path.parent / f"{tmp_path.name}-runtime_events.jsonl" provider = _FinalThenDoneProvider() agent = Agent( provider=provider, config=AgentConfig( max_iterations=3, flush_enabled=False, progress_watchdog_mode="log", final_diff_contract_mode="log", runtime_events_path=str(runtime_events_path), ), tool_context=ToolContext(workspace_dir=str(tmp_path)), ) events = [event async for event in agent.run_turn("Fix the failing parser test")] assert any(isinstance(event, DoneEvent) for event in events) assert len(provider.calls) == 1 assert not [ event for event in events if isinstance(event, WarningEvent) and event.code == "final_diff_contract_recovery" ] logged = [ json.loads(line) for line in runtime_events_path.read_text(encoding="utf-8").splitlines() ] assert any( event.get("feature") == "final_diff_contract" and event.get("injected_to_model") is False for event in logged ) final_diff_event = next( event for event in logged if event.get("feature") == "final_diff_contract" ) expected_receipt_summary = { "workspace_mutation_receipt_count": 0, "changed_receipt_count": 0, "noop_receipt_count": 0, "partial_receipt_count": 0, } for key, value in expected_receipt_summary.items(): assert final_diff_event["details"][key] == value assert final_diff_event["evidence"][key] == value @pytest.mark.asyncio async def test_agent_final_diff_contract_warns_for_empty_diff_after_workspace_write( tmp_path, ) -> None: _init_git_repo_with_source(tmp_path) (tmp_path / "src" / "parser.py").write_text("old\n", encoding="utf-8") runtime_events_path = tmp_path.parent / f"{tmp_path.name}-empty-diff-events.jsonl" provider = _FinalThenDoneProvider() agent = Agent( provider=provider, config=AgentConfig( max_iterations=3, flush_enabled=False, progress_watchdog_mode="log", final_diff_contract_mode="warn_model", runtime_events_path=str(runtime_events_path), ), tool_context=ToolContext( workspace_dir=str(tmp_path), workspace_file_writes=[ {"relative_path": "src/parser.py", "path": str(tmp_path / "src/parser.py")} ], ), ) events = [event async for event in agent.run_turn("Fix the failing parser test")] assert any(isinstance(event, DoneEvent) for event in events) assert len(provider.calls) == 2 assert any( isinstance(message.content, str) and "[Runtime final-diff check]" in message.content and "Current diff paths: " in message.content and "src/parser.py" in message.content for message in provider.calls[1] ) assert any( isinstance(event, WarningEvent) and event.code == "final_diff_contract_recovery" for event in events ) logged = [ json.loads(line) for line in runtime_events_path.read_text(encoding="utf-8").splitlines() ] assert any( event.get("feature") == "final_diff_contract" and event.get("injected_to_model") is True and event.get("reason") == "workspace_writes_without_final_diff" and event.get("diff_paths") == [] for event in logged ) @pytest.mark.asyncio async def test_agent_records_final_diff_contract_on_finish_error_with_diff(tmp_path) -> None: _init_git_repo_with_source(tmp_path) (tmp_path / "debug_case.py").write_text("print('repro')\n", encoding="utf-8") runtime_events_path = tmp_path.parent / f"{tmp_path.name}-error-runtime_events.jsonl" provider = _ProviderRaisesTimeout() agent = Agent( provider=provider, config=AgentConfig( max_provider_retries=0, timeout=60.0, iteration_timeout=30.0, flush_enabled=False, progress_watchdog_mode="log", final_diff_contract_mode="warn_model", runtime_events_path=str(runtime_events_path), ), tool_context=ToolContext(workspace_dir=str(tmp_path)), ) events = [event async for event in agent.run_turn("Fix the failing parser test")] assert len(provider.calls) == 1 assert any( isinstance(event, ErrorEvent) and event.code == "agent_runtime_timeout" for event in events ) assert not [ event for event in events if isinstance(event, WarningEvent) and event.code == "final_diff_contract_recovery" ] logged = [ json.loads(line) for line in runtime_events_path.read_text(encoding="utf-8").splitlines() ] assert any(event.get("reason") == "finish_error_with_non_empty_diff" for event in logged) final_diff_events = [ event for event in logged if event.get("feature") == "final_diff_contract" ] assert final_diff_events final_diff_event = final_diff_events[0] assert final_diff_event["mode"] == "warn_model" assert final_diff_event["action"] == "observe" assert final_diff_event["injected_to_model"] is False assert final_diff_event["reason"] == "scratch_artifact_in_final_diff" assert final_diff_event["diff_paths"] == ["debug_case.py", "src/parser.py"] assert final_diff_event["evidence"]["scratch_paths"] == ["debug_case.py"] assert final_diff_event["evidence"]["source_paths"] == ["src/parser.py"] @pytest.mark.asyncio async def test_agent_warn_model_recovers_before_final_after_failed_tool_with_diff( tmp_path, ) -> None: subprocess.run(["git", "init"], cwd=tmp_path, check=True, capture_output=True) source = tmp_path / "src.py" source.write_text("old\n", encoding="utf-8") subprocess.run(["git", "add", "src.py"], cwd=tmp_path, check=True, capture_output=True) subprocess.run( ["git", "commit", "-m", "init"], cwd=tmp_path, check=True, capture_output=True, env={ **dict(os.environ), "GIT_AUTHOR_NAME": "Test", "GIT_AUTHOR_EMAIL": "test@example.com", "GIT_COMMITTER_NAME": "Test", "GIT_COMMITTER_EMAIL": "test@example.com", }, ) provider = _FailedToolThenFinalProvider() tool_context = ToolContext(workspace_dir=str(tmp_path)) async def _failing_after_write(call: Any) -> ToolResult: tool_context.workspace_file_writes.append( {"relative_path": "src.py", "path": str(source)} ) source.write_text("new\n", encoding="utf-8") return ToolResult( tool_use_id=call.tool_use_id, tool_name=call.tool_name, content=( "[shell_warning:masked_pipeline_failure]\n" "error[E0308]: mismatched types" ), is_error=True, execution_status={ "version": 1, "status": "error", "exit_code": 0, "timed_out": False, "truncated": False, "reason": "masked_pipeline_failure", "source": "adapter", "preservation_class": "diagnostic", }, ) agent = Agent( provider=provider, config=AgentConfig( max_iterations=4, flush_enabled=False, progress_watchdog_mode="warn_model", tool_failure_loop_block_threshold=0, ), tool_handler=_failing_after_write, tool_context=tool_context, ) events = [event async for event in agent.run_turn("Fix the failing parser test")] assert any(isinstance(event, DoneEvent) for event in events) assert len(provider.calls) == 3 assert any( isinstance(message.content, str) and "[Runtime progress warning]" in message.content and "masked_pipeline_failure" in message.content and "Do not finalize this patch yet" in message.content for message in provider.calls[2] ) assert any( isinstance(event, WarningEvent) and event.code == "failed_tool_finalization_recovery" for event in events ) done_events = [event for event in events if isinstance(event, DoneEvent)] assert done_events[-1].text == "final attempt 3" assert agent.config.metadata["failed_tool_finalization_recoveries"] == 1 @pytest.mark.asyncio async def test_agent_rewarns_after_new_failed_focused_verification_with_diff( tmp_path, ) -> None: subprocess.run(["git", "init"], cwd=tmp_path, check=True, capture_output=True) source = tmp_path / "src.py" source.write_text("old\n", encoding="utf-8") subprocess.run(["git", "add", "src.py"], cwd=tmp_path, check=True, capture_output=True) subprocess.run( ["git", "commit", "-m", "init"], cwd=tmp_path, check=True, capture_output=True, env={ **dict(os.environ), "GIT_AUTHOR_NAME": "Test", "GIT_AUTHOR_EMAIL": "test@example.com", "GIT_COMMITTER_NAME": "Test", "GIT_COMMITTER_EMAIL": "test@example.com", }, ) tool_context = ToolContext(workspace_dir=str(tmp_path)) verification_calls = 0 async def _tool(call: Any) -> ToolResult: nonlocal verification_calls if call.tool_name == "edit_file": source.write_text("new\n", encoding="utf-8") tool_context.workspace_file_writes.append( {"relative_path": "src.py", "path": str(source)} ) return ToolResult( tool_use_id=call.tool_use_id, tool_name=call.tool_name, content="edited", ) if call.tool_name == "exec_command": verification_calls += 1 return ToolResult( tool_use_id=call.tool_use_id, tool_name=call.tool_name, content=f"error: focused validation failure {verification_calls}", is_error=True, ) raise AssertionError(f"unexpected tool: {call.tool_name}") provider = _RepeatedFailedVerificationFinalProvider() agent = Agent( provider=provider, config=AgentConfig( max_iterations=8, flush_enabled=False, progress_watchdog_mode="warn_model", tool_failure_loop_block_threshold=0, ), tool_handler=_tool, tool_context=tool_context, ) events = [event async for event in agent.run_turn("Fix the failing parser test")] warning_events = [ event for event in events if isinstance(event, WarningEvent) and event.code == "failed_tool_finalization_recovery" ] assert any(isinstance(event, DoneEvent) for event in events) assert len(provider.calls) == 6 assert len(warning_events) == 2 assert agent.config.metadata["failed_tool_finalization_recoveries"] == 2 assert any( isinstance(message.content, str) and "focused validation still failed" in message.content and "focused validation failure 1" in message.content for message in provider.calls[3] ) assert any( isinstance(message.content, str) and "focused validation still failed" in message.content and "focused validation failure 2" in message.content for message in provider.calls[5] ) done_events = [event for event in events if isinstance(event, DoneEvent)] assert done_events[-1].text == "final attempt 6" @pytest.mark.asyncio async def test_agent_does_not_warn_after_clean_maven_verification_summary( tmp_path, ) -> None: subprocess.run(["git", "init"], cwd=tmp_path, check=True, capture_output=True) source = tmp_path / "src.py" source.write_text("old\n", encoding="utf-8") subprocess.run(["git", "add", "src.py"], cwd=tmp_path, check=True, capture_output=True) subprocess.run( ["git", "commit", "-m", "init"], cwd=tmp_path, check=True, capture_output=True, env={ **dict(os.environ), "GIT_AUTHOR_NAME": "Test", "GIT_AUTHOR_EMAIL": "test@example.com", "GIT_COMMITTER_NAME": "Test", "GIT_COMMITTER_EMAIL": "test@example.com", }, ) tool_context = ToolContext(workspace_dir=str(tmp_path)) success_log = ( "exit_code=0\n" "[INFO] Results:\n" "[INFO] Tests run: 5, Failures: 0, Errors: 0, Skipped: 0\n" "[INFO] Scanned 862 class file(s) for forbidden API invocations, 0 error(s).\n" "[INFO] BUILD SUCCESS\n" ) assert Agent._tool_result_has_validation_success_signal(success_log) assert not Agent._tool_result_has_failure_signal(success_log) short_success_log = "test result: ok. 4 passed; 0 failed\n" assert Agent._tool_result_has_validation_success_signal(short_success_log) assert not Agent._tool_result_has_failure_signal(short_success_log) async def _tool(call: Any) -> ToolResult: if call.tool_name == "edit_file": source.write_text("new\n", encoding="utf-8") tool_context.workspace_file_writes.append( {"relative_path": "src.py", "path": str(source)} ) return ToolResult( tool_use_id=call.tool_use_id, tool_name=call.tool_name, content="edited", ) if call.tool_name == "exec_command": return ToolResult( tool_use_id=call.tool_use_id, tool_name=call.tool_name, content=success_log, ) raise AssertionError(f"unexpected tool: {call.tool_name}") provider = _PostWriteCleanMavenVerificationThenFinalProvider() agent = Agent( provider=provider, config=AgentConfig( max_iterations=5, flush_enabled=False, progress_watchdog_mode="warn_model", tool_failure_loop_block_threshold=0, ), tool_handler=_tool, tool_context=tool_context, ) events = [event async for event in agent.run_turn("Fix the failing parser test")] assert not [ event for event in events if isinstance(event, WarningEvent) and event.code == "failed_tool_finalization_recovery" ] assert len(provider.calls) == 3 done_events = [event for event in events if isinstance(event, DoneEvent)] assert done_events[-1].text == "final attempt 3" assert "failed_tool_finalization_recoveries" not in agent.config.metadata @pytest.mark.asyncio async def test_agent_warns_before_final_without_successful_focused_verification( tmp_path, ) -> None: subprocess.run(["git", "init"], cwd=tmp_path, check=True, capture_output=True) source = tmp_path / "src.py" source.write_text("old\n", encoding="utf-8") subprocess.run(["git", "add", "src.py"], cwd=tmp_path, check=True, capture_output=True) subprocess.run( ["git", "commit", "-m", "init"], cwd=tmp_path, check=True, capture_output=True, env={ **dict(os.environ), "GIT_AUTHOR_NAME": "Test", "GIT_AUTHOR_EMAIL": "test@example.com", "GIT_COMMITTER_NAME": "Test", "GIT_COMMITTER_EMAIL": "test@example.com", }, ) tool_context = ToolContext(workspace_dir=str(tmp_path)) async def _tool(call: Any) -> ToolResult: if call.tool_name == "edit_file": source.write_text("new\n", encoding="utf-8") tool_context.workspace_file_writes.append( {"relative_path": "src.py", "path": str(source)} ) return ToolResult( tool_use_id=call.tool_use_id, tool_name=call.tool_name, content="edited", ) raise AssertionError(f"unexpected tool: {call.tool_name}") provider = _EditThenFinalProvider() agent = Agent( provider=provider, config=AgentConfig( max_iterations=4, flush_enabled=False, progress_watchdog_mode="warn_model", ), tool_handler=_tool, tool_context=tool_context, ) events = [event async for event in agent.run_turn("Fix the failing parser test")] assert any(isinstance(event, DoneEvent) for event in events) assert len(provider.calls) == 3 assert any( isinstance(message.content, str) and "before any focused validation command succeeded" in message.content for message in provider.calls[2] ) assert agent.config.metadata["failed_tool_finalization_recoveries"] == 1 done_events = [event for event in events if isinstance(event, DoneEvent)] assert done_events[-1].text == "final attempt 3" def test_agent_focused_verification_recognizes_build_and_linter_checks() -> None: agent = object.__new__(Agent) assert agent._command_looks_like_focused_verification( "cd /testbed && cargo build --release --bin ruff 2>&1 | tail -30" ) assert agent._command_looks_like_focused_verification( "cd /testbed && cargo check -p ruff_linter" ) assert agent._command_looks_like_focused_verification( "./target/release/ruff check /tmp/repro.py --select=F523 --fix" ) assert agent._command_looks_like_focused_verification("make check") assert agent._command_looks_like_focused_verification( "./run-tests.py -i basics/try_finally_return.py" ) assert agent._command_looks_like_focused_verification("tests/jqtest") def test_focused_verification_classifier_success() -> None: result = ToolResult( tool_use_id="tool-1", tool_name="exec_command", content="exit_code=0\n3 passed\n", is_error=False, ) assert Agent._classify_focused_verification_result(result) == "success" def test_focused_verification_classifier_failure() -> None: result = ToolResult( tool_use_id="tool-1", tool_name="exec_command", content="exit_code=1\nFAILED tests/test_demo.py::test_demo\n", is_error=True, ) assert Agent._classify_focused_verification_result(result) == "failure" def test_focused_verification_classifier_unknown_without_success_signal() -> None: result = ToolResult( tool_use_id="tool-1", tool_name="exec_command", content="exit_code=0\nran command and wrote logs\n", is_error=False, ) assert Agent._classify_focused_verification_result(result) == "unknown" def test_agent_source_context_signature_includes_exec_source_reads() -> None: agent = object.__new__(Agent) source_result = ToolResult( tool_use_id="tool-1", tool_name="exec_command", content="1\tfn important() {}\n", ) signature = agent._source_context_signature( [ ToolCall( tool_use_id="tool-1", tool_name="exec_command", arguments={"command": "sed -n '1,20p' src/lib.rs"}, ) ], [source_result], ) assert signature is not None def test_agent_source_context_signature_ignores_non_source_exec_commands() -> None: agent = object.__new__(Agent) test_result = ToolResult( tool_use_id="tool-1", tool_name="exec_command", content="test result: ok. 4 passed; 0 failed\n", ) signature = agent._source_context_signature( [ ToolCall( tool_use_id="tool-1", tool_name="exec_command", arguments={"command": "cargo test -p parser"}, ) ], [test_result], ) assert signature is None def test_agent_filters_gitlink_only_porcelain_status() -> None: status = ( " m modules/oniguruma\n" " M src/parser.y\n" " M sample.json\n" "A sample2.json\n" "?? sample.json\n" "?? scratch.py\n" "?? src/new_module.py\n" ) filtered = Agent._filter_gitlink_porcelain_status( status, {"modules/oniguruma"}, ) assert "modules/oniguruma" not in filtered assert " M src/parser.y" in filtered assert " M sample.json" in filtered assert "A sample2.json" not in filtered assert "?? sample.json" not in filtered assert "?? scratch.py" not in filtered assert "?? src/new_module.py" in filtered assert Agent._filter_gitlink_porcelain_status( " m modules/oniguruma\n", {"modules/oniguruma"}, ) == "" @pytest.mark.asyncio async def test_agent_ignores_gitlink_only_workspace_diff(tmp_path) -> None: submodule = tmp_path / "submodule" repo = tmp_path / "repo" submodule.mkdir() repo.mkdir() for path in (submodule, repo): subprocess.run(["git", "init"], cwd=path, check=True, capture_output=True) subprocess.run( ["git", "config", "user.email", "test@example.com"], cwd=path, check=True, capture_output=True, ) subprocess.run( ["git", "config", "user.name", "Test"], cwd=path, check=True, capture_output=True, ) (submodule / "file.txt").write_text("clean\n", encoding="utf-8") subprocess.run(["git", "add", "file.txt"], cwd=submodule, check=True, capture_output=True) subprocess.run( ["git", "commit", "-m", "submodule init"], cwd=submodule, check=True, capture_output=True, ) subprocess.run( [ "git", "-c", "protocol.file.allow=always", "submodule", "add", str(submodule), "modules/oniguruma", ], cwd=repo, check=True, capture_output=True, ) subprocess.run(["git", "commit", "-m", "repo init"], cwd=repo, check=True, capture_output=True) (repo / "modules/oniguruma/file.txt").write_text("dirty\n", encoding="utf-8") (repo / "sample.json").write_text('{"repro": true}\n', encoding="utf-8") agent = Agent( provider=_FinalThenDoneProvider(), config=AgentConfig(flush_enabled=False), tool_context=ToolContext(workspace_dir=str(repo)), ) assert await agent._workspace_git_status_porcelain() == "" assert agent._workspace_diff_paths_for_runtime_event() == [] assert agent._workspace_diff_fingerprint_for_runtime_event() is None (repo / "src").mkdir() (repo / "src/new_module.py").write_text("value = 1\n", encoding="utf-8") status = await agent._workspace_git_status_porcelain() assert status == "?? src/new_module.py\n" assert agent._workspace_diff_paths_for_runtime_event() == ["src/new_module.py"] assert agent._workspace_diff_fingerprint_for_runtime_event() is not None def test_progress_watchdog_post_write_guidance_is_diff_focused() -> None: message = _progress_watchdog_guidance_message( "verified_workspace_diff_continued_tool_activity", { "count": 3, "workspace_write_count": 1, }, ) assert "You already have repository edits" in message assert "latest verification result" in message assert "Stop broad source exploration" in message def test_progress_watchdog_repeated_post_write_guidance_limits_source_tools() -> None: message = _progress_watchdog_guidance_message( "verified_workspace_diff_continued_tool_activity", { "count": 6, "workspace_write_count": 1, }, ) assert "have received this warning again" in message assert "Do not call read_file" in message assert "make a source edit" in message def test_progress_watchdog_code_fix_no_write_guidance_requires_workspace_edit() -> None: message = _progress_watchdog_guidance_message( "tool_activity_without_workspace_write", { "count": 16, "scratch_write_count": 4, "workspace_change_likely_required": True, }, ) assert "appears to require a repository patch" in message assert "no tracked workspace source file has been changed yet" in message assert "targeted source reads/searches" in message assert "writing more scratch notes" in message assert "use an available source-edit tool" in message assert "apply_patch, edit_file, or write_file" not in message def test_workspace_edit_gate_rejects_scratch_write_file(tmp_path) -> None: agent = Agent( provider=_ContextOverflowProvider(success_after=1), tool_context=ToolContext(workspace_dir=str(tmp_path)), ) gate_details = { "reason": "tool_activity_without_workspace_write", "count": 16, "threshold": 8, } scratch_result = agent._workspace_edit_gate_tool_result( ToolCall( tool_use_id="write-1", tool_name="write_file", arguments={"path": "/tmp/notes.md", "content": "notes"}, ), gate_details, recovery_read_paths=set(), recovery_reads_remaining=0, ) workspace_result = agent._workspace_edit_gate_tool_result( ToolCall( tool_use_id="write-2", tool_name="write_file", arguments={"path": str(tmp_path / "src.py"), "content": "patch"}, ), gate_details, recovery_read_paths=set(), recovery_reads_remaining=0, ) assert scratch_result is not None assert scratch_result.is_error is True assert scratch_result.execution_status["reason"] == "workspace_edit_required" assert workspace_result is None def test_workspace_edit_gate_rejects_synthetic_marker_write_file(tmp_path) -> None: agent = Agent( provider=_ContextOverflowProvider(success_after=1), tool_context=ToolContext(workspace_dir=str(tmp_path)), ) gate_details = { "reason": "tool_activity_without_workspace_write", "count": 16, "threshold": 8, } marker_result = agent._workspace_edit_gate_tool_result( ToolCall( tool_use_id="write-marker", tool_name="write_file", arguments={ "path": str(tmp_path / "src" / "debug_marker.h"), "content": "/* Placeholder for runtime guard unlock */\n", }, ), gate_details, recovery_read_paths=set(), recovery_reads_remaining=0, ) real_new_file_result = agent._workspace_edit_gate_tool_result( ToolCall( tool_use_id="write-real", tool_name="write_file", arguments={ "path": str(tmp_path / "src" / "feature_support.h"), "content": "int feature_support_enabled(void);\n", }, ), gate_details, recovery_read_paths=set(), recovery_reads_remaining=0, ) assert marker_result is not None assert marker_result.is_error is True assert "temporary marker" in marker_result.content assert real_new_file_result is None def test_effective_workspace_write_records_ignore_synthetic_new_files(tmp_path) -> None: tool_context = ToolContext(workspace_dir=str(tmp_path)) tool_context.workspace_file_writes.extend( [ { "relative_path": "src/debug_marker.h", "path": str(tmp_path / "src" / "debug_marker.h"), "created": True, }, { "relative_path": "src/parser.y", "path": str(tmp_path / "src" / "parser.y"), "created": False, }, ] ) agent = Agent( provider=_ContextOverflowProvider(success_after=1), tool_context=tool_context, ) assert [record["relative_path"] for record in agent._effective_workspace_write_records()] == [ "src/parser.y" ] def test_filter_ignored_porcelain_status_ignores_root_scratch_artifacts() -> None: status = "\n".join( [ "?? fix.patch", "?? src/parser.rs", "", ] ) assert Agent._filter_ignored_porcelain_status(status, set()) == "?? src/parser.rs\n" def test_filter_ignored_porcelain_status_can_make_scratch_only_diff_empty() -> None: status = "\n".join( [ "?? fix.patch", "", ] ) assert Agent._filter_ignored_porcelain_status(status, set()) == "" @pytest.mark.asyncio async def test_workspace_edit_gate_preserves_source_tools_after_repeated_no_write( tmp_path, ) -> None: source = tmp_path / "src.py" source.write_text("old\n", encoding="utf-8") tool_context = ToolContext(workspace_dir=str(tmp_path)) handler_calls: list[str] = [] async def _tool(call: Any) -> ToolResult: handler_calls.append(call.tool_name) if call.tool_name == "read_file": return ToolResult( tool_use_id=call.tool_use_id, tool_name=call.tool_name, content=source.read_text(encoding="utf-8"), ) if call.tool_name == "apply_patch": source.write_text("new\n", encoding="utf-8") tool_context.workspace_file_writes.append( {"relative_path": "src.py", "path": str(source)} ) return ToolResult( tool_use_id=call.tool_use_id, tool_name=call.tool_name, content="Applied patch: 1 file(s) modified [workspace]", ) raise AssertionError(f"unexpected tool: {call.tool_name}") provider = _NoWorkspaceWriteThenPatchProvider() agent = Agent( provider=provider, config=AgentConfig( max_iterations=25, flush_enabled=False, progress_watchdog_mode="warn_model", ), tool_definitions=[ ToolDefinition( name=name, description=f"{name} tool.", input_schema=ToolInputSchema(), ) for name in [ "read_file", "grep_search", "exec_command", "apply_patch", "edit_file", "write_file", ] ], tool_handler=_tool, tool_context=tool_context, ) events = [event async for event in agent.run_turn("Fix the failing parser test")] assert source.read_text(encoding="utf-8") == "new\n" assert handler_calls.count("read_file") == 17 assert handler_calls.count("apply_patch") == 1 assert agent.config.metadata["workspace_edit_gate_activations"] == 1 filtered_tool_names = {tool.name for tool in provider.tools_by_call[16] or []} assert filtered_tool_names == { "read_file", "grep_search", "exec_command", "apply_patch", "edit_file", "write_file", } gated_config = provider.configs[16] assert gated_config is not None assert "Runtime Patch Progress Guidance" in (gated_config.system or "") assert gated_config.tool_choice is None assert any(isinstance(event, DoneEvent) for event in events) assert not any( getattr(event, "kind", None) == "tool_result" and getattr(event, "tool_name", None) == "read_file" and (getattr(event, "execution_status", None) or {}).get("reason") == "workspace_edit_required" for event in events ) @pytest.mark.asyncio async def test_workspace_edit_gate_allows_target_read_after_patch_context_failure( tmp_path, ) -> None: source = tmp_path / "src.py" source.write_text("old\n", encoding="utf-8") tool_context = ToolContext(workspace_dir=str(tmp_path)) handler_calls: list[str] = [] patch_calls = 0 async def _tool(call: Any) -> ToolResult: nonlocal patch_calls handler_calls.append(call.tool_name) if call.tool_name == "read_file": return ToolResult( tool_use_id=call.tool_use_id, tool_name=call.tool_name, content=source.read_text(encoding="utf-8"), ) if call.tool_name == "apply_patch": patch_calls += 1 if patch_calls == 1: return ToolResult( tool_use_id=call.tool_use_id, tool_name=call.tool_name, content=( "apply_patch context mismatch at line 1: expected 'old', " "got 'older'. Read the current file content and retry with " "exact surrounding context." ), is_error=True, ) source.write_text("new\n", encoding="utf-8") tool_context.workspace_file_writes.append( {"relative_path": "src.py", "path": str(source)} ) return ToolResult( tool_use_id=call.tool_use_id, tool_name=call.tool_name, content="Applied patch: 1 file(s) modified [workspace]", ) raise AssertionError(f"unexpected tool: {call.tool_name}") provider = _PatchFailureRecoveryProvider() agent = Agent( provider=provider, config=AgentConfig( max_iterations=30, flush_enabled=False, progress_watchdog_mode="warn_model", ), tool_definitions=[ ToolDefinition( name=name, description=f"{name} tool.", input_schema=ToolInputSchema(), ) for name in [ "read_file", "grep_search", "exec_command", "apply_patch", "edit_file", "write_file", ] ], tool_handler=_tool, tool_context=tool_context, ) events = [event async for event in agent.run_turn("Fix the failing parser test")] assert source.read_text(encoding="utf-8") == "new\n" assert handler_calls.count("read_file") == 18 assert handler_calls.count("apply_patch") == 2 assert agent.config.metadata["workspace_edit_gate_activations"] == 1 assert agent.config.metadata["workspace_edit_gate_patch_recoveries"] == 1 first_gated_tools = {tool.name for tool in provider.tools_by_call[16] or []} assert first_gated_tools == { "read_file", "grep_search", "exec_command", "apply_patch", "edit_file", "write_file", } recovery_tools = {tool.name for tool in provider.tools_by_call[18] or []} assert recovery_tools == { "read_file", "grep_search", "exec_command", "apply_patch", "edit_file", "write_file", } post_recovery_tools = {tool.name for tool in provider.tools_by_call[19] or []} assert post_recovery_tools == { "read_file", "grep_search", "exec_command", "apply_patch", "edit_file", "write_file", } recovery_config = provider.configs[18] assert recovery_config is not None assert "failed edit target path" in (recovery_config.system or "") assert recovery_config.tool_choice is None post_recovery_config = provider.configs[19] assert post_recovery_config is not None assert post_recovery_config.tool_choice is None assert any(isinstance(event, DoneEvent) for event in events) @pytest.mark.asyncio async def test_workspace_edit_gate_allows_target_read_after_edit_context_failure( tmp_path, ) -> None: source = tmp_path / "src.py" source.write_text("old\n", encoding="utf-8") tool_context = ToolContext(workspace_dir=str(tmp_path)) handler_calls: list[str] = [] async def _tool(call: Any) -> ToolResult: handler_calls.append(call.tool_name) if call.tool_name == "read_file": return ToolResult( tool_use_id=call.tool_use_id, tool_name=call.tool_name, content=source.read_text(encoding="utf-8"), ) if call.tool_name == "edit_file": return ToolResult( tool_use_id=call.tool_use_id, tool_name=call.tool_name, content=( "edit_file could not find old_text in src.py. Read the current " "file content, then retry with exact text from that file." ), is_error=True, ) if call.tool_name == "apply_patch": source.write_text("new\n", encoding="utf-8") tool_context.workspace_file_writes.append( {"relative_path": "src.py", "path": str(source)} ) return ToolResult( tool_use_id=call.tool_use_id, tool_name=call.tool_name, content="Applied patch: 1 file(s) modified [workspace]", ) raise AssertionError(f"unexpected tool: {call.tool_name}") provider = _EditFailureRecoveryProvider() agent = Agent( provider=provider, config=AgentConfig( max_iterations=30, flush_enabled=False, progress_watchdog_mode="warn_model", ), tool_definitions=[ ToolDefinition( name=name, description=f"{name} tool.", input_schema=ToolInputSchema(), ) for name in [ "read_file", "grep_search", "exec_command", "apply_patch", "edit_file", "write_file", ] ], tool_handler=_tool, tool_context=tool_context, ) events = [event async for event in agent.run_turn("Fix the failing parser test")] assert source.read_text(encoding="utf-8") == "new\n" assert handler_calls.count("read_file") == 18 assert handler_calls.count("edit_file") == 1 assert handler_calls.count("apply_patch") == 1 assert agent.config.metadata["workspace_edit_gate_activations"] == 1 assert agent.config.metadata["workspace_edit_gate_patch_recoveries"] == 1 recovery_tools = {tool.name for tool in provider.tools_by_call[18] or []} assert recovery_tools == { "read_file", "grep_search", "exec_command", "apply_patch", "edit_file", "write_file", } assert any(isinstance(event, DoneEvent) for event in events) @pytest.mark.asyncio async def test_agent_failed_focused_verification_counts_after_workspace_write(tmp_path) -> None: tool_context = ToolContext(workspace_dir=str(tmp_path)) source = tmp_path / "src.py" source.write_text("old\n", encoding="utf-8") async def _tool(call: Any) -> ToolResult: if call.tool_name == "edit_file": source.write_text("new\n", encoding="utf-8") tool_context.workspace_file_writes.append( {"relative_path": "src.py", "path": str(source)} ) return ToolResult( tool_use_id=call.tool_use_id, tool_name=call.tool_name, content="edited", ) if call.tool_name == "exec_command": return ToolResult( tool_use_id=call.tool_use_id, tool_name=call.tool_name, content="error: build failed", is_error=True, ) return ToolResult( tool_use_id=call.tool_use_id, tool_name=call.tool_name, content="source", ) provider = _PostWriteFailedVerificationThenSourceProvider() agent = Agent( provider=provider, config=AgentConfig( max_iterations=8, flush_enabled=False, progress_watchdog_mode="warn_model", tool_failure_loop_block_threshold=0, ), tool_handler=_tool, tool_context=tool_context, ) events = [event async for event in agent.run_turn("Fix the failing parser test")] assert any(isinstance(event, DoneEvent) for event in events) assert len(provider.calls) == 6 assert any( isinstance(message.content, str) and "continued tool activity after a workspace diff and focused verification" in message.content and "Stop broad source exploration" in message.content for message in provider.calls[5] ) @pytest.mark.asyncio async def test_agent_converges_after_stable_verified_workspace_diff(tmp_path) -> None: subprocess.run(["git", "init"], cwd=tmp_path, check=True, capture_output=True) source = tmp_path / "src.py" source.write_text("old\n", encoding="utf-8") subprocess.run(["git", "add", "src.py"], cwd=tmp_path, check=True, capture_output=True) subprocess.run( ["git", "commit", "-m", "init"], cwd=tmp_path, check=True, capture_output=True, env={ **dict(os.environ), "GIT_AUTHOR_NAME": "Test", "GIT_AUTHOR_EMAIL": "test@example.com", "GIT_COMMITTER_NAME": "Test", "GIT_COMMITTER_EMAIL": "test@example.com", }, ) runtime_events_path = tmp_path / "runtime_events.jsonl" tool_context = ToolContext(workspace_dir=str(tmp_path)) handler_calls: list[str] = [] async def _tool(call: Any) -> ToolResult: handler_calls.append(call.tool_name) if call.tool_name == "edit_file": before = fingerprint_path(source) source.write_text("new\n", encoding="utf-8") after = fingerprint_path(source) record_semantic_mutation_receipt( tool_name="edit_file", path=source, operation="edit_file", before=before, after=after, partial=False, ctx=tool_context, ) return ToolResult( tool_use_id=call.tool_use_id, tool_name=call.tool_name, content="edited", ) if call.tool_name == "exec_command": return ToolResult( tool_use_id=call.tool_use_id, tool_name=call.tool_name, content="test result: ok. 4 passed; 0 failed\n", ) if call.tool_name == "read_file": return ToolResult( tool_use_id=call.tool_use_id, tool_name=call.tool_name, content=source.read_text(encoding="utf-8"), ) raise AssertionError(f"unexpected tool: {call.tool_name}") provider = _StableVerifiedDiffThenSourceProvider() agent = Agent( provider=provider, config=AgentConfig( max_iterations=12, flush_enabled=False, progress_watchdog_mode="warn_model", post_write_convergence_enabled=True, runtime_events_path=str(runtime_events_path), ), tool_handler=_tool, tool_context=tool_context, ) events = [event async for event in agent.run_turn("Fix the failing parser test")] assert any(isinstance(event, DoneEvent) for event in events) assert handler_calls == ["edit_file", "exec_command", *["read_file"] * 6] assert any( isinstance(message.content, str) and "[Runtime post-write convergence]" in message.content and "current diff has stayed unchanged" in message.content for call in provider.calls for message in call ) assert any( isinstance(message.content, str) and "[Runtime post-write convergence]" in message.content and "Do not call tools" in message.content for call in provider.calls for message in call ) assert provider.tool_lists[-1] is None done_events = [event for event in events if isinstance(event, DoneEvent)] assert done_events[-1].text == "final after convergence 9" runtime_events = [ json.loads(line) for line in runtime_events_path.read_text(encoding="utf-8").splitlines() ] assert any(row.get("name") == "post_write_convergence.warned" for row in runtime_events) assert any(row.get("name") == "post_write_convergence.finalized" for row in runtime_events) @pytest.mark.asyncio async def test_agent_blocks_repeated_missing_tool_handler_failures() -> None: agent = Agent( provider=_ContextOverflowProvider(success_after=1), config=AgentConfig(tool_failure_loop_block_threshold=3), ) tool_call = ToolCall( tool_use_id="missing-1", tool_name="missing_tool", arguments={"value": "same"}, ) await agent._execute_tool(tool_call) await agent._execute_tool(tool_call) third = await agent._execute_tool(tool_call) assert "tool_failure_loop_exhausted" not in third.content assert "Do not retry this exact call unchanged" in third.content assert third.execution_status is not None assert third.execution_status.get("reason") == "tool_failure_loop_exhausted" @pytest.mark.asyncio async def test_agent_provider_request_proof_budget_is_separate_from_tool_result_cap() -> None: provider = _ConfigCapturingProvider() agent = Agent( provider=provider, config=AgentConfig( context_window_tokens=200_000, max_tokens=8192, tool_result_provider_request_max_chars=96_000, flush_enabled=False, ), ) events = [event async for event in agent.run_turn("hello")] assert any(event.kind == "done" for event in events) assert provider.configs assert provider.configs[0] is not None assert provider.configs[0].provider_request_max_chars > 96_000 @pytest.mark.asyncio async def test_agent_provider_request_proof_budget_accepts_explicit_override() -> None: provider = _ConfigCapturingProvider() agent = Agent( provider=provider, config=AgentConfig( provider_request_proof_max_chars=123_456, tool_result_provider_request_max_chars=96_000, flush_enabled=False, ), ) events = [event async for event in agent.run_turn("hello")] assert any(event.kind == "done" for event in events) assert provider.configs assert provider.configs[0] is not None assert provider.configs[0].provider_request_max_chars == 123_456 def test_agent_child_config_inherits_tool_failure_loop_thresholds() -> None: agent = Agent( provider=_ContextOverflowProvider(success_after=1), config=AgentConfig( tool_failure_loop_block_threshold=7, provider_context_block_feedback=True, identical_request_loop_break_threshold=9, repeated_tool_call_recovery_threshold=11, progress_watchdog_mode="warn_model", progress_watchdog_repeated_tool_error_threshold=5, progress_watchdog_repeated_provider_failure_threshold=4, progress_watchdog_repeated_failure_anchor_threshold=6, tool_loop_observer_mode="log", runtime_recovery_mode="warn_model", runtime_recovery_source_loop_max_nudges=3, post_tool_empty_recovery_mode="warn_model", reasoning_prefill_recovery_mode="recover", runtime_events_path="/tmp/runtime-events.jsonl", ), ) child = agent._make_child_agent(SubagentSpec(task="child task"), depth=1) assert child.config.tool_failure_loop_block_threshold == 7 assert child.config.provider_context_block_feedback is True assert child.config.identical_request_loop_break_threshold == 9 assert child.config.repeated_tool_call_recovery_threshold == 11 assert child.config.progress_watchdog_mode == "warn_model" assert child.config.progress_watchdog_repeated_tool_error_threshold == 5 assert child.config.progress_watchdog_repeated_provider_failure_threshold == 4 assert child.config.progress_watchdog_repeated_failure_anchor_threshold == 6 assert child.config.tool_loop_observer_mode == "log" assert child.config.runtime_recovery_mode == "warn_model" assert child.config.runtime_recovery_source_loop_max_nudges == 3 assert child.config.post_tool_empty_recovery_mode == "warn_model" assert child.config.reasoning_prefill_recovery_mode == "recover" assert child.config.runtime_events_path == "/tmp/runtime-events.jsonl" def test_agent_config_normalizes_flush_triggers_and_clamps_compaction_tail() -> None: config = AgentConfig( flush_triggers=["reset", "inline_overflow"], compaction_protected_recent_messages=-4, ) assert config.flush_triggers == ["session_reset", "pre_compaction"] assert config.compaction_protected_recent_messages == 0 def test_agent_config_rejects_unknown_flush_triggers() -> None: with pytest.raises(ValueError, match="unknown flush trigger"): AgentConfig(flush_triggers=["manual", "bogus"]) def test_agent_child_config_inherits_context_and_flush_budget_policy() -> None: agent = Agent( provider=_ContextOverflowProvider(success_after=1), config=AgentConfig( context_window_tokens=200_000, max_tokens=8192, provider_request_proof_max_chars=123_456, tool_use_argument_provider_request_max_chars=12_345, tool_result_provider_request_max_chars=54_321, max_turn_llm_calls=9, max_turn_input_tokens=700_000, max_turn_output_tokens=70_000, max_turn_billed_cost_usd=0.75, max_turn_tool_errors=4, flush_enabled=True, flush_triggers=["session_reset", "manual", "idle", "pre_compaction"], flush_pre_compaction=True, flush_timeout_seconds=1.5, flush_background_timeout_seconds=15.0, flush_backoff_initial_seconds=3.0, flush_backoff_max_seconds=30.0, flush_archive_max_bytes=999_999, flush_compaction_requires_safe_receipt=False, ), ) child = agent._make_child_agent(SubagentSpec(task="child task"), depth=1) assert child.config.context_window_tokens == 200_000 assert child.config.provider_request_proof_max_chars == 123_456 assert child.config.tool_use_argument_provider_request_max_chars == 12_345 assert child.config.tool_result_provider_request_max_chars == 54_321 assert child.config.max_turn_llm_calls == 9 assert child.config.max_turn_input_tokens == 700_000 assert child.config.max_turn_output_tokens == 70_000 assert child.config.max_turn_billed_cost_usd == 0.75 assert child.config.max_turn_tool_errors == 4 assert child.config.flush_enabled is True assert child.config.flush_triggers == [ "session_reset", "manual", "idle", "pre_compaction", ] assert child.config.flush_pre_compaction is True assert child.config.flush_timeout_seconds == 1.5 assert child.config.flush_background_timeout_seconds == 15.0 assert child.config.flush_backoff_initial_seconds == 3.0 assert child.config.flush_backoff_max_seconds == 30.0 assert child.config.flush_archive_max_bytes == 999_999 assert child.config.flush_compaction_requires_safe_receipt is False def test_agent_config_max_turn_cost_usd_defaults_to_disabled() -> None: assert AgentConfig().max_turn_cost_usd == 0.0 @pytest.mark.asyncio async def test_agent_skips_price_resolution_per_event_when_turn_cost_budget_disabled( monkeypatch: pytest.MonkeyPatch, ) -> None: # max_turn_cost_usd left at its disabled default (0.0): the turn-cost # accumulator in the per-event ProviderDoneEvent branch must never touch # the (potentially network-blocking) price resolver, even across several # cost-blind LLM calls in the same turn. A single call to # resolve_model_price still happens once, at the very end of the turn, # for the pre-existing DoneEvent cost-reporting computation (out of scope # for this gate) — so this asserts the count does *not* scale with the # number of LLM calls, rather than asserting zero calls overall. import opensquilla.engine.pricing as pricing_module calls: list[tuple[str, str]] = [] real_resolve_model_price = pricing_module.resolve_model_price def _counting_resolve_model_price(model_id: str, provider: str = "") -> Any: calls.append((model_id, provider)) return real_resolve_model_price(model_id, provider) monkeypatch.setattr(pricing_module, "resolve_model_price", _counting_resolve_model_price) provider = _HighUsageToolLoopProvider(tool_rounds=3, input_tokens_per_call=1000) async def _tool(call: Any) -> ToolResult: return ToolResult(tool_use_id=call.tool_use_id, tool_name=call.tool_name, content="ok") agent = Agent( provider=provider, config=AgentConfig( model_id="deepseek/deepseek-v4-pro-20260423", flush_enabled=False, ), tool_handler=_tool, ) events = [event async for event in agent.run_turn("hello")] assert len(provider.calls) == 4 assert not any(event.kind == "error" for event in events) assert any(event.kind == "done" for event in events) # Exactly one call: the end-of-turn DoneEvent cost estimate. None of the # four per-event ProviderDoneEvent accumulation passes should have called # the resolver while the gate is disabled. assert len(calls) == 1 def test_agent_child_config_inherits_max_turn_cost_usd() -> None: agent = Agent( provider=_ContextOverflowProvider(success_after=1), config=AgentConfig(max_turn_cost_usd=0.42), ) child = agent._make_child_agent(SubagentSpec(task="child task"), depth=1) assert child.config.max_turn_cost_usd == 0.42 @pytest.mark.asyncio async def test_agent_stops_when_turn_estimated_cost_budget_is_exceeded() -> None: # No provider-billed cost at all (billed_cost=0.0 on every call), so the # gate must fall back to the estimator to know it is over budget. provider = _NoBilledCostUsageProvider(input_tokens=1000, output_tokens=1000) agent = Agent( provider=provider, config=AgentConfig( model_id="deepseek/deepseek-v4-pro-20260423", max_turn_cost_usd=0.001, ), ) events = [event async for event in agent.run_turn("hello")] assert len(provider.calls) == 1 error = next( event for event in events if event.kind == "error" and event.code == "turn_cost_budget_exceeded" ) assert "estimated cost basis" in error.message @pytest.mark.asyncio async def test_agent_labels_turn_cost_budget_error_as_mixed_when_billed_and_estimated() -> None: # Turn mixes a real billed cost (call 1) with an estimated cost (call 2, # cost-blind) — the gate's basis label must reflect both contributions. provider = _MixedBilledAndEstimatedCostProvider() async def _tool(call: Any) -> ToolResult: return ToolResult(tool_use_id=call.tool_use_id, tool_name=call.tool_name, content="ok") agent = Agent( provider=provider, config=AgentConfig( model_id="deepseek/deepseek-v4-pro-20260423", max_turn_cost_usd=0.001, ), tool_handler=_tool, ) events = [event async for event in agent.run_turn("hello")] assert len(provider.calls) == 2 error = next( event for event in events if event.kind == "error" and event.code == "turn_cost_budget_exceeded" ) assert "mixed cost basis" in error.message def test_with_model_usage_cost_fields_prices_unbilled_cache_reads_cache_aware() -> None: from opensquilla.engine import agent as agent_module blind_row = { "model": "deepseek/deepseek-v4-pro-20260423", "input_tokens": 1000, "output_tokens": 0, "billed_cost": 0.0, } cached_row = dict(blind_row, cache_read_tokens=800) blind = agent_module._with_model_usage_cost_fields([blind_row])[0] cached = agent_module._with_model_usage_cost_fields([cached_row])[0] assert blind["estimate_basis"] == "cache_aware" assert cached["estimate_basis"] == "cache_aware" # (200 * 0.435 + 800 * 0.003625) / 1e6 == 0.0000899, rounded to 6dp by # model_usage_cost_fields. assert cached["cost_usd"] == pytest.approx(0.00009) assert cached["cost_usd"] < blind["cost_usd"] class _BudgetCheckingProvider: provider_name = "openrouter" def __init__(self, *, proof_budget: int) -> None: self.proof_budget = proof_budget self.calls: list[list[Message]] = [] self.proofs: list[dict[str, Any]] = [] def chat( self, messages: list[Message], tools: list[Any] | None = None, config: ChatConfig | None = None, ) -> AsyncIterator[Any]: self.calls.append(messages) return self._stream(messages) async def _stream(self, messages: list[Message]) -> AsyncIterator[Any]: payload = {"messages": [message.model_dump(mode="json") for message in messages]} try: proof = prove_provider_payload( payload, projection_adapter="openrouter", proof_budget=self.proof_budget, ) except ProviderRequestBudgetExceeded as exc: self.proofs.append(exc.proof) yield ProviderError( message=json.dumps(exc.proof), code="provider_request_budget_exhausted", ) return self.proofs.append(proof) yield ProviderText(text="ok") yield ProviderDone(stop_reason="stop", input_tokens=1, output_tokens=1) async def list_models(self) -> list[Any]: return [] class _ProviderRaisesTimeout: provider_name = "fake" def __init__(self) -> None: self.calls: list[list[Message]] = [] def chat( self, messages: list[Message], tools: list[Any] | None = None, config: ChatConfig | None = None, ) -> AsyncIterator[Any]: self.calls.append(messages) return self._stream() async def _stream(self) -> AsyncIterator[Any]: raise TimeoutError("provider transport timeout") yield ProviderText(text="unreachable") async def list_models(self) -> list[Any]: return [] class _ProviderHeartbeatThenText: provider_name = "fake" def chat( self, messages: list[Message], tools: list[Any] | None = None, config: ChatConfig | None = None, ) -> AsyncIterator[Any]: return self._stream() async def _stream(self) -> AsyncIterator[Any]: yield ProviderHeartbeatEvent(phase="llm_fallback", message="retrying") yield ProviderText(text="ok") yield ProviderDone(stop_reason="stop", input_tokens=1, output_tokens=1) async def list_models(self) -> list[Any]: return [] class _ToolUseProvider: provider_name = "fake" def __init__(self) -> None: self.calls: list[list[Message]] = [] def chat( self, messages: list[Message], tools: list[Any] | None = None, config: ChatConfig | None = None, ) -> AsyncIterator[Any]: self.calls.append(messages) return self._stream() async def _stream(self) -> AsyncIterator[Any]: yield ProviderToolUseStart(tool_use_id="tool-1", tool_name="slow") yield ProviderToolUseEnd( tool_use_id="tool-1", tool_name="slow", arguments={}, ) yield ProviderDone(stop_reason="tool_use", input_tokens=1, output_tokens=1) async def list_models(self) -> list[Any]: return [] @pytest.mark.asyncio async def test_provider_heartbeat_reaches_agent_stream() -> None: agent = Agent( provider=_ProviderHeartbeatThenText(), config=AgentConfig(iteration_timeout=30.0, timeout=60.0, max_provider_retries=0), ) events = [event async for event in agent.run_turn("hello")] heartbeat_index = _event_index( events, lambda event: ( isinstance(event, RunHeartbeatEvent) and event.phase == "llm_fallback" and event.message == "retrying" ), ) text_index = _event_index( events, lambda event: ( getattr(event, "kind", None) == "text_delta" and getattr(event, "text", None) == "ok" ), ) assert heartbeat_index < text_index @pytest.mark.asyncio async def test_iteration_timeout_interrupts_stalled_provider_stream() -> None: provider = _StallingProvider() agent = Agent( provider=provider, config=AgentConfig(iteration_timeout=0.01, max_provider_retries=0), ) events = await asyncio.wait_for( _collect_events(agent.run_turn("hello")), timeout=0.5, ) error_index = _event_index( events, lambda event: isinstance(event, ErrorEvent) and event.code == "iteration_timeout", ) state_index = _event_index( events, lambda event: ( getattr(event, "kind", None) == "state_change" and getattr(event, "to_state", None) == AgentState.ERROR ), ) assert state_index < error_index assert len(provider.calls) == 1 assert provider.stream_closed is True assert not any(isinstance(event, DoneEvent) for event in events) @pytest.mark.asyncio async def test_iteration_timeout_does_not_interrupt_active_tool_argument_stream() -> None: async def write_file_tool(call: object) -> ToolResult: return ToolResult( tool_use_id=getattr(call, "tool_use_id"), tool_name=getattr(call, "tool_name"), content="written", ) provider = _ActiveLongToolArgumentProvider(fragment_delay=0.02) agent = Agent( provider=provider, config=AgentConfig( iteration_timeout=0.03, timeout=1.0, max_provider_retries=0, ), tool_definitions=[ ToolDefinition( name="write_file", description="Write a file.", input_schema=ToolInputSchema(), ) ], tool_handler=write_file_tool, ) events = await asyncio.wait_for(_collect_events(agent.run_turn("hello")), timeout=1.0) assert provider.calls assert any( getattr(event, "kind", None) == "tool_result" and getattr(event, "tool_name", None) == "write_file" and getattr(event, "result", None) == "written" for event in events ) assert any(isinstance(event, DoneEvent) for event in events) assert not any( isinstance(event, ErrorEvent) and event.code == "iteration_timeout" for event in events ) @pytest.mark.asyncio async def test_large_tool_argument_stream_emits_progress_heartbeat() -> None: async def write_file_tool(call: object) -> ToolResult: return ToolResult( tool_use_id=getattr(call, "tool_use_id"), tool_name=getattr(call, "tool_name"), content="written", ) provider = _ActiveLongToolArgumentProvider( fragment_delay=0.0, content="x" * 5000, ) agent = Agent( provider=provider, config=AgentConfig(iteration_timeout=1.0, timeout=2.0, max_provider_retries=0), tool_definitions=[ ToolDefinition( name="write_file", description="Write a file.", input_schema=ToolInputSchema(), ) ], tool_handler=write_file_tool, ) events = await asyncio.wait_for(_collect_events(agent.run_turn("hello")), timeout=1.0) heartbeat_index = _event_index( events, lambda event: ( isinstance(event, RunHeartbeatEvent) and event.phase == "llm_tool_arguments" and "write_file" in (event.message or "") ), ) done_index = _event_index(events, lambda event: isinstance(event, DoneEvent)) assert heartbeat_index < done_index @pytest.mark.asyncio async def test_iteration_timeout_caps_tool_execution() -> None: tool_started = asyncio.Event() tool_cancelled = asyncio.Event() never_complete = asyncio.Event() async def slow_tool(call: object) -> ToolResult: tool_started.set() try: await never_complete.wait() except asyncio.CancelledError: tool_cancelled.set() raise return ToolResult( tool_use_id=getattr(call, "tool_use_id"), tool_name=getattr(call, "tool_name"), content="late", ) agent = Agent( provider=_ToolUseProvider(), config=AgentConfig( iteration_timeout=0.1, timeout=5.0, tool_timeout=5.0, max_provider_retries=0, ), tool_definitions=[ ToolDefinition( name="slow", description="Slow tool.", input_schema=ToolInputSchema(), ) ], tool_handler=slow_tool, ) events = await asyncio.wait_for(_collect_events(agent.run_turn("hello")), timeout=2.0) assert tool_started.is_set() assert tool_cancelled.is_set() assert any( isinstance(event, ErrorEvent) and event.code == "iteration_timeout" for event in events ) @pytest.mark.asyncio async def test_provider_timeout_error_is_not_reclassified_as_iteration_timeout() -> None: provider = _ProviderRaisesTimeout() agent = Agent( provider=provider, config=AgentConfig(iteration_timeout=30.0, timeout=60.0, max_provider_retries=0), ) events = [event async for event in agent.run_turn("hello")] assert len(provider.calls) == 1 assert any( isinstance(event, ErrorEvent) and event.code == "agent_runtime_timeout" for event in events ) assert not any( isinstance(event, ErrorEvent) and event.code == "iteration_timeout" for event in events ) @pytest.mark.asyncio async def test_context_overflow_noop_compaction_does_not_resend_unchanged_context( monkeypatch: pytest.MonkeyPatch, ) -> None: async def _noop_compact(request: Any) -> CompactionResult: return CompactionResult( summary="", kept_entries=request.entries, removed_count=0, chunks_processed=0, ) monkeypatch.setattr("opensquilla.engine.agent.compact_context", _noop_compact) provider = _ContextOverflowProvider() agent = Agent( provider=provider, config=AgentConfig( max_provider_retries=0, max_overflow_retries=2, flush_enabled=False, ), ) events = [event async for event in agent.run_turn("hello")] assert len(provider.calls) == 1 assert any( isinstance(event, ErrorEvent) and event.code == "compaction_not_smaller" for event in events ) assert not any(getattr(event, "kind", None) == "compaction" for event in events) @pytest.mark.asyncio async def test_context_overflow_summary_only_larger_payload_does_not_retry( monkeypatch: pytest.MonkeyPatch, ) -> None: async def _summary_only_compact(request: Any) -> CompactionResult: return CompactionResult( summary="summary without reducing request payload", kept_entries=request.entries, removed_count=0, chunks_processed=1, ) monkeypatch.setattr("opensquilla.engine.agent.compact_context", _summary_only_compact) provider = _ContextOverflowProvider() agent = Agent( provider=provider, config=AgentConfig( max_provider_retries=0, max_overflow_retries=2, flush_enabled=False, ), ) events = [event async for event in agent.run_turn("hello")] assert len(provider.calls) == 1 assert any( isinstance(event, ErrorEvent) and event.code == "compaction_not_smaller" for event in events ) assert not any(getattr(event, "kind", None) == "compaction" for event in events) @pytest.mark.asyncio async def test_context_overflow_effective_compaction_allows_single_retry( monkeypatch: pytest.MonkeyPatch, ) -> None: async def _effective_compact(request: Any) -> CompactionResult: return CompactionResult( summary="short summary", kept_entries=[], removed_count=len(request.entries), chunks_processed=1, ) monkeypatch.setattr("opensquilla.engine.agent.compact_context", _effective_compact) provider = _ContextOverflowProvider(success_after=1) agent = Agent( provider=provider, config=AgentConfig( max_provider_retries=0, max_overflow_retries=2, flush_enabled=False, ), ) events = [event async for event in agent.run_turn("x" * 4000)] assert len(provider.calls) == 2 assert _provider_payload_is_smaller(provider.calls[0], provider.calls[1]) assert any(event.kind == "done" and getattr(event, "text", "") == "ok" for event in events) @pytest.mark.asyncio async def test_inline_overflow_compaction_reduces_tool_heavy_structured_context() -> None: big_output = ("synthetic log line: lorem ipsum dolor sit amet 0123456789\n" * 700)[:40_000] window_tokens = 8_000 messages: list[Message] = [ Message(role="user", content="Please analyze every log file in the workspace."), Message(role="assistant", content="Reading the logs now."), ] for i in range(6): messages.append( Message( role="assistant", content=[ ContentBlockToolUse( id=f"tool-{i}", name="read_file", input={"path": f"logs/part-{i}.log"}, ) ], ) ) messages.append( Message( role="user", content=[ContentBlockToolResult(tool_use_id=f"tool-{i}", content=big_output)], ) ) agent = Agent( provider=_ContextOverflowProvider(), config=AgentConfig(context_window_tokens=window_tokens, flush_enabled=False), ) original_chars = session_payload_chars(messages) outcome = await agent._check_context_overflow( messages, estimated_context_tokens=window_tokens + 1, compaction_window_tokens=window_tokens, ) assert outcome is not None assert outcome.compacted assert session_payload_chars(outcome.messages) < original_chars @pytest.mark.asyncio async def test_within_budget_skip_on_string_only_history_is_not_reported_as_compacted() -> None: agent = Agent( provider=_ContextOverflowProvider(), config=AgentConfig( context_window_tokens=1000, context_overflow_threshold=0.85, flush_enabled=False, ), ) messages = [ Message(role="user", content="my project is called Zephyr"), Message(role="assistant", content="Understood, the project is Zephyr."), Message(role="user", content="tell me a short story about it"), ] outcome = await agent._check_context_overflow(messages, estimated_context_tokens=900) assert outcome is not None assert outcome.removed_count == 0 assert outcome.summary == "" assert outcome.compacted is False @pytest.mark.asyncio async def test_inline_overflow_uses_live_context_not_cumulative_provider_usage( monkeypatch: pytest.MonkeyPatch, ) -> None: import opensquilla.engine.agent as agent_module provider = _HighUsageToolLoopProvider(tool_rounds=3, input_tokens_per_call=4000) flush_calls: list[int] = [] compact_requests: list[Any] = [] async def _tool(call: Any) -> ToolResult: return ToolResult( tool_use_id=call.tool_use_id, tool_name=call.tool_name, content="ok", ) async def _flush(_plan: Any, flush_messages: list[Message]) -> Any: flush_calls.append(len(flush_messages)) return SimpleNamespace( mode="llm", indexed_chunk_count=1, integrity_status="ok", output_coverage_status="ok", invalid_candidate_count=0, candidate_missing_ids=[], obligation_status="ok", obligation_missing_ids=[], ) async def _compact(request: Any) -> CompactionResult: compact_requests.append(request) return CompactionResult( summary="", kept_entries=request.entries, removed_count=0, chunks_processed=0, ) agent = Agent( provider=provider, config=AgentConfig( context_window_tokens=20_000, context_overflow_threshold=0.5, flush_enabled=True, flush_pre_compaction=True, flush_timeout_seconds=0.01, max_iterations=10, ), tool_handler=_tool, ) monkeypatch.setattr(agent, "_run_flush", _flush) monkeypatch.setattr(agent_module, "compact_context", _compact) events = [event async for event in agent.run_turn("read the files one by one")] done = next(event for event in events if isinstance(event, DoneEvent)) assert done.text == "done" assert done.input_tokens == 16_000 assert len(provider.calls) == 4 assert flush_calls == [] assert compact_requests == [] @pytest.mark.asyncio async def test_inline_overflow_still_triggers_for_large_live_provider_request( monkeypatch: pytest.MonkeyPatch, ) -> None: import opensquilla.engine.agent as agent_module provider = _HighUsageToolLoopProvider(tool_rounds=0, input_tokens_per_call=1) large_tool = ToolDefinition( name="large_context_tool", description="large live request surface " + ("z" * 6000), input_schema=ToolInputSchema(), ) flush_calls: list[int] = [] compact_requests: list[Any] = [] async def _flush(_plan: Any, flush_messages: list[Message]) -> Any: flush_calls.append(len(flush_messages)) return SimpleNamespace( mode="llm", indexed_chunk_count=1, integrity_status="ok", output_coverage_status="ok", invalid_candidate_count=0, candidate_missing_ids=[], obligation_status="ok", obligation_missing_ids=[], ) async def _compact(request: Any) -> CompactionResult: compact_requests.append(request) return CompactionResult( summary="", kept_entries=request.entries, removed_count=0, chunks_processed=0, ) agent = Agent( provider=provider, config=AgentConfig( context_window_tokens=3000, context_overflow_threshold=0.5, flush_enabled=True, flush_pre_compaction=True, flush_timeout_seconds=0.01, system_prompt="live request system context " + ("s" * 2000), ), tool_definitions=[large_tool], ) monkeypatch.setattr(agent, "_run_flush", _flush) monkeypatch.setattr(agent_module, "compact_context", _compact) events = [event async for event in agent.run_turn("hello")] assert any(isinstance(event, DoneEvent) for event in events) assert len(provider.calls) == 1 assert flush_calls == [1] assert len(compact_requests) == 1 @pytest.mark.asyncio async def test_inline_overflow_flush_enabled_without_trigger_skips_flush( monkeypatch: pytest.MonkeyPatch, ) -> None: import opensquilla.engine.agent as agent_module provider = _HighUsageToolLoopProvider(tool_rounds=0, input_tokens_per_call=1) flush_calls: list[int] = [] compact_requests: list[Any] = [] async def _flush(_plan: Any, flush_messages: list[Message]) -> Any: flush_calls.append(len(flush_messages)) async def _compact(request: Any) -> CompactionResult: compact_requests.append(request) return CompactionResult( summary="", kept_entries=request.entries, removed_count=0, chunks_processed=0, ) agent = Agent( provider=provider, config=AgentConfig( context_window_tokens=3000, context_overflow_threshold=0.5, flush_enabled=True, flush_pre_compaction=False, system_prompt="live request system context " + ("s" * 2000), ), tool_definitions=[ ToolDefinition( name="large_context_tool", description="large live request surface " + ("z" * 6000), input_schema=ToolInputSchema(), ) ], ) monkeypatch.setattr(agent, "_run_flush", _flush) monkeypatch.setattr(agent_module, "compact_context", _compact) events = [event async for event in agent.run_turn("hello")] assert any(isinstance(event, DoneEvent) for event in events) assert flush_calls == [] assert len(compact_requests) == 1 @pytest.mark.asyncio async def test_provider_request_budget_exhausted_compacts_warns_and_retries( monkeypatch: pytest.MonkeyPatch, ) -> None: compaction_events: list[tuple[str, dict[str, Any]]] = [] async def _effective_compact(request: Any) -> CompactionResult: return CompactionResult( summary="short summary", kept_entries=[], removed_count=len(request.entries), chunks_processed=1, ) monkeypatch.setattr("opensquilla.engine.agent.compact_context", _effective_compact) monkeypatch.setattr( "opensquilla.engine.agent.notify_compaction", lambda session_key, **payload: compaction_events.append((session_key, payload)), ) provider = _ProviderRequestBudgetExceededProvider(success_after=1) agent = Agent( provider=provider, config=AgentConfig( max_provider_retries=0, max_overflow_retries=2, flush_enabled=False, ), session_key="agent:main:budget", ) events = [event async for event in agent.run_turn("x" * 4000)] warning_codes = [event.code for event in events if isinstance(event, WarningEvent)] assert len(provider.calls) == 2 assert _provider_payload_is_smaller(provider.calls[0], provider.calls[1]) assert warning_codes == [ "context_auto_compaction_start", "context_auto_compaction_retry", ] assert any(event.kind == "done" and getattr(event, "text", "") == "ok" for event in events) assert not any( isinstance(event, ErrorEvent) and event.code == "provider_request_budget_exhausted" for event in events ) assert [(key, payload["status"]) for key, payload in compaction_events] == [ ("agent:main:budget", "started"), ("agent:main:budget", "observed"), ("agent:main:budget", "observed"), ] compaction_ids = {payload.get("compaction_id") for _, payload in compaction_events} assert len(compaction_ids) == 1 assert None not in compaction_ids assert [payload["event"] for _, payload in compaction_events] == [ "compaction.triggered", "compaction.chunk_summarized", "compaction.summary_verified", ] @pytest.mark.asyncio async def test_provider_request_budget_uses_provider_window_for_compaction( monkeypatch: pytest.MonkeyPatch, ) -> None: compaction_windows: list[int] = [] async def _effective_compact(request: Any) -> CompactionResult: compaction_windows.append(request.context_window_tokens) return CompactionResult( summary="short summary", kept_entries=[], removed_count=len(request.entries), chunks_processed=1, ) monkeypatch.setattr("opensquilla.engine.agent.compact_context", _effective_compact) provider = _ProviderRequestBudgetExceededProvider( success_after=1, proof={ "fallback_reason": "provider_request_budget_exhausted", "estimated_chars": 109_055, "estimated_tokens": 27_263, "proof_budget": 96_000, "recent_tail_too_large": True, }, ) agent = Agent( provider=provider, config=AgentConfig( context_window_tokens=1_048_576, max_provider_retries=0, max_overflow_retries=2, flush_enabled=False, ), ) events = [event async for event in agent.run_turn("x" * 4000)] assert compaction_windows assert compaction_windows[0] < 100_000 assert len(provider.calls) == 2 assert any(event.kind == "done" and getattr(event, "text", "") == "ok" for event in events) @pytest.mark.asyncio async def test_provider_request_budget_retry_payload_is_rechecked_against_budget( monkeypatch: pytest.MonkeyPatch, ) -> None: compaction_windows: list[int] = [] async def _effective_compact(request: Any) -> CompactionResult: compaction_windows.append(request.context_window_tokens) return CompactionResult( summary="short summary", kept_entries=[], removed_count=len(request.entries), chunks_processed=1, ) monkeypatch.setattr("opensquilla.engine.agent.compact_context", _effective_compact) provider = _BudgetCheckingProvider(proof_budget=2_500) agent = Agent( provider=provider, config=AgentConfig( context_window_tokens=1_048_576, max_provider_retries=0, max_overflow_retries=2, flush_enabled=False, ), ) events = [event async for event in agent.run_turn("x" * 4000)] assert [proof["fits"] for proof in provider.proofs] == [False, True] assert compaction_windows assert compaction_windows[0] < agent.config.context_window_tokens assert len(provider.calls) == 2 assert provider.proofs[1]["estimated_chars"] <= provider.proof_budget assert session_payload_chars(provider.calls[1]) < provider.proof_budget assert any(event.kind == "done" and getattr(event, "text", "") == "ok" for event in events) @pytest.mark.asyncio async def test_provider_budget_retry_uses_effective_proof_budget( monkeypatch: pytest.MonkeyPatch, ) -> None: compaction_windows: list[int] = [] async def _record_compaction_window(request: Any) -> CompactionResult: compaction_windows.append(request.context_window_tokens) return CompactionResult( summary="short summary", kept_entries=[], removed_count=len(request.entries), chunks_processed=1, ) monkeypatch.setattr("opensquilla.engine.agent.compact_context", _record_compaction_window) provider = _ProviderRequestBudgetExceededProvider( success_after=1, proof={ "fallback_reason": "provider_request_budget_exhausted", "estimated_chars": 100_000, "estimated_tokens": 25_000, "proof_budget": 96_000, "raw_proof_budget": 96_000, "effective_proof_budget": 86_400, "proof_headroom_chars": 9_600, "recent_tail_too_large": False, }, ) agent = Agent( provider=provider, config=AgentConfig( context_window_tokens=1_048_576, max_provider_retries=0, max_overflow_retries=2, flush_enabled=False, ), ) events = [event async for event in agent.run_turn("x" * 4000)] assert compaction_windows == [21_600] assert any(event.kind == "done" and getattr(event, "text", "") == "ok" for event in events) @pytest.mark.asyncio async def test_provider_request_budget_recent_tail_reason_survives_noop_compaction( monkeypatch: pytest.MonkeyPatch, ) -> None: async def _noop_compact(request: Any) -> CompactionResult: return CompactionResult( summary="", kept_entries=request.entries, removed_count=0, chunks_processed=0, ) monkeypatch.setattr("opensquilla.engine.agent.compact_context", _noop_compact) provider = _ProviderRequestBudgetExceededProvider( proof={ "fallback_reason": "provider_request_budget_exhausted", "estimated_chars": 109_055, "estimated_tokens": 27_263, "proof_budget": 96_000, "recent_tail_too_large": True, }, ) agent = Agent( provider=provider, config=AgentConfig( context_window_tokens=1_048_576, max_provider_retries=0, max_overflow_retries=2, flush_enabled=False, ), ) events = [event async for event in agent.run_turn("x" * 4000)] errors = [event for event in events if isinstance(event, ErrorEvent)] assert len(provider.calls) == 1 assert errors[-1].code == "provider_request_too_large" assert RAW_CURRENT_TURN_OVERFLOW_MESSAGE not in errors[-1].message assert not any( isinstance(event, ErrorEvent) and event.code == "compaction_not_smaller" for event in events ) @pytest.mark.asyncio async def test_provider_request_budget_recent_tail_exhaustion_is_reported_as_controlled_error( monkeypatch: pytest.MonkeyPatch, ) -> None: async def _effective_compact(request: Any) -> CompactionResult: return CompactionResult( summary="short summary", kept_entries=[], removed_count=len(request.entries), chunks_processed=1, ) monkeypatch.setattr("opensquilla.engine.agent.compact_context", _effective_compact) provider = _ProviderRequestBudgetExceededProvider( proof={ "fallback_reason": "provider_request_budget_exhausted", "recent_tail_too_large": True, "estimated_chars": 100_000, "proof_budget": 96_000, } ) agent = Agent( provider=provider, config=AgentConfig( max_provider_retries=0, max_overflow_retries=1, flush_enabled=False, ), ) events = [event async for event in agent.run_turn("x" * 4000)] errors = [event for event in events if isinstance(event, ErrorEvent)] assert len(provider.calls) == 2 assert errors[-1].code == "provider_request_too_large" assert "current turn" not in errors[-1].message.lower() assert RAW_CURRENT_TURN_OVERFLOW_MESSAGE not in errors[-1].message @pytest.mark.asyncio async def test_context_overflow_degraded_flush_still_runs_live_compaction_by_default( monkeypatch: pytest.MonkeyPatch, ) -> None: compact_called = False async def _compact_runs_after_degraded_flush(request: Any) -> CompactionResult: nonlocal compact_called compact_called = True return CompactionResult( summary="short summary", kept_entries=[], removed_count=len(request.entries), chunks_processed=1, ) monkeypatch.setattr( "opensquilla.engine.agent.compact_context", _compact_runs_after_degraded_flush, ) provider = _ContextOverflowProvider(success_after=1) agent = Agent( provider=provider, config=AgentConfig(max_provider_retries=0, max_overflow_retries=2), ) events = [event async for event in agent.run_turn("x" * 4000)] assert compact_called is True assert len(provider.calls) == 2 assert any(event.kind == "done" and getattr(event, "text", "") == "ok" for event in events) assert not any( isinstance(event, ErrorEvent) and event.code in {"compaction_refused_memory_flush", "compaction_refused_flush_timeout"} for event in events ) @pytest.mark.asyncio async def test_context_overflow_flush_timeout_records_backoff_and_retries( monkeypatch: pytest.MonkeyPatch, ) -> None: async def _compact_runs_after_flush_timeout(request: Any) -> CompactionResult: return CompactionResult( summary="short summary", kept_entries=[], removed_count=len(request.entries), chunks_processed=1, ) monkeypatch.setattr( "opensquilla.engine.agent.compact_context", _compact_runs_after_flush_timeout, ) provider = _ContextOverflowProvider(success_after=1) agent = Agent( provider=provider, config=AgentConfig( max_provider_retries=0, max_overflow_retries=2, flush_enabled=True, flush_pre_compaction=True, flush_timeout_seconds=0.01, flush_backoff_initial_seconds=10.0, ), ) async def slow_flush(_plan: Any, _messages: Any) -> None: await asyncio.sleep(1.0) monkeypatch.setattr(agent, "_run_flush", slow_flush) try: events = [event async for event in agent.run_turn("x" * 4000)] finally: task = agent._active_flush_task if task is not None and not task.done(): task.cancel() with pytest.raises(asyncio.CancelledError): await task assert len(provider.calls) == 2 assert agent._flush_backoff_seconds == 10.0 assert any(event.kind == "done" and getattr(event, "text", "") == "ok" for event in events) assert not any( isinstance(event, ErrorEvent) and event.code in {"compaction_refused_memory_flush", "compaction_refused_flush_timeout"} for event in events ) async def _collect_events(stream: AsyncIterator[Any]) -> list[Any]: return [event async for event in stream] def _event_index(events: list[Any], predicate: Any) -> int: return next(index for index, event in enumerate(events) if predicate(event)) def _provider_payload_is_smaller(before: list[Message], after: list[Message]) -> bool: return len(after) < len(before) or session_payload_chars(after) < session_payload_chars(before)