From df72c71182e94c6a29b46ff17626b3105c27d2d9 Mon Sep 17 00:00:00 2001 From: Shay Date: Tue, 23 Jun 2026 00:28:25 -0700 Subject: [PATCH] test(kernel): reconcile bound evidence spine static guards --- tests/test_candidate_operator.py | 36 +++++- tests/test_replay_adapter.py | 20 +++- tests/test_run_attempt_binding.py | 113 ++++++++++++++++++ ...est_sealed_practice_trace_bound_episode.py | 68 ++++++++++- 4 files changed, 229 insertions(+), 8 deletions(-) diff --git a/tests/test_candidate_operator.py b/tests/test_candidate_operator.py index db8a186e..5f992438 100644 --- a/tests/test_candidate_operator.py +++ b/tests/test_candidate_operator.py @@ -1108,15 +1108,47 @@ def test_module_coupling_and_side_effect_guards() -> None: "time", "uuid4", }.isdisjoint(calls) - for upstream in ( + allowed_records = { + "CandidateOperatorResult", + "CandidateReconstruction", + "CandidateOperatorRefusal", + } + forbidden_producers = { + "build_missing_role_candidate", + "candidate_operator_set_id", + "GroundedUnaryDeltaCue", + } + for downstream_path in ( "generate/geometric_search_run.py", "generate/replay_adapter.py", "generate/sealed_practice_trace.py", "generate/search_gate.py", "generate/compute_budget.py", "generate/contract_residuals.py", + "generate/run_attempt_binding.py", ): - assert "candidate_operator" not in Path(upstream).read_text("utf-8") + ds_source = Path(downstream_path).read_text("utf-8") + ds_tree = ast.parse(ds_source) + ds_imported_names: set[str] = set() + ds_calls: set[str] = set() + for node in ast.walk(ds_tree): + if isinstance(node, ast.ImportFrom): + if node.module == "generate.candidate_operator": + ds_imported_names.update(alias.name for alias in node.names) + elif isinstance(node, ast.Call): + if isinstance(node.func, ast.Name): + ds_calls.add(node.func.id) + elif isinstance(node.func, ast.Attribute): + ds_calls.add(node.func.attr) + + assert ds_imported_names <= allowed_records, ( + f"{downstream_path} imported forbidden candidate_operator records: " + f"{ds_imported_names - allowed_records}" + ) + assert forbidden_producers.isdisjoint(ds_calls), ( + f"{downstream_path} called forbidden candidate_operator producers: " + f"{ds_calls & forbidden_producers}" + ) def test_no_filesystem_network_clock_random_or_environment_identity() -> None: diff --git a/tests/test_replay_adapter.py b/tests/test_replay_adapter.py index b4c94f3d..eacb98d8 100644 --- a/tests/test_replay_adapter.py +++ b/tests/test_replay_adapter.py @@ -947,19 +947,35 @@ def test_module_coupling_and_side_effect_guards() -> None: "json", "generate.geometric_search_run", "generate.kernel_facts", + "generate.candidate_operator", + "generate.run_attempt_binding", } assert { + "build_missing_role_candidate", + "candidate_operator_set_id", + "bind_candidate_attempt_to_run", + "initialize_geometric_search_run", + "build_practice_trace_input", + "build_bound_practice_trace_input", + "seal_practice_trace", + "seal_bound_practice_trace", "decide_search_gate", "decide_compute_budget", - "initialize_geometric_search_run", "assess_contracts", "project_contract_residuals", "determine", }.isdisjoint(imported_names) assert { + "build_missing_role_candidate", + "candidate_operator_set_id", + "bind_candidate_attempt_to_run", + "initialize_geometric_search_run", + "build_practice_trace_input", + "build_bound_practice_trace_input", + "seal_practice_trace", + "seal_bound_practice_trace", "decide_search_gate", "decide_compute_budget", - "initialize_geometric_search_run", "assess_contracts", "project_contract_residuals", "determine", diff --git a/tests/test_run_attempt_binding.py b/tests/test_run_attempt_binding.py index 2b326749..53617227 100644 --- a/tests/test_run_attempt_binding.py +++ b/tests/test_run_attempt_binding.py @@ -312,3 +312,116 @@ def test_refusal_is_not_partial_binding() -> None: assert isinstance(outcome, CandidateAttemptRunBindingRefusal) assert not hasattr(outcome, "binding_id") assert not hasattr(outcome, "candidate_attempt_ref") + + +def test_module_coupling_and_side_effect_guards() -> None: + import ast + from pathlib import Path + + path = Path("generate/run_attempt_binding.py") + source = path.read_text("utf-8") + tree = ast.parse(source) + imports: set[str] = set() + imported_names: set[str] = set() + calls: set[str] = set() + + for node in ast.walk(tree): + if isinstance(node, ast.Import): + imports.update(alias.name for alias in node.names) + elif isinstance(node, ast.ImportFrom): + imports.add(node.module or "") + imported_names.update(alias.name for alias in node.names) + elif isinstance(node, ast.Call): + if isinstance(node.func, ast.Name): + calls.add(node.func.id) + elif isinstance(node.func, ast.Attribute): + calls.add(node.func.attr) + + assert imports <= { + "__future__", + "dataclasses", + "enum", + "hashlib", + "json", + "generate.candidate_operator", + "generate.geometric_search_run", + "generate.kernel_facts", + } + + assert { + "build_missing_role_candidate", + "candidate_operator_set_id", + "GroundedUnaryDeltaCue", + "initialize_geometric_search_run", + "build_replay_adapter_input", + "build_replay_adapter_input_from_binding", + "classify_replay_result", + "build_practice_trace_input", + "build_bound_practice_trace_input", + "seal_practice_trace", + "seal_bound_practice_trace", + }.isdisjoint(imported_names) + + assert { + "build_missing_role_candidate", + "candidate_operator_set_id", + "GroundedUnaryDeltaCue", + "initialize_geometric_search_run", + "build_replay_adapter_input", + "build_replay_adapter_input_from_binding", + "classify_replay_result", + "build_practice_trace_input", + "build_bound_practice_trace_input", + "seal_practice_trace", + "seal_bound_practice_trace", + "repair", + "serve", + "store", + "write", + "open", + "write_text", + "write_bytes", + "request", + "urlopen", + "sleep", + "time", + "uuid4", + "random", + "subprocess", + }.isdisjoint(calls) + + forbidden_module_families = ( + "runtime", "serving", "workbench", "teaching", "proposal", "pack", "policy", "identity", "eval" + ) + for imp in imports: + assert not any(imp.startswith(family) for family in forbidden_module_families), f"Imported forbidden module family: {imp}" + + +def test_no_filesystem_network_clock_random_or_environment_identity() -> None: + from pathlib import Path + source = Path("generate/run_attempt_binding.py").read_text("utf-8") + forbidden_fragments = ( + "import os", + "from os", + "import pathlib", + "from pathlib", + "import random", + "from random", + "import time", + "from time", + "import datetime", + "from datetime", + "import uuid", + "from uuid", + "import socket", + "from socket", + "import subprocess", + "from subprocess", + "import requests", + "from requests", + "os.environ", + "getenv(", + "gethostname(", + "Path(", + ) + assert not any(fragment in source for fragment in forbidden_fragments) diff --git a/tests/test_sealed_practice_trace_bound_episode.py b/tests/test_sealed_practice_trace_bound_episode.py index f4efb0ad..1f7c9f71 100644 --- a/tests/test_sealed_practice_trace_bound_episode.py +++ b/tests/test_sealed_practice_trace_bound_episode.py @@ -724,27 +724,87 @@ def test_seal_bound_refuses_invalid_binding_type_without_exception() -> None: def test_bound_module_does_not_call_upstream_producers() -> None: + # sealed_practice_trace consumes already-produced evidence. + # It does not produce candidates, bind attempts, build replay input, classify replay, + # or execute sealing side effects beyond constructing immutable records. path = Path("generate/sealed_practice_trace.py") source = path.read_text("utf-8") tree = ast.parse(source) + imports: set[str] = set() imported_names: set[str] = set() calls: set[str] = set() for node in ast.walk(tree): - if isinstance(node, ast.ImportFrom): + if isinstance(node, ast.Import): + imports.update(alias.name for alias in node.names) + elif isinstance(node, ast.ImportFrom): + imports.add(node.module or "") imported_names.update(alias.name for alias in node.names) elif isinstance(node, ast.Call): if isinstance(node.func, ast.Name): calls.add(node.func.id) elif isinstance(node.func, ast.Attribute): calls.add(node.func.attr) - forbidden = { + + # Allowed record imports and standard library / upstream modules + assert imports <= { + "__future__", + "dataclasses", + "enum", + "hashlib", + "json", + "generate.geometric_search_run", + "generate.kernel_facts", + "generate.replay_adapter", + "generate.candidate_operator", + "generate.run_attempt_binding", + } + + # Forbidden authority modules + forbidden_module_families = ( + "runtime", "serving", "workbench", "teaching", "proposal", "pack", "policy", "identity" + ) + for imp in imports: + assert not any(imp.startswith(family) for family in forbidden_module_families), f"Imported forbidden module family: {imp}" + + # Forbidden imports/calls + forbidden_imports = { "build_missing_role_candidate", + "candidate_operator_set_id", + "GroundedUnaryDeltaCue", "bind_candidate_attempt_to_run", + "build_replay_adapter_input", "build_replay_adapter_input_from_binding", "classify_replay_result", + "initialize_geometric_search_run", } - assert forbidden.isdisjoint(imported_names) - assert forbidden.isdisjoint(calls) + assert forbidden_imports.isdisjoint(imported_names) + assert forbidden_imports.isdisjoint(calls) + + # Allowed imported names from generate module family must be record types/enums only + allowed_imported_names = { + "CandidateOperatorResult", + "CandidateAttemptRunBinding", + "ReplayAdapterResult", + "ReplayAdapterRefusal", + "ReplayDisposition", + "GeometricSearchRun", + "SearchRunRefusal", + "SearchRunDisposition", + "SourceSpan", + # standard library imports and dataclass/enum decorators/helpers + "annotations", + "hashlib", + "json", + "dataclass", + "Enum", + "unique", + } + # Check that any imported name starting with generate or from a generate module is allowed + generate_imports = set() + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom) and node.module and node.module.startswith("generate"): + generate_imports.update(alias.name for alias in node.names) + assert generate_imports <= allowed_imported_names, f"Imported names that are not allowed records: {generate_imports - allowed_imported_names}" def test_bound_dataclasses_have_no_forbidden_authority_fields() -> None: