"""Deterministic tests for the closed benchmark connectivity contract.""" from __future__ import annotations import copy import dataclasses import hashlib import json import os import tempfile import unittest from dataclasses import FrozenInstanceError from pathlib import Path from scripts.agent_benchmark.connectivity import ( ISSUE_RESUME_CODES, MAX_EVIDENCE_BYTES, CallerCapability, ConnectivityEvidenceError, ConnectivityIssue, ConnectivityResult, ConnectivityValidationError, EffectiveBinding, RequestedEffectiveBinding, canonical_evidence_bytes, classify_issues, make_result, read_evidence, validate_binding, validate_result, write_evidence, ) from scripts.agent_benchmark.manifest import ExpectedBinding, IopCell, MatrixCell EFFECTIVE_SCALARS = ( "effective_route_kind", "effective_route_id", "effective_model", "effective_effort", ) def _identity(label: str) -> str: return "sha256:" + hashlib.sha256(label.encode("ascii")).hexdigest() def _cell(route_kind: str = "direct") -> MatrixCell: if route_kind == "direct": bindings = (ExpectedBinding("request", "gpt", "xhigh"),) else: bindings = ( ExpectedBinding("selector", "gpt", "xhigh"), ExpectedBinding("plan", "gpt", "high"), ExpectedBinding("work", "gpt", None), ExpectedBinding("review", "gpt", "high"), ) return MatrixCell("cell", "codex", IopCell("gpt", "xhigh", route_kind, "route", bindings)) def _capability() -> CallerCapability: return CallerCapability("codex", ("direct", "execution_preset"), ("high", "xhigh")) def _requested_only(cell: MatrixCell) -> RequestedEffectiveBinding: """Blocked observation: requested identity only, no effective group at all.""" return RequestedEffectiveBinding( cell.id, cell.caller, cell.iop.route_kind, cell.iop.route_id, cell.iop.request_model, cell.iop.requested_effort, ) def _binding(cell: MatrixCell) -> RequestedEffectiveBinding: return RequestedEffectiveBinding( cell.id, cell.caller, cell.iop.route_kind, cell.iop.route_id, cell.iop.request_model, cell.iop.requested_effort, cell.iop.route_kind, cell.iop.route_id, cell.iop.request_model, cell.iop.requested_effort, tuple(EffectiveBinding(item.stage, item.model, item.effort) for item in cell.iop.expected_bindings), ) def _issue(code: str) -> ConnectivityIssue: return ConnectivityIssue(code, ISSUE_RESUME_CODES[code]) class ConnectivityContractTest(unittest.TestCase): def test_direct_and_preset_exact_contracts_are_frozen(self): for route_kind in ("direct", "execution_preset"): cell = _cell(route_kind) result = make_result(cell, _capability(), _binding(cell)) self.assertEqual(result.status, "ready") self.assertEqual(result.binding.effective_bindings, tuple( EffectiveBinding(item.stage, item.model, item.effort) for item in cell.iop.expected_bindings )) with self.assertRaises(FrozenInstanceError): result.status = "implementation_gap" # type: ignore[misc] def test_every_requested_or_effective_substitution_fails_closed(self): cell = _cell() binding = _binding(cell) substitutions = ( {"cell_id": "other"}, {"caller": "agy"}, {"requested_route_kind": "execution_preset"}, {"requested_route_id": "other"}, {"requested_model": "alias"}, {"requested_effort": "high"}, {"effective_route_kind": "execution_preset"}, {"effective_route_id": "other"}, {"effective_model": "alias"}, {"effective_effort": "high"}, ) for replacement in substitutions: with self.subTest(replacement=replacement): mutated = RequestedEffectiveBinding(**{**binding.__dict__, **replacement}) with self.assertRaises(ConnectivityValidationError): validate_binding(cell, _capability(), mutated) def test_missing_extra_and_reordered_stage_bindings_fail_closed(self): cell = _cell("execution_preset") binding = _binding(cell) cases = ( binding.effective_bindings[:-1], binding.effective_bindings + (EffectiveBinding("repair", "gpt", "high"),), tuple(reversed(binding.effective_bindings)), binding.effective_bindings[:-1] + (EffectiveBinding("review", "alias", "high"),), ) for stages in cases: with self.subTest(stages=stages): observed = RequestedEffectiveBinding(**{**binding.__dict__, "effective_bindings": stages}) with self.assertRaises(ConnectivityValidationError): validate_binding(cell, _capability(), observed) def test_capability_is_closed_and_requires_requested_effort(self): cell = _cell() with self.assertRaises(ConnectivityValidationError): validate_binding(cell, CallerCapability("codex", ("direct",), ("high",)), _binding(cell)) with self.assertRaises(ConnectivityValidationError): validate_binding(cell, CallerCapability("codex", ("execution_preset", "direct"), ("high", "xhigh")), _binding(cell)) def test_malformed_capability_entries_raise_closed_error(self): cell = _cell() binding = _binding(cell) capabilities = ( CallerCapability("codex", ("unknown", "direct"), ("high", "xhigh")), CallerCapability("codex", ("direct", ["execution_preset"]), ("high", "xhigh")), CallerCapability("codex", ("direct", "direct"), ("high", "xhigh")), CallerCapability("codex", (), ("high", "xhigh")), CallerCapability("codex", ["direct"], ("high", "xhigh")), CallerCapability("codex", ("direct",), ("high", 3)), CallerCapability("codex", ("direct",), (["xhigh"],)), CallerCapability("codex", ("direct",), ("xhigh", "high")), CallerCapability("codex", ("direct",), ("Xhigh",)), CallerCapability("unknown", ("direct",), ("xhigh",)), "codex", ) for capability in capabilities: with self.subTest(capability=repr(capability)): with self.assertRaises(ConnectivityValidationError) as caught: validate_binding(cell, capability, binding) self.assertNotIsInstance(caught.exception, (ValueError, TypeError, KeyError)) def test_classifier_is_closed_and_implementation_gap_has_precedence(self): registration = (_issue("credential_missing"),) implementation = (_issue("stream_incompatible"),) self.assertEqual(classify_issues(()), "ready") self.assertEqual(classify_issues(registration), "registration_required") self.assertEqual(classify_issues(implementation), "implementation_gap") self.assertEqual(classify_issues(registration + implementation), "implementation_gap") with self.assertRaises(ConnectivityValidationError): classify_issues((ConnectivityIssue("unknown", "register_credential"),)) with self.assertRaises(ConnectivityValidationError): classify_issues([_issue("credential_missing")]) def test_issue_resume_pairs_are_closed_and_canonically_ordered(self): for code, resume_code in ISSUE_RESUME_CODES.items(): with self.subTest(code=code): self.assertIn( classify_issues((ConnectivityIssue(code, resume_code),)), ("registration_required", "implementation_gap"), ) for other_code, other_resume in ISSUE_RESUME_CODES.items(): if other_code == code: continue with self.assertRaises(ConnectivityValidationError): classify_issues((ConnectivityIssue(code, other_resume),)) for text in ("register credential", "sk-live-0000", "", "register_credential ", None): with self.assertRaises(ConnectivityValidationError): classify_issues((ConnectivityIssue(code, text),)) canonical = (_issue("model_missing"), _issue("stream_incompatible")) self.assertEqual(classify_issues(canonical), "implementation_gap") with self.assertRaises(ConnectivityValidationError): classify_issues(tuple(reversed(canonical))) with self.assertRaises(ConnectivityValidationError): classify_issues(canonical + canonical[:1]) def test_malformed_issue_entries_raise_closed_error(self): cell = _cell() capability = _capability() binding = _binding(cell) valid_resume = ISSUE_RESUME_CODES["credential_missing"] other_resume = ISSUE_RESUME_CODES["model_missing"] bad_code_entries = ( ConnectivityIssue(["credential_missing"], valid_resume), ConnectivityIssue({"code": "credential_missing"}, valid_resume), ConnectivityIssue(42, valid_resume), ConnectivityIssue(None, valid_resume), ConnectivityIssue("unknown", valid_resume), ) bad_resume_entries = ( ConnectivityIssue("credential_missing", ["register_credential"]), ConnectivityIssue("credential_missing", {"resume_code": "register_credential"}), ConnectivityIssue("credential_missing", 42), ConnectivityIssue("credential_missing", None), ConnectivityIssue("credential_missing", "register credential"), ConnectivityIssue("credential_missing", "sk-live-0000"), ConnectivityIssue("credential_missing", ""), ConnectivityIssue("credential_missing", other_resume), ) non_issue_objects = ( {"code": "credential_missing", "resume_code": valid_resume}, "credential_missing", 42, None, ("credential_missing", valid_resume), ) expected_message = { "code": "invalid issue code", "resume": "invalid issue resume_code", "object": "invalid issue", } forbidden_raw_values = ( "sk-live-0000", "register credential", "credential_missing", "register_credential", "register_model", "42", "None", ) labelled = ( *(("code", item) for item in bad_code_entries), *(("resume", item) for item in bad_resume_entries), *(("object", item) for item in non_issue_objects), ) for label, entry in labelled: with self.subTest(label=label, entry=repr(entry)): issues = (entry,) with self.assertRaises(ConnectivityValidationError) as classifier_caught: classify_issues(issues) self.assertNotIsInstance( classifier_caught.exception, (TypeError, KeyError, ValueError, AttributeError), ) self.assertEqual(str(classifier_caught.exception), expected_message[label]) for token in forbidden_raw_values: self.assertNotIn(token, str(classifier_caught.exception)) self.assertNotIn(repr(entry), str(classifier_caught.exception)) with self.assertRaises(ConnectivityValidationError) as result_caught: make_result(cell, capability, binding, issues) self.assertNotIsInstance( result_caught.exception, (TypeError, KeyError, ValueError, AttributeError), ) self.assertEqual(str(result_caught.exception), expected_message[label]) for token in forbidden_raw_values: self.assertNotIn(token, str(result_caught.exception)) self.assertNotIn(repr(entry), str(result_caught.exception)) def test_ready_requires_complete_exact_effective_observation(self): cell = _cell("execution_preset") binding = _binding(cell) with self.assertRaises(ConnectivityValidationError): make_result(cell, _capability(), _requested_only(cell)) partials = [{name: None} for name in EFFECTIVE_SCALARS] partials.append({"effective_bindings": ()}) partials.append({name: None for name in EFFECTIVE_SCALARS}) partials.append({"effective_model": "alias"}) partials.append({"effective_bindings": binding.effective_bindings[:-1]}) for replacement in partials: with self.subTest(replacement=replacement): mutated = RequestedEffectiveBinding(**{**binding.__dict__, **replacement}) with self.assertRaises(ConnectivityValidationError): make_result(cell, _capability(), mutated) with self.assertRaises(ConnectivityValidationError): make_result(cell, _capability(), mutated, (_issue("model_missing"),)) def test_blocked_results_stay_blocked_and_cannot_be_forged_ready(self): cell = _cell() issues = (_issue("credential_missing"),) blocked = make_result(cell, _capability(), _requested_only(cell), issues) self.assertEqual(blocked.status, "registration_required") self.assertIsNone(blocked.binding.effective_model) self.assertEqual(blocked.binding.effective_bindings, ()) validate_result(cell, blocked) forgeries = ( ConnectivityResult(_capability(), _requested_only(cell), issues, "ready"), ConnectivityResult(_capability(), _binding(cell), issues, "ready"), ConnectivityResult(_capability(), _binding(cell), (), "blocked"), ConnectivityResult(_capability(), _requested_only(cell), (), "ready"), ) for forged in forgeries: with self.subTest(status=forged.status, issues=len(forged.issues)): with self.assertRaises(ConnectivityValidationError): validate_result(cell, forged) def test_no_contract_field_accepts_opaque_caller_text(self): self.assertEqual( {field.name for field in dataclasses.fields(ConnectivityIssue)}, {"code", "resume_code"}, ) cell = _cell() binding = _binding(cell) opaque = "sk-live-0000000000000000" secret = "Authorization: Bearer should-not-appear" for text in (secret, opaque, "https://private.invalid"): with self.subTest(text=text): with self.assertRaises(ConnectivityValidationError) as caught: make_result(cell, _capability(), binding, (ConnectivityIssue("credential_missing", text),)) self.assertNotIn(text, str(caught.exception)) for replacement in ( {"requested_model": opaque}, {"effective_model": opaque}, {"effective_route_id": "private-endpoint"}, ): with self.subTest(replacement=replacement): mutated = RequestedEffectiveBinding(**{**binding.__dict__, **replacement}) with self.assertRaises(ConnectivityValidationError) as caught: make_result(cell, _capability(), mutated) self.assertNotIn(opaque, str(caught.exception)) self.assertNotIn("private-endpoint", str(caught.exception)) class ConnectivityEvidenceTest(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory(dir="/tmp", prefix="connectivity-") self.base = Path(os.path.realpath(self.temp.name)) self.root = self.base / "evidence" self.root.mkdir() self.cell = _cell() self.result = make_result(self.cell, _capability(), _binding(self.cell)) self.endpoint_identity = _identity("endpoint") self.config_identity = _identity("config") def tearDown(self): self.temp.cleanup() def _write(self, root, relative_path, cell=None, result=None): write_evidence( root, relative_path, cell or self.cell, result or self.result, self.endpoint_identity, self.config_identity, ) def test_canonical_evidence_is_deterministic_and_secret_safe(self): one = canonical_evidence_bytes(self.cell, self.result, self.endpoint_identity, self.config_identity) two = canonical_evidence_bytes(self.cell, self.result, self.endpoint_identity, self.config_identity) self.assertEqual(one, two) self.assertIn(self.endpoint_identity.encode(), one) for forbidden in (b"Authorization", b"Bearer", b"https://", b"prompt", b"tool"): self.assertNotIn(forbidden, one) def test_blocked_results_omit_effective_observations_and_round_trip(self): cases = ( ("direct", _issue("model_missing"), "registration_required"), ("execution_preset", _issue("endpoint_incompatible"), "implementation_gap"), ) for index, (route_kind, issue, status) in enumerate(cases): with self.subTest(route_kind=route_kind): cell = _cell(route_kind) result = make_result(cell, _capability(), _requested_only(cell), (issue,)) self.assertEqual(result.status, status) raw = canonical_evidence_bytes(cell, result, self.endpoint_identity, self.config_identity) for absent in EFFECTIVE_SCALARS: self.assertIn(f'"{absent}":null'.encode("ascii"), raw) self.assertIn(b'"effective_bindings":[]', raw) name = f"blocked-{index}.json" self._write(self.root, name, cell, result) parsed = read_evidence(self.root, name, cell) self.assertEqual(parsed["status"], status) self.assertIsNone(parsed["binding"]["effective_model"]) self.assertEqual(parsed["binding"]["effective_bindings"], []) self.assertEqual( parsed["issues"], [{"code": issue.code, "resume_code": issue.resume_code}] ) forged = ( (self.root / name).read_text(encoding="ascii") .replace(f'"status":"{status}"', '"status":"ready"') .replace(f'{{"code":"{issue.code}","resume_code":"{issue.resume_code}"}}', "") ) (self.root / f"forged-{index}.json").write_text(forged, encoding="ascii") with self.assertRaises(ConnectivityEvidenceError): read_evidence(self.root, f"forged-{index}.json", cell) def test_write_is_no_overwrite_and_read_rejects_corruption(self): self._write(self.root, "nested/preflight.json") raw = (self.root / "nested/preflight.json").read_bytes() self.assertEqual(read_evidence(self.root, "nested/preflight.json", self.cell)["status"], "ready") with self.assertRaises(ConnectivityEvidenceError): self._write(self.root, "nested/preflight.json") (self.root / "nested/preflight.json").write_bytes(raw + b" ") with self.assertRaises(ConnectivityEvidenceError): read_evidence(self.root, "nested/preflight.json", self.cell) def test_reader_rejects_canonical_schema_drift(self): self._write(self.root, "preflight.json") raw = (self.root / "preflight.json").read_text(encoding="ascii") (self.root / "preflight.json").write_text(raw.replace('"status":"ready"', '"status":"registration_required"'), encoding="ascii") with self.assertRaises(ConnectivityEvidenceError): read_evidence(self.root, "preflight.json", self.cell) def test_reader_rejects_canonical_binding_semantic_substitution(self): def _reseat(parsed: dict) -> bytes: return ( json.dumps( parsed, sort_keys=True, separators=(",", ":"), ensure_ascii=True ).encode("ascii") + b"\n" ) def _write_mutant(cell, result, name, mutate): self._write(self.root, f"{name}.json", cell, result) baseline = read_evidence(self.root, f"{name}.json", cell) self.assertEqual(baseline["status"], result.status) original = (self.root / f"{name}.json").read_text(encoding="ascii") parsed = json.loads(original) mutant = copy.deepcopy(parsed) mutate(mutant) mutant_name = f"mutant-{name}.json" (self.root / mutant_name).write_bytes(_reseat(mutant)) return mutant_name cases: list[tuple] = [] direct_cell = _cell("direct") direct_result = make_result(direct_cell, _capability(), _binding(direct_cell)) direct_scalars = ( ("requested_route_kind", "execution_preset"), ("requested_route_id", "route2"), ("requested_model", "gpth"), ("requested_effort", "high"), ("effective_route_kind", "execution_preset"), ("effective_route_id", "route2"), ("effective_model", "gpth"), ("effective_effort", "high"), ) for field, value in direct_scalars: cases.append(( "direct-ready-scalar", direct_cell, direct_result, f"direct-{field}", lambda p, f=field, v=value: p["binding"].__setitem__(f, v), )) cases.append(( "direct-ready-stage", direct_cell, direct_result, "direct-stage", lambda p: p["binding"]["effective_bindings"].__setitem__( 0, {"stage": "request", "model": "gpth", "effort": "xhigh"} ), )) preset_cell = _cell("execution_preset") preset_result = make_result(preset_cell, _capability(), _binding(preset_cell)) preset_scalars = ( ("requested_route_kind", "direct"), ("requested_route_id", "route2"), ("requested_model", "gpth"), ("requested_effort", "high"), ("effective_route_kind", "direct"), ("effective_route_id", "route2"), ("effective_model", "gpth"), ("effective_effort", "high"), ) for field, value in preset_scalars: cases.append(( "preset-ready-scalar", preset_cell, preset_result, f"preset-{field}", lambda p, f=field, v=value: p["binding"].__setitem__(f, v), )) cases.append(( "preset-ready-stage-model", preset_cell, preset_result, "preset-stage-model", lambda p: p["binding"]["effective_bindings"].__setitem__( 0, {"stage": "selector", "model": "gpth", "effort": "xhigh"} ), )) cases.append(( "preset-ready-stage-effort", preset_cell, preset_result, "preset-stage-effort", lambda p: p["binding"]["effective_bindings"].__setitem__( 1, {"stage": "plan", "model": "gpt", "effort": "xhigh"} ), )) cases.append(( "preset-ready-stage-set", preset_cell, preset_result, "preset-stage-set", lambda p: p["binding"]["effective_bindings"].__setitem__( 0, {"stage": "request", "model": "gpt", "effort": "xhigh"} ), )) cases.append(( "preset-ready-stage-order", preset_cell, preset_result, "preset-stage-order", lambda p: p["binding"]["effective_bindings"].reverse(), )) cases.append(( "preset-ready-stage-extra", preset_cell, preset_result, "preset-stage-extra", lambda p: p["binding"]["effective_bindings"].append( {"stage": "repair", "model": "gpt", "effort": "high"} ), )) cases.append(( "preset-ready-stage-missing", preset_cell, preset_result, "preset-stage-missing", lambda p: p["binding"]["effective_bindings"].pop(), )) blocked_cell = _cell("execution_preset") blocked_result = make_result( blocked_cell, _capability(), _binding(blocked_cell), (_issue("credential_missing"),) ) self.assertEqual(blocked_result.status, "registration_required") cases.append(( "blocked-observed-stage-model", blocked_cell, blocked_result, "blocked-stage-model", lambda p: p["binding"]["effective_bindings"].__setitem__( 0, {"stage": "selector", "model": "gpth", "effort": "xhigh"} ), )) cases.append(( "blocked-observed-stage-order", blocked_cell, blocked_result, "blocked-stage-order", lambda p: p["binding"]["effective_bindings"].reverse(), )) for label, cell, result, name, mutate in cases: with self.subTest(label=label, name=name): mutant_path = _write_mutant(cell, result, name, mutate) with self.assertRaises(ConnectivityEvidenceError): read_evidence(self.root, mutant_path, cell) def test_reader_rejects_noncanonical_issue_order(self): cell = _cell() issues = (_issue("model_missing"), _issue("stream_incompatible")) result = make_result(cell, _capability(), _requested_only(cell), issues) self._write(self.root, "ordered.json", cell, result) raw = (self.root / "ordered.json").read_text(encoding="ascii") first = '{"code":"model_missing","resume_code":"register_model"}' second = '{"code":"stream_incompatible","resume_code":"implement_stream_adapter"}' self.assertIn(f"{first},{second}", raw) (self.root / "swapped.json").write_text(raw.replace(f"{first},{second}", f"{second},{first}"), encoding="ascii") with self.assertRaises(ConnectivityEvidenceError): read_evidence(self.root, "swapped.json", cell) with self.assertRaises(ConnectivityValidationError): make_result(cell, _capability(), _requested_only(cell), tuple(reversed(issues))) def test_symlinked_roots_parents_and_targets_are_rejected(self): outside = self.base / "outside" outside.mkdir() real_root = self.base / "real" / "evidence" real_root.mkdir(parents=True) (self.base / "link").symlink_to(self.base / "real", target_is_directory=True) with self.assertRaises(ConnectivityEvidenceError): self._write(self.base / "link" / "evidence", "preflight.json") with self.assertRaises(ConnectivityEvidenceError): read_evidence(self.base / "link" / "evidence", "preflight.json", self.cell) self.assertFalse((real_root / "preflight.json").exists()) self._write(real_root, "preflight.json") self.assertEqual(read_evidence(real_root, "preflight.json", self.cell)["status"], "ready") (self.base / "root-link").symlink_to(self.root, target_is_directory=True) with self.assertRaises(ConnectivityEvidenceError): self._write(self.base / "root-link", "preflight.json") (self.root / "nested").symlink_to(outside, target_is_directory=True) with self.assertRaises(ConnectivityEvidenceError): self._write(self.root, "nested/preflight.json") self.assertFalse((outside / "preflight.json").exists()) (self.root / "link.json").symlink_to(outside / "escape.json") with self.assertRaises(ConnectivityEvidenceError): self._write(self.root, "link.json") with self.assertRaises(ConnectivityEvidenceError): read_evidence(self.root, "link.json", self.cell) self.assertFalse((outside / "escape.json").exists()) def test_oversized_directory_and_non_regular_targets_are_rejected(self): self._write(self.root, "small.json") self.assertEqual(read_evidence(self.root, "small.json", self.cell)["status"], "ready") (self.root / "big.json").write_bytes(b"{" + b" " * (MAX_EVIDENCE_BYTES + 16) + b"}") with self.assertRaises(ConnectivityEvidenceError): read_evidence(self.root, "big.json", self.cell) (self.root / "dir.json").mkdir() with self.assertRaises(ConnectivityEvidenceError): read_evidence(self.root, "dir.json", self.cell) with self.assertRaises(ConnectivityEvidenceError): self._write(self.root, "dir.json") if hasattr(os, "mkfifo"): os.mkfifo(self.root / "fifo.json") with self.assertRaises(ConnectivityEvidenceError): read_evidence(self.root, "fifo.json", self.cell) with self.assertRaises(ConnectivityEvidenceError): self._write(self.root, "fifo.json") def test_private_identity_and_escaping_paths_are_rejected(self): for identity in ("https://private.invalid", "sha256:" + "z" * 64, "", "sha256:abcd", None): with self.subTest(identity=identity): with self.assertRaises(ConnectivityEvidenceError): canonical_evidence_bytes(self.cell, self.result, identity, self.config_identity) with self.assertRaises(ConnectivityEvidenceError): canonical_evidence_bytes(self.cell, self.result, self.endpoint_identity, identity) for relative in ("../escape.json", "/abs.json", "nested/../escape.json", "preflight.txt", "preflight", "", ".json"): with self.subTest(relative=relative): with self.assertRaises(ConnectivityEvidenceError): self._write(self.root, relative) with self.assertRaises(ConnectivityEvidenceError): read_evidence(self.root, relative, self.cell) self.assertFalse((self.base / "escape.json").exists()) if __name__ == "__main__": unittest.main()