# Events and Cancellation

> Subscribe to live Framework session events, replay durable events, cancel a turn, and work with the canonical event protocol.

Source: <https://everruns.com/docs/framework/events-and-cancellation/>

Subscribe before sending a message to observe its live event projection:

```rust
use everruns::{Agent, Engine, OpenAI};


let agent = Agent::builder()
    .instructions("Be concise.")
    .provider(OpenAI::from_env()?)
    .model("gpt-6.1-sol")
    .build()?;
let session = Engine::new().create(agent);
let mut events = session.events();


let pending = session.send("Start.").await?;
while let Some(event) = events.recv().await? {
    println!("{}", event.event_type());
    if event.turn_id.as_deref() == Some(&pending.turn_id) && event.kind.is_terminal() {
        break;
    }
}
let turn = pending.wait().await?;
assert!(turn.success);
```

Known events have typed `SessionEventKind` values. Unknown canonical event types are preserved as `Other` with their payload, so the projection does not silently drop information. The feed is live and non-blocking: `send` starts execution in the session’s background actor, a slow or dropped consumer does not stop the turn, and the feed is not a durable replay API.

Use [lifecycle hooks](https://everruns.com/docs/framework/lifecycle-hooks/) instead when application work must be awaited at an execution boundary or its failure must affect the run.

See [Canonical events](#canonical-events) below for canonical envelopes, live-delta memory bounds, explicit lag handling, ordering, and the durability boundary. Use [Session history and resume](https://everruns.com/docs/framework/sessions/#history-and-resume) to rebuild a bounded persisted transcript after live lag or a process restart.

## Replay durable events

Durable events carry a dense per-session `sequence()` starting at 1. `session.events_after(n)` returns every durable event after sequence `n`, and it keeps working after `Engine::resume` in a new process when the backend persists events (for example `LocalConfig`). `session.events_from(n)` returns an `EventStream` that replays that backlog and then continues live. It never skips or repeats a durable event. Ephemeral deltas have no sequence and pass through live. A client that lagged or reconnected resumes from the last sequence it saw:

```rust
let last_seen = 2;
let mut stream = session.events_from(last_seen).await?;
while let Ok(Some(event)) = stream.try_recv() {
    println!("{:?} {}", event.sequence(), event.event_type());
}
```

## Cancel a turn

A message receipt exposes the specific accepting turn, so live applications can cancel without racing against whichever turn is active later:

```rust
let pending = session.send("Start.").await?;
pending.turn().cancel().await?;
let cancelled = pending.wait().await?;
assert!(!cancelled.success);
```

`run_with` retains cancellation-token convenience for request/response calls:

```rust
use everruns::{CancellationToken, RunOptions};


let cancel = CancellationToken::new();
let options = RunOptions::new().cancel_token(cancel.clone());
cancel.cancel();


let turn = session.run_with("Stop before starting.", options).await?;
assert!(!turn.success);
```

Cancellation is cooperative. Cancelling drops the in-flight turn future and tears down tool work through the same runtime path; it does not kill the host process or provide an independent transaction boundary.

## Observability

To send every session’s events to OpenTelemetry or Braintrust, or to your own listener, register it on the Engine; see [Framework observability](https://everruns.com/docs/framework/observability/).

## Canonical 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.

```rust
use everruns::prelude::*;


let agent = Agent::builder()
    .instructions("Answer concisely.")
    .provider(OpenAI::from_env()?)
    .model("gpt-6.1-sol")
    .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? {
        // Recording for replay, so take the canonical envelope: it withholds
        // nothing. `as_json()` carries the reviewed projection instead.
        canonical_events.push(event.canonical_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` exposes two surfaces, and which one you want depends on whether you are rendering or recording.

`SessionEvent::as_json()` and `SessionEvent::data()` return the **reviewed** form: the event envelope — id, type, timestamp, optional persisted sequence, correlation context, metadata, tags — with a `data` payload holding only the fields promoted onto `SessionEventKind`. Nothing else reaches it, so a field added inside the runtime cannot become part of the Framework’s public surface, or travel to wherever your application forwards these envelopes, without being promoted first. This is the form to log, forward, or expose to clients.

`SessionEvent::canonical_json()` returns the **canonical** envelope with the complete payload: prompts, tool arguments, tool results, structured assistant messages, and the payloads of event types this version does not recognize. Nothing observable is lost — this is the form for recording, auditing, and replay. It follows the runtime’s internal shape rather than the Framework’s reviewed surface, so treat what you read from it as unstable, and do not forward it anywhere the conversation itself should not go.

Live `output.message.delta` envelopes omit the redundant `data.accumulated` prefix on both surfaces; 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.

Model-generation accounting — model, provider, token counts, cost, and duration — is promoted onto `SessionEventKind::ModelGeneration`, so tracking spend never requires the unstable surface.

This bridge does not define a second wire schema. The canonical event contract, compatibility rules, and lifecycle semantics remain documented in [Events](https://everruns.com/docs/explanation/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()`](https://everruns.com/docs/framework/sessions/#history-and-resume). 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-core` (`host` feature) 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.

```rust
use async_trait::async_trait;
use everruns_core::events::{Event, EventRequest};
use everruns_contracts::typed_id::EventId;
use everruns_core::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.

The [external event-log example](https://github.com/everruns/everruns/tree/main/crates/everruns/tests/fixtures/external-consumer/event-log) provides a complete implementation in a standalone application.

### 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](https://everruns.com/docs/framework/sessions/#history-and-resume). 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-core` (`host` feature) 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`](https://github.com/everruns/everruns/blob/main/crates/everruns/examples/canonical_events.rs):

```bash
cargo run -p everruns --example canonical_events
```