From 5a26748c3eb6a6ee459d643832f68b2e7b961766 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sat, 11 Jul 2026 04:23:30 -0700 Subject: [PATCH] fix(commitments): isolate extraction batches by agent (#104426) * fix(commitments): isolate extraction batches by agent * chore: leave release notes to release workflow --- src/commitments/runtime.test.ts | 89 +++++++++++++++++++++++++++++++++ src/commitments/runtime.ts | 36 +++++++++++-- 2 files changed, 120 insertions(+), 5 deletions(-) diff --git a/src/commitments/runtime.test.ts b/src/commitments/runtime.test.ts index 47fb20482735..59cb6769ae92 100644 --- 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, + 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; + }, + 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 index 6da951bee270..9d779724d51b 100644 --- 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 & { 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 { if (draining) { @@ -303,9 +320,15 @@ export async function drainCommitmentExtractionQueue(): Promise { 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 { 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.