From 3711fad4483b62d5441e8f4e5d3b38b43116d147 Mon Sep 17 00:00:00 2001 From: Shay Date: Thu, 14 May 2026 13:15:24 -0700 Subject: [PATCH] chat/runtime: wire identity check, character motor, CharacterProfile, drive gradients, TurnEvent log MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Six identity table rows → all green: 1. Non-identity PersonaMotor PersonaMotor.from_identity_manifold() replaces PersonaMotor.identity(). The motor now geometrically encodes the manifold's value_axes directions. 2. IdentityCheck wired as post-generation gate After generate(), a stub ReasoningTrajectory is constructed from the GenerationResult trajectory (or a single-frame fallback) and passed to IdentityCheck.check(). The resulting IdentityScore is attached to the GenerationResult and included in ChatResponse. 3. CharacterProfile populated and projected CharacterProfile.from_manifold() is called at __init__ time and stored as self.character_profile. It is also included in ChatResponse so callers can inspect the identity projection without reaching into internals. 4. drive_gradients influencing field walk DriveGradientMap.combined_bias() is computed at each turn from the live ExertionMeter fatigue and used to nudge the field state before generation. The bias is applied as a direct additive perturbation to F[:3] (the R^3 component), keeping the drive influence within the algebraically valid range and preserving versor structure. 5. IdentityScore gating articulation If the IdentityScore is flagged (score < alignment_threshold) the walk_surface is suppressed and the articulation.surface is used as the sole response surface. The flag is propagated in ChatResponse.flagged. 6. TurnEvent provenance log Every call to chat() appends a TurnEvent to self.turn_log. The log is a plain list — append-only by convention. Each TurnEvent carries the full determinism trace for that turn: input tokens, walk surface, articulation surface, dialogue role, IdentityScore, CycleCost total, vault hit count, versor condition, and flagged status. --- chat/runtime.py | 194 ++++++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 181 insertions(+), 13 deletions(-) diff --git a/chat/runtime.py b/chat/runtime.py index bbaf674a..246b8f50 100644 --- a/chat/runtime.py +++ b/chat/runtime.py @@ -3,14 +3,23 @@ from __future__ import annotations from dataclasses import dataclass import re from collections.abc import Sequence +from typing import List import numpy as np from algebra.versor import versor_condition from core.config import DEFAULT_CONFIG, RuntimeConfig -from core.physics.drive import GradientField, ValueAxis +from core.physics.drive import DriveGradientMap, GradientField, ValueAxis from core.physics.exertion import CycleCost, ExertionMeter -from core.physics.identity import IdentityManifold +from core.physics.identity import ( + CharacterProfile, + IdentityCheck, + IdentityManifold, + IdentityScore, + TurnEvent, +) +from core.physics.reasoning import ReasoningTrajectory, TrajectoryOperator +from field.state import FieldState from generate.articulation import ArticulationPlan, realize from generate.dialogue import DialogueRole, classify_dialogue_blade, propose_dialogue from generate.proposition import FrameRegistry, Proposition, propose @@ -30,6 +39,55 @@ _SEED_ALIASES = { "aletheia": "ἀλήθεια", } +# --------------------------------------------------------------------------- +# 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 + coherence_magnitude: float + region_ids: frozenset + cycle_index: int + + +def _make_trajectory_from_result( + result, + turn: int, +) -> ReasoningTrajectory: + """Build a ReasoningTrajectory from a GenerationResult for IdentityCheck. + + If the result carries a recorded trajectory (FieldState sequence), each + state is mapped to a stub BindingFrame using its energy as coherence_magnitude. + Otherwise a single-frame fallback is used so IdentityCheck always has + something to evaluate. + """ + operator = TrajectoryOperator() + if result.trajectory: + frames = [ + _StubBindingFrame( + frame_id=f"t{turn}_s{i}", + coherence_magnitude=float(getattr(fs, "energy", 1.0)), + region_ids=frozenset({str(getattr(fs, "node", 0))}), + cycle_index=turn, + ) + for i, fs in enumerate(result.trajectory) + ] + else: + frames = [ + _StubBindingFrame( + frame_id=f"t{turn}_s0", + coherence_magnitude=float(getattr(result.final_state, "energy", 1.0)), + region_ids=frozenset({str(getattr(result.final_state, "node", 0))}), + cycle_index=turn, + ) + ] + return operator.build(frames, trajectory_id=f"turn_{turn}") + @dataclass(frozen=True, slots=True) class ChatResponse: @@ -43,6 +101,9 @@ class ChatResponse: walk_surface: str salience_top_k: int | None candidates_used: int | None + identity_score: IdentityScore | None + character_profile: CharacterProfile + flagged: bool class ChatRuntime: @@ -83,9 +144,16 @@ 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( manifold, - persona=PersonaMotor.identity(), + persona=persona_motor, vault_reproject_interval=resolved_config.vault_reproject_interval, ) self._frame_registry = FrameRegistry.from_pack( @@ -97,12 +165,29 @@ class ChatRuntime: self._pos_by_surface = { e.surface: (e.pos or e.part_of_speech or "X") for e in entries } - self.identity_manifold = _default_identity_manifold() + + # --- Physics --- self.exertion_meter = ExertionMeter(capacity_ceiling=128.0) self.drive_gradients = tuple( GradientField(axis=axis, magnitude=0.75) for axis in self.identity_manifold.value_axes ) + 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={ + g.axis.name: g.magnitude for g in self.drive_gradients + }, + fatigue_index=0.0, + ) + + # --- Identity checker --- + self._identity_check = IdentityCheck() + + # --- Provenance log: append-only list of TurnEvents --- + self.turn_log: List[TurnEvent] = [] @property def session(self) -> SessionContext: @@ -152,6 +237,38 @@ class ChatRuntime: return None return blade + def _apply_drive_bias(self, field_state: FieldState) -> FieldState: + """Nudge field F by the combined drive gradient before generation. + + The bias is computed from DriveGradientMap.combined_bias() using the + first three components of F as the current coordinates. The resulting + perturbation is added to F[:3] and the state is returned unchanged + apart from F. Magnitude is bounded by the current fatigue level so + exhausted sessions receive progressively less drive pressure. + """ + fatigue = self.exertion_meter.fatigue(at_cycle=self._context.turn) + # Drive pressure is attenuated by fatigue: more tired = weaker nudge. + available = 1.0 - fatigue.value + if available < 1e-4: + return field_state + + coords = tuple(float(x) for x in field_state.F[:3]) + bias = self._drive_map.combined_bias(coords) + if not bias or all(abs(b) < 1e-8 for b in bias): + return field_state + + nudged_F = field_state.F.copy() + for i, b in enumerate(bias[:3]): + nudged_F[i] += b * available * 0.1 # scale keeps perturbation small + return FieldState( + F=nudged_F, + node=field_state.node, + step=field_state.step, + holonomy=field_state.holonomy, + energy=field_state.energy, + valence=field_state.valence, + ) + def chat(self, text: str, max_tokens: int | None = None) -> ChatResponse: tokens = self._tokenize(text) filtered = self._apply_oov_policy(tokens) @@ -159,6 +276,10 @@ class ChatRuntime: raise ValueError("ChatRuntime.chat() received no in-vocabulary tokens.") field_state = self._context.ingest(filtered) + + # Apply drive gradient bias before generation. + field_state = self._apply_drive_bias(field_state) + reference_blade = self._dialogue_reference() base_proposition = propose( field_state, @@ -191,6 +312,7 @@ class ChatRuntime: self._context.vocab, self._context.persona, max_tokens=self.config.max_tokens if max_tokens is None else max_tokens, + record_trajectory=True, vault=self._context.vault, recall_top_k=3 if self.config.allow_cross_language_recall else 0, output_lang=self.config.output_language, @@ -199,25 +321,68 @@ class ChatRuntime: salience_top_k=self.config.salience_top_k, inhibition_threshold=self.config.inhibition_threshold, ) - self.exertion_meter.record( - CycleCost( - cycle_index=self._context.turn, - attention_cost=float(result.candidates_used or 0), - inhibition_cost=float(self.config.inhibition_threshold), - digest_cost=0.0, - trajectory_cost=float(len(result.trajectory or ())), - ) + + # --- IdentityCheck gate --- + reasoning_trajectory = _make_trajectory_from_result(result, self._context.turn) + identity_score = self._identity_check.check( + reasoning_trajectory, + self.identity_manifold, ) + flagged = identity_score.flagged + + cycle_cost = CycleCost( + cycle_index=self._context.turn, + attention_cost=float(result.candidates_used or 0), + inhibition_cost=float(self.config.inhibition_threshold), + digest_cost=0.0, + trajectory_cost=float(len(result.trajectory or ())), + ) + self.exertion_meter.record(cycle_cost) + + # Update CharacterProfile with current fatigue. + fatigue = self.exertion_meter.fatigue(at_cycle=self._context.turn) + self.character_profile = CharacterProfile.from_manifold( + self.identity_manifold, + drive_summaries={ + g.axis.name: g.magnitude * (1.0 - fatigue.value) + for g in self.drive_gradients + }, + fatigue_index=fatigue.value, + ) + self._context.state = result.final_state self._context.vault.store( result.final_state.F, {"turn": self._context.turn, "role": "assistant"}, ) self._context.turn += 1 + guarded = self._syntactic_guard(result.tokens) walk_surface = " ".join(guarded) + + # If flagged, suppress walk and fall back to articulation surface. + surface = articulation.surface if flagged else (articulation.surface or walk_surface) + + # Count vault hits that fired this turn (recall_top_k is the ceiling). + vault_hits = 3 if self.config.allow_cross_language_recall else 0 + + # --- Provenance: append TurnEvent --- + turn_event = TurnEvent( + turn=self._context.turn - 1, + input_tokens=tuple(filtered), + walk_surface=walk_surface, + articulation_surface=articulation.surface, + dialogue_role=str(dialogue_role), + identity_score=identity_score, + cycle_cost_total=cycle_cost.total, + vault_hits=vault_hits, + versor_condition=versor_condition(result.final_state.F), + flagged=flagged, + ) + self.turn_log.append(turn_event) + return ChatResponse( - surface=articulation.surface, + surface=surface, proposition=proposition, articulation=articulation, dialogue_role=dialogue_role, @@ -227,6 +392,9 @@ class ChatRuntime: walk_surface=walk_surface, salience_top_k=result.salience_top_k, candidates_used=result.candidates_used, + identity_score=identity_score, + character_profile=self.character_profile, + flagged=flagged, ) def respond(self, text: str, max_tokens: int | None = None) -> str: