fix(agents): recognize explicit current-source replies (#113554)

Use the canonical delivered current-source matcher for CLI, terminal, and subscription completion paths. Preserve provider, account, target, and thread identity so only the active source route suppresses stranded-reply recovery.

Punchcard-Session: golden-river-workshop-vj

Co-authored-by: Edward Abrams <102749+zeroaltitude@users.noreply.github.com>

src/agents/cli-runner/execute-tool-tracking.ts

@@ -17,6 +17,7 @@ import {
 import {
   extractMessagingToolSendResult,
   extractMessagingToolSourceReplyPayload,
+  isDeliveredMessagingToolSendToCurrentSource,
 } from "../embedded-agent-messaging-extraction.js";
 import {
   isMessagingTool,
@@ -246,6 +247,8 @@ export function createCliToolTracking(context: PreparedCliRunContext) {
     const toolArgs = params.args ?? {};
     const isMessagingSend = isMessagingToolSendAction(params.toolName, toolArgs);
     const content = isMessagingSend ? extractCliMessagingContent(toolArgs, params.result) : {};
+    const confirmedTarget =
+      params.target && extractMessagingToolSendResult(params.target, params.result);
     const deliveredCurrentSourceReply =
       isMessagingSend &&
       isDeliveredMessageToolOnlySourceReplyResult({
@@ -254,6 +257,16 @@ export function createCliToolTracking(context: PreparedCliRunContext) {
         args: params.args,
         result: params.result,
         isError: params.isError,
+        allowExplicitSourceRoute: isDeliveredMessagingToolSendToCurrentSource({
+          send: confirmedTarget,
+          config: context.params.config,
+          currentProvider: context.params.messageChannel ?? context.params.messageProvider,
+          currentAccountId: context.params.agentAccountId,
+          currentChannelId: context.params.currentChannelId,
+          currentThreadId: context.params.currentThreadTs,
+          sessionKey: context.params.sessionKey,
+          deliveredPayload: params.result,
+        }),
         deliveryConfirmed: true,
       });
     const sourceReplyFinal = deliveredCurrentSourceReply
@@ -286,11 +299,11 @@ export function createCliToolTracking(context: PreparedCliRunContext) {
         }
       }
     }
-    if (!params.target) {
+    if (!confirmedTarget) {
       return;
     }
     const targetWithContent = {
-      ...extractMessagingToolSendResult(params.target, params.result),
+      ...confirmedTarget,
       ...content,
       ...(sourceReplyFinal !== undefined ? { sourceReplyFinal } : {}),
     };

src/agents/cli-runner/execute.supervisor-capture.test.ts

@@ -2579,6 +2579,84 @@ describe("executePreparedCliRun supervisor output capture", () => {
     ]);
   });
 
+  it.each([
+    {
+      label: "the exact source route",
+      accountId: "account-1",
+      target: "chat123",
+      threadId: "thread-1",
+      expected: true,
+    },
+    {
+      label: "the same target in another account",
+      accountId: "account-2",
+      target: "chat123",
+      threadId: "thread-1",
+      expected: false,
+    },
+    {
+      label: "the same target in another thread",
+      accountId: "account-1",
+      target: "chat123",
+      threadId: "thread-2",
+      expected: false,
+    },
+    {
+      label: "another target",
+      accountId: "account-1",
+      target: "chat456",
+      threadId: "thread-1",
+      expected: false,
+    },
+  ])("records explicit message sends only for $label", async (testCase) => {
+    const context = buildPreparedCliRunContext({ output: "text", provider: "local-cli" });
+    context.mcpDeliveryCapture = true;
+    context.params.sourceReplyDeliveryMode = "message_tool_only";
+    context.params.messageChannel = TEST_MESSAGE_CHANNEL;
+    context.params.agentAccountId = "account-1";
+    context.params.currentChannelId = "chat123";
+    context.params.currentThreadTs = "thread-1";
+    supervisorSpawnMock.mockImplementationOnce(async (...args: unknown[]) => {
+      const input = args[0] as SupervisorSpawnInput;
+      recordMcpLoopbackToolCallResult({
+        captureKey: input.env?.OPENCLAW_MCP_CLI_CAPTURE_KEY ?? "",
+        toolName: "message",
+        args: {
+          action: "send",
+          channel: TEST_MESSAGE_CHANNEL,
+          accountId: testCase.accountId,
+          target: testCase.target,
+          threadId: testCase.threadId,
+          message: "explicit reply",
+        },
+        result: {
+          ok: true,
+          details: {
+            deliveryStatus: "sent",
+            sourceReplySink: "internal-ui",
+            sourceReply: { text: "explicit reply" },
+          },
+        },
+        isError: false,
+      });
+      input.onStdout?.("done");
+      return createManagedRun({
+        reason: "exit",
+        exitCode: 0,
+        exitSignal: null,
+        durationMs: 50,
+        stdout: "",
+        stderr: "",
+        timedOut: false,
+        noOutputTimedOut: false,
+      });
+    });
+
+    const result = await executePreparedCliRun(context);
+
+    expect(result.didDeliverSourceReplyViaMessageTool === true).toBe(testCase.expected);
+  });
+
   it("retains confirmed delivery for long non-streaming message calls", async () => {
     const context = buildPreparedCliRunContext({ output: "text", provider: "local-cli" });
     context.mcpDeliveryCapture = true;

src/agents/embedded-agent-messaging-extraction.ts

@@ -10,6 +10,7 @@ import { uniqueStrings } from "@openclaw/normalization-core/string-normalization
 import { getChannelPlugin, normalizeChannelId } from "../channels/plugins/index.js";
 import type { ChannelMessageActionName } from "../channels/plugins/types.public.js";
 import type { OpenClawConfig } from "../config/types.openclaw.js";
+import { isDeliveredCurrentSourceReply } from "../infra/outbound/source-reply-mirror.js";
 import { normalizeTargetForProvider } from "../infra/outbound/target-normalization.js";
 import {
   normalizeLegacyInteractiveReply,
@@ -350,3 +351,43 @@ export function extractMessagingToolSendResult(
     threadSuppressed: threadEvidence.threadSuppressed === true ? true : undefined,
   };
 }
+
+export function isDeliveredMessagingToolSendToCurrentSource(params: {
+  send: MessagingToolSend | undefined;
+  config?: OpenClawConfig;
+  currentProvider?: string;
+  currentAccountId?: string;
+  currentChannelId?: string;
+  currentMessagingTarget?: string;
+  currentThreadId?: string;
+  sessionKey?: string;
+  deliveredPayload?: unknown;
+}): boolean {
+  const send = params.send;
+  if (!send?.to) {
+    return false;
+  }
+  return isDeliveredCurrentSourceReply({
+    action: "send",
+    channel: send.provider,
+    accountId: send.accountId,
+    currentAccountId: params.currentAccountId,
+    actionParams: {
+      target: send.to,
+      ...(send.threadSuppressed
+        ? { topLevel: true }
+        : send.threadId
+          ? { threadId: send.threadId }
+          : {}),
+    },
+    cfg: params.config ?? {},
+    sessionKey: params.sessionKey,
+    toolContext: {
+      currentChannelProvider: params.currentProvider,
+      currentChannelId: params.currentChannelId,
+      currentMessagingTarget: params.currentMessagingTarget,
+      currentThreadTs: params.currentThreadId,
+    },
+    deliveredPayload: params.deliveredPayload,
+  });
+}

src/agents/embedded-agent-runner/run/attempt-session-prepare.ts

@@ -231,6 +231,16 @@ export async function prepareEmbeddedAttemptAgentSession(input: {
     agent: activeSession.agent,
     sourceReplyDeliveryMode: attempt.sourceReplyDeliveryMode,
     onDeliveredSourceReply: markSourceReplyDelivered,
+    config: attempt.config,
+    currentProvider: attempt.messageChannel ?? attempt.messageProvider,
+    currentAccountId: attempt.agentAccountId,
+    currentChannelId: attempt.currentChannelId,
+    currentMessagingTarget: attempt.currentMessagingTarget,
+    currentThreadId: attempt.currentThreadTs,
+    currentMessageId: attempt.currentMessageId,
+    replyToMode: attempt.replyToMode,
+    hasRepliedRef: attempt.hasRepliedRef,
+    sessionKey: attempt.sessionKey,
   });
   if (input.clientToolPreparation.codeModeControlsEnabledForRun) {
     installCodeModeRepairHook({

src/agents/embedded-agent-runner/run/attempt-stream-prepare.ts

@@ -319,6 +319,7 @@ export function prepareEmbeddedAttemptStream(input: {
     sessionKey: attempt.sessionKey,
     currentChannelId: attempt.currentChannelId,
     currentMessagingTarget: attempt.currentMessagingTarget,
+    currentAccountId: attempt.agentAccountId,
     currentThreadId: attempt.currentThreadTs,
     currentMessageId: attempt.currentMessageId,
     replyToMode: attempt.replyToMode,

src/agents/embedded-agent-runner/run/message-tool-terminal.test.ts

@@ -238,6 +238,74 @@ describe("message-tool-only source replies", () => {
     ).resolves.toBeUndefined();
   });
 
+  it.each([
+    {
+      label: "the exact source route",
+      accountId: "account-1",
+      target: "chat123",
+      threadId: "thread-1",
+      expected: true,
+    },
+    {
+      label: "the same target in another account",
+      accountId: "account-2",
+      target: "chat123",
+      threadId: "thread-1",
+      expected: false,
+    },
+    {
+      label: "the same target in another thread",
+      accountId: "account-1",
+      target: "chat123",
+      threadId: "thread-2",
+      expected: false,
+    },
+    {
+      label: "another target",
+      accountId: "account-1",
+      target: "chat456",
+      threadId: "thread-1",
+      expected: false,
+    },
+  ])("records explicit sends only for $label", async (testCase) => {
+    const agent = {} as unknown as Agent;
+    const onDeliveredSourceReply = vi.fn();
+    installMessageToolOnlyTerminalHook({
+      agent,
+      sourceReplyDeliveryMode: "message_tool_only",
+      onDeliveredSourceReply,
+      config: {},
+      currentProvider: "test-channel",
+      currentAccountId: "account-1",
+      currentChannelId: "chat123",
+      currentThreadId: "thread-1",
+      sessionKey: "agent:main:test-channel:chat123",
+    } as Parameters<typeof installMessageToolOnlyTerminalHook>[0] & {
+      config: object;
+      currentProvider: string;
+      currentAccountId: string;
+      currentChannelId: string;
+      currentThreadId: string;
+      sessionKey: string;
+    });
+
+    await agent.afterToolCall?.(
+      createAfterToolCallContext({
+        toolName: "message",
+        args: {
+          action: "send",
+          channel: "test-channel",
+          accountId: testCase.accountId,
+          target: testCase.target,
+          threadId: testCase.threadId,
+          message: "explicit reply",
+        },
+      }),
+    );
+
+    expect(onDeliveredSourceReply.mock.calls.length > 0).toBe(testCase.expected);
+  });
+
   it("leaves existing after-tool-call output alone when the send failed", async () => {
     const previousAfterToolCall = vi.fn(async () => ({
       content: [{ type: "text" as const, text: "failed" }],

src/agents/embedded-agent-runner/run/message-tool-terminal.ts

@@ -1,15 +1,27 @@
 import type { SourceReplyDeliveryMode } from "../../../auto-reply/get-reply-options.types.js";
 import { readEmbeddedMessageDeliveryFact } from "../../embedded-agent-message-delivery.js";
-/**
- * Detects message-tool-only sends that delivered a visible source reply.
- */
 import {
   isDeliveredMessageToolOnlySourceReplyResult,
   resolveMessageToolSourceReplyFinal,
 } from "../../embedded-agent-message-tool-source-reply.js";
+import {
+  extractMessagingToolSend,
+  extractMessagingToolSendResult,
+  isDeliveredMessagingToolSendToCurrentSource,
+} from "../../embedded-agent-messaging-extraction.js";
 import type { AfterToolCallContext, AfterToolCallResult, Agent } from "../../runtime/index.js";
 import { readToolResultDetails } from "../../tool-result-error.js";
 
+type MessageToolTerminalRoute = Omit<
+  Parameters<typeof isDeliveredMessagingToolSendToCurrentSource>[0],
+  "send" | "deliveredPayload"
+> & {
+  sourceReplyDeliveryMode?: SourceReplyDeliveryMode;
+  currentMessageId?: string | number;
+  replyToMode?: "off" | "first" | "all" | "batched";
+  hasRepliedRef?: { value: boolean };
+};
+
 function argsRecordForToolCall(context: AfterToolCallContext): Record<string, unknown> {
   if (context.args && typeof context.args === "object" && !Array.isArray(context.args)) {
     return context.args as Record<string, unknown>;
@@ -20,27 +32,55 @@ function argsRecordForToolCall(context: AfterToolCallContext): Record<string, un
     : {};
 }
 
-/**
- * Determines whether a `message.send` tool call delivered a visible source reply
- * in message-tool-only delivery mode. Only implicit-route, non-dry-run,
- * delivered sends qualify; explicit routes and errors are not source replies.
- */
-function isDeliveredMessageToolOnlySourceReply(params: {
-  sourceReplyDeliveryMode?: SourceReplyDeliveryMode;
-  context: AfterToolCallContext;
-  hookResult?: AfterToolCallResult;
-}): boolean {
+/** Detects message-tool-only sends that delivered a visible current-source reply. */
+function isDeliveredMessageToolOnlySourceReply(
+  params: MessageToolTerminalRoute & {
+    context: AfterToolCallContext;
+    hookResult?: AfterToolCallResult;
+  },
+): boolean {
+  const toolName = params.context.toolCall.name;
+  const toolArgs = argsRecordForToolCall(params.context);
+  const extractionArgs =
+    toolName === "message" &&
+    params.currentProvider &&
+    typeof toolArgs.provider !== "string" &&
+    typeof toolArgs.channel !== "string"
+      ? { ...toolArgs, provider: params.currentProvider }
+      : toolArgs;
+  const pendingSend = extractMessagingToolSend(toolName, extractionArgs, {
+    config: params.config,
+    currentChannelId: params.currentChannelId,
+    currentMessagingTarget: params.currentMessagingTarget,
+    currentThreadId: params.currentThreadId,
+    currentMessageId: params.currentMessageId,
+    replyToMode: params.replyToMode,
+    hasRepliedRef: params.hasRepliedRef,
+  });
+  const confirmedSend =
+    pendingSend && extractMessagingToolSendResult(pendingSend, params.context.result);
   const deliveryFact = readEmbeddedMessageDeliveryFact(
     readToolResultDetails(params.context.result)?.messageDelivery,
   );
   const isError = params.hookResult?.isError ?? params.context.isError;
   return isDeliveredMessageToolOnlySourceReplyResult({
     sourceReplyDeliveryMode: params.sourceReplyDeliveryMode,
-    toolName: params.context.toolCall.name,
-    args: argsRecordForToolCall(params.context),
+    toolName,
+    args: toolArgs,
     result: params.context.result,
     hookResult: params.hookResult,
     isError,
+    allowExplicitSourceRoute: isDeliveredMessagingToolSendToCurrentSource({
+      send: confirmedSend,
+      config: params.config,
+      currentProvider: params.currentProvider,
+      currentAccountId: params.currentAccountId,
+      currentChannelId: params.currentChannelId,
+      currentMessagingTarget: params.currentMessagingTarget,
+      currentThreadId: params.currentThreadId,
+      sessionKey: params.sessionKey,
+      deliveredPayload: params.context.result,
+    }),
     ...(deliveryFact
       ? {
           deliveryConfirmed:
@@ -50,12 +90,12 @@ function isDeliveredMessageToolOnlySourceReply(params: {
   });
 }
 
-/** Installs an after-tool hook that records source reply delivery evidence. */
-export function installMessageToolOnlyTerminalHook(params: {
-  agent: Agent;
-  sourceReplyDeliveryMode?: SourceReplyDeliveryMode;
-  onDeliveredSourceReply?: () => void;
-}): void {
+export function installMessageToolOnlyTerminalHook(
+  params: MessageToolTerminalRoute & {
+    agent: Agent;
+    onDeliveredSourceReply?: () => void;
+  },
+): void {
   if (params.sourceReplyDeliveryMode !== "message_tool_only") {
     return;
   }
@@ -64,7 +104,7 @@ export function installMessageToolOnlyTerminalHook(params: {
     const hookResult = await previousAfterToolCall?.(context, signal);
     if (
       isDeliveredMessageToolOnlySourceReply({
-        sourceReplyDeliveryMode: params.sourceReplyDeliveryMode,
+        ...params,
         context,
         hookResult,
       })

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

@@ -34,6 +34,7 @@ import {
   extractMessagingToolSend,
   extractMessagingToolSendResult,
   extractMessagingToolSourceReplyPayload,
+  isDeliveredMessagingToolSendToCurrentSource,
 } from "./embedded-agent-messaging-extraction.js";
 import {
   isMessagingTool,
@@ -299,6 +300,9 @@ export async function handleToolExecutionEnd(
     didDeliverMessagingResult && isMessagingSend
       ? [...argumentMediaUrls, ...collectMessagingMediaUrlsFromToolResult(result)]
       : [];
+  const extractionResult = applyToolSendReceiptForExtraction(result, toolSendReceiptResult);
+  const confirmedMessageTarget =
+    messageTarget && extractMessagingToolSendResult(messageTarget, extractionResult);
   const deliveredMessageToolSourceReply =
     didDeliverMessagingResult &&
     isDeliveredMessageToolOnlySourceReplyResult({
@@ -307,6 +311,18 @@ export async function handleToolExecutionEnd(
       args: startArgs,
       result,
       isError: isToolError,
+      allowExplicitSourceRoute: isDeliveredMessagingToolSendToCurrentSource({
+        send: confirmedMessageTarget,
+        config: ctx.params.config,
+        currentProvider: ctx.params.messageChannel,
+        currentAccountId: ctx.params.currentAccountId,
+        currentChannelId: ctx.params.currentChannelId,
+        currentMessagingTarget: ctx.params.currentMessagingTarget,
+        currentThreadId:
+          ctx.params.currentThreadId ?? parseSessionThreadInfoFast(ctx.params.sessionKey).threadId,
+        sessionKey: ctx.params.sessionKey,
+        deliveredPayload: extractionResult,
+      }),
       deliveryConfirmed: didDeliverMessagingResult,
     });
   const deliveredCurrentSourceReply =
@@ -330,11 +346,9 @@ export async function handleToolExecutionEnd(
     ctx.log.debug(`Committed messaging text: tool=${toolName} len=${messageText.length}`);
     ctx.trimMessagingToolSent();
   }
-  if (didDeliverMessagingResult && messageTarget) {
-    const extractionResult = applyToolSendReceiptForExtraction(result, toolSendReceiptResult);
-    const confirmedTarget = extractMessagingToolSendResult(messageTarget, extractionResult);
+  if (didDeliverMessagingResult && confirmedMessageTarget) {
     ctx.state.messagingToolSentTargets.push({
-      ...confirmedTarget,
+      ...confirmedMessageTarget,
       ...(messageText ? { text: messageText } : {}),
       ...(committedMediaUrls.length > 0 ? { mediaUrls: committedMediaUrls.slice() } : {}),
       ...(hasRichContent ? { hasRichContent: true as const } : {}),

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

@@ -1727,6 +1727,64 @@ describe("handleToolExecutionEnd mutating failure recovery", () => {
     expect(ctx.state.currentSourceMessagingToolSentTextsNormalized).toEqual(["qa-msteams-dm-ok"]);
   });
 
+  it.each([
+    {
+      label: "the exact source route",
+      accountId: "account-1",
+      target: "chat123",
+      threadId: "thread-1",
+      expected: true,
+    },
+    {
+      label: "the same target in another account",
+      accountId: "account-2",
+      target: "chat123",
+      threadId: "thread-1",
+      expected: false,
+    },
+    {
+      label: "the same target in another thread",
+      accountId: "account-1",
+      target: "chat123",
+      threadId: "thread-2",
+      expected: false,
+    },
+    {
+      label: "another target",
+      accountId: "account-1",
+      target: "chat456",
+      threadId: "thread-1",
+      expected: false,
+    },
+  ])("records explicit message sends only for $label", async (testCase) => {
+    const { ctx } = createTestContext();
+    Object.assign(ctx.params, {
+      config: {},
+      sourceReplyDeliveryMode: "message_tool_only",
+      messageChannel: "test-channel",
+      currentAccountId: "account-1",
+      currentChannelId: "chat123",
+      currentThreadId: "thread-1",
+    });
+
+    await executeTool(ctx, {
+      toolName: "message",
+      toolCallId: `tool-message-explicit-${testCase.label}`,
+      args: {
+        action: "send",
+        channel: "test-channel",
+        accountId: testCase.accountId,
+        target: testCase.target,
+        threadId: testCase.threadId,
+        message: "explicit reply",
+      },
+      isError: false,
+      result: { details: { ok: true } },
+    });
+
+    expect(ctx.state.messageToolOnlySourceReplyDelivered).toBe(testCase.expected);
+  });
+
   it("records rich-content delivery when visible text is blank", async () => {
     const { ctx } = createTestContext();
     const toolCallId = "tool-message-rich-content";

src/agents/embedded-agent-subscribe.handlers.types.ts

@@ -335,6 +335,7 @@ type ToolHandlerParams = Pick<
   | "sessionKey"
   | "currentChannelId"
   | "currentMessagingTarget"
+  | "currentAccountId"
   | "currentThreadId"
   | "currentMessageId"
   | "replyToMode"

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

@@ -113,6 +113,7 @@ export type SubscribeEmbeddedAgentSessionParams = {
   currentChannelId?: string;
   /** Routable target for the current conversation when it differs from the native channel ID. */
   currentMessagingTarget?: string;
+  currentAccountId?: string;
   /** Current transport thread resolved for this run. */
   currentThreadId?: string;
   /** Current inbound message id used to distinguish child replies from explicit roots. */