Worker Queues

OverviewCopy Anchor Link

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.

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

JobQueueCopy Anchor Link

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.

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 haveCopy Anchor Link

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:

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:

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.

WorkerPoolCopy Anchor Link

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).

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 WorkersCopy Anchor Link

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

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 for the coordinator side.

Redis ClusterCopy Anchor Link

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.

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 MetricsCopy Anchor Link

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:

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.

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 AutoscalingCopy Anchor Link

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

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'