feat: audio ear, cognition modules, dream engine, vitalis IDE, test encoder

pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+pythonpath = . src
+testpaths = tests src

src/audio_ear/feature_extractor.py

@@ -0,0 +1,34 @@
+import librosa
+import numpy as np
+from typing import Tuple, Dict
+from pathlib import Path
+
+def extract_features(wav_path: Path) -> Tuple[np.ndarray, Dict[str, float]]:
+    """
+    Extracts the 13-band Mel-frequency cepstral coefficients (MFCC) 
+    and heuristic prosody markers from a raw WAV file.
+    """
+    y, sr = librosa.load(str(wav_path), sr=16000)
+    
+    # Extract MFCC matrix
+    mfcc = librosa.feature.mfcc(y=y, sr=sr, n_mfcc=13)
+    
+    # Heuristic prosody extraction
+    pitches, magnitudes = librosa.piptrack(y=y, sr=sr)
+    valid_pitches = pitches[magnitudes > np.median(magnitudes)]
+    pitch = float(np.mean(valid_pitches)) if len(valid_pitches) > 0 else 0.0
+    
+    energy = float(np.mean(librosa.feature.rms(y=y)))
+    tempo, _ = librosa.beat.beat_track(y=y, sr=sr)
+    
+    # Calculate pause ratio based on silence thresholds
+    pause_ratio = float(np.sum(np.abs(y) < 0.01) / len(y)) if len(y) > 0 else 0.0
+    
+    prosody = {
+        "pitch": pitch,
+        "energy": energy,
+        "tempo": float(tempo[0] if isinstance(tempo, np.ndarray) else tempo),
+        "pause_ratio": pause_ratio
+    }
+    
+    return mfcc, prosody

src/audio_ear/recorder.py

@@ -0,0 +1,13 @@
+import sounddevice as sd
+import soundfile as sf
+import numpy as np
+from pathlib import Path
+
+def record_to_wav(duration_sec: int, out_path: Path, fs: int = 16000) -> None:
+    """
+    Interfaces directly with the local machine's sound architecture 
+    to capture a mono-channel audio array.
+    """
+    recording = sd.rec(int(duration_sec * fs), samplerate=fs, channels=1, dtype=np.float32)
+    sd.wait()
+    sf.write(str(out_path), recording, fs)

src/cognition/abstract_reasoner.py

