From ff759b3f1abf59bb43499d78266534afd2133b0d Mon Sep 17 00:00:00 2001 From: Masato Hoshino Date: Sun, 5 Jul 2026 23:18:03 +0900 Subject: [PATCH] fix(outbound): report honest message delivery status (#99928) * fix(outbound): report delivery status in best-effort message send results Best-effort sends (forced on for implicit message_tool_only source replies) previously collapsed failed and partial_failed durable send results into a success-shaped MessageSendResult, so agents saw a normal-looking envelope while delivery had actually failed. Surface deliveryStatus, a formatted error, sentBeforeError, and per-payload outcomes without changing throw semantics. * fix(outbound): centralize message delivery outcomes Co-authored-by: masatohoshino * docs(changelog): split aggregate entries from code landing --------- Co-authored-by: Peter Steinberger --- src/agents/command/delivery.ts | 62 +++---------------- src/channels/message/runtime.ts | 7 ++- src/channels/message/send.ts | 53 ++++++++++++++++ src/infra/outbound/message.test.ts | 98 ++++++++++++++++++++++++++++-- src/infra/outbound/message.ts | 21 ++++++- 5 files changed, 178 insertions(+), 63 deletions(-) diff --git a/src/agents/command/delivery.ts b/src/agents/command/delivery.ts index 84699e6596c..1f66ef98566 100644 --- a/src/agents/command/delivery.ts +++ b/src/agents/command/delivery.ts @@ -9,7 +9,11 @@ import { resolveSessionAgentId } from "../../agents/agent-scope.js"; import type { ReplyPayload } from "../../auto-reply/reply-payload.js"; import { normalizeReplyPayload } from "../../auto-reply/reply/normalize-reply.js"; import { createReplyMediaPathNormalizer } from "../../auto-reply/reply/reply-media-paths.runtime.js"; -import { sendDurableMessageBatch } from "../../channels/message/runtime.js"; +import { + sendDurableMessageBatch, + serializeDurableMessagePayloadOutcomes, + type SerializedDurableMessagePayloadOutcome, +} from "../../channels/message/runtime.js"; import { getChannelPlugin, normalizeChannelId } from "../../channels/plugins/index.js"; import { createReplyPrefixContext } from "../../channels/reply-prefix.js"; import { createOutboundSendDeps, type CliDeps } from "../../cli/outbound-send-deps.js"; @@ -65,24 +69,6 @@ function createRestartOnlyAbortSignal(source: AbortSignal | undefined): { }; } -/** Per-payload durable delivery status. */ -type AgentCommandDeliveryPayloadStatus = "sent" | "suppressed" | "failed"; - -/** Delivery outcome for one normalized outbound payload. */ -type AgentCommandDeliveryPayloadOutcome = { - index: number; - status: AgentCommandDeliveryPayloadStatus; - reason?: string; - resultCount?: number; - sentBeforeError?: boolean; - stage?: string; - error?: string; - hookEffect?: { - cancelReason?: string; - metadata?: Record; - }; -}; - /** Aggregate delivery status for an agent command result. */ type AgentCommandDeliveryStatus = { requested: true; @@ -96,7 +82,7 @@ type AgentCommandDeliveryStatus = { reason?: string; resultCount?: number; sentBeforeError?: true; - payloadOutcomes?: AgentCommandDeliveryPayloadOutcome[]; + payloadOutcomes?: SerializedDurableMessagePayloadOutcome[]; }; /** Agent command result after payload normalization and optional delivery. */ @@ -239,40 +225,10 @@ function buildDeliveryResult(params: { }; } -function serializeDeliveryPayloadOutcomes( - outcomes: DurableSendResult["payloadOutcomes"], -): AgentCommandDeliveryPayloadOutcome[] | undefined { - if (!outcomes || outcomes.length === 0) { - return undefined; - } - return outcomes.map((outcome) => { - if (outcome.status === "sent") { - return { - index: outcome.index, - status: "sent", - resultCount: outcome.results.length, - }; - } - if (outcome.status === "suppressed") { - return { - index: outcome.index, - status: "suppressed", - reason: outcome.reason, - ...(outcome.hookEffect ? { hookEffect: outcome.hookEffect } : {}), - }; - } - return { - index: outcome.index, - status: "failed", - error: formatErrorMessage(outcome.error), - sentBeforeError: outcome.sentBeforeError, - stage: outcome.stage, - }; - }); -} - function deliveryStatusFromDurableSend(send: DurableSendResult): AgentCommandDeliveryStatus { - const payloadOutcomes = serializeDeliveryPayloadOutcomes(send.payloadOutcomes); + const payloadOutcomes = serializeDurableMessagePayloadOutcomes(send.payloadOutcomes, { + includeHookEffect: true, + }); switch (send.status) { case "sent": return { diff --git a/src/channels/message/runtime.ts b/src/channels/message/runtime.ts index 122678fede8..b3f7606aed3 100644 --- a/src/channels/message/runtime.ts +++ b/src/channels/message/runtime.ts @@ -1,9 +1,14 @@ // Runtime-only barrel for durable message send helpers. Kept separate from the public message // contract barrel so hot imports can choose delivery runtime without pulling every type export. -export { sendDurableMessageBatch, withDurableMessageSendContext } from "./send.js"; +export { + sendDurableMessageBatch, + serializeDurableMessagePayloadOutcomes, + withDurableMessageSendContext, +} from "./send.js"; export type { DurableMessageBatchSendParams, DurableMessageBatchSendResult, DurableMessageSendContext, DurableMessageSendContextParams, + SerializedDurableMessagePayloadOutcome, } from "./send.js"; diff --git a/src/channels/message/send.ts b/src/channels/message/send.ts index 600d5d9419c..f23241cadce 100644 --- a/src/channels/message/send.ts +++ b/src/channels/message/send.ts @@ -104,6 +104,59 @@ export type DurableMessageBatchSendResult = payloadOutcomes?: DurableMessagePayloadDeliveryOutcome[]; }; +export type SerializedDurableMessagePayloadOutcome = + | { index: number; status: "sent"; resultCount: number } + | { + index: number; + status: "suppressed"; + reason: DurableMessageSuppressionReason; + hookEffect?: { + cancelReason?: string; + metadata?: Record; + }; + } + | { + index: number; + status: "failed"; + error: string; + sentBeforeError: boolean; + stage: DurableMessageFailureStage; + }; + +export function serializeDurableMessagePayloadOutcomes( + outcomes: DurableMessageBatchSendResult["payloadOutcomes"], + options?: { + /** Internal diagnostics may retain hook metadata; model-facing JSON results must omit it. */ + includeHookEffect?: boolean; + }, +): SerializedDurableMessagePayloadOutcome[] | undefined { + if (!outcomes || outcomes.length === 0) { + return undefined; + } + return outcomes.map((outcome): SerializedDurableMessagePayloadOutcome => { + if (outcome.status === "sent") { + return { index: outcome.index, status: "sent", resultCount: outcome.results.length }; + } + if (outcome.status === "suppressed") { + return { + index: outcome.index, + status: "suppressed", + reason: outcome.reason, + ...(options?.includeHookEffect === true && outcome.hookEffect + ? { hookEffect: outcome.hookEffect } + : {}), + }; + } + return { + index: outcome.index, + status: "failed", + error: formatErrorMessage(outcome.error), + sentBeforeError: outcome.sentBeforeError, + stage: outcome.stage, + }; + }); +} + const neverAbortedSignal = new AbortController().signal; function toDurableMessageIntent( diff --git a/src/infra/outbound/message.test.ts b/src/infra/outbound/message.test.ts index 22707310b76..deeb95a9f66 100644 --- a/src/infra/outbound/message.test.ts +++ b/src/infra/outbound/message.test.ts @@ -480,6 +480,7 @@ describe("sendMessage", () => { channel: "forum", to: "123456", via: "direct", + deliveryStatus: "sent", }, "send message result", ); @@ -488,7 +489,21 @@ describe("sendMessage", () => { }); it("preserves suppressed direct-send status", async () => { - mocks.deliverOutboundPayloads.mockResolvedValueOnce([]); + mocks.deliverOutboundPayloads.mockImplementationOnce(async (params: unknown) => { + const callbacks = params as { + onPayloadDeliveryOutcome?: (outcome: unknown) => void; + }; + callbacks.onPayloadDeliveryOutcome?.({ + index: 0, + status: "suppressed", + reason: "cancelled_by_message_sending_hook", + hookEffect: { + cancelReason: "owned-by-other-agent", + metadata: { unsafeForJson: 1n }, + }, + }); + return []; + }); const result = await sendMessage({ cfg: {}, @@ -498,26 +513,34 @@ describe("sendMessage", () => { }); expect(result.deliveryStatus).toBe("suppressed"); + expect(result.payloadOutcomes).toEqual([ + { + index: 0, + status: "suppressed", + reason: "cancelled_by_message_sending_hook", + }, + ]); + expect(() => JSON.stringify(result)).not.toThrow(); }); - it("does not throw best-effort direct send failures", async () => { + it("does not throw best-effort direct send failures but reports the failure", async () => { mocks.deliverOutboundPayloads.mockImplementationOnce(async (params: unknown) => { ( params as { onPayloadDeliveryOutcome?: (outcome: { index: number; - payload: { text: string }; status: "failed"; error: Error; - stage: "send"; + sentBeforeError: boolean; + stage: "platform_send"; }) => void; } ).onPayloadDeliveryOutcome?.({ index: 0, - payload: { text: "hi" }, status: "failed", error: new Error("transport unavailable"), - stage: "send", + sentBeforeError: false, + stage: "platform_send", }); return []; }); @@ -536,13 +559,76 @@ describe("sendMessage", () => { to: "123456", via: "direct", result: undefined, + deliveryStatus: "failed", + error: "transport unavailable", }, "best-effort send message result", ); + expect(result.payloadOutcomes).toEqual([ + { + index: 0, + status: "failed", + error: "transport unavailable", + sentBeforeError: false, + stage: "platform_send", + }, + ]); expectDeliveryCallFields({ bestEffort: true, queuePolicy: "best_effort", }); }); + + it("reports partial delivery on best-effort direct sends instead of plain success", async () => { + mocks.deliverOutboundPayloads.mockImplementationOnce(async (params: unknown) => { + const callbacks = params as { + onPayloadDeliveryOutcome?: (outcome: unknown) => void; + }; + callbacks.onPayloadDeliveryOutcome?.({ + index: 0, + status: "sent", + results: [{ channel: "forum", messageId: "m1" }], + }); + callbacks.onPayloadDeliveryOutcome?.({ + index: 1, + status: "failed", + error: new Error("chunk 2 rejected"), + sentBeforeError: true, + stage: "platform_send", + }); + return [{ channel: "forum", messageId: "m1" }]; + }); + + const result = await sendMessage({ + cfg: {}, + channel: "forum", + to: "123456", + content: "hi", + bestEffort: true, + }); + expectRecordFields( + result, + { + channel: "forum", + to: "123456", + via: "direct", + result: { channel: "forum", messageId: "m1" }, + deliveryStatus: "partial_failed", + error: "chunk 2 rejected", + sentBeforeError: true, + }, + "best-effort partial send message result", + ); + expect(result.payloadOutcomes).toEqual([ + { index: 0, status: "sent", resultCount: 1 }, + { + index: 1, + status: "failed", + error: "chunk 2 rejected", + sentBeforeError: true, + stage: "platform_send", + }, + ]); + }); }); diff --git a/src/infra/outbound/message.ts b/src/infra/outbound/message.ts index fbbf8ac03cf..9c6d0e2675f 100644 --- a/src/infra/outbound/message.ts +++ b/src/infra/outbound/message.ts @@ -2,12 +2,17 @@ // requirements, payload plans, gateway fallback, and optional mirroring. import type { ReplyPayload } from "../../auto-reply/reply-payload.js"; import { deriveDurableFinalDeliveryRequirements } from "../../channels/message/capabilities.js"; -import { sendDurableMessageBatch } from "../../channels/message/runtime.js"; +import { + sendDurableMessageBatch, + serializeDurableMessagePayloadOutcomes, + type SerializedDurableMessagePayloadOutcome, +} from "../../channels/message/runtime.js"; import type { OpenClawConfig } from "../../config/types.openclaw.js"; import type { OutboundMediaAccess } from "../../media/load-options.js"; import type { PollInput } from "../../polls.js"; import { normalizePollInput } from "../../polls.js"; import { createLazyRuntimeModule } from "../../shared/lazy-runtime.js"; +import { formatErrorMessage } from "../errors.js"; import { resolveOutboundChannelPlugin } from "./channel-resolution.js"; import { resolveMessageChannelSelection } from "./channel-selection.js"; import { @@ -95,7 +100,11 @@ export type MessageSendResult = { mediaUrl: string | null; mediaUrls?: string[]; result?: OutboundDeliveryResult | { messageId: string }; - deliveryStatus?: "suppressed"; + deliveryStatus?: "sent" | "suppressed" | "partial_failed" | "failed"; + /** Formatted send error when deliveryStatus is "failed" or "partial_failed". */ + error?: string; + sentBeforeError?: boolean; + payloadOutcomes?: SerializedDurableMessagePayloadOutcome[]; dryRun?: boolean; }; @@ -405,6 +414,7 @@ export async function sendMessage(params: MessageSendParams): Promise