From 3a4b3dfeccf721575c2ef422bad180fa8ba70c75 Mon Sep 17 00:00:00 2001 From: Edward Abrams Date: Sun, 23 Aug 2026 14:09:32 -0700 Subject: [PATCH] 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> --- .../cli-runner/execute-tool-tracking.ts | 17 +++- .../execute.supervisor-capture.test.ts | 78 +++++++++++++++++ .../embedded-agent-messaging-extraction.ts | 41 +++++++++ .../run/attempt-session-prepare.ts | 10 +++ .../run/attempt-stream-prepare.ts | 1 + .../run/message-tool-terminal.test.ts | 68 +++++++++++++++ .../run/message-tool-terminal.ts | 84 ++++++++++++++----- ...ent-subscribe.handlers.tools.completion.ts | 22 ++++- ...ded-agent-subscribe.handlers.tools.test.ts | 58 +++++++++++++ ...embedded-agent-subscribe.handlers.types.ts | 1 + src/agents/embedded-agent-subscribe.types.ts | 1 + 11 files changed, 353 insertions(+), 28 deletions(-) diff --git a/src/agents/cli-runner/execute-tool-tracking.ts b/src/agents/cli-runner/execute-tool-tracking.ts index 302b3904f4e9..f68497b04325 100644 --- a/src/agents/cli-runner/execute-tool-tracking.ts +++ b/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 } : {}), }; diff --git a/src/agents/cli-runner/execute.supervisor-capture.test.ts b/src/agents/cli-runner/execute.supervisor-capture.test.ts index 69443c75221d..832d1f849196 100644 --- a/src/agents/cli-runner/execute.supervisor-capture.test.ts +++ b/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; diff --git a/src/agents/embedded-agent-messaging-extraction.ts b/src/agents/embedded-agent-messaging-extraction.ts index 9c39589e0a58..b108a9bed236 100644 --- a/src/agents/embedded-agent-messaging-extraction.ts +++ b/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, + }); +} diff --git a/src/agents/embedded-agent-runner/run/attempt-session-prepare.ts b/src/agents/embedded-agent-runner/run/attempt-session-prepare.ts index 09791639de35..e9695dadbeb5 100644 --- a/src/agents/embedded-agent-runner/run/attempt-session-prepare.ts +++ b/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({ diff --git a/src/agents/embedded-agent-runner/run/attempt-stream-prepare.ts b/src/agents/embedded-agent-runner/run/attempt-stream-prepare.ts index 59ff61dc206e..d82484bdccae 100644 --- a/src/agents/embedded-agent-runner/run/attempt-stream-prepare.ts +++ b/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, diff --git a/src/agents/embedded-agent-runner/run/message-tool-terminal.test.ts b/src/agents/embedded-agent-runner/run/message-tool-terminal.test.ts index 9cb4f9eae7ab..1dabb70c0530 100644 --- a/src/agents/embedded-agent-runner/run/message-tool-terminal.test.ts +++ b/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[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" }], diff --git a/src/agents/embedded-agent-runner/run/message-tool-terminal.ts b/src/agents/embedded-agent-runner/run/message-tool-terminal.ts index daf59946066e..c9a79db49c4a 100644 --- a/src/agents/embedded-agent-runner/run/message-tool-terminal.ts +++ b/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[0], + "send" | "deliveredPayload" +> & { + sourceReplyDeliveryMode?: SourceReplyDeliveryMode; + currentMessageId?: string | number; + replyToMode?: "off" | "first" | "all" | "batched"; + hasRepliedRef?: { value: boolean }; +}; + function argsRecordForToolCall(context: AfterToolCallContext): Record { if (context.args && typeof context.args === "object" && !Array.isArray(context.args)) { return context.args as Record; @@ -20,27 +32,55 @@ function argsRecordForToolCall(context: AfterToolCallContext): Record 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, }) diff --git a/src/agents/embedded-agent-subscribe.handlers.tools.completion.ts b/src/agents/embedded-agent-subscribe.handlers.tools.completion.ts index 3d1721db5dce..9dce3e05a6ff 100644 --- a/src/agents/embedded-agent-subscribe.handlers.tools.completion.ts +++ b/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 } : {}), diff --git a/src/agents/embedded-agent-subscribe.handlers.tools.test.ts b/src/agents/embedded-agent-subscribe.handlers.tools.test.ts index c0873682890a..4862299eb385 100644 --- a/src/agents/embedded-agent-subscribe.handlers.tools.test.ts +++ b/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"; diff --git a/src/agents/embedded-agent-subscribe.handlers.types.ts b/src/agents/embedded-agent-subscribe.handlers.types.ts index 4e9b1e45d078..e59d789c932a 100644 --- a/src/agents/embedded-agent-subscribe.handlers.types.ts +++ b/src/agents/embedded-agent-subscribe.handlers.types.ts @@ -335,6 +335,7 @@ type ToolHandlerParams = Pick< | "sessionKey" | "currentChannelId" | "currentMessagingTarget" + | "currentAccountId" | "currentThreadId" | "currentMessageId" | "replyToMode" diff --git a/src/agents/embedded-agent-subscribe.types.ts b/src/agents/embedded-agent-subscribe.types.ts index bc3cf60bd163..439bdedca66a 100644 --- a/src/agents/embedded-agent-subscribe.types.ts +++ b/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. */