diff --git a/extensions/telegram/src/bot-message-dispatch.test.ts b/extensions/telegram/src/bot-message-dispatch.test.ts index 8323ec595f75..f5460098f9e2 100644 --- a/extensions/telegram/src/bot-message-dispatch.test.ts +++ b/extensions/telegram/src/bot-message-dispatch.test.ts @@ -2129,6 +2129,111 @@ describe("dispatchTelegramMessage draft streaming", () => { expect(deliverReplies).not.toHaveBeenCalled(); }); + it("falls back to normal delivery before rotating a stale queued block preview", async () => { + const { answerDraftStream } = setupDraftStreams({ answerMessageId: 2001 }); + let firstBlockPreviewWentStale = false; + answerDraftStream.lastDeliveredText.mockImplementation(() => + firstBlockPreviewWentStale ? "stale draft still visible" : "", + ); + dispatchReplyWithBufferedBlockDispatcher.mockImplementation( + async ({ dispatcherOptions, replyOptions }) => { + const firstPayload = setReplyPayloadMetadata( + { text: "Site A shows X." }, + { assistantMessageIndex: 0 }, + ); + const secondPayload = setReplyPayloadMetadata( + { text: "Site B shows Y." }, + { assistantMessageIndex: 1 }, + ); + await replyOptions?.onBlockReplyQueued?.(firstPayload, { assistantMessageIndex: 0 }); + await dispatcherOptions.deliver(firstPayload, { kind: "block" }); + firstBlockPreviewWentStale = true; + await replyOptions?.onBlockReplyQueued?.(secondPayload, { assistantMessageIndex: 1 }); + await dispatcherOptions.deliver(secondPayload, { kind: "block" }); + return { queuedFinal: true }; + }, + ); + + await dispatchWithContext({ context: createContext() }); + + expect(answerDraftStream.update).toHaveBeenNthCalledWith(1, "Site A shows X."); + expect(answerDraftStream.update).toHaveBeenNthCalledWith(2, "Site A shows X."); + expect(answerDraftStream.update).toHaveBeenNthCalledWith(3, "Site B shows Y."); + expect(answerDraftStream.clear).toHaveBeenCalled(); + expect(deliverReplies).toHaveBeenCalledTimes(1); + const fallbackDelivery = mockCallArg(deliverReplies) as { + replies?: Array<{ text?: string }>; + transcriptMirror?: unknown; + }; + expect(fallbackDelivery.replies?.[0]?.text).toBe("Site A shows X."); + expect(fallbackDelivery.transcriptMirror).toBeUndefined(); + const clearOrder = answerDraftStream.clear.mock.invocationCallOrder[0]; + const fallbackDeliveryOrder = deliverReplies.mock.invocationCallOrder[0]; + const rotationOrder = answerDraftStream.forceNewMessage.mock.invocationCallOrder[0]; + const secondBlockUpdateOrder = answerDraftStream.update.mock.invocationCallOrder[2]; + expect(clearOrder).toBeLessThan(fallbackDeliveryOrder); + expect(fallbackDeliveryOrder).toBeLessThan(rotationOrder); + expect(rotationOrder).toBeLessThan(secondBlockUpdateOrder); + }); + + it("keeps stale block materialization tied to the streamed block before later media sends", async () => { + const { answerDraftStream } = setupDraftStreams({ answerMessageId: 2001 }); + let firstBlockPreviewWentStale = false; + answerDraftStream.lastDeliveredText.mockImplementation(() => + firstBlockPreviewWentStale ? "stale draft still visible" : "", + ); + dispatchReplyWithBufferedBlockDispatcher.mockImplementation( + async ({ dispatcherOptions, replyOptions }) => { + const firstPayload = setReplyPayloadMetadata( + { text: "Site A shows X." }, + { assistantMessageIndex: 0 }, + ); + const mediaPayload = setReplyPayloadMetadata( + { + text: "Chart attached", + mediaUrl: "https://example.test/chart.png", + }, + { assistantMessageIndex: 0 }, + ); + const nextPayload = setReplyPayloadMetadata( + { text: "Site B shows Y." }, + { assistantMessageIndex: 1 }, + ); + await replyOptions?.onBlockReplyQueued?.(firstPayload, { assistantMessageIndex: 0 }); + await dispatcherOptions.deliver(firstPayload, { kind: "block" }); + await replyOptions?.onBlockReplyQueued?.(mediaPayload, { assistantMessageIndex: 0 }); + await dispatcherOptions.deliver(mediaPayload, { kind: "block" }); + firstBlockPreviewWentStale = true; + await replyOptions?.onBlockReplyQueued?.(nextPayload, { assistantMessageIndex: 1 }); + await dispatcherOptions.deliver(nextPayload, { kind: "block" }); + return { queuedFinal: true }; + }, + ); + + await dispatchWithContext({ context: createContext() }); + + expect(answerDraftStream.update).toHaveBeenNthCalledWith(1, "Site A shows X."); + expect(answerDraftStream.update).toHaveBeenNthCalledWith(2, "Site A shows X."); + expect(answerDraftStream.update).toHaveBeenNthCalledWith(3, "Site B shows Y."); + expect(deliverReplies).toHaveBeenCalledTimes(2); + expectDeliveredReply( + 0, + { text: "Chart attached", mediaUrl: "https://example.test/chart.png" }, + 0, + ); + const fallbackDelivery = mockCallArg(deliverReplies, 1) as { + replies?: Array<{ text?: string; mediaUrl?: string; mediaUrls?: string[] }>; + }; + expect(fallbackDelivery.replies?.[0]).toEqual({ text: "Site A shows X." }); + const mediaDeliveryOrder = deliverReplies.mock.invocationCallOrder[0]; + const fallbackDeliveryOrder = deliverReplies.mock.invocationCallOrder[1]; + const rotationOrder = answerDraftStream.forceNewMessage.mock.invocationCallOrder[0]; + const nextBlockUpdateOrder = answerDraftStream.update.mock.invocationCallOrder[2]; + expect(mediaDeliveryOrder).toBeLessThan(fallbackDeliveryOrder); + expect(fallbackDeliveryOrder).toBeLessThan(rotationOrder); + expect(rotationOrder).toBeLessThan(nextBlockUpdateOrder); + }); + it("rotates queued block boundaries before async block delivery drains", async () => { const { answerDraftStream } = setupDraftStreams({ answerMessageId: 2001 }); dispatchReplyWithBufferedBlockDispatcher.mockImplementation( @@ -2158,6 +2263,16 @@ describe("dispatchTelegramMessage draft streaming", () => { const secondBlockUpdateOrder = answerDraftStream.update.mock.invocationCallOrder[1]; expect(rotationOrder).toBeLessThan(secondBlockUpdateOrder); expect(deliverReplies).not.toHaveBeenCalled(); + expectRecordFields(mockCallArg(emitInternalMessageSentHook), { + content: "Site A shows X.", + messageId: 2001, + }); + expectRecordFields(mockCallArg(recordOutboundMessageForPromptContext), { + chatId: "123", + messageId: 2001, + text: "Site A shows X.", + messageThreadId: 777, + }); }); it("skips canceled queued block rotations when later delivery drains", async () => { @@ -2175,7 +2290,7 @@ describe("dispatchTelegramMessage draft streaming", () => { ); await dispatcherOptions.deliver( setReplyPayloadMetadata({ text: "Repeated block." }, { assistantMessageIndex: 1 }), - { kind: "block" }, + { kind: "block", assistantMessageIndex: 1 } as { kind: "block" }, ); return { queuedFinal: true }; }, @@ -2286,6 +2401,60 @@ describe("dispatchTelegramMessage draft streaming", () => { expect(deliverReplies).not.toHaveBeenCalled(); }); + it("expires rewritten unindexed queued block rotations after cancellation", async () => { + const { answerDraftStream } = setupDraftStreams({ answerMessageId: 2001 }); + dispatchReplyWithBufferedBlockDispatcher.mockImplementation( + async ({ dispatcherOptions, replyOptions }) => { + await replyOptions?.onPartialReply?.({ text: "Existing preview" }); + await replyOptions?.onBlockReplyQueued?.({ text: "Original block text" }); + await replyOptions?.onAssistantMessageStart?.(); + await dispatcherOptions.onBeforeDeliverCancelled?.( + { text: "PFX Original block text" }, + { kind: "block" }, + ); + await replyOptions?.onPartialReply?.({ text: "Second preview" }); + return { queuedFinal: true }; + }, + ); + + await dispatchWithContext({ context: createContext() }); + + expect(answerDraftStream.update).toHaveBeenNthCalledWith(1, "Existing preview"); + expect(answerDraftStream.update).toHaveBeenNthCalledWith(2, "Second preview"); + expect(answerDraftStream.forceNewMessage).toHaveBeenCalledTimes(1); + const rotationOrder = answerDraftStream.forceNewMessage.mock.invocationCallOrder[0]; + const secondPreviewUpdateOrder = answerDraftStream.update.mock.invocationCallOrder[1]; + expect(rotationOrder).toBeLessThan(secondPreviewUpdateOrder); + expect(deliverReplies).not.toHaveBeenCalled(); + }); + + it("expires rewritten unindexed queued block rotations after skip", async () => { + const { answerDraftStream } = setupDraftStreams({ answerMessageId: 2001 }); + dispatchReplyWithBufferedBlockDispatcher.mockImplementation( + async ({ dispatcherOptions, replyOptions }) => { + await replyOptions?.onPartialReply?.({ text: "Existing preview" }); + await replyOptions?.onBlockReplyQueued?.({ text: "Original block text" }); + await replyOptions?.onAssistantMessageStart?.(); + dispatcherOptions.onSkip?.( + { text: "PFX Original block text" }, + { kind: "block", reason: "silent" }, + ); + await replyOptions?.onPartialReply?.({ text: "Second preview" }); + return { queuedFinal: true }; + }, + ); + + await dispatchWithContext({ context: createContext() }); + + expect(answerDraftStream.update).toHaveBeenNthCalledWith(1, "Existing preview"); + expect(answerDraftStream.update).toHaveBeenNthCalledWith(2, "Second preview"); + expect(answerDraftStream.forceNewMessage).toHaveBeenCalledTimes(1); + const rotationOrder = answerDraftStream.forceNewMessage.mock.invocationCallOrder[0]; + const secondPreviewUpdateOrder = answerDraftStream.update.mock.invocationCallOrder[1]; + expect(rotationOrder).toBeLessThan(secondPreviewUpdateOrder); + expect(deliverReplies).not.toHaveBeenCalled(); + }); + it("expires canceled queued block rotations before later partial previews", async () => { const { answerDraftStream } = setupDraftStreams({ answerMessageId: 2001 }); dispatchReplyWithBufferedBlockDispatcher.mockImplementation( @@ -2314,6 +2483,36 @@ describe("dispatchTelegramMessage draft streaming", () => { expect(deliverReplies).not.toHaveBeenCalled(); }); + it("serializes canceled block cleanup behind queued assistant-boundary events", async () => { + const { answerDraftStream } = setupDraftStreams({ answerMessageId: 2001 }); + dispatchReplyWithBufferedBlockDispatcher.mockImplementation( + async ({ dispatcherOptions, replyOptions }) => { + const payload = setReplyPayloadMetadata( + { text: "Site A final" }, + { assistantMessageIndex: 0 }, + ); + void replyOptions?.onPartialReply?.({ text: "Site A partial" }); + void replyOptions?.onBlockReplyQueued?.(payload, { assistantMessageIndex: 0 }); + void replyOptions?.onAssistantMessageStart?.(); + await dispatcherOptions.onBeforeDeliverCancelled?.(payload, { kind: "block" }); + await replyOptions?.onPartialReply?.({ text: "Site B partial" }); + return { queuedFinal: true }; + }, + ); + + await dispatchWithContext({ context: createContext() }); + + expect(answerDraftStream.update).toHaveBeenNthCalledWith(1, "Site A partial"); + expect(answerDraftStream.update).toHaveBeenNthCalledWith(2, "Site B partial"); + expect(answerDraftStream.forceNewMessage).toHaveBeenCalledTimes(1); + const firstPartialUpdateOrder = answerDraftStream.update.mock.invocationCallOrder[0]; + const rotationOrder = answerDraftStream.forceNewMessage.mock.invocationCallOrder[0]; + const secondPartialUpdateOrder = answerDraftStream.update.mock.invocationCallOrder[1]; + expect(firstPartialUpdateOrder).toBeLessThan(rotationOrder); + expect(rotationOrder).toBeLessThan(secondPartialUpdateOrder); + expect(deliverReplies).not.toHaveBeenCalled(); + }); + it("preserves boundary rotation after a queued prior block is canceled", async () => { const { answerDraftStream } = setupDraftStreams({ answerMessageId: 2001 }); dispatchReplyWithBufferedBlockDispatcher.mockImplementation( @@ -2541,7 +2740,7 @@ describe("dispatchTelegramMessage draft streaming", () => { { mediaUrls: ["https://example.test/site-a.png"] }, { assistantMessageIndex: 0 }, ), - { kind: "block" }, + { kind: "block", assistantMessageIndex: 0 } as { kind: "block" }, ); await replyOptions?.onPartialReply?.({ text: "Site B partial" }); return { queuedFinal: true }; diff --git a/extensions/telegram/src/bot-message-dispatch.ts b/extensions/telegram/src/bot-message-dispatch.ts index e65cf355e915..3a2410ece2e5 100644 --- a/extensions/telegram/src/bot-message-dispatch.ts +++ b/extensions/telegram/src/bot-message-dispatch.ts @@ -19,8 +19,8 @@ import { import { CURRENT_MESSAGE_MARKER } from "openclaw/plugin-sdk/channel-mention-gating"; import { createChannelMessageReplyPipeline, - createOutboundPayloadPlan, createPreviewMessageReceipt, + createOutboundPayloadPlan, deriveDurableFinalDeliveryRequirements, projectOutboundPayloadPlanForDelivery, } from "openclaw/plugin-sdk/channel-outbound"; @@ -969,6 +969,10 @@ export const dispatchTelegramMessage = async ({ let lastAnswerPartialText = ""; let activeAnswerDraftIsToolProgressOnly = false; let activeAnswerBlockAssistantMessageIndex: number | undefined; + let lastAnswerBlockPayload: ReplyPayload | undefined; + let lastAnswerBlockText: string | undefined; + let lastAnswerBlockButtons: TelegramInlineButtons | undefined; + let materializeAnswerLaneBeforeRotation: (() => Promise) | undefined; type QueuedAnswerBlockRotation = { assistantMessageIndex?: number; text?: string; @@ -990,7 +994,7 @@ export const dispatchTelegramMessage = async ({ return; } if (answerLane.hasStreamedMessage) { - await rotateLaneForNewMessage(answerLane); + await rotateAnswerLaneForNewMessage(); } activeAnswerDraftIsToolProgressOnly = true; } @@ -1117,6 +1121,9 @@ export const dispatchTelegramMessage = async ({ if (lane === answerLane) { resetAnswerToolProgressDraft(); pendingAnswerBlockAssistantMessageIndex = undefined; + lastAnswerBlockPayload = undefined; + lastAnswerBlockText = undefined; + lastAnswerBlockButtons = undefined; } }; const rotateLaneForNewMessage = async (lane: DraftLaneState) => { @@ -1128,6 +1135,12 @@ export const dispatchTelegramMessage = async ({ lane.stream?.forceNewMessage(); resetDraftLaneState(lane); }; + const rotateAnswerLaneForNewMessage = async () => { + if (materializeAnswerLaneBeforeRotation) { + await materializeAnswerLaneBeforeRotation(); + } + await rotateLaneForNewMessage(answerLane); + }; const rotateAnswerLaneAfterToolProgress = async () => { nativeToolProgressDraft?.stop(); if (!activeAnswerDraftIsToolProgressOnly) { @@ -1148,7 +1161,7 @@ export const dispatchTelegramMessage = async ({ if (!answerLane.hasStreamedMessage || activeAnswerDraftIsToolProgressOnly) { return false; } - await rotateLaneForNewMessage(answerLane); + await rotateAnswerLaneForNewMessage(); return true; }; const prepareAnswerLaneForText = async (): Promise => { @@ -1216,23 +1229,34 @@ export const dispatchTelegramMessage = async ({ queuedAnswerBlockAssistantMessageIndex = entry.assistantMessageIndex; } }; - const queuedAnswerBlockRotationMatchesPayload = ( + const queuedAnswerBlockRotationTextMatchesPayload = ( entry: QueuedAnswerBlockRotation, payload: ReplyPayload, ) => { - return ( - entry.assistantMessageIndex === undefined || - (entry.text !== undefined && payload.text !== undefined && entry.text === payload.text) - ); + return entry.text !== undefined && payload.text !== undefined && entry.text === payload.text; }; - const takeQueuedAnswerBlockRotation = (payload: ReplyPayload): boolean => { - const matchIndex = queuedAnswerBlockRotations.findIndex((entry) => - queuedAnswerBlockRotationMatchesPayload(entry, payload), - ); - if (matchIndex < 0) { + const queuedAnswerBlockRotationMatchesDelivery = ( + entry: QueuedAnswerBlockRotation, + payload: ReplyPayload, + assistantMessageIndex?: number, + ) => { + if (assistantMessageIndex !== undefined && entry.assistantMessageIndex !== undefined) { + return assistantMessageIndex === entry.assistantMessageIndex; + } + return queuedAnswerBlockRotationTextMatchesPayload(entry, payload); + }; + const takeQueuedAnswerBlockRotation = ( + payload: ReplyPayload, + assistantMessageIndex?: number, + ): boolean => { + if (queuedAnswerBlockRotations.length === 0) { return false; } - const matchedEntries = queuedAnswerBlockRotations.splice(0, matchIndex + 1); + const matchIndex = queuedAnswerBlockRotations.findIndex((entry) => + queuedAnswerBlockRotationMatchesDelivery(entry, payload, assistantMessageIndex), + ); + const consumeIndex = Math.max(matchIndex, 0); + const matchedEntries = queuedAnswerBlockRotations.splice(0, consumeIndex + 1); const matchedEntry = matchedEntries.at(-1); const shouldRotateBeforeDelivery = matchedEntry?.shouldRotateBeforeDelivery ?? false; if (matchedEntry?.assistantMessageIndex !== undefined) { @@ -1242,10 +1266,18 @@ export const dispatchTelegramMessage = async ({ recomputeQueuedAnswerBlockRotations(); return shouldRotateBeforeDelivery; }; - const dropQueuedAnswerBlockRotation = (payload: ReplyPayload) => { - const matchIndex = queuedAnswerBlockRotations.findIndex((entry) => - queuedAnswerBlockRotationMatchesPayload(entry, payload), + const dropQueuedAnswerBlockRotation = ( + payload: ReplyPayload, + assistantMessageIndex?: number, + ) => { + let matchIndex = queuedAnswerBlockRotations.findIndex((entry) => + queuedAnswerBlockRotationMatchesDelivery(entry, payload, assistantMessageIndex), ); + if (matchIndex < 0 && assistantMessageIndex === undefined) { + matchIndex = queuedAnswerBlockRotations.findIndex( + (entry) => entry.assistantMessageIndex === undefined, + ); + } if (matchIndex >= 0) { const matchedEntry = queuedAnswerBlockRotations[matchIndex]; queuedAnswerBlockRotations.splice(matchIndex, 1); @@ -1261,6 +1293,10 @@ export const dispatchTelegramMessage = async ({ recomputeQueuedAnswerBlockRotations(); } }; + const getReplyDispatchAssistantMessageIndex = (info: object): number | undefined => { + const value = (info as { assistantMessageIndex?: unknown }).assistantMessageIndex; + return typeof value === "number" ? value : undefined; + }; const updateDraftFromPartial = (lane: DraftLaneState, update: DraftPartialTextUpdate) => { const laneStream = lane.stream; if (!laneStream || !update.text) { @@ -1713,6 +1749,60 @@ export const dispatchTelegramMessage = async ({ deliveryState.markDelivered(); }, }); + materializeAnswerLaneBeforeRotation = async () => { + if ( + !lastAnswerBlockPayload || + !answerLane.stream || + !answerLane.hasStreamedMessage || + answerLane.finalized || + activeAnswerDraftIsToolProgressOnly + ) { + return false; + } + const text = answerLane.lastPartialText || lastAnswerPartialText || lastAnswerBlockText; + if (!text?.trim()) { + return false; + } + // A block skipped by the duplicate-draft dedup was never rendered to its + // own draft update. Force the full delivery path (not the no-op finalize + // fast path) so the preserved intermediate block is materialized as a + // visible draft before the lane rotates for the next message. + const wasSkippedDuplicate = skippedDuplicateAnswerBlockDraftDelivery; + skippedDuplicateAnswerBlockDraftDelivery = false; + const deliveredText = answerLane.stream.lastDeliveredText?.(); + const messageId = answerLane.stream.messageId(); + if ( + !lastAnswerBlockButtons && + !wasSkippedDuplicate && + deliveredText === text.trimEnd() && + typeof messageId === "number" + ) { + await answerLane.stream.stop(); + answerLane.finalized = true; + deliveryState.markDelivered(); + await emitPreviewFinalizedHook({ + kind: "preview-finalized", + delivery: { + content: text, + promptContextContent: deliveredText, + messageId, + receipt: createPreviewMessageReceipt({ id: messageId }), + }, + }); + return true; + } + const result = await deliverLaneText({ + laneName: "answer", + text, + payload: lastAnswerBlockPayload, + infoKind: "block", + buttons: lastAnswerBlockButtons, + finalizePreview: true, + durable: false, + }); + await emitPreviewFinalizedHook(result); + return result.kind !== "skipped"; + }; const deliverProgressModeFinalAnswer = async ( payload: ReplyPayload, text: string, @@ -1810,8 +1900,14 @@ export const dispatchTelegramMessage = async ({ beforeDeliver: async (payload) => payload, onBeforeDeliverCancelled: (payload, info) => { if (info.kind === "block") { - dropQueuedAnswerBlockRotation(payload); + return enqueueDraftLaneEvent(async () => { + dropQueuedAnswerBlockRotation( + payload, + getReplyDispatchAssistantMessageIndex(info), + ); + }); } + return undefined; }, deliver: async (payload, info) => { if (isDispatchSuperseded()) { @@ -1925,7 +2021,10 @@ export const dispatchTelegramMessage = async ({ let blockDelivered = false; const hasAnswerSegment = segments.some((segment) => segment.lane === "answer"); if (info.kind === "block" && !hasAnswerSegment) { - dropQueuedAnswerBlockRotation(effectivePayload); + dropQueuedAnswerBlockRotation( + effectivePayload, + getReplyDispatchAssistantMessageIndex(info), + ); } for (const segment of segments) { if ( @@ -1969,6 +2068,15 @@ export const dispatchTelegramMessage = async ({ await prepareAnswerLaneForToolProgress(); } + const ownedByQueuedAnswerBlockRotation = + queuedAnswerBlockRotations.some((entry) => + queuedAnswerBlockRotationMatchesDelivery( + entry, + effectivePayload, + getReplyDispatchAssistantMessageIndex(info), + ), + ); + const skipTextOnlyBlock = streamMode === "partial" && info.kind === "block" && @@ -1978,20 +2086,35 @@ export const dispatchTelegramMessage = async ({ telegramButtons === undefined && answerLane.hasStreamedMessage && !activeAnswerDraftIsToolProgressOnly && + !ownedByQueuedAnswerBlockRotation && segment.update.text.trimEnd() === answerLane.lastPartialText.trimEnd(); if (skipTextOnlyBlock) { + // Defer the duplicate block: do not emit a redundant draft + // update now. Record it so that if a later rotation (tool + // progress / next assistant message) follows, the skipped + // block is materialized first instead of being lost, and so + // that the dispatch-end finalize can commit it when nothing + // else follows. Re-enable progress-draft state so a + // following tool-progress step can still rotate the lane. skippedDuplicateAnswerBlockDraftDelivery = true; + lastAnswerBlockPayload = effectivePayload; + lastAnswerBlockText = segment.update.text; + lastAnswerBlockButtons = telegramButtons; + resetAnswerToolProgressDraft(); + resetProgressDraftState(); blockDelivered = true; continue; } if (segment.lane === "answer" && info.kind === "block") { const preparedAnswerLane = await prepareAnswerLaneForText(); - const shouldRotateQueuedBlock = - takeQueuedAnswerBlockRotation(effectivePayload); + const shouldRotateQueuedBlock = takeQueuedAnswerBlockRotation( + effectivePayload, + getReplyDispatchAssistantMessageIndex(info), + ); if (shouldRotateQueuedBlock && !preparedAnswerLane) { - await rotateLaneForNewMessage(answerLane); + await rotateAnswerLaneForNewMessage(); rotateAnswerLaneWhenQueuedBlocksSettle = false; } resetAnswerToolProgressDraft(); @@ -2014,6 +2137,17 @@ export const dispatchTelegramMessage = async ({ if (segment.lane === "answer" && result.kind === "preview-finalized") { await emitPreviewFinalizedHook(result); } + if ( + segment.lane === "answer" && + info.kind === "block" && + (result.kind === "preview-updated" || + result.kind === "preview-finalized" || + result.kind === "preview-retained") + ) { + lastAnswerBlockPayload = effectivePayload; + lastAnswerBlockText = segment.update.text; + lastAnswerBlockButtons = telegramButtons; + } blockDelivered = blockDelivered || result.kind !== "skipped"; if (segment.lane === "reasoning") { if (result.kind !== "skipped") { @@ -2086,7 +2220,10 @@ export const dispatchTelegramMessage = async ({ onSkip: (payload, info) => { if (info.kind === "block") { void enqueueDraftLaneEvent(async () => { - dropQueuedAnswerBlockRotation(payload); + dropQueuedAnswerBlockRotation( + payload, + getReplyDispatchAssistantMessageIndex(info), + ); }); } if (payload.isError === true) { diff --git a/extensions/telegram/src/lane-delivery-text-deliverer.ts b/extensions/telegram/src/lane-delivery-text-deliverer.ts index c8bf417be6c2..65b667399626 100644 --- a/extensions/telegram/src/lane-delivery-text-deliverer.ts +++ b/extensions/telegram/src/lane-delivery-text-deliverer.ts @@ -78,6 +78,8 @@ type DeliverLaneTextParams = { payload: ReplyPayload; infoKind: string; buttons?: TelegramInlineButtons; + finalizePreview?: boolean; + durable?: boolean; }; function result( @@ -286,7 +288,7 @@ export function createLaneTextDeliverer(params: CreateLaneTextDelivererParams) { lane: DraftLaneState, text: string, payload: ReplyPayload, - isFinal: boolean, + useFinalTextRecovery: boolean, finalizePreview: boolean, buttons?: TelegramInlineButtons, ): Promise => { @@ -315,7 +317,7 @@ export function createLaneTextDeliverer(params: CreateLaneTextDelivererParams) { const finalText = activeFullText.trimEnd(); const deliveredStreamTextBeforeUpdate = stream.lastDeliveredText?.(); const deliveredPrefixBeforeUpdate = - isFinal && + useFinalTextRecovery && deliveredStreamTextBeforeUpdate !== undefined && isDeliveredPrefix({ deliveredText: deliveredStreamTextBeforeUpdate, @@ -362,7 +364,7 @@ export function createLaneTextDeliverer(params: CreateLaneTextDelivererParams) { }; const candidateTexts = [stream.lastDeliveredText?.(), lane.lastPartialText]; - if (isFinal && remainingChunks.length === 0 && isPotentialTruncatedFinal(activeFullText)) { + if (useFinalTextRecovery && remainingChunks.length === 0 && isPotentialTruncatedFinal(activeFullText)) { const resolvedFullCandidate = await params.resolveFinalTextCandidate?.({ finalText: text, laneName, @@ -377,7 +379,7 @@ export function createLaneTextDeliverer(params: CreateLaneTextDelivererParams) { } const retainedPreview = - isFinal && remainingChunks.length === 0 && isPotentialTruncatedFinal(activeFullText) + useFinalTextRecovery && remainingChunks.length === 0 && isPotentialTruncatedFinal(activeFullText) ? selectLongerFinalText({ finalText: activeFullText, candidateTexts, @@ -441,7 +443,7 @@ export function createLaneTextDeliverer(params: CreateLaneTextDelivererParams) { } else { await params.flushDraftLane(lane); } - const activeChunkIndexAfterStop = isFinal ? clampActiveChunkIndex() : activeChunkIndex; + const activeChunkIndexAfterStop = useFinalTextRecovery ? clampActiveChunkIndex() : activeChunkIndex; const activeChunkAfterStop = chunks[activeChunkIndexAfterStop] ?? activeChunk; const remainingChunksAfterStop = chunks.slice(activeChunkIndexAfterStop + 1); @@ -470,6 +472,7 @@ export function createLaneTextDeliverer(params: CreateLaneTextDelivererParams) { !retainedActiveChunkAfterStop ) { if ( + useFinalTextRecovery && isDeliveredPrefix({ deliveredText: deliveredStreamTextAfterStop, finalText }) && deliveredStreamTextAfterStop.length > activeChunkTextAfterStop.length ) { @@ -523,20 +526,23 @@ export function createLaneTextDeliverer(params: CreateLaneTextDelivererParams) { payload, infoKind, buttons, + finalizePreview: requestedFinalizePreview, + durable: requestedDurable, }: DeliverLaneTextParams): Promise => { const lane = params.lanes[laneName]; const reply = resolveSendableOutboundReplyParts(payload, { text }); - const isFinal = infoKind === "final"; - const finalizePreview = isFinal; + const isDurableFinal = infoKind === "final"; + const finalizePreview = requestedFinalizePreview ?? isDurableFinal; + const durable = requestedDurable ?? isDurableFinal; const streamed = !reply.hasMedia - ? await streamText(laneName, lane, text, payload, isFinal, finalizePreview, buttons) + ? await streamText(laneName, lane, text, payload, isDurableFinal, finalizePreview, buttons) : undefined; if (streamed) { return streamed; } if ( - isFinal && + finalizePreview && reply.hasMedia && lane.stream && lane.hasStreamedMessage && @@ -548,7 +554,7 @@ export function createLaneTextDeliverer(params: CreateLaneTextDelivererParams) { lane, text, textOnlyPayload(payload), - true, + isDurableFinal, true, buttons, ); @@ -564,7 +570,7 @@ export function createLaneTextDeliverer(params: CreateLaneTextDelivererParams) { fallbackButtons: stripButtons ? undefined : buttons, }), { - durable: true, + durable, }, ); return finalizedPreview; @@ -576,7 +582,7 @@ export function createLaneTextDeliverer(params: CreateLaneTextDelivererParams) { } const delivered = await params.sendPayload(params.applyTextToPayload(payload, text), { - durable: isFinal, + durable, }); if (delivered && finalizePreview) { lane.finalized = true; diff --git a/extensions/telegram/src/lane-delivery.test.ts b/extensions/telegram/src/lane-delivery.test.ts index 003ed34f2a08..8684b418b710 100644 --- a/extensions/telegram/src/lane-delivery.test.ts +++ b/extensions/telegram/src/lane-delivery.test.ts @@ -179,6 +179,65 @@ describe("createLaneTextDeliverer", () => { expect(harness.lanes.answer.finalized).toBe(true); }); + it("keeps media fallback non-durable when materializing an intermediate preview", async () => { + const harness = createHarness({ answerMessageId: 999 }); + harness.lanes.answer.hasStreamedMessage = true; + + const result = await harness.deliverLaneText({ + laneName: "answer", + text: "visible block", + payload: { text: "visible block", mediaUrls: ["file:///site-a.png"] }, + infoKind: "block", + finalizePreview: true, + durable: false, + }); + + const delivery = expectPreviewFinalized(result); + expect(delivery.content).toBe("visible block"); + expect(harness.sendPayload).toHaveBeenCalledWith( + { mediaUrls: ["file:///site-a.png"] }, + { durable: false }, + ); + expect(harness.lanes.answer.finalized).toBe(true); + }); + + it("does not use final transcript recovery when materializing an intermediate block preview", async () => { + const previousBlock = + "Here is the complete block preview with enough stable prefix text before the ellipsis..."; + const nextAssistantBlock = + "Here is the complete block preview with enough stable prefix text before the ellipsis and later assistant continuation text."; + const answer = createTestDraftStream({ messageId: 999 }); + answer.lastDeliveredText.mockReturnValue(nextAssistantBlock); + const harness = createHarness({ + answerStream: answer, + resolveFinalTextCandidate: () => nextAssistantBlock, + }); + harness.lanes.answer.lastPartialText = previousBlock; + harness.lanes.answer.hasStreamedMessage = true; + + const result = await harness.deliverLaneText({ + laneName: "answer", + text: previousBlock, + payload: { text: previousBlock }, + infoKind: "block", + finalizePreview: true, + durable: false, + }); + + expect(result.kind).toBe("sent"); + expect(answer.update).toHaveBeenCalledWith(previousBlock); + expect(answer.update).not.toHaveBeenCalledWith(nextAssistantBlock); + expect(harness.clearDraftLane).toHaveBeenCalledTimes(1); + expect(harness.sendPayload).toHaveBeenCalledWith( + { text: previousBlock }, + { durable: false }, + ); + expect(harness.sendPayload).not.toHaveBeenCalledWith( + { text: nextAssistantBlock }, + expect.anything(), + ); + }); + it("keeps block delivery in the draft lane when delivered text is stale", async () => { const answer = createTestDraftStream({ messageId: 999 }); answer.lastDeliveredText.mockReturnValue("working"); diff --git a/src/auto-reply/reply/before-deliver.test.ts b/src/auto-reply/reply/before-deliver.test.ts index f403e2dae65a..d18cd726922f 100644 --- a/src/auto-reply/reply/before-deliver.test.ts +++ b/src/auto-reply/reply/before-deliver.test.ts @@ -60,6 +60,62 @@ describe("beforeDeliver in reply dispatcher", () => { expect(dispatcher.getCancelledCounts?.()).toEqual({ tool: 0, block: 0, final: 1 }); }); + it("notifies cancellation when beforeDeliver throws before delivery", async () => { + const delivered: string[] = []; + const cancelled: Array<{ + assistantMessageIndex?: number; + kind: string; + text: string; + }> = []; + const errors: Array<{ + assistantMessageIndex?: number; + kind: string; + message: string; + }> = []; + + const dispatcher = createReplyDispatcher({ + deliver: async (payload) => { + delivered.push(payload.text ?? ""); + }, + onBeforeDeliverCancelled: (payload, info) => { + cancelled.push({ + assistantMessageIndex: (info as { assistantMessageIndex?: number }) + .assistantMessageIndex, + kind: info.kind, + text: payload.text ?? "", + }); + }, + onError: (err, info) => { + errors.push({ + assistantMessageIndex: (info as { assistantMessageIndex?: number }) + .assistantMessageIndex, + kind: info.kind, + message: err instanceof Error ? err.message : String(err), + }); + }, + beforeDeliver: async () => { + throw new Error("pre-delivery failed"); + }, + }); + + dispatcher.sendBlockReply( + setReplyPayloadMetadata({ text: "blocked block" }, { assistantMessageIndex: 9 }), + ); + dispatcher.markComplete(); + await dispatcher.waitForIdle(); + + expect(delivered).toEqual([]); + expect(cancelled).toEqual([ + { assistantMessageIndex: 9, kind: "block", text: "blocked block" }, + ]); + expect(errors).toEqual([ + { assistantMessageIndex: 9, kind: "block", message: "pre-delivery failed" }, + ]); + expect(dispatcher.getQueuedCounts()).toEqual({ tool: 0, block: 1, final: 0 }); + expect(dispatcher.getCancelledCounts?.()).toEqual({ tool: 0, block: 0, final: 0 }); + expect(dispatcher.getFailedCounts?.()).toEqual({ tool: 0, block: 1, final: 0 }); + }); + it("allows modifying payload in beforeDeliver", async () => { const delivered: string[] = []; @@ -84,10 +140,14 @@ describe("beforeDeliver in reply dispatcher", () => { it("preserves payload metadata through beforeDeliver rewrites", async () => { let deliveredMetadata: unknown; + let deliveredAssistantMessageIndex: unknown; const dispatcher = createReplyDispatcher({ - deliver: async (payload) => { + deliver: async (payload, info) => { deliveredMetadata = getReplyPayloadMetadata(payload); + deliveredAssistantMessageIndex = ( + info as { assistantMessageIndex?: unknown } + ).assistantMessageIndex; }, beforeDeliver: async () => ({ text: "rewritten" }), }); @@ -99,6 +159,7 @@ describe("beforeDeliver in reply dispatcher", () => { await dispatcher.waitForIdle(); expect(deliveredMetadata).toMatchObject({ assistantMessageIndex: 12 }); + expect(deliveredAssistantMessageIndex).toBe(12); }); it("delivers normally without beforeDeliver", async () => { diff --git a/src/auto-reply/reply/block-reply-coalescer.ts b/src/auto-reply/reply/block-reply-coalescer.ts index 6cae44eb4e64..992676ab9ce6 100644 --- a/src/auto-reply/reply/block-reply-coalescer.ts +++ b/src/auto-reply/reply/block-reply-coalescer.ts @@ -1,6 +1,6 @@ // Coalesces buffered block-streaming payloads into sendable reply parts. import { resolveSendableOutboundReplyParts } from "openclaw/plugin-sdk/reply-payload"; -import { isReplyPayloadStatusNotice } from "../reply-payload.js"; +import { copyReplyPayloadMetadata, isReplyPayloadStatusNotice } from "../reply-payload.js"; import type { ReplyPayload } from "../types.js"; import type { BlockStreamingCoalescing } from "./block-streaming.js"; @@ -32,6 +32,7 @@ export function createBlockReplyCoalescer(params: { let bufferIsCompactionNotice: ReplyPayload["isCompactionNotice"]; let bufferIsFallbackNotice: ReplyPayload["isFallbackNotice"]; let bufferIsStatusNotice: ReplyPayload["isStatusNotice"]; + let bufferMetadataSource: ReplyPayload | undefined; let idleTimer: NodeJS.Timeout | undefined; const clearIdleTimer = () => { @@ -50,6 +51,17 @@ export function createBlockReplyCoalescer(params: { bufferIsCompactionNotice = undefined; bufferIsFallbackNotice = undefined; bufferIsStatusNotice = undefined; + bufferMetadataSource = undefined; + }; + + const startBufferFromPayload = (payload: ReplyPayload) => { + bufferReplyToId = payload.replyToId; + bufferAudioAsVoice = payload.audioAsVoice; + bufferIsReasoning = payload.isReasoning; + bufferIsCompactionNotice = payload.isCompactionNotice; + bufferIsFallbackNotice = payload.isFallbackNotice; + bufferIsStatusNotice = payload.isStatusNotice; + bufferMetadataSource = payload; }; const scheduleIdleFlush = () => { @@ -84,8 +96,12 @@ export function createBlockReplyCoalescer(params: { isFallbackNotice: bufferIsFallbackNotice, isStatusNotice: bufferIsStatusNotice, }; + const metadataSource = bufferMetadataSource; + const payloadWithMetadata = metadataSource + ? copyReplyPayloadMetadata(metadataSource, payload) + : payload; resetBuffer(); - await onFlush(payload); + await onFlush(payloadWithMetadata); }; const canMergeBufferedTextWithMedia = (payload: ReplyPayload) => @@ -111,8 +127,11 @@ export function createBlockReplyCoalescer(params: { text: mergedText, replyToId: payload.replyToId ?? bufferReplyToId, }; + const metadataMergedPayload = bufferMetadataSource + ? copyReplyPayloadMetadata(bufferMetadataSource, mergedPayload) + : mergedPayload; resetBuffer(); - return mergedPayload; + return copyReplyPayloadMetadata(payload, metadataMergedPayload); }; const enqueue = (payload: ReplyPayload) => { @@ -142,12 +161,7 @@ export function createBlockReplyCoalescer(params: { if (bufferText) { void flush({ force: true }); } - bufferReplyToId = payload.replyToId; - bufferAudioAsVoice = payload.audioAsVoice; - bufferIsReasoning = payload.isReasoning; - bufferIsCompactionNotice = payload.isCompactionNotice; - bufferIsFallbackNotice = payload.isFallbackNotice; - bufferIsStatusNotice = payload.isStatusNotice; + startBufferFromPayload(payload); bufferText = text; void flush({ force: true }); return; @@ -177,24 +191,14 @@ export function createBlockReplyCoalescer(params: { } if (!bufferText) { - bufferReplyToId = payload.replyToId; - bufferAudioAsVoice = payload.audioAsVoice; - bufferIsReasoning = payload.isReasoning; - bufferIsCompactionNotice = payload.isCompactionNotice; - bufferIsFallbackNotice = payload.isFallbackNotice; - bufferIsStatusNotice = payload.isStatusNotice; + startBufferFromPayload(payload); } const nextText = bufferText ? `${bufferText}${joiner}${text}` : text; if (nextText.length > maxChars) { if (bufferText) { void flush({ force: true }); - bufferReplyToId = payload.replyToId; - bufferAudioAsVoice = payload.audioAsVoice; - bufferIsReasoning = payload.isReasoning; - bufferIsCompactionNotice = payload.isCompactionNotice; - bufferIsFallbackNotice = payload.isFallbackNotice; - bufferIsStatusNotice = payload.isStatusNotice; + startBufferFromPayload(payload); if (text.length >= maxChars) { void onFlush(payload); return; diff --git a/src/auto-reply/reply/block-reply-pipeline.test.ts b/src/auto-reply/reply/block-reply-pipeline.test.ts index 3198406257bc..20037178366a 100644 --- a/src/auto-reply/reply/block-reply-pipeline.test.ts +++ b/src/auto-reply/reply/block-reply-pipeline.test.ts @@ -1,6 +1,6 @@ /** Tests block reply pipeline buffering, dedupe, and final flush behavior. */ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -import { setReplyPayloadMetadata } from "../reply-payload.js"; +import { getReplyPayloadMetadata, setReplyPayloadMetadata } from "../reply-payload.js"; import { createBlockReplyContentKey, createBlockReplyPayloadKey, @@ -289,6 +289,31 @@ describe("createBlockReplyPipeline dedup with threading", () => { expect(sent).toEqual(["Alpha", "Beta"]); }); + + it("preserves assistant metadata on coalesced text flushes", async () => { + const sent: Array<{ assistantMessageIndex?: number; text?: string }> = []; + const pipeline = createBlockReplyPipeline({ + onBlockReply: async (payload) => { + sent.push({ + assistantMessageIndex: getReplyPayloadMetadata(payload)?.assistantMessageIndex, + text: payload.text, + }); + }, + timeoutMs: 5000, + coalescing: { + minChars: 100, + maxChars: 200, + idleMs: 1000, + joiner: " ", + }, + }); + + pipeline.enqueue(setReplyPayloadMetadata({ text: "Alpha" }, { assistantMessageIndex: 0 })); + pipeline.enqueue(setReplyPayloadMetadata({ text: "Beta" }, { assistantMessageIndex: 0 })); + await pipeline.flush({ force: true }); + + expect(sent).toEqual([{ assistantMessageIndex: 0, text: "Alpha Beta" }]); + }); }); describe("createBlockReplyPipeline content coverage dedup", () => { diff --git a/src/auto-reply/reply/reply-dispatcher.ts b/src/auto-reply/reply/reply-dispatcher.ts index f2fbc14aa142..2859b1e97842 100644 --- a/src/auto-reply/reply/reply-dispatcher.ts +++ b/src/auto-reply/reply/reply-dispatcher.ts @@ -6,7 +6,7 @@ import { generateSecureInt } from "../../infra/secure-random.js"; import { createSubsystemLogger } from "../../logging/subsystem.js"; import type { SilentReplyConversationType } from "../../shared/silent-reply-policy.js"; import { sleep } from "../../utils.js"; -import { copyReplyPayloadMetadata } from "../reply-payload.js"; +import { copyReplyPayloadMetadata, getReplyPayloadMetadata } from "../reply-payload.js"; import { isSilentReplyText, SILENT_REPLY_TOKEN } from "../tokens.js"; import type { GetReplyOptions, ReplyPayload } from "../types.js"; import { registerDispatcher } from "./dispatcher-registry.js"; @@ -47,6 +47,19 @@ const DEFAULT_HUMAN_DELAY_MIN_MS = 800; const DEFAULT_HUMAN_DELAY_MAX_MS = 2500; const silentReplyLogger = createSubsystemLogger("silent-reply/dispatcher"); +type ReplyDispatchRuntimeInfo = { kind: ReplyDispatchKind; assistantMessageIndex?: number }; + +function buildReplyDispatchRuntimeInfo( + payload: ReplyPayload, + kind: ReplyDispatchKind, +): ReplyDispatchRuntimeInfo { + const assistantMessageIndex = getReplyPayloadMetadata(payload)?.assistantMessageIndex; + return { + kind, + ...(assistantMessageIndex !== undefined ? { assistantMessageIndex } : {}), + }; +} + /** Generate a random delay within the configured range. */ function getHumanDelay(config: HumanDelayConfig | undefined): number { const mode = config?.mode ?? "off"; @@ -175,7 +188,11 @@ export function createReplyDispatcher(options: ReplyDispatcherOptions): ReplyDis responsePrefixContextProvider: options.responsePrefixContextProvider, transformReplyPayload: options.transformReplyPayload, onHeartbeatStrip: options.onHeartbeatStrip, - onSkip: (reason) => options.onSkip?.(payload, { kind, reason }), + onSkip: (reason) => + options.onSkip?.(payload, { + ...buildReplyDispatchRuntimeInfo(payload, kind), + reason, + }), }); if (!normalized) { if (kind === "final" && originalWasExactSilent) { @@ -205,25 +222,35 @@ export function createReplyDispatcher(options: ReplyDispatcherOptions): ReplyDis await sleep(delayMs); } } + const dispatchInfo = buildReplyDispatchRuntimeInfo(normalized, kind); let deliverPayload: ReplyPayload | null = normalized; if (beforeDeliver) { - deliverPayload = await beforeDeliver(normalized, { kind }); + try { + deliverPayload = await beforeDeliver(normalized, dispatchInfo); + } catch (err: unknown) { + try { + await options.onBeforeDeliverCancelled?.(normalized, dispatchInfo); + } catch (cancelErr: unknown) { + void options.onError?.(cancelErr, dispatchInfo); + } + throw err; + } if (!deliverPayload) { cancelledCounts[kind] += 1; try { - await options.onBeforeDeliverCancelled?.(normalized, { kind }); + await options.onBeforeDeliverCancelled?.(normalized, dispatchInfo); } catch (err: unknown) { - void options.onError?.(err, { kind }); + void options.onError?.(err, dispatchInfo); } return; } deliverPayload = copyReplyPayloadMetadata(normalized, deliverPayload); } - await options.deliver(deliverPayload, { kind }); + await options.deliver(deliverPayload, dispatchInfo); }) .catch((err: unknown) => { failedCounts[kind] += 1; - void options.onError?.(err, { kind }); + void options.onError?.(err, buildReplyDispatchRuntimeInfo(normalized, kind)); }) .finally(() => { pending -= 1;