toki
634531afca
실패를 완료되지 않은 실행과 혼동해 동일 사이클을 반복하던 조건을 분리하고, 이미 검증된 dispatcher 작업 이력과 benchmark 결과를 보존한다.
294 lines
15 KiB
Python
294 lines
15 KiB
Python
"""Credential-free tests for the isolated Codex IOP benchmark adapter."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from scripts.agent_benchmark.codex_iop import (
|
|
BASE_URL_ENV_KEY,
|
|
SECRET_ENV_KEY,
|
|
CodexJSONLError,
|
|
CodexJSONLParser,
|
|
CodexRuntimeError,
|
|
build_codex_invocation,
|
|
build_codex_spec,
|
|
redact_codex_jsonl,
|
|
run_codex_invocation,
|
|
runtime_from_environment,
|
|
)
|
|
from scripts.agent_benchmark.lifecycle import (
|
|
CallerEvent,
|
|
ParsedMetric,
|
|
REASON_DUPLICATE_EVENT,
|
|
REASON_NONZERO_EXIT,
|
|
REASON_PARSER_ERROR,
|
|
)
|
|
from scripts.agent_benchmark.manifest import ExpectedBinding, IopCell, MatrixCell, Timeout
|
|
from scripts.agent_benchmark.workspace import AttemptIdentity, PreparedWorkspace, TestbedProvenance
|
|
|
|
|
|
_ENDPOINT = "https://iop.private.example/v1"
|
|
_SECRET = "iop_test_secret_123456789"
|
|
_PROMPT = b"private benchmark prompt must not persist"
|
|
|
|
|
|
def _cell(*, effort: str = "xhigh") -> MatrixCell:
|
|
return MatrixCell(
|
|
"codex-gpt", "codex",
|
|
IopCell("gpt-5.6-luna", effort, "direct", "iop-gpt", (
|
|
ExpectedBinding("request", "gpt-5.6-luna", effort),
|
|
)),
|
|
)
|
|
|
|
|
|
class CodexIOPTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.tmp.name)
|
|
self.workspace = self.root / "workspace"
|
|
self.workspace.mkdir()
|
|
self.session = self.root / "session"
|
|
self.session.mkdir()
|
|
self.evidence = self.root / "attempt"
|
|
self.evidence.mkdir()
|
|
|
|
def tearDown(self) -> None:
|
|
self.tmp.cleanup()
|
|
|
|
def _runtime(self):
|
|
return runtime_from_environment({
|
|
BASE_URL_ENV_KEY: _ENDPOINT,
|
|
SECRET_ENV_KEY: _SECRET,
|
|
"PATH": os.environ.get("PATH", "/usr/bin:/bin"),
|
|
})
|
|
|
|
def _prepared(self) -> PreparedWorkspace:
|
|
return PreparedWorkspace(
|
|
identity=AttemptIdentity("run-0001", "codex-gpt", 1, 1),
|
|
attempt_root=str(self.evidence), workspace_dir=str(self.workspace),
|
|
session_dir=str(self.session), session_id="fresh-session", session_is_fresh=True,
|
|
workspace_checksum="sha256:" + "0" * 64, setup_cache_policy="isolated",
|
|
testbed_provenance=TestbedProvenance("../iop-s2", "main", "0" * 40, "0" * 64, True),
|
|
prepared_at="2026-08-10T00:00:00+00:00",
|
|
)
|
|
|
|
def _timeout(self) -> Timeout:
|
|
return Timeout(5, 2, 1, 1)
|
|
|
|
def _fake_codex(self, records: list[object], exit_code: int = 0) -> str:
|
|
path = self.root / f"fake-codex-{len(list(self.root.glob('fake-codex-*')))}.py"
|
|
lines = [
|
|
"#!/usr/bin/env python3", "import json, sys", "task = sys.stdin.read()",
|
|
]
|
|
for record in records:
|
|
if record == "TASK":
|
|
lines.append("print(json.dumps({'type': 'turn.completed', 'status': 'completed', 'content': task}), flush=True)")
|
|
elif isinstance(record, str):
|
|
lines.append(f"print({record!r}, flush=True)")
|
|
else:
|
|
lines.append(f"print(json.dumps({record!r}), flush=True)")
|
|
lines.append(f"raise SystemExit({exit_code})")
|
|
path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
return str(path)
|
|
|
|
def test_exact_isolated_responses_spec_uses_one_stdin_submission(self) -> None:
|
|
with patch.dict(os.environ, {"SSL_CERT_FILE": "/operator/dev-ca.pem", "NODE_EXTRA_CA_CERTS": "/operator/dev-ca.pem"}):
|
|
spec = build_codex_spec(_cell(), self._prepared(), self._runtime(), _PROMPT, self._timeout())
|
|
argv = list(spec.argv)
|
|
codex = argv[argv.index("--") + 1:]
|
|
self.assertFalse((self.workspace / ".git").exists())
|
|
self.assertEqual(codex.count("--skip-git-repo-check"), 1)
|
|
self.assertEqual(codex[:14], [
|
|
"codex", "exec", "--sandbox", "workspace-write", "--json", "--ephemeral", "--ignore-user-config",
|
|
"--strict-config", "--skip-git-repo-check", "-C", str(self.workspace),
|
|
"-m", "gpt-5.6-luna", "-c",
|
|
])
|
|
self.assertEqual(codex[-1], "-")
|
|
overrides = [codex[index + 1] for index, item in enumerate(codex[:-1]) if item == "-c"]
|
|
self.assertEqual(overrides, [
|
|
'model_provider="iop_benchmark"',
|
|
'model_providers.iop_benchmark.name="IOP Benchmark"',
|
|
f'model_providers.iop_benchmark.base_url="{_ENDPOINT}"',
|
|
'model_providers.iop_benchmark.env_key="IOP_BENCHMARK_API_KEY"',
|
|
'shell_environment_policy.filters.IOP_BENCHMARK_API_KEY="exclude"',
|
|
'model_providers.iop_benchmark.wire_api="responses"',
|
|
'model_reasoning_effort="xhigh"',
|
|
])
|
|
self.assertEqual(spec.submission_mode, "stdin_once")
|
|
self.assertEqual(spec.completion_mode, "exit_after_idle")
|
|
self.assertEqual(dict(spec.env)[SECRET_ENV_KEY], _SECRET)
|
|
self.assertEqual(dict(spec.env)["SSL_CERT_FILE"], "/operator/dev-ca.pem")
|
|
self.assertEqual(dict(spec.env)["NODE_EXTRA_CA_CERTS"], "/operator/dev-ca.pem")
|
|
self.assertNotIn(BASE_URL_ENV_KEY, dict(spec.env))
|
|
self.assertNotIn("OPENAI_API_KEY", dict(spec.env))
|
|
|
|
def test_provider_secret_is_explicitly_excluded_from_shell_environment(self) -> None:
|
|
"""Regression: the provider env_key injects the secret into the child
|
|
environment for authentication, but the same secret is explicitly
|
|
excluded from shell tool environments and snapshots.
|
|
|
|
Provider authentication, shell exclusion, and the exact invocation
|
|
spec are one invariant; both must hold together.
|
|
"""
|
|
with patch.dict(os.environ, {"SSL_CERT_FILE": "/operator/dev-ca.pem", "NODE_EXTRA_CA_CERTS": "/operator/dev-ca.pem"}):
|
|
spec = build_codex_spec(_cell(), self._prepared(), self._runtime(), _PROMPT, self._timeout())
|
|
env = dict(spec.env)
|
|
# Provider authentication: the secret must still reach the child process.
|
|
self.assertEqual(env[SECRET_ENV_KEY], _SECRET)
|
|
# Shell exclusion: the exact override must appear in the invocation.
|
|
argv = list(spec.argv)
|
|
codex = argv[argv.index("--") + 1:]
|
|
overrides = {codex[index + 1] for index, item in enumerate(codex[:-1]) if item == "-c"}
|
|
self.assertIn(
|
|
f'shell_environment_policy.filters.IOP_BENCHMARK_API_KEY="exclude"',
|
|
overrides,
|
|
"the benchmark secret must be explicitly excluded from shell environments",
|
|
)
|
|
# TLS variables, isolated HOME, and no leaking caller keys.
|
|
self.assertEqual(env["SSL_CERT_FILE"], "/operator/dev-ca.pem")
|
|
self.assertEqual(env["NODE_EXTRA_CA_CERTS"], "/operator/dev-ca.pem")
|
|
self.assertNotIn(BASE_URL_ENV_KEY, env)
|
|
self.assertNotIn("OPENAI_API_KEY", env)
|
|
|
|
def test_runtime_and_effort_are_closed(self) -> None:
|
|
with self.assertRaises(CodexRuntimeError):
|
|
runtime_from_environment({BASE_URL_ENV_KEY: _ENDPOINT, SECRET_ENV_KEY: _SECRET})
|
|
with self.assertRaises(CodexRuntimeError):
|
|
runtime_from_environment({BASE_URL_ENV_KEY: _ENDPOINT, SECRET_ENV_KEY: _SECRET, "PATH": "/bin", "EXTRA": "x"})
|
|
with self.assertRaises(CodexRuntimeError):
|
|
build_codex_spec(_cell(effort="high"), self._prepared(), self._runtime(), _PROMPT, self._timeout())
|
|
|
|
def test_fixture_finish_then_verified_idle_and_structural_redaction(self) -> None:
|
|
parser = CodexJSONLParser(_cell(), "fixture-nonce-0001")
|
|
fixture = Path("scripts/fixtures/agent-comparison-benchmark/codex-iop-stream.jsonl")
|
|
events = [parser.parse("stdout", line) for line in fixture.read_text(encoding="utf-8").splitlines()]
|
|
self.assertEqual([events[0], events[-1]], [None, CallerEvent("idle")])
|
|
tool = events[1]
|
|
self.assertEqual((tool.name, tool.value, tool.call_id), ("tool_duration", 7_250_000, "call-1"))
|
|
# A tool interval is reported inside the turn, so it is published as an
|
|
# overlapping interval instead of a subtractable slice.
|
|
self.assertTrue(tool.overlap)
|
|
self.assertEqual(events[2][-1], CallerEvent("finish"))
|
|
turn = {metric.name: metric.value for metric in events[2] if isinstance(metric, ParsedMetric)}
|
|
self.assertEqual(turn, {
|
|
"cache_write_tokens": 6, "cached_input_tokens": 8,
|
|
"input_tokens": 31, "output_tokens": 12, "reasoning_tokens": 4,
|
|
"model_calls": 1, "tool_calls": 1,
|
|
})
|
|
# The fixture omits the provider total, so it is never reconstructed
|
|
# from the reported categories.
|
|
self.assertNotIn("total_tokens", turn)
|
|
redacted = redact_codex_jsonl(
|
|
json.dumps({"type": "turn.completed", "content": _PROMPT.decode(), "endpoint": _ENDPOINT, "token": _SECRET}),
|
|
(_ENDPOINT, _SECRET, _PROMPT.decode()),
|
|
)
|
|
self.assertNotIn(_PROMPT.decode(), redacted)
|
|
self.assertNotIn(_ENDPOINT, redacted)
|
|
self.assertNotIn(_SECRET, redacted)
|
|
self.assertIn("[redacted]", redacted)
|
|
|
|
def test_bridge_proves_finish_then_idle_after_child_exit(self) -> None:
|
|
fake = self._fake_codex(["TASK"])
|
|
invocation = build_codex_invocation(_cell(), self._prepared(), self._runtime(), _PROMPT, self._timeout(), codex_executable=(sys.executable, fake))
|
|
result = run_codex_invocation(invocation, lambda _: None)
|
|
self.assertTrue(result.lifecycle.product.status == "succeeded")
|
|
self.assertEqual([event.kind for event in result.lifecycle.events], [
|
|
"submitted", "first_output", "metric:model_calls", "metric:tool_calls",
|
|
"caller_terminal", "finish", "idle", "exited", "quiet",
|
|
])
|
|
capture = result.lifecycle.stdout.text
|
|
self.assertNotIn(_PROMPT.decode(), capture)
|
|
self.assertNotIn(_ENDPOINT, capture)
|
|
self.assertNotIn(_SECRET, capture)
|
|
self.assertIsNone(result.effective_binding)
|
|
|
|
def test_child_failure_never_synthesizes_idle(self) -> None:
|
|
fake = self._fake_codex([{"type": "turn.completed", "status": "completed"}], exit_code=7)
|
|
result = run_codex_invocation(build_codex_invocation(_cell(), self._prepared(), self._runtime(), _PROMPT, self._timeout(), codex_executable=(sys.executable, fake)), lambda _: None)
|
|
self.assertFalse(result.lifecycle.product.status == "succeeded")
|
|
self.assertEqual(result.lifecycle.harness.reason, REASON_NONZERO_EXIT)
|
|
self.assertNotIn("idle", [event.kind for event in result.lifecycle.events])
|
|
|
|
def test_duplicate_malformed_and_unverified_idle_fail_closed(self) -> None:
|
|
cases = (
|
|
([{"type": "turn.completed", "status": "completed"}, {"type": "turn.completed", "status": "completed"}], REASON_DUPLICATE_EVENT),
|
|
(["not-json"], REASON_PARSER_ERROR),
|
|
([{"type": "adapter.idle", "adapter": "codex_iop", "nonce": "not-the-bridge-nonce", "child_exit": 0}], REASON_PARSER_ERROR),
|
|
)
|
|
for index, (records, reason) in enumerate(cases):
|
|
with self.subTest(reason=reason):
|
|
evidence = self.root / f"attempt-{reason}-{index}"
|
|
evidence.mkdir()
|
|
prepared = self._prepared().__class__(**{**self._prepared().__dict__, "attempt_root": str(evidence)})
|
|
fake = self._fake_codex(records)
|
|
result = run_codex_invocation(build_codex_invocation(_cell(), prepared, self._runtime(), _PROMPT, self._timeout(), codex_executable=(sys.executable, fake)), lambda _: None)
|
|
self.assertFalse(result.lifecycle.product.status == "succeeded")
|
|
self.assertEqual(result.lifecycle.harness.reason, reason)
|
|
|
|
def test_tool_intervals_require_one_explicit_unique_pairing(self) -> None:
|
|
parser = CodexJSONLParser(_cell(), "0123456789abcdef")
|
|
# A non-tool item carries neither a tool count nor an interval.
|
|
self.assertIsNone(parser.parse("stdout", json.dumps(
|
|
{"type": "item.completed", "item": {"type": "agent_message", "text": "x"}}
|
|
)))
|
|
# Real Codex command completions omit duration; they must still count.
|
|
self.assertIsNone(parser.parse("stdout", json.dumps(
|
|
{"type": "item.completed", "item": {"id": "call-0", "type": "command_execution"}}
|
|
)))
|
|
paired = json.dumps({"type": "item.completed", "item": {
|
|
"id": "call-1", "type": "command_execution", "duration_ms": 3,
|
|
}})
|
|
self.assertEqual(parser.parse("stdout", paired).call_id, "call-1")
|
|
for name, item in {
|
|
"duplicate-call": {"id": "call-1", "type": "command_execution", "duration_ms": 4},
|
|
"unpaired-duration": {"type": "command_execution", "duration_ms": 4},
|
|
"unsafe-call-id": {"id": "call 1", "type": "command_execution", "duration_ms": 4},
|
|
"string-duration": {"id": "call-2", "type": "command_execution", "duration_ms": "4"},
|
|
"negative-duration": {"id": "call-3", "type": "command_execution", "duration_ms": -4},
|
|
}.items():
|
|
with self.subTest(name=name):
|
|
with self.assertRaises(CodexJSONLError):
|
|
parser.parse("stdout", json.dumps({"type": "item.completed", "item": item}))
|
|
turn = parser.parse("stdout", json.dumps({"type": "turn.completed"}))
|
|
counts = {metric.name: metric.value for metric in turn if isinstance(metric, ParsedMetric)}
|
|
self.assertEqual(counts, {"model_calls": 1, "tool_calls": 2})
|
|
|
|
def test_unknown_or_fractional_turn_usage_fails_closed(self) -> None:
|
|
for usage in (
|
|
{"unknown_tokens": 1},
|
|
{"input_tokens": 1.5},
|
|
{"input_tokens": True},
|
|
{"input_tokens": -1},
|
|
):
|
|
with self.subTest(usage=usage):
|
|
parser = CodexJSONLParser(_cell(), "0123456789abcdef")
|
|
with self.assertRaises(CodexJSONLError):
|
|
parser.parse("stdout", json.dumps({
|
|
"type": "turn.completed", "status": "completed", "usage": usage,
|
|
}))
|
|
|
|
def test_effective_binding_is_optional_but_any_observation_is_exact(self) -> None:
|
|
parser = CodexJSONLParser(_cell(), "0123456789abcdef")
|
|
self.assertIsNone(parser.effective_binding)
|
|
self.assertEqual(parser.connectivity_result().status, "implementation_gap")
|
|
parser.parse("stdout", json.dumps({"type": "thread.started", "iop_effective_binding": {
|
|
"route_kind": "direct", "route_id": "iop-gpt", "model": "gpt-5.6-luna", "effort": "xhigh",
|
|
}}))
|
|
self.assertEqual(parser.effective_binding, ("direct", "iop-gpt", "gpt-5.6-luna", "xhigh"))
|
|
with self.assertRaises(CodexJSONLError):
|
|
parser.connectivity_result()
|
|
mismatch = CodexJSONLParser(_cell(), "0123456789abcdef")
|
|
with self.assertRaises(CodexJSONLError):
|
|
mismatch.parse("stdout", json.dumps({"type": "thread.started", "iop_effective_binding": {
|
|
"route_kind": "direct", "route_id": "iop-gpt", "model": "gpt-alias", "effort": "xhigh",
|
|
}}))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|