fix(commitments): preserve extraction batch on transient failure (#89817)

A drained extraction batch is spliced off the queue before the extractor runs. On a non-terminal failure the batch was never restored, so those items were silently lost and never retried. Restore the batch to the front of the queue (original order) on non-terminal errors and rethrow so the caller still logs; terminal model/auth errors keep the existing cooldown drop/stop behavior. Also ensure the single-slot debounce schedules a drain from the overflow branch, so a queue left full by a restored batch still gets retried instead of being stuck. Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-21 10:16:44 +00:00 · 2026-06-28 18:10:41 -07:00
parent ac6f249de1
commit be0c40bad2
2 changed files with 339 additions and 8 deletions
@@ -398,4 +398,313 @@ describe("commitment extraction runtime", () => {
    );
    expect(processed).toBe(DEFAULT_COMMITMENT_EXTRACTION_QUEUE_MAX_ITEMS);
  });
+
+  function mapBatchToCandidates({ items }: { items: CommitmentExtractionItem[] }) {
+    return {
+      candidates: items.map((item, index) => ({
+        itemId: item.itemId,
+        kind: "event_check_in" as const,
+        sensitivity: "routine" as const,
+        source: "inferred_user_context" as const,
+        reason: `Follow up ${index + 1}`,
+        suggestedText: `How did item ${index + 1} go?`,
+        dedupeKey: `event:${item.sourceMessageId ?? index}`,
+        confidence: 0.93,
+        dueWindow: {
+          earliest: "2026-04-30T17:00:00.000Z",
+          latest: "2026-04-30T23:00:00.000Z",
+          timezone: "America/Los_Angeles",
+        },
+      })),
+    };
+  }
+
+  it("restores and reprocesses a batch after a non-terminal extractor failure", async () => {
+    const cfg = await createConfig();
+    let attempts = 0;
+    const extractBatch = vi.fn(async (params: { items: CommitmentExtractionItem[] }) => {
+      attempts += 1;
+      if (attempts === 1) {
+        throw new Error("transient extraction failure");
+      }
+      return mapBatchToCandidates(params);
+    });
+    configureCommitmentExtractionRuntime({
+      forceInTests: true,
+      extractBatch,
+      setTimer: () => ({ unref() {} }) as ReturnType<typeof setTimeout>,
+      clearTimer: () => undefined,
+    });
+
+    expect(
+      enqueueCommitmentExtraction({
+        cfg,
+        nowMs,
+        agentId: "main",
+        sessionKey: "agent:main:telegram:user-1",
+        channel: "telegram",
+        sourceMessageId: "m1",
+        userText: "I have an interview tomorrow.",
+        assistantText: "Good luck.",
+      }),
+    ).toBe(true);
+    expect(
+      enqueueCommitmentExtraction({
+        cfg,
+        nowMs: nowMs + 1,
+        agentId: "main",
+        sessionKey: "agent:main:telegram:user-1",
+        channel: "telegram",
+        sourceMessageId: "m2",
+        userText: "I have a dentist appointment tomorrow.",
+        assistantText: "Hope it goes smoothly.",
+      }),
+    ).toBe(true);
+
+    // First drain: the extractor throws a non-terminal error and nothing persists.
+    await expect(drainCommitmentExtractionQueue()).rejects.toThrow("transient extraction failure");
+    expect(extractBatch).toHaveBeenCalledTimes(1);
+    const emptyStore = await loadCommitmentStore();
+    expect(emptyStore.commitments).toHaveLength(0);
+
+    // Retry: the restored batch is reprocessed once, in the same order, with no
+    // duplicate persistence or extraction.
+    await expect(drainCommitmentExtractionQueue()).resolves.toBe(2);
+    expect(extractBatch).toHaveBeenCalledTimes(2);
+
+    const firstCallIds = extractBatch.mock.calls[0]?.[0].items.map((item) => item.itemId);
+    const retryCallIds = extractBatch.mock.calls[1]?.[0].items.map((item) => item.itemId);
+    expect(retryCallIds).toEqual(firstCallIds);
+
+    const store = await loadCommitmentStore();
+    expect(store.commitments.map((commitment) => commitment.dedupeKey)).toEqual([
+      "event:m1",
+      "event:m2",
+    ]);
+
+    // A third drain has nothing left to do: no duplicate reprocessing.
+    await expect(drainCommitmentExtractionQueue()).resolves.toBe(0);
+    expect(extractBatch).toHaveBeenCalledTimes(2);
+  });
+
+  it("restores a failed batch to the front, preserving order across batches", async () => {
+    const cfg = await createConfig();
+    const seenOrder: string[] = [];
+    let attempts = 0;
+    const extractBatch = vi.fn(async ({ items }: { items: CommitmentExtractionItem[] }) => {
+      attempts += 1;
+      if (attempts === 1) {
+        // Fail the first batch only; record nothing so order reflects success runs.
+        throw new Error("transient extraction failure");
+      }
+      for (const item of items) {
+        seenOrder.push(item.sourceMessageId ?? "");
+      }
+      return { candidates: [] };
+    });
+    configureCommitmentExtractionRuntime({
+      forceInTests: true,
+      extractBatch,
+      setTimer: () => ({ unref() {} }) as ReturnType<typeof setTimeout>,
+      clearTimer: () => undefined,
+    });
+
+    // Enqueue more than one batch (batchMaxItems = 8) so the restored batch must
+    // land ahead of the tail rather than being appended.
+    const total = 10;
+    for (let index = 0; index < total; index += 1) {
+      expect(
+        enqueueCommitmentExtraction({
+          cfg,
+          nowMs: nowMs + index,
+          agentId: "main",
+          sessionKey: "agent:main:telegram:user-1",
+          channel: "telegram",
+          sourceMessageId: `m${index}`,
+          userText: `Commitment candidate ${index}`,
+          assistantText: "I will follow up.",
+        }),
+      ).toBe(true);
+    }
+
+    await expect(drainCommitmentExtractionQueue()).rejects.toThrow("transient extraction failure");
+    await expect(drainCommitmentExtractionQueue()).resolves.toBe(total);
+
+    const expectedOrder = Array.from({ length: total }, (_v, index) => `m${index}`);
+    expect(seenOrder).toEqual(expectedOrder);
+  });
+
+  it("keeps the existing drop/stop behavior on terminal extraction failures", async () => {
+    const cfg = await createConfig();
+    const extractBatch = vi.fn(async () => {
+      throw new Error('No API key found for provider "openai".');
+    });
+    configureCommitmentExtractionRuntime({
+      forceInTests: true,
+      extractBatch,
+      setTimer: () => ({ unref() {} }) as ReturnType<typeof setTimeout>,
+      clearTimer: () => undefined,
+    });
+
+    expect(
+      enqueueCommitmentExtraction({
+        cfg,
+        nowMs,
+        agentId: "main",
+        sessionKey: "agent:main:telegram:user-1",
+        channel: "telegram",
+        sourceMessageId: "m1",
+        userText: "I have an interview tomorrow.",
+        assistantText: "Good luck.",
+      }),
+    ).toBe(true);
+
+    await expect(drainCommitmentExtractionQueue()).rejects.toThrow("No API key found");
+    expect(extractBatch).toHaveBeenCalledTimes(1);
+
+    // Terminal failures drop the agent's queued work; a retry must not reprocess
+    // the dropped batch.
+    await expect(drainCommitmentExtractionQueue()).resolves.toBe(0);
+    expect(extractBatch).toHaveBeenCalledTimes(1);
+  });
+
+  it("schedules a retry when a non-terminal failure leaves the queue full", async () => {
+    const cfg = await createConfig();
+    const scheduled: Array<() => void> = [];
+    let attempts = 0;
+    const extractBatch = vi.fn(
+      async (_params: {
+        items: CommitmentExtractionItem[];
+      }): Promise<CommitmentExtractionBatchResult> => {
+        attempts += 1;
+        if (attempts === 1) {
+          throw new Error("transient extraction failure");
+        }
+        return { candidates: [] };
+      },
+    );
+    configureCommitmentExtractionRuntime({
+      forceInTests: true,
+      extractBatch,
+      setTimer: (callback) => {
+        scheduled.push(callback);
+        return { unref() {} } as ReturnType<typeof setTimeout>;
+      },
+      clearTimer: () => undefined,
+    });
+
+    for (let index = 0; index < DEFAULT_COMMITMENT_EXTRACTION_QUEUE_MAX_ITEMS; index += 1) {
+      expect(
+        enqueueCommitmentExtraction({
+          cfg,
+          nowMs: nowMs + index,
+          agentId: "main",
+          sessionKey: "agent:main:telegram:user-1",
+          channel: "telegram",
+          sourceMessageId: `m${index}`,
+          userText: `Commitment candidate ${index}`,
+          assistantText: "I will follow up.",
+        }),
+      ).toBe(true);
+    }
+    // The single-slot debounce schedules exactly one drain while filling.
+    expect(scheduled).toHaveLength(1);
+
+    // Fire the scheduled drain: the first batch fails (non-terminal) and is
+    // restored, leaving the queue full again with no pending timer.
+    scheduled[0]?.();
+    await vi.waitFor(() => {
+      expect(extractBatch).toHaveBeenCalledTimes(1);
+    });
+    await new Promise<void>((resolve) => {
+      setTimeout(resolve, 0);
+    });
+
+    // A new request is dropped because the queue is full, but the restored batch
+    // must still get a retry scheduled, otherwise it would be stuck forever.
+    expect(
+      enqueueCommitmentExtraction({
+        cfg,
+        nowMs: nowMs + DEFAULT_COMMITMENT_EXTRACTION_QUEUE_MAX_ITEMS,
+        agentId: "main",
+        sessionKey: "agent:main:telegram:user-1",
+        channel: "telegram",
+        sourceMessageId: "overflow",
+        userText: "Overflow candidate",
+        assistantText: "I will follow up.",
+      }),
+    ).toBe(false);
+    expect(scheduled).toHaveLength(2);
+
+    // The drain is healthy after the failure: the full queue reprocesses cleanly.
+    await expect(drainCommitmentExtractionQueue()).resolves.toBe(
+      DEFAULT_COMMITMENT_EXTRACTION_QUEUE_MAX_ITEMS,
+    );
+  });
+
+  it("re-arms the drain after a timer-fired non-terminal failure with no later enqueue", async () => {
+    const cfg = await createConfig();
+    const scheduled: Array<() => void> = [];
+    let attempts = 0;
+    const extractBatch = vi.fn(async (params: { items: CommitmentExtractionItem[] }) => {
+      attempts += 1;
+      if (attempts === 1) {
+        throw new Error("transient extraction failure");
+      }
+      return mapBatchToCandidates(params);
+    });
+    configureCommitmentExtractionRuntime({
+      forceInTests: true,
+      extractBatch,
+      setTimer: (callback) => {
+        scheduled.push(callback);
+        return { unref() {} } as ReturnType<typeof setTimeout>;
+      },
+      clearTimer: () => undefined,
+    });
+
+    expect(
+      enqueueCommitmentExtraction({
+        cfg,
+        nowMs,
+        agentId: "main",
+        sessionKey: "agent:main:telegram:user-1",
+        channel: "telegram",
+        sourceMessageId: "m1",
+        userText: "I have an interview tomorrow.",
+        assistantText: "Good luck.",
+      }),
+    ).toBe(true);
+    // The enqueue schedules exactly one drain via the single-slot debounce.
+    expect(scheduled).toHaveLength(1);
+
+    // Fire the scheduled drain. The timer callback clears the pending timer first,
+    // then the extractor throws a non-terminal error and the batch is restored.
+    scheduled[0]?.();
+    await vi.waitFor(() => {
+      expect(extractBatch).toHaveBeenCalledTimes(1);
+    });
+    await new Promise<void>((resolve) => {
+      setTimeout(resolve, 0);
+    });
+
+    // Regression guard for the timer-fired-failure path: with no later enqueue to
+    // reschedule it, the restored batch must still have a fresh drain armed, or it
+    // would sit only in memory and be lost on process exit.
+    expect(scheduled).toHaveLength(2);
+
+    // Firing that re-armed drain reprocesses the same batch and persists it.
+    scheduled[1]?.();
+    await vi.waitFor(() => {
+      expect(extractBatch).toHaveBeenCalledTimes(2);
+    });
+    await new Promise<void>((resolve) => {
+      setTimeout(resolve, 0);
+    });
+
+    const store = await loadCommitmentStore();
+    expect(store.commitments.map((commitment) => commitment.dedupeKey)).toEqual(["event:m1"]);
+    // The successful drain empties the queue, so no further retry is armed.
+    expect(scheduled).toHaveLength(2);
+  });
 });
