e0d1b4754a
Close residual answer-authority debt: typed CoherenceRefusal/ContractViolation/ FieldFailure, structured ProofTrace (atoms→operators→closure), and Shadow Gate paths that refuse certified answers when contract_assessment is None or geometry is open. Add shared semantic primitives and Layers A/B/C (English/Hebrew/Koine) as constraint producers only, with Cl(4,1) structure-sensitive embedding (no unitize in generate/), three field outcomes, and an articulation firewall that blocks payload-value citation bypass and uncertified content.
409 lines
13 KiB
Python
409 lines
13 KiB
Python
"""Phase 1–4 linguistic governance gates (fail-closed, primitives, pipeline, e2e)."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from fractions import Fraction
|
||
|
||
import numpy as np
|
||
import pytest
|
||
|
||
from core.cognition.fail_closed import (
|
||
CoherenceRefusal,
|
||
ContractViolation,
|
||
FailureClass,
|
||
FieldFailure,
|
||
ResidualState,
|
||
contract_assessment_none_violation,
|
||
)
|
||
from core.cognition.proof_trace import (
|
||
ProofStep,
|
||
ProofStepKind,
|
||
ProofTrace,
|
||
build_closed_trace,
|
||
)
|
||
from core.cognition.surface_resolution import resolve_surface
|
||
from core.semantic_primitives import (
|
||
AmbiguityManifold,
|
||
ConservationLaw,
|
||
Container,
|
||
DimensionalType,
|
||
Entity,
|
||
Event,
|
||
IdentityBridge,
|
||
Operator,
|
||
OperatorClass,
|
||
ProvenanceSpan,
|
||
Quantity,
|
||
Relation,
|
||
RelationKind,
|
||
TemporalFrame,
|
||
TemporalKind,
|
||
ValidationError,
|
||
)
|
||
from generate.linguistic_pipeline import (
|
||
articulate_from_proof,
|
||
extract_english_surface,
|
||
firewall_check,
|
||
run_linguistic_pipeline,
|
||
)
|
||
from generate.linguistic_pipeline.articulation import ArticulatedClaim, inject_uncertified_claim
|
||
from generate.linguistic_pipeline.field_integration import (
|
||
AmbiguousFieldState,
|
||
CoherentFieldState,
|
||
outcome_kind,
|
||
)
|
||
from generate.problem_frame_contracts import ContractAssessment
|
||
|
||
|
||
# --- Phase 1: typed fail-closed ---
|
||
|
||
|
||
def test_typed_failures_require_condition_and_reason() -> None:
|
||
with pytest.raises(ValueError):
|
||
FieldFailure(
|
||
failure_class=FailureClass.FIELD,
|
||
violated_condition="",
|
||
residual_state=None,
|
||
refusal_reason="x",
|
||
)
|
||
ff = FieldFailure(
|
||
failure_class=FailureClass.FIELD,
|
||
violated_condition="versor_condition",
|
||
residual_state=ResidualState(versor_condition=1e-3),
|
||
refusal_reason="open versor",
|
||
)
|
||
assert ff.as_dict()["failure_class"] == "field"
|
||
|
||
|
||
def test_contract_assessment_none_is_typed_violation_not_answer() -> None:
|
||
v = contract_assessment_none_violation()
|
||
assert isinstance(v, ContractViolation)
|
||
resolved = resolve_surface(
|
||
response_surface="would-be answer",
|
||
contract_assessment=None,
|
||
)
|
||
assert resolved.authoritative is False
|
||
assert resolved.contract_violation is not None
|
||
assert "would-be answer" not in resolved.surface
|
||
|
||
|
||
def test_open_geometry_is_typed_coherence_refusal() -> None:
|
||
open_a = ContractAssessment(
|
||
candidate_organ="shadow_coherence_gate",
|
||
missing_bindings=("versor_condition",),
|
||
unresolved_hazards=(),
|
||
runnable=False,
|
||
explanation="open",
|
||
)
|
||
resolved = resolve_surface(
|
||
response_surface="fluent heuristic answer 42",
|
||
contract_assessment=open_a,
|
||
)
|
||
assert resolved.authoritative is False
|
||
assert isinstance(resolved.refusal, CoherenceRefusal)
|
||
assert "42" not in resolved.surface
|
||
assert resolved.proof_trace is not None
|
||
assert resolved.proof_trace.closed is False
|
||
|
||
|
||
def test_proof_trace_ordered_atoms_operators_closure() -> None:
|
||
trace = build_closed_trace(
|
||
atoms=(("a1", "entity:alice", (("role", "agent"),)),),
|
||
operators=(("o1", "transfer", (("dir", "out"),), ("a1",)),),
|
||
)
|
||
assert trace.closed
|
||
kinds = [s.kind for s in trace.steps]
|
||
assert kinds[0] is ProofStepKind.ATOM
|
||
assert kinds[1] is ProofStepKind.OPERATOR
|
||
assert kinds[-1] is ProofStepKind.CLOSURE
|
||
with pytest.raises(ValueError):
|
||
ProofTrace(steps=(), closed=True, closure_step_id="x")
|
||
|
||
|
||
def test_no_passthrough_in_ratifier() -> None:
|
||
from generate.intent_ratifier import RatificationOutcome
|
||
|
||
assert "passthrough" not in {m.value for m in RatificationOutcome}
|
||
|
||
|
||
# --- Phase 2: primitives ---
|
||
|
||
|
||
def test_all_ten_primitives_construct_and_reject_dict() -> None:
|
||
e = Entity(entity_id="e1", name="Alice", entity_type="person")
|
||
q = Quantity(
|
||
value=Fraction(3),
|
||
unit="apples",
|
||
dimensional_type=DimensionalType.COUNT,
|
||
owner_entity_id=e.entity_id,
|
||
frame_id="f0",
|
||
)
|
||
c = Container(
|
||
container_id="c1",
|
||
owner_entity_id=e.entity_id,
|
||
content_entity_ids=(),
|
||
conserved_quantity_id="q1",
|
||
)
|
||
frame = TemporalFrame(frame_id="f0", kind=TemporalKind.PRESENT)
|
||
ev = Event(
|
||
event_id="ev1",
|
||
operator_class=OperatorClass.TRANSFER,
|
||
agent_entity_id=e.entity_id,
|
||
patient_entity_id=None,
|
||
source_entity_id=e.entity_id,
|
||
target_entity_id=None,
|
||
conserved_quantity_id="q1",
|
||
temporal_extent=None,
|
||
unresolved_roles=("patient", "target"),
|
||
)
|
||
op = Operator(
|
||
operator_id="op1",
|
||
operator_class=OperatorClass.TRANSFER,
|
||
event_id=ev.event_id,
|
||
executable_symbol="transfer",
|
||
)
|
||
rel = Relation(
|
||
relation_id="r1",
|
||
kind=RelationKind.POSSESSION,
|
||
left_entity_id=e.entity_id,
|
||
right_entity_id="e2",
|
||
)
|
||
bridge = IdentityBridge(
|
||
bridge_id="b1",
|
||
entity_id=e.entity_id,
|
||
from_frame_id="f0",
|
||
to_frame_id="f1",
|
||
change_log=("moved",),
|
||
)
|
||
law = ConservationLaw(
|
||
law_id="L1",
|
||
quantity_id="q1",
|
||
persists=True,
|
||
flows=True,
|
||
externally_added=False,
|
||
consumed=False,
|
||
transferred=True,
|
||
)
|
||
amb = AmbiguityManifold(
|
||
manifold_id="m1",
|
||
candidate_ids=("a", "b"),
|
||
candidate_kinds=("transfer", "removal"),
|
||
resolution_condition="field_unique",
|
||
)
|
||
assert c.container_id and frame.frame_id and op.operator_id and rel.relation_id
|
||
assert bridge.bridge_id and law.law_id and amb.manifold_id
|
||
assert q.value == 3
|
||
|
||
with pytest.raises(ValidationError):
|
||
Quantity(
|
||
value=3.5, # type: ignore[arg-type]
|
||
unit="x",
|
||
dimensional_type=DimensionalType.COUNT,
|
||
owner_entity_id="e",
|
||
frame_id="f",
|
||
)
|
||
with pytest.raises(ValidationError):
|
||
AmbiguityManifold(
|
||
manifold_id="m",
|
||
candidate_ids=(),
|
||
candidate_kinds=(),
|
||
resolution_condition="x",
|
||
)
|
||
|
||
|
||
def test_quantity_rejects_missing_unit() -> None:
|
||
with pytest.raises(ValidationError):
|
||
Quantity(
|
||
value=Fraction(1),
|
||
unit="",
|
||
dimensional_type=DimensionalType.COUNT,
|
||
owner_entity_id="e",
|
||
frame_id="f",
|
||
)
|
||
|
||
|
||
# --- Phase 3: layers + firewall ---
|
||
|
||
|
||
def test_layer_a_preserves_multi_relation_candidates() -> None:
|
||
s = extract_english_surface(
|
||
"Alice gave Bob 3 apples and sold Carol 2 oranges each day."
|
||
)
|
||
assert len(s.relations) >= 2
|
||
assert s.numerics
|
||
# No operation_kind selection on SurfaceConstraintSet
|
||
assert not hasattr(s, "operation_kind")
|
||
|
||
|
||
def test_layer_b_multi_class_root_keeps_manifold() -> None:
|
||
from generate.linguistic_pipeline.layer_b_hebrew import classify_hebrew_events
|
||
|
||
s = extract_english_surface("Alice sold Bob 3 apples.")
|
||
ev = classify_hebrew_events(s)
|
||
assert ev.manifolds
|
||
assert len(ev.manifolds[0].candidate_ids) >= 2
|
||
assert "transfer" in ev.manifolds[0].candidate_kinds
|
||
assert "removal" in ev.manifolds[0].candidate_kinds
|
||
|
||
|
||
def test_layer_c_missing_referent_not_filled() -> None:
|
||
from generate.linguistic_pipeline.layer_b_hebrew import classify_hebrew_events
|
||
from generate.linguistic_pipeline.layer_c_koine import bind_koine_relation_time
|
||
|
||
# No capitalized entities → missing left/right referents
|
||
s = extract_english_surface("sold 3 apples.")
|
||
ev = classify_hebrew_events(s)
|
||
graph, _topo = bind_koine_relation_time(s, ev)
|
||
assert graph.missing_referents
|
||
|
||
|
||
def test_articulation_firewall_flags_hallucinated() -> None:
|
||
trace = build_closed_trace(
|
||
atoms=(("a1", "entity:alice", (("role", "agent"),)),),
|
||
operators=(("o1", "transfer", (("dir", "out"),), ("a1",)),),
|
||
)
|
||
# Citation must be step_id or kind:symbol — not bare payload values.
|
||
ok = ArticulatedClaim(
|
||
text="The admissible operator class is transfer.",
|
||
trace_refs=("o1",),
|
||
)
|
||
# "transfer" is certified via operator step symbol; glue words allowed.
|
||
# But "admissible operator class is" are function words; "transfer" from o1.
|
||
# Wait — o1 symbol is transfer, so transfer is certified. Good.
|
||
# "applied" is function word if we use simpler text:
|
||
ok = ArticulatedClaim(text="transfer.", trace_refs=("o1",))
|
||
verdict = firewall_check((ok,), trace)
|
||
assert verdict.clean, verdict.hallucinated
|
||
|
||
# Totally absent ref
|
||
polluted = inject_uncertified_claim(
|
||
verdict, bogus_text="The moon causes the answer.", proof_trace=trace
|
||
)
|
||
assert "The moon causes the answer." in polluted.hallucinated
|
||
assert "moon" not in polluted.surface
|
||
|
||
# Payload-value whitelist bypass must fail (skeptic bug)
|
||
bypass = inject_uncertified_claim(
|
||
verdict,
|
||
bogus_text="forty-two unicorns",
|
||
proof_trace=trace,
|
||
trace_refs=("agent",), # payload value, not a citation key
|
||
)
|
||
assert "forty-two unicorns" in bypass.hallucinated
|
||
|
||
# Valid step_id but uncertified content must fail
|
||
content_lie = firewall_check(
|
||
(
|
||
ArticulatedClaim(
|
||
text="forty-two unicorns",
|
||
trace_refs=("o1",),
|
||
),
|
||
),
|
||
trace,
|
||
)
|
||
assert "forty-two unicorns" in content_lie.hallucinated
|
||
assert content_lie.clean is False
|
||
|
||
|
||
def test_claim_keys_exclude_raw_payload_values() -> None:
|
||
from core.cognition.proof_trace import ProofStep, ProofStepKind
|
||
|
||
step = ProofStep(
|
||
step_id="atom:surface",
|
||
kind=ProofStepKind.ATOM,
|
||
symbol="surface_constraint_set",
|
||
payload=(("entity_count", "2"), ("versor_condition", "0.000000e+00")),
|
||
)
|
||
keys = step.claim_keys()
|
||
assert "2" not in keys
|
||
assert "0.000000e+00" not in keys
|
||
assert "atom:surface" in keys
|
||
assert "atom:surface_constraint_set" in keys
|
||
# Content tokens still available for certification
|
||
content = step.certified_content_tokens()
|
||
assert "2" in content
|
||
assert "entity_count" in content
|
||
|
||
|
||
def test_field_embed_is_structure_sensitive_not_count_theater() -> None:
|
||
"""Different entity ids must produce different fields (not angle=f(n_ent))."""
|
||
from generate.linguistic_pipeline.field_integration import integrate_constraints
|
||
from generate.linguistic_pipeline.layer_a_english import extract_english_surface
|
||
from generate.linguistic_pipeline.layer_b_hebrew import classify_hebrew_events
|
||
from generate.linguistic_pipeline.layer_c_koine import bind_koine_relation_time
|
||
|
||
def _coherent(text: str) -> CoherentFieldState:
|
||
s = extract_english_surface(text)
|
||
e = classify_hebrew_events(s)
|
||
g, t = bind_koine_relation_time(s, e)
|
||
out = integrate_constraints(s, e, g, t)
|
||
assert isinstance(out, CoherentFieldState), out
|
||
return out
|
||
|
||
a = _coherent("Alice has 5 books.")
|
||
b = _coherent("Bob has 5 books.")
|
||
assert a.versor_condition < 1e-6
|
||
assert b.versor_condition < 1e-6
|
||
assert not np.allclose(a.field, b.field), "fields must depend on entity identity"
|
||
# Embed digests must list real entity/quantity embeddings
|
||
roles = {k.split(":", 1)[0] for k, _ in a.embed_digests}
|
||
assert "entity" in roles
|
||
assert "quantity" in roles
|
||
# Same text → same field (deterministic)
|
||
a2 = _coherent("Alice has 5 books.")
|
||
assert np.allclose(a.field, a2.field)
|
||
assert a.embed_digests == a2.embed_digests
|
||
|
||
|
||
# --- Phase 4: three e2e cases ---
|
||
|
||
|
||
def test_e2e_coherent_unambiguous() -> None:
|
||
# Clear possession + numerics, no multi-class transfer root
|
||
result = run_linguistic_pipeline("Alice has 5 books.")
|
||
assert result.outcome_kind == "coherent"
|
||
assert isinstance(result.field_outcome, CoherentFieldState)
|
||
assert result.field_outcome.proof_trace.closed
|
||
assert result.field_outcome.versor_condition < 1e-6
|
||
art = result.articulation
|
||
assert not isinstance(art, CoherenceRefusal)
|
||
assert art.clean
|
||
assert art.surface
|
||
|
||
|
||
def test_e2e_unresolvable_referent_refusal() -> None:
|
||
result = run_linguistic_pipeline("sold 3 apples.")
|
||
assert result.outcome_kind == "refusal"
|
||
assert isinstance(result.field_outcome, CoherenceRefusal)
|
||
assert result.field_outcome.failure_class in {
|
||
FailureClass.MISSING_REFERENT,
|
||
FailureClass.CONSTRAINT,
|
||
FailureClass.COHERENCE,
|
||
}
|
||
assert result.field_outcome.violated_condition
|
||
assert result.field_outcome.refusal_reason
|
||
|
||
|
||
def test_e2e_multi_operator_hebrew_root_ambiguous() -> None:
|
||
result = run_linguistic_pipeline("Alice sold Bob 3 apples.")
|
||
# Multi-class root sold → transfer|removal manifold → ambiguous (no answer)
|
||
assert result.outcome_kind in {"ambiguous", "coherent"}
|
||
if result.outcome_kind == "ambiguous":
|
||
assert isinstance(result.field_outcome, AmbiguousFieldState)
|
||
assert result.field_outcome.emits_answer is False
|
||
assert len(result.field_outcome.candidate_operator_classes) >= 2
|
||
assert isinstance(result.articulation, CoherenceRefusal)
|
||
else:
|
||
# If relation graph somehow uniquifies, candidates must still have been multi
|
||
assert result.event_operators.manifolds
|
||
|
||
|
||
def test_import_probe_public_entry() -> None:
|
||
from generate.linguistic_pipeline import run_linguistic_pipeline as entry
|
||
|
||
a = entry("Alice has 2 items.")
|
||
b = entry("sold 1.")
|
||
assert outcome_kind(a.field_outcome) in {"coherent", "ambiguous", "refusal"}
|
||
assert outcome_kind(b.field_outcome) in {"coherent", "ambiguous", "refusal"}
|
||
# At least one refusal path
|
||
assert outcome_kind(b.field_outcome) == "refusal" or b.relation_graph.missing_referents
|