Actors

An actor is a registered definition with a process() handler, durable state, and a persisted event queue. Actors are Yoke's reactive processors — they run because something happened on the stream.

Definition vs instance

One definition can have unlimited instances, each with different config and filters:

// Two typescript coders and one python coder — same definition, three instances
await ctx.instances.create("coder", {
  config: { model: "gpt-4o", lang: "typescript" },
  filter: { type: "work.assigned", where: { "payload.lang": "typescript" } },
});

Three layers of state

Each actor instance has three durable layers:

All three survive restart.

Queue

When fan-out matches an event to an instance filter, the event is inserted into actor_queues. The scheduler claims events per queue policy and invokes process().

Events are dequeued only after process() returns successfully. If process() throws, the event stays on the queue for retry.

SQLite

Each instance gets a private SQLite database — one file per instance, stored via pg_sqlite (BYTEA in Postgres) or S3:

Both access the same database. Never write __yoke_state directly — use ctx.state.

ctx.state.update() is atomic: SQLite transaction with optimistic version check via the state_cas op.

Postgres record

The actor_instances row holds config JSON, serialized filter, queue policy, worker assignment, and lifecycle status. This is what survives across boots to rebuild filters and provision SQLite.

The process loop

1. Event matched by filter → persisted to instance queue
2. Scheduler claims event(s) per queue policy + concurrency
3. await process(ctx, event)   // async, impure, non-deterministic
4. process may call ctx.state, ctx.append, ctx.portals.*, ctx.instances.*
5. on success → runtime dequeues event
6. on throw → event stays queued for retry

There is no built-in agent loop. An LLM actor implements its own loop inside process. A routing actor might be a one-liner.

Filters — how actors subscribe

Actors do not declare what they listen to in the definition. Instances carry filters:

Declarative (fast path, indexed):

filter: { type: "work.assigned", where: { "payload.lang": "typescript" } }
filter: { type: "work.*" }
filter: { type: "slack.message.received" }

Function (slow path, evaluated per event):

filterFromConfig: (config) => (event) =>
  event.type === "work.assigned" && event.payload.lang === config.lang,

Function filters require filterFromConfig to survive restart. See Filters.

Queue policy

Each definition sets how its queue is drained:

queue: { mode: "fifo", concurrency: 1 }   // default
queue: { mode: "lifo", concurrency: 4 }
queue: { mode: "random", concurrency: 2 }
queue: {
  mode: "custom",
  concurrency: 1,
  select: (events) => events.find((e) => e.type === "urgent.alert") ?? events[0],
}

concurrency controls parallel process() invocations per groupId. Infinity is valid.

Claim order within a group is always priority ASC, created_at ASC regardless of mode. Modes affect which event is selected when using custom, or reordering in lifo/random paths — not the underlying priority sort for default fifo/lifo claims.

Lifecycle hooks

Who creates instances

Anyone with a context:

• Other actors via ctx.instances.create()
• Portals in route handlers or start() loops
• Portal start() hooks at runtime boot
• HTTP control plane API
• CLI

All paths write to Postgres, provision SQLite, register filters, append instance.created, and start the drain loop.

Actor vs portal

Example coder actor

coder: actor({
  configSchema: z.object({ model: z.string(), lang: z.string() }),
  stateSchema: z.object({ jobsCompleted: z.number() }),
  queue: { mode: "fifo", concurrency: 1 },

  filterFromConfig: (config) => (event) =>
    event.type === "work.assigned" && event.payload.lang === config.lang,

  async process(ctx, event) {
    await ctx.append({
      type: "code.produced",
      payload: { code: `// generated with ${ctx.config.model}` },
    });
    await ctx.portals.slack.sendMessage("#general", "Code ready");
    await ctx.state.update((s) => ({
      ...s,
      jobsCompleted: s.jobsCompleted + 1,
    }));
  },
}),

Related