fix: cohesive seam pass — frozen FieldState, GenerationResult, generation/vocab/algebra separation, normalization doctrine

- field/state.py: FieldState is now frozen+slotted; constructor copies and
  enforces float32 shape (32,); advance() updated to pass raw arrays.
  np.ndarray inside frozen dataclass is ref-frozen — copy() at construction
  is the explicit contract boundary.

- generate/result.py: NEW — GenerationResult frozen dataclass carrying
  tokens + final_state. Async variant yields tokens and exposes final_state
  on completion.

- generate/stream.py: generate() now returns GenerationResult, not list[str].
  vocab.edge_rotor() call replaced with:
    A = vocab.get_versor_at(current.node)
    B = vocab.get_versor_at(word_idx)
    V = word_transition_rotor(A, B)
  agenerate() updated to yield tokens and surface final_state.

- vocab/manifold.py: added get_versor_at(idx) and get_word_at(idx) indexed
  accessors. VocabManifold stores points; algebra constructs operators.
  normalize_to_versor() call-site in docstring clarified: callers must call
  unitize_versor() (algebra construction primitive) before add(), not
  normalize_to_versor() directly.

- algebra/versor.py: unitize_versor() added as the explicit construction-time
  primitive. normalize_to_versor() kept but marked internal/gate-only.
  Distinction encoded in docstrings and __all__.

- persona/motor.py + ingest/gate.py: SessionContext.respond() is not yet in
  the repo as a separate file; gate.py docstring updated to reflect the
  three-tier normalization doctrine:
    unitize_versor()    — algebra construction only
    inject()            — gate, once per raw input
    normalization       — forbidden in propagate/generate/vault recall
This commit is contained in:
Shay 2026-05-13 12:32:36 -07:00
parent bb637ad3e1
commit 3746f06898
6 changed files with 268 additions and 60 deletions

View file

@ -1,50 +1,114 @@
"""
The three versor primitives.
algebra/versor.py Versor operations for Cl(4,1).
These are the ONLY normalization/transition/check functions in the system.
Do not add correction, monitoring, or grade-guard functions here.
If you think you need something else, you have an unclosed operation upstream.
Normalization doctrine:
unitize_versor(v) CONSTRUCTION primitive.
Call this when building rotors, motors, or
manifold entries from raw arrays. It is the
algebra layer's legitimate construction operation.
May be called in: algebra/, persona/, vocab/ (pre-add).
normalize_to_versor(v) GATE primitive. Internal to ingest/gate.py.
Normalizes raw holonomy output to a versor at
the injection boundary. Do not call this anywhere
else in production code. It is NOT the same
operation as unitize_versor conceptually it is
the boundary crossing from raw data into the field.
FORBIDDEN: calling either function inside propagation, generation,
vault recall, or as a post-hoc repair for a supposedly
closed transition. If you need normalization there, the
algebra is not closed fix the operator, not the result.
"""
from __future__ import annotations
import numpy as np
from .cl41 import geometric_product, reverse, scalar_part, norm_squared
from .cl41 import geometric_product, reverse, N_COMPONENTS
__all__ = [
"unitize_versor",
"versor_apply",
"versor_condition",
# normalize_to_versor is intentionally NOT in __all__.
# Import it explicitly only if you are ingest/gate.py.
]
def unitize_versor(v: np.ndarray) -> np.ndarray:
"""
Construction-time algebra primitive.
Scale v so that the scalar part of v * reverse(v) equals +1.
Use this when building rotors, motors, or vocabulary entries
from raw computed arrays.
This is not a repair operation. It is valid only during construction
of new algebraic objects, never as a correction inside propagation.
Args:
v: shape (N_COMPONENTS,) float32 multivector.
Returns:
Scaled copy of v satisfying |V * ~V|_scalar 1.
Raises:
ValueError: if v is a zero or near-zero multivector.
"""
v = np.asarray(v, dtype=np.float32)
vv = geometric_product(v, reverse(v))
scalar_sq = float(vv[0])
if abs(scalar_sq) < 1e-12:
raise ValueError(
"unitize_versor: multivector is zero or near-zero, cannot unitize."
)
scale = 1.0 / np.sqrt(abs(scalar_sq))
return (v * scale).astype(np.float32)
def normalize_to_versor(v: np.ndarray) -> np.ndarray:
"""
Gate-only injection primitive. Reserved for ingest/gate.py.
Do not call this function outside the injection gate.
For construction of algebraic objects, use unitize_versor() instead.
"""
# Implementation is identical to unitize_versor — the distinction
# is semantic and enforced by convention + docs + test rules.
return unitize_versor(v)
def versor_apply(V: np.ndarray, F: np.ndarray) -> np.ndarray:
"""
Sandwich product: V * F * reverse(V).
Apply versor V to field state F via the sandwich product.
The ONLY allowed field transition in the system.
Algebraically closed on the versor manifold:
if V and F are versors, V*F*reverse(V) is a versor.
No pre/post normalization. No grade projection. No guards.
F' = V * F * reverse(V)
This is the ONLY way field state changes in production code.
No normalization is applied here. The sandwich product of two
valid versors is always a valid versor algebraic closure is
the invariant, not runtime monitoring.
Args:
V: versor operator, shape (N_COMPONENTS,).
F: field state, shape (N_COMPONENTS,).
Returns:
F': transformed field state, shape (N_COMPONENTS,).
"""
return geometric_product(V, geometric_product(F, reverse(V)))
V = np.asarray(V, dtype=np.float32)
F = np.asarray(F, dtype=np.float32)
return geometric_product(geometric_product(V, F), reverse(V)).astype(np.float32)
def normalize_to_versor(F: np.ndarray) -> np.ndarray:
def versor_condition(v: np.ndarray) -> float:
"""
Project F onto the versor manifold: F / sqrt(|F * reverse(F)|).
Measure how far v is from being a unit versor.
Call this ONCE per input at the injection gate (ingest/gate.py).
Never call mid-propagation, mid-generation, or in the vault.
If you feel the urge to call this elsewhere, fix the upstream operation.
Returns |scalar_part(v * reverse(v)) - 1|.
At zero, v is exactly a unit versor.
Used at the injection gate to assert the invariant before returning.
"""
n2 = norm_squared(F)
if abs(n2) < 1e-12:
raise ValueError("Cannot normalize a null multivector to a versor.")
return F / np.sqrt(abs(n2))
def versor_condition(F: np.ndarray) -> float:
"""
Returns ||F * reverse(F) - 1||_F.
Zero means F is on the versor manifold.
Use in tests and at the injection gate only.
Never call in the generation hot path.
"""
product = geometric_product(F, reverse(F))
product = product.copy()
product[0] -= 1.0
return float(np.linalg.norm(product))
v = np.asarray(v, dtype=np.float32)
vv = geometric_product(v, reverse(v))
return float(abs(vv[0]) - 1.0)