@@ -0,0 +1,233 @@
+"""
+AbstractReasoner — Vitalis FSI
+
+Reasons about RELATIONSHIPS between concepts.
+Not pattern matching. Not retrieval.
+Genuine relational reasoning:
+  - Analogy: A is to B as C is to ?
+  - Composition: concept_A + concept_B = novel_concept
+  - Inversion: what is the opposite of this concept?
+  - Transitivity: if A relates to B and B relates to C, what does A relate to C?
+
+Built entirely on HDC operations. No external models.
+"""
+import numpy as np
+import os
+import json
+import time
+from vitalis_ide.math_core.kernel import VitalisKernel
+from src.cognition.abstraction import AbstractionEngine
+from src.hippocampus import Hippocampus
+
+
+class AbstractReasoner:
+    ANALOGY_THRESHOLD  = 0.25
+    COMPOSITION_DECAY  = 0.85
+    INVERSION_SHIFT    = 5000
+
+    def __init__(self):
+        self.kernel      = VitalisKernel()
+        self.abstraction = AbstractionEngine()
+        self.hippocampus = Hippocampus()
+        self.path        = os.path.expanduser(
+            "~/.vitalis_workspace/reasoning_log.json"
+        )
+        self._log = self._load_log()
+
+    def _load_log(self) -> list:
+        if os.path.exists(self.path):
+            with open(self.path) as f:
+                return json.load(f)
+        return []
+
+    def _save_log(self):
+        os.makedirs(os.path.dirname(self.path), exist_ok=True)
+        with open(self.path, "w") as f:
+            json.dump(self._log[-500:], f, indent=2)
+
+    # ------------------------------------------------------------------
+    # Core HDC reasoning operations
+    # ------------------------------------------------------------------
+    def _bind(self, a: np.ndarray, b: np.ndarray) -> np.ndarray:
+        """Bipolar binding: element-wise multiply."""
+        return (a.astype(np.int32) * b.astype(np.int32)).astype(np.int8)
+
+    def _bundle(self, vecs: list) -> np.ndarray:
+        """Bipolar bundling: sum then sign."""
+        stacked = np.stack(vecs).astype(np.int32).sum(axis=0)
+        result = np.sign(stacked).astype(np.int8)
+        result[result == 0] = 1
+        return result
+
+    def _invert(self, vec: np.ndarray) -> np.ndarray:
+        """
+        Semantic inversion: cyclic shift by half the vector length.
+        Produces a vector maximally dissimilar to the input.
+        """
+        return np.roll(vec, self.INVERSION_SHIFT)
+
+    # ------------------------------------------------------------------
+    # Analogy: A is to B as C is to ?
+    # ------------------------------------------------------------------
+    def analogy(
+        self,
+        concept_a: str,
+        concept_b: str,
+        concept_c: str,
+    ) -> dict:
+        """
+        Solves: A:B :: C:?
+        HDC method: ? = bind(bind(A, B), C)
+        Searches abstraction space and hippocampus for closest match.
+        """
+        vec_a = self.kernel.vectorize_tokens(concept_a.split(), positional=False)
+        vec_b = self.kernel.vectorize_tokens(concept_b.split(), positional=False)
+        vec_c = self.kernel.vectorize_tokens(concept_c.split(), positional=False)
+
+        # ? = B * A^-1 * C  (HDC analogy formula)
+        a_inv = self._bind(vec_a, vec_a)  # A bound with itself = identity-like
+        relation = self._bind(vec_a, vec_b)  # encode A→B relationship
+        answer_vec = self._bind(relation, vec_c)  # apply relation to C
+
+        # Search for closest concept
+        candidates = self.abstraction.query_abstractions(answer_vec, top_k=3)
+        hipp_results = self.hippocampus.similarity_search(answer_vec, top_k=3)
+
+        best_match = None
+        best_score = -1.0
+
+        for score, name, _ in candidates:
+            if score > best_score:
+                best_score = score
+                best_match = name
+
+        result = {
+            "type":       "analogy",
+            "query":      f"{concept_a}:{concept_b}::{concept_c}:?",
+            "answer_vec": answer_vec,
+            "best_match": best_match,
+            "confidence": round(float(best_score), 4),
+            "candidates": [(name, round(float(s), 4)) for s, name, _ in candidates],
+            "timestamp":  time.time(),
+        }
+
+        self._log.append({k: v for k, v in result.items() if k != "answer_vec"})
+        self._save_log()
+        return result
+
+    # ------------------------------------------------------------------
+    # Composition: merge two concepts into a novel one
+    # ------------------------------------------------------------------
+    def compose(self, concept_a: str, concept_b: str) -> dict:
+        """
+        Compose two concepts into a novel concept vector.
+        The result occupies a position in the space between both inputs.
+        Weighted by the COMPOSITION_DECAY to prevent drift.
+        """
+        vec_a = self.kernel.vectorize_tokens(concept_a.split(), positional=False)
+        vec_b = self.kernel.vectorize_tokens(concept_b.split(), positional=False)
+
+        # Bundle with decay weighting
+        composed = self._bundle([vec_a, vec_b])
+
+        # Apply composition decay — prevents the result from being
+        # too close to either parent
+        noise_mask = np.random.choice(
+            [-1, 1],
+            size=self.kernel.dim,
+            p=[1 - self.COMPOSITION_DECAY, self.COMPOSITION_DECAY]
+        ).astype(np.int8)
+        composed = self._bind(composed, noise_mask)
+
+        # Search for nearest existing concept
+        candidates = self.abstraction.query_abstractions(composed, top_k=3)
+
+        result = {
+            "type":        "composition",
+            "inputs":      [concept_a, concept_b],
+            "novel_vec":   composed,
+            "nearest":     [(name, round(float(s), 4)) for s, name, _ in candidates],
+            "novelty":     round(1.0 - (candidates[0][0] if candidates else 0.0), 4),
+            "timestamp":   time.time(),
+        }
+
+        self._log.append({k: v for k, v in result.items() if k != "novel_vec"})
+        self._save_log()
+        return result
+
+    # ------------------------------------------------------------------
+    # Inversion: what is the conceptual opposite?
+    # ------------------------------------------------------------------
+    def invert(self, concept: str) -> dict:
+        """
+        Find the conceptual opposite of a concept.
+        Uses cyclic shift inversion then searches concept space.
+        """
+        vec = self.kernel.vectorize_tokens(concept.split(), positional=False)
+        inverted = self._invert(vec)
+
+        candidates = self.abstraction.query_abstractions(inverted, top_k=3)
+        hipp_results = self.hippocampus.similarity_search(inverted, top_k=3)
+
+        result = {
+            "type":       "inversion",
+            "concept":    concept,
+            "opposites":  [(name, round(float(s), 4)) for s, name, _ in candidates],
+            "confidence": round(float(candidates[0][0]) if candidates else 0.0, 4),
+            "timestamp":  time.time(),
+        }
+
+        self._log.append(result)
+        self._save_log()
+        return result
+
+    # ------------------------------------------------------------------
+    # Transitivity: if A→B and B→C, what is A→C?
+    # ------------------------------------------------------------------
+    def transitive_chain(self, concepts: list) -> dict:
+        """
+        Chain reasoning: given [A, B, C, D...],
+        derive the relationship between A and the last element.
+        Each step binds the accumulated relationship with the next concept.
+        """
+        if len(concepts) < 2:
+            return {"error": "Need at least 2 concepts"}
+
+        vecs = [
+            self.kernel.vectorize_tokens(c.split(), positional=False)
+            for c in concepts
+        ]
+
+        # Accumulate relationship via sequential binding
+        accumulated = vecs[0].copy()
+        for i in range(1, len(vecs)):
+            accumulated = self._bind(accumulated, vecs[i])
+            # Apply position-aware permutation at each step
+            accumulated = np.roll(accumulated, i * 100)
+
+        candidates = self.abstraction.query_abstractions(accumulated, top_k=3)
+
+        result = {
+            "type":       "transitivity",
+            "chain":      concepts,
+            "conclusion": [(name, round(float(s), 4)) for s, name, _ in candidates],
+            "confidence": round(float(candidates[0][0]) if candidates else 0.0, 4),
+            "timestamp":  time.time(),
+        }
+
+        self._log.append(result)
+        self._save_log()
+        return result
+
+    def report(self) -> dict:
+        if not self._log:
+            return {"status": "No reasoning performed yet"}
+        type_counts = {}
+        for entry in self._log:
+            t = entry.get("type", "unknown")
+            type_counts[t] = type_counts.get(t, 0) + 1
+        return {
+            "total_reasoning_ops": len(self._log),
+            "by_type":             type_counts,
+            "recent":              self._log[-3:],
+        }

