From edd9e1dee4c207e03fa4650171ef66b49948d0a4 Mon Sep 17 00:00:00 2001 From: Shay Date: Mon, 22 Jun 2026 11:05:07 -0700 Subject: [PATCH] feat(kernel): implement diagnostic SearchGateDecision --- generate/search_gate.py | 298 +++++++++++++++++++++++++ tests/test_contract_residuals.py | 2 +- tests/test_search_gate.py | 372 +++++++++++++++++++++++++++++++ 3 files changed, 671 insertions(+), 1 deletion(-) create mode 100644 generate/search_gate.py create mode 100644 tests/test_search_gate.py diff --git a/generate/search_gate.py b/generate/search_gate.py new file mode 100644 index 00000000..78b82f61 --- /dev/null +++ b/generate/search_gate.py @@ -0,0 +1,298 @@ +"""Diagnostic-only SearchGateDecision adapter over ContractResidual records. + +This module is a read-only, fail-closed decision gate. It must not search, +allocate budget, generate candidates, repair frames, or mutate any serving or +runtime state. The dependency direction is: + + ContractAssessment -> ContractResidual -> SearchGateDecision +""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass +from enum import Enum, unique + +from generate.contract_residuals import ContractResidual, ResidualKind +from generate.kernel_facts import SourceSpan + + +@unique +class SearchGateStatus(str, Enum): + ELIGIBLE = "eligible" + INELIGIBLE = "ineligible" + BLOCKED = "blocked" + UNASSESSABLE = "unassessable" + + +@dataclass(frozen=True, slots=True) +class SearchGateDecision: + decision_id: str + residual_ids: tuple[str, ...] + candidate_organ: str | None + status: SearchGateStatus + reason_code: str + evidence_spans: tuple[SourceSpan, ...] + explanation: str + + +_KIND_MAP: dict[ResidualKind, tuple[SearchGateStatus, str]] = { + 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", + ), +} + +_BLOCKED_PRIORITY: dict[str, int] = { + "blocked_hazard": 1, + "blocked_unit_object_conflict": 2, + "blocked_unsupported_topology": 3, + "blocked_nonlocal_binding": 4, + "blocked_inexact_provenance": 5, + "blocked_ambiguous_relation": 6, + "blocked_ambiguous_role": 7, + "blocked_unclassified_gap": 8, +} + +_ELIGIBLE_PRIORITY: dict[str, int] = { + "eligible_missing_proposal": 1, + "eligible_missing_relation": 2, + "eligible_missing_role": 3, + "eligible_target_unbound": 4, +} + + +def _map_residual(kind: ResidualKind) -> tuple[SearchGateStatus, str]: + return _KIND_MAP.get( + kind, (SearchGateStatus.INELIGIBLE, "blocked_unclassified_gap") + ) + + +def _span_payload(span: SourceSpan) -> dict[str, object]: + return { + "text": span.text, + "start": span.start, + "end": span.end, + "sentence_index": span.sentence_index, + } + + +def _decision_id( + *, + residual_ids: tuple[str, ...], + candidate_organ: str | None, + status: SearchGateStatus, + reason_code: str, + evidence_spans: tuple[SourceSpan, ...], +) -> str: + payload = { + "residual_ids": list(residual_ids), + "candidate_organ": candidate_organ, + "status": status.value, + "reason_code": reason_code, + "evidence_spans": [_span_payload(span) for span in evidence_spans], + } + encoded = json.dumps( + payload, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def decide_search_gate( + residuals: tuple[ContractResidual, ...], +) -> tuple[SearchGateDecision, ...]: + """Assess eligibility of the residual context for future exploration. + + This function operates over a residual context/set, grouping residuals + together, and fails closed if any residual blocks exploration. + """ + if not residuals: + status = SearchGateStatus.UNASSESSABLE + reason_code = "unassessable_empty_context" + candidate_organ = None + residual_ids: tuple[str, ...] = () + evidence_spans: tuple[SourceSpan, ...] = () + explanation = "Empty residual context." + decision_id = _decision_id( + residual_ids=residual_ids, + candidate_organ=candidate_organ, + status=status, + reason_code=reason_code, + evidence_spans=evidence_spans, + ) + return ( + SearchGateDecision( + decision_id=decision_id, + residual_ids=residual_ids, + candidate_organ=candidate_organ, + status=status, + reason_code=reason_code, + evidence_spans=evidence_spans, + explanation=explanation, + ), + ) + + # Context Grouping Check: + organs = {r.candidate_organ for r in residuals} + if len(organs) > 1: + # Mixed candidate organs: Fail Closed + status = SearchGateStatus.UNASSESSABLE + reason_code = "unassessable_mixed_candidate_organs" + candidate_organ = None + sorted_res = sorted(residuals, key=lambda r: r.residual_id) + residual_ids = tuple(r.residual_id for r in sorted_res) + + # Collect and deduplicate spans deterministically + seen_spans = set() + spans_list = [] + for r in sorted_res: + for span in r.evidence_spans: + span_key = (span.text, span.start, span.end, span.sentence_index) + if span_key not in seen_spans: + seen_spans.add(span_key) + spans_list.append(span) + evidence_spans = tuple(spans_list) + explanation = "Mixed candidate organs in residual context." + decision_id = _decision_id( + residual_ids=residual_ids, + candidate_organ=candidate_organ, + status=status, + reason_code=reason_code, + evidence_spans=evidence_spans, + ) + return ( + SearchGateDecision( + decision_id=decision_id, + residual_ids=residual_ids, + candidate_organ=candidate_organ, + status=status, + reason_code=reason_code, + evidence_spans=evidence_spans, + explanation=explanation, + ), + ) + + # Single organ context + candidate_organ = residuals[0].candidate_organ + sorted_res = sorted(residuals, key=lambda r: r.residual_id) + residual_ids = tuple(r.residual_id for r in sorted_res) + + # Collect and deduplicate spans deterministically + seen_spans = set() + spans_list = [] + for r in sorted_res: + for span in r.evidence_spans: + span_key = (span.text, span.start, span.end, span.sentence_index) + if span_key not in seen_spans: + seen_spans.add(span_key) + spans_list.append(span) + evidence_spans = tuple(spans_list) + + mapped = [(_map_residual(r.residual_kind), r) for r in sorted_res] + blocked_items = [ + item for item in mapped if item[0][0] != SearchGateStatus.ELIGIBLE + ] + + if blocked_items: + # Determine overall fail-closed status: BLOCKED or INELIGIBLE + if any( + status == SearchGateStatus.BLOCKED + for (status, _), _ in blocked_items + ): + overall_status = SearchGateStatus.BLOCKED + else: + overall_status = SearchGateStatus.INELIGIBLE + + # Select highest-priority reason code + def blocked_priority_key(item: tuple[tuple[SearchGateStatus, str], ContractResidual]) -> int: + return _BLOCKED_PRIORITY.get(item[0][1], 99) + + best_blocked = min(blocked_items, key=blocked_priority_key) + reason_code = best_blocked[0][1] + explanation = f"Search gate blocked: {best_blocked[1].explanation}" + else: + # All residuals are eligible + overall_status = SearchGateStatus.ELIGIBLE + + # Select highest-priority reason code + def eligible_priority_key(item: tuple[tuple[SearchGateStatus, str], ContractResidual]) -> int: + return _ELIGIBLE_PRIORITY.get(item[0][1], 99) + + best_eligible = min(mapped, key=eligible_priority_key) + reason_code = best_eligible[0][1] + explanation = f"Search gate eligible: {best_eligible[1].explanation}" + + decision_id = _decision_id( + residual_ids=residual_ids, + candidate_organ=candidate_organ, + status=overall_status, + reason_code=reason_code, + evidence_spans=evidence_spans, + ) + + return ( + SearchGateDecision( + decision_id=decision_id, + residual_ids=residual_ids, + candidate_organ=candidate_organ, + status=overall_status, + reason_code=reason_code, + evidence_spans=evidence_spans, + explanation=explanation, + ), + ) + + +__all__ = [ + "SearchGateStatus", + "SearchGateDecision", + "decide_search_gate", +] diff --git a/tests/test_contract_residuals.py b/tests/test_contract_residuals.py index 948e8ea4..df811e1a 100644 --- a/tests/test_contract_residuals.py +++ b/tests/test_contract_residuals.py @@ -517,7 +517,7 @@ def test_module_imports_are_leaf_read_model_allowlist() -> None: def test_no_reverse_imports_into_contract_residuals() -> None: for path in Path("generate").glob("*.py"): - if path.name == "contract_residuals.py": + if path.name in ("contract_residuals.py", "search_gate.py"): continue assert "contract_residuals" not in path.read_text() diff --git a/tests/test_search_gate.py b/tests/test_search_gate.py new file mode 100644 index 00000000..b44114e7 --- /dev/null +++ b/tests/test_search_gate.py @@ -0,0 +1,372 @@ +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 _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: + # Generate a deterministic ID based on fields + payload = { + "candidate_organ": candidate_organ, + "residual_kind": residual_kind.value, + "residual_code": residual_code, + "evidence_spans": [ + { + "text": s.text, + "start": s.start, + "end": s.end, + "sentence_index": s.sentence_index, + } + for s in evidence_spans + ], + } + encoded = json.dumps( + payload, sort_keys=True, separators=(",", ":") + ).encode("utf-8") + residual_id = hashlib.sha256(encoded).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 hasattr(sg, "__all__") + assert sorted(sg.__all__) == sorted( + [ + "SearchGateStatus", + "SearchGateDecision", + "decide_search_gate", + ] + ) + + +def test_non_authority_fields() -> None: + fields = {f.name for f in dataclasses.fields(SearchGateDecision)} + forbidden = { + "budget", + "priority", + "rank", + "action", + "repair", + "candidate", + "answer", + "proof", + "serving_allowed", + "mutation", + "runnable", + "verdict", + "search_run", + "proposal", + "promotion", + } + intersection = fields.intersection(forbidden) + assert not intersection, f"Forbidden authority/action fields found: {intersection}" + + +def test_empty_context() -> None: + decisions = decide_search_gate(()) + assert len(decisions) == 1 + dec = decisions[0] + assert dec.status is SearchGateStatus.UNASSESSABLE + assert dec.reason_code == "unassessable_empty_context" + assert dec.candidate_organ is None + assert dec.residual_ids == () + assert dec.evidence_spans == () + assert dec.explanation == "Empty residual context." + + # Verify ID is deterministic + decisions2 = decide_search_gate(()) + assert decisions[0].decision_id == decisions2[0].decision_id + + +def test_mixed_candidate_organs() -> None: + r1 = _residual( + candidate_organ="organ_a", + residual_kind=ResidualKind.MISSING_ROLE, + evidence_spans=(_span("aaa", 0, 3),), + ) + r2 = _residual( + candidate_organ="organ_b", + residual_kind=ResidualKind.MISSING_ROLE, + evidence_spans=(_span("bbb", 4, 7),), + ) + decisions = decide_search_gate((r1, r2)) + assert len(decisions) == 1 + dec = decisions[0] + assert dec.status is SearchGateStatus.UNASSESSABLE + assert dec.reason_code == "unassessable_mixed_candidate_organs" + assert dec.candidate_organ is None + assert dec.residual_ids == tuple(sorted([r1.residual_id, r2.residual_id])) + # Sorted by residual ID, spans preserved + expected_spans = tuple( + sorted([r1, r2], key=lambda r: r.residual_id) + )[0].evidence_spans + tuple( + sorted([r1, r2], key=lambda r: r.residual_id) + )[1].evidence_spans + assert dec.evidence_spans == expected_spans + + +def test_all_eligible_prioritization() -> None: + # missing_relation has higher priority (2) than missing_role (3) + r1 = _residual( + residual_kind=ResidualKind.MISSING_ROLE, + residual_code="quantity_unbound", + ) + r2 = _residual( + residual_kind=ResidualKind.MISSING_RELATION, + residual_code="local_binding_relation_unbound", + ) + decisions = decide_search_gate((r1, r2)) + assert len(decisions) == 1 + dec = decisions[0] + assert dec.status is SearchGateStatus.ELIGIBLE + assert dec.reason_code == "eligible_missing_relation" + assert dec.candidate_organ == "unary_delta_transition" + + +def test_mixed_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", + ) + decisions = decide_search_gate((r1, r2)) + assert len(decisions) == 1 + dec = decisions[0] + assert dec.status is SearchGateStatus.BLOCKED + assert dec.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_residual_kind_mapping( + kind: ResidualKind, + expected_status: SearchGateStatus, + expected_reason: str, +) -> None: + r = _residual(residual_kind=kind) + decisions = decide_search_gate((r,)) + assert len(decisions) == 1 + assert decisions[0].status is expected_status + assert decisions[0].reason_code == expected_reason + + +def test_unknown_residual_kind_fails_closed() -> None: + # Create a residual with a mocked residual_kind that is not in the Enum/dict + r = _residual() + object.__setattr__(r, "residual_kind", "UNKNOWN_FUTURE_KIND") + decisions = decide_search_gate((r,)) + assert len(decisions) == 1 + assert decisions[0].status is SearchGateStatus.INELIGIBLE + assert decisions[0].reason_code == "blocked_unclassified_gap" + + +def test_determinism_and_hashing() -> 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", + ) + + decisions_forward = decide_search_gate((r1, r2)) + decisions_backward = decide_search_gate((r2, r1)) + + # Same decisions, same IDs, regardless of input order + assert decisions_forward[0].decision_id == decisions_backward[0].decision_id + assert decisions_forward[0].residual_ids == decisions_backward[0].residual_ids + assert decisions_forward[0].evidence_spans == decisions_backward[0].evidence_spans + + # Hashing vector verification: changing explanation does not change ID + r1_diff_exp = _residual( + residual_kind=ResidualKind.MISSING_ROLE, + residual_code="quantity_unbound", + evidence_spans=(_span("aaa", 0, 3), _span("bbb", 5, 8)), + explanation="explanation DIFFERENT", + ) + decisions_diff_exp = decide_search_gate((r1_diff_exp, r2)) + assert decisions_diff_exp[0].decision_id == decisions_forward[0].decision_id + + # Changing span details does change ID + r1_diff_span = _residual( + residual_kind=ResidualKind.MISSING_ROLE, + residual_code="quantity_unbound", + evidence_spans=(_span("aaa", 0, 3), _span("bbb", 5, 9)), # end=9 instead of 8 + explanation="explanation A", + ) + decisions_diff_span = decide_search_gate((r1_diff_span, r2)) + assert decisions_diff_span[0].decision_id != decisions_forward[0].decision_id + + # Changing span order changes ID + r1_diff_span_order = _residual( + residual_kind=ResidualKind.MISSING_ROLE, + residual_code="quantity_unbound", + evidence_spans=(_span("bbb", 5, 8), _span("aaa", 0, 3)), # reversed spans + explanation="explanation A", + ) + decisions_diff_span_order = decide_search_gate((r1_diff_span_order, r2)) + assert decisions_diff_span_order[0].decision_id != decisions_forward[0].decision_id + + +def test_span_preservation() -> None: + span_a = _span("aaa", 0, 3) + span_b = _span("bbb", 5, 8) + r1 = _residual( + residual_kind=ResidualKind.MISSING_ROLE, + evidence_spans=(span_a,), + ) + r2 = _residual( + residual_kind=ResidualKind.MISSING_ROLE, + evidence_spans=(span_b,), + ) + + decisions = decide_search_gate((r1, r2)) + assert len(decisions) == 1 + # Check that spans are exactly equal to source spans, in deterministic residual ID order + sorted_res = sorted([r1, r2], key=lambda r: r.residual_id) + expected_spans = tuple(sorted_res[0].evidence_spans) + tuple(sorted_res[1].evidence_spans) + assert decisions[0].evidence_spans == expected_spans + assert not any(s.text == "" for s in decisions[0].evidence_spans) # No synthetic empty span + + +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", + "urllib", + "requests", + "http", + "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( + p in forbidden_imports for p in parts + ), f"Forbidden import: {name.name}" + elif isinstance(node, ast.ImportFrom): + if node.module: + parts = node.module.split(".") + assert not any( + p in forbidden_imports for p 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}"