Actors

defineActor, process, lifecycle hooks, queue policy, filterFromConfig, queueSelectFromConfig, and sqliteStoreId.

An actor is a registered definition with a process() handler, durable state, and a persisted event queue. Definitions are fixed at build time; instances are created at runtime with per-instance config and filters.

defineActor

import { defineActor, defineEvents, z } from "@yoke/sdk";

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

const coder = defineActor({
  events,
  configSchema: z.object({
    model: z.string(),
    lang: z.string(),
  }),
  stateSchema: z.object({
    jobsCompleted: z.number(),
  }),
  filterFromConfig: (config) => (event) =>
    event.type === "work.assigned" && event.payload.lang === config.lang,
  async process(ctx, event) {
    await ctx.state.update((s) => ({
      ...s,
      jobsCompleted: s.jobsCompleted + 1,
    }));
  },
});

Inside defineRuntime().actors(), prefer defineFor(events, portals).actor({ ... }) — it omits the events field.

ActorDefinition fields

Field	Required	Description
`configSchema`	yes	Zod schema for instance config (validated on create)
`stateSchema`	yes	Zod schema for initial/default durable state
`process`	yes	`(ctx, event) => Promise<void>` — main handler
`queue`	no	Drain policy; default `{ mode: "fifo", concurrency: 1 }`
`filterFromConfig`	no	Factory for function filters (required for function-filter survival across restart)
`queueSelectFromConfig`	no	Factory for custom queue selectors after restart
`sqliteStoreId`	no	Resolve SQLite store segment per invocation
`onCreate`	no	Called once when instance is provisioned
`onDestroy`	no	Called before instance teardown

process

async process(ctx, event) {
  // ctx — full ActorCtx (see Context API)
  // event — StreamEventInput<EventMap> with typed payload for known types

  if (event.type === "work.assigned") {
    event.payload.lang; // typed when in EventMap
  }
}

Processing semantics:

Event matches instance filter → enqueued (Postgres actor_queues).
Scheduler claims event(s) per queue policy and concurrency group.
Worker calls process(ctx, event) in Deno.
Success → event dequeued.
Throw → event stays on queue for retry.

There is no early-ack API. Long work should be split across events or schedules.

matchEvent

Narrow event types in process():

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

async process(ctx, event) {
  if (matchEvent(event, "work.assigned")) {
    event.payload.lang; // narrowed
  }
}

onCreate / onDestroy

Lifecycle hooks receive the same ActorCtx as process() (without a triggering event — default priority/group, default correlation).

actor({
  configSchema: z.object({ webhookUrl: z.string() }),
  stateSchema: z.object({ ready: z.boolean() }),

  async onCreate(ctx) {
    await ctx.sqlite.exec(
      "CREATE TABLE IF NOT EXISTS jobs (id TEXT PRIMARY KEY, status TEXT)",
    );
    await ctx.state.set({ ready: true });
  },

  async onDestroy(ctx) {
    await ctx.portals.notify.send(ctx.config.webhookUrl, "instance stopped");
  },

  async process(ctx, event) { /* ... */ },
})

Hook	When	Typical use
`onCreate`	After instance row + SQLite provisioned	Schema migration, initial state, side subscriptions
`onDestroy`	Before rows deleted	Cleanup, final notifications

queue

Default queue policy:

import { DEFAULT_QUEUE_POLICY } from "@yoke/sdk";
// { mode: "fifo", concurrency: 1 }

Override on the definition:

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

See Queues for modes, priority groups, and custom selectors.

filterFromConfig

Function filters are not serializable. Persisted instances store config only; on restart the runtime rebuilds the filter by calling filterFromConfig(config):

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

Scenario	Behavior
Definition has `filterFromConfig`	Function filters work after restart via `eval_filter` RPC
Definition lacks `filterFromConfig`	Function-filter instances persist in Postgres but receive no events until recreated
Instance created with declarative filter	Fast-path fan-out; no `filterFromConfig` needed

At instance creation you can pass either filter style:

await ctx.instances.create("coder", {
  config: { model: "claude-sonnet", lang: "typescript" },
  filter: { type: "work.assigned", where: { "payload.lang": "typescript" } },
});

await ctx.instances.create("coder", {
  config: { model: "claude-sonnet", lang: "typescript" },
  filter: (event) =>
    event.type === "work.assigned" && event.payload.lang === "typescript",
});

Optional typeHint on CreateInstanceOpts narrows slow-path evaluation for function filters (optimization, not a guarantee).

queueSelectFromConfig

Custom queue.select functions are also not serializable. Provide a factory mirroring filterFromConfig:

queueSelectFromConfig: (config) => (pending) => {
  const preferred = pending.filter((e) => e.type === config.preferredType);
  return preferred[0] ?? pending[0];
},

At boot, get_actor_specs reports hasCustomQueueSelect: true and the runtime uses eval_queue_select RPC to run the selector over pending events.

If you inline queue: { mode: "custom", select, concurrency } without queueSelectFromConfig, custom selection works until restart, then fails unless the factory is added.

sqliteStoreId

By default each actor instance has one SQLite database keyed by instanceId. Use sqliteStoreId to shard storage within an instance:

import { sqliteStoreSegments, resolveSqliteStoreId } from "@yoke/sdk";

actor({
  configSchema: z.object({ tenantId: z.string() }),
  stateSchema: z.object({ count: z.number() }),

  sqliteStoreId: (config) => async (ctx, event) => {
    if (event?.type === "session.started") {
      return sqliteStoreSegments.perStint(event.payload.sessionId);
    }
    return sqliteStoreSegments.perInstance();
  },

  async process(ctx, event) {
    ctx.sqliteStoreId; // e.g. "inst-abc:stint:sess-123"
    await ctx.state.get(); // scoped to resolved store
  },
})

resolveSqliteStoreId

resolveSqliteStoreId("inst-abc", "");           // "inst-abc"
resolveSqliteStoreId("inst-abc", "default");    // "inst-abc"
resolveSqliteStoreId("inst-abc", "stint:x");    // "inst-abc:stint:x"

sqliteStoreSegments helpers

Helper	Segment	Use case
`perInstance()`	`""`	Default — one DB per instance
`perStint(id)`	`stint:${id}`	Agent monologue / forum stint isolation
`perInvocation(event)`	`inv:${event.id}`	One store per queued event
`perEventType(event)`	`event:${event.type}`	Partition by event type

Definition vs instance

	Definition	Instance
When	`defineRuntime().actors()`	`ctx.instances.create("name", opts)`
Stored	TypeScript module	Postgres (`instances`, queues, filters)
Contains	`process`, schemas, queue, factories	`config`, `filter`, runtime status
Count	Fixed set of names	Unlimited dynamic instances

ActorRegistration

type ActorRegistration<Name, TConfig, TState, TPortals, TEventMap> = {
  readonly name: Name;
  readonly definition: ActorDefinition<...>;
};

Inference helpers: InferActorConfig<R>, InferActorState<R>, InferActorConfigFromRegistry, InferActorStateFromRegistry.

Full example

export default defineRuntime()
  .events(events)
  .portals(({ events }) => {
    const { portal } = defineFor(events);
    return {
      slack: portal({
        configSchema: z.object({ botToken: z.string() }),
        interface: {
          async sendMessage(channel: string, text: string) {
            console.log(channel, text);
          },
        },
      }),
    };
  })
  .actors(({ events, portals }) => {
    const { actor } = defineFor(events, portals);

    return {
      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 onCreate(ctx) {
          await ctx.state.set({ jobsCompleted: 0 });
        },

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

On this page