Skip to content

Event System and Protocols

The infrastructure layer provides three cross-cutting systems: an async pub/sub EventBus for inter-module communication, a typed event schema with 120 event types across 23+ domains, and 22 @runtime_checkable Protocol classes for dependency injection.

EventBus — Async Pub/Sub

EventBus (core/infrastructure/event_bus.py) decouples producers from consumers. Components publish events without knowing which components subscribe.

Core API

from core.infrastructure.event_bus import EventBus
from core.infrastructure.event_schema import EventType

bus = EventBus(swallow_errors=True)  # Default: catch handler exceptions

# Subscribe to specific event types
bus.subscribe(EventType.TOOL_EXECUTION_SUCCESS, my_handler)

# Catch-all: receive every event published
bus.subscribe_all(audit_handler)

# Publish (async — awaits all async handlers)
await bus.publish(event)

Behavior

  • Typed subscription — Handlers subscribe by EventType enum value
  • Registration order — Handlers called in the order they were registered
  • Async awareness — Async handlers are awaited; sync handlers called directly
  • Error isolation — Handler exceptions caught/logged (one bad subscriber doesn't silence others)
  • Per-source sequence numbering — Monotonic _sequences[source] counters for replay debugging

Testing

Use swallow_errors=False in tests for fail-fast behavior. Production uses swallow_errors=True to prevent a single buggy handler from breaking the event pipeline.

Event Schema — 129 Types, 23+ Domains

EventType (core/infrastructure/event_schema.py) is an enum covering all observable system activities:

Domain Count Examples
agent ~5 Message start/chunk/end, tool invocation, memory context
audio ~4 Recording start/stop, transcription complete, TTS synthesis
config ~3 Parameter changed, config updated, model changed
stream 10 STREAM_* events for TTS pipeline coordination
tool ~6 Execution start/success/error/denied/granted/timeout
task ~4 Background task spawn/complete/cancel/error
migration ~3 Migration start/progress/complete
onboarding ~2 First-run detection, step completion
privacy ~2 Capability indicator state changes
telemetry ~2 Metrics export, benchmark results
isolation ~3 Sandbox execution, path validation, domain checks
approval ~3 Approval request/grant/deny
audit ~2 Audit log events for compliance
model ~2 Model download progress, setup check results
error ~2 Structured error reporting
system ~3 Lifecycle phases, health checks, shutdown

Event Structure

@dataclass
class Event:
    type: EventType         # Which event occurred
    source: str             # Component that published
    payload: dict           # Structured event data
    sequence: int           # Per-source monotonic counter
    timestamp: float        # Wall-clock time
    correlation_id: str     # Links related events across components

Service Protocols — 22 Interfaces

service_protocols.py (core/infrastructure/service_protocols.py) defines 22 @runtime_checkable Protocol classes for all injectable services:

Protocol Purpose
ModelBackend LLM inference (chat, embedding, structured output)
MemoryStore Vector memory operations (upsert, query)
ToolHost Tool execution with security pipeline
EventBusProtocol Async pub/sub interface
AuditLog Structured audit event persistence
WorkerPoolProtocol Background worker delegation
MetricsCollectorProtocol Performance telemetry collection
ProcessIsolationProtocol Sandboxed subprocess execution
LocalStoreProtocol SQLite canonical persistence
ApprovalWorkflowProtocol Human-in-the-loop approval gating
OrchestratorProtocol Conversational orchestration
... and 11 more Covering all major subsystem interfaces

Service Registry — Dependency Injection

ServiceRegistry (core/infrastructure/service_registry.py) provides runtime DI:

registry = ServiceRegistry()
registry.register(ModelBackend, ollama_backend)
backend = registry.resolve(ModelBackend)

Used in RuntimeApp but not enforced as the sole resolution mechanism (module-level singletons are simpler for a local desktop app). Provides infrastructure for future modularization without requiring all-or-nothing migration.

See Also