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

Fix main chat test regressions

Browse source 
- make async chat reuse initialized synchronous chat lifecycle
- restore make_rotor_from_angle compatibility helper
- restore identity ValueAxis export and legacy IdentityCheck call style
- preserve legacy reasoning trajectory fixtures for existing tests
This commit is contained in:
Shay 2026-05-14 15:48:45 -07:00 committed by GitHub
parent 216a789808
commit c46eae8fc8
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 133 additions and 140 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
algebra/rotor.py
View file

@ -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).

58
chat/runtime.py
View file

@ -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()."""

138
core/physics/identity.py
View file

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

60
core/physics/reasoning.py
View file

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