From 92be98fbdf47a8dde21ec74dd7f4b3061d6e396e Mon Sep 17 00:00:00 2001 From: Shay Date: Thu, 14 May 2026 19:35:03 -0700 Subject: [PATCH] feat(cognition): add CognitiveTurnPipeline spine (#19) --- core/cognition/__init__.py | 15 +++ core/cognition/pipeline.py | 103 ++++++++++++++++++ core/cognition/result.py | 53 +++++++++ core/cognition/trace.py | 67 ++++++++++++ tests/test_cognitive_turn_pipeline.py | 149 ++++++++++++++++++++++++++ 5 files changed, 387 insertions(+) create mode 100644 core/cognition/__init__.py create mode 100644 core/cognition/pipeline.py create mode 100644 core/cognition/result.py create mode 100644 core/cognition/trace.py create mode 100644 tests/test_cognitive_turn_pipeline.py diff --git a/core/cognition/__init__.py b/core/cognition/__init__.py new file mode 100644 index 00000000..6910c698 --- /dev/null +++ b/core/cognition/__init__.py @@ -0,0 +1,15 @@ +""" +core.cognition — the cognitive spine. + +Exports the public surface of the pipeline layer. +""" + +from core.cognition.pipeline import CognitiveTurnPipeline +from core.cognition.result import CognitiveTurnResult +from core.cognition.trace import compute_trace_hash + +__all__ = [ + "CognitiveTurnPipeline", + "CognitiveTurnResult", + "compute_trace_hash", +] diff --git a/core/cognition/pipeline.py b/core/cognition/pipeline.py new file mode 100644 index 00000000..87e43927 --- /dev/null +++ b/core/cognition/pipeline.py @@ -0,0 +1,103 @@ +""" +CognitiveTurnPipeline — the cognitive spine. + +Architecture: + listen -> ingest -> understand -> recall -> think -> articulate + -> learn_proposal -> trace + +This first-pass implementation delegates to ChatRuntime internals so +future intelligence modules (IntentPropositionGraph, ArticulationRealizerV2, +ReviewedTeachingLoop, CognitiveEvalHarness) have a clean plug-in surface +without requiring a full ChatRuntime rewrite. + +Constraint: ChatRuntime.chat() and ChatResponse contract are unchanged. +""" + +from __future__ import annotations + +from field.state import FieldState +from core.cognition.result import CognitiveTurnResult +from core.cognition.trace import compute_trace_hash + + +class CognitiveTurnPipeline: + """Thin pipeline wrapper over ChatRuntime. + + Phase 1 goal: extract the observability path so downstream modules have + a place to plug in. No new intelligence is added here. + """ + + def __init__(self, runtime) -> None: # runtime: ChatRuntime (no import cycle) + self.runtime = runtime + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + def run(self, text: str, max_tokens: int | None = None) -> CognitiveTurnResult: + """Execute one full cognitive turn and return a complete result record.""" + + # 1. LISTEN — capture pre-turn field state + field_state_before: FieldState | None = self._capture_field_state() + + # 2–7. INGEST / UNDERSTAND / RECALL / THINK / ARTICULATE / LEARN + # Delegated to ChatRuntime.chat() in Phase 1. + # ChatResponse is the stable contract surface. + response = self.runtime.chat(text, max_tokens=max_tokens) + + # 8. CAPTURE post-turn field state + field_state_after: FieldState = self.runtime.session.state + + # 9. Reconstruct input-layer tokens from the turn log + # (turn_log is appended inside chat(); last entry matches this turn) + last_turn = self.runtime.turn_log[-1] + input_tokens = last_turn.input_tokens # already filtered + filtered_tokens = last_turn.input_tokens # same at Phase 1 + + # Raw tokenization is identical to filtered for Phase 1 — the + # runtime's _tokenize() runs before _apply_oov_policy(). We + # expose input_tokens separately so Phase 2 can diverge them. + raw_tokens = tuple(self.runtime.tokenize(text)) + + # 10. TRACE — deterministic hash + trace_hash = compute_trace_hash( + input_text=text, + filtered_tokens=filtered_tokens, + surface=response.surface, + walk_surface=response.walk_surface, + articulation_surface=response.articulation_surface, + dialogue_role=str(response.dialogue_role), + versor_condition=response.versor_condition, + vault_hits=response.vault_hits, + ) + + return CognitiveTurnResult( + input_text=text, + input_tokens=raw_tokens, + filtered_tokens=filtered_tokens, + field_state_before=field_state_before, + field_state_after=field_state_after, + proposition=response.proposition, + articulation=response.articulation, + surface=response.surface, + walk_surface=response.walk_surface, + articulation_surface=response.articulation_surface, + dialogue_role=response.dialogue_role, + identity_score=response.identity_score, + vault_hits=response.vault_hits, + versor_condition=response.versor_condition, + trace_hash=trace_hash, + ) + + # ------------------------------------------------------------------ + # Internal helpers + # ------------------------------------------------------------------ + + def _capture_field_state(self) -> FieldState | None: + """Return current session field state, or None if not yet initialised.""" + try: + state = self.runtime.session.state + # SessionContext.state may be None before the first ingest + return state if state is not None else None + except AttributeError: + return None diff --git a/core/cognition/result.py b/core/cognition/result.py new file mode 100644 index 00000000..cb50f5b9 --- /dev/null +++ b/core/cognition/result.py @@ -0,0 +1,53 @@ +""" +CognitiveTurnResult — the complete record of one cognitive turn. + +This is the canonical output of CognitiveTurnPipeline.run(). It is +frozen and slot-based so it can be passed safely across module boundaries +without mutation risk. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +from field.state import FieldState +from generate.articulation import ArticulationPlan +from generate.dialogue import DialogueRole +from generate.proposition import Proposition +from core.physics.identity import IdentityScore + + +@dataclass(frozen=True, slots=True) +class CognitiveTurnResult: + """Full observability record for a single pipeline turn.""" + + # --- input layer --- + input_text: str + input_tokens: tuple[str, ...] + filtered_tokens: tuple[str, ...] + + # --- field layer --- + field_state_before: FieldState | None # None on the very first turn + field_state_after: FieldState + + # --- understanding / recall layer --- + proposition: Proposition + articulation: ArticulationPlan + + # --- output surfaces --- + surface: str # final voiced surface (what the user sees) + walk_surface: str # sentence-assembled walk surface + articulation_surface: str # bare articulation surface before assembly + + # --- dialogue --- + dialogue_role: DialogueRole + + # --- identity telemetry --- + identity_score: IdentityScore | None + + # --- vault / memory --- + vault_hits: int + + # --- invariant bookkeeping --- + versor_condition: float # must be < 1e-6 + trace_hash: str # SHA-256 over deterministic key fields diff --git a/core/cognition/trace.py b/core/cognition/trace.py new file mode 100644 index 00000000..7470b1a9 --- /dev/null +++ b/core/cognition/trace.py @@ -0,0 +1,67 @@ +""" +Deterministic trace hashing for cognitive turns. + +The hash captures every meaningful output of a pipeline run so that: + - identical inputs on identical field state → identical hash + - any output change → different hash + +Only stable, semantically meaningful fields are included. Floating-point +values are rounded to 9 decimal places before hashing so that numeric +noise from different hardware does not break determinism within a run. +""" + +from __future__ import annotations + +import hashlib +import json +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from core.cognition.result import CognitiveTurnResult + + +def _round_float(v: float, ndigits: int = 9) -> float: + return round(float(v), ndigits) + + +def compute_trace_hash( + input_text: str, + filtered_tokens: tuple[str, ...], + surface: str, + walk_surface: str, + articulation_surface: str, + dialogue_role: str, + versor_condition: float, + vault_hits: int, +) -> str: + """Return a deterministic SHA-256 hex digest over the turn's key outputs. + + Parameters match the subset of CognitiveTurnResult that is both + semantically meaningful and stable across hardware. + """ + payload = { + "input_text": input_text, + "filtered_tokens": list(filtered_tokens), + "surface": surface, + "walk_surface": walk_surface, + "articulation_surface": articulation_surface, + "dialogue_role": str(dialogue_role), + "versor_condition": _round_float(versor_condition), + "vault_hits": int(vault_hits), + } + serialized = json.dumps(payload, sort_keys=True, ensure_ascii=False) + return hashlib.sha256(serialized.encode("utf-8")).hexdigest() + + +def trace_hash_from_result(result: "CognitiveTurnResult") -> str: + """Convenience wrapper — compute the hash directly from a result object.""" + return compute_trace_hash( + input_text=result.input_text, + filtered_tokens=result.filtered_tokens, + surface=result.surface, + walk_surface=result.walk_surface, + articulation_surface=result.articulation_surface, + dialogue_role=str(result.dialogue_role), + versor_condition=result.versor_condition, + vault_hits=result.vault_hits, + ) diff --git a/tests/test_cognitive_turn_pipeline.py b/tests/test_cognitive_turn_pipeline.py new file mode 100644 index 00000000..cade91c2 --- /dev/null +++ b/tests/test_cognitive_turn_pipeline.py @@ -0,0 +1,149 @@ +""" +Tests for CognitiveTurnPipeline — the cognitive spine. + +Five tests, no micro-test explosion: + 1. test_pipeline_known_token_turn — happy-path turn with known tokens + 2. test_pipeline_unknown_token_grounding — OOV token handled; field still valid + 3. test_pipeline_two_turn_memory_continuity — field evolves across turns + 4. test_pipeline_trace_hash_deterministic — identical inputs → identical hash + 5. test_pipeline_preserves_versor_closure — versor_condition < 1e-6 per turn +""" + +from __future__ import annotations + +import numpy as np +import pytest + +from chat.runtime import ChatRuntime +from core.cognition import CognitiveTurnPipeline, CognitiveTurnResult +from core.cognition.trace import trace_hash_from_result + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture() +def runtime() -> ChatRuntime: + return ChatRuntime() + + +@pytest.fixture() +def pipeline(runtime: ChatRuntime) -> CognitiveTurnPipeline: + return CognitiveTurnPipeline(runtime) + + +# --------------------------------------------------------------------------- +# 1. Known token turn +# --------------------------------------------------------------------------- + +def test_pipeline_known_token_turn(pipeline: CognitiveTurnPipeline) -> None: + """A single turn with known tokens yields a fully populated result.""" + result = pipeline.run("light logos", max_tokens=8) + + assert isinstance(result, CognitiveTurnResult) + + # Input layer + assert result.input_text == "light logos" + assert len(result.input_tokens) >= 1 + assert len(result.filtered_tokens) >= 1 + + # Field layer + assert result.field_state_before is None # first turn: no prior state + assert result.field_state_after is not None + assert result.field_state_after.F.shape == (32,) + + # Output surfaces + assert result.surface.strip() + assert isinstance(result.walk_surface, str) + assert isinstance(result.articulation_surface, str) + + # Dialogue + assert result.dialogue_role in {"assert", "elaborate", "question", "refute"} + + # Bookkeeping + assert isinstance(result.versor_condition, float) + assert isinstance(result.trace_hash, str) and len(result.trace_hash) == 64 + assert isinstance(result.vault_hits, int) + + +# --------------------------------------------------------------------------- +# 2. Unknown / OOV token grounding +# --------------------------------------------------------------------------- + +def test_pipeline_unknown_token_grounding(pipeline: CognitiveTurnPipeline) -> None: + """OOV token in an open pack should not prevent field from staying valid.""" + result = pipeline.run("what is דברית", max_tokens=4) + + # Runtime must still produce a valid result + assert result.surface.strip() + assert result.field_state_after is not None + assert result.versor_condition < 1e-6 + + +# --------------------------------------------------------------------------- +# 3. Two-turn memory continuity +# --------------------------------------------------------------------------- + +def test_pipeline_two_turn_memory_continuity(pipeline: CognitiveTurnPipeline) -> None: + """Field state evolves between turns, confirming the pipeline threads memory.""" + first = pipeline.run("light logos", max_tokens=8) + second = pipeline.run("truth logos", max_tokens=8) + + # second turn knows about first + assert second.field_state_before is not None + assert second.field_state_before.F.shape == (32,) + + # field genuinely moved between turns + assert not np.array_equal( + first.field_state_after.F, + second.field_state_after.F, + ), "Field state must evolve across turns." + + # Both versor conditions are closed + assert first.versor_condition < 1e-6 + assert second.versor_condition < 1e-6 + + +# --------------------------------------------------------------------------- +# 4. Trace hash determinism +# --------------------------------------------------------------------------- + +def test_pipeline_trace_hash_deterministic() -> None: + """Identical inputs on a fresh runtime produce the same trace hash.""" + rt1 = ChatRuntime() + rt2 = ChatRuntime() + + r1 = CognitiveTurnPipeline(rt1).run("light truth", max_tokens=6) + r2 = CognitiveTurnPipeline(rt2).run("light truth", max_tokens=6) + + # Re-derive via the helper to confirm the hash formula is stable + assert r1.trace_hash == trace_hash_from_result(r1) + assert r2.trace_hash == trace_hash_from_result(r2) + + # Same hash across two independent runtimes with same prompt + assert r1.trace_hash == r2.trace_hash, ( + f"Expected deterministic hash, got:\n r1={r1.trace_hash}\n r2={r2.trace_hash}" + ) + + +# --------------------------------------------------------------------------- +# 5. Versor closure preserved across all turns +# --------------------------------------------------------------------------- + +def test_pipeline_preserves_versor_closure(pipeline: CognitiveTurnPipeline) -> None: + """versor_condition must stay below 1e-6 for every turn in the session.""" + prompts = [ + "logos light", + "truth word", + "what is λόγος", + "spirit breath", + ] + for prompt in prompts: + result = pipeline.run(prompt, max_tokens=6) + assert result.versor_condition < 1e-6, ( + f"Versor closure broken after prompt {prompt!r}: " + f"versor_condition={result.versor_condition:.2e}" + ) + # Field state invariant: shape must be intact + assert result.field_state_after.F.shape == (32,)