2125_GCE/glyphos/symbolic_pipeline.py

"""Symbolic Pipeline Abstraction for XIC.

Provides a structured, glyph-aware pipeline for symbolic cognition execution.
Routes prompts through the LAIN 8-lane cognition kernel with explicit step tracking
and comprehensive glyph resonance metrics.
"""

from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional


@dataclass
class GlyphResonanceMetrics:
    """Glyph resonance metrics from LAIN cognition layer."""
    weight: float
    lineage_score: float
    contributor_score: float
    frequency_score: float
    grammar_score: float


@dataclass
class GlyphResonanceMap:
    """Maps glyph IDs to their resonance metrics."""
    resonances: Dict[str, GlyphResonanceMetrics] = field(default_factory=dict)
    global_resonance_score: float = 0.0

    def get_glyph_resonance(self, glyph_id: str) -> Optional[GlyphResonanceMetrics]:
        """Get resonance metrics for a specific glyph."""
        return self.resonances.get(glyph_id)

    def get_top_glyphs(self, n: int = 5) -> List[tuple[str, GlyphResonanceMetrics]]:
        """Get top N glyphs by weight."""
        sorted_glyphs = sorted(
            self.resonances.items(),
            key=lambda x: x[1].weight,
            reverse=True
        )
        return sorted_glyphs[:n]

    def get_average_resonance(self) -> float:
        """Get average resonance across all glyphs."""
        if not self.resonances:
            return 0.0
        total = sum(m.weight for m in self.resonances.values())
        return total / len(self.resonances)


@dataclass
class FusedSymbol:
    """Fused symbolic representation from LAIN cognition."""
    summary: str
    glyph_ids: List[str] = field(default_factory=list)
    resonance_map: GlyphResonanceMap = field(default_factory=GlyphResonanceMap)

    @classmethod
    def from_lain_result(cls, lain_fused_symbol: Dict[str, Any]) -> "FusedSymbol":
        """Parse fused_symbol dict from LAIN result."""
        summary = lain_fused_symbol.get("summary", "")
        glyph_ids = lain_fused_symbol.get("glyph_ids", [])

        resonance_map = GlyphResonanceMap(
            global_resonance_score=lain_fused_symbol.get("global_resonance_score", 0.0)
        )

        raw_resonance = lain_fused_symbol.get("resonance_map", {})
        for glyph_id, metrics_dict in raw_resonance.items():
            if isinstance(metrics_dict, dict):
                resonance_map.resonances[glyph_id] = GlyphResonanceMetrics(
                    weight=metrics_dict.get("weight", 0.0),
                    lineage_score=metrics_dict.get("lineage_score", 0.0),
                    contributor_score=metrics_dict.get("contributor_score", 0.0),
                    frequency_score=metrics_dict.get("frequency_score", 0.0),
                    grammar_score=metrics_dict.get("grammar_score", 0.0),
                )

        return cls(summary=summary, glyph_ids=glyph_ids, resonance_map=resonance_map)


@dataclass
class SymbolicStep:
    """A single step in the symbolic pipeline execution."""
    name: str
    kind: str  # "prompt", "glyph_call", "fused_symbol"
    payload: Any
    context: Dict[str, Any] = field(default_factory=dict)


@dataclass
class SymbolicPipelineResult:
    """Result of a symbolic pipeline execution."""
    steps: List[SymbolicStep]
    output_text: str
    fused_symbol: Optional[FusedSymbol] = None


def extract_glyph_resonances(
    pipeline_result: "SymbolicPipelineResult",
) -> Dict[str, Dict[str, Any]]:
    """Extract glyph resonance metrics from a pipeline result.

    Returns dict mapping glyph_id → resonance metrics dict.
    """
    if not pipeline_result.fused_symbol:
        return {}

    result = {}
    for glyph_id, metrics in pipeline_result.fused_symbol.resonance_map.resonances.items():
        result[glyph_id] = {
            "weight": metrics.weight,
            "lineage_score": metrics.lineage_score,
            "contributor_score": metrics.contributor_score,
            "frequency_score": metrics.frequency_score,
            "grammar_score": metrics.grammar_score,
        }

    return result