src/cognition/complexity_reasoner.py

@@ -0,0 +1,217 @@
+"""
+ComplexityReasoner — Vitalis FSI
+
+Assesses how hard a problem is BEFORE attempting it.
+Allocates cognitive resources accordingly.
+Prevents wasted cycles on problems beyond current capability.
+Prevents under-allocation on trivial problems.
+
+Complexity dimensions:
+  - Structural: how many components does this problem have?
+  - Novelty: how far is this from known patterns?
+  - Depth: how many reasoning steps are required?
+  - Ambiguity: how many valid interpretations exist?
+"""
+import numpy as np
+import os
+import json
+import time
+from vitalis_ide.math_core.kernel import VitalisKernel
+from src.cognition.abstraction import AbstractionEngine
+from src.hippocampus import Hippocampus
+
+
+class ComplexityReasoner:
+    # Complexity tier thresholds
+    TRIVIAL_THRESHOLD    = 0.25
+    SIMPLE_THRESHOLD     = 0.45
+    MODERATE_THRESHOLD   = 0.65
+    COMPLEX_THRESHOLD    = 0.80
+
+    # Resource allocation per tier
+    RESOURCE_MAP = {
+        "TRIVIAL":  {"cycles": 1,  "abstraction_depth": 1, "analogy_search": False},
+        "SIMPLE":   {"cycles": 2,  "abstraction_depth": 2, "analogy_search": False},
+        "MODERATE": {"cycles": 4,  "abstraction_depth": 3, "analogy_search": True},
+        "COMPLEX":  {"cycles": 8,  "abstraction_depth": 4, "analogy_search": True},
+        "FRONTIER": {"cycles": 16, "abstraction_depth": 5, "analogy_search": True},
+    }
+
+    def __init__(self):
+        self.kernel      = VitalisKernel()
+        self.abstraction = AbstractionEngine()
+        self.hippocampus = Hippocampus()
+        self.path        = os.path.expanduser(
+            "~/.vitalis_workspace/complexity_log.json"
+        )
+        self._log        = self._load_log()
+        self._history    = []
+
+    def _load_log(self) -> list:
+        if os.path.exists(self.path):
+            with open(self.path) as f:
+                return json.load(f)
+        return []
+
+    def _save_log(self):
+        os.makedirs(os.path.dirname(self.path), exist_ok=True)
+        with open(self.path, "w") as f:
+            json.dump(self._log[-1000:], f, indent=2)
+
+    # ------------------------------------------------------------------
+    # Core assessment
+    # ------------------------------------------------------------------
+    def assess(self, intent: str, context: dict = None) -> dict:
+        """
+        Full complexity assessment for an intent.
+        Returns complexity score, tier, and resource allocation.
+        """
+        context = context or {}
+        tokens  = intent.lower().split()
+        vec     = self.kernel.vectorize_tokens(tokens, positional=False)
+
+        # 1. Structural complexity — token count and unique concepts
+        structural = self._structural_score(tokens)
+
+        # 2. Novelty — distance from known patterns
+        novelty = self._novelty_score(vec)
+
+        # 3. Depth — estimated reasoning steps needed
+        depth = self._depth_score(intent, tokens)
+
+        # 4. Ambiguity — spread across abstraction space
+        ambiguity = self._ambiguity_score(vec)
+
+        # Weighted composite
+        score = (
+            structural * 0.20 +
+            novelty    * 0.35 +
+            depth      * 0.25 +
+            ambiguity  * 0.20
+        )
+        score = float(np.clip(score, 0.0, 1.0))
+
+        tier      = self._tier(score)
+        resources = self.RESOURCE_MAP[tier].copy()
+
+        result = {
+            "intent":     intent,
+            "score":      round(score, 4),
+            "tier":       tier,
+            "dimensions": {
+                "structural": round(structural, 4),
+                "novelty":    round(novelty,    4),
+                "depth":      round(depth,      4),
+                "ambiguity":  round(ambiguity,  4),
+            },
+            "resources":  resources,
+            "timestamp":  time.time(),
+        }
+
+        self._log.append(result)
+        self._history.append(score)
+        if len(self._history) > 100:
+            self._history.pop(0)
+        self._save_log()
+        return result
+
+    # ------------------------------------------------------------------
+    # Dimension calculators
+    # ------------------------------------------------------------------
+    def _structural_score(self, tokens: list) -> float:
+        """More tokens + unique concepts = higher structural complexity."""
+        n      = len(tokens)
+        unique = len(set(tokens))
+        # Normalise: 10 tokens = moderate complexity
+        return float(np.clip((n / 10.0) * 0.5 + (unique / n if n > 0 else 0) * 0.5, 0, 1))
+
+    def _novelty_score(self, vec: np.ndarray) -> float:
+        """
+        How far is this from anything the system has seen before?
+        High novelty = far from known abstractions.
+        """
+        candidates = self.abstraction.query_abstractions(vec, top_k=1)
+        if not candidates:
+            return 1.0  # completely novel
+        best_sim = candidates[0][0]
+        return float(np.clip(1.0 - best_sim, 0.0, 1.0))
+
+    def _depth_score(self, intent: str, tokens: list) -> float:
+        """
+        Estimate reasoning depth from linguistic markers.
+        Multi-step intents score higher.
+        """
+        depth_markers = {
+            "high":   ["analyze", "verify", "optimize", "refactor",
+                       "debug", "compare", "evaluate", "synthesize"],
+            "medium": ["write", "scaffold", "create", "build",
+                       "generate", "implement"],
+            "low":    ["run", "check", "show", "list", "get"],
+        }
+        for token in tokens:
+            if token in depth_markers["high"]:
+                return 0.8
+            if token in depth_markers["medium"]:
+                return 0.5
+            if token in depth_markers["low"]:
+                return 0.2
+        # Connectives suggest multi-step reasoning
+        if any(w in intent.lower() for w in ["and then", "after", "before", "while"]):
+            return 0.9
+        return 0.4  # default moderate
+
+    def _ambiguity_score(self, vec: np.ndarray) -> float:
+        """
+        How spread are the top matches in abstraction space?
+        High spread = high ambiguity = harder problem.
+        """
+        candidates = self.abstraction.query_abstractions(vec, top_k=3)
+        if len(candidates) < 2:
+            return 0.5
+        scores = [c[0] for c in candidates]
+        spread = float(np.std(scores))
+        return float(np.clip(spread * 4.0, 0.0, 1.0))
+
+    def _tier(self, score: float) -> str:
+        if score < self.TRIVIAL_THRESHOLD:
+            return "TRIVIAL"
+        elif score < self.SIMPLE_THRESHOLD:
+            return "SIMPLE"
+        elif score < self.MODERATE_THRESHOLD:
+            return "MODERATE"
+        elif score < self.COMPLEX_THRESHOLD:
+            return "COMPLEX"
+        else:
+            return "FRONTIER"
+
+    # ------------------------------------------------------------------
+    # Trend analysis
+    # ------------------------------------------------------------------
+    def complexity_trend(self) -> dict:
+        """Is the system tackling harder or easier problems over time?"""
+        if len(self._history) < 5:
+            return {"status": "Insufficient data"}
+        recent   = float(np.mean(self._history[-5:]))
+        baseline = float(np.mean(self._history))
+        trend    = "increasing" if recent > baseline + 0.05 else \
+                   "decreasing" if recent < baseline - 0.05 else "stable"
+        return {
+            "recent_avg":  round(recent,   4),
+            "baseline":    round(baseline, 4),
+            "trend":       trend,
+            "sample_size": len(self._history),
+        }
+
+    def report(self) -> dict:
+        if not self._log:
+            return {"status": "No assessments yet"}
+        tiers = {}
+        for e in self._log:
+            t = e.get("tier", "UNKNOWN")
+            tiers[t] = tiers.get(t, 0) + 1
+        return {
+            "total_assessments": len(self._log),
+            "tier_distribution": tiers,
+            "avg_complexity":    round(float(np.mean([e["score"] for e in self._log])), 4),
+            "trend":             self.complexity_trend(),
+        }

