diff --git a/algebra/rotor.py b/algebra/rotor.py index 2b2a0e2f..9cd50c30 100644 --- a/algebra/rotor.py +++ b/algebra/rotor.py @@ -13,6 +13,23 @@ from .versor import unitize_versor _TRANSITION_BIVECTORS = (6, 7, 9, 10, 12, 14) +def make_rotor_from_angle(angle: float, bivector_idx: int = 6) -> np.ndarray: + """Construct a unit rotor from an angle and bivector component index. + + Compatibility helper for tests and low-level energy propagation checks. + It intentionally builds the same compact scalar+bivector rotor shape used + by the transition constructor and then unitizes it through the canonical + versor primitive. + """ + if not 0 <= int(bivector_idx) < N_COMPONENTS: + raise ValueError(f"bivector_idx out of range: {bivector_idx!r}") + rotor = np.zeros(N_COMPONENTS, dtype=np.float64) + half_angle = float(angle) / 2.0 + rotor[0] = np.cos(half_angle) + rotor[int(bivector_idx)] = np.sin(half_angle) + return unitize_versor(rotor) + + def word_transition_rotor(A: np.ndarray, B: np.ndarray) -> np.ndarray: """ Compute the rotor R that rotates versor A toward versor B in Cl(4,1). diff --git a/chat/runtime.py b/chat/runtime.py index 22b34f62..76185d6b 100644 --- a/chat/runtime.py +++ b/chat/runtime.py @@ -42,17 +42,8 @@ _SEED_ALIASES = { } -# --------------------------------------------------------------------------- -# Helper: safely extract a float from energy — handles EnergyProfile or float -# --------------------------------------------------------------------------- - def _energy_scalar(energy_obj) -> float: - """Return a plain float from a FieldState.energy value. - - FieldState.energy is typed as EnergyProfile | None. Older call sites - passed a raw float as a fallback default; both cases are handled here so - the caller never needs to branch. - """ + """Return a plain float from a FieldState.energy value.""" if energy_obj is None: return 1.0 if isinstance(energy_obj, EnergyProfile): @@ -63,14 +54,6 @@ def _energy_scalar(energy_obj) -> float: return 1.0 -# --------------------------------------------------------------------------- -# Stub BindingFrame for IdentityCheck — allows check() to run without a full -# reasoning pipeline being wired. Carries the minimum contract that -# ReasoningTrajectory.frames requires: frame_id, coherence_magnitude, -# region_ids, cycle_index. -# --------------------------------------------------------------------------- - - @dataclass class _StubBindingFrame: frame_id: str @@ -165,10 +148,7 @@ class ChatRuntime: manifold = manifolds[0] if len(pack_ids) == 1 else load_mounted_packs(pack_ids) self._manifests = tuple(manifests) - # --- Identity manifold (built first; persona motor derived from it) --- self.identity_manifold = _default_identity_manifold() - - # --- Persona motor: non-identity, derived from value_axes directions --- persona_motor = PersonaMotor.from_identity_manifold(self.identity_manifold) self._context = SessionContext( @@ -186,7 +166,6 @@ class ChatRuntime: e.surface: (e.pos or e.part_of_speech or "X") for e in entries } - # --- Physics --- self.exertion_meter = ExertionMeter(capacity_ceiling=128.0) self.drive_gradients = tuple( GradientField(axis=axis, magnitude=0.75) @@ -194,7 +173,6 @@ class ChatRuntime: ) self._drive_map = DriveGradientMap(gradients=self.drive_gradients) - # --- CharacterProfile: populated from live manifold at init --- self.character_profile = CharacterProfile.from_manifold( self.identity_manifold, drive_summaries={ @@ -202,11 +180,7 @@ class ChatRuntime: }, fatigue_index=0.0, ) - - # --- Identity checker --- self._identity_check = IdentityCheck() - - # --- Provenance log: append-only list of TurnEvents --- self.turn_log: List[TurnEvent] = [] @property @@ -332,7 +306,6 @@ class ChatRuntime: inhibition_threshold=self.config.inhibition_threshold, ) - # --- IdentityCheck gate --- reasoning_trajectory = _make_trajectory_from_result(result, self._context.turn) identity_score = self._identity_check.check( reasoning_trajectory, @@ -373,7 +346,6 @@ class ChatRuntime: ) walk_surface = sentence_plan.surface - # Identity flags are telemetry. They must not hide the manifold walk. surface = walk_surface or articulation.surface vault_hits = int(result.vault_hits) @@ -418,26 +390,14 @@ class ChatRuntime: return "" async def achat(self, text: str, max_tokens: int | None = None) -> ChatResponse: - """Async equivalent of chat() — drives agenerate() internally.""" - from generate.stream import agenerate - mt = max_tokens if max_tokens is not None else self.config.max_tokens - tokens: list[str] = [] - async for token in agenerate( - self._context.state, - self._context.vocab, - self._context.persona, - max_tokens=mt, - vault=self._context.vault, - ): - tokens.append(token) - result = self.chat(text, max_tokens=0) - sentence_plan = SentenceAssembler().assemble( - result.articulation, - tokens, - role=result.dialogue_role, - ) - from dataclasses import replace - return replace(result, surface=sentence_plan.surface, walk_surface=sentence_plan.surface) + """Async equivalent of chat(). + + The synchronous chat path owns ingest, drive bias, generation, + identity telemetry, vault storage, and turn-log accounting. Reusing it + keeps async semantics identical while avoiding an uninitialized + SessionContext.state on the first async turn. + """ + return self.chat(text, max_tokens=max_tokens) async def arespond(self, text: str, max_tokens: int | None = None) -> str: """Async equivalent of respond().""" diff --git a/core/physics/identity.py b/core/physics/identity.py index 51d0623a..66c32b7d 100644 --- a/core/physics/identity.py +++ b/core/physics/identity.py @@ -17,6 +17,25 @@ from dataclasses import dataclass from typing import Dict, FrozenSet, List, Optional, Tuple +@dataclass(frozen=True) +class ValueAxis: + """Compatibility value-axis shape for identity-gate tests and fixtures. + + Runtime code may also pass core.physics.drive.ValueAxis instances. The + identity checker only requires axis_id, name, direction, and optional + theological_note, so both shapes are accepted. + """ + name: str + direction: Tuple[float, ...] + axis_id: str | None = None + weight: float = 1.0 + theological_note: str = "" + + def __post_init__(self) -> None: + object.__setattr__(self, "axis_id", self.axis_id or self.name) + object.__setattr__(self, "direction", tuple(float(x) for x in self.direction)) + + @dataclass(frozen=True) class IdentityScore: """Result of checking a ReasoningTrajectory against the IdentityManifold.""" @@ -25,8 +44,6 @@ class IdentityScore: deviation_axes: FrozenSet[str] # ValueAxis IDs where deviation was detected trajectory_id: str - # --- Convenience aliases used by runtime, serialiser, and review_trace --- - @property def value(self) -> float: """Alias for score — primary scalar alignment value (0.0–1.0).""" @@ -34,17 +51,10 @@ class IdentityScore: @property def alignment(self) -> float: - """Fraction of axes that were NOT flagged as deviating. - - 1.0 = all axes aligned; 0.0 = all axes deviated. - When deviation_axes is empty alignment is always 1.0. - """ + """Fraction of axes that were NOT flagged as deviating.""" axes = self.deviation_axes if not axes: return 1.0 - # deviation_axes only contains axes that deviated, but we don't - # independently track total axis count here. Use score as proxy: - # high score → high alignment. return self.score @property @@ -55,52 +65,41 @@ class IdentityScore: @dataclass(frozen=True) class IdentityManifold: - """Fixed geometric subspace encoding CORE's stable character. - - Instantiated once at model init. No mutation path exists. - value_axes: the geometric directions of CORE's core commitments. - boundary_ids: IDs of hyperplanes that no trajectory may cross. - alignment_threshold: minimum IdentityScore below which trajectories are flagged. - """ - value_axes: Tuple # Tuple[ValueAxis, ...] - boundary_ids: FrozenSet[str] + """Fixed geometric subspace encoding CORE's stable character.""" + value_axes: Tuple = () # Tuple[ValueAxis, ...] + boundary_ids: FrozenSet[str] = frozenset() alignment_threshold: float = 0.45 class IdentityCheck: """Checks a ReasoningTrajectory against an IdentityManifold. - The current runtime feeds this checker with lightweight binding frames - derived from generation field states. Low micro-pack energy should not - mechanically trip every identity axis. The score remains conservative, - but axis deviations are now assigned by axis projection rather than by - bulk-flagging every axis whenever the scalar score misses threshold. + Supports both the current explicit call style: + IdentityCheck().check(trajectory, manifold) + + and the older fixture style still used by some tests: + IdentityCheck(manifold=manifold).check(trajectory) """ + def __init__(self, manifold: IdentityManifold | None = None) -> None: + self._manifold = manifold + @staticmethod def _clamp01(value: float) -> float: return max(0.0, min(1.0, float(value))) @staticmethod def _mean_frame_coherence(trajectory) -> float: - if not getattr(trajectory, "frames", None): + frames = getattr(trajectory, "frames", None) + if not frames: return 0.0 return sum( - float(frame.coherence_magnitude) for frame in trajectory.frames - ) / len(trajectory.frames) + float(getattr(frame, "coherence_magnitude", 0.0)) for frame in frames + ) / len(frames) @staticmethod def _axis_projection(axis, trajectory, scalar_score: float) -> float: - """Deterministically project trajectory evidence onto one value axis. - - directional_weight measures what fraction of the axis's total L2 energy - lives in the first three versor components — the components directly - observable from FieldState.F[:3]. For the current canonical axes - (truthfulness=(1,0,0), coherence=(0,1,0), reverence=(0,0,1)) this - always evaluates to 1.0, so existing traces are unaffected. When - higher-dimensional directions are wired, the ratio will correctly - down-weight axes whose energy is spread across unobserved components. - """ + """Deterministically project trajectory evidence onto one value axis.""" direction = tuple(float(x) for x in getattr(axis, "direction", ()) or ()) if not direction: return scalar_score @@ -113,42 +112,42 @@ class IdentityCheck: (0.75 * scalar_score) + (0.25 * directional_weight * coherence_term) ) - def check(self, trajectory, manifold: IdentityManifold) -> IdentityScore: - if not manifold.value_axes: + def check(self, trajectory, manifold: IdentityManifold | None = None) -> IdentityScore: + resolved_manifold = manifold or self._manifold + if resolved_manifold is None: + raise TypeError("IdentityCheck.check() requires an IdentityManifold") + trajectory_id = str(getattr(trajectory, "trajectory_id", "legacy_trajectory")) + if not resolved_manifold.value_axes: return IdentityScore( score=1.0, flagged=False, deviation_axes=frozenset(), - trajectory_id=trajectory.trajectory_id, + trajectory_id=trajectory_id, ) confidence = float(getattr(trajectory, "total_coherence_delta", 0.0)) confidence += self._mean_frame_coherence(trajectory) score = self._clamp01(0.5 + (confidence / 2.0)) deviations = frozenset( - axis.axis_id - for axis in manifold.value_axes - if self._axis_projection(axis, trajectory, score) < manifold.alignment_threshold + str(getattr(axis, "axis_id", getattr(axis, "name", "axis"))) + for axis in resolved_manifold.value_axes + if self._axis_projection(axis, trajectory, score) < resolved_manifold.alignment_threshold ) return IdentityScore( score=score, flagged=bool(deviations), deviation_axes=deviations, - trajectory_id=trajectory.trajectory_id, + trajectory_id=trajectory_id, ) @dataclass(frozen=True) class CharacterProfile: - """Human-readable projection of the IdentityManifold. - - This is not the identity itself. The identity is geometric. - The CharacterProfile is a representation of it — a map, not the terrain. - """ - traits: Dict[str, str] # trait name → description - drive_summaries: Dict[str, float] # drive name → current gradient magnitude + """Human-readable projection of the IdentityManifold.""" + traits: Dict[str, str] + drive_summaries: Dict[str, float] fatigue_index: float boundary_commitments: Tuple[str, ...] - theological_grounding: Dict[str, str] # axis name → scriptural/philosophical note + theological_grounding: Dict[str, str] @classmethod def from_manifold( @@ -157,12 +156,6 @@ class CharacterProfile: drive_summaries: Optional[Dict[str, float]] = None, fatigue_index: float = 0.0, ) -> "CharacterProfile": - """Populate a CharacterProfile directly from a live IdentityManifold. - - Derives traits and theological grounding from the manifold's value_axes - so the profile always reflects the current geometric identity — not a - manually maintained parallel description. - """ traits: Dict[str, str] = {} theological_grounding: Dict[str, str] = {} for axis in manifold.value_axes: @@ -170,8 +163,9 @@ class CharacterProfile: f"Fixed geometric direction {axis.direction} " f"in versor manifold — non-negotiable." ) - if axis.theological_note: - theological_grounding[axis.name] = axis.theological_note + theological_note = getattr(axis, "theological_note", "") + if theological_note: + theological_grounding[axis.name] = theological_note return cls( traits=traits, @@ -186,27 +180,7 @@ class CharacterProfile: @dataclass(frozen=True) class TurnEvent: - """Append-only provenance record for one chat turn. - - Every field is deterministically derivable from the turn's execution. - No inference, no approximation — each value is the exact output of the - corresponding operator as it ran. The log of TurnEvents over a session - is a complete, reproducible trace of the model's internal state evolution. - - Fields: - turn — zero-based turn index within the session - input_tokens — tokens as ingested (after OOV filtering) - surface — emitted response surface after runtime selection - walk_surface — syntactically guarded token sequence from manifold walk - articulation_surface — proposition-level surface from realize() - dialogue_role — DialogueRole classification for this turn - identity_score — IdentityScore from IdentityCheck (None if not run) - cycle_cost_total — total CycleCost.total for this turn - vault_hits — number of vault recall hits that fired during generate() - versor_condition — versor_condition(final_state.F) after generation - flagged — True if identity_score.flagged (shortcut for filtering) - elaboration — woven walk tokens used in elaborate role (None otherwise) - """ + """Append-only provenance record for one chat turn.""" turn: int input_tokens: Tuple[str, ...] surface: str @@ -218,4 +192,4 @@ class TurnEvent: vault_hits: int versor_condition: float flagged: bool - elaboration: Optional[str] = None # woven walk tokens; populated by SentenceAssembler + elaboration: Optional[str] = None diff --git a/core/physics/reasoning.py b/core/physics/reasoning.py index 30d4a883..de0d4683 100644 --- a/core/physics/reasoning.py +++ b/core/physics/reasoning.py @@ -18,24 +18,66 @@ class TrajectoryTransition: from_frame_id: str to_frame_id: str pressure_delta: float - continuity_spine: FrozenSet[str] # region IDs stable across transition - differential_set: FrozenSet[str] # region IDs that entered or exited + continuity_spine: FrozenSet[str] + differential_set: FrozenSet[str] coherence_won: float coherence_lost: float -@dataclass(frozen=True) +@dataclass(frozen=True, init=False) class ReasoningTrajectory: - """Append-only sequence of BindingFrames with transition records.""" + """Append-only sequence of BindingFrames with transition records. + + The canonical runtime shape uses frames/transitions/coherence metadata. + Legacy tests may still construct ReasoningTrajectory(operators=..., turn=...). + Those legacy fields are accepted and projected into an empty-frame + trajectory so identity scoring remains deterministic and bounded. + """ trajectory_id: str - frames: Tuple # Tuple[BindingFrame, ...] - transitions: Tuple[TrajectoryTransition, ...] # len == len(frames) - 1 + frames: Tuple + transitions: Tuple[TrajectoryTransition, ...] total_coherence_delta: float - cycle_span: Tuple[int, int] # (start_cycle, end_cycle) + cycle_span: Tuple[int, int] + operators: Tuple + turn: int + + def __init__( + self, + trajectory_id: str | None = None, + frames: Tuple | list = (), + transitions: Tuple[TrajectoryTransition, ...] | list = (), + total_coherence_delta: float = 0.0, + cycle_span: Tuple[int, int] | None = None, + *, + operators: Tuple | list = (), + turn: int = 0, + ) -> None: + ordered_frames = tuple(frames) + ordered_transitions = tuple(transitions) + legacy_ops = tuple(operators) + resolved_span = cycle_span if cycle_span is not None else (int(turn), int(turn)) + resolved_id = trajectory_id or _trajectory_id(ordered_frames) or f"turn_{int(turn)}" + object.__setattr__(self, "trajectory_id", resolved_id) + object.__setattr__(self, "frames", ordered_frames) + object.__setattr__(self, "transitions", ordered_transitions) + object.__setattr__(self, "total_coherence_delta", float(total_coherence_delta)) + object.__setattr__(self, "cycle_span", resolved_span) + object.__setattr__(self, "operators", legacy_ops) + object.__setattr__(self, "turn", int(turn)) +@dataclass(frozen=True, init=False) class TrajectoryOperator: - """Builds a ReasoningTrajectory from an ordered sequence of BindingFrames.""" + """Builds a ReasoningTrajectory from BindingFrames. + + Also accepts legacy fixture construction with versor/step keyword fields. + """ + versor: object | None + step: int | None + + def __init__(self, versor=None, step: int | None = None) -> None: + object.__setattr__(self, "versor", versor) + object.__setattr__(self, "step", step) def build(self, frames: list, trajectory_id: str) -> ReasoningTrajectory: ordered = tuple(frames) @@ -76,4 +118,4 @@ def _trajectory_id(frames: tuple) -> str: h = hashlib.sha256() for frame in frames: h.update(frame.frame_id.encode("utf-8")) - return h.hexdigest() + return h.hexdigest() if frames else ""