resolveChannelStreamingPreviewToolProgress,

resolveTranscriptBackedChannelFinalText,

} from "openclaw/plugin-sdk/channel-streaming";

import { isAbortRequestText } from "openclaw/plugin-sdk/command-primitives-runtime";

import type {

OpenClawConfig,

ReplyToMode,

splitTelegramReasoningText,

} from "./reasoning-lane-coordinator.js";

import { editMessageTelegram } from "./send.js";

import { getTelegramSequentialKey } from "./sequential-key.js";

import { cacheSticker, describeStickerImage } from "./sticker-cache.js";

import {

beginTelegramReplyFence,

buildTelegramReplyFenceLaneKey,

endTelegramReplyFence,

getTelegramReplyFenceSizeForTests,

isTelegramReplyFenceSuperseded,

releaseTelegramReplyFenceAbortController,

resetTelegramReplyFenceForTests,

resolveTelegramReplyFenceKey,

shouldSupersedeTelegramReplyFence,

supersedeTelegramReplyFence,

} from "./telegram-reply-fence.js";

export { pruneStickerMediaFromContext } from "./bot-message-dispatch.media.js";

export { getTelegramReplyFenceSizeForTests, resetTelegramReplyFenceForTests };

const EMPTY_RESPONSE_FALLBACK = "No response generated. Please try again.";

const silentReplyDispatchLogger = createSubsystemLogger("telegram/silent-reply-dispatch");

type TelegramTranscriptMirrorPayload = { text?: string; mediaUrls?: string[] };

type TelegramReplyFenceState = {

generation: number;

activeDispatches: number;

abortControllers?: Set<AbortController>;

};

type TelegramReplyFenceKey = {

activeKey: string;

roomEventKey: string;

};

// Newer accepted turns and authorized aborts can arrive ahead of older same-session reply work.

const telegramReplyFenceByKey = new Map<string, TelegramReplyFenceState>();

function normalizeTelegramFenceKey(value: unknown): string | undefined {

if (typeof value !== "string") {

return undefined;

}

const trimmed = value.trim();

return trimmed.length > 0 ? trimmed : undefined;

}

function resolveTelegramReplyFenceKey(params: {

ctxPayload: { SessionKey?: string; CommandTargetSessionKey?: string; InboundEventKind?: string };

chatId: number | string;

threadSpec: { id?: number | string | null; scope?: string };

}): TelegramReplyFenceKey {

const baseKey =

normalizeTelegramFenceKey(params.ctxPayload.CommandTargetSessionKey) ??

normalizeTelegramFenceKey(params.ctxPayload.SessionKey) ??

`telegram:${String(params.chatId)}:${params.threadSpec.scope ?? "default"}:${params.threadSpec.id ?? "root"}`;

const roomEventKey = `${baseKey}:room_event`;

return {

activeKey: params.ctxPayload.InboundEventKind === "room_event" ? roomEventKey : baseKey,

roomEventKey,

};

}

function abortTelegramReplyFenceControllers(state: TelegramReplyFenceState): void {

for (const controller of state.abortControllers ?? []) {

controller.abort();

}

state.abortControllers?.clear();

}

function beginTelegramReplyFence(params: {

key: string;

supersede: boolean;

abortController?: AbortController;

}): number {

const existing = telegramReplyFenceByKey.get(params.key);

const state: TelegramReplyFenceState = existing ?? {

generation: 0,

activeDispatches: 0,

};

if (params.supersede) {

state.generation += 1;

abortTelegramReplyFenceControllers(state);

}

if (params.abortController) {

(state.abortControllers ??= new Set()).add(params.abortController);

}

state.activeDispatches += 1;

telegramReplyFenceByKey.set(params.key, state);

return state.generation;

}

function supersedeTelegramReplyFence(key: string): void {

const state = telegramReplyFenceByKey.get(key);

if (!state) {

return;

}

state.generation += 1;

abortTelegramReplyFenceControllers(state);

if (state.activeDispatches <= 0 && (state.abortControllers?.size ?? 0) === 0) {

telegramReplyFenceByKey.delete(key);

} else {

telegramReplyFenceByKey.set(key, state);

}

}

