AgentQueue

AgentQueue is the turn scheduler inside every agent. It serializes top-level turns, drains queued input into the active turn, and handles aborts and resets.

Interface

interface AgentQueue {
  abort: (reason?: unknown) => void
  clear: () => Promise<void>
  getActiveTurnId: () => string | undefined
  interrupt: (reason?: unknown) => MaybePromise<string | undefined>
  isIdle: () => boolean
  send: (item: AgentInput, options?: AgentSignalOptions) => string
  wait: (options?: AgentSignalOptions) => Promise<void>
}

Turn lifecycle

When you call send(), the queue either:

1. Enqueues a new top-level turn if no turn is active.
2. Appends the input to the active turn's pending input buffer and emits turn.input_queued.

The active turn loop:

1. Emits turn.start.
2. Calls the runner for one model step with the current input and cumulative stepNumber.
3. Executes the returned toolCalls and appends their outputs.
4. If the model needs a tool follow-up or new input was queued while the runner was running, drains that input (turn.input_drained) and starts the next step.
5. Emits turn.done on success, turn.failed on error, or turn.aborted if aborted.

The queue owns this loop so tool follow-ups and input submitted with send() share the same boundary. Built-in runners only execute one model request; the Agent layer handles their returned toolCalls uniformly.

Queueing semantics

Top-level turns on the same agent run one at a time. Different agents run concurrently.

import { createAgent, run } from '@apeira/core'
import { responses } from '@apeira/core/responses'

const agent = createAgent({
  instructions: 'You are a helpful assistant.',
  runner: responses({
    apiKey: process.env.OPENAI_API_KEY,
    baseURL: 'https://api.openai.com/v1/',
    model: 'gpt-5.5',
  }),
})

const input = { content: 'Hello.', role: 'user', type: 'message' } as const

const first = run(agent, input)
const second = run(agent, input) // waits for first

run() builds a ReadableStream around send() and subscribe(), filtering events to the submitted turn.

Abort and interrupt

• abort(reason) aborts the active turn without recording a boundary.
• interrupt(reason) aborts the active turn and lets the agent record a <turn_aborted> boundary in storage.
• clear() aborts the active turn and drops all queued turns and pending input. It is used by reset().
• wait() returns a promise that resolves when the queue becomes idle.

import { createAgent } from '@apeira/core'
import { responses } from '@apeira/core/responses'

const agent = createAgent({
  instructions: 'You are a helpful assistant.',
  runner: responses({
    apiKey: process.env.OPENAI_API_KEY,
    baseURL: 'https://api.openai.com/v1/',
    model: 'gpt-5.5',
  }),
})

await agent.wait() // resolves when no turn is running

Design notes

• Pending top-level turns are stored in a yocto-queue; each turn also owns its pending input buffer.
• If the active turn has already been aborted, send() targets the next pending turn when one exists.
• send() is synchronous and returns immediately with a crypto.randomUUID() turn id.
• The queue uses an internal pumping flag so concurrent send() calls share the same pump loop.