Merge pull request 'feat(cognition): fail-closed linguistic governance + trilingual constraint pipeline' (#96) from feat/linguistic-governance-fail-closed-ir into main
Reviewed-on: #96
This commit is contained in:
commit
25504c9d0b
14 changed files with 2852 additions and 19 deletions
256
core/cognition/fail_closed.py
Normal file
256
core/cognition/fail_closed.py
Normal file
|
|
@ -0,0 +1,256 @@
|
|||
"""Typed fail-closed outcomes for geometry / coherence / contract gates.
|
||||
|
||||
These types are the only admissible non-answer results on answer-authority
|
||||
paths. Silent defaults and heuristic fills are forbidden when the field
|
||||
cannot close.
|
||||
|
||||
Fields required by linguistic governance Phase 1:
|
||||
failure_class, violated_condition, residual_state, refusal_reason
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
from typing import Any, Mapping
|
||||
|
||||
|
||||
class FailureClass(str, Enum):
|
||||
"""Machine-stable failure taxonomy for answer-authority paths."""
|
||||
|
||||
COHERENCE = "coherence"
|
||||
CONTRACT = "contract"
|
||||
FIELD = "field"
|
||||
CONSTRAINT = "constraint"
|
||||
MISSING_REFERENT = "missing_referent"
|
||||
AMBIGUITY = "ambiguity"
|
||||
ARTICULATION = "articulation"
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ResidualState:
|
||||
"""Optional residual snapshot when a gate refuses.
|
||||
|
||||
Coordinates are never required; digests / scalar residuals preferred.
|
||||
"""
|
||||
|
||||
versor_condition: float | None = None
|
||||
goldtether_residual: float | None = None
|
||||
missing_bindings: tuple[str, ...] = ()
|
||||
unresolved_hazards: tuple[str, ...] = ()
|
||||
detail: str = ""
|
||||
|
||||
def as_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"versor_condition": self.versor_condition,
|
||||
"goldtether_residual": self.goldtether_residual,
|
||||
"missing_bindings": list(self.missing_bindings),
|
||||
"unresolved_hazards": list(self.unresolved_hazards),
|
||||
"detail": self.detail,
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class FieldFailure:
|
||||
"""Geometry / field construction could not produce a closed state."""
|
||||
|
||||
failure_class: FailureClass
|
||||
violated_condition: str
|
||||
residual_state: ResidualState | None
|
||||
refusal_reason: str
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if not self.violated_condition:
|
||||
raise ValueError("FieldFailure.violated_condition must be non-empty")
|
||||
if not self.refusal_reason:
|
||||
raise ValueError("FieldFailure.refusal_reason must be non-empty")
|
||||
if not isinstance(self.failure_class, FailureClass):
|
||||
raise TypeError("FieldFailure.failure_class must be a FailureClass")
|
||||
|
||||
def as_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"kind": "field_failure",
|
||||
"failure_class": self.failure_class.value,
|
||||
"violated_condition": self.violated_condition,
|
||||
"residual_state": None
|
||||
if self.residual_state is None
|
||||
else self.residual_state.as_dict(),
|
||||
"refusal_reason": self.refusal_reason,
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class CoherenceRefusal:
|
||||
"""No admissible configuration — typed abstention, not a default answer."""
|
||||
|
||||
failure_class: FailureClass
|
||||
violated_condition: str
|
||||
residual_state: ResidualState | None
|
||||
refusal_reason: str
|
||||
surface_message: str = ""
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if not self.violated_condition:
|
||||
raise ValueError("CoherenceRefusal.violated_condition must be non-empty")
|
||||
if not self.refusal_reason:
|
||||
raise ValueError("CoherenceRefusal.refusal_reason must be non-empty")
|
||||
if not isinstance(self.failure_class, FailureClass):
|
||||
raise TypeError("CoherenceRefusal.failure_class must be a FailureClass")
|
||||
|
||||
@property
|
||||
def message(self) -> str:
|
||||
if self.surface_message:
|
||||
return self.surface_message
|
||||
return (
|
||||
f"Abstaining: {self.refusal_reason} "
|
||||
f"(condition={self.violated_condition})."
|
||||
)
|
||||
|
||||
def as_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"kind": "coherence_refusal",
|
||||
"failure_class": self.failure_class.value,
|
||||
"violated_condition": self.violated_condition,
|
||||
"residual_state": None
|
||||
if self.residual_state is None
|
||||
else self.residual_state.as_dict(),
|
||||
"refusal_reason": self.refusal_reason,
|
||||
"surface_message": self.surface_message,
|
||||
"message": self.message,
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ContractViolation:
|
||||
"""Contract assessment missing or structurally incomplete — never silent pass."""
|
||||
|
||||
failure_class: FailureClass
|
||||
violated_condition: str
|
||||
residual_state: ResidualState | None
|
||||
refusal_reason: str
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if self.failure_class is not FailureClass.CONTRACT:
|
||||
raise ValueError("ContractViolation.failure_class must be CONTRACT")
|
||||
if not self.violated_condition:
|
||||
raise ValueError("ContractViolation.violated_condition must be non-empty")
|
||||
if not self.refusal_reason:
|
||||
raise ValueError("ContractViolation.refusal_reason must be non-empty")
|
||||
|
||||
def as_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"kind": "contract_violation",
|
||||
"failure_class": self.failure_class.value,
|
||||
"violated_condition": self.violated_condition,
|
||||
"residual_state": None
|
||||
if self.residual_state is None
|
||||
else self.residual_state.as_dict(),
|
||||
"refusal_reason": self.refusal_reason,
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ConstraintViolation:
|
||||
"""A typed semantic constraint could not be satisfied."""
|
||||
|
||||
failure_class: FailureClass
|
||||
violated_condition: str
|
||||
residual_state: ResidualState | None
|
||||
refusal_reason: str
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if not self.violated_condition:
|
||||
raise ValueError("ConstraintViolation.violated_condition must be non-empty")
|
||||
if not self.refusal_reason:
|
||||
raise ValueError("ConstraintViolation.refusal_reason must be non-empty")
|
||||
|
||||
def as_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"kind": "constraint_violation",
|
||||
"failure_class": self.failure_class.value,
|
||||
"violated_condition": self.violated_condition,
|
||||
"residual_state": None
|
||||
if self.residual_state is None
|
||||
else self.residual_state.as_dict(),
|
||||
"refusal_reason": self.refusal_reason,
|
||||
}
|
||||
|
||||
|
||||
def contract_assessment_none_violation() -> ContractViolation:
|
||||
"""Canonical violation when assessment is absent on an answer-authority path."""
|
||||
return ContractViolation(
|
||||
failure_class=FailureClass.CONTRACT,
|
||||
violated_condition="contract_assessment_present",
|
||||
residual_state=ResidualState(
|
||||
detail="contract_assessment is None — geometric conjugate cannot pass"
|
||||
),
|
||||
refusal_reason=(
|
||||
"contract_assessment is None; substrate and certified answers are refused"
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def open_geometry_refusal(
|
||||
*,
|
||||
missing_bindings: tuple[str, ...] = (),
|
||||
unresolved_hazards: tuple[str, ...] = (),
|
||||
explanation: str = "",
|
||||
versor_condition: float | None = None,
|
||||
goldtether_residual: float | None = None,
|
||||
) -> CoherenceRefusal:
|
||||
"""Typed abstention when versor / GoldTether / binding contract is open."""
|
||||
conditions: list[str] = []
|
||||
if missing_bindings:
|
||||
conditions.extend(missing_bindings)
|
||||
if unresolved_hazards:
|
||||
conditions.extend(unresolved_hazards)
|
||||
violated = "|".join(conditions) if conditions else "geometric_contract_closed"
|
||||
return CoherenceRefusal(
|
||||
failure_class=FailureClass.COHERENCE,
|
||||
violated_condition=violated,
|
||||
residual_state=ResidualState(
|
||||
versor_condition=versor_condition,
|
||||
goldtether_residual=goldtether_residual,
|
||||
missing_bindings=missing_bindings,
|
||||
unresolved_hazards=unresolved_hazards,
|
||||
detail=explanation,
|
||||
),
|
||||
refusal_reason=(
|
||||
"field geometric contract is not closed; no certified answer is available"
|
||||
),
|
||||
surface_message=(
|
||||
"I cannot certify an answer: the geometric coherence contract is open "
|
||||
f"({violated})."
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def residual_from_mapping(data: Mapping[str, Any] | None) -> ResidualState | None:
|
||||
if not data:
|
||||
return None
|
||||
return ResidualState(
|
||||
versor_condition=_opt_float(data.get("versor_condition")),
|
||||
goldtether_residual=_opt_float(data.get("goldtether_residual")),
|
||||
missing_bindings=tuple(data.get("missing_bindings") or ()),
|
||||
unresolved_hazards=tuple(data.get("unresolved_hazards") or ()),
|
||||
detail=str(data.get("detail") or ""),
|
||||
)
|
||||
|
||||
|
||||
def _opt_float(value: Any) -> float | None:
|
||||
if value is None:
|
||||
return None
|
||||
return float(value)
|
||||
|
||||
|
||||
__all__ = [
|
||||
"FailureClass",
|
||||
"ResidualState",
|
||||
"FieldFailure",
|
||||
"CoherenceRefusal",
|
||||
"ContractViolation",
|
||||
"ConstraintViolation",
|
||||
"contract_assessment_none_violation",
|
||||
"open_geometry_refusal",
|
||||
"residual_from_mapping",
|
||||
]
|
||||
283
core/cognition/proof_trace.py
Normal file
283
core/cognition/proof_trace.py
Normal file
|
|
@ -0,0 +1,283 @@
|
|||
"""Minimal structured proof trace for proof-preserving articulation.
|
||||
|
||||
Ordered sequence: semantic atoms → field operators → closure result.
|
||||
No free-text-only steps as authoritative proof content.
|
||||
|
||||
Citation rules (firewall):
|
||||
* Valid citation keys are ONLY step_id and ``kind:symbol``.
|
||||
* Raw payload values are never citation keys (prevents whitelist bypass
|
||||
via incidental values like ``"2"`` or ``"0.000000e+00"``).
|
||||
* Payload content is available separately for content certification.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
from typing import Any, Iterable, Sequence
|
||||
|
||||
|
||||
class ProofStepKind(str, Enum):
|
||||
ATOM = "atom"
|
||||
OPERATOR = "operator"
|
||||
CLOSURE = "closure"
|
||||
REFUSAL = "refusal"
|
||||
|
||||
|
||||
_TOKEN_SPLIT = re.compile(r"[^a-z0-9_.:-]+", re.IGNORECASE)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ProofStep:
|
||||
"""One ordered, typed proof step."""
|
||||
|
||||
step_id: str
|
||||
kind: ProofStepKind
|
||||
symbol: str
|
||||
"""Machine id: entity id, operator class, closure predicate, etc."""
|
||||
payload: tuple[tuple[str, str], ...] = ()
|
||||
"""Stringly-serialised structured payload pairs (key, value) only."""
|
||||
parent_ids: tuple[str, ...] = ()
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if not self.step_id:
|
||||
raise ValueError("ProofStep.step_id must be non-empty")
|
||||
if not isinstance(self.kind, ProofStepKind):
|
||||
raise TypeError("ProofStep.kind must be a ProofStepKind")
|
||||
if not self.symbol:
|
||||
raise ValueError("ProofStep.symbol must be non-empty")
|
||||
for key, value in self.payload:
|
||||
if not isinstance(key, str) or not isinstance(value, str):
|
||||
raise TypeError("ProofStep.payload must be tuple[tuple[str, str], ...]")
|
||||
|
||||
def as_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"step_id": self.step_id,
|
||||
"kind": self.kind.value,
|
||||
"symbol": self.symbol,
|
||||
"payload": [[k, v] for k, v in self.payload],
|
||||
"parent_ids": list(self.parent_ids),
|
||||
}
|
||||
|
||||
def citation_keys(self) -> frozenset[str]:
|
||||
"""Keys an articulation claim may *cite* as trace_refs.
|
||||
|
||||
Only step_id and kind:symbol — never bare payload values.
|
||||
"""
|
||||
return frozenset(
|
||||
{
|
||||
self.step_id,
|
||||
f"{self.kind.value}:{self.symbol}",
|
||||
}
|
||||
)
|
||||
|
||||
# Back-compat alias used by older call sites / tests.
|
||||
def claim_keys(self) -> frozenset[str]:
|
||||
return self.citation_keys()
|
||||
|
||||
def certified_content_tokens(self) -> frozenset[str]:
|
||||
"""Tokens that may appear in claim *text* when this step is cited.
|
||||
|
||||
Includes step_id parts, symbol parts, and payload keys/values —
|
||||
but only as content vocabulary, not as citation keys.
|
||||
"""
|
||||
tokens: set[str] = set()
|
||||
for raw in (self.step_id, self.symbol, self.kind.value):
|
||||
tokens |= _tokenize(raw)
|
||||
for k, v in self.payload:
|
||||
tokens |= _tokenize(k)
|
||||
tokens |= _tokenize(v)
|
||||
return frozenset(tokens)
|
||||
|
||||
|
||||
def _tokenize(text: str) -> set[str]:
|
||||
out: set[str] = set()
|
||||
for part in _TOKEN_SPLIT.split(text.lower()):
|
||||
if part:
|
||||
out.add(part)
|
||||
# Also keep dotted/colon segments whole when present.
|
||||
# Whole lowercased string if multi-token machine id
|
||||
if text and " " not in text:
|
||||
out.add(text.lower())
|
||||
return out
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ProofTrace:
|
||||
"""Ordered proof trace. Empty only for non-authoritative turns."""
|
||||
|
||||
steps: tuple[ProofStep, ...] = ()
|
||||
closed: bool = False
|
||||
closure_step_id: str | None = None
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if self.closed and not self.steps:
|
||||
raise ValueError("closed ProofTrace must contain at least one step")
|
||||
if self.closed:
|
||||
if self.closure_step_id is None:
|
||||
raise ValueError("closed ProofTrace requires closure_step_id")
|
||||
ids = {s.step_id for s in self.steps}
|
||||
if self.closure_step_id not in ids:
|
||||
raise ValueError("closure_step_id must reference a step in the trace")
|
||||
closure = next(s for s in self.steps if s.step_id == self.closure_step_id)
|
||||
if closure.kind is not ProofStepKind.CLOSURE:
|
||||
raise ValueError("closure_step_id must point at a CLOSURE step")
|
||||
|
||||
def as_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"closed": self.closed,
|
||||
"closure_step_id": self.closure_step_id,
|
||||
"steps": [s.as_dict() for s in self.steps],
|
||||
}
|
||||
|
||||
def all_citation_keys(self) -> frozenset[str]:
|
||||
keys: set[str] = set()
|
||||
for step in self.steps:
|
||||
keys |= set(step.citation_keys())
|
||||
return frozenset(keys)
|
||||
|
||||
def all_claim_keys(self) -> frozenset[str]:
|
||||
"""Alias for citation keys (not payload-value whitelist)."""
|
||||
return self.all_citation_keys()
|
||||
|
||||
def steps_for_refs(self, refs: Sequence[str]) -> tuple[ProofStep, ...]:
|
||||
"""Resolve citation refs to steps; unknown refs yield empty match list."""
|
||||
by_key: dict[str, list[ProofStep]] = {}
|
||||
for step in self.steps:
|
||||
for key in step.citation_keys():
|
||||
by_key.setdefault(key, []).append(step)
|
||||
found: list[ProofStep] = []
|
||||
seen: set[str] = set()
|
||||
for ref in refs:
|
||||
for step in by_key.get(ref, ()):
|
||||
if step.step_id not in seen:
|
||||
seen.add(step.step_id)
|
||||
found.append(step)
|
||||
return tuple(found)
|
||||
|
||||
def certified_content_for_refs(self, refs: Sequence[str]) -> frozenset[str]:
|
||||
tokens: set[str] = set()
|
||||
for step in self.steps_for_refs(refs):
|
||||
tokens |= set(step.certified_content_tokens())
|
||||
return frozenset(tokens)
|
||||
|
||||
def extend(self, extra: Sequence[ProofStep]) -> "ProofTrace":
|
||||
return ProofTrace(
|
||||
steps=self.steps + tuple(extra),
|
||||
closed=self.closed,
|
||||
closure_step_id=self.closure_step_id,
|
||||
)
|
||||
|
||||
|
||||
def build_closed_trace(
|
||||
atoms: Iterable[tuple[str, str, Sequence[tuple[str, str]]]],
|
||||
operators: Iterable[tuple[str, str, Sequence[tuple[str, str]], Sequence[str]]],
|
||||
*,
|
||||
closure_symbol: str = "versor_and_goldtether_closed",
|
||||
closure_payload: Sequence[tuple[str, str]] = (),
|
||||
) -> ProofTrace:
|
||||
"""Build a closed proof from atom and operator descriptors.
|
||||
|
||||
atoms: (step_id, symbol, payload)
|
||||
operators: (step_id, symbol, payload, parent_ids)
|
||||
"""
|
||||
steps: list[ProofStep] = []
|
||||
for step_id, symbol, payload in atoms:
|
||||
steps.append(
|
||||
ProofStep(
|
||||
step_id=step_id,
|
||||
kind=ProofStepKind.ATOM,
|
||||
symbol=symbol,
|
||||
payload=tuple((str(k), str(v)) for k, v in payload),
|
||||
)
|
||||
)
|
||||
for step_id, symbol, payload, parents in operators:
|
||||
steps.append(
|
||||
ProofStep(
|
||||
step_id=step_id,
|
||||
kind=ProofStepKind.OPERATOR,
|
||||
symbol=symbol,
|
||||
payload=tuple((str(k), str(v)) for k, v in payload),
|
||||
parent_ids=tuple(parents),
|
||||
)
|
||||
)
|
||||
closure_id = "closure:0"
|
||||
parent_ids = tuple(s.step_id for s in steps if s.kind is ProofStepKind.OPERATOR)
|
||||
if not parent_ids:
|
||||
parent_ids = tuple(s.step_id for s in steps)
|
||||
steps.append(
|
||||
ProofStep(
|
||||
step_id=closure_id,
|
||||
kind=ProofStepKind.CLOSURE,
|
||||
symbol=closure_symbol,
|
||||
payload=tuple((str(k), str(v)) for k, v in closure_payload),
|
||||
parent_ids=parent_ids,
|
||||
)
|
||||
)
|
||||
return ProofTrace(steps=tuple(steps), closed=True, closure_step_id=closure_id)
|
||||
|
||||
|
||||
def build_refusal_trace(
|
||||
*,
|
||||
reason: str,
|
||||
violated_condition: str,
|
||||
) -> ProofTrace:
|
||||
"""Trace that certifies only the refusal itself (no answer claims)."""
|
||||
step = ProofStep(
|
||||
step_id="refusal:0",
|
||||
kind=ProofStepKind.REFUSAL,
|
||||
symbol="coherence_refusal",
|
||||
payload=(
|
||||
("reason", reason),
|
||||
("violated_condition", violated_condition),
|
||||
),
|
||||
)
|
||||
return ProofTrace(steps=(step,), closed=False, closure_step_id=None)
|
||||
|
||||
|
||||
def geometry_contract_trace(
|
||||
*,
|
||||
versor_condition: float,
|
||||
goldtether_residual: float,
|
||||
closed: bool,
|
||||
) -> ProofTrace:
|
||||
"""Proof fragment from live shadow-gate scalars."""
|
||||
payload = (
|
||||
("versor_condition", f"{versor_condition:.6e}"),
|
||||
("goldtether_residual", f"{goldtether_residual:.6e}"),
|
||||
)
|
||||
if closed:
|
||||
return build_closed_trace(
|
||||
atoms=(
|
||||
(
|
||||
"atom:field",
|
||||
"field_state",
|
||||
payload,
|
||||
),
|
||||
),
|
||||
operators=(
|
||||
(
|
||||
"op:geometry_gate",
|
||||
"shadow_coherence_gate",
|
||||
payload,
|
||||
("atom:field",),
|
||||
),
|
||||
),
|
||||
closure_symbol="geometric_contract_closed",
|
||||
closure_payload=payload,
|
||||
)
|
||||
return build_refusal_trace(
|
||||
reason="geometric_contract_open",
|
||||
violated_condition="versor_condition_and_goldtether",
|
||||
)
|
||||
|
||||
|
||||
__all__ = [
|
||||
"ProofStepKind",
|
||||
"ProofStep",
|
||||
"ProofTrace",
|
||||
"build_closed_trace",
|
||||
"build_refusal_trace",
|
||||
"geometry_contract_trace",
|
||||
]
|
||||
|
|
@ -8,6 +8,11 @@ The pipeline produces several candidate surfaces in one turn:
|
|||
|
||||
Historically these mutated one string in evaluation order. This module
|
||||
centralizes the policy so fold behavior is declared and unit-testable.
|
||||
|
||||
Phase 1 linguistic governance:
|
||||
* ``contract_assessment is None`` → typed ContractViolation; no certified answer
|
||||
* open geometric contract → typed CoherenceRefusal; no certified answer
|
||||
* walk/compose folds never upgrade an uncertified base into authority
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
|
@ -16,11 +21,24 @@ from dataclasses import dataclass
|
|||
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from core.cognition.fail_closed import (
|
||||
CoherenceRefusal,
|
||||
ContractViolation,
|
||||
FailureClass,
|
||||
ResidualState,
|
||||
contract_assessment_none_violation,
|
||||
open_geometry_refusal,
|
||||
)
|
||||
from core.cognition.proof_trace import ProofTrace, build_refusal_trace, geometry_contract_trace
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from generate.graph_planner import PropositionGraph
|
||||
from generate.problem_frame_contracts import ContractAssessment
|
||||
|
||||
|
||||
_ABSTENTION_AUTHORITY = "coherence_abstention"
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class SurfaceResolution:
|
||||
"""Resolved user-facing and articulation surfaces.
|
||||
|
|
@ -31,14 +49,21 @@ class SurfaceResolution:
|
|||
|
||||
When authority == "substrate_realizer", the PropositionGraph +
|
||||
realize_semantic path was granted supremacy by the Shadow Coherence
|
||||
Gate (strict structural + contract + coherence proof). Legacy runtime
|
||||
and walk/compose folds are still applied after, never before.
|
||||
Gate (strict structural + contract + coherence proof).
|
||||
|
||||
``authoritative`` is True only when a certified answer may leave the
|
||||
system. Geometry-open and assessment-missing paths set this False and
|
||||
attach a typed refusal/violation.
|
||||
"""
|
||||
|
||||
surface: str
|
||||
articulation_surface: str
|
||||
authority: str
|
||||
fold_sources: tuple[str, ...] = ()
|
||||
authoritative: bool = True
|
||||
refusal: CoherenceRefusal | None = None
|
||||
contract_violation: ContractViolation | None = None
|
||||
proof_trace: ProofTrace | None = None
|
||||
|
||||
|
||||
def _base_runtime_surface(
|
||||
|
|
@ -57,6 +82,60 @@ def _base_runtime_surface(
|
|||
return response_surface, response_articulation_surface, "runtime"
|
||||
|
||||
|
||||
def _assessment_residual(
|
||||
contract_assessment: "ContractAssessment | None",
|
||||
) -> ResidualState | None:
|
||||
if contract_assessment is None:
|
||||
return ResidualState(detail="contract_assessment is None")
|
||||
return ResidualState(
|
||||
missing_bindings=tuple(contract_assessment.missing_bindings),
|
||||
unresolved_hazards=tuple(contract_assessment.unresolved_hazards),
|
||||
detail=str(contract_assessment.explanation or ""),
|
||||
)
|
||||
|
||||
|
||||
def _abstention_resolution(
|
||||
*,
|
||||
refusal: CoherenceRefusal | None,
|
||||
violation: ContractViolation | None,
|
||||
) -> SurfaceResolution:
|
||||
if violation is not None:
|
||||
msg = (
|
||||
f"I cannot certify an answer: {violation.refusal_reason} "
|
||||
f"(condition={violation.violated_condition})."
|
||||
)
|
||||
trace = build_refusal_trace(
|
||||
reason=violation.refusal_reason,
|
||||
violated_condition=violation.violated_condition,
|
||||
)
|
||||
return SurfaceResolution(
|
||||
surface=msg,
|
||||
articulation_surface=msg,
|
||||
authority=_ABSTENTION_AUTHORITY,
|
||||
fold_sources=(),
|
||||
authoritative=False,
|
||||
refusal=None,
|
||||
contract_violation=violation,
|
||||
proof_trace=trace,
|
||||
)
|
||||
assert refusal is not None
|
||||
msg = refusal.message
|
||||
trace = build_refusal_trace(
|
||||
reason=refusal.refusal_reason,
|
||||
violated_condition=refusal.violated_condition,
|
||||
)
|
||||
return SurfaceResolution(
|
||||
surface=msg,
|
||||
articulation_surface=msg,
|
||||
authority=_ABSTENTION_AUTHORITY,
|
||||
fold_sources=(),
|
||||
authoritative=False,
|
||||
refusal=refusal,
|
||||
contract_violation=None,
|
||||
proof_trace=trace,
|
||||
)
|
||||
|
||||
|
||||
def resolve_surface(
|
||||
*,
|
||||
canonical_surface: str = "",
|
||||
|
|
@ -70,6 +149,7 @@ def resolve_surface(
|
|||
compose_surface: str = "",
|
||||
proposition_graph: "PropositionGraph | None" = None,
|
||||
contract_assessment: "ContractAssessment | None" = None,
|
||||
require_closed_geometry: bool = True,
|
||||
) -> SurfaceResolution:
|
||||
"""Resolve the final turn surface under dual-competing Shadow Coherence Gate.
|
||||
|
||||
|
|
@ -84,14 +164,38 @@ def resolve_surface(
|
|||
``contract_assessment``. Assessment is **required** for substrate
|
||||
commit; ``None`` refuses geometric authority (fail-closed).
|
||||
|
||||
When either competitor fails, authority stays on the runtime base surface.
|
||||
The transitional ``realizer_useful`` shim is admitted only when conjugate
|
||||
coherence still passes (never as a substitute for a failed geometric gate).
|
||||
|
||||
Walk/compose folds are *always* suffixes — they never affect the
|
||||
authority prefix decision.
|
||||
When ``require_closed_geometry`` is True (default), a missing or open
|
||||
geometric contract yields a typed abstention — no runtime fluent answer
|
||||
is emitted as certified content. Walk/compose folds are suppressed on
|
||||
abstention paths.
|
||||
"""
|
||||
|
||||
# --- Fail-closed: missing assessment never silently passes ---
|
||||
if require_closed_geometry and contract_assessment is None:
|
||||
return _abstention_resolution(
|
||||
refusal=None,
|
||||
violation=contract_assessment_none_violation(),
|
||||
)
|
||||
|
||||
conjugate_ok = _conjugate_coherence_ok(contract_assessment)
|
||||
if require_closed_geometry and not conjugate_ok:
|
||||
if contract_assessment is None:
|
||||
# Unreachable when require_closed_geometry handled None above,
|
||||
# but keep branch explicit for non-require callers.
|
||||
return _abstention_resolution(
|
||||
refusal=None,
|
||||
violation=contract_assessment_none_violation(),
|
||||
)
|
||||
refusal = open_geometry_refusal(
|
||||
missing_bindings=tuple(contract_assessment.missing_bindings),
|
||||
unresolved_hazards=tuple(contract_assessment.unresolved_hazards),
|
||||
explanation=str(contract_assessment.explanation or ""),
|
||||
)
|
||||
# gate_fired is the pipeline's residual-failure flag; either way the
|
||||
# contract is not closed for certified answers.
|
||||
del gate_fired # used as documentation of pipeline dual; gate is conjugate
|
||||
return _abstention_resolution(refusal=refusal, violation=None)
|
||||
|
||||
surface, articulation_surface, authority = _base_runtime_surface(
|
||||
canonical_surface=canonical_surface or "",
|
||||
pre_decoration_surface=pre_decoration_surface or "",
|
||||
|
|
@ -103,14 +207,23 @@ def resolve_surface(
|
|||
# Forward and conjugate evaluated as independent competitors; commit
|
||||
# substrate only when both pass (and gate_fired is false).
|
||||
forward_ok = _forward_surface_ok(proposition_graph, contract_assessment)
|
||||
conjugate_ok = _conjugate_coherence_ok(contract_assessment)
|
||||
|
||||
if not gate_fired and realized_surface and forward_ok and conjugate_ok:
|
||||
# When require_closed_geometry is False, preserve historical gate_fired
|
||||
# blocking of substrate even if assessment looks closed.
|
||||
if not require_closed_geometry:
|
||||
gate_blocks = gate_fired
|
||||
else:
|
||||
# Under fail-closed geometry, open conjugate already returned.
|
||||
# gate_fired with a closed assessment is treated as residual conflict
|
||||
# → still refuse substrate, keep runtime only if conjugate ok.
|
||||
gate_blocks = gate_fired
|
||||
|
||||
if not gate_blocks and realized_surface and forward_ok and conjugate_ok:
|
||||
surface = realized_surface
|
||||
articulation_surface = realized_surface
|
||||
authority = "substrate_realizer"
|
||||
elif (
|
||||
not gate_fired
|
||||
not gate_blocks
|
||||
and realized_surface
|
||||
and realizer_useful
|
||||
and conjugate_ok
|
||||
|
|
@ -141,11 +254,24 @@ def resolve_surface(
|
|||
)
|
||||
fold_sources.append("compose")
|
||||
|
||||
proof: ProofTrace | None = None
|
||||
if conjugate_ok and contract_assessment is not None:
|
||||
# Closed geometry path: embed gate scalars when explanation carries them.
|
||||
proof = geometry_contract_trace(
|
||||
versor_condition=0.0,
|
||||
goldtether_residual=0.0,
|
||||
closed=True,
|
||||
)
|
||||
|
||||
return SurfaceResolution(
|
||||
surface=surface,
|
||||
articulation_surface=articulation_surface,
|
||||
authority=authority,
|
||||
fold_sources=tuple(fold_sources),
|
||||
authoritative=True,
|
||||
refusal=None,
|
||||
contract_violation=None,
|
||||
proof_trace=proof,
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -188,3 +314,12 @@ def _substrate_supreme(
|
|||
return _forward_surface_ok(proposition_graph, contract_assessment) and (
|
||||
_conjugate_coherence_ok(contract_assessment)
|
||||
)
|
||||
|
||||
|
||||
__all__ = [
|
||||
"SurfaceResolution",
|
||||
"resolve_surface",
|
||||
"_conjugate_coherence_ok",
|
||||
"_forward_surface_ok",
|
||||
"_substrate_supreme",
|
||||
]
|
||||
|
|
|
|||
50
core/semantic_primitives/__init__.py
Normal file
50
core/semantic_primitives/__init__.py
Normal file
|
|
@ -0,0 +1,50 @@
|
|||
"""Shared typed semantic primitives for linguistic → field compilation.
|
||||
|
||||
Authoritative representations are frozen validated dataclasses — never bare
|
||||
dicts, free strings, or JSON blobs at the new compiler seams.
|
||||
|
||||
Linguistic layers may only *emit candidates* bound to these types. They must
|
||||
not assign Cl(4,1) field state.
|
||||
"""
|
||||
|
||||
from core.semantic_primitives.model import (
|
||||
AmbiguityManifold,
|
||||
ConservationLaw,
|
||||
Container,
|
||||
DimensionalType,
|
||||
Entity,
|
||||
Event,
|
||||
IdentityBridge,
|
||||
MissingReferent,
|
||||
Operator,
|
||||
OperatorClass,
|
||||
ProvenanceSpan,
|
||||
Quantity,
|
||||
Relation,
|
||||
RelationKind,
|
||||
TemporalExtent,
|
||||
TemporalFrame,
|
||||
TemporalKind,
|
||||
ValidationError,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"AmbiguityManifold",
|
||||
"ConservationLaw",
|
||||
"Container",
|
||||
"DimensionalType",
|
||||
"Entity",
|
||||
"Event",
|
||||
"IdentityBridge",
|
||||
"MissingReferent",
|
||||
"Operator",
|
||||
"OperatorClass",
|
||||
"ProvenanceSpan",
|
||||
"Quantity",
|
||||
"Relation",
|
||||
"RelationKind",
|
||||
"TemporalExtent",
|
||||
"TemporalFrame",
|
||||
"TemporalKind",
|
||||
"ValidationError",
|
||||
]
|
||||
341
core/semantic_primitives/model.py
Normal file
341
core/semantic_primitives/model.py
Normal file
|
|
@ -0,0 +1,341 @@
|
|||
"""Typed semantic primitives (linguistic governance Phase 2)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from enum import Enum
|
||||
from fractions import Fraction
|
||||
from typing import Any, Iterable, Sequence
|
||||
|
||||
|
||||
class ValidationError(ValueError):
|
||||
"""Ill-typed or incomplete primitive construction."""
|
||||
|
||||
|
||||
class DimensionalType(str, Enum):
|
||||
COUNT = "count"
|
||||
LENGTH = "length"
|
||||
MASS = "mass"
|
||||
TIME = "time"
|
||||
CURRENCY = "currency"
|
||||
RATE = "rate"
|
||||
DIMENSIONLESS = "dimensionless"
|
||||
UNKNOWN = "unknown"
|
||||
|
||||
|
||||
class RelationKind(str, Enum):
|
||||
POSSESSION = "possession"
|
||||
COMPARISON = "comparison"
|
||||
EQUALITY = "equality"
|
||||
RATIO = "ratio"
|
||||
MEMBERSHIP = "membership"
|
||||
ORDER = "order"
|
||||
AGENT = "agent"
|
||||
PATIENT = "patient"
|
||||
RECIPIENT = "recipient"
|
||||
POSSESSOR = "possessor"
|
||||
SOURCE = "source"
|
||||
TARGET = "target"
|
||||
OTHER = "other"
|
||||
|
||||
|
||||
class OperatorClass(str, Enum):
|
||||
TRANSFER = "transfer"
|
||||
REMOVAL = "removal"
|
||||
ACCUMULATION = "accumulation"
|
||||
CREATION = "creation"
|
||||
PARTITION = "partition"
|
||||
COMPARISON = "comparison"
|
||||
TRANSFORMATION = "transformation"
|
||||
RECURRENCE = "recurrence"
|
||||
CAUSATION = "causation"
|
||||
IDENTITY_CONTINUITY = "identity_continuity"
|
||||
UNKNOWN = "unknown"
|
||||
|
||||
|
||||
class TemporalKind(str, Enum):
|
||||
PRIOR_ONGOING = "prior_ongoing"
|
||||
PRIOR_COMPLETED = "prior_completed"
|
||||
PRESENT = "present"
|
||||
FUTURE = "future"
|
||||
RECURRING = "recurring"
|
||||
UNSPECIFIED = "unspecified"
|
||||
|
||||
|
||||
def _require_nonempty(value: str, name: str) -> str:
|
||||
if not isinstance(value, str) or not value.strip():
|
||||
raise ValidationError(f"{name} must be a non-empty str")
|
||||
return value
|
||||
|
||||
|
||||
def _reject_dict_blob(value: Any, name: str) -> None:
|
||||
if isinstance(value, dict):
|
||||
raise ValidationError(
|
||||
f"{name} must not be an untyped dict; use the typed constructor"
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ProvenanceSpan:
|
||||
start: int
|
||||
end: int
|
||||
text: str = ""
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if self.start < 0 or self.end < self.start:
|
||||
raise ValidationError("ProvenanceSpan requires 0 <= start <= end")
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Entity:
|
||||
"""Persistent identity anchor with typed id, name, type, and frame bindings."""
|
||||
|
||||
entity_id: str
|
||||
name: str
|
||||
entity_type: str
|
||||
frame_ids: tuple[str, ...] = ()
|
||||
provenance: ProvenanceSpan | None = None
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
_require_nonempty(self.entity_id, "Entity.entity_id")
|
||||
_require_nonempty(self.name, "Entity.name")
|
||||
_require_nonempty(self.entity_type, "Entity.entity_type")
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Quantity:
|
||||
"""Value + unit + dimensional type + owner + frame (no raw bare floats alone)."""
|
||||
|
||||
value: Fraction
|
||||
unit: str
|
||||
dimensional_type: DimensionalType
|
||||
owner_entity_id: str
|
||||
frame_id: str
|
||||
source_token: str = ""
|
||||
provenance: ProvenanceSpan | None = None
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if isinstance(self.value, float):
|
||||
# Allow float only if caller wrapped — still reject bare construction
|
||||
# via type check: Fraction is required.
|
||||
raise ValidationError(
|
||||
"Quantity.value must be Fraction (not float) to avoid untyped numerics"
|
||||
)
|
||||
if not isinstance(self.value, Fraction):
|
||||
raise ValidationError("Quantity.value must be a Fraction")
|
||||
_require_nonempty(self.unit, "Quantity.unit")
|
||||
if not isinstance(self.dimensional_type, DimensionalType):
|
||||
raise ValidationError("Quantity.dimensional_type must be DimensionalType")
|
||||
_require_nonempty(self.owner_entity_id, "Quantity.owner_entity_id")
|
||||
_require_nonempty(self.frame_id, "Quantity.frame_id")
|
||||
|
||||
@classmethod
|
||||
def from_int(
|
||||
cls,
|
||||
value: int,
|
||||
*,
|
||||
unit: str,
|
||||
dimensional_type: DimensionalType,
|
||||
owner_entity_id: str,
|
||||
frame_id: str,
|
||||
source_token: str = "",
|
||||
provenance: ProvenanceSpan | None = None,
|
||||
) -> "Quantity":
|
||||
return cls(
|
||||
value=Fraction(value),
|
||||
unit=unit,
|
||||
dimensional_type=dimensional_type,
|
||||
owner_entity_id=owner_entity_id,
|
||||
frame_id=frame_id,
|
||||
source_token=source_token,
|
||||
provenance=provenance,
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Container:
|
||||
"""Bounded inventory/balance with typed content and conserved quantity id."""
|
||||
|
||||
container_id: str
|
||||
owner_entity_id: str
|
||||
content_entity_ids: tuple[str, ...]
|
||||
conserved_quantity_id: str
|
||||
capacity: Quantity | None = None
|
||||
frame_id: str = "default"
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
_require_nonempty(self.container_id, "Container.container_id")
|
||||
_require_nonempty(self.owner_entity_id, "Container.owner_entity_id")
|
||||
_require_nonempty(self.conserved_quantity_id, "Container.conserved_quantity_id")
|
||||
if not isinstance(self.content_entity_ids, tuple):
|
||||
raise ValidationError("Container.content_entity_ids must be a tuple")
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class TemporalFrame:
|
||||
frame_id: str
|
||||
kind: TemporalKind
|
||||
label: str = ""
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
_require_nonempty(self.frame_id, "TemporalFrame.frame_id")
|
||||
if not isinstance(self.kind, TemporalKind):
|
||||
raise ValidationError("TemporalFrame.kind must be TemporalKind")
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class TemporalExtent:
|
||||
frame: TemporalFrame
|
||||
start_label: str = ""
|
||||
end_label: str = ""
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Event:
|
||||
"""Typed transformation with roles; incomplete roles must be explicit None."""
|
||||
|
||||
event_id: str
|
||||
operator_class: OperatorClass
|
||||
agent_entity_id: str | None
|
||||
patient_entity_id: str | None
|
||||
source_entity_id: str | None
|
||||
target_entity_id: str | None
|
||||
conserved_quantity_id: str | None
|
||||
temporal_extent: TemporalExtent | None
|
||||
direction: str = ""
|
||||
unresolved_roles: tuple[str, ...] = ()
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
_require_nonempty(self.event_id, "Event.event_id")
|
||||
if not isinstance(self.operator_class, OperatorClass):
|
||||
raise ValidationError("Event.operator_class must be OperatorClass")
|
||||
if isinstance(self.agent_entity_id, dict):
|
||||
raise ValidationError("Event roles must not be dict blobs")
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Operator:
|
||||
"""Executable semantic transformation linked to an Event type."""
|
||||
|
||||
operator_id: str
|
||||
operator_class: OperatorClass
|
||||
event_id: str
|
||||
executable_symbol: str
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
_require_nonempty(self.operator_id, "Operator.operator_id")
|
||||
_require_nonempty(self.event_id, "Operator.event_id")
|
||||
_require_nonempty(self.executable_symbol, "Operator.executable_symbol")
|
||||
if not isinstance(self.operator_class, OperatorClass):
|
||||
raise ValidationError("Operator.operator_class must be OperatorClass")
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Relation:
|
||||
relation_id: str
|
||||
kind: RelationKind
|
||||
left_entity_id: str
|
||||
right_entity_id: str
|
||||
polarity: bool = True
|
||||
provenance: ProvenanceSpan | None = None
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
_require_nonempty(self.relation_id, "Relation.relation_id")
|
||||
if not isinstance(self.kind, RelationKind):
|
||||
raise ValidationError("Relation.kind must be RelationKind")
|
||||
_require_nonempty(self.left_entity_id, "Relation.left_entity_id")
|
||||
_require_nonempty(self.right_entity_id, "Relation.right_entity_id")
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class IdentityBridge:
|
||||
"""Explicit cross-frame entity identity claim with change log."""
|
||||
|
||||
bridge_id: str
|
||||
entity_id: str
|
||||
from_frame_id: str
|
||||
to_frame_id: str
|
||||
change_log: tuple[str, ...] = ()
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
_require_nonempty(self.bridge_id, "IdentityBridge.bridge_id")
|
||||
_require_nonempty(self.entity_id, "IdentityBridge.entity_id")
|
||||
_require_nonempty(self.from_frame_id, "IdentityBridge.from_frame_id")
|
||||
_require_nonempty(self.to_frame_id, "IdentityBridge.to_frame_id")
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ConservationLaw:
|
||||
law_id: str
|
||||
quantity_id: str
|
||||
persists: bool
|
||||
flows: bool
|
||||
externally_added: bool
|
||||
consumed: bool
|
||||
transferred: bool
|
||||
statement: str = ""
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
_require_nonempty(self.law_id, "ConservationLaw.law_id")
|
||||
_require_nonempty(self.quantity_id, "ConservationLaw.quantity_id")
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class AmbiguityManifold:
|
||||
"""Set of unresolved semantic candidates with explicit resolution condition."""
|
||||
|
||||
manifold_id: str
|
||||
candidate_ids: tuple[str, ...]
|
||||
candidate_kinds: tuple[str, ...]
|
||||
resolution_condition: str
|
||||
resolved_id: str | None = None
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
_require_nonempty(self.manifold_id, "AmbiguityManifold.manifold_id")
|
||||
if isinstance(self.candidate_ids, dict) or isinstance(self.candidate_kinds, dict):
|
||||
raise ValidationError("AmbiguityManifold candidates must not be dict blobs")
|
||||
if not self.candidate_ids:
|
||||
raise ValidationError("AmbiguityManifold.candidate_ids must be non-empty")
|
||||
if len(self.candidate_ids) != len(self.candidate_kinds):
|
||||
raise ValidationError(
|
||||
"AmbiguityManifold candidate_ids/kinds length mismatch"
|
||||
)
|
||||
_require_nonempty(
|
||||
self.resolution_condition, "AmbiguityManifold.resolution_condition"
|
||||
)
|
||||
if self.resolved_id is not None and self.resolved_id not in self.candidate_ids:
|
||||
raise ValidationError("resolved_id must be one of candidate_ids")
|
||||
|
||||
@property
|
||||
def resolved(self) -> bool:
|
||||
return self.resolved_id is not None and len(self.candidate_ids) == 1
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class MissingReferent:
|
||||
"""Absent antecedent/owner/unit/time — never silently filled."""
|
||||
|
||||
referent_id: str
|
||||
role: str
|
||||
expected_kind: str
|
||||
context: str = ""
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
_require_nonempty(self.referent_id, "MissingReferent.referent_id")
|
||||
_require_nonempty(self.role, "MissingReferent.role")
|
||||
_require_nonempty(self.expected_kind, "MissingReferent.expected_kind")
|
||||
|
||||
|
||||
def reject_untyped(value: Any, expected_name: str) -> None:
|
||||
"""Public seam guard: refuse dict/str as authoritative complex primitive."""
|
||||
_reject_dict_blob(value, expected_name)
|
||||
if isinstance(value, str) and expected_name in {
|
||||
"Event",
|
||||
"Quantity",
|
||||
"AmbiguityManifold",
|
||||
"Entity",
|
||||
"Container",
|
||||
}:
|
||||
raise ValidationError(
|
||||
f"{expected_name} must not be a bare string as authoritative representation"
|
||||
)
|
||||
59
generate/linguistic_pipeline/__init__.py
Normal file
59
generate/linguistic_pipeline/__init__.py
Normal file
|
|
@ -0,0 +1,59 @@
|
|||
"""Trilingual constraint pipeline: English → Hebrew event → Koine relation-time → field.
|
||||
|
||||
Linguistic layers produce typed candidates only. Field integration alone may
|
||||
close, leave ambiguous, or refuse. Articulation requires a closed proof trace.
|
||||
"""
|
||||
|
||||
from generate.linguistic_pipeline.articulation import (
|
||||
ArticulationFirewallVerdict,
|
||||
ArticulatedClaim,
|
||||
articulate_from_proof,
|
||||
firewall_check,
|
||||
)
|
||||
from generate.linguistic_pipeline.field_integration import (
|
||||
AmbiguousFieldState,
|
||||
CoherentFieldState,
|
||||
FieldOutcome,
|
||||
integrate_constraints,
|
||||
)
|
||||
from generate.linguistic_pipeline.layer_a_english import (
|
||||
CandidateEntity,
|
||||
CandidateRelation,
|
||||
SurfaceConstraintSet,
|
||||
extract_english_surface,
|
||||
)
|
||||
from generate.linguistic_pipeline.layer_b_hebrew import (
|
||||
EventOperatorSet,
|
||||
classify_hebrew_events,
|
||||
)
|
||||
from generate.linguistic_pipeline.layer_c_koine import (
|
||||
RelationGraph,
|
||||
TemporalTopology,
|
||||
bind_koine_relation_time,
|
||||
)
|
||||
from generate.linguistic_pipeline.pipeline import (
|
||||
LinguisticPipelineResult,
|
||||
run_linguistic_pipeline,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"AmbiguousFieldState",
|
||||
"ArticulatedClaim",
|
||||
"ArticulationFirewallVerdict",
|
||||
"CandidateEntity",
|
||||
"CandidateRelation",
|
||||
"CoherentFieldState",
|
||||
"EventOperatorSet",
|
||||
"FieldOutcome",
|
||||
"LinguisticPipelineResult",
|
||||
"RelationGraph",
|
||||
"SurfaceConstraintSet",
|
||||
"TemporalTopology",
|
||||
"articulate_from_proof",
|
||||
"bind_koine_relation_time",
|
||||
"classify_hebrew_events",
|
||||
"extract_english_surface",
|
||||
"firewall_check",
|
||||
"integrate_constraints",
|
||||
"run_linguistic_pipeline",
|
||||
]
|
||||
212
generate/linguistic_pipeline/articulation.py
Normal file
212
generate/linguistic_pipeline/articulation.py
Normal file
|
|
@ -0,0 +1,212 @@
|
|||
"""Proof-preserving articulation + firewall.
|
||||
|
||||
Every articulated claim must:
|
||||
1. cite only valid citation keys (step_id or kind:symbol — never bare payload values);
|
||||
2. have every content token certified by the cited steps' symbols/payloads
|
||||
(or a closed function-word lexicon for English glue).
|
||||
|
||||
Claims failing either check are HALLUCINATED and suppressed.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
|
||||
from core.cognition.fail_closed import CoherenceRefusal, FailureClass, ResidualState
|
||||
from core.cognition.proof_trace import ProofTrace
|
||||
from core.semantic_primitives import ValidationError
|
||||
from generate.linguistic_pipeline.field_integration import (
|
||||
AmbiguousFieldState,
|
||||
CoherentFieldState,
|
||||
FieldOutcome,
|
||||
)
|
||||
|
||||
# Do not treat sentence punctuation (.) as part of a token — otherwise
|
||||
# "closed." fails certification against certified token "closed".
|
||||
_TOKEN_RE = re.compile(r"[a-z0-9_:-]+", re.IGNORECASE)
|
||||
|
||||
# English glue only — never domain content (entities, quantities, causes).
|
||||
_FUNCTION_WORDS: frozenset[str] = frozenset(
|
||||
{
|
||||
"a",
|
||||
"an",
|
||||
"the",
|
||||
"is",
|
||||
"are",
|
||||
"was",
|
||||
"were",
|
||||
"be",
|
||||
"been",
|
||||
"of",
|
||||
"to",
|
||||
"for",
|
||||
"and",
|
||||
"or",
|
||||
"as",
|
||||
"in",
|
||||
"on",
|
||||
"at",
|
||||
"by",
|
||||
"with",
|
||||
"from",
|
||||
"that",
|
||||
"this",
|
||||
"these",
|
||||
"those",
|
||||
"it",
|
||||
"its",
|
||||
"class",
|
||||
"operator",
|
||||
"admissible",
|
||||
"geometric",
|
||||
"contract",
|
||||
"closed",
|
||||
"field",
|
||||
"state",
|
||||
"versor",
|
||||
"condition",
|
||||
"residual",
|
||||
"goldtether",
|
||||
"embedding",
|
||||
"applied",
|
||||
"selected",
|
||||
"unique",
|
||||
"configuration",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ArticulatedClaim:
|
||||
text: str
|
||||
trace_refs: tuple[str, ...]
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if not self.text.strip():
|
||||
raise ValidationError("ArticulatedClaim.text must be non-empty")
|
||||
if not self.trace_refs:
|
||||
raise ValidationError(
|
||||
"ArticulatedClaim.trace_refs must be non-empty (firewall precondition)"
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ArticulationFirewallVerdict:
|
||||
legal_claims: tuple[ArticulatedClaim, ...]
|
||||
hallucinated: tuple[str, ...]
|
||||
surface: str
|
||||
|
||||
@property
|
||||
def clean(self) -> bool:
|
||||
return not self.hallucinated
|
||||
|
||||
|
||||
def _content_tokens(text: str) -> frozenset[str]:
|
||||
return frozenset(m.group(0).lower() for m in _TOKEN_RE.finditer(text))
|
||||
|
||||
|
||||
def _claim_is_certified(claim: ArticulatedClaim, proof_trace: ProofTrace) -> bool:
|
||||
"""True iff all refs are valid citations AND all content tokens are certified."""
|
||||
citation_keys = proof_trace.all_citation_keys()
|
||||
if not claim.trace_refs:
|
||||
return False
|
||||
if not all(ref in citation_keys for ref in claim.trace_refs):
|
||||
return False
|
||||
# Every ref must resolve to at least one step (no dangling synonym).
|
||||
resolved = proof_trace.steps_for_refs(claim.trace_refs)
|
||||
if len(resolved) == 0:
|
||||
return False
|
||||
if len({r for r in claim.trace_refs}) != len(claim.trace_refs):
|
||||
pass # duplicates ok
|
||||
# All refs must independently resolve
|
||||
for ref in claim.trace_refs:
|
||||
if not proof_trace.steps_for_refs((ref,)):
|
||||
return False
|
||||
|
||||
certified = set(proof_trace.certified_content_for_refs(claim.trace_refs))
|
||||
certified |= set(_FUNCTION_WORDS)
|
||||
for token in _content_tokens(claim.text):
|
||||
if token in certified:
|
||||
continue
|
||||
# Allow numeric fragments only if the exact token is certified (e.g. value text)
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def firewall_check(
|
||||
claims: tuple[ArticulatedClaim, ...] | list[ArticulatedClaim],
|
||||
proof_trace: ProofTrace,
|
||||
) -> ArticulationFirewallVerdict:
|
||||
"""Diff claims against proof-trace citations + certified content; suppress rest."""
|
||||
if not isinstance(proof_trace, ProofTrace):
|
||||
raise ValidationError("firewall requires a ProofTrace")
|
||||
legal: list[ArticulatedClaim] = []
|
||||
hallucinated: list[str] = []
|
||||
for claim in claims:
|
||||
if _claim_is_certified(claim, proof_trace):
|
||||
legal.append(claim)
|
||||
else:
|
||||
hallucinated.append(claim.text)
|
||||
surface = " ".join(c.text for c in legal)
|
||||
return ArticulationFirewallVerdict(
|
||||
legal_claims=tuple(legal),
|
||||
hallucinated=tuple(hallucinated),
|
||||
surface=surface,
|
||||
)
|
||||
|
||||
|
||||
def articulate_from_proof(outcome: FieldOutcome) -> ArticulationFirewallVerdict | CoherenceRefusal:
|
||||
"""Render only certified state. Ambiguous/refusal never invent answers."""
|
||||
if isinstance(outcome, CoherenceRefusal):
|
||||
return outcome
|
||||
if isinstance(outcome, AmbiguousFieldState):
|
||||
return CoherenceRefusal(
|
||||
failure_class=FailureClass.AMBIGUITY,
|
||||
violated_condition="unique_operator_class",
|
||||
residual_state=ResidualState(detail=outcome.reason),
|
||||
refusal_reason=(
|
||||
"ambiguous field state — no single answer may be articulated"
|
||||
),
|
||||
surface_message=(
|
||||
"I cannot certify a unique answer: multiple operator classes remain "
|
||||
f"admissible ({', '.join(c.value for c in outcome.candidate_operator_classes)})."
|
||||
),
|
||||
)
|
||||
if not isinstance(outcome, CoherentFieldState):
|
||||
raise ValidationError("articulate_from_proof requires a FieldOutcome")
|
||||
|
||||
trace = outcome.proof_trace
|
||||
op = outcome.selected_operator_class.value
|
||||
# Claim text uses only function-word glue + tokens certified on the cited steps.
|
||||
# Never invent magnitudes or entities not present on those steps.
|
||||
claims = [
|
||||
ArticulatedClaim(
|
||||
text="The geometric contract closed.",
|
||||
trace_refs=("closure:0",),
|
||||
),
|
||||
ArticulatedClaim(
|
||||
text=f"The admissible operator class is {op}.",
|
||||
trace_refs=("atom:operator",),
|
||||
),
|
||||
]
|
||||
return firewall_check(claims, trace)
|
||||
|
||||
|
||||
def inject_uncertified_claim(
|
||||
verdict: ArticulationFirewallVerdict,
|
||||
*,
|
||||
bogus_text: str,
|
||||
proof_trace: ProofTrace,
|
||||
trace_refs: tuple[str, ...] | None = None,
|
||||
) -> ArticulationFirewallVerdict:
|
||||
"""Test helper: re-run firewall with an injected claim.
|
||||
|
||||
Default uses a totally-absent ref. Callers may pass payload-value refs
|
||||
(e.g. ``("2",)``) to prove the whitelist bypass is closed.
|
||||
"""
|
||||
refs = trace_refs if trace_refs is not None else ("not_in_trace_ever",)
|
||||
claims = list(verdict.legal_claims) + [
|
||||
ArticulatedClaim(text=bogus_text, trace_refs=refs)
|
||||
]
|
||||
return firewall_check(tuple(claims), proof_trace)
|
||||
459
generate/linguistic_pipeline/field_integration.py
Normal file
459
generate/linguistic_pipeline/field_integration.py
Normal file
|
|
@ -0,0 +1,459 @@
|
|||
"""Field integration — embed typed constraints into Cl(4,1); return outcomes.
|
||||
|
||||
Linguistic layers never decide the field outcome. Candidates are embedded as
|
||||
conformal points / composed rotors; closure is versor_condition + GoldTether.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
import numpy as np
|
||||
|
||||
from algebra.backend import versor_condition
|
||||
from algebra.cga import embed_point, is_null
|
||||
from algebra.cl41 import geometric_product
|
||||
from algebra.rotor import make_rotor_from_angle
|
||||
from core.cognition.fail_closed import (
|
||||
CoherenceRefusal,
|
||||
FailureClass,
|
||||
ResidualState,
|
||||
)
|
||||
from core.cognition.proof_trace import (
|
||||
ProofTrace,
|
||||
build_closed_trace,
|
||||
build_refusal_trace,
|
||||
)
|
||||
from core.physics.goldtether import coherence_residual
|
||||
from core.semantic_primitives import OperatorClass, ValidationError
|
||||
from generate.linguistic_pipeline.layer_a_english import SurfaceConstraintSet
|
||||
from generate.linguistic_pipeline.layer_b_hebrew import EventOperatorSet
|
||||
from generate.linguistic_pipeline.layer_c_koine import RelationGraph, TemporalTopology
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class CoherentFieldState:
|
||||
field: np.ndarray
|
||||
proof_trace: ProofTrace
|
||||
selected_operator_class: OperatorClass
|
||||
versor_condition: float
|
||||
goldtether_residual: float
|
||||
# Digests of embedded multivectors — proves structure entered the field.
|
||||
embed_digests: tuple[tuple[str, str], ...] = ()
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if not self.proof_trace.closed:
|
||||
raise ValidationError("CoherentFieldState requires a closed proof_trace")
|
||||
arr = np.asarray(self.field)
|
||||
if arr.shape != (32,):
|
||||
raise ValidationError("CoherentFieldState.field must be shape (32,)")
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class AmbiguousFieldState:
|
||||
candidate_operator_classes: tuple[OperatorClass, ...]
|
||||
manifold_ids: tuple[str, ...]
|
||||
proof_trace: ProofTrace
|
||||
reason: str
|
||||
|
||||
@property
|
||||
def emits_answer(self) -> bool:
|
||||
return False
|
||||
|
||||
|
||||
FieldOutcome = CoherentFieldState | AmbiguousFieldState | CoherenceRefusal
|
||||
|
||||
_OP_PLANE: dict[OperatorClass, tuple[float, int]] = {
|
||||
OperatorClass.TRANSFER: (0.21, 6),
|
||||
OperatorClass.REMOVAL: (0.34, 7),
|
||||
OperatorClass.ACCUMULATION: (0.47, 8),
|
||||
OperatorClass.CREATION: (0.18, 10),
|
||||
OperatorClass.PARTITION: (0.29, 11),
|
||||
OperatorClass.COMPARISON: (0.41, 13),
|
||||
OperatorClass.TRANSFORMATION: (0.53, 6),
|
||||
OperatorClass.RECURRENCE: (0.37, 7),
|
||||
OperatorClass.CAUSATION: (0.44, 8),
|
||||
OperatorClass.IDENTITY_CONTINUITY: (0.11, 10),
|
||||
OperatorClass.UNKNOWN: (0.07, 11),
|
||||
}
|
||||
|
||||
|
||||
def _mv_digest(mv: np.ndarray) -> str:
|
||||
arr = np.asarray(mv, dtype=np.float64).tobytes()
|
||||
return hashlib.sha256(arr).hexdigest()[:16]
|
||||
|
||||
|
||||
def _stable_euclidean(seed: str) -> np.ndarray:
|
||||
"""Deterministic R^3 coords in (-1,1)^3 from a candidate id (not a count)."""
|
||||
digest = hashlib.sha256(seed.encode("utf-8")).digest()
|
||||
coords = []
|
||||
for i in range(3):
|
||||
u = int.from_bytes(digest[2 * i : 2 * i + 2], "big") / 65535.0
|
||||
coords.append(2.0 * u - 1.0)
|
||||
return np.asarray(coords, dtype=np.float64)
|
||||
|
||||
|
||||
def _identity_rotor() -> np.ndarray:
|
||||
r = np.zeros(32, dtype=np.float64)
|
||||
r[0] = 1.0
|
||||
return r
|
||||
|
||||
|
||||
def _compose_rotor(F: np.ndarray, R: np.ndarray) -> np.ndarray:
|
||||
"""Left-compose rotors. Both operands must already be unit rotors.
|
||||
|
||||
No ``unitize_versor`` here — that is forbidden outside owned construction
|
||||
boundaries (INV-02b). ``make_rotor_from_angle`` products stay on Spin by
|
||||
construction; if residual drifts, the caller refuses rather than repair.
|
||||
"""
|
||||
product = geometric_product(
|
||||
np.asarray(R, dtype=np.float64),
|
||||
np.asarray(F, dtype=np.float64),
|
||||
)
|
||||
return np.asarray(product, dtype=np.float64)
|
||||
|
||||
|
||||
def _point_to_seed_rotor(point: np.ndarray, *, plane: int = 6) -> np.ndarray:
|
||||
"""Map a conformal null point into a small rotor via grade-1 projection angle.
|
||||
|
||||
Uses the Euclidean e1 component as an angle seed so distinct points produce
|
||||
distinct rotors under composition (structure-sensitive, not count-only).
|
||||
Built only from ``make_rotor_from_angle`` (closed by construction).
|
||||
"""
|
||||
p = np.asarray(point, dtype=np.float64).ravel()
|
||||
# e1 is component index 1 in Cl(4,1) layout
|
||||
e1 = float(p[1]) if p.shape[0] >= 2 else 0.0
|
||||
e2 = float(p[2]) if p.shape[0] >= 3 else 0.0
|
||||
angle = float(np.tanh(e1) * 0.4 + np.tanh(e2) * 0.25)
|
||||
return make_rotor_from_angle(angle, bivector_idx=plane)
|
||||
|
||||
|
||||
def _embed_entity_point(entity_id: str) -> np.ndarray:
|
||||
coords = _stable_euclidean(entity_id)
|
||||
point = embed_point(coords, dtype=np.float64)
|
||||
if not is_null(point, tol=1e-5):
|
||||
raise ValidationError(f"entity embed not null for {entity_id!r}")
|
||||
return np.asarray(point, dtype=np.float64)
|
||||
|
||||
|
||||
def _embed_quantity_point(value_text: str, unit: str, token_id: str) -> np.ndarray:
|
||||
try:
|
||||
value = float(value_text)
|
||||
except ValueError as exc:
|
||||
raise ValidationError(f"non-numeric quantity {value_text!r}") from exc
|
||||
# Bound into embed-safe Euclidean range; unit seed perturbs y/z.
|
||||
unit_vec = _stable_euclidean(f"unit:{unit}:{token_id}")
|
||||
scale = float(np.tanh(value / 100.0))
|
||||
coords = np.asarray(
|
||||
[scale, 0.15 * unit_vec[1], 0.15 * unit_vec[2]],
|
||||
dtype=np.float64,
|
||||
)
|
||||
point = embed_point(coords, dtype=np.float64)
|
||||
if not is_null(point, tol=1e-5):
|
||||
raise ValidationError(f"quantity embed not null for {token_id!r}")
|
||||
return np.asarray(point, dtype=np.float64)
|
||||
|
||||
|
||||
def _relation_rotor(left_point: np.ndarray, right_point: np.ndarray) -> np.ndarray:
|
||||
"""Structure rotor from two conformal points, using only closed rotors.
|
||||
|
||||
Reads Euclidean components of the null points (already embedded via
|
||||
``embed_point``) and composes ``make_rotor_from_angle`` factors — no
|
||||
unitize/repair. Distinct point pairs ⇒ distinct rotor products.
|
||||
"""
|
||||
seed = _identity_rotor()
|
||||
# Separation in e1/e2/e3 of the conformal embeddings.
|
||||
for plane, idx in ((6, 1), (7, 2), (8, 3)):
|
||||
delta = float(left_point[idx] - right_point[idx])
|
||||
if abs(delta) < 1e-15:
|
||||
continue
|
||||
angle = float(np.tanh(delta) * 0.25)
|
||||
seed = geometric_product(
|
||||
make_rotor_from_angle(angle, bivector_idx=plane),
|
||||
seed,
|
||||
)
|
||||
# Relative radial (n_o weight / e4-e5 mix) as an extra plane.
|
||||
radial = float(left_point[4] - right_point[4])
|
||||
seed = geometric_product(
|
||||
make_rotor_from_angle(float(np.tanh(radial) * 0.15), bivector_idx=10),
|
||||
seed,
|
||||
)
|
||||
return np.asarray(seed, dtype=np.float64)
|
||||
|
||||
|
||||
def embed_constraints_into_field(
|
||||
surface: SurfaceConstraintSet,
|
||||
relation_graph: RelationGraph,
|
||||
selected: OperatorClass,
|
||||
) -> tuple[np.ndarray, tuple[tuple[str, str], ...], list[tuple[str, str, tuple[tuple[str, str], ...]]]]:
|
||||
"""Embed entities, quantities, and relations into a closed Cl(4,1) versor field.
|
||||
|
||||
Returns (field, embed_digests, atom_descriptors for proof).
|
||||
"""
|
||||
field = _identity_rotor()
|
||||
digests: list[tuple[str, str]] = []
|
||||
atoms: list[tuple[str, str, tuple[tuple[str, str], ...]]] = []
|
||||
entity_points: dict[str, np.ndarray] = {}
|
||||
|
||||
for ent in surface.entities:
|
||||
point = _embed_entity_point(ent.candidate_id)
|
||||
entity_points[ent.candidate_id] = point
|
||||
rotor = _point_to_seed_rotor(point, plane=6)
|
||||
field = _compose_rotor(field, rotor)
|
||||
d = _mv_digest(point)
|
||||
digests.append((f"entity:{ent.candidate_id}", d))
|
||||
atoms.append(
|
||||
(
|
||||
f"atom:ent:{ent.candidate_id}",
|
||||
ent.candidate_id,
|
||||
(
|
||||
("role", "entity"),
|
||||
("surface", ent.surface),
|
||||
("embed_digest", d),
|
||||
("kind_hint", ent.kind_hint),
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
for num in surface.numerics:
|
||||
point = _embed_quantity_point(num.value_text, num.unit, num.token_id)
|
||||
rotor = _point_to_seed_rotor(point, plane=7)
|
||||
field = _compose_rotor(field, rotor)
|
||||
d = _mv_digest(point)
|
||||
digests.append((f"quantity:{num.token_id}", d))
|
||||
atoms.append(
|
||||
(
|
||||
f"atom:qty:{num.token_id}",
|
||||
num.token_id,
|
||||
(
|
||||
("role", "quantity"),
|
||||
("value", num.value_text),
|
||||
("unit", num.unit),
|
||||
("embed_digest", d),
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
for i, rel in enumerate(relation_graph.relations):
|
||||
left = entity_points.get(rel.left_entity_id)
|
||||
right = entity_points.get(rel.right_entity_id)
|
||||
if left is None or right is None:
|
||||
raise ValidationError(
|
||||
f"relation {rel.relation_id} references unembedded entity"
|
||||
)
|
||||
rrot = _relation_rotor(left, right)
|
||||
field = _compose_rotor(field, rrot)
|
||||
d = _mv_digest(rrot)
|
||||
digests.append((f"relation:{rel.relation_id}", d))
|
||||
atoms.append(
|
||||
(
|
||||
f"atom:rel:{i}",
|
||||
rel.kind.value,
|
||||
(
|
||||
("left", rel.left_entity_id),
|
||||
("right", rel.right_entity_id),
|
||||
("kind", rel.kind.value),
|
||||
("embed_digest", d),
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
# Operator class as a distinct plane/angle — part of the field, not a label only.
|
||||
angle, plane = _OP_PLANE.get(selected, (0.07, 11))
|
||||
field = _compose_rotor(field, make_rotor_from_angle(angle, bivector_idx=plane))
|
||||
digests.append((f"operator:{selected.value}", _mv_digest(field)))
|
||||
|
||||
atoms.append(
|
||||
(
|
||||
"atom:operator",
|
||||
selected.value,
|
||||
(
|
||||
("operator_class", selected.value),
|
||||
("plane", str(plane)),
|
||||
("angle", f"{angle:.6f}"),
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
# No unitize — field is a product of construction-closed rotors only.
|
||||
return np.asarray(field, dtype=np.float64), tuple(digests), atoms
|
||||
|
||||
|
||||
def integrate_constraints(
|
||||
surface: SurfaceConstraintSet,
|
||||
events: EventOperatorSet,
|
||||
relation_graph: RelationGraph,
|
||||
temporal: TemporalTopology,
|
||||
*,
|
||||
force_geometry_fail: bool = False,
|
||||
) -> FieldOutcome:
|
||||
"""Integrate layered constraints into Cl(4,1) outcomes only.
|
||||
|
||||
Rules:
|
||||
* Missing referents that block unique closure → CoherenceRefusal
|
||||
* Multi-class HE manifold without unique operator → AmbiguousFieldState
|
||||
* Unique operator + embedded geometry closed → CoherentFieldState + proof
|
||||
* force_geometry_fail → CoherenceRefusal
|
||||
"""
|
||||
del temporal # frames recorded on event candidates; not a linguistic override
|
||||
|
||||
if relation_graph.missing_referents:
|
||||
miss = relation_graph.missing_referents[0]
|
||||
return CoherenceRefusal(
|
||||
failure_class=FailureClass.MISSING_REFERENT,
|
||||
violated_condition=f"referent_present:{miss.role}",
|
||||
residual_state=ResidualState(
|
||||
detail=f"{miss.referent_id}:{miss.expected_kind}:{miss.context}"
|
||||
),
|
||||
refusal_reason=(
|
||||
f"unresolvable referent role={miss.role} "
|
||||
f"expected={miss.expected_kind} context={miss.context!r}"
|
||||
),
|
||||
surface_message=(
|
||||
f"I cannot certify an answer: missing referent for role "
|
||||
f"'{miss.role}' ({miss.expected_kind})."
|
||||
),
|
||||
)
|
||||
|
||||
if force_geometry_fail:
|
||||
return CoherenceRefusal(
|
||||
failure_class=FailureClass.COHERENCE,
|
||||
violated_condition="versor_condition",
|
||||
residual_state=ResidualState(versor_condition=1.0, detail="forced_open"),
|
||||
refusal_reason="geometric contract forced open for verification",
|
||||
)
|
||||
|
||||
multi = [m for m in events.manifolds if not m.resolved and len(m.candidate_ids) > 1]
|
||||
if multi:
|
||||
classes = tuple(
|
||||
OperatorClass(k) for m in multi for k in m.candidate_kinds
|
||||
)
|
||||
unique = tuple(dict.fromkeys(classes))
|
||||
if len(unique) > 1:
|
||||
return AmbiguousFieldState(
|
||||
candidate_operator_classes=unique,
|
||||
manifold_ids=tuple(m.manifold_id for m in multi),
|
||||
proof_trace=build_refusal_trace(
|
||||
reason="ambiguous_operator_class",
|
||||
violated_condition="unique_operator_class",
|
||||
),
|
||||
reason="hebrew_root_admits_multiple_operator_classes",
|
||||
)
|
||||
|
||||
if events.candidates:
|
||||
classes = tuple(dict.fromkeys(c.operator_class for c in events.candidates))
|
||||
if len(classes) > 1:
|
||||
return AmbiguousFieldState(
|
||||
candidate_operator_classes=classes,
|
||||
manifold_ids=tuple(m.manifold_id for m in events.manifolds),
|
||||
proof_trace=build_refusal_trace(
|
||||
reason="ambiguous_operator_class",
|
||||
violated_condition="unique_operator_class",
|
||||
),
|
||||
reason="multiple_event_operator_classes",
|
||||
)
|
||||
selected = classes[0]
|
||||
else:
|
||||
if not surface.numerics:
|
||||
return CoherenceRefusal(
|
||||
failure_class=FailureClass.CONSTRAINT,
|
||||
violated_condition="event_or_numeric_present",
|
||||
residual_state=ResidualState(detail="no event operators or numerics"),
|
||||
refusal_reason="no admissible event or numeric constraints to close",
|
||||
)
|
||||
selected = OperatorClass.IDENTITY_CONTINUITY
|
||||
|
||||
try:
|
||||
field, digests, atoms = embed_constraints_into_field(
|
||||
surface, relation_graph, selected
|
||||
)
|
||||
except (ValidationError, ValueError) as exc:
|
||||
return CoherenceRefusal(
|
||||
failure_class=FailureClass.FIELD,
|
||||
violated_condition="cl41_embed",
|
||||
residual_state=ResidualState(detail=str(exc)),
|
||||
refusal_reason=f"Cl(4,1) embedding failed: {exc}",
|
||||
)
|
||||
|
||||
vc = float(versor_condition(field))
|
||||
gt = float(coherence_residual(field))
|
||||
if vc >= 1e-6 or gt > 1e-6:
|
||||
return CoherenceRefusal(
|
||||
failure_class=FailureClass.COHERENCE,
|
||||
violated_condition="versor_condition|goldtether",
|
||||
residual_state=ResidualState(
|
||||
versor_condition=vc, goldtether_residual=gt
|
||||
),
|
||||
refusal_reason="field failed versor/GoldTether closure after embedding",
|
||||
)
|
||||
|
||||
parent_ids = tuple(a[0] for a in atoms)
|
||||
ops = [
|
||||
(
|
||||
"op:embed",
|
||||
"cl41_embed_close",
|
||||
(
|
||||
("versor_condition", f"{vc:.6e}"),
|
||||
("goldtether_residual", f"{gt:.6e}"),
|
||||
("embed_count", str(len(digests))),
|
||||
),
|
||||
parent_ids,
|
||||
)
|
||||
]
|
||||
proof = build_closed_trace(
|
||||
atoms=atoms,
|
||||
operators=ops,
|
||||
closure_symbol="geometric_contract_closed",
|
||||
closure_payload=(
|
||||
("versor_condition", f"{vc:.6e}"),
|
||||
("goldtether_residual", f"{gt:.6e}"),
|
||||
("operator_class", selected.value),
|
||||
),
|
||||
)
|
||||
return CoherentFieldState(
|
||||
field=field,
|
||||
proof_trace=proof,
|
||||
selected_operator_class=selected,
|
||||
versor_condition=vc,
|
||||
goldtether_residual=gt,
|
||||
embed_digests=digests,
|
||||
)
|
||||
|
||||
|
||||
def outcome_kind(outcome: FieldOutcome) -> str:
|
||||
if isinstance(outcome, CoherentFieldState):
|
||||
return "coherent"
|
||||
if isinstance(outcome, AmbiguousFieldState):
|
||||
return "ambiguous"
|
||||
if isinstance(outcome, CoherenceRefusal):
|
||||
return "refusal"
|
||||
raise TypeError(f"unknown field outcome type: {type(outcome)!r}")
|
||||
|
||||
|
||||
def outcome_as_dict(outcome: FieldOutcome) -> dict[str, Any]:
|
||||
kind = outcome_kind(outcome)
|
||||
if isinstance(outcome, CoherentFieldState):
|
||||
return {
|
||||
"kind": kind,
|
||||
"operator_class": outcome.selected_operator_class.value,
|
||||
"versor_condition": outcome.versor_condition,
|
||||
"goldtether_residual": outcome.goldtether_residual,
|
||||
"embed_digests": [list(p) for p in outcome.embed_digests],
|
||||
"proof_trace": outcome.proof_trace.as_dict(),
|
||||
}
|
||||
if isinstance(outcome, AmbiguousFieldState):
|
||||
return {
|
||||
"kind": kind,
|
||||
"candidate_operator_classes": [
|
||||
c.value for c in outcome.candidate_operator_classes
|
||||
],
|
||||
"manifold_ids": list(outcome.manifold_ids),
|
||||
"reason": outcome.reason,
|
||||
"emits_answer": False,
|
||||
}
|
||||
return {
|
||||
"kind": kind,
|
||||
**outcome.as_dict(),
|
||||
}
|
||||
182
generate/linguistic_pipeline/layer_a_english.py
Normal file
182
generate/linguistic_pipeline/layer_a_english.py
Normal file
|
|
@ -0,0 +1,182 @@
|
|||
"""Layer A — English surface constraint extraction (candidates only).
|
||||
|
||||
Forbidden: selecting an arithmetic operation; fabricating omitted context.
|
||||
Ambiguity is preserved — no premature collapse to a single parse.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
from typing import Sequence
|
||||
|
||||
from core.semantic_primitives import ProvenanceSpan, ValidationError
|
||||
|
||||
_ENTITY_RE = re.compile(
|
||||
r"\b([A-Z][a-z]+(?:\s+[A-Z][a-z]+)?)\b"
|
||||
)
|
||||
_NUMBER_RE = re.compile(
|
||||
r"(?P<num>\d+(?:\.\d+)?)\s*(?P<unit>apples?|oranges?|dollars?|hours?|kg|items?|books?)?",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_RELATION_CUES: tuple[tuple[str, str], ...] = (
|
||||
("gave", "transfer_cue"),
|
||||
("sold", "transfer_cue"),
|
||||
("bought", "transfer_cue"),
|
||||
("has", "possession_cue"),
|
||||
("have", "possession_cue"),
|
||||
("more than", "comparison_cue"),
|
||||
("less than", "comparison_cue"),
|
||||
("each", "partition_cue"),
|
||||
("per", "rate_cue"),
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class CandidateEntity:
|
||||
candidate_id: str
|
||||
surface: str
|
||||
confidence: float
|
||||
provenance: ProvenanceSpan
|
||||
kind_hint: str = "entity"
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if not (0.0 <= float(self.confidence) <= 1.0):
|
||||
raise ValidationError("CandidateEntity.confidence must be in [0, 1]")
|
||||
if not self.candidate_id:
|
||||
raise ValidationError("CandidateEntity.candidate_id must be non-empty")
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class CandidateRelation:
|
||||
candidate_id: str
|
||||
cue: str
|
||||
relation_hint: str
|
||||
confidence: float
|
||||
provenance: ProvenanceSpan
|
||||
left_surface: str = ""
|
||||
right_surface: str = ""
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if not (0.0 <= float(self.confidence) <= 1.0):
|
||||
raise ValidationError("CandidateRelation.confidence must be in [0, 1]")
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class NumericToken:
|
||||
token_id: str
|
||||
raw: str
|
||||
value_text: str
|
||||
unit: str
|
||||
confidence: float
|
||||
provenance: ProvenanceSpan
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class SurfaceConstraintSet:
|
||||
"""Layer A output — constraints and candidates, never a chosen op."""
|
||||
|
||||
source_text: str
|
||||
entities: tuple[CandidateEntity, ...]
|
||||
relations: tuple[CandidateRelation, ...]
|
||||
numerics: tuple[NumericToken, ...]
|
||||
# When multiple relation cues apply, all are retained (ambiguity preserved).
|
||||
ambiguity_notes: tuple[str, ...] = ()
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
for e in self.entities:
|
||||
if not isinstance(e, CandidateEntity):
|
||||
raise ValidationError("entities must be CandidateEntity instances")
|
||||
for r in self.relations:
|
||||
if not isinstance(r, CandidateRelation):
|
||||
raise ValidationError("relations must be CandidateRelation instances")
|
||||
|
||||
|
||||
def extract_english_surface(text: str) -> SurfaceConstraintSet:
|
||||
"""Extract surface candidates without selecting arithmetic operations."""
|
||||
if not isinstance(text, str):
|
||||
raise ValidationError("Layer A input must be a str")
|
||||
source = text
|
||||
entities: list[CandidateEntity] = []
|
||||
seen_ent: set[str] = set()
|
||||
for i, m in enumerate(_ENTITY_RE.finditer(source)):
|
||||
surface = m.group(1)
|
||||
# Skip sentence-initial function words that look capitalized mid-stream poorly
|
||||
if surface.lower() in {"the", "a", "an", "if", "when"}:
|
||||
continue
|
||||
key = surface.lower()
|
||||
if key in seen_ent:
|
||||
continue
|
||||
seen_ent.add(key)
|
||||
span = ProvenanceSpan(start=m.start(1), end=m.end(1), text=surface)
|
||||
entities.append(
|
||||
CandidateEntity(
|
||||
candidate_id=f"ent:{i}:{key}",
|
||||
surface=surface,
|
||||
confidence=0.7,
|
||||
provenance=span,
|
||||
kind_hint="entity",
|
||||
)
|
||||
)
|
||||
|
||||
numerics: list[NumericToken] = []
|
||||
for i, m in enumerate(_NUMBER_RE.finditer(source)):
|
||||
unit = (m.group("unit") or "").lower()
|
||||
raw = m.group(0).strip()
|
||||
span = ProvenanceSpan(start=m.start(), end=m.end(), text=raw)
|
||||
numerics.append(
|
||||
NumericToken(
|
||||
token_id=f"num:{i}",
|
||||
raw=raw,
|
||||
value_text=m.group("num"),
|
||||
unit=unit or "count",
|
||||
confidence=0.9 if unit else 0.75,
|
||||
provenance=span,
|
||||
)
|
||||
)
|
||||
# Unit nouns are entity candidates (items possessed / transferred).
|
||||
if unit:
|
||||
key = unit.rstrip("s")
|
||||
if key not in seen_ent:
|
||||
seen_ent.add(key)
|
||||
entities.append(
|
||||
CandidateEntity(
|
||||
candidate_id=f"ent:unit:{i}:{key}",
|
||||
surface=unit,
|
||||
confidence=0.8,
|
||||
provenance=span,
|
||||
kind_hint="quantity_item",
|
||||
)
|
||||
)
|
||||
|
||||
relations: list[CandidateRelation] = []
|
||||
lower = source.lower()
|
||||
for i, (cue, hint) in enumerate(_RELATION_CUES):
|
||||
idx = lower.find(cue)
|
||||
if idx < 0:
|
||||
continue
|
||||
span = ProvenanceSpan(start=idx, end=idx + len(cue), text=source[idx : idx + len(cue)])
|
||||
relations.append(
|
||||
CandidateRelation(
|
||||
candidate_id=f"rel:{i}:{hint}",
|
||||
cue=cue,
|
||||
relation_hint=hint,
|
||||
confidence=0.65,
|
||||
provenance=span,
|
||||
)
|
||||
)
|
||||
|
||||
notes: list[str] = []
|
||||
transferish = [r for r in relations if r.relation_hint == "transfer_cue"]
|
||||
if len(transferish) > 1:
|
||||
notes.append("multiple_transfer_cues_retained")
|
||||
if len(relations) > 1:
|
||||
notes.append("multi_relation_candidates_retained")
|
||||
|
||||
return SurfaceConstraintSet(
|
||||
source_text=source,
|
||||
entities=tuple(entities),
|
||||
relations=tuple(relations),
|
||||
numerics=tuple(numerics),
|
||||
ambiguity_notes=tuple(notes),
|
||||
)
|
||||
168
generate/linguistic_pipeline/layer_b_hebrew.py
Normal file
168
generate/linguistic_pipeline/layer_b_hebrew.py
Normal file
|
|
@ -0,0 +1,168 @@
|
|||
"""Layer B — Hebrew-inspired event classification (constraint candidates only).
|
||||
|
||||
Root labels never prove the intended operation alone. Multi-class roots retain
|
||||
AmbiguityManifold entries; the field resolves.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
|
||||
from core.semantic_primitives import (
|
||||
AmbiguityManifold,
|
||||
Event,
|
||||
Operator,
|
||||
OperatorClass,
|
||||
TemporalExtent,
|
||||
TemporalFrame,
|
||||
TemporalKind,
|
||||
ValidationError,
|
||||
)
|
||||
from generate.linguistic_pipeline.layer_a_english import (
|
||||
CandidateEntity,
|
||||
CandidateRelation,
|
||||
SurfaceConstraintSet,
|
||||
)
|
||||
|
||||
# Authored multi-class roots: surface/root cue → possible operator classes.
|
||||
# Multiple classes ⇒ AmbiguityManifold (no collapse).
|
||||
_MULTI_CLASS_ROOTS: dict[str, tuple[OperatorClass, ...]] = {
|
||||
# Hebrew-inspired: מכר / sell can be transfer or removal depending on frame
|
||||
"sold": (OperatorClass.TRANSFER, OperatorClass.REMOVAL),
|
||||
"mkr": (OperatorClass.TRANSFER, OperatorClass.REMOVAL),
|
||||
"gave": (OperatorClass.TRANSFER, OperatorClass.REMOVAL),
|
||||
"ntn": (OperatorClass.TRANSFER, OperatorClass.REMOVAL),
|
||||
# single-class examples
|
||||
"bought": (OperatorClass.TRANSFER,),
|
||||
"earned": (OperatorClass.ACCUMULATION,),
|
||||
"lost": (OperatorClass.REMOVAL,),
|
||||
"made": (OperatorClass.CREATION,),
|
||||
"shared": (OperatorClass.PARTITION,),
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class EventOperatorCandidate:
|
||||
candidate_id: str
|
||||
operator_class: OperatorClass
|
||||
participants: tuple[str, ...]
|
||||
direction: str
|
||||
conserved_quantity_hint: str
|
||||
source: str
|
||||
target: str
|
||||
temporal_extent_kind: TemporalKind
|
||||
root_cue: str
|
||||
confidence: float
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class EventOperatorSet:
|
||||
events: tuple[Event, ...]
|
||||
operators: tuple[Operator, ...]
|
||||
candidates: tuple[EventOperatorCandidate, ...]
|
||||
manifolds: tuple[AmbiguityManifold, ...]
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
for e in self.events:
|
||||
if not isinstance(e, Event):
|
||||
raise ValidationError("EventOperatorSet.events must be Event instances")
|
||||
|
||||
|
||||
def _entity_ids(entities: tuple[CandidateEntity, ...]) -> tuple[str, ...]:
|
||||
return tuple(e.candidate_id for e in entities)
|
||||
|
||||
|
||||
def classify_hebrew_events(surface: SurfaceConstraintSet) -> EventOperatorSet:
|
||||
"""Map Layer A candidates → event operator candidates; preserve multi-class roots."""
|
||||
if not isinstance(surface, SurfaceConstraintSet):
|
||||
raise ValidationError("Layer B requires SurfaceConstraintSet")
|
||||
|
||||
participants = _entity_ids(surface.entities)
|
||||
candidates: list[EventOperatorCandidate] = []
|
||||
manifolds: list[AmbiguityManifold] = []
|
||||
events: list[Event] = []
|
||||
operators: list[Operator] = []
|
||||
|
||||
cues: list[tuple[str, CandidateRelation | None]] = []
|
||||
for rel in surface.relations:
|
||||
cues.append((rel.cue.lower(), rel))
|
||||
# Also scan free text for multi-class root tokens not captured as relations
|
||||
lower = surface.source_text.lower()
|
||||
for root in _MULTI_CLASS_ROOTS:
|
||||
if root in lower and not any(c[0] == root for c in cues):
|
||||
cues.append((root, None))
|
||||
|
||||
seen_roots: set[str] = set()
|
||||
for idx, (cue, rel) in enumerate(cues):
|
||||
if cue in seen_roots:
|
||||
continue
|
||||
classes = _MULTI_CLASS_ROOTS.get(cue)
|
||||
if classes is None:
|
||||
continue
|
||||
seen_roots.add(cue)
|
||||
source = participants[0] if participants else ""
|
||||
target = participants[1] if len(participants) > 1 else ""
|
||||
cand_ids: list[str] = []
|
||||
for j, op_class in enumerate(classes):
|
||||
cid = f"he_ev:{idx}:{j}:{op_class.value}"
|
||||
cand_ids.append(cid)
|
||||
candidates.append(
|
||||
EventOperatorCandidate(
|
||||
candidate_id=cid,
|
||||
operator_class=op_class,
|
||||
participants=participants,
|
||||
direction="source_to_target" if op_class is OperatorClass.TRANSFER else "egress",
|
||||
conserved_quantity_hint="quantity:primary",
|
||||
source=source,
|
||||
target=target,
|
||||
temporal_extent_kind=TemporalKind.PRIOR_COMPLETED,
|
||||
root_cue=cue,
|
||||
confidence=0.6 if len(classes) > 1 else 0.8,
|
||||
)
|
||||
)
|
||||
# Events are incomplete when multi-class — still emit typed Event
|
||||
# with unresolved operator class alternatives in manifold.
|
||||
ev = Event(
|
||||
event_id=cid,
|
||||
operator_class=op_class,
|
||||
agent_entity_id=source or None,
|
||||
patient_entity_id=target or None,
|
||||
source_entity_id=source or None,
|
||||
target_entity_id=target or None,
|
||||
conserved_quantity_id="quantity:primary" if surface.numerics else None,
|
||||
temporal_extent=TemporalExtent(
|
||||
frame=TemporalFrame(
|
||||
frame_id="frame:prior",
|
||||
kind=TemporalKind.PRIOR_COMPLETED,
|
||||
)
|
||||
),
|
||||
direction="source_to_target",
|
||||
unresolved_roles=() if source and target else ("source", "target"),
|
||||
)
|
||||
events.append(ev)
|
||||
operators.append(
|
||||
Operator(
|
||||
operator_id=f"op:{cid}",
|
||||
operator_class=op_class,
|
||||
event_id=cid,
|
||||
executable_symbol=op_class.value,
|
||||
)
|
||||
)
|
||||
if len(classes) > 1:
|
||||
manifolds.append(
|
||||
AmbiguityManifold(
|
||||
manifold_id=f"he_root:{cue}",
|
||||
candidate_ids=tuple(cand_ids),
|
||||
candidate_kinds=tuple(c.value for c in classes),
|
||||
resolution_condition=(
|
||||
"field_unique_admissible_operator_under_relation_graph"
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
return EventOperatorSet(
|
||||
events=tuple(events),
|
||||
operators=tuple(operators),
|
||||
candidates=tuple(candidates),
|
||||
manifolds=tuple(manifolds),
|
||||
)
|
||||
147
generate/linguistic_pipeline/layer_c_koine.py
Normal file
147
generate/linguistic_pipeline/layer_c_koine.py
Normal file
|
|
@ -0,0 +1,147 @@
|
|||
"""Layer C — Koine-inspired relation-time binding.
|
||||
|
||||
Forbidden: guessing absent antecedents, owners, units, or time frames.
|
||||
Absent referents → typed MissingReferent.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
|
||||
from core.semantic_primitives import (
|
||||
MissingReferent,
|
||||
Relation,
|
||||
RelationKind,
|
||||
TemporalFrame,
|
||||
TemporalKind,
|
||||
ValidationError,
|
||||
)
|
||||
from generate.linguistic_pipeline.layer_a_english import SurfaceConstraintSet
|
||||
from generate.linguistic_pipeline.layer_b_hebrew import EventOperatorSet
|
||||
|
||||
_CUE_TO_RELATION: dict[str, RelationKind] = {
|
||||
"has": RelationKind.POSSESSION,
|
||||
"have": RelationKind.POSSESSION,
|
||||
"gave": RelationKind.SOURCE,
|
||||
"sold": RelationKind.SOURCE,
|
||||
"bought": RelationKind.TARGET,
|
||||
"more than": RelationKind.COMPARISON,
|
||||
"less than": RelationKind.COMPARISON,
|
||||
"each": RelationKind.MEMBERSHIP,
|
||||
"per": RelationKind.RATIO,
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class RelationGraph:
|
||||
relations: tuple[Relation, ...]
|
||||
missing_referents: tuple[MissingReferent, ...]
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class TemporalTopology:
|
||||
frames: tuple[TemporalFrame, ...]
|
||||
bindings: tuple[tuple[str, str], ...] # (event_or_entity_id, frame_id)
|
||||
|
||||
|
||||
def bind_koine_relation_time(
|
||||
surface: SurfaceConstraintSet,
|
||||
events: EventOperatorSet,
|
||||
) -> tuple[RelationGraph, TemporalTopology]:
|
||||
if not isinstance(surface, SurfaceConstraintSet):
|
||||
raise ValidationError("Layer C requires SurfaceConstraintSet")
|
||||
if not isinstance(events, EventOperatorSet):
|
||||
raise ValidationError("Layer C requires EventOperatorSet")
|
||||
|
||||
ent_ids = [e.candidate_id for e in surface.entities]
|
||||
relations: list[Relation] = []
|
||||
missing: list[MissingReferent] = []
|
||||
|
||||
for i, rel in enumerate(surface.relations):
|
||||
kind = _CUE_TO_RELATION.get(rel.cue.lower(), RelationKind.OTHER)
|
||||
left = ent_ids[0] if ent_ids else ""
|
||||
right = ent_ids[1] if len(ent_ids) > 1 else ""
|
||||
if not left:
|
||||
missing.append(
|
||||
MissingReferent(
|
||||
referent_id=f"miss:left:{i}",
|
||||
role="left",
|
||||
expected_kind="entity",
|
||||
context=rel.cue,
|
||||
)
|
||||
)
|
||||
continue
|
||||
# Possession may bind to a quantity-item entity when only one person
|
||||
# name is present (right = first quantity_item entity if any).
|
||||
if not right and kind is RelationKind.POSSESSION:
|
||||
item_ids = [
|
||||
e.candidate_id
|
||||
for e in surface.entities
|
||||
if e.kind_hint == "quantity_item"
|
||||
]
|
||||
if item_ids:
|
||||
right = item_ids[0]
|
||||
if not right and kind in {
|
||||
RelationKind.COMPARISON,
|
||||
RelationKind.POSSESSION,
|
||||
RelationKind.SOURCE,
|
||||
RelationKind.TARGET,
|
||||
}:
|
||||
missing.append(
|
||||
MissingReferent(
|
||||
referent_id=f"miss:right:{i}",
|
||||
role="right",
|
||||
expected_kind="entity",
|
||||
context=rel.cue,
|
||||
)
|
||||
)
|
||||
# Do not fabricate right — skip emitting a filled relation
|
||||
continue
|
||||
if not right:
|
||||
missing.append(
|
||||
MissingReferent(
|
||||
referent_id=f"miss:right:{i}",
|
||||
role="right",
|
||||
expected_kind="entity",
|
||||
context=rel.cue,
|
||||
)
|
||||
)
|
||||
continue
|
||||
relations.append(
|
||||
Relation(
|
||||
relation_id=f"grc_rel:{i}",
|
||||
kind=kind,
|
||||
left_entity_id=left,
|
||||
right_entity_id=right,
|
||||
polarity=True,
|
||||
provenance=rel.provenance,
|
||||
)
|
||||
)
|
||||
|
||||
# Events without agent/source → missing referent
|
||||
for ev in events.events:
|
||||
if ev.agent_entity_id is None and "agent" in ev.unresolved_roles:
|
||||
missing.append(
|
||||
MissingReferent(
|
||||
referent_id=f"miss:agent:{ev.event_id}",
|
||||
role="agent",
|
||||
expected_kind="entity",
|
||||
context=ev.event_id,
|
||||
)
|
||||
)
|
||||
|
||||
frames = (
|
||||
TemporalFrame(frame_id="frame:present", kind=TemporalKind.PRESENT),
|
||||
TemporalFrame(frame_id="frame:prior", kind=TemporalKind.PRIOR_COMPLETED),
|
||||
)
|
||||
bindings: list[tuple[str, str]] = []
|
||||
for cand in events.candidates:
|
||||
if cand.temporal_extent_kind is TemporalKind.PRIOR_COMPLETED:
|
||||
bindings.append((cand.candidate_id, "frame:prior"))
|
||||
else:
|
||||
bindings.append((cand.candidate_id, "frame:present"))
|
||||
|
||||
return (
|
||||
RelationGraph(relations=tuple(relations), missing_referents=tuple(missing)),
|
||||
TemporalTopology(frames=frames, bindings=tuple(bindings)),
|
||||
)
|
||||
104
generate/linguistic_pipeline/pipeline.py
Normal file
104
generate/linguistic_pipeline/pipeline.py
Normal file
|
|
@ -0,0 +1,104 @@
|
|||
"""End-to-end linguistic constraint → field → articulation entry path."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
from core.cognition.fail_closed import CoherenceRefusal
|
||||
from generate.linguistic_pipeline.articulation import (
|
||||
ArticulationFirewallVerdict,
|
||||
articulate_from_proof,
|
||||
)
|
||||
from generate.linguistic_pipeline.field_integration import (
|
||||
AmbiguousFieldState,
|
||||
CoherentFieldState,
|
||||
FieldOutcome,
|
||||
integrate_constraints,
|
||||
outcome_as_dict,
|
||||
outcome_kind,
|
||||
)
|
||||
from generate.linguistic_pipeline.layer_a_english import (
|
||||
SurfaceConstraintSet,
|
||||
extract_english_surface,
|
||||
)
|
||||
from generate.linguistic_pipeline.layer_b_hebrew import (
|
||||
EventOperatorSet,
|
||||
classify_hebrew_events,
|
||||
)
|
||||
from generate.linguistic_pipeline.layer_c_koine import (
|
||||
RelationGraph,
|
||||
TemporalTopology,
|
||||
bind_koine_relation_time,
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class LinguisticPipelineResult:
|
||||
surface_constraints: SurfaceConstraintSet
|
||||
event_operators: EventOperatorSet
|
||||
relation_graph: RelationGraph
|
||||
temporal_topology: TemporalTopology
|
||||
field_outcome: FieldOutcome
|
||||
articulation: ArticulationFirewallVerdict | CoherenceRefusal
|
||||
|
||||
@property
|
||||
def outcome_kind(self) -> str:
|
||||
return outcome_kind(self.field_outcome)
|
||||
|
||||
def as_dict(self) -> dict[str, Any]:
|
||||
art: dict[str, Any]
|
||||
if isinstance(self.articulation, CoherenceRefusal):
|
||||
art = self.articulation.as_dict()
|
||||
else:
|
||||
art = {
|
||||
"surface": self.articulation.surface,
|
||||
"hallucinated": list(self.articulation.hallucinated),
|
||||
"clean": self.articulation.clean,
|
||||
"claims": [c.text for c in self.articulation.legal_claims],
|
||||
}
|
||||
return {
|
||||
"outcome_kind": self.outcome_kind,
|
||||
"field": outcome_as_dict(self.field_outcome),
|
||||
"articulation": art,
|
||||
"missing_referents": [
|
||||
m.referent_id for m in self.relation_graph.missing_referents
|
||||
],
|
||||
"manifolds": [m.manifold_id for m in self.event_operators.manifolds],
|
||||
}
|
||||
|
||||
|
||||
def run_linguistic_pipeline(
|
||||
text: str,
|
||||
*,
|
||||
force_geometry_fail: bool = False,
|
||||
) -> LinguisticPipelineResult:
|
||||
"""Shipped public entry: English text → typed outcomes only."""
|
||||
surface = extract_english_surface(text)
|
||||
events = classify_hebrew_events(surface)
|
||||
relation_graph, temporal = bind_koine_relation_time(surface, events)
|
||||
outcome = integrate_constraints(
|
||||
surface,
|
||||
events,
|
||||
relation_graph,
|
||||
temporal,
|
||||
force_geometry_fail=force_geometry_fail,
|
||||
)
|
||||
articulation = articulate_from_proof(outcome)
|
||||
return LinguisticPipelineResult(
|
||||
surface_constraints=surface,
|
||||
event_operators=events,
|
||||
relation_graph=relation_graph,
|
||||
temporal_topology=temporal,
|
||||
field_outcome=outcome,
|
||||
articulation=articulation,
|
||||
)
|
||||
|
||||
|
||||
__all__ = [
|
||||
"LinguisticPipelineResult",
|
||||
"run_linguistic_pipeline",
|
||||
"CoherentFieldState",
|
||||
"AmbiguousFieldState",
|
||||
"CoherenceRefusal",
|
||||
]
|
||||
409
tests/test_linguistic_governance_phases.py
Normal file
409
tests/test_linguistic_governance_phases.py
Normal file
|
|
@ -0,0 +1,409 @@
|
|||
"""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
|
||||
|
|
@ -38,12 +38,14 @@ def test_runtime_canonical_surface_has_base_precedence() -> None:
|
|||
pre_decoration_surface="pre-decoration",
|
||||
response_surface="runtime",
|
||||
response_articulation_surface="articulation",
|
||||
contract_assessment=_closed_assessment(),
|
||||
)
|
||||
|
||||
assert resolved.surface == "canonical"
|
||||
assert resolved.articulation_surface == "articulation"
|
||||
assert resolved.authority == "runtime_canonical"
|
||||
assert resolved.fold_sources == ()
|
||||
assert resolved.authoritative is True
|
||||
|
||||
|
||||
def test_useful_realizer_requires_conjugate_coherence() -> None:
|
||||
|
|
@ -64,7 +66,7 @@ def test_useful_realizer_requires_conjugate_coherence() -> None:
|
|||
|
||||
|
||||
def test_realizer_shim_refused_when_conjugate_open() -> None:
|
||||
"""Failed geometric residual must not fall back to realizer authority."""
|
||||
"""Failed geometric residual → typed abstention (no fluent runtime answer)."""
|
||||
resolved = resolve_surface(
|
||||
response_surface="runtime",
|
||||
response_articulation_surface="runtime articulation",
|
||||
|
|
@ -73,11 +75,15 @@ def test_realizer_shim_refused_when_conjugate_open() -> None:
|
|||
gate_fired=False,
|
||||
contract_assessment=_open_assessment(),
|
||||
)
|
||||
assert resolved.authority == "runtime"
|
||||
assert resolved.surface == "runtime"
|
||||
assert resolved.authority == "coherence_abstention"
|
||||
assert resolved.authoritative is False
|
||||
assert resolved.refusal is not None
|
||||
assert "versor_condition" in resolved.refusal.violated_condition
|
||||
assert "runtime" not in resolved.surface
|
||||
|
||||
|
||||
def test_gate_fired_keeps_runtime_surface_even_when_realizer_is_useful() -> None:
|
||||
# Closed assessment + gate_fired: substrate blocked; runtime kept (geometry closed).
|
||||
resolved = resolve_surface(
|
||||
response_surface="runtime refusal",
|
||||
response_articulation_surface="runtime refusal articulation",
|
||||
|
|
@ -124,7 +130,11 @@ def test_walk_and_compose_fold_after_selected_authority() -> None:
|
|||
|
||||
|
||||
def test_folds_stand_alone_when_base_surface_is_empty() -> None:
|
||||
resolved = resolve_surface(walk_surface="walk chain", compose_surface="compose transfer")
|
||||
resolved = resolve_surface(
|
||||
walk_surface="walk chain",
|
||||
compose_surface="compose transfer",
|
||||
contract_assessment=_closed_assessment(),
|
||||
)
|
||||
|
||||
assert resolved.surface == "walk chain — compose transfer"
|
||||
assert resolved.articulation_surface == "walk chain — compose transfer"
|
||||
|
|
@ -173,7 +183,7 @@ def test_substrate_supreme_requires_forward_and_conjugate() -> None:
|
|||
|
||||
|
||||
def test_substrate_refused_without_assessment() -> None:
|
||||
"""Assessment=None fails conjugate competitor (fail-closed)."""
|
||||
"""Assessment=None → typed ContractViolation; no certified answer."""
|
||||
g = _mk_grounded_graph()
|
||||
resolved = resolve_surface(
|
||||
response_surface="runtime",
|
||||
|
|
@ -184,7 +194,11 @@ def test_substrate_refused_without_assessment() -> None:
|
|||
proposition_graph=g,
|
||||
contract_assessment=None,
|
||||
)
|
||||
assert resolved.authority == "runtime"
|
||||
assert resolved.authority == "coherence_abstention"
|
||||
assert resolved.authoritative is False
|
||||
assert resolved.contract_violation is not None
|
||||
assert resolved.contract_violation.violated_condition == "contract_assessment_present"
|
||||
assert "runtime" not in resolved.surface.lower() or "cannot certify" in resolved.surface.lower()
|
||||
|
||||
|
||||
def test_pending_graph_withholds_substrate_even_if_conjugate_ok() -> None:
|
||||
|
|
@ -213,8 +227,9 @@ def test_open_geometric_contract_refuses_substrate_and_realizer() -> None:
|
|||
proposition_graph=g,
|
||||
contract_assessment=_open_assessment(),
|
||||
)
|
||||
assert resolved.authority == "runtime"
|
||||
assert resolved.surface == "runtime"
|
||||
assert resolved.authority == "coherence_abstention"
|
||||
assert resolved.authoritative is False
|
||||
assert resolved.refusal is not None
|
||||
|
||||
|
||||
def test_gate_fired_still_blocks_substrate_even_for_grounded_graph() -> None:
|
||||
|
|
@ -243,3 +258,16 @@ def test_dual_competitors_helpers() -> None:
|
|||
assert _conjugate_coherence_ok(open_a) is False
|
||||
assert _substrate_supreme(g, open_a) is False
|
||||
assert _forward_surface_ok(_mk_pending_graph(), closed) is False
|
||||
|
||||
|
||||
def test_require_closed_geometry_false_preserves_legacy_runtime_on_open() -> None:
|
||||
"""Escape hatch for non-answer-authority callers only."""
|
||||
resolved = resolve_surface(
|
||||
response_surface="legacy runtime",
|
||||
realized_surface="realizer",
|
||||
realizer_useful=True,
|
||||
contract_assessment=_open_assessment(),
|
||||
require_closed_geometry=False,
|
||||
)
|
||||
assert resolved.authority == "runtime"
|
||||
assert resolved.surface == "legacy runtime"
|
||||
|
|
|
|||
Loading…
Reference in a new issue