2125_GCE/LLMCompress/llm_compressor.py

"""LLM symbolic compressor.

Feeds LLM interactions through the GlyphOS Cognitive Kernel and produces
a symbolic CompressionReport.
"""

from __future__ import annotations

import json
from typing import Any, Dict, List, Optional

from glyphos.cognitive_kernel import get_kernel
from glyphos.events import emit

from .llm_adapter import LLMAdapter, LLMResponse
from .compression_report import CompressionReport


def _interaction_to_payload(interaction: LLMResponse) -> bytes:
    """Serialize a single interaction into a payload for symbolic analysis."""
    data = interaction.to_dict()
    return json.dumps(data, ensure_ascii=False, sort_keys=True).encode("utf-8")


def compress_interaction(
    adapter: LLMAdapter,
    prompt: str,
    *,
    mode: str = "analyze",
    context: Optional[Dict[str, Any]] = None,
    **llm_kwargs: Any,
) -> CompressionReport:
    """Run a single LLM interaction through the symbolic stack."""
    interaction = adapter.run(prompt, **llm_kwargs)

    emit(
        "cognition.started",
        {
            "source": "LLMCompress",
            "mode": mode,
            "prompt_preview": prompt[:120],
        },
    )

    manifest: Dict[str, Any] = {
        "type": "llm_interaction",
        "source": "LLMCompress",
        "model_name": interaction.model_name,
    }
    segments: List[Dict[str, Any]] = []
    payload: bytes = _interaction_to_payload(interaction)

    kernel = get_kernel()
    exec_context: Dict[str, Any] = context.copy() if context else {}
    exec_context.setdefault("source", "LLMCompress")
    exec_context.setdefault("interaction_type", "single")

    result = kernel.execute_symbolic(
        manifest=manifest,
        segments=segments,
        payload=payload,
        mode=mode,
        context=exec_context,
    )

    fused_symbol = result.get("fused_symbol", {})
    diagnostics = result.get("diagnostics", {})
    cognition_trace = result.get("cognition_trace", [])

    glyph_res = diagnostics.get("glyph_resonance") or {}
    glyph_ids: List[str] = []
    if isinstance(glyph_res, dict):
        gid = glyph_res.get("glyph_id")
        if isinstance(gid, str):
            glyph_ids.append(gid)

    report = CompressionReport(
        interactions=[interaction.to_dict()],
        fused_symbol=fused_symbol,
        diagnostics=diagnostics,
        cognition_trace=cognition_trace,
        glyph_ids=glyph_ids,
        glyph_resonance=glyph_res or None,
        tags=["llm_compress", mode],
        metadata={"model_name": interaction.model_name},
    )

    emit(
        "cognition.completed",
        {
            "source": "LLMCompress",
            "mode": mode,
            "model_name": interaction.model_name,
            "glyph_resonance": glyph_res or None,
            "summary": fused_symbol.get("summary"),
        },
    )

    return report


def compress_session(
    adapter: LLMAdapter,
    prompts: List[str],
    *,
    mode: str = "analyze",
    context: Optional[Dict[str, Any]] = None,
    **llm_kwargs: Any,
) -> CompressionReport:
    """Compress a multi-turn LLM session into a single symbolic report."""
    interactions = [adapter.run(p, **llm_kwargs) for p in prompts]

    session_data = [i.to_dict() for i in interactions]
    payload = json.dumps(
        {"session": session_data},
        ensure_ascii=False,
        sort_keys=True,
    ).encode("utf-8")

    manifest: Dict[str, Any] = {
        "type": "llm_session",
        "source": "LLMCompress",
        "turns": len(interactions),
        "model_name": interactions[0].model_name if interactions else None,
    }
    segments: List[Dict[str, Any]] = []

    kernel = get_kernel()
    exec_context: Dict[str, Any] = context.copy() if context else {}
    exec_context.setdefault("source", "LLMCompress")
    exec_context.setdefault("interaction_type", "session")

    emit(
        "cognition.started",
        {
            "source": "LLMCompress",
            "mode": mode,
            "turns": len(interactions),
        },
    )

    result = kernel.execute_symbolic(
        manifest=manifest,
        segments=segments,
        payload=payload,
        mode=mode,
        context=exec_context,
    )

    fused_symbol = result.get("fused_symbol", {})
    diagnostics = result.get("diagnostics", {})
    cognition_trace = result.get("cognition_trace", [])

    glyph_res = diagnostics.get("glyph_resonance") or {}
    glyph_ids: List[str] = []
    if isinstance(glyph_res, dict):
        gid = glyph_res.get("glyph_id")
        if isinstance(gid, str):
            glyph_ids.append(gid)

    report = CompressionReport(
        interactions=session_data,
        fused_symbol=fused_symbol,
        diagnostics=diagnostics,
        cognition_trace=cognition_trace,
        glyph_ids=glyph_ids,
        glyph_resonance=glyph_res or None,
        tags=["llm_compress", "session", mode],
        metadata={
            "model_name": manifest.get("model_name"),
            "turns": len(interactions),
        },
    )

    emit(
        "cognition.completed",
        {
            "source": "LLMCompress",
            "mode": mode,
            "turns": len(interactions),
            "glyph_resonance": glyph_res or None,
            "summary": fused_symbol.get("summary"),
        },
    )

    return report