diff --git a/generate/search_gate.py b/generate/search_gate.py new file mode 100644 index 00000000..3731fe91 --- /dev/null +++ b/generate/search_gate.py @@ -0,0 +1,308 @@ +"""Diagnostic-only SearchGateDecision adapter over ContractResidual records.""" + +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 + +SEARCH_GATE_POLICY_VERSION = "search_gate.v1" + + +@unique +class SearchGateStatus(str, Enum): + ELIGIBLE = "eligible" + INELIGIBLE = "ineligible" + BLOCKED = "blocked" + UNASSESSABLE = "unassessable" + + +@dataclass(frozen=True, slots=True) +class SearchGateDecision: + decision_id: str + policy_version: str + input_digest: 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 _enum_value(value: object) -> object: + return getattr(value, "value", value) + + +def _map_residual(kind: object) -> tuple[SearchGateStatus, str]: + return _KIND_MAP.get( + kind, (SearchGateStatus.BLOCKED, "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 _ordered_residuals( + residuals: tuple[ContractResidual, ...], +) -> tuple[ContractResidual, ...]: + return tuple(sorted(residuals, key=lambda residual: residual.residual_id)) + + +def _collect_evidence_spans( + residuals: tuple[ContractResidual, ...], +) -> tuple[SourceSpan, ...]: + return tuple( + span + for residual in residuals + for span in residual.evidence_spans + ) + + +def _residual_payload(residual: ContractResidual) -> dict[str, object]: + return { + "residual_id": residual.residual_id, + "candidate_organ": residual.candidate_organ, + "family_id": residual.family_id, + "residual_kind": _enum_value(residual.residual_kind), + "residual_code": residual.residual_code, + "source_axis": _enum_value(residual.source_axis), + "evidence_spans": [ + _span_payload(span) for span in residual.evidence_spans + ], + } + + +def _sha256_json(payload: dict[str, object]) -> str: + encoded = json.dumps( + payload, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def _input_digest(residuals: tuple[ContractResidual, ...]) -> str: + return _sha256_json( + {"residuals": [_residual_payload(residual) for residual in residuals]} + ) + + +def _decision_id( + *, + policy_version: str, + input_digest: str, + residual_ids: tuple[str, ...], + candidate_organ: str | None, + status: SearchGateStatus, + reason_code: str, + evidence_spans: tuple[SourceSpan, ...], +) -> str: + return _sha256_json( + { + "policy_version": policy_version, + "input_digest": input_digest, + "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], + } + ) + + +def _make_decision( + *, + ordered_residuals: tuple[ContractResidual, ...], + candidate_organ: str | None, + status: SearchGateStatus, + reason_code: str, + explanation: str, +) -> SearchGateDecision: + residual_ids = tuple(residual.residual_id for residual in ordered_residuals) + evidence_spans = _collect_evidence_spans(ordered_residuals) + input_digest = _input_digest(ordered_residuals) + decision_id = _decision_id( + policy_version=SEARCH_GATE_POLICY_VERSION, + input_digest=input_digest, + residual_ids=residual_ids, + candidate_organ=candidate_organ, + status=status, + reason_code=reason_code, + evidence_spans=evidence_spans, + ) + return SearchGateDecision( + decision_id=decision_id, + policy_version=SEARCH_GATE_POLICY_VERSION, + input_digest=input_digest, + residual_ids=residual_ids, + candidate_organ=candidate_organ, + status=status, + reason_code=reason_code, + evidence_spans=evidence_spans, + explanation=explanation, + ) + + +def decide_search_gate( + residuals: tuple[ContractResidual, ...], +) -> tuple[SearchGateDecision, ...]: + if not residuals: + return ( + _make_decision( + ordered_residuals=(), + candidate_organ=None, + status=SearchGateStatus.UNASSESSABLE, + reason_code="unassessable_empty_context", + explanation="Empty residual context.", + ), + ) + + ordered_residuals = _ordered_residuals(residuals) + organs = {residual.candidate_organ for residual in ordered_residuals} + if len(organs) > 1: + return ( + _make_decision( + ordered_residuals=ordered_residuals, + candidate_organ=None, + status=SearchGateStatus.UNASSESSABLE, + reason_code="unassessable_mixed_candidate_organs", + explanation="Mixed candidate organs in residual context.", + ), + ) + + candidate_organ = ordered_residuals[0].candidate_organ + mapped = [ + (_map_residual(residual.residual_kind), residual) + for residual in ordered_residuals + ] + blocked_items = [ + item for item in mapped if item[0][0] != SearchGateStatus.ELIGIBLE + ] + + if blocked_items: + if any( + status == SearchGateStatus.BLOCKED + for (status, _), _ in blocked_items + ): + overall_status = SearchGateStatus.BLOCKED + else: + overall_status = SearchGateStatus.INELIGIBLE + + 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: + overall_status = SearchGateStatus.ELIGIBLE + + 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}" + + return ( + _make_decision( + ordered_residuals=ordered_residuals, + candidate_organ=candidate_organ, + status=overall_status, + reason_code=reason_code, + 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..2f5e6c7d --- /dev/null +++ b/tests/test_search_gate.py @@ -0,0 +1,346 @@ +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}" + )