"""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 from glyphos.events import emit 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 Emits: - kernel.warmup.completed event """ 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 # Emit warmup completed event emit("kernel.warmup.completed", { "glyph_stats": self._glyph_stats_cache, "startup_time": self._startup_time, }) 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 Emits: - cognition.started event - cognition.completed event - glyph.resonance.updated event (if glyph resonance present) """ if not self._warmed_up: self.warmup() # Emit cognition started event emit("cognition.started", { "gx_path": gx_path, "mode": mode, "context": context, }) # 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 # Extract event payload from result fused_symbol = result.get("fused_symbol", {}) diagnostics = result.get("diagnostics", {}) # Emit cognition completed event emit("cognition.completed", { "gx_path": gx_path, "mode": mode, "elapsed": diagnostics.get("elapsed"), "glyph_resonance": diagnostics.get("glyph_resonance"), "summary": fused_symbol.get("summary"), }) # Emit glyph resonance event if present glyph_resonance = diagnostics.get("glyph_resonance") if glyph_resonance and glyph_resonance.get("glyph_found"): emit("glyph.resonance.updated", { "glyph_id": glyph_resonance.get("glyph_id"), "glyph_score": glyph_resonance.get("glyph_score"), "glyph_resonance": glyph_resonance, }) 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). Supports both single-glyph and multi-glyph resonance modes. Args: manifest: GX manifest dict segments: GX segments list payload: Compressed GX payload bytes mode: Cognitive mode context: Optional execution context. May contain: - glyph_id: Single glyph for glyph-aware cognition - glyph_ids: List of glyphs for multi-glyph resonance Returns: ExecutionResult dict with fused_symbol containing: - Single-glyph: summary, glyph_ids=[glyph_id], resonance_map - Multi-glyph: summary, glyph_ids=[...], resonance_map with all metrics """ if not self._warmed_up: self.warmup() # Build context with mode exec_context = context or {} exec_context["cognitive_mode"] = mode # Check for multi-glyph resonance context glyph_ids = exec_context.get("glyph_ids") is_multi_glyph = glyph_ids is not None and len(glyph_ids) > 0 # Normalize segments normalized_segs = normalize_segments(manifest, segments, payload) # Map to lanes (0-7) lane_assignments = map_lanes(normalized_segs) # Build envelope envelope = build_envelope(manifest, lane_assignments, payload, context=exec_context) # Execute through LAIN with glyph bridge result = execute_with_lain(envelope) # Post-process for multi-glyph resonance if requested if is_multi_glyph: multi_glyph_metrics = self.compute_multi_glyph_resonance(glyph_ids, result) # Merge multi-glyph resonance into fused_symbol if "fused_symbol" not in result: result["fused_symbol"] = {} fused = result["fused_symbol"] fused["glyph_ids"] = glyph_ids fused["global_resonance_score"] = multi_glyph_metrics["global_resonance_score"] # Build resonance_map from computed metrics if "resonance_map" not in fused: fused["resonance_map"] = {} for glyph_id, metrics in multi_glyph_metrics["resonances"].items(): fused["resonance_map"][glyph_id] = metrics # Store guardrails info if any triggered if multi_glyph_metrics["guardrails_triggered"]: if "diagnostics" not in result: result["diagnostics"] = {} result["diagnostics"]["guardrails"] = multi_glyph_metrics["guardrails_triggered"] # 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"), } def compute_multi_glyph_resonance( self, glyph_ids: List[str], result: Dict[str, Any] ) -> Dict[str, Any]: """Compute multi-glyph resonance metrics from execution result. Uses actual glyph metadata from the registry to compute real resonance scores: - weight: Based on glyph score and activation state - lineage_score: From lineage.inheritanceWeight - contributor_score: From originalMetrics connectivity - frequency_score: From praw vector magnitude - grammar_score: From originalMetrics stability Args: glyph_ids: List of glyph IDs to compute resonance for result: Execution result dict from LAIN Returns: Dict with: - glyph_ids: Input glyph list - resonances: Dict mapping glyph_id → metrics - global_resonance_score: Weighted average across glyphs - guardrails_triggered: List of guardrail messages """ from glyphs import get_super resonances = {} scores = [] for glyph_id in glyph_ids: glyph = get_super(glyph_id) if not glyph: continue metrics = glyph.get('originalMetrics', {}) activation = glyph.get('activation', {}) lineage = glyph.get('lineage', {}) praw = glyph.get('praw', {}) # Compute weight from glyph score (max 335) and activation score = glyph.get('score', 0) activation_score = activation.get('score', 0) weight = min(1.0, (score / 335) * 0.7 + (activation_score / 100) * 0.3) # Compute lineage score from inheritance weight inheritance_weight = lineage.get('inheritanceWeight', 0) lineage_score = inheritance_weight # Compute contributor score from connectivity metric connectivity = metrics.get('connectivity', 50) contributor_score = connectivity / 100 # Compute frequency score from praw vector magnitude praw_values = [praw.get('P', 0), praw.get('R', 0), praw.get('A', 0), praw.get('W', 0)] praw_magnitude = (sum(v * v for v in praw_values) ** 0.5) / 200 frequency_score = min(1.0, praw_magnitude) # Compute grammar score from stability metric stability = metrics.get('stability', 50) grammar_score = stability / 100 metrics = { "weight": round(weight, 4), "lineage_score": round(lineage_score, 4), "contributor_score": round(contributor_score, 4), "frequency_score": round(frequency_score, 4), "grammar_score": round(grammar_score, 4), } resonances[glyph_id] = metrics scores.append(metrics["weight"]) # Compute global resonance as weighted average global_resonance = sum(scores) / len(scores) if scores else 0.0 return { "glyph_ids": glyph_ids, "resonances": resonances, "global_resonance_score": round(min(1.0, global_resonance), 4), "guardrails_triggered": [], } # 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, }