Publish Shadow-X v14 CCS model card + code + tests + architecture

README.md

@@ -0,0 +1,113 @@
+---
+language:
+- en
+- it
+license: other
+library_name: none
+tags:
+- deterministic-ai
+- no-llm
+- cognitive-control
+- topological-memory
+- out-of-core
+---
+
+# Shadow-X v14 CCS
+
+Deterministic Cognitive Control System (CCS-v2) for Shadow-X.
+
+This repository contains the v14 executive controller layer that upgrades Shadow-X from:
+- topological associative memory + basic deterministic flow,
+to:
+- full deterministic cognitive control with hierarchical goals, attention routing, conflict inhibition, self-monitoring, task prospection, and third-order deterministic rule updates.
+
+No probabilities, no gradients, no softmax.
+
+## Core Architecture
+
+```mermaid
+graph TD
+    A[Input Pulse] --> B[Phantom Kernel]
+    B --> C[Episodic Memory Slabs and Z-Hypergrid]
+
+    D[Deterministic Cognitive Controller CCS-v2] -->|Goal Maintenance| B
+    D -->|Attention Allocation| C
+    D -->|Conflict Inhibition| B
+    D -->|Error Monitor and Correction| B
+    D -->|Task Switching and Prospection| C
+    D -->|Meta Rule Update| E[Third-Order Rule Modifier]
+
+    F[goal_vector.dat Hierarchical Goal Stack] --> D
+    D --> G[trace_log.jsonl Full Audit Trail]
+```
+
+## Deterministic Control Loop
+
+```mermaid
+flowchart LR
+    G1[Load Active Goal] --> A1[Allocate Active Zones]
+    A1 --> C1[Query Candidate Bindings]
+    C1 --> I1[Resolve Conflicts]
+    I1 --> P1[Prospection 2-3 Futures]
+    P1 --> E1[Error Monitor vs Goal]
+    E1 -->|Deviation High| R1[Deterministic Rebind]
+    E1 -->|Deviation Low| O1[Materialize Output]
+    R1 --> O1
+    O1 --> T1[Append Trace Log]
+    T1 --> M1[Third-Order Rule Update]
+```
+
+## Modules
+
+1. Goal Vector + Hierarchical Goal Stack  
+Persistent goals in `goal_vector.dat`, parent-child relations, deterministic selection.
+
+2. Deterministic Attention Allocator  
+Goal-projected geometric routing to activate relevant slabs/zones only.
+
+3. Conflict Resolver (Inhibition)  
+Suppression of contradictory candidates via fixed opposition threshold.
+
+4. Error Monitor + Self-Correction  
+Output/goal deviation check with deterministic rebind and constraint strengthening.
+
+5. Task Switcher + Prospection  
+Deterministic branch simulation (2-3 futures) and best-branch selection.
+
+6. Third-Order Rule Modifier  
+Deterministic threshold adaptation from recurrent error windows.
+
+## Repository Content
+
+- `shadow_x_v14_ccs/cognitive_controller.py`: CCS-v2 implementation.
+- `shadow_x_v14_ccs/phantom_bridge.py`: bridge from CCS-v2 to `phantom_kernel.py`.
+- `shadow_x_v14_ccs/demo_cli.py`: one-shot + goal query + trace + self-correction demo.
+- `shadow_x_v14_ccs/ARCHITECTURE.md`: architecture notes.
+- `shadow_x_v14_ccs/README.md`: package-focused documentation.
+- `tests/test_shadow_x_v14_ccs.py`: controller unit/integration tests.
+- `tests/test_shadow_x_v14_phantom_bridge.py`: bridge tests + real-dataset smoke.
+- `tests/test_shadow_x_v14_demo_cli.py`: public demo flow tests.
+
+## Quickstart
+
+```python
+from shadow_x_v14_ccs import create_v14_controller_with_phantom_kernel
+
+controller = create_v14_controller_with_phantom_kernel(
+    model_path="v11_Morpho_Optimized",
+    state_dir="shadow_x_v14_ccs/state",
+    max_zone_candidates=128,
+    max_candidates_per_zone=16,
+)
+```
+
+## Public Demo
+
+- Hugging Face Space: `https://huggingface.co/spaces/RthItalia/shadow-x-v14-mini-demo`
+- GitHub Gist: `https://gist.github.com/rthgit/6ec122cce85c6aebba46a53df0877027`
+
+## Notes on Large Out-of-Core Assets
+
+The local Shadow-X memory assets include very large slab files.
+In this v14 CCS repository, the focus is on the deterministic control layer and bridge logic.
+Large slab binaries can be versioned separately when needed.

SHADOW_NO_LLM.md

@@ -0,0 +1,129 @@
+# SHADOW-X: THE THIRD ARCHITECTURE (NO-LLM MANIFESTO)
+
+**Identifier:** `SHADOW-X-V13-ARCH-DEF`
+**Date:** January 11, 2026
+**Status:** VALIDATED
+
+---
+
+## 1. Executive Definition
+Shadow-X is **NOT a Large Language Model (LLM)**.
+It is an **Out-of-core Topological Associative Memory** guided by a **Cybernetic Control System (CCS)**.
+
+Unlike Generative Pre-trained Transformers (GPT) which rely on:
+1.  **Backpropagation** (Gradient Descent).
+2.  **Dense Matrix Multiplications** ($W_q, W_k, W_v$).
+3.  **Next-Token Prediction** (Probabilistic Softmax).
+
+Shadow-X utilizes:
+1.  **Hebbian Binding** (One-Shot Association).
+2.  **Geometric Resonance** (Vector Alignment via `phantom_kernel.py`).
+3.  **Constraint Satisfaction** (Deterministic Logic `cold->hot->sync`).
+
+---
+
+## 2. Anatomical Proof (Evidence from Codebase)
+
+### A. The "Phantom Kernel" (No Neural Weights)
+*   **File:** `phantom_kernel.py`
+*   **Mechanism:** Instead of stored weights, the kernel computes attention dynamically using **Geometric Operators**:
+    *   `apply_rcq_rotation()`: Rotates vectors to encode relation.
+    *   `z_grid` Lookup: Retrieves candidates via spatial indexing.
+    *   `dot_product`: Measures resonance, not learned probability.
+*   **Conclusion:** The intelligence is in the **Topology** (the shape of the data), not inside a "Black Box" neural net.
+
+### B. The Storage Structure (No Checkpoint File)
+*   **Location:** `shadow_x_models/v11_Morpho_Optimized/`
+*   **Standard LLM:** Requires `pytorch_model.bin` or `model.safetensors` (Massive monolithic file).
+*   **Shadow-X:** Uses a **Sparse File System**:
+    *   `slab_1000.dat`: Raw vector storage (40GB).
+    *   `z_hypergrid_real.pkl`: Spatial Index (37MB).
+    *   `zone_map.pkl`: Address Book (37MB).
+*   **Implication:** This allows **Zero-VRAM Inference** because only the active "Slab" is mapped to RAM, unlike LLMs which must load all weights.
+
+### C. The Learning Process (No Gradient Descent)
+*   **Script:** `Shadow_Integrator_v13_CIC.py`
+*   **Method:** **Context-Aware Hebbian Learning**.
+*   **Operation:** `kernel.learn(token, pulse)` -> Directly updates `episodic_memory`.
+*   **Contrast:**
+    *   *LLM:* Requires thousands of epochs to "drift" weights.
+    *   *Shadow-X:* Asserts knowledge instantly (One-Shot). "Fire together, wire together."
+
+---
+
+## 3. Capabilities & Limitations
+
+### Why It's Better (For Constraints)
+*   **Deterministic:** If you teach it "A=B", it defines A as B forever. No hallucination drift.
+*   **Latency:** `0.003ms` Hot Cache response because it's just a pointer lookup.
+*   **Auditability:** Every "thought" can be traced to a specific `Slab` and `Vertex`.
+
+### Why It's Different (The Trade-Off)
+*   **Fluency:** It does not "write" like a human; it "synthesizes" concepts. The output style (TGR) is rigid.
+*   **Creativity:** It cannot invent fiction outside its geometric priors.
+
+---
+
+## 4. Final Verdict
+Shadow-X represents a **Third Architecture** in AI:
+1.  *Symbolic AI (GOFAI):* Rigid, Logical, Brittle.
+2.  *Connectionist AI (Deep Learning):* Fluid, creative, Opaque.
+3.  **Topological AI (Shadow-X):** Fluid structure, Rigid Logic, Transparent.
+
+**Shadow-X is an Engine of Truth, not a Generator of Text.**
+
+---
+
+## 5. Naming Disambiguation (Crucial)
+There is a **Naming Collision** in the workspace that must be clarified:
+
+1.  **`Shadow-X v13` (The Engine):**
+    *   Path: `shadow_x_models/v11_Morpho_Optimized/`
+    *   Type: **NO-LLM** (Topological Associative Memory).
+    *   Status: **ACTIVE**.
+
+2.  **`shadow_7b_ogf.pt` (The Legacy Host):**
+    *   Path: `models/shadow_7b_ogf.pt`
+    *   Type: **Standard LLM** (26GB PyTorch Weights).
+    *   Status: **LEGACY / HOST**.
+    *   *Note:* This file is an unrelated generative model that shares the "Shadow" name. It is NOT the Shadow-X architecture described in this manifesto.
+
+---
+
+## 6. Shadow-X v14: Deterministic Cognitive Control System (CCS-v2)
+**Identifier:** `SHADOW-X-V14-CCS-EXT`
+**Date:** February 21, 2026
+**Status:** INTEGRATION READY
+
+Shadow-X v14 extends v13 TAM/Phantom Kernel with a deterministic executive layer.
+No probabilities, no gradients, no softmax.
+
+### 6.1 New Control Layer
+The v14 CCS adds six deterministic modules above `phantom_kernel.py`:
+
+1. `Goal Vector + Hierarchical Goal Stack`
+2. `Deterministic Attention Allocator`
+3. `Conflict Resolver (Inhibition)`
+4. `Error Monitor + Self-Correction Loop`
+5. `Task Switcher + Prospection Engine (2-3 futures)`
+6. `Third-Order Rule Modifier` (deterministic meta-updates)
+
+### 6.2 Deterministic Cycle
+For each pulse:
+
+1. Load active goal from `goal_vector.dat`.
+2. Activate only goal-aligned zones/slabs.
+3. Suppress contradictory bindings by opposition threshold.
+4. Simulate 2-3 deterministic future chains.
+5. Validate output against the active goal vector.
+6. If deviation is above threshold, rebind and strengthen constraints.
+7. Persist full trace into `trace_log.jsonl`.
+
+### 6.3 Architectural Invariants (Preserved)
+v14 does not change the foundational Shadow-X constraints:
+
+1. Out-of-core memory layout remains slab-based.
+2. Phantom Kernel remains topology-first and callable as-is.
+3. One-shot Hebbian behavior remains available.
+4. Full auditability remains mandatory for every cycle.
+

