core/tests/test_search_gate.py

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
            ],
        }
        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 tuple(sg.__all__) == (
        "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_policy_version_and_input_digest_are_present() -> None:
    fields = {f.name for f in dataclasses.fields(SearchGateDecision)}
    assert "policy_version" in fields
    assert "input_digest" in fields

    decision = decide_search_gate((_residual(),))[0]

    assert decision.policy_version == "search_gate.v1"
    assert len(decision.input_digest) == 64
    assert len(decision.decision_id) == 64
    assert decision.input_digest != decision.decision_id


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."
    assert dec.policy_version == "search_gate.v1"
    assert len(dec.input_digest) == 64

    decisions2 = decide_search_gate(())
    assert decisions[0].decision_id == decisions2[0].decision_id
    assert decisions[0].input_digest == decisions2[0].input_digest


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_res = tuple(sorted([r1, r2], key=lambda r: r.residual_id))
    expected_spans = sorted_res[0].evidence_spans + sorted_res[1].evidence_spans
    assert dec.evidence_spans == expected_spans


def test_all_eligible_prioritization() -> 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",
    )
    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:
    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))

    assert decisions_forward[0].decision_id == decisions_backward[0].decision_id
    assert decisions_forward[0].input_digest == decisions_backward[0].input_digest
    assert decisions_forward[0].residual_ids == decisions_backward[0].residual_ids
    assert decisions_forward[0].evidence_spans == decisions_backward[0].evidence_spans

    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
    assert decisions_diff_exp[0].input_digest == decisions_forward[0].input_digest

    r1_diff_span = _residual(
        residual_kind=ResidualKind.MISSING_ROLE,
        residual_code="quantity_unbound",
        evidence_spans=(_span("aaa", 0, 3), _span("bbb", 5, 9)),
        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
    assert decisions_diff_span[0].input_digest != decisions_forward[0].input_digest

    r1_diff_span_order = _residual(
        residual_kind=ResidualKind.MISSING_ROLE,
        residual_code="quantity_unbound",
        evidence_spans=(_span("bbb", 5, 8), _span("aaa", 0, 3)),
        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
    assert decisions_diff_span_order[0].input_digest != decisions_forward[0].input_digest


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


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