function isTelegramReplyFenceSuperseded(params: { key: string; generation: number }): boolean {

return (telegramReplyFenceByKey.get(params.key)?.generation ?? 0) !== params.generation;

}

function endTelegramReplyFence(key: string, abortController?: AbortController): void {

const state = telegramReplyFenceByKey.get(key);

if (!state) {

return;

}

if (abortController) {

state.abortControllers?.delete(abortController);

}

state.activeDispatches = Math.max(0, state.activeDispatches - 1);

if (state.activeDispatches <= 0 && (state.abortControllers?.size ?? 0) === 0) {

telegramReplyFenceByKey.delete(key);

}

}

function releaseTelegramReplyFenceAbortController(

key: string,

abortController?: AbortController,

): void {

if (!abortController) {

return;

}

const state = telegramReplyFenceByKey.get(key);

if (!state) {

return;

}

state.abortControllers?.delete(abortController);

if (state.activeDispatches <= 0 && (state.abortControllers?.size ?? 0) === 0) {

telegramReplyFenceByKey.delete(key);

}

}

function shouldSupersedeTelegramReplyFence(ctxPayload: {

Body?: string;

RawBody?: string;

CommandBody?: string;

CommandAuthorized: boolean;

}): boolean {

const dispatchText = ctxPayload.CommandBody ?? ctxPayload.RawBody ?? ctxPayload.Body ?? "";

return !isAbortRequestText(dispatchText) || ctxPayload.CommandAuthorized;

}

export function getTelegramReplyFenceSizeForTests(): number {

return telegramReplyFenceByKey.size;

}

export function resetTelegramReplyFenceForTests(): void {

telegramReplyFenceByKey.clear();

}

function resolveTelegramReasoningLevel(params: {

cfg: OpenClawConfig;

sessionKey?: string;

chatId,

threadSpec,

});

const replyFenceLaneKey = getTelegramSequentialKey({

message: msg,

...(context.primaryCtx.me ? { me: context.primaryCtx.me } : {}),

});

const scopedReplyFenceLaneKey = buildTelegramReplyFenceLaneKey({

accountId: route.accountId,

sequentialKey: replyFenceLaneKey,

});

let replyFenceGeneration: number | undefined;

const roomEventAbortController = isRoomEvent ? new AbortController() : undefined;

let roomEventAbortControllerQueued = false;

const replyAbortController = new AbortController();

let replyAbortControllerQueued = false;

let dispatchWasSuperseded = false;

const isDispatchSuperseded = () =>

replyFenceGeneration !== undefined &&

}

endTelegramReplyFence(

replyFenceKey.activeKey,

roomEventAbortControllerQueued ? undefined : roomEventAbortController,

replyAbortControllerQueued ? undefined : replyAbortController,

);

replyFenceGeneration = undefined;

};

replyFenceGeneration = beginTelegramReplyFence({

key: replyFenceKey.activeKey,

supersede: supersedeReplyFence,

abortController: roomEventAbortController,

abortController: replyAbortController,

laneKey: scopedReplyFenceLaneKey,

});

const implicitQuoteReplyTargetId =

replyOptions: {

skillFilter,

disableBlockStreaming,

abortSignal: roomEventAbortController?.signal,

abortSignal: replyAbortController.signal,

sourceReplyDeliveryMode: isRoomEvent ? "message_tool_only" : undefined,

queuedDeliveryCorrelations: isRoomEvent

? [{ begin: beginDeliveryCorrelation }]

: undefined,

queuedFollowupLifecycle:

isRoomEvent && roomEventAbortController

? {

onEnqueued: () => {

roomEventAbortControllerQueued = true;

},

onComplete: () => {

roomEventAbortControllerQueued = false;

releaseTelegramReplyFenceAbortController(

replyFenceKey.activeKey,

roomEventAbortController,

);

},

}

: undefined,

queuedFollowupLifecycle: isRoomEvent

? {

onEnqueued: () => {

replyAbortControllerQueued = true;

},

onComplete: () => {

replyAbortControllerQueued = false;

releaseTelegramReplyFenceAbortController(

replyFenceKey.activeKey,

replyAbortController,

);

},

}

: undefined,

suppressTyping: isRoomEvent,

onPartialReply:

answerLane.stream || reasoningLane.stream