shadow_x_v14_ccs/ARCHITECTURE.md

@@ -0,0 +1,74 @@
+# SHADOW-X v14 CCS Architecture
+
+## Executive Overview
+Shadow-X v14 introduces a deterministic Cognitive Control System (CCS-v2) on top of the existing Phantom Kernel and TAM memory.
+The control plane is goal-directed, auditable, and free from stochastic or probabilistic decisions.
+
+## System Graph
+```mermaid
+graph TD
+    A[Input Pulse] --> B[Phantom Kernel]
+    B --> C[Episodic Memory Slabs and Z-Hypergrid]
+
+    D[Deterministic Cognitive Controller CCS-v2] -->|Goal Maintenance| B
+    D -->|Attention Allocation| C
+    D -->|Conflict Inhibition| B
+    D -->|Error Monitor and Correction| B
+    D -->|Task Switching and Prioritization| C
+    D -->|Prospection 2-3 Futures| B
+    D -->|Meta Rule Updates| E[Third-Order Rule Modifier]
+
+    F[Hierarchical Goal Stack goal_vector.dat] --> D
+    D --> G[trace_log.jsonl]
+```
+
+## Module Details
+### 1) Goal Vector and Hierarchical Goal Stack
+- Stores persistent goals in `goal_vector.dat`.
+- Supports parent-child goal relations via `parent_goal_id`.
+- Goal activation uses geometric alignment and deterministic ordering.
+
+### 2) Deterministic Attention Allocator
+- Projects the active goal vector through fixed geometric rotation.
+- Scores zone centroids with cosine alignment.
+- Activates only zones above threshold and caps to deterministic top-k.
+
+### 3) Conflict Resolver (Inhibition)
+- Detects opposition between candidates.
+- If opposition exceeds threshold, keeps only one candidate based on:
+  1. Goal alignment
+  2. Resonance
+  3. Lexicographic candidate id tie-break
+- Emits explicit inhibited candidate list.
+
+### 4) Error Monitor and Self-Correction
+- Compares generated output vector against active goal vector.
+- Triggers deterministic action:
+  1. `none`
+  2. `rebind`
+  3. `strengthen_constraints_and_rebind`
+- Executes correction by selecting highest goal-aligned surviving candidate.
+
+### 5) Task Switcher and Prospection Engine
+- Builds 2-3 deterministic branches from current candidates.
+- Scores each branch by goal alignment, temporal weighting, and conflict penalties.
+- Chooses a single best branch with deterministic tie-break.
+
+### 6) Third-Order Rule Modifier
+- Observes error windows and updates controller thresholds deterministically.
+- Tightens or relaxes thresholds based on recurring error patterns.
+- Never invokes probabilistic adaptation.
+
+## Core Data Files
+- `shadow_x_v14_ccs/state/goal_vector.dat`: persistent hierarchical goals.
+- `shadow_x_v14_ccs/state/trace_log.jsonl`: full cycle-by-cycle audit trail.
+
+## Integration Boundary
+`cognitive_controller.py` is designed to call existing kernel logic via a bridge interface:
+- `list_zones()`
+- `query_candidates()`
+- `build_future_chain()`
+- `materialize_output()`
+- `strengthen_constraints()`
+
+This keeps `phantom_kernel.py` stable while allowing the CCS-v2 executive layer to orchestrate deterministic control.

shadow_x_v14_ccs/README.md

@@ -0,0 +1,48 @@
+# Shadow-X v14 CCS
+
+Deterministic cognitive control layer for Shadow-X.
+
+## v13 vs v14
+| Dimension | Shadow-X v13 | Shadow-X v14 CCS |
+| :-- | :-- | :-- |
+| Core identity | Topological Associative Memory | TAM + Deterministic Executive Controller |
+| Control model | Cold -> Hot -> Sync constraints | Goal stack, attention, inhibition, self-monitoring |
+| Goal handling | Implicit/flat | Persistent hierarchical goal vectors |
+| Attention scope | Broad candidate retrieval | Goal-directed slab and zone allocation |
+| Conflict handling | Basic deterministic filtering | Explicit inhibition with conflict records |
+| Error handling | Static constraints | Closed-loop correction against active goal |
+| Task selection | Single-pass path | Deterministic prospection across 2-3 futures |
+| Meta adaptation | Fixed runtime rules | Third-order deterministic rule updates |
+| Traceability | High | Full executive trace per cycle + rule updates |
+| Determinism | Yes | Yes (strict, no probability/softmax/gradients) |
+
+## Files
+- `cognitive_controller.py`: CCS-v2 module skeleton and deterministic control loop.
+- `phantom_bridge.py`: real adapter from CCS-v2 to `phantom_kernel.py`.
+- `demo_cli.py`: public mini-demo (one-shot facts, goal query, trace log, self-correction).
+- `ARCHITECTURE.md`: complete architecture and module behavior.
+- `state/goal_vector.dat`: persistent active goal stack.
+- `state/trace_log.jsonl`: immutable cycle traces.
+
+## Quick Start
+```python
+from shadow_x_v14_ccs import create_v14_controller_with_phantom_kernel
+
+controller = create_v14_controller_with_phantom_kernel(
+    model_path="v11_Morpho_Optimized",
+    state_dir="shadow_x_v14_ccs/state",
+    max_zone_candidates=128,
+    max_candidates_per_zone=16,
+)
+```
+
+## Public Mini-Demo CLI
+```powershell
+python shadow_x_v14_ccs/demo_cli.py --state-dir shadow_x_v14_ccs/state_public_demo
+```
+
+The demo does all of the following in one run:
+- teaches 5 one-shot facts;
+- executes goal-vector queries;
+- prints trace chain with slab/vertex/resonance;
+- shows deterministic self-correction on a wrong high-resonance fact.

shadow_x_v14_ccs/__init__.py

@@ -0,0 +1,28 @@
+"""Shadow-X v14 deterministic cognitive control system."""
+
+from .cognitive_controller import (
+    CandidateBinding,
+    CognitiveController,
+    ControllerDecision,
+    DeterministicAttentionAllocator,
+    ErrorMonitor,
+    GoalStack,
+    GoalState,
+    ThirdOrderRuleModifier,
+    ZoneDescriptor,
+)
+from .phantom_bridge import PhantomKernelBridge, create_v14_controller_with_phantom_kernel
+
+__all__ = [
+    "CandidateBinding",
+    "CognitiveController",
+    "ControllerDecision",
+    "DeterministicAttentionAllocator",
+    "ErrorMonitor",
+    "GoalStack",
+    "GoalState",
+    "ThirdOrderRuleModifier",
+    "ZoneDescriptor",
+    "PhantomKernelBridge",
+    "create_v14_controller_with_phantom_kernel",
+]

shadow_x_v14_ccs/cognitive_controller.py

