# Worker Queues

URL: https://cogitator.app/docs/deployment/worker-queues

> Distributed job processing with BullMQ — background execution for agents, workflows, and swarms.

## Overview

The `@cogitator-ai/worker` package provides distributed job processing backed by BullMQ and Redis. It supports background execution of agents, workflows, and swarms with automatic retries, job priorities, and Prometheus metrics for autoscaling.

```bash
pnpm add @cogitator-ai/worker @cogitator-ai/core ioredis
```

## JobQueue

`JobQueue` dispatches serialized agent, workflow and swarm configurations as jobs to Redis. Without options, the queue is named `cogitator-jobs` and jobs get 3 attempts with exponential backoff from 1 s, keeping the last 100 completed and 500 failed jobs.

```typescript
import { JobQueue } from '@cogitator-ai/worker';

const queue = new JobQueue({
  name: 'cogitator-jobs',
  redis: { host: 'localhost', port: 6379 },
  defaultJobOptions: {
    attempts: 3,
    backoff: { type: 'exponential', delay: 1000 },
    removeOnComplete: 100,
    removeOnFail: 500,
  },
});

const job = await queue.addAgentJob(
  {
    name: 'researcher',
    instructions: 'You are a research assistant.',
    model: 'gpt-5.5',
    provider: 'openai', // prepended when `model` names no provider the worker routes to
    tools: [],
  },
  'Summarize the latest AI papers on reasoning',
  { priority: 1, threadId: 'thread_abc123', userId: 'user_42', delay: 5_000 }
);

console.log('State:', await queue.getJobState(job.id!));
```

The worker routes `model` like the same agent in-process: a prefix naming a built-in provider, a backend in the worker Cogitator's `llm.backends` or a registered plugin picks that provider, so `openrouter/deepseek/deepseek-v4-pro` runs on an `openrouter` backend whatever `provider` says. The optional `provider` (any provider the worker routes to, custom backends and plugins included) is prepended only to a model whose prefix names none, such as `meta-llama/llama-4-scout` with `provider: 'openrouter'`. Without it such a model runs on the worker's `llm.defaultProvider`, and a `provider` the worker cannot route to fails the job.

### Queueing an agent you already have

`serializeAgent` turns an `Agent` into a job: its model, instructions, sampling, reasoning effort, tool schemas and response format, with a Zod response schema turned into JSON Schema. The worker turns the schema back into a Zod schema, so a queued run validates and repairs its structured answer like an in-process run, and the job result carries it:

```typescript
import { serializeAgent, type AgentJobResult } from '@cogitator-ai/worker';

const job = await queue.addAgentJob(serializeAgent(chiefEditor), 'Pitches: ...');
// once the job completes
const result = (await queue.getJob(job.id!))?.returnvalue as AgentJobResult;
result.structured; // the validated answer of a json_schema agent, when it fits
result.usage; // { inputTokens, outputTokens, totalTokens, cost, reasoningTokens? }
```

The redis config also takes a `url` (`redis://user:password@host:port/db`, or `rediss://` for TLS); explicit `host`, `port`, `username`, `password`, `db` and `tls` fields override what it says.

`getJobState(id)` returns `'waiting'`, `'prioritized'` (added with a `priority`), `'delayed'`, `'active'`, `'completed'`, `'failed'`, `'waiting-children'` (the parent of a BullMQ flow) or `'unknown'` (the `JobState` type).

`threadId` defaults to a fresh id; `userId` makes the run act for that user (it owns the thread), and `metadata` is stored on the payload. Other queue methods: `getJob(id)`, `getMetrics()`, `pause()`, `resume()`, `clean(grace, limit, state)`, `getQueue()` (the underlying BullMQ `Queue`) and `close()`.

Workflow and swarm jobs follow the same pattern:

```typescript
await queue.addWorkflowJob(workflowConfig, { query: 'input' }, { priority: 2, runId: 'run_1' });
await queue.addSwarmJob(
  { topology: 'debate', agents: [proAgent, conAgent], coordinator: moderatorAgent, maxRounds: 2 },
  'Debate the pros and cons of microservices'
);
```

Workflow jobs execute a serialized DAG (`agent`, `transform`, `condition` and `parallel` nodes over a shared state object); swarm topologies map to strategies (`sequential` → pipeline, `hierarchical` → hierarchical with the coordinator as supervisor, `collaborative` → round-robin, `debate` → debate, `voting` → consensus). See the package README for the full workflow node reference.

## WorkerPool

`WorkerPool` spawns BullMQ workers that consume jobs from the queue. Built-in processors handle agent, workflow, swarm, and distributed swarm-agent job types. Without a `cogitator`, the pool runs agents with `new Cogitator()` (provider keys from the environment).