src/cognition/self_model.py

@@ -0,0 +1,203 @@
+"""
+SelfModel — Vitalis FSI
+
+Vitalis maintains a coherent understanding of its own architecture,
+capabilities, limitations, and growth trajectory.
+
+This is not a static config file.
+This is a living self-representation that updates as the system evolves.
+
+The SelfModel answers:
+  - What am I capable of right now?
+  - What are my current limitations?
+  - How have I grown since initialization?
+  - What is the next capability boundary I should push?
+  - Am I operating within my identity alignment?
+"""
+import numpy as np
+import os
+import json
+import time
+from vitalis_ide.math_core.kernel import VitalisKernel
+from src.brain.resonance import ResonanceEngine
+from src.hippocampus import Hippocampus
+
+
+class SelfModel:
+    # Capability thresholds — evolve as resonance weights grow
+    CAPABILITY_THRESHOLD  = 1.2   # resonance weight above this = capable
+    LIMITATION_THRESHOLD  = 0.5   # resonance weight below this = limited
+    FRONTIER_THRESHOLD    = 0.8   # between = frontier (learning zone)
+
+    def __init__(self):
+        self.kernel    = VitalisKernel()
+        self.resonance = ResonanceEngine()
+        self.hippocampus = Hippocampus()
+        self.path      = os.path.expanduser(
+            "~/.vitalis_workspace/self_model.json"
+        )
+        self._birth_time = self._load_birth_time()
+        self._snapshots  = self._load_snapshots()
+
+    def _load_birth_time(self) -> float:
+        if os.path.exists(self.path):
+            with open(self.path) as f:
+                data = json.load(f)
+                return data.get("birth_time", time.time())
+        birth = time.time()
+        self._save({"birth_time": birth, "snapshots": []})
+        return birth
+
+    def _load_snapshots(self) -> list:
+        if os.path.exists(self.path):
+            with open(self.path) as f:
+                return json.load(f).get("snapshots", [])
+        return []
+
+    def _save(self, data: dict = None):
+        os.makedirs(os.path.dirname(self.path), exist_ok=True)
+        if data is None:
+            data = {
+                "birth_time": self._birth_time,
+                "snapshots":  self._snapshots[-100:],
+            }
+        with open(self.path, "w") as f:
+            json.dump(data, f, indent=2)
+
+    # ------------------------------------------------------------------
+    # Core self-assessment
+    # ------------------------------------------------------------------
+    def assess(self) -> dict:
+        """
+        Full self-assessment. Returns current capability map,
+        limitations, frontier zones, and growth trajectory.
+        """
+        resonance_report = self.resonance.report()
+        weights          = self.resonance.weights
+        memory_report    = self.hippocampus.memory_report()
+
+        # Capability map
+        capabilities = []
+        limitations  = []
+        frontiers    = []
+
+        for pattern, weight in weights.items():
+            if weight >= self.CAPABILITY_THRESHOLD:
+                capabilities.append((pattern, round(weight, 3)))
+            elif weight <= self.LIMITATION_THRESHOLD:
+                limitations.append((pattern, round(weight, 3)))
+            else:
+                frontiers.append((pattern, round(weight, 3)))
+
+        capabilities.sort(key=lambda x: x[1], reverse=True)
+        limitations.sort(key=lambda x: x[1])
+        frontiers.sort(key=lambda x: x[1], reverse=True)
+
+        # Growth metrics
+        age_hours   = (time.time() - self._birth_time) / 3600
+        mem_count   = len(memory_report)
+        avg_strength = float(np.mean([
+            v["strength"] for v in memory_report.values()
+        ])) if memory_report else 0.0
+
+        # Next capability boundary — highest frontier item
+        next_boundary = frontiers[0][0] if frontiers else "unexplored"
+
+        # Identity coherence — how aligned are capabilities with identity?
+        identity_vec = self._load_identity_vec()
+        coherence    = self._identity_coherence(identity_vec, capabilities)
+
+        assessment = {
+            "timestamp":       time.time(),
+            "age_hours":       round(age_hours, 2),
+            "capabilities":    capabilities[:10],
+            "limitations":     limitations[:5],
+            "frontiers":       frontiers[:5],
+            "next_boundary":   next_boundary,
+            "memory_count":    mem_count,
+            "memory_strength": round(avg_strength, 4),
+            "identity_coherence": round(coherence, 4),
+            "total_patterns":  len(weights),
+            "growth_index":    self._growth_index(),
+        }
+
+        # Snapshot for trajectory tracking
+        self._snapshots.append({
+            "timestamp":    assessment["timestamp"],
+            "capabilities": len(capabilities),
+            "limitations":  len(limitations),
+            "growth_index": assessment["growth_index"],
+        })
+        self._save()
+
+        return assessment
+
+    # ------------------------------------------------------------------
+    # Internal
+    # ------------------------------------------------------------------
+    def _load_identity_vec(self) -> np.ndarray:
+        path = os.path.expanduser("~/.vitalis_workspace/identity.npy")
+        if os.path.exists(path):
+            return np.load(path)
+        return np.ones(self.kernel.dim, dtype=np.int8)
+
+    def _identity_coherence(
+        self, identity_vec: np.ndarray, capabilities: list
+    ) -> float:
+        """
+        How aligned are current capabilities with the system's identity?
+        High coherence = acting true to itself.
+        """
+        if not capabilities:
+            return 0.5
+        sims = []
+        for pattern, _ in capabilities[:5]:
+            cap_vec = self.kernel.vectorize_tokens(
+                pattern.split("_"), positional=False
+            )
+            sims.append(self.kernel.similarity(identity_vec, cap_vec))
+        return float(np.mean(sims))
+
+    def _growth_index(self) -> float:
+        """
+        Single metric for overall growth.
+        Combines: total patterns, avg weight, memory count.
+        """
+        weights = self.resonance.weights
+        if not weights:
+            return 0.0
+        avg_w    = float(np.mean(list(weights.values())))
+        n        = len(weights)
+        mem      = len(self.hippocampus.all_slots())
+        # Normalised growth index
+        return round(float(np.log1p(n) * avg_w + np.log1p(mem) * 0.1), 4)
+
+    def growth_trajectory(self) -> dict:
+        """How has the system grown over time?"""
+        if len(self._snapshots) < 2:
+            return {"status": "Insufficient snapshots"}
+        first = self._snapshots[0]
+        last  = self._snapshots[-1]
+        return {
+            "snapshots":          len(self._snapshots),
+            "growth_index_delta": round(
+                last["growth_index"] - first["growth_index"], 4
+            ),
+            "capability_delta":   last["capabilities"] - first["capabilities"],
+            "limitation_delta":   last["limitations"]  - first["limitations"],
+            "time_span_hours":    round(
+                (last["timestamp"] - first["timestamp"]) / 3600, 2
+            ),
+        }
+
+    def report(self) -> dict:
+        assessment = self.assess()
+        return {
+            "age_hours":          assessment["age_hours"],
+            "growth_index":       assessment["growth_index"],
+            "capabilities":       len(assessment["capabilities"]),
+            "limitations":        len(assessment["limitations"]),
+            "next_boundary":      assessment["next_boundary"],
+            "identity_coherence": assessment["identity_coherence"],
+            "trajectory":         self.growth_trajectory(),
+        }

