fix(reply): preserve pending thread evidence when reconciling partial send results (#93291)

* fix(reply): preserve pending thread evidence when reconciling partial send results

extractMessagingToolSendResult re-derived threadId/threadImplicit/threadSuppressed
straight from the provider result. Mattermost is the only production provider that
implements extractToolSendResult, and for an implicitly threaded send it reports only
{ to }, so the reconciler overwrote the correct pending thread evidence with undefined.
That defeated same-thread reply suppression in reply-payloads dedupe and delivered the
agent's final reply twice in the thread, on both the native and Codex harnesses.

A partial provider result now keeps the pending thread evidence it does not speak to: a
provider-reported threadId still wins (and clears the implicit flag), but an absent one
no longer erases the pending threadId/threadImplicit/threadSuppressed.

Regression introduced by c67dc59b02 (#90943).

* test(reply): use a core-local stub provider instead of the bundled Mattermost import

The reconcile-thread regression test deep-imported extensions/mattermost from a
core test, which trips the core/extension package boundary (boundary-invariants
"keeps core tests off bundled extension deep imports", extension-test-boundary,
and check-tsgo-core-boundary pulling extensions/mattermost transitively).

Replace it with a core-local channel test plugin that reproduces the same
contract: an implicit-threading extractToolSend, a partial extractToolSendResult
that reports only { to, threadId? }, and no targetsMatchForReplySuppression
matcher. The test now exercises the generic reconciler contract with no
extension dependency. It still fails on pristine main and passes with the fix.

* fix(reply): reconcile thread evidence atomically

---------

Co-authored-by: Vincent Koc <25068+vincentkoc@users.noreply.github.com>

src/agents/embedded-agent-subscribe.tools.reconcile-thread.test.ts

@@ -0,0 +1,164 @@
+import { afterEach, describe, expect, it } from "vitest";
+import { getMatchingMessagingToolReplyTargets } from "../auto-reply/reply/reply-payloads-dedupe.js";
+import { setActivePluginRegistry } from "../plugins/runtime.js";
+import { createChannelTestPluginBase, createTestRegistry } from "../test-utils/channel-plugins.js";
+import {
+  extractMessagingToolSend,
+  extractMessagingToolSendResult,
+} from "./embedded-agent-subscribe.tools.js";
+
+const PARTIAL_RESULT_PROVIDER = "partialthreadprovider";
+
+function createPartialResultPlugin(): unknown {
+  return {
+    ...createChannelTestPluginBase({ id: PARTIAL_RESULT_PROVIDER }),
+    actions: {
+      extractToolSend: ({ args }: { args: Record<string, unknown> }) =>
+        args.action === "send" && typeof args.to === "string"
+          ? { to: args.to, threadImplicit: true }
+          : null,
+      extractToolSendResult: ({ result }: { result: unknown }) => {
+        const toolSend = (result as { details?: { toolSend?: Record<string, unknown> } })?.details
+          ?.toolSend;
+        const to = typeof toolSend?.to === "string" ? toolSend.to : undefined;
+        if (!to) {
+          return null;
+        }
+        const threadId = typeof toolSend?.threadId === "string" ? toolSend.threadId : undefined;
+        return {
+          to,
+          ...(threadId ? { threadId } : {}),
+          ...(toolSend?.threadImplicit === true ? { threadImplicit: true } : {}),
+          ...(toolSend?.threadSuppressed === true ? { threadSuppressed: true } : {}),
+        };
+      },
+    },
+    threading: {
+      resolveAutoThreadId: ({ toolContext }: { toolContext?: { currentThreadTs?: string } }) =>
+        toolContext?.currentThreadTs,
+    },
+  };
+}
+
+function registerPartialResultProvider(): void {
+  setActivePluginRegistry(
+    createTestRegistry([
+      { pluginId: PARTIAL_RESULT_PROVIDER, source: "test", plugin: createPartialResultPlugin() },
+    ]),
+  );
+}
+
+describe("extractMessagingToolSendResult thread evidence", () => {
+  afterEach(() => {
+    setActivePluginRegistry(createTestRegistry());
+  });
+
+  it("preserves implicit thread evidence when the provider result omits it", () => {
+    registerPartialResultProvider();
+
+    const pending = extractMessagingToolSend(
+      "message",
+      { action: "send", provider: PARTIAL_RESULT_PROVIDER, to: "channel:abc", message: "answer" },
+      {
+        currentChannelId: "channel:abc",
+        currentMessagingTarget: "channel:abc",
+        currentThreadId: "root-1",
+        replyToMode: "all",
+      },
+    );
+    expect(pending?.threadImplicit).toBe(true);
+    expect(pending?.threadId).toBe("root-1");
+
+    const confirmed = extractMessagingToolSendResult(pending!, {
+      details: { toolSend: { to: "channel:abc" } },
+    });
+    expect(confirmed.threadImplicit).toBe(true);
+    expect(confirmed.threadId).toBe("root-1");
+
+    const matches = getMatchingMessagingToolReplyTargets({
+      messageProvider: PARTIAL_RESULT_PROVIDER,
+      originatingTo: "channel:abc",
+      originatingThreadId: "root-1",
+      messagingToolSentTargets: [confirmed],
+    });
+    expect(matches).toHaveLength(1);
+  });
+
+  it("lets an explicit provider-reported thread override pending implicit evidence", () => {
+    registerPartialResultProvider();
+
+    const confirmed = extractMessagingToolSendResult(
+      {
+        tool: "message",
+        provider: PARTIAL_RESULT_PROVIDER,
+        to: "channel:abc",
+        threadImplicit: true,
+      },
+      { details: { toolSend: { to: "channel:abc", threadId: "root-9" } } },
+    );
+    expect(confirmed.threadId).toBe("root-9");
+    expect(confirmed.threadImplicit).toBeUndefined();
+  });
+
+  it.each([
+    {
+      name: "provider suppression replaces pending implicit evidence",
+      pending: {
+        threadId: "root-1",
+        threadImplicit: true,
+      },
+      result: {
+        threadSuppressed: true,
+      },
+      expected: {
+        threadId: undefined,
+        threadImplicit: undefined,
+        threadSuppressed: true,
+      },
+    },
+    {
+      name: "provider implicit evidence replaces pending suppression",
+      pending: {
+        threadSuppressed: true,
+      },
+      result: {
+        threadImplicit: true,
+      },
+      expected: {
+        threadId: undefined,
+        threadImplicit: true,
+        threadSuppressed: undefined,
+      },
+    },
+    {
+      name: "a partial result preserves pending suppression",
+      pending: {
+        threadSuppressed: true,
+      },
+      result: {},
+      expected: {
+        threadId: undefined,
+        threadImplicit: undefined,
+        threadSuppressed: true,
+      },
+    },
+  ])("$name", ({ pending, result, expected }) => {
+    registerPartialResultProvider();
+
+    const confirmed = extractMessagingToolSendResult(
+      {
+        tool: "message",
+        provider: PARTIAL_RESULT_PROVIDER,
+        to: "channel:abc",
+        ...pending,
+      },
+      { details: { toolSend: { to: "channel:abc", ...result } } },
+    );
+
+    expect({
+      threadId: confirmed.threadId,
+      threadImplicit: confirmed.threadImplicit,
+      threadSuppressed: confirmed.threadSuppressed,
+    }).toEqual(expected);
+  });
+});

src/agents/embedded-agent-subscribe.tools.ts

@@ -972,13 +972,21 @@ export function extractMessagingToolSendResult(
   if (!extracted?.to) {
     return pending;
   }
+  const extractedThreadId = normalizeOptionalString(extracted.threadId);
+  const providerReportedThread =
+    extractedThreadId != null ||
+    extracted.threadImplicit === true ||
+    extracted.threadSuppressed === true;
+  // Thread route fields are one state. Mixing provider and pending values can
+  // create contradictory implicit and suppressed evidence.
+  const threadEvidence = providerReportedThread ? extracted : pending;
   return {
     ...pending,
     ...extracted,
     accountId: normalizeOptionalString(extracted.accountId) ?? pending.accountId,
     to: normalizeTargetForProvider(providerId ?? pending.provider, extracted.to),
-    threadId: normalizeOptionalString(extracted.threadId),
-    threadImplicit: extracted.threadImplicit === true ? true : undefined,
-    threadSuppressed: extracted.threadSuppressed === true ? true : undefined,
+    threadId: normalizeOptionalString(threadEvidence.threadId),
+    threadImplicit: threadEvidence.threadImplicit === true ? true : undefined,
+    threadSuppressed: threadEvidence.threadSuppressed === true ? true : undefined,
   };
 }