GlyphRunner System
4bc49c90b3
New module: - gx_lain/lain_glyph_bridge.py: Bridge connecting LedoGlyph600 to LAIN cognition Functions: - load_glyph_context(manifest, context): Load relevant glyph from registry - inject_glyph_metadata_into_lane(lane_result, glyph_context): Add glyph fields to lane - compute_glyph_resonance(glyph_context): Calculate glyph resonance metrics - augment_fused_symbol_with_glyphs(fused_symbol, glyph_context): Add glyph to final output Modified: - gx_lain/runtime.py: Integrate glyph bridge into execute_with_lain() * Load glyph context as step 1 of cognition * Inject glyph metadata into each lane result * Augment fused symbol with glyph context * Add glyph_resonance to diagnostics * Track glyph loading in cognition_trace Tests: - tests/test_lain_glyph_bridge.py: 10 comprehensive tests * Context loading (with/without glyph) * Metadata injection (preserves existing fields) * Resonance computation (4-component metric) * Symbol augmentation * Full integration test Features: - Glyph metadata: id, name, category, score, period, band - Frequency signatures: praw (P, R, A, W) - Activation envelopes: mode, score - Lineage: signature, inheritance weight - Symbolic anatomy: power, complexity, resonance, stability, connectivity, affinity - Resonance profile: activation + frequency + symbolic metrics (0.0-1.0) All 18 integration tests still passing (no regressions).
613 lines
17 KiB
Python
613 lines
17 KiB
Python
from __future__ import annotations
|
||
from typing import Dict, List, Tuple, Any, Optional
|
||
import uuid
|
||
import time
|
||
import logging
|
||
from pathlib import Path
|
||
|
||
INTERFACE_VERSION = "1.0"
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def load_gx(path: str) -> Tuple[dict, List[dict], bytes]:
|
||
"""Load a .gx file and return (manifest, segments, payload).
|
||
|
||
Thin wrapper around runtime_executor.gx_loader.load_gx().
|
||
Reconstructs segments list from manifest['codex_lineage']['segments'].
|
||
|
||
Args:
|
||
path: Path to .gx file
|
||
|
||
Returns:
|
||
Tuple of (manifest, segments_list, payload_bytes)
|
||
|
||
Raises:
|
||
FileNotFoundError: If .gx file doesn't exist
|
||
RuntimeError: If manifest is malformed
|
||
"""
|
||
from runtime_executor.gx_loader import load_gx as loader_load_gx
|
||
|
||
manifest, payload = loader_load_gx(path)
|
||
|
||
# Extract segments from manifest['codex_lineage']['segments']
|
||
codex_lineage = manifest.get("codex_lineage", {})
|
||
segments = codex_lineage.get("segments", [])
|
||
|
||
if not segments:
|
||
logger.warning(f"No segments found in {path}, continuing with empty list")
|
||
|
||
return manifest, segments, payload
|
||
|
||
|
||
def normalize_segments(
|
||
manifest: dict,
|
||
raw_segments: List[dict],
|
||
payload: bytes,
|
||
) -> List[dict]:
|
||
"""Normalize raw segments into canonical Segment schema.
|
||
|
||
Converts raw segment metadata from codex_lineage.segments into normalized form
|
||
with required keys: id, start_line, end_line, text, symbolic_lane, semantic_role.
|
||
|
||
Assumptions:
|
||
- Raw segments have: id, start, end (0-based line indices)
|
||
- Optional: lane, role metadata
|
||
- Text extraction is stubbed (reserved for future payload analysis)
|
||
|
||
Args:
|
||
manifest: GX manifest dict
|
||
raw_segments: List of raw segment dicts from codex_lineage
|
||
payload: Compressed payload bytes (reserved for text extraction)
|
||
|
||
Returns:
|
||
List of normalized segment dicts conforming to Segment schema
|
||
"""
|
||
normalized = []
|
||
|
||
for raw_seg in raw_segments:
|
||
seg_id = str(raw_seg.get("id", "unknown"))
|
||
start_line = int(raw_seg.get("start", 0))
|
||
end_line = int(raw_seg.get("end", 0))
|
||
|
||
# Explicit lane from metadata, or infer
|
||
explicit_lane = raw_seg.get("lane")
|
||
if explicit_lane is not None:
|
||
symbolic_lane = int(explicit_lane)
|
||
else:
|
||
symbolic_lane = _infer_lane(raw_seg)
|
||
|
||
# Clamp to valid range [0, 7]
|
||
symbolic_lane = max(0, min(7, symbolic_lane))
|
||
|
||
# Semantic role: infer from segment metadata
|
||
semantic_role = _infer_semantic_role(raw_seg)
|
||
|
||
# Text: stub for now; in production extract from payload via byte ranges
|
||
text = f"[segment {seg_id}: lines {start_line}–{end_line}]"
|
||
|
||
normalized_seg = {
|
||
"id": seg_id,
|
||
"start_line": start_line,
|
||
"end_line": end_line,
|
||
"text": text,
|
||
"symbolic_lane": symbolic_lane,
|
||
"semantic_role": semantic_role,
|
||
"_raw": raw_seg, # preserve original for debugging
|
||
}
|
||
|
||
normalized.append(normalized_seg)
|
||
|
||
return normalized
|
||
|
||
|
||
def _infer_lane(segment: dict) -> int:
|
||
"""Infer lane assignment from segment metadata (heuristic).
|
||
|
||
Rules (minimum viable):
|
||
- Structural markers → lane 0
|
||
- Main content → lane 1 (default)
|
||
- Comments/annotations → lane 3
|
||
- Hints → lane 4
|
||
- Author/meta → lane 6
|
||
- Time/epoch → lane 7
|
||
|
||
Args:
|
||
segment: Raw segment dict
|
||
|
||
Returns:
|
||
Lane id 0–7
|
||
"""
|
||
seg_id = str(segment.get("id", "")).lower()
|
||
|
||
# Structural
|
||
if any(x in seg_id for x in ["struct", "class", "def", "header", "header"]):
|
||
return 0
|
||
|
||
# Annotations/comments
|
||
if any(x in seg_id for x in ["comment", "annotation", "tag"]):
|
||
return 3
|
||
|
||
# Hints
|
||
if any(x in seg_id for x in ["hint", "note", "tip", "warn"]):
|
||
return 4
|
||
|
||
# Author/meta
|
||
if any(x in seg_id for x in ["meta", "author", "signature"]):
|
||
return 6
|
||
|
||
# Time/epoch
|
||
if any(x in seg_id for x in ["epoch", "time", "date", "version"]):
|
||
return 7
|
||
|
||
# Default: semantic flow
|
||
return 1
|
||
|
||
|
||
def _infer_semantic_role(segment: dict) -> str:
|
||
"""Infer semantic role from segment metadata (heuristic).
|
||
|
||
Args:
|
||
segment: Raw segment dict
|
||
|
||
Returns:
|
||
One of: "definition", "constraint", "example", "meta", "unknown"
|
||
"""
|
||
seg_id = str(segment.get("id", "")).lower()
|
||
|
||
if any(x in seg_id for x in ["def", "class", "function", "declaration"]):
|
||
return "definition"
|
||
|
||
if any(x in seg_id for x in ["constraint", "rule", "assertion", "require"]):
|
||
return "constraint"
|
||
|
||
if any(x in seg_id for x in ["example", "test", "sample", "demo"]):
|
||
return "example"
|
||
|
||
if any(x in seg_id for x in ["meta", "note", "comment", "annotation", "tag"]):
|
||
return "meta"
|
||
|
||
return "unknown"
|
||
|
||
|
||
def map_lanes(segments: List[dict]) -> Dict[int, List[dict]]:
|
||
"""Map normalized segments into 0–7 lane model.
|
||
|
||
Organizes segments by symbolic_lane into a dict where:
|
||
- Keys: lane numbers 0–7
|
||
- Values: lists of segments assigned to that lane
|
||
|
||
Lane semantics (from spec):
|
||
- 0: structural_logic
|
||
- 1: semantic_flow
|
||
- 2: compression_residue
|
||
- 3: symbolic_metadata
|
||
- 4: execution_hints
|
||
- 5: predictive_scaffolding
|
||
- 6: contributor_imprint
|
||
- 7: epoch_resonance
|
||
|
||
Args:
|
||
segments: List of normalized segment dicts
|
||
|
||
Returns:
|
||
Dict[int, List[dict]] where keys are 0–7
|
||
"""
|
||
lanes: Dict[int, List[dict]] = {i: [] for i in range(8)}
|
||
|
||
for seg in segments:
|
||
lane = seg.get("symbolic_lane", 1)
|
||
# Safety clamp
|
||
lane = max(0, min(7, lane))
|
||
lanes[lane].append(seg)
|
||
|
||
return lanes
|
||
|
||
|
||
def build_envelope(
|
||
manifest: dict,
|
||
lanes: Dict[int, List[dict]],
|
||
payload: bytes,
|
||
context: Optional[dict] = None,
|
||
) -> dict:
|
||
"""Build ExecutionEnvelope for LAIN from components.
|
||
|
||
Constructs the envelope that LAIN will consume. The envelope is immutable
|
||
from LAIN's perspective and includes all necessary context.
|
||
|
||
Args:
|
||
manifest: GX manifest dict
|
||
lanes: Dict[int, List[dict]] where keys are lane ids 0–7
|
||
payload: Compressed payload bytes
|
||
context: Optional context overrides (runtime_flags, epoch, cognitive_mode, invocation_id)
|
||
|
||
Returns:
|
||
ExecutionEnvelope dict ready for LAIN.execute()
|
||
"""
|
||
if context is None:
|
||
context = {}
|
||
|
||
base_context = {
|
||
"runtime_flags": context.get("runtime_flags", {}),
|
||
"contributor": manifest.get("contributor", "unknown"),
|
||
"epoch": context.get("epoch"),
|
||
"cognitive_mode": context.get("cognitive_mode", "analyze"),
|
||
"invocation_id": context.get("invocation_id", str(uuid.uuid4())),
|
||
"interface_version": INTERFACE_VERSION,
|
||
}
|
||
|
||
envelope = {
|
||
"manifest": manifest,
|
||
"lanes": lanes,
|
||
"payload": payload,
|
||
"context": base_context,
|
||
}
|
||
|
||
return envelope
|
||
|
||
|
||
def fuse_lanes(lane_results: Dict[int, dict]) -> dict:
|
||
"""Fuse lane results into final fused_symbol.
|
||
|
||
Merges summaries, key_points, constraints, and open_questions
|
||
from all lanes into a single coherent representation.
|
||
|
||
Args:
|
||
lane_results: Dict[lane_id, lane_result]
|
||
|
||
Returns:
|
||
Fused symbol dict with summary, key_points, constraints, open_questions
|
||
"""
|
||
summaries = []
|
||
all_key_points = []
|
||
all_constraints = []
|
||
all_questions = []
|
||
|
||
for lane_id in sorted(lane_results.keys()):
|
||
result = lane_results[lane_id]
|
||
|
||
# Collect summaries
|
||
if result.get("summary"):
|
||
summaries.append(result["summary"])
|
||
|
||
# Collect key points
|
||
all_key_points.extend(result.get("key_points", []))
|
||
|
||
# Collect constraints
|
||
all_constraints.extend(result.get("constraints", []))
|
||
|
||
# Collect open questions
|
||
all_questions.extend(result.get("open_questions", []))
|
||
|
||
# Merge summary
|
||
if summaries:
|
||
combined_summary = " | ".join(summaries)
|
||
else:
|
||
combined_summary = "No lanes processed"
|
||
|
||
# Deduplicate and limit key points
|
||
unique_key_points = list(dict.fromkeys(all_key_points))[:10]
|
||
|
||
# Deduplicate constraints
|
||
unique_constraints = list(dict.fromkeys(all_constraints))
|
||
|
||
# Deduplicate questions
|
||
unique_questions = list(dict.fromkeys(all_questions))
|
||
|
||
return {
|
||
"summary": combined_summary,
|
||
"key_points": unique_key_points,
|
||
"constraints": unique_constraints,
|
||
"open_questions": unique_questions,
|
||
}
|
||
|
||
|
||
def compute_resonance(lane_results: Dict[int, dict], context: Dict[str, Any]) -> dict:
|
||
"""Compute resonance metrics for lanes.
|
||
|
||
Simple rule: resonance[lane] = 1.0 if lane produced content, else 0.0
|
||
|
||
Args:
|
||
lane_results: Dict[lane_id, lane_result]
|
||
context: Execution context
|
||
|
||
Returns:
|
||
Dict[str, float] with resonance metrics
|
||
"""
|
||
resonance = {}
|
||
|
||
for lane_id in sorted(lane_results.keys()):
|
||
result = lane_results[lane_id]
|
||
|
||
# Lane has content if it produced a non-empty summary
|
||
has_content = bool(result.get("summary", "").strip())
|
||
resonance[f"lane_{lane_id}"] = 1.0 if has_content else 0.0
|
||
|
||
return resonance
|
||
|
||
|
||
def render_output_text(fused_symbol: dict, context: Dict[str, Any]) -> str:
|
||
"""Render human-facing output text from fused_symbol.
|
||
|
||
Format varies by cognitive_mode.
|
||
|
||
Args:
|
||
fused_symbol: Fused symbol dict
|
||
context: Execution context
|
||
|
||
Returns:
|
||
Human-readable output string
|
||
"""
|
||
mode = context.get("cognitive_mode", "analyze")
|
||
|
||
mode_label = mode.upper()
|
||
summary = fused_symbol.get("summary", "No summary")
|
||
|
||
lines = [
|
||
f"[{mode_label}]",
|
||
f"{summary}",
|
||
]
|
||
|
||
key_points = fused_symbol.get("key_points", [])
|
||
if key_points:
|
||
lines.append("")
|
||
lines.append("Key Points:")
|
||
for point in key_points[:5]:
|
||
lines.append(f" • {point}")
|
||
|
||
constraints = fused_symbol.get("constraints", [])
|
||
if constraints:
|
||
lines.append("")
|
||
lines.append("Constraints:")
|
||
for constraint in constraints[:5]:
|
||
lines.append(f" • {constraint}")
|
||
|
||
questions = fused_symbol.get("open_questions", [])
|
||
if questions:
|
||
lines.append("")
|
||
lines.append("Open Questions:")
|
||
for question in questions[:5]:
|
||
lines.append(f" ? {question}")
|
||
|
||
return "\n".join(lines)
|
||
|
||
|
||
def execute_with_lain(envelope: dict) -> dict:
|
||
"""Execute ExecutionEnvelope through LAIN cognition engine.
|
||
|
||
Real implementation: iterate through lanes, process each via lane processors,
|
||
fuse results, integrate glyph metadata, and return full ExecutionResult.
|
||
|
||
Contract:
|
||
- Does not mutate input envelope
|
||
- Deterministic for a given envelope
|
||
- Errors are returned in result['diagnostics']['errors'], not raised
|
||
|
||
Args:
|
||
envelope: ExecutionEnvelope dict from build_envelope()
|
||
|
||
Returns:
|
||
ExecutionResult dict with cognition_trace, fused_symbol, output_text, diagnostics
|
||
"""
|
||
from .lane_processors import process_lane
|
||
from .lain_glyph_bridge import (
|
||
load_glyph_context,
|
||
inject_glyph_metadata_into_lane,
|
||
compute_glyph_resonance,
|
||
augment_fused_symbol_with_glyphs,
|
||
)
|
||
|
||
start_time = time.time()
|
||
|
||
manifest = envelope.get("manifest", {})
|
||
lanes = envelope.get("lanes", {})
|
||
payload = envelope.get("payload", b"")
|
||
context = envelope.get("context", {})
|
||
|
||
# Initialize tracking
|
||
lane_timings: Dict[int, float] = {}
|
||
lane_results: Dict[int, dict] = {}
|
||
errors: List[dict] = []
|
||
cognition_trace = []
|
||
|
||
# Step 0: Load envelope
|
||
cognition_trace.append({
|
||
"step": 0,
|
||
"lane": -1,
|
||
"segment_id": None,
|
||
"operation": "load_envelope",
|
||
"input": {
|
||
"lanes": sorted(lanes.keys()),
|
||
"num_segments": sum(len(segs) for segs in lanes.values()),
|
||
"manifest_version": manifest.get("version"),
|
||
},
|
||
"output": {},
|
||
"note": "Loaded ExecutionEnvelope into LAIN cognition engine.",
|
||
})
|
||
|
||
# Step 1: Load glyph context
|
||
glyph_context = load_glyph_context(manifest, context)
|
||
cognition_trace.append({
|
||
"step": 1,
|
||
"lane": -1,
|
||
"segment_id": None,
|
||
"operation": "glyph_context_loaded",
|
||
"input": {"glyph_found": glyph_context.get("found", False)},
|
||
"output": {
|
||
"glyph_id": glyph_context.get("id"),
|
||
"glyph_name": glyph_context.get("name"),
|
||
"glyph_score": glyph_context.get("score"),
|
||
},
|
||
"note": f"Loaded glyph context: {glyph_context.get('name')}",
|
||
})
|
||
|
||
# Process each lane
|
||
step_num = 2
|
||
for lane_id in sorted(lanes.keys()):
|
||
lane_start = time.time()
|
||
lane_segments = lanes.get(lane_id, [])
|
||
|
||
try:
|
||
# Call lane processor
|
||
lane_result = process_lane(
|
||
lane_id,
|
||
lane_segments,
|
||
context,
|
||
manifest,
|
||
)
|
||
|
||
# Inject glyph metadata into lane result
|
||
lane_result = inject_glyph_metadata_into_lane(lane_result, glyph_context)
|
||
|
||
lane_results[lane_id] = lane_result
|
||
|
||
# Record timing
|
||
lane_elapsed = time.time() - lane_start
|
||
lane_timings[lane_id] = lane_elapsed
|
||
|
||
# Trace entry
|
||
cognition_trace.append({
|
||
"step": step_num,
|
||
"lane": lane_id,
|
||
"segment_id": None,
|
||
"operation": f"process_lane_{lane_id}",
|
||
"input": {"segments": len(lane_segments)},
|
||
"output": {
|
||
"summary_length": len(lane_result.get("summary", "")),
|
||
"key_points": len(lane_result.get("key_points", [])),
|
||
},
|
||
"note": f"Processed lane {lane_id} with {len(lane_segments)} segments in {lane_elapsed:.4f}s.",
|
||
})
|
||
|
||
except Exception as e:
|
||
lane_elapsed = time.time() - lane_start
|
||
lane_timings[lane_id] = lane_elapsed
|
||
|
||
err = make_error(
|
||
"LaneProcessorError",
|
||
f"Lane {lane_id} processing failed: {e}",
|
||
lane=lane_id,
|
||
recoverable=True,
|
||
)
|
||
errors.append(err)
|
||
|
||
lane_results[lane_id] = {
|
||
"summary": f"Error processing lane {lane_id}",
|
||
"key_points": [],
|
||
"constraints": [],
|
||
"open_questions": [],
|
||
}
|
||
|
||
cognition_trace.append({
|
||
"step": step_num,
|
||
"lane": lane_id,
|
||
"segment_id": None,
|
||
"operation": f"process_lane_{lane_id}",
|
||
"input": {"segments": len(lane_segments)},
|
||
"output": {"error": str(e)},
|
||
"note": f"Lane {lane_id} processing failed (recoverable).",
|
||
})
|
||
|
||
step_num += 1
|
||
|
||
# Fuse lane results
|
||
fused_symbol = fuse_lanes(lane_results)
|
||
|
||
# Augment fused symbol with glyph metadata
|
||
fused_symbol = augment_fused_symbol_with_glyphs(fused_symbol, glyph_context)
|
||
|
||
# Compute lane resonance
|
||
resonance = compute_resonance(lane_results, context)
|
||
|
||
# Compute glyph resonance
|
||
glyph_resonance = compute_glyph_resonance(glyph_context)
|
||
|
||
# Render output text
|
||
output_text = render_output_text(fused_symbol, context)
|
||
|
||
elapsed = time.time() - start_time
|
||
|
||
# Build diagnostics
|
||
diagnostics = {
|
||
"lane_timings": lane_timings,
|
||
"errors": errors,
|
||
"resonance": resonance,
|
||
"glyph_resonance": glyph_resonance,
|
||
"interface_version": INTERFACE_VERSION,
|
||
"elapsed": elapsed,
|
||
}
|
||
|
||
# Return ExecutionResult
|
||
result = {
|
||
"cognition_trace": cognition_trace,
|
||
"fused_symbol": fused_symbol,
|
||
"output_text": output_text,
|
||
"diagnostics": diagnostics,
|
||
}
|
||
|
||
return result
|
||
|
||
|
||
def execute_gx_path(
|
||
gx_path: str,
|
||
context: Optional[dict] = None,
|
||
) -> dict:
|
||
"""Main entry point: load .gx file and execute through LAIN.
|
||
|
||
Pipeline: load → normalize → map_lanes → build_envelope → execute
|
||
|
||
Args:
|
||
gx_path: Path to .gx file
|
||
context: Optional context overrides
|
||
|
||
Returns:
|
||
ExecutionResult dict
|
||
|
||
Raises:
|
||
Exceptions from load_gx() if file is invalid or missing
|
||
"""
|
||
# Load
|
||
manifest, raw_segments, payload = load_gx(gx_path)
|
||
|
||
# Normalize
|
||
segments = normalize_segments(manifest, raw_segments, payload)
|
||
|
||
# Map to lanes
|
||
lanes = map_lanes(segments)
|
||
|
||
# Build envelope
|
||
envelope = build_envelope(manifest, lanes, payload, context)
|
||
|
||
# Execute
|
||
result = execute_with_lain(envelope)
|
||
|
||
return result
|
||
|
||
|
||
def make_error(
|
||
error_type: str,
|
||
message: str,
|
||
segment_id: Optional[str] = None,
|
||
lane: Optional[int] = None,
|
||
recoverable: bool = True,
|
||
) -> dict:
|
||
"""Construct a structured GXRuntimeError dict.
|
||
|
||
Args:
|
||
error_type: Error class (e.g. "DecodeError", "LaneError", "SegmentError")
|
||
message: Human-readable message
|
||
segment_id: Optional segment id if segment-specific
|
||
lane: Optional lane id if lane-specific
|
||
recoverable: Whether error is recoverable
|
||
|
||
Returns:
|
||
GXRuntimeError dict
|
||
"""
|
||
return {
|
||
"type": error_type,
|
||
"message": message,
|
||
"segment_id": segment_id,
|
||
"lane": lane,
|
||
"recoverable": recoverable,
|
||
}
|