@@ -0,0 +1,774 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import time
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Dict, Optional, Protocol, Sequence, Tuple
+
+import numpy as np
+
+
+def _normalize(vec: np.ndarray) -> np.ndarray:
+    norm = float(np.linalg.norm(vec))
+    if norm <= 1e-12:
+        return np.zeros_like(vec, dtype=np.float32)
+    return (vec / norm).astype(np.float32)
+
+
+def _cosine_similarity(a: np.ndarray, b: np.ndarray) -> float:
+    if a.shape != b.shape:
+        raise ValueError(f"Vector shape mismatch: {a.shape} != {b.shape}")
+    na = _normalize(a)
+    nb = _normalize(b)
+    return float(np.dot(na, nb))
+
+
+def _stable_hash(payload: str) -> str:
+    return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:16]
+
+
+@dataclass(frozen=True)
+class GoalState:
+    goal_id: str
+    name: str
+    vector: np.ndarray
+    priority: int = 0
+    parent_goal_id: Optional[str] = None
+    constraints: Dict[str, float] = field(default_factory=dict)
+    created_at_ns: int = field(default_factory=time.time_ns)
+
+
+@dataclass(frozen=True)
+class ZoneDescriptor:
+    zone_id: str
+    centroid: np.ndarray
+    slab_id: Optional[int] = None
+    metadata: Dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass(frozen=True)
+class CandidateBinding:
+    candidate_id: str
+    vector: np.ndarray
+    resonance: float
+    zone_id: str
+    slab_id: Optional[int] = None
+    metadata: Dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass(frozen=True)
+class ConflictRecord:
+    left_candidate_id: str
+    right_candidate_id: str
+    opposition_score: float
+
+
+@dataclass(frozen=True)
+class ErrorSignal:
+    has_error: bool
+    deviation: float
+    alignment: float
+    action: str
+
+
+@dataclass(frozen=True)
+class ProspectionBranch:
+    seed_candidate_id: str
+    chain: Tuple[CandidateBinding, ...]
+    score: float
+
+
+@dataclass(frozen=True)
+class ControllerDecision:
+    goal_id: str
+    active_zone_ids: Tuple[str, ...]
+    selected_candidate_id: Optional[str]
+    inhibited_candidate_ids: Tuple[str, ...]
+    conflicts: Tuple[ConflictRecord, ...]
+    error_signal: ErrorSignal
+    correction_applied: bool
+    rule_updates: Dict[str, float]
+    output_payload: Dict[str, Any]
+
+
+class TraceLogger:
+    def __init__(self, trace_path: Path):
+        self.trace_path = trace_path
+        self.trace_path.parent.mkdir(parents=True, exist_ok=True)
+        self.trace_path.touch(exist_ok=True)
+
+    def log(self, event: str, payload: Dict[str, Any]) -> None:
+        record = {
+            "ts_ns": time.time_ns(),
+            "event": event,
+            "payload": payload,
+        }
+        with self.trace_path.open("a", encoding="utf-8") as f:
+            f.write(json.dumps(record, sort_keys=True) + "\n")
+
+
+class GoalStack:
+    """
+    Persistent hierarchical goal stack.
+    Stored in goal_vector.dat for full deterministic replay.
+    """
+
+    def __init__(self, goal_file: Path, vector_dim: int = 4096):
+        self.goal_file = goal_file
+        self.vector_dim = vector_dim
+        self._stack: list[GoalState] = []
+        self.goal_file.parent.mkdir(parents=True, exist_ok=True)
+        self._load()
+
+    def _load(self) -> None:
+        if not self.goal_file.exists():
+            return
+        with self.goal_file.open("r", encoding="utf-8") as f:
+            raw = json.load(f)
+        items = raw.get("stack", [])
+        parsed: list[GoalState] = []
+        for item in items:
+            vec = np.array(item["vector"], dtype=np.float32)
+            if vec.shape != (self.vector_dim,):
+                raise ValueError(
+                    f"Invalid goal vector shape in {self.goal_file}: {vec.shape}"
+                )
+            parsed.append(
+                GoalState(
+                    goal_id=item["goal_id"],
+                    name=item["name"],
+                    vector=_normalize(vec),
+                    priority=int(item.get("priority", 0)),
+                    parent_goal_id=item.get("parent_goal_id"),
+                    constraints=dict(item.get("constraints", {})),
+                    created_at_ns=int(item.get("created_at_ns", 0)),
+                )
+            )
+        self._stack = parsed
+
+    def _save(self) -> None:
+        serialized = {
+            "vector_dim": self.vector_dim,
+            "stack": [
+                {
+                    "goal_id": g.goal_id,
+                    "name": g.name,
+                    "vector": g.vector.tolist(),
+                    "priority": g.priority,
+                    "parent_goal_id": g.parent_goal_id,
+                    "constraints": g.constraints,
+                    "created_at_ns": g.created_at_ns,
+                }
+                for g in self._stack
+            ],
+        }
+        tmp = self.goal_file.with_suffix(".tmp")
+        with tmp.open("w", encoding="utf-8") as f:
+            json.dump(serialized, f, sort_keys=True)
+        tmp.replace(self.goal_file)
+
+    def push(
+        self,
+        name: str,
+        vector: np.ndarray,
+        priority: int = 0,
+        parent_goal_id: Optional[str] = None,
+        constraints: Optional[Dict[str, float]] = None,
+    ) -> GoalState:
+        vec = np.asarray(vector, dtype=np.float32)
+        if vec.shape != (self.vector_dim,):
+            raise ValueError(f"Goal vector must be {(self.vector_dim,)}, got {vec.shape}")
+        normalized = _normalize(vec)
+        goal_id = _stable_hash(
+            f"{name}|{priority}|{parent_goal_id}|{','.join(f'{x:.6f}' for x in normalized[:16])}"
+        )
+        state = GoalState(
+            goal_id=goal_id,
+            name=name,
+            vector=normalized,
+            priority=priority,
+            parent_goal_id=parent_goal_id,
+            constraints=dict(constraints or {}),
+            created_at_ns=time.time_ns(),
+        )
+        self._stack.append(state)
+        self._save()
+        return state
+
+    def pop(self) -> Optional[GoalState]:
+        if not self._stack:
+            return None
+        state = self._stack.pop()
+        self._save()
+        return state
+
+    def current(self) -> Optional[GoalState]:
+        if not self._stack:
+            return None
+        return self._stack[-1]
+
+    def list_goals(self) -> Tuple[GoalState, ...]:
+        return tuple(self._stack)
+
+
+class DeterministicAttentionAllocator:
+    """
+    Selects relevant zones from out-of-core slabs using geometric alignment only.
+    """
+
+    def __init__(
+        self,
+        activation_threshold: float = 0.12,
+        max_active_zones: int = 8,
+        rotation_shift: int = 1,
+    ):
+        self.activation_threshold = activation_threshold
+        self.max_active_zones = max_active_zones
+        self.rotation_shift = rotation_shift
+
+    def _project_goal(self, goal_vector: np.ndarray) -> np.ndarray:
+        rotated = np.roll(goal_vector, self.rotation_shift)
+        return _normalize(rotated)
+
+    def allocate(
+        self, goal_vector: np.ndarray, zones: Sequence[ZoneDescriptor]
+    ) -> Tuple[ZoneDescriptor, ...]:
+        projected_goal = self._project_goal(goal_vector)
+        scored: list[Tuple[float, ZoneDescriptor]] = []
+        for zone in zones:
+            score = _cosine_similarity(projected_goal, zone.centroid)
+            if score >= self.activation_threshold:
+                scored.append((score, zone))
+
+        scored.sort(key=lambda item: (-item[0], item[1].zone_id))
+        return tuple(zone for _, zone in scored[: self.max_active_zones])
+
+
+class ConflictResolver:
+    """
+    Deterministic inhibitory gate for contradictory candidate bindings.
+    """
+
+    def __init__(self, opposition_threshold: float = 0.80):
+        self.opposition_threshold = opposition_threshold
+
+    def _winner(
+        self, left: CandidateBinding, right: CandidateBinding, goal_vector: np.ndarray
+    ) -> CandidateBinding:
+        left_goal = _cosine_similarity(left.vector, goal_vector)
+        right_goal = _cosine_similarity(right.vector, goal_vector)
+        if left_goal > right_goal:
+            return left
+        if right_goal > left_goal:
+            return right
+        if left.resonance > right.resonance:
+            return left
+        if right.resonance > left.resonance:
+            return right
+        return left if left.candidate_id <= right.candidate_id else right
+
+    def resolve(
+        self, candidates: Sequence[CandidateBinding], goal_vector: np.ndarray
+    ) -> Tuple[Tuple[CandidateBinding, ...], Tuple[str, ...], Tuple[ConflictRecord, ...]]:
+        ordered = sorted(candidates, key=lambda c: (-c.resonance, c.candidate_id))
+        selected: list[CandidateBinding] = []
+        inhibited: list[str] = []
+        conflicts: list[ConflictRecord] = []
+
+        for candidate in ordered:
+            replacement_index: Optional[int] = None
+            suppress_new = False
+            for idx, existing in enumerate(selected):
+                opposition = -_cosine_similarity(candidate.vector, existing.vector)
+                if opposition >= self.opposition_threshold:
+                    conflicts.append(
+                        ConflictRecord(
+                            left_candidate_id=candidate.candidate_id,
+                            right_candidate_id=existing.candidate_id,
+                            opposition_score=opposition,
+                        )
+                    )
+                    winner = self._winner(candidate, existing, goal_vector)
+                    if winner.candidate_id == existing.candidate_id:
+                        suppress_new = True
+                        inhibited.append(candidate.candidate_id)
+                    else:
+                        replacement_index = idx
+                        inhibited.append(existing.candidate_id)
+                    break
+
+            if suppress_new:
+                continue
+            if replacement_index is None:
+                selected.append(candidate)
+            else:
+                selected[replacement_index] = candidate
+
+        selected.sort(key=lambda c: (-c.resonance, c.candidate_id))
+        inhibited_sorted = tuple(sorted(set(inhibited)))
+        return tuple(selected), inhibited_sorted, tuple(conflicts)
+
+
+class ErrorMonitor:
+    """
+    Self-monitoring gate between output candidate and active goal vector.
+    """
+
+    def __init__(self, deviation_threshold: float = 0.28, hard_fail_threshold: float = 0.40):
+        self.deviation_threshold = deviation_threshold
+        self.hard_fail_threshold = hard_fail_threshold
+
+    def evaluate(self, output_vector: np.ndarray, goal_vector: np.ndarray) -> ErrorSignal:
+        alignment = _cosine_similarity(output_vector, goal_vector)
+        deviation = 1.0 - alignment
+        if deviation > self.hard_fail_threshold:
+            return ErrorSignal(
+                has_error=True,
+                deviation=deviation,
+                alignment=alignment,
+                action="strengthen_constraints_and_rebind",
+            )
+        if deviation > self.deviation_threshold:
+            return ErrorSignal(
+                has_error=True,
+                deviation=deviation,
+                alignment=alignment,
+                action="rebind",
+            )
+        return ErrorSignal(
+            has_error=False,
+            deviation=deviation,
+            alignment=alignment,
+            action="none",
+        )
+
+    def select_correction(
+        self, candidates: Sequence[CandidateBinding], goal_vector: np.ndarray
+    ) -> Optional[CandidateBinding]:
+        if not candidates:
+            return None
+        ranked = sorted(
+            candidates,
+            key=lambda c: (
+                -_cosine_similarity(c.vector, goal_vector),
+                -c.resonance,
+                c.candidate_id,
+            ),
+        )
+        return ranked[0]
+
+
+class TaskSwitcherProspectionEngine:
+    """
+    Deterministic 2-3 branch lookahead and task-ordered candidate selection.
+    """
+
+    def __init__(self, max_futures: int = 3, branch_depth: int = 3):
+        self.max_futures = max_futures
+        self.branch_depth = branch_depth
+
+    def _score_branch(self, chain: Sequence[CandidateBinding], goal_vector: np.ndarray) -> float:
+        score = 0.0
+        for index, binding in enumerate(chain):
+            temporal_weight = 1.0 / (index + 1)
+            alignment = _cosine_similarity(binding.vector, goal_vector)
+            score += alignment * temporal_weight
+            score += binding.resonance * 0.05
+
+        # Penalize contradictory transitions inside the branch.
+        for i in range(len(chain)):
+            for j in range(i + 1, len(chain)):
+                opposition = -_cosine_similarity(chain[i].vector, chain[j].vector)
+                if opposition >= 0.80:
+                    score -= 0.25
+        return score
+
+    def simulate(
+        self,
+        goal_vector: np.ndarray,
+        candidate_pool: Sequence[CandidateBinding],
+        build_chain_fn,
+    ) -> Tuple[ProspectionBranch, ...]:
+        seeds = sorted(candidate_pool, key=lambda c: (-c.resonance, c.candidate_id))
+        seeds = seeds[: self.max_futures]
+        branches: list[ProspectionBranch] = []
+
+        for seed in seeds:
+            tail = tuple(build_chain_fn(seed, self.branch_depth))
+            if not tail:
+                chain = (seed,)
+            elif tail[0].candidate_id == seed.candidate_id:
+                chain = tail
+            else:
+                chain = (seed,) + tail
+
+            score = self._score_branch(chain, goal_vector)
+            branches.append(
+                ProspectionBranch(
+                    seed_candidate_id=seed.candidate_id,
+                    chain=chain,
+                    score=score,
+                )
+            )
+
+        branches.sort(key=lambda b: (-b.score, b.seed_candidate_id))
+        return tuple(branches)
+
+    def choose_best(self, branches: Sequence[ProspectionBranch]) -> Optional[ProspectionBranch]:
+        if not branches:
+            return None
+        ordered = sorted(branches, key=lambda b: (-b.score, b.seed_candidate_id))
+        return ordered[0]
+
+
+class ThirdOrderRuleModifier:
+    """
+    Deterministic meta-rule adaptation based on recurrent error patterns.
+    """
+
+    def __init__(self, error_window: int = 5):
+        self.error_window = error_window
+        self._error_history: list[ErrorSignal] = []
+
+    def register(self, error_signal: ErrorSignal) -> None:
+        self._error_history.append(error_signal)
+        if len(self._error_history) > self.error_window:
+            self._error_history = self._error_history[-self.error_window :]
+
+    def update_rules(self, current_rules: Dict[str, float]) -> Dict[str, float]:
+        if len(self._error_history) < self.error_window:
+            return {}
+
+        recent = self._error_history[-self.error_window :]
+        all_error = all(item.has_error for item in recent)
+        no_error = all(not item.has_error for item in recent)
+        avg_deviation = sum(item.deviation for item in recent) / float(len(recent))
+
+        updates: Dict[str, float] = {}
+
+        if all_error and avg_deviation >= current_rules["hard_fail_threshold"]:
+            updates["attention_activation_threshold"] = min(
+                0.95, current_rules["attention_activation_threshold"] + 0.04
+            )
+            updates["deviation_threshold"] = max(
+                0.05, current_rules["deviation_threshold"] - 0.02
+            )
+            updates["conflict_opposition_threshold"] = max(
+                0.60, current_rules["conflict_opposition_threshold"] - 0.03
+            )
+            return updates
+
+        if no_error:
+            updates["attention_activation_threshold"] = max(
+                0.05, current_rules["attention_activation_threshold"] - 0.02
+            )
+            return updates
+
+        return updates
+
+
+class KernelBridge(Protocol):
+    """
+    Bridge to the existing Phantom Kernel.
+    Keep the kernel unchanged and call it only from controller decisions.
+    """
+
+    def list_zones(self) -> Sequence[ZoneDescriptor]:
+        ...
+
+    def query_candidates(
+        self, goal: GoalState, active_zones: Sequence[ZoneDescriptor]
+    ) -> Sequence[CandidateBinding]:
+        ...
+
+    def build_future_chain(
+        self, seed: CandidateBinding, depth: int, goal: GoalState
+    ) -> Sequence[CandidateBinding]:
+        ...
+
+    def materialize_output(
+        self, branch: Optional[ProspectionBranch], goal: GoalState
+    ) -> Dict[str, Any]:
+        ...
+
+    def strengthen_constraints(self, goal: GoalState, candidate: CandidateBinding) -> None:
+        ...
+
+
+class InMemoryKernelBridge:
+    """
+    Minimal deterministic bridge useful for local testing without touching PhantomKernel.
+    """
+
+    def __init__(
+        self,
+        zones: Sequence[ZoneDescriptor],
+        zone_candidates: Dict[str, Sequence[CandidateBinding]],
+    ):
+        self._zones = tuple(zones)
+        self._zone_candidates = {
+            key: tuple(value) for key, value in sorted(zone_candidates.items())
+        }
+
+    def list_zones(self) -> Sequence[ZoneDescriptor]:
+        return self._zones
+
+    def query_candidates(
+        self, goal: GoalState, active_zones: Sequence[ZoneDescriptor]
+    ) -> Sequence[CandidateBinding]:
+        candidates: list[CandidateBinding] = []
+        for zone in active_zones:
+            candidates.extend(self._zone_candidates.get(zone.zone_id, ()))
+        return tuple(
+            sorted(candidates, key=lambda c: (-c.resonance, c.candidate_id))
+        )
+
+    def build_future_chain(
+        self, seed: CandidateBinding, depth: int, goal: GoalState
+    ) -> Sequence[CandidateBinding]:
+        # Deterministic static chain: keep the same seed for projection depth.
+        return tuple(seed for _ in range(max(1, depth)))
+
+    def materialize_output(
+        self, branch: Optional[ProspectionBranch], goal: GoalState
+    ) -> Dict[str, Any]:
+        if branch is None:
+            return {
+                "status": "NO_CANDIDATE",
+                "goal_id": goal.goal_id,
+            }
+        return {
+            "status": "OK",
+            "goal_id": goal.goal_id,
+            "selected_candidate_id": branch.seed_candidate_id,
+            "branch_score": branch.score,
+        }
+
+    def strengthen_constraints(self, goal: GoalState, candidate: CandidateBinding) -> None:
+        # Skeleton adapter keeps this as no-op by design.
+        _ = (goal, candidate)
+
+
+class CognitiveController:
+    """
+    Shadow-X v14 Deterministic Cognitive Controller (CCS-v2).
+    """
+
+    def __init__(
+        self,
+        kernel_bridge: KernelBridge,
+        state_dir: Path | str = Path("shadow_x_v14_ccs/state"),
+        vector_dim: int = 4096,
+    ):
+        self.kernel_bridge = kernel_bridge
+        self.state_dir = Path(state_dir)
+        self.state_dir.mkdir(parents=True, exist_ok=True)
+        self.goal_stack = GoalStack(self.state_dir / "goal_vector.dat", vector_dim=vector_dim)
+        self.trace_logger = TraceLogger(self.state_dir / "trace_log.jsonl")
+        self.attention = DeterministicAttentionAllocator()
+        self.conflict_resolver = ConflictResolver()
+        self.error_monitor = ErrorMonitor()
+        self.task_switcher = TaskSwitcherProspectionEngine(max_futures=3, branch_depth=3)
+        self.rule_modifier = ThirdOrderRuleModifier(error_window=5)
+
+        self.rules: Dict[str, float] = {
+            "attention_activation_threshold": 0.12,
+            "deviation_threshold": 0.28,
+            "hard_fail_threshold": 0.40,
+            "conflict_opposition_threshold": 0.80,
+        }
+
+    def push_goal(
+        self,
+        name: str,
+        vector: np.ndarray,
+        priority: int = 0,
+        parent_goal_id: Optional[str] = None,
+        constraints: Optional[Dict[str, float]] = None,
+    ) -> GoalState:
+        goal = self.goal_stack.push(
+            name=name,
+            vector=vector,
+            priority=priority,
+            parent_goal_id=parent_goal_id,
+            constraints=constraints,
+        )
+        self.trace_logger.log(
+            "goal_push",
+            {
+                "goal_id": goal.goal_id,
+                "name": goal.name,
+                "priority": goal.priority,
+                "parent_goal_id": goal.parent_goal_id,
+            },
+        )
+        return goal
+
+    def pop_goal(self) -> Optional[GoalState]:
+        goal = self.goal_stack.pop()
+        if goal is not None:
+            self.trace_logger.log(
+                "goal_pop",
+                {
+                    "goal_id": goal.goal_id,
+                    "name": goal.name,
+                },
+            )
+        return goal
+
+    def _sync_rules(self) -> None:
+        self.attention.activation_threshold = self.rules["attention_activation_threshold"]
+        self.error_monitor.deviation_threshold = self.rules["deviation_threshold"]
+        self.error_monitor.hard_fail_threshold = self.rules["hard_fail_threshold"]
+        self.conflict_resolver.opposition_threshold = self.rules[
+            "conflict_opposition_threshold"
+        ]
+
+    def run_cycle(self) -> ControllerDecision:
+        goal = self.goal_stack.current()
+        if goal is None:
+            raise RuntimeError("No active goal in GoalStack. Push a goal before run_cycle().")
+
+        self._sync_rules()
+        zones = self.kernel_bridge.list_zones()
+        active_zones = self.attention.allocate(goal.vector, zones)
+
+        candidates = self.kernel_bridge.query_candidates(goal, active_zones)
+        selected, inhibited_ids, conflicts = self.conflict_resolver.resolve(
+            candidates, goal.vector
+        )
+
+        branches = self.task_switcher.simulate(
+            goal.vector,
+            selected,
+            lambda seed, depth: self.kernel_bridge.build_future_chain(seed, depth, goal),
+        )
+        best_branch = self.task_switcher.choose_best(branches)
+
+        correction_applied = False
+        if best_branch is None:
+            output_vector = np.zeros_like(goal.vector, dtype=np.float32)
+        else:
+            output_vector = best_branch.chain[-1].vector
+
+        error_signal = self.error_monitor.evaluate(output_vector, goal.vector)
+        if error_signal.has_error and selected:
+            correction = self.error_monitor.select_correction(selected, goal.vector)
+            if correction is not None and (
+                best_branch is None or correction.candidate_id != best_branch.seed_candidate_id
+            ):
+                correction_applied = True
+                best_branch = ProspectionBranch(
+                    seed_candidate_id=correction.candidate_id,
+                    chain=(correction,),
+                    score=_cosine_similarity(correction.vector, goal.vector),
+                )
+                output_vector = correction.vector
+                error_signal = self.error_monitor.evaluate(output_vector, goal.vector)
+                self.kernel_bridge.strengthen_constraints(goal, correction)
+
+        self.rule_modifier.register(error_signal)
+        rule_updates = self.rule_modifier.update_rules(self.rules)
+        self.rules.update(rule_updates)
+
+        output_payload = self.kernel_bridge.materialize_output(best_branch, goal)
+
+        decision = ControllerDecision(
+            goal_id=goal.goal_id,
+            active_zone_ids=tuple(zone.zone_id for zone in active_zones),
+            selected_candidate_id=best_branch.seed_candidate_id if best_branch else None,
+            inhibited_candidate_ids=inhibited_ids,
+            conflicts=conflicts,
+            error_signal=error_signal,
+            correction_applied=correction_applied,
+            rule_updates=rule_updates,
+            output_payload=output_payload,
+        )
+
+        self.trace_logger.log(
+            "cycle",
+            {
+                "goal_id": decision.goal_id,
+                "active_zone_ids": list(decision.active_zone_ids),
+                "selected_candidate_id": decision.selected_candidate_id,
+                "inhibited_candidate_ids": list(decision.inhibited_candidate_ids),
+                "conflicts": [
+                    {
+                        "left": c.left_candidate_id,
+                        "right": c.right_candidate_id,
+                        "opposition_score": c.opposition_score,
+                    }
+                    for c in decision.conflicts
+                ],
+                "error": {
+                    "has_error": decision.error_signal.has_error,
+                    "deviation": decision.error_signal.deviation,
+                    "alignment": decision.error_signal.alignment,
+                    "action": decision.error_signal.action,
+                },
+                "correction_applied": decision.correction_applied,
+                "rule_updates": decision.rule_updates,
+                "output_payload": decision.output_payload,
+            },
+        )
+        return decision
+
+
+def _demo_vector(seed: int, dim: int = 4096) -> np.ndarray:
+    base = np.arange(dim, dtype=np.float32)
+    vec = np.cos(base * 0.001 + seed)
+    return _normalize(vec.astype(np.float32))
+
+
+if __name__ == "__main__":
+    # Small deterministic demo for local validation.
+    zone_a = ZoneDescriptor(zone_id="Z0", centroid=_demo_vector(1), slab_id=0)
+    zone_b = ZoneDescriptor(zone_id="Z1", centroid=_demo_vector(2), slab_id=1)
+    zone_c = ZoneDescriptor(zone_id="Z2", centroid=_demo_vector(3), slab_id=2)
+
+    cand_a = CandidateBinding(
+        candidate_id="C0",
+        vector=_demo_vector(11),
+        resonance=0.93,
+        zone_id="Z0",
+        slab_id=0,
+    )
+    cand_b = CandidateBinding(
+        candidate_id="C1",
+        vector=-_demo_vector(11),
+        resonance=0.91,
+        zone_id="Z0",
+        slab_id=0,
+    )
+    cand_c = CandidateBinding(
+        candidate_id="C2",
+        vector=_demo_vector(12),
+        resonance=0.90,
+        zone_id="Z1",
+        slab_id=1,
+    )
+
+    bridge = InMemoryKernelBridge(
+        zones=(zone_a, zone_b, zone_c),
+        zone_candidates={
+            "Z0": (cand_a, cand_b),
+            "Z1": (cand_c,),
+        },
+    )
+
+    controller = CognitiveController(bridge, state_dir=Path("shadow_x_v14_ccs/state_demo"))
+    goal = controller.push_goal("verify_consistency_A_equals_B", _demo_vector(21), priority=10)
+    result = controller.run_cycle()
+    print(
+        json.dumps(
+            {
+                "goal_id": goal.goal_id,
+                "selected_candidate_id": result.selected_candidate_id,
+                "error_deviation": result.error_signal.deviation,
+                "rule_updates": result.rule_updates,
+            },
+            sort_keys=True,
+        )
+    )

