Skip to content
ManykindManykindmanykind

Event streams

Events are how data enters a vault and how agents react to it. A stream is every event in a scope that shares one context value. You never create a stream: you publish events with a context, and they form one.

Each connected agent handles the events of one stream in one Job. A new context starts a new Job.

1. Publish events

From a client or server, use @cef-ai/vault-sdk:

import { VaultSDK, KeypairWallet } from "@cef-ai/vault-sdk";


const sdk = new VaultSDK({ endpoint: vaultApiUrl, signer: await KeypairWallet.fromSeed(seed) });
const vault = await sdk.vault.current();


const { eventId } = await vault.scope("default").publish({
  type: "reading.received",
  context: "sensor:42",
  payload: { schema_version: 1, celsius: 21.5 },
});
Field Required Meaning
type yes The event type. A code agent handles it with @OnEvent("<type>").
context yes The stream key. Events that share it form one stream.
payload yes Your data.
target no <asPubkey>:<alias>. Delivers the event only to that agent. Without it, every connected agent that handles the type receives it.
role no "source", "user", or "agent".
correlationId, metadata, parents, timestamp no Tracing, labels, causal lineage, and the source time.

publish throws if the vault rejects the event. Publishing returns before any agent runs: dispatch is asynchronous.

An agent publishes with ctx.vault.publish(type, payload, opts?). The event joins the stream of the event being handled.

2. Handle them

import { Engagement, OnEvent, type Context, type Event } from "@cef-ai/agent-sdk";


type Reading = { schema_version: number; celsius: number };


@Engagement({ id: "default", goal: "Track sensor readings" })
export default class SensorAgent {
  @OnEvent("reading.received")
  async onReading(event: Event<Reading>, ctx: Context) {
    if (event.payload?.schema_version !== 1) {
      console.warn("unknown schema_version", event.payload);
      return;
    }
    await ctx.cubby("readings").exec(
      "INSERT INTO sensor_readings(event_id, context, celsius, at) VALUES (?, ?, ?, ?) ON CONFLICT(event_id) DO NOTHING",
      [event.eventId, event.context, event.payload.celsius, event.timestamp],
    );
  }
}

event carries type, payload, timestamp (ISO 8601), context, role, and, when present, from, eventId, and parents.

Give each payload a schema_version and ignore versions you do not know. Old and new shapes then coexist safely during a rollout.

Make handlers idempotent

Delivery is at-least-once, so the same event can arrive twice. Make every write safe to repeat:

  • Key rows on event.eventId or a natural key, and write with ON CONFLICT … DO NOTHING, ON CONFLICT … DO UPDATE, or INSERT OR IGNORE.
  • Keep no state in memory between events. Each event can run in a fresh isolate. Read what you need from the cubby when the handler starts.

Who receives an event

The vault delivers an event to every agent connected in that scope that handles its type, unless the event names a target. LLM agents receive only events targeted at them. An event no connected agent handles reaches nobody, and nothing reports an error.

Follow a stream

const sub = vault.scope("default").subscribe(
  { context: "sensor:42", types: ["reading.alert"] },
  (event) => console.log(event.type, event.payload),
  { intervalMs: 1000 },
);
// later
sub.unsubscribe();

subscribe polls the stream, every second by default, and delivers each event once. subscribeAll({ types }, handler) follows every stream in the scope. To read history, use vault.scope(s).stream(context).events.list().

A widget follows a stream with WidgetRuntime.subscribe; see Visualize vault data.

Test with the Sandbox

In ROC, open Sandbox, pick the agent and a vault, and use the simulator: Single event sends one event, Data stream sends a sequence, and Replay replays events from the vault into a new context.

Failure modes

Symptom Cause Fix
Event accepted, no Job No connected agent handles that type in that scope, or the event targets another agent. Connect the agent; check the type string and target.
Duplicate rows A redelivered event. Make the write idempotent.
PAYLOAD_TOO_LARGE The event is too big. Upload the data to vault objects and publish its path.

Limits

Limit Value
Publish request body 1 MiB
Events per publish call 100
Event retention per stream 7 days

See Runs and events.