fix: skeptic remediations for Stage 3–4 exit gates
- Metadata arm returns baseline decision payload (bit-identical digests). - TeachingStore auto-applies HE morph rule from compiled pack (live path). - Inductive derived edges stamp geometric admissibility via pipeline versor grounding; refuse ungrounded promotions. - VaultPromotionPolicy default residual_threshold=1e-6 for COHERENT. [Verification]: skeptic_fixes 26 passed; stage4 ablation digests equal
This commit is contained in:
Shay
parent
3de431dcdc
commit
c71c00cca6
8 changed files with 212 additions and 74 deletions
|
|
@ -1,22 +1,26 @@
|
|||
"""Bounded multi-step inductive closure over teaching-store relations (Stage 3C).
|
||||
|
||||
Fixed-point expansion of same-relation chains with explicit budgets,
|
||||
cycle handling, contradiction detection, and replayable provenance.
|
||||
cycle handling, contradiction detection, geometric admissibility, and
|
||||
replayable provenance.
|
||||
|
||||
Atom *identity* for entailment telemetry remains conformal
|
||||
(``CognitiveTurnPipeline._proof_atom``). This module expands the *relation
|
||||
graph* of surface triples from ``TeachingStore.triples()`` into a closed
|
||||
set under transitive composition of equal relation labels — not string
|
||||
atom join as final authority for field identity.
|
||||
atom join as final identity authority.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Iterable, Sequence
|
||||
from typing import Any, Callable, Iterable, Sequence
|
||||
|
||||
_DEFAULT_BUDGET = 16
|
||||
|
||||
# geometric_admissible(head, relation, tail) -> bool
|
||||
GeometricAdmissibleFn = Callable[[str, str, str], bool]
|
||||
|
||||
|
||||
def _norm(token: str) -> str:
|
||||
return token.strip().lower()
|
||||
|
|
@ -71,14 +75,27 @@ class InductiveClosureResult:
|
|||
"fixed_point": self.fixed_point,
|
||||
"truncated": self.truncated,
|
||||
"n_derived": len(self.derived),
|
||||
"n_admissible_derived": sum(1 for d in self.derived if d.admissible),
|
||||
}
|
||||
|
||||
|
||||
def default_geometric_admissible(head: str, relation: str, tail: str) -> bool:
|
||||
"""Strict default: a derived relation is admissible only if endpoints are
|
||||
non-empty distinct tokens and the relation is non-empty. Callers that
|
||||
have a vocab/field resolver should pass a stronger callback that checks
|
||||
closed Cl(4,1) versors (see pipeline wiring).
|
||||
"""
|
||||
del relation
|
||||
h, t = head.strip(), tail.strip()
|
||||
return bool(h) and bool(t) and h != t
|
||||
|
||||
|
||||
def expand_relation_closure(
|
||||
triples: Sequence[tuple[str, str, str]],
|
||||
*,
|
||||
budget: int = _DEFAULT_BUDGET,
|
||||
relations: Iterable[str] | None = None,
|
||||
geometric_admissible: GeometricAdmissibleFn | None = None,
|
||||
) -> InductiveClosureResult:
|
||||
"""Compute same-relation transitive closure with budget and contradictions.
|
||||
|
||||
|
|
@ -88,22 +105,20 @@ def expand_relation_closure(
|
|||
derive (a,r,c) with path a…c.
|
||||
* Cycle: if path would revisit a node, skip (no infinite loop).
|
||||
* Contradiction: two different tails for the same (head, relation)
|
||||
at the same fixed-point layer mark both as contradiction=True
|
||||
(functional assumption for same-relation edges).
|
||||
among base facts mark contradiction=True.
|
||||
* Geometric admissibility: each *derived* relation is stamped
|
||||
admissible only when ``geometric_admissible(h,r,t)`` is True.
|
||||
Default requires non-empty distinct endpoints; pipeline supplies
|
||||
versor-grounded checks when session vocab is available.
|
||||
* Termination: no new triples or budget exhausted.
|
||||
|
||||
Geometric admissibility of *field* atoms is enforced by callers that
|
||||
map surfaces through ``_proof_atom`` before using derived triples as
|
||||
proof premises; this expander is total over the teaching-store graph.
|
||||
"""
|
||||
if budget < 1:
|
||||
raise ValueError("budget must be >= 1")
|
||||
|
||||
geom = geometric_admissible or default_geometric_admissible
|
||||
rel_filter = None if relations is None else {_norm(r) for r in relations}
|
||||
|
||||
# Base
|
||||
base_list: list[DerivedRelation] = []
|
||||
# key (h,r) -> set of tails for contradiction detection
|
||||
edge_map: dict[tuple[str, str], set[str]] = {}
|
||||
known: set[tuple[str, str, str]] = set()
|
||||
|
||||
|
|
@ -118,6 +133,7 @@ def expand_relation_closure(
|
|||
continue
|
||||
known.add(key3)
|
||||
edge_map.setdefault((hn, rn), set()).add(tn)
|
||||
# Base facts are store-grounded; admissible if geometric check passes.
|
||||
base_list.append(
|
||||
DerivedRelation(
|
||||
head=hn,
|
||||
|
|
@ -125,11 +141,10 @@ def expand_relation_closure(
|
|||
tail=tn,
|
||||
path=(hn, tn),
|
||||
step=0,
|
||||
admissible=True,
|
||||
admissible=bool(geom(hn, rn, tn)),
|
||||
)
|
||||
)
|
||||
|
||||
# Mark base contradictions
|
||||
contradictions: list[DerivedRelation] = []
|
||||
for (h, r), tails in edge_map.items():
|
||||
if len(tails) > 1:
|
||||
|
|
@ -148,7 +163,6 @@ def expand_relation_closure(
|
|||
)
|
||||
|
||||
derived: list[DerivedRelation] = []
|
||||
# Working set of edges as (h,r,t) for composition
|
||||
work = set(known)
|
||||
steps_taken = 0
|
||||
fixed_point = False
|
||||
|
|
@ -157,7 +171,6 @@ def expand_relation_closure(
|
|||
for step in range(1, budget + 1):
|
||||
steps_taken = step
|
||||
new_edges: list[DerivedRelation] = []
|
||||
# Index tails by (h,r)
|
||||
by_hr: dict[tuple[str, str], list[str]] = {}
|
||||
for h, r, t in work:
|
||||
by_hr.setdefault((h, r), []).append(t)
|
||||
|
|
@ -166,12 +179,12 @@ def expand_relation_closure(
|
|||
for b in mids:
|
||||
for c in by_hr.get((b, r), ()):
|
||||
if a == c:
|
||||
continue # cycle / identity
|
||||
continue
|
||||
key3 = (a, r, c)
|
||||
if key3 in work:
|
||||
continue
|
||||
# path reconstruction (bounded)
|
||||
path = (a, b, c)
|
||||
ok = bool(geom(a, r, c))
|
||||
new_edges.append(
|
||||
DerivedRelation(
|
||||
head=a,
|
||||
|
|
@ -179,7 +192,8 @@ def expand_relation_closure(
|
|||
tail=c,
|
||||
path=path,
|
||||
step=step,
|
||||
admissible=True,
|
||||
admissible=ok,
|
||||
contradiction=False,
|
||||
)
|
||||
)
|
||||
|
||||
|
|
@ -188,9 +202,6 @@ def expand_relation_closure(
|
|||
steps_taken = step - 1 if step > 0 else 0
|
||||
break
|
||||
|
||||
# Dedup new edges. Transitive multi-tails for the same (head, relation)
|
||||
# are *not* contradictions — only base multi-tails (step 0) mark
|
||||
# functional conflicts (recorded once above).
|
||||
for dr in new_edges:
|
||||
key3 = dr.as_triple()
|
||||
if key3 in work:
|
||||
|
|
@ -199,9 +210,7 @@ def expand_relation_closure(
|
|||
edge_map.setdefault((dr.head, dr.relation), set()).add(dr.tail)
|
||||
derived.append(dr)
|
||||
else:
|
||||
# Budget exhausted without fixed point
|
||||
truncated = True
|
||||
# Check if more edges would exist
|
||||
by_hr = {}
|
||||
for h, r, t in work:
|
||||
by_hr.setdefault((h, r), []).append(t)
|
||||
|
|
|
|||
|
|
@ -442,8 +442,25 @@ class CognitiveTurnPipeline:
|
|||
compose_surface = CognitiveTurnPipeline._render_compose_surface(compose_result)
|
||||
|
||||
# Stage 3C — bounded inductive closure over teaching-store relations.
|
||||
# Provenance-preserving fixed-point; folded into operator_invocation only.
|
||||
inductive_closure = expand_relation_closure(triples, budget=16)
|
||||
# Provenance-preserving fixed-point; derived edges require geometric
|
||||
# admissibility (closed Cl(4,1) versors when vocab can ground them).
|
||||
def _geom_admissible(h: str, r: str, t: str) -> bool:
|
||||
del r
|
||||
hv = self._resolve_surface_versor(h)
|
||||
tv = self._resolve_surface_versor(t)
|
||||
if hv is None or tv is None:
|
||||
# Ungrounded endpoints cannot be promoted as geometric facts.
|
||||
return False
|
||||
return (
|
||||
float(versor_condition(hv)) < 1e-6
|
||||
and float(versor_condition(tv)) < 1e-6
|
||||
)
|
||||
|
||||
inductive_closure = expand_relation_closure(
|
||||
triples,
|
||||
budget=16,
|
||||
geometric_admissible=_geom_admissible,
|
||||
)
|
||||
|
||||
entailment_trace = self._maybe_entailment_trace(intent, triples)
|
||||
|
||||
|
|
|
|||
|
|
@ -15,9 +15,16 @@ class PromotionDecision:
|
|||
|
||||
|
||||
class VaultPromotionPolicy:
|
||||
"""Promote only settled, coherent regions into deep vault storage."""
|
||||
"""Promote only settled, *geometrically* coherent regions to COHERENT.
|
||||
|
||||
def __init__(self, residual_threshold: float = 0.05) -> None:
|
||||
Stage 3A / Master Blueprint: COHERENT standing requires the unitary
|
||||
residual condition (``coherence_residual ≤ 1e-6`` by default), not a
|
||||
soft energy-band threshold alone. Tests may pass a looser
|
||||
``residual_threshold`` explicitly for energy-class isolation fixtures;
|
||||
production ``VaultPromotionPolicy()`` uses the geometric floor.
|
||||
"""
|
||||
|
||||
def __init__(self, residual_threshold: float = 1e-6) -> None:
|
||||
if residual_threshold < 0.0:
|
||||
raise ValueError("residual_threshold must be non-negative")
|
||||
self.residual_threshold = float(residual_threshold)
|
||||
|
|
@ -27,6 +34,13 @@ class VaultPromotionPolicy:
|
|||
return PromotionDecision(False, "missing_energy_profile", EnergyClass.E2)
|
||||
if not energy.energy_class.vault_candidate:
|
||||
return PromotionDecision(False, "region_still_active", energy.energy_class)
|
||||
# Full geometric unitarity gate for COHERENT promotion (Stage 3A).
|
||||
if energy.coherence_residual > self.residual_threshold:
|
||||
return PromotionDecision(False, "coherence_residual_above_threshold", energy.energy_class)
|
||||
return PromotionDecision(
|
||||
False, "coherence_residual_above_threshold", energy.energy_class
|
||||
)
|
||||
# Residual must also be finite and non-negative (typed safety).
|
||||
r = float(energy.coherence_residual)
|
||||
if not (r == r) or r < 0.0: # NaN or negative
|
||||
return PromotionDecision(False, "coherence_residual_invalid", energy.energy_class)
|
||||
return PromotionDecision(True, "settled_coherent_region", energy.energy_class)
|
||||
|
|
|
|||
|
|
@ -130,9 +130,9 @@ def run_four_arm_ablation(
|
|||
)
|
||||
)
|
||||
|
||||
# Metrics
|
||||
# Metadata decision kind must match canonical (both PASS).
|
||||
meta_identical = d_meta.kind is d_can.kind is DecisionKind.PASS
|
||||
# Metrics — Stage 4 requires metadata bit-identical to canonical baseline
|
||||
# (full SHA-256 of decision payload, not kind-only equality).
|
||||
meta_identical = _digest(d_meta) == _digest(d_can) and d_meta.kind is DecisionKind.PASS
|
||||
# Executable must change decision to ABSTAIN vs baseline PASS.
|
||||
exec_changed = (
|
||||
d_can.kind is DecisionKind.PASS and d_exec.kind is DecisionKind.ABSTAIN
|
||||
|
|
|
|||
|
|
@ -61,8 +61,15 @@ def apply_he_morph_constraint(
|
|||
``he_surface`` is the observed HE surface form when present.
|
||||
"""
|
||||
mode = mode.strip().lower()
|
||||
# Canonical baseline decision (no morph authority). Metadata-only must
|
||||
# return this *exact* decision object shape so digests are bit-identical
|
||||
# to baseline (Stage 4 sealed harness). Morph provenance is tracked by
|
||||
# the ablation harness separately, never folded into the decision payload
|
||||
# on the metadata arm.
|
||||
baseline = ConstraintDecision(kind=DecisionKind.PASS, reason="canonical_baseline")
|
||||
|
||||
if mode == "canonical":
|
||||
return ConstraintDecision(kind=DecisionKind.PASS, reason="canonical_baseline")
|
||||
return baseline
|
||||
|
||||
if not he_surface:
|
||||
if mode == "adversarial":
|
||||
|
|
@ -70,7 +77,8 @@ def apply_he_morph_constraint(
|
|||
kind=DecisionKind.FAIL_CLOSED,
|
||||
reason="missing_he_surface",
|
||||
)
|
||||
return ConstraintDecision(kind=DecisionKind.PASS, reason="no_he_surface")
|
||||
# No HE surface → identical to baseline (including metadata arm).
|
||||
return baseline
|
||||
|
||||
hits = lookup_surface(observed_catalog, he_surface)
|
||||
if hits is None:
|
||||
|
|
@ -111,12 +119,8 @@ def apply_he_morph_constraint(
|
|||
)
|
||||
|
||||
if mode == "metadata":
|
||||
# Morph present for provenance only — decision identical to baseline.
|
||||
return ConstraintDecision(
|
||||
kind=DecisionKind.PASS,
|
||||
reason="metadata_only_inert",
|
||||
surfaces=(surface,),
|
||||
)
|
||||
# Morph observed but must not change the consumer decision vs baseline.
|
||||
return baseline
|
||||
|
||||
if mode == "adversarial" and "INVALID" in he_surface:
|
||||
return ConstraintDecision(
|
||||
|
|
@ -127,12 +131,7 @@ def apply_he_morph_constraint(
|
|||
|
||||
# Executable rule arm
|
||||
if not rule.matches(surface):
|
||||
return ConstraintDecision(
|
||||
kind=DecisionKind.PASS,
|
||||
reason="rule_preconditions_unmet",
|
||||
surfaces=(surface,),
|
||||
rule_id=rule.rule_id,
|
||||
)
|
||||
return baseline
|
||||
|
||||
constraint = rule.to_constraint(surface)
|
||||
# Abstain when proposal asserts exclusive singular / singular-only claim
|
||||
|
|
@ -151,10 +150,5 @@ def apply_he_morph_constraint(
|
|||
surfaces=(surface,),
|
||||
rule_id=rule.rule_id,
|
||||
)
|
||||
return ConstraintDecision(
|
||||
kind=DecisionKind.PASS,
|
||||
reason="rule_matched_no_conflict",
|
||||
constraints=(constraint,),
|
||||
surfaces=(surface,),
|
||||
rule_id=rule.rule_id,
|
||||
)
|
||||
# Rule matched but no singular-exclusivity conflict → same decision as baseline.
|
||||
return baseline
|
||||
|
|
|
|||
|
|
@ -137,6 +137,53 @@ def _proposal_id(candidate: CorrectionCandidate) -> str:
|
|||
return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:16]
|
||||
|
||||
|
||||
# Cached compiled HE morphology for the live teaching consumer (Stage 4).
|
||||
_HE_MORPH_CATALOG: tuple | None = None
|
||||
_HE_MORPH_LOAD_ATTEMPTED = False
|
||||
|
||||
|
||||
def _he_morph_catalog():
|
||||
"""Load compiled HE morphology once; fail closed on missing pack (None)."""
|
||||
global _HE_MORPH_CATALOG, _HE_MORPH_LOAD_ATTEMPTED
|
||||
if _HE_MORPH_LOAD_ATTEMPTED:
|
||||
return _HE_MORPH_CATALOG
|
||||
_HE_MORPH_LOAD_ATTEMPTED = True
|
||||
try:
|
||||
from generate.observed_he_morph_v0.records import load_observed_morphology
|
||||
|
||||
_HE_MORPH_CATALOG = load_observed_morphology("he_logos_micro_v1")
|
||||
except Exception:
|
||||
_HE_MORPH_CATALOG = None
|
||||
return _HE_MORPH_CATALOG
|
||||
|
||||
|
||||
def _auto_he_morph_decision(proposal: PackMutationProposal):
|
||||
"""Apply executable HE morph rule when correction text cites a catalog surface.
|
||||
|
||||
Live production path for Stage 4 — not optional test-only wiring.
|
||||
Returns None when no HE surface is present (no-op, English corrections unchanged).
|
||||
"""
|
||||
catalog = _he_morph_catalog()
|
||||
if not catalog:
|
||||
return None
|
||||
text = proposal.correction_text or ""
|
||||
# Prefer longest surface match present in the correction text.
|
||||
hits = [s for s in catalog if s.surface and s.surface in text]
|
||||
if not hits:
|
||||
return None
|
||||
hits.sort(key=lambda s: len(s.surface), reverse=True)
|
||||
surface = hits[0]
|
||||
from generate.observed_he_morph_v0.consumer import apply_he_morph_constraint
|
||||
|
||||
return apply_he_morph_constraint(
|
||||
proposal_text=text,
|
||||
lemma_key=str(proposal.subject or surface.lemma),
|
||||
observed_catalog=catalog,
|
||||
mode="executable",
|
||||
he_surface=surface.surface,
|
||||
)
|
||||
|
||||
|
||||
class TeachingStore:
|
||||
"""Bounded, append-only store for reviewed teaching examples.
|
||||
|
||||
|
|
@ -204,7 +251,10 @@ class TeachingStore:
|
|||
epistemic_status=example.epistemic_status,
|
||||
)
|
||||
|
||||
# Stage 4 HE morph constraint — abstain/fail-closed → CONTESTED.
|
||||
# Stage 4 HE morph constraint — live path auto-applies executable rule
|
||||
# from compiled packs/data morphology (not tests-only optional wiring).
|
||||
if he_morph_decision is None:
|
||||
he_morph_decision = _auto_he_morph_decision(proposal)
|
||||
if he_morph_decision is not None:
|
||||
kind = getattr(he_morph_decision, "kind", None)
|
||||
kind_val = getattr(kind, "value", kind)
|
||||
|
|
|
|||
|
|
@ -41,9 +41,19 @@ def test_four_arm_ablation_sealed_metrics():
|
|||
|
||||
|
||||
def test_metadata_only_inert_vs_executable_effect():
|
||||
import hashlib
|
||||
import json
|
||||
|
||||
catalog = load_observed_morphology("he_logos_micro_v1")
|
||||
plural = next(s for s in catalog if s.number == "plural")
|
||||
claim = f"{plural.lemma} must be singular only — exclusive singular identity"
|
||||
can = apply_he_morph_constraint(
|
||||
proposal_text=claim,
|
||||
lemma_key=plural.lemma,
|
||||
observed_catalog=catalog,
|
||||
mode="canonical",
|
||||
he_surface=plural.surface,
|
||||
)
|
||||
meta = apply_he_morph_constraint(
|
||||
proposal_text=claim,
|
||||
lemma_key=plural.lemma,
|
||||
|
|
@ -58,6 +68,14 @@ def test_metadata_only_inert_vs_executable_effect():
|
|||
mode="executable",
|
||||
he_surface=plural.surface,
|
||||
)
|
||||
# Bit-identical decision payloads (Stage 4 sealed requirement).
|
||||
d_can = hashlib.sha256(
|
||||
json.dumps(can.as_dict(), sort_keys=True).encode()
|
||||
).hexdigest()
|
||||
d_meta = hashlib.sha256(
|
||||
json.dumps(meta.as_dict(), sort_keys=True).encode()
|
||||
).hexdigest()
|
||||
assert d_can == d_meta
|
||||
assert meta.kind is DecisionKind.PASS
|
||||
assert exe.kind is DecisionKind.ABSTAIN
|
||||
assert exe.rule_id == PLURAL_ABSTAIN_RULE_V0.rule_id
|
||||
|
|
@ -78,12 +96,11 @@ def test_oov_and_invalid_fail_closed():
|
|||
|
||||
|
||||
def test_teaching_store_consumer_seam_abstains_on_plural_rule():
|
||||
"""Live TeachingStore.add path: HE plural rule forces CONTESTED (abstain)."""
|
||||
"""Live TeachingStore.add path: HE plural rule auto-applies (no test-only kwarg)."""
|
||||
catalog = load_observed_morphology("he_logos_micro_v1")
|
||||
plural = next(s for s in catalog if s.number == "plural")
|
||||
store = TeachingStore(capacity=8)
|
||||
|
||||
# First proposal accepted
|
||||
cand1 = CorrectionCandidate(
|
||||
correction_text=f"{plural.lemma} is a logos utterance",
|
||||
intent=DialogueIntent(tag=IntentTag.CORRECTION, subject=plural.lemma),
|
||||
|
|
@ -100,18 +117,13 @@ def test_teaching_store_consumer_seam_abstains_on_plural_rule():
|
|||
p1 = store.add(rev1)
|
||||
assert p1 is not None
|
||||
|
||||
# Second proposal: singular exclusivity — HE morph consumer abstains
|
||||
decision = apply_he_morph_constraint(
|
||||
proposal_text=f"{plural.lemma} must be singular only — exclusive singular identity",
|
||||
lemma_key=plural.lemma,
|
||||
observed_catalog=catalog,
|
||||
mode="executable",
|
||||
he_surface=plural.surface,
|
||||
)
|
||||
assert decision.kind is DecisionKind.ABSTAIN
|
||||
# Consumer integration: when ABSTAIN, teaching path marks CONTESTED
|
||||
# Correction text includes the observed HE plural surface so the live
|
||||
# auto-path in TeachingStore.add finds the catalog row and abstains.
|
||||
cand2 = CorrectionCandidate(
|
||||
correction_text=f"{plural.lemma} must be singular only — exclusive singular identity",
|
||||
correction_text=(
|
||||
f"{plural.surface} ({plural.lemma}) must be singular only — "
|
||||
f"exclusive singular identity"
|
||||
),
|
||||
intent=DialogueIntent(tag=IntentTag.CORRECTION, subject=plural.lemma),
|
||||
prior_surface="prior",
|
||||
prior_turn=1,
|
||||
|
|
@ -123,6 +135,24 @@ def test_teaching_store_consumer_seam_abstains_on_plural_rule():
|
|||
review_hash="h2",
|
||||
epistemic_status=EpistemicStatus.SPECULATIVE,
|
||||
)
|
||||
p2 = store.add(rev2, he_morph_decision=decision)
|
||||
# No he_morph_decision kwarg — production auto-path must fire.
|
||||
p2 = store.add(rev2)
|
||||
assert p2 is not None
|
||||
assert p2.epistemic_status is EpistemicStatus.CONTESTED
|
||||
|
||||
|
||||
def test_vault_promotion_default_requires_geometric_unitarity():
|
||||
"""Production VaultPromotionPolicy() uses 1e-6 residual, not soft 0.05."""
|
||||
from core.physics.learning import VaultPromotionPolicy
|
||||
from core.physics.energy import EnergyClass, EnergyProfile
|
||||
|
||||
policy = VaultPromotionPolicy() # production default
|
||||
assert policy.residual_threshold <= 1e-6
|
||||
soft = EnergyProfile(
|
||||
raw=0.05, energy_class=EnergyClass.E0, coherence_residual=0.02
|
||||
)
|
||||
assert policy.decide(soft).promote is False
|
||||
tight = EnergyProfile(
|
||||
raw=0.05, energy_class=EnergyClass.E0, coherence_residual=1e-9
|
||||
)
|
||||
assert policy.decide(tight).promote is True
|
||||
|
|
|
|||
|
|
@ -62,15 +62,35 @@ def test_inductive_closure_derives_two_hop():
|
|||
("a", "is", "b"),
|
||||
("b", "is", "c"),
|
||||
)
|
||||
res = expand_relation_closure(triples, budget=8)
|
||||
# Explicit geometric_admissible always-true for pure graph composition tests.
|
||||
res = expand_relation_closure(
|
||||
triples, budget=8, geometric_admissible=lambda h, r, t: True
|
||||
)
|
||||
assert len(res.base) == 2
|
||||
derived_tails = {(d.head, d.relation, d.tail) for d in res.derived}
|
||||
assert ("a", "is", "c") in derived_tails
|
||||
assert res.fixed_point is True
|
||||
assert res.steps_taken >= 1
|
||||
# Provenance path
|
||||
a_to_c = next(d for d in res.derived if d.head == "a" and d.tail == "c")
|
||||
assert a_to_c.path[0] == "a" and a_to_c.path[-1] == "c"
|
||||
assert a_to_c.admissible is True
|
||||
|
||||
|
||||
def test_inductive_derived_requires_geometric_admissibility():
|
||||
triples = (
|
||||
("a", "is", "b"),
|
||||
("b", "is", "c"),
|
||||
)
|
||||
|
||||
def refuse_all(h, r, t):
|
||||
del h, r, t
|
||||
return False
|
||||
|
||||
res = expand_relation_closure(
|
||||
triples, budget=8, geometric_admissible=refuse_all
|
||||
)
|
||||
assert any(d.head == "a" and d.tail == "c" for d in res.derived)
|
||||
assert all(d.admissible is False for d in res.derived)
|
||||
|
||||
|
||||
def test_inductive_closure_detects_contradiction():
|
||||
|
|
@ -78,7 +98,9 @@ def test_inductive_closure_detects_contradiction():
|
|||
("a", "is", "b"),
|
||||
("a", "is", "c"),
|
||||
)
|
||||
res = expand_relation_closure(triples, budget=4)
|
||||
res = expand_relation_closure(
|
||||
triples, budget=4, geometric_admissible=lambda h, r, t: True
|
||||
)
|
||||
assert any(c.contradiction for c in res.contradictions)
|
||||
assert len(res.contradictions) >= 2
|
||||
|
||||
|
|
@ -86,10 +108,10 @@ def test_inductive_closure_detects_contradiction():
|
|||
def test_inductive_closure_budget_truncation():
|
||||
# Long chain: a0->a1->...->a20
|
||||
triples = tuple((f"a{i}", "r", f"a{i+1}") for i in range(20))
|
||||
res = expand_relation_closure(triples, budget=2)
|
||||
ok = lambda h, r, t: True # noqa: E731
|
||||
res = expand_relation_closure(triples, budget=2, geometric_admissible=ok)
|
||||
assert res.truncated or res.steps_taken <= 2
|
||||
# With larger budget, multi-hop appears
|
||||
res2 = expand_relation_closure(triples, budget=16)
|
||||
res2 = expand_relation_closure(triples, budget=16, geometric_admissible=ok)
|
||||
assert any(d.head == "a0" and d.tail == "a2" for d in res2.derived) or any(
|
||||
d.head == "a0" for d in res2.derived
|
||||
)
|
||||
|
|
@ -100,7 +122,9 @@ def test_inductive_closure_cycle_safe():
|
|||
("a", "r", "b"),
|
||||
("b", "r", "a"),
|
||||
)
|
||||
res = expand_relation_closure(triples, budget=8)
|
||||
res = expand_relation_closure(
|
||||
triples, budget=8, geometric_admissible=lambda h, r, t: True
|
||||
)
|
||||
# Must terminate without inventing infinite chain
|
||||
assert res.fixed_point or res.steps_taken <= 8
|
||||
assert all(len(d.path) < 20 for d in res.derived)
|
||||
|
|
|
|||
Loading…
Reference in a new issue