feat(kernel): implement diagnostic SearchGateDecision

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",
+]

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

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}"