fix(qa-channel): prevent duplicate agent replies (#114910)

* fix(qa-channel): prevent duplicate final replies

* fix(qa-channel): compare durable tool traces

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);

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<string, unknown>)
+        .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;