iop/scripts/agent_benchmark/claude_iop_test.py

"""Hermetic contract tests for the Claude Code IOP benchmark adapter."""

from __future__ import annotations

import json
import os
import tempfile
import textwrap
import unittest
from pathlib import Path
from unittest.mock import patch

from scripts.agent_benchmark.claude_iop import (
    ClaudeIopAdapter,
    ClaudeIopProtocolError,
    ClaudeIopRuntime,
    ClaudeIopValidationError,
    ClaudeStreamParser,
    parse_preflight_binding,
    redact_claude_event,
)
from scripts.agent_benchmark.lifecycle import (
    CallerEvent, CallerTerminal, ParsedMetric,
    REASON_PARSER_ERROR, REASON_SUCCESS, run_invocation,
)
from scripts.agent_benchmark.manifest import ExpectedBinding, IopCell, MatrixCell, Timeout
from scripts.agent_benchmark.workspace import (
    AttemptIdentity,
    PreparedWorkspace,
    TestbedProvenance,
)


SENTINELS = ("prompt-secret-sentinel", "https://private.iop.invalid", "api-secret-sentinel")
ARBITRARY_SENTINELS = (
    "tool-secret-sentinel", "result-secret-sentinel", "error-secret-sentinel",
)


def _cell(route_kind: str = "direct") -> MatrixCell:
    bindings = (ExpectedBinding("request", "claude-sonnet", "high"),)
    if route_kind == "execution_preset":
        bindings = (
            ExpectedBinding("selector", "claude-sonnet", "high"),
            ExpectedBinding("plan", "claude-sonnet", "high"),
            ExpectedBinding("work", "claude-sonnet", "high"),
            ExpectedBinding("review", "claude-sonnet", "high"),
        )
    return MatrixCell("claude-direct", "claude", IopCell(
        "claude-sonnet", "high", route_kind, "iop-route", bindings,
    ))


def _workspace(root: Path, session_id: str = "session-fixture") -> PreparedWorkspace:
    workspace = root / "workspace"
    workspace.mkdir()
    return PreparedWorkspace(
        AttemptIdentity("run-20260102T030405Z-abcdef123456", "claude-direct", 1, 1),
        str(root), str(workspace), str(root / "session"), session_id, True,
        "sha256:" + "0" * 64, "isolated",
        TestbedProvenance("/testbed", "main", "0" * 40, "sha256:" + "1" * 64, True),
        "2026-01-02T03:04:05Z",
    )


def _binding_event(cell: MatrixCell) -> str:
    return json.dumps({
        "type": "system", "subtype": "iop_binding", "binding": {
            "cell_id": cell.id, "caller": cell.caller,
            "requested_route_kind": cell.iop.route_kind, "requested_route_id": cell.iop.route_id,
            "requested_model": cell.iop.request_model, "requested_effort": cell.iop.requested_effort,
            "effective_route_kind": cell.iop.route_kind, "effective_route_id": cell.iop.route_id,
            "effective_model": cell.iop.request_model, "effective_effort": cell.iop.requested_effort,
            "effective_bindings": [
                {"stage": item.stage, "model": item.model, "effort": item.effort}
                for item in cell.iop.expected_bindings
            ],
        },
    })


