toki
de4d8f4ff8
세 Agent의 direct route를 동일한 fail-closed preflight와 격리 실행 경계에서 비교하고, 관측되지 않은 preset 셀이 실행되는 것을 막기 위해 연결 계약과 증거 수집 흐름을 고정한다.
284 lines
13 KiB
Python
284 lines
13 KiB
Python
"""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 scripts.agent_benchmark.claude_iop import (
|
|
ClaudeIopAdapter,
|
|
ClaudeIopProtocolError,
|
|
ClaudeIopRuntime,
|
|
ClaudeIopValidationError,
|
|
ClaudeStreamParser,
|
|
parse_preflight_binding,
|
|
redact_claude_event,
|
|
)
|
|
from scripts.agent_benchmark.lifecycle import 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]):
|
|
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)
|
|
"""), 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()
|
|
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=",
|
|
))
|
|
env = dict(spec.env)
|
|
self.assertEqual(env["ANTHROPIC_BASE_URL"], SENTINELS[1])
|
|
self.assertEqual(env["ANTHROPIC_API_KEY"], SENTINELS[2])
|
|
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")
|
|
self.assertEqual([parser("stdout", line) for line in lines], [None, "finish", "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, "finish"])
|
|
|
|
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.success, result)
|
|
self.assertEqual(result.terminal_reason, REASON_SUCCESS)
|
|
self.assertTrue(result.submitted)
|
|
self.assertTrue(result.finish_then_idle_then_quiet)
|
|
for sentinel in (*SENTINELS, *ARBITRARY_SENTINELS):
|
|
self.assertNotIn(sentinel, 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.success, 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.success, outcome)
|
|
self.assertEqual(outcome.terminal_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()
|