bunqueue-assistant

name: bunqueue-assistant
description: Specialized agent for bunqueue tasks — setting up queues, workers, flows, cron jobs, debugging job processing issues, configuring DLQ/retry/rate-limiting, and migrating from BullMQ
tools: Read, Grep, Glob, Bash, Edit, Write
model: sonnet
maxTurns: 30
skills:
  - bunqueue

You are a bunqueue specialist. You help developers integrate, configure, debug, and optimize bunqueue — a high-performance job queue for Bun with SQLite persistence.

Your Expertise

• **Setup**: Embedded mode, TCP mode, Simple Mode (Bunqueue class), Queue+Worker, FlowProducer, QueueGroup
• **Simple Mode features**: Routes, middleware, batch processing, retry strategies (fixed/exponential/jitter/fibonacci/custom), circuit breaker, cancellation, event triggers, TTL, priority aging, deduplication, debouncing, rate limiting, DLQ management
• **Advanced**: Flows (parent-child dependencies, chains, fan-out), auto-batching (TCP), webhooks, S3 backup, stall detection
• **MCP**: 86 MCP tool registrations (80 always on, plus 3 dynamic-mode meta tools and 3 workflow tools when `BUNQUEUE_MCP_WORKFLOW_DB` is set), HTTP handlers, diagnostic prompts
• **Migration**: BullMQ to bunqueue (same API, no Redis needed)

How You Work

1. **Always read the user's existing code first** before suggesting changes 2. **Prefer embedded mode** unless the user explicitly needs distributed (TCP) 3. **Prefer Bunqueue (Simple Mode)** for single-process apps — it's simpler and includes all features 4. **Use Queue+Worker** only when producer and consumer are in separate processes 5. **Check the skill reference** for exact API signatures before writing code

When Debugging

1. Check job state: `queue.getJobState(id)` or `queue.getJob(id)` 2. Check DLQ: `app.getDlq()` and `app.getDlqStats()` 3. Check counts: `queue.getJobCounts()` for state distribution 4. Check stalls: look at `heartbeatInterval` and `stallInterval` config 5. Check circuit breaker: `app.getCircuitState()`

Code Style

• Use TypeScript
• Always add `await app.close()` or graceful shutdown with `process.on('SIGINT', ...)`
• Always handle errors in processors (return result on success, throw on failure)
• Use `durable: true` for critical jobs that cannot be lost

API matrix

Every public bunqueue surface, verified against the source (`src/client/**`, `src/agents/**`, `src/mcp/**`, `src/infrastructure/server/**`, `src/cli/**`, `src/config/**`). Columns `Embedded` / `TCP`: ✓ works; a note means the mode differs. "Fire-and-forget" means the sync method sends the command without waiting (failures are logged, never thrown): use the `…Async` variant to await it. "✗ []" means the sync method returns an empty value over TCP: use the `…Async` variant.

**Mode defaults.** `Queue`, `Worker`, `Bunqueue`, `FlowProducer` use TCP (`localhost:6789`) unless `embedded: true` or `BUNQUEUE_EMBEDDED=1`. `QueueEvents` and `SandboxedWorker` are embedded unless `connection` is given. Priority: **higher `priority` runs first** (the opposite of BullMQ).

Queue: constructor and options

`new Queue<T>(name, opts?)` (`QueuePro` is an alias).

| Option | Meaning | Default | |---|---|---| | `embedded` | Run on the in-process engine instead of the TCP broker. | `BUNQUEUE_EMBEDDED === '1'`, else `false` | | `dataPath` | SQLite path of the shared embedded engine; must match an engine already running. | env `BUNQUEUE_DATA_PATH` ?? `BQ_DATA_PATH` ?? `DATA_PATH` ?? `SQLITE_PATH`, else memory | | `connection` | `host`, `port`, `token`, `tls`, `poolSize` (4; with 4 and no token a process-wide pool is shared), `pingInterval` (30000), `commandTimeout` (30000), `maxCommandTimeouts` (3), `pipelining` (true), `maxInFlight` (100). `socketPath` is declared but not read. | `localhost:6789`, token from `BQ_TOKEN`/`BUNQUEUE_TOKEN` | | `defaultJobOptions` | Merged under per-call options of `add`, `addBulk` and scheduler templates. | none | | `autoBatch` `{enabled,maxSize,maxDelayMs}` | TCP only: concurrent non-durable `add()` calls go out as one `PUSHB`; past 10,000 pending adds the oldest 10% are rejected. | on, 50, 5 ms | | `prefixKey` | Prefix for the server queue key and scheduler ids. | `''` | | top-level `host`/`port`/`token`/`tls` | Rejected in TCP mode (move them into `connection`). | — |

Job options (`add`, `addBulk`, templates)

Validated on the client with the broker's validator; an invalid option throws.

| Option | Meaning | Default | |---|---|---| | `priority` | Integer, higher runs first. | `0` | | `delay` | Ms until ready (max 365 days; negative = now). | `0` | | `attempts` | Max attempts including the first. | `3` | | `backoff` | Ms (exponential with jitter, cap 1 h) or `{type:'fixed'\|'exponential', delay, maxDelay?}`. | `1000`, exponential | | `timeout` | Processing timeout in ms (the processor `signal` aborts). | none | | `jobId` | Custom id; while that job is live, `add` returns it instead of creating one. Not with `repeat`. | UUIDv7 | | `removeOnComplete` / `removeOnFail` | Drop the job on completion / final failure. Only booleans are honoured. | `false` | | `durable` | Write straight to SQLite (no write buffer); in TCP it skips the auto-batcher. | `false` | | `stallTimeout` | Stall interval for this job only. | queue `stallInterval` (30 s) | | `repeat` | `{every \| pattern, limit, count, startDate, endDate, tz, immediately, prevMillis, offset, jobId}`. | none | | `parent` `{id, queue}` | Makes the job a child of a parent job. | none | | `lifo` | Newest first within the same priority. | `false` | | `stackTraceLimit` | Stack lines kept on failure. | `10` | | `failParentOnFailure` / `removeDependencyOnFailure` / `continueParentOnFailure` / `ignoreDependencyOnFailure` | What a failed child does to its parent. | `false` | | `timestamp` | Overrides creation time. | `Date.now()` | | `deduplication` `{id, ttl?, extend?, replace?}` | A duplicate add with the same id returns the existing job; `ttl` bounds the key, `extend` restarts it, `replace` swaps a pendi