[AI-assisted] fix(reply): wait for block replies before tools (#83722)

IWhatsskill · clawsweeper[bot] · Takhoffman · web-flow · commit 0a4de3de5761 · 2026-05-22T03:32:45.000Z
Summary: - The branch adds an abort-aware dispatcher-idle wait after successful same-channel and direct ACP block replies, plus regression tests and a changelog entry. - Reproducibility: yes. Current main source shows the same-channel block callback queues dispatcher delivery w ... spatcher idle, and the PR body supplies before/after diagnostic output for the tool-start ordering failure. Automerge notes: - PR branch already contained follow-up commit before automerge: [AI-assisted] fix(reply): wait for block replies before tools Validation: - ClawSweeper review passed for head 3257620. - Required merge gates passed before the squash merge. Prepared head SHA: 3257620 Review: #83722 (comment) Co-authored-by: JARVIS-Glasses <284122573+JARVIS-Glasses@users.noreply.github.com> Co-authored-by: clawsweeper <274271284+clawsweeper[bot]@users.noreply.github.com> Co-authored-by: clawsweeper[bot] <274271284+clawsweeper[bot]@users.noreply.github.com> Approved-by: takhoffman Co-authored-by: takhoffman <781889+takhoffman@users.noreply.github.com>
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -24,6 +24,7 @@ Docs: https://docs.openclaw.ai
 ### Fixes
 
 - CLI/agents: allow `openclaw agent --session-key` to target explicit session keys, including agent-scoped legacy keys. (#85121) Thanks @Kaspre.
+- Auto-reply/ACP: wait for same-channel block reply delivery before starting tool work, while still honoring ACP dispatch aborts so stopped turns do not wait on slow channel sends. (#83722) Thanks @IWhatsskill.
 - Agents/subagents: surface blocked child-run completions as errors instead of successful subagent finishes. (#80886) Thanks @TurboTheTurtle.
 - Agents/Pi: treat accepted embedded `sessions_spawn` child-session handoffs as terminal progress so parent turns no longer report false non-deliverable failures. (#85054) Thanks @samzong.
 - WhatsApp: update Baileys to `7.0.0-rc13` and drop the obsolete logger type patch.
diff --git a/src/auto-reply/reply/dispatch-acp-delivery.test.ts b/src/auto-reply/reply/dispatch-acp-delivery.test.ts
@@ -1,7 +1,7 @@
 import { beforeEach, describe, expect, it, vi } from "vitest";
 import type { OpenClawConfig } from "../../config/config.js";
 import { createAcpDispatchDeliveryCoordinator } from "./dispatch-acp-delivery.js";
-import type { ReplyDispatcher } from "./reply-dispatcher.js";
+import { createReplyDispatcher, type ReplyDispatcher } from "./reply-dispatcher.js";
 import { buildTestCtx } from "./test-ctx.js";
 import { createAcpTestConfig } from "./test-fixtures/acp-runtime.js";
 
@@ -240,6 +240,93 @@ describe("createAcpDispatchDeliveryCoordinator", () => {
     expect(coordinator.getRoutedCounts().block).toBe(0);
   });
 
+  it("waits for direct block dispatcher delivery before resolving block delivery", async () => {
+    const delivered: unknown[] = [];
+    let releaseDelivery: (() => void) | undefined;
+    let markDeliveryStarted: (() => void) | undefined;
+    const deliveryStarted = new Promise<void>((resolve) => {
+      markDeliveryStarted = resolve;
+    });
+    const deliveryGate = new Promise<void>((resolve) => {
+      releaseDelivery = resolve;
+    });
+    const dispatcher = createReplyDispatcher({
+      deliver: async (payload) => {
+        delivered.push(payload);
+        markDeliveryStarted?.();
+        await deliveryGate;
+      },
+    });
+    const coordinator = createAcpDispatchDeliveryCoordinator({
+      cfg: createAcpTestConfig(),
+      ctx: buildTestCtx({
+        Provider: "visiblechat",
+        Surface: "visiblechat",
+        SessionKey: "agent:codex-acp:session-1",
+      }),
+      dispatcher,
+      inboundAudio: false,
+      shouldRouteToOriginating: false,
+    });
+
+    let deliverySettled = false;
+    const deliveryPromise = coordinator.deliver("block", { text: "hello" }, { skipTts: true });
+    void deliveryPromise.then(() => {
+      deliverySettled = true;
+    });
+
+    await deliveryStarted;
+
+    expect(delivered).toEqual([{ text: "hello" }]);
+    expect(deliverySettled).toBe(false);
+
+    releaseDelivery?.();
+    await expect(deliveryPromise).resolves.toBe(true);
+    expect(deliverySettled).toBe(true);
+  });
+
+  it("stops waiting for direct block delivery when the ACP dispatch aborts", async () => {
+    const delivered: unknown[] = [];
+    const controller = new AbortController();
+    let releaseDelivery: (() => void) | undefined;
+    let markDeliveryStarted: (() => void) | undefined;
+    const deliveryStarted = new Promise<void>((resolve) => {
+      markDeliveryStarted = resolve;
+    });
+    const deliveryGate = new Promise<void>((resolve) => {
+      releaseDelivery = resolve;
+    });
+    const dispatcher = createReplyDispatcher({
+      deliver: async (payload) => {
+        delivered.push(payload);
+        markDeliveryStarted?.();
+        await deliveryGate;
+      },
+    });
+    const coordinator = createAcpDispatchDeliveryCoordinator({
+      cfg: createAcpTestConfig(),
+      ctx: buildTestCtx({
+        Provider: "visiblechat",
+        Surface: "visiblechat",
+        SessionKey: "agent:codex-acp:session-1",
+      }),
+      dispatcher,
+      inboundAudio: false,
+      shouldRouteToOriginating: false,
+      abortSignal: controller.signal,
+    });
+
+    const deliveryPromise = coordinator.deliver("block", { text: "hello" }, { skipTts: true });
+    await deliveryStarted;
+    controller.abort();
+
+    await expect(deliveryPromise).resolves.toBe(true);
+    expect(delivered).toEqual([{ text: "hello" }]);
+
+    releaseDelivery?.();
+    await dispatcher.waitForIdle();
+  });
+
   it("strips split TTS directives from visible ACP block delivery", async () => {
     const dispatcher = createDispatcher();
     const coordinator = createAcpDispatchDeliveryCoordinator({
diff --git a/src/auto-reply/reply/dispatch-acp-delivery.ts b/src/auto-reply/reply/dispatch-acp-delivery.ts
@@ -13,6 +13,7 @@ import { resolveStatusTtsSnapshot } from "../../tts/status-config.js";
 import { resolveConfiguredTtsMode, shouldCleanTtsDirectiveText } from "../../tts/tts-config.js";
 import type { FinalizedMsgContext } from "../templating.js";
 import type { ReplyPayload } from "../types.js";
+import { waitForReplyDispatcherIdle } from "./reply-dispatcher.js";
 import type { ReplyDispatchKind, ReplyDispatcher } from "./reply-dispatcher.types.js";
 import { resolveRoutedDeliveryThreadId } from "./routed-delivery-thread.js";
 
@@ -191,6 +192,7 @@ export function createAcpDispatchDeliveryCoordinator(params: {
   originatingChannel?: string;
   originatingTo?: string;
   onReplyStart?: () => Promise<void> | void;
+  abortSignal?: AbortSignal;
 }): AcpDispatchDeliveryCoordinator {
   const directChannel = normalizeOptionalLowercaseString(params.ctx.Provider ?? params.ctx.Surface);
   const routedChannel = normalizeOptionalLowercaseString(params.originatingChannel);
@@ -458,6 +460,9 @@ export function createAcpDispatchDeliveryCoordinator(params: {
     } else if (!delivered && tracksVisibleText) {
       state.failedVisibleTextDelivery = true;
     }
+    if (kind === "block" && delivered) {
+      await waitForReplyDispatcherIdle(params.dispatcher, params.abortSignal);
+    }
     return delivered;
   };
 
diff --git a/src/auto-reply/reply/dispatch-acp.ts b/src/auto-reply/reply/dispatch-acp.ts
@@ -386,6 +386,7 @@ export async function tryDispatchAcpReply(params: {
     originatingChannel: params.originatingChannel,
     originatingTo: params.originatingTo,
     onReplyStart: params.onReplyStart,
+    abortSignal: params.abortSignal,
   });
 
   const identityPendingBeforeTurn = isSessionIdentityPending(
diff --git a/src/auto-reply/reply/dispatch-from-config.test.ts b/src/auto-reply/reply/dispatch-from-config.test.ts
@@ -27,7 +27,7 @@ import { createInternalHookEventPayload } from "../../test-utils/internal-hook-e
 import type { MsgContext } from "../templating.js";
 import { setReplyPayloadMetadata, type GetReplyOptions, type ReplyPayload } from "../types.js";
 import { PROVIDER_CONVERSATION_STATE_ERROR_USER_MESSAGE } from "./provider-request-error-classifier.js";
-import type { ReplyDispatcher } from "./reply-dispatcher.js";
+import { createReplyDispatcher, type ReplyDispatcher } from "./reply-dispatcher.js";
 import { buildTestCtx } from "./test-ctx.js";
 
 type AbortResult = { handled: boolean; aborted: boolean; stoppedSubagents?: number };
@@ -4590,6 +4590,56 @@ describe("dispatchReplyFromConfig", () => {
     expect(callOrder).toEqual(["queued:The answer is 42", "dispatch:The answer is 42"]);
   });
 
+  it("waits for same-channel block dispatcher delivery before resolving block replies", async () => {
+    setNoAbort();
+    const ctx = buildTestCtx({ Provider: "whatsapp" });
+    const delivered: ReplyPayload[] = [];
+    let releaseDelivery: (() => void) | undefined;
+    let markDeliveryStarted: (() => void) | undefined;
+    const deliveryStarted = new Promise<void>((resolve) => {
+      markDeliveryStarted = resolve;
+    });
+    const deliveryGate = new Promise<void>((resolve) => {
+      releaseDelivery = resolve;
+    });
+    const dispatcher = createReplyDispatcher({
+      deliver: async (payload) => {
+        delivered.push(payload);
+        markDeliveryStarted?.();
+        await deliveryGate;
+      },
+    });
+    let blockReplySettled = false;
+    const replyResolver = async (
+      _ctx: MsgContext,
+      opts?: GetReplyOptions,
+    ): Promise<ReplyPayload | undefined> => {
+      const blockReplyPromise = Promise.resolve(opts?.onBlockReply?.({ text: "before tool" })).then(
+        () => {
+          blockReplySettled = true;
+        },
+      );
+
+      await deliveryStarted;
+
+      expect(delivered).toEqual([{ text: "before tool" }]);
+      expect(blockReplySettled).toBe(false);
+
+      releaseDelivery?.();
+      await blockReplyPromise;
+      return undefined;
+    };
+
+    await dispatchReplyFromConfig({
+      ctx,
+      cfg: emptyConfig,
+      dispatcher,
+      replyResolver,
+    });
+
+    expect(blockReplySettled).toBe(true);
+  });
+
   it("forwards payload metadata into onBlockReplyQueued context", async () => {
     setNoAbort();
     const dispatcher = createDispatcher();
diff --git a/src/auto-reply/reply/dispatch-from-config.ts b/src/auto-reply/reply/dispatch-from-config.ts
@@ -121,6 +121,7 @@ import { resolveEffectiveReplyRoute } from "./effective-reply-route.js";
 import { withFullRuntimeReplyConfig } from "./get-reply-fast-path.js";
 import { claimInboundDedupe, commitInboundDedupe, releaseInboundDedupe } from "./inbound-dedupe.js";
 import { resolveOriginMessageProvider } from "./origin-routing.js";
+import { waitForReplyDispatcherIdle } from "./reply-dispatcher.js";
 import type { ReplyDispatcher } from "./reply-dispatcher.types.js";
 import {
   createReplyOperation,
@@ -2066,7 +2067,10 @@ export async function dispatchReplyFromConfig(
                   await sendPayloadAsync(normalizedPayload, context?.abortSignal, false);
                 } else {
                   markInboundDedupeReplayUnsafe();
-                  dispatcher.sendBlockReply(normalizedPayload);
+                  const delivered = dispatcher.sendBlockReply(normalizedPayload);
+                  if (delivered) {
+                    await waitForReplyDispatcherIdle(dispatcher, context?.abortSignal);
+                  }
                 }
               };
               return run();
diff --git a/src/auto-reply/reply/reply-dispatcher.ts b/src/auto-reply/reply/reply-dispatcher.ts
@@ -254,6 +254,30 @@ export function createReplyDispatcher(options: ReplyDispatcherOptions): ReplyDis
   };
 }
 
+export async function waitForReplyDispatcherIdle(
+  dispatcher: Pick<ReplyDispatcher, "waitForIdle">,
+  abortSignal?: AbortSignal,
+): Promise<void> {
+  if (!abortSignal) {
+    await dispatcher.waitForIdle();
+    return;
+  }
+  if (abortSignal.aborted) {
+    return;
+  }
+  let removeAbortListener: (() => void) | undefined;
+  const aborted = new Promise<void>((resolve) => {
+    const onAbort = () => resolve();
+    abortSignal.addEventListener("abort", onAbort, { once: true });
+    removeAbortListener = () => abortSignal.removeEventListener("abort", onAbort);
+  });
+  try {
+    await Promise.race([dispatcher.waitForIdle(), aborted]);
+  } finally {
+    removeAbortListener?.();
+  }
+}
+
 export function createReplyDispatcherWithTyping(
   options: ReplyDispatcherWithTypingOptions,
 ): ReplyDispatcherWithTypingResult {
diff --git a/src/auto-reply/reply/reply-flow.test.ts b/src/auto-reply/reply/reply-flow.test.ts
@@ -1,7 +1,7 @@
 import { describe, expect, it, vi } from "vitest";
 import type { OpenClawConfig } from "../../config/types.openclaw.js";
 import { HEARTBEAT_TOKEN, SILENT_REPLY_TOKEN } from "../tokens.js";
-import { createReplyDispatcher } from "./reply-dispatcher.js";
+import { createReplyDispatcher, waitForReplyDispatcherIdle } from "./reply-dispatcher.js";
 import { createReplyToModeFilter } from "./reply-threading.js";
 
 type DeliverPayload = Parameters<Parameters<typeof createReplyDispatcher>[0]["deliver"]>[0];
@@ -217,6 +217,32 @@ describe("createReplyDispatcher", () => {
   });
 });
 
+describe("waitForReplyDispatcherIdle", () => {
+  it("returns when the abort signal fires before the dispatcher becomes idle", async () => {
+    const controller = new AbortController();
+    const waitForIdle = vi.fn(
+      () =>
+        new Promise<void>(() => {
+          // Keep the dispatcher busy until the abort path wins.
+        }),
+    );
+
+    let settled = false;
+    const waitPromise = waitForReplyDispatcherIdle({ waitForIdle }, controller.signal).then(() => {
+      settled = true;
+    });
+
+    await Promise.resolve();
+    expect(settled).toBe(false);
+
+    controller.abort();
+    await waitPromise;
+
+    expect(settled).toBe(true);
+    expect(waitForIdle).toHaveBeenCalledTimes(1);
+  });
+});
+
 describe("createReplyToModeFilter", () => {
   it("handles off/all mode behavior for replyToId", () => {
     const cases: Array<{
