feat(kernel): implement diagnostic-only SearchGateDecision
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
This commit is contained in:
Shay
GitHub
parent
8fc5c46cc2
commit
7b2456d402
3 changed files with 655 additions and 1 deletions
308
generate/search_gate.py
Normal file
308
generate/search_gate.py
Normal file
|
|
@ -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",
|
||||
]
|
||||
|
|
@ -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()
|
||||
|
||||
|
|
|
|||
346
tests/test_search_gate.py
Normal file
346
tests/test_search_gate.py
Normal file
|
|
@ -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}"
|
||||
)
|
||||
Loading…
Reference in a new issue