src/cognitive/reasoning_engine.py

@@ -0,0 +1,21 @@
+from ..dream_engine.helix_memory import HelixMemory
+import numpy as np
+
+class ReasoningEngine:
+    MODE_MAP = {
+        "question":      "EXECUTION",
+        "instruction":   "RECOVERY",
+        "explanation":   "ANALYTICAL",
+        "unknown":       "EXPLORATORY",
+    }
+
+    def __init__(self, helix_path):
+        self.helix = HelixMemory(helix_path)
+
+    def select_mode(self, hv: np.ndarray) -> str:
+        prototypes = self.helix.retrieve(hv, top_k=1)
+        if not prototypes:
+            return "EXPLORATORY"
+        proto, meta = prototypes[0]
+        label = meta.get("label", "unknown")
+        return self.MODE_MAP.get(label, "EXPLORATORY")

src/dream_engine/helix_memory.py

@@ -0,0 +1,52 @@
+import numpy as np
+import pickle
+from pathlib import Path
+from typing import List, Tuple, Dict, Optional
+
+class HelixMemory:
+    def __init__(self, storage_path: Path):
+        self.storage_path = storage_path
+        self._load()
+
+    def _load(self) -> None:
+        if self.storage_path.exists():
+            with self.storage_path.open("rb") as f:
+                self.entries: List[Tuple[int, np.ndarray, int, dict]] = pickle.load(f)
+        else:
+            self.entries = []
+
+    def _save(self) -> None:
+        self.storage_path.parent.mkdir(parents=True, exist_ok=True)
+        with self.storage_path.open("wb") as f:
+            pickle.dump(self.entries, f)
+
+    def add(self, hv: np.ndarray, meta: Optional[dict] = None) -> None:
+        meta = meta or {}
+        for i, (code, proto, cnt, old_meta) in enumerate(self.entries):
+            sim = np.mean(hv == proto)
+            if sim > 0.85:
+                merged = {**old_meta, **meta}
+                self.entries[i] = (code, proto, cnt + 1, merged)
+                self._save()
+                return
+
+        new_code = len(self.entries)
+        self.entries.append((new_code, hv.copy(), 1, meta))
+        self._save()
+
+    def retrieve(self, hv: np.ndarray, top_k: int = 3) -> List[Tuple[np.ndarray, dict]]:
+        sims = [(np.mean(hv == proto), proto, meta) for _, proto, _, meta in self.entries]
+        sims.sort(key=lambda x: x[0], reverse=True)
+        return [(proto, meta) for _, proto, meta in sims[:top_k]]
+
+    def reconstruct(self, code_id: int) -> np.ndarray:
+        for cid, proto, _, _ in self.entries:
+            if cid == code_id:
+                return proto.copy()
+        raise KeyError(f"Helix code {code_id} not found")
+
+    def most_uncertain(self) -> Tuple[int, np.ndarray]:
+        if not self.entries:
+            raise RuntimeError("HelixMemory empty")
+        entry = min(self.entries, key=lambda e: e[2])
+        return entry[0], entry[1]

