Canonical Framework events

Session::events() installs an in-process subscriber without exposing runtime event buses or core event types. Subscribe before Session::run() so the stream sees the turn from its first event.

use everruns::prelude::*;

let agent = Agent::builder()
    .instructions("Answer concisely.")
    .model(Model::simulated("Hello!"))
    .build()?;
let engine = Engine::new();
let mut session = engine.create(agent);
let mut events = session.events();

let observer = tokio::spawn(async move {
    let mut canonical_events = Vec::new();
    while let Some(event) = events.recv().await? {
        // Record the public event envelope.
        canonical_events.push(event.as_json().clone());

        // Typed rendering for common terminal/service UI concerns.
        match event.kind {
            SessionEventKind::TextDelta { delta } => print!("{delta}"),
            SessionEventKind::ToolStarted { tool_name, .. } => {
                eprintln!("starting {tool_name}");
            }
            SessionEventKind::ToolCompleted {
                tool_name,
                success,
                ..
            } => eprintln!("{tool_name}: {success}"),
            SessionEventKind::TurnFailed { error } => eprintln!("failed: {error}"),
            SessionEventKind::TurnCancelled => eprintln!("cancelled"),
            _ => {}
        }
    }
    Ok::<_, EventStreamError>(canonical_events)
});

let turn = session.run("hello").await?;
drop(session); // closes the subscriber once buffered events are drained
let recorded = observer.await??;
assert!(!recorded.is_empty());

One protocol, two views

SessionEventKind is a convenience projection for application renderers. It promotes assistant output lifecycle and deltas, model reasoning/generation, tool lifecycle/progress/output, and turn terminal states. It is non-exhaustive, so match it with a fallback arm.

SessionEvent::as_json() returns the canonical envelope used by the public Everruns event protocol, including the event id and type, timestamp, optional persisted sequence, correlation context, typed payload, metadata, and tags. Live output.message.delta envelopes omit the redundant data.accumulated prefix; retaining every growing prefix in a slow subscriber’s buffer would use quadratic memory. Concatenate the typed TextDelta::delta values to reconstruct streamed text, or use the subsequent output.message.completed event for the complete message. SessionEvent::data() accesses the retained payload directly. Use it for details intentionally not duplicated into the convenience projection, such as:

• tool arguments, results, status, narration, and duration;
• complete structured assistant messages, phases, model metadata, and usage;
• stable failure codes and localization fields;
• newly added event types unknown to this version of the Framework.

This bridge does not define a second wire schema. The canonical event contract, compatibility rules, and lifecycle semantics remain documented in Events.

Durability, observation, and derived history

These roles are deliberately separate:

• EventLog is the host’s sole durable conversation write authority. It stores complete canonical event envelopes and provides bounded cursor replay.
• EventSink is the host’s post-commit, nonblocking live-delivery boundary. Session::events() exposes that observation path as an ergonomic EventStream subscriber. Neither sink nor subscriber is durable or authoritative.
• EventHistory is one read-only message projection rebuilt from EventLog replay. It is an index/view, never a second writable message store.

Framework applications read that bounded projection through Session::history(). It pages messages from a stable event-log snapshot; it does not maintain or write an independent transcript.

Rebuild a transcript in persisted sequence order from input.message, output.message.completed, and relevant tool.completed events. An output.message.replaced event alone creates no history message; the subsequent completed message contains the safe replacement. If a crash leaves a replacement without completion, replay correctly omits that incomplete output. The Framework stream exposes canonical payloads, subject to the live-delta exception above, and introduces no independent writable message history.

Canonical recordings can contain user messages, agent instructions, model inputs, tool arguments, and tool results. Treat them as application data with the same access controls and retention policy as the session itself; do not log them indiscriminately. Model and tool text is untrusted: terminal renderers should strip or escape control sequences, and web renderers should escape it as content rather than interpreting it as markup or commands. Provider credentials are not part of the event protocol.

Implementing a custom event log

An advanced host can store canonical events itself. everruns-host exposes EventReader and EventLog as a public SPI: an external crate implements both against its own storage and supplies the result to composition through HostBackends::with_event_log. No in-crate access is required, cursors and pages are built with EventCursor::continuation, EventCursor::after, and EventPage::new, which validate the shared invariants.

Three request shapes are distinguished by EventReadRequest::cursor():

• no cursor is an initial read that captures the session’s current high-watermark and reports it as EventPage::snapshot_high_watermark();
• a cursor whose snapshot_high_watermark() is Some is a continuation pinned to that snapshot, so appends committed later stay invisible and paging neither skips nor duplicates;
• a cursor whose snapshot_high_watermark() is None, built by EventCursor::after, is a poll that captures a fresh snapshot and therefore does observe those later appends.

