Events & the event stream

The event stream (event bus) is the central nervous system of Yoke. Everything communicates through it. One interface, many backends — the code that talks to the bus never changes, only the delivery guarantees and durability do.

Events

An event is an envelope:

{
  id: string;              // unique id
  topic: string;           // e.g. "messages.incoming"
  type: string;            // e.g. "user.message"
  source: string;          // who published it
  timestamp: string;       // ISO 8601
  payload: unknown;        // your data
  metadata?: Record<string, unknown>;
}

Every runtime can define its own events, each with a type, a description, and an optional payload schema (zod) for validation and tooling:

runtime.event({
  type: "order.placed",
  description: "A customer placed an order.",
  schema: z.object({ orderId: z.string(), total: z.number() }),
});

Publishing

await ctx.emit({
  topic: "orders.created",
  type: "order.placed",
  payload: { orderId: "ord_123", total: 19.99 },
});

When a schema is provided, the payload is validated at publish time.

Subscribing

Actors subscribe on your behalf, but you can subscribe directly too:

const subscription = await bus.subscribe(
  ["orders.created", "invoices.*"],
  (envelope) => {
    console.log(envelope.payload);
  },
);
// later:
await subscription.unsubscribe();

Topic conventions

Topics are dot-separated namespaces: messages.incoming, yoke.runtime.*.
A trailing * component is a wildcard that matches any one segment.
** matches any suffix.

The interface

interface EventBus {
  readonly name: string;
  publish<P>(event: PublishEvent<P>): Promise<EventEnvelope<P>>;
  subscribe<P>(
    topics: string | string[],
    handler: (env: EventEnvelope<P>) => void | Promise<void>,
    options?: SubscribeOptions,
  ): Promise<EventSubscription>;
}

SubscribeOptions supports resuming from a cursor (startAt) and stream consumer/group names (consumer) for durable backends.

Adapters

Adapter	Module	Delivery model	Durable
Memory	`MemoryEventBus`	in-process pub/sub, async FIFO per subscriber	no
Redis Streams	`RedisEventBus`	stream groups + consumers (`ioredis`)	yes
NATS JetStream	`JetStreamEventBus`	core NATS + JetStream consumers (`nats`)	yes
Nano	`NanoEventBus`	provider contract needed	yes

import { RedisEventBus } from "yoke";

const bus = new RedisEventBus({
  redis: { host: "localhost", port: 6379 },
  stream: "yoke-events",
});

New providers implement the EventBus interface — never the other way around.

Standard events

The runtime emits these on its own stream (yoke.<scope>.<action>):

Type	Emitted when	Payload
`runtime.started`	runtime boots	`{ runtimeId }`
`runtime.stopping`	shutdown begins	`{ runtimeId }`
`runtime.stopped`	shutdown finished	`{ runtimeId }`
`agent.spawned`	an agent instance is created	`{ agentName, instanceId }`
`agent.message.received`	a message is dispatched to an agent	`{ agentName, instanceId, message }`
`agent.replied`	an agent produces a reply	`{ agentName, instanceId, reply }`
`agent.failed`	an agent invocation throws	`{ agentName, instanceId, error }`
`actor.started`	an actor begins listening	`{ actorName }`
`actor.stopped`	an actor stops listening	`{ actorName }`
`actor.failed`	an actor handler throws	`{ actorName, eventId, error }`
`tool.called`	a tool is invoked	`{ toolName, agentName? }`
`tool.failed`	a tool throws	`{ toolName, error }`
`event.defined`	a custom event is registered	`{ type, description? }`