opensquilla--opensquilla/tests/test_engine/test_agent_llm_budget.py

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": "<html>bad</html>"},
    )

    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: <none>" 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)