mirror of
https://github.com/openclaw/openclaw.git
synced 2026-03-17 21:10:54 +00:00
3fe4c19305
* fix(telegram): prevent duplicate messages in DM draft streaming mode When using sendMessageDraft for DM streaming (streaming: 'partial'), the draft bubble auto-converts to the final message. The code was incorrectly falling through to sendPayload() after the draft was finalized, causing a duplicate message. This fix checks if we're in draft preview mode with hasStreamedMessage and skips the sendPayload call, returning "preview-finalized" directly. Key changes: - Use hasStreamedMessage flag instead of previewRevision comparison - Avoids double stopDraftLane calls by returning early - Prevents duplicate messages when final text equals last streamed text Root cause: In lane-delivery.ts, the final message handling logic did not properly handle the DM draft flow where sendMessageDraft creates a transient bubble that doesn't need a separate final send. * fix(telegram): harden DM draft finalization path * fix(telegram): require emitted draft preview for unchanged finals * fix(telegram): require final draft text emission before finalize * fix: update changelog for telegram draft finalization (#32118) (thanks @OpenCils) --------- Co-authored-by: Ayaan Zaidi <zaidi@uplause.io>
733 lines
25 KiB
TypeScript
733 lines
25 KiB
TypeScript
import type { Bot } from "grammy";
|
|
import { resolveAgentDir } from "../agents/agent-scope.js";
|
|
import {
|
|
findModelInCatalog,
|
|
loadModelCatalog,
|
|
modelSupportsVision,
|
|
} from "../agents/model-catalog.js";
|
|
import { resolveDefaultModelForAgent } from "../agents/model-selection.js";
|
|
import { resolveChunkMode } from "../auto-reply/chunk.js";
|
|
import { clearHistoryEntriesIfEnabled } from "../auto-reply/reply/history.js";
|
|
import { dispatchReplyWithBufferedBlockDispatcher } from "../auto-reply/reply/provider-dispatcher.js";
|
|
import type { ReplyPayload } from "../auto-reply/types.js";
|
|
import { removeAckReactionAfterReply } from "../channels/ack-reactions.js";
|
|
import { logAckFailure, logTypingFailure } from "../channels/logging.js";
|
|
import { createReplyPrefixOptions } from "../channels/reply-prefix.js";
|
|
import { createTypingCallbacks } from "../channels/typing.js";
|
|
import { resolveMarkdownTableMode } from "../config/markdown-tables.js";
|
|
import { loadSessionStore, resolveStorePath } from "../config/sessions.js";
|
|
import type { OpenClawConfig, ReplyToMode, TelegramAccountConfig } from "../config/types.js";
|
|
import { danger, logVerbose } from "../globals.js";
|
|
import { getAgentScopedMediaLocalRoots } from "../media/local-roots.js";
|
|
import type { RuntimeEnv } from "../runtime.js";
|
|
import type { TelegramMessageContext } from "./bot-message-context.js";
|
|
import type { TelegramBotOptions } from "./bot.js";
|
|
import { deliverReplies } from "./bot/delivery.js";
|
|
import type { TelegramStreamMode } from "./bot/types.js";
|
|
import type { TelegramInlineButtons } from "./button-types.js";
|
|
import { createTelegramDraftStream } from "./draft-stream.js";
|
|
import { renderTelegramHtmlText } from "./format.js";
|
|
import {
|
|
type ArchivedPreview,
|
|
createLaneDeliveryStateTracker,
|
|
createLaneTextDeliverer,
|
|
type DraftLaneState,
|
|
type LaneName,
|
|
} from "./lane-delivery.js";
|
|
import {
|
|
createTelegramReasoningStepState,
|
|
splitTelegramReasoningText,
|
|
} from "./reasoning-lane-coordinator.js";
|
|
import { editMessageTelegram } from "./send.js";
|
|
import { cacheSticker, describeStickerImage } from "./sticker-cache.js";
|
|
|
|
const EMPTY_RESPONSE_FALLBACK = "No response generated. Please try again.";
|
|
|
|
/** Minimum chars before sending first streaming message (improves push notification UX) */
|
|
const DRAFT_MIN_INITIAL_CHARS = 30;
|
|
|
|
async function resolveStickerVisionSupport(cfg: OpenClawConfig, agentId: string) {
|
|
try {
|
|
const catalog = await loadModelCatalog({ config: cfg });
|
|
const defaultModel = resolveDefaultModelForAgent({ cfg, agentId });
|
|
const entry = findModelInCatalog(catalog, defaultModel.provider, defaultModel.model);
|
|
if (!entry) {
|
|
return false;
|
|
}
|
|
return modelSupportsVision(entry);
|
|
} catch {
|
|
return false;
|
|
}
|
|
}
|
|
|
|
export function pruneStickerMediaFromContext(
|
|
ctxPayload: {
|
|
MediaPath?: string;
|
|
MediaUrl?: string;
|
|
MediaType?: string;
|
|
MediaPaths?: string[];
|
|
MediaUrls?: string[];
|
|
MediaTypes?: string[];
|
|
},
|
|
opts?: { stickerMediaIncluded?: boolean },
|
|
) {
|
|
if (opts?.stickerMediaIncluded === false) {
|
|
return;
|
|
}
|
|
const nextMediaPaths = Array.isArray(ctxPayload.MediaPaths)
|
|
? ctxPayload.MediaPaths.slice(1)
|
|
: undefined;
|
|
const nextMediaUrls = Array.isArray(ctxPayload.MediaUrls)
|
|
? ctxPayload.MediaUrls.slice(1)
|
|
: undefined;
|
|
const nextMediaTypes = Array.isArray(ctxPayload.MediaTypes)
|
|
? ctxPayload.MediaTypes.slice(1)
|
|
: undefined;
|
|
ctxPayload.MediaPaths = nextMediaPaths && nextMediaPaths.length > 0 ? nextMediaPaths : undefined;
|
|
ctxPayload.MediaUrls = nextMediaUrls && nextMediaUrls.length > 0 ? nextMediaUrls : undefined;
|
|
ctxPayload.MediaTypes = nextMediaTypes && nextMediaTypes.length > 0 ? nextMediaTypes : undefined;
|
|
ctxPayload.MediaPath = ctxPayload.MediaPaths?.[0];
|
|
ctxPayload.MediaUrl = ctxPayload.MediaUrls?.[0] ?? ctxPayload.MediaPath;
|
|
ctxPayload.MediaType = ctxPayload.MediaTypes?.[0];
|
|
}
|
|
|
|
type DispatchTelegramMessageParams = {
|
|
context: TelegramMessageContext;
|
|
bot: Bot;
|
|
cfg: OpenClawConfig;
|
|
runtime: RuntimeEnv;
|
|
replyToMode: ReplyToMode;
|
|
streamMode: TelegramStreamMode;
|
|
textLimit: number;
|
|
telegramCfg: TelegramAccountConfig;
|
|
opts: Pick<TelegramBotOptions, "token">;
|
|
};
|
|
|
|
type TelegramReasoningLevel = "off" | "on" | "stream";
|
|
|
|
function resolveTelegramReasoningLevel(params: {
|
|
cfg: OpenClawConfig;
|
|
sessionKey?: string;
|
|
agentId: string;
|
|
}): TelegramReasoningLevel {
|
|
const { cfg, sessionKey, agentId } = params;
|
|
if (!sessionKey) {
|
|
return "off";
|
|
}
|
|
try {
|
|
const storePath = resolveStorePath(cfg.session?.store, { agentId });
|
|
const store = loadSessionStore(storePath, { skipCache: true });
|
|
const entry = store[sessionKey.toLowerCase()] ?? store[sessionKey];
|
|
const level = entry?.reasoningLevel;
|
|
if (level === "on" || level === "stream") {
|
|
return level;
|
|
}
|
|
} catch {
|
|
// Fall through to default.
|
|
}
|
|
return "off";
|
|
}
|
|
|
|
export const dispatchTelegramMessage = async ({
|
|
context,
|
|
bot,
|
|
cfg,
|
|
runtime,
|
|
replyToMode,
|
|
streamMode,
|
|
textLimit,
|
|
telegramCfg,
|
|
opts,
|
|
}: DispatchTelegramMessageParams) => {
|
|
const {
|
|
ctxPayload,
|
|
msg,
|
|
chatId,
|
|
isGroup,
|
|
threadSpec,
|
|
historyKey,
|
|
historyLimit,
|
|
groupHistories,
|
|
route,
|
|
skillFilter,
|
|
sendTyping,
|
|
sendRecordVoice,
|
|
ackReactionPromise,
|
|
reactionApi,
|
|
removeAckAfterReply,
|
|
statusReactionController,
|
|
} = context;
|
|
|
|
const draftMaxChars = Math.min(textLimit, 4096);
|
|
const tableMode = resolveMarkdownTableMode({
|
|
cfg,
|
|
channel: "telegram",
|
|
accountId: route.accountId,
|
|
});
|
|
const renderDraftPreview = (text: string) => ({
|
|
text: renderTelegramHtmlText(text, { tableMode }),
|
|
parseMode: "HTML" as const,
|
|
});
|
|
const accountBlockStreamingEnabled =
|
|
typeof telegramCfg.blockStreaming === "boolean"
|
|
? telegramCfg.blockStreaming
|
|
: cfg.agents?.defaults?.blockStreamingDefault === "on";
|
|
const resolvedReasoningLevel = resolveTelegramReasoningLevel({
|
|
cfg,
|
|
sessionKey: ctxPayload.SessionKey,
|
|
agentId: route.agentId,
|
|
});
|
|
const forceBlockStreamingForReasoning = resolvedReasoningLevel === "on";
|
|
const streamReasoningDraft = resolvedReasoningLevel === "stream";
|
|
const previewStreamingEnabled = streamMode !== "off";
|
|
const canStreamAnswerDraft =
|
|
previewStreamingEnabled && !accountBlockStreamingEnabled && !forceBlockStreamingForReasoning;
|
|
const canStreamReasoningDraft = canStreamAnswerDraft || streamReasoningDraft;
|
|
const draftReplyToMessageId =
|
|
replyToMode !== "off" && typeof msg.message_id === "number" ? msg.message_id : undefined;
|
|
const draftMinInitialChars = DRAFT_MIN_INITIAL_CHARS;
|
|
const mediaLocalRoots = getAgentScopedMediaLocalRoots(cfg, route.agentId);
|
|
const archivedAnswerPreviews: ArchivedPreview[] = [];
|
|
const archivedReasoningPreviewIds: number[] = [];
|
|
const createDraftLane = (laneName: LaneName, enabled: boolean): DraftLaneState => {
|
|
const useMessagePreviewTransportForDmReasoning =
|
|
laneName === "reasoning" && threadSpec?.scope === "dm" && canStreamAnswerDraft;
|
|
const stream = enabled
|
|
? createTelegramDraftStream({
|
|
api: bot.api,
|
|
chatId,
|
|
maxChars: draftMaxChars,
|
|
thread: threadSpec,
|
|
previewTransport: useMessagePreviewTransportForDmReasoning ? "message" : "auto",
|
|
replyToMessageId: draftReplyToMessageId,
|
|
minInitialChars: draftMinInitialChars,
|
|
renderText: renderDraftPreview,
|
|
onSupersededPreview:
|
|
laneName === "answer" || laneName === "reasoning"
|
|
? (preview) => {
|
|
if (laneName === "reasoning") {
|
|
if (!archivedReasoningPreviewIds.includes(preview.messageId)) {
|
|
archivedReasoningPreviewIds.push(preview.messageId);
|
|
}
|
|
return;
|
|
}
|
|
archivedAnswerPreviews.push({
|
|
messageId: preview.messageId,
|
|
textSnapshot: preview.textSnapshot,
|
|
});
|
|
}
|
|
: undefined,
|
|
log: logVerbose,
|
|
warn: logVerbose,
|
|
})
|
|
: undefined;
|
|
return {
|
|
stream,
|
|
lastPartialText: "",
|
|
hasStreamedMessage: false,
|
|
previewRevisionBaseline: stream?.previewRevision?.() ?? 0,
|
|
};
|
|
};
|
|
const lanes: Record<LaneName, DraftLaneState> = {
|
|
answer: createDraftLane("answer", canStreamAnswerDraft),
|
|
reasoning: createDraftLane("reasoning", canStreamReasoningDraft),
|
|
};
|
|
const answerLane = lanes.answer;
|
|
const reasoningLane = lanes.reasoning;
|
|
let splitReasoningOnNextStream = false;
|
|
const reasoningStepState = createTelegramReasoningStepState();
|
|
type SplitLaneSegment = { lane: LaneName; text: string };
|
|
type SplitLaneSegmentsResult = {
|
|
segments: SplitLaneSegment[];
|
|
suppressedReasoningOnly: boolean;
|
|
};
|
|
const splitTextIntoLaneSegments = (text?: string): SplitLaneSegmentsResult => {
|
|
const split = splitTelegramReasoningText(text);
|
|
const segments: SplitLaneSegment[] = [];
|
|
const suppressReasoning = resolvedReasoningLevel === "off";
|
|
if (split.reasoningText && !suppressReasoning) {
|
|
segments.push({ lane: "reasoning", text: split.reasoningText });
|
|
}
|
|
if (split.answerText) {
|
|
segments.push({ lane: "answer", text: split.answerText });
|
|
}
|
|
return {
|
|
segments,
|
|
suppressedReasoningOnly:
|
|
Boolean(split.reasoningText) && suppressReasoning && !split.answerText,
|
|
};
|
|
};
|
|
const resetDraftLaneState = (lane: DraftLaneState) => {
|
|
lane.lastPartialText = "";
|
|
lane.hasStreamedMessage = false;
|
|
lane.previewRevisionBaseline = lane.stream?.previewRevision?.() ?? lane.previewRevisionBaseline;
|
|
};
|
|
const updateDraftFromPartial = (lane: DraftLaneState, text: string | undefined) => {
|
|
const laneStream = lane.stream;
|
|
if (!laneStream || !text) {
|
|
return;
|
|
}
|
|
if (text === lane.lastPartialText) {
|
|
return;
|
|
}
|
|
// Mark that we've received streaming content (for forceNewMessage decision).
|
|
lane.hasStreamedMessage = true;
|
|
// Some providers briefly emit a shorter prefix snapshot (for example
|
|
// "Sure." -> "Sure" -> "Sure."). Keep the longer preview to avoid
|
|
// visible punctuation flicker.
|
|
if (
|
|
lane.lastPartialText &&
|
|
lane.lastPartialText.startsWith(text) &&
|
|
text.length < lane.lastPartialText.length
|
|
) {
|
|
return;
|
|
}
|
|
lane.lastPartialText = text;
|
|
laneStream.update(text);
|
|
};
|
|
const ingestDraftLaneSegments = (text: string | undefined) => {
|
|
const split = splitTextIntoLaneSegments(text);
|
|
for (const segment of split.segments) {
|
|
if (segment.lane === "reasoning") {
|
|
reasoningStepState.noteReasoningHint();
|
|
reasoningStepState.noteReasoningDelivered();
|
|
}
|
|
updateDraftFromPartial(lanes[segment.lane], segment.text);
|
|
}
|
|
};
|
|
const flushDraftLane = async (lane: DraftLaneState) => {
|
|
if (!lane.stream) {
|
|
return;
|
|
}
|
|
await lane.stream.flush();
|
|
};
|
|
|
|
const disableBlockStreaming = !previewStreamingEnabled
|
|
? true
|
|
: forceBlockStreamingForReasoning
|
|
? false
|
|
: typeof telegramCfg.blockStreaming === "boolean"
|
|
? !telegramCfg.blockStreaming
|
|
: canStreamAnswerDraft
|
|
? true
|
|
: undefined;
|
|
|
|
const { onModelSelected, ...prefixOptions } = createReplyPrefixOptions({
|
|
cfg,
|
|
agentId: route.agentId,
|
|
channel: "telegram",
|
|
accountId: route.accountId,
|
|
});
|
|
const chunkMode = resolveChunkMode(cfg, "telegram", route.accountId);
|
|
|
|
// Handle uncached stickers: get a dedicated vision description before dispatch
|
|
// This ensures we cache a raw description rather than a conversational response
|
|
const sticker = ctxPayload.Sticker;
|
|
if (sticker?.fileId && sticker.fileUniqueId && ctxPayload.MediaPath) {
|
|
const agentDir = resolveAgentDir(cfg, route.agentId);
|
|
const stickerSupportsVision = await resolveStickerVisionSupport(cfg, route.agentId);
|
|
let description = sticker.cachedDescription ?? null;
|
|
if (!description) {
|
|
description = await describeStickerImage({
|
|
imagePath: ctxPayload.MediaPath,
|
|
cfg,
|
|
agentDir,
|
|
agentId: route.agentId,
|
|
});
|
|
}
|
|
if (description) {
|
|
// Format the description with sticker context
|
|
const stickerContext = [sticker.emoji, sticker.setName ? `from "${sticker.setName}"` : null]
|
|
.filter(Boolean)
|
|
.join(" ");
|
|
const formattedDesc = `[Sticker${stickerContext ? ` ${stickerContext}` : ""}] ${description}`;
|
|
|
|
sticker.cachedDescription = description;
|
|
if (!stickerSupportsVision) {
|
|
// Update context to use description instead of image
|
|
ctxPayload.Body = formattedDesc;
|
|
ctxPayload.BodyForAgent = formattedDesc;
|
|
// Drop only the sticker attachment; keep replied media context if present.
|
|
pruneStickerMediaFromContext(ctxPayload, {
|
|
stickerMediaIncluded: ctxPayload.StickerMediaIncluded,
|
|
});
|
|
}
|
|
|
|
// Cache the description for future encounters
|
|
if (sticker.fileId) {
|
|
cacheSticker({
|
|
fileId: sticker.fileId,
|
|
fileUniqueId: sticker.fileUniqueId,
|
|
emoji: sticker.emoji,
|
|
setName: sticker.setName,
|
|
description,
|
|
cachedAt: new Date().toISOString(),
|
|
receivedFrom: ctxPayload.From,
|
|
});
|
|
logVerbose(`telegram: cached sticker description for ${sticker.fileUniqueId}`);
|
|
} else {
|
|
logVerbose(`telegram: skipped sticker cache (missing fileId)`);
|
|
}
|
|
}
|
|
}
|
|
|
|
const replyQuoteText =
|
|
ctxPayload.ReplyToIsQuote && ctxPayload.ReplyToBody
|
|
? ctxPayload.ReplyToBody.trim() || undefined
|
|
: undefined;
|
|
const deliveryState = createLaneDeliveryStateTracker();
|
|
const finalizedPreviewByLane: Record<LaneName, boolean> = {
|
|
answer: false,
|
|
reasoning: false,
|
|
};
|
|
const clearGroupHistory = () => {
|
|
if (isGroup && historyKey) {
|
|
clearHistoryEntriesIfEnabled({ historyMap: groupHistories, historyKey, limit: historyLimit });
|
|
}
|
|
};
|
|
const deliveryBaseOptions = {
|
|
chatId: String(chatId),
|
|
accountId: route.accountId,
|
|
token: opts.token,
|
|
runtime,
|
|
bot,
|
|
mediaLocalRoots,
|
|
replyToMode,
|
|
textLimit,
|
|
thread: threadSpec,
|
|
tableMode,
|
|
chunkMode,
|
|
linkPreview: telegramCfg.linkPreview,
|
|
replyQuoteText,
|
|
};
|
|
const applyTextToPayload = (payload: ReplyPayload, text: string): ReplyPayload => {
|
|
if (payload.text === text) {
|
|
return payload;
|
|
}
|
|
return { ...payload, text };
|
|
};
|
|
const sendPayload = async (payload: ReplyPayload) => {
|
|
const result = await deliverReplies({
|
|
...deliveryBaseOptions,
|
|
replies: [payload],
|
|
onVoiceRecording: sendRecordVoice,
|
|
});
|
|
if (result.delivered) {
|
|
deliveryState.markDelivered();
|
|
}
|
|
return result.delivered;
|
|
};
|
|
const deliverLaneText = createLaneTextDeliverer({
|
|
lanes,
|
|
archivedAnswerPreviews,
|
|
finalizedPreviewByLane,
|
|
draftMaxChars,
|
|
applyTextToPayload,
|
|
sendPayload,
|
|
flushDraftLane,
|
|
stopDraftLane: async (lane) => {
|
|
await lane.stream?.stop();
|
|
},
|
|
editPreview: async ({ messageId, text, previewButtons }) => {
|
|
await editMessageTelegram(chatId, messageId, text, {
|
|
api: bot.api,
|
|
cfg,
|
|
accountId: route.accountId,
|
|
linkPreview: telegramCfg.linkPreview,
|
|
buttons: previewButtons,
|
|
});
|
|
},
|
|
deletePreviewMessage: async (messageId) => {
|
|
await bot.api.deleteMessage(chatId, messageId);
|
|
},
|
|
log: logVerbose,
|
|
markDelivered: () => {
|
|
deliveryState.markDelivered();
|
|
},
|
|
});
|
|
|
|
let queuedFinal = false;
|
|
|
|
if (statusReactionController) {
|
|
void statusReactionController.setThinking();
|
|
}
|
|
|
|
const typingCallbacks = createTypingCallbacks({
|
|
start: sendTyping,
|
|
onStartError: (err) => {
|
|
logTypingFailure({
|
|
log: logVerbose,
|
|
channel: "telegram",
|
|
target: String(chatId),
|
|
error: err,
|
|
});
|
|
},
|
|
});
|
|
|
|
try {
|
|
({ queuedFinal } = await dispatchReplyWithBufferedBlockDispatcher({
|
|
ctx: ctxPayload,
|
|
cfg,
|
|
dispatcherOptions: {
|
|
...prefixOptions,
|
|
typingCallbacks,
|
|
deliver: async (payload, info) => {
|
|
const previewButtons = (
|
|
payload.channelData?.telegram as { buttons?: TelegramInlineButtons } | undefined
|
|
)?.buttons;
|
|
const split = splitTextIntoLaneSegments(payload.text);
|
|
const segments = split.segments;
|
|
const hasMedia = Boolean(payload.mediaUrl) || (payload.mediaUrls?.length ?? 0) > 0;
|
|
|
|
const flushBufferedFinalAnswer = async () => {
|
|
const buffered = reasoningStepState.takeBufferedFinalAnswer();
|
|
if (!buffered) {
|
|
return;
|
|
}
|
|
const bufferedButtons = (
|
|
buffered.payload.channelData?.telegram as
|
|
| { buttons?: TelegramInlineButtons }
|
|
| undefined
|
|
)?.buttons;
|
|
await deliverLaneText({
|
|
laneName: "answer",
|
|
text: buffered.text,
|
|
payload: buffered.payload,
|
|
infoKind: "final",
|
|
previewButtons: bufferedButtons,
|
|
});
|
|
reasoningStepState.resetForNextStep();
|
|
};
|
|
|
|
for (const segment of segments) {
|
|
if (
|
|
segment.lane === "answer" &&
|
|
info.kind === "final" &&
|
|
reasoningStepState.shouldBufferFinalAnswer()
|
|
) {
|
|
reasoningStepState.bufferFinalAnswer({ payload, text: segment.text });
|
|
continue;
|
|
}
|
|
if (segment.lane === "reasoning") {
|
|
reasoningStepState.noteReasoningHint();
|
|
}
|
|
const result = await deliverLaneText({
|
|
laneName: segment.lane,
|
|
text: segment.text,
|
|
payload,
|
|
infoKind: info.kind,
|
|
previewButtons,
|
|
allowPreviewUpdateForNonFinal: segment.lane === "reasoning",
|
|
});
|
|
if (segment.lane === "reasoning") {
|
|
if (result !== "skipped") {
|
|
reasoningStepState.noteReasoningDelivered();
|
|
await flushBufferedFinalAnswer();
|
|
}
|
|
continue;
|
|
}
|
|
if (info.kind === "final") {
|
|
if (reasoningLane.hasStreamedMessage) {
|
|
finalizedPreviewByLane.reasoning = true;
|
|
}
|
|
reasoningStepState.resetForNextStep();
|
|
}
|
|
}
|
|
if (segments.length > 0) {
|
|
return;
|
|
}
|
|
if (split.suppressedReasoningOnly) {
|
|
if (hasMedia) {
|
|
const payloadWithoutSuppressedReasoning =
|
|
typeof payload.text === "string" ? { ...payload, text: "" } : payload;
|
|
await sendPayload(payloadWithoutSuppressedReasoning);
|
|
}
|
|
if (info.kind === "final") {
|
|
await flushBufferedFinalAnswer();
|
|
}
|
|
return;
|
|
}
|
|
|
|
if (info.kind === "final") {
|
|
await answerLane.stream?.stop();
|
|
await reasoningLane.stream?.stop();
|
|
reasoningStepState.resetForNextStep();
|
|
}
|
|
const canSendAsIs =
|
|
hasMedia || (typeof payload.text === "string" && payload.text.length > 0);
|
|
if (!canSendAsIs) {
|
|
if (info.kind === "final") {
|
|
await flushBufferedFinalAnswer();
|
|
}
|
|
return;
|
|
}
|
|
await sendPayload(payload);
|
|
if (info.kind === "final") {
|
|
await flushBufferedFinalAnswer();
|
|
}
|
|
},
|
|
onSkip: (_payload, info) => {
|
|
if (info.reason !== "silent") {
|
|
deliveryState.markNonSilentSkip();
|
|
}
|
|
},
|
|
onError: (err, info) => {
|
|
deliveryState.markNonSilentFailure();
|
|
runtime.error?.(danger(`telegram ${info.kind} reply failed: ${String(err)}`));
|
|
},
|
|
},
|
|
replyOptions: {
|
|
skillFilter,
|
|
disableBlockStreaming,
|
|
onPartialReply:
|
|
answerLane.stream || reasoningLane.stream
|
|
? (payload) => ingestDraftLaneSegments(payload.text)
|
|
: undefined,
|
|
onReasoningStream: reasoningLane.stream
|
|
? (payload) => {
|
|
// Split between reasoning blocks only when the next reasoning
|
|
// stream starts. Splitting at reasoning-end can orphan the active
|
|
// preview and cause duplicate reasoning sends on reasoning final.
|
|
if (splitReasoningOnNextStream) {
|
|
reasoningLane.stream?.forceNewMessage();
|
|
resetDraftLaneState(reasoningLane);
|
|
splitReasoningOnNextStream = false;
|
|
}
|
|
ingestDraftLaneSegments(payload.text);
|
|
}
|
|
: undefined,
|
|
onAssistantMessageStart: answerLane.stream
|
|
? async () => {
|
|
reasoningStepState.resetForNextStep();
|
|
if (answerLane.hasStreamedMessage) {
|
|
const previewMessageId = answerLane.stream?.messageId();
|
|
// Only archive previews that still need a matching final text update.
|
|
// Once a preview has already been finalized, archiving it here causes
|
|
// cleanup to delete a user-visible final message on later media-only turns.
|
|
if (typeof previewMessageId === "number" && !finalizedPreviewByLane.answer) {
|
|
archivedAnswerPreviews.push({
|
|
messageId: previewMessageId,
|
|
textSnapshot: answerLane.lastPartialText,
|
|
});
|
|
}
|
|
answerLane.stream?.forceNewMessage();
|
|
}
|
|
resetDraftLaneState(answerLane);
|
|
// New assistant message boundary: this lane now tracks a fresh preview lifecycle.
|
|
finalizedPreviewByLane.answer = false;
|
|
}
|
|
: undefined,
|
|
onReasoningEnd: reasoningLane.stream
|
|
? () => {
|
|
// Split when/if a later reasoning block begins.
|
|
splitReasoningOnNextStream = reasoningLane.hasStreamedMessage;
|
|
}
|
|
: undefined,
|
|
onToolStart: statusReactionController
|
|
? async (payload) => {
|
|
await statusReactionController.setTool(payload.name);
|
|
}
|
|
: undefined,
|
|
onModelSelected,
|
|
},
|
|
}));
|
|
} finally {
|
|
// Must stop() first to flush debounced content before clear() wipes state.
|
|
const streamCleanupStates = new Map<
|
|
NonNullable<DraftLaneState["stream"]>,
|
|
{ shouldClear: boolean }
|
|
>();
|
|
const lanesToCleanup: Array<{ laneName: LaneName; lane: DraftLaneState }> = [
|
|
{ laneName: "answer", lane: answerLane },
|
|
{ laneName: "reasoning", lane: reasoningLane },
|
|
];
|
|
for (const laneState of lanesToCleanup) {
|
|
const stream = laneState.lane.stream;
|
|
if (!stream) {
|
|
continue;
|
|
}
|
|
const shouldClear = !finalizedPreviewByLane[laneState.laneName];
|
|
const existing = streamCleanupStates.get(stream);
|
|
if (!existing) {
|
|
streamCleanupStates.set(stream, { shouldClear });
|
|
continue;
|
|
}
|
|
existing.shouldClear = existing.shouldClear && shouldClear;
|
|
}
|
|
for (const [stream, cleanupState] of streamCleanupStates) {
|
|
await stream.stop();
|
|
if (cleanupState.shouldClear) {
|
|
await stream.clear();
|
|
}
|
|
}
|
|
for (const archivedPreview of archivedAnswerPreviews) {
|
|
try {
|
|
await bot.api.deleteMessage(chatId, archivedPreview.messageId);
|
|
} catch (err) {
|
|
logVerbose(
|
|
`telegram: archived answer preview cleanup failed (${archivedPreview.messageId}): ${String(err)}`,
|
|
);
|
|
}
|
|
}
|
|
for (const messageId of archivedReasoningPreviewIds) {
|
|
try {
|
|
await bot.api.deleteMessage(chatId, messageId);
|
|
} catch (err) {
|
|
logVerbose(
|
|
`telegram: archived reasoning preview cleanup failed (${messageId}): ${String(err)}`,
|
|
);
|
|
}
|
|
}
|
|
}
|
|
let sentFallback = false;
|
|
const deliverySummary = deliveryState.snapshot();
|
|
if (
|
|
!deliverySummary.delivered &&
|
|
(deliverySummary.skippedNonSilent > 0 || deliverySummary.failedNonSilent > 0)
|
|
) {
|
|
const result = await deliverReplies({
|
|
replies: [{ text: EMPTY_RESPONSE_FALLBACK }],
|
|
...deliveryBaseOptions,
|
|
});
|
|
sentFallback = result.delivered;
|
|
}
|
|
|
|
const hasFinalResponse = queuedFinal || sentFallback;
|
|
|
|
if (statusReactionController && !hasFinalResponse) {
|
|
void statusReactionController.setError().catch((err) => {
|
|
logVerbose(`telegram: status reaction error finalize failed: ${String(err)}`);
|
|
});
|
|
}
|
|
|
|
if (!hasFinalResponse) {
|
|
clearGroupHistory();
|
|
return;
|
|
}
|
|
|
|
if (statusReactionController) {
|
|
void statusReactionController.setDone().catch((err) => {
|
|
logVerbose(`telegram: status reaction finalize failed: ${String(err)}`);
|
|
});
|
|
} else {
|
|
removeAckReactionAfterReply({
|
|
removeAfterReply: removeAckAfterReply,
|
|
ackReactionPromise,
|
|
ackReactionValue: ackReactionPromise ? "ack" : null,
|
|
remove: () => reactionApi?.(chatId, msg.message_id ?? 0, []) ?? Promise.resolve(),
|
|
onError: (err) => {
|
|
if (!msg.message_id) {
|
|
return;
|
|
}
|
|
logAckFailure({
|
|
log: logVerbose,
|
|
channel: "telegram",
|
|
target: `${chatId}/${msg.message_id}`,
|
|
error: err,
|
|
});
|
|
},
|
|
});
|
|
}
|
|
clearGroupHistory();
|
|
};
|