"""Credential-free tests for the official agy 1.1.12 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 unittest.mock import patch from scripts.agent_benchmark.agy_iop import ( AGY_AUTH_ENV, AGY_ENDPOINT_ENV, AGY_KNOWN_VERSION, AGY_SETTINGS_RELATIVE_PATH, AgyAdapterError, AgyEventParser, AgyRuntimeInputs, AgyRuntimeObservation, _runtime_identity, build_agy_invocation, inspect_agy_iop_capability, preflight_agy_iop, redact_agy_event, run_agy_invocation, ) from scripts.agent_benchmark.connectivity import EffectiveBinding, RequestedEffectiveBinding from scripts.agent_benchmark.lifecycle import ( REASON_MALFORMED_EVENT, SUBMISSION_ARGV_TASK, InvocationSpec, env_pairs, ) from scripts.agent_benchmark.manifest import ExpectedBinding, IopCell, MatrixCell, Timeout from scripts.agent_benchmark.workspace import AttemptIdentity, PreparedWorkspace, TestbedProvenance def _help() -> str: return "--print --output-format stream-json --sandbox --model --effort" def _cell() -> MatrixCell: return MatrixCell( "agy-direct", "agy", IopCell("gemini-3.6-flash", "high", "direct", "agy-direct", ( ExpectedBinding("request", "gemini-3.6-flash", "high"), )), ) def _binding() -> RequestedEffectiveBinding: return RequestedEffectiveBinding( "agy-direct", "agy", "direct", "agy-direct", "gemini-3.6-flash", "high", "direct", "agy-direct", "gemini-3.6-flash", "high", (EffectiveBinding("request", "gemini-3.6-flash", "high"),), ) 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.session = self.root / "session" self.session.mkdir() self.runtime = AgyRuntimeInputs( sys.executable, "https://private.invalid/gemini/agy-direct", "iop_secret_123456789" ) def tearDown(self) -> None: self.temp.cleanup() def _observation(self, runtime: AgyRuntimeInputs | None = None) -> AgyRuntimeObservation: value = runtime or self.runtime return AgyRuntimeObservation( "agy-direct", "direct", "agy-direct", _runtime_identity("endpoint", value.endpoint), _runtime_identity("credential", value.credential), "sha256:" + "c" * 64, ) def _prepared(self) -> PreparedWorkspace: return PreparedWorkspace( AttemptIdentity("run", "agy-direct", 1, 1), str(self.root), str(self.workspace), str(self.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): value = runtime or self.runtime return preflight_agy_iop( _cell(), inspect_agy_iop_capability("agy 1.1.12", _help()), value, self._observation(value), ) def _run_lines(self, lines: list[str], parser: AgyEventParser): evidence = self.root / f"evidence-{len(list(self.root.glob('evidence-*')))}" evidence.mkdir() source = "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, self._preflight(), lambda _: None) def test_official_public_surface_is_pinned_without_invented_environment(self) -> None: capability = inspect_agy_iop_capability("1.1.12", _help()) self.assertEqual(capability.version, AGY_KNOWN_VERSION) self.assertTrue(capability.iop_transport_supported) self.assertTrue(capability.endpoint_supported) self.assertTrue(capability.auth_supported) self.assertFalse(inspect_agy_iop_capability("1.1.11", _help()).iop_transport_supported) self.assertFalse(inspect_agy_iop_capability("1.1.12", _help().replace("stream-json", "json")).stream_supported) def test_build_uses_official_gemini_api_key_transport(self) -> None: with patch.dict(os.environ, {"SSL_CERT_FILE": "/operator/dev-ca.pem", "NODE_EXTRA_CA_CERTS": "/operator/dev-ca.pem"}): spec = build_agy_invocation(_cell(), self._prepared(), b"one task", Timeout(5, 1, 1, 1), self._preflight()) environment = dict(spec.env) self.assertEqual(environment[AGY_ENDPOINT_ENV], self.runtime.endpoint) self.assertEqual(environment[AGY_AUTH_ENV], self.runtime.credential) self.assertEqual(environment["HOME"], str(self.session)) self.assertNotIn("AGY_PROVIDER", environment) self.assertNotIn("AGY_OPENAI_BASE_URL", environment) self.assertNotIn("AGY_OPENAI_API_KEY", environment) self.assertNotIn("--effort", spec.argv) self.assertEqual(spec.argv[spec.argv.index("--model") + 1], "Gemini 3.6 Flash") self.assertEqual(spec.argv[-2:], ("--print", "one task")) self.assertEqual(spec.submission_mode, SUBMISSION_ARGV_TASK) self.assertEqual(spec.task_payload, b"") self.assertEqual(environment["SSL_CERT_FILE"], "/operator/dev-ca.pem") self.assertEqual(environment["NODE_EXTRA_CA_CERTS"], "/operator/dev-ca.pem") self.assertEqual(spec.env_allowlist, (AGY_ENDPOINT_ENV, AGY_AUTH_ENV, "SSL_CERT_FILE", "NODE_EXTRA_CA_CERTS")) settings_path = self.session / AGY_SETTINGS_RELATIVE_PATH self.assertEqual(json.loads(settings_path.read_text(encoding="utf-8")), { "enableTelemetry": False, "modelProvider": "gemini", "toolPermission": "always-proceed", }) self.assertEqual(settings_path.stat().st_mode & 0o777, 0o600) settings_text = settings_path.read_text(encoding="utf-8") self.assertNotIn(self.runtime.endpoint, settings_text) self.assertNotIn(self.runtime.credential, settings_text) def test_build_rejects_preexisting_isolated_settings(self) -> None: settings_path = self.session / AGY_SETTINGS_RELATIVE_PATH settings_path.parent.mkdir(parents=True) settings_path.write_text('{"modelProvider":"other"}\n', encoding="utf-8") with self.assertRaisesRegex(AgyAdapterError, "isolated provider settings"): build_agy_invocation( _cell(), self._prepared(), b"task", Timeout(5, 1, 1, 1), self._preflight() ) def test_build_rejects_missing_prepared_workspace_or_session(self) -> None: prepared = self._prepared() for field in ("workspace_dir", "session_dir", "attempt_root"): missing = prepared.__class__(**{ **prepared.__dict__, field: str(self.root / f"missing-{field}"), }) with self.subTest(field=field): with self.assertRaisesRegex(AgyAdapterError, "prepared workspace"): build_agy_invocation( _cell(), missing, b"task", Timeout(5, 1, 1, 1), self._preflight() ) def test_build_rejects_non_utf8_task_for_print_argument(self) -> None: with self.assertRaisesRegex(AgyAdapterError, "must be UTF-8"): build_agy_invocation( _cell(), self._prepared(), b"\xff", Timeout(5, 1, 1, 1), self._preflight() ) def test_route_qualified_https_runtime_is_required(self) -> None: for endpoint in ( "http://private.invalid/gemini/agy-direct", "https://private.invalid/v1", "https://private.invalid/gemini/other", ): runtime = replace(self.runtime, endpoint=endpoint) result = self._preflight(runtime) self.assertEqual(result.status, "implementation_gap") self.assertEqual([issue.code for issue in result.issues], ["endpoint_incompatible"]) missing = replace(self.runtime, credential="") result = preflight_agy_iop( _cell(), inspect_agy_iop_capability("1.1.12", _help()), missing, self._observation(self.runtime), ) self.assertEqual(result.status, "registration_required") self.assertEqual([issue.code for issue in result.issues], ["credential_missing"]) def test_unknown_model_and_unvalidated_observation_fail_closed(self) -> None: unsupported = replace(_cell(), iop=replace(_cell().iop, request_model="gemini-unknown")) result = preflight_agy_iop( unsupported, inspect_agy_iop_capability("1.1.12", _help()), self.runtime, self._observation(), ) self.assertIn("model_missing", [issue.code for issue in result.issues]) mismatched = replace(self._observation(), endpoint_identity="sha256:" + "d" * 64) result = preflight_agy_iop( _cell(), inspect_agy_iop_capability("1.1.12", _help()), self.runtime, mismatched, ) self.assertIsNone(result.runtime) with self.assertRaises(AgyAdapterError): build_agy_invocation(_cell(), self._prepared(), b"task", Timeout(5, 1, 1, 1), result) def test_official_fixture_completes_and_preserves_metrics(self) -> None: parser = AgyEventParser(_cell(), _binding()) fixture = Path("scripts/fixtures/agent-comparison-benchmark/agy-iop-stream.jsonl") result = self._run_lines(fixture.read_text(encoding="utf-8").splitlines(), parser) self.assertTrue(result.success) self.assertTrue(result.finish_then_idle_then_quiet) metrics = {metric.name: metric for metric in result.metrics} self.assertEqual( set(metrics), {"total_duration", "model_calls", "input_tokens", "cached_input_tokens", "output_tokens", "reasoning_tokens", "total_tokens"}, ) self.assertEqual(metrics["total_duration"].value, 12_000_000) self.assertEqual(metrics["model_calls"].value, 1) self.assertEqual(metrics["total_tokens"].value, 12) capability = inspect_agy_iop_capability("1.1.12", _help()) self.assertEqual(parser.observed_result(capability, result).binding, _binding()) def test_result_must_follow_init_be_unique_and_successful(self) -> None: result = { "event": "result", "result": { "status": "SUCCESS", "duration_seconds": 0.1, "num_turns": 1, "usage": {"input_tokens": 1}, "response": "secret content", }, } for lines, reason in ( ([json.dumps(result)], REASON_MALFORMED_EVENT), ([json.dumps({"event": "init", "init": {}}), json.dumps(result), json.dumps(result)], REASON_MALFORMED_EVENT), ([json.dumps({"event": "init", "init": {}}), json.dumps({**result, "result": {**result["result"], "status": "ERROR"}})], REASON_MALFORMED_EVENT), ): parser = AgyEventParser(_cell(), _binding()) invocation = self._run_lines(lines, parser) self.assertFalse(invocation.success) self.assertEqual(invocation.terminal_reason, reason) def test_latest_step_usage_is_used_only_when_result_omits_usage(self) -> None: parser = AgyEventParser(_cell(), _binding()) lines = [ json.dumps({"event": "init", "conversation_id": "c", "init": {"model": "Gemini 3.6 Flash"}}), json.dumps({"event": "step_update", "step_update": {"state": "DONE", "step_index": 0, "step_type": "agent_response", "text_delta": "private", "usage": {"input_tokens": 2, "output_tokens": 1, "total_tokens": 3}}}), json.dumps({"event": "result", "result": {"status": "SUCCESS", "duration_seconds": 0.2, "num_turns": 1, "response": "private"}}), ] invocation = self._run_lines(lines, parser) self.assertTrue(invocation.success) self.assertEqual({m.name: m.value for m in invocation.metrics}["total_tokens"], 3) def test_malformed_usage_fails_without_partial_metric(self) -> None: parser = AgyEventParser(_cell(), _binding()) invocation = self._run_lines([ '{"event":"init","init":{}}', '{"event":"result","result":{"status":"SUCCESS","usage":{"input_tokens":"1"}}}', ], parser) self.assertFalse(invocation.success) self.assertEqual(invocation.terminal_reason, REASON_MALFORMED_EVENT) self.assertEqual(invocation.metrics, ()) def test_structural_redaction_excludes_response_tools_endpoint_and_secret(self) -> None: raw = json.dumps({ "event": "result", "result": { "status": "SUCCESS", "response": "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, '{"event":"result","status":"SUCCESS"}') for forbidden in ("raw prompt", "tool_input", self.runtime.endpoint, self.runtime.credential): self.assertNotIn(forbidden, redacted) if __name__ == "__main__": unittest.main()