use async_trait::async_trait;
use everruns_core::events::{Event, EventRequest};
use everruns_provider::typed_id::EventId;
use everruns_host::{
    EventCursor, EventDurability, EventLog, EventLogError, EventPage, EventReadRequest,
    EventReader,
};

#[async_trait]
impl EventReader for MyEventLog {
    async fn read_page(&self, request: EventReadRequest) -> Result<EventPage, EventLogError> {
        let session_id = request.session_id();
        let current_high = self.high_watermark(session_id);
        let (after, snapshot) = match request.cursor() {
            None => (0, current_high),
            Some(cursor) => {
                if cursor.session_id() != session_id {
                    return Err(EventLogError::CrossSessionCursor {
                        detail: "cursor belongs to another session".into(),
                    });
                }
                match cursor.snapshot_high_watermark() {
                    Some(snapshot) if snapshot > current_high => {
                        return Err(EventLogError::ExpiredCursor {
                            detail: "cursor snapshot is not available".into(),
                        });
                    }
                    // Pinned continuation, then the polling form.
                    Some(snapshot) => (cursor.after_sequence(), snapshot),
                    None => (cursor.after_sequence(), current_high),
                }
            }
        };

        let limit = request.limit().get();
        let mut events = self.events_in(session_id, after, snapshot, limit + 1);
        let has_more = events.len() > limit;
        if has_more {
            events.pop();
        }
        let next_cursor = has_more
            .then(|| {
                let last = events.last().and_then(|event: &Event| event.sequence).unwrap_or(after);
                EventCursor::continuation(session_id, last, snapshot)
            })
            .transpose()?;
        EventPage::new(events, next_cursor, snapshot)
    }
}

#[async_trait]
impl EventLog for MyEventLog {
    async fn append(&self, request: EventRequest) -> Result<Event, EventLogError> {
        if request.is_ephemeral() {
            return Err(EventLogError::InvalidAppend {
                detail: "ephemeral events are sink-only".into(),
            });
        }
        // The log owns identity: assign the event id and the next per-session
        // sequence, persist, and return the finalized canonical envelope.
        let sequence = self.next_sequence(request.session_id);
        let event = request.into_event(EventId::new(), sequence);
        self.persist(&event)?;
        Ok(event)
    }

    fn durability(&self) -> EventDurability {
        EventDurability::CrashDurable
    }
}

The contract an implementation must uphold:

• an accepted append owns id and sequence assignment and returns the finalized canonical Event, visible to the next read of that session;
• durable sequences are unique and strictly increasing per session, and need not be contiguous, gaps are expected when a reader projects an append-only physical log into a filtered logical event sequence;
• a continuation stays pinned to the first page’s high-watermark and cannot observe concurrent appends; a poll cursor can;
• cursor/session mismatches and inconsistent positions return the typed EventLogError variants above rather than panicking;
• the log is append-only. There is no truncate, rewind, or mutation contract, and EventHistory remains a read-only projection rather than a second writable message store.

tests/fixtures/external-consumer/event-log in the repository is a complete out-of-workspace implementation exercised by repository CI.

Ordering and bounded delivery

A subscriber receives events in channel arrival order and each session has its own stream. The canonical sequence field is a replay position, not a live delivery counter: durable events carry Some(sequence) and live-only ephemeral events such as streaming deltas carry no sequence. Persisted sequences increase monotonically per session and may have gaps. Ephemeral events do not consume replay positions.

The live stream has a bounded buffer and never applies backpressure to the agent turn. A dropped or slow subscriber cannot stall model or tool execution. If a subscriber falls behind, recv() and try_recv() return EventStreamError::Lagged { missed }; loss is never hidden. The next receive can continue from the oldest retained event, but the renderer must treat its live projection as incomplete.

Streaming deltas are provisional and sink-only. Completed assistant/tool events are authoritative, and an output-replacement event means accumulated text for that message must be discarded. Durable events reach the live sink only after their log append commits; ephemeral events go directly to the sink and never enter history. After live lag, a Framework application can rebuild its persisted transcript with bounded Session::history() pages. That projection excludes ephemeral deltas by design. Applications that need raw durable envelopes rather than derived messages can provide and read an EventLog through the advanced everruns-host SPI; neither recovery path relies on the in-process subscriber.

Cancellation and failure

Pass a CancellationToken through RunOptions to stop a turn. Cancellation produces both a Turn with TurnStopReason::Cancelled and a correlated turn.cancelled event carrying the same turn_id.

Runtime failures remain available through Session::run()’s outcome/error semantics and the event stream. Subscribe before running and continue draining the stream after the run resolves to retain the terminal failure event and its full structured payload.

The complete runnable example is canonical_events.rs:

cargo run -p everruns --example canonical_events