Implement GlyphOS Cognitive Kernel

Add a system service layer on top of LAIN cognition and Supercharged Glyph Registry: Components: - glyphos/cognitive_kernel.py: CognitiveKernel class + functional API * CognitiveKernel: Main orchestrator with execute_gx(), execute_symbolic() * Result accessors: get_last_result(), get_last_trace(), get_last_fused_symbol() * get_kernel(): Singleton kernel instance * run_gx(): Convenience function for global kernel * kernel_status(): Status introspection - glyphos/__init__.py: Package initialization - tests/test_cognitive_kernel.py: Comprehensive test suite (8 tests, 100% pass) * Kernel initialization and warmup * GX execution and result validation * Result accessor methods * Singleton pattern * Functional API - COGNITIVE_KERNEL.md: Complete documentation Test Results: - 12 registry tests ✅ - 10 glyph bridge tests ✅ - 6 integration suites ✅ - 8 cognitive kernel tests ✅ - Total: 36 tests, 0 failures No breaking changes - all existing tests pass.
2026-05-20 18:03:25 -04:00
parent 02a298f44c
commit 5c4bfb2dc1
6 changed files with 1159 additions and 0 deletions
@@ -0,0 +1,285 @@
+"""GlyphOS Cognitive Kernel
+
+Orchestrates LAIN cognition engine with Supercharged Glyph Registry.
+Provides a clean service API for executing GX files and managing glyph-aware analysis.
+"""
+
+from typing import Optional, Dict, Any, List
+import time
+from pathlib import Path
+
+from gx_lain.runtime import execute_gx_path, load_gx, normalize_segments, map_lanes, build_envelope, execute_with_lain
+from glyphs.super_registry import load_all_supercharged, super_stats
+
+
+class CognitiveKernel:
+    """System service for GlyphOS cognition pipeline.
+
+    Orchestrates:
+    - LAIN 8-lane symbolic cognition
+    - Supercharged Glyph Registry integration
+    - Result caching and introspection
+    """
+
+    def __init__(self, *, auto_load_glyphs: bool = True):
+        """Initialize the Cognitive Kernel.
+
+        Args:
+            auto_load_glyphs: If True, load Supercharged Glyphs during warmup.
+                             Defaults to True.
+        """
+        self._auto_load_glyphs = auto_load_glyphs
+        self._last_result: Optional[Dict[str, Any]] = None
+        self._startup_time: Optional[float] = None
+        self._glyph_stats_cache: Optional[Dict[str, Any]] = None
+        self._warmed_up = False
+        self._last_mode: Optional[str] = None
+
+    def warmup(self) -> None:
+        """Perform one-time initialization.
+
+        Loads:
+        - Supercharged Glyphs (if auto_load_glyphs)
+        - Registry statistics
+
+        Records:
+        - Kernel startup time
+        """
+        if self._warmed_up:
+            return
+
+        self._startup_time = time.time()
+
+        if self._auto_load_glyphs:
+            load_all_supercharged()
+
+        # Cache registry stats
+        self._glyph_stats_cache = super_stats()
+
+        self._warmed_up = True
+
+    def execute_gx(
+        self,
+        gx_path: str,
+        *,
+        mode: str = "analyze",
+        context: Optional[Dict[str, Any]] = None
+    ) -> Dict[str, Any]:
+        """Execute a .gx file through the full cognition pipeline.
+
+        Args:
+            gx_path: Path to .gx file
+            mode: Cognitive mode (e.g., "analyze", "debug")
+            context: Optional execution context dict
+
+        Returns:
+            ExecutionResult dict with:
+            - fused_symbol: Combined 8-lane analysis
+            - output_text: Rendered analysis
+            - cognition_trace: Step-by-step processing
+            - diagnostics: Performance metrics + glyph resonance
+        """
+        if not self._warmed_up:
+            self.warmup()
+
+        # Build context with mode
+        exec_context = context or {}
+        exec_context["cognitive_mode"] = mode
+
+        # Execute through LAIN pipeline
+        result = execute_gx_path(gx_path, context=exec_context)
+
+        # Cache result
+        self._last_result = result
+        self._last_mode = mode
+
+        return result
+
+    def execute_symbolic(
+        self,
+        manifest: Dict[str, Any],
+        segments: List[Dict[str, Any]],
+        payload: bytes,
+        *,
+        mode: str = "analyze",
+        context: Optional[Dict[str, Any]] = None
+    ) -> Dict[str, Any]:
+        """Execute cognition on in-memory GX components (no filesystem).
+
+        Args:
+            manifest: GX manifest dict
+            segments: GX segments list
+            payload: Compressed GX payload bytes
+            mode: Cognitive mode
+            context: Optional execution context
+
+        Returns:
+            ExecutionResult dict
+        """
+        if not self._warmed_up:
+            self.warmup()
+
+        # Build context with mode
+        exec_context = context or {}
+        exec_context["cognitive_mode"] = mode
+
+        # Normalize segments
+        normalized_segs = normalize_segments(segments, payload)
+
+        # Map to lanes (0-7)
+        lane_assignments = map_lanes(manifest, normalized_segs)
+
+        # Build envelope
+        envelope = build_envelope(manifest, normalized_segs)
+
+        # Execute through LAIN with glyph bridge
+        result = execute_with_lain(manifest, envelope, lane_assignments, exec_context)
+
+        # Cache result
+        self._last_result = result
+        self._last_mode = mode
+
+        return result
+
+    def get_glyph_stats(self) -> Dict[str, Any]:
+        """Get Supercharged Glyph Registry statistics.
+
+        Returns:
+            Dict with:
+            - total_glyphs: 600
+            - categories: List of category names
+            - fields_present: All fields in registry
+            - sample_ids: First 5 glyph IDs
+            - loaded: Whether registry is loaded
+            - load_path: Path to data file
+            - kernel_startup_time: Kernel warmup timestamp
+        """
+        if not self._warmed_up:
+            self.warmup()
+
+        stats = self._glyph_stats_cache or super_stats()
+
+        # Add kernel metadata
+        stats["kernel_startup_time"] = self._startup_time
+
+        return stats
+
+    def get_last_result(self) -> Optional[Dict[str, Any]]:
+        """Return the last ExecutionResult, if any.
+
+        Returns:
+            Full ExecutionResult dict or None
+        """
+        return self._last_result
+
+    def get_last_trace(self) -> Optional[List[Dict[str, Any]]]:
+        """Return cognition_trace from last ExecutionResult, if present.
+
+        Returns:
+            List of trace steps or None
+        """
+        if self._last_result is None:
+            return None
+        return self._last_result.get("cognition_trace")
+
+    def get_last_fused_symbol(self) -> Optional[Dict[str, Any]]:
+        """Return fused_symbol from last ExecutionResult, if present.
+
+        Returns:
+            Fused symbol dict or None
+        """
+        if self._last_result is None:
+            return None
+        return self._last_result.get("fused_symbol")
+
+    def get_last_resonance(self) -> Optional[Dict[str, Any]]:
+        """Return resonance metrics from last ExecutionResult, if present.
+
+        Returns:
+            Dict with:
+            - resonance: Overall resonance metrics (if present)
+            - glyph_resonance: Glyph-specific metrics (if glyph was used)
+            Or None if no result
+        """
+        if self._last_result is None:
+            return None
+
+        diagnostics = self._last_result.get("diagnostics", {})
+
+        return {
+            "resonance": diagnostics.get("resonance"),
+            "glyph_resonance": diagnostics.get("glyph_resonance"),
+            "elapsed": diagnostics.get("elapsed"),
+        }
+
+
+# Global singleton kernel instance
+_GLOBAL_KERNEL: Optional[CognitiveKernel] = None
+
+
+def get_kernel() -> CognitiveKernel:
+    """Get or create the singleton CognitiveKernel instance.
+
+    On first call:
+    - Creates a new CognitiveKernel
+    - Calls warmup() to initialize glyphs
+
+    Returns:
+        Singleton CognitiveKernel instance
+    """
+    global _GLOBAL_KERNEL
+
+    if _GLOBAL_KERNEL is None:
+        _GLOBAL_KERNEL = CognitiveKernel(auto_load_glyphs=True)
+        _GLOBAL_KERNEL.warmup()
+
+    return _GLOBAL_KERNEL
+
+
+def run_gx(
+    gx_path: str,
+    *,
+    mode: str = "analyze",
+    context: Optional[Dict[str, Any]] = None
+) -> Dict[str, Any]:
+    """Convenience function: execute .gx through the global kernel.
+
+    Equivalent to: get_kernel().execute_gx(gx_path, mode=mode, context=context)
+
+    Args:
+        gx_path: Path to .gx file
+        mode: Cognitive mode
+        context: Optional execution context
+
+    Returns:
+        ExecutionResult dict
+    """
+    kernel = get_kernel()
+    return kernel.execute_gx(gx_path, mode=mode, context=context)
+
+
+def kernel_status() -> Dict[str, Any]:
+    """Get status of the global CognitiveKernel.
+
+    Returns:
+        Dict with:
+        - glyph_stats: Registry metadata (total_glyphs, categories, etc.)
+        - last_run_present: Whether a result has been cached
+        - last_mode: Mode of last execution (or None)
+        - last_elapsed: Elapsed time from last run (or None)
+        - startup_time: Kernel warmup timestamp
+        - is_warmed_up: Whether kernel has been initialized
+    """
+    kernel = get_kernel()
+    glyph_stats = kernel.get_glyph_stats()
+    last_result = kernel.get_last_result()
+    last_resonance = kernel.get_last_resonance()
+
+    return {
+        "glyph_stats": glyph_stats,
+        "last_run_present": last_result is not None,
+        "last_mode": kernel._last_mode,
+        "last_elapsed": last_resonance.get("elapsed") if last_resonance else None,
+        "startup_time": kernel._startup_time,
+        "is_warmed_up": kernel._warmed_up,
+    }