GlyphRunner System
69c97e125a
New instructions: - STREAM: Line-by-line execution and output - CHAIN: Named execution boundaries - CALL_GLYPH: Invoke glyph-aware cognition - SET_CONTEXT: Set symbolic/cognitive context metadata - LOG: Structured logging Symbolic execution mode: - SET_MODE "symbolic" routes prompts through LAIN 8-lane cognition pipeline - run_symbolic_prompt() compresses prompt, builds manifest, executes via execute_symbolic() - Full integration with glyphos/cognitive_kernel.py GPU-accelerated path: - xic_extensions/gpu_runtime.py: has_gpu() probes torch.cuda, run_on_gpu() executes - SET_PARAM "use_gpu" true enables GPU (auto-fallback to CPU if unavailable) - No required GPU dependencies; system works equally on CPU Demo programs: - demo_symbolic.gx.json: Shows symbolic mode through LAIN pipeline - demo_gpu.gx.json: Shows GPU mode with CPU fallback Backward compatibility: - All 4 original ops unchanged; 5 new ops added to OP_TABLE - xic_vm.py, xic_executor.py: No changes (pure dispatcher pattern holds) - demo_chat.gx.json: Still executes identically - All existing GlyphRunner commands: Unchanged behavior Architecture: - Lazy imports prevent circular dependencies (xic_ops, glyphos, xic_extensions) - Clean separation: XIC is client of cognition layer - Zero breaking changes; additive extension only - No XIC v2 binary format; all within v1 JSON+.gx architecture Validation: - 10 integration tests: all passing - Backward compat verified with original demo - Symbolic and GPU modes tested end-to-end - No external dependencies required (GPU optional) Co-contributors: LAIN cognition engine, gx_compiler GSZ3, glyphos event system
376 lines
11 KiB
Python
376 lines
11 KiB
Python
"""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).
|
|
|
|
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(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)
|
|
|
|
# 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 run_symbolic_prompt(prompt: str, context: dict | None = None) -> str:
|
|
"""Entry point for symbolic execution from XIC.
|
|
|
|
Compresses the prompt text into GSZ3 bytes, builds a minimal manifest,
|
|
and routes through the full LAIN 8-lane cognition pipeline via
|
|
CognitiveKernel.execute_symbolic(). Returns the output_text string.
|
|
|
|
Args:
|
|
prompt: User or system prompt text
|
|
context: Optional symbolic/cognitive context dict
|
|
|
|
Returns:
|
|
String result from the 8-lane cognition pipeline
|
|
"""
|
|
from gx_compiler.compressor import GXCompressor
|
|
|
|
kernel = get_kernel()
|
|
prompt_bytes = prompt.encode("utf-8")
|
|
|
|
try:
|
|
payload = GXCompressor.compress(prompt)
|
|
except Exception as e:
|
|
return f"[Symbolic Error] Compression failed: {e}"
|
|
|
|
manifest = {
|
|
"source_file": "<symbolic>",
|
|
"source_type": "symbolic",
|
|
"version": "1.0.0",
|
|
"contributor": "XIC-symbolic",
|
|
"segments": [{"id": "seg_0", "start": 0, "end": 1,
|
|
"start_byte": 0, "end_byte": len(prompt_bytes)}],
|
|
}
|
|
segments = [{"id": "seg_0", "start": 0, "end": 1,
|
|
"start_byte": 0, "end_byte": len(prompt_bytes)}]
|
|
|
|
result = kernel.execute_symbolic(
|
|
manifest=manifest,
|
|
segments=segments,
|
|
payload=payload,
|
|
mode="symbolic",
|
|
context=context or {},
|
|
)
|
|
|
|
return result.get("output_text") or result.get("fused_symbol", {}).get("summary", prompt)
|
|
|
|
|
|
# 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,
|
|
}
|