Event System

The event module implements an event-driven architecture that decouples Aleph's subsystems. Instead of direct method calls between modules, components publish events to buses and subscribe to the events they care about. src/event/ ships two related but distinct buses:

• EventBus (src/event/bus.rs) -- per-agent tokio broadcast channel; created via EventBus::new() / EventBus::with_buffer_size / EventBus::with_config. Optionally connects to a GlobalBus for cross-agent aggregation. It is not a singleton.
• GlobalBus (src/event/global_bus.rs) -- a process-wide singleton guarded by once_cell::sync::Lazy, accessed via GlobalBus::global(). It aggregates events from multiple EventBus instances and routes callbacks through EventFilters.

AlephEvent (src/event/types.rs) is the shared payload type for both buses; #[serde(tag = "type", content = "data")] makes it directly JSON-serializable.

Design Philosophy

Event-driven architecture provides three benefits in Aleph:

1. Decoupling — The Memory system doesn't need to know about the Gateway; both just use events
2. Observability — Every significant action produces an event that can be logged, audited, or streamed to clients
3. Extensibility — New features can subscribe to existing events without modifying publishers

EventBus defaults to a 1024-slot broadcast buffer; it retains history up to MAX_HISTORY_SIZE = 10000. GlobalBus also defaults to 1024 slots. All event handling is asynchronous and non-blocking.

Architecture

