diff --git a/extensions/telegram/src/outbound-adapter.normalize-payload.test.ts b/extensions/telegram/src/outbound-adapter.normalize-payload.test.ts new file mode 100644 index 000000000000..cffab99dd884 --- /dev/null +++ b/extensions/telegram/src/outbound-adapter.normalize-payload.test.ts @@ -0,0 +1,228 @@ +import { describe, expect, it } from "vitest"; +import { telegramOutbound } from "./outbound-adapter.js"; + +describe("telegramOutbound normalizePayload", () => { + it("normalizes metadata-only direct payloads with provided fallback text", () => { + const normalized = telegramOutbound.normalizePayload?.({ + cfg: {} as never, + payload: { + text: " ", + fallbackText: { text: "Pablo Daily Summary\n- Review the stuck cron." }, + channelData: { + telegram: { + buttons: [[{ text: "Open task", url: "https://example.test/task" }]], + }, + }, + }, + }); + + expect(normalized).toEqual({ + text: "Pablo Daily Summary\n- Review the stuck cron.", + fallbackText: { text: "Pablo Daily Summary\n- Review the stuck cron." }, + channelData: { + telegram: { + buttons: [[{ text: "Open task", url: "https://example.test/task" }]], + }, + }, + }); + }); + + it("keeps reaction-only payloads textless during payload normalization", () => { + const normalized = telegramOutbound.normalizePayload?.({ + cfg: {} as never, + payload: { + fallbackText: { text: "Pablo Daily Summary\n- Review the stuck cron." }, + channelData: { + telegram: { + reaction: { emoji: "+1", replyToId: "123" }, + }, + }, + }, + }); + + expect(normalized).toEqual({ + fallbackText: { text: "Pablo Daily Summary\n- Review the stuck cron." }, + channelData: { + telegram: { + reaction: { emoji: "+1", replyToId: "123" }, + }, + }, + }); + }); + + it("suppresses metadata-only button payloads when no fallback text exists", () => { + const normalized = telegramOutbound.normalizePayload?.({ + cfg: {} as never, + payload: { + channelData: { + telegram: { + buttons: [[{ text: "Open task", url: "https://example.test/task" }]], + }, + }, + }, + }); + + expect(normalized).toBeNull(); + }); + + it("suppresses unrelated metadata-only payloads even when fallback text exists", () => { + const normalized = telegramOutbound.normalizePayload?.({ + cfg: {} as never, + payload: { + fallbackText: { text: "Pablo Daily Summary\n- Review the stuck cron." }, + channelData: { plugin: { traceId: "trace-1" } }, + }, + }); + + expect(normalized).toBeNull(); + }); + + it.each([ + { name: "media", payload: { mediaUrl: "https://example.test/report.png" } }, + { name: "location", payload: { location: { latitude: 1, longitude: 2 } } }, + { + name: "portable buttons", + payload: { + presentation: { + blocks: [{ type: "buttons" as const, buttons: [{ label: "Retry", value: "retry" }] }], + }, + }, + }, + ])("preserves $name payloads without Telegram metadata", ({ payload }) => { + const normalized = telegramOutbound.normalizePayload?.({ cfg: {} as never, payload }); + + expect(normalized).toEqual(payload); + }); + + it("merges all fallback adopters into the linked summary and keeps reactions separate", () => { + const payloads = [ + { text: "Pablo Daily Summary" }, + { + fallbackText: { text: "Pablo Daily Summary", replacesPayloadIndex: 0 }, + channelData: { telegram: { reaction: { emoji: "+1", replyToId: "123" } } }, + }, + { + fallbackText: { text: "Pablo Daily Summary", replacesPayloadIndex: 0 }, + channelData: { telegram: { buttons: [[{ text: "Open task 1", callback_data: "one" }]] } }, + }, + { + fallbackText: { text: "Pablo Daily Summary", replacesPayloadIndex: 0 }, + channelData: { telegram: { buttons: [[{ text: "Open task 2", callback_data: "two" }]] } }, + }, + ]; + const normalizedPayloads = payloads.map( + (payload) => telegramOutbound.normalizePayload?.({ cfg: {} as never, payload }) ?? payload, + ); + const normalized = telegramOutbound.normalizePayloadBatch?.({ + cfg: {} as never, + payloads: normalizedPayloads.map((payload, index) => ({ index, payload })), + }); + + expect(normalized).toEqual([ + { + text: "Pablo Daily Summary", + fallbackText: { text: "Pablo Daily Summary", replacesPayloadIndex: 0 }, + channelData: { + telegram: { + buttons: [ + [{ text: "Open task 1", callback_data: "one" }], + [{ text: "Open task 2", callback_data: "two" }], + ], + }, + }, + }, + { + fallbackText: { text: "Pablo Daily Summary", replacesPayloadIndex: 0 }, + channelData: { telegram: { reaction: { emoji: "+1", replyToId: "123" } } }, + }, + null, + null, + ]); + }); + + it("merges fallback buttons into a linked captioned media payload", () => { + const payloads = [ + { text: "Pablo Daily Summary", mediaUrl: "https://example.test/report.png" }, + { + text: "Pablo Daily Summary", + fallbackText: { text: "Pablo Daily Summary", replacesPayloadIndex: 0 }, + channelData: { telegram: { buttons: [[{ text: "Open task" }]] } }, + }, + ]; + + expect( + telegramOutbound.normalizePayloadBatch?.({ + cfg: {} as never, + payloads: payloads.map((payload, index) => ({ index, payload })), + }), + ).toEqual([ + { + text: "Pablo Daily Summary", + mediaUrl: "https://example.test/report.png", + fallbackText: { text: "Pablo Daily Summary", replacesPayloadIndex: 0 }, + channelData: { telegram: { buttons: [[{ text: "Open task" }]] } }, + }, + null, + ]); + }); + + it("does not merge a fallback adopter with independent media", () => { + const payloads = [ + { text: "Pablo Daily Summary" }, + { + text: "Pablo Daily Summary", + mediaUrl: "https://example.test/detail.png", + fallbackText: { text: "Pablo Daily Summary", replacesPayloadIndex: 0 }, + channelData: { telegram: { buttons: [[{ text: "Open task" }]] } }, + }, + ]; + + expect( + telegramOutbound.normalizePayloadBatch?.({ + cfg: {} as never, + payloads: payloads.map((payload, index) => ({ index, payload })), + }), + ).toEqual(payloads); + }); + + it("does not merge an adopted fallback without an explicit source link", () => { + const payloads = [ + { text: "Pablo Daily Summary" }, + { + text: "Pablo Daily Summary", + fallbackText: { text: "Pablo Daily Summary" }, + channelData: { telegram: { buttons: [[{ text: "Open task" }]] } }, + }, + ]; + + expect( + telegramOutbound.normalizePayloadBatch?.({ + cfg: {} as never, + payloads: payloads.map((payload, index) => ({ index, payload })), + }), + ).toEqual(payloads); + }); + + it("keeps fallback adopters with distinct quote metadata separate", () => { + const payloads = [ + { text: "Pablo Daily Summary" }, + { + text: "Pablo Daily Summary", + fallbackText: { text: "Pablo Daily Summary", replacesPayloadIndex: 0 }, + channelData: { telegram: { quoteText: "First quote" } }, + }, + { + text: "Pablo Daily Summary", + fallbackText: { text: "Pablo Daily Summary", replacesPayloadIndex: 0 }, + channelData: { telegram: { quoteText: "Second quote" } }, + }, + ]; + + expect( + telegramOutbound.normalizePayloadBatch?.({ + cfg: {} as never, + payloads: payloads.map((payload, index) => ({ index, payload })), + }), + ).toEqual(payloads); + }); +}); diff --git a/extensions/telegram/src/outbound-adapter.ts b/extensions/telegram/src/outbound-adapter.ts index efd2db0a6df7..17326e7f5dcc 100644 --- a/extensions/telegram/src/outbound-adapter.ts +++ b/extensions/telegram/src/outbound-adapter.ts @@ -137,6 +137,133 @@ type CreateTelegramOutboundAdapterOptions = { preferFinalAssistantVisibleText?: boolean; }; +function normalizeTelegramMetadataOnlyPayload(payload: ReplyPayload): ReplyPayload | null { + const telegramData = payload.channelData?.telegram as + | { + buttons?: TelegramInlineButtons; + quoteText?: string; + reaction?: { emoji?: unknown; replyToId?: unknown; replyToCurrent?: unknown }; + } + | undefined; + const text = resolveTelegramInteractiveTextFallback({ + text: payload.text, + interactive: payload.interactive, + presentation: payload.presentation, + }); + if ( + text?.trim() || + resolveSendableOutboundReplyParts(payload).mediaUrls.length > 0 || + payload.location || + payload.audioAsVoice === true || + payload.videoAsNote === true || + payload.presentation || + payload.interactive + ) { + return payload; + } + const buttons = resolveTelegramInlineButtons({ + buttons: telegramData?.buttons, + presentation: payload.presentation, + interactive: payload.interactive, + }); + const hasQuoteText = + typeof telegramData?.quoteText === "string" && Boolean(telegramData.quoteText.trim()); + const hasReaction = + typeof telegramData?.reaction?.emoji === "string" && + Boolean(telegramData.reaction.emoji.trim()); + if (hasReaction && !buttons?.length && !hasQuoteText) { + return payload; + } + const fallbackText = payload.fallbackText?.text.trim(); + if (!buttons?.length && !hasQuoteText) { + return null; + } + return fallbackText ? { ...payload, text: fallbackText } : null; +} + +function mergeTelegramFallbackPayloads(source: ReplyPayload, adopter: ReplyPayload): ReplyPayload { + const sourceTelegram = source.channelData?.telegram as + | { buttons?: TelegramInlineButtons; quoteText?: string } + | undefined; + const adopterTelegram = adopter.channelData?.telegram as + | { buttons?: TelegramInlineButtons; quoteText?: string } + | undefined; + const buttons = [...(sourceTelegram?.buttons ?? []), ...(adopterTelegram?.buttons ?? [])]; + const quoteText = sourceTelegram?.quoteText?.trim() + ? sourceTelegram.quoteText + : adopterTelegram?.quoteText; + const telegram = + sourceTelegram || adopterTelegram + ? { + ...adopterTelegram, + ...sourceTelegram, + ...(buttons.length > 0 ? { buttons } : {}), + ...(quoteText ? { quoteText } : {}), + } + : undefined; + return { + ...adopter, + ...source, + fallbackText: adopter.fallbackText, + channelData: { + ...adopter.channelData, + ...source.channelData, + ...(telegram ? { telegram } : {}), + }, + }; +} + +function normalizeTelegramFallbackPayloadBatch( + entries: readonly { index: number; payload: ReplyPayload }[], +): ReadonlyArray { + const normalized: Array = entries.map((entry) => entry.payload); + const positions = new Map(entries.map((entry, position) => [entry.index, position])); + for (const [position, entry] of entries.entries()) { + const fallback = entry.payload.fallbackText; + if ( + fallback?.replacesPayloadIndex === undefined || + entry.payload.text?.trim() !== fallback.text.trim() || + entry.payload.interactive || + entry.payload.presentation || + resolveSendableOutboundReplyParts(entry.payload).mediaUrls.length > 0 || + entry.payload.location || + entry.payload.audioAsVoice === true || + entry.payload.videoAsNote === true + ) { + continue; + } + const channelData = entry.payload.channelData; + const channelDataKeys = channelData ? Object.keys(channelData) : []; + const telegramData = channelData?.telegram as + | { + buttons?: TelegramInlineButtons; + quoteText?: string; + reaction?: unknown; + } + | undefined; + if ( + channelDataKeys.length !== 1 || + channelDataKeys[0] !== "telegram" || + !telegramData?.buttons?.length || + telegramData.quoteText?.trim() || + telegramData.reaction + ) { + continue; + } + const sourcePosition = positions.get(fallback.replacesPayloadIndex); + if (sourcePosition === undefined) { + continue; + } + const source = normalized[sourcePosition]; + if (!source || source.text?.trim() !== fallback.text.trim()) { + continue; + } + normalized[sourcePosition] = mergeTelegramFallbackPayloads(source, entry.payload); + normalized[position] = null; + } + return normalized; +} + export async function sendTelegramPayloadMessages(params: { send: TelegramSendFn; sendLocation: TelegramLocationFn; @@ -302,6 +429,8 @@ export function createTelegramOutboundAdapter( shouldTreatDeliveredTextAsVisible: options.shouldTreatDeliveredTextAsVisible, targetsMatchForReplySuppression: options.targetsMatchForReplySuppression, preferFinalAssistantVisibleText: options.preferFinalAssistantVisibleText, + normalizePayload: ({ payload }) => normalizeTelegramMetadataOnlyPayload(payload), + normalizePayloadBatch: ({ payloads }) => normalizeTelegramFallbackPayloadBatch(payloads), presentationCapabilities: TELEGRAM_PRESENTATION_CAPABILITIES, deliveryCapabilities: { pin: true, diff --git a/src/agents/runtime-plan/types.ts b/src/agents/runtime-plan/types.ts index 6fb18aca30d6..4979818ec1a0 100644 --- a/src/agents/runtime-plan/types.ts +++ b/src/agents/runtime-plan/types.ts @@ -264,6 +264,10 @@ type AgentRuntimeReplyPayloadLocation = { /** Portable reply payload emitted by agent runtimes before channel rendering. */ type AgentRuntimeReplyPayload = { text?: string; + fallbackText?: { + text: string; + replacesPayloadIndex?: number; + }; mediaUrl?: string; mediaUrls?: string[]; trustedLocalMedia?: boolean; diff --git a/src/auto-reply/reply-payload.ts b/src/auto-reply/reply-payload.ts index 4e449b0610e7..c024cd8f85b2 100644 --- a/src/auto-reply/reply-payload.ts +++ b/src/auto-reply/reply-payload.ts @@ -10,6 +10,12 @@ import type { /** Channel-agnostic assistant reply payload. */ export type ReplyPayload = { text?: string; + /** Visible body a channel adapter may use when native structured content requires text. */ + fallbackText?: { + text: string; + /** Batch payload replaced when the adapter adopts this fallback body. */ + replacesPayloadIndex?: number; + }; mediaUrl?: string; mediaUrls?: string[]; /** Internal-only trust signal for gateway webchat local media embedding. */ diff --git a/src/channels/plugins/outbound.types.ts b/src/channels/plugins/outbound.types.ts index 99b341ca0011..b540d6529964 100644 --- a/src/channels/plugins/outbound.types.ts +++ b/src/channels/plugins/outbound.types.ts @@ -160,6 +160,12 @@ type ChannelOutboundNormalizePayloadParams = { accountId?: string | null; }; +type ChannelOutboundNormalizePayloadBatchParams = { + payloads: readonly { index: number; payload: ReplyPayload }[]; + cfg: OpenClawConfig; + accountId?: string | null; +}; + export type ChannelOutboundAdapter = { deliveryMode: "direct" | "gateway" | "hybrid"; chunker?: ((text: string, limit: number, ctx?: ChannelOutboundChunkContext) => string[]) | null; @@ -189,6 +195,10 @@ export type ChannelOutboundAdapter = { supportsPollDurationSeconds?: boolean; supportsAnonymousPolls?: boolean; normalizePayload?: (params: ChannelOutboundNormalizePayloadParams) => ReplyPayload | null; + /** Normalize an ordered batch in place. Return one entry per input; null suppresses that send. */ + normalizePayloadBatch?: ( + params: ChannelOutboundNormalizePayloadBatchParams, + ) => ReadonlyArray; sendTextOnlyErrorPayloads?: boolean; shouldSkipPlainTextSanitization?: (params: { payload: ReplyPayload }) => boolean; resolveEffectiveTextChunkLimit?: (params: { diff --git a/src/cron/isolated-agent/delivery-dispatch.double-announce.test.ts b/src/cron/isolated-agent/delivery-dispatch.double-announce.test.ts index 513097fa70b8..1318e3c0d646 100644 --- a/src/cron/isolated-agent/delivery-dispatch.double-announce.test.ts +++ b/src/cron/isolated-agent/delivery-dispatch.double-announce.test.ts @@ -397,6 +397,296 @@ describe("dispatchCronDelivery — double-announce guard", () => { expect(state.delivered).toBe(true); }); + it("uses non-empty summary text when structured direct payloads are textless", async () => { + const params = makeBaseParams({ synthesizedText: undefined }); + params.summary = "Pablo Daily Summary\n- One task needs attention."; + params.outputText = "Pablo Daily Summary\n- One task needs attention."; + params.deliveryPayloadHasStructuredContent = true; + params.deliveryPayloads = [{ text: " " }, {}] as never; + + const state = await dispatchCronDelivery(params); + + expect(deliverOutboundPayloads).toHaveBeenCalledTimes(1); + expectDeliveryCall(0, { + channel: "telegram", + to: "123456", + payloads: [{ text: "Pablo Daily Summary\n- One task needs attention." }], + skipQueue: true, + }); + expect(state.deliveryAttempted).toBe(true); + expect(state.delivered).toBe(true); + }); + + it("adds generic fallback text to metadata-only direct payloads", async () => { + const params = makeBaseParams({ synthesizedText: undefined }); + params.summary = "Pablo Daily Summary\n- Review the stuck cron."; + params.outputText = "Pablo Daily Summary\n- Review the stuck cron."; + params.deliveryPayloadHasStructuredContent = true; + params.deliveryPayloads = [ + { + text: " ", + channelData: { + telegram: { + buttons: [[{ text: "Open task", url: "https://example.test/task" }]], + }, + }, + }, + ] as never; + + const state = await dispatchCronDelivery(params); + + expect(deliverOutboundPayloads).toHaveBeenCalledTimes(1); + expectDeliveryCall(0, { + channel: "telegram", + to: "123456", + payloads: [ + { text: "Pablo Daily Summary\n- Review the stuck cron." }, + { + fallbackText: { + text: "Pablo Daily Summary\n- Review the stuck cron.", + replacesPayloadIndex: 0, + }, + channelData: { + telegram: { + buttons: [[{ text: "Open task", url: "https://example.test/task" }]], + }, + }, + }, + ], + skipQueue: true, + }); + expect(state.deliveryAttempted).toBe(true); + expect(state.delivered).toBe(true); + }); + + it("leaves portable button-only payloads for channel presentation rendering", async () => { + const params = makeBaseParams({ synthesizedText: undefined }); + params.summary = "Pablo Daily Summary"; + params.outputText = "Pablo Daily Summary"; + params.deliveryPayloadHasStructuredContent = true; + params.deliveryPayloads = [ + { + presentation: { + blocks: [{ type: "buttons", buttons: [{ label: "Retry", value: "retry" }] }], + }, + }, + ]; + + const state = await dispatchCronDelivery(params); + + expectDeliveryCall(0, { + channel: "telegram", + to: "123456", + payloads: [ + { + presentation: { + blocks: [{ type: "buttons", buttons: [{ label: "Retry", value: "retry" }] }], + }, + }, + ], + skipQueue: true, + }); + expect(state.delivered).toBe(true); + }); + + it("leaves channel metadata payload text decisions to the channel adapter", async () => { + const params = makeBaseParams({ synthesizedText: undefined }); + params.summary = "Pablo Daily Summary\n- Review the stuck cron."; + params.outputText = "Pablo Daily Summary\n- Review the stuck cron."; + params.deliveryPayloadHasStructuredContent = true; + params.deliveryPayloads = [ + { + channelData: { + telegram: { + reaction: { emoji: "👍", replyToId: "123" }, + }, + }, + }, + ] as never; + + const state = await dispatchCronDelivery(params); + + expect(deliverOutboundPayloads).toHaveBeenCalledTimes(1); + expectDeliveryCall(0, { + channel: "telegram", + to: "123456", + payloads: [ + { text: "Pablo Daily Summary\n- Review the stuck cron." }, + { + fallbackText: { + text: "Pablo Daily Summary\n- Review the stuck cron.", + replacesPayloadIndex: 0, + }, + channelData: { + telegram: { + reaction: { emoji: "👍", replyToId: "123" }, + }, + }, + }, + ], + skipQueue: true, + }); + expect(state.deliveryAttempted).toBe(true); + expect(state.delivered).toBe(true); + }); + + it("carries the summary payload index into channel-owned fallback normalization", async () => { + const params = makeBaseParams({ synthesizedText: undefined }); + params.summary = "Pablo Daily Summary\n- Review the stuck cron."; + params.outputText = "Pablo Daily Summary\n- Review the stuck cron."; + params.deliveryPayloadHasStructuredContent = true; + params.deliveryPayloads = [ + { text: " " }, + { text: "Pablo Daily Summary\n- Review the stuck cron." }, + { + channelData: { + telegram: { + reaction: { emoji: "👍", replyToId: "123" }, + }, + }, + }, + { + text: " ", + channelData: { + telegram: { + buttons: [[{ text: "Open task", url: "https://example.test/task" }]], + }, + }, + }, + ] as never; + + const state = await dispatchCronDelivery(params); + + expect(deliverOutboundPayloads).toHaveBeenCalledTimes(1); + expectDeliveryCall(0, { + channel: "telegram", + to: "123456", + payloads: [ + { text: "Pablo Daily Summary\n- Review the stuck cron." }, + { + fallbackText: { + text: "Pablo Daily Summary\n- Review the stuck cron.", + replacesPayloadIndex: 0, + }, + channelData: { + telegram: { + reaction: { emoji: "👍", replyToId: "123" }, + }, + }, + }, + { + fallbackText: { + text: "Pablo Daily Summary\n- Review the stuck cron.", + replacesPayloadIndex: 0, + }, + channelData: { + telegram: { + buttons: [[{ text: "Open task", url: "https://example.test/task" }]], + }, + }, + }, + ], + skipQueue: true, + }); + expect(state.deliveryAttempted).toBe(true); + expect(state.delivered).toBe(true); + }); + + it("reuses captioned media as the source for metadata fallback", async () => { + const params = makeBaseParams({ synthesizedText: undefined }); + params.summary = "Pablo Daily Summary"; + params.outputText = "Pablo Daily Summary"; + params.deliveryPayloadHasStructuredContent = true; + params.deliveryPayloads = [ + { text: "Pablo Daily Summary", mediaUrl: "https://example.test/report.png" }, + { + channelData: { + telegram: { buttons: [[{ text: "Open task", url: "https://example.test/task" }]] }, + }, + }, + ]; + + const state = await dispatchCronDelivery(params); + + expect(deliverOutboundPayloads).toHaveBeenCalledTimes(1); + expectDeliveryCall(0, { + channel: "telegram", + to: "123456", + payloads: [ + { text: "Pablo Daily Summary", mediaUrl: "https://example.test/report.png" }, + { + fallbackText: { text: "Pablo Daily Summary", replacesPayloadIndex: 0 }, + channelData: { + telegram: { buttons: [[{ text: "Open task", url: "https://example.test/task" }]] }, + }, + }, + ], + skipQueue: true, + }); + expect(state.delivered).toBe(true); + }); + + it("does not attach fallback hints when the direct summary is silent", async () => { + const params = makeBaseParams({ synthesizedText: undefined }); + params.summary = SILENT_REPLY_TOKEN; + params.outputText = SILENT_REPLY_TOKEN; + params.deliveryPayloadHasStructuredContent = true; + params.deliveryPayloads = [ + { + text: SILENT_REPLY_TOKEN, + channelData: { + telegram: { + buttons: [[{ text: "Open task", url: "https://example.test/task" }]], + }, + }, + }, + ] as never; + + const state = await dispatchCronDelivery(params); + + expect(deliverOutboundPayloads).toHaveBeenCalledTimes(1); + expectDeliveryCall(0, { + channel: "telegram", + to: "123456", + payloads: [ + { + channelData: { + telegram: { + buttons: [[{ text: "Open task", url: "https://example.test/task" }]], + }, + }, + }, + ], + skipQueue: true, + }); + expect(state.deliveryAttempted).toBe(true); + expect(state.delivered).toBe(true); + }); + + it("uses summary fallback for non-Telegram direct payloads that normalize away", async () => { + const params = makeBaseParams({ synthesizedText: undefined }); + params.resolvedDelivery = makeResolvedDelivery({ + channel: "discord", + to: "channel-123", + }) as never; + params.summary = "Pablo Daily Summary\n- Non-Telegram fallback."; + params.outputText = "Pablo Daily Summary\n- Non-Telegram fallback."; + params.deliveryPayloadHasStructuredContent = true; + params.deliveryPayloads = [{ text: " " }] as never; + + const state = await dispatchCronDelivery(params); + + expect(deliverOutboundPayloads).toHaveBeenCalledTimes(1); + expectDeliveryCall(0, { + channel: "discord", + to: "channel-123", + payloads: [{ text: "Pablo Daily Summary\n- Non-Telegram fallback." }], + skipQueue: true, + }); + expect(state.deliveryAttempted).toBe(true); + expect(state.delivered).toBe(true); + }); + it("skips announce fallback after verified message-tool source delivery", async () => { const params = makeBaseParams({ synthesizedText: "Fallback cron summary." }); params.sourceDeliveryOutcome = { diff --git a/src/cron/isolated-agent/delivery-dispatch.ts b/src/cron/isolated-agent/delivery-dispatch.ts index f620ef78a9eb..f39f3dbe48be 100644 --- a/src/cron/isolated-agent/delivery-dispatch.ts +++ b/src/cron/isolated-agent/delivery-dispatch.ts @@ -455,6 +455,38 @@ function resolveCronAwarenessText(params: { normalizeOptionalString(params.synthesizedText)); } +function resolveDirectCronSummaryFallbackText(params: { + outputText?: string; + summary?: string; + synthesizedText?: string; +}): string | undefined { + return ( + normalizeOptionalString(params.outputText) ?? + normalizeOptionalString(params.summary) ?? + normalizeOptionalString(params.synthesizedText) + ); +} + +function shouldAttachDirectCronFallbackText(payload: ReplyPayload): boolean { + return ( + Boolean(payload.channelData) && + !hasReplyPayloadContent(payload, { trimText: true, hasChannelData: false }) + ); +} + +function resolveDirectCronFallbackSourceIndex( + payloads: ReplyPayload[], + fallbackText: string | undefined, +): number | undefined { + if (!fallbackText) { + return undefined; + } + const index = payloads.findLastIndex( + (payload) => normalizeOptionalString(payload.text) === fallbackText, + ); + return index >= 0 ? index : undefined; +} + function formatTargetCronDeliveryAwarenessText(text: string): string { return `A scheduled cron job delivered this message to this channel:\n${text}`; } @@ -1096,23 +1128,61 @@ export async function dispatchCronDelivery( delivery, }); try { - const rawPayloads = - deliveryPayloads.length > 0 - ? deliveryPayloads - : synthesizedText - ? [{ text: synthesizedText }] - : []; - const normalizedPayloads = rawPayloads - .map((p) => { - if (!p.text) { - return p; - } - const normalized = normalizeSilentReplyText(p.text); - return Object.assign({}, p, { - text: normalized.strippedTrailingSilentToken ? undefined : normalized.text, - }); - }) - .filter((p) => hasReplyPayloadContent(p, { trimText: true })); + const summaryFallbackText = resolveDirectCronSummaryFallbackText({ + outputText, + summary, + synthesizedText, + }); + const normalizedSummaryFallback = summaryFallbackText + ? normalizeSilentReplyText(summaryFallbackText) + : undefined; + const normalizedSummaryFallbackText = + normalizedSummaryFallback?.strippedTrailingSilentToken === true + ? undefined + : normalizedSummaryFallback?.text; + const normalizeDirectPayload = (payload: ReplyPayload): ReplyPayload => { + const normalized = payload.text ? normalizeSilentReplyText(payload.text) : undefined; + return normalized + ? { + ...payload, + text: normalized.strippedTrailingSilentToken ? undefined : normalized.text, + } + : payload; + }; + const normalizedDeliveryPayloads = deliveryPayloads + .map(normalizeDirectPayload) + .filter((payload) => hasReplyPayloadContent(payload, { trimText: true })); + const existingFallbackSourceIndex = resolveDirectCronFallbackSourceIndex( + normalizedDeliveryPayloads, + normalizedSummaryFallbackText, + ); + const needsFallbackSource = + Boolean(normalizedSummaryFallbackText) && + normalizedDeliveryPayloads.some(shouldAttachDirectCronFallbackText) && + existingFallbackSourceIndex === undefined; + const fallbackSourceIndex = needsFallbackSource ? 0 : existingFallbackSourceIndex; + const directPayloads = needsFallbackSource + ? [{ text: normalizedSummaryFallbackText }, ...normalizedDeliveryPayloads] + : normalizedDeliveryPayloads; + let normalizedPayloads: ReplyPayload[] = []; + for (const payload of directPayloads) { + normalizedPayloads.push( + shouldAttachDirectCronFallbackText(payload) && normalizedSummaryFallbackText + ? { + ...payload, + fallbackText: { + text: normalizedSummaryFallbackText, + ...(fallbackSourceIndex !== undefined + ? { replacesPayloadIndex: fallbackSourceIndex } + : {}), + }, + } + : payload, + ); + } + if (normalizedPayloads.length === 0 && normalizedSummaryFallbackText) { + normalizedPayloads = [{ text: normalizedSummaryFallbackText }]; + } if (normalizedPayloads.length === 0) { return await finishSilentReplyDelivery(); } diff --git a/src/infra/outbound/deliver.test.ts b/src/infra/outbound/deliver.test.ts index 66b94b1097d3..3d2cedd4a325 100644 --- a/src/infra/outbound/deliver.test.ts +++ b/src/infra/outbound/deliver.test.ts @@ -3281,6 +3281,54 @@ describe("deliverOutboundPayloads", () => { }); }); + it("passes ordered source indexes through adapter batch normalization", async () => { + const normalizePayloadBatch = vi.fn< + NonNullable + >(({ payloads }) => [ + { + ...payloads[0]?.payload, + channelData: { merged: payloads.map((entry) => entry.index) }, + }, + null, + ]); + const sendPayload = vi.fn().mockResolvedValue({ + channel: "matrix" as const, + messageId: "merged", + roomId: "!room", + }); + setActivePluginRegistry( + createTestRegistry([ + { + pluginId: "matrix", + source: "test", + plugin: createOutboundTestPlugin({ + id: "matrix", + outbound: { + deliveryMode: "direct", + normalizePayloadBatch, + sendText: vi.fn(), + sendPayload, + }, + }), + }, + ]), + ); + + await deliverOutboundPayloads({ + cfg: {}, + channel: "matrix", + to: "!room", + payloads: [{ text: "First" }, { text: "Second" }], + }); + + expect(normalizePayloadBatch).toHaveBeenCalledTimes(1); + expect(sendPayload).toHaveBeenCalledTimes(1); + expect(requireMockCallArg(sendPayload, "sendPayload").payload).toMatchObject({ + text: "First", + channelData: { merged: [0, 1] }, + }); + }); + it("strips internal runtime scaffolding copied into rendered and normalized nested payloads", async () => { const sendPayload = vi.fn().mockResolvedValue({ channel: "matrix" as const, diff --git a/src/infra/outbound/deliver.ts b/src/infra/outbound/deliver.ts index d400e7273c80..a2e1582853c4 100644 --- a/src/infra/outbound/deliver.ts +++ b/src/infra/outbound/deliver.ts @@ -171,6 +171,9 @@ type ChannelHandler = { supportsMedia: boolean; sanitizeText?: (payload: ReplyPayload) => string; normalizePayload?: (payload: ReplyPayload) => ReplyPayload | null; + normalizePayloadBatch?: ( + payloads: NormalizedPayloadForChannelDelivery[], + ) => NormalizedPayloadForChannelDelivery[]; sendTextOnlyErrorPayloads?: boolean; renderPresentation?: (payload: ReplyPayload) => Promise; presentationCapabilities?: ChannelOutboundAdapter["presentationCapabilities"]; @@ -471,6 +474,19 @@ function createPluginHandler( accountId: params.accountId, }) : undefined, + normalizePayloadBatch: outbound?.normalizePayloadBatch + ? (payloads) => { + const normalized = outbound.normalizePayloadBatch!({ + payloads, + cfg: params.cfg, + accountId: params.accountId, + }); + return payloads.flatMap((entry, index) => { + const payload = normalized[index]; + return payload ? [{ ...entry, payload }] : []; + }); + } + : undefined, sendTextOnlyErrorPayloads: outbound?.sendTextOnlyErrorPayloads === true, presentationCapabilities: outbound?.presentationCapabilities, renderPresentation: outbound?.renderPresentation @@ -947,7 +963,9 @@ function normalizePayloadsForChannelDelivery( normalizedPayloads.push({ index: entry.sourceIndex, payload: normalized }); } } - return normalizedPayloads; + return handler.normalizePayloadBatch + ? handler.normalizePayloadBatch(normalizedPayloads) + : normalizedPayloads; } function stripInternalRuntimeScaffoldingFromValue(value: unknown): unknown { diff --git a/src/plugin-sdk/reply-payload.ts b/src/plugin-sdk/reply-payload.ts index a3a9fd6f4a43..47b66137592d 100644 --- a/src/plugin-sdk/reply-payload.ts +++ b/src/plugin-sdk/reply-payload.ts @@ -26,6 +26,12 @@ export { export type OutboundReplyPayload = { /** Plain text reply body. */ text?: string; + /** Visible body a channel adapter may use when native structured content requires text. */ + fallbackText?: { + text: string; + /** Batch payload replaced when the adapter adopts this fallback body. */ + replacesPayloadIndex?: number; + }; /** Ordered media attachments for channels that can send multiple media items. */ mediaUrls?: string[]; /** Legacy single media attachment. */