fix(auto-reply): unify overflow protection with in-flight drop policy, exclude heartbeats from stranded recovery

obviyus · obviyus · commit 11ac9f4a9730 · 2026-07-10T11:58:33.000+05:30
diff --git a/src/auto-reply/reply/agent-runner.misc.runreplyagent.test.ts b/src/auto-reply/reply/agent-runner.misc.runreplyagent.test.ts
@@ -174,6 +174,24 @@ vi.mock("../../cli/command-secret-gateway.js", () => ({
   }),
 }));
 
+// Dedicated suites cover these sidecars; misc runner cases keep them inert to avoid unrelated graphs.
+vi.mock("../../cli/command-secret-targets.js", () => ({
+  getAgentRuntimeCommandSecretTargetIds: () => new Set<string>(),
+  getScopedChannelsCommandSecretTargets: () => ({ targetIds: new Set<string>() }),
+}));
+
+vi.mock("../../agents/harness/runtime-plugin.js", () => ({
+  ensureSelectedAgentHarnessPlugin: async () => undefined,
+}));
+
+vi.mock("../../commitments/runtime.js", () => ({
+  enqueueCommitmentExtraction: () => false,
+}));
+
+vi.mock("./followup-runner.js", () => ({
+  createFollowupRunner: () => vi.fn(async () => undefined),
+}));
+
 vi.mock("../../utils/provider-utils.js", () => ({
   isReasoningTagProvider: (provider: string | undefined | null) =>
     provider === "google" || provider === "google-gemini-cli",
@@ -3467,6 +3485,7 @@ describe("runReplyAgent private message_tool_only final warning (#85714)", () =>
     summaryLine?: string;
     strandedReplyRetry?: boolean;
     sendPolicyDenied?: boolean;
+    isHeartbeat?: boolean;
     replyOperation?: ReturnType<typeof createReplyOperation>;
   }) {
     const tmp = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-stranded-"));
@@ -3571,6 +3590,7 @@ describe("runReplyAgent private message_tool_only final warning (#85714)", () =>
       resolvedBlockStreamingBreak: "message_end",
       shouldInjectGroupIntro: false,
       typingMode: "instant",
+      ...(params.isHeartbeat ? { opts: { isHeartbeat: true } } : {}),
       ...(params.replyOperation ? { replyOperation: params.replyOperation } : {}),
     });
     return { storePath, tmp, sessionKey, result, finalAssistantText };
@@ -3708,6 +3728,14 @@ describe("runReplyAgent private message_tool_only final warning (#85714)", () =>
     expect(vi.mocked(enqueueFollowupRun)).not.toHaveBeenCalled();
   });
 
+  it("does not warn, enqueue retry, or emit diagnostic for heartbeat runs", async () => {
+    const { result } = await runPrivateFinalCase({ isHeartbeat: true });
+    expect(warnPrivateFinalSpy).not.toHaveBeenCalled();
+    expect(vi.mocked(enqueueFollowupRun)).not.toHaveBeenCalled();
+    const payloads = result === undefined ? [] : normalizeReplyPayloads(result);
+    expect(payloads.some((payload) => payload.text === strandedDiagnosticText)).toBe(false);
+  });
+
   it("does not warn or enqueue retry when send policy denied source delivery", async () => {
     await runPrivateFinalCase({ sendPolicyDenied: true });
     expect(warnPrivateFinalSpy).not.toHaveBeenCalled();
diff --git a/src/auto-reply/reply/agent-runner.ts b/src/auto-reply/reply/agent-runner.ts
@@ -120,8 +120,8 @@ import {
   type FollowupRun,
   type QueueSettings,
 } from "./queue.js";
-import { createReplyMediaContext } from "./reply-media-paths.js";
 import { normalizeReplyPayloadDirectives } from "./reply-delivery.js";
+import { createReplyMediaContext } from "./reply-media-paths.js";
 import { resolveReplyOperationRunState } from "./reply-operation-run-state.js";
 import {
   replyRunRegistry,
@@ -136,7 +136,7 @@ import { incrementRunCompactionCount, persistRunSessionUsage } from "./session-r
 import { resolveSourceReplyVisibilityPolicy } from "./source-reply-delivery-mode.js";
 import {
   buildStrandedReplyDeliveryFailurePayload,
-  STRANDED_REPLY_RETRY_MARKER,
+  buildStrandedReplyRetryFollowupRun,
 } from "./stranded-reply-recovery.js";
 import { createTypingSignaler } from "./typing-mode.js";
 import type { TypingController } from "./typing.js";
@@ -2051,10 +2051,7 @@ export async function runReplyAgent(params: {
       }
       return buildStrandedReplyDeliveryFailurePayload();
     };
-    if (
-      opts?.sourceReplyDeliveryMode === "message_tool_only" &&
-      committedSourceReplyDelivery
-    ) {
+    if (opts?.sourceReplyDeliveryMode === "message_tool_only" && committedSourceReplyDelivery) {
       await opts.onObservedReplyDelivery?.();
     }
     const currentMessageId = sessionCtx.MessageSidFull ?? sessionCtx.MessageSid;
@@ -2634,7 +2631,10 @@ export async function runReplyAgent(params: {
           : (rawAssistantText ?? ""),
       );
       const isRoomEvent = sessionCtx.InboundEventKind === "room_event";
+      // Heartbeats already deliver fallback finals via sendDurableMessageBatch;
+      // recovering here would duplicate that message.
       const isStrandedReply =
+        !isHeartbeat &&
         !isRoomEvent &&
         shouldWarnAboutPrivateMessageToolFinal({
           sourceReplyDeliveryMode: sourceReplyPolicy.sourceReplyDeliveryMode,
@@ -2644,6 +2644,7 @@ export async function runReplyAgent(params: {
         });
       const retryMissingSourceDelivery =
         isStrandedReplyRetryRun &&
+        !isHeartbeat &&
         !isRoomEvent &&
         sourceReplyPolicy.sourceReplyDeliveryMode === "message_tool_only" &&
         !sourceReplyPolicy.sendPolicyDenied &&
@@ -2663,30 +2664,12 @@ export async function runReplyAgent(params: {
         if (isStrandedReplyRetryRun) {
           finalPayloads = [...finalPayloads, buildStrandedReplyDeliveryFailurePayload()];
         } else {
-          const retryDeliveryText = assistantFinalText;
-          const retryPrompt =
-            `[System] Your previous reply was not delivered to the conversation because ` +
-            `you did not call message(action=send). Your reply text was:\n\n` +
-            `"${retryDeliveryText}"\n\n` +
-            `Please deliver this reply now by calling message(action=send). ` +
-            `Do not add any extra commentary; just deliver the original reply.`;
           const retryEnqueued = enqueueFollowupRun(
             queueKey,
-            {
-              ...followupRun,
-              prompt: retryPrompt,
-              summaryLine: STRANDED_REPLY_RETRY_MARKER,
-              strandedReplyRetry: true,
-              disableCollectBatching: true,
-              transcriptPrompt: undefined,
-              userTurnTranscriptRecorder: undefined,
-              currentInboundContext: undefined,
-              run: {
-                ...followupRun.run,
-                sourceReplyDeliveryMode: sourceReplyPolicy.sourceReplyDeliveryMode,
-                suppressNextUserMessagePersistence: true,
-              },
-            },
+            buildStrandedReplyRetryFollowupRun(followupRun, {
+              finalText: assistantFinalText,
+              sourceReplyDeliveryMode: sourceReplyPolicy.sourceReplyDeliveryMode,
+            }),
             resolvedQueue,
             "none",
             runFollowupTurn,
diff --git a/src/auto-reply/reply/followup-runner.ts b/src/auto-reply/reply/followup-runner.ts
@@ -123,7 +123,7 @@ import { incrementRunCompactionCount, persistRunSessionUsage } from "./session-r
 import { resolveSourceReplyVisibilityPolicy } from "./source-reply-delivery-mode.js";
 import {
   buildStrandedReplyDeliveryFailurePayload,
-  STRANDED_REPLY_RETRY_MARKER,
+  buildStrandedReplyRetryFollowupRun,
 } from "./stranded-reply-recovery.js";
 import { createTypingSignaler } from "./typing-mode.js";
 import type { TypingController } from "./typing.js";
@@ -1606,6 +1606,13 @@ export function createFollowupRunner(params: {
         if (isStrandedReplyRetryFollowup(effectiveQueued)) {
           return false;
         }
+        // Heartbeat turns can reach this path: runReplyAgent builds the
+        // followup runner with opts.isHeartbeat and may enqueue-followup while
+        // another run is active. Heartbeats already deliver fallback finals
+        // via sendDurableMessageBatch, so recovery would duplicate delivery.
+        if (opts?.isHeartbeat === true) {
+          return false;
+        }
         const sourceReplyPolicy = resolveSourceReplyVisibilityPolicy({
           cfg: runtimeConfig,
           ctx: {
@@ -1647,33 +1654,15 @@ export function createFollowupRunner(params: {
           channel: queued.originatingChannel ?? run.messageProvider ?? activeSessionEntry?.channel,
           finalTextLength: assistantFinalText.trim().length,
         });
-        const retryDeliveryText = assistantFinalText;
-        const retryPrompt =
-          `[System] Your previous reply was not delivered to the conversation because ` +
-          `you did not call message(action=send). Your reply text was:\n\n` +
-          `"${retryDeliveryText}"\n\n` +
-          `Please deliver this reply now by calling message(action=send). ` +
-          `Do not add any extra commentary; just deliver the original reply.`;
         const retryEnqueued =
           typeof replySessionKey === "string" &&
           replySessionKey.length > 0 &&
           enqueueFollowupRun(
             replySessionKey,
-            {
-              ...effectiveQueued,
-              prompt: retryPrompt,
-              summaryLine: STRANDED_REPLY_RETRY_MARKER,
-              strandedReplyRetry: true,
-              disableCollectBatching: true,
-              transcriptPrompt: undefined,
-              userTurnTranscriptRecorder: undefined,
-              currentInboundContext: undefined,
-              run: {
-                ...run,
-                sourceReplyDeliveryMode: sourceReplyPolicy.sourceReplyDeliveryMode,
-                suppressNextUserMessagePersistence: true,
-              },
-            },
+            buildStrandedReplyRetryFollowupRun(effectiveQueued, {
+              finalText: assistantFinalText,
+              sourceReplyDeliveryMode: sourceReplyPolicy.sourceReplyDeliveryMode,
+            }),
             resolveQueueSettings({
               cfg: runtimeConfig,
               channel: queued.originatingChannel ?? run.messageProvider,
diff --git a/src/auto-reply/reply/queue.collect.test.ts b/src/auto-reply/reply/queue.collect.test.ts
@@ -1993,6 +1993,38 @@ describe("followup queue collect routing", () => {
     expect(getExistingFollowupQueue(key)?.summarySources).toHaveLength(0);
   });
 
+  it("does not advance debounce stamp when overflow rejects an incoming message", () => {
+    const key = `test-priority-followup-debounce-reject-${Date.now()}`;
+    const settings: QueueSettings = {
+      mode: "followup",
+      debounceMs: 5_000,
+      cap: 1,
+      dropPolicy: "old",
+    };
+
+    const priorityAccepted = enqueueFollowupRun(
+      key,
+      createRun({ prompt: "priority retry" }),
+      settings,
+      "none",
+      undefined,
+      false,
+      { position: "front" },
+    );
+    const queue = getExistingFollowupQueue(key);
+    expect(priorityAccepted).toBe(true);
+    expect(queue).toBeDefined();
+    const stampedAt = queue!.lastEnqueuedAt;
+    expect(stampedAt).toBeGreaterThan(0);
+
+    const rejected = enqueueFollowupRun(key, createRun({ prompt: "busy chat noise" }), settings);
+    expect(rejected).toBe(false);
+    expect(getExistingFollowupQueue(key)?.lastEnqueuedAt).toBe(stampedAt);
+    expect(getExistingFollowupQueue(key)?.items.map((item) => item.prompt)).toEqual([
+      "priority retry",
+    ]);
+  });
+
   it("leaves the queue untouched when protected overflow cannot drop enough items", () => {
     const key = `test-priority-followup-atomic-overflow-${Date.now()}`;
     const initialSettings: QueueSettings = {
diff --git a/src/auto-reply/reply/queue/enqueue.ts b/src/auto-reply/reply/queue/enqueue.ts
@@ -137,8 +137,6 @@ export function enqueueFollowupRun(
   if (!markFollowupRunEnqueued(run)) {
     return false;
   }
-  queue.lastEnqueuedAt = Date.now();
-  queue.lastRun = run.run;
 
   const shouldEnqueue = applyQueueDropPolicy({
     queue,
@@ -190,6 +188,10 @@ export function enqueueFollowupRun(
     completeFollowupRunLifecycle(run);
     return false;
   }
+  // Only admitted items refresh debounce; rejected overflow must not starve
+  // protected stranded-reply retries waiting for the quiet window.
+  queue.lastEnqueuedAt = Date.now();
+  queue.lastRun = run.run;
 
   run.queueAbortSignal = queue.abortController.signal;
   if (options.position === "front") {
diff --git a/src/auto-reply/reply/stranded-reply-recovery.ts b/src/auto-reply/reply/stranded-reply-recovery.ts
@@ -1,5 +1,7 @@
+import type { SourceReplyDeliveryMode } from "../get-reply-options.types.js";
 import { markReplyPayloadForSourceSuppressionDelivery } from "../reply-payload.js";
 import type { ReplyPayload } from "../types.js";
+import type { FollowupRun } from "./queue/types.js";
 
 export const STRANDED_REPLY_RETRY_MARKER = "stranded-reply-retry";
 export const STRANDED_REPLY_DELIVERY_FAILURE_TEXT =
@@ -12,3 +14,38 @@ export function buildStrandedReplyDeliveryFailurePayload(): ReplyPayload {
     isStatusNotice: true,
   });
 }
+
+export function buildStrandedReplyRetryPrompt(finalText: string): string {
+  return (
+    `[System] Your previous reply was not delivered to the conversation because ` +
+    `you did not call message(action=send). Your reply text was:\n\n` +
+    `"${finalText}"\n\n` +
+    `Please deliver this reply now by calling message(action=send). ` +
+    `Do not add any extra commentary; just deliver the original reply.`
+  );
+}
+
+/** Build the one-shot recovery followup that re-prompts message(action=send). */
+export function buildStrandedReplyRetryFollowupRun(
+  base: FollowupRun,
+  params: {
+    finalText: string;
+    sourceReplyDeliveryMode: SourceReplyDeliveryMode | undefined;
+  },
+): FollowupRun {
+  return {
+    ...base,
+    prompt: buildStrandedReplyRetryPrompt(params.finalText),
+    summaryLine: STRANDED_REPLY_RETRY_MARKER,
+    strandedReplyRetry: true,
+    disableCollectBatching: true,
+    transcriptPrompt: undefined,
+    userTurnTranscriptRecorder: undefined,
+    currentInboundContext: undefined,
+    run: {
+      ...base.run,
+      sourceReplyDeliveryMode: params.sourceReplyDeliveryMode,
+      suppressNextUserMessagePersistence: true,
+    },
+  };
+}
diff --git a/src/utils/queue-helpers.test.ts b/src/utils/queue-helpers.test.ts