```typescript
import { Cogitator } from '@cogitator-ai/core';
import { WorkerPool } from '@cogitator-ai/worker';

const pool = new WorkerPool(
  {
    name: 'cogitator-jobs',
    redis: { host: 'localhost', port: 6379 },
    workerCount: 4,
    concurrency: 5,
    cogitator: new Cogitator({
      llm: { providers: { openai: { apiKey: process.env.OPENAI_API_KEY! } } },
    }),
    tools: [webSearch, calculator],
  },
  {
    onJobStarted: (jobId, type) => console.log(`${type} job ${jobId} started`),
    onJobCompleted: (jobId, result) => console.log(`${jobId} done`, result.type),
    onJobFailed: (jobId, error) => console.error(`${jobId} failed:`, error.message),
    onWorkerError: (error) => console.error(error),
  }
);

await pool.start();

process.on('SIGTERM', async () => {
  await pool.stop(30_000); // graceful, force-closes after the timeout
  await queue.close();
});
```

`lockDuration` and `stalledInterval` are passed to BullMQ. `forceStop()` closes workers without waiting; `getWorkerCount()` and `isPoolRunning()` report pool state.

Serialized agents carry tool **schemas** only. The pool resolves tools by name from its `tools` option; jobs that need tools the worker does not provide fail with a clear error instead of running with stubs.

## Distributed Swarm Workers

Swarms with `distributed.enabled` dispatch each agent turn to Redis. Run `DistributedSwarmWorker` on worker nodes to execute them:

```typescript
import { DistributedSwarmWorker } from '@cogitator-ai/worker';

const worker = new DistributedSwarmWorker({
  redis: { host: 'localhost', port: 6379 },
  keyPrefix: 'swarm', // must match the swarm's distributed.redis.keyPrefix
  queue: 'swarm-agent-jobs', // must match distributed.queue
  concurrency: 4, // default 4
  cogitator,
  tools: [webSearch],
});

await worker.start();
```

`redis` also takes `password` and `db`; `pollTimeout` (seconds, default 1) controls how long a blocking poll waits before re-checking state. A second constructor argument takes `onJobStarted`, `onJobCompleted`, `onJobFailed` and `onError` callbacks.

See [Distributed Swarms](/docs/swarms/distributed) for the coordinator side.

## Redis Cluster

Pass cluster nodes to both queue and workers. They connect to every listed node through an ioredis `Cluster`, and cluster mode uses the `{cogitator}` hash tag prefix to keep keys on the same slot.

```typescript
const redisConfig = {
  password: process.env.REDIS_PASSWORD,
  cluster: {
    nodes: [
      { host: 'redis-1.example.com', port: 6379 },
      { host: 'redis-2.example.com', port: 6379 },
    ],
  },
};

const queue = new JobQueue({ redis: redisConfig });
const pool = new WorkerPool({ redis: redisConfig, workerCount: 4, concurrency: 10 });
```

## Prometheus Metrics

`queue.getMetrics()` returns the queue counts; `WorkerPool` records job durations, per-type counts and final failures in `pool.metrics` (a `MetricsCollector`: `recordJob(type, durationMs)` on completion, `recordFailure(type)` when a job fails its last attempt). Combine them in the Prometheus exposition format for monitoring and Kubernetes HPA:

```typescript
app.get('/metrics', async (req, res) => {
  const metrics = await queue.getMetrics();
  res.set('Content-Type', 'text/plain');
  res.send(pool.metrics.format(metrics, { queue: 'cogitator-jobs' }));
});
```

Label values are escaped per the Prometheus exposition format. Without a pool, `formatPrometheusMetrics(await queue.getMetrics(), labels)` renders the queue metrics alone.

| Metric                           | Type      | Description                                                 |
| -------------------------------- | --------- | ----------------------------------------------------------- |
| `cogitator_queue_depth`          | gauge     | Jobs waiting + delayed                                      |
| `cogitator_queue_waiting`        | gauge     | Jobs ready to run, prioritized jobs included                |
| `cogitator_queue_delayed`        | gauge     | Delayed / scheduled jobs                                    |
| `cogitator_queue_active`         | gauge     | Jobs currently processing                                   |
| `cogitator_queue_completed`      | gauge     | Completed jobs kept in Redis (capped by `removeOnComplete`) |
| `cogitator_queue_failed`         | gauge     | Failed jobs kept in Redis (capped by `removeOnFail`)        |
| `cogitator_workers_total`        | gauge     | Workers connected to the queue                              |
| `cogitator_jobs_by_type_total`   | counter   | Jobs processed by type                                      |
| `cogitator_jobs_failed_total`    | counter   | Jobs that failed their last attempt, by type                |
| `cogitator_job_duration_seconds` | histogram | Processing time distribution                                |

`cogitator_queue_completed` and `cogitator_queue_failed` come from BullMQ and drop as old jobs are trimmed, so they are not counters. Alert on `cogitator_jobs_failed_total` instead, for example `increase(cogitator_jobs_failed_total[5m]) > 5`. The per-type counters appear after the first job of that type.

### HPA Autoscaling

Use `cogitator_queue_depth` as the scaling signal in a Kubernetes HorizontalPodAutoscaler (through a custom-metrics adapter such as prometheus-adapter or KEDA):

```yaml
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: cogitator-worker
  minReplicas: 1
  maxReplicas: 20
  metrics:
    - type: Pods
      pods:
        metric:
          name: cogitator_queue_depth
        target:
          type: AverageValue
          averageValue: '10'
```