core-labs/core
core-labs/core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

feat(cognition): fail-closed linguistic governance + trilingual constraint pipeline

Browse source 
Close residual answer-authority debt: typed CoherenceRefusal/ContractViolation/
FieldFailure, structured ProofTrace (atoms→operators→closure), and Shadow Gate
paths that refuse certified answers when contract_assessment is None or geometry
is open. Add shared semantic primitives and Layers A/B/C (English/Hebrew/Koine)
as constraint producers only, with Cl(4,1) structure-sensitive embedding (no
unitize in generate/), three field outcomes, and an articulation firewall that
blocks payload-value citation bypass and uncertified content.
This commit is contained in:
Shay 2026-07-20 16:16:56 -07:00
parent 18c578d960
commit e0d1b4754a
14 changed files with 2852 additions and 19 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

256
core/cognition/fail_closed.py Normal file
View 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
View 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",
]

157
core/cognition/surface_resolution.py
View file

@ -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
View 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
View 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
View 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
View 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
View 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
View 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
View 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
View 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
View 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
View 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

44
tests/test_surface_resolution.py
View file

@ -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"