7b2456d402
Implements a diagnostic-only SearchGateDecision adapter over ContractResidual records.
Preserves the authority boundary:
ContractAssessment -> ContractResidual -> SearchGateDecision.
SearchGateDecision is not proof, repair, budget, search, serving, candidate generation, or mutation authority.
Validation reported on head 1d68292c7f:
- tests/test_search_gate.py: 23 passed
- tests/test_contract_residuals.py: 75 passed
- adjacent contract/proposal tests: 48 passed
- ruff: passed
- compileall: passed
- smoke: 108 passed
- git diff --check: clean
346 lines
12 KiB
Python
346 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import ast
|
|
import dataclasses
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from generate.contract_residuals import ContractResidual, ResidualKind, ResidualSourceAxis
|
|
from generate.kernel_facts import SourceSpan
|
|
from generate.search_gate import SearchGateDecision, SearchGateStatus, decide_search_gate
|
|
|
|
|
|
def _span(text: str = "test", start: int = 0, end: int = 4, sentence_index: int | None = None) -> SourceSpan:
|
|
return SourceSpan(text=text, start=start, end=end, sentence_index=sentence_index)
|
|
|
|
|
|
def _span_payload(span: SourceSpan) -> dict[str, object]:
|
|
return {
|
|
"text": span.text,
|
|
"start": span.start,
|
|
"end": span.end,
|
|
"sentence_index": span.sentence_index,
|
|
}
|
|
|
|
|
|
def _residual(
|
|
*,
|
|
residual_id: str | None = None,
|
|
candidate_organ: str = "unary_delta_transition",
|
|
family_id: str | None = "state_change.unary_delta",
|
|
residual_kind: ResidualKind = ResidualKind.MISSING_ROLE,
|
|
residual_code: str = "changed_object_unbound",
|
|
source_axis: ResidualSourceAxis = ResidualSourceAxis.ROLE,
|
|
evidence_spans: tuple[SourceSpan, ...] = (),
|
|
explanation: str = "explanation text",
|
|
) -> ContractResidual:
|
|
if residual_id is None:
|
|
payload = {
|
|
"candidate_organ": candidate_organ,
|
|
"residual_kind": residual_kind.value,
|
|
"residual_code": residual_code,
|
|
"evidence_spans": [_span_payload(span) for span in evidence_spans],
|
|
}
|
|
residual_id = hashlib.sha256(
|
|
json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
|
|
).hexdigest()
|
|
return ContractResidual(
|
|
residual_id=residual_id,
|
|
candidate_organ=candidate_organ,
|
|
family_id=family_id,
|
|
residual_kind=residual_kind,
|
|
residual_code=residual_code,
|
|
source_axis=source_axis,
|
|
evidence_spans=evidence_spans,
|
|
explanation=explanation,
|
|
)
|
|
|
|
|
|
def test_public_api() -> None:
|
|
import generate.search_gate as sg
|
|
|
|
assert tuple(sg.__all__) == (
|
|
"SearchGateStatus",
|
|
"SearchGateDecision",
|
|
"decide_search_gate",
|
|
)
|
|
|
|
|
|
def test_decision_fields_are_diagnostic_only() -> None:
|
|
fields = {field.name for field in dataclasses.fields(SearchGateDecision)}
|
|
assert {"policy_version", "input_digest"}.issubset(fields)
|
|
assert fields.isdisjoint(
|
|
{
|
|
"budget",
|
|
"priority",
|
|
"rank",
|
|
"action",
|
|
"repair",
|
|
"candidate",
|
|
"answer",
|
|
"proof",
|
|
"serving_allowed",
|
|
"mutation",
|
|
"runnable",
|
|
"verdict",
|
|
"search_run",
|
|
"proposal",
|
|
"promotion",
|
|
}
|
|
)
|
|
|
|
|
|
def test_empty_context_is_unassessable_and_replay_stable() -> None:
|
|
first = decide_search_gate(())[0]
|
|
second = decide_search_gate(())[0]
|
|
|
|
assert first.status is SearchGateStatus.UNASSESSABLE
|
|
assert first.reason_code == "unassessable_empty_context"
|
|
assert first.candidate_organ is None
|
|
assert first.residual_ids == ()
|
|
assert first.evidence_spans == ()
|
|
assert first.policy_version == "search_gate.v1"
|
|
assert len(first.input_digest) == 64
|
|
assert len(first.decision_id) == 64
|
|
assert first == second
|
|
|
|
|
|
def test_mixed_candidate_organs_fail_closed_with_preserved_spans() -> None:
|
|
r1 = _residual(candidate_organ="organ_a", evidence_spans=(_span("aaa", 0, 3),))
|
|
r2 = _residual(candidate_organ="organ_b", evidence_spans=(_span("bbb", 4, 7),))
|
|
|
|
decision = decide_search_gate((r1, r2))[0]
|
|
sorted_residuals = tuple(sorted([r1, r2], key=lambda residual: residual.residual_id))
|
|
|
|
assert decision.status is SearchGateStatus.UNASSESSABLE
|
|
assert decision.reason_code == "unassessable_mixed_candidate_organs"
|
|
assert decision.candidate_organ is None
|
|
assert decision.residual_ids == tuple(residual.residual_id for residual in sorted_residuals)
|
|
assert decision.evidence_spans == (
|
|
sorted_residuals[0].evidence_spans + sorted_residuals[1].evidence_spans
|
|
)
|
|
|
|
|
|
def test_context_with_only_eligible_residuals_is_eligible() -> None:
|
|
r1 = _residual(residual_kind=ResidualKind.MISSING_ROLE, residual_code="quantity_unbound")
|
|
r2 = _residual(
|
|
residual_kind=ResidualKind.MISSING_RELATION,
|
|
residual_code="local_binding_relation_unbound",
|
|
)
|
|
|
|
decision = decide_search_gate((r1, r2))[0]
|
|
|
|
assert decision.status is SearchGateStatus.ELIGIBLE
|
|
assert decision.reason_code == "eligible_missing_relation"
|
|
assert decision.candidate_organ == "unary_delta_transition"
|
|
|
|
|
|
def test_context_with_any_blocker_fails_closed() -> None:
|
|
r1 = _residual(residual_kind=ResidualKind.MISSING_ROLE, residual_code="quantity_unbound")
|
|
r2 = _residual(
|
|
residual_kind=ResidualKind.AMBIGUOUS_ROLE,
|
|
residual_code="quantity_ambiguous",
|
|
)
|
|
|
|
decision = decide_search_gate((r1, r2))[0]
|
|
|
|
assert decision.status is SearchGateStatus.BLOCKED
|
|
assert decision.reason_code == "blocked_ambiguous_role"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("kind", "expected_status", "expected_reason"),
|
|
[
|
|
(ResidualKind.MISSING_PROPOSAL, SearchGateStatus.ELIGIBLE, "eligible_missing_proposal"),
|
|
(ResidualKind.MISSING_RELATION, SearchGateStatus.ELIGIBLE, "eligible_missing_relation"),
|
|
(ResidualKind.MISSING_ROLE, SearchGateStatus.ELIGIBLE, "eligible_missing_role"),
|
|
(ResidualKind.TARGET_UNBOUND, SearchGateStatus.ELIGIBLE, "eligible_target_unbound"),
|
|
(ResidualKind.AMBIGUOUS_RELATION, SearchGateStatus.BLOCKED, "blocked_ambiguous_relation"),
|
|
(ResidualKind.AMBIGUOUS_ROLE, SearchGateStatus.BLOCKED, "blocked_ambiguous_role"),
|
|
(ResidualKind.INEXACT_PROVENANCE, SearchGateStatus.BLOCKED, "blocked_inexact_provenance"),
|
|
(ResidualKind.NONLOCAL_BINDING, SearchGateStatus.BLOCKED, "blocked_nonlocal_binding"),
|
|
(ResidualKind.UNSUPPORTED_TOPOLOGY, SearchGateStatus.BLOCKED, "blocked_unsupported_topology"),
|
|
(ResidualKind.UNIT_OBJECT_CONFLICT, SearchGateStatus.BLOCKED, "blocked_unit_object_conflict"),
|
|
(ResidualKind.HAZARD_BLOCKED, SearchGateStatus.BLOCKED, "blocked_hazard"),
|
|
(ResidualKind.CONTRACT_GAP_UNCLASSIFIED, SearchGateStatus.BLOCKED, "blocked_unclassified_gap"),
|
|
],
|
|
)
|
|
def test_every_residual_kind_has_explicit_policy(
|
|
kind: ResidualKind,
|
|
expected_status: SearchGateStatus,
|
|
expected_reason: str,
|
|
) -> None:
|
|
decision = decide_search_gate((_residual(residual_kind=kind),))[0]
|
|
|
|
assert decision.status is expected_status
|
|
assert decision.reason_code == expected_reason
|
|
|
|
|
|
def test_unknown_residual_kind_fails_closed() -> None:
|
|
residual = _residual()
|
|
object.__setattr__(residual, "residual_kind", "UNKNOWN_FUTURE_KIND")
|
|
|
|
decision = decide_search_gate((residual,))[0]
|
|
|
|
assert decision.status is SearchGateStatus.BLOCKED
|
|
assert decision.reason_code == "blocked_unclassified_gap"
|
|
|
|
|
|
def test_decision_and_input_digest_are_deterministic() -> None:
|
|
r1 = _residual(
|
|
residual_kind=ResidualKind.MISSING_ROLE,
|
|
residual_code="quantity_unbound",
|
|
evidence_spans=(_span("aaa", 0, 3), _span("bbb", 5, 8)),
|
|
explanation="explanation A",
|
|
)
|
|
r2 = _residual(
|
|
residual_kind=ResidualKind.MISSING_ROLE,
|
|
residual_code="entity_unbound",
|
|
evidence_spans=(_span("ccc", 10, 13),),
|
|
explanation="explanation B",
|
|
)
|
|
|
|
forward = decide_search_gate((r1, r2))[0]
|
|
backward = decide_search_gate((r2, r1))[0]
|
|
|
|
assert forward.decision_id == backward.decision_id
|
|
assert forward.input_digest == backward.input_digest
|
|
assert forward.residual_ids == backward.residual_ids
|
|
assert forward.evidence_spans == backward.evidence_spans
|
|
|
|
r1_different_explanation = _residual(
|
|
residual_kind=ResidualKind.MISSING_ROLE,
|
|
residual_code="quantity_unbound",
|
|
evidence_spans=(_span("aaa", 0, 3), _span("bbb", 5, 8)),
|
|
explanation="different explanation",
|
|
)
|
|
different_explanation = decide_search_gate((r1_different_explanation, r2))[0]
|
|
assert different_explanation.decision_id == forward.decision_id
|
|
assert different_explanation.input_digest == forward.input_digest
|
|
|
|
r1_different_span = _residual(
|
|
residual_kind=ResidualKind.MISSING_ROLE,
|
|
residual_code="quantity_unbound",
|
|
evidence_spans=(_span("aaa", 0, 3), _span("bbb", 5, 9)),
|
|
explanation="explanation A",
|
|
)
|
|
different_span = decide_search_gate((r1_different_span, r2))[0]
|
|
assert different_span.decision_id != forward.decision_id
|
|
assert different_span.input_digest != forward.input_digest
|
|
|
|
|
|
def test_duplicate_evidence_spans_are_preserved_per_residual() -> None:
|
|
span = _span("same", 0, 4, 0)
|
|
r1 = _residual(residual_id="a", evidence_spans=(span,))
|
|
r2 = _residual(residual_id="b", residual_code="entity_unbound", evidence_spans=(span,))
|
|
|
|
decision = decide_search_gate((r2, r1))[0]
|
|
|
|
assert decision.residual_ids == ("a", "b")
|
|
assert decision.evidence_spans == (span, span)
|
|
|
|
|
|
def test_input_digest_uses_complete_context_without_explanation() -> None:
|
|
span = _span("aaa", 0, 3, 0)
|
|
residual = _residual(
|
|
residual_id="residual-a",
|
|
candidate_organ="unary_delta_transition",
|
|
family_id="state_change.unary_delta",
|
|
residual_kind=ResidualKind.MISSING_ROLE,
|
|
residual_code="changed_object_unbound",
|
|
source_axis=ResidualSourceAxis.ROLE,
|
|
evidence_spans=(span,),
|
|
explanation="excluded explanation",
|
|
)
|
|
|
|
decision = decide_search_gate((residual,))[0]
|
|
expected_payload = {
|
|
"residuals": [
|
|
{
|
|
"residual_id": "residual-a",
|
|
"candidate_organ": "unary_delta_transition",
|
|
"family_id": "state_change.unary_delta",
|
|
"residual_kind": "missing_role",
|
|
"residual_code": "changed_object_unbound",
|
|
"source_axis": "role",
|
|
"evidence_spans": [_span_payload(span)],
|
|
}
|
|
]
|
|
}
|
|
expected_digest = hashlib.sha256(
|
|
json.dumps(
|
|
expected_payload,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
ensure_ascii=False,
|
|
).encode("utf-8")
|
|
).hexdigest()
|
|
|
|
assert decision.input_digest == expected_digest
|
|
|
|
|
|
def test_coupling_guards() -> None:
|
|
path = Path(__file__).parent.parent / "generate" / "search_gate.py"
|
|
tree = ast.parse(path.read_text("utf-8"))
|
|
|
|
forbidden_imports = {
|
|
"runtime",
|
|
"serving",
|
|
"workbench",
|
|
"teaching",
|
|
"eval",
|
|
"report",
|
|
"vault",
|
|
"recall",
|
|
"field",
|
|
"algebra",
|
|
"subprocess",
|
|
"socket",
|
|
"random",
|
|
"datetime",
|
|
"time",
|
|
}
|
|
forbidden_calls = {
|
|
"assess_contracts",
|
|
"project_contract_residuals",
|
|
"build_problem_frame",
|
|
"determine",
|
|
"search",
|
|
"repair",
|
|
"serve",
|
|
"store",
|
|
"write",
|
|
"open",
|
|
"write_text",
|
|
"write_bytes",
|
|
}
|
|
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Import):
|
|
for name in node.names:
|
|
parts = name.name.split(".")
|
|
assert not any(
|
|
part in forbidden_imports for part in parts
|
|
), f"Forbidden import: {name.name}"
|
|
elif isinstance(node, ast.ImportFrom):
|
|
if node.module:
|
|
parts = node.module.split(".")
|
|
assert not any(
|
|
part in forbidden_imports for part in parts
|
|
), f"Forbidden import: {node.module}"
|
|
for name in node.names:
|
|
assert name.name not in forbidden_calls, (
|
|
f"Forbidden import of function: {name.name}"
|
|
)
|
|
elif isinstance(node, ast.Call):
|
|
if isinstance(node.func, ast.Name):
|
|
assert node.func.id not in forbidden_calls, (
|
|
f"Forbidden call: {node.func.id}"
|
|
)
|
|
elif isinstance(node.func, ast.Attribute):
|
|
assert node.func.attr not in forbidden_calls, (
|
|
f"Forbidden call/method: {node.func.attr}"
|
|
)
|