From b13993af15543a22aadcaffde47129a4e71d122b Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Wed, 26 Aug 2026 18:14:06 -0700 Subject: [PATCH] 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 --- .../reply/agent-runner-payloads.test.ts | 22 +++--- .../reply/followup-delivery.channel.test.ts | 15 +++- src/auto-reply/reply/reply-payloads-dedupe.ts | 6 +- .../reply-payloads-transport-dedupe.test.ts | 78 +++++++++++++++++++ 4 files changed, 104 insertions(+), 17 deletions(-) create mode 100644 src/auto-reply/reply/reply-payloads-transport-dedupe.test.ts diff --git a/src/auto-reply/reply/agent-runner-payloads.test.ts b/src/auto-reply/reply/agent-runner-payloads.test.ts index 475626ad62dc..1a23844aeda1 100644 --- a/src/auto-reply/reply/agent-runner-payloads.test.ts +++ b/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", diff --git a/src/auto-reply/reply/followup-delivery.channel.test.ts b/src/auto-reply/reply/followup-delivery.channel.test.ts index 450d740272d7..c6b000dbad79 100644 --- a/src/auto-reply/reply/followup-delivery.channel.test.ts +++ b/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" }]), diff --git a/src/auto-reply/reply/reply-payloads-dedupe.ts b/src/auto-reply/reply/reply-payloads-dedupe.ts index 49179f0f96a9..fa7317729d0c 100644 --- a/src/auto-reply/reply/reply-payloads-dedupe.ts +++ b/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, diff --git a/src/auto-reply/reply/reply-payloads-transport-dedupe.test.ts b/src/auto-reply/reply/reply-payloads-transport-dedupe.test.ts new file mode 100644 index 000000000000..cfc7e65914e6 --- /dev/null +++ b/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); + }, + ); +});