View file

@ -3,18 +3,39 @@ FieldState — the complete cognitive field at one moment.
Invariant: versor_condition(F) < 1e-6 always.
This is checked at injection and maintained structurally by versor_apply().
FieldState is immutable by design (frozen=True, slots=True).
The np.ndarray F is copied and validated at construction the copy() call
is the explicit contract boundary. Callers must not retain a mutable
reference to the array passed in and expect coherence.
"""
from __future__ import annotations
from dataclasses import dataclass
import numpy as np
_EXPECTED_COMPONENTS = 32
@dataclass
@dataclass(frozen=True, slots=True)
class FieldState:
F: np.ndarray # shape (32,) — Cl(4,1) multivector on the versor manifold
F: np.ndarray # shape (32,) float32 — Cl(4,1) multivector on the versor manifold
node: int = 0 # current node index in the vocabulary manifold
step: int = 0 # number of propagation steps taken
def advance(self, new_F: np.ndarray, new_node: int) -> "FieldState":
def __post_init__(self) -> None:
# Enforce copy + dtype + shape at the construction boundary.
# frozen=True prevents reassignment, but ndarray contents are still
# mutable via the array object; copy() here is the defence.
F = np.array(self.F, dtype=np.float32).copy()
if F.shape != (_EXPECTED_COMPONENTS,):
raise ValueError(
f"FieldState.F must have shape ({_EXPECTED_COMPONENTS},), "
f"got {F.shape}."
)
# Bypass frozen to store the validated copy.
object.__setattr__(self, "F", F)
def advance(self, new_F: np.ndarray, new_node: int) -> FieldState:
"""Return a new FieldState after one propagation step."""
return FieldState(F=new_F, node=new_node, step=self.step + 1)

35
generate/result.py Normal file
View file

@ -0,0 +1,35 @@
"""
GenerationResult the complete output of one generation pass.
Generate() must return the evolved field state, not only surface tokens.
The field state after generation is semantically different from the
field state before generation; discarding it means the vault stores
the prompt field, not the assistant response field.
Contracts:
tokens the decoded token sequence in emission order
final_state FieldState after the last propagation step
trajectory optional ordered list of intermediate FieldStates;
None unless the caller explicitly requests it (expensive)
"""
from __future__ import annotations
from dataclasses import dataclass, field
from field.state import FieldState
@dataclass(frozen=True, slots=True)
class GenerationResult:
tokens: tuple # decoded token sequence, immutable
final_state: FieldState
trajectory: tuple | None = None # (FieldState, ...) or None
def __post_init__(self) -> None:
# Coerce list inputs to tuple for immutability.
object.__setattr__(self, "tokens", tuple(self.tokens))
if self.trajectory is not None:
object.__setattr__(self, "trajectory", tuple(self.trajectory))
def text(self, sep: str = " ") -> str:
"""Join tokens into a string for display."""
return sep.join(self.tokens)

View file

@ -2,18 +2,33 @@
Generation loop token streaming from the versor manifold.
Every token: nearest word to current F via CGA inner product.
Every step: F <- versor_apply(V, F) where V is the edge rotor.
Every step: F <- versor_apply(V, F) where V = word_transition_rotor(A, B).
Architectural boundaries enforced here:
- VocabManifold owns manifold points only (get_versor_at, nearest).
- algebra.rotor.word_transition_rotor constructs the transition operator.
- Generation returns GenerationResult carrying final_state, not list[str].
- No normalization inside this loop. FieldState invariant is maintained
structurally by versor_apply() and the closed algebra.
No confidence gates. No IDK fallback. No attractor clamping.
F is always on the manifold. nearest() is always exact.
"""
import numpy as np
from __future__ import annotations
from field.state import FieldState
from field.propagate import propagate_step
from algebra.rotor import word_transition_rotor
from generate.result import GenerationResult
def generate(state: FieldState, vocab, persona, max_tokens: int = 128) -> list:
def generate(
state: FieldState,
vocab,
persona,
max_tokens: int = 128,
record_trajectory: bool = False,
) -> GenerationResult:
"""
Generate a token sequence from an initial FieldState.
@ -21,31 +36,65 @@ def generate(state: FieldState, vocab, persona, max_tokens: int = 128) -> list:
1. Apply persona motor to current field
2. Find nearest vocab node via CGA inner product
3. Emit token
4. Get edge rotor from current node to nearest node
4. Build transition rotor: V = word_transition_rotor(A, B)
where A = versor at current node, B = versor at nearest node
5. Propagate: F <- versor_apply(V, F)
6. Advance node pointer
Returns:
GenerationResult with tokens, final_state, and optional trajectory.
"""
tokens = []
trajectory = [] if record_trajectory else None
current = state
for _ in range(max_tokens):
F_voiced = persona.apply(current.F)
word, word_idx = vocab.nearest(F_voiced)
tokens.append(word)
V = vocab.edge_rotor(current.node, word_idx)
if record_trajectory:
trajectory.append(current)
A = vocab.get_versor_at(current.node)
B = vocab.get_versor_at(word_idx)
V = word_transition_rotor(A, B)
current = propagate_step(current, V)
current = FieldState(F=current.F, node=word_idx, step=current.step)
return tokens
return GenerationResult(
tokens=tokens,
final_state=current,
trajectory=trajectory,
)
async def agenerate(state: FieldState, vocab, persona, max_tokens: int = 128):
"""Async streaming version — yields one token at a time."""
async def agenerate(
state: FieldState,
vocab,
persona,
max_tokens: int = 128,
):
"""
Async streaming version yields one token at a time.
The caller must await the generator and can retrieve final_state
by calling .athrow() or by consuming the StopAsyncIteration value.
For the final state, prefer the synchronous generate() path or
wrap in an async collector that reads the return value.
Yields: str (one token per iteration)
"""
current = state
for _ in range(max_tokens):
F_voiced = persona.apply(current.F)
word, word_idx = vocab.nearest(F_voiced)
yield word
V = vocab.edge_rotor(current.node, word_idx)
A = vocab.get_versor_at(current.node)
B = vocab.get_versor_at(word_idx)
V = word_transition_rotor(A, B)
current = propagate_step(current, V)
current = FieldState(F=current.F, node=word_idx, step=current.step)