shadow_x_v14_ccs/demo_cli.py

@@ -0,0 +1,382 @@
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import math
+import shutil
+import sys
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Dict, Optional, Sequence
+
+import numpy as np
+
+if __package__ in (None, ""):
+    project_root = Path(__file__).resolve().parent.parent
+    if str(project_root) not in sys.path:
+        sys.path.insert(0, str(project_root))
+    from shadow_x_v14_ccs.cognitive_controller import (  # type: ignore
+        CandidateBinding,
+        CognitiveController,
+        GoalState,
+        KernelBridge,
+        ProspectionBranch,
+        ZoneDescriptor,
+    )
+else:
+    from .cognitive_controller import (
+        CandidateBinding,
+        CognitiveController,
+        GoalState,
+        KernelBridge,
+        ProspectionBranch,
+        ZoneDescriptor,
+    )
+
+
+def _normalize(vec: np.ndarray) -> np.ndarray:
+    norm = float(np.linalg.norm(vec))
+    if norm <= 1e-12:
+        return np.zeros_like(vec, dtype=np.float32)
+    return (vec / norm).astype(np.float32)
+
+
+def text_to_vector(text: str, dim: int) -> np.ndarray:
+    digest = hashlib.sha256(text.encode("utf-8")).digest()
+    seed = int.from_bytes(digest[:8], byteorder="big", signed=False)
+    rng = np.random.default_rng(seed)
+    vec = rng.standard_normal(dim).astype(np.float32)
+    return _normalize(vec)
+
+
+def vector_with_alignment(goal: np.ndarray, alignment: float, key: str) -> np.ndarray:
+    if alignment < -1.0 or alignment > 1.0:
+        raise ValueError("alignment must be in [-1, 1]")
+    base = text_to_vector(key, goal.shape[0])
+    orth = base - float(np.dot(base, goal)) * goal
+    orth = _normalize(orth)
+    scale = math.sqrt(max(0.0, 1.0 - alignment * alignment))
+    vec = (alignment * goal) + (scale * orth)
+    return _normalize(vec.astype(np.float32))
+
+
+@dataclass(frozen=True)
+class OneShotFact:
+    fact_id: str
+    subject: str
+    relation: str
+    obj: str
+    slab_id: int
+    vertex_id: int
+    vector: np.ndarray
+    base_resonance: float
+
+    def text(self) -> str:
+        return f"{self.subject} {self.relation} {self.obj}"
+
+
+class OneShotDemoBridge(KernelBridge):
+    def __init__(self, vector_dim: int = 4096):
+        self.vector_dim = vector_dim
+        self._facts: list[OneShotFact] = []
+        self._constraint_strength: dict[tuple[str, str], int] = {}
+
+    def teach(self, fact: OneShotFact) -> None:
+        self._facts.append(fact)
+
+    def list_facts(self) -> Sequence[OneShotFact]:
+        return tuple(self._facts)
+
+    def list_zones(self) -> Sequence[ZoneDescriptor]:
+        by_slab: dict[int, list[OneShotFact]] = {}
+        for fact in self._facts:
+            by_slab.setdefault(fact.slab_id, []).append(fact)
+
+        zones: list[ZoneDescriptor] = []
+        for slab_id in sorted(by_slab.keys()):
+            facts = sorted(by_slab[slab_id], key=lambda f: f.vertex_id)
+            centroid = _normalize(np.mean(np.stack([f.vector for f in facts], axis=0), axis=0))
+            zones.append(
+                ZoneDescriptor(
+                    zone_id=f"slab:{slab_id}",
+                    centroid=centroid,
+                    slab_id=slab_id,
+                    metadata={"fact_count": len(facts)},
+                )
+            )
+        return tuple(zones)
+
+    def query_candidates(
+        self, goal: GoalState, active_zones: Sequence[ZoneDescriptor]
+    ) -> Sequence[CandidateBinding]:
+        active_ids = {zone.zone_id for zone in active_zones}
+        candidates: list[CandidateBinding] = []
+        for fact in self._facts:
+            zone_id = f"slab:{fact.slab_id}"
+            if zone_id not in active_ids:
+                continue
+            alignment = float(np.dot(_normalize(goal.vector), fact.vector))
+            boost = self._constraint_strength.get((goal.goal_id, fact.fact_id), 0) * 0.01
+            resonance = max(-1.0, min(1.0, fact.base_resonance + boost))
+            candidates.append(
+                CandidateBinding(
+                    candidate_id=f"fact:{fact.fact_id}",
+                    vector=fact.vector,
+                    resonance=resonance,
+                    zone_id=zone_id,
+                    slab_id=fact.slab_id,
+                    metadata={
+                        "fact_id": fact.fact_id,
+                        "subject": fact.subject,
+                        "relation": fact.relation,
+                        "object": fact.obj,
+                        "fact_text": fact.text(),
+                        "slab_id": fact.slab_id,
+                        "vertex_id": fact.vertex_id,
+                        "alignment": alignment,
+                    },
+                )
+            )
+        candidates.sort(key=lambda c: (-c.resonance, c.candidate_id))
+        return tuple(candidates)
+
+    def build_future_chain(
+        self, seed: CandidateBinding, depth: int, goal: GoalState
+    ) -> Sequence[CandidateBinding]:
+        if depth <= 0:
+            return ()
+        return tuple(seed for _ in range(depth))
+
+    def materialize_output(
+        self, branch: Optional[ProspectionBranch], goal: GoalState
+    ) -> Dict[str, Any]:
+        if branch is None or not branch.chain:
+            return {"status": "NO_CANDIDATE", "goal_id": goal.goal_id}
+
+        terminal = branch.chain[-1]
+        chain_trace: list[Dict[str, Any]] = []
+        for step in branch.chain:
+            chain_trace.append(
+                {
+                    "candidate_id": step.candidate_id,
+                    "fact": step.metadata.get("fact_text"),
+                    "slab_id": step.metadata.get("slab_id"),
+                    "vertex_id": step.metadata.get("vertex_id"),
+                    "resonance": step.resonance,
+                    "alignment": step.metadata.get("alignment"),
+                }
+            )
+
+        return {
+            "status": "OK",
+            "goal_id": goal.goal_id,
+            "selected_candidate_id": branch.seed_candidate_id,
+            "branch_score": branch.score,
+            "answer": terminal.metadata.get("fact_text", "UNKNOWN"),
+            "chain_trace": chain_trace,
+        }
+
+    def strengthen_constraints(self, goal: GoalState, candidate: CandidateBinding) -> None:
+        fact_id = candidate.metadata.get("fact_id")
+        if not fact_id:
+            return
+        key = (goal.goal_id, str(fact_id))
+        self._constraint_strength[key] = self._constraint_strength.get(key, 0) + 1
+
+
+def build_demo_bridge(vector_dim: int = 4096) -> tuple[OneShotDemoBridge, np.ndarray, np.ndarray]:
+    bridge = OneShotDemoBridge(vector_dim=vector_dim)
+
+    goal_general = text_to_vector("goal:shadowx:deterministic-control", vector_dim)
+    goal_capital = text_to_vector("goal:paris-capital-france", vector_dim)
+
+    facts = [
+        OneShotFact(
+            fact_id="f0",
+            subject="shadow-x",
+            relation="uses",
+            obj="deterministic_control",
+            slab_id=0,
+            vertex_id=10,
+            vector=vector_with_alignment(goal_general, 0.92, "f0"),
+            base_resonance=0.86,
+        ),
+        OneShotFact(
+            fact_id="f1",
+            subject="earth",
+            relation="orbits",
+            obj="sun",
+            slab_id=1,
+            vertex_id=20,
+            vector=text_to_vector("earth-orbits-sun", vector_dim),
+            base_resonance=0.42,
+        ),
+        OneShotFact(
+            fact_id="f2",
+            subject="python",
+            relation="supports",
+            obj="type_hints",
+            slab_id=2,
+            vertex_id=30,
+            vector=text_to_vector("python-type-hints", vector_dim),
+            base_resonance=0.38,
+        ),
+        # Wrong but high resonance: used to trigger self-correction.
+        OneShotFact(
+            fact_id="f3",
+            subject="paris",
+            relation="is_capital_of",
+            obj="germany",
+            slab_id=3,
+            vertex_id=41,
+            vector=vector_with_alignment(goal_capital, 0.24, "f3_wrong"),
+            base_resonance=0.99,
+        ),
+        # Correct but lower resonance: correction should switch to this.
+        OneShotFact(
+            fact_id="f4",
+            subject="paris",
+            relation="is_capital_of",
+            obj="france",
+            slab_id=3,
+            vertex_id=42,
+            vector=vector_with_alignment(goal_capital, 0.28, "f4_correct"),
+            base_resonance=0.05,
+        ),
+    ]
+
+    for fact in facts:
+        bridge.teach(fact)
+
+    return bridge, goal_general, goal_capital
+
+
+def load_cycle_events(trace_file: Path) -> list[dict]:
+    if not trace_file.exists():
+        return []
+    events: list[dict] = []
+    with trace_file.open("r", encoding="utf-8") as f:
+        for raw in f:
+            raw = raw.strip()
+            if not raw:
+                continue
+            record = json.loads(raw)
+            if record.get("event") == "cycle":
+                events.append(record)
+    return events
+
+
+def run_demo(
+    state_dir: Path,
+    vector_dim: int = 4096,
+    reset_state: bool = True,
+) -> Dict[str, Any]:
+    if reset_state and state_dir.exists():
+        shutil.rmtree(state_dir)
+    state_dir.mkdir(parents=True, exist_ok=True)
+
+    bridge, goal_general, goal_capital = build_demo_bridge(vector_dim=vector_dim)
+    controller = CognitiveController(bridge, state_dir=state_dir, vector_dim=vector_dim)
+
+    # Keep all slabs visible in the demo.
+    controller.rules["attention_activation_threshold"] = -1.0
+
+    taught_facts = [fact.text() for fact in bridge.list_facts()]
+
+    # Scenario A: direct deterministic retrieval.
+    controller.push_goal("query_general_control", goal_general, priority=5)
+    decision_general = controller.run_cycle()
+    controller.pop_goal()
+
+    # Scenario B: self-correction over wrong high-resonance input.
+    controller.push_goal("query_capital_paris_france", goal_capital, priority=10)
+    decision_correction = controller.run_cycle()
+
+    trace_events = load_cycle_events(state_dir / "trace_log.jsonl")
+    last_cycle = trace_events[-1] if trace_events else {}
+    chain_trace = (
+        last_cycle.get("payload", {})
+        .get("output_payload", {})
+        .get("chain_trace", [])
+    )
+
+    summary = {
+        "taught_facts": taught_facts,
+        "general_answer": decision_general.output_payload.get("answer"),
+        "general_selected": decision_general.selected_candidate_id,
+        "correction_answer": decision_correction.output_payload.get("answer"),
+        "correction_selected": decision_correction.selected_candidate_id,
+        "correction_applied": decision_correction.correction_applied,
+        "correction_error_action": decision_correction.error_signal.action,
+        "trace_cycle_count": len(trace_events),
+        "trace_chain": chain_trace,
+        "trace_file": str((state_dir / "trace_log.jsonl").resolve()),
+    }
+    return summary
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Shadow-X v14 public mini-demo CLI (one-shot + goal + trace + self-correction)."
+    )
+    parser.add_argument(
+        "--state-dir",
+        type=Path,
+        default=Path("shadow_x_v14_ccs/state_public_demo"),
+        help="State directory for goal stack and trace log.",
+    )
+    parser.add_argument(
+        "--vector-dim",
+        type=int,
+        default=4096,
+        help="Vector dimensionality for the demo.",
+    )
+    parser.add_argument(
+        "--no-reset",
+        action="store_true",
+        help="Do not reset existing state directory before the run.",
+    )
+    args = parser.parse_args()
+
+    summary = run_demo(
+        state_dir=args.state_dir,
+        vector_dim=args.vector_dim,
+        reset_state=not args.no_reset,
+    )
+
+    print("\n=== SHADOW-X v14 MINI DEMO ===")
+    print("One-shot taught facts:")
+    for idx, fact in enumerate(summary["taught_facts"], start=1):
+        print(f"  {idx}. {fact}")
+
+    print("\nScenario A - Goal query:")
+    print(f"  Selected: {summary['general_selected']}")
+    print(f"  Answer:   {summary['general_answer']}")
+
+    print("\nScenario B - Self-correction:")
+    print(f"  Selected after cycle: {summary['correction_selected']}")
+    print(f"  Answer after cycle:   {summary['correction_answer']}")
+    print(f"  Correction applied:   {summary['correction_applied']}")
+    print(f"  Error action:         {summary['correction_error_action']}")
+
+    print("\nTrace (latest cycle chain, slab/vertex/resonance):")
+    chain = summary["trace_chain"]
+    if not chain:
+        print("  (empty)")
+    else:
+        for step in chain:
+            print(
+                "  "
+                f"{step['candidate_id']} | "
+                f"slab={step['slab_id']} vertex={step['vertex_id']} "
+                f"res={step['resonance']:.4f} align={float(step['alignment']):.4f} | "
+                f"{step['fact']}"
+            )
+
+    print(f"\nTrace file: {summary['trace_file']}")
+    print(f"Trace cycle events: {summary['trace_cycle_count']}")
+
+
+if __name__ == "__main__":
+    main()

