test(kernel): reconcile bound evidence spine static guards
This commit is contained in:
parent
ff1280ffa2
commit
df72c71182
4 changed files with 229 additions and 8 deletions
|
|
@ -1108,15 +1108,47 @@ def test_module_coupling_and_side_effect_guards() -> None:
|
||||||
"time",
|
"time",
|
||||||
"uuid4",
|
"uuid4",
|
||||||
}.isdisjoint(calls)
|
}.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/geometric_search_run.py",
|
||||||
"generate/replay_adapter.py",
|
"generate/replay_adapter.py",
|
||||||
"generate/sealed_practice_trace.py",
|
"generate/sealed_practice_trace.py",
|
||||||
"generate/search_gate.py",
|
"generate/search_gate.py",
|
||||||
"generate/compute_budget.py",
|
"generate/compute_budget.py",
|
||||||
"generate/contract_residuals.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:
|
def test_no_filesystem_network_clock_random_or_environment_identity() -> None:
|
||||||
|
|
|
||||||
|
|
@ -947,19 +947,35 @@ def test_module_coupling_and_side_effect_guards() -> None:
|
||||||
"json",
|
"json",
|
||||||
"generate.geometric_search_run",
|
"generate.geometric_search_run",
|
||||||
"generate.kernel_facts",
|
"generate.kernel_facts",
|
||||||
|
"generate.candidate_operator",
|
||||||
|
"generate.run_attempt_binding",
|
||||||
}
|
}
|
||||||
assert {
|
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_search_gate",
|
||||||
"decide_compute_budget",
|
"decide_compute_budget",
|
||||||
"initialize_geometric_search_run",
|
|
||||||
"assess_contracts",
|
"assess_contracts",
|
||||||
"project_contract_residuals",
|
"project_contract_residuals",
|
||||||
"determine",
|
"determine",
|
||||||
}.isdisjoint(imported_names)
|
}.isdisjoint(imported_names)
|
||||||
assert {
|
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_search_gate",
|
||||||
"decide_compute_budget",
|
"decide_compute_budget",
|
||||||
"initialize_geometric_search_run",
|
|
||||||
"assess_contracts",
|
"assess_contracts",
|
||||||
"project_contract_residuals",
|
"project_contract_residuals",
|
||||||
"determine",
|
"determine",
|
||||||
|
|
|
||||||
|
|
@ -312,3 +312,116 @@ def test_refusal_is_not_partial_binding() -> None:
|
||||||
assert isinstance(outcome, CandidateAttemptRunBindingRefusal)
|
assert isinstance(outcome, CandidateAttemptRunBindingRefusal)
|
||||||
assert not hasattr(outcome, "binding_id")
|
assert not hasattr(outcome, "binding_id")
|
||||||
assert not hasattr(outcome, "candidate_attempt_ref")
|
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)
|
||||||
|
|
|
||||||
|
|
@ -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:
|
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")
|
path = Path("generate/sealed_practice_trace.py")
|
||||||
source = path.read_text("utf-8")
|
source = path.read_text("utf-8")
|
||||||
tree = ast.parse(source)
|
tree = ast.parse(source)
|
||||||
|
imports: set[str] = set()
|
||||||
imported_names: set[str] = set()
|
imported_names: set[str] = set()
|
||||||
calls: set[str] = set()
|
calls: set[str] = set()
|
||||||
for node in ast.walk(tree):
|
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)
|
imported_names.update(alias.name for alias in node.names)
|
||||||
elif isinstance(node, ast.Call):
|
elif isinstance(node, ast.Call):
|
||||||
if isinstance(node.func, ast.Name):
|
if isinstance(node.func, ast.Name):
|
||||||
calls.add(node.func.id)
|
calls.add(node.func.id)
|
||||||
elif isinstance(node.func, ast.Attribute):
|
elif isinstance(node.func, ast.Attribute):
|
||||||
calls.add(node.func.attr)
|
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",
|
"build_missing_role_candidate",
|
||||||
|
"candidate_operator_set_id",
|
||||||
|
"GroundedUnaryDeltaCue",
|
||||||
"bind_candidate_attempt_to_run",
|
"bind_candidate_attempt_to_run",
|
||||||
|
"build_replay_adapter_input",
|
||||||
"build_replay_adapter_input_from_binding",
|
"build_replay_adapter_input_from_binding",
|
||||||
"classify_replay_result",
|
"classify_replay_result",
|
||||||
|
"initialize_geometric_search_run",
|
||||||
}
|
}
|
||||||
assert forbidden.isdisjoint(imported_names)
|
assert forbidden_imports.isdisjoint(imported_names)
|
||||||
assert forbidden.isdisjoint(calls)
|
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:
|
def test_bound_dataclasses_have_no_forbidden_authority_fields() -> None:
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue