fix(auto-reply): prevent duplicate rewritten-command replies (#114379)

* fix(auto-reply): dedupe settled block finals

* chore: leave release note to release flow

qa/scenarios/channels/rewritten-command-single-delivery.yaml

@@ -0,0 +1,75 @@
+title: Rewritten command replies deliver once
+
+scenario:
+  id: rewritten-command-single-delivery
+  surface: channels
+  coverage:
+    primary:
+      - channels.streaming-final-reply
+    secondary:
+      - channels.qa-channel-final-reply
+  objective: Verify a chat command that rewrites into an agent work order produces one final channel reply.
+  successCriteria:
+    - The rewritten /learn turn reaches the agent and returns the requested marker.
+    - Exactly one marker-bearing qa-channel outbound remains after delivery settles.
+  docsRefs:
+    - docs/channels/qa-channel.md
+    - docs/concepts/streaming.md
+  codeRefs:
+    - src/auto-reply/reply/commands-learn.ts
+    - src/auto-reply/reply/get-reply-inline-actions.ts
+    - src/auto-reply/reply/dispatch-from-config.execute.ts
+    - extensions/qa-channel/src/inbound.ts
+  execution:
+    kind: flow
+    channel: qa-channel
+    summary: Send a /learn work-order rewrite and prove its agent reply is delivered once.
+    config:
+      requiredProviderMode: mock-openai
+      conversationId: qa-rewritten-command-single-delivery
+      marker: QA-REWRITTEN-COMMAND-SINGLE-DELIVERY-OK
+
+flow:
+  steps:
+    - name: delivers one rewritten-command reply
+      actions:
+        - assert:
+            expr: env.providerMode === config.requiredProviderMode
+            message: this deterministic rewritten-command proof requires mock-openai
+        - call: waitForGatewayHealthy
+          args:
+            - ref: env
+            - 60000
+        - call: waitForQaChannelReady
+          args:
+            - ref: env
+            - 60000
+        - call: reset
+        - set: startIndex
+          value:
+            expr: state.getSnapshot().messages.length
+        - sendInbound:
+            conversation:
+              id:
+                ref: config.conversationId
+              kind: direct
+            senderId: qa-rewritten-command-operator
+            senderName: QA Rewritten Command Operator
+            text:
+              expr: "`/learn marker check. Reply exactly: ${config.marker}`"
+        - call: waitForCondition
+          saveAs: outbound
+          args:
+            - lambda:
+                expr: "state.getSnapshot().messages.slice(startIndex).find((candidate) => candidate.direction === 'outbound' && candidate.conversation.id === config.conversationId && candidate.text.includes(config.marker))"
+            - expr: liveTurnTimeoutMs(env, 45000)
+        - call: sleep
+          args: [1000]
+        - set: markerOutbounds
+          value:
+            expr: "state.getSnapshot().messages.slice(startIndex).filter((candidate) => candidate.direction === 'outbound' && candidate.conversation.id === config.conversationId && candidate.text.includes(config.marker))"
+        - assert:
+            expr: markerOutbounds.length === 1
+            message:
+              expr: "`expected one rewritten-command reply, got ${markerOutbounds.length}; transcript=${formatTransportTranscript(state, { conversationId: config.conversationId })}`"
+      detailsExpr: outbound.text

src/auto-reply/reply/dispatch-from-config.choose-route.ts

@@ -6,10 +6,12 @@ import {
 import { logVerbose } from "../../globals.js";
 import { registerReplyDispatcherSettledTask } from "../dispatch-dispatcher.js";
 import {
+  copyReplyPayloadMetadata,
   getReplyPayloadMetadata,
   setReplyPayloadMetadata,
   type ReplyPayload,
 } from "../reply-payload.js";
+import { createBlockReplyContentKey } from "./block-reply-pipeline.js";
 import type { CommandSessionMetadataChange } from "./command-session-metadata.js";
 import {
   DispatchReplyOperationAbortedError,
@@ -220,10 +222,79 @@ export async function chooseDispatchRoute(state: PrepareDispatchOperationReadySt
     const reply = resolveSendableOutboundReplyParts(payload);
     return !reply.hasMedia && !hasExecApprovalPayload(payload);
   };
+  const deliveredBlockContentKeys = new Set<string>();
+  const pendingBlockDeliveryOutcomes = new Map<
+    string,
+    Array<Promise<ReplyDispatchDeliveryOutcome>>
+  >();
+  const sendTrackedBlockReply = (payload: ReplyPayload): boolean => {
+    const contentKey = createBlockReplyContentKey(payload);
+    const delivery = captureReplyDispatchDeliveryOutcome(payload);
+    const queued = dispatcher.sendBlockReply(payload);
+    if (!queued || !delivery.isTracked()) {
+      return queued;
+    }
+    const outcomes = pendingBlockDeliveryOutcomes.get(contentKey);
+    if (outcomes) {
+      outcomes.push(delivery.promise);
+    } else {
+      pendingBlockDeliveryOutcomes.set(contentKey, [delivery.promise]);
+    }
+    return queued;
+  };
+  const recordRoutedBlockReplyDelivery = (
+    payload: ReplyPayload,
+    result: Awaited<ReturnType<typeof sendPayloadAsync>>,
+  ): void => {
+    if (result && isRoutedReplyDelivered(result)) {
+      deliveredBlockContentKeys.add(createBlockReplyContentKey(payload));
+    }
+  };
+  const wasReplyDeliveredAsBlock = async (
+    payload: ReplyPayload,
+    abortSignal?: AbortSignal,
+  ): Promise<boolean> => {
+    const contentKey = createBlockReplyContentKey(payload);
+    if (deliveredBlockContentKeys.has(contentKey)) {
+      return true;
+    }
+    const outcomes = pendingBlockDeliveryOutcomes.get(contentKey);
+    if (!outcomes) {
+      return false;
+    }
+    pendingBlockDeliveryOutcomes.delete(contentKey);
+    const settlement = Promise.all(outcomes).then((settledOutcomes) => ({
+      kind: "settled" as const,
+      outcomes: settledOutcomes,
+    }));
+    if (abortSignal?.aborted) {
+      return false;
+    }
+    let removeAbortListener: (() => void) | undefined;
+    const result = abortSignal
+      ? await Promise.race([
+          settlement,
+          new Promise<{ kind: "aborted" }>((resolve) => {
+            const onAbort = () => resolve({ kind: "aborted" });
+            abortSignal.addEventListener("abort", onAbort, { once: true });
+            removeAbortListener = () => abortSignal.removeEventListener("abort", onAbort);
+          }),
+        ]).finally(() => removeAbortListener?.())
+      : await settlement;
+    if (result.kind === "aborted") {
+      return false;
+    }
+    const delivered = result.outcomes.some((outcome) => outcome === "delivered");
+    if (delivered) {
+      deliveredBlockContentKeys.add(contentKey);
+    }
+    return delivered;
+  };
   const sendFinalPayload = async (
     payload: ReplyPayload,
     options: { abortSignal?: AbortSignal; deliveryId?: string } = {},
   ): Promise<{
+    dedupedAgainstBlock?: boolean;
     queuedFinal: boolean;
     routedFinalCount: number;
     dispatcherOutcome?: Promise<ReplyDispatchDeliveryOutcome>;
@@ -272,8 +343,24 @@ export async function chooseDispatchRoute(state: PrepareDispatchOperationReadySt
             accountId: replyRoute.accountId,
           });
     throwIfFinalDeliveryAborted();
-    const normalizedPayload = await normalizeReplyMediaPayload(ttsPayload);
+    let normalizedPayload = await normalizeReplyMediaPayload(ttsPayload);
     throwIfFinalDeliveryAborted();
+    const deliveredAsBlock = await wasReplyDeliveredAsBlock(payload, abortSignal);
+    throwIfFinalDeliveryAborted();
+    if (deliveredAsBlock) {
+      if (createBlockReplyContentKey(normalizedPayload) === createBlockReplyContentKey(payload)) {
+        return { dedupedAgainstBlock: true, queuedFinal: false, routedFinalCount: 0 };
+      }
+      // Final-only transforms such as TTS still need delivery, but the block already
+      // made the text visible. Preserve only the newly added media/rich payload.
+      normalizedPayload = copyReplyPayloadMetadata(normalizedPayload, {
+        ...normalizedPayload,
+        text: undefined,
+      });
+      if (!hasOutboundReplyContent(normalizedPayload, { trimText: true })) {
+        return { dedupedAgainstBlock: true, queuedFinal: false, routedFinalCount: 0 };
+      }
+    }
     const result = await routeReplyToOriginating(normalizedPayload, {
       abortSignal,
       kind: "final",
@@ -520,6 +607,9 @@ export async function chooseDispatchRoute(state: PrepareDispatchOperationReadySt
       flushPendingCommentaryProgress,
       noteCommentaryProgress,
       shouldSuppressMessageToolOnlyTextErrorProgress,
+      sendTrackedBlockReply,
+      recordRoutedBlockReplyDelivery,
+      wasReplyDeliveredAsBlock,
       sendFinalPayload,
     },
     {

src/auto-reply/reply/dispatch-from-config.delivery-and-tts.test-utils.ts

@@ -1219,6 +1219,156 @@ describe("dispatchReplyFromConfig", () => {
     expect(blockReplySentTexts).toContain("The answer is 42");
   });
 
+  it("does not redeliver a final that already settled as an identical block", async () => {
+    setNoAbort();
+    const delivered: Array<{ kind: string; text?: string }> = [];
+    const dispatcher = createReplyDispatcher({
+      deliver: async (payload, info) => {
+        delivered.push({ kind: info.kind, text: payload.text });
+      },
+    });
+    const replyResolver = async (
+      _ctx: MsgContext,
+      opts?: GetReplyOptions,
+    ): Promise<ReplyPayload> => {
+      await opts?.onBlockReply?.({ text: "rewritten command answer" });
+      return { text: "rewritten command answer" };
+    };
+
+    const result = await dispatchReplyFromConfig({
+      ctx: buildTestCtx({ Provider: "qa-channel", Surface: "qa-channel" }),
+      cfg: emptyConfig,
+      dispatcher,
+      replyResolver,
+    });
+    dispatcher.markComplete();
+    await dispatcher.waitForIdle();
+
+    expect(delivered).toEqual([{ kind: "block", text: "rewritten command answer" }]);
+    expect(result.counts).toEqual({ tool: 0, block: 1, final: 0 });
+  });
+
+  it("keeps the final fallback when an identical block delivery fails", async () => {
+    setNoAbort();
+    const delivered: Array<{ kind: string; text?: string }> = [];
+    const dispatcher = createReplyDispatcher({
+      deliver: async (payload, info) => {
+        if (info.kind === "block") {
+          throw new Error("block delivery failed");
+        }
+        delivered.push({ kind: info.kind, text: payload.text });
+      },
+    });
+    const replyResolver = async (
+      _ctx: MsgContext,
+      opts?: GetReplyOptions,
+    ): Promise<ReplyPayload> => {
+      await opts?.onBlockReply?.({ text: "retry this final" });
+      return { text: "retry this final" };
+    };
+
+    const result = await dispatchReplyFromConfig({
+      ctx: buildTestCtx({ Provider: "qa-channel", Surface: "qa-channel" }),
+      cfg: emptyConfig,
+      dispatcher,
+      replyResolver,
+    });
+    dispatcher.markComplete();
+    await dispatcher.waitForIdle();
+
+    expect(delivered).toEqual([{ kind: "final", text: "retry this final" }]);
+    expect(result.counts).toEqual({ tool: 0, block: 1, final: 1 });
+  });
+
+  it("does not send the final fallback when aborted during block settlement", async () => {
+    setNoAbort();
+    let markBlockStarted: (() => void) | undefined;
+    let releaseBlock: (() => void) | undefined;
+    const blockStarted = new Promise<void>((resolve) => {
+      markBlockStarted = resolve;
+    });
+    const blockRelease = new Promise<void>((resolve) => {
+      releaseBlock = resolve;
+    });
+    const delivered: Array<{ kind: string; text?: string }> = [];
+    const dispatcher = createReplyDispatcher({
+      deliver: async (payload, info) => {
+        if (info.kind === "block") {
+          markBlockStarted?.();
+          await blockRelease;
+          throw new Error("block delivery failed after abort");
+        }
+        delivered.push({ kind: info.kind, text: payload.text });
+      },
+    });
+    const abortController = new AbortController();
+    const replyResolver = async (
+      _ctx: MsgContext,
+      opts?: GetReplyOptions,
+    ): Promise<ReplyPayload> => {
+      await opts?.onBlockReply?.({ text: "cancelled rewritten answer" });
+      return { text: "cancelled rewritten answer" };
+    };
+
+    const dispatch = dispatchReplyFromConfig({
+      ctx: buildTestCtx({ Provider: "qa-channel", Surface: "qa-channel" }),
+      cfg: emptyConfig,
+      dispatcher,
+      replyOptions: { abortSignal: abortController.signal },
+      replyResolver,
+    });
+    await blockStarted;
+    abortController.abort();
+    await dispatch;
+    expect(delivered).toEqual([]);
+
+    releaseBlock?.();
+    dispatcher.markComplete();
+    await dispatcher.waitForIdle();
+
+    expect(delivered).toEqual([]);
+  });
+
+  it("keeps final-only TTS media after deduping identical block text", async () => {
+    setNoAbort();
+    ttsMocks.state.synthesizeFinalAudio = true;
+    const delivered: Array<{ kind: string; payload: ReplyPayload }> = [];
+    const dispatcher = createReplyDispatcher({
+      deliver: async (payload, info) => {
+        delivered.push({ kind: info.kind, payload });
+      },
+    });
+    const replyResolver = async (
+      _ctx: MsgContext,
+      opts?: GetReplyOptions,
+    ): Promise<ReplyPayload> => {
+      await opts?.onBlockReply?.({ text: "spoken rewritten answer" });
+      return { text: "spoken rewritten answer" };
+    };
+
+    const result = await dispatchReplyFromConfig({
+      ctx: buildTestCtx({ Provider: "qa-channel", Surface: "qa-channel" }),
+      cfg: emptyConfig,
+      dispatcher,
+      replyResolver,
+    });
+    dispatcher.markComplete();
+    await dispatcher.waitForIdle();
+
+    expect(delivered).toEqual([
+      { kind: "block", payload: { text: "spoken rewritten answer" } },
+      {
+        kind: "final",
+        payload: expect.objectContaining({
+          text: undefined,
+          mediaUrl: "https://example.com/tts-synth.opus",
+          audioAsVoice: true,
+        }),
+      },
+    ]);
+    expect(result.counts).toEqual({ tool: 0, block: 1, final: 1 });
+  });
+
   it("delivers opted-in block reasoning payloads without applying TTS", async () => {
     setNoAbort();
     const dispatcher = createDispatcher();

src/auto-reply/reply/dispatch-from-config.execute.ts

@@ -71,6 +71,7 @@ export async function executeDispatch(state: PrepareDispatchExecutionReadyState)
     reasoningPayloadsEnabled,
     recordAgentDispatchCompleted,
     recordProcessed,
+    recordRoutedBlockReplyDelivery,
     replyConfig,
     replyContextAccountId,
     replyResolver,
@@ -81,6 +82,7 @@ export async function executeDispatch(state: PrepareDispatchExecutionReadyState)
     routeReplyTo,
     runWithDispatchLifecycleAdmission,
     sendPayloadAsync,
+    sendTrackedBlockReply,
     sendPlanUpdate,
     sendPolicy,
     sendPolicyDenied,
@@ -541,15 +543,16 @@ export async function executeDispatch(state: PrepareDispatchExecutionReadyState)
                       return;
                     }
                     if (shouldRouteToOriginating) {
-                      await sendPayloadAsync(
+                      const result = await sendPayloadAsync(
                         normalizedPayload,
                         context?.abortSignal,
                         false,
                         "block",
                       );
+                      recordRoutedBlockReplyDelivery(normalizedPayload, result);
                     } else {
                       markInboundDedupeReplayUnsafe();
-                      const delivered = dispatcher.sendBlockReply(normalizedPayload);
+                      const delivered = sendTrackedBlockReply(normalizedPayload);
                       if (delivered) {
                         state.hasPendingDirectBlockReplyDelivery = true;
                       }

src/auto-reply/reply/dispatch-from-config.finalize.ts

@@ -141,8 +141,11 @@ export async function finalizeDispatchAndAudit(state: ExecuteDispatchReadyState)
       continue;
     }
     sentFinalPayloadDedupeKeys.add(finalPayloadDedupeKey);
-    attemptedFinalDelivery = true;
     const finalReply = await sendFinalPayload(reply, { deliveryId: String(replyIndex) });
+    if (finalReply.dedupedAgainstBlock) {
+      continue;
+    }
+    attemptedFinalDelivery = true;
     queuedFinal = finalReply.queuedFinal || queuedFinal;
     routedFinalCount += finalReply.routedFinalCount;
     if (finalReply.queuedFinal) {

src/auto-reply/reply/dispatch-from-config.prepare-delivery.ts

@@ -197,15 +197,15 @@ export async function prepareDispatchDelivery(state: GatherDispatchRequestReadyS
     abortSignal?: AbortSignal,
     mirror?: boolean,
     kind: ReplyDispatchKind = "tool",
-  ): Promise<void> => {
+  ) => {
     // Keep the runtime guard explicit because this helper is called from nested
     // reply callbacks where TypeScript cannot narrow shouldRouteToOriginating.
     if (!routeReplyRuntime || !routeReplyChannel || !routeReplyTo) {
-      return;
+      return null;
     }
     const effectiveAbortSignal = abortSignal ?? getDispatchAbortSignal();
     if (effectiveAbortSignal?.aborted) {
-      return;
+      return null;
     }
     const result = await routeReplyToOriginating(payload, {
       abortSignal: effectiveAbortSignal,
@@ -215,6 +215,7 @@ export async function prepareDispatchDelivery(state: GatherDispatchRequestReadyS
     if (result && !result.ok) {
       logVerbose(`dispatch-from-config: route-reply failed: ${result.error ?? "unknown error"}`);
     }
+    return result;
   };
 
   type PluginBindingTranscriptOwner = {