shadow_x_v14_ccs/phantom_bridge.py

@@ -0,0 +1,283 @@
+from __future__ import annotations
+
+from collections import defaultdict
+from pathlib import Path
+from typing import Any, Dict, Optional, Sequence, Tuple
+
+import numpy as np
+
+from .cognitive_controller import (
+    CandidateBinding,
+    CognitiveController,
+    GoalState,
+    KernelBridge,
+    ProspectionBranch,
+    ZoneDescriptor,
+)
+
+
+def _normalize(vec: np.ndarray) -> np.ndarray:
+    norm = float(np.linalg.norm(vec))
+    if norm <= 1e-12:
+        return np.zeros_like(vec, dtype=np.float32)
+    return (vec / norm).astype(np.float32)
+
+
+class PhantomKernelBridge(KernelBridge):
+    """
+    Deterministic bridge between CCS-v2 and the existing Phantom Kernel.
+
+    The bridge keeps Phantom Kernel unchanged and exposes deterministic
+    zone/candidate/future-chain operations to the cognitive controller.
+    """
+
+    def __init__(
+        self,
+        kernel,
+        max_zone_candidates: int = 256,
+        centroid_sample_size: int = 64,
+        max_candidates_per_zone: int = 32,
+        candidate_activation_threshold: float = -1.0,
+    ):
+        self.kernel = kernel
+        self.max_zone_candidates = max_zone_candidates
+        self.centroid_sample_size = centroid_sample_size
+        self.max_candidates_per_zone = max_candidates_per_zone
+        self.candidate_activation_threshold = candidate_activation_threshold
+
+        self._zones: Tuple[ZoneDescriptor, ...] = ()
+        self._zone_candidates: Dict[str, Tuple[Tuple[int, np.ndarray], ...]] = {}
+        self._constraint_strength: Dict[Tuple[str, int], int] = {}
+        self._build_zone_index()
+
+    def _deterministic_sample(self, values: Sequence[int], limit: int) -> Tuple[int, ...]:
+        if limit <= 0:
+            return ()
+        if len(values) <= limit:
+            return tuple(values)
+
+        stride = len(values) / float(limit)
+        sampled = [values[int(i * stride)] for i in range(limit)]
+        return tuple(sampled)
+
+    def _build_zone_index(self) -> None:
+        by_slab: Dict[int, list[int]] = defaultdict(list)
+        for gid_raw, loc in self.kernel.zone_map.items():
+            try:
+                gid = int(gid_raw)
+                slab_id = int(loc[0])
+            except Exception:
+                continue
+
+            if slab_id in self.kernel.slabs:
+                by_slab[slab_id].append(gid)
+
+        zones: list[ZoneDescriptor] = []
+        zone_candidates: Dict[str, Tuple[Tuple[int, np.ndarray], ...]] = {}
+
+        for slab_id in sorted(by_slab.keys()):
+            gids = sorted(by_slab[slab_id])
+            sampled = self._deterministic_sample(gids, self.max_zone_candidates)
+
+            vector_entries: list[Tuple[int, np.ndarray]] = []
+            for gid in sampled:
+                vec = self.kernel.get_vector(gid)
+                if vec is None:
+                    continue
+                nvec = _normalize(np.asarray(vec, dtype=np.float32))
+                if not np.any(nvec):
+                    continue
+                vector_entries.append((gid, nvec))
+
+            if not vector_entries:
+                continue
+
+            centroid_count = min(self.centroid_sample_size, len(vector_entries))
+            centroid_stack = np.stack(
+                [vector_entries[i][1] for i in range(centroid_count)],
+                axis=0,
+            )
+            centroid = _normalize(centroid_stack.mean(axis=0))
+
+            zone_id = f"slab:{slab_id}"
+            zones.append(
+                ZoneDescriptor(
+                    zone_id=zone_id,
+                    centroid=centroid,
+                    slab_id=slab_id,
+                    metadata={
+                        "sample_count": len(vector_entries),
+                        "max_candidates_per_zone": self.max_candidates_per_zone,
+                    },
+                )
+            )
+            zone_candidates[zone_id] = tuple(vector_entries)
+
+        self._zones = tuple(zones)
+        self._zone_candidates = zone_candidates
+
+    def list_zones(self) -> Sequence[ZoneDescriptor]:
+        return self._zones
+
+    def _candidate_id(self, zone_id: str, gid: int) -> str:
+        return f"{zone_id}:gid:{gid}"
+
+    def _candidate_binding(
+        self,
+        goal: GoalState,
+        zone: ZoneDescriptor,
+        gid: int,
+        vec: np.ndarray,
+        resonance: float,
+    ) -> CandidateBinding:
+        boost = self._constraint_strength.get((goal.goal_id, gid), 0) * 0.01
+        adjusted_resonance = max(-1.0, min(1.0, resonance + boost))
+        token = self.kernel.get_token(gid)
+        return CandidateBinding(
+            candidate_id=self._candidate_id(zone.zone_id, gid),
+            vector=vec,
+            resonance=adjusted_resonance,
+            zone_id=zone.zone_id,
+            slab_id=zone.slab_id,
+            metadata={
+                "gid": gid,
+                "token": token,
+                "constraint_strength": self._constraint_strength.get((goal.goal_id, gid), 0),
+            },
+        )
+
+    def query_candidates(
+        self, goal: GoalState, active_zones: Sequence[ZoneDescriptor]
+    ) -> Sequence[CandidateBinding]:
+        goal_vector = _normalize(goal.vector)
+        all_candidates: list[CandidateBinding] = []
+
+        for zone in active_zones:
+            zone_entries = self._zone_candidates.get(zone.zone_id, ())
+            zone_scored: list[CandidateBinding] = []
+
+            for gid, vec in zone_entries:
+                resonance = float(np.dot(goal_vector, vec))
+                if resonance < self.candidate_activation_threshold:
+                    continue
+                zone_scored.append(self._candidate_binding(goal, zone, gid, vec, resonance))
+
+            zone_scored.sort(key=lambda c: (-c.resonance, c.candidate_id))
+            all_candidates.extend(zone_scored[: self.max_candidates_per_zone])
+
+        all_candidates.sort(key=lambda c: (-c.resonance, c.candidate_id))
+        return tuple(all_candidates)
+
+    def _zone_from_seed(self, seed: CandidateBinding) -> Optional[ZoneDescriptor]:
+        for zone in self._zones:
+            if zone.zone_id == seed.zone_id:
+                return zone
+        return None
+
+    def build_future_chain(
+        self, seed: CandidateBinding, depth: int, goal: GoalState
+    ) -> Sequence[CandidateBinding]:
+        if depth <= 0:
+            return ()
+
+        zone = self._zone_from_seed(seed)
+        if zone is None:
+            return (seed,)
+
+        zone_entries = self._zone_candidates.get(zone.zone_id, ())
+        goal_vector = _normalize(goal.vector)
+
+        used_gids: set[int] = set()
+        if "gid" in seed.metadata:
+            used_gids.add(int(seed.metadata["gid"]))
+
+        chain: list[CandidateBinding] = [seed]
+        prev_vector = seed.vector
+
+        for _ in range(depth - 1):
+            best_gid: Optional[int] = None
+            best_vec: Optional[np.ndarray] = None
+            best_score = float("-inf")
+
+            for gid, vec in zone_entries:
+                if gid in used_gids:
+                    continue
+                alignment = float(np.dot(goal_vector, vec))
+                continuity = float(np.dot(prev_vector, vec))
+                boost = self._constraint_strength.get((goal.goal_id, gid), 0) * 0.01
+                score = (0.7 * alignment) + (0.3 * continuity) + boost
+
+                if (score > best_score) or (
+                    abs(score - best_score) <= 1e-12
+                    and best_gid is not None
+                    and gid < best_gid
+                ):
+                    best_gid = gid
+                    best_vec = vec
+                    best_score = score
+
+            if best_gid is None or best_vec is None:
+                break
+
+            used_gids.add(best_gid)
+            chain.append(self._candidate_binding(goal, zone, best_gid, best_vec, best_score))
+            prev_vector = best_vec
+
+        return tuple(chain)
+
+    def materialize_output(
+        self, branch: Optional[ProspectionBranch], goal: GoalState
+    ) -> Dict[str, Any]:
+        if branch is None or not branch.chain:
+            return {
+                "status": "NO_CANDIDATE",
+                "goal_id": goal.goal_id,
+            }
+
+        terminal = branch.chain[-1]
+        chain_ids = [item.candidate_id for item in branch.chain]
+        chain_tokens = [str(item.metadata.get("token", "UNK")) for item in branch.chain]
+
+        return {
+            "status": "OK",
+            "goal_id": goal.goal_id,
+            "selected_candidate_id": branch.seed_candidate_id,
+            "terminal_candidate_id": terminal.candidate_id,
+            "terminal_token": terminal.metadata.get("token", "UNK"),
+            "terminal_gid": terminal.metadata.get("gid"),
+            "branch_score": branch.score,
+            "chain_candidate_ids": chain_ids,
+            "chain_tokens": chain_tokens,
+        }
+
+    def strengthen_constraints(self, goal: GoalState, candidate: CandidateBinding) -> None:
+        gid = candidate.metadata.get("gid")
+        if gid is None:
+            return
+        key = (goal.goal_id, int(gid))
+        self._constraint_strength[key] = self._constraint_strength.get(key, 0) + 1
+
+
+def create_v14_controller_with_phantom_kernel(
+    model_path: str = "v11_Morpho_Optimized",
+    state_dir: Path | str = Path("shadow_x_v14_ccs/state"),
+    *,
+    max_zone_candidates: int = 256,
+    centroid_sample_size: int = 64,
+    max_candidates_per_zone: int = 32,
+    candidate_activation_threshold: float = -1.0,
+) -> CognitiveController:
+    """
+    Convenience factory for end-to-end CCS-v14 + Phantom Kernel wiring.
+    """
+    from phantom_kernel import PhantomKernel
+
+    kernel = PhantomKernel(model_path=model_path)
+    bridge = PhantomKernelBridge(
+        kernel=kernel,
+        max_zone_candidates=max_zone_candidates,
+        centroid_sample_size=centroid_sample_size,
+        max_candidates_per_zone=max_candidates_per_zone,
+        candidate_activation_threshold=candidate_activation_threshold,
+    )
+    return CognitiveController(kernel_bridge=bridge, state_dir=state_dir)

tests/test_shadow_x_v14_ccs.py

@@ -0,0 +1,238 @@
+import json
+import shutil
+import tempfile
+import unittest
+from pathlib import Path
+
+import numpy as np
+
+from shadow_x_v14_ccs.cognitive_controller import (
+    CandidateBinding,
+    CognitiveController,
+    ConflictResolver,
+    DeterministicAttentionAllocator,
+    ErrorMonitor,
+    GoalStack,
+    InMemoryKernelBridge,
+    TaskSwitcherProspectionEngine,
+    ThirdOrderRuleModifier,
+    ZoneDescriptor,
+)
+
+
+def make_vec(seed: int, dim: int = 4096) -> np.ndarray:
+    base = np.arange(dim, dtype=np.float32)
+    vec = np.cos(base * 0.001 + seed).astype(np.float32)
+    norm = np.linalg.norm(vec)
+    if norm <= 1e-12:
+        return np.zeros(dim, dtype=np.float32)
+    return (vec / norm).astype(np.float32)
+
+
+class GoalStackTests(unittest.TestCase):
+    def setUp(self):
+        self.tmpdir = tempfile.mkdtemp(prefix="shadowx_v14_goalstack_")
+        self.goal_file = Path(self.tmpdir) / "goal_vector.dat"
+
+    def tearDown(self):
+        shutil.rmtree(self.tmpdir, ignore_errors=True)
+
+    def test_push_pop_and_persistence(self):
+        stack = GoalStack(self.goal_file, vector_dim=4096)
+        g1 = stack.push("root_goal", make_vec(1), priority=9)
+        g2 = stack.push("child_goal", make_vec(2), priority=4, parent_goal_id=g1.goal_id)
+
+        self.assertEqual(stack.current().goal_id, g2.goal_id)
+        self.assertEqual(len(stack.list_goals()), 2)
+
+        reloaded = GoalStack(self.goal_file, vector_dim=4096)
+        self.assertEqual(len(reloaded.list_goals()), 2)
+        self.assertEqual(reloaded.current().goal_id, g2.goal_id)
+
+        popped = reloaded.pop()
+        self.assertEqual(popped.goal_id, g2.goal_id)
+        self.assertEqual(reloaded.current().goal_id, g1.goal_id)
+
+
+class ModuleUnitTests(unittest.TestCase):
+    def test_attention_allocator_threshold_and_order(self):
+        allocator = DeterministicAttentionAllocator(activation_threshold=0.80, max_active_zones=2)
+        goal = make_vec(10)
+        projected_1 = np.roll(goal, 1).astype(np.float32)
+        projected_1 /= np.linalg.norm(projected_1)
+        projected_2 = np.roll(goal, 2).astype(np.float32)
+        projected_2 /= np.linalg.norm(projected_2)
+        zones = (
+            ZoneDescriptor("z_good_1", projected_1, slab_id=0),
+            ZoneDescriptor("z_good_2", projected_2, slab_id=1),
+            ZoneDescriptor("z_bad", -make_vec(10), slab_id=2),
+        )
+        selected = allocator.allocate(goal, zones)
+        ids = tuple(z.zone_id for z in selected)
+        self.assertEqual(len(ids), 2)
+        self.assertIn("z_good_1", ids)
+        self.assertIn("z_good_2", ids)
+        self.assertNotIn("z_bad", ids)
+
+    def test_conflict_resolver_inhibits_opposition(self):
+        resolver = ConflictResolver(opposition_threshold=0.80)
+        goal = make_vec(20)
+        c_good = CandidateBinding("c_good", make_vec(20), 0.92, "z0")
+        c_opp = CandidateBinding("c_opp", -make_vec(20), 0.95, "z0")
+        c_neutral = CandidateBinding("c_neutral", make_vec(21), 0.90, "z1")
+
+        selected, inhibited, conflicts = resolver.resolve((c_good, c_opp, c_neutral), goal)
+        selected_ids = {c.candidate_id for c in selected}
+        self.assertIn("c_good", selected_ids)
+        self.assertNotIn("c_opp", selected_ids)
+        self.assertIn("c_opp", inhibited)
+        self.assertGreaterEqual(len(conflicts), 1)
+
+    def test_error_monitor_states(self):
+        monitor = ErrorMonitor(deviation_threshold=0.2, hard_fail_threshold=0.4)
+        goal = make_vec(30)
+
+        ok = monitor.evaluate(goal, goal)
+        self.assertFalse(ok.has_error)
+        self.assertEqual(ok.action, "none")
+
+        rebind = monitor.evaluate(make_vec(31), goal)
+        self.assertTrue(rebind.has_error)
+        self.assertIn(rebind.action, {"rebind", "strengthen_constraints_and_rebind"})
+
+        hard_fail = monitor.evaluate(-goal, goal)
+        self.assertTrue(hard_fail.has_error)
+        self.assertEqual(hard_fail.action, "strengthen_constraints_and_rebind")
+
+    def test_task_switcher_prospection_returns_deterministic_best(self):
+        engine = TaskSwitcherProspectionEngine(max_futures=3, branch_depth=3)
+        goal = make_vec(40)
+        c0 = CandidateBinding("c0", make_vec(40), 0.93, "z0")
+        c1 = CandidateBinding("c1", make_vec(41), 0.91, "z1")
+        c2 = CandidateBinding("c2", -make_vec(40), 0.95, "z2")
+
+        def build_chain(seed: CandidateBinding, depth: int):
+            return tuple(seed for _ in range(depth))
+
+        branches = engine.simulate(goal, (c0, c1, c2), build_chain)
+        best = engine.choose_best(branches)
+        self.assertIsNotNone(best)
+        self.assertEqual(best.seed_candidate_id, "c0")
+
+    def test_rule_modifier_updates_after_error_window(self):
+        mod = ThirdOrderRuleModifier(error_window=5)
+        rules = {
+            "attention_activation_threshold": 0.12,
+            "deviation_threshold": 0.28,
+            "hard_fail_threshold": 0.40,
+            "conflict_opposition_threshold": 0.80,
+        }
+
+        monitor = ErrorMonitor(deviation_threshold=0.2, hard_fail_threshold=0.4)
+        goal = make_vec(50)
+        for _ in range(5):
+            mod.register(monitor.evaluate(-goal, goal))
+        updates = mod.update_rules(rules)
+
+        self.assertTrue(updates)
+        self.assertGreaterEqual(
+            updates["attention_activation_threshold"], rules["attention_activation_threshold"]
+        )
+        self.assertLessEqual(
+            updates["deviation_threshold"], rules["deviation_threshold"]
+        )
+
+
+class ControllerIntegrationTests(unittest.TestCase):
+    def setUp(self):
+        self.tmpdir = tempfile.mkdtemp(prefix="shadowx_v14_controller_")
+        self.state_dir = Path(self.tmpdir) / "state"
+        self.goal = make_vec(100)
+
+        zones = (
+            ZoneDescriptor("Z0", make_vec(100), slab_id=0),
+            ZoneDescriptor("Z1", make_vec(101), slab_id=1),
+            ZoneDescriptor("Z2", -make_vec(100), slab_id=2),
+        )
+        candidates = {
+            "Z0": (
+                CandidateBinding("C0", make_vec(100), 0.94, "Z0", slab_id=0),
+                CandidateBinding("C1", -make_vec(100), 0.93, "Z0", slab_id=0),
+            ),
+            "Z1": (
+                CandidateBinding("C2", make_vec(101), 0.92, "Z1", slab_id=1),
+            ),
+            "Z2": (
+                CandidateBinding("C3", -make_vec(100), 0.97, "Z2", slab_id=2),
+            ),
+        }
+        bridge = InMemoryKernelBridge(zones, candidates)
+        self.controller = CognitiveController(bridge, state_dir=self.state_dir, vector_dim=4096)
+
+    def tearDown(self):
+        shutil.rmtree(self.tmpdir, ignore_errors=True)
+
+    def test_run_cycle_generates_decision_and_trace(self):
+        self.controller.push_goal("consistency_check", self.goal, priority=8)
+        decision = self.controller.run_cycle()
+
+        self.assertEqual(decision.goal_id, self.controller.goal_stack.current().goal_id)
+        self.assertIsNotNone(decision.selected_candidate_id)
+        self.assertIn("C1", decision.inhibited_candidate_ids)
+        self.assertIsInstance(decision.output_payload, dict)
+
+        trace_file = self.state_dir / "trace_log.jsonl"
+        self.assertTrue(trace_file.exists())
+        lines = trace_file.read_text(encoding="utf-8").strip().splitlines()
+        self.assertGreaterEqual(len(lines), 2)  # goal_push + cycle
+
+        parsed = [json.loads(line) for line in lines]
+        events = [p["event"] for p in parsed]
+        self.assertIn("goal_push", events)
+        self.assertIn("cycle", events)
+
+    def test_determinism_across_fresh_controllers_with_same_input(self):
+        self.controller.push_goal("determinism_goal", self.goal, priority=7)
+        first = self.controller.run_cycle()
+
+        zones = self.controller.kernel_bridge.list_zones()
+        candidates = {
+            z.zone_id: tuple(self.controller.kernel_bridge._zone_candidates.get(z.zone_id, ()))
+            for z in zones
+        }
+        bridge2 = InMemoryKernelBridge(zones, candidates)
+        controller2 = CognitiveController(
+            bridge2, state_dir=Path(self.tmpdir) / "state2", vector_dim=4096
+        )
+        controller2.push_goal("determinism_goal", self.goal, priority=7)
+        second = controller2.run_cycle()
+
+        self.assertEqual(first.selected_candidate_id, second.selected_candidate_id)
+        self.assertEqual(first.active_zone_ids, second.active_zone_ids)
+        self.assertEqual(first.inhibited_candidate_ids, second.inhibited_candidate_ids)
+        self.assertAlmostEqual(
+            first.error_signal.deviation, second.error_signal.deviation, places=8
+        )
+
+    def test_error_when_running_without_goal(self):
+        with self.assertRaises(RuntimeError):
+            self.controller.run_cycle()
+
+    def test_no_candidate_path_is_handled(self):
+        empty_bridge = InMemoryKernelBridge(
+            zones=(ZoneDescriptor("Z9", make_vec(900), slab_id=9),),
+            zone_candidates={},
+        )
+        controller = CognitiveController(
+            empty_bridge, state_dir=Path(self.tmpdir) / "state_empty", vector_dim=4096
+        )
+        controller.push_goal("empty_case", self.goal, priority=1)
+        decision = controller.run_cycle()
+
+        self.assertIsNone(decision.selected_candidate_id)
+        self.assertEqual(decision.output_payload.get("status"), "NO_CANDIDATE")
+        self.assertTrue(decision.error_signal.has_error)
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)

