openclaw/extensions/telegram/src/bot-message-dispatch.ts

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<typeof resolveDispatchTelegramContext>;
  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<typeof resolveDispatchTelegramContext>;
}) {
  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<typeof resolveDispatchTelegramContext>;
  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<TelegramDispatchResult> => {
  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" };
};