src/vitalis_ide/cli/main.py

@@ -0,0 +1,68 @@
+import argparse
+import sys
+import time
+from pathlib import Path
+
+from audio_ear.recorder import record_to_wav
+from audio_ear.feature_extractor import extract_features
+from hdc_encoder.encoder import encode
+from dream_engine.consolidator import DreamEngine
+from dream_engine.helix_memory import HelixMemory
+
+def log_success(intent: str, details: dict) -> None:
+    from cognition.mind import VitalisMind
+    mind = VitalisMind()
+    mind.ledger.append({"type": "success", "intent": intent, "details": details})
+
+def log_failure(intent: str, error: str) -> None:
+    from cognition.mind import VitalisMind
+    mind = VitalisMind()
+    mind.ledger.append({"type": "failure", "intent": intent, "error": error})
+
+def cmd_listen(args: argparse.Namespace) -> None:
+    out_path = Path(args.output or f"/tmp/live_{int(time.time())}.wav")
+    print(f"🔊 Recording {args.duration}s -> {out_path}")
+    record_to_wav(duration_sec=args.duration, out_path=out_path)
+    mfcc, prosody = extract_features(out_path)
+    hv = encode(mfcc, prosody)
+    
+    helix_path = Path.home() / ".vitalis_workspace" / "helix_memory.pkl"
+    helix = HelixMemory(helix_path)
+    dreamer = DreamEngine(helix)
+    dreamer.ingest(hv, meta={"source": str(out_path), "prosody": prosody})
+    
+    log_success("listen_live", {"file": str(out_path)})
+    print("✅ Ingested into DreamEngine.")
+
+def cmd_dream(args: argparse.Namespace) -> None:
+    helix_path = Path.home() / ".vitalis_workspace" / "helix_memory.pkl"
+    helix = HelixMemory(helix_path)
+    dreamer = DreamEngine(helix)
+    dreamer.dream(force=True)
+    print("💤 Consolidation complete.")
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(prog="vitalis")
+    sub = parser.add_subparsers(dest="command", required=True)
+
+    p_listen = sub.add_parser("listen")
+    p_listen.add_argument("-d", "--duration", type=int, default=10)
+    p_listen.add_argument("-o", "--output", type=str)
+    p_listen.set_defaults(func=cmd_listen)
+
+    p_dream = sub.add_parser("dream")
+    p_dream.set_defaults(func=cmd_dream)
+
+    return parser
+
+def main(argv: list | None = None) -> None:
+    parser = build_parser()
+    args = parser.parse_args(argv)
+    try: args.func(args)
+    except Exception as exc:
+        log_failure(intent=args.command, error=str(exc))
+        print(f"❌ Failed: {exc}", file=sys.stderr)
+        sys.exit(1)
+
+if __name__ == "__main__":
+    main()