tests/test_shadow_x_v14_demo_cli.py

@@ -0,0 +1,34 @@
+import shutil
+import tempfile
+import unittest
+from pathlib import Path
+
+from shadow_x_v14_ccs.demo_cli import run_demo
+
+
+class DemoCliTests(unittest.TestCase):
+    def setUp(self):
+        self.tmpdir = tempfile.mkdtemp(prefix="shadowx_v14_demo_cli_")
+        self.state_dir = Path(self.tmpdir) / "state"
+
+    def tearDown(self):
+        shutil.rmtree(self.tmpdir, ignore_errors=True)
+
+    def test_run_demo_covers_required_public_flow(self):
+        summary = run_demo(self.state_dir, vector_dim=512, reset_state=True)
+
+        self.assertEqual(len(summary["taught_facts"]), 5)
+        self.assertTrue(summary["general_answer"])
+        self.assertTrue(summary["correction_answer"])
+        self.assertTrue(summary["correction_applied"])
+        self.assertGreaterEqual(summary["trace_cycle_count"], 2)
+        self.assertTrue(summary["trace_chain"])
+
+        first = summary["trace_chain"][0]
+        self.assertIn("slab_id", first)
+        self.assertIn("vertex_id", first)
+        self.assertIn("resonance", first)
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)

