Events & Stream

The global stream is Yoke's communication backbone. Every action that matters in the system becomes a durable, sequenced event in Postgres. Actors never communicate directly — they publish events and subscribe via filters.

Event shape

Every persisted event has this structure:

type StreamEvent = {
  seq: number;           // global monotonic position (per tenant)
  id: string;            // unique event id (UUID)
  at: number;            // unix timestamp ms
  type: string;          // open namespace, e.g. "work.assigned"
  payload: unknown;      // JSON, validated by Zod at append
  source?: {
    kind: "actor" | "portal";
    instanceId: string;
  };
  correlation?: string;  // optional grouping for related work
  priority?: number;     // queue ordering hint (lower = sooner)
  groupId?: string;      // concurrency group for queue processing
};

Append-only API

User code can only append to the stream. There is no tail, subscribe, or read API exposed to TypeScript:

await ctx.append({
  type: "work.assigned",
  payload: { role: "coder", lang: "typescript" },
});

The append path:

1. Validate payload against Zod schema (if registered)
2. Insert row into stream_events with next seq
3. Return immediately to caller
4. Async fan-out evaluates actor filters and enqueues matches

Typed events with defineEvents

Register schemas once for compile-time safety and runtime validation:

const events = defineEvents({
  "slack.message.received": z.object({
    channel: z.string(),
    text: z.string(),
  }),
  "work.assigned": z.object({
    role: z.string(),
    lang: z.string().optional(),
  }),
});

export default defineRuntime().events(events) /* ... */;

Inside process(), known event types narrow automatically:

async process(ctx, event) {
  if (event.type === "work.assigned") {
    event.payload.lang; // typed as string | undefined
  }
}

Unregistered event types still append and process — payload is unknown.

Correlation

Group related events into one piece of work:

ctx.correlation.current;    // id of event being processed, if any
ctx.correlation.inherit();  // reuse current or create new
ctx.correlation.create();   // explicit new chain

// Default: inherit correlation while processing
await ctx.append({ type: "step.started", payload: {} });

// Start a new chain
await ctx.append({ type: "step.started", payload: {} }, { correlation: "new" });

// Explicit id
await ctx.append({ type: "step.started", payload: {} }, { correlation: "order-42" });

Correlation is stored on the event row and passed through fan-out. It does not affect delivery — only observability and downstream logic that chooses to read it.

Priority and groupId

When fan-out enqueues an event to an actor queue, priority and groupId control drain behavior:

await ctx.append(
  { type: "urgent.alert", payload: { text: "down" } },
  { priority: 0, groupId: "alerts" },
);

Queue claim order is priority ASC, created_at ASC — lower priority numbers are claimed first; ties break on enqueue time. This is independent of queue policy mode (fifo, lifo, random, custom).

groupId scopes concurrency: the scheduler drains one concurrency slot per group. Use "inherit" in conversation postMessage to propagate the triggering event's group.

Stream lifecycle events

The runtime appends system events:

Conversation and agent packages add more — see Event catalog.

Fan-out delivery

After append commits, the fan-out engine:

1. Looks up declarative filters in the fast-path index (by type prefix)
2. Evaluates function filters on the slow path (via Deno eval_filter)
3. Inserts matching events into actor_queues per instance
4. Records delivery in fanout_delivered for idempotency

Fan-out is per-tenant and idempotent — duplicate delivery for the same (tenant_id, stream_seq, instance_id) is prevented.

See Fan-out for implementation details.

Testing with CLI append

Append events without writing code:

yoke append \
  --database "$DATABASE_URL" \
  --type slack.message.received \
  --payload '{"channel":"#general","text":"hello"}'

Multi-tenant deployments scope by --tenant-id.

What the stream is not

Related