tests/test_agent_loop.py

from __future__ import annotations

import asyncio
import base64
import hashlib
import inspect
import json
import ssl
from copy import deepcopy
from typing import Any

import httpx
import pytest

import autocad_bench.harness.agent_loop as agent_loop_module
from autocad_bench.harness.agent_loop import (
    AnthropicMessagesComputerUseModel,
    BedrockComputerUseModel,
    FireworksChatCompletionsComputerUseModel,
    MantleChatCompletionsComputerUseModel,
    MantleGrokResponsesComputerUseModel,
    MantleResponsesComputerUseModel,
    ModelTurn,
    OpenAIResponsesComputerUseModel,
    OpenRouterResponsesComputerUseModel,
    run_agent_loop,
)
from autocad_bench.sandbox.broker.fake import FakeDesktopBroker
from autocad_bench.sandbox.broker.models import (
    BrokerError,
    BrokerErrorCode,
    DesktopScreenshot,
)

_PNG = base64.b64decode(
    "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwC"
    "AAAAC0lEQVR42mNk+A8AAQUBAScY42YAAAAASUVORK5CYII="
)


class ScriptedModel:
    def __init__(self, actions: list[dict[str, Any]]) -> None:
        self.actions = list(actions)
        self.observations: list[tuple[str, bool]] = []
        self.observation_messages: list[str] = []

    async def start(self, **_: Any) -> None:
        return None

    async def next_action(self, **_: Any) -> ModelTurn:
        payload = self.actions.pop(0)
        return ModelTurn(
            tool_use_id=f"tool-{len(self.observations)}",
            action_payload=payload,
            stop_reason="tool_use",
            usage={"totalTokens": 10},
            latency_ms=1.0,
        )

    async def observe(
        self,
        *,
        tool_use_id: str,
        screenshot: DesktopScreenshot,
        message: str,
        is_error: bool,
    ) -> None:
        del screenshot
        self.observations.append((tool_use_id, is_error))
        self.observation_messages.append(message)


def test_model_system_prompt_documents_keyboard_allowlist() -> None:
    prompt = agent_loop_module._MODEL_SYSTEM_PROMPT

    assert "KEYBOARD ALLOWLIST" in prompt
    assert "ALT, BACKSPACE, CTRL, DELETE" in prompt
    assert "F1 through F12" in prompt
    assert "single ASCII letter or digit" in prompt
    assert "CONTROL, ESCAPE, RETURN, PGDN, and PGUP" in prompt
    assert "META, SUPER, WIN, and WINDOWS are prohibited" in prompt
    assert "ENTERHead" in prompt
    assert "CTRL+ALT+DELETE" in prompt
    assert "must not save" in prompt
    assert "controller saves the active DWG" in prompt


def test_responses_tool_uses_a_cross_provider_strict_action_schema() -> None:
    tool = agent_loop_module._OPENAI_COMPUTER_ACTION_TOOL

    assert tool["strict"] is True
    parameters = tool["parameters"]
    assert parameters["additionalProperties"] is False
    assert parameters["required"] == ["actions"]
    actions = parameters["properties"]["actions"]
    assert "maxItems" not in actions

    variants = actions["items"]["anyOf"]
    by_action = {
        variant["properties"]["action"]["enum"][0]: variant for variant in variants
    }
    assert set(by_action) == {
        "click",
        "double_click",
        "drag",
        "move",
        "scroll",
        "key",
        "type",
        "wait",
        "done",
    }
    for variant in variants:
        assert variant["additionalProperties"] is False
        assert set(variant["required"]) == set(variant["properties"])

    assert set(by_action["click"]["required"]) == {
        "action",
        "intent",
        "x",
        "y",
        "button",
    }
    assert "text" in by_action["type"]["required"]
    key_schema = by_action["key"]["properties"]["keys"]
    assert key_schema["maxItems"] == 4
    assert "ENTER" in key_schema["items"]["enum"]
    assert "ENTERHead" not in key_schema["items"]["enum"]


def test_all_model_provider_read_timeouts_default_to_thirty_minutes() -> None:
    expected = 30.0 * 60.0

    assert agent_loop_module.MODEL_RESPONSE_READ_TIMEOUT_S == expected
    for model_type in (
        OpenAIResponsesComputerUseModel,
        OpenRouterResponsesComputerUseModel,
        FireworksChatCompletionsComputerUseModel,
        MantleChatCompletionsComputerUseModel,
        MantleResponsesComputerUseModel,
        AnthropicMessagesComputerUseModel,
        BedrockComputerUseModel,
    ):
        assert inspect.signature(model_type).parameters["timeout_s"].default == expected


def test_agent_loop_retrieves_artifact_and_both_event_logs() -> None:
    async def run() -> None:
        broker = FakeDesktopBroker(width=320, height=200)
        model = ScriptedModel(
            [
                {"action": "type", "text": "LINE"},
                {"action": "key", "keys": ["ENTER"]},
                {"action": "done"},
            ]
        )
        result = await run_agent_loop(
            broker=broker,
            model=model,
            task_id="task-001",
            attempt_id="pilot-test",
            prompt="draw",
            reference_png=_PNG,
            max_actions=10,
            wall_timeout_s=10,
        )
        assert result.completed is True
        assert result.stop_reason == "done"
        assert result.accepted_actions == 3
        assert result.artifact is not None
        assert result.artifact.startswith(b"AC1032")
        assert result.events
        assert result.frame_events
        assert len(result.screenshots) == 4
        assert len(result.input_receipts) == 3
        assert result.status is not None
        assert result.status.checkpoint_saved is True
        assert result.cleanup_errors == ()
        assert "confirms input delivery only" in model.observation_messages[0]
        assert (
            "one outcome frame returned after the complete batch"
            in model.observation_messages[0]
        )

    asyncio.run(run())


def test_agent_loop_can_stream_trajectory_without_retaining_png_history() -> None:
    async def run() -> None:
        progress: list[str] = []
        result = await run_agent_loop(
            broker=FakeDesktopBroker(width=320, height=200),
            model=ScriptedModel(
                [
                    {"action": "type", "text": "LINE"},
                    {"action": "key", "keys": ["ENTER"]},
                    {"action": "done"},
                ]
            ),
            task_id="task-001",
            attempt_id="streamed-trajectory",
            prompt="draw",
            reference_png=_PNG,
            max_actions=10,
            wall_timeout_s=10,
            progress_callback=lambda event, _payload: progress.append(event),
            retain_in_memory_trajectory=False,
        )

        assert result.completed is True
        assert result.accepted_actions == 3
        assert result.screenshots == ()
        assert result.turns == ()
        assert result.input_receipts == ()
        assert progress.count("action_result") == 3

    asyncio.run(run())


def test_agent_loop_finalization_has_a_hard_deadline(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    class HangingCleanupBroker(FakeDesktopBroker):
        async def checkpoint_save(self, session_id: str) -> None:
            del session_id
            await asyncio.sleep(60)

        async def release_session(self, session_id: str) -> None:
            del session_id
            await asyncio.sleep(60)

    async def run() -> None:
        monkeypatch.setattr(agent_loop_module, "FINALIZATION_EVIDENCE_BUDGET_S", 0.02)
        monkeypatch.setattr(agent_loop_module, "FINALIZATION_RELEASE_TIMEOUT_S", 0.02)
        result = await asyncio.wait_for(
            run_agent_loop(
                broker=HangingCleanupBroker(width=320, height=200),
                model=ScriptedModel([{"action": "done"}]),
                task_id="task-001",
                attempt_id="bounded-finalization",
                prompt="draw",
                reference_png=_PNG,
                max_actions=2,
                wall_timeout_s=10,
            ),
            timeout=1,
        )
        assert any(
            error.startswith("checkpoint: TimeoutError")
            for error in result.cleanup_errors
        )
        assert any(
            error.startswith("release: TimeoutError") for error in result.cleanup_errors
        )

    asyncio.run(run())


def test_agent_loop_executes_batch_sequentially_and_only_observes_final_frame() -> None:
    class RecordingModel(ScriptedModel):
        def __init__(self) -> None:
            super().__init__(
                [
                    {
                        "actions": [
                            {
                                "action": "type",
                                "text": "LINE",
                                "intent": "Type the LINE command.",
                            },
                            {
                                "action": "key",
                                "keys": ["ENTER"],
                                "intent": "Submit the LINE command.",
                            },
                        ]
                    },
                    {"actions": [{"action": "done", "intent": "Finish."}]},
                ]
            )
            self.observed_sequences: list[int] = []

        async def observe(self, **kwargs: Any) -> None:
            self.observed_sequences.append(kwargs["screenshot"].sequence)
            await super().observe(**kwargs)

    class TrackingBroker(FakeDesktopBroker):
        def __init__(self) -> None:
            super().__init__(width=320, height=200)
            self.requests: list[tuple[str | None, int | None]] = []

        async def execute_action(self, session_id, action, **kwargs):
            self.requests.append(
                (kwargs.get("request_id"), kwargs.get("expected_action_index"))
            )
            return await super().execute_action(session_id, action, **kwargs)

    async def run() -> None:
        broker = TrackingBroker()
        model = RecordingModel()
        progress: list[tuple[str, Any]] = []
        result = await run_agent_loop(
            broker=broker,
            model=model,
            task_id="task-001",
            attempt_id="pilot-batch",
            prompt="draw",
            reference_png=_PNG,
            max_actions=5,
            wall_timeout_s=10,
            progress_callback=lambda event, payload: progress.append((event, payload)),
        )

        assert result.completed is True
        assert result.accepted_actions == 3
        assert [
            entry.action.action for entry in broker.audit_log(result.session_id)
        ] == [
            "type",
            "key",
            "done",
        ]
        assert model.observed_sequences == [2]
        assert (
            "No intermediate screenshot or result was provided"
            in model.observation_messages[0]
        )
        assert [expected for _, expected in broker.requests] == [1, 2, 3]
        request_ids = [request_id for request_id, _ in broker.requests]
        assert len(set(request_ids)) == 3
        action_events = [
            payload for event, payload in progress if event == "action_result"
        ]
        assert [event["batch_index"] for event in action_events] == [1, 2, 1]
        assert [event["batch_size"] for event in action_events] == [2, 2, 1]

    asyncio.run(run())


def test_agent_loop_does_not_trust_repeated_provider_tool_ids_for_idempotency() -> None:
    class RepeatedToolIdModel(ScriptedModel):
        async def next_action(self, **_: Any) -> ModelTurn:
            payload = self.actions.pop(0)
            return ModelTurn(
                tool_use_id="provider-reused-this-id",
                action_payload=payload,
                stop_reason="tool_use",
                usage={},
                latency_ms=1.0,
            )

    class TrackingBroker(FakeDesktopBroker):
        def __init__(self) -> None:
            super().__init__(width=320, height=200)
            self.request_ids: list[str | None] = []

        async def execute_action(self, session_id, action, **kwargs):
            self.request_ids.append(kwargs.get("request_id"))
            return await super().execute_action(session_id, action, **kwargs)

    async def run() -> None:
        broker = TrackingBroker()
        result = await run_agent_loop(
            broker=broker,
            model=RepeatedToolIdModel(
                [
                    {"action": "wait", "milliseconds": 0},
                    {"action": "done"},
                ]
            ),
            task_id="task-001",
            attempt_id="repeated-provider-tool-id",
            prompt="draw",
            reference_png=_PNG,
            max_actions=3,
            wall_timeout_s=10,
        )

        assert result.completed is True
        assert len(broker.request_ids) == 2
        assert len(set(broker.request_ids)) == 2

    asyncio.run(run())


def test_agent_loop_reports_nonfatal_post_action_instability_to_model() -> None:
    class UnstableReceiptBroker(FakeDesktopBroker):
        async def execute_action(self, session_id, action, **kwargs):
            result = await super().execute_action(session_id, action, **kwargs)
            return result.model_copy(
                update={"input_receipt": {"post_action_visual_stable": False}}
            )

    async def run() -> None:
        model = ScriptedModel(
            [
                {"action": "wait", "milliseconds": 0},
                {"action": "done"},
            ]
        )
        result = await run_agent_loop(
            broker=UnstableReceiptBroker(width=320, height=200),
            model=model,
            task_id="task-001",
            attempt_id="unstable-post-action",
            prompt="draw",
            reference_png=_PNG,
            max_actions=2,
            wall_timeout_s=10,
        )

        assert result.completed is True
        assert (
            "could not prove that whole-desktop pixels settled"
            in (model.observation_messages[0])
        )
        assert "freshest available frame" in model.observation_messages[0]

    asyncio.run(run())


def test_agent_loop_prevalidates_entire_batch_before_injecting_any_item() -> None:
    async def run() -> None:
        broker = FakeDesktopBroker(width=320, height=200)
        model = ScriptedModel(
            [
                {
                    "actions": [
                        {
                            "action": "click",
                            "x": 20,
                            "y": 20,
                            "intent": "Click the canvas.",
                        },
                        {
                            "action": "click",
                            "x": 999,
                            "y": 20,
                            "intent": "Click outside the desktop.",
                        },
                    ]
                },
                {"action": "done"},
            ]
        )
        result = await run_agent_loop(
            broker=broker,
            model=model,
            task_id="task-001",
            attempt_id="pilot-invalid-batch",
            prompt="draw",
            reference_png=_PNG,
            max_actions=5,
            wall_timeout_s=10,
        )

        assert result.completed is True
        assert result.invalid_actions == 1
        assert result.accepted_actions == 1
        assert [
            entry.action.action for entry in broker.audit_log(result.session_id)
        ] == ["done"]
        assert "before batch execution" in model.observation_messages[0]

    asyncio.run(run())


def test_agent_loop_recovers_from_broker_rejection_after_partial_batch() -> None:
    class DriftedBroker(FakeDesktopBroker):
        async def execute_action(self, session_id, action, **kwargs):
            if action.action == "type" and action.text == " ":
                raise BrokerError(
                    BrokerErrorCode.PROHIBITED_ACTION,
                    "simulated remote validator drift",
                )
            return await super().execute_action(session_id, action, **kwargs)

    async def run() -> None:
        broker = DriftedBroker(width=320, height=200)
        model = ScriptedModel(
            [
                {
                    "actions": [
                        {
                            "action": "click",
                            "x": 20,
                            "y": 20,
                            "intent": "Focus the command line.",
                        },
                        {
                            "action": "type",
                            "text": " ",
                            "intent": "Type one literal space.",
                        },
                        {
                            "action": "key",
                            "keys": ["ESC"],
                            "intent": "Cancel the current command.",
                        },
                    ]
                },
                {"actions": [{"action": "done", "intent": "Finish."}]},
            ]
        )
        result = await run_agent_loop(
            broker=broker,
            model=model,
            task_id="task-001",
            attempt_id="pilot-recover-partial-batch",
            prompt="draw",
            reference_png=_PNG,
            max_actions=5,
            wall_timeout_s=10,
        )

        assert result.completed is True
        assert result.stop_reason == "done"
        assert result.invalid_actions == 1
        assert result.accepted_actions == 2
        assert [
            entry.action.action for entry in broker.audit_log(result.session_id)
        ] == [
            "click",
            "done",
        ]
        assert model.observations[0][1] is True
        assert (
            "first 1 action(s) were executed exactly once"
            in (model.observation_messages[0])
        )
        assert "every later item were not executed" in model.observation_messages[0]

    asyncio.run(run())


def test_agent_loop_has_no_separate_per_tool_call_action_cap() -> None:
    async def run() -> None:
        broker = FakeDesktopBroker(width=320, height=200)
        model = ScriptedModel(
            [
                {
                    "actions": [
                        {
                            "action": "wait",
                            "milliseconds": 0,
                            "intent": f"Execute allowed batch item {index + 1}.",
                        }
                        for index in range(20)
                    ]
                },
                {
                    "actions": [
                        {
                            "action": "wait",
                            "milliseconds": 0,
                            "intent": f"Execute uncapped batch item {index + 1}.",
                        }
                        for index in range(21)
                    ]
                },
                {"action": "done"},
            ]
        )
        result = await run_agent_loop(
            broker=broker,
            model=model,
            task_id="task-001",
            attempt_id="pilot-unlimited-action-batch",
            prompt="draw",
            reference_png=_PNG,
            max_actions=45,
            wall_timeout_s=10,
        )

        assert result.completed is True
        assert result.invalid_actions == 0
        assert result.accepted_actions == 42
        assert len(broker.audit_log(result.session_id)) == 42
        assert all("discarded" not in message for message in model.observation_messages)

    asyncio.run(run())


def test_agent_loop_rejects_non_ascii_type_before_injecting_batch() -> None:
    async def run() -> None:
        broker = FakeDesktopBroker(width=320, height=200)
        model = ScriptedModel(
            [
                {
                    "actions": [
                        {
                            "action": "click",
                            "x": 20,
                            "y": 20,
                            "intent": "Focus the command line.",
                        },
                        {
                            "action": "type",
                            "text": "90°",
                            "intent": "Type a dimension label.",
                        },
                    ]
                },
                {"action": "done"},
            ]
        )
        result = await run_agent_loop(
            broker=broker,
            model=model,
            task_id="task-001",
            attempt_id="pilot-non-ascii-type",
            prompt="draw",
            reference_png=_PNG,
            max_actions=5,
            wall_timeout_s=10,
        )

        assert result.completed is True
        assert result.invalid_actions == 1
        assert result.accepted_actions == 1
        assert [
            entry.action.action for entry in broker.audit_log(result.session_id)
        ] == ["done"]
        assert "printable ASCII only" in model.observation_messages[0]

    asyncio.run(run())


def test_agent_loop_rejects_batch_larger_than_remaining_budget_without_partial_execution() -> (
    None
):
    async def run() -> None:
        broker = FakeDesktopBroker(width=320, height=200)
        model = ScriptedModel(
            [
                {
                    "actions": [
                        {
                            "action": "type",
                            "text": "LINE",
                            "intent": "Type LINE.",
                        },
                        {
                            "action": "key",
                            "keys": ["ENTER"],
                            "intent": "Submit LINE.",
                        },
                    ]
                },
                {"action": "done"},
            ]
        )
        result = await run_agent_loop(
            broker=broker,
            model=model,
            task_id="task-001",
            attempt_id="pilot-budgeted-batch",
            prompt="draw",
            reference_png=_PNG,
            max_actions=1,
            wall_timeout_s=10,
        )

        assert result.completed is True
        assert result.invalid_actions == 1
        assert result.accepted_actions == 1
        assert [
            entry.action.action for entry in broker.audit_log(result.session_id)
        ] == ["done"]

    asyncio.run(run())


def test_agent_loop_allows_done_only_as_final_batch_item() -> None:
    async def run() -> None:
        broker = FakeDesktopBroker(width=320, height=200)
        model = ScriptedModel(
            [
                {
                    "actions": [
                        {"action": "done", "intent": "Finish."},
                        {
                            "action": "wait",
                            "milliseconds": 1,
                            "intent": "Wait briefly.",
                        },
                    ]
                },
                {
                    "actions": [
                        {
                            "action": "type",
                            "text": "QSAVE",
                            "intent": "Type QSAVE.",
                        },
                        {
                            "action": "key",
                            "keys": ["ENTER"],
                            "intent": "Submit QSAVE.",
                        },
                        {"action": "done", "intent": "Finish."},
                    ]
                },
            ]
        )
        result = await run_agent_loop(
            broker=broker,
            model=model,
            task_id="task-001",
            attempt_id="pilot-final-done",
            prompt="draw",
            reference_png=_PNG,
            max_actions=4,
            wall_timeout_s=10,
        )

        assert result.completed is True
        assert result.invalid_actions == 1
        assert result.accepted_actions == 3
        assert [
            entry.action.action for entry in broker.audit_log(result.session_id)
        ] == [
            "type",
            "key",
            "done",
        ]

    asyncio.run(run())


def test_agent_wall_clock_starts_after_verified_handoff() -> None:
    class SlowResetBroker(FakeDesktopBroker):
        async def reset_session(self, session_id):
            await asyncio.sleep(0.04)
            return await super().reset_session(session_id)

    async def run() -> None:
        broker = SlowResetBroker(width=320, height=200)
        model = ScriptedModel([{"action": "done"}])
        result = await run_agent_loop(
            broker=broker,
            model=model,
            task_id="task-001",
            attempt_id="pilot-slow-handoff",
            prompt="draw",
            reference_png=_PNG,
            max_actions=1,
            wall_timeout_s=0.02,
        )

        assert result.completed is True
        assert result.stop_reason == "done"
        assert result.setup_elapsed_s >= 0.03
        assert result.rollout_elapsed_s < result.setup_elapsed_s

    asyncio.run(run())


def test_agent_loop_can_run_without_a_wall_clock_timeout() -> None:
    async def run() -> None:
        broker = FakeDesktopBroker(width=320, height=200)
        model = ScriptedModel(
            [{"action": "click", "x": 20, "y": 20}, {"action": "done"}]
        )
        result = await run_agent_loop(
            broker=broker,
            model=model,
            task_id="task-001",
            attempt_id="pilot-no-wall-timeout",
            prompt="draw",
            reference_png=_PNG,
            max_actions=10,
            wall_timeout_s=None,
        )

        assert result.completed is True
        assert result.stop_reason == "done"
        assert "9 actions remain. Inspect" in model.observation_messages[0]
        assert "seconds remain" not in model.observation_messages[0]

    asyncio.run(run())


def test_agent_loop_cancellation_still_checkpoints_retrieves_and_releases() -> None:
    class TrackingBroker(FakeDesktopBroker):
        def __init__(self) -> None:
            super().__init__(width=320, height=200)
            self.events_retrieved = False

        async def retrieve_events(self, session_id: str) -> bytes:
            self.events_retrieved = True
            return await super().retrieve_events(session_id)

    class BlockingModel(ScriptedModel):
        def __init__(self) -> None:
            super().__init__([])
            self.waiting = asyncio.Event()

        async def next_action(self, **_: Any) -> ModelTurn:
            self.waiting.set()
            await asyncio.Event().wait()
            raise AssertionError("unreachable")

    async def run() -> None:
        broker = TrackingBroker()
        model = BlockingModel()
        task = asyncio.create_task(
            run_agent_loop(
                broker=broker,
                model=model,
                task_id="task-001",
                attempt_id="pilot-operator-stop",
                prompt="draw",
                reference_png=_PNG,
                max_actions=10,
                wall_timeout_s=None,
            )
        )
        await model.waiting.wait()
        task.cancel()

        result = await task

        assert result.completed is False
        assert result.stop_reason == "operator_stopped"
        assert result.failure is None
        assert result.artifact is not None
        assert result.events == b""
        assert broker.events_retrieved is True
        assert result.frame_events
        assert result.status is not None
        assert result.status.checkpoint_saved is True
        assert result.session_id is not None
        assert (await broker.get_status(result.session_id)).state.value == "released"
        assert result.cleanup_errors == ()

    asyncio.run(run())


def test_agent_loop_rejects_a_stale_post_action_frame_as_infrastructure_error() -> None:
    class StaleFrameBroker(FakeDesktopBroker):
        async def execute_action(self, session_id, action, **kwargs):
            result = await super().execute_action(session_id, action, **kwargs)
            stale = result.screenshot.model_copy(
                update={"sequence": result.screenshot.sequence - 1}
            )
            return result.model_copy(update={"screenshot": stale})

    async def run() -> None:
        broker = StaleFrameBroker(width=320, height=200)
        model = ScriptedModel([{"action": "click", "x": 20, "y": 20}])
        result = await run_agent_loop(
            broker=broker,
            model=model,
            task_id="task-001",
            attempt_id="pilot-stale-frame",
            prompt="draw",
            reference_png=_PNG,
            max_actions=10,
            wall_timeout_s=10,
        )

        assert result.completed is False
        assert result.stop_reason == "controller_error"
        assert result.accepted_actions == 0
        assert result.failure is not None
        assert "fresh post-action screenshot" in result.failure

    asyncio.run(run())


def test_agent_loop_classifies_provider_payload_failure_as_model_error() -> None:
    class BrokenModel(ScriptedModel):
        async def next_action(self, **_: Any) -> ModelTurn:
            raise ValueError("model returned invalid computer-action arguments")

    async def run() -> None:
        result = await run_agent_loop(
            broker=FakeDesktopBroker(width=320, height=200),
            model=BrokenModel([]),
            task_id="task-001",
            attempt_id="provider-payload-failure",
            prompt="draw",
            reference_png=_PNG,
            max_actions=10,
            wall_timeout_s=10,
        )

        assert result.completed is False
        assert result.stop_reason == "model_error"
        assert result.failure is not None
        assert "invalid computer-action arguments" in result.failure

    asyncio.run(run())


def test_agent_loop_bounds_invalid_actions_and_still_retrieves_logs() -> None:
    async def run() -> None:
        broker = FakeDesktopBroker(width=320, height=200)
        model = ScriptedModel(
            [
                {"action": "type", "text": "bad\ncontrol"},
                {"action": "click", "x": 999, "y": 10},
            ]
        )
        result = await run_agent_loop(
            broker=broker,
            model=model,
            task_id="task-001",
            attempt_id="pilot-invalid",
            prompt="draw",
            reference_png=_PNG,
            max_actions=10,
            wall_timeout_s=10,
            invalid_action_limit=2,
        )
        assert result.completed is False
        assert result.stop_reason == "invalid_action_limit"
        assert result.invalid_actions == 2
        assert result.frame_events
        assert result.artifact is not None

    asyncio.run(run())


def test_agent_loop_returns_rejected_action_to_model_and_continues() -> None:
    async def run() -> None:
        broker = FakeDesktopBroker(width=320, height=200)
        model = ScriptedModel(
            [
                {"action": "click", "x": 999, "y": 10},
                {"action": "click", "x": 20, "y": 20},
                {"action": "done"},
            ]
        )
        result = await run_agent_loop(
            broker=broker,
            model=model,
            task_id="task-001",
            attempt_id="pilot-recover-invalid",
            prompt="draw",
            reference_png=_PNG,
            max_actions=10,
            wall_timeout_s=10,
            invalid_action_limit=3,
        )

        assert result.completed is True
        assert result.stop_reason == "done"
        assert result.invalid_actions == 1
        assert result.consecutive_invalid_actions == 0
        assert result.accepted_actions == 2
        assert model.observations[0][1] is True
        assert "Consecutive invalid action 1 of 3" in model.observation_messages[0]
        assert (
            "Action rejected by the desktop controller" in model.observation_messages[0]
        )

    asyncio.run(run())


def test_agent_loop_invalid_limit_counts_only_consecutive_invalid_turns() -> None:
    async def run() -> None:
        broker = FakeDesktopBroker(width=320, height=200)
        model = ScriptedModel(
            [
                {"action": "click", "x": 999, "y": 10},
                {"action": "click", "x": 20, "y": 20},
                {"action": "click", "x": 999, "y": 10},
                {"action": "done"},
            ]
        )
        progress: list[tuple[str, Any]] = []
        result = await run_agent_loop(
            broker=broker,
            model=model,
            task_id="task-001",
            attempt_id="pilot-consecutive-invalid",
            prompt="draw",
            reference_png=_PNG,
            max_actions=10,
            wall_timeout_s=10,
            invalid_action_limit=2,
            progress_callback=lambda event, payload: progress.append((event, payload)),
        )

        assert result.completed is True
        assert result.stop_reason == "done"
        assert result.invalid_actions == 2
        assert result.consecutive_invalid_actions == 0
        assert result.accepted_actions == 2
        assert [
            payload["consecutive_count"]
            for event, payload in progress
            if event == "invalid_action"
        ] == [1, 1]

    asyncio.run(run())


class FakeBedrockClient:
    def __init__(self) -> None:
        self.requests: list[dict[str, Any]] = []

    def converse(self, **request: Any) -> dict[str, Any]:
        self.requests.append(request)
        return {
            "stopReason": "tool_use",
            "output": {
                "message": {
                    "role": "assistant",
                    "content": [
                        {
                            "toolUse": {
                                "toolUseId": "tool-1",
                                "name": "computer_action",
                                "input": {
                                    "actions": [{"action": "done", "intent": "Finish."}]
                                },
                            }
                        },
                        {
                            "toolUse": {
                                "toolUseId": "tool-2",
                                "name": "computer_action",
                                "input": {"action": "wait", "milliseconds": 1},
                            }
                        },
                    ],
                }
            },
            "usage": {"inputTokens": 10, "outputTokens": 2, "totalTokens": 12},
        }


def test_bedrock_adapter_forces_one_normalized_action_without_broker_secrets() -> None:
    async def run() -> None:
        png = _PNG
        screenshot = DesktopScreenshot(
            session_id="session",
            sequence=0,
            captured_at="2026-07-15T00:00:00Z",
            width=320,
            height=200,
            autocad_maximized=True,
            png_sha256=hashlib.sha256(png).hexdigest(),
            png=png,
        )
        client = FakeBedrockClient()
        model = BedrockComputerUseModel(client=client)
        await model.start(
            prompt="draw",
            reference_png=png,
            screenshot=screenshot,
            max_actions=10,
            wall_timeout_s=60,
        )
        turn = await model.next_action(remaining_actions=10, remaining_time_s=60)
        assert turn.action_payload == {
            "actions": [{"action": "done", "intent": "Finish."}]
        }
        assert turn.discarded_tool_uses == 1
        assert sum("toolUse" in block for block in model.messages[-1]["content"]) == 1
        request = client.requests[0]
        assert request["toolConfig"]["toolChoice"]["tool"]["name"] == "computer_action"
        assert "broker" not in str(request).lower()
        await model.observe(
            tool_use_id=turn.tool_use_id,
            screenshot=screenshot,
            message="accepted",
            is_error=False,
        )
        assert "toolResult" in model.messages[-1]["content"][0]

    asyncio.run(run())


def test_bedrock_adapter_resets_ten_attempt_streaks_after_valid_action() -> None:
    class TransientError(Exception):
        status_code = 529

    class FlakyBedrockClient:
        def __init__(self) -> None:
            self.requests = 0
            self.transient_remaining = 9
            self.malformed_remaining = 9

        def converse(self, **request: Any) -> dict[str, Any]:
            del request
            self.requests += 1
            if self.transient_remaining:
                self.transient_remaining -= 1
                raise TransientError()
            if self.malformed_remaining:
                self.malformed_remaining -= 1
                return {
                    "stopReason": "end_turn",
                    "output": {
                        "message": {
                            "role": "assistant",
                            "content": [{"text": "no tool"}],
                        }
                    },
                    "usage": {},
                }
            self.transient_remaining = 9
            self.malformed_remaining = 9
            return {
                "stopReason": "tool_use",
                "output": {
                    "message": {
                        "role": "assistant",
                        "content": [
                            {
                                "toolUse": {
                                    "toolUseId": f"tool-{self.requests}",
                                    "name": "computer_action",
                                    "input": {
                                        "actions": [
                                            {
                                                "action": "done",
                                                "intent": "Finish.",
                                            }
                                        ]
                                    },
                                }
                            }
                        ],
                    }
                },
                "usage": {},
            }

    async def run() -> None:
        screenshot = DesktopScreenshot(
            session_id="session",
            sequence=0,
            captured_at="2026-07-19T00:00:00Z",
            width=320,
            height=200,
            autocad_maximized=True,
            png_sha256=hashlib.sha256(_PNG).hexdigest(),
            png=_PNG,
        )
        client = FlakyBedrockClient()
        model = BedrockComputerUseModel(
            client=client,
            retry_backoff_s=0,
        )
        await model.start(
            prompt="draw",
            reference_png=_PNG,
            screenshot=screenshot,
            max_actions=10,
            wall_timeout_s=None,
        )

        first = await model.next_action(remaining_actions=10, remaining_time_s=None)
        await model.observe(
            tool_use_id=first.tool_use_id,
            screenshot=screenshot,
            message="input injected",
            is_error=False,
        )
        second = await model.next_action(remaining_actions=9, remaining_time_s=None)

        assert first.request_attempts == 19
        assert second.request_attempts == 19
        assert first.response_validation_retries == 9
        assert second.response_validation_retries == 9
        assert first.model_input is not None
        assert second.model_input is not None
        assert first.model_input["transient_error_retries"] == 9
        assert second.model_input["transient_error_retries"] == 9

    asyncio.run(run())


def test_openai_adapter_uses_gpt_5_6_sol_xhigh_and_forced_action_tool() -> None:
    async def run() -> None:
        png = _PNG
        screenshot = DesktopScreenshot(
            session_id="session",
            sequence=0,
            captured_at="2026-07-15T00:00:00Z",
            width=320,
            height=200,
            autocad_maximized=True,
            png_sha256=hashlib.sha256(png).hexdigest(),
            png=png,
        )
        requests: list[dict[str, Any]] = []

        def handler(request: httpx.Request) -> httpx.Response:
            assert request.url == httpx.URL("https://openai.test/v1/responses")
            assert request.headers["authorization"] == "Bearer test-key"
            body = json.loads(request.content)
            requests.append(body)
            return httpx.Response(
                200,
                json={
                    "id": f"resp_{len(requests)}",
                    "status": "completed",
                    "output": [
                        {
                            "type": "function_call",
                            "id": f"fc_{len(requests)}",
                            "call_id": f"call_{len(requests)}",
                            "name": "computer_action",
                            "arguments": json.dumps(
                                {
                                    "actions": [
                                        {
                                            "action": "key",
                                            "keys": ["ENTER"],
                                            "intent": "Submit the current value.",
                                        }
                                    ]
                                }
                            ),
                        }
                    ],
                    "usage": {
                        "input_tokens": 10,
                        "input_tokens_details": {"cached_tokens": 6},
                        "prompt_tokens_details": {"cache_write_tokens": 8},
                        "output_tokens": 20,
                        "output_tokens_details": {"reasoning_tokens": 12},
                        "total_tokens": 30,
                    },
                },
            )

        model = OpenAIResponsesComputerUseModel(
            api_key="test-key",
            base_url="https://openai.test/v1",
            transport=httpx.MockTransport(handler),
        )
        await model.start(
            prompt="draw",
            reference_png=png,
            screenshot=screenshot,
            max_actions=10,
            wall_timeout_s=60,
        )
        first = await model.next_action(remaining_actions=10, remaining_time_s=60)
        assert first.action_payload == {
            "actions": [
                {
                    "action": "key",
                    "keys": ["ENTER"],
                    "intent": "Submit the current value.",
                }
            ]
        }
        assert first.request_attempts == 1
        assert first.usage == {
            "input_tokens": 10,
            "cached_input_tokens": 6,
            "cache_write_input_tokens": 8,
            "output_tokens": 20,
            "reasoning_output_tokens": 12,
            "total_tokens": 30,
        }
        assert first.model_input is not None
        assert first.model_input["model"] == "gpt-5.6-sol"
        assert "test-key" not in json.dumps(first.model_input)
        assert "data:image/png;base64" not in json.dumps(first.model_input)
        input_images = [
            block
            for block in first.model_input["input"][0]["content"]
            if block["type"] == "input_image"
        ]
        assert [block["artifact_path"] for block in input_images] == [
            "reference.png",
            "frames/frame_0000.png",
        ]
        request = requests[0]
        assert request["model"] == "gpt-5.6-sol"
        assert request["reasoning"] == {"effort": "xhigh"}
        assert request["parallel_tool_calls"] is False
        assert request["tool_choice"] == {
            "type": "function",
            "name": "computer_action",
        }
        assert request["tools"][0]["name"] == "computer_action"
        assert request["tools"][0]["parameters"]["required"] == ["actions"]
        assert (
            "maxItems" not in request["tools"][0]["parameters"]["properties"]["actions"]
        )
        assert "no separate per-tool-call action cap" in request["instructions"]
        assert "temperature" not in request
        assert request["store"] is False
        assert (
            "Keyboard input goes to whichever control currently has focus"
            in request["instructions"]
        )
        assert "click the intended target" in request["instructions"]
        assert (
            "reference drawing and task statement remain available on every turn"
            in request["instructions"]
        )
        assert "supplied task statement is authoritative" in request["instructions"]
        assert (
            "inspect it from the views appropriate to the work"
            in request["instructions"]
        )
        assert "do not receive screenshots, tool results" in request["instructions"]
        assert "immediately and sequentially" in request["instructions"]
        image_blocks = [
            block
            for block in request["input"][0]["content"]
            if block["type"] == "input_image"
        ]
        assert len(image_blocks) == 2
        assert all(
            block["image_url"].startswith("data:image/png;base64,")
            for block in image_blocks
        )
        assert "test-key" not in json.dumps(request)

        await model.observe(
            tool_use_id=first.tool_use_id,
            screenshot=screenshot,
            message="accepted",
            is_error=False,
        )
        await model.next_action(remaining_actions=9, remaining_time_s=55)
        first_content = requests[0]["input"][0]["content"]
        second_content = requests[1]["input"][0]["content"]
        assert first_content[:3] == second_content[:3]
        second_text = second_content[3]["text"]
        assert "Remaining budget: 9 actions, 55.0 seconds." in second_text
        assert '"action":{"actions":[{"action":"key"' in second_text
        assert '"controller_result":"input_injected"' in second_text
        assert '"screen_changed":false' in second_text
        assert '"before_frame":{"sequence":0' in second_text
        assert '"after_frame":{"sequence":0' in second_text
        second_images = [
            block
            for block in requests[1]["input"][0]["content"]
            if block["type"] == "input_image"
        ]
        assert len(second_images) == 2
        assert second_images[0]["image_url"] == image_blocks[0]["image_url"]

    asyncio.run(run())


def test_openai_adapter_retries_transient_response_before_desktop_action() -> None:
    async def run() -> None:
        screenshot = DesktopScreenshot(
            session_id="session",
            sequence=0,
            captured_at="2026-07-15T00:00:00Z",
            width=320,
            height=200,
            autocad_maximized=True,
            png_sha256=hashlib.sha256(_PNG).hexdigest(),
            png=_PNG,
        )
        calls = 0

        def handler(_: httpx.Request) -> httpx.Response:
            nonlocal calls
            calls += 1
            if calls == 1:
                return httpx.Response(503, json={"error": {"message": "retry"}})
            return httpx.Response(
                200,
                json={
                    "status": "completed",
                    "output": [
                        {
                            "type": "function_call",
                            "call_id": "call-retried",
                            "name": "computer_action",
                            "arguments": '{"action":"done"}',
                        }
                    ],
                    "usage": {"total_tokens": 5},
                },
            )

        model = OpenAIResponsesComputerUseModel(
            api_key="test-key",
            base_url="https://openai.test/v1",
            transport=httpx.MockTransport(handler),
            retry_backoff_s=0,
        )
        await model.start(
            prompt="draw",
            reference_png=_PNG,
            screenshot=screenshot,
            max_actions=10,
            wall_timeout_s=60,
        )

        turn = await model.next_action(remaining_actions=10, remaining_time_s=60)

        assert turn.action_payload == {"action": "done"}
        assert turn.request_attempts == 2
        assert calls == 2

    asyncio.run(run())


def test_responses_adapters_reset_ten_attempt_streaks_after_valid_action() -> None:
    async def run() -> None:
        screenshot = DesktopScreenshot(
            session_id="session",
            sequence=0,
            captured_at="2026-07-19T00:00:00Z",
            width=320,
            height=200,
            autocad_maximized=True,
            png_sha256=hashlib.sha256(_PNG).hexdigest(),
            png=_PNG,
        )
        transient_remaining = 9
        malformed_remaining = 9
        requests = 0

        def handler(_: httpx.Request) -> httpx.Response:
            nonlocal transient_remaining, malformed_remaining, requests
            requests += 1
            if transient_remaining:
                transient_remaining -= 1
                return httpx.Response(529, json={"error": {"message": "overloaded"}})
            if malformed_remaining:
                malformed_remaining -= 1
                return httpx.Response(
                    200,
                    json={"status": "completed", "output": [], "usage": {}},
                )
            transient_remaining = 9
            malformed_remaining = 9
            return httpx.Response(
                200,
                json={
                    "status": "completed",
                    "output": [
                        {
                            "type": "function_call",
                            "call_id": f"call-{requests}",
                            "name": "computer_action",
                            "arguments": (
                                '{"actions":[{"action":"done","intent":"Finish."}]}'
                            ),
                        }
                    ],
                    "usage": {},
                },
            )

        model = OpenAIResponsesComputerUseModel(
            api_key="test-key",
            base_url="https://openai.test/v1",
            transport=httpx.MockTransport(handler),
            retry_backoff_s=0,
        )
        await model.start(
            prompt="draw",
            reference_png=_PNG,
            screenshot=screenshot,
            max_actions=10,
            wall_timeout_s=None,
        )

        first = await model.next_action(remaining_actions=10, remaining_time_s=None)
        await model.observe(
            tool_use_id=first.tool_use_id,
            screenshot=screenshot,
            message="input injected",
            is_error=False,
        )
        second = await model.next_action(remaining_actions=9, remaining_time_s=None)

        assert first.request_attempts == 19
        assert second.request_attempts == 19
        assert first.response_validation_retries == 9
        assert second.response_validation_retries == 9
        assert first.model_input is not None
        assert second.model_input is not None
        assert first.model_input["transient_error_retries"] == 9
        assert second.model_input["transient_error_retries"] == 9

    asyncio.run(run())


def test_openrouter_adapter_caps_reasoning_and_honors_provider_retry() -> None:
    async def run() -> None:
        screenshot = DesktopScreenshot(
            session_id="session",
            sequence=0,
            captured_at="2026-07-15T00:00:00Z",
            width=320,
            height=200,
            autocad_maximized=True,
            png_sha256=hashlib.sha256(_PNG).hexdigest(),
            png=_PNG,
        )
        requests: list[dict[str, Any]] = []

        def handler(request: httpx.Request) -> httpx.Response:
            requests.append(json.loads(request.content))
            if len(requests) == 1:
                return httpx.Response(
                    429,
                    headers={"Retry-After": "0"},
                    json={
                        "error": {
                            "message": "Provider returned error",
                            "metadata": {
                                "provider_name": "Moonshot AI",
                                "retry_after_seconds": 0,
                            },
                        }
                    },
                )
            return httpx.Response(
                200,
                json={
                    "status": "completed",
                    "output": [
                        {
                            "type": "function_call",
                            "call_id": "call-after-provider-retry",
                            "name": "computer_action",
                            "arguments": (
                                '{"actions":[{"action":"done","intent":"Finish."}]}'
                            ),
                        }
                    ],
                    "usage": {"total_tokens": 5},
                },
            )

        model = OpenRouterResponsesComputerUseModel(
            api_key="test-key",
            model_id="moonshotai/kimi-k3",
            reasoning_max_tokens=512,
            tool_choice_mode="required",
            transport=httpx.MockTransport(handler),
            retry_backoff_s=0,
        )
        await model.start(
            prompt="draw",
            reference_png=_PNG,
            screenshot=screenshot,
            max_actions=10,
            wall_timeout_s=None,
        )

        turn = await model.next_action(remaining_actions=10, remaining_time_s=None)

        assert turn.request_attempts == 2
        assert turn.action_payload["actions"][0]["action"] == "done"
        assert requests[0]["reasoning"] == {
            "enabled": True,
            "max_tokens": 512,
        }
        assert "max_output_tokens" not in requests[0]
        assert requests[0]["tool_choice"] == "required"
        assert requests[0]["session_id"] == requests[0]["prompt_cache_key"]
        assert requests[0]["session_id"].startswith("autocad-bench-")
        assert "Remaining budget:" not in requests[0]["instructions"]
        assert requests[0]["input"][0]["content"][2] == {
            "type": "input_text",
            "text": (
                "The task instructions and fixed reference drawing above are "
                "reusable context for every turn in this rollout."
            ),
        }
        assert "cache_control" not in requests[0]

        await model.observe(
            tool_use_id=turn.tool_use_id,
            screenshot=screenshot,
            message="accepted",
            is_error=False,
        )
        await model.next_action(remaining_actions=9, remaining_time_s=None)
        first_success = requests[1]
        next_turn = requests[2]
        assert first_success["session_id"] == next_turn["session_id"]
        assert first_success["prompt_cache_key"] == next_turn["prompt_cache_key"]
        assert next_turn["input"][0] == first_success["input"][0]
        assert [item["type"] for item in next_turn["input"]] == [
            "message",
            "function_call",
            "function_call_output",
            "message",
        ]
        assert next_turn["input"][1]["call_id"] == turn.tool_use_id
        assert next_turn["input"][2]["call_id"] == turn.tool_use_id
        outcome = json.loads(next_turn["input"][2]["output"])
        assert outcome["controller_result"] == "input_injected"
        assert outcome["screen_changed"] is False
        assert next_turn["input"][3]["content"][0]["text"] == (
            "Remaining budget: 9 actions."
        )

    asyncio.run(run())


def test_openrouter_adapter_sends_literal_auto_tool_choice() -> None:
    async def run() -> None:
        screenshot = DesktopScreenshot(
            session_id="session",
            sequence=0,
            captured_at="2026-07-15T00:00:00Z",
            width=320,
            height=200,
            autocad_maximized=True,
            png_sha256=hashlib.sha256(_PNG).hexdigest(),
            png=_PNG,
        )
        requests: list[dict[str, Any]] = []

        def handler(request: httpx.Request) -> httpx.Response:
            requests.append(json.loads(request.content))
            return httpx.Response(
                200,
                json={
                    "status": "completed",
                    "output": [
                        {
                            "type": "function_call",
                            "call_id": "call-auto",
                            "name": "computer_action",
                            "arguments": (
                                '{"actions":[{"action":"done","intent":"Finish."}]}'
                            ),
                        }
                    ],
                    "usage": {"total_tokens": 5},
                },
            )

        model = OpenRouterResponsesComputerUseModel(
            api_key="test-key",
            model_id="z-ai/glm-5v-turbo",
            tool_choice_mode="auto",
            transport=httpx.MockTransport(handler),
            retry_backoff_s=0,
        )
        await model.start(
            prompt="draw",
            reference_png=_PNG,
            screenshot=screenshot,
            max_actions=10,
            wall_timeout_s=None,
        )

        turn = await model.next_action(
            remaining_actions=10,
            remaining_time_s=None,
        )

        assert turn.action_payload["actions"][0]["action"] == "done"
        assert requests[0]["tool_choice"] == "auto"

    asyncio.run(run())


def test_mantle_chat_adapter_sends_images_strict_tools_and_compacts_history() -> None:
    from botocore.credentials import Credentials

    class FakeSession:
        def get_credentials(self) -> Credentials:
            return Credentials("access", "secret", "token")

    async def run() -> None:
        screenshot = DesktopScreenshot(
            session_id="mantle-chat-session",
            sequence=0,
            captured_at="2026-07-19T00:00:00Z",
            width=320,
            height=200,
            autocad_maximized=True,
            png_sha256=hashlib.sha256(_PNG).hexdigest(),
            png=_PNG,
        )
        requests: list[dict[str, Any]] = []

        def handler(request: httpx.Request) -> httpx.Response:
            assert request.url == httpx.URL("https://mantle.test/v1/chat/completions")
            assert request.headers["authorization"].startswith("AWS4-HMAC-SHA256")
            payload = json.loads(request.content)
            requests.append(payload)
            call_number = len(requests)
            return httpx.Response(
                200,
                json={
                    "choices": [
                        {
                            "finish_reason": "tool_calls",
                            "message": {
                                "role": "assistant",
                                "content": None,
                                "tool_calls": [
                                    {
                                        "id": f"mantle-call-{call_number}",
                                        "type": "function",
                                        "function": {
                                            "name": "computer_action",
                                            "arguments": (
                                                '{"actions":[{"action":"wait",'
                                                '"intent":"Observe the desktop.",'
                                                '"milliseconds":0}]}'
                                            ),
                                        },
                                    }
                                ],
                            },
                        }
                    ],
                    "usage": {
                        "prompt_tokens": 100,
                        "completion_tokens": 10,
                        "total_tokens": 110,
                        "prompt_tokens_details": {"cached_tokens": 20},
                    },
                },
            )

        model = MantleChatCompletionsComputerUseModel(
            model_id="moonshotai.kimi-k2.5",
            base_url="https://mantle.test/v1",
            session=FakeSession(),
            compaction_messages=8,
            compaction_recent_outcomes=2,
            retry_backoff_s=0,
            transport=httpx.MockTransport(handler),
        )
        await model.start(
            prompt="draw",
            reference_png=_PNG,
            screenshot=screenshot,
            max_actions=10,
            wall_timeout_s=None,
        )

        turns: list[ModelTurn] = []
        for index in range(4):
            turn = await model.next_action(
                remaining_actions=10 - index,
                remaining_time_s=None,
            )
            turns.append(turn)
            await model.observe(
                tool_use_id=turn.tool_use_id,
                screenshot=screenshot,
                message=f"accepted-{index}",
                is_error=False,
            )

        first = requests[0]
        assert first["model"] == "moonshotai.kimi-k2.5"
        assert first["stream"] is False
        assert first["parallel_tool_calls"] is False
        assert first["tools"][0]["function"]["strict"] is True
        assert first["tool_choice"] == {
            "type": "function",
            "function": {"name": "computer_action"},
        }
        first_content = first["messages"][1]["content"]
        assert sum(block["type"] == "image_url" for block in first_content) == 2
        assert turns[0].usage["cached_input_tokens"] == 20

        compacted = requests[3]
        assert [item["role"] for item in compacted["messages"]] == [
            "system",
            "user",
        ]
        content = compacted["messages"][1]["content"]
        assert sum(block["type"] == "image_url" for block in content) == 2
        summary = next(
            block["text"]
            for block in content
            if block["type"] == "text"
            and block["text"].startswith("PROMPT HISTORY COMPACTION")
        )
        assert '"completed_controller_outcomes":3' in summary
        assert '"older_outcomes_compacted":1' in summary
        assert "accepted-2" in summary
        metadata = turns[3].model_input["history_compaction"]
        assert metadata["round"] == 1
        assert metadata["strategy"] == "deterministic_controller_checkpoint"
        assert turns[3].model_input["prompt_cache"] == {
            "strategy": "byte_stable_append_only_prefix",
            "explicit_provider_flag": False,
            "usage_telemetry": "prompt_tokens_details.cached_tokens",
        }

    asyncio.run(run())


def test_mantle_kimi_compacts_before_complete_request_exceeds_safe_limit() -> None:
    from botocore.credentials import Credentials

    class FakeSession:
        def get_credentials(self) -> Credentials:
            return Credentials("access", "secret", "token")

    async def run() -> None:
        reference_png = _PNG + (b"reference" * 30_000)
        desktop_png = _PNG + (b"desktop" * 35_000)

        def screenshot(sequence: int) -> DesktopScreenshot:
            return DesktopScreenshot(
                session_id="mantle-kimi-long-history",
                sequence=sequence,
                captured_at="2026-07-19T00:00:00Z",
                width=1920,
                height=1080,
                autocad_maximized=True,
                png_sha256=hashlib.sha256(desktop_png).hexdigest(),
                png=desktop_png,
            )

        request_sizes: list[int] = []
        image_counts: list[int] = []
        message_roles: list[list[str]] = []

        def handler(request: httpx.Request) -> httpx.Response:
            request_sizes.append(len(request.content))
            payload = json.loads(request.content)
            message_roles.append([item["role"] for item in payload["messages"]])
            image_counts.append(
                sum(
                    block.get("type") == "image_url"
                    for message in payload["messages"]
                    if isinstance(message.get("content"), list)
                    for block in message["content"]
                    if isinstance(block, dict)
                )
            )
            call_number = len(request_sizes)
            return httpx.Response(
                200,
                json={
                    "choices": [
                        {
                            "finish_reason": "tool_calls",
                            "message": {
                                "role": "assistant",
                                "content": None,
                                "tool_calls": [
                                    {
                                        "id": f"mantle-long-{call_number}",
                                        "type": "function",
                                        "function": {
                                            "name": "computer_action",
                                            "arguments": (
                                                '{"actions":[{"action":"wait",'
                                                '"intent":"Observe.",'
                                                '"milliseconds":0}]}'
                                            ),
                                        },
                                    }
                                ],
                            },
                        }
                    ],
                    "usage": {"total_tokens": 1},
                },
            )

        safe_limit = 3 * 1024 * 1024
        model = MantleChatCompletionsComputerUseModel(
            model_id="moonshotai.kimi-k2.5",
            base_url="https://mantle.test/v1",
            session=FakeSession(),
            # Isolate the serialized-body guard from the message-count guard.
            compaction_messages=10_000,
            compaction_recent_outcomes=4,
            request_soft_limit_bytes=safe_limit,
            retry_backoff_s=0,
            transport=httpx.MockTransport(handler),
        )
        await model.start(
            prompt="draw",
            reference_png=reference_png,
            screenshot=screenshot(0),
            max_actions=200,
            wall_timeout_s=None,
        )

        turns: list[ModelTurn] = []
        for index in range(100):
            turn = await model.next_action(
                remaining_actions=200 - index,
                remaining_time_s=None,
            )
            turns.append(turn)
            await model.observe(
                tool_use_id=turn.tool_use_id,
                screenshot=screenshot(index + 1),
                message=f"accepted-{index}",
                is_error=False,
            )

        assert len(request_sizes) == 100
        assert max(request_sizes) <= safe_limit
        assert all(
            turn.model_input["serialized_request_bytes"] <= safe_limit
            for turn in turns
        )
        assert model.compaction_round > 0
        assert model.serialized_body_guard_compactions > 0
        compacted_indexes = [
            index
            for index, roles in enumerate(message_roles)
            if index > 0 and roles == ["system", "user"]
        ]
        assert compacted_indexes
        assert all(image_counts[index] == 2 for index in compacted_indexes)

    asyncio.run(run())


def test_mantle_kimi_retries_body_limit_400_after_forced_compaction() -> None:
    from botocore.credentials import Credentials

    class FakeSession:
        def get_credentials(self) -> Credentials:
            return Credentials("access", "secret", "token")

    async def run() -> None:
        screenshot = DesktopScreenshot(
            session_id="mantle-kimi-body-limit-retry",
            sequence=0,
            captured_at="2026-07-19T00:00:00Z",
            width=320,
            height=200,
            autocad_maximized=True,
            png_sha256=hashlib.sha256(_PNG).hexdigest(),
            png=_PNG,
        )
        requests: list[dict[str, Any]] = []

        def handler(request: httpx.Request) -> httpx.Response:
            payload = json.loads(request.content)
            requests.append(payload)
            call_number = len(requests)
            if call_number == 2:
                return httpx.Response(
                    400,
                    text=(
                        "Failed to buffer the request body: "
                        "length limit exceeded"
                    ),
                )
            return httpx.Response(
                200,
                json={
                    "choices": [
                        {
                            "finish_reason": "tool_calls",
                            "message": {
                                "role": "assistant",
                                "content": None,
                                "tool_calls": [
                                    {
                                        "id": f"mantle-retry-{call_number}",
                                        "type": "function",
                                        "function": {
                                            "name": "computer_action",
                                            "arguments": (
                                                '{"actions":[{"action":"wait",'
                                                '"intent":"Observe.",'
                                                '"milliseconds":0}]}'
                                            ),
                                        },
                                    }
                                ],
                            },
                        }
                    ],
                    "usage": {"total_tokens": 1},
                },
            )

        model = MantleChatCompletionsComputerUseModel(
            model_id="moonshotai.kimi-k2.5",
            base_url="https://mantle.test/v1",
            session=FakeSession(),
            compaction_messages=10_000,
            retry_backoff_s=0,
            transport=httpx.MockTransport(handler),
        )
        await model.start(
            prompt="draw",
            reference_png=_PNG,
            screenshot=screenshot,
            max_actions=10,
            wall_timeout_s=None,
        )

        first = await model.next_action(
            remaining_actions=10,
            remaining_time_s=None,
        )
        await model.observe(
            tool_use_id=first.tool_use_id,
            screenshot=screenshot,
            message="accepted-first",
            is_error=False,
        )
        recovered = await model.next_action(
            remaining_actions=9,
            remaining_time_s=None,
        )

        assert recovered.action_payload["actions"][0]["action"] == "wait"
        assert len(requests) == 3
        assert [item["role"] for item in requests[2]["messages"]] == [
            "system",
            "user",
        ]
        compacted_content = requests[2]["messages"][1]["content"]
        assert sum(
            block["type"] == "image_url" for block in compacted_content
        ) == 2
        assert model.compaction_round == 1
        assert model.provider_body_limit_retries == 1
        assert recovered.model_input["history_compaction"]["round"] == 1
        assert recovered.model_input["history_compaction"][
            "provider_body_limit_retries"
        ] == 1

    asyncio.run(run())


def test_mantle_chat_computer_use_is_scoped_to_kimi_k2_5() -> None:
    with pytest.raises(ValueError, match="scoped to moonshotai.kimi-k2.5"):
        MantleChatCompletionsComputerUseModel(model_id="other.model")


def test_fireworks_fast_chat_uses_images_tools_cache_affinity_and_append_only_history() -> None:
    async def run() -> None:
        screenshot = DesktopScreenshot(
            session_id="fireworks-session",
            sequence=0,
            captured_at="2026-07-20T00:00:00Z",
            width=320,
            height=200,
            autocad_maximized=True,
            png_sha256=hashlib.sha256(_PNG).hexdigest(),
            png=_PNG,
        )
        requests: list[dict[str, Any]] = []
        request_headers: list[dict[str, str]] = []

        def handler(request: httpx.Request) -> httpx.Response:
            assert request.url == httpx.URL(
                "https://fireworks.test/inference/v1/chat/completions"
            )
            request_headers.append(dict(request.headers))
            payload = json.loads(request.content)
            requests.append(payload)
            call_number = len(requests)
            return httpx.Response(
                200,
                headers={"x-ratelimit-limit-tokens": "21600000"},
                json={
                    "perf_metrics": {"time_to_first_token_ms": 12},
                    "choices": [
                        {
                            "finish_reason": "tool_calls",
                            "message": {
                                "role": "assistant",
                                "content": None,
                                "reasoning_content": f"reasoning-{call_number}",
                                "tool_calls": [
                                    {
                                        "id": f"fireworks-call-{call_number}",
                                        "type": "function",
                                        "function": {
                                            "name": "computer_action",
                                            "arguments": (
                                                '{"actions":[{"action":"wait",'
                                                '"intent":"Observe.",'
                                                '"milliseconds":0}]}'
                                            ),
                                        },
                                    }
                                ],
                            },
                        }
                    ],
                    "usage": {
                        "prompt_tokens": 100,
                        "completion_tokens": 10,
                        "total_tokens": 110,
                        "prompt_tokens_details": {"cached_tokens": 64},
                    },
                },
            )

        model = FireworksChatCompletionsComputerUseModel(
            api_key="fireworks-secret",
            base_url="https://fireworks.test/inference/v1",
            retry_backoff_s=0,
            transport=httpx.MockTransport(handler),
        )
        await model.start(
            prompt="draw",
            reference_png=_PNG,
            screenshot=screenshot,
            max_actions=10,
            wall_timeout_s=None,
        )

        first = await model.next_action(
            remaining_actions=10,
            remaining_time_s=None,
        )
        await model.observe(
            tool_use_id=first.tool_use_id,
            screenshot=screenshot.model_copy(update={"sequence": 1}),
            message="accepted",
            is_error=False,
        )
        second = await model.next_action(
            remaining_actions=9,
            remaining_time_s=None,
        )

        assert len(requests) == 2
        first_request, second_request = requests
        assert first_request["model"] == (
            "accounts/fireworks/routers/kimi-k2p6-fast"
        )
        assert "service_tier" not in first_request
        assert first_request["max_tokens"] == 2048
        assert first_request["thinking"] == {
            "type": "enabled",
            "budget_tokens": 1024,
        }
        assert "reasoning_effort" not in first_request
        assert first_request["perf_metrics_in_response"] is True
        assert first_request["tools"][0]["function"]["strict"] is True
        assert first_request["tool_choice"] == {
            "type": "function",
            "function": {"name": "computer_action"},
        }
        first_content = first_request["messages"][1]["content"]
        assert sum(block["type"] == "image_url" for block in first_content) == 2
        assert [message["role"] for message in second_request["messages"]] == [
            "system",
            "user",
            "assistant",
            "tool",
            "user",
        ]
        assert first_request["messages"] == second_request["messages"][:2]
        assert second_request["messages"][2]["reasoning_content"] == "reasoning-1"
        assert request_headers[0]["authorization"] == "Bearer fireworks-secret"
        assert (
            request_headers[0]["x-multi-turn-session-id"]
            == request_headers[1]["x-multi-turn-session-id"]
            == request_headers[0]["x-session-affinity"]
            == first_request["prompt_cache_key"]
        )
        assert first.usage["cached_input_tokens"] == 64
        assert first.model_input["history_mode"] == "fireworks_chat_completions"
        assert first.model_input["provider_metadata"]["serving_path_requested"] == (
            "fast_router"
        )
        assert first.model_input["provider_metadata"]["rate_limit_headers"] == {
            "x-ratelimit-limit-tokens": "21600000"
        }
        assert second.action_payload["actions"][0]["action"] == "wait"

    asyncio.run(run())


def test_fireworks_retries_checkpoint_swap_425_consecutively() -> None:
    async def run() -> None:
        screenshot = DesktopScreenshot(
            session_id="fireworks-425",
            sequence=0,
            captured_at="2026-07-20T00:00:00Z",
            width=320,
            height=200,
            autocad_maximized=True,
            png_sha256=hashlib.sha256(_PNG).hexdigest(),
            png=_PNG,
        )
        attempts = 0

        def handler(_request: httpx.Request) -> httpx.Response:
            nonlocal attempts
            attempts += 1
            if attempts == 1:
                return httpx.Response(425, text="checkpoint swap in progress")
            return httpx.Response(
                200,
                json={
                    "choices": [
                        {
                            "finish_reason": "tool_calls",
                            "message": {
                                "role": "assistant",
                                "content": None,
                                "tool_calls": [
                                    {
                                        "id": "fireworks-call",
                                        "type": "function",
                                        "function": {
                                            "name": "computer_action",
                                            "arguments": '{"actions":[{"action":"done","intent":"Done."}]}',
                                        },
                                    }
                                ],
                            },
                        }
                    ],
                    "usage": {"total_tokens": 1},
                },
            )

        model = FireworksChatCompletionsComputerUseModel(
            api_key="secret",
            base_url="https://fireworks.test/inference/v1",
            retry_backoff_s=0,
            transport=httpx.MockTransport(handler),
        )
        await model.start(
            prompt="draw",
            reference_png=_PNG,
            screenshot=screenshot,
            max_actions=10,
            wall_timeout_s=None,
        )
        turn = await model.next_action(
            remaining_actions=10,
            remaining_time_s=None,
        )
        assert attempts == 2
        assert turn.request_attempts == 2
        assert turn.model_input["transient_error_retries"] == 1

    asyncio.run(run())


def test_fireworks_qwen_uses_preserved_reasoning_and_standard_serving() -> None:
    model = FireworksChatCompletionsComputerUseModel(
        api_key="fireworks-secret",
        model_id="accounts/fireworks/models/qwen3p7-plus",
        base_url="https://fireworks.test/inference/v1",
    )
    model.cache_affinity_key = "qwen-session"

    assert model._additional_request_fields() == {
        "thinking": {"type": "enabled", "budget_tokens": 1024},
        "reasoning_history": "preserved",
        "prompt_cache_key": "qwen-session",
        "perf_metrics_in_response": True,
    }
    metadata = model._provider_trace_metadata(
        httpx.Response(200),
        {},
    )
    assert metadata["serving_path_requested"] == "standard"


def test_openrouter_anthropic_route_enables_strict_output_beta() -> None:
    anthropic = OpenRouterResponsesComputerUseModel(
        api_key="test-key",
        model_id="anthropic/claude-sonnet-5",
    )
    non_anthropic = OpenRouterResponsesComputerUseModel(
        api_key="test-key",
        model_id="google/gemini-3.5-flash",
    )

    assert anthropic._additional_request_headers() == {
        "x-anthropic-beta": "structured-outputs-2025-11-13"
    }
    assert non_anthropic._additional_request_headers() == {}
    anthropic_tool = anthropic._computer_action_tool()
    assert anthropic_tool["strict"] is True
    assert '"maximum"' not in json.dumps(anthropic_tool["parameters"])
    assert '"maxItems"' not in json.dumps(anthropic_tool["parameters"])
    assert '"maximum"' in json.dumps(non_anthropic._computer_action_tool())
    anthropic.screenshot = DesktopScreenshot(
        session_id="anthropic-cache-session",
        sequence=0,
        captured_at="2026-07-19T00:00:00Z",
        width=320,
        height=200,
        autocad_maximized=True,
        png_sha256=hashlib.sha256(_PNG).hexdigest(),
        png=_PNG,
    )
    cache_fields = anthropic._prompt_cache_request_fields()
    assert cache_fields["session_id"] == cache_fields["prompt_cache_key"]
    assert cache_fields["cache_control"] == {"type": "ephemeral"}
    assert "prompt_cache_breakpoint" not in anthropic._static_context_marker()


def test_openrouter_adapter_retries_malformed_json_without_desktop_action() -> None:
    async def run() -> None:
        screenshot = DesktopScreenshot(
            session_id="session",
            sequence=0,
            captured_at="2026-07-15T00:00:00Z",
            width=320,
            height=200,
            autocad_maximized=True,
            png_sha256=hashlib.sha256(_PNG).hexdigest(),
            png=_PNG,
        )
        requests: list[httpx.Request] = []

        def handler(request: httpx.Request) -> httpx.Response:
            requests.append(request)
            if len(requests) == 1:
                return httpx.Response(
                    200,
                    headers={
                        "Content-Type": "application/json",
                        "X-Generation-Id": "gen-truncated",
                    },
                    content=b'{"status":"completed","output":',
                )
            return httpx.Response(
                200,
                json={
                    "status": "completed",
                    "output": [
                        {
                            "type": "function_call",
                            "call_id": "call-after-json-retry",
                            "name": "computer_action",
                            "arguments": '{"actions":[{"action":"done"}]}',
                        }
                    ],
                    "usage": {"total_tokens": 5},
                },
            )

        model = OpenRouterResponsesComputerUseModel(
            api_key="test-key",
            model_id="moonshotai/kimi-k3",
            transport=httpx.MockTransport(handler),
            retry_backoff_s=0,
        )
        await model.start(
            prompt="draw",
            reference_png=_PNG,
            screenshot=screenshot,
            max_actions=10,
            wall_timeout_s=None,
        )

        turn = await model.next_action(remaining_actions=10, remaining_time_s=None)

        assert turn.request_attempts == 2
        assert turn.action_payload["actions"][0]["action"] == "done"
        assert all(
            request.headers["accept"] == "text/event-stream" for request in requests
        )
        assert all(
            json.loads(request.content)["stream"] is True for request in requests
        )

    asyncio.run(run())


def test_openrouter_adapter_decodes_sse_terminal_response() -> None:
    async def run() -> None:
        screenshot = DesktopScreenshot(
            session_id="session",
            sequence=0,
            captured_at="2026-07-15T00:00:00Z",
            width=320,
            height=200,
            autocad_maximized=True,
            png_sha256=hashlib.sha256(_PNG).hexdigest(),
            png=_PNG,
        )

        def handler(request: httpx.Request) -> httpx.Response:
            assert json.loads(request.content)["stream"] is True
            completed = {
                "type": "response.done",
                "response": {
                    "status": "completed",
                    "output": [
                        {
                            "type": "function_call",
                            "call_id": "call-from-sse",
                            "name": "computer_action",
                            "arguments": '{"actions":[{"action":"done"}]}',
                        }
                    ],
                    "usage": {"total_tokens": 7},
                },
            }
            body = (
                ": OPENROUTER PROCESSING\n\n"
                'data: {"type":"response.created"}\n\n'
                f"data: {json.dumps(completed)}\n\n"
                "data: [DONE]\n\n"
            )
            return httpx.Response(
                200,
                headers={"Content-Type": "text/event-stream"},
                text=body,
            )

        model = OpenRouterResponsesComputerUseModel(
            api_key="test-key",
            model_id="moonshotai/kimi-k3",
            transport=httpx.MockTransport(handler),
            retry_backoff_s=0,
        )
        await model.start(
            prompt="draw",
            reference_png=_PNG,
            screenshot=screenshot,
            max_actions=10,
            wall_timeout_s=None,
        )

        turn = await model.next_action(remaining_actions=10, remaining_time_s=None)

        assert turn.request_attempts == 1
        assert turn.tool_use_id == "call-from-sse"
        assert turn.action_payload["actions"][0]["action"] == "done"
        assert turn.usage == {"total_tokens": 7}

    asyncio.run(run())


def test_openrouter_adapter_merges_output_item_done_into_terminal_response() -> None:
    async def run() -> None:
        screenshot = DesktopScreenshot(
            session_id="session",
            sequence=0,
            captured_at="2026-07-15T00:00:00Z",
            width=320,
            height=200,
            autocad_maximized=True,
            png_sha256=hashlib.sha256(_PNG).hexdigest(),
            png=_PNG,
        )

        def handler(_: httpx.Request) -> httpx.Response:
            output_item = {
                "type": "response.output_item.done",
                "response_id": "resp-1",
                "output_index": 0,
                "item": {
                    "type": "function_call",
                    "id": "item-1",
                    "call_id": "call-separated-from-terminal",
                    "name": "computer_action",
                    "arguments": '{"actions":[{"action":"done"}]}',
                },
            }
            terminal = {
                "type": "response.done",
                "response": {
                    "id": "resp-1",
                    "status": "completed",
                    "usage": {"total_tokens": 11},
                },
            }
            # Include a malformed intermediary record and omit blank lines to
            # exercise OpenRouter/proxy recovery without replaying the request.
            body = "\n".join(
                [
                    ": OPENROUTER PROCESSING",
                    "data: {malformed-intermediary-json}",
                    f"data: {json.dumps(output_item)}",
                    f"data: {json.dumps(terminal)}",
                    "data: [DONE]",
                ]
            )
            return httpx.Response(
                200,
                headers={"Content-Type": "text/event-stream"},
                text=body,
            )

        model = OpenRouterResponsesComputerUseModel(
            api_key="test-key",
            model_id="moonshotai/kimi-k3",
            transport=httpx.MockTransport(handler),
            retry_backoff_s=0,
        )
        await model.start(
            prompt="draw",
            reference_png=_PNG,
            screenshot=screenshot,
            max_actions=10,
            wall_timeout_s=None,
        )

        turn = await model.next_action(remaining_actions=10, remaining_time_s=None)

        assert turn.request_attempts == 1
        assert turn.tool_use_id == "call-separated-from-terminal"
        assert turn.action_payload["actions"][0]["action"] == "done"
        assert turn.usage == {"total_tokens": 11}

    asyncio.run(run())


def test_openrouter_adapter_retains_nonretryable_upstream_error_detail() -> None:
    async def run() -> None:
        screenshot = DesktopScreenshot(
            session_id="session",
            sequence=0,
            captured_at="2026-07-15T00:00:00Z",
            width=320,
            height=200,
            autocad_maximized=True,
            png_sha256=hashlib.sha256(_PNG).hexdigest(),
            png=_PNG,
        )

        def handler(_: httpx.Request) -> httpx.Response:
            return httpx.Response(
                400,
                json={
                    "error": {
                        "message": "Provider returned error",
                        "metadata": {
                            "provider_name": "Moonshot AI",
                            "raw": "temporarily rate-limited upstream",
                        },
                    }
                },
            )

        model = OpenRouterResponsesComputerUseModel(
            api_key="test-key",
            model_id="moonshotai/kimi-k3",
            reasoning_max_tokens=512,
            transport=httpx.MockTransport(handler),
            max_request_attempts=1,
        )
        await model.start(
            prompt="draw",
            reference_png=_PNG,
            screenshot=screenshot,
            max_actions=10,
            wall_timeout_s=None,
        )

        with pytest.raises(RuntimeError, match="temporarily rate-limited upstream"):
            await model.next_action(remaining_actions=10, remaining_time_s=None)

    asyncio.run(run())


def test_openai_adapter_retries_missing_computer_action_before_failing_rollout() -> (
    None
):
    async def run() -> None:
        screenshot = DesktopScreenshot(
            session_id="session",
            sequence=0,
            captured_at="2026-07-15T00:00:00Z",
            width=320,
            height=200,
            autocad_maximized=True,
            png_sha256=hashlib.sha256(_PNG).hexdigest(),
            png=_PNG,
        )
        calls = 0
        requests: list[dict[str, Any]] = []

        def handler(request: httpx.Request) -> httpx.Response:
            nonlocal calls
            calls += 1
            requests.append(json.loads(request.content))
            if calls == 1:
                return httpx.Response(
                    200,
                    json={
                        "status": "incomplete",
                        "output": [],
                        "usage": {"input_tokens": 10, "output_tokens": 20},
                    },
                )
            return httpx.Response(
                200,
                json={
                    "status": "completed",
                    "output": [
                        {
                            "type": "function_call",
                            "call_id": "call-repaired",
                            "name": "computer_action",
                            "arguments": '{"actions":[{"action":"done","intent":"Finish."}]}',
                        }
                    ],
                    "usage": {"input_tokens": 11, "output_tokens": 21},
                },
            )

        model = OpenAIResponsesComputerUseModel(
            api_key="test-key",
            base_url="https://openai.test/v1",
            transport=httpx.MockTransport(handler),
            retry_backoff_s=0,
        )
        await model.start(
            prompt="draw",
            reference_png=_PNG,
            screenshot=screenshot,
            max_actions=10,
            wall_timeout_s=60,
        )

        turn = await model.next_action(remaining_actions=10, remaining_time_s=60)

        assert calls == 2
        assert turn.request_attempts == 2
        assert turn.response_validation_retries == 1
        assert turn.action_payload["actions"][0]["action"] == "done"
        assert turn.usage == {"input_tokens": 21, "output_tokens": 41}
        assert requests[0]["max_output_tokens"] == 8192
        assert requests[1]["max_output_tokens"] == 16384
        assert "RESPONSE VALIDATION RETRY" in requests[1]["instructions"]
        assert turn.model_input is not None
        assert turn.model_input["response_validation_retries"] == 1

    asyncio.run(run())


def test_openai_adapter_retries_raw_tls_failure_with_fresh_client() -> None:
    class FlakyTlsTransport(httpx.AsyncBaseTransport):
        def __init__(self) -> None:
            self.calls = 0

        async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
            self.calls += 1
            if self.calls == 1:
                raise ssl.SSLError("ssl/tls alert bad record mac")
            return httpx.Response(
                200,
                request=request,
                json={
                    "status": "completed",
                    "output": [
                        {
                            "type": "function_call",
                            "call_id": "call-after-tls-retry",
                            "name": "computer_action",
                            "arguments": '{"action":"done"}',
                        }
                    ],
                    "usage": {"total_tokens": 5},
                },
            )

    async def run() -> None:
        screenshot = DesktopScreenshot(
            session_id="session",
            sequence=0,
            captured_at="2026-07-15T00:00:00Z",
            width=320,
            height=200,
            autocad_maximized=True,
            png_sha256=hashlib.sha256(_PNG).hexdigest(),
            png=_PNG,
        )
        transport = FlakyTlsTransport()
        model = OpenAIResponsesComputerUseModel(
            api_key="test-key",
            base_url="https://openai.test/v1",
            transport=transport,
            retry_backoff_s=0,
        )
        await model.start(
            prompt="draw",
            reference_png=_PNG,
            screenshot=screenshot,
            max_actions=10,
            wall_timeout_s=None,
        )

        turn = await model.next_action(remaining_actions=10, remaining_time_s=None)

        assert turn.action_payload == {"action": "done"}
        assert turn.request_attempts == 2
        assert transport.calls == 2

    asyncio.run(run())


def test_anthropic_adapter_uses_official_messages_shape_and_secret_free_trace() -> None:
    class FakeAnthropicClient:
        def __init__(self) -> None:
            self.beta = self
            self.messages = self
            self.requests: list[dict[str, Any]] = []

        async def create(self, **request: Any) -> dict[str, Any]:
            self.requests.append(dict(request))
            return {
                "id": "msg-test",
                "stop_reason": "tool_use",
                "content": [
                    {"type": "text", "text": "working"},
                    {
                        "type": "tool_use",
                        "id": "tool-test",
                        "name": "computer_action",
                        "input": {
                            "actions": [
                                {
                                    "action": "click",
                                    "x": 10,
                                    "y": 20,
                                    "intent": "Focus the command line.",
                                }
                            ]
                        },
                    },
                ],
                "usage": {
                    "input_tokens": 11,
                    "output_tokens": 7,
                    "cache_creation_input_tokens": 5,
                    "cache_read_input_tokens": 3,
                },
            }

    async def run() -> None:
        screenshot = DesktopScreenshot(
            session_id="session",
            sequence=0,
            captured_at="2026-07-16T00:00:00Z",
            width=320,
            height=200,
            autocad_maximized=True,
            png_sha256=hashlib.sha256(_PNG).hexdigest(),
            png=_PNG,
        )
        client = FakeAnthropicClient()
        model = AnthropicMessagesComputerUseModel(
            api_key="test-key",
            model_id="claude-test",
            client=client,
        )
        await model.start(
            prompt="draw",
            reference_png=_PNG,
            screenshot=screenshot,
            max_actions=10,
            wall_timeout_s=60,
        )
        turn = await model.next_action(remaining_actions=10, remaining_time_s=60)

        assert turn.action_payload == {
            "actions": [
                {
                    "action": "click",
                    "x": 10,
                    "y": 20,
                    "intent": "Focus the command line.",
                }
            ]
        }
        assert turn.usage == {
            "input_tokens": 11,
            "output_tokens": 7,
            "cache_creation_input_tokens": 5,
            "cache_read_input_tokens": 3,
            "cached_input_tokens": 3,
        }
        request = client.requests[0]
        assert request["model"] == "claude-test"
        assert request["betas"] == ["compact-2026-01-12"]
        assert request["cache_control"] == {"type": "ephemeral"}
        assert "Remaining budget:" not in json.dumps(request["system"])
        assert request["system"][0]["cache_control"] == {"type": "ephemeral"}
        assert request["context_management"] == {
            "edits": [
                {
                    "type": "compact_20260112",
                    "trigger": {"type": "input_tokens", "value": 150_000},
                }
            ]
        }
        assert request["tool_choice"] == {
            "type": "tool",
            "name": "computer_action",
            "disable_parallel_tool_use": True,
        }
        assert request["tools"][0]["name"] == "computer_action"
        assert request["tools"][0]["strict"] is True
        assert request["tools"][0]["input_schema"]["required"] == ["actions"]
        assert '"maximum"' not in json.dumps(request["tools"][0]["input_schema"])
        assert (
            len(
                [
                    block
                    for block in request["messages"][0]["content"]
                    if block["type"] == "image"
                ]
            )
            == 2
        )
        assert request["messages"][0]["content"][2]["cache_control"] == {
            "type": "ephemeral"
        }
        assert turn.model_input is not None
        serialized = json.dumps(turn.model_input)
        assert "test-key" not in serialized
        assert base64.b64encode(_PNG).decode("ascii") not in serialized
        assert "reference.png" in serialized

        await model.observe(
            tool_use_id=turn.tool_use_id,
            screenshot=screenshot,
            message="input injected",
            is_error=False,
        )
        second_turn = await model.next_action(
            remaining_actions=9,
            remaining_time_s=55,
        )
        first_content = client.requests[0]["messages"][0]["content"]
        second_messages = client.requests[1]["messages"]
        assert [message["role"] for message in second_messages] == [
            "user",
            "assistant",
            "user",
        ]
        assert second_messages[0]["content"] == first_content
        assert second_messages[1]["content"][1]["type"] == "tool_use"
        assert second_messages[1]["content"][1]["id"] == turn.tool_use_id
        second_content = second_messages[2]["content"]
        assert second_content[0]["type"] == "tool_result"
        assert second_content[0]["tool_use_id"] == turn.tool_use_id
        assert second_content[0]["is_error"] is False
        outcome = json.loads(second_content[0]["content"])
        assert outcome["controller_result"] == "input_injected"
        assert outcome["screen_changed"] is False
        assert second_content[1]["text"] == (
            "Remaining budget: 9 actions, 55.0 seconds."
        )
        assert second_turn.model_input is not None
        second_serialized = json.dumps(second_turn.model_input)
        assert base64.b64encode(_PNG).decode("ascii") not in second_serialized
        # The exact committed Anthropic state is persisted alongside the
        # auditable request history for interruption recovery.
        assert second_serialized.count('"artifact_path"') == 6

    asyncio.run(run())


def test_fable_uses_deterministic_client_compaction_without_server_sampling() -> None:
    class UnexpectedBetaMessages:
        def __init__(self) -> None:
            self.calls = 0

        async def create(self, **request: Any) -> dict[str, Any]:
            del request
            self.calls += 1
            raise AssertionError("Fable must not use Anthropic server compaction")

    class FakeAnthropicClient:
        def __init__(self) -> None:
            self.messages = self
            self.beta_messages = UnexpectedBetaMessages()
            self.beta = type(
                "Beta",
                (),
                {"messages": self.beta_messages},
            )()
            self.requests: list[dict[str, Any]] = []

        async def create(self, **request: Any) -> dict[str, Any]:
            self.requests.append(deepcopy(request))
            return {
                "stop_reason": "tool_use",
                "content": [
                    {
                        "type": "tool_use",
                        "id": f"tool-{len(self.requests)}",
                        "name": "computer_action",
                        "input": {
                            "actions": [
                                {
                                    "action": "click",
                                    "x": 10,
                                    "y": 20,
                                    "intent": "Continue the drawing.",
                                }
                            ]
                        },
                    }
                ],
                "usage": {"input_tokens": 101, "output_tokens": 3},
            }

    async def run() -> None:
        screenshot = DesktopScreenshot(
            session_id="session",
            sequence=0,
            captured_at="2026-07-20T00:00:00Z",
            width=320,
            height=200,
            autocad_maximized=True,
            png_sha256=hashlib.sha256(_PNG).hexdigest(),
            png=_PNG,
        )
        client = FakeAnthropicClient()
        model = AnthropicMessagesComputerUseModel(
            api_key="test-key",
            model_id="claude-fable-5",
            client=client,
            client_compaction_trigger_tokens=100,
            client_compaction_turns=100,
        )
        await model.start(
            prompt="draw",
            reference_png=_PNG,
            screenshot=screenshot,
            max_actions=10,
            wall_timeout_s=None,
        )

        first = await model.next_action(
            remaining_actions=10,
            remaining_time_s=None,
        )
        assert client.beta_messages.calls == 0
        assert "betas" not in client.requests[0]
        assert "context_management" not in client.requests[0]
        assert client.requests[0]["cache_control"] == {"type": "ephemeral"}
        assert first.model_input is not None
        assert first.model_input["history_compaction"] == {
            "strategy": "deterministic_controller_checkpoint",
            "client_enabled": True,
            "server_enabled": False,
            "round": 0,
            "applied_to_request": False,
            "trigger_input_tokens": 100,
            "trigger_turns": 100,
            "recent_outcomes_retained": 0,
            "older_outcomes_compacted": 0,
            "turns_since_compaction": 0,
            "last_effective_input_tokens": 101,
            "serialized_body_guard_compactions": 0,
        }

        await model.observe(
            tool_use_id=first.tool_use_id,
            screenshot=screenshot,
            message="input injected",
            is_error=False,
        )
        second = await model.next_action(
            remaining_actions=9,
            remaining_time_s=None,
        )

        assert client.beta_messages.calls == 0
        assert len(client.requests) == 2
        second_request = client.requests[1]
        assert "betas" not in second_request
        assert "context_management" not in second_request
        assert [message["role"] for message in second_request["messages"]] == [
            "user"
        ]
        second_content = second_request["messages"][0]["content"]
        assert sum(block["type"] == "image" for block in second_content) == 2
        summary = next(
            block["text"]
            for block in second_content
            if block.get("type") == "text"
            and block.get("text", "").startswith("PROMPT HISTORY COMPACTION:")
        )
        assert '"completed_controller_outcomes":1' in summary
        assert '"action":"click"' in summary
        assert "input_injected" in summary
        assert second.model_input is not None
        assert second.model_input["history_compaction"]["round"] == 1
        assert second.model_input["history_compaction"]["applied_to_request"] is True
        assert second.model_input["history_compaction"]["server_enabled"] is False
        assert second.model_input["client_compaction_state"]["outcomes"][0][
            "action"
        ] == first.action_payload

    asyncio.run(run())


def test_anthropic_adapter_repairs_missing_tool_before_desktop_action() -> None:
    class FakeAnthropicClient:
        def __init__(self) -> None:
            self.messages = self
            self.requests: list[dict[str, Any]] = []

        async def create(self, **request: Any) -> dict[str, Any]:
            self.requests.append(dict(request))
            if len(self.requests) == 1:
                return {
                    "stop_reason": "max_tokens",
                    "content": [{"type": "text", "text": "truncated"}],
                    "usage": {"input_tokens": 10, "output_tokens": 20},
                }
            return {
                "stop_reason": "tool_use",
                "content": [
                    {
                        "type": "tool_use",
                        "id": "tool-repaired",
                        "name": "computer_action",
                        "input": {"actions": [{"action": "done", "intent": "Finish."}]},
                    }
                ],
                "usage": {"input_tokens": 11, "output_tokens": 21},
            }

    async def run() -> None:
        screenshot = DesktopScreenshot(
            session_id="session",
            sequence=0,
            captured_at="2026-07-16T00:00:00Z",
            width=320,
            height=200,
            autocad_maximized=True,
            png_sha256=hashlib.sha256(_PNG).hexdigest(),
            png=_PNG,
        )
        client = FakeAnthropicClient()
        model = AnthropicMessagesComputerUseModel(
            api_key="test-key",
            model_id="claude-test",
            client=client,
            response_retry_backoff_s=0,
        )
        await model.start(
            prompt="draw",
            reference_png=_PNG,
            screenshot=screenshot,
            max_actions=10,
            wall_timeout_s=60,
        )

        turn = await model.next_action(remaining_actions=10, remaining_time_s=60)

        assert turn.action_payload["actions"][0]["action"] == "done"
        assert turn.request_attempts == 2
        assert turn.response_validation_retries == 1
        assert turn.usage == {"input_tokens": 21, "output_tokens": 41}
        assert client.requests[0]["max_tokens"] == 8192
        assert client.requests[1]["max_tokens"] == 16384
        assert "RESPONSE VALIDATION RETRY" in json.dumps(client.requests[1]["system"])

    asyncio.run(run())


def test_anthropic_adapter_retries_rate_limits_without_consuming_validation_attempts() -> (
    None
):
    class RateLimitError(Exception):
        status_code = 429

        def __init__(self) -> None:
            self.response = type(
                "Response",
                (),
                {"headers": {"retry-after": "0"}},
            )()

    class FakeAnthropicClient:
        def __init__(self) -> None:
            self.messages = self
            self.requests: list[dict[str, Any]] = []

        async def create(self, **request: Any) -> dict[str, Any]:
            self.requests.append(dict(request))
            if len(self.requests) <= 4:
                raise RateLimitError()
            return {
                "stop_reason": "tool_use",
                "content": [
                    {
                        "type": "tool_use",
                        "id": "tool-after-rate-limit",
                        "name": "computer_action",
                        "input": {"actions": [{"action": "done", "intent": "Finish."}]},
                    }
                ],
                "usage": {"input_tokens": 11, "output_tokens": 7},
            }

    async def run() -> None:
        screenshot = DesktopScreenshot(
            session_id="session",
            sequence=0,
            captured_at="2026-07-19T00:00:00Z",
            width=320,
            height=200,
            autocad_maximized=True,
            png_sha256=hashlib.sha256(_PNG).hexdigest(),
            png=_PNG,
        )
        client = FakeAnthropicClient()
        model = AnthropicMessagesComputerUseModel(
            api_key="test-key",
            model_id="claude-test",
            client=client,
            response_retry_backoff_s=0,
        )
        await model.start(
            prompt="draw",
            reference_png=_PNG,
            screenshot=screenshot,
            max_actions=10,
            wall_timeout_s=None,
        )

        turn = await model.next_action(remaining_actions=10, remaining_time_s=None)

        assert len(client.requests) == 5
        assert turn.request_attempts == 5
        assert turn.response_validation_retries == 0
        assert turn.model_input is not None
        assert turn.model_input["rate_limit_retries"] == 4

    asyncio.run(run())


def test_anthropic_adapter_allows_nine_consecutive_transient_failures_and_resets() -> (
    None
):
    class TransientError(Exception):
        status_code = 529
        response = type("Response", (), {"headers": {"x-should-retry": "false"}})()

    class FakeAnthropicClient:
        def __init__(self) -> None:
            self.messages = self
            self.requests: list[dict[str, Any]] = []
            self.failures_remaining = 9

        async def create(self, **request: Any) -> dict[str, Any]:
            self.requests.append(dict(request))
            if self.failures_remaining:
                self.failures_remaining -= 1
                raise TransientError()
            self.failures_remaining = 9
            return {
                "stop_reason": "tool_use",
                "content": [
                    {
                        "type": "tool_use",
                        "id": f"tool-{len(self.requests)}",
                        "name": "computer_action",
                        "input": {"actions": [{"action": "done", "intent": "Finish."}]},
                    }
                ],
                "usage": {"input_tokens": 1, "output_tokens": 1},
            }

    async def run() -> None:
        screenshot = DesktopScreenshot(
            session_id="session",
            sequence=0,
            captured_at="2026-07-19T00:00:00Z",
            width=320,
            height=200,
            autocad_maximized=True,
            png_sha256=hashlib.sha256(_PNG).hexdigest(),
            png=_PNG,
        )
        client = FakeAnthropicClient()
        model = AnthropicMessagesComputerUseModel(
            api_key="test-key",
            model_id="claude-test",
            client=client,
            response_retry_backoff_s=0,
        )
        await model.start(
            prompt="draw",
            reference_png=_PNG,
            screenshot=screenshot,
            max_actions=10,
            wall_timeout_s=None,
        )

        first = await model.next_action(remaining_actions=10, remaining_time_s=None)
        await model.observe(
            tool_use_id=first.tool_use_id,
            screenshot=screenshot,
            message="input injected",
            is_error=False,
        )
        second = await model.next_action(remaining_actions=9, remaining_time_s=None)

        assert first.request_attempts == 10
        assert second.request_attempts == 10
        assert first.model_input is not None
        assert second.model_input is not None
        assert first.model_input["transient_error_retries"] == 9
        assert second.model_input["transient_error_retries"] == 9

    asyncio.run(run())


def test_anthropic_adapter_stops_after_ten_consecutive_transient_failures() -> None:
    class ConnectionError(Exception):
        pass

    class FakeAnthropicClient:
        def __init__(self) -> None:
            self.messages = self
            self.requests = 0

        async def create(self, **request: Any) -> dict[str, Any]:
            del request
            self.requests += 1
            error = ConnectionError("connection failed")
            error.__cause__ = httpx.ReadTimeout("read failed")
            raise error

    async def run() -> None:
        screenshot = DesktopScreenshot(
            session_id="session",
            sequence=0,
            captured_at="2026-07-19T00:00:00Z",
            width=320,
            height=200,
            autocad_maximized=True,
            png_sha256=hashlib.sha256(_PNG).hexdigest(),
            png=_PNG,
        )
        client = FakeAnthropicClient()
        model = AnthropicMessagesComputerUseModel(
            api_key="test-key",
            model_id="claude-test",
            client=client,
            response_retry_backoff_s=0,
        )
        await model.start(
            prompt="draw",
            reference_png=_PNG,
            screenshot=screenshot,
            max_actions=10,
            wall_timeout_s=None,
        )

        with pytest.raises(ConnectionError, match="connection failed"):
            await model.next_action(remaining_actions=10, remaining_time_s=None)
        assert client.requests == 10

    asyncio.run(run())


def test_anthropic_adapter_persistently_retries_http_500_with_backoff() -> None:
    class InternalServerError(Exception):
        status_code = 500

    class FakeAnthropicClient:
        def __init__(self) -> None:
            self.messages = self
            self.requests = 0

        async def create(self, **request: Any) -> dict[str, Any]:
            del request
            self.requests += 1
            if self.requests <= 12:
                raise InternalServerError("provider unavailable")
            return {
                "stop_reason": "tool_use",
                "content": [
                    {
                        "type": "tool_use",
                        "id": "tool-recovered",
                        "name": "computer_action",
                        "input": {
                            "actions": [{"action": "done", "intent": "Finish."}]
                        },
                    }
                ],
                "usage": {"input_tokens": 1, "output_tokens": 1},
            }

    async def run() -> None:
        screenshot = DesktopScreenshot(
            session_id="session",
            sequence=0,
            captured_at="2026-07-19T00:00:00Z",
            width=320,
            height=200,
            autocad_maximized=True,
            png_sha256=hashlib.sha256(_PNG).hexdigest(),
            png=_PNG,
        )
        client = FakeAnthropicClient()
        model = AnthropicMessagesComputerUseModel(
            api_key="test-key",
            model_id="claude-test",
            client=client,
            response_retry_backoff_s=0,
        )
        await model.start(
            prompt="draw",
            reference_png=_PNG,
            screenshot=screenshot,
            max_actions=10,
            wall_timeout_s=None,
        )

        turn = await model.next_action(remaining_actions=10, remaining_time_s=None)

        assert client.requests == 13
        assert turn.request_attempts == 13
        assert turn.model_input is not None
        assert turn.model_input["transient_error_retries"] == 12
        assert turn.action_payload == {
            "actions": [{"action": "done", "intent": "Finish."}]
        }

    asyncio.run(run())


def test_anthropic_adapter_allows_nine_consecutive_malformed_responses_and_resets() -> (
    None
):
    class FakeAnthropicClient:
        def __init__(self) -> None:
            self.messages = self
            self.requests: list[dict[str, Any]] = []
            self.malformed_remaining = 9

        async def create(self, **request: Any) -> dict[str, Any]:
            self.requests.append(dict(request))
            if self.malformed_remaining:
                self.malformed_remaining -= 1
                return {
                    "stop_reason": "max_tokens",
                    "content": [{"type": "text", "text": "no tool"}],
                    "usage": {"input_tokens": 1, "output_tokens": 1},
                }
            self.malformed_remaining = 9
            return {
                "stop_reason": "tool_use",
                "content": [
                    {
                        "type": "tool_use",
                        "id": f"tool-{len(self.requests)}",
                        "name": "computer_action",
                        "input": {"actions": [{"action": "done", "intent": "Finish."}]},
                    }
                ],
                "usage": {"input_tokens": 1, "output_tokens": 1},
            }

    async def run() -> None:
        screenshot = DesktopScreenshot(
            session_id="session",
            sequence=0,
            captured_at="2026-07-19T00:00:00Z",
            width=320,
            height=200,
            autocad_maximized=True,
            png_sha256=hashlib.sha256(_PNG).hexdigest(),
            png=_PNG,
        )
        client = FakeAnthropicClient()
        model = AnthropicMessagesComputerUseModel(
            api_key="test-key",
            model_id="claude-test",
            client=client,
            response_retry_backoff_s=0,
        )
        await model.start(
            prompt="draw",
            reference_png=_PNG,
            screenshot=screenshot,
            max_actions=10,
            wall_timeout_s=None,
        )

        first = await model.next_action(remaining_actions=10, remaining_time_s=None)
        await model.observe(
            tool_use_id=first.tool_use_id,
            screenshot=screenshot,
            message="input injected",
            is_error=False,
        )
        second = await model.next_action(remaining_actions=9, remaining_time_s=None)

        assert first.request_attempts == 10
        assert second.request_attempts == 10
        assert first.response_validation_retries == 9
        assert second.response_validation_retries == 9

    asyncio.run(run())


def test_anthropic_adapter_compacts_server_history_and_reattaches_reference() -> None:
    class FakeAnthropicClient:
        def __init__(self) -> None:
            self.beta = self
            self.messages = self
            self.requests: list[dict[str, Any]] = []

        async def create(self, **request: Any) -> dict[str, Any]:
            self.requests.append(deepcopy(request))
            tool_id = f"tool-{len(self.requests)}"
            content: list[dict[str, Any]] = []
            usage: dict[str, Any] = {"input_tokens": 2, "output_tokens": 3}
            if len(self.requests) == 1:
                content.append(
                    {
                        "type": "compaction",
                        "content": "Task and drawing progress summary.",
                        "encrypted_content": None,
                    }
                )
                usage["iterations"] = [
                    {
                        "type": "compaction",
                        "input_tokens": 100,
                        "output_tokens": 20,
                        "cache_read_input_tokens": 50,
                    },
                    {
                        "type": "message",
                        "input_tokens": 2,
                        "output_tokens": 3,
                        "cache_read_input_tokens": 1,
                    },
                ]
            content.append(
                {
                    "type": "tool_use",
                    "id": tool_id,
                    "name": "computer_action",
                    "input": {"actions": [{"action": "done", "intent": "Done."}]},
                }
            )
            return {
                "stop_reason": "tool_use",
                "content": content,
                "usage": usage,
            }

    async def run() -> None:
        screenshot = DesktopScreenshot(
            session_id="session",
            sequence=0,
            captured_at="2026-07-19T00:00:00Z",
            width=320,
            height=200,
            autocad_maximized=True,
            png_sha256=hashlib.sha256(_PNG).hexdigest(),
            png=_PNG,
        )
        client = FakeAnthropicClient()
        model = AnthropicMessagesComputerUseModel(
            api_key="test-key",
            model_id="claude-test",
            client=client,
        )
        await model.start(
            prompt="draw",
            reference_png=_PNG,
            screenshot=screenshot,
            max_actions=10,
            wall_timeout_s=None,
        )

        first = await model.next_action(remaining_actions=10, remaining_time_s=None)
        assert first.usage == {
            "input_tokens": 102,
            "output_tokens": 23,
            "cache_read_input_tokens": 51,
            "cached_input_tokens": 51,
        }
        assert len(model.messages) == 1
        assert model.messages[0]["role"] == "assistant"
        assert model.messages[0]["content"][0]["type"] == "compaction"
        assert model.messages[0]["content"][0]["cache_control"] == {"type": "ephemeral"}
        assert "encrypted_content" not in model.messages[0]["content"][0]
        assert model.static_context_required is True

        await model.observe(
            tool_use_id=first.tool_use_id,
            screenshot=screenshot,
            message="input injected",
            is_error=False,
        )
        await model.next_action(remaining_actions=9, remaining_time_s=None)
        second_messages = client.requests[1]["messages"]
        assert [message["role"] for message in second_messages] == [
            "assistant",
            "user",
        ]
        second_content = second_messages[1]["content"]
        assert second_content[0]["type"] == "tool_result"
        assert sum(block["type"] == "image" for block in second_content) == 2
        assert any(
            block.get("text") == agent_loop_module._STATIC_CONTEXT_MARKER
            for block in second_content
        )
        assert client.requests[0]["betas"] == ["compact-2026-01-12"]

    asyncio.run(run())


def test_anthropic_adapter_treats_null_compaction_as_noop() -> None:
    class FakeAnthropicClient:
        def __init__(self) -> None:
            self.messages = self
            self.requests: list[dict[str, Any]] = []

        async def create(self, **request: Any) -> dict[str, Any]:
            self.requests.append(deepcopy(request))
            return {
                "stop_reason": "tool_use",
                "content": [
                    {
                        "type": "compaction",
                        "content": None,
                        "encrypted_content": None,
                    },
                    {
                        "type": "tool_use",
                        "id": f"tool-{len(self.requests)}",
                        "name": "computer_action",
                        "input": {"actions": [{"action": "done", "intent": "Done."}]},
                    },
                ],
                "usage": {"input_tokens": 1, "output_tokens": 1},
            }

    async def run() -> None:
        screenshot = DesktopScreenshot(
            session_id="session",
            sequence=0,
            captured_at="2026-07-19T00:00:00Z",
            width=320,
            height=200,
            autocad_maximized=True,
            png_sha256=hashlib.sha256(_PNG).hexdigest(),
            png=_PNG,
        )
        client = FakeAnthropicClient()
        model = AnthropicMessagesComputerUseModel(
            api_key="test-key", model_id="claude-test", client=client
        )
        await model.start(
            prompt="draw",
            reference_png=_PNG,
            screenshot=screenshot,
            max_actions=10,
            wall_timeout_s=None,
        )

        first = await model.next_action(remaining_actions=10, remaining_time_s=None)

        assert len(model.messages) == 2
        assert model.messages[0]["role"] == "user"
        assert all(
            block.get("type") != "compaction" for block in model.messages[1]["content"]
        )
        assert model.static_context_required is False
        await model.observe(
            tool_use_id=first.tool_use_id,
            screenshot=screenshot,
            message="input injected",
            is_error=False,
        )
        await model.next_action(remaining_actions=9, remaining_time_s=None)
        assert "compaction" not in json.dumps(client.requests[1]["messages"])

    asyncio.run(run())


def test_anthropic_adapter_prunes_images_before_soft_request_limit(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    class FakeAnthropicClient:
        def __init__(self) -> None:
            self.messages = self
            self.requests: list[dict[str, Any]] = []

        async def create(self, **request: Any) -> dict[str, Any]:
            self.requests.append(deepcopy(request))
            return {
                "stop_reason": "tool_use",
                "content": [
                    {
                        "type": "tool_use",
                        "id": "tool-new",
                        "name": "computer_action",
                        "input": {"actions": [{"action": "done", "intent": "Done."}]},
                    }
                ],
                "usage": {"input_tokens": 1, "output_tokens": 1},
            }

    async def run() -> None:
        monkeypatch.setattr(
            agent_loop_module,
            "_ANTHROPIC_REQUEST_SOFT_LIMIT_BYTES",
            100_000,
        )
        screenshot = DesktopScreenshot(
            session_id="session",
            sequence=1,
            captured_at="2026-07-19T00:00:00Z",
            width=320,
            height=200,
            autocad_maximized=True,
            png_sha256=hashlib.sha256(_PNG).hexdigest(),
            png=_PNG,
        )
        client = FakeAnthropicClient()
        model = AnthropicMessagesComputerUseModel(
            api_key="test-key", model_id="claude-test", client=client
        )
        await model.start(
            prompt="draw",
            reference_png=_PNG,
            screenshot=screenshot,
            max_actions=10,
            wall_timeout_s=None,
        )
        model.static_context_required = False
        model.messages = [
            {
                "role": "user",
                "content": [
                    {
                        "type": "image",
                        "source": {
                            "type": "base64",
                            "media_type": "image/png",
                            "data": "A" * 200_000,
                        },
                    }
                ],
            },
            {
                "role": "assistant",
                "content": [
                    {
                        "type": "tool_use",
                        "id": "tool-old",
                        "name": "computer_action",
                        "input": {"actions": [{"action": "wait", "milliseconds": 1}]},
                    }
                ],
            },
        ]
        model.trace_messages = [
            {
                "role": "user",
                "content": [
                    {
                        "type": "image",
                        "artifact_path": "frames/frame_0000.png",
                        "sha256": "old",
                    }
                ],
            },
            deepcopy(model.messages[1]),
        ]
        model.pending_tool_result = {
            "type": "tool_result",
            "tool_use_id": "tool-old",
            "content": "{}",
            "is_error": False,
        }

        turn = await model.next_action(remaining_actions=9, remaining_time_s=None)

        assert turn.model_input is not None
        assert turn.model_input["historical_desktop_images_pruned"] is True
        assert turn.model_input["request_size_bytes"] < 100_000
        old_content = client.requests[0]["messages"][0]["content"]
        assert old_content == [
            {
                "type": "text",
                "text": agent_loop_module._ANTHROPIC_OMITTED_SCREENSHOT_TEXT,
            }
        ]

    asyncio.run(run())


def test_anthropic_adapter_recovers_once_from_http_413() -> None:
    class RequestTooLargeError(Exception):
        status_code = 413

    class FakeAnthropicClient:
        def __init__(self) -> None:
            self.messages = self
            self.requests: list[dict[str, Any]] = []

        async def create(self, **request: Any) -> dict[str, Any]:
            self.requests.append(deepcopy(request))
            if len(self.requests) == 1:
                raise RequestTooLargeError("too large")
            return {
                "stop_reason": "tool_use",
                "content": [
                    {
                        "type": "tool_use",
                        "id": "tool-new",
                        "name": "computer_action",
                        "input": {"actions": [{"action": "done", "intent": "Done."}]},
                    }
                ],
                "usage": {"input_tokens": 1, "output_tokens": 1},
            }

    async def run() -> None:
        screenshot = DesktopScreenshot(
            session_id="session",
            sequence=1,
            captured_at="2026-07-19T00:00:00Z",
            width=320,
            height=200,
            autocad_maximized=True,
            png_sha256=hashlib.sha256(_PNG).hexdigest(),
            png=_PNG,
        )
        client = FakeAnthropicClient()
        model = AnthropicMessagesComputerUseModel(
            api_key="test-key", model_id="claude-test", client=client
        )
        await model.start(
            prompt="draw",
            reference_png=_PNG,
            screenshot=screenshot,
            max_actions=10,
            wall_timeout_s=None,
        )
        model.static_context_required = False
        historical = model._image_block(_PNG)
        model.messages = [
            {"role": "user", "content": [historical]},
            {
                "role": "assistant",
                "content": [
                    {
                        "type": "tool_use",
                        "id": "tool-old",
                        "name": "computer_action",
                        "input": {"actions": [{"action": "wait", "milliseconds": 1}]},
                    }
                ],
            },
        ]
        model.trace_messages = [
            {
                "role": "user",
                "content": [
                    {
                        "type": "image",
                        "artifact_path": "frames/frame_0000.png",
                        "sha256": hashlib.sha256(_PNG).hexdigest(),
                    }
                ],
            },
            deepcopy(model.messages[1]),
        ]
        model.pending_tool_result = {
            "type": "tool_result",
            "tool_use_id": "tool-old",
            "content": "{}",
            "is_error": False,
        }

        turn = await model.next_action(remaining_actions=9, remaining_time_s=None)

        assert len(client.requests) == 2
        assert turn.request_attempts == 2
        assert turn.model_input is not None
        assert turn.model_input["request_size_retries"] == 1
        assert client.requests[0]["messages"][0]["content"][0]["type"] == "image"
        assert client.requests[1]["messages"][0]["content"][0]["type"] == "text"

    asyncio.run(run())


def test_generic_mantle_responses_adapter_signs_request_and_omits_reasoning() -> None:
    from botocore.credentials import Credentials

    class FakeSession:
        def get_credentials(self) -> Credentials:
            return Credentials("access", "secret", "token")

    async def run() -> None:
        screenshot = DesktopScreenshot(
            session_id="session",
            sequence=0,
            captured_at="2026-07-16T00:00:00Z",
            width=320,
            height=200,
            autocad_maximized=True,
            png_sha256=hashlib.sha256(_PNG).hexdigest(),
            png=_PNG,
        )
        captured: list[dict[str, Any]] = []

        def handler(request: httpx.Request) -> httpx.Response:
            assert request.url == httpx.URL("https://mantle.test/v1/responses")
            assert request.headers["authorization"].startswith("AWS4-HMAC-SHA256")
            assert request.headers["x-amz-security-token"] == "token"
            body = json.loads(request.content)
            captured.append(body)
            return httpx.Response(
                200,
                json={
                    "status": "completed",
                    "output": [
                        {
                            "type": "function_call",
                            "call_id": "call-mantle",
                            "name": "computer_action",
                            "arguments": '{"action":"done"}',
                        }
                    ],
                    "usage": {"total_tokens": 4},
                },
            )

        model = MantleResponsesComputerUseModel(
            model_id="openai.gpt-5.6-sol",
            base_url="https://mantle.test/v1",
            session=FakeSession(),
            transport=httpx.MockTransport(handler),
        )
        await model.start(
            prompt="draw",
            reference_png=_PNG,
            screenshot=screenshot,
            max_actions=10,
            wall_timeout_s=60,
        )
        turn = await model.next_action(remaining_actions=10, remaining_time_s=60)

        assert turn.action_payload == {"action": "done"}
        assert captured[0]["model"] == "openai.gpt-5.6-sol"
        assert "reasoning" not in captured[0]
        assert captured[0]["tools"][0]["strict"] is False
        assert "aws-sigv4" not in json.dumps(turn.model_input)

    asyncio.run(run())


def test_mantle_grok_uses_stored_state_cache_strict_tools_and_compaction() -> None:
    from botocore.credentials import Credentials

    class FakeSession:
        def get_credentials(self) -> Credentials:
            return Credentials("access", "secret", "token")

    async def run() -> None:
        screenshots = [
            DesktopScreenshot(
                session_id="grok-session",
                sequence=index,
                captured_at="2026-07-20T00:00:00Z",
                width=320,
                height=200,
                autocad_maximized=True,
                png_sha256=hashlib.sha256(_PNG).hexdigest(),
                png=_PNG,
            )
            for index in range(4)
        ]
        captured: list[dict[str, Any]] = []

        def handler(request: httpx.Request) -> httpx.Response:
            assert request.url == httpx.URL(
                "https://mantle.test/openai/v1/responses"
            )
            body = json.loads(request.content)
            captured.append(body)
            number = len(captured)
            return httpx.Response(
                200,
                json={
                    "id": f"resp-{number}",
                    "status": "completed",
                    "output": [
                        {
                            "type": "function_call",
                            "call_id": f"call-{number}",
                            "name": "computer_action",
                            "arguments": (
                                '{"actions":[{"action":"wait",'
                                '"milliseconds":1,"intent":"test"}]}'
                            ),
                        }
                    ],
                    "usage": {
                        "input_tokens": 100,
                        "input_tokens_details": {"cached_tokens": 80},
                        "output_tokens": 10,
                    },
                },
            )

        model = MantleGrokResponsesComputerUseModel(
            base_url="https://mantle.test/openai/v1",
            session=FakeSession(),
            transport=httpx.MockTransport(handler),
        )
        await model.start(
            prompt="draw",
            reference_png=_PNG,
            screenshot=screenshots[0],
            max_actions=100,
            wall_timeout_s=60,
        )
        first = await model.next_action(remaining_actions=100, remaining_time_s=60)
        await model.observe(
            tool_use_id=first.tool_use_id,
            screenshot=screenshots[1],
            message="ok",
            is_error=False,
        )
        second = await model.next_action(remaining_actions=99, remaining_time_s=59)
        await model.observe(
            tool_use_id=second.tool_use_id,
            screenshot=screenshots[2],
            message="ok",
            is_error=False,
        )
        model.server_state_turns_since_compaction = 24
        third = await model.next_action(remaining_actions=98, remaining_time_s=58)

        first_request, second_request, compacted_request = captured
        assert first_request["store"] is True
        assert first_request["reasoning"] == {"effort": "high"}
        assert first_request["include"] == ["reasoning.encrypted_content"]
        assert first_request["prompt_cache_retention"] == "in_memory"
        assert first_request["prompt_cache_key"].startswith("acb-grok43-")
        assert len(first_request["prompt_cache_key"]) <= 64
        assert first_request["tools"][0]["strict"] is True
        assert "previous_response_id" not in first_request

        assert second_request["previous_response_id"] == "resp-1"
        assert second_request["input"][0]["type"] == "function_call_output"
        assert "fixed reference drawing" not in json.dumps(second_request["input"])
        assert second.usage["cached_input_tokens"] == 80

        assert "previous_response_id" not in compacted_request
        compacted_text = json.dumps(compacted_request["input"])
        assert "fixed reference drawing" in compacted_text
        assert "SERVER-STATE COMPACTION" in compacted_text
        assert third.model_input["history_mode"] == "mantle_grok_server_state"
        assert third.model_input["server_state_compacted"] is True
        assert third.model_input["server_state_response_id"] == "resp-3"

    asyncio.run(run())