tests/test_shadow_x_v14_phantom_bridge.py

@@ -0,0 +1,144 @@
+import shutil
+import tempfile
+import unittest
+from pathlib import Path
+
+import numpy as np
+
+from shadow_x_v14_ccs.cognitive_controller import CognitiveController, GoalState, ProspectionBranch
+from shadow_x_v14_ccs.phantom_bridge import PhantomKernelBridge
+
+
+def make_vec(seed: int, dim: int = 64) -> np.ndarray:
+    base = np.arange(dim, dtype=np.float32)
+    vec = np.cos(base * 0.07 + seed).astype(np.float32)
+    vec /= (np.linalg.norm(vec) + 1e-12)
+    return vec
+
+
+class FakeKernel:
+    def __init__(self):
+        self.slabs = {0: object(), 1: object()}
+        self.zone_map = {
+            0: (0, 0),
+            1: (0, 1),
+            2: (1, 0),
+            3: (1, 1),
+        }
+        self._vectors = {
+            0: make_vec(1),
+            1: make_vec(2),
+            2: make_vec(10),
+            3: -make_vec(1),
+        }
+
+    def get_vector(self, gid: int):
+        return self._vectors.get(gid)
+
+    def get_token(self, gid: int):
+        return f"T{gid}"
+
+
+class PhantomBridgeUnitTests(unittest.TestCase):
+    def setUp(self):
+        self.kernel = FakeKernel()
+        self.bridge = PhantomKernelBridge(
+            kernel=self.kernel,
+            max_zone_candidates=4,
+            centroid_sample_size=2,
+            max_candidates_per_zone=4,
+            candidate_activation_threshold=-1.0,
+        )
+        self.goal = GoalState(goal_id="g1", name="goal", vector=make_vec(1))
+
+    def test_zone_index_and_query(self):
+        zones = self.bridge.list_zones()
+        self.assertEqual(len(zones), 2)
+
+        candidates = self.bridge.query_candidates(self.goal, zones)
+        self.assertGreaterEqual(len(candidates), 1)
+
+        top = candidates[0]
+        self.assertEqual(top.metadata.get("gid"), 0)
+        self.assertEqual(top.metadata.get("token"), "T0")
+
+    def test_strengthen_constraints_biases_resonance(self):
+        zones = self.bridge.list_zones()
+        before = self.bridge.query_candidates(self.goal, zones)
+        c1_before = next(c for c in before if c.metadata.get("gid") == 1)
+
+        for _ in range(3):
+            self.bridge.strengthen_constraints(self.goal, c1_before)
+
+        after = self.bridge.query_candidates(self.goal, zones)
+        c1_after = next(c for c in after if c.metadata.get("gid") == 1)
+
+        self.assertAlmostEqual(c1_after.resonance, c1_before.resonance + 0.03, places=6)
+
+    def test_future_chain_and_materialize_output(self):
+        zones = self.bridge.list_zones()
+        candidates = self.bridge.query_candidates(self.goal, zones)
+        seed = candidates[0]
+
+        chain = self.bridge.build_future_chain(seed, depth=3, goal=self.goal)
+        self.assertGreaterEqual(len(chain), 1)
+        self.assertLessEqual(len(chain), 3)
+
+        branch = ProspectionBranch(
+            seed_candidate_id=seed.candidate_id,
+            chain=tuple(chain),
+            score=0.91,
+        )
+        payload = self.bridge.materialize_output(branch, self.goal)
+        self.assertEqual(payload["status"], "OK")
+        self.assertIn("chain_candidate_ids", payload)
+        self.assertIn("terminal_gid", payload)
+
+
+@unittest.skipUnless(
+    Path("v11_Morpho_Optimized").exists(),
+    "Real Phantom dataset folder not found.",
+)
+class PhantomBridgeRealDatasetSmokeTests(unittest.TestCase):
+    def setUp(self):
+        self.tmpdir = tempfile.mkdtemp(prefix="shadowx_v14_real_bridge_")
+
+    def tearDown(self):
+        shutil.rmtree(self.tmpdir, ignore_errors=True)
+
+    def test_real_kernel_bridge_controller_cycle(self):
+        from phantom_kernel import PhantomKernel
+
+        kernel = PhantomKernel(model_path="v11_Morpho_Optimized")
+        try:
+            bridge = PhantomKernelBridge(
+                kernel=kernel,
+                max_zone_candidates=64,
+                centroid_sample_size=16,
+                max_candidates_per_zone=8,
+                candidate_activation_threshold=-1.0,
+            )
+
+            zones = bridge.list_zones()
+            self.assertGreater(len(zones), 0)
+
+            controller = CognitiveController(
+                bridge,
+                state_dir=Path(self.tmpdir) / "state",
+                vector_dim=4096,
+            )
+            controller.rules["attention_activation_threshold"] = -1.0
+
+            goal_vector = zones[0].centroid
+            controller.push_goal("real_dataset_smoke", goal_vector, priority=5)
+            decision = controller.run_cycle()
+
+            self.assertEqual(decision.goal_id, controller.goal_stack.current().goal_id)
+            self.assertIsInstance(decision.output_payload, dict)
+            self.assertIn("status", decision.output_payload)
+        finally:
+            kernel.close()
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)