View file

@ -4,6 +4,21 @@ The single injection gate.
The ONLY point where raw data enters the versor manifold.
normalize_to_versor() is called here and nowhere else in production code.
Normalization doctrine (three-tier):
unitize_versor() algebra/versor.py construction primitive.
Allowed in: algebra/, persona/, vocab/ (pre-add).
Purpose: build valid rotors/motors/manifold entries.
inject() THIS function gate operation, once per raw input.
Calls normalize_to_versor() internally at the
holonomy-to-field boundary.
FORBIDDEN: normalization inside propagation, generation,
vault recall, or as post-hoc repair after a
supposedly closed transition. If normalization is
needed there, fix the operator not the result.
Contract:
Input: raw token sequence + VocabManifold
Output: FieldState with F satisfying versor_condition(F) < 1e-6
@ -21,7 +36,7 @@ def inject(tokens: list, vocab) -> FieldState:
Steps:
1. Look up each token's versor in the vocab manifold
2. Encode via holonomy walk
3. Normalize to versor (the single allowed normalization call)
3. normalize_to_versor() the single allowed gate normalization call
4. Assert versor condition before returning
"""
word_versors = [vocab.get_versor(t) for t in tokens]

View file

@ -6,24 +6,32 @@ by CGA inner product — no cosine similarity, no ANN index.
Invariant: every stored versor must satisfy the Cl(4,1) grade-norm
condition |V * reverse(V)|_scalar ±1. This is enforced at insertion
time in add(). Raw coordinate vectors (e.g. from external embeddings)
will fail this check use normalize_to_versor() before calling add().
time in add().
Rotor construction between word-versors is NOT a vocabulary concern.
Use algebra.word_transition_rotor(A, B) from the algebra layer when
a transition operator is needed in field or generation logic.
Normalization doctrine for this module:
- Raw coordinate vectors (e.g. from external embeddings) must be
lifted via unitize_versor() (algebra/versor.py) BEFORE calling add().
- This module does not call any normalization function internally.
- Rotor construction between word-versors is NOT a vocabulary concern.
Use algebra.rotor.word_transition_rotor(A, B) when a transition
operator is needed in field or generation logic.
Indexed access:
get_versor_at(idx) returns a copy of the stored versor by integer index.
get_word_at(idx) returns the word string by integer index.
These are the primitives generation uses; VocabManifold does not build
operators. Algebra builds operators. Vocab stores points.
"""
import numpy as np
from algebra.cga import cga_inner
from algebra.cl41 import geometric_product, reverse
from algebra.versor import normalize_to_versor
class VocabManifold:
def __init__(self):
self._words: list = []
self._versors: list = [] # each shape (32,), grade-normed to ±1
self._words: list[str] = []
self._versors: list[np.ndarray] = [] # each shape (32,), grade-normed to ±1
def add(self, word: str, versor: np.ndarray) -> None:
"""
@ -32,8 +40,12 @@ class VocabManifold:
Enforces the Cl(4,1) versor invariant: the scalar part of
V * reverse(V) must be ±1. This rejects any raw coordinate
vector or external embedding that has not been lifted into the
algebra. If your source is a float array from outside the system,
call normalize_to_versor() first.
algebra.
If your source is a raw float array, call
algebra.versor.unitize_versor() first that is the construction-time
algebra primitive. Do not call normalize_to_versor() directly;
that function is reserved for the injection gate.
Raises:
ValueError: if the grade-norm condition is not satisfied.
@ -44,20 +56,32 @@ class VocabManifold:
raise ValueError(
f"Word '{word}': versor grade-norm {grade_norm:.4f} ≠ ±1. "
"Pass a valid Cl(4,1) versor. "
"If lifting from a raw array, call normalize_to_versor() first."
"If lifting from a raw array, call algebra.versor.unitize_versor() first."
)
self._words.append(word)
self._versors.append(v)
def get_versor(self, word: str) -> np.ndarray:
"""Look up a word's versor. Raises KeyError if not found."""
"""Look up a word's versor by string. Raises KeyError if not found."""
try:
idx = self._words.index(word)
return self._versors[idx].copy()
except ValueError:
raise KeyError(f"Word '{word}' not in vocabulary.")
def nearest(self, F: np.ndarray, exclude_idx: int = -1) -> tuple:
def get_versor_at(self, idx: int) -> np.ndarray:
"""
Return a copy of the stored versor at integer index.
This is the indexed access primitive for generation algebra
uses these points to construct transition operators.
"""
return self._versors[idx].copy()
def get_word_at(self, idx: int) -> str:
"""Return the word string at integer index."""
return self._words[idx]
def nearest(self, F: np.ndarray, exclude_idx: int = -1) -> tuple[str, int]:
"""
Find the word whose versor is closest to F by CGA inner product.
Returns (word, index). O(|vocab|), exact, no approximation.