261 lines
12 KiB
Python
261 lines
12 KiB
Python
"""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,
|
|
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"))
|
|
|
|
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()
|