Persist batch item checkpoints during recovery (#5753)

* fix(sse): checkpoint batch item recovery * fix(db): renumber batch checkpoints migration 110→112 (collision with #5667) 110 was taken by 110_model_context_overrides.sql (#5667), which landed on the release branch after this PR branched. migrationRunner throws a hard version- collision error on startup when two files share a numeric prefix. 112 is the next free slot (110/111 taken on the release tip). Co-authored-by: diegosouzapw <diegosouza.pw@gmail.com> --------- Co-authored-by: diegosouzapw <diegosouza.pw@gmail.com>
2026-08-04 06:12:10 +03:00 · 2026-06-30 23:24:22 -05:00
parent a176e0e67e
commit a219ab5ee3
6 changed files with 484 additions and 113 deletions
--- a/open-sse/services/batchProcessor.ts
+++ b/open-sse/services/batchProcessor.ts
@@ -1,14 +1,20 @@
 import { v4 as uuidv4 } from "uuid";
-import type { BatchRecord } from "@/lib/localDb";
+import type { BatchItemCheckpoint, BatchRecord } from "@/lib/localDb";
 import {
+  countBatchItemCheckpoints,
  createFile,
  deleteFile,
+  ensureBatchItemCheckpoints,
  getApiKeyById,
  getBatch,
  getFileContent,
  getPendingBatches,
  getTerminalBatches,
+  listBatchItemCheckpoints,
  listFiles,
+  markBatchItemError,
+  markBatchItemProcessing,
+  markBatchItemResult,
  updateBatch,
 } from "@/lib/localDb";
 import { dispatch } from "@/lib/batches/dispatch";
@@ -67,28 +73,12 @@ export async function processPendingBatches(): Promise<void> {
  const pending = getPendingBatches();

  // Phase 1: Stale recovery — in_progress/finalizing batches not in activeBatches
-  // are from a previous session; reset them to validating so they get picked up fresh
+  // are from a previous session; reset checkpointed batches to validating so they
+  // can be completed without replaying already-dispatched items.
  for (const batch of pending) {
    if (batch.status === "in_progress" || batch.status === "finalizing") {
      if (!activeBatches.has(batch.id)) {
-        console.log(`[BATCH] Recovering stale batch ${batch.id} (${batch.status}) → validating`);
-
-        if (batch.outputFileId) {
-          deleteFile(batch.outputFileId);
-        }
-        if (batch.errorFileId) {
-          deleteFile(batch.errorFileId);
-        }
-
-        updateBatch(batch.id, {
-          status: "validating",
-          inProgressAt: null,
-          finalizingAt: null,
-          outputFileId: null,
-          errorFileId: null,
-          requestCountsCompleted: 0,
-          requestCountsFailed: 0,
-        });
+        recoverStaleBatch(batch);
      }
    }
  }
@@ -116,6 +106,49 @@ export async function processPendingBatches(): Promise<void> {
  await cleanupExpiredBatches();
 }

+function recoverStaleBatch(batch: BatchRecord): void {
+  const checkpointCount = countBatchItemCheckpoints(batch.id);
+  const hasPotentialExternalEffects =
+    batch.requestCountsTotal > 0 ||
+    batch.requestCountsCompleted > 0 ||
+    batch.requestCountsFailed > 0 ||
+    batch.status === "finalizing";
+
+  if (checkpointCount === 0 && hasPotentialExternalEffects) {
+    console.warn(
+      `[BATCH] Stale batch ${batch.id} has no item checkpoints; failing instead of replaying provider calls`
+    );
+    failBatch(
+      batch.id,
+      "Cannot safely recover stale batch because item checkpoints are unavailable; create a new batch to retry intentionally."
+    );
+    return;
+  }
+
+  console.log(`[BATCH] Recovering stale batch ${batch.id} (${batch.status}) → validating`);
+
+  if (batch.outputFileId) {
+    deleteFile(batch.outputFileId);
+  }
+  if (batch.errorFileId) {
+    deleteFile(batch.errorFileId);
+  }
+
+  updateBatch(batch.id, {
+    status: "validating",
+    inProgressAt: null,
+    finalizingAt: null,
+    outputFileId: null,
+    errorFileId: null,
+    ...(checkpointCount === 0
+      ? {
+          requestCountsCompleted: 0,
+          requestCountsFailed: 0,
+        }
+      : {}),
+  });
+}
+
 function parseBatchWindowSeconds(window: string | null | undefined): number {
  if (!window) return DEFAULT_BATCH_WINDOW_SECONDS;
  const match = /^(\d+)([hdm])$/.exec(window);
@@ -285,6 +318,8 @@ async function startBatch(batch: any): Promise<void> {

    console.log(`[BATCH] Batch ${batch.id} contains (${total} items)`);

+    ensureBatchItemCheckpoints(batch.id, parsedItems.items);
+
    updateBatch(batch.id, {
      status: "in_progress",
      inProgressAt: Math.floor(Date.now() / 1000),
@@ -316,12 +351,21 @@ const HEADERS_CACHE_TTL_MS = 60_000;

 async function processBatchItems(batch: BatchRecord, items: BatchRequestItem[]): Promise<void> {
  const state = createBatchState(batch);
+  const checkpoints = new Map<number, BatchItemCheckpoint>(
+    listBatchItemCheckpoints(batch.id).map((checkpoint) => [checkpoint.lineNumber, checkpoint])
+  );

  const apiKey = await resolveApiKey(batch);

  for (const item of items) {
    if (isBatchCancelled(batch.id)) break;

+    const checkpoint = checkpoints.get(item.lineNumber);
+    if (checkpoint && applyRecoveredCheckpoint(batch.id, item, checkpoint, state)) {
+      maybePersistProgress(batch.id, state);
+      continue;
+    }
+
    const cachedHeaders =
      prevHeaders && Date.now() - prevHeadersTimestamp < HEADERS_CACHE_TTL_MS ? prevHeaders : null;
    if (cachedHeaders) {
@@ -331,6 +375,8 @@ async function processBatchItems(batch: BatchRecord, items: BatchRequestItem[]):
      }
    }

+    markBatchItemProcessing(batch.id, item);
+
    try {
      const response = await processSingleItemWithRetry(item, apiKey);
      let responseBody: unknown;
@@ -352,13 +398,17 @@ async function processBatchItems(batch: BatchRecord, items: BatchRequestItem[]):
        },
      };

+      markBatchItemResult(batch.id, item, wrapped);
      state.results.push(wrapped);
      applyItemResult(state, response.status, responseBody);
      prevHeaders = response.headers;
      prevHeadersTimestamp = Date.now();
    } catch (exception) {
      // Track processing-level errors separately (items that failed to be processed)
-      state.errors.push({ custom_id: item.customId ?? null, error: String(exception) });
+      const error = { custom_id: item.customId ?? null, error: String(exception) };
+      markBatchItemError(batch.id, item, error);
+      state.errors.push(error);
+      state.failed++;
      prevHeaders = null;
      prevHeadersTimestamp = 0;
    }
@@ -369,6 +419,43 @@ async function processBatchItems(batch: BatchRecord, items: BatchRequestItem[]):
  return finalizeBatch(batch.id, state.results, state.errors);
 }

+function applyRecoveredCheckpoint(
+  batchId: string,
+  item: BatchRequestItem,
+  checkpoint: BatchItemCheckpoint,
+  state: ReturnType<typeof createBatchState>
+): boolean {
+  if (checkpoint.status === "completed" && checkpoint.result) {
+    state.results.push(checkpoint.result);
+    applyItemResult(
+      state,
+      checkpoint.result.response?.status_code ?? 500,
+      checkpoint.result.response?.body
+    );
+    return true;
+  }
+
+  if (checkpoint.status === "errored" && checkpoint.error) {
+    state.errors.push(checkpoint.error);
+    state.failed++;
+    return true;
+  }
+
+  if (checkpoint.status === "processing") {
+    const error = {
+      custom_id: item.customId ?? null,
+      error:
+        "Batch item was interrupted before its provider response was recorded; it was not replayed to avoid duplicate provider work.",
+    };
+    markBatchItemError(batchId, item, error);
+    state.errors.push(error);
+    state.failed++;
+    return true;
+  }
+
+  return false;
+}
+
 function isBatchCancelled(batchId: string): boolean {
  const current = getBatch(batchId);

--- a/src/lib/db/batches.ts
+++ b/src/lib/db/batches.ts
@@ -82,6 +82,42 @@ export interface BatchRecord {
  outputExpiresAfterAnchor?: string | null;
 }

+export type BatchItemCheckpointStatus = "pending" | "processing" | "completed" | "errored";
+
+export interface BatchItemCheckpoint {
+  batchId: string;
+  lineNumber: number;
+  customId: string | null;
+  status: BatchItemCheckpointStatus;
+  result: any | null;
+  error: any | null;
+  createdAt: number;
+  updatedAt: number;
+}
+
+function parseJsonColumn(value: unknown): any | null {
+  if (value == null) return null;
+  if (typeof value !== "string") return value;
+  try {
+    return JSON.parse(value);
+  } catch {
+    return null;
+  }
+}
+
+function parseBatchItemCheckpoint(row: any): BatchItemCheckpoint {
+  return {
+    batchId: row.batch_id,
+    lineNumber: Number(row.line_number),
+    customId: row.custom_id ?? null,
+    status: row.status,
+    result: parseJsonColumn(row.result_json),
+    error: parseJsonColumn(row.error_json),
+    createdAt: Number(row.created_at),
+    updatedAt: Number(row.updated_at),
+  };
+}
+
 export function createBatch(
  batch: Omit<
    BatchRecord,
@@ -156,6 +192,128 @@ export function updateBatch(id: string, updates: Partial<BatchRecord>): boolean
  return result.changes > 0;
 }

+export function ensureBatchItemCheckpoints(
+  batchId: string,
+  items: Array<{ lineNumber: number; customId: string | null }>
+): void {
+  if (items.length === 0) return;
+
+  const db = getDbInstance();
+  const now = Math.floor(Date.now() / 1000);
+  const insert = db.prepare(`
+    INSERT OR IGNORE INTO batch_item_checkpoints (
+      batch_id,
+      line_number,
+      custom_id,
+      status,
+      result_json,
+      error_json,
+      created_at,
+      updated_at
+    )
+    VALUES (?, ?, ?, 'pending', NULL, NULL, ?, ?)
+  `);
+
+  const tx = db.transaction(() => {
+    for (const item of items) {
+      insert.run(batchId, item.lineNumber, item.customId, now, now);
+    }
+  });
+  tx();
+}
+
+export function countBatchItemCheckpoints(batchId: string): number {
+  const db = getDbInstance();
+  const row = db
+    .prepare("SELECT COUNT(*) AS c FROM batch_item_checkpoints WHERE batch_id = ?")
+    .get(batchId) as { c: number } | undefined;
+  return row ? Number(row.c) : 0;
+}
+
+export function listBatchItemCheckpoints(batchId: string): BatchItemCheckpoint[] {
+  const db = getDbInstance();
+  const rows = db
+    .prepare(
+      `
+      SELECT batch_id, line_number, custom_id, status, result_json, error_json, created_at, updated_at
+      FROM batch_item_checkpoints
+      WHERE batch_id = ?
+      ORDER BY line_number ASC
+    `
+    )
+    .all(batchId);
+  return rows.map((row) => parseBatchItemCheckpoint(row));
+}
+
+export function markBatchItemProcessing(
+  batchId: string,
+  item: { lineNumber: number; customId: string | null }
+): void {
+  const db = getDbInstance();
+  const now = Math.floor(Date.now() / 1000);
+  db.prepare(
+    `
+    INSERT INTO batch_item_checkpoints (
+      batch_id,
+      line_number,
+      custom_id,
+      status,
+      result_json,
+      error_json,
+      created_at,
+      updated_at
+    )
+    VALUES (?, ?, ?, 'processing', NULL, NULL, ?, ?)
+    ON CONFLICT(batch_id, line_number) DO UPDATE SET
+      custom_id = excluded.custom_id,
+      status = 'processing',
+      result_json = NULL,
+      error_json = NULL,
+      updated_at = excluded.updated_at
+  `
+  ).run(batchId, item.lineNumber, item.customId, now, now);
+}
+
+export function markBatchItemResult(
+  batchId: string,
+  item: { lineNumber: number; customId: string | null },
+  result: any
+): void {
+  const db = getDbInstance();
+  const now = Math.floor(Date.now() / 1000);
+  db.prepare(
+    `
+    UPDATE batch_item_checkpoints
+    SET custom_id = ?,
+        status = 'completed',
+        result_json = ?,
+        error_json = NULL,
+        updated_at = ?
+    WHERE batch_id = ? AND line_number = ?
+  `
+  ).run(item.customId, JSON.stringify(result), now, batchId, item.lineNumber);
+}
+
+export function markBatchItemError(
+  batchId: string,
+  item: { lineNumber: number; customId: string | null },
+  error: any
+): void {
+  const db = getDbInstance();
+  const now = Math.floor(Date.now() / 1000);
+  db.prepare(
+    `
+    UPDATE batch_item_checkpoints
+    SET custom_id = ?,
+        status = 'errored',
+        result_json = NULL,
+        error_json = ?,
+        updated_at = ?
+    WHERE batch_id = ? AND line_number = ?
+  `
+  ).run(item.customId, JSON.stringify(error), now, batchId, item.lineNumber);
+}
+
 export function listBatches(apiKeyId?: string, limit: number = 20, after?: string): BatchRecord[] {
  const db = getDbInstance();
  const afterBatch = after ? getBatch(after) : null;
@@ -224,6 +382,8 @@ export function deleteBatch(id: string): boolean {
  const batch = getBatch(id);
  if (!batch) return false;

+  db.prepare("DELETE FROM batch_item_checkpoints WHERE batch_id = ?").run(id);
+
  // Soft-delete associated files (input, output, error)
  if (batch.inputFileId) {
    try {
@@ -281,6 +441,10 @@ export function deleteCompletedBatches(): { deletedBatches: number; deletedFiles
    }
  }

+  db.prepare(
+    "DELETE FROM batch_item_checkpoints WHERE batch_id IN (SELECT id FROM batches WHERE status = 'completed')"
+  ).run();
+
  const result = db.prepare("DELETE FROM batches WHERE status = 'completed'").run();
  return { deletedBatches: result.changes, deletedFiles };
 }
--- a/src/lib/db/migrations/112_batch_item_checkpoints.sql
+++ b/src/lib/db/migrations/112_batch_item_checkpoints.sql
@@ -0,0 +1,18 @@
+-- 110_batch_item_checkpoints.sql
+-- Durable per-item checkpoints for OpenAI-compatible batch processing.
+
+CREATE TABLE IF NOT EXISTS batch_item_checkpoints (
+  batch_id TEXT NOT NULL,
+  line_number INTEGER NOT NULL,
+  custom_id TEXT,
+  status TEXT NOT NULL CHECK(status IN ('pending', 'processing', 'completed', 'errored')),
+  result_json TEXT,
+  error_json TEXT,
+  created_at INTEGER NOT NULL,
+  updated_at INTEGER NOT NULL,
+  PRIMARY KEY (batch_id, line_number),
+  FOREIGN KEY(batch_id) REFERENCES batches(id) ON DELETE CASCADE
+);
+
+CREATE INDEX IF NOT EXISTS idx_batch_item_checkpoints_batch_status
+  ON batch_item_checkpoints(batch_id, status);
--- a/src/lib/localDb.ts
+++ b/src/lib/localDb.ts
@@ -290,12 +290,18 @@ export {
  countBatches,
  getPendingBatches,
  getTerminalBatches,
+  ensureBatchItemCheckpoints,
+  countBatchItemCheckpoints,
+  listBatchItemCheckpoints,
+  markBatchItemProcessing,
+  markBatchItemResult,
+  markBatchItemError,
  deleteBatch,
  deleteCompletedBatches,
 } from "./db/batches";

 export type { FileRecord } from "./db/files";
-export type { BatchRecord } from "./db/batches";
+export type { BatchItemCheckpoint, BatchRecord } from "./db/batches";

 export type { ModelComboMapping } from "./db/modelComboMappings";

@@ -578,12 +584,7 @@ export {
  listAllocationsForApiKey,
 } from "./db/quotaPools";
 // Quota per-(key, model) caps — Group B Fase 3 #7
-export {
-  getModelCap,
-  listModelCaps,
-  setModelCap,
-  deleteModelCap,
-} from "./db/quotaModelCaps";
+export { getModelCap, listModelCaps, setModelCap, deleteModelCap } from "./db/quotaModelCaps";

 export {
  // Quota Groups (B2)
--- a/tests/unit/batch-processor.test.ts
+++ b/tests/unit/batch-processor.test.ts
@@ -319,10 +319,147 @@ test("processPendingBatches caches rate-limit headers across sequential batches"
  assert.strictEqual(afterReset.timestamp, 0, "reset should clear cached timestamp");
 });

-test("processPendingBatches should recover stale in_progress batches", async () => {
+test("processPendingBatches should recover checkpointed stale batches without replaying completed items", async () => {
+  const lines = [
+    JSON.stringify({
+      custom_id: "already-done",
+      method: "POST",
+      url: "/v1/chat/completions",
+      body: { model: "gpt-4", messages: [{ role: "user", content: "first" }] },
+    }),
+    JSON.stringify({
+      custom_id: "needs-dispatch",
+      method: "POST",
+      url: "/v1/chat/completions",
+      body: { model: "gpt-4", messages: [{ role: "user", content: "second" }] },
+    }),
+  ];
+  const file = await localDb.createFile({
+    bytes: Buffer.byteLength(lines.join("\n") + "\n"),
+    filename: "checkpointed_stale_test.jsonl",
+    purpose: "batch_input",
+    content: Buffer.from(lines.join("\n") + "\n"),
+  });
+
+  const batch = await localDb.createBatch({
+    endpoint: "/v1/chat/completions",
+    status: "in_progress",
+    inputFileId: file.id,
+    completionWindow: "24h",
+    inProgressAt: Math.floor(Date.now() / 1000),
+  });
+  await localDb.updateBatch(batch.id, {
+    requestCountsTotal: 2,
+    requestCountsCompleted: 1,
+  });
+  await localDb.ensureBatchItemCheckpoints(batch.id, [
+    { lineNumber: 1, customId: "already-done" },
+    { lineNumber: 2, customId: "needs-dispatch" },
+  ]);
+  await localDb.markBatchItemResult(
+    batch.id,
+    { lineNumber: 1, customId: "already-done" },
+    {
+      id: "req_checkpointed",
+      custom_id: "already-done",
+      response: {
+        status_code: 200,
+        body: {
+          id: "chatcmpl-checkpointed",
+          choices: [{ message: { content: "from checkpoint" } }],
+          usage: { prompt_tokens: 10, completion_tokens: 5 },
+        },
+      },
+    }
+  );
+
+  let callCount = 0;
+  mock.method(dispatch, "dispatchBatchApiRequest", async () => {
+    callCount++;
+    return new Response(
+      JSON.stringify({
+        id: "chatcmpl-live",
+        choices: [{ message: { content: "processed once" } }],
+        usage: { prompt_tokens: 8, completion_tokens: 4 },
+      }),
+      { status: 200, headers: { "Content-Type": "application/json" } }
+    );
+  });
+
+  await batchProcessor.processPendingBatches();
+
+  await waitForAllBatches();
+
+  const updated = await localDb.getBatch(batch.id);
+  assert.strictEqual(updated?.status, "completed", "Stale batch should be recovered and completed");
+  assert.strictEqual(callCount, 1, "only the unchecked item should be dispatched");
+  assert.strictEqual(updated?.requestCountsCompleted, 2);
+  assert.ok(updated?.outputFileId, "recovered batch should emit an output file");
+
+  const output = localDb.getFileContent(updated!.outputFileId!);
+  assert.ok(output, "output file content should exist");
+  const outputRows = output
+    .toString()
+    .trim()
+    .split("\n")
+    .map((line) => JSON.parse(line));
+  assert.deepStrictEqual(
+    outputRows.map((row) => row.custom_id),
+    ["already-done", "needs-dispatch"]
+  );
+});
+
+test("processPendingBatches should not replay interrupted checkpoint items", async () => {
  const file = await localDb.createFile({
    bytes: 10,
-    filename: "stale_test.jsonl",
+    filename: "stale_processing_checkpoint.jsonl",
+    purpose: "batch_input",
+    content: Buffer.from(
+      JSON.stringify({
+        custom_id: "in-flight",
+        method: "POST",
+        url: "/v1/chat/completions",
+        body: { model: "gpt-4", messages: [{ role: "user", content: "hi" }] },
+      }) + "\n"
+    ),
+  });
+
+  const batch = await localDb.createBatch({
+    endpoint: "/v1/chat/completions",
+    status: "in_progress",
+    inputFileId: file.id,
+    completionWindow: "24h",
+    inProgressAt: Math.floor(Date.now() / 1000),
+  });
+  await localDb.updateBatch(batch.id, { requestCountsTotal: 1 });
+  await localDb.ensureBatchItemCheckpoints(batch.id, [{ lineNumber: 1, customId: "in-flight" }]);
+  await localDb.markBatchItemProcessing(batch.id, { lineNumber: 1, customId: "in-flight" });
+
+  let callCount = 0;
+  mock.method(dispatch, "dispatchBatchApiRequest", async () => {
+    callCount++;
+    throw new Error("interrupted checkpoint should not dispatch");
+  });
+
+  await batchProcessor.processPendingBatches();
+
+  await waitForAllBatches();
+
+  const updated = await localDb.getBatch(batch.id);
+  assert.strictEqual(updated?.status, "completed");
+  assert.strictEqual(callCount, 0, "interrupted checkpoint should not be replayed");
+  assert.strictEqual(updated?.requestCountsFailed, 1);
+  assert.ok(updated?.errorFileId, "interrupted item should be emitted as an error row");
+
+  const errorOutput = localDb.getFileContent(updated!.errorFileId!);
+  assert.ok(errorOutput, "error file content should exist");
+  assert.match(errorOutput.toString(), /not replayed to avoid duplicate provider work/);
+});
+
+test("processPendingBatches should fail stale batches without checkpoints instead of replaying", async () => {
+  const file = await localDb.createFile({
+    bytes: 10,
+    filename: "legacy_stale_no_checkpoints.jsonl",
    purpose: "batch_input",
    content: Buffer.from(
      JSON.stringify({
@@ -340,77 +477,20 @@ test("processPendingBatches should recover stale in_progress batches", async ()
    completionWindow: "24h",
    inProgressAt: Math.floor(Date.now() / 1000),
  });
+  await localDb.updateBatch(batch.id, { requestCountsTotal: 1 });

+  let callCount = 0;
  mock.method(dispatch, "dispatchBatchApiRequest", async () => {
-    return new Response(
-      JSON.stringify({
-        id: "chatcmpl-stale",
-        choices: [{ message: { content: "recovered" } }],
-        usage: { prompt_tokens: 10, completion_tokens: 5 },
-      }),
-      { status: 200, headers: { "Content-Type": "application/json" } }
-    );
+    callCount++;
+    throw new Error("legacy stale batch should not dispatch");
  });

  await batchProcessor.processPendingBatches();

-  await waitForAllBatches();
-
  const updated = await localDb.getBatch(batch.id);
-  assert.strictEqual(updated?.status, "completed", "Stale batch should be recovered and completed");
-  assert.ok(
-    updated?.inProgressAt != null,
-    "inProgressAt should be set (from fresh start, not stale)"
-  );
-});
-
-test("processPendingBatches should recover stale finalizing batches", async () => {
-  const file = await localDb.createFile({
-    bytes: 10,
-    filename: "stale_finalizing.jsonl",
-    purpose: "batch_input",
-    content: Buffer.from(
-      JSON.stringify({
-        method: "POST",
-        url: "/v1/chat/completions",
-        body: { model: "gpt-4", messages: [{ role: "user", content: "hi" }] },
-      }) + "\n"
-    ),
-  });
-
-  const batch = await localDb.createBatch({
-    endpoint: "/v1/chat/completions",
-    status: "finalizing",
-    inputFileId: file.id,
-    completionWindow: "24h",
-    finalizingAt: Math.floor(Date.now() / 1000),
-  });
-
-  mock.method(dispatch, "dispatchBatchApiRequest", async () => {
-    return new Response(
-      JSON.stringify({
-        id: "chatcmpl-stale-finalizing",
-        choices: [{ message: { content: "recovered" } }],
-        usage: { prompt_tokens: 10, completion_tokens: 5 },
-      }),
-      { status: 200, headers: { "Content-Type": "application/json" } }
-    );
-  });
-
-  await batchProcessor.processPendingBatches();
-
-  await waitForAllBatches();
-
-  const updated = await localDb.getBatch(batch.id);
-  assert.strictEqual(
-    updated?.status,
-    "completed",
-    "Stale finalizing batch should be recovered and completed"
-  );
-  assert.ok(
-    updated?.inProgressAt != null,
-    "inProgressAt should be set by fresh start after recovery"
-  );
+  assert.strictEqual(updated?.status, "failed");
+  assert.strictEqual(callCount, 0, "legacy stale batch should not be replayed");
+  assert.match(updated?.errors?.[0]?.message ?? "", /Cannot safely recover stale batch/);
 });

 test("processPendingBatches should respect BATCH_MAX_CONCURRENT (default 1)", async () => {
--- a/tests/unit/batch_api.test.ts
+++ b/tests/unit/batch_api.test.ts
@@ -21,6 +21,8 @@ const {
  formatFileResponse,
  deleteFile,
  getTerminalBatches,
+  ensureBatchItemCheckpoints,
+  markBatchItemResult,
 } = await import("../../src/lib/localDb.ts");
 const { getDbInstance } = await import("../../src/lib/db/core.ts");
 const {
@@ -45,6 +47,7 @@ test.afterEach(async () => {
  }
  try {
    const db = getDbInstance();
+    db.prepare("DELETE FROM batch_item_checkpoints").run();
    db.prepare("DELETE FROM batches").run();
    db.prepare("DELETE FROM files").run();
    db.prepare("DELETE FROM provider_connections").run();
@@ -695,28 +698,45 @@ test("Batch processor recovers orphaned finalizing batches during startup recove
  updateBatch(batch.id, {
    status: "finalizing",
    finalizingAt: Math.floor(Date.now() / 1000),
+    requestCountsTotal: 1,
+    requestCountsCompleted: 1,
  });
+  ensureBatchItemCheckpoints(batch.id, [{ lineNumber: 1, customId: "req-recovery" }]);
+  markBatchItemResult(
+    batch.id,
+    { lineNumber: 1, customId: "req-recovery" },
+    {
+      id: "req_checkpointed_recovery",
+      custom_id: "req-recovery",
+      response: {
+        status_code: 200,
+        body: {
+          id: "chatcmpl-mock-recovery",
+          object: "chat.completion",
+          model: "gpt-4o-mini",
+          choices: [
+            {
+              index: 0,
+              message: { role: "assistant", content: "recovered ok" },
+              finish_reason: "stop",
+            },
+          ],
+          usage: {
+            prompt_tokens: 1,
+            completion_tokens: 1,
+            total_tokens: 2,
+          },
+        },
+      },
+    }
+  );

  const originalFetch = globalThis.fetch;
-  (globalThis as any).fetch = async () => {
-    return Response.json({
-      id: "chatcmpl-mock-recovery",
-      object: "chat.completion",
-      model: "gpt-4o-mini",
-      choices: [
-        {
-          index: 0,
-          message: { role: "assistant", content: "recovered ok" },
-          finish_reason: "stop",
-        },
-      ],
-      usage: {
-        prompt_tokens: 1,
-        completion_tokens: 1,
-        total_tokens: 2,
-      },
-    });
-  };
+  let fetchCount = 0;
+  globalThis.fetch = (async () => {
+    fetchCount++;
+    throw new Error("checkpointed finalizing batch should not dispatch");
+  }) as typeof fetch;

  try {
    await processPendingBatches();
@@ -724,6 +744,7 @@ test("Batch processor recovers orphaned finalizing batches during startup recove

    const recoveredBatch = getBatch(batch.id);
    assert.strictEqual(recoveredBatch?.status, "completed");
+    assert.strictEqual(fetchCount, 0);
  } finally {
    globalThis.fetch = originalFetch;
  }