class ClaudeIopTest(unittest.TestCase):
    def setUp(self) -> None:
        # The lifecycle intentionally requires an executable fake CLI.  Some
        # CI hosts mount /tmp noexec, so keep this short-lived directory under
        # the repository worktree instead.
        self.temp = tempfile.TemporaryDirectory(dir=Path.cwd(), prefix=".claude-iop-test-")
        self.root = Path(self.temp.name)
        self.cell = _cell()
        self.workspace = _workspace(self.root)
        self.binary = self.root / "claude"
        self.binary.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
        self.binary.chmod(0o700)
        self.runtime = ClaudeIopRuntime(str(self.binary), SENTINELS[1], SENTINELS[2])

    def tearDown(self) -> None:
        self.temp.cleanup()

    def _adapter(self, *, cell: MatrixCell | None = None) -> ClaudeIopAdapter:
        return ClaudeIopAdapter(cell or self.cell, self.workspace, self.runtime)

    def _fixture_lines(self) -> list[str]:
        fixture = Path("scripts/fixtures/agent-comparison-benchmark/claude-iop-stream.jsonl")
        return fixture.read_text(encoding="utf-8").splitlines()

    def _run_fake(self, lines: list[str], *, exit_code: int = 0):
        self.binary.write_text(textwrap.dedent(f"""\
            #!/usr/bin/env python3
            import os, sys
            assert sys.stdin.read() == {SENTINELS[0]!r}
            assert os.environ["ANTHROPIC_BASE_URL"] == {SENTINELS[1]!r}
            assert os.environ["ANTHROPIC_API_KEY"] == {SENTINELS[2]!r}
            for line in {lines!r}:
                print(line)
            raise SystemExit({exit_code!r})
        """), encoding="utf-8")
        evidence = self.root / f"evidence-{len(tuple(self.root.glob('evidence-*')))}"
        evidence.mkdir()
        adapter = self._adapter()
        result = run_invocation(
            adapter.invocation(SENTINELS[0], evidence, Timeout(5, 1, 1, 1)),
            parse_event=adapter.parser(), redact=adapter.redactor(SENTINELS[0]),
            on_started=lambda _: None,
        )
        durable = "\n".join(
            path.read_text(encoding="utf-8") for path in sorted(evidence.iterdir())
        )
        return result, durable

    def test_exact_iop_only_invocation_and_fresh_workspace(self) -> None:
        adapter = self._adapter()
        with patch.dict(os.environ, {"SSL_CERT_FILE": "/operator/dev-ca.pem", "NODE_EXTRA_CA_CERTS": "/operator/dev-ca.pem"}):
            spec = adapter.invocation(SENTINELS[0], self.root / "evidence", Timeout(5, 1, 1, 1))
        self.assertEqual(spec.cwd, self.workspace.workspace_dir)
        self.assertEqual(spec.submission_mode, "stdin_once")
        self.assertEqual(spec.task_payload, SENTINELS[0].encode())
        self.assertEqual(spec.argv[1:], (
            "--bare", "--print", "--verbose", "--input-format", "text",
            "--output-format", "stream-json", "--model", "claude-sonnet", "--effort", "high",
            "--no-session-persistence", "--permission-mode", "dontAsk",
            "--tools", "Read,Write,Edit", "--allowedTools", "Read,Write,Edit",
        ))
        env = dict(spec.env)
        self.assertEqual(env["ANTHROPIC_BASE_URL"], SENTINELS[1])
        self.assertEqual(env["ANTHROPIC_API_KEY"], SENTINELS[2])
        self.assertEqual(env["SSL_CERT_FILE"], "/operator/dev-ca.pem")
        self.assertEqual(env["NODE_EXTRA_CA_CERTS"], "/operator/dev-ca.pem")
        self.assertNotIn("ANTHROPIC_AUTH_TOKEN", env)
        self.assertNotIn("CLAUDE_CONFIG_DIR", env)
        (self.root / "other").mkdir()
        not_fresh = _workspace(self.root / "other", "session-other")
        object.__setattr__(not_fresh, "session_is_fresh", False)
        with self.assertRaises(ClaudeIopValidationError):
            ClaudeIopAdapter(self.cell, not_fresh, self.runtime)

    def test_direct_and_preset_preflight_are_exact_without_substitution(self) -> None:
        for route_kind in ("direct", "execution_preset"):
            cell = _cell(route_kind)
            event = _binding_event(cell)
            adapter = self._adapter(cell=cell)
            self.assertEqual(adapter.preflight(event).status, "ready")
            mutated = json.loads(event)
            mutated["binding"]["effective_model"] = "fallback"
            with self.assertRaises(ClaudeIopProtocolError):
                parse_preflight_binding(json.dumps(mutated), cell)
        incomplete = json.loads(_binding_event(self.cell))
        del incomplete["binding"]["effective_effort"]
        with self.assertRaises(ClaudeIopProtocolError):
            self._adapter().preflight(json.dumps(incomplete))

    def test_runtime_requires_available_binary_and_complete_iop_config(self) -> None:
        missing_binary = ClaudeIopRuntime(
            str(self.root / "missing-claude"), SENTINELS[1], SENTINELS[2]
        )
        with self.assertRaises(ClaudeIopValidationError):
            ClaudeIopAdapter(self.cell, self.workspace, missing_binary)
        for base_url, api_key in (("", SENTINELS[2]), (SENTINELS[1], "")):
            with self.assertRaises(ClaudeIopValidationError):
                ClaudeIopAdapter(
                    self.cell, self.workspace,
                    ClaudeIopRuntime(str(self.binary), base_url, api_key),
                )

    def test_fixture_uses_production_shaped_ordered_terminal_evidence(self) -> None:
        lines = self._fixture_lines()
        parser = ClaudeStreamParser(self.cell, "session-fixture")
        parsed = [parser("stdout", line) for line in lines]
        self.assertEqual(parsed[:2], [None, CallerEvent("finish")])
        self.assertEqual(parsed[2][-1], CallerEvent("idle"))

        malformed = json.loads(lines[1])
        del malformed["message"]["model"]
        with self.assertRaises(ClaudeIopProtocolError):
            parser = ClaudeStreamParser(self.cell, "session-fixture")
            parser("stdout", lines[0])
            parser("stdout", json.dumps(malformed))
        with self.assertRaises(ClaudeIopProtocolError):
            parser("stdout", "not-json")

    def test_parser_rejects_missing_duplicate_mismatched_and_out_of_order_evidence(self) -> None:
        init, assistant, result = self._fixture_lines()
        wrong_session = json.loads(assistant)
        wrong_session["session_id"] = "other-claude-session"
        wrong_model = json.loads(assistant)
        wrong_model["message"]["model"] = "fallback-model"
        missing_session = json.loads(result)
        del missing_session["session_id"]
        missing_result = [init, assistant]
        cases = {
            "assistant-before-init": [assistant],
            "result-before-assistant": [init, result],
            "duplicate-init": [init, init],
            "duplicate-assistant": [init, assistant, assistant],
            "session-mismatch": [init, json.dumps(wrong_session)],
            "nested-model-mismatch": [init, json.dumps(wrong_model)],
            "missing-result-session": [init, assistant, json.dumps(missing_session)],
        }
        for name, events in cases.items():
            with self.subTest(name=name):
                parser = ClaudeStreamParser(self.cell, "session-fixture")
                with self.assertRaises(ClaudeIopProtocolError):
                    for event in events:
                        parser("stdout", event)
        parser = ClaudeStreamParser(self.cell, "session-fixture")
        self.assertEqual([parser("stdout", event) for event in missing_result], [None, CallerEvent("finish")])

    def test_parser_accepts_partial_snapshots_and_tool_result_cycles(self) -> None:
        init, _, result = self._fixture_lines()
        parser = ClaudeStreamParser(self.cell, "session-fixture")
        partial = json.dumps({
            "type": "assistant", "session_id": "claude-session-fixture",
            "message": {"id": "msg-tool-1", "model": "claude-sonnet", "stop_reason": None,
                        "content": [{"type": "tool_use", "id": "tool-1"}]},
        })
        user = json.dumps({
            "type": "user", "session_id": "claude-session-fixture",
            "message": {"content": [{"type": "tool_result", "tool_use_id": "tool-1"}]},
        })
        final = json.dumps({
            "type": "assistant", "session_id": "claude-session-fixture",
            "message": {"id": "msg-final", "model": "claude-sonnet",
                        "stop_reason": "end_turn", "content": []},
        })
        self.assertEqual(
            [parser("stdout", line) for line in (init, partial, partial, user, final)],
            [None, None, None, None, CallerEvent("finish")],
        )
        observations = parser("stdout", result)
        metrics = {
            metric.name: metric.value for metric in observations
            if isinstance(metric, ParsedMetric)
        }
        self.assertEqual(metrics["model_calls"], 2)

    def test_parser_accepts_cumulative_message_ids_and_direct_result(self) -> None:
        init, _, result = self._fixture_lines()
        parser = ClaudeStreamParser(self.cell, "session-fixture")
        partial = json.dumps({
            "type": "assistant", "session_id": "claude-session-fixture",
            "message": {"id": "msg-cumulative", "model": "claude-sonnet",
                        "stop_reason": None, "content": []},
        })
        user = json.dumps({
            "type": "user", "session_id": "claude-session-fixture",
            "message": {"content": []},
        })
        terminal = json.loads(result)
        terminal["usage"].update({
            "cache_creation": {"ephemeral_1h_input_tokens": 0, "ephemeral_5m_input_tokens": 0},
            "inference_geo": "", "iterations": [],
            "output_tokens_details": {"thinking_tokens": 0},
            "server_tool_use": {"web_fetch_requests": 0, "web_search_requests": 0},
            "service_tier": "standard", "speed": "standard",
        })
        parsed = [
            parser("stdout", line)
            for line in (init, partial, partial, user, partial, user, user, partial)
        ]
        self.assertEqual(parsed, [None] * len(parsed))
        observations = parser("stdout", json.dumps(terminal))
        self.assertEqual(observations[-2:], (CallerEvent("finish"), CallerEvent("idle")))
        self.assertEqual(
            {metric.name: metric.value for metric in observations if isinstance(metric, ParsedMetric)}["model_calls"], 1
        )

    def test_parser_classifies_synthetic_api_error_without_parser_failure(self) -> None:
        init, _, _ = self._fixture_lines()
        parser = ClaudeStreamParser(self.cell, "session-fixture")
        synthetic = json.dumps({
            "type": "assistant", "session_id": "claude-session-fixture",
            "is_api_error_message": True,
            "message": {"id": "synthetic", "model": "<synthetic>",
                        "stop_reason": "stop_sequence", "content": []},
        })
        terminal = json.dumps({
            "type": "result", "session_id": "claude-session-fixture",
            "subtype": "success", "is_error": True, "terminal_reason": "api_error",
        })
        self.assertEqual(
            [parser("stdout", line) for line in (init, synthetic, terminal)],
            [None, None, (
                CallerTerminal("failed", "caller_error"),
                CallerEvent("finish"), CallerEvent("idle"),
            )],
        )

        outcome, _ = self._run_fake([init, synthetic, terminal], exit_code=1)
        self.assertFalse(outcome.product.status == "succeeded")
        self.assertEqual(outcome.product.status, "failed")
        self.assertEqual(outcome.harness.status, "passed")

    def test_lifecycle_accepts_result_direct_active_snapshot(self) -> None:
        init, _, result = self._fixture_lines()
        partial = json.dumps({
            "type": "assistant", "session_id": "claude-session-fixture",
            "message": {"id": "msg-direct", "model": "claude-sonnet",
                        "stop_reason": None, "content": []},
        })
        outcome, _ = self._run_fake([init, partial, result])
        self.assertTrue(outcome.product.status == "succeeded", outcome)
        kinds = [event.kind for event in outcome.events]
        self.assertLess(kinds.index("finish"), kinds.index("idle"))

    def test_parser_accepts_explicit_tool_use_terminal_once(self) -> None:
        init, _, result = self._fixture_lines()
        parser = ClaudeStreamParser(self.cell, "session-fixture")
        tool_use = json.dumps({
            "type": "assistant", "session_id": "claude-session-fixture",
            "message": {"id": "msg-tool-1", "model": "claude-sonnet",
                        "stop_reason": "tool_use", "content": []},
        })
        user = json.dumps({
            "type": "user", "session_id": "claude-session-fixture", "message": {"content": []},
        })
        final = json.dumps({
            "type": "assistant", "session_id": "claude-session-fixture",
            "message": {"id": "msg-final", "model": "claude-sonnet",
                        "stop_reason": "end_turn", "content": []},
        })
        self.assertEqual(
            [parser("stdout", line) for line in (init, tool_use, user, final)],
            [None, None, None, CallerEvent("finish")],
        )
        self.assertEqual(
            {metric.name: metric.value for metric in parser("stdout", result) if isinstance(metric, ParsedMetric)}["model_calls"],
            2,
        )
        duplicate = ClaudeStreamParser(self.cell, "session-fixture")
        duplicate("stdout", init)
        duplicate("stdout", tool_use)
        duplicate("stdout", user)
        with self.assertRaises(ClaudeIopProtocolError):
            duplicate("stdout", tool_use)

    def test_reported_result_values_become_bound_normalized_observations(self) -> None:
        init, assistant, result = self._fixture_lines()
        parser = ClaudeStreamParser(self.cell, "session-fixture")
        parser("stdout", init)
        parser("stdout", assistant)
        parsed = parser("stdout", result)
        self.assertEqual(parsed[-1], CallerEvent("idle"))
        observed = {metric.name: metric for metric in parsed if isinstance(metric, ParsedMetric)}
        self.assertEqual(observed["total_duration"].value, 1234 * 10 ** 6)
        self.assertFalse(observed["total_duration"].overlap)
        # The reported API duration is inside the reported total, so it is
        # published as an overlapping interval and never subtracted from it.
        self.assertEqual(observed["model_duration"].value, 1000 * 10 ** 6)
        self.assertTrue(observed["model_duration"].overlap)
        self.assertEqual(observed["model_calls"].value, 1)
        self.assertEqual(observed["input_tokens"].value, 11)
        self.assertEqual(observed["output_tokens"].value, 22)
        self.assertEqual(observed["cached_input_tokens"].value, 5)
        # The fixture omits cache creation, so that category stays unreported
        # rather than being reported as zero.
        self.assertNotIn("cache_write_tokens", observed)
        for metric in parsed:
            if not isinstance(metric, ParsedMetric):
                continue
            self.assertEqual(metric.model, "claude-sonnet")
            self.assertEqual(metric.source, "caller_output")
        self.assertEqual(observed["input_tokens"].clock, "none")
        self.assertEqual(observed["total_duration"].clock, "caller_reported")

    def test_unknown_negative_boolean_and_fractional_usage_fails_closed(self) -> None:
        init, assistant, result = self._fixture_lines()
        base = json.loads(result)
        cases = {
            "unknown-usage-key": {"usage": {"input_tokens": 1, "web_search_requests": 2}},
            "fractional-token": {"usage": {"input_tokens": 1.5}},
            "boolean-token": {"usage": {"input_tokens": True}},
            "negative-token": {"usage": {"input_tokens": -1}},
            "string-duration": {"duration_ms": "1234"},
            "boolean-duration": {"duration_ms": True},
            "negative-duration": {"duration_ms": -5},
            "non-object-usage": {"usage": [1, 2]},
        }
        for name, override in cases.items():
            with self.subTest(name=name):
                parser = ClaudeStreamParser(self.cell, "session-fixture")
                parser("stdout", init)
                parser("stdout", assistant)
                with self.assertRaises(ClaudeIopProtocolError):
                    parser("stdout", json.dumps({**base, **override}))

    def test_structural_redaction_never_retains_sensitive_content(self) -> None:
        raw = json.dumps({
            "type": "result", "result": "result-secret-sentinel",
            "message": "error-secret-sentinel",
            "content": "prompt-secret-sentinel",
            "tool_input": {"arguments": "tool-secret-sentinel"},
            "diagnostic": "https://private.iop.invalid api-secret-sentinel",
        })
        redacted = redact_claude_event(raw, SENTINELS)
        error_redacted = redact_claude_event(json.dumps({
            "type": "error", "message": "error-secret-sentinel",
            "error": {"detail": "tool-secret-sentinel"},
        }), SENTINELS)
        for sentinel in (*SENTINELS, *ARBITRARY_SENTINELS):
            self.assertNotIn(sentinel, redacted)
            self.assertNotIn(sentinel, error_redacted)
        self.assertIn("[redacted]", redacted)
        self.assertEqual(redact_claude_event("raw prompt-secret-sentinel", SENTINELS),
                         '{"type":"invalid_claude_json"}')

    def test_fake_cli_runs_once_and_durable_evidence_is_redacted(self) -> None:
        init, assistant, result = (json.loads(line) for line in self._fixture_lines())
        assistant["message"]["content"] = SENTINELS[0]
        assistant["tool_input"] = {"arguments": ARBITRARY_SENTINELS[0]}
        result["result"] = ARBITRARY_SENTINELS[1]
        diagnostic = {"type": "system", "subtype": "notice", "error": ARBITRARY_SENTINELS[2]}
        outcome, durable = self._run_fake([
            json.dumps(init), json.dumps(diagnostic), json.dumps(assistant), json.dumps(result),
        ])
        result = outcome
        self.assertTrue(result.product.status == "succeeded", result)
        self.assertEqual(result.harness.reason, REASON_SUCCESS)
        self.assertTrue(result.submitted)
        self.assertTrue(result.harness.ordered_terminal)
        for sentinel in (*SENTINELS, *ARBITRARY_SENTINELS):
            self.assertNotIn(sentinel, durable)
        self.assertEqual(
            [metric.name for metric in result.metrics],
            [
                "total_duration", "model_duration", "model_calls",
                "cached_input_tokens", "input_tokens", "output_tokens",
            ],
        )
        # Numeric observations survive the content redactors that removed every
        # sentinel above, and the durable journal keeps them in one closed shape.
        self.assertIn('"kind": "metric:total_duration"', durable)
        self.assertIn('\\"value\\":1234000000', durable)

    def test_lifecycle_rejects_boundary_violations(self) -> None:
        init, assistant, result = self._fixture_lines()
        bad_model = json.loads(assistant)
        bad_model["message"]["model"] = "fallback-model"
        cases = {
            "missing-init": [assistant, result],
            "missing-result": [init, assistant],
            "duplicate-init": [init, init, assistant, result],
            "out-of-order-result": [init, result, assistant],
            "mismatched-model": [init, json.dumps(bad_model), result],
            "malformed-json": [init, "not-json", assistant, result],
        }
        for name, lines in cases.items():
            with self.subTest(name=name):
                outcome, durable = self._run_fake(lines)
                self.assertFalse(outcome.product.status == "succeeded", outcome)
                for sentinel in (*SENTINELS, *ARBITRARY_SENTINELS):
                    self.assertNotIn(sentinel, durable)

    def test_metric_prefixed_malformed_output_is_redacted_before_durable_capture(self) -> None:
        metric_sentinel = "metric:" + SENTINELS[0]
        outcome, durable = self._run_fake([metric_sentinel])
        self.assertFalse(outcome.product.status == "succeeded", outcome)
        self.assertEqual(outcome.harness.reason, REASON_PARSER_ERROR)
        self.assertTrue(outcome.submitted)
        self.assertIn("invalid_claude_json", durable)
        for sentinel in (*SENTINELS, metric_sentinel, SENTINELS[0]):
            self.assertNotIn(sentinel, durable)


if __name__ == "__main__":
    unittest.main()