Enqueue and inspect jobs

AgentQueue writes jobs to Redis. A worker must use the same connection and queue name and register the names you enqueue. Use the background-job quickstart for a complete producer and worker.

Submit a job

Save this as producer.ts. The worker must register an agent under assistant.

producer.ts

import { AgentQueue } from "@agentium/queue";

const queue = new AgentQueue({
  connection: process.env.REDIS_URL ?? "redis://127.0.0.1:6379",
  queueName: "agentium-jobs",
  defaultJobOptions: {
    attempts: 3,
    backoff: { type: "exponential", delay: 1000 },
  },
});

try {
  const { jobId } = await queue.enqueueAgentRun({
    agentName: "assistant",
    input: "Summarize the release notes.",
    sessionId: "release-review",
    userId: "local-user",
    tenantId: "local-development",
  });
  console.log("Queued:", jobId);
} finally {
  await queue.close();
}

Closing a producer releases its connections; it does not remove enqueued work. Store the returned job ID in your application if you need to query it later.

Configure delivery

All three enqueue methods accept priority, delay in milliseconds, attempts, and backoff. Agent and team jobs take input; workflow jobs take initialState. See the exact producer signatures. Use defaultJobOptions for queue-wide BullMQ options. Per-job fields override the corresponding defaults. Retries belong here, not on AgentWorker. A retry repeats execution and may repeat an external effect. For recurring work, use schedule() with a stable ID and exactly one agent, team, or workflow target. See schedules for timezone and migration behavior.

Inspect a job

This complete helper can be called with a producer your application owns:

import type { AgentQueue } from "@agentium/queue";

export async function inspectJob(queue: AgentQueue, jobId: string) {
  const status = await queue.getJobStatus(jobId);
  if (status.state === "failed") throw new Error(status.error ?? "Job failed");
  return status;
}

Field	Meaning
`state`	Published union: `waiting`, `active`, `completed`, `failed`, or `delayed`
`progress`	Numeric progress when present. Agent workers count stream chunks; this is not a percentage.
`result`	Stored return value after completion
`error`	Failure reason when present
`createdAt`, `processedAt`, `finishedAt`	JavaScript `Date` values; the last two can be absent

getJobStatus() throws if the job does not exist. Configure retention deliberately: removing completed jobs also removes their queryable results. The published JobStatus.result and onCompleted callback use RunOutput. Agent and team runs return that shape; workflow jobs store their WorkflowResult. When consuming a mixed queue, validate the stored value for the job type your application submitted rather than assuming every result has text.

Observe completion

onCompleted((jobId, result) => ...) and onFailed((jobId, error) => ...) register queue-wide callbacks. Filter by your job ID, keep the producer open while observing, and handle work inside callbacks without leaving rejected promises unobserved. Use status queries to recover state after a consumer reconnects.

Remove a job

await queue.cancelJob(jobId) removes an existing BullMQ job; an absent job is a no-op. Removal can reject when a job is locked by an active worker. It does not abort the agent, roll back effects, or remove a recurring schedule. Use unschedule(id) to stop future scheduled deliveries. The durable queue provides a separate task-cancellation contract.

Build

Agent applications

Harness runtime

Teams and workflows

Data and capabilities

Interfaces and jobs

Voice and telephony

Verify and manage

Enqueue and inspect jobs

Submit a job

Configure delivery

Inspect a job

Observe completion

Remove a job

​Submit a job

​Configure delivery

​Inspect a job

​Observe completion

​Remove a job

Submit a job

Configure delivery

Inspect a job

Observe completion

Remove a job