feat(queue): Deliver local dev work through BullMQ on Redis

apps/example/scripts/check-vercel-output.mjs

@@ -1,4 +1,4 @@
-import { existsSync, readFileSync, statSync } from "node:fs";
+import { existsSync, readdirSync, readFileSync, statSync } from "node:fs";
 import path from "node:path";
 import { fileURLToPath } from "node:url";
 
@@ -102,6 +102,17 @@ function assertFunctionHasJuniorContent(functionDir, pluginPackages) {
   }
 }
 
+// The BullMQ queue provider is for local dev only. Junior imports it at run
+// time so bundlers skip it. Fail if BullMQ or its msgpackr dependency still
+// reaches a deployed function.
+function assertNoDevQueuePackages(functionDir) {
+  for (const entry of readdirSync(functionDir, { recursive: true })) {
+    if (/(^|[\\/])(bullmq|msgpackr)[^\\/]*$/.test(entry)) {
+      fail(`${path.relative(appRoot, functionDir)} contains ${entry}`);
+    }
+  }
+}
+
 if (existsSync(path.join(appRoot, "api"))) {
   fail(
     "apps/example/api exists; Vercel would route source functions before Nitro",
@@ -119,6 +130,7 @@ if (pluginPackages.length === 0) {
 
 for (const functionDir of [serverFunctionDir, queueFunctionDir]) {
   assertFunctionHasJuniorContent(functionDir, pluginPackages);
+  assertNoDevQueuePackages(functionDir);
 }
 
 console.log(

ast-grep/rules/no-queue-provider-module-imports.yml

@@ -1,7 +1,7 @@
 id: no-queue-provider-module-imports
 language: TypeScript
 severity: error
-message: Only packages/junior/src/chat/queue/provider.ts may import a queue provider module such as chat/queue/vercel.ts. Other code calls getQueueProvider(). See packages/junior/src/chat/queue/README.md.
+message: Only packages/junior/src/chat/queue/provider.ts may import a queue provider module such as chat/queue/vercel.ts or chat/queue/bullmq.ts. Other code calls getQueueProvider(). See packages/junior/src/chat/queue/README.md.
 files:
   - packages/junior/src/**/*.ts
   - packages/junior/src/**/*.tsx
@@ -13,4 +13,4 @@ rule:
     - kind: export_statement
   has:
     kind: string
-    regex: '^["''](@/chat/queue/vercel|(\.{1,2}/)+(.+/)?queue/vercel)["'']$'
+    regex: '^["''](@/chat/queue/(vercel|bullmq)|(\.{1,2}/)+(.+/)?queue/(vercel|bullmq))["'']$'

packages/docs/src/content/docs/contribute/development.md

@@ -41,6 +41,10 @@ pnpm dev
 
 This starts the example app on `http://localhost:3000` by default. It also rebuilds and watches the workspace packages that the example app consumes, so dashboard and runtime package edits are reflected without manually rebuilding first.
 
+`pnpm dev` delivers background work through BullMQ on the Redis in `compose.yaml` (`redis://localhost:6382`). Start it with `docker compose up -d redis`. Local work then needs no Vercel Queues access. To use Vercel Queues, set `JUNIOR_QUEUE_PROVIDER=vercel` and run `pnpm dev:env` for the Vercel token. To use another Redis, set `JUNIOR_QUEUE_REDIS_URL`.
+
+With BullMQ, `pnpm dev` starts one queue worker process next to Nitro. Nitro accepts requests and sends jobs. The worker runs them. A new Junior build restarts both processes. A change to `apps/example/server.ts`, `plugins.ts`, `dashboard.ts` or `apps/example/app/` restarts only the worker. The worker and Nitro must share state, so `pnpm dev` keeps state in that Redis unless you set `JUNIOR_STATE_ADAPTER`. `JUNIOR_STATE_ADAPTER=memory` does not work with BullMQ. Each queue topic runs up to 4 jobs at the same time. Set `JUNIOR_QUEUE_WORKER_CONCURRENCY` to change that number.
+
 To also open the Sentry Local UI in your browser, set `OPEN_SPOTLIGHT_WEB=1`:
 
 ```bash

packages/docs/src/content/docs/reference/config-and-env.md

@@ -51,6 +51,9 @@ related:
 | `JUNIOR_S3_ACCESS_KEY_ID`                   | Conditional | S3 access key ID. Required when the provider is `s3`.                                                                                                                       |
 | `JUNIOR_S3_SECRET_ACCESS_KEY`               | Conditional | S3 secret access key. Required when the provider is `s3`.                                                                                                                   |
 | `JUNIOR_S3_REGION`                          | No          | S3 signing region. Defaults to `us-east-1`.                                                                                                                                 |
+| `JUNIOR_QUEUE_PROVIDER`                     | No          | Queue that delivers background work: `vercel` or `bullmq`. Defaults to `vercel`. Use `bullmq` only for local development.                                                   |
+| `JUNIOR_QUEUE_REDIS_URL`                    | Conditional | Redis for the BullMQ queue, such as `redis://localhost:6382`. Required when the provider is `bullmq`.                                                                       |
+| `JUNIOR_QUEUE_WORKER_CONCURRENCY`           | No          | Jobs that the BullMQ worker runs at the same time for each queue topic. A positive integer. Defaults to `4`.                                                                |
 
 For Vercel deployments, create a private Blob store and connect it to the
 project before using `sendFiles` or `publishImage`. Prefer an OIDC connection.
@@ -68,6 +71,16 @@ visible after you change to another provider.
 Junior loses them when the process stops, and each process has its own copy.
 Do not use it for a deployed app.
 
+`JUNIOR_QUEUE_PROVIDER=bullmq` delivers background work through BullMQ on
+Redis. Local development then needs no Vercel Queues access. The app only
+sends jobs. One separate worker process runs them. `pnpm dev` starts that
+process, and the worker logs `queue.bullmq.worker.started` with its
+concurrency for each queue topic. Because the worker is a separate process,
+state must be in Redis, not memory. Install the optional `bullmq` and `redis` packages to use
+it. Junior loads them only when this provider is selected, so they never enter a
+deployed bundle. Give the queue its own Redis URL. If two dev servers share one
+Redis, each one can take the other's jobs.
+
 Junior applies `JUNIOR_SQL_STATEMENT_TIMEOUT_MS` through PostgreSQL `statement_timeout` for both the Neon and node-postgres drivers. `junior upgrade` does not apply this runtime limit because schema migrations can legitimately take longer.
 
 `JUNIOR_CROSS_ACTOR_MID_RUN_MODE=follow_up` keeps another actor's mid-run ask

packages/junior/package.json

@@ -109,7 +109,17 @@
     "zod": "catalog:"
   },
   "peerDependencies": {
-    "nitro": "3.0.260415-beta || 3.0.260429-beta || 3.0.260522-beta || 3.0.260603-beta || 3.0.260610-beta || 3.0.260903-beta"
+    "bullmq": "^6.5.0",
+    "nitro": "3.0.260415-beta || 3.0.260429-beta || 3.0.260522-beta || 3.0.260603-beta || 3.0.260610-beta || 3.0.260903-beta",
+    "redis": "^5.12.1"
+  },
+  "peerDependenciesMeta": {
+    "bullmq": {
+      "optional": true
+    },
+    "redis": {
+      "optional": true
+    }
   },
   "devDependencies": {
     "@emnapi/core": "^1.10.0",
@@ -123,11 +133,13 @@
     "@types/pg": "^8.15.6",
     "@typescript/typescript6": "6.0.2",
     "@vitest/coverage-v8": "4.1.12",
+    "bullmq": "6.5.0",
     "dependency-cruiser": "^17.4.0",
     "drizzle-kit": "catalog:",
     "msw": "^2.14.6",
     "nitro": "3.0.260522-beta",
     "oxlint": "^1.79.0",
+    "redis": "5.12.1",
     "tsdown": "0.23.1",
     "typescript": "^7.0.2",
     "vercel": "^54.4.0",

packages/junior/src/app.ts

@@ -77,6 +77,7 @@ import {
 } from "@/deployment";
 import {
   consumeConversationQueueMessage,
+  conversationWorkQueueConsumer,
   createVercelConversationWorkCallback,
   registerVercelConversationWorkDevConsumer,
 } from "@/chat/task-execution/vercel-callback";
@@ -89,14 +90,23 @@ import { bindMessageAgent } from "@/chat/agent-invocations/message";
 import { bindSpawnAgent } from "@/chat/agent-invocations/spawn";
 import {
   createVercelPluginTaskCallback,
+  pluginTaskQueueConsumer,
   registerVercelPluginTaskDevConsumer,
   type PluginTaskQueueMessage,
 } from "@/chat/plugins/task-queue";
 import { processPluginTask } from "@/chat/plugins/task-runner";
 import {
   createVercelWorkspaceSnapshotJobCallback,
   registerVercelWorkspaceSnapshotJobDevConsumer,
+  workspaceSnapshotJobQueueConsumer,
 } from "@/chat/sandbox/snapshot/job-callback";
+import {
+  getQueueProvider,
+  type QueueConsumer,
+  type QueueWorker,
+} from "@/chat/queue/provider";
+import { closeDb } from "@/chat/db";
+import { disconnectStateAdapter } from "@/chat/state/adapter";
 import {
   createProductionConversationWorkOptions,
   createProductionSlackWebhookServices,
@@ -410,8 +420,55 @@ function mountRoutes(app: Hono, routes: HostRouteRegistration[]): void {
   }
 }
 
+/** A Hono app from `createApp()` that can also run its queue consumers. */
+export type JuniorApp = Hono & {
+  /**
+   * Pull this app's queue messages in this process until `close()`.
+   *
+   * Only for a queue provider whose consumers pull messages, such as the
+   * local BullMQ queue. The dev server runs it in its own worker process, so
+   * the Nitro process only sends messages. `close()` also closes the database
+   * and state connections of this process.
+   */
+  startQueueWorker(): Promise<QueueWorker>;
+};
+
+async function startQueueWorker(
+  consumers: QueueConsumer[],
+  backgroundWork: Set<Promise<unknown>>,
+): Promise<QueueWorker> {
+  const provider = getQueueProvider();
+  if (!provider.startWorker) {
+    throw new Error(
+      `The ${provider.messagingSystem} queue provider has no queue worker`,
+    );
+  }
+  const worker = await provider.startWorker(consumers);
+  return {
+    // Try every step, so one failure does not leave the other resources open.
+    close: async () => {
+      try {
+        await worker.close();
+      } finally {
+        // A job can end before the work it started with `waitUntil`, such
+        // as title work. Let that work finish before its connections close.
+        while (backgroundWork.size > 0) {
+          await Promise.allSettled(backgroundWork);
+        }
+        try {
+          await disconnectStateAdapter();
+        } finally {
+          await closeDb();
+        }
+      }
+    },
+  };
+}
+
 /** Create a Hono app with all Junior routes. */
-export async function createApp(options?: JuniorAppOptions): Promise<Hono> {
+export async function createApp(
+  options?: JuniorAppOptions,
+): Promise<JuniorApp> {
   const virtualConfig = await resolveVirtualConfig();
   const dashboard = options?.dashboard ?? virtualConfig?.dashboard;
   const configuredPlugins = options?.plugins ?? virtualConfig?.pluginSet;
@@ -539,6 +596,9 @@ export async function createApp(options?: JuniorAppOptions): Promise<Hono> {
     sandbox: { tracePropagation },
     visionContext: { attachmentStorage },
   };
+  // Background work that Conversation work starts. A queue worker waits for
+  // it before it closes its connections.
+  const backgroundWork = new Set<Promise<unknown>>();
   let conversationWorkOptions: ConversationWorkCallbackOptions | undefined;
   const getConversationWorkOptions = (): ConversationWorkCallbackOptions => {
     conversationWorkOptions ??=
@@ -550,7 +610,13 @@ export async function createApp(options?: JuniorAppOptions): Promise<Hono> {
           ? { sendPluginTask: (message) => pluginTaskQueue.send(message) }
           : undefined),
         services: runtimeServiceOverrides,
-        waitUntil: (task) => waitUntil(task),
+        waitUntil: (task) => {
+          backgroundWork.add(task);
+          void task
+            .finally(() => backgroundWork.delete(task))
+            .catch(() => undefined);
+          waitUntil(task);
+        },
       });
     return conversationWorkOptions;
   };
@@ -742,5 +808,15 @@ export async function createApp(options?: JuniorAppOptions): Promise<Hono> {
     });
   });
 
-  return app;
+  return Object.assign(app, {
+    startQueueWorker: () =>
+      startQueueWorker(
+        [
+          conversationWorkQueueConsumer(getConversationWorkOptions()),
+          pluginTaskQueueConsumer(),
+          workspaceSnapshotJobQueueConsumer(),
+        ],
+        backgroundWork,
+      ),
+  });
 }

