diff --git a/README.md b/README.md
index 5e526f2..a8361ad 100644
--- a/README.md
+++ b/README.md
@@ -230,6 +230,69 @@ await ping(ctx, components.batchWorker, {
 });
 ```
 
+### Sharing a Workpool to limit parallelism
+
+Pass an optional `Workpool` instance to `ping` to limit concurrent batch
+mutations across named workers. Mount the pool alongside BatchWorker in your
+app's `convex.config.ts`:
+
+```ts
+import workpool from "@convex-dev/workpool/convex.config.js";
+
+app.use(workpool);
+```
+
+This requires Workpool's `onFailure` option, currently available in the
+[PR #231 build](https://github.com/get-convex/workpool/pull/231):
+
+```sh
+npm i https://pkg.pr.new/get-convex/workpool/@convex-dev/workpool@231
+```
+
+```ts
+import { Workpool } from "@convex-dev/workpool";
+import { ping } from "@convex-dev/batch-worker";
+
+const pool = new Workpool(components.workpool, { maxParallelism: 10 });
+
+// Inside the mutation that inserts work:
+await ping(ctx, components.batchWorker, {
+  name: `user:${userId}`,
+  workQuery: internal.jobs.getBatch,
+  workerMutation: internal.jobs.processBatch,
+  workpool: pool,
+});
+```
+
+Each pool task represents one loop iteration, carrying only the worker name
+and generation. It fetches its batch when capacity is available, so work added
+while waiting can join the batch. All workers using the same mounted pool
+share its capacity with any other jobs in that pool. Each named worker still
+processes batches serially. The limit covers the batch mutation, not actions
+it schedules separately; `status: running` also includes workers queued for
+capacity.
+
+Pass the pool consistently on every `ping`. Omitting it selects the scheduler.
+Changing pools moves the current continuation and preserves the cursor and
+debounce eligibility. `start` and `kick` reuse the stored pool. Only enqueue
+and cancel function handles and the pool's concurrency/logging options are
+stored; BatchWorker has no runtime dependency on Workpool.
+
+Pooled workers use `onFailure` instead of a monitor or status polling. A failed
+iteration retries after `config.monitorLagMs` (default 60 seconds, minimum 10
+seconds). Workpool's action retry options do not apply to these mutations.
+`stop` invalidates the current generation and requests cancellation, so even
+an admitted stale mutation cannot commit another batch after the stop commits.
+
+Canceling a job directly through the Workpool API does **not** invoke
+`onFailure`. Recover that worker manually:
+
+```ts
+await ctx.runMutation(components.batchWorker.lib.kick, { name: `user:${userId}` });
+```
+
+See [pooled.ts](./example/convex/pooled.ts) for a complete per-user queue example.
+
 ### The cursor
 
 The component stores the cursor for you: whatever your work query returns
@@ -420,7 +483,8 @@ Tips for rate-limiting LLM calls:
 ### Failure handling
 
 If your work query or worker mutation throws, the loop dies and the liveness
-monitor restarts it after ~`monitorLagMs`. The unprocessed rows are still in
+monitor (or Workpool's `onFailure` callback) restarts it after ~`monitorLagMs`.
+The unprocessed rows are still in
 your table and the cursor stayed where it was, so the query hands out the **same
 batch again**. That gives you at-least-once processing, but it also means one
 poison item that always throws can wedge the queue. For work that can fail per
@@ -513,6 +577,25 @@ The demo is hosted on Convex itself via
 published to `https://<deployment>.convex.site`. See
 [example/README.md](./example/README.md#deploy) for the deploy commands.
 
+### Benchmarking execution cadence
+
+For a live comparison of scheduler and Workpool iteration/polling cadence,
+deploy the example to a development deployment and run:
+
+```sh
+npm run build
+npx convex dev --once
+node execution-benchmark.mjs --duration-ms 10000 --trials 3
+```
+
+The benchmark uses a dedicated pool and internal functions. It compares busy
+iterations and empty-queue polling at 0 and 200 ms, with pool capacities of 1
+and 10, and includes ten-worker throughput. Server execution logs provide the
+timings; instrumentation adds no per-iteration database writes. Results,
+raw logs, and a report go to `.context/execution-benchmark/`. For a shorter run,
+append `--pilot --duration-ms 3000 --trials 1`. Run one benchmark at a time:
+its worker names are reused between trials.
+
 ### How it works
 
 | Table         | Written by                          | Read by                 |
diff --git a/example/convex/_generated/api.d.ts b/example/convex/_generated/api.d.ts
index bfd09a4..7b9ee52 100644
--- a/example/convex/_generated/api.d.ts
+++ b/example/convex/_generated/api.d.ts
@@ -11,6 +11,8 @@
 import type * as aggregates from "../aggregates.js";
 import type * as e2e from "../e2e.js";
 import type * as example from "../example.js";
+import type * as executionBenchmark from "../executionBenchmark.js";
+import type * as pooled from "../pooled.js";
 import type * as rateLimited from "../rateLimited.js";
 
 import type {
@@ -23,6 +25,8 @@ declare const fullApi: ApiFromModules<{
   aggregates: typeof aggregates;
   e2e: typeof e2e;
   example: typeof example;
+  executionBenchmark: typeof executionBenchmark;
+  pooled: typeof pooled;
   rateLimited: typeof rateLimited;
 }>;
 
@@ -54,6 +58,8 @@ export declare const internal: FilterApi<
 
 export declare const components: {
   batchWorker: import("@convex-dev/batch-worker/_generated/component.js").ComponentApi<"batchWorker">;
+  workpool: import("@convex-dev/workpool/_generated/component.js").ComponentApi<"workpool">;
+  benchmarkPool: import("@convex-dev/workpool/_generated/component.js").ComponentApi<"benchmarkPool">;
   rateLimiter: import("@convex-dev/rate-limiter/_generated/component.js").ComponentApi<"rateLimiter">;
   staticHosting: import("@convex-dev/static-hosting/_generated/component.js").ComponentApi<"staticHosting">;
 };
diff --git a/example/convex/convex.config.ts b/example/convex/convex.config.ts
index 7785f51..6252457 100644
--- a/example/convex/convex.config.ts
+++ b/example/convex/convex.config.ts
@@ -2,11 +2,16 @@ import { defineApp } from "convex/server";
 import batchWorker from "@convex-dev/batch-worker/convex.config.js";
 import rateLimiter from "@convex-dev/rate-limiter/convex.config.js";
 import staticHosting from "@convex-dev/static-hosting/convex.config.js";
+import workpool from "@convex-dev/workpool/convex.config.js";
 
 // The static-hosting component owns `/` and serves the built demo UI; any app
 // HTTP routes would move under `/api`. This is its fastest serving mode.
 const app = defineApp({ httpPrefix: "/api" });
 app.use(batchWorker);
+// Shared by the per-user workers in pooled.ts.
+app.use(workpool);
+// Isolated from example workloads while comparing execution cadence.
+app.use(workpool, { name: "benchmarkPool" });
 // Used by the rate-limited worker example in rateLimited.ts.
 app.use(rateLimiter);
 // Serves the Vite app in example/src — see `npm run deploy`.
diff --git a/example/convex/executionBenchmark.ts b/example/convex/executionBenchmark.ts
new file mode 100644
index 0000000..0e5bea1
--- /dev/null
+++ b/example/convex/executionBenchmark.ts
@@ -0,0 +1,156 @@
+import { Workpool } from "@convex-dev/workpool";
+import { defineBatchWorkerValidators, ping } from "@convex-dev/batch-worker";
+import { v } from "convex/values";
+import { components, internal } from "./_generated/api.js";
+import { internalMutation, internalQuery } from "./_generated/server.js";
+
+// Console markers instrument the real query path without adding per-iteration
+// writes or turning an idle poll into a batch mutation. The Node runner reads
+// successful execution logs and retains their execution timestamps as well.
+const validators = defineBatchWorkerValidators({
+  cursor: v.number(),
+  batch: { iteration: v.number() },
+});
+
+export const start = internalMutation({
+  args: {
+    runId: v.string(),
+    backend: v.union(v.literal("scheduler"), v.literal("workpool")),
+    mode: v.union(v.literal("busy"), v.literal("poll")),
+    durationMs: v.number(),
+    pollIntervalMs: v.number(),
+    workers: v.number(),
+    maxParallelism: v.number(),
+  },
+  returns: v.object({
+    names: v.array(v.string()),
+    startedAt: v.number(),
+    deadline: v.number(),
+  }),
+  handler: async (ctx, args) => {
+    if (
+      !Number.isFinite(args.durationMs) ||
+      args.durationMs < 1000 ||
+      args.durationMs > 60_000 ||
+      !Number.isInteger(args.workers) ||
+      args.workers < 1 ||
+      args.workers > 10 ||
+      !Number.isInteger(args.maxParallelism) ||
+      args.maxParallelism < 1 ||
+      args.maxParallelism > 10 ||
+      !Number.isFinite(args.pollIntervalMs) ||
+      args.pollIntervalMs < 0 ||
+      args.pollIntervalMs > 1000
+    ) {
+      throw new Error(
+        "Benchmark limits: 1–60s, 1–10 workers/slots, 0–1000ms polls",
+      );
+    }
+    const startedAt = Date.now();
+    const deadline = startedAt + args.durationMs;
+    const pool = new Workpool(components.benchmarkPool, {
+      maxParallelism: args.maxParallelism,
+      logLevel: "ERROR",
+    });
+    const names: string[] = [];
+    for (let i = 0; i < args.workers; i++) {
+      // Reuse names between trials to bound benchmark metadata growth.
+      const name = `benchmark:${args.backend}:${args.mode}:${args.pollIntervalMs}:${i}`;
+      names.push(name);
+      const previous = await ctx.db
+        .query("executionBenchmarks")
+        .withIndex("name", (q) => q.eq("name", name))
+        .unique();
+      const config = {
+        name,
+        runId: args.runId,
+        mode: args.mode,
+        startedAt,
+        deadline,
+        pollIntervalMs: args.pollIntervalMs,
+      };
+      if (previous)
+        await ctx.db.replace("executionBenchmarks", previous._id, config);
+      else await ctx.db.insert("executionBenchmarks", config);
+      await ctx.runMutation(components.batchWorker.lib.stop, { name });
+      await ping(ctx, components.batchWorker, {
+        name,
+        workQuery: internal.executionBenchmark.getBatch,
+        workerMutation: internal.executionBenchmark.processBatch,
+        ...(args.backend === "workpool" ? { workpool: pool } : {}),
+      });
+      await ctx.runMutation(components.batchWorker.lib.setCursor, { name });
+      await ctx.runMutation(components.batchWorker.lib.start, { name });
+    }
+    return { names, startedAt, deadline };
+  },
+});
+
+export const getBatch = internalQuery({
+  args: validators.vQueryArgs,
+  returns: validators.vQueryReturns,
+  handler: async (ctx, { name, cursor }) => {
+    const config = await ctx.db
+      .query("executionBenchmarks")
+      .withIndex("name", (q) => q.eq("name", name))
+      .unique();
+    if (!config || Date.now() >= config.deadline || (cursor ?? 0) >= 10_000) {
+      return { kind: "idle" as const, cooldownMs: 0 };
+    }
+    const phase =
+      config.mode === "poll"
+        ? cursor === undefined
+          ? "bootstrap"
+          : "poll"
+        : "iteration";
+    console.log(
+      "BATCH_BENCH",
+      JSON.stringify({
+        runId: config.runId,
+        name,
+        phase,
+        atMs: Date.now(),
+        iteration: cursor ?? 0,
+      }),
+    );
+    if (phase === "poll") {
+      return {
+        kind: "idle" as const,
+        cooldownMs: config.deadline - config.startedAt + 1000,
+        pollIntervalMs: config.pollIntervalMs,
+      };
+    }
+    // A single bootstrap batch starts the cooldown clock for polling trials.
+    return { batch: { iteration: cursor ?? 0 }, cursor: (cursor ?? 0) + 1 };
+  },
+});
+
+export const processBatch = internalMutation({
+  args: validators.vMutationArgs,
+  returns: validators.vMutationReturns,
+  handler: async () => null,
+});
+
+export const status = internalQuery({
+  args: { names: v.array(v.string()) },
+  returns: v.boolean(),
+  handler: async (ctx, { names }) => {
+    for (const name of names) {
+      const state = await ctx.runQuery(components.batchWorker.lib.status, {
+        name,
+      });
+      if (state?.kind === "running") return false;
+    }
+    return true;
+  },
+});
+
+export const stop = internalMutation({
+  args: { names: v.array(v.string()) },
+  returns: v.null(),
+  handler: async (ctx, { names }) => {
+    for (const name of names)
+      await ctx.runMutation(components.batchWorker.lib.stop, { name });
+    return null;
+  },
+});
diff --git a/example/convex/pooled.ts b/example/convex/pooled.ts
new file mode 100644
index 0000000..39d4106
--- /dev/null
+++ b/example/convex/pooled.ts
@@ -0,0 +1,88 @@
+import { Workpool } from "@convex-dev/workpool";
+import { defineBatchWorkerValidators, ping } from "@convex-dev/batch-worker";
+import { v } from "convex/values";
+import { components, internal } from "./_generated/api.js";
+import { internalMutation, internalQuery } from "./_generated/server.js";
+
+const pool = new Workpool(components.workpool, { maxParallelism: 10 });
+const validators = defineBatchWorkerValidators({
+  batch: {
+    user: v.string(),
+    events: v.array(v.object({ id: v.id("userEvents"), value: v.number() })),
+  },
+});
+
+// Internal demo entry point. A public endpoint should derive the user from auth.
+// npx convex run pooled:addEvents '{"user":"alice","values":[1,2,3]}'
+export const addEvents = internalMutation({
+  args: { user: v.string(), values: v.array(v.number()) },
+  returns: v.null(),
+  handler: async (ctx, { user, values }) => {
+    for (const value of values) {
+      await ctx.db.insert("userEvents", {
+        user,
+        value,
+        insertedAt: ctx.db.vars.commitTs,
+      });
+    }
+    await ping(ctx, components.batchWorker, {
+      name: `user:${user}`,
+      workQuery: internal.pooled.getBatch,
+      workerMutation: internal.pooled.processBatch,
+      workpool: pool,
+    });
+    return null;
+  },
+});
+
+export const getBatch = internalQuery({
+  args: validators.vQueryArgs,
+  returns: validators.vQueryReturns,
+  handler: async (ctx, { name, cursor }) => {
+    const user = name.slice("user:".length);
+    const events = await ctx.db
+      .query("userEvents")
+      .withIndex("user_insertedAt", (q) =>
+        q.eq("user", user).gte("insertedAt", cursor ?? 0n),
+      )
+      .take(100);
+    if (!events.length) return { kind: "idle" as const };
+    return {
+      batch: {
+        user,
+        events: events.map((event) => ({ id: event._id, value: event.value })),
+      },
+      cursor: events.at(-1)!.insertedAt,
+    };
+  },
+});
+
+export const processBatch = internalMutation({
+  args: validators.vMutationArgs,
+  returns: validators.vMutationReturns,
+  handler: async (ctx, { user, events }) => {
+    const previous = await ctx.db
+      .query("userTotals")
+      .withIndex("user", (q) => q.eq("user", user))
+      .unique();
+    const total =
+      (previous?.total ?? 0) +
+      events.reduce((sum, event) => sum + event.value, 0);
+    if (previous) await ctx.db.patch("userTotals", previous._id, { total });
+    else await ctx.db.insert("userTotals", { user, total });
+    for (const event of events) await ctx.db.delete("userEvents", event.id);
+    return null;
+  },
+});
+
+export const getTotal = internalQuery({
+  args: { user: v.string() },
+  returns: v.number(),
+  handler: async (ctx, { user }) => {
+    const total = await ctx.db
+      .query("userTotals")
+      .withIndex("user", (q) => q.eq("user", user))
+      .unique();
+    return total?.total ?? 0;
+  },
+});
diff --git a/example/convex/schema.ts b/example/convex/schema.ts
index 244a190..529483f 100644
--- a/example/convex/schema.ts
+++ b/example/convex/schema.ts
@@ -2,6 +2,24 @@ import { defineSchema, defineTable } from "convex/server";
 import { v } from "convex/values";
 
 export default defineSchema({
+  executionBenchmarks: defineTable({
+    name: v.string(),
+    runId: v.string(),
+    mode: v.union(v.literal("busy"), v.literal("poll")),
+    startedAt: v.number(),
+    deadline: v.number(),
+    pollIntervalMs: v.number(),
+  }).index("name", ["name"]),
+  // Independent per-user queues sharing one Workpool (see pooled.ts).
+  userEvents: defineTable({
+    user: v.string(),
+    value: v.number(),
+    insertedAt: v.commitTs(),
+  }).index("user_insertedAt", ["user", "insertedAt"]),
+  userTotals: defineTable({ user: v.string(), total: v.number() }).index(
+    "user",
+    ["user"],
+  ),
   // A simple work queue: each row is one event to be summed.
   //
   // `insertedAt` is written as `ctx.db.vars.commitTs` and resolves, when the
diff --git a/example/convex/setup.test.ts b/example/convex/setup.test.ts
index eb796c4..fddee7e 100644
--- a/example/convex/setup.test.ts
+++ b/example/convex/setup.test.ts
@@ -4,6 +4,7 @@ import { convexTest } from "convex-test";
 import schema from "./schema.js";
 import component from "@convex-dev/batch-worker/test";
 import rateLimiter from "@convex-dev/rate-limiter/test";
+import workpool from "@convex-dev/workpool/test";
 
 const modules = import.meta.glob("./**/*.*s");
 // When users want to write tests that use your component, they need to
@@ -12,6 +13,7 @@ export function initConvexTest() {
   const t = convexTest(schema, modules);
   component.register(t);
   rateLimiter.register(t);
+  workpool.register(t);
   return t;
 }
 
diff --git a/execution-benchmark.mjs b/execution-benchmark.mjs
new file mode 100644
index 0000000..9676a2b
--- /dev/null
+++ b/execution-benchmark.mjs
@@ -0,0 +1,438 @@
+#!/usr/bin/env node
+// Live scheduler/Workpool cadence comparison. Internal functions are invoked
+// through the authenticated Convex CLI; no admin key is extracted or persisted.
+import { spawn, execFile } from "node:child_process";
+import { promisify } from "node:util";
+import { readFileSync, mkdirSync, writeFileSync } from "node:fs";
+import { parseArgs } from "node:util";
+import { createRequire } from "node:module";
+
+const { values } = parseArgs({
+  options: {
+    "duration-ms": { type: "string", default: "10000" },
+    trials: { type: "string", default: "3" },
+    out: { type: "string", default: ".context/execution-benchmark" },
+    pilot: { type: "boolean", default: false },
+    summarize: { type: "boolean", default: false },
+  },
+});
+if (values.summarize) {
+  writeReport(
+    JSON.parse(readFileSync(`${values.out}/results.json`, "utf8")),
+    values.out,
+  );
+  process.exit(0);
+}
+const durationMs = Number(values["duration-ms"]);
+const trials = Number(values.trials);
+if (
+  !Number.isInteger(durationMs) ||
+  durationMs < 2000 ||
+  durationMs > 60_000 ||
+  !Number.isInteger(trials) ||
+  trials < 1 ||
+  trials > 10
+) {
+  throw new Error("Use duration-ms 2000–60000 and trials 1–10");
+}
+const envFile = readFileSync(".env.local", "utf8");
+const deployment =
+  process.env.CONVEX_DEPLOYMENT ??
+  envFile.match(/^CONVEX_DEPLOYMENT=([^\s#]+)/m)?.[1];
+if (!deployment?.startsWith("dev:"))
+  throw new Error("This benchmark requires a configured dev deployment");
+const deploymentName = deployment.slice(4);
+const cli = "node_modules/convex/bin/main.js";
+const exec = promisify(execFile);
+const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
+const out = values.out;
+mkdirSync(out, { recursive: true });
+async function run(fn, args) {
+  const { stdout } = await exec(
+    process.execPath,
+    [
+      cli,
+      "run",
+      "--deployment",
+      deploymentName,
+      `executionBenchmark:${fn}`,
+      JSON.stringify(args),
+    ],
+    { maxBuffer: 2_000_000 },
+  );
+  return stdout.trim() ? JSON.parse(stdout) : null;
+}
+
+const raw = [];
+let buffer = "";
+const logs = spawn(
+  process.execPath,
+  [
+    cli,
+    "logs",
+    "--deployment",
+    deploymentName,
+    "--history",
+    "1",
+    "--success",
+    "--jsonl",
+  ],
+  { stdio: ["ignore", "pipe", "pipe"] },
+);
+let ready;
+const connected = new Promise((resolve) => {
+  ready = resolve;
+});
+let logErrors = "";
+logs.stderr.on("data", (chunk) => {
+  logErrors += chunk;
+});
+logs.stdout.on("data", (chunk) => {
+  buffer += chunk.toString();
+  let newline;
+  while ((newline = buffer.indexOf("\n")) !== -1) {
+    const line = buffer.slice(0, newline);
+    buffer = buffer.slice(newline + 1);
+    try {
+      raw.push(JSON.parse(line));
+      ready();
+    } catch {
+      /* CLI notices */
+    }
+  }
+});
+
+function completions() {
+  const unique = new Map();
+  for (const event of raw)
+    if (event.kind === "Completion") unique.set(event.executionId, event);
+  return [...unique.values()];
+}
+function markers(events, runId) {
+  const points = [];
+  for (const event of events) {
+    if (event.error || event.willRetry || event.parentExecutionId) continue;
+    for (const line of event.logLines ?? []) {
+      const message = (line.messages ?? []).join(" ");
+      if (!message.includes("BATCH_BENCH")) continue;
+      const marker = JSON.parse(
+        message.slice(message.indexOf("{"), message.lastIndexOf("}") + 1),
+      );
+      if (marker.runId !== runId) continue;
+      points.push({
+        ...marker,
+        startMs: event.executionTimestamp * 1000,
+        finishedMs: event.timestamp * 1000,
+        executionMs: event.executionTime * 1000,
+        executionId: event.executionId,
+        requestId: event.requestId,
+      });
+    }
+  }
+  return points;
+}
+function stats(nums) {
+  if (!nums.length) return null;
+  const sorted = [...nums].sort((a, b) => a - b);
+  const percentile = (p) =>
+    sorted[Math.min(sorted.length - 1, Math.ceil(p * sorted.length) - 1)];
+  return {
+    n: nums.length,
+    mean: nums.reduce((a, b) => a + b, 0) / nums.length,
+    p50: percentile(0.5),
+    p95: percentile(0.95),
+    min: sorted[0],
+    max: sorted.at(-1),
+  };
+}
+function summarize(config, started, points, events) {
+  const warmupMs = Math.min(1000, durationMs / 4);
+  const phase = config.mode === "busy" ? "iteration" : "poll";
+  const measured = points.filter(
+    (p) =>
+      p.phase === phase &&
+      p.startMs >= started.startedAt + warmupMs &&
+      p.startMs < started.deadline,
+  );
+  const gaps = [],
+    queueGaps = [];
+  const byWorker = started.names.map((name) => {
+    const ordered = measured
+      .filter((p) => p.name === name)
+      .sort((a, b) => a.startMs - b.startMs);
+    if (ordered.length < 3)
+      throw new Error(
+        `Too few successful samples for ${name}: ${ordered.length}`,
+      );
+    for (let i = 1; i < ordered.length; i++) {
+      gaps.push(ordered[i].startMs - ordered[i - 1].startMs);
+      queueGaps.push(ordered[i].startMs - ordered[i - 1].finishedMs);
+    }
+    if (phase === "iteration")
+      for (let i = 1; i < ordered.length; i++) {
+        if (ordered[i].iteration !== ordered[i - 1].iteration + 1)
+          throw new Error("Missing/duplicate iteration logs");
+      }
+    return {
+      name,
+      samples: ordered.length,
+      rate:
+        ((ordered.length - 1) * 1000) /
+        (ordered.at(-1).startMs - ordered[0].startMs),
+    };
+  });
+  const functionCalls = {};
+  for (const event of events) {
+    const t = event.executionTimestamp * 1000;
+    if (
+      t < started.startedAt + warmupMs ||
+      t >= started.deadline ||
+      event.caller !== "Scheduler"
+    )
+      continue;
+    if (
+      !(
+        event.componentPath === "batchWorker" ||
+        event.componentPath?.startsWith("benchmarkPool")
+      )
+    )
+      continue;
+    const key = `${event.componentPath}/${event.identifier}`;
+    functionCalls[key] = (functionCalls[key] ?? 0) + 1;
+  }
+  return {
+    ...config,
+    ...started,
+    warmupMs,
+    successfulSamples: measured.length,
+    cadenceMs: stats(gaps),
+    executionMs: stats(measured.map((p) => p.executionMs)),
+    queueGapMs: stats(queueGaps),
+    aggregateIterationsPerSec: byWorker.reduce((n, w) => n + w.rate, 0),
+    byWorker,
+    functionCalls,
+    failedExecutions: events.filter(
+      (e) =>
+        e.error &&
+        e.executionTimestamp * 1000 >= started.startedAt &&
+        e.executionTimestamp * 1000 < started.deadline,
+    ).length,
+    points,
+  };
+}
+
+const backends = [
+  { backend: "scheduler", maxParallelism: 10 },
+  { backend: "workpool", maxParallelism: 10 },
+  { backend: "workpool", maxParallelism: 1 },
+];
+const workloads = [
+  { mode: "busy", pollIntervalMs: 0, workers: 1 },
+  { mode: "poll", pollIntervalMs: 0, workers: 1 },
+  { mode: "poll", pollIntervalMs: 200, workers: 1 },
+  { mode: "busy", pollIntervalMs: 0, workers: 10 },
+];
+const cases = workloads.flatMap((workload) =>
+  backends
+    .filter((b) => workload.workers === 1 || b.maxParallelism === 10)
+    .map((backend) => ({ ...workload, ...backend })),
+);
+const selected = values.pilot
+  ? cases.filter(
+      (c) =>
+        c.workers === 1 && c.maxParallelism === 10 && c.pollIntervalMs === 0,
+    )
+  : cases;
+const results = {
+  deployment: deploymentName,
+  startedAt: new Date().toISOString(),
+  durationMs,
+  trials,
+  packageVersions: {
+    convex: JSON.parse(readFileSync("node_modules/convex/package.json", "utf8"))
+      .version,
+    workpool: JSON.parse(
+      readFileSync("node_modules/@convex-dev/workpool/package.json", "utf8"),
+    ).version,
+    workpoolDriver: createRequire(
+      createRequire(import.meta.url).resolve(
+        "@convex-dev/workpool/package.json",
+      ),
+    )("@convex-dev/batch-worker/package.json").version,
+  },
+  runs: [],
+};
+let activeNames = [];
+function save() {
+  writeFileSync(`${out}/results.json`, JSON.stringify(results, null, 2));
+  // Retain raw benchmark-related completion events for reproducibility.
+  const relevant = completions().filter(
+    (e) =>
+      e.componentPath === "batchWorker" ||
+      e.componentPath?.startsWith("benchmarkPool") ||
+      e.identifier.startsWith("executionBenchmark:"),
+  );
+  writeFileSync(
+    `${out}/executions.jsonl`,
+    relevant.map((e) => JSON.stringify(e)).join("\n") + "\n",
+  );
+}
+try {
+  await Promise.race([
+    connected,
+    sleep(10_000).then(() => {
+      throw new Error(`Log stream did not connect: ${logErrors}`);
+    }),
+  ]);
+  for (let trial = 0; trial < trials; trial++) {
+    // Alternate execution order to reduce time/order bias.
+    const ordered = trial % 2 ? [...selected].reverse() : selected;
+    for (const config of ordered) {
+      const runId = `${Date.now()}-${trial}-${config.backend}-${config.mode}-${config.pollIntervalMs}-${config.workers}-${config.maxParallelism}`;
+      console.log(
+        `Starting trial ${trial + 1}: ${config.backend} slots=${config.maxParallelism}, ${config.mode}/${config.pollIntervalMs}ms, workers=${config.workers}`,
+      );
+      const started = await run("start", { ...config, durationMs, runId });
+      activeNames = started.names;
+      await sleep(durationMs + 300);
+      const timeout = Date.now() + 15_000;
+      while (!(await run("status", { names: activeNames }))) {
+        if (Date.now() >= timeout)
+          throw new Error("Benchmark did not become idle");
+        await sleep(500);
+      }
+      activeNames = [];
+      await sleep(2000); // finish pool completions and flush the log stream
+      const events = completions();
+      const result = summarize(
+        { ...config, runId, trial: trial + 1 },
+        started,
+        markers(events, runId),
+        events,
+      );
+      results.runs.push(result);
+      save();
+      console.log(
+        `  ${result.successfulSamples} samples; interval p50=${result.cadenceMs.p50.toFixed(1)}ms p95=${result.cadenceMs.p95.toFixed(1)}ms; aggregate=${result.aggregateIterationsPerSec.toFixed(1)}/s`,
+      );
+    }
+  }
+} finally {
+  try {
+    if (activeNames.length) await run("stop", { names: activeNames });
+  } finally {
+    logs.kill("SIGTERM");
+    save();
+  }
+}
+writeReport(results, out);
+console.log(`Results: ${out}/results.json`);
+
+function writeReport(results, output) {
+  const groups = new Map();
+  for (const run of results.runs) {
+    const key = [
+      run.mode,
+      run.pollIntervalMs,
+      run.workers,
+      run.backend,
+      run.maxParallelism,
+    ].join(":");
+    if (!groups.has(key)) groups.set(key, []);
+    groups.get(key).push(run);
+  }
+  const rows = [...groups.values()].map((runs) => {
+    const gaps = [],
+      execution = [],
+      queueGaps = [];
+    for (const run of runs) {
+      const phase = run.mode === "busy" ? "iteration" : "poll";
+      for (const name of run.names) {
+        const points = run.points
+          .filter(
+            (p) =>
+              p.name === name &&
+              p.phase === phase &&
+              p.startMs >= run.startedAt + run.warmupMs &&
+              p.startMs < run.deadline,
+          )
+          .sort((a, b) => a.startMs - b.startMs);
+        execution.push(...points.map((p) => p.executionMs));
+        for (let i = 1; i < points.length; i++) {
+          gaps.push(points[i].startMs - points[i - 1].startMs);
+          queueGaps.push(points[i].startMs - points[i - 1].finishedMs);
+        }
+      }
+    }
+    const { mode, pollIntervalMs, workers, backend, maxParallelism } = runs[0];
+    const samples = runs.reduce((n, r) => n + r.successfulSamples, 0);
+    return {
+      mode,
+      pollIntervalMs,
+      workers,
+      backend,
+      maxParallelism,
+      trials: runs.length,
+      samples,
+      cadenceMs: stats(gaps),
+      executionMs: stats(execution),
+      queueGapMs: stats(queueGaps),
+      aggregateIterationsPerSec: stats(
+        runs.map((r) => r.aggregateIterationsPerSec),
+      ),
+      scheduledExecutionsPerIteration:
+        runs.reduce(
+          (n, r) =>
+            n + Object.values(r.functionCalls).reduce((a, b) => a + b, 0),
+          0,
+        ) / samples,
+      failedExecutions: runs.reduce((n, r) => n + r.failedExecutions, 0),
+    };
+  });
+  const summary = {
+    ...results,
+    rows,
+    runs: results.runs.map((run) => {
+      const summary = { ...run };
+      delete summary.points;
+      return summary;
+    }),
+  };
+  writeFileSync(`${output}/summary.json`, JSON.stringify(summary, null, 2));
+  const lines = [
+    "# Scheduler vs Workpool execution cadence",
+    "",
+    `Deployment: ${results.deployment}. Started: ${results.startedAt}.`,
+    "",
+    `${results.trials} trials per case, ${results.durationMs / 1000}s each, first ${(results.runs[0]?.warmupMs ?? 0) / 1000}s excluded for warmup. Cases run sequentially; order reversed on alternating trials.`,
+    "",
+    "| Workload | Workers | Backend / slots | Samples | Median gap (ms) | p95 gap (ms) | Aggregate iterations/s | Scheduled executions/iteration |",
+    "| --- | ---: | --- | ---: | ---: | ---: | ---: | ---: |",
+    ...rows.map(
+      (r) =>
+        `| ${r.mode === "busy" ? "Busy" : `Poll ${r.pollIntervalMs}ms`} | ${r.workers} | ${r.backend}${r.backend === "workpool" ? ` / ${r.maxParallelism}` : ""} | ${r.samples} | ${r.cadenceMs.p50.toFixed(1)} | ${r.cadenceMs.p95.toFixed(1)} | ${r.aggregateIterationsPerSec.mean.toFixed(1)} | ${r.scheduledExecutionsPerIteration.toFixed(2)} |`,
+    ),
+    "",
+    "## Method",
+    "",
+    "These are minimal-loop measurements, not application throughput. Each query reads one fixed benchmark configuration row and logs a marker. Busy iterations call a no-op worker mutation and advance the existing cursor. Polling iterations return idle with the requested interval; one bootstrap batch starts the cooldown clock and is excluded. No per-iteration instrumentation writes are added.",
+    "",
+    "Intervals use successful outer mutation execution timestamps from Convex logs (scheduler loop or Workpool wrapper), grouped by named worker. Median and p95 pool all within-trial, within-worker intervals. Rates average the per-trial sum of worker rates. Scheduled-execution counts include the loop/wrapper, Workpool driver and completion functions; they exclude nested runQuery/runMutation calls and are not billing totals. The harness verifies contiguous busy iteration numbers and all workers return to idle.",
+    "",
+    "Both paths use the same app callbacks and deployment. Workpool uses a dedicated component, with no unrelated queued work. Timings include scheduler delay and database overhead, but exclude client request latency and the warmup window. Console logging and the fixed config read add equal instrumentation work to both paths. Results depend on deployment load and are not a service guarantee.",
+    "",
+    "## Reproduce",
+    "",
+    "```sh",
+    "npm ci",
+    "npm run build",
+    "npx convex dev --once",
+    "node execution-benchmark.mjs --duration-ms 10000 --trials 3",
+    "```",
+    "",
+    "The runner requires a dev deployment, uses authenticated CLI calls to internal functions, and bounds each run to at most 60 seconds and 10 workers. Raw execution logs and timestamps are written alongside this report as executions.jsonl and results.json. Regenerate the report with `node execution-benchmark.mjs --summarize --out <output-directory>`.",
+    "",
+  ];
+  writeFileSync(`${output}/report.md`, lines.join("\n"));
+  console.log(lines.slice(6, 8 + rows.length).join("\n"));
+}
diff --git a/package-lock.json b/package-lock.json
index e2aff7c..5020921 100644
--- a/package-lock.json
+++ b/package-lock.json
@@ -15,6 +15,7 @@
         "@convex-dev/eslint-plugin": "^2.0.0",
         "@convex-dev/rate-limiter": "^0.3.2",
         "@convex-dev/static-hosting": "^0.2.0-alpha.2",
+        "@convex-dev/workpool": "https://pkg.pr.new/get-convex/workpool/@convex-dev/workpool@231",
         "@edge-runtime/vm": "^5.0.0",
         "@eslint/eslintrc": "^3.3.5",
         "@eslint/js": "9.39.4",
@@ -342,6 +343,17 @@
         "node": ">=6.9.0"
       }
     },
+    "node_modules/@convex-dev/batch-worker": {
+      "version": "0.2.2",
+      "resolved": "https://registry.npmjs.org/@convex-dev/batch-worker/-/batch-worker-0.2.2.tgz",
+      "integrity": "sha512-3knigElzTVhWw1sN28EYkGBTdrJZ03vMj2PJPuFq3V0Tg41i7tcv/EWDBFQWcUaKzBWWVw693hcJtxCF3CRtEw==",
+      "dev": true,
+      "license": "Apache-2.0",
+      "peerDependencies": {
+        "convex": "^1.36.1",
+        "react": "^18.3.1 || ^19.0.0"
+      }
+    },
     "node_modules/@convex-dev/eslint-plugin": {
       "version": "2.0.0",
       "resolved": "https://registry.npmjs.org/@convex-dev/eslint-plugin/-/eslint-plugin-2.0.0.tgz",
@@ -386,6 +398,20 @@
         "react": "^18.3.1 || ^19.0.0"
       }
     },
+    "node_modules/@convex-dev/workpool": {
+      "version": "0.4.11",
+      "resolved": "https://pkg.pr.new/get-convex/workpool/@convex-dev/workpool@231",
+      "integrity": "sha512-vyBfL5uyLY/85AtOhfmfwN0OyxeVQJu08naAAG3LQ5E+T0frRE0/v12BQ5Y8H1+xgd2o95onTnxpllTRXrUIGA==",
+      "dev": true,
+      "license": "Apache-2.0",
+      "dependencies": {
+        "@convex-dev/batch-worker": "^0.2.2"
+      },
+      "peerDependencies": {
+        "convex": "^1.36.1",
+        "convex-helpers": "^0.1.94"
+      }
+    },
     "node_modules/@edge-runtime/primitives": {
       "version": "6.0.0",
       "resolved": "https://registry.npmjs.org/@edge-runtime/primitives/-/primitives-6.0.0.tgz",
diff --git a/package.json b/package.json
index 5f0d1e6..719ee11 100644
--- a/package.json
+++ b/package.json
@@ -70,6 +70,7 @@
     "@convex-dev/eslint-plugin": "^2.0.0",
     "@convex-dev/rate-limiter": "^0.3.2",
     "@convex-dev/static-hosting": "^0.2.0-alpha.2",
+    "@convex-dev/workpool": "https://pkg.pr.new/get-convex/workpool/@convex-dev/workpool@231",
     "@edge-runtime/vm": "^5.0.0",
     "@eslint/eslintrc": "^3.3.5",
     "@eslint/js": "9.39.4",
diff --git a/src/client/index.ts b/src/client/index.ts
index d2a741f..575811a 100644
--- a/src/client/index.ts
+++ b/src/client/index.ts
@@ -13,8 +13,24 @@ import {
   type Config,
   type DefaultCursor,
   type WorkerResult,
+  type WorkpoolConfig,
 } from "../component/shared.js";
 
+/**
+ * The portion of a Workpool instance used by BatchWorker. Kept structural so
+ * scheduler-only users don't need to install Workpool, even for its types.
+ * Requires a Workpool version supporting `onFailure`.
+ */
+export type BatchWorkpool = {
+  component: {
+    lib: {
+      enqueue: FunctionReference<"mutation", "internal">;
+      cancel: FunctionReference<"mutation", "internal">;
+    };
+  };
+  options: Pick<WorkpoolConfig, "maxParallelism" | "logLevel">;
+};
+
 export {
   defineBatchWorkerValidators,
   vBatchQueryArgs,
@@ -85,17 +101,41 @@ export async function ping<
     >;
     /** Loop configuration. */
     config?: Partial<Config>;
+    /**
+     * Execute each iteration in this shared Workpool. Batches are fetched
+     * only when admitted. Omit to use the scheduler directly.
+     * Failures retry via onFailure; direct pool cancellations need `lib.kick`.
+     */
+    workpool?: BatchWorkpool;
   },
 ): Promise<void> {
   const [workQuery, workerMutation] = await Promise.all([
     createFunctionHandle(args.workQuery),
     createFunctionHandle(args.workerMutation),
   ]);
+  let workpool: WorkpoolConfig | undefined;
+  if (args.workpool) {
+    const [enqueue, cancel] = await Promise.all([
+      createFunctionHandle(args.workpool.component.lib.enqueue),
+      createFunctionHandle(args.workpool.component.lib.cancel),
+    ]);
+    workpool = {
+      enqueue,
+      cancel,
+      ...(args.workpool.options.maxParallelism !== undefined
+        ? { maxParallelism: args.workpool.options.maxParallelism }
+        : {}),
+      ...(args.workpool.options.logLevel !== undefined
+        ? { logLevel: args.workpool.options.logLevel }
+        : {}),
+    };
+  }
   await ctx.runMutation(component.lib.ping, {
     name: args.name,
     workQuery,
     workerMutation,
     config: args.config ?? {},
+    ...(workpool ? { workpool } : {}),
   });
 }
 
diff --git a/src/client/workpool.test.ts b/src/client/workpool.test.ts
new file mode 100644
index 0000000..fc2e71a
--- /dev/null
+++ b/src/client/workpool.test.ts
@@ -0,0 +1,485 @@
+/// <reference types="vite/client" />
+import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
+import {
+  anyApi,
+  componentsGeneric,
+  defineSchema,
+  defineTable,
+  internalMutationGeneric,
+  internalQueryGeneric,
+  type ApiFromModules,
+  type DataModelFromSchemaDefinition,
+  type QueryBuilder,
+  type MutationBuilder,
+} from "convex/server";
+import { v } from "convex/values";
+import { convexTest } from "convex-test";
+import {
+  Workpool,
+  type WorkpoolComponent,
+  type WorkId,
+} from "@convex-dev/workpool";
+import workpoolTest from "@convex-dev/workpool/test";
+import batchWorkerTest from "../test.js";
+import { defineBatchWorkerValidators, ping } from "./index.js";
+import type { ComponentApi } from "../component/_generated/component.js";
+import type { Doc } from "../component/_generated/dataModel.js";
+import type { internal as batchInternal } from "../component/_generated/api.js";
+
+const schema = defineSchema({
+  items: defineTable({ name: v.string(), seq: v.int64() }).index("name_seq", [
+    "name",
+    "seq",
+  ]),
+  batches: defineTable({ name: v.string(), seqs: v.array(v.int64()) }),
+  behavior: defineTable({
+    name: v.string(),
+    failQuery: v.optional(v.boolean()),
+    failMutation: v.optional(v.boolean()),
+    timeoutMs: v.optional(v.number()),
+    debounceMs: v.optional(v.number()),
+  }).index("name", ["name"]),
+});
+type DataModel = DataModelFromSchemaDefinition<typeof schema>;
+const query = internalQueryGeneric as QueryBuilder<DataModel, "internal">;
+const mutation = internalMutationGeneric as MutationBuilder<
+  DataModel,
+  "internal"
+>;
+const validators = defineBatchWorkerValidators({
+  cursor: v.int64(),
+  batch: {
+    name: v.string(),
+    items: v.array(v.object({ id: v.id("items"), seq: v.int64() })),
+  },
+});
+
+export const getBatch = query({
+  args: validators.vQueryArgs,
+  returns: validators.vQueryReturns,
+  handler: async (ctx, { name, cursor }) => {
+    const behavior = await ctx.db
+      .query("behavior")
+      .withIndex("name", (q) => q.eq("name", name))
+      .unique();
+    if (behavior?.failQuery) throw new Error("query failure");
+    const items = await ctx.db
+      .query("items")
+      .withIndex("name_seq", (q) => q.eq("name", name).gte("seq", cursor ?? 0n))
+      .take(3);
+    if (!items.length)
+      return {
+        kind: "idle" as const,
+        cooldownMs: 0,
+        ...(behavior?.timeoutMs !== undefined
+          ? { timeoutMs: behavior.timeoutMs }
+          : {}),
+      };
+    return {
+      batch: {
+        name,
+        items: items.map((item) => ({ id: item._id, seq: item.seq as bigint })),
+      },
+      cursor: (items.at(-1)!.seq as bigint) + 1n,
+    };
+  },
+});
+export const processBatch = mutation({
+  args: validators.vMutationArgs,
+  returns: validators.vMutationReturns,
+  handler: async (ctx, { name, items }) => {
+    await ctx.db.insert("batches", {
+      name,
+      seqs: items.map((item) => item.seq),
+    });
+    for (const item of items) await ctx.db.delete("items", item.id);
+    const behavior = await ctx.db
+      .query("behavior")
+      .withIndex("name", (q) => q.eq("name", name))
+      .unique();
+    if (behavior?.failMutation)
+      throw new Error("mutation failure after writes");
+    return behavior?.debounceMs !== undefined
+      ? { debounceMs: behavior.debounceMs }
+      : null;
+  },
+});
+const testApi = (
+  anyApi as unknown as ApiFromModules<{
+    "workpool.test": {
+      getBatch: typeof getBatch;
+      processBatch: typeof processBatch;
+    };
+  }>
+)["workpool.test"];
+
+// Test-only inspection functions mounted inside the actual component namespaces.
+const workerQueries = {
+  state: internalQueryGeneric({
+    args: { name: v.string() },
+    handler: async (ctx, { name }) => {
+      const worker = (await ctx.db
+        .query("workers")
+        .withIndex("name", (q) => q.eq("name", name))
+        .unique()) as Doc<"workers"> | null;
+      const state =
+        worker &&
+        ((await ctx.db.get(
+          "workerState",
+          worker.stateId,
+        )) as Doc<"workerState"> | null);
+      return { worker, state };
+    },
+  }),
+};
+const poolQueries = {
+  state: internalQueryGeneric({
+    args: {},
+    handler: async (ctx) => ({
+      work: await ctx.db.query("work").take(100),
+      state: await ctx.db.query("internalState").unique(),
+      globals: await ctx.db.query("globals").unique(),
+    }),
+  }),
+};
+const components = componentsGeneric() as unknown as {
+  batchWorker: ComponentApi &
+    typeof batchInternal &
+    ApiFromModules<{ _test: typeof workerQueries }>;
+  pool: WorkpoolComponent & ApiFromModules<{ _test: typeof poolQueries }>;
+  otherPool: WorkpoolComponent;
+};
+
+function setup() {
+  const t = convexTest(schema, {
+    ...import.meta.glob("./**/*.ts"),
+    "./workpool.test.ts": async () => ({ getBatch, processBatch }),
+  });
+  t.registerComponent("batchWorker", batchWorkerTest.schema, {
+    ...batchWorkerTest.modules,
+    "./component/_test.ts": async () => workerQueries,
+  });
+  for (const name of ["pool", "otherPool"]) {
+    t.registerComponent(name, workpoolTest.schema, {
+      ...workpoolTest.modules,
+      "./component/_test.ts": async () => poolQueries,
+    });
+    batchWorkerTest.register(t, `${name}/batchWorker`);
+  }
+  // Omit maxParallelism so tests can pause/resume the pool dynamically.
+  const pool = new Workpool(components.pool, { logLevel: "ERROR" });
+  const register = (
+    name = "user",
+    workpool: Workpool | undefined = pool,
+    debounceMs = 0,
+  ) =>
+    t.mutation((ctx) =>
+      ping(ctx, components.batchWorker, {
+        name,
+        workQuery: testApi.getBatch,
+        workerMutation: testApi.processBatch,
+        config: { debounceMs, monitorLagMs: 10_000 },
+        ...(workpool ? { workpool } : {}),
+      }),
+    );
+  const seed = (name: string, seqs: bigint[]) =>
+    t.mutation(async (ctx) => {
+      for (const seq of seqs) await ctx.db.insert("items", { name, seq });
+    });
+  const state = (name = "user") =>
+    t.query(components.batchWorker._test.state, { name });
+  const capacity = (maxParallelism: number) =>
+    t.mutation(components.pool.config.update, { maxParallelism });
+  const batches = () =>
+    t.query(async (ctx) =>
+      (await ctx.db.query("batches").take(100)).map((b) => ({
+        name: b.name,
+        seqs: b.seqs,
+      })),
+    );
+  // Advance individual scheduler waves so we can inspect admission and failure
+  // before a delayed retry runs. convex-test serializes mutations, so checking
+  // Workpool's admitted set is more meaningful than timing handler overlap.
+  const until = async (condition: () => Promise<boolean>) => {
+    for (let i = 0; i < 150; i++) {
+      if (await condition()) return;
+      vi.advanceTimersToNextTimer();
+      await t.finishInProgressScheduledFunctions();
+    }
+    throw new Error("condition did not become true");
+  };
+  const drain = () => t.finishAllScheduledFunctions(vi.runAllTimers);
+  return { t, pool, register, seed, state, capacity, batches, until, drain };
+}
+
+beforeEach(() => {
+  vi.useFakeTimers();
+  vi.setSystemTime(new Date("2026-09-09T00:00:00Z"));
+});
+afterEach(() => {
+  vi.useRealTimers();
+});
+
+describe("Workpool execution", () => {
+  test("queues one lazy continuation per worker and caps shared admission", async () => {
+    const s = setup();
+    await s.capacity(0);
+    for (const name of ["alice", "bob", "carol"]) {
+      await s.seed(name, [0n]);
+      await s.register(name);
+      await s.register(name);
+      await s.seed(name, [1n, 2n, 3n]);
+    }
+    const queued = await s.t.query(components.pool._test.state, {});
+    expect(queued.work).toHaveLength(3);
+    expect(
+      queued.work.every(
+        (w) => Object.keys(w.fnArgs).sort().join() === "generation,name",
+      ),
+    ).toBe(true);
+    expect(
+      queued.work.every(
+        (w) => w.onComplete.onStatusHandle.failed && !w.onComplete.fnHandle,
+      ),
+    ).toBe(true);
+    expect(await s.batches()).toEqual([]);
+    const before = await s.state("alice");
+    expect(before.state?.runnerId).toBeUndefined();
+    expect(before.state?.monitorId).toBeUndefined();
+    // Backlog older than the scheduler watchdog's threshold is healthy.
+    vi.setSystemTime(Date.now() + 120_000);
+    await s.register("alice");
+    await s.t.mutation(components.batchWorker.monitor.monitor, {
+      name: "alice",
+    });
+    expect((await s.state("alice")).state?.generation).toBe(
+      before.state?.generation,
+    );
+
+    await s.capacity(2);
+    await s.until(
+      async () =>
+        (await s.t.query(components.pool._test.state, {})).state?.running
+          .length === 2,
+    );
+    expect(
+      (await s.t.query(components.pool._test.state, {})).state.running,
+    ).toHaveLength(2);
+    await s.drain();
+    for (const name of ["alice", "bob", "carol"]) {
+      expect(
+        (await s.batches()).filter((b) => b.name === name).map((b) => b.seqs),
+      ).toEqual([[0n, 1n, 2n], [3n]]);
+      expect((await s.state(name)).worker?.status.kind).toBe("idle");
+      expect((await s.state(name)).state?.workpoolJob).toBeUndefined();
+    }
+  });
+
+  test.each(["failQuery", "failMutation"] as const)(
+    "recovers %s with delayed retry and atomic cursor/batch writes",
+    async (failure) => {
+      const s = setup();
+      const behavior = await s.t.mutation((ctx) =>
+        ctx.db.insert("behavior", { name: "user", [failure]: true }),
+      );
+      await s.seed("user", [0n, 1n]);
+      await s.register();
+      const before = (await s.state()).state!;
+      await s.until(
+        async () => (await s.state()).state!.generation > before.generation,
+      );
+      const after = (await s.state()).state!;
+      expect(after.cursor).toBeUndefined();
+      expect(after.workpoolJob!.runAt).toBeGreaterThanOrEqual(
+        Date.now() + 10_000,
+      );
+      expect(after.monitorId).toBeUndefined();
+      expect(await s.batches()).toEqual([]);
+      expect(
+        await s.t.query((ctx) => ctx.db.query("items").take(10)),
+      ).toHaveLength(2);
+      // Duplicate notifications for the failed generation cannot create jobs.
+      await s.t.mutation(components.batchWorker.workpool.onFailure, {
+        workId: before.workpoolJob!.id,
+        context: { name: "user", generation: before.generation },
+        result: { kind: "failed", error: "duplicate" },
+      });
+      expect((await s.state()).state?.generation).toBe(after.generation);
+      await s.t.mutation((ctx) =>
+        ctx.db.patch("behavior", behavior, { [failure]: false }),
+      );
+      await s.drain();
+      expect(await s.batches()).toEqual([{ name: "user", seqs: [0n, 1n] }]);
+      expect((await s.state()).state?.cursor).toBe(2n);
+    },
+  );
+
+  test("stop invalidates an admitted mutation and late failures across start", async () => {
+    const s = setup();
+    await s.seed("user", [0n]);
+    await s.register();
+    const before = (await s.state()).state!;
+    await s.until(
+      async () =>
+        (await s.t.query(components.pool._test.state, {})).state?.running
+          .length === 1,
+    );
+    await s.t.mutation(components.batchWorker.lib.stop, { name: "user" });
+    await s.register(); // ping cannot resume a stopped worker
+    expect((await s.state()).worker?.status.kind).toBe("stopped");
+    await s.drain(); // admitted wrapper may run, but the batch must not
+    expect(await s.batches()).toEqual([]);
+    await s.t.mutation(components.batchWorker.lib.start, { name: "user" });
+    const restarted = (await s.state()).state!;
+    await s.t.mutation(components.batchWorker.workpool.onFailure, {
+      workId: before.workpoolJob!.id,
+      context: { name: "user", generation: before.generation },
+      result: { kind: "failed", error: "late failure" },
+    });
+    expect((await s.state()).state?.generation).toBe(restarted.generation);
+    await s.drain();
+    expect(await s.batches()).toEqual([{ name: "user", seqs: [0n] }]);
+  });
+
+  test("interrupts idle timeouts but keeps imminent pooled wakeups", async () => {
+    const s = setup();
+    const behavior = await s.t.mutation((ctx) =>
+      ctx.db.insert("behavior", { name: "user", timeoutMs: 30_000 }),
+    );
+    await s.register();
+    await s.until(async () => (await s.state()).worker?.status.kind === "idle");
+    const waiting = (await s.state()).state!;
+    await s.seed("user", [0n]);
+    await s.register();
+    expect((await s.state()).state!.workpoolJob!.id).not.toBe(
+      waiting.workpoolJob!.id,
+    );
+    await s.until(async () => (await s.state()).worker?.status.kind === "idle");
+    const imminent = (await s.state()).state!;
+    vi.setSystemTime(imminent.workpoolJob!.runAt - 500);
+    await s.register();
+    expect((await s.state()).state!.workpoolJob!.id).toBe(
+      imminent.workpoolJob!.id,
+    );
+    await s.t.mutation((ctx) =>
+      ctx.db.patch("behavior", behavior, { timeoutMs: undefined }),
+    );
+    await s.drain();
+    expect(await s.batches()).toEqual([{ name: "user", seqs: [0n] }]);
+  });
+
+  test("keeps debounce windows when pinged", async () => {
+    const s = setup();
+    await s.t.mutation((ctx) =>
+      ctx.db.insert("behavior", { name: "user", debounceMs: 20_000 }),
+    );
+    await s.seed("user", [0n, 1n, 2n, 3n]);
+    await s.register("user", s.pool, 5000);
+    const initial = (await s.state()).state!;
+    await s.register("user", s.pool, 5000);
+    expect((await s.state()).state?.workpoolJob).toEqual(initial.workpoolJob);
+    await s.until(async () => (await s.batches()).length === 1);
+    const next = (await s.state()).state!;
+    await s.register("user", s.pool, 5000);
+    expect((await s.state()).state?.workpoolJob).toEqual(next.workpoolJob);
+    expect(next.workpoolJob!.runAt).toBeGreaterThanOrEqual(Date.now() + 20_000);
+    await s.drain();
+    expect((await s.batches()).map((b) => b.seqs)).toEqual([
+      [0n, 1n, 2n],
+      [3n],
+    ]);
+  });
+
+  test("recovers a failed timeout wake while the worker is marked idle", async () => {
+    const s = setup();
+    const behavior = await s.t.mutation((ctx) =>
+      ctx.db.insert("behavior", { name: "user", timeoutMs: 30_000 }),
+    );
+    await s.register();
+    await s.until(async () => (await s.state()).worker?.status.kind === "idle");
+    const waiting = (await s.state()).state!;
+    await s.t.mutation((ctx) =>
+      ctx.db.patch("behavior", behavior, { failQuery: true }),
+    );
+    await s.until(
+      async () => (await s.state()).state!.generation > waiting.generation,
+    );
+    const recovering = await s.state();
+    expect(recovering.worker?.status.kind).toBe("running");
+    expect(recovering.state?.monitorId).toBeUndefined();
+    expect(recovering.state!.workpoolJob!.runAt).toBeGreaterThanOrEqual(
+      Date.now() + 10_000,
+    );
+    await s.t.mutation((ctx) =>
+      ctx.db.patch("behavior", behavior, {
+        failQuery: false,
+        timeoutMs: undefined,
+      }),
+    );
+    await s.seed("user", [0n]);
+    await s.drain();
+    expect(await s.batches()).toEqual([{ name: "user", seqs: [0n] }]);
+  });
+
+  test("moves between pools and the scheduler without losing the cursor", async () => {
+    const s = setup();
+    await s.capacity(0);
+    await s.seed("user", [0n, 1n, 2n, 3n]);
+    await s.register();
+    const first = (await s.state()).state!;
+    const otherPool = new Workpool(components.otherPool, {
+      maxParallelism: 0,
+      logLevel: "ERROR",
+    });
+    await s.register("user", otherPool);
+    const moved = (await s.state()).state!;
+    expect(moved.workpoolJob!.cancel).not.toBe(first.workpoolJob!.cancel);
+    // Explicitly omit the pool to switch to scheduler execution.
+    await s.t.mutation((ctx) =>
+      ping(ctx, components.batchWorker, {
+        name: "user",
+        workQuery: testApi.getBatch,
+        workerMutation: testApi.processBatch,
+        config: { debounceMs: 0 },
+      }),
+    );
+    const scheduled = (await s.state()).state!;
+    expect(scheduled.workpoolJob).toBeUndefined();
+    expect(scheduled.runnerId).toBeDefined();
+    expect(scheduled.monitorId).toBeDefined();
+    await s.until(async () => (await s.batches()).length === 1);
+    await s.register();
+    const pooled = (await s.state()).state!;
+    expect(pooled.cursor).toBe(3n);
+    expect(pooled.monitorId).toBeUndefined();
+    await s.capacity(1);
+    await s.drain();
+    expect((await s.batches()).map((b) => b.seqs)).toEqual([
+      [0n, 1n, 2n],
+      [3n],
+    ]);
+    expect(
+      await s.t.query((ctx) =>
+        s.pool.status(ctx, first.workpoolJob!.id as WorkId),
+      ),
+    ).toEqual({ state: "finished" });
+  });
+
+  test("direct Workpool cancellation requires manual kick", async () => {
+    const s = setup();
+    await s.capacity(0);
+    await s.seed("user", [0n]);
+    await s.register();
+    const before = (await s.state()).state!;
+    await s.t.mutation((ctx) =>
+      s.pool.cancel(ctx, before.workpoolJob!.id as WorkId),
+    );
+    await s.drain();
+    expect((await s.state()).state?.generation).toBe(before.generation);
+    expect(await s.batches()).toEqual([]);
+    await s.t.mutation(components.batchWorker.lib.kick, { name: "user" });
+    await s.capacity(1);
+    await s.drain();
+    expect(await s.batches()).toEqual([{ name: "user", seqs: [0n] }]);
+  });
+});
diff --git a/src/component/_generated/api.ts b/src/component/_generated/api.ts
index 75e9b10..c4dd100 100644
--- a/src/component/_generated/api.ts
+++ b/src/component/_generated/api.ts
@@ -15,6 +15,7 @@ import type * as logging from "../logging.js";
 import type * as loop from "../loop.js";
 import type * as monitor from "../monitor.js";
 import type * as shared from "../shared.js";
+import type * as workpool from "../workpool.js";
 
 import type {
   ApiFromModules,
@@ -31,6 +32,7 @@ const fullApi: ApiFromModules<{
   loop: typeof loop;
   monitor: typeof monitor;
   shared: typeof shared;
+  workpool: typeof workpool;
 }> = anyApi as any;
 
 /**
diff --git a/src/component/_generated/component.ts b/src/component/_generated/component.ts
index da091e7..f1eb4f8 100644
--- a/src/component/_generated/component.ts
+++ b/src/component/_generated/component.ts
@@ -31,6 +31,13 @@ export type ComponentApi<Name extends string | undefined = string | undefined> =
         any,
         Name
       >;
+      kick: FunctionReference<
+        "mutation",
+        "internal",
+        { name: string },
+        null,
+        Name
+      >;
       ping: FunctionReference<
         "mutation",
         "internal",
@@ -39,6 +46,12 @@ export type ComponentApi<Name extends string | undefined = string | undefined> =
           name: string;
           workQuery: string;
           workerMutation: string;
+          workpool?: {
+            cancel: string;
+            enqueue: string;
+            logLevel?: "DEBUG" | "TRACE" | "INFO" | "REPORT" | "WARN" | "ERROR";
+            maxParallelism?: number;
+          };
         },
         null,
         Name
diff --git a/src/component/kick.ts b/src/component/kick.ts
index 97a1815..2ed8f25 100644
--- a/src/component/kick.ts
+++ b/src/component/kick.ts
@@ -8,7 +8,9 @@ import {
   MONITOR_REFRESH_WITHIN_MS,
   RUNNING_THRESHOLD_MS,
   MONITOR_LAG_MS,
+  type WorkpoolConfig,
 } from "./shared.js";
+import { cancelWorkpoolJob, enqueueLoop } from "./workpool.js";
 
 export async function getWorker(ctx: QueryCtx, name: string) {
   return ctx.db
@@ -48,6 +50,7 @@ export async function ping(
     workQuery: string;
     workerMutation: string;
     config?: Partial<Config> | undefined;
+    workpool?: WorkpoolConfig | undefined;
   },
 ): Promise<void> {
   const worker = await getWorker(ctx, args.name);
@@ -62,6 +65,7 @@ export async function ping(
       workQuery: args.workQuery,
       workerMutation: args.workerMutation,
       config: args.config ?? {},
+      ...(args.workpool ? { workpool: args.workpool } : {}),
       status: { kind: "running" },
       stateId,
     });
@@ -71,7 +75,9 @@ export async function ping(
     return;
   }
 
+  const poolChanged = !sameWorkpool(worker.workpool, args.workpool);
   if (
+    poolChanged ||
     args.workQuery !== worker.workQuery ||
     args.workerMutation !== worker.workerMutation ||
     (args.config &&
@@ -80,11 +86,33 @@ export async function ping(
   ) {
     worker.workQuery = args.workQuery;
     worker.workerMutation = args.workerMutation;
+    if (args.workpool) worker.workpool = args.workpool;
+    else delete worker.workpool;
     if (args.config) {
       worker.config = args.config;
     }
     await ctx.db.replace("workers", worker._id, worker);
   }
+  if (poolChanged) {
+    const state = await getOrCreateWorkerState(ctx, worker);
+    const scheduled =
+      state.runnerId &&
+      (await ctx.db.system.get("_scheduled_functions", state.runnerId));
+    const runAt =
+      state.workpoolJob?.runAt ?? scheduled?.scheduledTime ?? Date.now();
+    await cancelLoop(ctx, state);
+    await cancelMonitor(ctx, state);
+    if (
+      worker.status.kind !== "stopped" &&
+      (state.runnerId || state.workpoolJob || worker.status.kind === "running")
+    ) {
+      // Preserve debounce/timeout eligibility while moving an active worker.
+      // An idle worker's ping below may still interrupt its wait.
+      await scheduleLoopRun(ctx, worker, {
+        delayMs: Math.max(0, runAt - Date.now()),
+      });
+    }
+  }
   if (worker.status.kind !== "idle") {
     ctx.log.debug(`[ping] "${worker.name}" ${worker.status.kind} — no-op`);
     return;
@@ -92,6 +120,15 @@ export async function ping(
   await wake(ctx, worker);
 }
 
+function sameWorkpool(a?: WorkpoolConfig, b?: WorkpoolConfig): boolean {
+  return (
+    a?.enqueue === b?.enqueue &&
+    a?.cancel === b?.cancel &&
+    a?.maxParallelism === b?.maxParallelism &&
+    a?.logLevel === b?.logLevel
+  );
+}
+
 /**
  * Resume an existing worker (e.g. after `stop`) using its stored handles and
  * config. No-ops if the worker was never created with `ping`.
@@ -117,7 +154,7 @@ export async function kick(ctx: MutationCtx, name: string): Promise<void> {
   const worker = await getWorker(ctx, name);
   if (!worker) return;
   const state = await getOrCreateWorkerState(ctx, worker);
-  if (state.runnerId) await cancelIfPending(ctx, state.runnerId);
+  await cancelLoop(ctx, state);
   // Clear the monitor state so scheduleLoopRun arms a fresh monitor even if
   // monitorRunAtMs still looks healthy.
   await cancelMonitor(ctx, state);
@@ -126,20 +163,14 @@ export async function kick(ctx: MutationCtx, name: string): Promise<void> {
 }
 
 /**
- * Stop the worker: cancel its loop and monitor and mark it idle.
+ * Stop the worker: cancel its loop and monitor and mark it stopped.
  * Only `start` will resume it.
  */
 export async function stop(ctx: MutationCtx, name: string): Promise<void> {
   const worker = await getWorker(ctx, name);
   if (!worker) return;
   const state = await getOrCreateWorkerState(ctx, worker);
-  if (state?.runnerId) {
-    await cancelIfPending(ctx, state.runnerId);
-    await ctx.db.patch("workerState", state._id, {
-      runnerId: undefined,
-      generation: state.generation + 1n,
-    });
-  }
+  await cancelLoop(ctx, state);
   await cancelMonitor(ctx, state);
   await ctx.db.patch("workers", worker._id, { status: { kind: "stopped" } });
 }
@@ -158,26 +189,29 @@ async function wake(ctx: MutationCtx, worker: Doc<"workers">): Promise<void> {
   const loop =
     state.runnerId &&
     (await ctx.db.system.get("_scheduled_functions", state.runnerId));
+  const runAt =
+    state.workpoolJob?.runAt ??
+    (loop?.state.kind === "pending" ? loop.scheduledTime : undefined);
   // Possibly wait for a debounce window before running
   const delayMs = worker.config.debounceMs ?? DEFAULT_CONFIG.debounceMs;
   await ctx.db.patch("workers", worker._id, { status: { kind: "running" } });
   // Rescheduling would run at `now + delayMs`; if the pending run is already
   // sooner than that (or imminent), canceling it would only delay work.
   if (
-    loop?.state.kind === "pending" &&
-    loop.scheduledTime < now + Math.max(delayMs, RUNNING_THRESHOLD_MS)
+    runAt !== undefined &&
+    runAt < now + Math.max(delayMs, RUNNING_THRESHOLD_MS)
   ) {
     ctx.log.debug(
       `[wake] "${worker.name}" already scheduled to run sooner — keeping it`,
     );
     // The kept run gets a fresh cooldown window too.
     await ctx.db.patch("workerState", state._id, {
-      lastWorkTs: Math.max(state.lastWorkTs, loop.scheduledTime),
+      lastWorkTs: Math.max(state.lastWorkTs, runAt),
     });
     return;
   }
   ctx.log.debug(`[wake] "${worker.name}" interrupting wait`);
-  if (loop) await cancelIfPending(ctx, loop._id);
+  if (state.runnerId || state.workpoolJob) await cancelLoop(ctx, state);
   await scheduleLoopRun(ctx, worker, { delayMs, lastWorkTs: now + delayMs });
 }
 
@@ -229,6 +263,7 @@ export async function goIdle(
   await ctx.db.patch("workerState", state._id, {
     generation: state.generation + 1n,
     runnerId: undefined,
+    workpoolJob: undefined,
   });
   await cancelMonitor(ctx, state);
   await ctx.db.patch("workers", worker._id, { status: { kind: "idle" } });
@@ -245,22 +280,48 @@ async function scheduleLoopRun(
 ): Promise<void> {
   const state = await getOrCreateWorkerState(ctx, worker);
   const generation = state.generation + 1n;
-  const runnerId = await ctx.scheduler.runAfter(
-    opts.delayMs,
-    internal.loop.loop,
-    { name: worker.name, generation },
-  );
+  const runAt = Date.now() + opts.delayMs;
+  let runnerId: Id<"_scheduled_functions"> | undefined;
+  let workpoolJob: Doc<"workerState">["workpoolJob"];
+  if (worker.workpool) {
+    const id = await enqueueLoop(
+      ctx,
+      worker.workpool,
+      { name: worker.name, generation },
+      runAt,
+    );
+    workpoolJob = { id, runAt, cancel: worker.workpool.cancel };
+  } else {
+    runnerId = await ctx.scheduler.runAfter(opts.delayMs, internal.loop.loop, {
+      name: worker.name,
+      generation,
+    });
+  }
   // The cursor rides along on the patch this already does every iteration.
   // `undefined` leaves the stored cursor as it was.
   await ctx.db.patch("workerState", state._id, {
     generation,
     runnerId,
+    workpoolJob,
     ...(opts.lastWorkTs !== undefined ? { lastWorkTs: opts.lastWorkTs } : {}),
     ...(opts.cursor !== undefined ? { cursor: opts.cursor } : {}),
   });
 
-  // await ctx.db.patch("workers", worker._id, { status: worker.status });
-  await ensureMonitored(ctx, worker, Date.now() + opts.delayMs);
+  if (!worker.workpool) await ensureMonitored(ctx, worker, runAt);
+}
+
+/** Invalidation is transactional even if Workpool already admitted the job. */
+async function cancelLoop(
+  ctx: MutationCtx,
+  state: Doc<"workerState">,
+): Promise<void> {
+  if (state.runnerId) await cancelIfPending(ctx, state.runnerId);
+  if (state.workpoolJob) await cancelWorkpoolJob(ctx, state.workpoolJob);
+  await ctx.db.patch("workerState", state._id, {
+    runnerId: undefined,
+    workpoolJob: undefined,
+    generation: state.generation + 1n,
+  });
 }
 
 // ── Monitor ────────────────────────────────────────────────────────────────
@@ -276,6 +337,7 @@ export async function ensureMonitored(
   worker: Doc<"workers">,
   loopRunAtMs: number,
 ): Promise<void> {
+  if (worker.workpool) return;
   const state = await ctx.db.get("workerState", worker.stateId);
   if (!state) return;
 
diff --git a/src/component/lib.ts b/src/component/lib.ts
index 62c0d1e..1b5c7bb 100644
--- a/src/component/lib.ts
+++ b/src/component/lib.ts
@@ -8,7 +8,7 @@ import {
   start as startHelper,
   stop as stopHelper,
 } from "./kick.js";
-import { vConfig, vStatus } from "./shared.js";
+import { vConfig, vStatus, vWorkpool } from "./shared.js";
 
 /**
  * The public component API. Apps call `ping` from `@convex-dev/batch-worker`
@@ -23,6 +23,7 @@ export const ping = mutation({
     workQuery: v.string(),
     workerMutation: v.string(),
     config: v.optional(vConfig.partial()),
+    workpool: v.optional(vWorkpool),
   },
   returns: v.null(),
   handler: async (ctx, args) => pingHelper(ctx, args),
diff --git a/src/component/loop.ts b/src/component/loop.ts
index d2243bd..ec2cebc 100644
--- a/src/component/loop.ts
+++ b/src/component/loop.ts
@@ -41,7 +41,7 @@ export const loop = internalMutation({
       return; // worker was deleted
     }
 
-    if (state.generation !== generation) {
+    if (state.generation !== generation || worker.status.kind === "stopped") {
       ctx.log.debug(
         `[loop] "${name}" superseded (gen ${generation} !== ${state.generation})`,
       );
@@ -62,8 +62,8 @@ export const loop = internalMutation({
 
     // Stale snapshot read: no OCC dependency, so concurrent inserts while we
     // drain don't force this loop to retry. If the query or worker mutation
-    // throws, this loop fails (and doesn't reschedule) — the monitor restarts
-    // it.
+    // throws, this loop fails (and doesn't reschedule) — the monitor or the
+    // Workpool onFailure callback restarts it.
     const result = await ctx.runQuery(queryRef, queryArgs, {
       useStaleSnapshot: true,
     });
diff --git a/src/component/monitor.ts b/src/component/monitor.ts
index 5a2f701..5b39171 100644
--- a/src/component/monitor.ts
+++ b/src/component/monitor.ts
@@ -29,6 +29,11 @@ export const monitor = internalMutation({
     }
 
     const state = await getOrCreateWorkerState(ctx, worker);
+    // A monitor scheduled before switching to a pool must not restart its job.
+    if (worker.workpool) {
+      await cancelMonitor(ctx, state);
+      return;
+    }
     const loop =
       state.runnerId &&
       (await ctx.db.system.get("_scheduled_functions", state.runnerId));
diff --git a/src/component/schema.ts b/src/component/schema.ts
index 3a6259f..532fa90 100644
--- a/src/component/schema.ts
+++ b/src/component/schema.ts
@@ -1,6 +1,6 @@
 import { defineSchema, defineTable } from "convex/server";
 import { v } from "convex/values";
-import { vConfig, vStatus } from "./shared.js";
+import { vConfig, vStatus, vWorkpool } from "./shared.js";
 
 export default defineSchema({
   // One row per named worker. Written rarely — only on create/reconfigure,
@@ -13,6 +13,7 @@ export default defineSchema({
     workQuery: v.string(),
     workerMutation: v.string(),
     config: vConfig.partial(),
+    workpool: v.optional(vWorkpool),
     status: vStatus,
     stateId: v.id("workerState"),
   }).index("name", ["name"]),
@@ -30,6 +31,16 @@ export default defineSchema({
     // The currently-scheduled loop invocation, checked by the monitor and
     // canceled when a ping interrupts a wait.
     runnerId: v.optional(v.id("_scheduled_functions")),
+    // Alternative to runnerId. Keep the cancel handle with the job so a pool
+    // change can still cancel work in the old pool. runAt is its eligibility
+    // time, not a promise that the pool has capacity then.
+    workpoolJob: v.optional(
+      v.object({
+        id: v.string(),
+        runAt: v.number(),
+        cancel: v.string(),
+      }),
+    ),
     // The monitor that restarts the loop if it dies, scheduled to fire
     // `monitorLagMs` after the loop's next run and refreshed as it approaches.
     monitorId: v.optional(v.id("_scheduled_functions")),
diff --git a/src/component/shared.ts b/src/component/shared.ts
index c3f8659..d183689 100644
--- a/src/component/shared.ts
+++ b/src/component/shared.ts
@@ -6,6 +6,16 @@ import {
   type VCommitTs,
   type Validator,
 } from "convex/values";
+import { logLevel } from "./logging.js";
+
+/** Capabilities for a sibling Workpool, passed as handles by the client. */
+export const vWorkpool = v.object({
+  enqueue: v.string(),
+  cancel: v.string(),
+  maxParallelism: v.optional(v.number()),
+  logLevel: v.optional(logLevel),
+});
+export type WorkpoolConfig = Infer<typeof vWorkpool>;
 
 export const MS = 1;
 export const SECOND = 1000 * MS;
@@ -40,6 +50,8 @@ export const vConfig = v.object({
    * monitor restarts the loop if it didn't run (and push the monitor back) by
    * then — this is also the effective retry cadence when the work query or
    * worker mutation throws.
+   * With a Workpool, no monitor is scheduled; this controls the delay before
+   * retrying a failed iteration (still at least 10 seconds).
    */
   monitorLagMs: v.number(),
 });
diff --git a/src/component/workpool.ts b/src/component/workpool.ts
new file mode 100644
index 0000000..bf87b75
--- /dev/null
+++ b/src/component/workpool.ts
@@ -0,0 +1,94 @@
+import { createFunctionHandle, type FunctionHandle } from "convex/server";
+import { v } from "convex/values";
+import { internal } from "./_generated/api.js";
+import { internalMutation, type MutationCtx } from "./functions.js";
+import { continueRunning, getWorker } from "./kick.js";
+import {
+  MONITOR_LAG_MS,
+  MONITOR_REFRESH_WITHIN_MS,
+  type WorkpoolConfig,
+} from "./shared.js";
+
+// Workpool's enqueue wire protocol. No runtime dependency: Workpool itself
+// uses BatchWorker as its scheduler-backed driver.
+type EnqueueArgs = {
+  fnHandle: string;
+  fnName: string;
+  fnType: "mutation";
+  fnArgs: { name: string; generation: bigint };
+  runAt: number;
+  config: Pick<WorkpoolConfig, "maxParallelism" | "logLevel">;
+  onComplete: {
+    onStatusHandle: { failed: string };
+    context: { name: string; generation: bigint };
+  };
+};
+
+export async function enqueueLoop(
+  ctx: MutationCtx,
+  pool: WorkpoolConfig,
+  args: { name: string; generation: bigint },
+  runAt: number,
+): Promise<string> {
+  const [fnHandle, failed] = await Promise.all([
+    createFunctionHandle(internal.loop.loop),
+    createFunctionHandle(internal.workpool.onFailure),
+  ]);
+  return ctx.runMutation(
+    pool.enqueue as FunctionHandle<"mutation", EnqueueArgs, string>,
+    {
+      fnHandle,
+      fnName: `batchWorker:${args.name}`,
+      fnType: "mutation",
+      fnArgs: args,
+      runAt,
+      config: {
+        ...(pool.maxParallelism !== undefined
+          ? { maxParallelism: pool.maxParallelism }
+          : {}),
+        ...(pool.logLevel !== undefined ? { logLevel: pool.logLevel } : {}),
+      },
+      // This is how Workpool's client serializes its onFailure option.
+      onComplete: { onStatusHandle: { failed }, context: args },
+    },
+  );
+}
+
+export async function cancelWorkpoolJob(
+  ctx: MutationCtx,
+  job: { id: string; cancel: string },
+): Promise<void> {
+  await ctx.runMutation(
+    job.cancel as FunctionHandle<"mutation", { id: string }, null>,
+    { id: job.id },
+  );
+}
+
+/** Only failed iterations invoke this; success schedules its own successor. */
+export const onFailure = internalMutation({
+  args: {
+    workId: v.string(),
+    context: v.object({ name: v.string(), generation: v.int64() }),
+    result: v.object({ kind: v.literal("failed"), error: v.string() }),
+  },
+  returns: v.null(),
+  handler: async (ctx, { workId, context: { name, generation }, result }) => {
+    const worker = await getWorker(ctx, name);
+    if (!worker?.workpool || worker.status.kind === "stopped") return null;
+    const state = await ctx.db.get("workerState", worker.stateId);
+    if (state?.generation !== generation || state.workpoolJob?.id !== workId) {
+      return null;
+    }
+    ctx.log.error(`[onFailure] "${name}" loop failed: ${result.error}`);
+    ctx.log.event("restart", { name });
+    await continueRunning(
+      ctx,
+      worker,
+      Math.max(
+        MONITOR_REFRESH_WITHIN_MS,
+        worker.config.monitorLagMs ?? MONITOR_LAG_MS,
+      ),
+    );
+    return null;
+  },
+});