"""Agent-loop tests for the finalize-time red-evidence gate. Scripted-provider tests covering the loop-level contract: off by default, challenge injection with polarity-correct message, same-state dedup, the 2-challenge cap, and that the gate never blocks finalization. """ from __future__ import annotations import json import os import subprocess from collections.abc import AsyncIterator from typing import Any import pytest from opensquilla.engine import ( Agent, AgentConfig, DoneEvent, ToolResult, WarningEvent, ) from opensquilla.provider import ChatConfig, Message from opensquilla.provider import DoneEvent as ProviderDone from opensquilla.provider import TextDeltaEvent as ProviderText from opensquilla.provider import ToolUseEndEvent as ProviderToolUseEnd from opensquilla.provider import ToolUseStartEvent as ProviderToolUseStart from opensquilla.tools.mutation_receipts import ( fingerprint_path, record_semantic_mutation_receipt, ) from opensquilla.tools.types import ToolContext _RED_MARKER = "fail-run" def _init_git_workspace(tmp_path) -> Any: 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", }, ) return source class _ScriptedProvider: """Replays a fixed per-call script of tool calls and final texts. Script entries are ``("exec", command)``, ``("edit", path)``, ``("read", path)``, or ``("final",)``. Any call past the end of the script yields a final text. """ provider_name = "fake" def __init__(self, script: list[tuple[str, ...]]) -> None: self.calls: list[list[Message]] = [] self._script = script 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]: entry: tuple[str, ...] = ("final",) if call_number <= len(self._script): entry = self._script[call_number - 1] if entry[0] == "edit": tool_use_id = f"edit-{call_number}" 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": entry[1], "old_text": "old", "new_text": "new"}, ) yield ProviderDone(stop_reason="tool_calls", input_tokens=1, output_tokens=1) return if entry[0] == "read": 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": entry[1]}, ) yield ProviderDone(stop_reason="tool_calls", input_tokens=1, output_tokens=1) return if entry[0] == "exec": 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": entry[1]}, ) 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 [] def _make_tool_handler(tmp_path, tool_context: ToolContext): source = tmp_path / "src.py" 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": command = str(call.arguments.get("command") or "") red = _RED_MARKER in command return ToolResult( tool_use_id=call.tool_use_id, tool_name=call.tool_name, content=( "exit_code=1\nFAILED: assertion did not hold" if red else "exit_code=0\nok" ), is_error=red, execution_status={ "version": 1, "status": "error" if red else "success", "exit_code": 1 if red else 0, "timed_out": False, "truncated": False, "reason": "nonzero_exit" if red else None, "source": "adapter", "preservation_class": "diagnostic" if red else "normal", }, ) raise AssertionError(f"unexpected tool: {call.tool_name}") return _tool def _gate_config(tmp_path, *, enabled: bool = True, **overrides: Any) -> AgentConfig: return AgentConfig( max_iterations=10, flush_enabled=False, # "log" keeps the pre-existing failed-tool/empty-diff warn_model # recoveries out of the way so only the gate injects here. progress_watchdog_mode="log", tool_failure_loop_block_threshold=0, finalize_evidence_gate_enabled=enabled, **overrides, ) def _gate_warnings(events: list[Any]) -> list[WarningEvent]: return [ event for event in events if isinstance(event, WarningEvent) and event.code == "finalize_evidence_gate_recovery" ] @pytest.mark.asyncio async def test_gate_off_by_default_red_final_is_accepted(tmp_path) -> None: _init_git_workspace(tmp_path) provider = _ScriptedProvider( [ ("edit", "src.py"), ("exec", "python /tmp/squilla-scratch/fail-run.py"), ("final",), ] ) tool_context = ToolContext(workspace_dir=str(tmp_path)) config = AgentConfig( max_iterations=10, flush_enabled=False, progress_watchdog_mode="log", tool_failure_loop_block_threshold=0, ) assert config.finalize_evidence_gate_enabled is False agent = Agent( provider=provider, config=config, tool_handler=_make_tool_handler(tmp_path, tool_context), tool_context=tool_context, ) events = [event async for event in agent.run_turn("Fix the bug")] assert len(provider.calls) == 3 assert _gate_warnings(events) == [] assert "finalize_evidence_gate_detections" not in agent.config.metadata done_events = [event for event in events if isinstance(event, DoneEvent)] assert done_events[-1].text == "final attempt 3" @pytest.mark.asyncio async def test_gate_challenges_red_final_then_accepts_verified_final(tmp_path) -> None: _init_git_workspace(tmp_path) runtime_events_path = tmp_path / "runtime_events.jsonl" provider = _ScriptedProvider( [ ("edit", "src.py"), ("exec", "python /tmp/squilla-scratch/fail-run.py"), ("final",), ("exec", "python /tmp/squilla-scratch/rerun_fixed.py"), ("final",), ] ) tool_context = ToolContext(workspace_dir=str(tmp_path)) agent = Agent( provider=provider, config=_gate_config(tmp_path, runtime_events_path=str(runtime_events_path)), tool_handler=_make_tool_handler(tmp_path, tool_context), tool_context=tool_context, ) events = [event async for event in agent.run_turn("Fix the bug")] assert len(provider.calls) == 5 challenge_messages = [ message.content for message in provider.calls[3] if isinstance(message.content, str) and message.content.startswith("[Finalize evidence check]") ] assert len(challenge_messages) == 1 assert "python /tmp/squilla-scratch/fail-run.py" in challenge_messages[0] assert "binding evidence" in challenge_messages[0] assert "minimal" not in challenge_messages[0].lower() assert len(_gate_warnings(events)) == 1 assert agent.config.metadata["finalize_evidence_gate_detections"] == 1 assert agent.config.metadata["finalize_evidence_gate_recoveries"] == 1 done_events = [event for event in events if isinstance(event, DoneEvent)] assert done_events[-1].text == "final attempt 5" logged = [ json.loads(line) for line in runtime_events_path.read_text().splitlines() ] challenges = [ event for event in logged if event.get("name") == "finalize_evidence_gate.challenge" ] assert len(challenges) == 1 assert challenges[0]["feature"] == "finalize_evidence_gate" assert challenges[0]["reason"] == "red_execution_after_final_edit" assert challenges[0]["injected_to_model"] is True assert challenges[0]["details"]["triggers"] == ["red_execution_after_final_edit"] assert challenges[0]["details"]["red_exit_code"] == 1 @pytest.mark.asyncio async def test_gate_same_red_state_challenges_once_and_never_blocks(tmp_path) -> None: _init_git_workspace(tmp_path) runtime_events_path = tmp_path / "runtime_events.jsonl" provider = _ScriptedProvider( [ ("edit", "src.py"), ("exec", "python /tmp/squilla-scratch/fail-run.py"), ("final",), # The model finalizes again without re-running anything: the # observation key is unchanged, so the gate stays quiet. ("final",), ] ) tool_context = ToolContext(workspace_dir=str(tmp_path)) agent = Agent( provider=provider, config=_gate_config(tmp_path, runtime_events_path=str(runtime_events_path)), tool_handler=_make_tool_handler(tmp_path, tool_context), tool_context=tool_context, ) events = [event async for event in agent.run_turn("Fix the bug")] assert len(provider.calls) == 4 assert len(_gate_warnings(events)) == 1 assert agent.config.metadata["finalize_evidence_gate_detections"] == 2 assert agent.config.metadata["finalize_evidence_gate_recoveries"] == 1 done_events = [event for event in events if isinstance(event, DoneEvent)] assert done_events[-1].text == "final attempt 4" logged = [ json.loads(line) for line in runtime_events_path.read_text().splitlines() ] challenges = [ event for event in logged if event.get("name") == "finalize_evidence_gate.challenge" ] assert [event["injected_to_model"] for event in challenges] == [True, False] @pytest.mark.asyncio async def test_gate_caps_at_two_challenges_then_accepts_red_final(tmp_path) -> None: _init_git_workspace(tmp_path) provider = _ScriptedProvider( [ ("edit", "src.py"), ("exec", "python /tmp/squilla-scratch/fail-run-a.py"), ("final",), ("exec", "python /tmp/squilla-scratch/fail-run-b.py"), ("final",), ("exec", "python /tmp/squilla-scratch/fail-run-c.py"), # Third distinct red state: the cap (2) is reached, the run must # finish even though the evidence is still red. ("final",), ] ) tool_context = ToolContext(workspace_dir=str(tmp_path)) agent = Agent( provider=provider, config=_gate_config(tmp_path), tool_handler=_make_tool_handler(tmp_path, tool_context), tool_context=tool_context, ) events = [event async for event in agent.run_turn("Fix the bug")] assert len(provider.calls) == 7 assert len(_gate_warnings(events)) == 2 assert agent.config.metadata["finalize_evidence_gate_detections"] == 3 assert agent.config.metadata["finalize_evidence_gate_recoveries"] == 2 done_events = [event for event in events if isinstance(event, DoneEvent)] assert done_events[-1].text == "final attempt 7" @pytest.mark.asyncio async def test_gate_quiet_when_final_edit_is_verified_green(tmp_path) -> None: _init_git_workspace(tmp_path) provider = _ScriptedProvider( [ ("edit", "src.py"), ("exec", "python /tmp/squilla-scratch/repro_check.py"), ("final",), ] ) tool_context = ToolContext(workspace_dir=str(tmp_path)) agent = Agent( provider=provider, config=_gate_config(tmp_path), tool_handler=_make_tool_handler(tmp_path, tool_context), tool_context=tool_context, ) events = [event async for event in agent.run_turn("Fix the bug")] assert len(provider.calls) == 3 assert _gate_warnings(events) == [] assert "finalize_evidence_gate_recoveries" not in agent.config.metadata done_events = [event for event in events if isinstance(event, DoneEvent)] assert done_events[-1].text == "final attempt 3" @pytest.mark.asyncio async def test_gate_suppressed_without_llm_call_budget_headroom(tmp_path) -> None: """A challenge must never spend the run's last LLM call: with no headroom for a follow-up call the injection would discard the model's final answer and end the turn in a hard budget error instead of a submission.""" _init_git_workspace(tmp_path) runtime_events_path = tmp_path / "runtime_events.jsonl" provider = _ScriptedProvider( [ ("edit", "src.py"), ("exec", "python /tmp/squilla-scratch/fail-run.py"), ("final",), ] ) tool_context = ToolContext(workspace_dir=str(tmp_path)) agent = Agent( provider=provider, config=_gate_config( tmp_path, max_turn_llm_calls=3, runtime_events_path=str(runtime_events_path), ), tool_handler=_make_tool_handler(tmp_path, tool_context), tool_context=tool_context, ) events = [event async for event in agent.run_turn("Fix the bug")] # The red state is detected but not injected: the final answer stands. assert len(provider.calls) == 3 assert _gate_warnings(events) == [] assert agent.config.metadata["finalize_evidence_gate_detections"] == 1 assert "finalize_evidence_gate_recoveries" not in agent.config.metadata done_events = [event for event in events if isinstance(event, DoneEvent)] assert done_events[-1].text == "final attempt 3" logged = [ json.loads(line) for line in runtime_events_path.read_text().splitlines() ] challenges = [ event for event in logged if event.get("name") == "finalize_evidence_gate.challenge" ] assert len(challenges) == 1 assert challenges[0]["injected_to_model"] is False @pytest.mark.asyncio async def test_gate_defers_to_post_write_convergence_finalization(tmp_path) -> None: """When post-write convergence has already forced a tools-disabled wrap-up, the gate must not challenge that wrap-up: it would contradict the convergence instruction ("do not call tools") with its own ("re-run the command"), and the model cannot satisfy both.""" source = _init_git_workspace(tmp_path) tool_context = ToolContext(workspace_dir=str(tmp_path)) provider = _ScriptedProvider( [ ("edit", "src.py"), # Self-written repro stays red: the gate would normally hold the # run open as red_repro_outstanding_after_final_edit. ("edit", "/tmp/squilla-scratch/fail-run-repro.py"), ("exec", "python /tmp/squilla-scratch/fail-run-repro.py"), # Green focused verification: convergence-eligible from here on. ("exec", "pytest tests/test_src.py"), *[("read", "src.py")] * 6, ] ) async def _tool(call: Any) -> ToolResult: if call.tool_name == "edit_file": path = str(call.arguments.get("path") or "") if path == "src.py": 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 == "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 == "exec_command": command = str(call.arguments.get("command") or "") red = _RED_MARKER in command return ToolResult( tool_use_id=call.tool_use_id, tool_name=call.tool_name, content=( "exit_code=1\nFAILED: assertion did not hold" if red else "test result: ok. 4 passed; 0 failed\n" ), is_error=red, execution_status={ "version": 1, "status": "error" if red else "success", "exit_code": 1 if red else 0, "timed_out": False, "truncated": False, "reason": "nonzero_exit" if red else None, "source": "adapter", "preservation_class": "diagnostic" if red else "normal", }, ) raise AssertionError(f"unexpected tool: {call.tool_name}") agent = Agent( provider=provider, config=AgentConfig( max_iterations=15, flush_enabled=False, progress_watchdog_mode="warn_model", post_write_convergence_enabled=True, finalize_evidence_gate_enabled=True, tool_failure_loop_block_threshold=0, ), tool_handler=_tool, tool_context=tool_context, ) events = [event async for event in agent.run_turn("Fix the bug")] # The convergence wrap-up finished the run in one final call. assert agent.config.metadata["post_write_convergence_finalizations"] == 1 assert len(provider.calls) == 11 done_events = [event for event in events if isinstance(event, DoneEvent)] assert done_events[-1].text == "final attempt 11" # The gate never evaluated the wrap-up, despite the outstanding red repro. assert _gate_warnings(events) == [] assert "finalize_evidence_gate_detections" not in agent.config.metadata assert not any( isinstance(message.content, str) and message.content.startswith("[Finalize evidence check]") for call in provider.calls for message in call ) _DENIED_MARKER = "blocked-run" @pytest.mark.asyncio async def test_gate_quiet_when_trailing_execution_was_denied(tmp_path) -> None: """A policy-denied execution is not red evidence: with a green verification before it, the gate must accept the final answer instead of challenging on the denied command (live valid1 defect: sandbox blocks dominated challenge reds).""" _init_git_workspace(tmp_path) provider = _ScriptedProvider( [ ("edit", "src.py"), ("exec", "pytest tests/test_src.py"), ("exec", f"python /tmp/{_DENIED_MARKER}.py"), ("final",), ] ) tool_context = ToolContext(workspace_dir=str(tmp_path)) base_handler = _make_tool_handler(tmp_path, tool_context) async def _tool(call: Any) -> ToolResult: if call.tool_name == "exec_command" and _DENIED_MARKER in str( call.arguments.get("command") or "" ): return ToolResult( tool_use_id=call.tool_use_id, tool_name=call.tool_name, content='{"status": "policy_denied", "message": "blocked by sandbox policy"}', is_error=True, execution_status={ "version": 1, "status": "error", "exit_code": None, "timed_out": False, "truncated": False, "reason": "denied", "source": "tool_runtime", "preservation_class": "diagnostic", }, ) return await base_handler(call) agent = Agent( provider=provider, config=_gate_config(tmp_path), tool_handler=_tool, tool_context=tool_context, ) events = [event async for event in agent.run_turn("Fix the bug")] assert len(provider.calls) == 4 assert _gate_warnings(events) == [] assert "finalize_evidence_gate_detections" not in agent.config.metadata done_events = [event for event in events if isinstance(event, DoneEvent)] assert done_events[-1].text == "final attempt 4"