Merge ede2b5d into 7a602c7

yetval · web-flow · commit f57c9784256a · 2026-06-08T15:46:21.000Z
diff --git a/src/auto-reply/reply/queue/drain.ts b/src/auto-reply/reply/queue/drain.ts
@@ -9,6 +9,7 @@ import {
   drainCollectQueueStep,
   drainNextQueueItem,
   hasCrossChannelItems,
+  removeQueuedItemsByRef,
   previewQueueSummaryPrompt,
   waitForQueueDebounce,
 } from "../../../utils/queue-helpers.js";
@@ -497,7 +498,7 @@ export function scheduleFollowupDrain(
             } else {
               await drainGroup();
             }
-            queue.items.splice(0, groupItems.length);
+            removeQueuedItemsByRef(queue.items, groupItems);
             if (pendingSummary) {
               clearFollowupQueueSummaryState(queue);
               pendingSummary = undefined;
diff --git a/src/utils/queue-helpers.test.ts b/src/utils/queue-helpers.test.ts
@@ -1,10 +1,12 @@
 // Queue helper tests cover queue ordering and dedupe utility behavior.
 import { describe, expect, it } from "vitest";
 import {
+  applyQueueDropPolicy,
   applyQueueRuntimeSettings,
   buildQueueSummaryPrompt,
   clearQueueSummaryState,
   drainCollectItemIfNeeded,
+  drainNextQueueItem,
   hasCrossChannelItems,
   previewQueueSummaryPrompt,
 } from "./queue-helpers.js";
@@ -170,6 +172,57 @@ describe("drainCollectItemIfNeeded", () => {
   });
 });
 
+describe("drainNextQueueItem", () => {
+  it("keeps overflow survivors when the queue mutates during an awaited drain", async () => {
+    type Item = { id: string };
+    const queue = {
+      items: [{ id: "m1" }],
+      cap: 3,
+      dropPolicy: "summarize" as const,
+      droppedCount: 0,
+      summaryLines: [],
+    };
+    const delivered: string[] = [];
+    const dropped: string[] = [];
+    let release!: () => void;
+    const gate = new Promise<void>((resolve) => {
+      release = resolve;
+    });
+
+    const firstDrain = drainNextQueueItem(queue.items, async (item: Item) => {
+      delivered.push(item.id);
+      await gate;
+    });
+    await Promise.resolve();
+
+    for (let index = 2; index <= 8; index += 1) {
+      const item = { id: `m${index}` };
+      const shouldEnqueue = applyQueueDropPolicy({
+        queue,
+        summarize: (queued) => queued.id,
+        onDrop: (items) => {
+          dropped.push(...items.map((queued) => queued.id));
+        },
+      });
+      if (shouldEnqueue) {
+        queue.items.push(item);
+      }
+    }
+
+    release();
+    await firstDrain;
+    while (
+      await drainNextQueueItem(queue.items, async (item) => {
+        delivered.push(item.id);
+      })
+    ) {}
+
+    expect(delivered).toEqual(["m1", "m6", "m7", "m8"]);
+    expect(dropped).toEqual(["m1", "m2", "m3", "m4", "m5"]);
+    expect(queue.items).toEqual([]);
+  });
+});
+
 describe("hasCrossChannelItems", () => {
   it("lets unresolved items join an otherwise single keyed route", () => {
     const items = [
diff --git a/src/utils/queue-helpers.ts b/src/utils/queue-helpers.ts
@@ -166,6 +166,15 @@ export function beginQueueDrain<T extends { draining: boolean }>(
   return queue;
 }
 
+export function removeQueuedItemsByRef<T>(items: T[], processed: readonly T[]): void {
+  for (const item of processed) {
+    const idx = items.indexOf(item);
+    if (idx !== -1) {
+      items.splice(idx, 1);
+    }
+  }
+}
+
 /** Run and remove the next queued item, returning false when empty. */
 export async function drainNextQueueItem<T>(
   items: T[],
@@ -176,7 +185,7 @@ export async function drainNextQueueItem<T>(
     return false;
   }
   await run(next);
-  items.shift();
+  removeQueuedItemsByRef(items, [next]);
   return true;
 }
 
