diff --git a/extensions/qa-channel/src/inbound.test.ts b/extensions/qa-channel/src/inbound.test.ts index 592f4d111b14..bcfecca78368 100644 --- a/extensions/qa-channel/src/inbound.test.ts +++ b/extensions/qa-channel/src/inbound.test.ts @@ -162,6 +162,111 @@ describe("handleQaInbound", () => { ); }); + it("delivers identical block and final replies exactly once", async () => { + const runtime = createPluginRuntimeMock(); + setQaChannelRuntime(runtime); + + await handleQaInbound(createQaInboundParams()); + + const assembled = firstRunAssembledParams(runtime); + await assembled.delivery.deliver({ text: "single answer" }, { kind: "block" }); + await assembled.delivery.deliver({ text: "single answer" }, { kind: "final" }); + + expect(sendQaBusMessage).toHaveBeenCalledOnce(); + expect(sendQaBusMessage).toHaveBeenCalledWith( + expect.objectContaining({ text: "single answer" }), + ); + }); + + it("delivers an identical final when it adds tool-call trace data", async () => { + const runtime = createPluginRuntimeMock(); + setQaChannelRuntime(runtime); + + await handleQaInbound(createQaInboundParams()); + + const assembled = firstRunAssembledParams(runtime); + await assembled.delivery.deliver({ text: "single answer" }, { kind: "block" }); + await assembled.replyOptions?.onToolStart?.({ phase: "start", name: "search" }); + await assembled.delivery.deliver({ text: "single answer" }, { kind: "final" }); + + expect(sendQaBusMessage).toHaveBeenCalledTimes(2); + expect(sendQaBusMessage).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ text: "single answer", toolCalls: [{ name: "search" }] }), + ); + }); + + it("suppresses an identical normalized tool-call snapshot", async () => { + const runtime = createPluginRuntimeMock(); + setQaChannelRuntime(runtime); + + await handleQaInbound(createQaInboundParams()); + + const assembled = firstRunAssembledParams(runtime); + await assembled.replyOptions?.onToolStart?.({ + phase: "start", + name: "search", + args: { second: 2, first: 1 }, + }); + await assembled.delivery.deliver({ text: "single answer" }, { kind: "block" }); + const toolCalls = vi.mocked(sendQaBusMessage).mock.calls[0]?.[0].toolCalls; + if (!toolCalls?.[0]) { + throw new Error("expected durable tool-call trace"); + } + toolCalls[0].arguments = { first: 1, second: 2 }; + await assembled.delivery.deliver({ text: "single answer" }, { kind: "final" }); + + expect(sendQaBusMessage).toHaveBeenCalledOnce(); + }); + + it("delivers a same-count final when its tool-call record changes", async () => { + const runtime = createPluginRuntimeMock(); + setQaChannelRuntime(runtime); + + await handleQaInbound(createQaInboundParams()); + + const assembled = firstRunAssembledParams(runtime); + await assembled.replyOptions?.onToolStart?.({ + phase: "start", + name: "search", + args: { attempt: 1 }, + }); + await assembled.delivery.deliver({ text: "single answer" }, { kind: "block" }); + const toolCalls = vi.mocked(sendQaBusMessage).mock.calls[0]?.[0].toolCalls; + if (!toolCalls?.[0]) { + throw new Error("expected durable tool-call trace"); + } + toolCalls[0] = { name: "search", arguments: { attempt: 2 } }; + await assembled.delivery.deliver({ text: "single answer" }, { kind: "final" }); + + expect(sendQaBusMessage).toHaveBeenCalledTimes(2); + expect(sendQaBusMessage).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ + text: "single answer", + toolCalls: [{ name: "search", arguments: { attempt: 2 } }], + }), + ); + }); + + it("clears an active preview before suppressing an identical final", async () => { + const runtime = createPluginRuntimeMock(); + setQaChannelRuntime(runtime); + + await handleQaInbound(createQaInboundParams()); + + const assembled = firstRunAssembledParams(runtime); + await assembled.delivery.deliver({ text: "single answer" }, { kind: "block" }); + await assembled.replyOptions?.onPartialReply?.({ text: "new preview" }); + await assembled.delivery.deliver({ text: "single answer" }, { kind: "final" }); + + expect(sendQaBusMessage).toHaveBeenCalledTimes(2); + expect(deleteQaBusMessage).toHaveBeenCalledOnce(); + expect(deleteQaBusMessage).toHaveBeenCalledWith( + expect.objectContaining({ messageId: "preview-1" }), + ); + }); + it("deletes an active preview when reply dispatch fails", async () => { const runtime = createPluginRuntimeMock(); setQaChannelRuntime(runtime); diff --git a/extensions/qa-channel/src/inbound.ts b/extensions/qa-channel/src/inbound.ts index c55bd41153bc..d99850cbbbba 100644 --- a/extensions/qa-channel/src/inbound.ts +++ b/extensions/qa-channel/src/inbound.ts @@ -111,6 +111,33 @@ function formatQaErrorForLog(error: unknown): string { return escaped; } +function normalizeQaToolCallSnapshotValue(value: unknown): unknown { + if (Array.isArray(value)) { + return value.map(normalizeQaToolCallSnapshotValue); + } + if (value && typeof value === "object") { + return Object.fromEntries( + Object.entries(value as Record) + .toSorted(([left], [right]) => (left < right ? -1 : left > right ? 1 : 0)) + .map(([key, entry]) => [key, normalizeQaToolCallSnapshotValue(entry)]), + ); + } + return value; +} + +function serializeQaToolCallSnapshot(toolCalls: QaBusToolCall[]): string { + // Call order is chronological trace data; nested argument keys are the + // unordered surface that must be canonicalized before comparison. + return JSON.stringify( + toolCalls.map((toolCall) => ({ + name: toolCall.name, + ...(toolCall.arguments + ? { arguments: normalizeQaToolCallSnapshotValue(toolCall.arguments) } + : {}), + })), + ); +} + function createQaReplyPreview(params: { account: ResolvedQaChannelAccount; inbound: QaBusMessage; @@ -119,6 +146,8 @@ function createQaReplyPreview(params: { }) { let messageId: string | null = null; let currentText = ""; + let lastDurableText = ""; + let lastDurableToolCallSnapshot = "[]"; let pending = Promise.resolve(); const write = (text: string) => { @@ -170,6 +199,7 @@ function createQaReplyPreview(params: { if (!text.trim()) { return; } + const toolCallSnapshot = serializeQaToolCallSnapshot(params.toolCalls); await sendQaBusMessage({ baseUrl: params.account.baseUrl, accountId: params.account.accountId, @@ -181,12 +211,26 @@ function createQaReplyPreview(params: { replyToId: params.inbound.id, toolCalls: params.toolCalls, }); + lastDurableText = text; + lastDurableToolCallSnapshot = toolCallSnapshot; }; return { clear, async deliver(text: string, kind: string) { await pending; + // Core may close a streamed block with an identical final payload. + // The block is already durable, so posting the final again duplicates the reply. + if ( + kind === "final" && + text === lastDurableText && + serializeQaToolCallSnapshot(params.toolCalls) === lastDurableToolCallSnapshot + ) { + // Count equality is not record equality: a same-count final with changed + // tool records must still be delivered. + await clear(); + return; + } if (kind === "final" && messageId && params.toolCalls.length === 0) { await write(text); return;