core-labs/core
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

test(kernel): reconcile bound evidence spine static guards

Browse source
This commit is contained in:
Shay 2026-06-23 00:28:25 -07:00
parent ff1280ffa2
commit df72c71182
4 changed files with 229 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

36
tests/test_candidate_operator.py
View file

@ -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:

20
tests/test_replay_adapter.py
View file

@ -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",

113
tests/test_run_attempt_binding.py
View file

@ -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)

68
tests/test_sealed_practice_trace_bound_episode.py
View file

@ -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: