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())