fix: preserve debounce and followup ordering (#52998) (thanks @osolmaz)

steipete · steipete · commit af209382fe4b · 2026-03-23T11:10:00.000-07:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -27,6 +27,7 @@ Docs: https://docs.openclaw.ai
 - Voice-call/Plivo: stabilize Plivo v2 replay keys so webhook retries and replay protection stop colliding on valid follow-up deliveries.
 - Release/install: keep previously released bundled plugins and Control UI assets in published openclaw npm installs, and fail release checks when those shipped artifacts are missing. Thanks @vincentkoc.
 - Mistral/models: lower bundled Mistral max-token defaults to safe output budgets and teach `openclaw doctor --fix` to repair old persisted Mistral provider configs that still carry context-sized output limits, avoiding deterministic Mistral 422 rejects on fresh and existing setups. Fixes #52599. Thanks @vincentkoc.
+- Telegram/auto-reply: preserve same-chat inbound debounce order without stranding stale busy-session followups, and keep same-key overflow turns ordered when tracked debounce keys are saturated. (#52998) Thanks @osolmaz.
 
 ## 2026.3.22
 
diff --git a/src/auto-reply/inbound-debounce.ts b/src/auto-reply/inbound-debounce.ts
@@ -203,7 +203,11 @@ export function createInboundDebouncer<T>(params: InboundDebounceCreateParams<T>
       return;
     }
     if (!canTrackKey(key)) {
-      await runFlush([item]);
+      // When the debounce map is saturated, fall back to immediate keyed work
+      // instead of buffering, but still preserve same-key ordering.
+      await enqueueKeyTask(key, async () => {
+        await runFlush([item]);
+      });
       return;
     }
 
diff --git a/src/auto-reply/inbound.test.ts b/src/auto-reply/inbound.test.ts
@@ -543,6 +543,69 @@ describe("createInboundDebouncer", () => {
 
     vi.useRealTimers();
   });
+
+  it("keeps same-key overflow work ordered after falling back to immediate flushes", async () => {
+    const started: string[] = [];
+    const finished: string[] = [];
+    let releaseOverflow!: () => void;
+    const overflowGate = new Promise<void>((resolve) => {
+      releaseOverflow = resolve;
+    });
+
+    const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout");
+    const debouncer = createInboundDebouncer<{ key: string; id: string }>({
+      debounceMs: 50,
+      maxTrackedKeys: 1,
+      buildKey: (item) => item.key,
+      onFlush: async (items) => {
+        const ids = items.map((entry) => entry.id).join(",");
+        started.push(ids);
+        if (ids === "2") {
+          await overflowGate;
+        }
+        finished.push(ids);
+      },
+    });
+
+    try {
+      await debouncer.enqueue({ key: "a", id: "1" });
+      const callCountBeforeOverflow = setTimeoutSpy.mock.calls.length;
+      clearTimeout(
+        setTimeoutSpy.mock.results[callCountBeforeOverflow - 1]?.value as ReturnType<
+          typeof setTimeout
+        >,
+      );
+
+      const overflowEnqueue = debouncer.enqueue({ key: "b", id: "2" });
+      await vi.waitFor(() => {
+        expect(started).toEqual(["2"]);
+      });
+
+      const bufferedEnqueue = debouncer.enqueue({ key: "b", id: "3" });
+      const bufferedTimerIndex = setTimeoutSpy.mock.calls.findLastIndex(
+        (call, index) => index >= callCountBeforeOverflow && call[1] === 50,
+      );
+      expect(bufferedTimerIndex).toBeGreaterThanOrEqual(callCountBeforeOverflow);
+      clearTimeout(
+        setTimeoutSpy.mock.results[bufferedTimerIndex]?.value as ReturnType<typeof setTimeout>,
+      );
+      const bufferedFlush = (
+        setTimeoutSpy.mock.calls[bufferedTimerIndex]?.[0] as (() => Promise<void>) | undefined
+      )?.();
+
+      await Promise.resolve();
+      expect(started).toEqual(["2"]);
+      expect(finished).toEqual([]);
+
+      releaseOverflow();
+      await Promise.all([overflowEnqueue, bufferedEnqueue, bufferedFlush]);
+
+      expect(started).toEqual(["2", "3"]);
+      expect(finished).toEqual(["2", "3"]);
+    } finally {
+      setTimeoutSpy.mockRestore();
+    }
+  });
 });
 
 describe("initSessionState BodyStripped", () => {
diff --git a/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts b/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts
@@ -114,6 +114,7 @@ function createMinimalRun(params?: {
   typingMode?: TypingMode;
   blockStreamingEnabled?: boolean;
   isActive?: boolean;
+  isRunActive?: () => boolean;
   shouldFollowup?: boolean;
   resolvedQueueMode?: string;
   runOverrides?: Partial<FollowupRun["run"]>;
@@ -169,6 +170,7 @@ function createMinimalRun(params?: {
         shouldSteer: false,
         shouldFollowup: params?.shouldFollowup ?? false,
         isActive: params?.isActive ?? false,
+        isRunActive: params?.isRunActive,
         isStreaming: false,
         opts,
         typing,
@@ -326,6 +328,23 @@ describe("runReplyAgent heartbeat followup guard", () => {
     expect(state.runEmbeddedPiAgentMock).not.toHaveBeenCalled();
   });
 
+  it("starts draining immediately when the active snapshot is already stale", async () => {
+    const { run } = createMinimalRun({
+      opts: { isHeartbeat: false },
+      isActive: true,
+      isRunActive: () => false,
+      shouldFollowup: true,
+      resolvedQueueMode: "collect",
+    });
+
+    const result = await run();
+
+    expect(result).toBeUndefined();
+    expect(vi.mocked(enqueueFollowupRun)).toHaveBeenCalledTimes(1);
+    expect(vi.mocked(scheduleFollowupDrain)).toHaveBeenCalledTimes(1);
+    expect(state.runEmbeddedPiAgentMock).not.toHaveBeenCalled();
+  });
+
   it("drains followup queue when an unexpected exception escapes the run path", async () => {
     const accounting = await import("./session-run-accounting.js");
     const persistSpy = vi
diff --git a/src/auto-reply/reply/agent-runner.ts b/src/auto-reply/reply/agent-runner.ts
@@ -73,6 +73,7 @@ export async function runReplyAgent(params: {
   shouldSteer: boolean;
   shouldFollowup: boolean;
   isActive: boolean;
+  isRunActive?: () => boolean;
   isStreaming: boolean;
   opts?: GetReplyOptions;
   typing: TypingController;
@@ -104,6 +105,7 @@ export async function runReplyAgent(params: {
     shouldSteer,
     shouldFollowup,
     isActive,
+    isRunActive,
     isStreaming,
     opts,
     typing,
@@ -241,6 +243,11 @@ export async function runReplyAgent(params: {
       queuedRunFollowupTurn,
       false,
     );
+    // Re-check liveness after enqueue so a stale active snapshot cannot leave
+    // the followup queue idle if the original run already finished.
+    if (!isRunActive?.()) {
+      finalizeWithFollowup(undefined, queueKey, queuedRunFollowupTurn);
+    }
     await touchActiveSessionEntry();
     typing.cleanup();
     return undefined;
diff --git a/src/auto-reply/reply/get-reply-run.ts b/src/auto-reply/reply/get-reply-run.ts
@@ -585,6 +585,7 @@ export async function runPreparedReply(
     shouldSteer,
     shouldFollowup,
     isActive,
+    isRunActive: () => isEmbeddedPiRunActive(sessionIdFinal),
     isStreaming,
     opts,
     typing,

Original file line number	Diff line number	Diff line change
`@@ -203,7 +203,11 @@ export function createInboundDebouncer<T>(params: InboundDebounceCreateParams<T>`
`203`	`203`	`return;`
`204`	`204`	`}`
`205`	`205`	`if (!canTrackKey(key)) {`
`206`		`- await runFlush([item]);`
	`206`	`+ // When the debounce map is saturated, fall back to immediate keyed work`
	`207`	`+ // instead of buffering, but still preserve same-key ordering.`
	`208`	`+ await enqueueKeyTask(key, async () => {`
	`209`	`+ await runFlush([item]);`
	`210`	`+ });`
`207`	`211`	`return;`
`208`	`212`	`}`
`209`	`213`