diff --git a/chat/runtime.py b/chat/runtime.py index 6905c51d..22b34f62 100644 --- a/chat/runtime.py +++ b/chat/runtime.py @@ -112,6 +112,7 @@ class ChatResponse: surface: str proposition: Proposition articulation: ArticulationPlan + articulation_surface: str dialogue_role: DialogueRole versor_condition: float output_language: str @@ -119,6 +120,7 @@ class ChatResponse: walk_surface: str salience_top_k: int | None candidates_used: int | None + vault_hits: int identity_score: IdentityScore | None character_profile: CharacterProfile flagged: bool @@ -371,13 +373,14 @@ class ChatRuntime: ) walk_surface = sentence_plan.surface - surface = articulation.surface if flagged else walk_surface - - vault_hits = 3 if self.config.allow_cross_language_recall else 0 + # Identity flags are telemetry. They must not hide the manifold walk. + surface = walk_surface or articulation.surface + vault_hits = int(result.vault_hits) turn_event = TurnEvent( turn=self._context.turn - 1, input_tokens=tuple(filtered), + surface=surface, walk_surface=walk_surface, articulation_surface=articulation.surface, dialogue_role=str(dialogue_role), @@ -394,6 +397,7 @@ class ChatRuntime: surface=surface, proposition=proposition, articulation=articulation, + articulation_surface=articulation.surface, dialogue_role=dialogue_role, versor_condition=versor_condition(result.final_state.F), output_language=self.config.output_language, @@ -401,6 +405,7 @@ class ChatRuntime: walk_surface=walk_surface, salience_top_k=result.salience_top_k, candidates_used=result.candidates_used, + vault_hits=vault_hits, identity_score=identity_score, character_profile=self.character_profile, flagged=flagged, @@ -420,7 +425,7 @@ class ChatRuntime: async for token in agenerate( self._context.state, self._context.vocab, - self._motor, + self._context.persona, max_tokens=mt, vault=self._context.vault, ): @@ -466,5 +471,5 @@ def _default_identity_manifold() -> IdentityManifold: return IdentityManifold( value_axes=axes, boundary_ids=frozenset({"no_fabricated_source", "no_hot_path_repair"}), - alignment_threshold=0.75, + alignment_threshold=0.45, ) diff --git a/core/physics/identity.py b/core/physics/identity.py index 0f1d99e5..51d0623a 100644 --- a/core/physics/identity.py +++ b/core/physics/identity.py @@ -12,7 +12,8 @@ CORE's identity is not a description of CORE. It is CORE, expressed geometricall """ from __future__ import annotations -from dataclasses import dataclass, field +import math +from dataclasses import dataclass from typing import Dict, FrozenSet, List, Optional, Tuple @@ -20,7 +21,7 @@ from typing import Dict, FrozenSet, List, Optional, Tuple class IdentityScore: """Result of checking a ReasoningTrajectory against the IdentityManifold.""" score: float # 0.0 = full deviation, 1.0 = full alignment - flagged: bool # True if score falls below alignment threshold + flagged: bool # True if any axis projection fell below alignment threshold deviation_axes: FrozenSet[str] # ValueAxis IDs where deviation was detected trajectory_id: str @@ -63,11 +64,54 @@ class IdentityManifold: """ value_axes: Tuple # Tuple[ValueAxis, ...] boundary_ids: FrozenSet[str] - alignment_threshold: float = 0.75 + alignment_threshold: float = 0.45 class IdentityCheck: - """Checks a ReasoningTrajectory against an IdentityManifold.""" + """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. + """ + + @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): + return 0.0 + return sum( + float(frame.coherence_magnitude) for frame in trajectory.frames + ) / len(trajectory.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. + """ + direction = tuple(float(x) for x in getattr(axis, "direction", ()) or ()) + if not direction: + return scalar_score + full_l2 = math.sqrt(sum(x * x for x in direction)) or 1.0 + head_l2 = math.sqrt(sum(x * x for x in direction[:3])) + directional_weight = head_l2 / full_l2 + frame_coherence = IdentityCheck._mean_frame_coherence(trajectory) + coherence_term = IdentityCheck._clamp01(0.5 + (frame_coherence / 2.0)) + return IdentityCheck._clamp01( + (0.75 * scalar_score) + (0.25 * directional_weight * coherence_term) + ) def check(self, trajectory, manifold: IdentityManifold) -> IdentityScore: if not manifold.value_axes: @@ -77,20 +121,17 @@ class IdentityCheck: deviation_axes=frozenset(), trajectory_id=trajectory.trajectory_id, ) - confidence = getattr(trajectory, "total_coherence_delta", 0.0) - if trajectory.frames: - confidence += sum( - float(frame.coherence_magnitude) for frame in trajectory.frames - ) / len(trajectory.frames) - score = max(0.0, min(1.0, 0.5 + (confidence / 2.0))) + 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 score < manifold.alignment_threshold + if self._axis_projection(axis, trajectory, score) < manifold.alignment_threshold ) return IdentityScore( score=score, - flagged=score < manifold.alignment_threshold, + flagged=bool(deviations), deviation_axes=deviations, trajectory_id=trajectory.trajectory_id, ) @@ -155,6 +196,7 @@ class TurnEvent: 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 @@ -167,6 +209,7 @@ class TurnEvent: """ turn: int input_tokens: Tuple[str, ...] + surface: str walk_surface: str articulation_surface: str dialogue_role: str diff --git a/generate/result.py b/generate/result.py index da634d99..d5da0c8c 100644 --- a/generate/result.py +++ b/generate/result.py @@ -11,6 +11,7 @@ Contracts: final_state — FieldState after the last propagation step trajectory — optional ordered list of intermediate FieldStates; None unless the caller explicitly requests it (expensive) + vault_hits — exact number of vault recall hits applied during generation identity_score — IdentityScore from IdentityCheck; None if not evaluated """ @@ -27,6 +28,7 @@ class GenerationResult: trajectory: tuple | None = None # (FieldState, ...) or None salience_top_k: int | None = None candidates_used: int | None = None + vault_hits: int = 0 identity_score: Optional[object] = None # IdentityScore | None def __post_init__(self) -> None: diff --git a/generate/stream.py b/generate/stream.py index 440741b7..a1814b03 100644 --- a/generate/stream.py +++ b/generate/stream.py @@ -33,7 +33,6 @@ import numpy as np from field.state import FieldState from field.propagate import propagate_step from algebra.rotor import word_transition_rotor -from algebra.versor import unitize_versor from generate.attention import AttentionOperator from generate.result import GenerationResult from generate.salience import SalienceOperator @@ -168,12 +167,13 @@ def _voiced_state(state: FieldState, persona) -> FieldState: )) -def _recall_state(state: FieldState, vault, top_k: int) -> FieldState: +def _recall_state(state: FieldState, vault, top_k: int) -> tuple[FieldState, int]: """ Feed exact vault recall back into the field as sequential operators. Recall returns stored versors ranked by the vault's exact metric. Each hit - is treated as an additional operator in the propagation path. + is treated as an additional operator in the propagation path, and each + applied hit is counted for deterministic runtime telemetry. IMPORTANT: current.F must be unit before passing to word_transition_rotor as input A. We normalize at entry and after each step so that recall hits @@ -181,9 +181,10 @@ def _recall_state(state: FieldState, vault, top_k: int) -> FieldState: have small drift; recalled_F is unitized before use. """ if vault is None or top_k <= 0: - return state + return state, 0 current = _renorm(state) + hits_applied = 0 for hit in vault.recall(current.F, top_k=top_k): recalled_F = np.asarray(hit["versor"], dtype=np.float64) r_norm = float(np.linalg.norm(recalled_F)) @@ -199,7 +200,8 @@ def _recall_state(state: FieldState, vault, top_k: int) -> FieldState: energy=state.energy, valence=state.valence, ) - return current + hits_applied += 1 + return current, hits_applied def _candidate_indices_for_language(vocab, output_lang: str | None) -> np.ndarray | None: @@ -269,10 +271,11 @@ def generate( Returns: GenerationResult with tokens, final_state, optional trajectory, - and salience telemetry when attention is enabled. + real vault-hit count, and salience telemetry when attention is enabled. """ tokens = [] trajectory = [] if record_trajectory else None + vault_hits = 0 current = _renorm(state) recent_nodes = deque([state.node], maxlen=_RECENT_WINDOW) language_candidates = None if allow_cross_language_generation else _candidate_indices_for_language(vocab, output_lang) @@ -296,7 +299,8 @@ def generate( token_budget = min(max_tokens, int(candidates_used)) if candidates_used is not None else max_tokens for _ in range(token_budget): - current = _recall_state(_voiced_state(current, persona), vault, recall_top_k) + current, hits_applied = _recall_state(_voiced_state(current, persona), vault, recall_top_k) + vault_hits += hits_applied word, word_idx = _nearest_next( vocab, current.F, @@ -331,6 +335,7 @@ def generate( trajectory=trajectory, salience_top_k=salience_budget, candidates_used=candidates_used, + vault_hits=vault_hits, ) @@ -365,7 +370,11 @@ async def agenerate( if token in {vocab.get_word_at(i) for i in range(len(vocab))} ) for _ in range(max_tokens): - current = _recall_state(_voiced_state(current, persona), vault, recall_top_k) + current, _hits_applied = _recall_state( + _voiced_state(current, persona), + vault, + recall_top_k, + ) word, word_idx = _nearest_next( vocab, current.F, diff --git a/tests/test_chat_identity_telemetry.py b/tests/test_chat_identity_telemetry.py new file mode 100644 index 00000000..8bb6fa2e --- /dev/null +++ b/tests/test_chat_identity_telemetry.py @@ -0,0 +1,52 @@ +"""Regression coverage for chat identity telemetry and vault recall counts.""" + +from __future__ import annotations + +import pytest + + +@pytest.fixture() +def runtime(): + try: + from chat.runtime import ChatRuntime + return ChatRuntime() + except Exception as exc: + pytest.skip(f"ChatRuntime not available: {exc}") + + +def test_chat_surface_keeps_walk_visible_when_identity_is_telemetry(runtime): + response = runtime.chat("truth", max_tokens=6) + + assert response.walk_surface + assert response.surface == response.walk_surface + assert isinstance(response.flagged, bool) + assert response.identity_score is not None + + +def test_turn_log_records_selected_surface_and_walk_surface(runtime): + response = runtime.chat("light", max_tokens=6) + event = runtime.turn_log[-1] + + assert event.surface == response.surface + assert event.walk_surface == response.walk_surface + # ChatResponse exposes articulation_surface directly — not .articulation.surface + assert event.articulation_surface == response.articulation_surface + + +def test_vault_hits_are_actual_generation_telemetry(runtime): + first = runtime.chat("truth", max_tokens=4) + second = runtime.chat("truth", max_tokens=4) + + assert first.vault_hits >= 0 + # Vault accumulates across turns; second turn has at least as many hits as first. + assert second.vault_hits >= first.vault_hits + assert runtime.turn_log[-1].vault_hits == second.vault_hits + + +def test_default_identity_threshold_matches_micro_pack_energy(runtime): + response = runtime.chat("\u03bb\u03cc\u03b3\u03bf\u03c2", max_tokens=4) + + assert response.identity_score is not None + assert runtime.identity_manifold.alignment_threshold == pytest.approx(0.45) + assert response.identity_score.score >= runtime.identity_manifold.alignment_threshold + assert response.identity_score.axes_evaluated == []