def get_dominant_glyphs(
    pipeline_result: "SymbolicPipelineResult",
    n: int = 3,
) -> List[tuple[str, float]]:
    """Get top N glyphs by resonance weight from a pipeline result.

    Returns list of (glyph_id, weight) tuples sorted by weight descending.
    """
    if not pipeline_result.fused_symbol:
        return []

    return [
        (glyph_id, metrics.weight)
        for glyph_id, metrics in pipeline_result.fused_symbol.resonance_map.get_top_glyphs(n)
    ]


def format_glyph_resonance_report(
    pipeline_result: "SymbolicPipelineResult",
) -> str:
    """Format a human-readable glyph resonance report."""
    if not pipeline_result.fused_symbol:
        return "No glyph resonance data."

    resonance = pipeline_result.fused_symbol.resonance_map
    lines = [
        f"Global Resonance Score: {resonance.global_resonance_score:.3f}",
        f"Glyphs Engaged: {len(resonance.resonances)}",
        "",
        "Top Glyphs by Weight:",
    ]

    for glyph_id, metrics in resonance.get_top_glyphs(5):
        lines.append(
            f"  {glyph_id}: weight={metrics.weight:.3f}, "
            f"lineage={metrics.lineage_score:.3f}, "
            f"contributor={metrics.contributor_score:.3f}"
        )

    return "\n".join(lines)


def run_symbolic_pipeline(
    prompt: str,
    context: Optional[Dict[str, Any]] = None,
    glyph_id: Optional[str] = None,
) -> SymbolicPipelineResult:
    """
    High-level symbolic pipeline entrypoint for XIC.

    Accepts a prompt and optional symbolic/glyph context, routes through
    the LAIN 8-lane cognition kernel via CognitiveKernel.execute_symbolic(),
    and returns a structured SymbolicPipelineResult with execution steps,
    final output text, and fused symbolic representation.

    Args:
        prompt: User or system prompt text.
        context: Optional dict of symbolic/cognitive context metadata.
        glyph_id: Optional glyph identifier for glyph-aware cognition.

    Returns:
        SymbolicPipelineResult with:
        - steps: List of SymbolicStep objects tracking execution flow.
        - output_text: Final text result from cognition layer.
        - fused_symbol: Fused symbolic representation (if produced by LAIN).
    """
    from gx_compiler.compressor import GXCompressor
    from .cognitive_kernel import get_kernel

    steps: List[SymbolicStep] = []
    kernel = get_kernel()
    prompt_bytes = prompt.encode("utf-8")

    # Step 1: Initial prompt
    steps.append(SymbolicStep(
        name="initial_prompt",
        kind="prompt",
        payload=prompt,
        context=dict(context or {})
    ))

    # Step 2: Prepare context for glyph-aware processing
    exec_context = dict(context or {})
    if glyph_id:
        exec_context["glyph_id"] = glyph_id
        steps.append(SymbolicStep(
            name=f"glyph:{glyph_id}",
            kind="glyph_call",
            payload=prompt,
            context=exec_context
        ))

    # Step 3: Compress prompt and build manifest
    try:
        payload = GXCompressor.compress(prompt)
    except Exception as e:
        return SymbolicPipelineResult(
            steps=steps,
            output_text=f"[Pipeline Error] Compression failed: {e}",
            fused_symbol=None
        )

    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)}]

    # Step 4: Execute through LAIN cognition pipeline
    result = kernel.execute_symbolic(
        manifest=manifest,
        segments=segments,
        payload=payload,
        mode="symbolic",
        context=exec_context,
    )

    # Step 5: Extract and parse results
    lain_fused_symbol = result.get("fused_symbol")
    fused_symbol = None

    if lain_fused_symbol:
        fused_symbol = FusedSymbol.from_lain_result(lain_fused_symbol)
        output_text = result.get("output_text") or fused_symbol.summary
    else:
        output_text = result.get("output_text") or prompt

    # Step 6: Record fusion step if fused_symbol present
    if fused_symbol:
        steps.append(SymbolicStep(
            name="fusion",
            kind="fused_symbol",
            payload={
                "summary": fused_symbol.summary,
                "glyph_ids": fused_symbol.glyph_ids,
                "global_resonance_score": fused_symbol.resonance_map.global_resonance_score,
            },
            context={}
        ))

    return SymbolicPipelineResult(
        steps=steps,
        output_text=output_text,
        fused_symbol=fused_symbol
    )