From 32b9e3c4afea803d7af20819bf6574c76dbe8f8d Mon Sep 17 00:00:00 2001
From: Ian Macartney <366683+ianmacartney@users.noreply.github.com>
Date: Thu, 17 Sep 2026 23:23:14 -0700
Subject: [PATCH 1/3] fix(examples): validate batch candidates in mutation
 snapshots

---
 example/convex/e2e.ts              | 21 ++++++------
 example/convex/example.test.ts     | 31 +++++++++++++++++-
 example/convex/example.ts          | 19 +++++++----
 example/convex/rateLimited.test.ts | 51 +++++++++++++++++++++++++++++-
 example/convex/rateLimited.ts      | 38 +++++++++++-----------
 5 files changed, 121 insertions(+), 39 deletions(-)

diff --git a/example/convex/e2e.ts b/example/convex/e2e.ts
index 734e314..b083218 100644
--- a/example/convex/e2e.ts
+++ b/example/convex/e2e.ts
@@ -54,9 +54,7 @@ export const enqueue = mutation({
 const { vQueryArgs, vQueryReturns, vMutationArgs } =
   defineBatchWorkerValidators({
     batch: {
-      items: v.array(
-        v.object({ id: v.id("e2eEvents"), creationTime: v.number() }),
-      ),
+      ids: v.array(v.id("e2eEvents")),
     },
   });
 
@@ -76,10 +74,7 @@ export const getBatch = internalQuery({
     return {
       kind: "work" as const,
       batch: {
-        items: events.map((e) => ({
-          id: e._id,
-          creationTime: e._creationTime,
-        })),
+        ids: events.map((e) => e._id),
       },
       cursor: events.at(-1)!.updatedAt,
     };
@@ -88,9 +83,15 @@ export const getBatch = internalQuery({
 
 export const processBatch = internalMutation({
   args: vMutationArgs,
-  handler: async (ctx, { items }) => {
+  handler: async (ctx, { ids }) => {
+    // Match the queue example: measure only rows still present in this snapshot.
+    const candidates = await Promise.all(
+      ids.map((id) => ctx.db.get("e2eEvents", id)),
+    );
+    const items = candidates.filter((item) => item !== null);
+    if (items.length === 0) return;
     const now = Date.now();
-    const latencies = items.map((b) => now - b.creationTime);
+    const latencies = items.map((b) => now - b._creationTime);
     await ctx.db.insert("e2eSamples", {
       processedAt: now,
       batchSize: items.length,
@@ -98,7 +99,7 @@ export const processBatch = internalMutation({
       newestLatencyMs: Math.min(...latencies),
     });
     for (const b of items) {
-      await ctx.db.delete("e2eEvents", b.id);
+      await ctx.db.delete("e2eEvents", b._id);
     }
   },
 });
diff --git a/example/convex/example.test.ts b/example/convex/example.test.ts
index 465ecfb..78fefe8 100644
--- a/example/convex/example.test.ts
+++ b/example/convex/example.test.ts
@@ -1,6 +1,6 @@
 import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
 import { initConvexTest } from "./setup.test";
-import { api, components } from "./_generated/api";
+import { api, components, internal } from "./_generated/api";
 
 describe("example worker", () => {
   beforeEach(() => {
@@ -21,6 +21,35 @@ describe("example worker", () => {
     expect(totals).toEqual({ total: 5, count: 1, pending: 0 });
   });
 
+  test("re-fetches changed events and skips deleted or already processed candidates", async () => {
+    const t = initConvexTest();
+    const ids = await t.run(async (ctx) => {
+      const first = await ctx.db.insert("events", {
+        value: 1,
+        insertedAt: ctx.db.vars.commitTs,
+      });
+      const second = await ctx.db.insert("events", {
+        value: 2,
+        insertedAt: ctx.db.vars.commitTs,
+      });
+      return [first, second];
+    });
+    const batch = await t.query(internal.example.getBatch, { name: "events" });
+    if (!batch.batch) throw new Error("Expected work");
+    // Explicitly create the gap between the query and mutation snapshots.
+    await t.run(async (ctx) => {
+      await ctx.db.patch("events", ids[0], { value: 10 });
+      await ctx.db.delete("events", ids[1]);
+    });
+    await t.mutation(internal.example.processBatch, batch.batch);
+    await t.mutation(internal.example.processBatch, batch.batch);
+    expect(await t.query(api.example.getTotals, {})).toEqual({
+      total: 10,
+      count: 1,
+      pending: 0,
+    });
+  });
+
   test("batches many events across iterations", async () => {
     const t = initConvexTest();
     // 25 events => 3 batches of 10/10/5.
diff --git a/example/convex/example.ts b/example/convex/example.ts
index 1e0bc46..a24adbd 100644
--- a/example/convex/example.ts
+++ b/example/convex/example.ts
@@ -33,10 +33,8 @@ export const addEvent = mutation({
   },
 });
 
-const vEvent = v.object({ id: v.id("events"), value: v.number() });
-
 const { vQueryArgs, vQueryReturns, vMutationArgs, vMutationReturns } =
-  defineBatchWorkerValidators({ batch: { events: v.array(vEvent) } });
+  defineBatchWorkerValidators({ batch: { ids: v.array(v.id("events")) } });
 
 /**
  * The work query: returns the next batch of work, or `idle` when the queue is
@@ -59,7 +57,7 @@ export const getBatch = internalQuery({
     }
     return {
       kind: "work" as const,
-      batch: { events: events.map((e) => ({ id: e._id, value: e.value })) },
+      batch: { ids: events.map((e) => e._id) },
       // Rows come back in commit order, so the last one is how far we got.
       // The component commits this with the batch and hands it back above.
       cursor: events.at(-1)!.insertedAt,
@@ -75,7 +73,14 @@ export const getBatch = internalQuery({
 export const processBatch = internalMutation({
   args: vMutationArgs,
   returns: vMutationReturns,
-  handler: async (ctx, { events }) => {
+  handler: async (ctx, { ids }) => {
+    // The query can read an older snapshot. Use current values and skip rows
+    // already deleted by a previous batch or another mutation.
+    const candidates = await Promise.all(
+      ids.map((id) => ctx.db.get("events", id)),
+    );
+    const events = candidates.filter((event) => event !== null);
+    if (events.length === 0) return null;
     const sum = events.reduce((a, e) => a + e.value, 0);
     const totals = await ctx.db
       .query("totals")
@@ -93,8 +98,8 @@ export const processBatch = internalMutation({
         count: events.length,
       });
     }
-    for (const { id } of events) {
-      await ctx.db.delete("events", id);
+    for (const { _id } of events) {
+      await ctx.db.delete("events", _id);
     }
   },
 });
diff --git a/example/convex/rateLimited.test.ts b/example/convex/rateLimited.test.ts
index 385e25b..0c29307 100644
--- a/example/convex/rateLimited.test.ts
+++ b/example/convex/rateLimited.test.ts
@@ -1,6 +1,6 @@
 import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
 import { initConvexTest } from "./setup.test";
-import { api } from "./_generated/api";
+import { api, internal } from "./_generated/api";
 
 describe("rate-limited LLM batches", () => {
   beforeEach(() => {
@@ -27,6 +27,55 @@ describe("rate-limited LLM batches", () => {
     expect(requests.every((r) => r.response?.startsWith("Echo:"))).toBe(true);
   });
 
+  test("re-checks stale candidates before reserving tokens and scheduling work", async () => {
+    const t = initConvexTest();
+    const ids = await t.run(async (ctx) => {
+      const ids = [];
+      for (let i = 0; i < 3; i++) {
+        ids.push(
+          await ctx.db.insert("llmRequests", {
+            prompt: `old ${i}`,
+            inputTokens: 1,
+            state: "pending",
+            updatedAt: ctx.db.vars.commitTs,
+          }),
+        );
+      }
+      return ids;
+    });
+    const batch = await t.query(internal.rateLimited.getBatch, { name: "llm" });
+    if (!batch.batch) throw new Error("Expected work");
+    await t.run(async (ctx) => {
+      await ctx.db.patch("llmRequests", ids[0], {
+        prompt: "current prompt",
+        inputTokens: 2,
+      });
+      await ctx.db.patch("llmRequests", ids[1], {
+        state: "finished",
+        response: "already done",
+      });
+      await ctx.db.delete("llmRequests", ids[2]);
+    });
+    await t.mutation(internal.rateLimited.startBatch, batch.batch);
+    const scheduled = await t.run((ctx) =>
+      ctx.db.system.query("_scheduled_functions").collect(),
+    );
+    await t.mutation(internal.rateLimited.startBatch, batch.batch);
+    expect(
+      await t.run((ctx) =>
+        ctx.db.system.query("_scheduled_functions").collect(),
+      ),
+    ).toEqual(scheduled);
+    expect(scheduled).toHaveLength(1);
+    expect(scheduled[0].args).toEqual([
+      { prompts: [{ id: ids[0], prompt: "current prompt" }] },
+    ]);
+    await t.finishAllScheduledFunctions(vi.runAllTimers);
+    expect(
+      await t.run((ctx) => ctx.db.get("llmRequests", ids[1])),
+    ).toMatchObject({ response: "already done" });
+  });
+
   test("claims every request once when one transaction's requests span batches", async () => {
     const t = initConvexTest();
     // 11 requests in one transaction — one commit timestamp, split by
diff --git a/example/convex/rateLimited.ts b/example/convex/rateLimited.ts
index 844bafb..d0f643b 100644
--- a/example/convex/rateLimited.ts
+++ b/example/convex/rateLimited.ts
@@ -68,16 +68,10 @@ export const submitRequest = mutation({
   },
 });
 
-// Carry everything the batch needs — including the prompt — so nothing has to
-// go back to the database to re-read it.
-const vLlmRequest = v.object({
-  id: v.id("llmRequests"),
-  prompt: v.string(),
-  inputTokens: v.number(),
-});
-
+// Return candidate IDs. The mutation checks current state and reads the prompt
+// and token estimate from its own snapshot before claiming each request.
 const { vQueryArgs, vQueryReturns, vMutationArgs, vMutationReturns } =
-  defineBatchWorkerValidators({ batch: { requests: v.array(vLlmRequest) } });
+  defineBatchWorkerValidators({ batch: { ids: v.array(v.id("llmRequests")) } });
 
 export const getBatch = internalQuery({
   args: vQueryArgs,
@@ -103,11 +97,7 @@ export const getBatch = internalQuery({
     return {
       kind: "work" as const,
       batch: {
-        requests: pending.map((r) => ({
-          id: r._id,
-          prompt: r.prompt,
-          inputTokens: r.inputTokens,
-        })),
+        ids: pending.map((r) => r._id),
       },
       // Rows come back in commit order, so the last one is how far we got.
       cursor: pending.at(-1)!.updatedAt,
@@ -122,7 +112,15 @@ export const getBatch = internalQuery({
 export const startBatch = internalMutation({
   args: vMutationArgs,
   returns: vMutationReturns,
-  handler: async (ctx, { requests }) => {
+  handler: async (ctx, { ids }) => {
+    // A stale query can return requests already claimed, finished, or deleted.
+    const candidates = await Promise.all(
+      ids.map((id) => ctx.db.get("llmRequests", id)),
+    );
+    const requests = candidates.filter(
+      (r): r is NonNullable<typeof r> => r !== null && r.state === "pending",
+    );
+    if (requests.length === 0) return null;
     const totalInputTokens = requests.reduce((a, r) => a + r.inputTokens, 0);
 
     // Reserve the whole batch's input tokens up front. `reserve: true` never
@@ -133,13 +131,13 @@ export const startBatch = internalMutation({
       reserve: true,
     });
 
-    // Mark them started so getBatch won't hand them out again. Every request
-    // in the batch gets claimed, so it's safe to move the cursor past them.
+    // Claim only requests that are still pending in this mutation's snapshot.
+    // Previously claimed or deleted candidates can also be passed by the cursor.
     // Patches refresh `updatedAt` too, so the "started" range stays in claim
     // order for the recovery scan sketched in getBatch's TODO.
     const startedAt = Date.now();
-    for (const { id } of requests) {
-      await ctx.db.patch("llmRequests", id, {
+    for (const { _id } of requests) {
+      await ctx.db.patch("llmRequests", _id, {
         state: "started",
         startedAt,
         updatedAt: ctx.db.vars.commitTs,
@@ -152,7 +150,7 @@ export const startBatch = internalMutation({
       retryAfter ?? 0,
       internal.rateLimited.runBatch,
       {
-        prompts: requests.map((r) => ({ id: r.id, prompt: r.prompt })),
+        prompts: requests.map((r) => ({ id: r._id, prompt: r.prompt })),
       },
     );
 

From 9bef49a4c49e4688305c07f2812d14da88ef49e3 Mon Sep 17 00:00:00 2001
From: Ian Macartney <366683+ianmacartney@users.noreply.github.com>
Date: Thu, 17 Sep 2026 23:23:19 -0700
Subject: [PATCH 2/3] docs: clarify that work queries may read older snapshots

---
 README.md                    | 80 ++++++++++++++++++++++++------------
 example/README.md            |  5 +++
 example/convex/aggregates.ts |  5 +++
 src/client/index.ts          |  4 ++
 src/component/loop.ts        |  4 +-
 src/component/shared.ts      |  4 ++
 6 files changed, 74 insertions(+), 28 deletions(-)

diff --git a/README.md b/README.md
index 5e526f2..ee445a3 100644
--- a/README.md
+++ b/README.md
@@ -90,12 +90,12 @@ export const addEvent = mutation({
   },
 });
 
-const vEvents = v.array(v.object({ id: v.id("events"), value: v.number() }));
+const vEventIds = v.array(v.id("events"));
 
 // One declaration gives the query's args (`{ name, cursor? }`) and returns,
 // plus the mutation's args. `cursor` defaults to a commit timestamp.
 const { vQueryArgs, vQueryReturns, vMutationArgs } =
-  defineBatchWorkerValidators({ batch: { events: vEvents } });
+  defineBatchWorkerValidators({ batch: { ids: vEventIds } });
 
 // Return the next batch of work, or `idle` when there's nothing to do.
 export const getBatch = internalQuery({
@@ -114,7 +114,7 @@ export const getBatch = internalQuery({
     }
     return {
       kind: "work" as const,
-      batch: { events: events.map((e) => ({ id: e._id, value: e.value })) },
+      batch: { ids: events.map((e) => e._id) },
       // Rows come back in commit order, so the last one is how far we got.
       cursor: events.at(-1)!.insertedAt,
     };
@@ -124,9 +124,12 @@ export const getBatch = internalQuery({
 // Process one batch. The worker owns cleanup — delete what you process!
 export const processBatch = internalMutation({
   args: vMutationArgs,
-  handler: async (ctx, { events }) => {
-    for (const { value, id } of events) {
-      // ... do the work (sum, call an API, schedule downstream jobs, etc.) ...
+  handler: async (ctx, { ids }) => {
+    for (const id of ids) {
+      // The query may have read an older snapshot. Check the current row.
+      const event = await ctx.db.get("events", id);
+      if (!event) continue; // already processed or deleted
+      // ... do the work using event.value (sum, schedule downstream jobs, etc.) ...
 
       // Clean up the work you processed.
       await ctx.db.delete("events", id);
@@ -140,33 +143,53 @@ The component **does not clean up your work for you** — your worker mutation i
 responsible for deleting (or marking complete / advancing past) the rows it
 processed, otherwise the next query will return them again.
 
-### Fetch work in the query, process it in the mutation
+### Find work in the query, validate it in the mutation
 
-Think of the work query and the worker mutation as **two separate transactions**
-(under the hood they run against the same database snapshot, but they behave
-independently for conflict purposes). This split enables concurrent processing
-without database conflicts on work being added while being fetched.
+Treat the work query and the worker mutation as **two separate transactions**.
+The query may read an older database snapshot than the mutation. There is **no
+guarantee that the rows or values returned by the query are still current** when
+the mutation runs: rows may have changed, been deleted, or already been
+processed.
 
 - The **work query** runs as a snapshot read that takes **no read
   dependencies**. It can scan the queue (`.take(BATCH_SIZE)` over an index)
   alongside concurrent inserts.
 - The **worker mutation** is a normal transaction: every range it reads becomes
   a read dependency. If it re-queries the range of work in the queue, concurrent
-  inserts would conflict, so do range reads in the work query only.
-
-So: **fetch the batch in the query, and pass everything the mutation needs
-through `batch`**, so the mutation doesn't need to re-fetch it. Note: when the
-mutation updates rows by `_id`, it will depend on those documents, so it is
-protected against concurrent changes to those documents.
+  inserts would conflict. Keep queue scans in the query and use point reads by
+  `_id` to validate selected rows in the mutation.
+
+The default pattern is to **return IDs from the query, then re-fetch those rows
+in the mutation**. Check that each row still exists and is eligible for work
+(for example, `state === "pending"`), and compute updates from its current
+values. Skip rows that have already been processed. These point reads
+participate in the mutation's conflict detection without adding a dependency on
+the whole queue.
+
+You can pass values through `batch` when your application can safely process
+older values, such as immutable, append-only events tracked with an exclusive
+cursor. Document that assumption. Patching a row by `_id` takes a dependency in
+the mutation's snapshot; it **does not validate values from the query's older
+snapshot** or establish that the row is still eligible for processing.
 
 ```ts
-// ✅ Query scans; mutation gets the data handed to it.
+// Query: find candidate rows without adding queue-wide read dependencies.
 return {
   kind: "work" as const,
-  batch: { events: rows.map((e) => ({ id: e._id, value: e.value })) },
+  batch: { ids: rows.map((row) => row._id) },
 };
 ```
 
+```ts
+// Mutation: validate each candidate in the snapshot used for the writes.
+for (const id of ids) {
+  const row = await ctx.db.get("tasks", id);
+  if (!row || row.state !== "pending") continue;
+  await ctx.db.patch("tasks", id, { state: "started" });
+  // Schedule work using the current row's values.
+}
+```
+
 ### Steering the loop dynamically
 
 Your worker mutation may return `{ debounceMs }` to throttle the loop, and a
@@ -235,7 +258,8 @@ await ping(ctx, components.batchWorker, {
 The component stores the cursor for you: whatever your work query returns
 alongside a batch comes back as `args.cursor` on the next call. It's committed
 in the same transaction as the batch, so if your worker mutation throws, the
-cursor stays where it was and the batch is handed out again.
+cursor stays where it was and the query retries from that position. The retry
+may return a different batch as the query's snapshot changes.
 
 The default cursor type is a
 [commit timestamp](https://docs.convex.dev/database/advanced/commit-timestamp)
@@ -364,10 +388,11 @@ the value. If you need the fully-up-to-date value, you have a couple options:
 
 The worker mutation is a transaction, so it can't call external APIs itself. To
 batch work that requires `fetch`— calling an LLM, hitting a third-party API —
-have the worker mutation _claim_ a batch (mark the rows started so the query
-won't hand them out again) and then schedule the action, or enqueue it in a
-Workpool. The action does the work and calls a mutation to commit the results
-back.
+have the worker mutation _claim_ a batch: re-fetch the candidate IDs, check that
+each row is still pending, and mark those rows started. Then schedule the
+action, or enqueue it in a Workpool, using only the rows actually claimed. A
+stale query can return previously claimed rows, so the mutation must skip them.
+The action does the work and calls a mutation to commit the results back.
 
 Full working code is in [rateLimited.ts](./example/convex/rateLimited.ts). It
 also rate-limits the batches, covered next.
@@ -420,9 +445,10 @@ 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
-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
+monitor restarts it after ~`monitorLagMs`. The failed mutation's writes and
+cursor advance roll back, so the query retries from the previous cursor. Its
+next batch may differ because it can read a different snapshot. Design for
+repeated delivery and re-check rows in the mutation before applying effects. One
 poison item that always throws can wedge the queue. For work that can fail per
 item, catch errors inside the worker mutation, and isolate bad docs in a table
 for async debugging.
diff --git a/example/README.md b/example/README.md
index 27f41ed..ffdd9ee 100644
--- a/example/README.md
+++ b/example/README.md
@@ -13,6 +13,11 @@ All three drain their queue by returning a `v.commitTs()` cursor from the work
 query. The component commits it with the batch and hands it back on the next
 call, so each scan resumes where the last one stopped.
 
+The work query may read an older snapshot than the worker mutation. The queue
+and LLM examples return IDs and re-fetch current rows in the mutation, skipping
+deleted or already-claimed work. The scoreboard passes values directly because
+its score events are immutable and its exclusive cursor tracks processed events.
+
 Run all commands from the **root of the repo**.
 
 ## Develop
diff --git a/example/convex/aggregates.ts b/example/convex/aggregates.ts
index 7211b17..7af2c67 100644
--- a/example/convex/aggregates.ts
+++ b/example/convex/aggregates.ts
@@ -14,6 +14,11 @@ import {
 // batches of them into per-team totals. Because exactly one loop runs at a
 // time, those hot aggregate rows have a single writer — so they never OCC
 // against each other or against incoming scores, no matter the rate.
+//
+// This example treats scoreEvents as immutable, append-only events. Passing
+// their values through the batch is safe under that assumption; an exclusive
+// cursor skips events already folded into the totals. If events can be edited
+// or deleted, return IDs and re-check their current state in the mutation.
 
 const WORKER = "aggregates";
 
diff --git a/src/client/index.ts b/src/client/index.ts
index d2a741f..ec310c7 100644
--- a/src/client/index.ts
+++ b/src/client/index.ts
@@ -41,6 +41,10 @@ export type {
  *  - a **worker mutation** that processes a batch and owns its cleanup. It may
  *    return `{ debounceMs }` to throttle the loop.
  *
+ * The work query may read an older snapshot than the worker mutation. Return
+ * candidate IDs and re-fetch them in the mutation when processing depends on
+ * current values or state. A write by ID does not validate the query's values.
+ *
  * Validate both with {@link defineBatchWorkerValidators}. The cursor type is
  * taken from the work query's `cursor` arg, and both return types are checked
  * against it.
diff --git a/src/component/loop.ts b/src/component/loop.ts
index d2243bd..066640d 100644
--- a/src/component/loop.ts
+++ b/src/component/loop.ts
@@ -61,7 +61,9 @@ 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
+    // drain don't force this loop to retry. The query's snapshot may be older
+    // than the worker mutation's; the worker must validate candidate rows when
+    // it needs current state. If the query or worker mutation
     // throws, this loop fails (and doesn't reschedule) — the monitor restarts
     // it.
     const result = await ctx.runQuery(queryRef, queryArgs, {
diff --git a/src/component/shared.ts b/src/component/shared.ts
index c3f8659..fb880ca 100644
--- a/src/component/shared.ts
+++ b/src/component/shared.ts
@@ -97,6 +97,10 @@ export type BatchQueryArgs<Cursor = DefaultCursor> = {
  * explicit `idle` (optionally with a `timeoutMs` hint for when to check again
  * — e.g. when the next item is scheduled).
  *
+ * The query may read an older snapshot than the worker mutation. Treat the
+ * batch as candidates: re-fetch by ID and validate current state in the mutation
+ * unless the application explicitly tolerates processing older values.
+ *
  * Alongside a batch, return a `cursor` saying how far this batch got. The
  * component commits it with the batch, and hands it back as `args.cursor` on
  * the next call so the scan resumes there. It is committed only if the worker

From e7d291c3af83633d895c0568a6615f7d500452b9 Mon Sep 17 00:00:00 2001
From: Ian Macartney <366683+ianmacartney@users.noreply.github.com>
Date: Fri, 18 Sep 2026 00:19:34 -0700
Subject: [PATCH 3/3] docs: clarify read-your-writes guarantees between batches

---
 README.md                     | 71 +++++++++++++++++++++--------------
 example/README.md             | 11 ++++--
 example/convex/example.ts     |  4 +-
 example/convex/rateLimited.ts |  9 +++--
 src/client/index.ts           |  8 ++--
 src/component/loop.ts         |  8 ++--
 src/component/shared.ts       |  7 ++--
 7 files changed, 69 insertions(+), 49 deletions(-)

diff --git a/README.md b/README.md
index ee445a3..2b672a7 100644
--- a/README.md
+++ b/README.md
@@ -126,9 +126,9 @@ export const processBatch = internalMutation({
   args: vMutationArgs,
   handler: async (ctx, { ids }) => {
     for (const id of ids) {
-      // The query may have read an older snapshot. Check the current row.
+      // Re-fetch if other mutations may edit or delete queued events.
       const event = await ctx.db.get("events", id);
-      if (!event) continue; // already processed or deleted
+      if (!event) continue; // deleted by another mutation
       // ... do the work using event.value (sum, schedule downstream jobs, etc.) ...
 
       // Clean up the work you processed.
@@ -143,13 +143,17 @@ The component **does not clean up your work for you** — your worker mutation i
 responsible for deleting (or marking complete / advancing past) the rows it
 processed, otherwise the next query will return them again.
 
-### Find work in the query, validate it in the mutation
+### Find work in the query, process it in the mutation
 
 Treat the work query and the worker mutation as **two separate transactions**.
-The query may read an older database snapshot than the mutation. There is **no
-guarantee that the rows or values returned by the query are still current** when
-the mutation runs: rows may have changed, been deleted, or already been
-processed.
+The query may read an older database snapshot than the mutation, but **each
+round's query sees the committed writes from this worker's previous rounds**. If
+a batch deletes rows or marks them started, the next query observes those
+changes, so it can exclude that work from its results.
+
+Other mutations can still change, delete, or claim rows between the query's
+snapshot and the worker mutation's snapshot. Revalidate in the mutation when
+processing depends on those concurrent changes.
 
 - The **work query** runs as a snapshot read that takes **no read
   dependencies**. It can scan the queue (`.take(BATCH_SIZE)` over an index)
@@ -157,23 +161,27 @@ processed.
 - The **worker mutation** is a normal transaction: every range it reads becomes
   a read dependency. If it re-queries the range of work in the queue, concurrent
   inserts would conflict. Keep queue scans in the query and use point reads by
-  `_id` to validate selected rows in the mutation.
+  `_id` when selected rows need revalidation in the mutation.
+
+If other writers can change the work, **return IDs from the query, then re-fetch
+those rows in the mutation**. Check that each row still exists and is eligible
+for work (for example, `state === "pending"`), and compute updates from its
+current values. These point reads participate in the mutation's conflict
+detection without adding a dependency on the whole queue.
 
-The default pattern is to **return IDs from the query, then re-fetch those rows
-in the mutation**. Check that each row still exists and is eligible for work
-(for example, `state === "pending"`), and compute updates from its current
-values. Skip rows that have already been processed. These point reads
-participate in the mutation's conflict detection without adding a dependency on
-the whole queue.
+You can **pass values directly through `batch`** when they cannot change before
+processing, or when your application can safely process older values. For
+example, if events are immutable until this worker processes and deletes them,
+and no other consumer edits or deletes them, there is no need to re-fetch just
+to check for deletes from earlier rounds. If events are retained, an exclusive
+cursor can track which ones have been processed. Document these assumptions.
 
-You can pass values through `batch` when your application can safely process
-older values, such as immutable, append-only events tracked with an exclusive
-cursor. Document that assumption. Patching a row by `_id` takes a dependency in
-the mutation's snapshot; it **does not validate values from the query's older
-snapshot** or establish that the row is still eligible for processing.
+Patching a row by `_id` takes a dependency in the mutation's snapshot; it **does
+not validate values from the query's older snapshot** or establish that another
+writer has left the row eligible for processing.
 
 ```ts
-// Query: find candidate rows without adding queue-wide read dependencies.
+// When other writers can change the work, return IDs for revalidation.
 return {
   kind: "work" as const,
   batch: { ids: rows.map((row) => row._id) },
@@ -181,7 +189,7 @@ return {
 ```
 
 ```ts
-// Mutation: validate each candidate in the snapshot used for the writes.
+// Mutation: check for concurrent changes before applying effects.
 for (const id of ids) {
   const row = await ctx.db.get("tasks", id);
   if (!row || row.state !== "pending") continue;
@@ -390,9 +398,13 @@ The worker mutation is a transaction, so it can't call external APIs itself. To
 batch work that requires `fetch`— calling an LLM, hitting a third-party API —
 have the worker mutation _claim_ a batch: re-fetch the candidate IDs, check that
 each row is still pending, and mark those rows started. Then schedule the
-action, or enqueue it in a Workpool, using only the rows actually claimed. A
-stale query can return previously claimed rows, so the mutation must skip them.
-The action does the work and calls a mutation to commit the results back.
+action, or enqueue it in a Workpool, using only the rows actually claimed. The
+next round's query sees this worker's claims and excludes those rows from its
+pending results. Re-fetching protects against other mutations changing or
+claiming requests between snapshots; if this worker is the only consumer and
+pending requests cannot change, it can claim them using values passed through
+`batch`. The action does the work and calls a mutation to commit the results
+back.
 
 Full working code is in [rateLimited.ts](./example/convex/rateLimited.ts). It
 also rate-limits the batches, covered next.
@@ -447,11 +459,12 @@ Tips for rate-limiting LLM calls:
 If your work query or worker mutation throws, the loop dies and the liveness
 monitor restarts it after ~`monitorLagMs`. The failed mutation's writes and
 cursor advance roll back, so the query retries from the previous cursor. Its
-next batch may differ because it can read a different snapshot. Design for
-repeated delivery and re-check rows in the mutation before applying effects. One
-poison item that always throws can wedge the queue. For work that can fail per
-item, catch errors inside the worker mutation, and isolate bad docs in a table
-for async debugging.
+next batch may differ because other mutations can change the work before the
+retry. Successful batches' committed writes remain visible to later rounds, so
+their cleanup is not lost on retry. Re-check rows if other writers can change
+them. One poison item that always throws can wedge the queue. For work that can
+fail per item, catch errors inside the worker mutation, and isolate bad docs in
+a table for async debugging.
 
 This is a low-level primitive, relative to components like Workpool or Workflow,
 so you have to handle exceptional cases yourself.
diff --git a/example/README.md b/example/README.md
index ffdd9ee..1ef22b6 100644
--- a/example/README.md
+++ b/example/README.md
@@ -13,10 +13,13 @@ All three drain their queue by returning a `v.commitTs()` cursor from the work
 query. The component commits it with the batch and hands it back on the next
 call, so each scan resumes where the last one stopped.
 
-The work query may read an older snapshot than the worker mutation. The queue
-and LLM examples return IDs and re-fetch current rows in the mutation, skipping
-deleted or already-claimed work. The scoreboard passes values directly because
-its score events are immutable and its exclusive cursor tracks processed events.
+The work query may read an older snapshot than the worker mutation, but each
+round sees this worker's committed writes from previous rounds, including
+deletes and claims. The queue and LLM examples return IDs and re-fetch current
+rows to handle edits, deletes, or claims by other mutations between snapshots.
+If only this worker can change or remove queued work, values can be passed
+directly through the batch. The scoreboard does this because its score events
+are immutable and its exclusive cursor tracks processed events.
 
 Run all commands from the **root of the repo**.
 
diff --git a/example/convex/example.ts b/example/convex/example.ts
index a24adbd..57c5e58 100644
--- a/example/convex/example.ts
+++ b/example/convex/example.ts
@@ -74,8 +74,8 @@ export const processBatch = internalMutation({
   args: vMutationArgs,
   returns: vMutationReturns,
   handler: async (ctx, { ids }) => {
-    // The query can read an older snapshot. Use current values and skip rows
-    // already deleted by a previous batch or another mutation.
+    // Re-fetch to handle edits or deletes by other mutations between snapshots.
+    // The query already sees this worker's deletes from previous rounds.
     const candidates = await Promise.all(
       ids.map((id) => ctx.db.get("events", id)),
     );
diff --git a/example/convex/rateLimited.ts b/example/convex/rateLimited.ts
index d0f643b..56bee8d 100644
--- a/example/convex/rateLimited.ts
+++ b/example/convex/rateLimited.ts
@@ -68,8 +68,8 @@ export const submitRequest = mutation({
   },
 });
 
-// Return candidate IDs. The mutation checks current state and reads the prompt
-// and token estimate from its own snapshot before claiming each request.
+// Return IDs so the mutation can check for changes by other mutations before
+// claiming each request, using its current prompt and token estimate.
 const { vQueryArgs, vQueryReturns, vMutationArgs, vMutationReturns } =
   defineBatchWorkerValidators({ batch: { ids: v.array(v.id("llmRequests")) } });
 
@@ -113,7 +113,8 @@ export const startBatch = internalMutation({
   args: vMutationArgs,
   returns: vMutationReturns,
   handler: async (ctx, { ids }) => {
-    // A stale query can return requests already claimed, finished, or deleted.
+    // Other mutations may have changed, claimed, or deleted these requests
+    // since the query's snapshot. This worker's prior claims are already visible.
     const candidates = await Promise.all(
       ids.map((id) => ctx.db.get("llmRequests", id)),
     );
@@ -132,7 +133,7 @@ export const startBatch = internalMutation({
     });
 
     // Claim only requests that are still pending in this mutation's snapshot.
-    // Previously claimed or deleted candidates can also be passed by the cursor.
+    // The next round's query sees these claims and excludes them from pending.
     // Patches refresh `updatedAt` too, so the "started" range stays in claim
     // order for the recovery scan sketched in getBatch's TODO.
     const startedAt = Date.now();
diff --git a/src/client/index.ts b/src/client/index.ts
index ec310c7..60b0d32 100644
--- a/src/client/index.ts
+++ b/src/client/index.ts
@@ -41,9 +41,11 @@ export type {
  *  - a **worker mutation** that processes a batch and owns its cleanup. It may
  *    return `{ debounceMs }` to throttle the loop.
  *
- * The work query may read an older snapshot than the worker mutation. Return
- * candidate IDs and re-fetch them in the mutation when processing depends on
- * current values or state. A write by ID does not validate the query's values.
+ * The work query may read an older snapshot than the worker mutation, but each
+ * round sees this worker's committed writes from previous rounds. Return IDs
+ * and re-fetch them when processing depends on changes by other writers;
+ * otherwise, values can be passed through the batch. A write by ID does not
+ * validate the query's values.
  *
  * Validate both with {@link defineBatchWorkerValidators}. The cursor type is
  * taken from the work query's `cursor` arg, and both return types are checked
diff --git a/src/component/loop.ts b/src/component/loop.ts
index 066640d..4fcb29a 100644
--- a/src/component/loop.ts
+++ b/src/component/loop.ts
@@ -62,10 +62,10 @@ export const loop = internalMutation({
 
     // Stale snapshot read: no OCC dependency, so concurrent inserts while we
     // drain don't force this loop to retry. The query's snapshot may be older
-    // than the worker mutation's; the worker must validate candidate rows when
-    // it needs current state. If the query or worker mutation
-    // throws, this loop fails (and doesn't reschedule) — the monitor restarts
-    // it.
+    // than the worker mutation's, but includes this worker's committed writes
+    // from previous rounds. Revalidate rows when processing depends on changes
+    // by other writers. If the query or worker mutation throws, this loop fails
+    // (and doesn't reschedule) — the monitor restarts it.
     const result = await ctx.runQuery(queryRef, queryArgs, {
       useStaleSnapshot: true,
     });
diff --git a/src/component/shared.ts b/src/component/shared.ts
index fb880ca..17526ea 100644
--- a/src/component/shared.ts
+++ b/src/component/shared.ts
@@ -97,9 +97,10 @@ export type BatchQueryArgs<Cursor = DefaultCursor> = {
  * explicit `idle` (optionally with a `timeoutMs` hint for when to check again
  * — e.g. when the next item is scheduled).
  *
- * The query may read an older snapshot than the worker mutation. Treat the
- * batch as candidates: re-fetch by ID and validate current state in the mutation
- * unless the application explicitly tolerates processing older values.
+ * The query may read an older snapshot than the worker mutation, but each round
+ * sees this worker's committed writes from previous rounds. Re-fetch by ID when
+ * processing depends on changes by other writers. Values can be passed through
+ * when they cannot change before processing or older values are acceptable.
  *
  * Alongside a batch, return a `cursor` saying how far this batch got. The
  * component commits it with the batch, and hands it back as `args.cursor` on