@@ -77,6 +77,22 @@ function clearTimer(handle: TimerHandle): void {
  (runtime.clearTimer ?? clearTimeout)(handle);
 }

+// Single-slot debounce: schedule one drain unless one is already pending. Shared
+// by enqueue (new work), the overflow branch, and the drain's non-terminal
+// failure path (so a batch restored after a timer-fired failure still gets
+// retried even when no later enqueue arrives).
+function scheduleDrainSoon(debounceMs: number): void {
+  if (timer) {
+    return;
+  }
+  timer = setTimer(() => {
+    timer = null;
+    void drainCommitmentExtractionQueue().catch((err: unknown) => {
+      log.warn("commitment extraction failed", { error: String(err) });
+    });
+  }, debounceMs);
+}
+
 /** Installs runtime hooks for extraction tests or alternate batch extraction. */
 export function configureCommitmentExtractionRuntime(next: CommitmentExtractionRuntime): void {
  runtime = next;
@@ -131,6 +147,10 @@ export function enqueueCommitmentExtraction(input: CommitmentExtractionEnqueueIn
      });
      queueOverflowWarned = true;
    }
+    // The queue can be full because a non-terminal failure restored its batch
+    // (see drainCommitmentExtractionQueue). Dropping this request must not also
+    // drop the retry: make sure a drain is scheduled before returning.
+    scheduleDrainSoon(resolved.extraction.debounceMs);
    return false;
  }
  queue.push({
@@ -150,14 +170,7 @@ export function enqueueCommitmentExtraction(input: CommitmentExtractionEnqueueIn
    ...(input.sourceRunId?.trim() ? { sourceRunId: input.sourceRunId.trim() } : {}),
    cfg: input.cfg,
  });
-  if (!timer) {
-    timer = setTimer(() => {
-      timer = null;
-      void drainCommitmentExtractionQueue().catch((err: unknown) => {
-        log.warn("commitment extraction failed", { error: String(err) });
-      });
-    }, resolved.extraction.debounceMs);
-  }
+  scheduleDrainSoon(resolved.extraction.debounceMs);
  return true;
 }

@@ -306,6 +319,15 @@ export async function drainCommitmentExtractionQueue(): Promise<number> {
            Date.now(),
            items[0]?.nowMs ?? Date.now(),
          );
+        } else {
+          // Non-terminal failure (e.g. transient model/network error): the batch
+          // was already spliced out, so restore it to the front in original order.
+          // A timer-fired drain has already cleared `timer`, so also re-arm the
+          // debounce; otherwise the restored batch sits only in memory and is lost
+          // on process exit if no later enqueue happens to reschedule a drain.
+          // Rethrow so the caller still logs; the next drain reprocesses it in order.
+          queue.unshift(...batch);
+          scheduleDrainSoon(resolved.extraction.debounceMs);
        }
        throw error;
      }