fix(commitments): isolate extraction batches by agent (#104426)

steipete · web-flow · commit 5a26748c3eb6 · 2026-07-11T04:23:30.000-07:00
* fix(commitments): isolate extraction batches by agent

* chore: leave release notes to release workflow
diff --git a/src/commitments/runtime.test.ts b/src/commitments/runtime.test.ts
@@ -190,6 +190,43 @@ describe("commitment extraction runtime", () => {
     expect(store.commitments[0]).not.toHaveProperty("sourceAssistantText");
   });
 
+  it("partitions extraction batches by agent", async () => {
+    const cfg = await createConfig();
+    const extractBatch = vi.fn(async (_params: { items: CommitmentExtractionItem[] }) => ({
+      candidates: [],
+    }));
+    configureCommitmentExtractionRuntime({
+      forceInTests: true,
+      extractBatch,
+      setTimer: () => ({ unref() {} }) as ReturnType<typeof setTimeout>,
+      clearTimer: () => undefined,
+    });
+
+    for (const [index, agentId] of ["alpha", "beta", "alpha", "beta"].entries()) {
+      expect(
+        enqueueCommitmentExtraction({
+          cfg,
+          nowMs: nowMs + index,
+          agentId,
+          sessionKey: `agent:${agentId}:telegram:user-1`,
+          channel: "telegram",
+          sourceMessageId: `m${index}`,
+          userText: `Commitment candidate ${index}`,
+          assistantText: "I will follow up.",
+        }),
+      ).toBe(true);
+    }
+
+    await expect(drainCommitmentExtractionQueue()).resolves.toBe(4);
+    expect(extractBatch).toHaveBeenCalledTimes(2);
+    expect(
+      extractBatch.mock.calls.map(([params]) => params.items.map((item) => item.agentId)),
+    ).toEqual([
+      ["alpha", "alpha"],
+      ["beta", "beta"],
+    ]);
+  });
+
   it("uses the configured agent model for the hidden extractor run", async () => {
     const cfg = await createConfig();
     cfg.agents = {
@@ -568,6 +605,58 @@ describe("commitment extraction runtime", () => {
     expect(extractBatch).toHaveBeenCalledTimes(1);
   });
 
+  it("keeps other agents queued after a terminal extraction failure", async () => {
+    const cfg = await createConfig();
+    const scheduled: Array<() => void> = [];
+    const extractBatch = vi.fn(async ({ items }: { items: CommitmentExtractionItem[] }) => {
+      if (items[0]?.agentId === "alpha") {
+        throw new Error('No API key found for provider "openai".');
+      }
+      return { candidates: [] };
+    });
+    configureCommitmentExtractionRuntime({
+      forceInTests: true,
+      extractBatch,
+      setTimer: (callback) => {
+        scheduled.push(callback);
+        return { unref() {} } as ReturnType<typeof setTimeout>;
+      },
+      clearTimer: () => undefined,
+    });
+
+    for (const agentId of ["alpha", "beta"]) {
+      expect(
+        enqueueCommitmentExtraction({
+          cfg,
+          nowMs,
+          agentId,
+          sessionKey: `agent:${agentId}:telegram:user-1`,
+          channel: "telegram",
+          userText: "I have an interview tomorrow.",
+          assistantText: "Good luck.",
+        }),
+      ).toBe(true);
+    }
+
+    expect(scheduled).toHaveLength(1);
+    scheduled[0]?.();
+    await vi.waitFor(() => {
+      expect(extractBatch).toHaveBeenCalledTimes(1);
+    });
+    await vi.waitFor(() => {
+      expect(scheduled).toHaveLength(2);
+    });
+
+    scheduled[1]?.();
+    await vi.waitFor(() => {
+      expect(extractBatch).toHaveBeenCalledTimes(2);
+    });
+    expect(extractBatch.mock.calls.map(([params]) => params.items[0]?.agentId)).toEqual([
+      "alpha",
+      "beta",
+    ]);
+  });
+
   it("schedules a retry when a non-terminal failure leaves the queue full", async () => {
     const cfg = await createConfig();
     const scheduled: Array<() => void> = [];
diff --git a/src/commitments/runtime.ts b/src/commitments/runtime.ts
@@ -78,9 +78,8 @@ function clearTimer(handle: TimerHandle): void {
 }
 
 // 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).
+// by enqueue (new work), the overflow branch, and drain failure paths so queued
+// work still progresses after a timer-fired extraction failure.
 function scheduleDrainSoon(debounceMs: number): void {
   if (timer) {
     return;
@@ -294,6 +293,24 @@ async function hydrateBatch(
   );
 }
 
+function takeAgentBatch(
+  agentId: string,
+  maxItems: number,
+): Array<Omit<CommitmentExtractionItem, "existingPending"> & { cfg?: OpenClawConfig }> {
+  const batch = [];
+  for (let index = 0; index < queue.length && batch.length < maxItems;) {
+    if (queue[index]?.agentId !== agentId) {
+      index += 1;
+      continue;
+    }
+    const [item] = queue.splice(index, 1);
+    if (item) {
+      batch.push(item);
+    }
+  }
+  return batch;
+}
+
 /** Drains queued extraction work in batches and returns processed item count. */
 export async function drainCommitmentExtractionQueue(): Promise<number> {
   if (draining) {
@@ -303,9 +320,15 @@ export async function drainCommitmentExtractionQueue(): Promise<number> {
   try {
     let processed = 0;
     while (queue.length > 0) {
-      const firstCfg = queue[0]?.cfg;
+      const first = queue[0];
+      if (!first) {
+        break;
+      }
+      const firstCfg = first.cfg;
       const resolved = resolveCommitmentsConfig(firstCfg);
-      const batch = queue.splice(0, resolved.extraction.batchMaxItems);
+      // Extraction inherits the first item's model, credentials, workspace, and
+      // session file. Keep every prompt and failure policy scoped to that agent.
+      const batch = takeAgentBatch(first.agentId, resolved.extraction.batchMaxItems);
       const items = await hydrateBatch(batch);
       const extractor = runtime.extractBatch ?? defaultExtractBatch;
       let result: CommitmentExtractionBatchResult;
@@ -319,6 +342,9 @@ export async function drainCommitmentExtractionQueue(): Promise<number> {
             Date.now(),
             items[0]?.nowMs ?? Date.now(),
           );
+          if (queue.length > 0) {
+            scheduleDrainSoon(resolved.extraction.debounceMs);
+          }
         } 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.