tests/test_encoder.py

@@ -0,0 +1,43 @@
+import numpy as np
+from src.hdc_encoder.encoder import encode, DIM
+
+class TestHDCEncoder:
+    def setup_method(self):
+        """Initialize base parameters before each test to guarantee state isolation."""
+        self.prosody = {"pitch": 140.0, "energy": 0.25, "tempo": 115.0, "pause_ratio": 0.15}
+        self.frames = 40
+        self.mfcc = np.random.rand(13, self.frames).astype(np.float32)
+
+    def test_encoder_dimensionality(self):
+        """Verify the output vector matches the strict dimensional parameters."""
+        hv = encode(self.mfcc, self.prosody)
+        
+        assert hv.shape == (DIM,), f"Dimensionality failure: Expected {DIM}, got {hv.shape[0]}"
+        assert hv.dtype == np.uint8, f"Type architecture failure: Expected uint8, got {hv.dtype}"
+
+    def test_temporal_sequence_asymmetry(self):
+        """
+        Verify that reversing the audio sequence produces a fundamentally different 
+        hypervector. This proves the encoder captures the temporal direction of speech.
+        """
+        mfcc_reversed = self.mfcc[:, ::-1]
+        
+        hv_forward = encode(self.mfcc, self.prosody)
+        hv_reversed = encode(mfcc_reversed, self.prosody)
+        
+        similarity = np.mean(hv_forward == hv_reversed)
+        
+        # In an orthogonal 10k dimensional space, random vectors share ~50% similarity.
+        # We enforce a strict threshold to ensure the temporal shift breaks vector alignment.
+        assert similarity < 0.60, f"Commutativity failure: Temporal sequence not isolated. Similarity: {similarity}"
+
+    def test_prosody_modulation(self):
+        """Verify that altering emotional/prosodic intent alters the final memory vector."""
+        angry_prosody = {"pitch": 240.0, "energy": 0.85, "tempo": 160.0, "pause_ratio": 0.05}
+        
+        hv_base = encode(self.mfcc, self.prosody)
+        hv_angry = encode(self.mfcc, angry_prosody)
+        
+        similarity = np.mean(hv_base == hv_angry)
+        
+        assert similarity < 0.95, "Modulation failure: Prosody shifts did not impact the spatial vector."