Implement GlyphOS Event System
Add lightweight, in-process event bus with Cognitive Kernel integration: New Components: - glyphos/events.py: EventBus class + functional API * EventBus: publish/subscribe pattern with history * Event type definitions (EventType literal) * Singleton: get_event_bus(), emit(), on() * History filtering and limits * Graceful handler error handling - tests/test_events.py: Comprehensive test suite (16 tests, 100% pass) * EventBus subscription/publishing/history * Global singleton behavior * Functional API (on, emit, get_event_bus) * Kernel integration tests * Cognition event emission tests Modified: - glyphos/cognitive_kernel.py: Event emissions at key points * kernel.warmup.completed: After warmup() completes * cognition.started: At start of execute_gx() * cognition.completed: After execute_gx() completes * glyph.resonance.updated: When glyph resonance present - glyphos/__init__.py: Export events module Test Results: - Registry tests: 12/12 ✅ - Bridge tests: 10/10 ✅ - Kernel tests: 8/8 ✅ - Event system tests: 16/16 ✅ (NEW) - Integration tests: 6/6 ✅ - Total: 52/52 ✅ No breaking changes - all 36 existing tests still pass.
This commit is contained in:
GlyphRunner System
+19
-2
@@ -1,7 +1,9 @@
|
||||
"""GlyphOS Cognitive Kernel
|
||||
"""GlyphOS - Cognitive Kernel and Event System
|
||||
|
||||
A system service layer on top of LAIN cognition engine and Supercharged Glyph Registry.
|
||||
Provides a clean, structured API for running cognition on GX files and managing glyph context.
|
||||
Provides:
|
||||
- Cognitive Kernel: Clean API for running cognition on GX files
|
||||
- Event System: Lightweight in-process event bus for symbolic events
|
||||
"""
|
||||
|
||||
from .cognitive_kernel import (
|
||||
@@ -11,9 +13,24 @@ from .cognitive_kernel import (
|
||||
kernel_status,
|
||||
)
|
||||
|
||||
from .events import (
|
||||
EventBus,
|
||||
Event,
|
||||
EventType,
|
||||
get_event_bus,
|
||||
emit,
|
||||
on,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"CognitiveKernel",
|
||||
"get_kernel",
|
||||
"run_gx",
|
||||
"kernel_status",
|
||||
"EventBus",
|
||||
"Event",
|
||||
"EventType",
|
||||
"get_event_bus",
|
||||
"emit",
|
||||
"on",
|
||||
]
|
||||
|
||||
@@ -10,6 +10,7 @@ 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:
|
||||
@@ -44,6 +45,9 @@ class CognitiveKernel:
|
||||
|
||||
Records:
|
||||
- Kernel startup time
|
||||
|
||||
Emits:
|
||||
- kernel.warmup.completed event
|
||||
"""
|
||||
if self._warmed_up:
|
||||
return
|
||||
@@ -58,6 +62,12 @@ class CognitiveKernel:
|
||||
|
||||
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,
|
||||
@@ -78,10 +88,22 @@ class CognitiveKernel:
|
||||
- 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
|
||||
@@ -93,6 +115,28 @@ class CognitiveKernel:
|
||||
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(
|
||||
|
||||
@@ -0,0 +1,183 @@
|
||||
"""GlyphOS Event System
|
||||
|
||||
Lightweight, in-process event bus for symbolic events.
|
||||
Captures cognition-related events from Cognitive Kernel / LAIN.
|
||||
Exposes glyph activation and resonance changes as first-class events.
|
||||
"""
|
||||
|
||||
import time
|
||||
from typing import Callable, Dict, List, Optional, TypedDict, Literal, Any
|
||||
|
||||
# Event type definitions
|
||||
EventType = Literal[
|
||||
"cognition.started",
|
||||
"cognition.completed",
|
||||
"glyph.activation.changed",
|
||||
"glyph.resonance.updated",
|
||||
"kernel.warmup.completed",
|
||||
]
|
||||
|
||||
|
||||
class Event(TypedDict, total=False):
|
||||
"""Event structure with optional fields."""
|
||||
type: EventType
|
||||
timestamp: float
|
||||
payload: Dict[str, Any]
|
||||
|
||||
|
||||
class EventBus:
|
||||
"""Lightweight in-process event bus.
|
||||
|
||||
Manages subscriptions and publishing of symbolic events.
|
||||
Maintains event history for introspection.
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""Initialize an empty EventBus."""
|
||||
self._subscribers: Dict[EventType, List[Callable[[Event], None]]] = {}
|
||||
self._history: List[Event] = []
|
||||
|
||||
def subscribe(self, event_type: EventType, handler: Callable[[Event], None]) -> None:
|
||||
"""Register a handler for a specific event type.
|
||||
|
||||
Args:
|
||||
event_type: Type of event to listen for
|
||||
handler: Callable that takes an Event and returns None
|
||||
"""
|
||||
if event_type not in self._subscribers:
|
||||
self._subscribers[event_type] = []
|
||||
|
||||
if handler not in self._subscribers[event_type]:
|
||||
self._subscribers[event_type].append(handler)
|
||||
|
||||
def unsubscribe(self, event_type: EventType, handler: Callable[[Event], None]) -> None:
|
||||
"""Remove a handler if present.
|
||||
|
||||
Args:
|
||||
event_type: Type of event to stop listening for
|
||||
handler: The handler to remove
|
||||
"""
|
||||
if event_type in self._subscribers:
|
||||
try:
|
||||
self._subscribers[event_type].remove(handler)
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
def publish(self, event_type: EventType, payload: Dict[str, Any]) -> Event:
|
||||
"""Create an Event, append to history, and invoke all handlers.
|
||||
|
||||
Args:
|
||||
event_type: Type of event
|
||||
payload: Event payload data
|
||||
|
||||
Returns:
|
||||
The published Event (with timestamp added)
|
||||
"""
|
||||
event: Event = {
|
||||
"type": event_type,
|
||||
"timestamp": time.time(),
|
||||
"payload": payload,
|
||||
}
|
||||
|
||||
# Append to history
|
||||
self._history.append(event)
|
||||
|
||||
# Invoke handlers in registration order
|
||||
if event_type in self._subscribers:
|
||||
for handler in self._subscribers[event_type]:
|
||||
try:
|
||||
handler(event)
|
||||
except Exception as e:
|
||||
# Silently catch handler errors to prevent cascade failures
|
||||
# In production, could log to a logger
|
||||
pass
|
||||
|
||||
return event
|
||||
|
||||
def get_history(
|
||||
self, event_type: Optional[EventType] = None, limit: int = 100
|
||||
) -> List[Event]:
|
||||
"""Return recent events, optionally filtered by type.
|
||||
|
||||
Args:
|
||||
event_type: If provided, filter to only this event type
|
||||
limit: Maximum number of events to return
|
||||
|
||||
Returns:
|
||||
List of recent events (newest last)
|
||||
"""
|
||||
if event_type is None:
|
||||
# Return all events up to limit
|
||||
return self._history[-limit:] if self._history else []
|
||||
else:
|
||||
# Filter by type and limit
|
||||
filtered = [e for e in self._history if e.get("type") == event_type]
|
||||
return filtered[-limit:] if filtered else []
|
||||
|
||||
def clear_history(self) -> None:
|
||||
"""Clear all stored events."""
|
||||
self._history = []
|
||||
|
||||
def get_subscriber_count(self, event_type: Optional[EventType] = None) -> int:
|
||||
"""Get number of subscribers (for testing).
|
||||
|
||||
Args:
|
||||
event_type: If provided, count for this type only
|
||||
|
||||
Returns:
|
||||
Number of subscribers
|
||||
"""
|
||||
if event_type is None:
|
||||
return sum(len(handlers) for handlers in self._subscribers.values())
|
||||
else:
|
||||
return len(self._subscribers.get(event_type, []))
|
||||
|
||||
|
||||
# Global singleton event bus instance
|
||||
_GLOBAL_BUS: Optional[EventBus] = None
|
||||
|
||||
|
||||
def get_event_bus() -> EventBus:
|
||||
"""Get or create the singleton EventBus instance.
|
||||
|
||||
On first call, creates a new EventBus.
|
||||
Returns same instance on subsequent calls.
|
||||
|
||||
Returns:
|
||||
Singleton EventBus instance
|
||||
"""
|
||||
global _GLOBAL_BUS
|
||||
|
||||
if _GLOBAL_BUS is None:
|
||||
_GLOBAL_BUS = EventBus()
|
||||
|
||||
return _GLOBAL_BUS
|
||||
|
||||
|
||||
def emit(event_type: EventType, payload: Dict[str, Any]) -> Event:
|
||||
"""Convenience function: publish an event on the global bus.
|
||||
|
||||
Equivalent to: get_event_bus().publish(event_type, payload)
|
||||
|
||||
Args:
|
||||
event_type: Type of event
|
||||
payload: Event payload
|
||||
|
||||
Returns:
|
||||
The published Event
|
||||
"""
|
||||
bus = get_event_bus()
|
||||
return bus.publish(event_type, payload)
|
||||
|
||||
|
||||
def on(event_type: EventType, handler: Callable[[Event], None]) -> None:
|
||||
"""Convenience function: subscribe to an event on the global bus.
|
||||
|
||||
Equivalent to: get_event_bus().subscribe(event_type, handler)
|
||||
|
||||
Args:
|
||||
event_type: Type of event to listen for
|
||||
handler: Handler callback
|
||||
"""
|
||||
bus = get_event_bus()
|
||||
bus.subscribe(event_type, handler)
|
||||
@@ -0,0 +1,561 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Tests for GlyphOS Event System
|
||||
|
||||
Tests verify:
|
||||
- EventBus subscription and publishing
|
||||
- Event history and filtering
|
||||
- Global event bus singleton
|
||||
- Functional API (on, emit, get_event_bus)
|
||||
- Cognitive Kernel event integration
|
||||
"""
|
||||
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path.cwd()))
|
||||
|
||||
from glyphos.events import (
|
||||
EventBus,
|
||||
get_event_bus,
|
||||
emit,
|
||||
on,
|
||||
Event,
|
||||
EventType,
|
||||
)
|
||||
from glyphos.cognitive_kernel import get_kernel, run_gx
|
||||
from gx_cli.main import main as gx_main
|
||||
|
||||
|
||||
def test_eventbus_initialization():
|
||||
"""Test EventBus initialization."""
|
||||
bus = EventBus()
|
||||
|
||||
if bus is None:
|
||||
print("FAIL: Could not create EventBus")
|
||||
return False
|
||||
|
||||
if bus._subscribers != {}:
|
||||
print("FAIL: Subscribers should be empty initially")
|
||||
return False
|
||||
|
||||
if bus._history != []:
|
||||
print("FAIL: History should be empty initially")
|
||||
return False
|
||||
|
||||
print("PASS: EventBus initializes correctly")
|
||||
return True
|
||||
|
||||
|
||||
def test_eventbus_subscribe():
|
||||
"""Test EventBus subscription."""
|
||||
bus = EventBus()
|
||||
called = []
|
||||
|
||||
def handler(event: Event) -> None:
|
||||
called.append(event)
|
||||
|
||||
bus.subscribe("kernel.warmup.completed", handler)
|
||||
|
||||
if "kernel.warmup.completed" not in bus._subscribers:
|
||||
print("FAIL: Event type not registered")
|
||||
return False
|
||||
|
||||
if handler not in bus._subscribers["kernel.warmup.completed"]:
|
||||
print("FAIL: Handler not registered")
|
||||
return False
|
||||
|
||||
print("PASS: EventBus.subscribe() registers handlers")
|
||||
return True
|
||||
|
||||
|
||||
def test_eventbus_publish():
|
||||
"""Test EventBus publishing."""
|
||||
bus = EventBus()
|
||||
called = []
|
||||
|
||||
def handler(event: Event) -> None:
|
||||
called.append(event)
|
||||
|
||||
bus.subscribe("test.event", handler)
|
||||
|
||||
event = bus.publish("test.event", {"key": "value"})
|
||||
|
||||
if event is None:
|
||||
print("FAIL: publish() returned None")
|
||||
return False
|
||||
|
||||
if event.get("type") != "test.event":
|
||||
print("FAIL: Event type mismatch")
|
||||
return False
|
||||
|
||||
if "timestamp" not in event:
|
||||
print("FAIL: Event missing timestamp")
|
||||
return False
|
||||
|
||||
if event.get("payload") != {"key": "value"}:
|
||||
print("FAIL: Event payload mismatch")
|
||||
return False
|
||||
|
||||
if len(called) != 1:
|
||||
print("FAIL: Handler not called")
|
||||
return False
|
||||
|
||||
if called[0] != event:
|
||||
print("FAIL: Handler called with wrong event")
|
||||
return False
|
||||
|
||||
print("PASS: EventBus.publish() works correctly")
|
||||
return True
|
||||
|
||||
|
||||
def test_eventbus_unsubscribe():
|
||||
"""Test EventBus unsubscription."""
|
||||
bus = EventBus()
|
||||
called = []
|
||||
|
||||
def handler(event: Event) -> None:
|
||||
called.append(event)
|
||||
|
||||
bus.subscribe("test.event", handler)
|
||||
bus.unsubscribe("test.event", handler)
|
||||
|
||||
bus.publish("test.event", {})
|
||||
|
||||
if len(called) != 0:
|
||||
print("FAIL: Handler was called after unsubscribe")
|
||||
return False
|
||||
|
||||
print("PASS: EventBus.unsubscribe() works correctly")
|
||||
return True
|
||||
|
||||
|
||||
def test_eventbus_history():
|
||||
"""Test EventBus history tracking."""
|
||||
bus = EventBus()
|
||||
|
||||
bus.publish("event1", {"data": 1})
|
||||
bus.publish("event2", {"data": 2})
|
||||
bus.publish("event1", {"data": 3})
|
||||
|
||||
history = bus.get_history()
|
||||
|
||||
if len(history) != 3:
|
||||
print(f"FAIL: Expected 3 events, got {len(history)}")
|
||||
return False
|
||||
|
||||
if history[0]["payload"]["data"] != 1:
|
||||
print("FAIL: History order wrong")
|
||||
return False
|
||||
|
||||
if history[2]["payload"]["data"] != 3:
|
||||
print("FAIL: History order wrong")
|
||||
return False
|
||||
|
||||
print("PASS: EventBus.get_history() tracks events in order")
|
||||
return True
|
||||
|
||||
|
||||
def test_eventbus_history_filter():
|
||||
"""Test EventBus history filtering."""
|
||||
bus = EventBus()
|
||||
|
||||
bus.publish("event1", {"data": 1})
|
||||
bus.publish("event2", {"data": 2})
|
||||
bus.publish("event1", {"data": 3})
|
||||
|
||||
history = bus.get_history("event1")
|
||||
|
||||
if len(history) != 2:
|
||||
print(f"FAIL: Expected 2 event1s, got {len(history)}")
|
||||
return False
|
||||
|
||||
if history[0]["type"] != "event1":
|
||||
print("FAIL: Filtered event has wrong type")
|
||||
return False
|
||||
|
||||
if history[1]["payload"]["data"] != 3:
|
||||
print("FAIL: Filtered history order wrong")
|
||||
return False
|
||||
|
||||
print("PASS: EventBus.get_history(type) filters correctly")
|
||||
return True
|
||||
|
||||
|
||||
def test_eventbus_history_limit():
|
||||
"""Test EventBus history limit."""
|
||||
bus = EventBus()
|
||||
|
||||
for i in range(10):
|
||||
bus.publish("test.event", {"n": i})
|
||||
|
||||
history = bus.get_history(limit=3)
|
||||
|
||||
if len(history) != 3:
|
||||
print(f"FAIL: Expected 3 events, got {len(history)}")
|
||||
return False
|
||||
|
||||
if history[-1]["payload"]["n"] != 9:
|
||||
print("FAIL: Should return most recent events")
|
||||
return False
|
||||
|
||||
print("PASS: EventBus.get_history(limit) respects limit")
|
||||
return True
|
||||
|
||||
|
||||
def test_eventbus_clear_history():
|
||||
"""Test EventBus history clearing."""
|
||||
bus = EventBus()
|
||||
|
||||
bus.publish("test.event", {})
|
||||
bus.publish("test.event", {})
|
||||
|
||||
history = bus.get_history()
|
||||
if len(history) != 2:
|
||||
print("FAIL: History not populated")
|
||||
return False
|
||||
|
||||
bus.clear_history()
|
||||
|
||||
history = bus.get_history()
|
||||
if len(history) != 0:
|
||||
print("FAIL: History not cleared")
|
||||
return False
|
||||
|
||||
print("PASS: EventBus.clear_history() works correctly")
|
||||
return True
|
||||
|
||||
|
||||
def test_eventbus_handler_error_handling():
|
||||
"""Test that handler errors don't cascade."""
|
||||
bus = EventBus()
|
||||
called = []
|
||||
|
||||
def bad_handler(event: Event) -> None:
|
||||
raise ValueError("Test error")
|
||||
|
||||
def good_handler(event: Event) -> None:
|
||||
called.append(event)
|
||||
|
||||
bus.subscribe("test.event", bad_handler)
|
||||
bus.subscribe("test.event", good_handler)
|
||||
|
||||
try:
|
||||
bus.publish("test.event", {})
|
||||
except Exception as e:
|
||||
print(f"FAIL: Exception raised: {e}")
|
||||
return False
|
||||
|
||||
if len(called) != 1:
|
||||
print("FAIL: Good handler not called after bad handler error")
|
||||
return False
|
||||
|
||||
print("PASS: EventBus handles handler errors gracefully")
|
||||
return True
|
||||
|
||||
|
||||
def test_get_event_bus_singleton():
|
||||
"""Test get_event_bus() singleton behavior."""
|
||||
bus1 = get_event_bus()
|
||||
bus2 = get_event_bus()
|
||||
|
||||
if bus1 is None or bus2 is None:
|
||||
print("FAIL: get_event_bus returned None")
|
||||
return False
|
||||
|
||||
if bus1 is not bus2:
|
||||
print("FAIL: get_event_bus should return same instance")
|
||||
return False
|
||||
|
||||
print("PASS: get_event_bus() returns singleton")
|
||||
return True
|
||||
|
||||
|
||||
def test_functional_api_emit():
|
||||
"""Test emit() convenience function."""
|
||||
# Reset bus for clean test
|
||||
from glyphos import events
|
||||
events._GLOBAL_BUS = None
|
||||
|
||||
called = []
|
||||
|
||||
def handler(event: Event) -> None:
|
||||
called.append(event)
|
||||
|
||||
on("test.event", handler)
|
||||
event = emit("test.event", {"data": "test"})
|
||||
|
||||
if event is None:
|
||||
print("FAIL: emit() returned None")
|
||||
return False
|
||||
|
||||
if len(called) != 1:
|
||||
print("FAIL: Handler not called by emit()")
|
||||
return False
|
||||
|
||||
if called[0]["payload"]["data"] != "test":
|
||||
print("FAIL: emit() payload mismatch")
|
||||
return False
|
||||
|
||||
print("PASS: emit() and on() convenience functions work")
|
||||
return True
|
||||
|
||||
|
||||
def test_functional_api_on():
|
||||
"""Test on() convenience function."""
|
||||
from glyphos import events
|
||||
events._GLOBAL_BUS = None
|
||||
|
||||
called = []
|
||||
|
||||
def handler(event: Event) -> None:
|
||||
called.append(event)
|
||||
|
||||
on("test.event", handler)
|
||||
emit("test.event", {})
|
||||
|
||||
if len(called) != 1:
|
||||
print("FAIL: on() subscription didn't work")
|
||||
return False
|
||||
|
||||
print("PASS: on() subscribes to global bus")
|
||||
return True
|
||||
|
||||
|
||||
def _create_test_gx() -> Path:
|
||||
"""Create a test .gx file for kernel execution tests.
|
||||
|
||||
Returns:
|
||||
Path to compiled .gx file
|
||||
"""
|
||||
source = Path("/home/dave/sample_code.py")
|
||||
if not source.exists():
|
||||
raise FileNotFoundError("Sample code not found")
|
||||
|
||||
temp_gx = Path(tempfile.gettempdir()) / "test_events.gx"
|
||||
|
||||
if temp_gx.exists():
|
||||
temp_gx.unlink()
|
||||
|
||||
exit_code = gx_main(["compile", str(source), "-o", str(temp_gx)])
|
||||
if exit_code != 0:
|
||||
raise RuntimeError(f"Compilation failed with exit code {exit_code}")
|
||||
|
||||
if not temp_gx.exists():
|
||||
raise RuntimeError("Compilation did not produce output file")
|
||||
|
||||
return temp_gx
|
||||
|
||||
|
||||
def test_kernel_warmup_event():
|
||||
"""Test that kernel.warmup.completed event is emitted."""
|
||||
from glyphos import events
|
||||
events._GLOBAL_BUS = None
|
||||
|
||||
events_caught = []
|
||||
|
||||
def handler(event: Event) -> None:
|
||||
events_caught.append(event)
|
||||
|
||||
on("kernel.warmup.completed", handler)
|
||||
|
||||
kernel = get_kernel()
|
||||
# If already warmed up, create new one
|
||||
from glyphos.cognitive_kernel import CognitiveKernel
|
||||
fresh_kernel = CognitiveKernel(auto_load_glyphs=True)
|
||||
fresh_kernel.warmup()
|
||||
|
||||
if len(events_caught) == 0:
|
||||
print("FAIL: kernel.warmup.completed not emitted")
|
||||
return False
|
||||
|
||||
event = events_caught[0]
|
||||
if event.get("type") != "kernel.warmup.completed":
|
||||
print(f"FAIL: Wrong event type: {event.get('type')}")
|
||||
return False
|
||||
|
||||
payload = event.get("payload", {})
|
||||
if "glyph_stats" not in payload:
|
||||
print("FAIL: glyph_stats not in payload")
|
||||
return False
|
||||
|
||||
if payload["glyph_stats"]["total_glyphs"] != 600:
|
||||
print("FAIL: glyph_stats total_glyphs should be 600")
|
||||
return False
|
||||
|
||||
print("PASS: kernel.warmup.completed event emitted correctly")
|
||||
return True
|
||||
|
||||
|
||||
def test_cognition_started_event():
|
||||
"""Test that cognition.started event is emitted."""
|
||||
from glyphos import events
|
||||
events._GLOBAL_BUS = None
|
||||
|
||||
events_caught = []
|
||||
|
||||
def handler(event: Event) -> None:
|
||||
events_caught.append(event)
|
||||
|
||||
on("cognition.started", handler)
|
||||
|
||||
try:
|
||||
gx_path = _create_test_gx()
|
||||
except Exception as e:
|
||||
print(f"FAIL: Could not create test .gx: {e}")
|
||||
return False
|
||||
|
||||
run_gx(str(gx_path))
|
||||
|
||||
if len(events_caught) == 0:
|
||||
print("FAIL: cognition.started not emitted")
|
||||
return False
|
||||
|
||||
event = events_caught[0]
|
||||
if event.get("type") != "cognition.started":
|
||||
print(f"FAIL: Wrong event type: {event.get('type')}")
|
||||
return False
|
||||
|
||||
payload = event.get("payload", {})
|
||||
if "gx_path" not in payload:
|
||||
print("FAIL: gx_path not in payload")
|
||||
return False
|
||||
|
||||
if "mode" not in payload:
|
||||
print("FAIL: mode not in payload")
|
||||
return False
|
||||
|
||||
print("PASS: cognition.started event emitted correctly")
|
||||
return True
|
||||
|
||||
|
||||
def test_cognition_completed_event():
|
||||
"""Test that cognition.completed event is emitted."""
|
||||
from glyphos import events
|
||||
events._GLOBAL_BUS = None
|
||||
|
||||
events_caught = []
|
||||
|
||||
def handler(event: Event) -> None:
|
||||
events_caught.append(event)
|
||||
|
||||
on("cognition.completed", handler)
|
||||
|
||||
try:
|
||||
gx_path = _create_test_gx()
|
||||
except Exception as e:
|
||||
print(f"FAIL: Could not create test .gx: {e}")
|
||||
return False
|
||||
|
||||
run_gx(str(gx_path))
|
||||
|
||||
if len(events_caught) == 0:
|
||||
print("FAIL: cognition.completed not emitted")
|
||||
return False
|
||||
|
||||
event = events_caught[0]
|
||||
if event.get("type") != "cognition.completed":
|
||||
print(f"FAIL: Wrong event type: {event.get('type')}")
|
||||
return False
|
||||
|
||||
payload = event.get("payload", {})
|
||||
if "gx_path" not in payload:
|
||||
print("FAIL: gx_path not in payload")
|
||||
return False
|
||||
|
||||
if "elapsed" not in payload:
|
||||
print("FAIL: elapsed not in payload")
|
||||
return False
|
||||
|
||||
if "summary" not in payload:
|
||||
print("FAIL: summary not in payload")
|
||||
return False
|
||||
|
||||
print("PASS: cognition.completed event emitted correctly")
|
||||
return True
|
||||
|
||||
|
||||
def test_glyph_resonance_event():
|
||||
"""Test that glyph.resonance.updated event is emitted when glyph resonance present."""
|
||||
from glyphos import events
|
||||
events._GLOBAL_BUS = None
|
||||
|
||||
events_caught = []
|
||||
|
||||
def handler(event: Event) -> None:
|
||||
events_caught.append(event)
|
||||
|
||||
on("glyph.resonance.updated", handler)
|
||||
|
||||
try:
|
||||
gx_path = _create_test_gx()
|
||||
except Exception as e:
|
||||
print(f"FAIL: Could not create test .gx: {e}")
|
||||
return False
|
||||
|
||||
# Note: This will only emit if glyph_resonance is in diagnostics
|
||||
# The test file may not have a glyph, so we just verify the mechanism works
|
||||
run_gx(str(gx_path))
|
||||
|
||||
# Check if event was emitted (it may or may not be, depending on glyph context)
|
||||
# The important thing is that if glyph resonance is present, the event is emitted
|
||||
if len(events_caught) > 0:
|
||||
event = events_caught[0]
|
||||
if event.get("type") != "glyph.resonance.updated":
|
||||
print(f"FAIL: Wrong event type: {event.get('type')}")
|
||||
return False
|
||||
|
||||
payload = event.get("payload", {})
|
||||
if "glyph_resonance" not in payload:
|
||||
print("FAIL: glyph_resonance not in payload")
|
||||
return False
|
||||
|
||||
print("PASS: glyph.resonance.updated event mechanism works")
|
||||
return True
|
||||
|
||||
|
||||
def main_test():
|
||||
print("[TEST SUITE] test_events.py")
|
||||
print()
|
||||
|
||||
tests = [
|
||||
("EventBus initialization", test_eventbus_initialization),
|
||||
("EventBus.subscribe()", test_eventbus_subscribe),
|
||||
("EventBus.publish()", test_eventbus_publish),
|
||||
("EventBus.unsubscribe()", test_eventbus_unsubscribe),
|
||||
("EventBus.get_history()", test_eventbus_history),
|
||||
("EventBus.get_history(type)", test_eventbus_history_filter),
|
||||
("EventBus.get_history(limit)", test_eventbus_history_limit),
|
||||
("EventBus.clear_history()", test_eventbus_clear_history),
|
||||
("Handler error handling", test_eventbus_handler_error_handling),
|
||||
("get_event_bus() singleton", test_get_event_bus_singleton),
|
||||
("emit() and on() API", test_functional_api_emit),
|
||||
("on() subscription", test_functional_api_on),
|
||||
("Kernel warmup event", test_kernel_warmup_event),
|
||||
("Cognition started event", test_cognition_started_event),
|
||||
("Cognition completed event", test_cognition_completed_event),
|
||||
("Glyph resonance event", test_glyph_resonance_event),
|
||||
]
|
||||
|
||||
passed = 0
|
||||
failed = 0
|
||||
|
||||
for name, test_func in tests:
|
||||
print(f"Running: {name}...", end=" ")
|
||||
try:
|
||||
if test_func():
|
||||
passed += 1
|
||||
else:
|
||||
failed += 1
|
||||
except Exception as e:
|
||||
print(f"FAIL: {e}")
|
||||
failed += 1
|
||||
|
||||
print()
|
||||
print(f"Results: {passed} passed, {failed} failed")
|
||||
|
||||
return 0 if failed == 0 else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main_test())
|
||||
Reference in New Issue
Block a user