Add LLMCompress subsystem - sandbox for symbolic compression of LLM behavior
New subsystem fully self-contained: Components: - LLMCompress/llm_adapter.py: LLMAdapter + LLMResponse (abstract over LLM backends) - LLMCompress/compression_report.py: CompressionReport (symbolic analysis results) - LLMCompress/llm_compressor.py: compress_interaction() and compress_session() - LLMCompress/tests/test_llm_compress.py: 5 comprehensive tests Integration: - Uses GlyphOS Cognitive Kernel for symbolic analysis - Integrates with GlyphOS Event System - Emits cognition.started and cognition.completed events - Supports in-memory GX execution via execute_symbolic() Test Coverage: - LLMCompress tests: 5/5 PASS - All existing tests still pass (52/52) - Total: 57 tests passing Bug fixes in cognitive_kernel.py: - Fixed execute_symbolic() method calls to use correct function signatures - normalize_segments(manifest, segments, payload) - map_lanes(segments) - build_envelope(manifest, lanes, payload, context) - execute_with_lain(envelope) Constraints: - No modifications to gx_compiler/* - No modifications to glyphs/super_registry.py - Self-contained subsystem with proper isolation - Full backward compatibility maintained
This commit is contained in:
@@ -0,0 +1,133 @@
|
||||
"""Tests for LLMCompress subsystem.
|
||||
|
||||
These tests verify:
|
||||
|
||||
- LLMAdapter normalization
|
||||
- compress_interaction() end-to-end flow
|
||||
- compress_session() multi-turn flow
|
||||
- Integration with GlyphOS Cognitive Kernel
|
||||
- Event emission during compression
|
||||
|
||||
This uses a fake LLM backend so tests run without external dependencies.
|
||||
"""
|
||||
|
||||
from typing import Any, Dict
|
||||
|
||||
from LLMCompress import (
|
||||
LLMAdapter,
|
||||
compress_interaction,
|
||||
compress_session,
|
||||
CompressionReport,
|
||||
)
|
||||
|
||||
from glyphos.events import get_event_bus
|
||||
from glyphos.cognitive_kernel import get_kernel # noqa: F401 # imported to ensure availability
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fake LLM backend
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def fake_llm_backend(prompt: str, **kwargs: Any) -> Dict[str, Any]:
|
||||
"""A deterministic fake LLM backend for testing."""
|
||||
return {
|
||||
"response": f"FAKE_RESPONSE({prompt})",
|
||||
"tokens_prompt": len(prompt.split()),
|
||||
"tokens_response": 3,
|
||||
"model_name": "fake-llm-test",
|
||||
"extra_info": "ok",
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_adapter_normalization():
|
||||
adapter = LLMAdapter(fake_llm_backend, model_name="fake-llm-test")
|
||||
out = adapter.run("hello world")
|
||||
|
||||
assert out.prompt == "hello world"
|
||||
assert out.response.startswith("FAKE_RESPONSE")
|
||||
assert out.tokens_prompt == 2
|
||||
assert out.tokens_response == 3
|
||||
assert out.model_name == "fake-llm-test"
|
||||
assert isinstance(out.metadata, dict)
|
||||
|
||||
|
||||
def test_compress_interaction_basic():
|
||||
adapter = LLMAdapter(fake_llm_backend, model_name="fake-llm-test")
|
||||
|
||||
bus = get_event_bus()
|
||||
bus.clear_history()
|
||||
|
||||
report = compress_interaction(adapter, "hello test")
|
||||
|
||||
# Report structure
|
||||
assert isinstance(report, CompressionReport)
|
||||
assert len(report.interactions) == 1
|
||||
assert "prompt" in report.interactions[0]
|
||||
assert "response" in report.interactions[0]
|
||||
|
||||
# Kernel output
|
||||
assert isinstance(report.fused_symbol, dict)
|
||||
assert isinstance(report.diagnostics, dict)
|
||||
|
||||
# Events fired
|
||||
history = bus.get_history(limit=10)
|
||||
types = [e["type"] for e in history]
|
||||
|
||||
assert "cognition.started" in types
|
||||
assert "cognition.completed" in types
|
||||
|
||||
|
||||
def test_compress_session_multi_turn():
|
||||
adapter = LLMAdapter(fake_llm_backend, model_name="fake-llm-test")
|
||||
|
||||
bus = get_event_bus()
|
||||
bus.clear_history()
|
||||
|
||||
prompts = ["turn one", "turn two", "turn three"]
|
||||
report = compress_session(adapter, prompts)
|
||||
|
||||
assert isinstance(report, CompressionReport)
|
||||
assert len(report.interactions) == 3
|
||||
|
||||
# Kernel output
|
||||
assert isinstance(report.fused_symbol, dict)
|
||||
assert isinstance(report.diagnostics, dict)
|
||||
|
||||
# Events fired
|
||||
history = bus.get_history(limit=10)
|
||||
types = [e["type"] for e in history]
|
||||
|
||||
assert "cognition.started" in types
|
||||
assert "cognition.completed" in types
|
||||
|
||||
|
||||
def test_payload_encoding_is_valid_json():
|
||||
adapter = LLMAdapter(fake_llm_backend)
|
||||
report = compress_interaction(adapter, "encode me")
|
||||
|
||||
# Ensure payload was JSON-serializable and processed by kernel
|
||||
assert isinstance(report.fused_symbol, dict)
|
||||
assert isinstance(report.diagnostics, dict)
|
||||
|
||||
|
||||
def test_event_metadata_includes_source():
|
||||
adapter = LLMAdapter(fake_llm_backend)
|
||||
|
||||
bus = get_event_bus()
|
||||
bus.clear_history()
|
||||
|
||||
compress_interaction(adapter, "metadata test")
|
||||
|
||||
events = bus.get_history(limit=10)
|
||||
found = False
|
||||
for e in events:
|
||||
if e["type"] == "cognition.started":
|
||||
assert e["payload"]["source"] == "LLMCompress"
|
||||
found = True
|
||||
break
|
||||
|
||||
assert found, "Expected cognition.started event with source=LLMCompress"
|
||||
Reference in New Issue
Block a user