packages/junior/src/chat/config.ts

@@ -3,6 +3,7 @@ import {
   type AttachmentStorageConfig,
 } from "@/chat/attachments/config";
 import { toOptionalTrimmed } from "@/chat/optional-string";
+import { readQueueConfig, type QueueConfig } from "@/chat/queue/config";
 import { resolveGatewayModel } from "@/chat/pi/client";
 import { normalizeSlackEmojiName } from "@/chat/slack/emoji";
 import { logWarn } from "@/chat/logging";
@@ -131,6 +132,7 @@ export type SqlDriver = "neon" | "postgres";
 export interface ChatConfig {
   attachmentStorage: AttachmentStorageConfig;
   bot: BotConfig;
+  queue: QueueConfig;
   functionMaxDurationSeconds: number;
   conversationWorkEnabled: boolean;
   conversationWorkSoftYieldAfterMs: number;
@@ -612,6 +614,7 @@ export function readChatConfig(
   return {
     attachmentStorage: readAttachmentStorageConfig(env),
     bot: readBotConfig(env, resolvedFunctionMaxDurationSeconds),
+    queue: readQueueConfig(env),
     functionMaxDurationSeconds: resolvedFunctionMaxDurationSeconds,
     conversationWorkEnabled: parseConversationWorkEnabled(
       env.JUNIOR_CONVERSATION_WORK_ENABLED,

packages/junior/src/chat/plugins/task-queue.ts

@@ -10,6 +10,7 @@ import { logWarn } from "@/chat/logging";
 import { queueCallback } from "@/chat/queue/callback";
 import {
   getQueueProvider,
+  type QueueConsumer,
   type QueueMessageMetadata,
 } from "@/chat/queue/provider";
 import {
@@ -139,6 +140,11 @@ export function createVercelPluginTaskCallback(): (
   return pluginTaskCallback().create();
 }
 
+/** Return the queue consumer for plugin tasks, for a pulling queue worker. */
+export function pluginTaskQueueConsumer(): QueueConsumer {
+  return pluginTaskCallback().consumer;
+}
+
 /** Register the local-dev consumer for plugin tasks. */
 export function registerVercelPluginTaskDevConsumer():
   | (() => void)

packages/junior/src/chat/queue/README.md

@@ -14,11 +14,35 @@ Conversation work is different. It only wakes a conversation. Keep using
 ## Provider
 
 - `provider.ts` owns the `QueueProvider` contract and `getQueueProvider()`.
-  Queue code calls `getQueueProvider()`, not a provider module. Today it
-  always returns the Vercel provider.
+  Queue code calls `getQueueProvider()`, not a provider module.
+  `JUNIOR_QUEUE_PROVIDER` selects the provider. The default is `vercel`.
 - `vercel.ts` implements `QueueProvider`. It is the only module that imports
   `@vercel/queue`, and only `provider.ts` imports it. Two ast-grep rules
   enforce these boundaries.
+- `bullmq.ts` implements `QueueProvider` with BullMQ on Redis, for local
+  development only. `pnpm dev` selects it. The Nitro process only sends jobs.
+  Each `send()` opens a Redis client and closes it after the job is added, so
+  the module keeps no connection between calls.
+  `scripts/dev-queue-worker.mjs` runs in its own process, creates the app, and
+  calls `app.startQueueWorker()`. That calls the provider's `startWorker()`,
+  which pulls jobs from Redis. Nothing pushes to the callback routes.
+- The worker from `startWorker()` owns its Redis client and BullMQ workers.
+  Its `close()` waits for running jobs, then closes the client, even when a
+  BullMQ worker fails to close. `app.startQueueWorker()` then closes the state
+  and database connections of the process. `close()` has no time limit. The
+  dev server sends `SIGTERM` to the worker process and kills it if it has not
+  exited after 10 seconds.
+- `registerDevConsumer` is for a provider that delivers each message in the
+  process that sent it, like the Vercel SDK in dev. BullMQ does not implement
+  it.
+- `bullmq.ts` imports `bullmq` and `redis` by a name in a variable, so
+  bundlers do not put them into the production server.
+  `apps/example/scripts/check-vercel-output.mjs` fails if they reach a
+  deployed function.
+- BullMQ is not the same as Vercel Queues. An idempotency key drops a second
+  send only while BullMQ still stores the first job (one hour after it
+  completes). BullMQ renews the lock of a running job, so it does not send a
+  slow job again when `visibilityTimeoutSeconds` runs out.
 - Shared types, such as `QueueMessageMetadata` and `QueueSendOptions`, live in
   `provider.ts`.
 - `nitro.ts` still writes the Vercel `queue/v2beta` triggers into the build