fix(queue): prevent applyQueueDropPolicy from selecting in-flight items as overflow victims (#103284)

* fix(queue): prevent applyQueueDropPolicy from selecting in-flight items as overflow victims

When a burst of inbound messages hits the followup queue cap while the
head item is mid-delivery, applyQueueDropPolicy can select that same
in-flight item as an overflow victim. With the default
dropPolicy: "summarize", the in-flight item ends up recorded in the
overflow summary as dropped even though it is still being delivered,
producing a contradictory record where the same message is both
answered and reported as unanswered-due-to-overflow.

The fix introduces an optional `inFlight` Set parameter to
applyQueueDropPolicy and drainNextQueueItem. The followup queue state
now owns a shared inFlight set that is:
- Populated by drainNextQueueItem during the await run(next) window
- Populated by the collect-merge drain path for activeGroupItems
- Passed to applyQueueDropPolicy in enqueueFollowupRun

The drop policy now computes an effective queue length that excludes
in-flight items, and skips them when selecting splice victims.

Fixes #103246

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* fix(queue): align pending depth with active deliveries

---------

Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>
Co-authored-by: Peter Steinberger <steipete@gmail.com>

src/auto-reply/reply/queue.in-flight.test.ts

@@ -0,0 +1,250 @@
+// Proves queue caps and depth describe pending work while active identities remain in shared state.
+import { afterEach, describe, expect, it, vi } from "vitest";
+import {
+  clearFollowupQueue,
+  completeFollowupRunLifecycle,
+  enqueueFollowupRun,
+  getFollowupQueueDepth,
+  scheduleFollowupDrain,
+} from "./queue.js";
+import { createDeferred, createQueueTestRun as createRun } from "./queue.test-helpers.js";
+import { getExistingFollowupQueue } from "./queue/state.js";
+import type { FollowupRun, QueueDropPolicy, QueueSettings } from "./queue/types.js";
+
+describe("followup queue in-flight ownership", () => {
+  const keys = new Set<string>();
+
+  afterEach(() => {
+    for (const key of keys) {
+      clearFollowupQueue(key);
+    }
+    keys.clear();
+  });
+
+  const createKey = (suffix: string) => {
+    const key = `test-in-flight-${suffix}-${Date.now()}-${Math.random()}`;
+    keys.add(key);
+    return key;
+  };
+
+  const createSettings = (dropPolicy: QueueDropPolicy): QueueSettings => ({
+    mode: "followup",
+    debounceMs: 0,
+    cap: 1,
+    dropPolicy,
+  });
+
+  it.each(["old", "summarize"] as const)(
+    "keeps an active single delivery out of %s overflow victims",
+    async (dropPolicy) => {
+      const key = createKey(dropPolicy);
+      const entered = createDeferred<void>();
+      const release = createDeferred<void>();
+      const activeComplete = vi.fn();
+      const pendingComplete = vi.fn();
+      const calls: FollowupRun[] = [];
+      const active = {
+        ...createRun({ prompt: "active" }),
+        queuedLifecycle: { onComplete: activeComplete },
+      };
+      const runFollowup = async (run: FollowupRun) => {
+        calls.push(run);
+        run.queuedLifecycle?.onAdmitted?.();
+        if (run === active) {
+          entered.resolve();
+          await release.promise;
+        }
+        completeFollowupRunLifecycle(run);
+      };
+
+      try {
+        expect(
+          enqueueFollowupRun(key, active, createSettings(dropPolicy), "none", runFollowup),
+        ).toBe(true);
+        await entered.promise;
+
+        expect(getFollowupQueueDepth(key)).toBe(0);
+        expect(
+          enqueueFollowupRun(
+            key,
+            {
+              ...createRun({ prompt: "pending" }),
+              queuedLifecycle: { onComplete: pendingComplete },
+            },
+            createSettings(dropPolicy),
+            "none",
+          ),
+        ).toBe(true);
+        expect(
+          enqueueFollowupRun(
+            key,
+            createRun({ prompt: "survivor" }),
+            createSettings(dropPolicy),
+            "none",
+          ),
+        ).toBe(true);
+
+        const queue = getExistingFollowupQueue(key);
+        expect(queue?.inFlight.has(active)).toBe(true);
+        expect(queue?.items.map((item) => item.prompt)).toEqual(["active", "survivor"]);
+        expect(getFollowupQueueDepth(key)).toBe(1);
+        expect(activeComplete).not.toHaveBeenCalled();
+        expect(pendingComplete).toHaveBeenCalledTimes(dropPolicy === "old" ? 1 : 0);
+        expect(queue?.summarySources.map((item) => item.prompt)).toEqual(
+          dropPolicy === "summarize" ? ["pending"] : [],
+        );
+      } finally {
+        release.resolve();
+      }
+
+      await expect.poll(() => getExistingFollowupQueue(key)).toBeUndefined();
+      expect(activeComplete).toHaveBeenCalledOnce();
+      expect(pendingComplete).toHaveBeenCalledOnce();
+      expect(calls.at(-1)?.prompt).toBe("survivor");
+    },
+  );
+
+  it("admits one pending item under drop:new while another item is active", async () => {
+    const key = createKey("new");
+    const entered = createDeferred<void>();
+    const release = createDeferred<void>();
+    const rejectedEnqueued = vi.fn();
+    const rejectedComplete = vi.fn();
+    const active = createRun({ prompt: "active" });
+    const runFollowup = async (run: FollowupRun) => {
+      run.queuedLifecycle?.onAdmitted?.();
+      if (run === active) {
+        entered.resolve();
+        await release.promise;
+      }
+      completeFollowupRunLifecycle(run);
+    };
+
+    try {
+      expect(enqueueFollowupRun(key, active, createSettings("new"), "none", runFollowup)).toBe(
+        true,
+      );
+      await entered.promise;
+
+      expect(getFollowupQueueDepth(key)).toBe(0);
+      expect(
+        enqueueFollowupRun(key, createRun({ prompt: "pending" }), createSettings("new"), "none"),
+      ).toBe(true);
+      expect(
+        enqueueFollowupRun(
+          key,
+          {
+            ...createRun({ prompt: "rejected" }),
+            queuedLifecycle: {
+              onEnqueued: rejectedEnqueued,
+              onComplete: rejectedComplete,
+            },
+          },
+          createSettings("new"),
+          "none",
+        ),
+      ).toBe(false);
+
+      expect(getFollowupQueueDepth(key)).toBe(1);
+      expect(getExistingFollowupQueue(key)?.items.map((item) => item.prompt)).toEqual([
+        "active",
+        "pending",
+      ]);
+      expect(rejectedEnqueued).not.toHaveBeenCalled();
+      expect(rejectedComplete).toHaveBeenCalledOnce();
+    } finally {
+      release.resolve();
+    }
+
+    await expect.poll(() => getExistingFollowupQueue(key)).toBeUndefined();
+  });
+
+  it("protects a collect group and counts only active identities still present", async () => {
+    const key = createKey("collect");
+    const entered = createDeferred<void>();
+    const release = createDeferred<void>();
+    const groupCompletions = [vi.fn(), vi.fn()];
+    const pendingComplete = vi.fn();
+    const rejectedComplete = vi.fn();
+    let aggregate: FollowupRun | undefined;
+    const initialSettings: QueueSettings = {
+      mode: "collect",
+      debounceMs: 0,
+      cap: 50,
+      dropPolicy: "summarize",
+    };
+    const group = groupCompletions.map((onComplete, index) => ({
+      ...createRun({
+        prompt: `group-${index + 1}`,
+        originatingChannel: "slack" as const,
+        originatingTo: "channel:A",
+        originatingChatType: "channel",
+      }),
+      queuedLifecycle: { onComplete },
+    }));
+    const runFollowup = async (run: FollowupRun) => {
+      if (!aggregate) {
+        aggregate = run;
+        entered.resolve();
+        await release.promise;
+      }
+      completeFollowupRunLifecycle(run);
+    };
+
+    for (const run of group) {
+      expect(enqueueFollowupRun(key, run, initialSettings, "none", undefined, false)).toBe(true);
+    }
+    scheduleFollowupDrain(key, runFollowup);
+
+    try {
+      await entered.promise;
+      const queue = getExistingFollowupQueue(key);
+      expect(queue?.inFlight.size).toBe(2);
+      expect(getFollowupQueueDepth(key)).toBe(0);
+
+      const oldSettings: QueueSettings = { ...initialSettings, cap: 1, dropPolicy: "old" };
+      expect(
+        enqueueFollowupRun(
+          key,
+          {
+            ...createRun({ prompt: "pending-old" }),
+            queuedLifecycle: { onComplete: pendingComplete },
+          },
+          oldSettings,
+          "none",
+        ),
+      ).toBe(true);
+      expect(enqueueFollowupRun(key, createRun({ prompt: "survivor" }), oldSettings, "none")).toBe(
+        true,
+      );
+
+      expect(queue?.items.map((item) => item.prompt)).toEqual(["group-1", "group-2", "survivor"]);
+      expect(pendingComplete).toHaveBeenCalledOnce();
+      expect(groupCompletions.map((complete) => complete.mock.calls.length)).toEqual([0, 0]);
+
+      aggregate?.queuedLifecycle?.onAdmitted?.();
+      expect(queue?.items.map((item) => item.prompt)).toEqual(["survivor"]);
+      expect(queue?.inFlight.size).toBe(2);
+      expect(getFollowupQueueDepth(key)).toBe(1);
+
+      expect(
+        enqueueFollowupRun(
+          key,
+          {
+            ...createRun({ prompt: "rejected-new" }),
+            queuedLifecycle: { onComplete: rejectedComplete },
+          },
+          { ...initialSettings, cap: 1, dropPolicy: "new" },
+          "none",
+        ),
+      ).toBe(false);
+      expect(rejectedComplete).toHaveBeenCalledOnce();
+      expect(getFollowupQueueDepth(key)).toBe(1);
+    } finally {
+      release.resolve();
+    }
+
+    await expect.poll(() => getExistingFollowupQueue(key)).toBeUndefined();
+    expect(groupCompletions.map((complete) => complete.mock.calls.length)).toEqual([1, 1]);
+  });
+});

src/auto-reply/reply/queue/drain.ts

@@ -1106,6 +1106,7 @@ export function scheduleFollowupDrain(
             isCrossChannel,
             items: queue.items,
             run: effectiveRunFollowup,
+            inFlight: queue.inFlight,
           });
           if (collectDrainResult === "empty") {
             break;
@@ -1194,6 +1195,11 @@ export function scheduleFollowupDrain(
               });
             };
             try {
+              // Mark active group items as in-flight so the drop policy does not
+              // select them as overflow victims while the group drain is awaited.
+              for (const item of activeGroupItems) {
+                queue.inFlight.add(item);
+              }
               await drainGroup();
             } catch (err) {
               if (admitted) {
@@ -1201,6 +1207,9 @@ export function scheduleFollowupDrain(
               }
               throw err;
             } finally {
+              for (const item of activeGroupItems) {
+                queue.inFlight.delete(item);
+              }
               cancellation.dispose();
             }
             if (!admitted) {
@@ -1218,7 +1227,7 @@ export function scheduleFollowupDrain(
           continue;
         }
 
-        if (!(await drainNextQueueItem(queue.items, effectiveRunFollowup))) {
+        if (!(await drainNextQueueItem(queue.items, effectiveRunFollowup, queue.inFlight))) {
           break;
         }
       }

src/auto-reply/reply/queue/enqueue.ts

@@ -3,7 +3,11 @@ import { normalizeOptionalString } from "@openclaw/normalization-core/string-coe
 import { normalizeChatType } from "../../../channels/chat-type.js";
 import { resolveGlobalDedupeCache } from "../../../infra/dedupe.js";
 import { channelRouteDedupeKey } from "../../../plugin-sdk/channel-route.js";
-import { applyQueueDropPolicy, shouldSkipQueueItem } from "../../../utils/queue-helpers.js";
+import {
+  applyQueueDropPolicy,
+  countPendingQueueItems,
+  shouldSkipQueueItem,
+} from "../../../utils/queue-helpers.js";
 import {
   createOverflowSummaryRetrySource,
   kickFollowupDrainIfIdle,
@@ -120,7 +124,8 @@ export function enqueueFollowupRun(
   }
   // drop:new rejects this source without mutating the existing queue. Do not
   // publish an external queued identity for work that will never be admitted.
-  if (queue.dropPolicy === "new" && queue.cap > 0 && queue.items.length >= queue.cap) {
+  const pendingCount = countPendingQueueItems(queue.items, queue.inFlight);
+  if (queue.dropPolicy === "new" && queue.cap > 0 && pendingCount >= queue.cap) {
     completeFollowupRunLifecycle(run);
     return false;
   }
@@ -132,6 +137,7 @@ export function enqueueFollowupRun(
 
   const shouldEnqueue = applyQueueDropPolicy({
     queue,
+    inFlight: queue.inFlight,
     summarize: (item) => normalizeOptionalString(item.summaryLine) || item.prompt.trim(),
     onDrop: (dropped) => {
       if (queue.dropPolicy === "summarize") {
@@ -201,7 +207,7 @@ export function getFollowupQueueDepth(key: string): number {
   if (!queue) {
     return 0;
   }
-  return queue.items.length;
+  return countPendingQueueItems(queue.items, queue.inFlight);
 }
 
 export function resetRecentQueuedMessageIdDedupe(): void {

src/auto-reply/reply/queue/state.ts

@@ -14,6 +14,8 @@ export type FollowupQueueState = {
   abortController: AbortController;
   items: FollowupRun[];
   draining: boolean;
+  /** Identities retained in `items` while delivery awaits; pending cap and depth must exclude them. */
+  inFlight: Set<FollowupRun>;
   lastEnqueuedAt: number;
   mode: QueueMode;
   debounceMs: number;
@@ -112,6 +114,7 @@ export function getFollowupQueue(key: string, settings: QueueSettings): Followup
     abortController: new AbortController(),
     items: [],
     draining: false,
+    inFlight: new Set(),
     lastEnqueuedAt: 0,
     mode: settings.mode,
     debounceMs:
@@ -158,6 +161,7 @@ export function clearFollowupQueue(key: string): number {
     }
   }
   queue.items.length = 0;
+  queue.inFlight.clear();
   queue.droppedCount = 0;
   queue.summaryLines = [];
   queue.summarySources = [];

src/utils/queue-helpers.test.ts

@@ -4,6 +4,7 @@ import {
   applyQueueDropPolicy,
   applyQueueRuntimeSettings,
   clearQueueSummaryState,
+  countPendingQueueItems,
   drainCollectQueueStep,
   drainNextQueueItem,
   hasCrossChannelItems,
@@ -165,14 +166,22 @@ describe("drainCollectQueueStep", () => {
 });
 
 describe("drainNextQueueItem", () => {
+  it("counts only in-flight identities that still intersect the queue", () => {
+    const active = { id: "active" };
+    const pending = { id: "pending" };
+    const alreadyRemoved = { id: "already-removed" };
+
+    expect(countPendingQueueItems([active, pending], new Set([active, alreadyRemoved]))).toBe(1);
+  });
+
   it("keeps overflow survivors when the queue mutates during an awaited drain", async () => {
     type Item = { id: string };
     const queue = {
-      items: [{ id: "m1" }],
+      items: [{ id: "m1" }] as Item[],
       cap: 3,
       dropPolicy: "summarize" as const,
       droppedCount: 0,
-      summaryLines: [],
+      summaryLines: [] as string[],
     };
     const delivered: string[] = [];
     const dropped: string[] = [];
@@ -180,11 +189,16 @@ describe("drainNextQueueItem", () => {
     const gate = new Promise<void>((resolve) => {
       release = resolve;
     });
+    const inFlight = new Set<Item>();
 
-    const firstDrain = drainNextQueueItem(queue.items, async (item: Item) => {
-      delivered.push(item.id);
-      await gate;
-    });
+    const firstDrain = drainNextQueueItem(
+      queue.items,
+      async (item: Item) => {
+        delivered.push(item.id);
+        await gate;
+      },
+      inFlight,
+    );
     await Promise.resolve();
 
     for (let index = 2; index <= 8; index += 1) {
@@ -192,6 +206,7 @@ describe("drainNextQueueItem", () => {
       const shouldEnqueue = applyQueueDropPolicy({
         queue,
         summarize: (queued) => queued.id,
+        inFlight,
         onDrop: (items) => {
           dropped.push(...items.map((queued) => queued.id));
         },
@@ -204,15 +219,48 @@ describe("drainNextQueueItem", () => {
     release();
     await firstDrain;
     while (
-      await drainNextQueueItem(queue.items, async (item) => {
-        delivered.push(item.id);
-      })
+      await drainNextQueueItem(
+        queue.items,
+        async (item) => {
+          delivered.push(item.id);
+        },
+        inFlight,
+      )
     ) {}
 
     expect(delivered).toEqual(["m1", "m6", "m7", "m8"]);
-    expect(dropped).toEqual(["m1", "m2", "m3", "m4", "m5"]);
+    expect(dropped).toEqual(["m2", "m3", "m4", "m5"]);
     expect(queue.items).toEqual([]);
   });
+
+  it("skips in-flight items when selecting drop victims", () => {
+    type Item = { id: string };
+    const m1: Item = { id: "m1" };
+    const m2: Item = { id: "m2" };
+    const m3: Item = { id: "m3" };
+    const m4: Item = { id: "m4" };
+    const queue = {
+      items: [m1, m2, m3, m4],
+      cap: 2,
+      dropPolicy: "old" as const,
+      droppedCount: 0,
+      summaryLines: [] as string[],
+    };
+    const inFlight = new Set<Item>([m1]);
+    const dropped: string[] = [];
+
+    applyQueueDropPolicy({
+      queue,
+      inFlight,
+      summarize: (item) => item.id,
+      onDrop: (items) => {
+        dropped.push(...items.map((item) => item.id));
+      },
+    });
+
+    expect(dropped).toEqual(["m2", "m3"]);
+    expect(queue.items).toEqual([m1, m4]);
+  });
 });
 
 describe("hasCrossChannelItems", () => {

src/utils/queue-helpers.ts

@@ -98,22 +98,41 @@ export function shouldSkipQueueItem<T>(params: {
   return params.dedupe(params.item, params.items);
 }
 
+/** Count identities that are still pending in the queue, excluding active deliveries. */
+export function countPendingQueueItems<T>(items: readonly T[], inFlight?: ReadonlySet<T>): number {
+  if (!inFlight || inFlight.size === 0) {
+    return items.length;
+  }
+  return items.reduce((count, item) => count + (inFlight.has(item) ? 0 : 1), 0);
+}
+
 /** Apply overflow policy before enqueueing another item. */
 export function applyQueueDropPolicy<T>(params: {
   queue: QueueState<T>;
   summarize: (item: T) => string;
   summaryLimit?: number;
   onDrop?: (items: T[]) => void;
+  inFlight?: ReadonlySet<T>;
 }): boolean {
   const cap = params.queue.cap;
-  if (cap <= 0 || params.queue.items.length < cap) {
+  const pendingCount = countPendingQueueItems(params.queue.items, params.inFlight);
+  if (cap <= 0 || pendingCount < cap) {
     return true;
   }
   if (params.queue.dropPolicy === "new") {
     return false;
   }
-  const dropCount = params.queue.items.length - cap + 1;
-  const dropped = params.queue.items.splice(0, dropCount);
+  const dropCount = pendingCount - cap + 1;
+  const dropped: T[] = [];
+  // Active identities remain in the shared array until delivery succeeds; evict only pending work.
+  for (let index = 0; dropped.length < dropCount; ) {
+    const item = params.queue.items[index];
+    if (params.inFlight?.has(item)) {
+      index += 1;
+      continue;
+    }
+    dropped.push(...params.queue.items.splice(index, 1));
+  }
   params.onDrop?.(dropped);
   if (params.queue.dropPolicy === "summarize") {
     for (const item of dropped) {
@@ -181,13 +200,22 @@ export function removeQueuedItemsByRef<T>(items: T[], processed: readonly T[]):
 export async function drainNextQueueItem<T>(
   items: T[],
   run: (item: T) => Promise<void>,
+  inFlight?: Set<T>,
 ): Promise<boolean> {
   const next = items[0];
   if (!next) {
     return false;
   }
-  await run(next);
-  removeQueuedItemsByRef(items, [next]);
+  // Mark the item as in-flight so applyQueueDropPolicy skips it during the
+  // await window when the shared items array is still mutated by enqueuers.
+  inFlight?.add(next);
+  try {
+    await run(next);
+    // Keep the identity protected until its successful by-reference removal.
+    removeQueuedItemsByRef(items, [next]);
+  } finally {
+    inFlight?.delete(next);
+  }
   return true;
 }
 
@@ -198,6 +226,7 @@ async function drainCollectItemIfNeeded<T>(params: {
   setForceIndividualCollect?: (next: boolean) => void;
   items: T[];
   run: (item: T) => Promise<void>;
+  inFlight?: Set<T>;
 }): Promise<"skipped" | "drained" | "empty"> {
   if (!params.forceIndividualCollect && !params.isCrossChannel) {
     return "skipped";
@@ -206,7 +235,7 @@ async function drainCollectItemIfNeeded<T>(params: {
     // Once cross-channel items appear, future collection stays individual to preserve ordering.
     params.setForceIndividualCollect?.(true);
   }
-  const drained = await drainNextQueueItem(params.items, params.run);
+  const drained = await drainNextQueueItem(params.items, params.run, params.inFlight);
   return drained ? "drained" : "empty";
 }
 
@@ -216,6 +245,7 @@ export async function drainCollectQueueStep<T>(params: {
   isCrossChannel: boolean;
   items: T[];
   run: (item: T) => Promise<void>;
+  inFlight?: Set<T>;
 }): Promise<"skipped" | "drained" | "empty"> {
   return await drainCollectItemIfNeeded({
     forceIndividualCollect: params.collectState.forceIndividualCollect,
@@ -225,6 +255,7 @@ export async function drainCollectQueueStep<T>(params: {
     },
     items: params.items,
     run: params.run,
+    inFlight: params.inFlight,
   });
 }