fix: dedupe replies against their resolved delivery thread (#130510)

* fix: dedupe replies against their resolved delivery thread

Compare message-tool deliveries with the channel-owned automatic reply transport before suppressing duplicate output. Preserve explicit thread targets and the existing behavior of channels without a transport hook. Related: #120339.

* test: type the resolved reply transport fixture

* test: preserve Slack thread fallback in follow-up fixture

src/auto-reply/reply/agent-runner-payloads.test.ts

@@ -97,16 +97,20 @@ describe("buildReplyPayloads media filter integration", () => {
               resolveReplyTransport: ({
                 threadId,
                 replyToId,
+                replyToIsExplicit,
                 replyDelivery,
-              }: ResolveReplyTransportParams) => ({
-                replyToId:
-                  replyDelivery?.replyToMode === "off"
-                    ? threadId != null
-                      ? String(threadId)
-                      : undefined
-                    : (replyToId ?? (threadId != null ? String(threadId) : undefined)),
-                threadId: null,
-              }),
+              }: ResolveReplyTransportParams) => {
+                const allowedReply = replyDelivery?.replyToMode === "off" ? undefined : replyToId;
+                // Slack uses the known root for inherited replies, but explicit targets win.
+                const resolved =
+                  replyToIsExplicit === false
+                    ? (threadId ?? allowedReply)
+                    : (allowedReply ?? threadId);
+                return {
+                  replyToId: resolved == null ? undefined : String(resolved),
+                  threadId: null,
+                };
+              },
             },
           },
           source: "test",

src/auto-reply/reply/followup-delivery.channel.test.ts

@@ -168,10 +168,17 @@ describe("follow-up delivery channel boundary", () => {
   it("dedupes later Slack replies against their actual first-mode transport thread", () => {
     const slack = createChannelPlugin("slack");
     slack.threading = {
-      resolveReplyTransport: ({ threadId, replyToId, replyToIsExplicit }) => ({
-        threadId: null,
-        replyToId: replyToIsExplicit ? replyToId : threadId == null ? undefined : String(threadId),
-      }),
+      resolveReplyTransport: ({ threadId, replyToId, replyToIsExplicit }) => {
+        const inheritedThread = threadId == null ? undefined : String(threadId);
+        // First-mode can clear replyToId; Slack still falls back to the inherited thread.
+        return {
+          threadId: null,
+          replyToId:
+            replyToIsExplicit === false
+              ? (inheritedThread ?? replyToId)
+              : (replyToId ?? inheritedThread),
+        };
+      },
     };
     setActivePluginRegistry(
       createTestRegistry([{ pluginId: "slack", plugin: slack, source: "test" }]),

src/auto-reply/reply/reply-payloads-dedupe.ts

@@ -200,14 +200,12 @@ function resolveOriginThreadIdForPayload(params: {
   replyDelivery?: ReplyDeliveryContext;
 }): string | undefined {
   const originThreadId = normalizeThreadIdForComparison(params.originatingThreadId);
-  if (originThreadId && !params.replyToIsExplicit) {
-    return originThreadId;
-  }
   const replyToId = normalizeThreadIdForComparison(params.replyToId);
   const resolveReplyTransport = getChannelPlugin(params.provider)?.threading?.resolveReplyTransport;
-  if (!replyToId || !params.config || !resolveReplyTransport) {
+  if (!params.config || !resolveReplyTransport) {
     return originThreadId;
   }
+  // Implicit replies can leave the inbound thread; dedupe must use the same transport as delivery.
   const transport = resolveReplyTransport({
     cfg: params.config,
     accountId: params.accountId,

src/auto-reply/reply/reply-payloads-transport-dedupe.test.ts

@@ -0,0 +1,78 @@
+import { beforeEach, describe, expect, it } from "vitest";
+import type { ChannelThreadingAdapter } from "../../channels/plugins/types.public.js";
+import { resetPluginRuntimeStateForTest, setActivePluginRegistry } from "../../plugins/runtime.js";
+import {
+  createChannelTestPluginBase,
+  createTestRegistry,
+} from "../../test-utils/channel-plugins.js";
+import { buildReplyPayloads } from "./agent-runner-payloads.js";
+import { resolveFollowupDeliveryPayloads } from "./followup-delivery-payloads.js";
+
+describe("reply dedupe uses the plugin's delivery destination", () => {
+  beforeEach(() => {
+    resetPluginRuntimeStateForTest();
+    setActivePluginRegistry(
+      createTestRegistry([
+        {
+          pluginId: "test-flat",
+          source: "test",
+          plugin: {
+            ...createChannelTestPluginBase({ id: "test-flat" }),
+            threading: {
+              resolveReplyTransport: ({ replyDelivery }) =>
+                replyDelivery?.replyToMode === "off" ? { threadId: null, replyToId: null } : null,
+            } satisfies ChannelThreadingAdapter,
+          },
+        },
+      ]),
+    );
+  });
+
+  it.each([
+    { mode: "off", toolThread: undefined, count: 0 },
+    { mode: "off", toolThread: "inbound-thread", count: 1 },
+    { mode: "all", toolThread: undefined, count: 1 },
+    { mode: "all", toolThread: "inbound-thread", count: 0 },
+  ] as const)(
+    "mode=$mode toolThread=$toolThread across immediate and queued replies",
+    async ({ mode, toolThread, count }) => {
+      const payloads = [{ text: "The completed answer." }];
+      const targets = [
+        {
+          tool: "message",
+          provider: "test-flat",
+          to: "room",
+          threadId: toolThread,
+          text: "The completed answer.",
+        },
+      ];
+      const result = await buildReplyPayloads({
+        config: {},
+        payloads,
+        isHeartbeat: false,
+        didLogHeartbeatStrip: false,
+        blockStreamingEnabled: false,
+        blockReplyPipeline: null,
+        replyToMode: mode,
+        replyToChannel: "test-flat",
+        currentMessageId: "current-message",
+        messageProvider: "test-flat",
+        originatingTo: "room",
+        originatingThreadId: "inbound-thread",
+        messagingToolSentTargets: targets,
+      });
+      expect(result.replyPayloads).toHaveLength(count);
+      expect(
+        resolveFollowupDeliveryPayloads({
+          cfg: {},
+          payloads,
+          messageProvider: "test-flat",
+          originatingTo: "room",
+          originatingThreadId: "inbound-thread",
+          originatingReplyToMode: mode,
+          sentTargets: targets,
+        }),
+      ).toHaveLength(count);
+    },
+  );
+});