Events

EventInput, StreamEvent, AppendInput, Zod validation, correlation, priority, and groupId.

All communication in Yoke flows through a global append-only event stream. The SDK types distinguish persisted events, append inputs, and the narrowed shapes received in process().

StreamEvent

A persisted event after append (assigned by the runtime):

type StreamEvent<T = unknown> = {
  seq: number;       // monotonic stream sequence
  id: string;        // unique event id
  at: number;        // epoch ms timestamp
  type: string;      // event type key
  payload: T;
  source?: EventSource;
  correlation?: string;
  priority?: number;
  groupId?: string;
};

type EventSource = {
  kind: "actor" | "portal";
  instanceId: string;
};

Field	Description
`seq`	Global ordering in the tenant stream
`id`	Stable id used for queue deduplication and custom selectors
`at`	Wall-clock append time (ms)
`source`	Which instance appended the event
`correlation`	Logical work-chain id
`priority`	MX-style priority (lower runs sooner within a group)
`groupId`	Concurrency partition within an actor instance queue

EventInput

Minimal shape for new events (runtime assigns seq, id, at):

type EventInput<T = unknown> = {
  type: string;
  payload: T;
  correlation?: string;
  priority?: number;
  groupId?: string;
};

AppendInput

Typed append union — known event types from your EventMap plus an open fallback:

type KnownAppendInput<TMap extends EventMap> = {
  [K in keyof TMap & string]: {
    type: K;
    payload: z.core.output<TMap[K]>;
    correlation?: string;
    priority?: number;
    groupId?: string;
  };
}[keyof TMap & string];

type AppendInput<TMap extends EventMap = EmptyEventMap> =
  | KnownAppendInput<TMap>
  | EventInput;

Example with registered types:

const events = defineEvents({
  "work.assigned": z.object({ role: z.string(), lang: z.string() }),
});

// Typed payload — editor knows fields
await ctx.append({
  type: "work.assigned",
  payload: { role: "coder", lang: "typescript" },
});

// Unregistered types still allowed
await ctx.append({
  type: "custom.debug",
  payload: { note: "anything" },
});

StreamEventInput

Shape received in process(ctx, event):

type KnownStreamEvent<TMap extends EventMap> = {
  [K in keyof TMap & string]: {
    seq: number;
    id: string;
    at: number;
    type: K;
    payload: z.core.output<TMap[K]>;
    source?: EventSource;
    correlation?: string;
    priority?: number;
    groupId?: string;
  };
}[keyof TMap & string];

type StreamEventInput<TMap extends EventMap = EmptyEventMap> =
  keyof TMap & string extends never
    ? StreamEvent
    : KnownStreamEvent<TMap>;

When an EventMap is wired through the actor definition, only known types appear in the union — full discriminated narrowing on event.type.

AppendResult

type AppendResult = {
  seq: number;
  id: string;
};

Returned from ctx.append().

Zod validation

Event payloads are validated in the Deno worker at:

ctx.append() — before stream_append RPC
process() — before handler runs (re-validates dequeued payload)

Validation logic:

function validateEvent(input: AppendInput): AppendInput {
  const schema = loaded.eventSchemas[input.type];
  if (!schema) return input;           // unknown type — pass through
  const payload = schema.parse(input.payload);  // throws on invalid
  return { ...input, payload };
}

const events = defineEvents({
  "order.created": z.object({
    orderId: z.string().uuid(),
    total: z.number().positive(),
  }),
});

export default defineRuntime().events(events).build();

Event type	Schema registered	Validation
Known	yes	Strict Zod parse; throws `ZodError` on failure
Unknown	no	Payload accepted as-is

Actor configSchema and stateSchema are separate — validated on instance create and state operations, not on the stream.

ctx.append options

await ctx.append(
  { type: "work.assigned", payload: { role: "coder", lang: "ts" } },
  {
    correlation: "inherit" | "new" | "my-chain-id",
    priority: 0 | "inherit",
    groupId: "default" | "inherit" | "monologue-abc",
  },
);

Correlation

Group related events into one logical piece of work:

Value	Behavior
`"inherit"` (default while processing)	Reuse `ctx.correlation.current`, or create if none
`"new"`	Start a fresh correlation chain
explicit string	Use that id

await ctx.append({ type: "step.started", payload: {} }); // inherits
await ctx.append({ type: "step.done", payload: {} }, { correlation: "new" });

priority

MX-style ordering within a concurrency group on an actor queue:

Lower number runs sooner (default 0)
"inherit" copies ctx.priority from the event being processed
Omitted on append → resolved via resolveQueueDispatch (inherit while processing, else default 0)

await ctx.append(
  { type: "guardian.run", payload: {} },
  { priority: -10 }, // urgent — runs before priority 0 in same group
);

groupId

Partitions concurrency within one actor instance:

Events with different groupId values drain in parallel up to queue.concurrency per group
Default group: "default" (DEFAULT_GROUP_ID)
"inherit" copies ctx.groupId from the current event

await ctx.append(
  { type: "agent.think", payload: { monologueId: "m-1" } },
  { groupId: "m-1" }, // isolated drain lane for this monologue
);

Scheduler claims pending events with:

ORDER BY priority ASC, created_at ASC

within each (instance_id, group_id) partition.

resolveQueueDispatch

Utility used internally by append and conversation postMessage:

import { resolveQueueDispatch, DEFAULT_PRIORITY, DEFAULT_GROUP_ID } from "@yoke/sdk";

const { priority, groupId } = resolveQueueDispatch(
  { priority: ctx.priority, groupId: ctx.groupId },
  { priority: "inherit", groupId: "my-group" },
);

matchEvent

Type guard for known event types:

import { matchEvent } from "@yoke/sdk";

async process(ctx, event) {
  if (matchEvent(event, "work.assigned")) {
    event.payload.lang; // narrowed
    return;
  }
  // other known types or exhaustiveness checking
}

Event flow

ctx.append(input)
  → validateEvent (Zod if registered)
  → stream_append RPC
  → Postgres stream_events
  → fan-out to matching instance queues
  → scheduler → process(ctx, event)

See Filters for subscription matching and Queues for drain ordering.

On this page