fix(imessage): clamp catchup watermark to last dispatched row when truncating

omarshahine · claude · omarshahine · commit 92c243a11e7d · 2026-05-08T18:39:55.000-07:00
clawsweeper review on PR #79387 caught a regression I introduced when adding the parse-rejected high-watermark fix in 8c6421b: when the cross-chat fetch returns more valid rows than `perRunLimit`, the fetcher caps `capped = collected.slice(0, limit)` but `rawWatermarkRowid` still reflects the highest rowid across ALL raw rows — including the cap-truncated tail. The catchup loop applies that watermark as a cursor floor, so the persisted cursor leapfrogs past valid-but- undispatched rows. Same silent message-loss class as the original codex finding, just on a different code path. Concrete repro in `catchup-bridge.test.ts > caps cross-chat results at perRunLimit, oldest first`: 8 valid rows (chat 1: rowids 100-103, chat 2: rowids 200-203), perRunLimit=5. Dispatched: 100, 101, 102, 103, 200. Before this fix: cursor persisted at 203 (raw watermark of chat 2's last row), so rowids 201, 202, 203 — explicitly promised by the WARN log to be picked up on the next startup — were silently dropped. After this fix: cursor stops at 200 (last dispatched rowid). Fix: in `catchup-bridge.ts`, derive `effectiveWatermarkRowid` / `effectiveWatermarkMs` from `rawWatermarkRowid` / `rawWatermarkMs` clamped to `capped[capped.length-1]` when truncation hits. When no truncation, the watermark still covers parse-rejected rows interspersed with the dispatched batch (the original forward-progress guarantee from 8c6421b). When `capped.length === 0` (pathological limit=0 or all rows parse-rejected past the cap boundary), suppress the watermark entirely so the cursor stays put. Regression test: extended the existing "caps cross-chat results at perRunLimit, oldest first" test to assert `summary.cursorAfter.lastSeenRowid === 200` (the last dispatched rowid), not 203 (the raw watermark). Acceptance criteria from clawsweeper: - pnpm vitest run extensions/imessage/src/monitor/catchup.test.ts extensions/imessage/src/monitor/catchup-bridge.test.ts → 29/29 pass - pnpm exec oxfmt --check --threads=1 ... → All matched files use the correct format - pnpm lint:extensions:bundled → 0 warnings, 0 errors Also drops `[...collected].sort()` for `collected.toSorted()` in the same block (oxlint no-array-sort, latent in the 8c6421b commit because oxlint missed it on one path). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
diff --git a/extensions/imessage/src/monitor/catchup-bridge.test.ts b/extensions/imessage/src/monitor/catchup-bridge.test.ts
@@ -270,6 +270,15 @@ describe("runIMessageCatchup", () => {
     expect(summary.replayed).toBe(5);
     // Oldest-first by rowid: 100, 101, 102, 103, 200 (chat 1's first 4, then chat 2's first).
     expect(dispatched).toEqual(["g-100", "g-101", "g-102", "g-103", "g-200"]);
+    // Regression for clawsweeper #79387 finding: the cursor must NOT
+    // advance past the last dispatched row when perRunLimit truncates
+    // the cross-chat page. Without the cap-aware watermark clamp, the
+    // bridge would emit a watermark covering the raw rows it dropped
+    // (rowids 201, 202, 203 from chat 2), and the catchup loop would
+    // persist `lastSeenRowid` past them — so the promised "next startup
+    // picks up the rest" warning would lie and those rows would be
+    // permanently lost. Cursor must stop at the last dispatched rowid (200).
+    expect(summary.cursorAfter.lastSeenRowid).toBe(200);
   });
 
   it("treats a dispatch throw as a failure and holds the cursor", async () => {
diff --git a/extensions/imessage/src/monitor/catchup-bridge.ts b/extensions/imessage/src/monitor/catchup-bridge.ts
@@ -193,17 +193,16 @@ export async function runIMessageCatchup(
       }
     }
 
-    collected.sort((a, b) => a.rowid - b.rowid);
-    const capped = collected.slice(0, limit);
-    if (capped.length < collected.length) {
+    const sorted = collected.toSorted((a, b) => a.rowid - b.rowid);
+    const capped = sorted.slice(0, limit);
+    const isCapTruncated = capped.length < sorted.length;
+    if (isCapTruncated) {
       warnLog(
-        `imessage catchup: fetched ${collected.length} rows across chats, ` +
+        `imessage catchup: fetched ${sorted.length} rows across chats, ` +
           `capped to perRunLimit=${limit} (oldest first); next startup picks up the rest`,
       );
-    }
-    // Drop payloads we are no longer going to dispatch so the dispatch
-    // adapter does not have to defend against the discarded ones.
-    if (capped.length < collected.length) {
+      // Drop payloads we are no longer going to dispatch so the dispatch
+      // adapter does not have to defend against the discarded ones.
       const keep = new Set(capped.map((row) => row.guid));
       for (const guid of payloadByGuid.keys()) {
         if (!keep.has(guid)) {
@@ -212,11 +211,36 @@ export async function runIMessageCatchup(
       }
     }
 
+    // Clamp the raw watermark when cap-truncation hits so the catchup loop
+    // cannot persist a cursor past undispatched valid rows. Without this,
+    // a `messages.history` page wider than `perRunLimit` would silently
+    // skip the cap-truncated tail forever — the WARN above promises the
+    // next startup picks up the rest, and that promise relies on the
+    // cursor staying at the last dispatched rowid. When no truncation
+    // happens, the watermark covers parse-rejected rows interspersed
+    // with the dispatched batch (the original forward-progress fix).
+    let effectiveWatermarkRowid = rawWatermarkRowid;
+    let effectiveWatermarkMs = rawWatermarkMs;
+    if (isCapTruncated && capped.length > 0) {
+      const last = capped.at(-1);
+      if (last) {
+        effectiveWatermarkRowid = Math.min(effectiveWatermarkRowid, last.rowid);
+        effectiveWatermarkMs = Math.min(effectiveWatermarkMs, last.date);
+      }
+    } else if (isCapTruncated && capped.length === 0) {
+      // Pathological: cap=0. Don't emit any watermark; preserve the prior
+      // cursor and let the next pass try again.
+      effectiveWatermarkRowid = Number.NaN;
+      effectiveWatermarkMs = Number.NaN;
+    }
+
     return {
       resolved: true,
       rows: capped,
-      ...(Number.isFinite(rawWatermarkRowid) ? { highWatermarkRowid: rawWatermarkRowid } : {}),
-      ...(Number.isFinite(rawWatermarkMs) ? { highWatermarkMs: rawWatermarkMs } : {}),
+      ...(Number.isFinite(effectiveWatermarkRowid)
+        ? { highWatermarkRowid: effectiveWatermarkRowid }
+        : {}),
+      ...(Number.isFinite(effectiveWatermarkMs) ? { highWatermarkMs: effectiveWatermarkMs } : {}),
     };
   };
 
diff --git a/src/config/bundled-channel-config-metadata.generated.ts b/src/config/bundled-channel-config-metadata.generated.ts
