diff --git a/qa/scenarios/channels/rewritten-command-single-delivery.yaml b/qa/scenarios/channels/rewritten-command-single-delivery.yaml new file mode 100644 index 000000000000..b6e21e7796d8 --- /dev/null +++ b/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 diff --git a/src/auto-reply/reply/dispatch-from-config.choose-route.ts b/src/auto-reply/reply/dispatch-from-config.choose-route.ts index 65fa662608ce..61003c81e79e 100644 --- a/src/auto-reply/reply/dispatch-from-config.choose-route.ts +++ b/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(); + const pendingBlockDeliveryOutcomes = new Map< + string, + Array> + >(); + 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>, + ): void => { + if (result && isRoutedReplyDelivered(result)) { + deliveredBlockContentKeys.add(createBlockReplyContentKey(payload)); + } + }; + const wasReplyDeliveredAsBlock = async ( + payload: ReplyPayload, + abortSignal?: AbortSignal, + ): Promise => { + 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; @@ -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, }, { diff --git a/src/auto-reply/reply/dispatch-from-config.delivery-and-tts.test-utils.ts b/src/auto-reply/reply/dispatch-from-config.delivery-and-tts.test-utils.ts index 508fa192acb3..b5511e5e0f0f 100644 --- a/src/auto-reply/reply/dispatch-from-config.delivery-and-tts.test-utils.ts +++ b/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 => { + 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 => { + 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((resolve) => { + markBlockStarted = resolve; + }); + const blockRelease = new Promise((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 => { + 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 => { + 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(); diff --git a/src/auto-reply/reply/dispatch-from-config.execute.ts b/src/auto-reply/reply/dispatch-from-config.execute.ts index f1c65d156576..cc66421701c7 100644 --- a/src/auto-reply/reply/dispatch-from-config.execute.ts +++ b/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; } diff --git a/src/auto-reply/reply/dispatch-from-config.finalize.ts b/src/auto-reply/reply/dispatch-from-config.finalize.ts index c8c66d21efc4..7e74277f9b60 100644 --- a/src/auto-reply/reply/dispatch-from-config.finalize.ts +++ b/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) { diff --git a/src/auto-reply/reply/dispatch-from-config.prepare-delivery.ts b/src/auto-reply/reply/dispatch-from-config.prepare-delivery.ts index e1135c34e52a..7f17236cb280 100644 --- a/src/auto-reply/reply/dispatch-from-config.prepare-delivery.ts +++ b/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 => { + ) => { // 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 = {