"""Credential-free tests for the fail-closed agy IOP adapter.""" from __future__ import annotations import json import os import sys import tempfile import unittest from dataclasses import replace from pathlib import Path from scripts.agent_benchmark.agy_iop import ( AGY_AUTH_ENV, AGY_ENDPOINT_ENV, AGY_KNOWN_VERSION, AGY_PROVIDER_ENV, AgyAdapterError, AgyEventParser, AgyRuntimeInputs, AgyRuntimeObservation, build_agy_invocation, inspect_agy_iop_capability, preflight_agy_iop, redact_agy_event, run_agy_invocation, ) from scripts.agent_benchmark.lifecycle import ( REASON_DUPLICATE_EVENT, REASON_MALFORMED_EVENT, InvocationSpec, env_pairs, ) from scripts.agent_benchmark.manifest import ExpectedBinding, IopCell, MatrixCell, Timeout from scripts.agent_benchmark.workspace import AttemptIdentity, PreparedWorkspace, TestbedProvenance def _help(*, transport: bool = True) -> str: basic = "--print --output-format stream-json --sandbox --model --effort" return basic + (f" {AGY_PROVIDER_ENV} {AGY_ENDPOINT_ENV} {AGY_AUTH_ENV}" if transport else "") def _cell() -> MatrixCell: return MatrixCell( "agy-direct", "agy", IopCell("gemini-2.0-flash", "high", "direct", "agy-direct", ( ExpectedBinding("request", "gemini-2.0-flash", "high"), )), ) def _iop_config_observation() -> AgyRuntimeObservation: """Fixed evidence from the independent IOP config owner for this cell.""" return AgyRuntimeObservation( "agy-direct", "direct", "agy-direct", "sha256:feb4c33d4e775c775bfb3c333fdb7d4f97069af31c8e824094fb13181fad53d3", "sha256:ab1b96f33fc4a662c870f349d92c54bc8e2574028fa41b79526d4edaf6f49daa", "sha256:" + "c" * 64, ) class AgyIopTest(unittest.TestCase): def setUp(self) -> None: self.temp = tempfile.TemporaryDirectory() self.root = Path(self.temp.name) self.workspace = self.root / "workspace" self.workspace.mkdir() self.runtime = AgyRuntimeInputs(sys.executable, "https://private.invalid/v1", "iop_secret_123456789") def tearDown(self) -> None: self.temp.cleanup() def _prepared(self) -> PreparedWorkspace: return PreparedWorkspace( AttemptIdentity("run", "agy-direct", 1, 1), str(self.root), str(self.workspace), str(self.root / "session"), "fresh-session", True, "sha256:" + "0" * 64, "isolated", TestbedProvenance("/testbed", "main", "0" * 40, "sha256:" + "1" * 64, True), "2026-01-01T00:00:00+00:00", ) def _preflight(self, *, runtime: AgyRuntimeInputs | None = None, help_text: str | None = None): values = self.runtime if runtime is None else runtime return preflight_agy_iop( _cell(), inspect_agy_iop_capability("agy 1.1.11", _help() if help_text is None else help_text), values, _iop_config_observation(), ) def _run_lines(self, lines: list[str], parser: AgyEventParser, preflight): evidence = self.root / f"evidence-{len(list(self.root.glob('evidence-*')))}" evidence.mkdir() source = "import sys; lines=" + repr(lines) + "; [print(line) for line in lines]" spec = InvocationSpec( argv=(sys.executable, "-u", "-c", source), cwd=str(self.root), env=env_pairs({"PATH": os.environ.get("PATH", "/usr/bin:/bin")}), submission_mode="stdin_once", completion_mode="exit_after_idle", timeout=Timeout(5, 1, 1, 1), evidence_dir=str(evidence), task_payload=b"task", ) return run_agy_invocation(spec, parser, preflight, lambda _: None) def test_absent_or_unknown_transport_never_constructs_launch(self) -> None: for version, help_text, expected in ( ("agy 1.1.11", _help(transport=False), ("endpoint_incompatible", "auth_incompatible", "protocol_incompatible")), ("agy 9.9.9", _help(), "protocol_incompatible"), ): with self.subTest(version=version): preflight = preflight_agy_iop( _cell(), inspect_agy_iop_capability(version, help_text), self.runtime, _iop_config_observation(), ) self.assertEqual(preflight.status, "implementation_gap") expected_codes = (expected,) if isinstance(expected, str) else expected self.assertEqual([item.code for item in preflight.issues], list(expected_codes)) with self.assertRaises(AgyAdapterError): build_agy_invocation(_cell(), self._prepared(), b"task", Timeout(5, 1, 1, 1), preflight) def test_installed_public_surface_is_exact_fail_closed_gap(self) -> None: public_help = "--print --output-format stream-json --sandbox --model --effort" for transport_name in (AGY_PROVIDER_ENV, AGY_ENDPOINT_ENV, AGY_AUTH_ENV): self.assertNotIn(transport_name, public_help) capability = inspect_agy_iop_capability("1.1.11", public_help) self.assertEqual(capability.version, AGY_KNOWN_VERSION) self.assertTrue(capability.stream_supported) self.assertFalse(capability.endpoint_supported) self.assertFalse(capability.auth_supported) self.assertFalse(capability.protocol_supported) self.assertFalse(capability.iop_transport_supported) preflight = preflight_agy_iop( _cell(), capability, self.runtime, _iop_config_observation() ) self.assertEqual(preflight.status, "implementation_gap") self.assertEqual( [item.code for item in preflight.issues], ["endpoint_incompatible", "auth_incompatible", "protocol_incompatible"], ) with self.assertRaises(AgyAdapterError): build_agy_invocation( _cell(), self._prepared(), b"task", Timeout(5, 1, 1, 1), preflight ) def test_non_ready_preflight_cannot_start_supplied_invocation(self) -> None: preflight = self._preflight(help_text=_help(transport=False)) self.assertEqual(preflight.status, "implementation_gap") self.assertIsNotNone(preflight.runtime) marker = self.root / "caller-launched" evidence = self.root / "blocked-evidence" source = "from pathlib import Path; Path(" + repr(str(marker)) + ").write_text('launched')" spec = InvocationSpec( argv=(sys.executable, "-u", "-c", source), cwd=str(self.root), env=env_pairs({"PATH": os.environ.get("PATH", "/usr/bin:/bin")}), submission_mode="stdin_once", completion_mode="exit_after_idle", timeout=Timeout(5, 1, 1, 1), evidence_dir=str(evidence), task_payload=b"task", ) started: list[object] = [] with self.assertRaisesRegex(AgyAdapterError, "agy IOP transport is not proven"): run_agy_invocation(spec, AgyEventParser(_cell()), preflight, started.append) self.assertFalse(marker.exists()) self.assertEqual(started, []) self.assertFalse(evidence.exists()) def test_registration_gaps_remain_distinct_from_implementation_gap(self) -> None: no_credential = AgyRuntimeInputs(sys.executable, self.runtime.endpoint, "") supported = inspect_agy_iop_capability("agy 1.1.11", _help()) result = preflight_agy_iop( _cell(), supported, no_credential, _iop_config_observation() ) self.assertEqual(result.status, "registration_required") self.assertEqual([item.code for item in result.issues], ["credential_missing"]) gap = preflight_agy_iop( _cell(), inspect_agy_iop_capability("agy 1.1.11", _help(transport=False)), no_credential, _iop_config_observation(), ) self.assertEqual(gap.status, "implementation_gap") self.assertEqual([item.code for item in gap.issues], ["credential_missing", "endpoint_incompatible", "auth_incompatible", "protocol_incompatible"]) def test_endpoint_auth_and_protocol_gaps_are_exact(self) -> None: cases = ( (_help().replace(AGY_ENDPOINT_ENV, ""), "endpoint_incompatible"), (_help().replace(AGY_AUTH_ENV, ""), "auth_incompatible"), (_help().replace("--sandbox", ""), "protocol_incompatible"), ) for help_text, expected in cases: with self.subTest(expected=expected): outcome = self._preflight(help_text=help_text) self.assertEqual([item.code for item in outcome.issues], [expected]) unknown = inspect_agy_iop_capability(None, None) # type: ignore[arg-type] self.assertFalse(unknown.iop_transport_supported) def test_build_is_fresh_stdin_sandbox_and_iop_only(self) -> None: preflight = self._preflight() spec = build_agy_invocation(_cell(), self._prepared(), b"one task", Timeout(5, 1, 1, 1), preflight) self.assertEqual(spec.submission_mode, "stdin_once") self.assertIn("--print", spec.argv) self.assertIn("--sandbox", spec.argv) self.assertNotIn("--resume", spec.argv) environment = dict(spec.env) self.assertEqual(environment[AGY_PROVIDER_ENV], "iop-openai") self.assertEqual(environment[AGY_ENDPOINT_ENV], self.runtime.endpoint) self.assertEqual(environment[AGY_AUTH_ENV], self.runtime.credential) def test_exact_help_tokens_and_stream_format_gate(self) -> None: lookalike = _help().replace("--print", "--print-json").replace( AGY_ENDPOINT_ENV, AGY_ENDPOINT_ENV + "_EXTRA" ).replace("stream-json", "stream-jsonl") capability = inspect_agy_iop_capability("agy 1.1.11", lookalike) self.assertFalse(capability.iop_transport_supported) self.assertFalse(capability.endpoint_supported) self.assertFalse(capability.stream_supported) missing_stream = self._preflight(help_text=_help().replace("stream-json", "")) self.assertEqual([issue.code for issue in missing_stream.issues], ["stream_incompatible"]) def test_unvalidated_runtime_cannot_launch(self) -> None: observation = _iop_config_observation() for mismatched in ( replace(observation, cell_id="other-cell"), replace(observation, route_id="other-route"), replace(observation, endpoint_identity="sha256:" + "d" * 64), replace(observation, config_identity="not-a-config-identity"), ): with self.subTest(observation=mismatched): preflight = preflight_agy_iop( _cell(), inspect_agy_iop_capability("agy 1.1.11", _help()), self.runtime, mismatched ) self.assertEqual(preflight.status, "implementation_gap") self.assertIsNone(preflight.runtime) with self.assertRaises(AgyAdapterError): build_agy_invocation(_cell(), self._prepared(), b"task", Timeout(5, 1, 1, 1), preflight) def test_arbitrary_runtime_cannot_self_issue_iop_proof(self) -> None: arbitrary = AgyRuntimeInputs(sys.executable, "https://api.openai.com/v1", "unrelated_token_123456789") preflight = preflight_agy_iop( _cell(), inspect_agy_iop_capability("agy 1.1.11", _help()), arbitrary, _iop_config_observation(), ) self.assertEqual(preflight.status, "implementation_gap") self.assertIsNone(preflight.runtime) with self.assertRaises(AgyAdapterError): build_agy_invocation(_cell(), self._prepared(), b"task", Timeout(5, 1, 1, 1), preflight) def test_lifecycle_fixture_success_and_metric_preservation(self) -> None: parser = AgyEventParser(_cell()) fixture = Path("scripts/fixtures/agent-comparison-benchmark/agy-iop-stream.jsonl") result = self._run_lines(fixture.read_text(encoding="utf-8").splitlines(), parser, self._preflight()) self.assertTrue(result.success) self.assertTrue(result.finish_then_idle_then_quiet) self.assertIn('"kind": "metric:duration_ms"', Path(result.journal_path).read_text(encoding="utf-8")) capability = inspect_agy_iop_capability("agy 1.1.11", _help()) self.assertEqual(parser.observed_result(capability, result).status, "ready") def test_metric_prefix_cannot_bypass_durable_redaction(self) -> None: parser = AgyEventParser(_cell()) raw_lines = [f"metric:{self.runtime.endpoint}", f"metric:{self.runtime.credential}", "metric:not-json"] result = self._run_lines(raw_lines, parser, self._preflight()) self.assertFalse(result.success) self.assertEqual(result.terminal_reason, REASON_MALFORMED_EVENT) persisted = Path(result.journal_path).read_text(encoding="utf-8") + Path(result.result_path).read_text(encoding="utf-8") for forbidden in (*raw_lines, self.runtime.endpoint, self.runtime.credential): self.assertNotIn(forbidden, persisted) def test_mismatch_duplicate_and_quota_cannot_pass(self) -> None: event = {"type": "result", "subtype": "success", "model": "other", "effort": "high", "route_kind": "direct", "route_id": "agy-direct"} self.assertEqual(AgyEventParser(_cell())("stdout", json.dumps(event)), "malformed") parser = AgyEventParser(_cell()) finish = {"type": "result", "subtype": "success", "model": "gemini-2.0-flash", "effort": "high", "route_kind": "direct", "route_id": "agy-direct"} self.assertEqual(parser("stdout", json.dumps(finish)), "finish") self.assertEqual(parser("stdout", json.dumps(finish)), "finish") self.assertEqual(AgyEventParser(_cell())("stdout", '{"type":"result","subtype":"error","reason":"quota"}'), "quota_error") evidence = self.root / "duplicate-evidence" evidence.mkdir() source = "import json; event=" + repr(finish) + "; print(json.dumps(event)); print(json.dumps(event))" spec = InvocationSpec( argv=(sys.executable, "-u", "-c", source), cwd=str(self.root), env=env_pairs({"PATH": os.environ.get("PATH", "/usr/bin:/bin")}), submission_mode="stdin_once", completion_mode="exit_after_idle", timeout=Timeout(5, 1, 1, 1), evidence_dir=str(evidence), task_payload=b"task", ) duplicate = run_agy_invocation(spec, AgyEventParser(_cell()), self._preflight(), lambda _: None) self.assertFalse(duplicate.success) self.assertEqual(duplicate.terminal_reason, REASON_DUPLICATE_EVENT) def test_structural_redaction_excludes_content_tools_endpoints_and_secrets(self) -> None: raw = json.dumps({"type": "result", "subtype": "success", "model": "gemini-2.0-flash", "content": "raw prompt", "tool_input": {"secret": "x"}, "endpoint": self.runtime.endpoint, "token": self.runtime.credential}) redacted = redact_agy_event(raw, (self.runtime.endpoint, self.runtime.credential)) self.assertEqual(redacted, '{"model":"gemini-2.0-flash","subtype":"success","type":"result"}') for forbidden in ("raw prompt", "tool_input", self.runtime.endpoint, self.runtime.credential): self.assertNotIn(forbidden, redacted) def test_lifecycle_rejects_quota_without_durable_leak(self) -> None: evidence = self.root / "evidence" evidence.mkdir() parser = AgyEventParser(_cell()) secret = self.runtime.credential endpoint = self.runtime.endpoint source = "import json; print(json.dumps(" + repr({ "type": "result", "subtype": "error", "reason": "quota", "content": secret, "endpoint": endpoint, }) + "))" spec = InvocationSpec( argv=(sys.executable, "-u", "-c", source), cwd=str(self.root), env=env_pairs({"PATH": os.environ.get("PATH", "/usr/bin:/bin")}), submission_mode="stdin_once", completion_mode="exit_after_idle", timeout=Timeout(5, 1, 1, 1), evidence_dir=str(evidence), task_payload=b"task", ) result = run_agy_invocation(spec, parser, self._preflight(), lambda _: None) self.assertFalse(result.success) self.assertEqual(result.terminal_reason, REASON_MALFORMED_EVENT) persisted = (Path(result.journal_path).read_text(encoding="utf-8") + Path(result.result_path).read_text(encoding="utf-8")) self.assertNotIn(secret, persisted) self.assertNotIn(endpoint, persisted) def test_ready_requires_observed_stage_binding_and_successful_lifecycle(self) -> None: capability = inspect_agy_iop_capability("agy 1.1.11", _help()) finish = {"type": "result", "subtype": "success", "model": "gemini-2.0-flash", "effort": "high", "route_kind": "direct", "route_id": "agy-direct"} idle = {"type": "system", "subtype": "idle", "model": "gemini-2.0-flash", "effort": "high", "route_kind": "direct", "route_id": "agy-direct"} binding = {"type": "iop", "subtype": "effective_binding", "route_kind": "direct", "route_id": "agy-direct", "model": "gemini-2.0-flash", "effort": "high", "stages": [{"stage": "request", "model": "gemini-2.0-flash", "effort": "high"}]} for lines in ( [json.dumps(finish), json.dumps(idle)], [json.dumps(binding), json.dumps(idle), json.dumps(finish)], [json.dumps(binding), json.dumps(finish), json.dumps(finish), json.dumps(idle)], [json.dumps({**binding, "route_id": "other"}), json.dumps(finish), json.dumps(idle)], ): with self.subTest(lines=lines): parser = AgyEventParser(_cell()) result = self._run_lines(lines, parser, self._preflight()) self.assertEqual(parser.observed_result(capability, result).status, "implementation_gap") if __name__ == "__main__": unittest.main()