> ## Documentation Index
> Fetch the complete documentation index at: https://docs.agentium.in/llms.txt
> Use this file to discover all available pages before exploring further.

# @agentium/queue

> Complete API reference for the queue package

## Installation

```bash theme={null}
npm install @agentium/queue bullmq ioredis
```

Requires a running Redis server.

***

## AgentQueue

The producer that enqueues agent and workflow jobs.

```typescript theme={null}
import { AgentQueue } from "@agentium/queue";
```

### Constructor

```typescript theme={null}
new AgentQueue(config: QueueConfig)
```

<ParamField body="connection" type="{ host: string; port: number }" required>
  Redis connection details.
</ParamField>

<ParamField body="queueName" type="string" default="agentium:jobs">
  Name of the BullMQ queue.
</ParamField>

<ParamField body="defaultJobOptions" type="Record<string, unknown>">
  Default options applied to all jobs (priority, attempts, delay, etc.).
</ParamField>

### Methods

#### enqueueAgentRun

```typescript theme={null}
async enqueueAgentRun(opts: {
  agentName: string;
  input: string;
  sessionId?: string;
  userId?: string;
  priority?: number;
  delay?: number;
}): Promise<{ jobId: string }>
```

Enqueue an agent execution job.

#### enqueueWorkflow

```typescript theme={null}
async enqueueWorkflow(opts: {
  workflowName: string;
  initialState?: Record<string, unknown>;
  sessionId?: string;
  priority?: number;
  delay?: number;
}): Promise<{ jobId: string }>
```

Enqueue a workflow execution job.

#### getJobStatus

```typescript theme={null}
async getJobStatus(jobId: string): Promise<JobStatus>
```

Get the current status of a job.

#### cancelJob

```typescript theme={null}
async cancelJob(jobId: string): Promise<void>
```

Cancel a pending or active job.

#### Event Handlers

```typescript theme={null}
queue.onCompleted((jobId: string, result: RunOutput) => {
  console.log(`Job ${jobId} completed:`, result.text);
});

queue.onFailed((jobId: string, error: Error) => {
  console.error(`Job ${jobId} failed:`, error.message);
});
```

#### close

```typescript theme={null}
async close(): Promise<void>
```

Close the queue connection.

***

## AgentWorker

The consumer that processes jobs from the queue.

```typescript theme={null}
import { AgentWorker } from "@agentium/queue";
```

### Constructor

```typescript theme={null}
new AgentWorker(config: WorkerConfig)
```

<ParamField body="connection" type="{ host: string; port: number }" required>
  Redis connection details.
</ParamField>

<ParamField body="queueName" type="string" default="agentium:jobs">
  Queue name to consume from.
</ParamField>

<ParamField body="concurrency" type="number" default="5">
  Number of jobs processed in parallel.
</ParamField>

<ParamField body="agentRegistry" type="Record<string, Agent>" required>
  Map of agent name to Agent instance.
</ParamField>

<ParamField body="workflowRegistry" type="Record<string, Workflow>">
  Map of workflow name to Workflow instance.
</ParamField>

### Methods

#### start

```typescript theme={null}
start(): void
```

Begin processing jobs from the queue.

#### stop

```typescript theme={null}
async stop(): Promise<void>
```

Gracefully stop the worker.

***

## Types

### JobPayload

```typescript theme={null}
type JobPayload = AgentJobPayload | WorkflowJobPayload;

interface AgentJobPayload {
  type: "agent";
  agentName: string;
  input: string;
  sessionId?: string;
  userId?: string;
}

interface WorkflowJobPayload {
  type: "workflow";
  workflowName: string;
  initialState?: Record<string, unknown>;
  sessionId?: string;
}
```

### JobStatus

```typescript theme={null}
interface JobStatus {
  jobId: string;
  state: "waiting" | "active" | "completed" | "failed" | "delayed";
  progress?: number;
  result?: RunOutput;
  error?: string;
  createdAt: Date;
  processedAt?: Date;
  finishedAt?: Date;
}
```

***

## Utilities

### bridgeEventBusToJob

```typescript theme={null}
import { bridgeEventBusToJob } from "@agentium/queue";

const cleanup = bridgeEventBusToJob(eventBus, job, runId);
// Returns cleanup function to remove listeners
```

Bridges agent `EventBus` events to BullMQ job progress and logs, enabling real-time job status tracking.

***

## Example

```typescript theme={null}
import { Agent, openai } from "@agentium/core";
import { AgentQueue, AgentWorker } from "@agentium/queue";

const agent = new Agent({
  name: "processor",
  model: openai("gpt-4o"),
  instructions: "Process the given data.",
});

// Producer
const queue = new AgentQueue({
  connection: { host: "localhost", port: 6379 },
});

const { jobId } = await queue.enqueueAgentRun({
  agentName: "processor",
  input: "Analyze this dataset",
});

console.log("Enqueued job:", jobId);

// Worker (can be in a separate process)
const worker = new AgentWorker({
  connection: { host: "localhost", port: 6379 },
  agentRegistry: { processor: agent },
  concurrency: 3,
});

worker.start();

// Monitor
queue.onCompleted((id, result) => {
  console.log(`Job ${id}:`, result.text);
});
```