import { getGlobalHookRunner } from "openclaw/plugin-sdk/plugin-runtime"; import { createSubsystemLogger, danger, logVerbose } from "openclaw/plugin-sdk/runtime-env"; import { truncateUtf16Safe } from "openclaw/plugin-sdk/text-utility-runtime"; import { resolveDispatchTelegramContext } from "./bot-message-dispatch-context.js"; import { createDeliveryState, deliverFallback, finalizePendingAnswerBlockDraft, } from "./bot-message-dispatch-delivery.js"; import { cleanupDrafts, createDraftState, prepareAnswerLaneForToolProgress, waitForDraftEvents, } from "./bot-message-dispatch-draft.js"; import { createProgressState } from "./bot-message-dispatch-progress.js"; import { createReplyState } from "./bot-message-dispatch-reply.js"; import { createFreshTelegramSessionEntryLoader, resolveTelegramReasoningLevel, } from "./bot-message-dispatch-session.js"; import { createTelegramDispatchStatus } from "./bot-message-dispatch-status.js"; import { runTelegramDispatchTurn } from "./bot-message-dispatch-turn.js"; import { findModelInCatalog, loadPreparedModelCatalog, modelSupportsVision, resolveAgentDir, resolveDefaultModelForAgent, } from "./bot-message-dispatch.agent.runtime.js"; import { generateTopicLabel, getAgentScopedMediaLocalRoots, resolveAutoTopicLabelConfig, resolveChunkMode, resolveMarkdownTableMode, } from "./bot-message-dispatch.runtime.js"; import type { DispatchTelegramMessageParams, TelegramDispatchTurn, TelegramDispatchResult, } from "./bot-message-dispatch.types.js"; import { getTelegramTextParts, resolveTelegramReplyId } from "./bot/helpers.js"; import { addTelegramNativeQuoteCandidate, buildTelegramNativeQuoteCandidate, type TelegramNativeQuoteCandidateByMessageId, } from "./bot/native-quote.js"; import { cacheSticker, describeStickerImage } from "./sticker-cache.js"; const EMPTY_RESPONSE_FALLBACK = "No response generated. Please try again."; const silentReplyDispatchLogger = createSubsystemLogger("telegram/silent-reply-dispatch"); async function resolveStickerVisionSupport( cfg: DispatchTelegramMessageParams["cfg"], agentId: string, ) { try { const catalog = await loadPreparedModelCatalog({ config: cfg, agentId, agentDir: resolveAgentDir(cfg, agentId), readOnly: true, }); const defaultModel = resolveDefaultModelForAgent({ cfg, agentId }); const entry = findModelInCatalog(catalog, defaultModel.provider, defaultModel.model); return entry ? modelSupportsVision(entry) : false; } catch { return false; } } function includeStickerDescription(params: { body: string | undefined; formattedDescription: string; }): string { if (!params.body) { return params.formattedDescription; } const current = params.body.trim(); if (!current) { return params.formattedDescription; } if (params.body.includes(params.formattedDescription)) { return params.body; } return `${params.formattedDescription}\n${params.body}`; } function resolveTelegramQuoteContext(params: { context: ReturnType; replyToMode: DispatchTelegramMessageParams["replyToMode"]; }) { const { context, replyToMode } = params; const rawReplyQuoteText = context.ctxPayload.ReplyToIsQuote && typeof context.ctxPayload.ReplyToQuoteText === "string" ? context.ctxPayload.ReplyToQuoteText : undefined; const replyQuoteText = context.ctxPayload.ReplyToIsQuote ? rawReplyQuoteText?.trim() ? rawReplyQuoteText : context.ctxPayload.ReplyToBody?.trim() || undefined : undefined; const replyQuoteMessageId = replyQuoteText && !context.ctxPayload.ReplyToIsExternal ? resolveTelegramReplyId(context.ctxPayload.ReplyToId) : undefined; const replyQuoteTargetsBotMessage = context.msg.reply_to_message?.from?.is_bot === true; const replyQuoteByMessageId: TelegramNativeQuoteCandidateByMessageId = {}; if (replyToMode !== "off") { if (replyQuoteText && replyQuoteMessageId != null) { addTelegramNativeQuoteCandidate(replyQuoteByMessageId, replyQuoteMessageId, { text: replyQuoteText, ...(typeof context.ctxPayload.ReplyToQuotePosition === "number" ? { position: context.ctxPayload.ReplyToQuotePosition } : {}), ...(Array.isArray(context.ctxPayload.ReplyToQuoteEntities) ? { entities: context.ctxPayload.ReplyToQuoteEntities } : {}), }); } addTelegramNativeQuoteCandidate( replyQuoteByMessageId, context.ctxPayload.MessageSid ?? context.msg.message_id, buildTelegramNativeQuoteCandidate(getTelegramTextParts(context.msg)), ); if ( !context.ctxPayload.ReplyToIsExternal && typeof context.ctxPayload.ReplyToQuoteSourceText === "string" ) { addTelegramNativeQuoteCandidate( replyQuoteByMessageId, context.ctxPayload.ReplyToId, buildTelegramNativeQuoteCandidate({ text: context.ctxPayload.ReplyToQuoteSourceText, entities: Array.isArray(context.ctxPayload.ReplyToQuoteSourceEntities) ? context.ctxPayload.ReplyToQuoteSourceEntities : undefined, }), ); } } const draftReplyToMessageId = replyToMode !== "off" && typeof context.msg.message_id === "number" ? replyQuoteTargetsBotMessage ? context.msg.message_id : (replyQuoteMessageId ?? context.msg.message_id) : undefined; return { draftReplyToMessageId, replyQuoteByMessageId, replyQuoteEntities: Array.isArray(context.ctxPayload.ReplyToQuoteEntities) ? context.ctxPayload.ReplyToQuoteEntities : undefined, replyQuoteMessageId, replyQuotePosition: typeof context.ctxPayload.ReplyToQuotePosition === "number" ? context.ctxPayload.ReplyToQuotePosition : undefined, replyQuoteText, }; } async function prepareTelegramSticker(params: { cfg: DispatchTelegramMessageParams["cfg"]; context: ReturnType; }) { const { context } = params; const sticker = context.ctxPayload.Sticker; const stickerFact = context.ctxPayload.media?.find((media) => media.kind === "sticker"); const stickerPath = stickerFact?.path ?? (!stickerFact && context.ctxPayload.StickerMediaIncluded ? context.ctxPayload.media?.[0]?.path : undefined); if (!sticker?.fileId || !sticker.fileUniqueId || !stickerPath) { return; } const agentDir = resolveAgentDir(params.cfg, context.route.agentId); const stickerSupportsVision = await resolveStickerVisionSupport( params.cfg, context.route.agentId, ); const description = sticker.cachedDescription || (await describeStickerImage({ imagePath: stickerPath, cfg: params.cfg, agentDir, agentId: context.route.agentId, })); if (!description) { return; } const stickerContext = [sticker.emoji, sticker.setName ? `from "${sticker.setName}"` : null] .filter(Boolean) .join(" "); const formattedDescription = `[Sticker${stickerContext ? ` ${stickerContext}` : ""}] ${description}`; sticker.cachedDescription = description; if (!stickerSupportsVision) { const isCaptionlessSticker = !context.ctxPayload.RawBody?.trim() && context.ctxPayload.StickerMediaIncluded === true; context.ctxPayload.Body = includeStickerDescription({ body: context.ctxPayload.Body, formattedDescription, }); context.ctxPayload.BodyForAgent = isCaptionlessSticker && !context.ctxPayload.BodyForAgent?.trim() ? formattedDescription : includeStickerDescription({ body: context.ctxPayload.BodyForAgent, formattedDescription, }); context.ctxPayload.SkipStickerMediaUnderstanding = true; } cacheSticker({ fileId: sticker.fileId, fileUniqueId: sticker.fileUniqueId, emoji: sticker.emoji, setName: sticker.setName, description, cachedAt: new Date().toISOString(), receivedFrom: context.ctxPayload.From, }); logVerbose(`telegram: cached sticker description for ${sticker.fileUniqueId}`); } function scheduleDmTopicLabel(params: { bot: DispatchTelegramMessageParams["bot"]; cfg: DispatchTelegramMessageParams["cfg"]; context: ReturnType; isFirstTurnInSession: boolean; telegramCfg: DispatchTelegramMessageParams["telegramCfg"]; }) { const { context } = params; const isDmTopic = !context.isGroup && context.threadSpec.scope === "dm" && context.threadSpec.id != null; if (!isDmTopic || !params.isFirstTurnInSession) { return; } const userMessage = truncateUtf16Safe( context.ctxPayload.RawBody ?? context.ctxPayload.Body ?? "", 500, ); if (!userMessage.trim()) { return; } const directAutoTopicLabel = context.groupConfig && "autoTopicLabel" in context.groupConfig ? context.groupConfig.autoTopicLabel : undefined; const autoTopicConfig = resolveAutoTopicLabelConfig( directAutoTopicLabel, params.telegramCfg.autoTopicLabel, ); if (!autoTopicConfig) { return; } const topicThreadId = context.threadSpec.id!; void (async () => { try { const label = await generateTopicLabel({ userMessage, prompt: autoTopicConfig.prompt, cfg: params.cfg, agentId: context.route.agentId, agentDir: resolveAgentDir(params.cfg, context.route.agentId), }); if (!label) { logVerbose("auto-topic-label: LLM returned empty label"); return; } logVerbose(`auto-topic-label: generated label (len=${label.length})`); await params.bot.api.editForumTopic(context.chatId, topicThreadId, { name: label }); logVerbose(`auto-topic-label: renamed topic ${context.chatId}/${topicThreadId}`); } catch (err) { logVerbose(`auto-topic-label: failed: ${String(err)}`); } })(); } export const dispatchTelegramMessage = async ( dispatchParams: DispatchTelegramMessageParams, ): Promise => { const { context, bot, cfg, runtime, replyToMode, telegramCfg, telegramDeps: injectedTelegramDeps, retryDispatchErrors = false, suppressFailureFallback = false, turnAdoptionLifecycle, } = dispatchParams; const dispatchStartedAt = Date.now(); const dispatchContext = resolveDispatchTelegramContext({ context }); const telegramDeps = injectedTelegramDeps ?? (await import("./bot-deps.js")).defaultTelegramBotDeps; const loadFreshSessionEntry = createFreshTelegramSessionEntryLoader({ cfg, telegramDeps }); const isRoomEvent = dispatchContext.ctxPayload.InboundEventKind === "room_event"; const status = createTelegramDispatchStatus({ context: dispatchContext }); const tableMode = resolveMarkdownTableMode({ cfg, channel: "telegram", accountId: dispatchContext.route.accountId, supportsBlockTables: telegramCfg.richMessages === true, }); const resolvedReasoningLevel = resolveTelegramReasoningLevel({ cfg, sessionKey: dispatchContext.ctxPayload.SessionKey, agentId: dispatchContext.route.agentId, loadFreshSessionEntry, }); const quote = resolveTelegramQuoteContext({ context: dispatchContext, replyToMode }); // Draft messages are provider-visible before final modifiers run. Suppress them when a hook // can rewrite or cancel, or the original payload can flash before the normal delivery gate. const hookRunner = getGlobalHookRunner(); const allowProviderPreview = !( (hookRunner?.hasHooks("reply_payload_sending") ?? false) || (hookRunner?.hasHooks("message_sending") ?? false) ); const isDispatchSuperseded = () => turnAdoptionLifecycle?.abortSignal?.aborted === true; const turnConfig = { ...dispatchParams, allowProviderPreview, chunkMode: resolveChunkMode(cfg, "telegram", dispatchContext.route.accountId), context: dispatchContext, dispatchStartedAt, draftReplyToMessageId: quote.draftReplyToMessageId, isSuperseded: isDispatchSuperseded, loadFreshSessionEntry, mediaLocalRoots: getAgentScopedMediaLocalRoots(cfg, dispatchContext.route.agentId), replyQuoteByMessageId: quote.replyQuoteByMessageId, replyQuoteEntities: quote.replyQuoteEntities, replyQuoteMessageId: quote.replyQuoteMessageId, replyQuotePosition: quote.replyQuotePosition, replyQuoteText: quote.replyQuoteText, resolvedReasoningLevel, statusReactionController: status.controller, tableMode, telegramDeps, }; const draftState = createDraftState(turnConfig); const progressState = createProgressState( turnConfig, draftState, () => turn, async () => await prepareAnswerLaneForToolProgress(turn), ); const deliveryState = createDeliveryState({ ...turnConfig, lanes: draftState.lanes }, () => turn); const turn: TelegramDispatchTurn = { ...turnConfig, ...draftState, ...progressState, ...deliveryState, ...createReplyState(), queuedFinal: false, noVisibleReplyFallbackEligible: false, suppressSilentReplyFallback: false, hadErrorReplyFailureOrSkip: false, }; let isFirstTurnInSession = false; let dispatchWasSuperseded: boolean; let turnDispatched: boolean | undefined; const isDmTopic = !dispatchContext.isGroup && dispatchContext.threadSpec.scope === "dm" && dispatchContext.threadSpec.id != null; try { await prepareTelegramSticker({ cfg, context: dispatchContext }); if (isDmTopic) { try { const sessionKey = dispatchContext.ctxPayload.SessionKey; if (sessionKey) { isFirstTurnInSession = !loadFreshSessionEntry(dispatchContext.route.agentId, sessionKey) .entry?.systemSent; } else { logVerbose("auto-topic-label: SessionKey is absent, skipping first-turn detection"); } } catch (err) { logVerbose(`auto-topic-label: session store error: ${String(err)}`); } } loadFreshSessionEntry.clear(); if (status.controller && !isRoomEvent) { void status.controller.setThinking(); } try { turnDispatched = await runTelegramDispatchTurn(turn); } catch (err) { turn.dispatchError = err; runtime.error?.(danger(`telegram dispatch failed: ${String(err)}`)); } finally { // Terminal order: stop producers, drain queued drafts, materialize accepted text, // clean previews, then collapse the progress window. turn.progressCompositor.cancel(); await waitForDraftEvents(turn); try { await finalizePendingAnswerBlockDraft(turn); } catch (err) { turn.dispatchError ??= err; runtime.error?.(danger(`telegram terminal block delivery failed: ${String(err)}`)); } await cleanupDrafts(turn, isDispatchSuperseded()); } } finally { dispatchWasSuperseded = isDispatchSuperseded(); } if (turnDispatched === false) { return { kind: "completed" }; } if (dispatchWasSuperseded) { if (status.controller) { status.finalizeInBackground({ outcome: "done" }, "finalize"); } return { kind: "completed" }; } const deliverySummary = turn.deliveryState.snapshot(); let sentFallback = false; const shouldSendFailureFallback = !isRoomEvent && !suppressFailureFallback && !turn.finalAnswerDelivered && (turn.dispatchError || deliverySummary.failedNonSilent > 0 || (deliverySummary.skippedNonSilent > 0 && !turn.suppressSilentReplyFallback)); if (shouldSendFailureFallback) { const fallbackText = turn.dispatchError ? "Something went wrong while processing your request. Please try again." : EMPTY_RESPONSE_FALLBACK; const result = await deliverFallback( turn, [{ text: fallbackText }], telegramCfg.silentErrorReplies === true && (turn.dispatchError != null || turn.hadErrorReplyFailureOrSkip), ); sentFallback = result.delivered; } if ( !sentFallback && !turn.dispatchError && !deliverySummary.delivered && !turn.suppressSilentReplyFallback && !turn.queuedFinal && turn.noVisibleReplyFallbackEligible ) { sentFallback = (await deliverFallback(turn, [{ text: EMPTY_RESPONSE_FALLBACK }], false)) .delivered; silentReplyDispatchLogger.debug("telegram recovered eligible turn without visible response", { hasSessionKey: Boolean(dispatchContext.ctxPayload.SessionKey), hasChatId: dispatchContext.chatId != null, queuedFinal: turn.queuedFinal, sentFallback, }); } const hasFinalResponse = turn.finalAnswerDelivered || sentFallback || turn.suppressSilentReplyFallback || turn.queuedFinal; const hasVisibleResponse = deliverySummary.delivered || sentFallback || turn.suppressSilentReplyFallback || turn.queuedFinal; const deliveryFailureWithoutFinalResponse = !turn.finalAnswerDelivered && (deliverySummary.skippedNonSilent > 0 || deliverySummary.failedNonSilent > 0); const retryableDispatchFailure = turn.dispatchError ?? (deliveryFailureWithoutFinalResponse ? new Error( `Telegram reply delivery failed without a final response (failed=${deliverySummary.failedNonSilent}, skipped=${deliverySummary.skippedNonSilent})`, ) : null); if (status.controller && !hasVisibleResponse) { status.finalizeInBackground({ outcome: "error" }, "error finalize"); } const shouldReturnRetryableDispatchFailure = retryDispatchErrors && ((turn.dispatchError != null && !hasFinalResponse) || (turn.dispatchError == null && deliveryFailureWithoutFinalResponse && !hasVisibleResponse)); if (retryableDispatchFailure && shouldReturnRetryableDispatchFailure) { return { kind: "failed-retryable", error: retryableDispatchFailure }; } if (!hasVisibleResponse) { return { kind: "completed" }; } scheduleDmTopicLabel({ bot, cfg, context: dispatchContext, isFirstTurnInSession, telegramCfg, }); if (status.controller) { status.finalizeInBackground( { outcome: turn.agentRunFailed || turn.dispatchError != null || (!turn.finalAnswerDelivered && sentFallback) ? "error" : "done", }, "finalize", ); } return { kind: "completed" }; };