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

chat/runtime: wire identity check, character motor, CharacterProfile, drive gradients, TurnEvent log

Browse source 
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.
This commit is contained in:
Shay 2026-05-14 13:15:24 -07:00
parent 6cb28566ec
commit 3711fad448
1 changed files with 181 additions and 13 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

194
chat/runtime.py
View file

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