Queues

Each actor definition can specify a queue policy that controls how the runtime drains matched events from the per-instance queue. Queue policy is set at definition time and serialized on each instance record.

QueuePolicy type

type QueuePolicy =
  | { mode: "fifo"; concurrency: number }
  | { mode: "lifo"; concurrency: number }
  | { mode: "random"; concurrency: number }
  | { mode: "custom"; select: (queue) => event | event[]; concurrency: number };

Default: { mode: "fifo", concurrency: 1 } (DEFAULT_QUEUE_POLICY).

Modes

fifo

First-in-first-out among pending events, respecting priority sort:

queue: { mode: "fifo", concurrency: 1 },

Best for: ordered processing, chat ingestion, sequential workflows.

lifo

Most recently enqueued events processed first (after priority sort):

queue: { mode: "lifo", concurrency: 2 },

Best for: "latest state wins" patterns, stack-like processing.

random

Random selection among pending events:

queue: { mode: "random", concurrency: 4 },

Best for: load spreading when order does not matter.

custom

Application-defined selection over the pending queue:

queue: {
  mode: "custom",
  concurrency: 1,
  select: (events) => {
    const urgent = events.find((e) => e.type === "alert.critical");
    return urgent ?? events[0];
  },
},

The runtime lists pending events, sends them to Deno via eval_queue_select, and claims the returned event_id.

Concurrency

concurrency controls parallel process() invocations per group_id:

queue: { mode: "fifo", concurrency: Infinity }

Use groupId on appended events to scope concurrency:

await ctx.append(
  { type: "work.batch", payload: { id: "a" } },
  { groupId: "worker-a" },
);

Priority at append time

await ctx.append(
  { type: "alert.warning", payload: {} },
  { priority: 10 },
);

await ctx.append(
  { type: "alert.critical", payload: {} },
  { priority: 0 },
);

Lower priority values are claimed first. Tie-break: created_at ASC.

queueSelectFromConfig

Custom select functions are not serializable. Survive restart with a factory:

const worker = defineActor({
  configSchema: z.object({ preferredType: z.string() }),
  stateSchema: z.object({}),
  queue: { mode: "custom", concurrency: 1, select: () => { throw new Error("rebuilt"); } },
  queueSelectFromConfig: (config) => (events) =>
    events.find((e) => e.type === config.preferredType) ?? events[0],
  // ...
});

On boot, the runtime calls queueSelectFromConfig(config) to rebuild the selector.

Without the factory, custom-queue instances fall back to broken inline select after restart.

Definition example

From a minimal router/coder package:

router: actor({
  queue: { mode: "fifo", concurrency: 1 },
  // ...
}),

coder: actor({
  queue: { mode: "fifo", concurrency: 1 },
  // ...
}),

Agent SDK actors default to { mode: "fifo", concurrency: 1 } — preserving monologue message order.

Interaction with dequeue semantics

Regardless of mode:

• Event dequeued only after process() succeeds
• Throw → event stays on queue
• No early-ack API

Related