┌─────────────────────────────────────────┐
│         GlobalBus (singleton, Lazy)      │
│  ┌─────────────┐  ┌──────────────────┐ │
│  │  broadcast  │  │  Subscription    │ │
│  │  (GlobalEvt)│  │  Registry        │ │
│  └─────────────┘  └──────────────────┘ │
│       ▲                                 │
│       │ auto-broadcast when EventBus    │
│       │ is connected                    │
│       │                                 │
│ ┌──────────────────────────────────────┐│
│ │ EventBus (per-agent, NOT singleton)  ││
│ │  broadcast(TimestampedEvent)         ││
│ │  + history ring                      ││
│ └──────────────────────────────────────┘│
│       │                           ▲     │
│       │ publish(AlephEvent)       │ recv│
│       ▼                           │     │
│ ┌──────────┐              ┌─────────────┐
│ │ Publisher│              │  Handler    │
│ │ (any     │              │ (impl       │
│ │ module)  │              │ EventHandler│
│ └──────────┘              └─────────────┘

Core Components

GlobalBus (singleton)

Process-wide cross-agent aggregation bus:

use alephcore::event::global_bus::GlobalBus;
use alephcore::event::filter::EventFilter;
use alephcore::event::EventType;

let bus = GlobalBus::global();
let sub_id = bus.subscribe(
    EventFilter::new(vec![EventType::ToolCallStarted, EventType::ToolCallCompleted]),
    |event| println!("from agent={}", event.source_agent_id),
).await;

GlobalBus::broadcast(agent_id, session_id, event) wraps the event with source_agent_id / source_session_id / timestamp / sequence and dispatches it to all subscribers whose filters match.

EventBus (per-agent)

use alephcore::event::{EventBus, AlephEvent, SessionInfo};

let bus = EventBus::with_buffer_size(2048);
bus.publish(AlephEvent::SessionCreated(
    SessionInfo {
        id: session_id.clone(),
        parent_id: None,
        agent_id: "main".into(),
        model: "gpt-4-turbo".into(),
        created_at: chrono::Utc::now().timestamp_millis(),
    }
)).await;

The EventBus:

• Holds a tokio broadcast::Sender<TimestampedEvent> with a configurable buffer (default 1024)
• Wraps each published event as TimestampedEvent { event, timestamp, sequence }; sequence is monotonically incremented by an internal AtomicU64
• Retains history up to MAX_HISTORY_SIZE = 10000; configurable via EventBusConfig { buffer_size, enable_history, max_history_size }
• Can optionally connect to a GlobalBus (after setting agent_id / session_id); connected buses auto-broadcast through GlobalBus::broadcast

EventHandler Trait

#[async_trait]
pub trait EventHandler: Send + Sync {
    fn name(&self) -> &'static str;
    fn subscriptions(&self) -> Vec<EventType>;
    async fn handle(
        &self,
        event: &AlephEvent,
        ctx: &EventContext,
    ) -> Result<Vec<AlephEvent>, HandlerError>;
}

Implementors receive AlephEvent and may return new events to publish:

• Logging handlers write to disk
• WebSocket handlers forward to connected clients
• Memory handlers update the knowledge graph

Note that the trait is not generic over E: Event -- the receiver signature is &AlephEvent, and subscriptions are declared via subscriptions() -> Vec<EventType>. EventContext (src/event/handler.rs) carries bus: EventBus, abort_signal: Arc<AtomicBool>, and session_id: Arc<RwLock<Option<String>>>.

EventFilter

EventFilter (src/event/filter.rs) is a struct, not an enum; the three filtering dimensions are AND-combined:

pub struct EventFilter {
    pub session_ids: Option<HashSet<String>>,   // None = all sessions
    pub agent_ids: Option<HashSet<String>>,     // None = all agents
    pub event_types: Vec<EventType>,            // empty = matches nothing
}

Construction:

let filter = EventFilter::new(vec![EventType::ToolCallStarted, EventType::ToolCallCompleted])
    .with_session("session-123")
    .with_agent("agent-1");
// or
let filter = EventFilter::all();   // equivalent to EventFilter::new(vec![EventType::All])

Example: The Panel's WebSocket subscription uses EventFilter::new(vec![EventType::ToolCallStarted, ...]).with_session(...) to stream the relevant tool events for that session to the client.

Event Types

AlephEvent (src/event/types.rs) is a #[serde(tag = "type", content = "data")] #[non_exhaustive] enum. Variants grouped by purpose:

EventType is the subscription-filter discriminant. There is no MemoryUpdated or ConfigReloaded on this bus -- the latter is private to McpManagerEvent and is not bridged into the global stream.

Event Ordering and Timestamps

Each event is wrapped in TimestampedEvent:

pub struct TimestampedEvent {
    pub event: AlephEvent,
    pub timestamp: i64,   // chrono::Utc::now().timestamp_millis()
    pub sequence: u64,    // monotonically incremented by EventBus's AtomicU64
}

EventBus retains an in-process history ring (Arc<RwLock<Vec<TimestampedEvent>>>); new subscribers replay from there. GlobalEvent (src/event/global_bus.rs) adds source_agent_id / source_session_id and its own sequence, giving a deterministic ordering for cross-agent delivery.

Usage Example

use alephcore::event::{EventBus, EventHandler, EventContext, EventType, AlephEvent, HandlerError};
use async_trait::async_trait;
use std::sync::Arc;

struct MemoryLogger;

#[async_trait]
impl EventHandler for MemoryLogger {
    fn name(&self) -> &'static str { "MemoryLogger" }
    fn subscriptions(&self) -> Vec<EventType> {
        vec![EventType::ToolCallCompleted]
    }
    async fn handle(
        &self,
        event: &AlephEvent,
        _ctx: &EventContext,
    ) -> Result<Vec<AlephEvent>, HandlerError> {
        if let AlephEvent::ToolCallCompleted(t) = event {
            tracing::info!(call_id = %t.call_id, tool = %t.tool, "tool finished");
        }
        Ok(vec![])
    }
}

let bus = EventBus::new();
let mut registry = EventHandlerRegistry::new();
registry.register(Arc::new(MemoryLogger));
let _handles = registry.start(EventContext::new(bus.clone())).await;

Code Location

• src/event/mod.rs — Module entry point, re-exports
• src/event/bus.rs — EventBus (per-agent) + EventBusConfig + EventSubscriber
• src/event/global_bus.rs — GlobalBus singleton + GlobalEvent + Subscription + SubscriptionId
• src/event/handler.rs — EventHandler trait + EventHandlerRegistry + EventContext
• src/event/filter.rs — EventFilter struct
• src/event/types.rs — AlephEvent + EventType + payload types

See Also

• Gateway — How events are forwarded to WebSocket clients
• Session Service — Session lifecycle events
• Memory System — Memory update events

26.7.x Addendum

broadcast::RecvError::Lagged Must Recover

26.7.21 fix: the channel forwarder no longer dies permanently on broadcast::RecvError::Lagged. The per-connection event-forward task no longer leaks on WebSocket disconnect. Webhook receivers return 503 under congestion rather than silently dropping inbound messages.

Single-Source Error Receipt

26.7.21 fix: the gateway returns a single-source user-readable error to the client. The raw error chain never leaks out to the Panel.

SessionEvent

26.7.7+ session_events is the canonical event log -- MessageProjector materializes events into the messages table; a boot-time ProjectionReconciler back-fills transcripts. SessionEventRecord { seq, event, created_at_ms } is the log payload; this is a parallel event stream to AlephEvent, separating "session events" from "runtime events".

Gateway Dead-Letter Forensics

26.7.17+: undeliverable messages on the channel path are captured in a gateway-side dead-letter forensics trail, inspectable from the Panel, with safe re-drive semantics.

Loop-Graph Governance Events

26.7.21+ governance events (victory claim, goal settled, team disbanded, watcher triggered) flow over the same event bus, subscribed by watchers in the loop_graph layer. NodeKind::Team and NodeKind::LoopGoal are members of the non_exhaustive enum in loop_graph/types.rs.

See Also

• Gateway Protocol — error-receipt single-source
• Messages — SessionEvent and MessageProjector
• Architecture Overview — how governance events flow through the graph