mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-28 05:16:23 -06:00
refactor(telegram): split message dispatch (#108228)
* refactor(telegram): split message dispatch * fix(telegram): keep dispatch types internal * fix(telegram): satisfy dispatch lint
This commit is contained in:
Peter Steinberger
committed by
GitHub
GitHub
parent
b4d82f4729
commit
bb3fae8348
@@ -305,7 +305,6 @@ extensions/telegram/src/action-runtime.test.ts
|
||||
extensions/telegram/src/action-runtime.ts
|
||||
extensions/telegram/src/bot-message-context.session.ts
|
||||
extensions/telegram/src/bot-message-dispatch.test.ts
|
||||
extensions/telegram/src/bot-message-dispatch.ts
|
||||
extensions/telegram/src/bot-native-commands.session-meta.test.ts
|
||||
extensions/telegram/src/bot-native-commands.ts
|
||||
extensions/telegram/src/bot.create-telegram-bot.test.ts
|
||||
|
||||
@@ -0,0 +1,292 @@
|
||||
// Telegram plugin module recovers dispatch routing and group-history context.
|
||||
import { CURRENT_MESSAGE_MARKER } from "openclaw/plugin-sdk/channel-mention-gating";
|
||||
import { parseStrictPositiveInteger } from "openclaw/plugin-sdk/number-runtime";
|
||||
import { createChannelHistoryWindow } from "openclaw/plugin-sdk/reply-history";
|
||||
import { logVerbose } from "openclaw/plugin-sdk/runtime-env";
|
||||
import { withTelegramApiErrorLogging } from "./api-logging.js";
|
||||
import type { TelegramMessageContext } from "./bot-message-context.js";
|
||||
import {
|
||||
buildTelegramGroupFrom,
|
||||
buildTelegramGroupPeerId,
|
||||
buildTelegramInboundOriginTarget,
|
||||
buildTypingThreadParams,
|
||||
type TelegramThreadSpec,
|
||||
} from "./bot/helpers.js";
|
||||
import {
|
||||
isTelegramHistoryEntryAfterAmbientWatermark,
|
||||
mergeTelegramGroupHistoryPromptContext,
|
||||
retainTelegramGroupHistoryPromptContext,
|
||||
selectTelegramGroupHistoryAfterLastSelf,
|
||||
} from "./group-history-window.js";
|
||||
|
||||
const TELEGRAM_GENERAL_TOPIC_ID = 1;
|
||||
|
||||
function normalizeTelegramThreadId(value: unknown): number | undefined {
|
||||
return parseStrictPositiveInteger(value);
|
||||
}
|
||||
|
||||
function resolveTelegramForumThreadScopeFromSessionKey(
|
||||
sessionKey: unknown,
|
||||
): { chatId: string; threadId: number } | undefined {
|
||||
if (typeof sessionKey !== "string") {
|
||||
return undefined;
|
||||
}
|
||||
const match = /:telegram:group:(-?\d+):topic:(\d+)(?::|$)/.exec(sessionKey);
|
||||
const threadId = normalizeTelegramThreadId(match?.[2]);
|
||||
if (!match?.[1] || threadId == null) {
|
||||
return undefined;
|
||||
}
|
||||
return { chatId: match[1], threadId };
|
||||
}
|
||||
|
||||
function resolveDispatchTelegramThreadSpec(params: {
|
||||
chatId: TelegramMessageContext["chatId"];
|
||||
ctxPayload: TelegramMessageContext["ctxPayload"];
|
||||
threadSpec: TelegramThreadSpec;
|
||||
}): TelegramThreadSpec {
|
||||
if (
|
||||
params.threadSpec.scope !== "forum" ||
|
||||
(params.threadSpec.id != null && params.threadSpec.id !== TELEGRAM_GENERAL_TOPIC_ID)
|
||||
) {
|
||||
return params.threadSpec;
|
||||
}
|
||||
const scopedThread = resolveTelegramForumThreadScopeFromSessionKey(params.ctxPayload.SessionKey);
|
||||
const scopedThreadId =
|
||||
scopedThread?.chatId === String(params.chatId) ? scopedThread.threadId : undefined;
|
||||
const payloadThreadId =
|
||||
normalizeTelegramThreadId(params.ctxPayload.MessageThreadId) ??
|
||||
normalizeTelegramThreadId(params.ctxPayload.TransportThreadId);
|
||||
// Missing forum IDs are normalized to General; topic-scoped turn facts are more specific.
|
||||
const recoveredThreadId = scopedThreadId ?? payloadThreadId;
|
||||
return recoveredThreadId == null || recoveredThreadId === params.threadSpec.id
|
||||
? params.threadSpec
|
||||
: { ...params.threadSpec, id: recoveredThreadId };
|
||||
}
|
||||
|
||||
function normalizeDispatchTelegramThreadPayload(params: {
|
||||
context: TelegramMessageContext;
|
||||
threadSpec: TelegramThreadSpec;
|
||||
}): TelegramMessageContext {
|
||||
if (params.threadSpec.scope !== "forum" || params.threadSpec.id == null) {
|
||||
return params.context;
|
||||
}
|
||||
const messageThreadId = normalizeTelegramThreadId(params.context.ctxPayload.MessageThreadId);
|
||||
const transportThreadId = normalizeTelegramThreadId(params.context.ctxPayload.TransportThreadId);
|
||||
if (messageThreadId === params.threadSpec.id && transportThreadId === params.threadSpec.id) {
|
||||
return params.context;
|
||||
}
|
||||
return {
|
||||
...params.context,
|
||||
ctxPayload: {
|
||||
...params.context.ctxPayload,
|
||||
MessageThreadId: params.threadSpec.id,
|
||||
TransportThreadId: params.threadSpec.id,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function extractCurrentTelegramBody(body: string | undefined): string {
|
||||
if (!body) {
|
||||
return "";
|
||||
}
|
||||
const markerIndex = body.lastIndexOf(CURRENT_MESSAGE_MARKER);
|
||||
if (markerIndex === -1) {
|
||||
return body;
|
||||
}
|
||||
return body.slice(markerIndex + CURRENT_MESSAGE_MARKER.length).trimStart();
|
||||
}
|
||||
|
||||
function buildRecoveredTelegramChatActionSender(params: {
|
||||
context: TelegramMessageContext;
|
||||
threadId?: number;
|
||||
action: "typing" | "record_voice";
|
||||
}): () => Promise<void> {
|
||||
return async () => {
|
||||
try {
|
||||
await withTelegramApiErrorLogging({
|
||||
operation: "sendChatAction",
|
||||
fn: () =>
|
||||
params.context.sendChatActionHandler.sendChatAction(
|
||||
params.context.chatId,
|
||||
params.action,
|
||||
buildTypingThreadParams(params.threadId),
|
||||
),
|
||||
});
|
||||
} catch (err) {
|
||||
if (params.action !== "record_voice") {
|
||||
throw err;
|
||||
}
|
||||
logVerbose(
|
||||
`telegram record_voice cue failed for chat ${params.context.chatId}: ${String(err)}`,
|
||||
);
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
function migrateRecoveredTelegramGroupHistory(params: {
|
||||
context: TelegramMessageContext;
|
||||
recoveredHistoryKey?: string;
|
||||
}) {
|
||||
const originalHistoryKey = params.context.historyKey;
|
||||
const recoveredHistoryKey = params.recoveredHistoryKey;
|
||||
if (
|
||||
!params.context.isGroup ||
|
||||
!originalHistoryKey ||
|
||||
!recoveredHistoryKey ||
|
||||
originalHistoryKey === recoveredHistoryKey ||
|
||||
params.context.historyLimit <= 0
|
||||
) {
|
||||
return;
|
||||
}
|
||||
// Topic recovery mutates the raw in-memory buffer before any prompt is built;
|
||||
// prompt readers apply the ambient transcript watermark after recovery.
|
||||
const originalEntries = params.context.groupHistories.get(originalHistoryKey);
|
||||
if (!originalEntries?.length) {
|
||||
return;
|
||||
}
|
||||
const messageId = params.context.ctxPayload.MessageSid;
|
||||
const rawBody = params.context.ctxPayload.RawBody;
|
||||
const entryIndex = originalEntries.findLastIndex((entry) => {
|
||||
if (messageId && entry.messageId === messageId) {
|
||||
return true;
|
||||
}
|
||||
return !messageId && typeof rawBody === "string" && entry.body === rawBody;
|
||||
});
|
||||
if (entryIndex === -1) {
|
||||
return;
|
||||
}
|
||||
const [entry] = originalEntries.splice(entryIndex, 1);
|
||||
if (!entry) {
|
||||
return;
|
||||
}
|
||||
createChannelHistoryWindow({ historyMap: params.context.groupHistories }).record({
|
||||
historyKey: recoveredHistoryKey,
|
||||
limit: params.context.historyLimit,
|
||||
entry,
|
||||
});
|
||||
}
|
||||
|
||||
export function resolveDispatchTelegramContext(params: {
|
||||
context: TelegramMessageContext;
|
||||
}): TelegramMessageContext {
|
||||
const threadSpec = resolveDispatchTelegramThreadSpec({
|
||||
chatId: params.context.chatId,
|
||||
ctxPayload: params.context.ctxPayload,
|
||||
threadSpec: params.context.threadSpec,
|
||||
});
|
||||
if (threadSpec === params.context.threadSpec || threadSpec.scope !== "forum") {
|
||||
return normalizeDispatchTelegramThreadPayload({ context: params.context, threadSpec });
|
||||
}
|
||||
const recoveredRoutingTarget = buildTelegramInboundOriginTarget(
|
||||
params.context.chatId,
|
||||
threadSpec,
|
||||
);
|
||||
const recoveredFrom = params.context.isGroup
|
||||
? buildTelegramGroupFrom(params.context.chatId, threadSpec.id)
|
||||
: params.context.ctxPayload.From;
|
||||
const recoveredUpdateLastRoute =
|
||||
params.context.turn.record.updateLastRoute && threadSpec.id != null
|
||||
? {
|
||||
...params.context.turn.record.updateLastRoute,
|
||||
to: `telegram:${params.context.chatId}:topic:${threadSpec.id}`,
|
||||
threadId: String(threadSpec.id),
|
||||
}
|
||||
: params.context.turn.record.updateLastRoute;
|
||||
const recoveredHistoryKey = params.context.isGroup
|
||||
? buildTelegramGroupPeerId(params.context.chatId, threadSpec.id)
|
||||
: params.context.historyKey;
|
||||
const recoveredHistoryEntries =
|
||||
recoveredHistoryKey && params.context.historyLimit > 0
|
||||
? (params.context.groupHistories.get(recoveredHistoryKey) ?? [])
|
||||
.filter((entry) =>
|
||||
isTelegramHistoryEntryAfterAmbientWatermark(
|
||||
entry,
|
||||
params.context.ctxPayload.AmbientTranscriptPreviousMessageId
|
||||
? {
|
||||
messageId: params.context.ctxPayload.AmbientTranscriptPreviousMessageId,
|
||||
...(params.context.ctxPayload.AmbientTranscriptPreviousTimestampMs !== undefined
|
||||
? {
|
||||
timestampMs:
|
||||
params.context.ctxPayload.AmbientTranscriptPreviousTimestampMs,
|
||||
}
|
||||
: {}),
|
||||
}
|
||||
: undefined,
|
||||
),
|
||||
)
|
||||
.slice(-params.context.historyLimit)
|
||||
: [];
|
||||
const recoveredWatermarkedHistoryEntries = selectTelegramGroupHistoryAfterLastSelf(
|
||||
recoveredHistoryEntries,
|
||||
).slice(-params.context.historyLimit);
|
||||
const recoveredPromptHistoryEntries =
|
||||
params.context.isGroup && recoveredHistoryKey && params.context.historyLimit > 0
|
||||
? params.context.ctxPayload.InboundEventKind === "room_event"
|
||||
? recoveredHistoryEntries
|
||||
: recoveredWatermarkedHistoryEntries
|
||||
: [];
|
||||
const recoveredInboundHistory =
|
||||
params.context.isGroup && recoveredHistoryKey && params.context.historyLimit > 0
|
||||
? recoveredPromptHistoryEntries.length > 0
|
||||
? recoveredPromptHistoryEntries
|
||||
: undefined
|
||||
: params.context.ctxPayload.InboundHistory;
|
||||
const recoveredBodyForAgent = extractCurrentTelegramBody(
|
||||
params.context.ctxPayload.BodyForAgent ?? params.context.ctxPayload.Body,
|
||||
);
|
||||
const recoveredPromptContextBase = retainTelegramGroupHistoryPromptContext({
|
||||
promptContext: params.context.ctxPayload.UntrustedStructuredContext ?? [],
|
||||
entries: recoveredPromptHistoryEntries,
|
||||
});
|
||||
const recoveredPromptContext =
|
||||
recoveredPromptHistoryEntries.length > 0
|
||||
? mergeTelegramGroupHistoryPromptContext({
|
||||
promptContext: recoveredPromptContextBase ?? [],
|
||||
entries: recoveredPromptHistoryEntries,
|
||||
})
|
||||
: recoveredPromptContextBase?.length
|
||||
? recoveredPromptContextBase
|
||||
: undefined;
|
||||
const recoveredSendTyping = buildRecoveredTelegramChatActionSender({
|
||||
context: params.context,
|
||||
threadId: threadSpec.id,
|
||||
action: "typing",
|
||||
});
|
||||
const recoveredSendRecordVoice = buildRecoveredTelegramChatActionSender({
|
||||
context: params.context,
|
||||
threadId: threadSpec.id,
|
||||
action: "record_voice",
|
||||
});
|
||||
migrateRecoveredTelegramGroupHistory({ context: params.context, recoveredHistoryKey });
|
||||
return {
|
||||
...params.context,
|
||||
historyKey: recoveredHistoryKey,
|
||||
threadSpec,
|
||||
resolvedThreadId: threadSpec.id,
|
||||
replyThreadId: threadSpec.id,
|
||||
sendTyping: recoveredSendTyping,
|
||||
sendRecordVoice: recoveredSendRecordVoice,
|
||||
turn: {
|
||||
...params.context.turn,
|
||||
record: {
|
||||
...params.context.turn.record,
|
||||
updateLastRoute: recoveredUpdateLastRoute,
|
||||
},
|
||||
},
|
||||
ctxPayload:
|
||||
threadSpec.id == null
|
||||
? params.context.ctxPayload
|
||||
: {
|
||||
...params.context.ctxPayload,
|
||||
Body: recoveredBodyForAgent,
|
||||
BodyForAgent: recoveredBodyForAgent,
|
||||
From: recoveredFrom,
|
||||
InboundHistory: recoveredInboundHistory,
|
||||
MessageThreadId: threadSpec.id,
|
||||
OriginatingTo: recoveredRoutingTarget,
|
||||
To: recoveredRoutingTarget,
|
||||
TransportThreadId: threadSpec.id,
|
||||
UntrustedStructuredContext: recoveredPromptContext,
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,570 @@
|
||||
// Telegram plugin module owns final payload projection and Telegram delivery.
|
||||
import type { Bot } from "grammy";
|
||||
import type { Message } from "grammy/types";
|
||||
import {
|
||||
createOutboundPayloadPlan,
|
||||
deriveDurableFinalDeliveryRequirements,
|
||||
projectOutboundPayloadPlanForDelivery,
|
||||
resolveTranscriptBackedChannelFinalText,
|
||||
} from "openclaw/plugin-sdk/channel-outbound";
|
||||
import type {
|
||||
OpenClawConfig,
|
||||
ReplyToMode,
|
||||
TelegramAccountConfig,
|
||||
} from "openclaw/plugin-sdk/config-contracts";
|
||||
import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime";
|
||||
import type { ReplyPayload } from "openclaw/plugin-sdk/reply-payload";
|
||||
import { isSingleUseReplyToMode } from "openclaw/plugin-sdk/reply-reference";
|
||||
import type { RuntimeEnv } from "openclaw/plugin-sdk/runtime-env";
|
||||
import { logVerbose } from "openclaw/plugin-sdk/runtime-env";
|
||||
import type { TelegramBotDeps } from "./bot-deps.js";
|
||||
import type { TelegramMessageContext } from "./bot-message-context.js";
|
||||
import type { TelegramDraftController } from "./bot-message-dispatch-draft.js";
|
||||
import type { TelegramProgressController } from "./bot-message-dispatch-progress.js";
|
||||
import {
|
||||
mirrorTelegramAssistantReplyToTranscript,
|
||||
createCurrentTurnTranscriptFinalResolver,
|
||||
} from "./bot-message-dispatch-session.js";
|
||||
import type {
|
||||
CurrentTurnTranscriptFinal,
|
||||
FreshTelegramSessionEntryLoader,
|
||||
TelegramTranscriptMirrorPayload,
|
||||
} from "./bot-message-dispatch.types.js";
|
||||
import type { TelegramBotOptions } from "./bot.types.js";
|
||||
import { deliverReplies, emitInternalMessageSentHook } from "./bot/delivery.js";
|
||||
import type { TelegramThreadSpec } from "./bot/helpers.js";
|
||||
import { resolveTelegramReplyId } from "./bot/helpers.js";
|
||||
import type { TelegramNativeQuoteCandidateByMessageId } from "./bot/native-quote.js";
|
||||
import type { TelegramInlineButtons } from "./button-types.js";
|
||||
import { canonicalizeTelegramPresentationPayload } from "./interactive-fallback.js";
|
||||
import {
|
||||
createLaneDeliveryStateTracker,
|
||||
createLaneTextDeliverer,
|
||||
type LaneDeliveryResult,
|
||||
} from "./lane-delivery.js";
|
||||
import { recordOutboundMessageForPromptContext } from "./outbound-message-context.js";
|
||||
import {
|
||||
createTelegramPromptContextProjectionSequence,
|
||||
resolveTelegramPromptContextDeliverySignature,
|
||||
withTelegramPromptContextSource,
|
||||
type TelegramPromptContextProjection,
|
||||
type TelegramPromptContextProjectionSequence,
|
||||
type TelegramPromptContextSource,
|
||||
} from "./prompt-context-projection.js";
|
||||
import { editMessageTelegram } from "./send.js";
|
||||
|
||||
export function createTelegramDeliveryController(params: {
|
||||
bot: Bot;
|
||||
cfg: OpenClawConfig;
|
||||
chunkMode: ReturnType<typeof import("./bot-message-dispatch.runtime.js").resolveChunkMode>;
|
||||
context: TelegramMessageContext;
|
||||
dispatchStartedAt: number;
|
||||
draft: TelegramDraftController;
|
||||
isDispatchSuperseded: () => boolean;
|
||||
loadFreshSessionEntry: FreshTelegramSessionEntryLoader;
|
||||
mediaLocalRoots: readonly string[];
|
||||
opts: Pick<TelegramBotOptions, "token" | "mediaMaxMb">;
|
||||
progress: TelegramProgressController;
|
||||
draftReplyToMessageId?: number;
|
||||
replyQuoteByMessageId: TelegramNativeQuoteCandidateByMessageId;
|
||||
replyQuoteEntities?: Message["entities"];
|
||||
replyQuoteMessageId?: number;
|
||||
replyQuotePosition?: number;
|
||||
replyQuoteText?: string;
|
||||
replyToMode: ReplyToMode;
|
||||
runtime: RuntimeEnv;
|
||||
streamMode: import("./bot/types.js").TelegramStreamMode;
|
||||
tableMode: Parameters<typeof deliverReplies>[0]["tableMode"];
|
||||
telegramCfg: TelegramAccountConfig;
|
||||
telegramDeps: TelegramBotDeps;
|
||||
textLimit: number;
|
||||
threadSpec: TelegramThreadSpec;
|
||||
}) {
|
||||
const { context } = params;
|
||||
const sessionKey = context.ctxPayload.SessionKey;
|
||||
const deliveryState = createLaneDeliveryStateTracker();
|
||||
const resolveCurrentTurnTranscriptFinal = createCurrentTurnTranscriptFinalResolver({
|
||||
agentId: context.route.agentId,
|
||||
dispatchStartedAt: params.dispatchStartedAt,
|
||||
loadFreshSessionEntry: params.loadFreshSessionEntry,
|
||||
sessionKey,
|
||||
});
|
||||
let transcriptMirrorSequence = 0;
|
||||
const transcriptMirrorTurnId = `${context.chatId}:${context.ctxPayload.MessageSid ?? context.msg.message_id ?? params.dispatchStartedAt}`;
|
||||
const implicitQuoteReplyTargetId =
|
||||
context.ctxPayload.ReplyToIsQuote &&
|
||||
!context.msg.reply_to_message?.from?.is_bot &&
|
||||
params.replyQuoteMessageId != null
|
||||
? String(params.replyQuoteMessageId)
|
||||
: undefined;
|
||||
const currentMessageIdForQuoteReply =
|
||||
implicitQuoteReplyTargetId && context.ctxPayload.MessageSid
|
||||
? context.ctxPayload.MessageSid
|
||||
: undefined;
|
||||
|
||||
const projectPayloadForDelivery = (payload: ReplyPayload): ReplyPayload | undefined =>
|
||||
projectOutboundPayloadPlanForDelivery(
|
||||
createOutboundPayloadPlan([payload], {
|
||||
cfg: params.cfg,
|
||||
sessionKey,
|
||||
surface: "telegram",
|
||||
}),
|
||||
)[0];
|
||||
const promptContextDeliverySignature = (payload: ReplyPayload): string | undefined => {
|
||||
const projected = projectPayloadForDelivery(payload);
|
||||
return projected ? resolveTelegramPromptContextDeliverySignature(projected) : undefined;
|
||||
};
|
||||
const resolvePromptContextSource = (
|
||||
final: CurrentTurnTranscriptFinal | undefined,
|
||||
...payloads: ReplyPayload[]
|
||||
): TelegramPromptContextSource | undefined => {
|
||||
const finalSignature = final ? promptContextDeliverySignature({ text: final.text }) : undefined;
|
||||
if (!final?.messageId || !finalSignature) {
|
||||
return undefined;
|
||||
}
|
||||
return payloads.some((payload) => promptContextDeliverySignature(payload) === finalSignature)
|
||||
? { transcriptMessageId: final.messageId }
|
||||
: undefined;
|
||||
};
|
||||
const recordPromptContextMessage = (record: {
|
||||
messageId: number;
|
||||
message?: Message;
|
||||
text?: string;
|
||||
projection?: TelegramPromptContextProjection;
|
||||
}): Promise<boolean> =>
|
||||
(
|
||||
params.telegramDeps.recordOutboundMessageForPromptContext ??
|
||||
recordOutboundMessageForPromptContext
|
||||
)({
|
||||
cfg: params.cfg,
|
||||
account: {
|
||||
accountId: context.route.accountId,
|
||||
...(params.telegramCfg.name !== undefined ? { name: params.telegramCfg.name } : {}),
|
||||
...(context.primaryCtx.me ? { bot: context.primaryCtx.me } : {}),
|
||||
},
|
||||
...(context.primaryCtx.me?.id !== undefined ? { botUserId: context.primaryCtx.me.id } : {}),
|
||||
chatId: String(context.chatId),
|
||||
message: record.message ?? { message_id: record.messageId },
|
||||
messageId: record.messageId,
|
||||
...(record.text ? { text: record.text } : {}),
|
||||
...(record.projection ? { promptContextProjection: record.projection } : {}),
|
||||
...(params.threadSpec.id !== undefined ? { messageThreadId: params.threadSpec.id } : {}),
|
||||
});
|
||||
const createPromptContextSequence = (source?: TelegramPromptContextSource) =>
|
||||
createTelegramPromptContextProjectionSequence({
|
||||
...(source ? { source } : {}),
|
||||
record: recordPromptContextMessage,
|
||||
});
|
||||
const transcriptMirror = sessionKey
|
||||
? async (payload: TelegramTranscriptMirrorPayload) => {
|
||||
const idempotencyKey = `telegram-final:${sessionKey}:${transcriptMirrorTurnId}:${transcriptMirrorSequence++}`;
|
||||
await mirrorTelegramAssistantReplyToTranscript({
|
||||
cfg: params.cfg,
|
||||
idempotencyKey,
|
||||
loadFreshSessionEntry: params.loadFreshSessionEntry,
|
||||
route: context.route,
|
||||
sessionKey,
|
||||
payload,
|
||||
});
|
||||
}
|
||||
: undefined;
|
||||
const deliveryBaseOptions = {
|
||||
chatId: String(context.chatId),
|
||||
accountId: context.route.accountId,
|
||||
sessionKeyForInternalHooks: sessionKey,
|
||||
mirrorIsGroup: context.isGroup,
|
||||
mirrorGroupId: context.isGroup ? String(context.chatId) : undefined,
|
||||
token: params.opts.token,
|
||||
runtime: params.runtime,
|
||||
bot: params.bot,
|
||||
mediaLocalRoots: params.mediaLocalRoots,
|
||||
mediaMaxBytes: (params.opts.mediaMaxMb ?? params.telegramCfg.mediaMaxMb ?? 100) * 1024 * 1024,
|
||||
replyToMode: params.replyToMode,
|
||||
textLimit: params.textLimit,
|
||||
thread: params.threadSpec,
|
||||
tableMode: params.tableMode,
|
||||
chunkMode: params.chunkMode,
|
||||
richMessages: params.telegramCfg.richMessages,
|
||||
linkPreview: params.telegramCfg.linkPreview,
|
||||
replyQuoteMessageId: params.replyQuoteMessageId,
|
||||
replyQuoteText: params.replyQuoteText,
|
||||
replyQuotePosition: params.replyQuotePosition,
|
||||
replyQuoteEntities: params.replyQuoteEntities,
|
||||
replyQuoteByMessageId: params.replyQuoteByMessageId,
|
||||
transcriptMirror,
|
||||
};
|
||||
|
||||
const applyTextToPayload = (payload: ReplyPayload, text: string): ReplyPayload =>
|
||||
payload.text === text ? payload : { ...payload, text };
|
||||
const applyQuoteReplyTarget = (payload: ReplyPayload): ReplyPayload => {
|
||||
if (
|
||||
!implicitQuoteReplyTargetId ||
|
||||
!currentMessageIdForQuoteReply ||
|
||||
payload.replyToId !== currentMessageIdForQuoteReply ||
|
||||
payload.replyToTag ||
|
||||
payload.replyToCurrent
|
||||
) {
|
||||
return payload;
|
||||
}
|
||||
return { ...payload, replyToId: implicitQuoteReplyTargetId };
|
||||
};
|
||||
const usesNativeTelegramQuote = (payload: ReplyPayload): boolean =>
|
||||
params.replyQuoteText != null ||
|
||||
(payload.replyToId != null && params.replyQuoteByMessageId[payload.replyToId] != null);
|
||||
|
||||
const sendPayload = async (
|
||||
payload: ReplyPayload,
|
||||
options?: {
|
||||
afterAcceptedDraft?: boolean;
|
||||
durable?: boolean;
|
||||
silent?: boolean;
|
||||
mirrorTranscript?: boolean;
|
||||
promptContextSequence?: TelegramPromptContextProjectionSequence;
|
||||
textMode?: "html";
|
||||
},
|
||||
) => {
|
||||
if (params.isDispatchSuperseded()) {
|
||||
await options?.promptContextSequence?.fail();
|
||||
return false;
|
||||
}
|
||||
const targetedPayload = applyQuoteReplyTarget(payload);
|
||||
const finalReplyTargetId = resolveTelegramReplyId(targetedPayload.replyToId);
|
||||
const targetsDifferentMessage =
|
||||
finalReplyTargetId != null && finalReplyTargetId !== params.draftReplyToMessageId;
|
||||
const consumedSingleUseReply =
|
||||
options?.afterAcceptedDraft === true &&
|
||||
isSingleUseReplyToMode(params.replyToMode) &&
|
||||
!targetsDifferentMessage;
|
||||
const deliverablePayload = consumedSingleUseReply
|
||||
? (({ replyToId: _, replyToTag: _tag, replyToCurrent: _current, ...rest }) => rest)(
|
||||
targetedPayload,
|
||||
)
|
||||
: targetedPayload;
|
||||
const effectiveReplyToMode = consumedSingleUseReply ? "off" : params.replyToMode;
|
||||
const projectionSequence =
|
||||
options?.promptContextSequence ??
|
||||
createPromptContextSequence(
|
||||
options?.durable
|
||||
? resolvePromptContextSource(
|
||||
await resolveCurrentTurnTranscriptFinal(),
|
||||
deliverablePayload,
|
||||
)
|
||||
: undefined,
|
||||
);
|
||||
const effectivePayload = withTelegramPromptContextSource(
|
||||
deliverablePayload,
|
||||
projectionSequence.source,
|
||||
);
|
||||
const silent =
|
||||
options?.silent ??
|
||||
(params.telegramCfg.silentErrorReplies === true && payload.isError === true);
|
||||
const durableDelivery = params.telegramDeps.deliverInboundReplyWithMessageSendContext;
|
||||
if (options?.durable && durableDelivery && projectionSequence.isFresh()) {
|
||||
const durable = await durableDelivery({
|
||||
cfg: params.cfg,
|
||||
channel: "telegram",
|
||||
to: String(context.chatId),
|
||||
accountId: context.route.accountId,
|
||||
agentId: context.route.agentId,
|
||||
ctxPayload: context.ctxPayload,
|
||||
payload: effectivePayload,
|
||||
info: { kind: "final" },
|
||||
replyToMode: effectiveReplyToMode,
|
||||
threadId: params.threadSpec.id,
|
||||
formatting: {
|
||||
textLimit: params.textLimit,
|
||||
tableMode: params.tableMode,
|
||||
chunkMode: params.chunkMode,
|
||||
...(options?.textMode === "html" ? { parseMode: "HTML" as const } : {}),
|
||||
},
|
||||
silent,
|
||||
requiredCapabilities: deriveDurableFinalDeliveryRequirements({
|
||||
payload: effectivePayload,
|
||||
replyToId: effectivePayload.replyToId,
|
||||
threadId: params.threadSpec.id,
|
||||
silent,
|
||||
payloadTransport: true,
|
||||
extraCapabilities: {
|
||||
nativeQuote: !consumedSingleUseReply && usesNativeTelegramQuote(effectivePayload),
|
||||
},
|
||||
}),
|
||||
});
|
||||
if (durable.status === "failed") {
|
||||
await projectionSequence.fail();
|
||||
throw durable.error;
|
||||
}
|
||||
if (durable.status === "handled_visible") {
|
||||
deliveryState.markDelivered();
|
||||
return true;
|
||||
}
|
||||
if (durable.status === "handled_no_send") {
|
||||
await projectionSequence.fail();
|
||||
return false;
|
||||
}
|
||||
}
|
||||
try {
|
||||
const result = await (params.telegramDeps.deliverReplies ?? deliverReplies)({
|
||||
...deliveryBaseOptions,
|
||||
replyToMode: effectiveReplyToMode,
|
||||
transcriptMirror:
|
||||
options?.durable && options?.mirrorTranscript !== false ? transcriptMirror : undefined,
|
||||
replies: [effectivePayload],
|
||||
onVoiceRecording: context.sendRecordVoice,
|
||||
silent,
|
||||
mediaLoader: params.telegramDeps.loadWebMedia,
|
||||
promptContextSequence: projectionSequence,
|
||||
...(options?.textMode ? { textMode: options.textMode } : {}),
|
||||
});
|
||||
if (!result.delivered) {
|
||||
await projectionSequence.fail();
|
||||
return false;
|
||||
}
|
||||
await projectionSequence.finish();
|
||||
deliveryState.markDelivered();
|
||||
return true;
|
||||
} catch (error) {
|
||||
await projectionSequence.fail();
|
||||
throw error;
|
||||
}
|
||||
};
|
||||
|
||||
const emitPreviewFinalizedHook = async (result: LaneDeliveryResult) => {
|
||||
if (params.isDispatchSuperseded() || result.kind !== "preview-finalized") {
|
||||
return;
|
||||
}
|
||||
(params.telegramDeps.emitInternalMessageSentHook ?? emitInternalMessageSentHook)({
|
||||
sessionKeyForInternalHooks: sessionKey,
|
||||
chatId: String(context.chatId),
|
||||
accountId: context.route.accountId,
|
||||
content: result.delivery.content,
|
||||
success: true,
|
||||
messageId: result.delivery.messageId,
|
||||
isGroup: context.isGroup,
|
||||
groupId: context.isGroup ? String(context.chatId) : undefined,
|
||||
});
|
||||
if (transcriptMirror && result.delivery.content) {
|
||||
void transcriptMirror({ text: result.delivery.content }).catch((err: unknown) => {
|
||||
logVerbose(
|
||||
`telegram preview-finalized transcriptMirror failed: ${formatErrorMessage(err)}`,
|
||||
);
|
||||
});
|
||||
}
|
||||
};
|
||||
const deliverLaneText = createLaneTextDeliverer({
|
||||
lanes: params.draft.lanes,
|
||||
applyTextToPayload,
|
||||
sendPayload,
|
||||
flushDraftLane: params.draft.flushLane,
|
||||
stopDraftLane: async (lane) => await lane.stream?.stop(),
|
||||
clearDraftLane: async (lane) => await lane.stream?.clear(),
|
||||
editStreamMessage: async ({ messageId, text, textMode, buttons }) => {
|
||||
if (!params.isDispatchSuperseded()) {
|
||||
await (params.telegramDeps.editMessageTelegram ?? editMessageTelegram)(
|
||||
context.chatId,
|
||||
messageId,
|
||||
text,
|
||||
{
|
||||
api: params.bot.api,
|
||||
cfg: params.cfg,
|
||||
accountId: context.route.accountId,
|
||||
linkPreview: params.telegramCfg.linkPreview,
|
||||
textMode,
|
||||
buttons,
|
||||
},
|
||||
);
|
||||
}
|
||||
},
|
||||
createPromptContextSequence,
|
||||
resolveFinalTextCandidate: async () => (await resolveCurrentTurnTranscriptFinal())?.text,
|
||||
log: logVerbose,
|
||||
markDelivered: deliveryState.markDelivered,
|
||||
});
|
||||
|
||||
const materializeAnswerLaneBeforeRotation = async () => {
|
||||
const block = params.draft.activeAnswerBlockDelivery();
|
||||
const lane = params.draft.answerLane;
|
||||
if (
|
||||
!block ||
|
||||
!lane.stream ||
|
||||
!lane.hasStreamedMessage ||
|
||||
lane.finalized ||
|
||||
params.draft.isAnswerToolProgressOnly()
|
||||
) {
|
||||
return;
|
||||
}
|
||||
const text = lane.lastPartialText || params.draft.lastAnswerPartialText() || block.text;
|
||||
if (!text?.trim()) {
|
||||
return;
|
||||
}
|
||||
const result = await deliverLaneText({
|
||||
laneName: "answer",
|
||||
text,
|
||||
payload: block.payload,
|
||||
infoKind: "block",
|
||||
buttons: block.buttons,
|
||||
finalizePreview: true,
|
||||
durable: false,
|
||||
});
|
||||
params.draft.setActiveAnswerBlockDelivery();
|
||||
await emitPreviewFinalizedHook(result);
|
||||
};
|
||||
params.draft.setMaterializeBeforeRotation(materializeAnswerLaneBeforeRotation);
|
||||
|
||||
const postCosmeticSummaryBar = async (line: string) => {
|
||||
try {
|
||||
await sendPayload({ text: line }, { durable: true, mirrorTranscript: false });
|
||||
} catch (err) {
|
||||
logVerbose(`telegram: collapse summary bar send failed: ${formatErrorMessage(err)}`);
|
||||
}
|
||||
};
|
||||
const deliverProgressCollapseSummary = async () => {
|
||||
const line = params.progress.resolveCollapseSummaryLine();
|
||||
if (line) {
|
||||
await postCosmeticSummaryBar(line);
|
||||
}
|
||||
};
|
||||
const deliverProgressModeFinalAnswer = async (
|
||||
payload: ReplyPayload,
|
||||
text: string,
|
||||
promptContextSequence: TelegramPromptContextProjectionSequence,
|
||||
): Promise<LaneDeliveryResult> => {
|
||||
const afterAcceptedDraft = params.draft.answerLane.stream?.hasConsumedReplyTarget?.() === true;
|
||||
if (payload.isError === true) {
|
||||
params.progress.setSummaryDelivered();
|
||||
await params.progress.teardownWindow();
|
||||
const delivered = await sendPayload(applyTextToPayload(payload, text), {
|
||||
afterAcceptedDraft,
|
||||
durable: true,
|
||||
promptContextSequence,
|
||||
});
|
||||
if (!delivered) {
|
||||
return { kind: "skipped" };
|
||||
}
|
||||
params.draft.answerLane.finalized = true;
|
||||
params.progress.markFinalDelivered();
|
||||
return { kind: "sent" };
|
||||
}
|
||||
const barLine = params.progress.resolveCollapseSummaryLine();
|
||||
const delivered = await sendPayload(applyTextToPayload(payload, text), {
|
||||
afterAcceptedDraft,
|
||||
durable: true,
|
||||
promptContextSequence,
|
||||
});
|
||||
if (barLine) {
|
||||
await params.progress.applyCollapseSummary(barLine, postCosmeticSummaryBar);
|
||||
params.progress.resetAnswerLaneAfterCollapse();
|
||||
} else {
|
||||
await params.progress.teardownWindow();
|
||||
}
|
||||
if (!delivered) {
|
||||
return { kind: "skipped" };
|
||||
}
|
||||
params.draft.answerLane.finalized = true;
|
||||
params.progress.markFinalDelivered();
|
||||
return { kind: "sent" };
|
||||
};
|
||||
const deliverFinalAnswerText = async (
|
||||
answerPayload: ReplyPayload,
|
||||
text: string,
|
||||
buttons?: TelegramInlineButtons,
|
||||
): Promise<LaneDeliveryResult> => {
|
||||
const transcriptFinal = await resolveCurrentTurnTranscriptFinal();
|
||||
const finalText = await resolveTranscriptBackedChannelFinalText({
|
||||
finalText: text,
|
||||
resolveCandidateText: async () => transcriptFinal?.text,
|
||||
});
|
||||
const source = resolvePromptContextSource(
|
||||
transcriptFinal,
|
||||
answerPayload,
|
||||
applyTextToPayload(answerPayload, finalText),
|
||||
);
|
||||
const promptContextSequence = createPromptContextSequence(source);
|
||||
const isFollowUp = params.progress.finalAnswerDelivered();
|
||||
let result: LaneDeliveryResult;
|
||||
if (!isFollowUp && params.streamMode === "progress") {
|
||||
result = await deliverProgressModeFinalAnswer(
|
||||
answerPayload,
|
||||
finalText,
|
||||
promptContextSequence,
|
||||
);
|
||||
} else {
|
||||
if (isFollowUp) {
|
||||
await params.draft.prepareAnswerLaneForText();
|
||||
} else if (!(await params.draft.rotateAnswerLaneAfterToolProgress())) {
|
||||
await params.draft.rotateAnswerLaneAfterQueuedBlocksSettle();
|
||||
}
|
||||
result = await deliverLaneText({
|
||||
laneName: "answer",
|
||||
text: finalText,
|
||||
payload: answerPayload,
|
||||
infoKind: "final",
|
||||
buttons,
|
||||
allowStream: !usesNativeTelegramQuote(answerPayload),
|
||||
promptContextSequence,
|
||||
});
|
||||
if (!isFollowUp && result.kind !== "skipped") {
|
||||
params.progress.markFinalDelivered();
|
||||
}
|
||||
}
|
||||
if (result.kind === "preview-finalized") {
|
||||
await emitPreviewFinalizedHook(result);
|
||||
}
|
||||
return result;
|
||||
};
|
||||
const finalizePendingAnswerBlockDraft = async (final: {
|
||||
queuedFinal: boolean;
|
||||
dispatchError?: unknown;
|
||||
}) => {
|
||||
const block = params.draft.activeAnswerBlockDelivery();
|
||||
if (
|
||||
!block ||
|
||||
final.queuedFinal ||
|
||||
final.dispatchError ||
|
||||
params.isDispatchSuperseded() ||
|
||||
params.draft.answerLane.finalized
|
||||
) {
|
||||
return;
|
||||
}
|
||||
const content = block.text.trimEnd();
|
||||
if (!content) {
|
||||
return;
|
||||
}
|
||||
params.progress.markFinalStarted();
|
||||
await deliverFinalAnswerText(block.payload, content, block.buttons);
|
||||
params.draft.setActiveAnswerBlockDelivery();
|
||||
};
|
||||
|
||||
return {
|
||||
applyTextToPayload,
|
||||
createPromptContextSequence,
|
||||
deliverFallback: async (replies: ReplyPayload[], silent: boolean) =>
|
||||
await (params.telegramDeps.deliverReplies ?? deliverReplies)({
|
||||
replies,
|
||||
...deliveryBaseOptions,
|
||||
silent,
|
||||
mediaLoader: params.telegramDeps.loadWebMedia,
|
||||
}),
|
||||
deliverFinalAnswerText,
|
||||
deliverLaneText,
|
||||
deliverProgressCollapseSummary,
|
||||
emitPreviewFinalizedHook,
|
||||
finalizePendingAnswerBlockDraft,
|
||||
markDelivered: deliveryState.markDelivered,
|
||||
markNonSilentFailure: deliveryState.markNonSilentFailure,
|
||||
markNonSilentSkip: deliveryState.markNonSilentSkip,
|
||||
normalizeDeliveryPayload: (payload: ReplyPayload): ReplyPayload | undefined => {
|
||||
const keepReasoningLane =
|
||||
payload.isReasoning === true && params.draft.durableReasoningPayloadsEnabled;
|
||||
const payloadForPlan = keepReasoningLane ? { ...payload } : payload;
|
||||
if (keepReasoningLane) {
|
||||
delete payloadForPlan.isReasoning;
|
||||
}
|
||||
const normalized = projectPayloadForDelivery(payloadForPlan);
|
||||
return normalized ? canonicalizeTelegramPresentationPayload(normalized) : undefined;
|
||||
},
|
||||
sendPayload,
|
||||
snapshot: deliveryState.snapshot,
|
||||
};
|
||||
}
|
||||
|
||||
export type TelegramDeliveryController = ReturnType<typeof createTelegramDeliveryController>;
|
||||
@@ -0,0 +1,505 @@
|
||||
// Telegram plugin module owns answer/reasoning draft lanes and rotation state.
|
||||
import type { Bot } from "grammy";
|
||||
import { resolveChannelStreamingBlockEnabled } from "openclaw/plugin-sdk/channel-outbound";
|
||||
import type {
|
||||
OpenClawConfig,
|
||||
ReplyToMode,
|
||||
TelegramAccountConfig,
|
||||
} from "openclaw/plugin-sdk/config-contracts";
|
||||
import type { ReplyPayload } from "openclaw/plugin-sdk/reply-payload";
|
||||
import type { BlockReplyContext } from "openclaw/plugin-sdk/reply-runtime";
|
||||
import { logVerbose } from "openclaw/plugin-sdk/runtime-env";
|
||||
import type { TelegramBotDeps } from "./bot-deps.js";
|
||||
import { resolveMarkdownTableMode } from "./bot-message-dispatch.runtime.js";
|
||||
import type {
|
||||
TelegramReasoningLevel,
|
||||
TelegramAnswerBlockDelivery,
|
||||
} from "./bot-message-dispatch.types.js";
|
||||
import type { TelegramThreadSpec } from "./bot/helpers.js";
|
||||
import type { TelegramStreamMode } from "./bot/types.js";
|
||||
import { resolveTelegramDraftStreamingChunking } from "./draft-chunking.js";
|
||||
import { createTelegramDraftStream, type TelegramDraftPreview } from "./draft-stream.js";
|
||||
import { renderTelegramHtmlText } from "./format.js";
|
||||
import type { DraftLaneState, LaneName } from "./lane-delivery.js";
|
||||
import { TELEGRAM_TEXT_CHUNK_LIMIT } from "./outbound-adapter.js";
|
||||
import { splitTelegramReasoningText } from "./reasoning-lane-coordinator.js";
|
||||
import { buildTelegramRichMarkdown, TELEGRAM_RICH_TEXT_LIMIT } from "./rich-message.js";
|
||||
|
||||
const DRAFT_MIN_INITIAL_CHARS = 30;
|
||||
|
||||
type DraftPartialTextUpdate = {
|
||||
text: string;
|
||||
delta?: string;
|
||||
replace?: true;
|
||||
isReasoningSnapshot?: boolean;
|
||||
};
|
||||
|
||||
type SplitLaneSegment = { lane: LaneName; update: DraftPartialTextUpdate };
|
||||
type SplitLaneSegmentsResult = {
|
||||
segments: SplitLaneSegment[];
|
||||
suppressedReasoningOnly: boolean;
|
||||
};
|
||||
|
||||
type QueuedAnswerBlockRotation = {
|
||||
assistantMessageIndex?: number;
|
||||
text?: string;
|
||||
shouldRotateBeforeDelivery: boolean;
|
||||
};
|
||||
|
||||
function resolveDraftPartialText(
|
||||
previous: string,
|
||||
update: DraftPartialTextUpdate,
|
||||
): string | undefined {
|
||||
const nextText =
|
||||
update.replace || update.isReasoningSnapshot || update.delta === undefined
|
||||
? update.text
|
||||
: `${previous}${update.delta}`;
|
||||
return nextText === previous ? undefined : nextText;
|
||||
}
|
||||
|
||||
export function createTelegramDraftController(params: {
|
||||
accountId: string;
|
||||
bot: Bot;
|
||||
cfg: OpenClawConfig;
|
||||
chatId: number;
|
||||
draftReplyToMessageId?: number;
|
||||
forceBlockStreamingForReasoning: boolean;
|
||||
hasTelegramQuoteReply: boolean;
|
||||
isDispatchSuperseded: () => boolean;
|
||||
isRoomEvent: boolean;
|
||||
replyToMode: ReplyToMode;
|
||||
resolvedReasoningLevel: TelegramReasoningLevel;
|
||||
streamMode: TelegramStreamMode;
|
||||
tableMode: ReturnType<typeof resolveMarkdownTableMode>;
|
||||
telegramCfg: TelegramAccountConfig;
|
||||
telegramDeps: TelegramBotDeps;
|
||||
textLimit: number;
|
||||
threadSpec: TelegramThreadSpec;
|
||||
}) {
|
||||
const streamDeliveryEnabled = !params.isRoomEvent && params.streamMode !== "off";
|
||||
const accountBlockStreamingEnabled =
|
||||
resolveChannelStreamingBlockEnabled(params.telegramCfg) ??
|
||||
params.cfg.agents?.defaults?.blockStreamingDefault === "on";
|
||||
const canStreamAnswerDraft =
|
||||
streamDeliveryEnabled &&
|
||||
!params.hasTelegramQuoteReply &&
|
||||
!accountBlockStreamingEnabled &&
|
||||
!params.forceBlockStreamingForReasoning;
|
||||
const streamReasoningDraft = params.resolvedReasoningLevel === "stream";
|
||||
const streamReasoningInProgressDraft =
|
||||
streamReasoningDraft && params.streamMode === "progress" && canStreamAnswerDraft;
|
||||
const canStreamReasoningDraft =
|
||||
!params.isRoomEvent && streamReasoningDraft && !streamReasoningInProgressDraft;
|
||||
const draftMaxChars =
|
||||
params.streamMode === "block"
|
||||
? Math.min(
|
||||
resolveTelegramDraftStreamingChunking(params.cfg, params.accountId).maxChars,
|
||||
params.textLimit,
|
||||
)
|
||||
: Math.min(
|
||||
params.textLimit,
|
||||
params.telegramCfg.richMessages === true
|
||||
? TELEGRAM_RICH_TEXT_LIMIT
|
||||
: TELEGRAM_TEXT_CHUNK_LIMIT,
|
||||
);
|
||||
const renderStreamText = (text: string): TelegramDraftPreview =>
|
||||
params.telegramCfg.richMessages === true
|
||||
? {
|
||||
text,
|
||||
richMessage: buildTelegramRichMarkdown(text, {
|
||||
tableMode: params.tableMode,
|
||||
skipEntityDetection: params.telegramCfg.linkPreview === false,
|
||||
}),
|
||||
}
|
||||
: {
|
||||
text: renderTelegramHtmlText(text, { tableMode: params.tableMode }),
|
||||
parseMode: "HTML",
|
||||
markdownSource: { text, tableMode: params.tableMode },
|
||||
};
|
||||
|
||||
const createDraftLane = (laneName: LaneName, enabled: boolean): DraftLaneState => {
|
||||
const stream = enabled
|
||||
? (params.telegramDeps.createTelegramDraftStream ?? createTelegramDraftStream)({
|
||||
api: params.bot.api,
|
||||
chatId: params.chatId,
|
||||
maxChars: draftMaxChars,
|
||||
thread: params.threadSpec,
|
||||
replyToMessageId: params.draftReplyToMessageId,
|
||||
replyToMode: params.replyToMode,
|
||||
richMessages: params.telegramCfg.richMessages,
|
||||
minInitialChars: params.streamMode === "progress" ? 0 : DRAFT_MIN_INITIAL_CHARS,
|
||||
renderText: renderStreamText,
|
||||
onRetainedPage: (page) => {
|
||||
lanes[laneName].retainedPromptContextPages.push({
|
||||
messageId: page.messageId,
|
||||
text: page.textSnapshot,
|
||||
});
|
||||
},
|
||||
log: logVerbose,
|
||||
warn: logVerbose,
|
||||
})
|
||||
: undefined;
|
||||
return {
|
||||
stream,
|
||||
lastPartialText: "",
|
||||
hasStreamedMessage: false,
|
||||
finalized: false,
|
||||
retainedPromptContextPages: [],
|
||||
};
|
||||
};
|
||||
const lanes: Record<LaneName, DraftLaneState> = {
|
||||
answer: createDraftLane("answer", canStreamAnswerDraft),
|
||||
reasoning: createDraftLane("reasoning", canStreamReasoningDraft),
|
||||
};
|
||||
const answerLane = lanes.answer;
|
||||
const reasoningLane = lanes.reasoning;
|
||||
let lastAnswerPartialText = "";
|
||||
let activeAnswerDraftIsToolProgressOnly = false;
|
||||
let activeAnswerBlockAssistantMessageIndex: number | undefined;
|
||||
let activeAnswerBlockDelivery: TelegramAnswerBlockDelivery | undefined;
|
||||
let materializeAnswerLaneBeforeRotation: (() => Promise<void>) | undefined;
|
||||
const queuedAnswerBlockRotations: QueuedAnswerBlockRotation[] = [];
|
||||
let queuedAnswerBlockAssistantMessageIndex: number | undefined;
|
||||
let pendingAnswerBlockAssistantMessageIndex: number | undefined;
|
||||
let rotateAnswerLaneWhenQueuedBlocksSettle = false;
|
||||
let eventQueue = Promise.resolve();
|
||||
let resetProgress = () => {};
|
||||
let suppressProgress = () => {};
|
||||
let noteReasoningHint = () => {};
|
||||
let noteReasoningDelivered = () => {};
|
||||
|
||||
const resetAnswerToolProgressDraft = () => {
|
||||
activeAnswerDraftIsToolProgressOnly = false;
|
||||
};
|
||||
const resetLaneState = (lane: DraftLaneState) => {
|
||||
lane.lastPartialText = "";
|
||||
if (lane === answerLane) {
|
||||
lastAnswerPartialText = "";
|
||||
}
|
||||
lane.hasStreamedMessage = false;
|
||||
lane.finalized = false;
|
||||
lane.retainedPromptContextPages = [];
|
||||
if (lane === answerLane) {
|
||||
resetAnswerToolProgressDraft();
|
||||
pendingAnswerBlockAssistantMessageIndex = undefined;
|
||||
activeAnswerBlockDelivery = undefined;
|
||||
}
|
||||
};
|
||||
const repositionLaneForNewMessage = (lane: DraftLaneState) => {
|
||||
// Reposition instead of delete-then-repost: the replacement must land
|
||||
// before deferred cleanup or Telegram can jump and retain a stale preview.
|
||||
lane.stream?.rotateToNewMessageDeferringDelete();
|
||||
resetLaneState(lane);
|
||||
};
|
||||
const rotateLaneForNewMessage = async (lane: DraftLaneState) => {
|
||||
if (!lane.hasStreamedMessage && typeof lane.stream?.messageId() !== "number") {
|
||||
resetLaneState(lane);
|
||||
return;
|
||||
}
|
||||
await lane.stream?.stop();
|
||||
lane.stream?.forceNewMessage();
|
||||
resetLaneState(lane);
|
||||
};
|
||||
const rotateAnswerLaneForNewMessage = async () => {
|
||||
await materializeAnswerLaneBeforeRotation?.();
|
||||
await rotateLaneForNewMessage(answerLane);
|
||||
};
|
||||
const rotateAnswerLaneAfterToolProgress = async () => {
|
||||
if (!activeAnswerDraftIsToolProgressOnly) {
|
||||
return false;
|
||||
}
|
||||
repositionLaneForNewMessage(answerLane);
|
||||
suppressProgress();
|
||||
rotateAnswerLaneWhenQueuedBlocksSettle = false;
|
||||
return true;
|
||||
};
|
||||
const rotateAnswerLaneAfterQueuedBlocksSettle = async () => {
|
||||
if (!rotateAnswerLaneWhenQueuedBlocksSettle || queuedAnswerBlockRotations.length > 0) {
|
||||
return false;
|
||||
}
|
||||
rotateAnswerLaneWhenQueuedBlocksSettle = false;
|
||||
if (!answerLane.hasStreamedMessage || activeAnswerDraftIsToolProgressOnly) {
|
||||
return false;
|
||||
}
|
||||
await rotateAnswerLaneForNewMessage();
|
||||
return true;
|
||||
};
|
||||
const prepareAnswerLaneForText = async (): Promise<boolean> => {
|
||||
// Progress mode owns one stationary activity window; answer text never rotates it.
|
||||
if (params.streamMode === "progress") {
|
||||
return false;
|
||||
}
|
||||
if (await rotateAnswerLaneAfterToolProgress()) {
|
||||
return true;
|
||||
}
|
||||
if (await rotateAnswerLaneAfterQueuedBlocksSettle()) {
|
||||
return true;
|
||||
}
|
||||
if (!answerLane.finalized) {
|
||||
return false;
|
||||
}
|
||||
answerLane.stream?.forceNewMessage();
|
||||
resetLaneState(answerLane);
|
||||
rotateAnswerLaneWhenQueuedBlocksSettle = false;
|
||||
return true;
|
||||
};
|
||||
const prepareAnswerLaneForToolProgress = async () => {
|
||||
if (answerLane.finalized) {
|
||||
answerLane.stream?.forceNewMessage();
|
||||
resetLaneState(answerLane);
|
||||
}
|
||||
if (activeAnswerDraftIsToolProgressOnly) {
|
||||
return;
|
||||
}
|
||||
if (params.streamMode !== "progress" && answerLane.hasStreamedMessage) {
|
||||
await rotateAnswerLaneForNewMessage();
|
||||
}
|
||||
activeAnswerDraftIsToolProgressOnly = true;
|
||||
};
|
||||
|
||||
const splitTextIntoLaneSegments = (
|
||||
update: { text?: string; delta?: string; replace?: true; isReasoningSnapshot?: boolean },
|
||||
isReasoning?: boolean,
|
||||
): SplitLaneSegmentsResult => {
|
||||
const split = splitTelegramReasoningText(update.text, isReasoning);
|
||||
const splitSegments: Array<{ lane: LaneName; text: string }> = [];
|
||||
const useDelta =
|
||||
!update.replace && update.isReasoningSnapshot !== true && update.delta !== undefined;
|
||||
const suppressReasoning = params.resolvedReasoningLevel === "off";
|
||||
if (split.reasoningText && !suppressReasoning) {
|
||||
splitSegments.push({ lane: "reasoning", text: split.reasoningText });
|
||||
}
|
||||
if (split.answerText) {
|
||||
splitSegments.push({ lane: "answer", text: split.answerText });
|
||||
}
|
||||
return {
|
||||
segments: splitSegments.map((segment) => ({
|
||||
lane: segment.lane,
|
||||
update: {
|
||||
text: segment.text,
|
||||
...(!useDelta || splitSegments.length !== 1 ? {} : { delta: update.delta }),
|
||||
...(update.replace ? { replace: true as const } : {}),
|
||||
...(update.isReasoningSnapshot ? { isReasoningSnapshot: true } : {}),
|
||||
},
|
||||
})),
|
||||
suppressedReasoningOnly:
|
||||
Boolean(split.reasoningText) && suppressReasoning && !split.answerText,
|
||||
};
|
||||
};
|
||||
const updateDraftFromPartial = (lane: DraftLaneState, update: DraftPartialTextUpdate) => {
|
||||
if (!lane.stream || !update.text) {
|
||||
return;
|
||||
}
|
||||
const previousText = lane === answerLane ? lastAnswerPartialText : lane.lastPartialText;
|
||||
const nextText = resolveDraftPartialText(previousText, update);
|
||||
if (!nextText || (lane === answerLane && params.streamMode === "progress")) {
|
||||
return;
|
||||
}
|
||||
if (lane === answerLane) {
|
||||
resetAnswerToolProgressDraft();
|
||||
suppressProgress();
|
||||
lastAnswerPartialText = nextText;
|
||||
}
|
||||
lane.hasStreamedMessage = true;
|
||||
lane.finalized = false;
|
||||
lane.lastPartialText = nextText;
|
||||
lane.stream.update(nextText);
|
||||
};
|
||||
const ingestDraftLaneSegments = async (
|
||||
update: { text?: string; delta?: string; replace?: true; isReasoningSnapshot?: boolean },
|
||||
isReasoning?: boolean,
|
||||
) => {
|
||||
const split = splitTextIntoLaneSegments(update, isReasoning);
|
||||
for (const segment of split.segments) {
|
||||
if (segment.lane === "answer") {
|
||||
await prepareAnswerLaneForText();
|
||||
}
|
||||
if (segment.lane === "reasoning") {
|
||||
noteReasoningHint();
|
||||
noteReasoningDelivered();
|
||||
}
|
||||
updateDraftFromPartial(lanes[segment.lane], segment.update);
|
||||
}
|
||||
};
|
||||
const enqueueEvent = (task: () => Promise<void>): Promise<void> => {
|
||||
const next = eventQueue.then(async () => {
|
||||
if (!params.isDispatchSuperseded()) {
|
||||
await task();
|
||||
}
|
||||
});
|
||||
eventQueue = next.catch((err: unknown) => {
|
||||
logVerbose(`telegram: draft lane callback failed: ${String(err)}`);
|
||||
});
|
||||
return eventQueue;
|
||||
};
|
||||
|
||||
const recomputeQueuedAnswerBlockRotations = () => {
|
||||
let previous =
|
||||
activeAnswerBlockAssistantMessageIndex ?? pendingAnswerBlockAssistantMessageIndex;
|
||||
queuedAnswerBlockAssistantMessageIndex = undefined;
|
||||
for (const entry of queuedAnswerBlockRotations) {
|
||||
if (entry.assistantMessageIndex === undefined) {
|
||||
continue;
|
||||
}
|
||||
entry.shouldRotateBeforeDelivery =
|
||||
previous !== undefined && entry.assistantMessageIndex !== previous;
|
||||
previous = entry.assistantMessageIndex;
|
||||
queuedAnswerBlockAssistantMessageIndex = entry.assistantMessageIndex;
|
||||
}
|
||||
};
|
||||
const rotationMatches = (
|
||||
entry: QueuedAnswerBlockRotation,
|
||||
payload: ReplyPayload,
|
||||
assistantMessageIndex?: number,
|
||||
) =>
|
||||
assistantMessageIndex !== undefined && entry.assistantMessageIndex !== undefined
|
||||
? assistantMessageIndex === entry.assistantMessageIndex
|
||||
: entry.text !== undefined && payload.text !== undefined && entry.text === payload.text;
|
||||
const prepareQueuedAnswerBlock = async (
|
||||
payload: ReplyPayload,
|
||||
blockContext?: BlockReplyContext,
|
||||
) => {
|
||||
if (
|
||||
!splitTextIntoLaneSegments({ text: payload.text }, payload.isReasoning).segments.some(
|
||||
(segment) => segment.lane === "answer",
|
||||
)
|
||||
) {
|
||||
return;
|
||||
}
|
||||
resetProgress();
|
||||
const assistantMessageIndex = blockContext?.assistantMessageIndex;
|
||||
if (assistantMessageIndex === undefined) {
|
||||
queuedAnswerBlockRotations.push({ text: payload.text, shouldRotateBeforeDelivery: false });
|
||||
return;
|
||||
}
|
||||
const previous =
|
||||
queuedAnswerBlockAssistantMessageIndex ??
|
||||
activeAnswerBlockAssistantMessageIndex ??
|
||||
pendingAnswerBlockAssistantMessageIndex;
|
||||
queuedAnswerBlockRotations.push({
|
||||
assistantMessageIndex,
|
||||
text: payload.text,
|
||||
shouldRotateBeforeDelivery: previous !== undefined && assistantMessageIndex !== previous,
|
||||
});
|
||||
queuedAnswerBlockAssistantMessageIndex = assistantMessageIndex;
|
||||
};
|
||||
const takeQueuedAnswerBlockRotation = (payload: ReplyPayload, index?: number): boolean => {
|
||||
if (queuedAnswerBlockRotations.length === 0) {
|
||||
return false;
|
||||
}
|
||||
const matchIndex = queuedAnswerBlockRotations.findIndex((entry) =>
|
||||
rotationMatches(entry, payload, index),
|
||||
);
|
||||
const matched = queuedAnswerBlockRotations.splice(0, Math.max(matchIndex, 0) + 1).at(-1);
|
||||
if (matched?.assistantMessageIndex !== undefined) {
|
||||
activeAnswerBlockAssistantMessageIndex = matched.assistantMessageIndex;
|
||||
pendingAnswerBlockAssistantMessageIndex = undefined;
|
||||
}
|
||||
recomputeQueuedAnswerBlockRotations();
|
||||
return matched?.shouldRotateBeforeDelivery ?? false;
|
||||
};
|
||||
const dropQueuedAnswerBlockRotation = (payload: ReplyPayload, index?: number) => {
|
||||
let matchIndex = queuedAnswerBlockRotations.findIndex((entry) =>
|
||||
rotationMatches(entry, payload, index),
|
||||
);
|
||||
if (matchIndex < 0 && index === undefined) {
|
||||
matchIndex = queuedAnswerBlockRotations.findIndex(
|
||||
(entry) => entry.assistantMessageIndex === undefined,
|
||||
);
|
||||
}
|
||||
if (matchIndex < 0) {
|
||||
return;
|
||||
}
|
||||
const [matched] = queuedAnswerBlockRotations.splice(matchIndex, 1);
|
||||
if (
|
||||
matchIndex === 0 &&
|
||||
matched?.assistantMessageIndex !== undefined &&
|
||||
rotateAnswerLaneWhenQueuedBlocksSettle &&
|
||||
activeAnswerBlockAssistantMessageIndex === undefined &&
|
||||
answerLane.hasStreamedMessage
|
||||
) {
|
||||
pendingAnswerBlockAssistantMessageIndex = matched.assistantMessageIndex;
|
||||
}
|
||||
recomputeQueuedAnswerBlockRotations();
|
||||
};
|
||||
|
||||
const resolvedBlockStreamingEnabled = resolveChannelStreamingBlockEnabled(params.telegramCfg);
|
||||
const disableBlockStreaming = !streamDeliveryEnabled
|
||||
? true
|
||||
: params.forceBlockStreamingForReasoning
|
||||
? false
|
||||
: typeof resolvedBlockStreamingEnabled === "boolean"
|
||||
? !resolvedBlockStreamingEnabled
|
||||
: canStreamAnswerDraft
|
||||
? true
|
||||
: undefined;
|
||||
|
||||
return {
|
||||
answerLane,
|
||||
reasoningLane,
|
||||
lanes,
|
||||
canPushAnswerDraft: () => Boolean(answerLane.stream),
|
||||
cleanup: async (superseded: boolean) => {
|
||||
for (const lane of [answerLane, reasoningLane]) {
|
||||
const stream = lane.stream;
|
||||
if (!stream) {
|
||||
continue;
|
||||
}
|
||||
if (superseded) {
|
||||
await (typeof stream.discard === "function" ? stream.discard() : stream.stop());
|
||||
} else if (lane.finalized) {
|
||||
await stream.stop();
|
||||
} else {
|
||||
await stream.clear();
|
||||
}
|
||||
}
|
||||
},
|
||||
disableBlockStreaming,
|
||||
durableReasoningPayloadsEnabled:
|
||||
params.resolvedReasoningLevel === "on" || Boolean(reasoningLane.stream),
|
||||
enqueueEvent,
|
||||
ingestDraftLaneSegments,
|
||||
isAnswerToolProgressOnly: () => activeAnswerDraftIsToolProgressOnly,
|
||||
isQueuedAnswerBlock: (payload: ReplyPayload, index?: number) =>
|
||||
queuedAnswerBlockRotations.some((entry) => rotationMatches(entry, payload, index)),
|
||||
lastAnswerPartialText: () => lastAnswerPartialText,
|
||||
prepareAnswerLaneForText,
|
||||
prepareAnswerLaneForToolProgress,
|
||||
prepareQueuedAnswerBlock,
|
||||
dropQueuedAnswerBlockRotation,
|
||||
takeQueuedAnswerBlockRotation,
|
||||
renderStreamText,
|
||||
repositionLaneForNewMessage,
|
||||
resetAnswerToolProgressDraft,
|
||||
resetLaneState,
|
||||
rotateAnswerLaneAfterQueuedBlocksSettle,
|
||||
rotateAnswerLaneAfterToolProgress,
|
||||
rotateAnswerLaneForNewMessage,
|
||||
rotateLaneForNewMessage,
|
||||
setActiveAnswerBlockDelivery: (delivery?: TelegramAnswerBlockDelivery) => {
|
||||
activeAnswerBlockDelivery = delivery;
|
||||
},
|
||||
activeAnswerBlockDelivery: () => activeAnswerBlockDelivery,
|
||||
setMaterializeBeforeRotation: (materialize: () => Promise<void>) => {
|
||||
materializeAnswerLaneBeforeRotation = materialize;
|
||||
},
|
||||
setProgressLifecycle: (lifecycle: { reset: () => void; suppress: () => void }) => {
|
||||
resetProgress = lifecycle.reset;
|
||||
suppressProgress = lifecycle.suppress;
|
||||
},
|
||||
setReasoningStepCallbacks: (callbacks: { noteHint: () => void; noteDelivered: () => void }) => {
|
||||
noteReasoningHint = callbacks.noteHint;
|
||||
noteReasoningDelivered = callbacks.noteDelivered;
|
||||
},
|
||||
setRotateWhenQueuedBlocksSettle: (value: boolean) => {
|
||||
rotateAnswerLaneWhenQueuedBlocksSettle = value;
|
||||
},
|
||||
splitTextIntoLaneSegments,
|
||||
streamDeliveryEnabled,
|
||||
streamReasoningInProgressDraft,
|
||||
waitForEvents: async () => await eventQueue,
|
||||
flushLane: async (lane: DraftLaneState) => await lane.stream?.flush(),
|
||||
};
|
||||
}
|
||||
|
||||
export type TelegramDraftController = ReturnType<typeof createTelegramDraftController>;
|
||||
@@ -0,0 +1,117 @@
|
||||
// Telegram plugin module owns pre-adoption reply-fence authority.
|
||||
import type { TelegramMessageContext } from "./bot-message-context.js";
|
||||
import type { DispatchTelegramMessageParams } from "./bot-message-dispatch.types.js";
|
||||
import { getTelegramSequentialKey } from "./sequential-key.js";
|
||||
import {
|
||||
beginTelegramReplyFence,
|
||||
buildTelegramNonInterruptingReplyFenceKey,
|
||||
buildTelegramReplyFenceLaneKey,
|
||||
endTelegramReplyFence,
|
||||
isTelegramReplyFenceSuperseded,
|
||||
releaseTelegramReplyFenceAbortController,
|
||||
resolveTelegramReplyFenceKey,
|
||||
shouldSupersedeTelegramReplyFence,
|
||||
supersedeTelegramReplyFence,
|
||||
} from "./telegram-reply-fence.js";
|
||||
|
||||
type CreateTelegramReplyFenceParams = Pick<
|
||||
DispatchTelegramMessageParams,
|
||||
"onTurnAdopted" | "onTurnDeferred" | "onTurnAbandoned" | "turnAbortSignal"
|
||||
> & {
|
||||
context: TelegramMessageContext;
|
||||
};
|
||||
|
||||
export function createTelegramReplyFenceController(params: CreateTelegramReplyFenceParams) {
|
||||
const { context } = params;
|
||||
const replyFenceKey = resolveTelegramReplyFenceKey({
|
||||
ctxPayload: context.ctxPayload,
|
||||
chatId: context.chatId,
|
||||
threadSpec: context.threadSpec,
|
||||
});
|
||||
const sequentialKey = getTelegramSequentialKey({
|
||||
message: context.msg,
|
||||
...(context.primaryCtx.me ? { me: context.primaryCtx.me } : {}),
|
||||
});
|
||||
const laneKey = buildTelegramReplyFenceLaneKey({
|
||||
accountId: context.route.accountId,
|
||||
sequentialKey,
|
||||
});
|
||||
const supersedes = shouldSupersedeTelegramReplyFence(context.ctxPayload);
|
||||
const activeKey = supersedes
|
||||
? replyFenceKey.activeKey
|
||||
: buildTelegramNonInterruptingReplyFenceKey({
|
||||
activeKey: replyFenceKey.activeKey,
|
||||
laneKey,
|
||||
});
|
||||
// Ambient room-event work uses a separate fence key. Any non-room-event
|
||||
// inbound may cancel it without owning abort authority over adopted user turns.
|
||||
if (context.ctxPayload.InboundEventKind !== "room_event") {
|
||||
supersedeTelegramReplyFence(replyFenceKey.roomEventKey);
|
||||
}
|
||||
|
||||
const abortController = new AbortController();
|
||||
const abortSignal = params.turnAbortSignal
|
||||
? AbortSignal.any([abortController.signal, params.turnAbortSignal])
|
||||
: abortController.signal;
|
||||
let generation: number | undefined = beginTelegramReplyFence({
|
||||
key: activeKey,
|
||||
supersede: supersedes,
|
||||
abortController,
|
||||
laneKey,
|
||||
});
|
||||
let abortControllerQueued = false;
|
||||
let queuedTurnAdmitted = false;
|
||||
|
||||
const isSuperseded = () =>
|
||||
abortController.signal.aborted ||
|
||||
(generation !== undefined && isTelegramReplyFenceSuperseded({ key: activeKey, generation }));
|
||||
|
||||
const release = () => {
|
||||
if (generation === undefined) {
|
||||
return;
|
||||
}
|
||||
endTelegramReplyFence(activeKey, abortControllerQueued ? undefined : abortController);
|
||||
generation = undefined;
|
||||
};
|
||||
|
||||
const adoptTurn = async () => {
|
||||
await params.onTurnAdopted?.();
|
||||
// Fence abort and supersession authority end after durable adoption.
|
||||
// Core then owns all interruption of the adopted run.
|
||||
release();
|
||||
releaseTelegramReplyFenceAbortController(activeKey, abortController);
|
||||
};
|
||||
|
||||
return {
|
||||
abortSignal,
|
||||
adoptTurn,
|
||||
generation: () => generation,
|
||||
isSuperseded,
|
||||
release,
|
||||
queuedFollowupLifecycle:
|
||||
context.ctxPayload.InboundEventKind === "room_event" ||
|
||||
params.onTurnAdopted ||
|
||||
params.onTurnDeferred ||
|
||||
params.onTurnAbandoned
|
||||
? {
|
||||
onEnqueued: () => {
|
||||
abortControllerQueued = true;
|
||||
params.onTurnDeferred?.();
|
||||
},
|
||||
onAdmitted: async () => {
|
||||
await adoptTurn();
|
||||
queuedTurnAdmitted = true;
|
||||
},
|
||||
onComplete: () => {
|
||||
abortControllerQueued = false;
|
||||
releaseTelegramReplyFenceAbortController(activeKey, abortController);
|
||||
if (!queuedTurnAdmitted) {
|
||||
params.onTurnAbandoned?.();
|
||||
}
|
||||
},
|
||||
}
|
||||
: undefined,
|
||||
};
|
||||
}
|
||||
|
||||
export type TelegramReplyFenceController = ReturnType<typeof createTelegramReplyFenceController>;
|
||||
@@ -0,0 +1,360 @@
|
||||
// Telegram plugin module owns the ephemeral progress window and collapse summary.
|
||||
import {
|
||||
buildChannelProgressDraftLine,
|
||||
buildChannelProgressDraftLineForEntry,
|
||||
createChannelProgressDraftCompositor,
|
||||
isChannelProgressDraftWorkToolName,
|
||||
resolveChannelStreamingPreviewToolProgress,
|
||||
type ChannelProgressDraftLine,
|
||||
} from "openclaw/plugin-sdk/channel-outbound";
|
||||
import type { TelegramAccountConfig } from "openclaw/plugin-sdk/config-contracts";
|
||||
import type { TelegramBotDeps } from "./bot-deps.js";
|
||||
import type { TelegramMessageContext } from "./bot-message-context.js";
|
||||
import type { TelegramDraftController } from "./bot-message-dispatch-draft.js";
|
||||
import type { TelegramStreamMode } from "./bot/types.js";
|
||||
import {
|
||||
formatTelegramProgressLine,
|
||||
renderTelegramProgressDraftPreview,
|
||||
} from "./progress-draft-preview.js";
|
||||
import {
|
||||
createTelegramProgressSummaryTracker,
|
||||
formatTelegramProgressSummaryLine,
|
||||
} from "./progress-summary.js";
|
||||
|
||||
type BufferedDispatchParams = Parameters<
|
||||
TelegramBotDeps["dispatchReplyWithBufferedBlockDispatcher"]
|
||||
>[0];
|
||||
type ReplyOptions = NonNullable<BufferedDispatchParams["replyOptions"]>;
|
||||
type CallbackPayload<K extends keyof ReplyOptions> =
|
||||
NonNullable<ReplyOptions[K]> extends (...args: infer Args) => unknown ? Args[0] : never;
|
||||
|
||||
function buildTelegramThinkingProgressLine(progressTokens: number): ChannelProgressDraftLine {
|
||||
const label = `Thinking… (~${Math.round(progressTokens)} tokens)`;
|
||||
return {
|
||||
id: "reasoning:token-progress",
|
||||
kind: "item",
|
||||
icon: "🧠",
|
||||
label,
|
||||
text: `🧠 ${label}`,
|
||||
prefix: false,
|
||||
};
|
||||
}
|
||||
|
||||
export function createTelegramProgressController(params: {
|
||||
accountId: string;
|
||||
chatId: TelegramMessageContext["chatId"];
|
||||
draft: TelegramDraftController;
|
||||
statusReactionController: TelegramMessageContext["statusReactionController"];
|
||||
streamMode: TelegramStreamMode;
|
||||
streamReasoningInProgressDraft: boolean;
|
||||
telegramCfg: TelegramAccountConfig;
|
||||
threadId?: number;
|
||||
}) {
|
||||
const { answerLane } = params.draft;
|
||||
const summaryStartedAt = Date.now();
|
||||
const summary = createTelegramProgressSummaryTracker();
|
||||
let summaryDelivered = false;
|
||||
let draftEverRendered = false;
|
||||
let finalAnswerDeliveryStarted = false;
|
||||
let finalAnswerDelivered = false;
|
||||
let sawProgressFinal = false;
|
||||
let verboseProgressActive: () => boolean = () => false;
|
||||
|
||||
const compositor = createChannelProgressDraftCompositor({
|
||||
entry: params.telegramCfg,
|
||||
mode: params.streamMode,
|
||||
active: Boolean(answerLane.stream),
|
||||
seed: `${params.accountId}:${params.chatId}:${params.threadId ?? ""}`,
|
||||
formatLine: (text) => (compositor.hasStatusHeadline ? text : formatTelegramProgressLine(text)),
|
||||
reasoningGate: params.streamReasoningInProgressDraft,
|
||||
reasoningLinePrefix: "🧠 ",
|
||||
commentaryLinePrefix: "💬 ",
|
||||
commentaryItalics: false,
|
||||
update: async (streamText, options) => {
|
||||
draftEverRendered = true;
|
||||
await params.draft.prepareAnswerLaneForToolProgress();
|
||||
answerLane.lastPartialText = streamText;
|
||||
answerLane.hasStreamedMessage = true;
|
||||
answerLane.finalized = false;
|
||||
answerLane.stream?.updatePreview(
|
||||
renderTelegramProgressDraftPreview(
|
||||
streamText,
|
||||
options?.lines ?? [],
|
||||
params.telegramCfg.richMessages === true,
|
||||
compositor.hasStatusHeadline,
|
||||
),
|
||||
);
|
||||
if (options?.flush) {
|
||||
await answerLane.stream?.flush();
|
||||
}
|
||||
},
|
||||
});
|
||||
|
||||
params.draft.setProgressLifecycle({
|
||||
reset: () => compositor.reset(),
|
||||
suppress: () => compositor.suppress(),
|
||||
});
|
||||
|
||||
const canPushToolProgress = () =>
|
||||
Boolean(
|
||||
answerLane.stream &&
|
||||
!verboseProgressActive() &&
|
||||
!answerLane.finalized &&
|
||||
!finalAnswerDeliveryStarted &&
|
||||
!finalAnswerDelivered,
|
||||
);
|
||||
const pushToolProgress = async (
|
||||
line?: string | ChannelProgressDraftLine,
|
||||
options?: { toolName?: string; startImmediately?: boolean },
|
||||
) => {
|
||||
if (!canPushToolProgress()) {
|
||||
return false;
|
||||
}
|
||||
return await compositor.pushToolProgress(line, options);
|
||||
};
|
||||
const pushReasoningProgress = async (payload: {
|
||||
text?: string;
|
||||
isReasoningSnapshot?: boolean;
|
||||
}) => {
|
||||
if (params.streamReasoningInProgressDraft && payload.text) {
|
||||
summary.noteReasoningActivity();
|
||||
}
|
||||
return await compositor.pushReasoningProgress(payload.text, {
|
||||
snapshot: payload.isReasoningSnapshot === true,
|
||||
});
|
||||
};
|
||||
const pushThinkingTokenProgress = async (progressTokens: number) => {
|
||||
const rendered = await pushToolProgress(buildTelegramThinkingProgressLine(progressTokens), {
|
||||
startImmediately: true,
|
||||
});
|
||||
if (rendered) {
|
||||
summary.noteReasoningActivity();
|
||||
}
|
||||
return rendered;
|
||||
};
|
||||
|
||||
const markFinalStarted = () => {
|
||||
finalAnswerDeliveryStarted = true;
|
||||
compositor.markFinalReplyStarted();
|
||||
};
|
||||
const markFinalDelivered = () => {
|
||||
finalAnswerDelivered = true;
|
||||
sawProgressFinal = true;
|
||||
compositor.markFinalReplyDelivered();
|
||||
};
|
||||
const resolveCollapseSummaryLine = (): string | undefined => {
|
||||
if (summaryDelivered) {
|
||||
return undefined;
|
||||
}
|
||||
summaryDelivered = true;
|
||||
if (!draftEverRendered) {
|
||||
return undefined;
|
||||
}
|
||||
return (
|
||||
formatTelegramProgressSummaryLine(summary.counts(), Date.now() - summaryStartedAt) ||
|
||||
undefined
|
||||
);
|
||||
};
|
||||
const applyCollapseSummary = async (
|
||||
line: string,
|
||||
postCosmeticSummary: (line: string) => Promise<void>,
|
||||
) => {
|
||||
const messageId = await answerLane.stream?.finalizeToPreview(
|
||||
params.draft.renderStreamText(line),
|
||||
);
|
||||
if (typeof messageId !== "number") {
|
||||
await postCosmeticSummary(line);
|
||||
}
|
||||
};
|
||||
const resetAnswerLaneAfterCollapse = () => {
|
||||
if (params.draft.isAnswerToolProgressOnly()) {
|
||||
params.draft.resetAnswerToolProgressDraft();
|
||||
compositor.suppress();
|
||||
params.draft.setRotateWhenQueuedBlocksSettle(false);
|
||||
}
|
||||
answerLane.stream?.forceNewMessage();
|
||||
params.draft.resetLaneState(answerLane);
|
||||
};
|
||||
const teardownWindow = async () => {
|
||||
if (params.draft.isAnswerToolProgressOnly()) {
|
||||
await params.draft.rotateAnswerLaneAfterToolProgress();
|
||||
return;
|
||||
}
|
||||
await answerLane.stream?.clear();
|
||||
params.draft.resetLaneState(answerLane);
|
||||
};
|
||||
|
||||
const handleToolStart = async (payload: CallbackPayload<"onToolStart">) => {
|
||||
const toolName = payload.name?.trim();
|
||||
if (payload.phase === "start") {
|
||||
const windowRendersTool =
|
||||
canPushToolProgress() &&
|
||||
resolveChannelStreamingPreviewToolProgress(params.telegramCfg) &&
|
||||
isChannelProgressDraftWorkToolName(toolName);
|
||||
if (windowRendersTool) {
|
||||
summary.noteToolCall();
|
||||
} else {
|
||||
summary.closeReasoningBurst();
|
||||
summary.closeCommentaryBurst();
|
||||
}
|
||||
}
|
||||
const progressPromise = pushToolProgress(
|
||||
buildChannelProgressDraftLineForEntry(
|
||||
params.telegramCfg,
|
||||
{
|
||||
event: "tool",
|
||||
itemId: payload.itemId,
|
||||
toolCallId: payload.toolCallId,
|
||||
name: toolName,
|
||||
phase: payload.phase,
|
||||
args: payload.args,
|
||||
},
|
||||
payload.detailMode ? { detailMode: payload.detailMode } : undefined,
|
||||
),
|
||||
{ toolName, startImmediately: true },
|
||||
);
|
||||
if (params.statusReactionController && toolName) {
|
||||
await params.statusReactionController.setTool(toolName);
|
||||
}
|
||||
await progressPromise;
|
||||
};
|
||||
const handleItemEvent = async (payload: CallbackPayload<"onItemEvent">) => {
|
||||
if (payload.kind === "preamble") {
|
||||
if (verboseProgressActive()) {
|
||||
return;
|
||||
}
|
||||
if (params.streamMode === "progress") {
|
||||
await compositor.pushPreambleHeadline(payload.progressText, { itemId: payload.itemId });
|
||||
}
|
||||
if (params.streamMode === "progress" && compositor.commentaryProgressEnabled) {
|
||||
const accepted = await compositor.pushCommentaryProgress(payload.progressText, {
|
||||
itemId: payload.itemId,
|
||||
});
|
||||
if (accepted) {
|
||||
summary.noteCommentary(payload.itemId, payload.progressText);
|
||||
}
|
||||
}
|
||||
return;
|
||||
}
|
||||
await pushToolProgress(
|
||||
buildChannelProgressDraftLineForEntry(params.telegramCfg, {
|
||||
event: "item",
|
||||
itemId: payload.itemId,
|
||||
toolCallId: payload.toolCallId,
|
||||
itemKind: payload.kind,
|
||||
title: payload.title,
|
||||
name: payload.name,
|
||||
phase: payload.phase,
|
||||
status: payload.status,
|
||||
summary: payload.summary,
|
||||
progressText: payload.progressText,
|
||||
meta: payload.meta,
|
||||
}),
|
||||
);
|
||||
};
|
||||
const handlePlanUpdate = async (payload: CallbackPayload<"onPlanUpdate">) => {
|
||||
if (payload.phase === "update") {
|
||||
await pushToolProgress(
|
||||
buildChannelProgressDraftLine({
|
||||
event: "plan",
|
||||
phase: payload.phase,
|
||||
title: payload.title,
|
||||
explanation: payload.explanation,
|
||||
steps: payload.steps,
|
||||
}),
|
||||
);
|
||||
}
|
||||
};
|
||||
const handleApprovalEvent = async (payload: CallbackPayload<"onApprovalEvent">) => {
|
||||
if (payload.phase === "requested") {
|
||||
await pushToolProgress(
|
||||
buildChannelProgressDraftLine({
|
||||
event: "approval",
|
||||
phase: payload.phase,
|
||||
title: payload.title,
|
||||
command: payload.command,
|
||||
reason: payload.reason,
|
||||
message: payload.message,
|
||||
}),
|
||||
);
|
||||
}
|
||||
};
|
||||
const handleCommandOutput = async (payload: CallbackPayload<"onCommandOutput">) => {
|
||||
if (payload.phase === "end") {
|
||||
await pushToolProgress(
|
||||
buildChannelProgressDraftLineForEntry(params.telegramCfg, {
|
||||
event: "command-output",
|
||||
itemId: payload.itemId,
|
||||
toolCallId: payload.toolCallId,
|
||||
phase: payload.phase,
|
||||
title: payload.title,
|
||||
name: payload.name,
|
||||
status: payload.status,
|
||||
exitCode: payload.exitCode,
|
||||
}),
|
||||
);
|
||||
}
|
||||
};
|
||||
const handlePatchSummary = async (payload: CallbackPayload<"onPatchSummary">) => {
|
||||
if (payload.phase === "end") {
|
||||
await pushToolProgress(
|
||||
buildChannelProgressDraftLine({
|
||||
event: "patch",
|
||||
itemId: payload.itemId,
|
||||
toolCallId: payload.toolCallId,
|
||||
phase: payload.phase,
|
||||
title: payload.title,
|
||||
name: payload.name,
|
||||
added: payload.added,
|
||||
modified: payload.modified,
|
||||
deleted: payload.deleted,
|
||||
summary: payload.summary,
|
||||
}),
|
||||
);
|
||||
}
|
||||
};
|
||||
|
||||
return {
|
||||
applyCollapseSummary,
|
||||
canPushToolProgress,
|
||||
cancel: () => compositor.cancel(),
|
||||
closeReasoningBurst: () => summary.closeReasoningBurst(),
|
||||
commentaryProgressEnabled: compositor.commentaryProgressEnabled,
|
||||
finalAnswerDelivered: () => finalAnswerDelivered,
|
||||
finalAnswerDeliveryStarted: () => finalAnswerDeliveryStarted,
|
||||
handleApprovalEvent,
|
||||
handleCommandOutput,
|
||||
handleItemEvent,
|
||||
handlePatchSummary,
|
||||
handlePlanUpdate,
|
||||
handleToolStart,
|
||||
markFinalDelivered,
|
||||
markFinalStarted,
|
||||
markSawFinal: () => {
|
||||
sawProgressFinal = true;
|
||||
},
|
||||
progressPreambleEnabled:
|
||||
params.streamMode === "progress" && answerLane.stream ? true : undefined,
|
||||
pushReasoningProgress,
|
||||
pushThinkingTokenProgress,
|
||||
pushToolProgress,
|
||||
reset: () => compositor.reset(),
|
||||
resetAnswerLaneAfterCollapse,
|
||||
resolveCollapseSummaryLine,
|
||||
sawProgressFinal: () => sawProgressFinal,
|
||||
setFinalAnswerDelivered: (value: boolean) => {
|
||||
finalAnswerDelivered = value;
|
||||
},
|
||||
setSummaryDelivered: () => {
|
||||
summaryDelivered = true;
|
||||
},
|
||||
setVerboseProgressActive: (isActive: () => boolean) => {
|
||||
verboseProgressActive = isActive;
|
||||
},
|
||||
suppress: () => compositor.suppress(),
|
||||
teardownWindow,
|
||||
verboseProgressActive: () => verboseProgressActive(),
|
||||
};
|
||||
}
|
||||
|
||||
export type TelegramProgressController = ReturnType<typeof createTelegramProgressController>;
|
||||
@@ -0,0 +1,417 @@
|
||||
import type { OpenClawConfig, TelegramAccountConfig } from "openclaw/plugin-sdk/config-contracts";
|
||||
// Telegram plugin module owns buffered reply payload delivery decisions.
|
||||
import { normalizeMessagePresentation } from "openclaw/plugin-sdk/interactive-runtime";
|
||||
import {
|
||||
isFastModeAutoProgressPayload,
|
||||
isReplyPayloadNonTerminalToolErrorWarning,
|
||||
resolveSendableOutboundReplyParts,
|
||||
} from "openclaw/plugin-sdk/reply-payload";
|
||||
import type { ReplyPayload } from "openclaw/plugin-sdk/reply-payload";
|
||||
import type { RuntimeEnv } from "openclaw/plugin-sdk/runtime-env";
|
||||
import { danger } from "openclaw/plugin-sdk/runtime-env";
|
||||
import type { TelegramBotDeps } from "./bot-deps.js";
|
||||
import type { TelegramMessageContext } from "./bot-message-context.js";
|
||||
import type { TelegramDeliveryController } from "./bot-message-dispatch-delivery.js";
|
||||
import type { TelegramDraftController } from "./bot-message-dispatch-draft.js";
|
||||
import type { TelegramReplyFenceController } from "./bot-message-dispatch-fence.js";
|
||||
import type { TelegramProgressController } from "./bot-message-dispatch-progress.js";
|
||||
import { deduplicateBlockSentMedia } from "./bot-message-dispatch.media-dedup.js";
|
||||
import type { TelegramDispatchTurnState } from "./bot-message-dispatch.types.js";
|
||||
import type { TelegramStreamMode } from "./bot/types.js";
|
||||
import { resolveTelegramInlineButtons, type TelegramInlineButtons } from "./button-types.js";
|
||||
import {
|
||||
buildTelegramErrorScopeKey,
|
||||
isSilentErrorPolicy,
|
||||
resolveTelegramErrorPolicy,
|
||||
shouldSuppressTelegramError,
|
||||
} from "./error-policy.js";
|
||||
import { shouldSuppressLocalTelegramExecApprovalPrompt } from "./exec-approvals.js";
|
||||
import { createTelegramReasoningStepState } from "./reasoning-lane-coordinator.js";
|
||||
|
||||
type BufferedDispatchParams = Parameters<
|
||||
TelegramBotDeps["dispatchReplyWithBufferedBlockDispatcher"]
|
||||
>[0];
|
||||
type DispatcherOptions = BufferedDispatchParams["dispatcherOptions"];
|
||||
type Deliver = DispatcherOptions["deliver"];
|
||||
type Skip = NonNullable<DispatcherOptions["onSkip"]>;
|
||||
type ErrorCallback = NonNullable<DispatcherOptions["onError"]>;
|
||||
type Cancel = NonNullable<DispatcherOptions["onBeforeDeliverCancelled"]>;
|
||||
|
||||
function resolvePayloadTelegramInlineButtons(
|
||||
payload: ReplyPayload,
|
||||
): TelegramInlineButtons | undefined {
|
||||
const telegramData = payload.channelData?.telegram as
|
||||
| { buttons?: TelegramInlineButtons }
|
||||
| undefined;
|
||||
return resolveTelegramInlineButtons({
|
||||
buttons: telegramData?.buttons,
|
||||
presentation: normalizeMessagePresentation(payload.presentation),
|
||||
interactive: payload.interactive,
|
||||
});
|
||||
}
|
||||
|
||||
function hasExecApprovalPayload(payload: ReplyPayload): boolean {
|
||||
return payload.channelData?.execApproval !== undefined;
|
||||
}
|
||||
|
||||
export function createTelegramReplyDelivery(params: {
|
||||
cfg: OpenClawConfig;
|
||||
context: TelegramMessageContext;
|
||||
delivery: TelegramDeliveryController;
|
||||
draft: TelegramDraftController;
|
||||
fence: Pick<TelegramReplyFenceController, "generation" | "isSuperseded">;
|
||||
progress: TelegramProgressController;
|
||||
runtime: RuntimeEnv;
|
||||
state: TelegramDispatchTurnState;
|
||||
streamMode: TelegramStreamMode;
|
||||
telegramCfg: TelegramAccountConfig;
|
||||
}) {
|
||||
const reasoningStepState = createTelegramReasoningStepState();
|
||||
const sentBlockMediaUrls = new Set<string>();
|
||||
params.draft.setReasoningStepCallbacks({
|
||||
noteHint: () => reasoningStepState.noteReasoningHint(),
|
||||
noteDelivered: () => reasoningStepState.noteReasoningDelivered(),
|
||||
});
|
||||
|
||||
const flushBufferedFinalAnswer = async () => {
|
||||
const buffered = reasoningStepState.takeBufferedFinalAnswer(params.fence.generation());
|
||||
if (!buffered) {
|
||||
return;
|
||||
}
|
||||
await params.delivery.deliverFinalAnswerText(
|
||||
buffered.payload,
|
||||
buffered.text,
|
||||
resolvePayloadTelegramInlineButtons(buffered.payload),
|
||||
);
|
||||
reasoningStepState.resetForNextStep();
|
||||
};
|
||||
const trackBlockMedia = (delivered: boolean, kind: string, payload: ReplyPayload) => {
|
||||
if (delivered && kind === "block" && payload.mediaUrls?.length) {
|
||||
for (const url of payload.mediaUrls) {
|
||||
sentBlockMediaUrls.add(url);
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
const deliver: Deliver = async (payload, info) => {
|
||||
if (params.fence.isSuperseded()) {
|
||||
return;
|
||||
}
|
||||
const normalizedPayload = params.delivery.normalizeDeliveryPayload(payload);
|
||||
if (!normalizedPayload) {
|
||||
return;
|
||||
}
|
||||
const deduped =
|
||||
info.kind === "final"
|
||||
? deduplicateBlockSentMedia(normalizedPayload, sentBlockMediaUrls)
|
||||
: normalizedPayload;
|
||||
if (!deduped) {
|
||||
return;
|
||||
}
|
||||
const effectivePayload = deduped;
|
||||
if (
|
||||
shouldSuppressLocalTelegramExecApprovalPrompt({
|
||||
cfg: params.cfg,
|
||||
accountId: params.context.route.accountId,
|
||||
payload: effectivePayload,
|
||||
})
|
||||
) {
|
||||
params.state.queuedFinal = true;
|
||||
return;
|
||||
}
|
||||
const telegramButtons = resolvePayloadTelegramInlineButtons(effectivePayload);
|
||||
const lanePayload =
|
||||
info.kind === "block" &&
|
||||
typeof payload.text === "string" &&
|
||||
typeof effectivePayload.text === "string" &&
|
||||
payload.text !== effectivePayload.text &&
|
||||
payload.text.trimEnd() === effectivePayload.text &&
|
||||
!effectivePayload.mediaUrl &&
|
||||
!effectivePayload.mediaUrls?.length
|
||||
? { ...effectivePayload, text: payload.text }
|
||||
: effectivePayload;
|
||||
const split = params.draft.splitTextIntoLaneSegments(
|
||||
{ text: lanePayload.text },
|
||||
payload.isReasoning,
|
||||
);
|
||||
const segments = split.segments;
|
||||
const reply = resolveSendableOutboundReplyParts(effectivePayload);
|
||||
if (info.kind === "final" && (reply.text.length > 0 || reply.hasMedia)) {
|
||||
params.progress.markFinalStarted();
|
||||
}
|
||||
if (info.kind === "final") {
|
||||
await params.draft.enqueueEvent(async () => {});
|
||||
}
|
||||
const isToolPayloadAfterFinal =
|
||||
info.kind === "tool" &&
|
||||
(params.progress.finalAnswerDeliveryStarted() || params.progress.finalAnswerDelivered());
|
||||
const isNonTerminalWarningAfterDeliveredFinal =
|
||||
isReplyPayloadNonTerminalToolErrorWarning(payload) && params.progress.finalAnswerDelivered();
|
||||
if (
|
||||
(isToolPayloadAfterFinal || isNonTerminalWarningAfterDeliveredFinal) &&
|
||||
!reply.hasMedia &&
|
||||
!hasExecApprovalPayload(effectivePayload)
|
||||
) {
|
||||
return;
|
||||
}
|
||||
if (payload.isError === true) {
|
||||
params.state.hadErrorReplyFailureOrSkip = true;
|
||||
}
|
||||
|
||||
let blockDelivered = false;
|
||||
const hasAnswerSegment = segments.some((segment) => segment.lane === "answer");
|
||||
if (info.kind === "block" && !hasAnswerSegment) {
|
||||
params.draft.dropQueuedAnswerBlockRotation(effectivePayload, info.assistantMessageIndex);
|
||||
}
|
||||
for (const segment of segments) {
|
||||
if (
|
||||
segment.lane === "answer" &&
|
||||
info.kind === "final" &&
|
||||
reasoningStepState.shouldBufferFinalAnswer()
|
||||
) {
|
||||
reasoningStepState.bufferFinalAnswer({
|
||||
payload: effectivePayload,
|
||||
text: segment.update.text,
|
||||
bufferedGeneration: params.fence.generation(),
|
||||
});
|
||||
continue;
|
||||
}
|
||||
if (segment.lane === "reasoning") {
|
||||
reasoningStepState.noteReasoningHint();
|
||||
}
|
||||
if (segment.lane === "answer" && info.kind === "tool") {
|
||||
if (params.progress.verboseProgressActive()) {
|
||||
if (
|
||||
await params.delivery.sendPayload(
|
||||
params.delivery.applyTextToPayload(effectivePayload, segment.update.text),
|
||||
)
|
||||
) {
|
||||
blockDelivered = true;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
const canRepresentAsTransientProgress =
|
||||
!reply.hasMedia &&
|
||||
telegramButtons === undefined &&
|
||||
!hasExecApprovalPayload(effectivePayload);
|
||||
const isFastModeProgressPayload = isFastModeAutoProgressPayload(effectivePayload);
|
||||
if (params.streamMode === "progress") {
|
||||
if (
|
||||
canRepresentAsTransientProgress &&
|
||||
params.draft.answerLane.stream &&
|
||||
!isFastModeProgressPayload
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
if (
|
||||
(canRepresentAsTransientProgress || isFastModeProgressPayload) &&
|
||||
(await params.progress.pushToolProgress(segment.update.text, {
|
||||
startImmediately: true,
|
||||
}))
|
||||
) {
|
||||
blockDelivered = true;
|
||||
continue;
|
||||
}
|
||||
}
|
||||
await params.draft.prepareAnswerLaneForToolProgress();
|
||||
}
|
||||
|
||||
const ownedByQueuedRotation = params.draft.isQueuedAnswerBlock(
|
||||
lanePayload,
|
||||
info.assistantMessageIndex,
|
||||
);
|
||||
const skipTextOnlyBlock =
|
||||
params.streamMode === "partial" &&
|
||||
info.kind === "block" &&
|
||||
segment.lane === "answer" &&
|
||||
!reply.hasMedia &&
|
||||
!hasExecApprovalPayload(effectivePayload) &&
|
||||
telegramButtons === undefined &&
|
||||
params.draft.answerLane.hasStreamedMessage &&
|
||||
!params.draft.isAnswerToolProgressOnly() &&
|
||||
!ownedByQueuedRotation &&
|
||||
segment.update.text.trimEnd() === params.draft.answerLane.lastPartialText.trimEnd();
|
||||
const suppressProgressAnswerBlock =
|
||||
params.streamMode === "progress" &&
|
||||
info.kind === "block" &&
|
||||
segment.lane === "answer" &&
|
||||
!reply.hasMedia &&
|
||||
!hasExecApprovalPayload(effectivePayload) &&
|
||||
telegramButtons === undefined;
|
||||
if (skipTextOnlyBlock || suppressProgressAnswerBlock) {
|
||||
params.draft.setActiveAnswerBlockDelivery({
|
||||
payload: effectivePayload,
|
||||
text: segment.update.text,
|
||||
buttons: telegramButtons,
|
||||
});
|
||||
params.draft.resetAnswerToolProgressDraft();
|
||||
params.progress.reset();
|
||||
blockDelivered = true;
|
||||
continue;
|
||||
}
|
||||
|
||||
if (segment.lane === "answer" && info.kind === "block") {
|
||||
const prepared = await params.draft.prepareAnswerLaneForText();
|
||||
const shouldRotate = params.draft.takeQueuedAnswerBlockRotation(
|
||||
lanePayload,
|
||||
info.assistantMessageIndex,
|
||||
);
|
||||
if (params.streamMode !== "progress" && shouldRotate && !prepared) {
|
||||
await params.draft.rotateAnswerLaneForNewMessage();
|
||||
params.draft.setRotateWhenQueuedBlocksSettle(false);
|
||||
}
|
||||
params.draft.resetAnswerToolProgressDraft();
|
||||
params.progress.reset();
|
||||
}
|
||||
const result =
|
||||
segment.lane === "answer" && info.kind === "final"
|
||||
? await params.delivery.deliverFinalAnswerText(
|
||||
effectivePayload,
|
||||
segment.update.text,
|
||||
telegramButtons,
|
||||
)
|
||||
: await params.delivery.deliverLaneText({
|
||||
laneName: segment.lane,
|
||||
text: segment.update.text,
|
||||
payload: lanePayload,
|
||||
infoKind: info.kind,
|
||||
buttons: telegramButtons,
|
||||
});
|
||||
if (
|
||||
segment.lane === "answer" &&
|
||||
info.kind !== "final" &&
|
||||
result.kind === "preview-finalized"
|
||||
) {
|
||||
await params.delivery.emitPreviewFinalizedHook(result);
|
||||
}
|
||||
if (segment.lane === "answer" && info.kind === "block" && result.kind === "preview-updated") {
|
||||
params.draft.setActiveAnswerBlockDelivery({
|
||||
payload: lanePayload,
|
||||
text: segment.update.text,
|
||||
buttons: telegramButtons,
|
||||
});
|
||||
}
|
||||
blockDelivered ||= result.kind !== "skipped";
|
||||
if (segment.lane === "reasoning") {
|
||||
if (result.kind !== "skipped") {
|
||||
reasoningStepState.noteReasoningDelivered();
|
||||
await flushBufferedFinalAnswer();
|
||||
}
|
||||
} else if (info.kind === "final") {
|
||||
reasoningStepState.resetForNextStep();
|
||||
}
|
||||
}
|
||||
if (segments.length > 0) {
|
||||
trackBlockMedia(blockDelivered, info.kind, effectivePayload);
|
||||
return;
|
||||
}
|
||||
|
||||
if (split.suppressedReasoningOnly) {
|
||||
let delivered = false;
|
||||
if (reply.hasMedia) {
|
||||
if (info.kind === "final") {
|
||||
await params.draft.rotateAnswerLaneAfterToolProgress();
|
||||
await params.draft.answerLane.stream?.stop();
|
||||
await params.draft.reasoningLane.stream?.stop();
|
||||
reasoningStepState.resetForNextStep();
|
||||
}
|
||||
const payloadWithoutReasoning =
|
||||
typeof effectivePayload.text === "string"
|
||||
? { ...effectivePayload, text: "" }
|
||||
: effectivePayload;
|
||||
delivered = await params.delivery.sendPayload(payloadWithoutReasoning, {
|
||||
durable: info.kind === "final",
|
||||
});
|
||||
}
|
||||
if (info.kind === "final" && delivered) {
|
||||
params.progress.markFinalDelivered();
|
||||
}
|
||||
if (info.kind === "final") {
|
||||
await flushBufferedFinalAnswer();
|
||||
}
|
||||
trackBlockMedia(delivered, info.kind, effectivePayload);
|
||||
return;
|
||||
}
|
||||
|
||||
if (info.kind === "final") {
|
||||
await params.draft.rotateAnswerLaneAfterToolProgress();
|
||||
await params.draft.answerLane.stream?.stop();
|
||||
await params.draft.reasoningLane.stream?.stop();
|
||||
reasoningStepState.resetForNextStep();
|
||||
}
|
||||
if (!reply.hasMedia && reply.text.length === 0) {
|
||||
if (info.kind === "final") {
|
||||
await flushBufferedFinalAnswer();
|
||||
}
|
||||
return;
|
||||
}
|
||||
const delivered = await params.delivery.sendPayload(effectivePayload, {
|
||||
durable: info.kind === "final",
|
||||
});
|
||||
if (info.kind === "final" && delivered) {
|
||||
params.progress.markFinalDelivered();
|
||||
}
|
||||
if (info.kind === "final") {
|
||||
await flushBufferedFinalAnswer();
|
||||
}
|
||||
trackBlockMedia(delivered, info.kind, effectivePayload);
|
||||
};
|
||||
|
||||
const onSkip: Skip = (payload, info) => {
|
||||
if (info.kind === "block") {
|
||||
void params.draft.enqueueEvent(async () => {
|
||||
params.draft.dropQueuedAnswerBlockRotation(payload, info.assistantMessageIndex);
|
||||
});
|
||||
}
|
||||
if (payload.isError === true) {
|
||||
params.state.hadErrorReplyFailureOrSkip = true;
|
||||
}
|
||||
if (info.reason !== "silent") {
|
||||
params.delivery.markNonSilentSkip();
|
||||
}
|
||||
};
|
||||
|
||||
const onError: ErrorCallback = (err, info) => {
|
||||
const errorPolicy = resolveTelegramErrorPolicy({
|
||||
accountConfig: params.telegramCfg,
|
||||
groupConfig: params.context.groupConfig,
|
||||
topicConfig: params.context.topicConfig,
|
||||
});
|
||||
if (isSilentErrorPolicy(errorPolicy.policy)) {
|
||||
return;
|
||||
}
|
||||
if (
|
||||
errorPolicy.policy === "once" &&
|
||||
shouldSuppressTelegramError({
|
||||
scopeKey: buildTelegramErrorScopeKey({
|
||||
accountId: params.context.route.accountId,
|
||||
chatId: params.context.chatId,
|
||||
threadId: params.context.threadSpec.id,
|
||||
}),
|
||||
cooldownMs: errorPolicy.cooldownMs,
|
||||
errorMessage: String(err),
|
||||
})
|
||||
) {
|
||||
return;
|
||||
}
|
||||
params.delivery.markNonSilentFailure();
|
||||
params.runtime.error?.(danger(`telegram ${info.kind} reply failed: ${String(err)}`));
|
||||
};
|
||||
|
||||
return {
|
||||
deliver,
|
||||
onBeforeDeliverCancelled: (payload: Parameters<Cancel>[0], info: Parameters<Cancel>[1]) => {
|
||||
if (info.kind === "block") {
|
||||
return params.draft.enqueueEvent(async () => {
|
||||
params.draft.dropQueuedAnswerBlockRotation(payload, info.assistantMessageIndex);
|
||||
});
|
||||
}
|
||||
return undefined;
|
||||
},
|
||||
onError,
|
||||
onSkip,
|
||||
reasoningStepState,
|
||||
};
|
||||
}
|
||||
|
||||
export type TelegramReplyDelivery = ReturnType<typeof createTelegramReplyDelivery>;
|
||||
@@ -0,0 +1,155 @@
|
||||
// Telegram plugin module owns dispatch-time session and transcript access.
|
||||
import path from "node:path";
|
||||
import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts";
|
||||
import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime";
|
||||
import { logVerbose } from "openclaw/plugin-sdk/runtime-env";
|
||||
import {
|
||||
appendAssistantMirrorMessageByIdentity,
|
||||
readLatestAssistantTextByIdentity,
|
||||
} from "openclaw/plugin-sdk/session-transcript-runtime";
|
||||
import { resolveTelegramConfigReasoningDefault } from "./agent-config.js";
|
||||
import type { TelegramBotDeps } from "./bot-deps.js";
|
||||
import type { TelegramMessageContext } from "./bot-message-context.js";
|
||||
import { getSessionEntry } from "./bot-message-dispatch.runtime.js";
|
||||
import type {
|
||||
CurrentTurnTranscriptFinal,
|
||||
FreshTelegramSessionEntryLoader,
|
||||
TelegramReasoningLevel,
|
||||
TelegramScopedTranscriptSession,
|
||||
TelegramTranscriptMirrorPayload,
|
||||
} from "./bot-message-dispatch.types.js";
|
||||
|
||||
export function createFreshTelegramSessionEntryLoader(params: {
|
||||
cfg: OpenClawConfig;
|
||||
telegramDeps: TelegramBotDeps;
|
||||
}): FreshTelegramSessionEntryLoader {
|
||||
const entriesByPathAndKey = new Map<string, ReturnType<typeof getSessionEntry>>();
|
||||
const load = ((agentId: string, sessionKey: string) => {
|
||||
const storePath = params.telegramDeps.resolveStorePath(params.cfg.session?.store, { agentId });
|
||||
const cacheKey = `${storePath}\0${sessionKey}`;
|
||||
if (entriesByPathAndKey.has(cacheKey)) {
|
||||
return { storePath, entry: entriesByPathAndKey.get(cacheKey) };
|
||||
}
|
||||
const entry = (params.telegramDeps.getSessionEntry ?? getSessionEntry)({
|
||||
storePath,
|
||||
sessionKey,
|
||||
readConsistency: "latest",
|
||||
});
|
||||
entriesByPathAndKey.set(cacheKey, entry);
|
||||
return { storePath, entry };
|
||||
}) as FreshTelegramSessionEntryLoader;
|
||||
load.clear = () => entriesByPathAndKey.clear();
|
||||
return load;
|
||||
}
|
||||
|
||||
export function resolveTelegramReasoningLevel(params: {
|
||||
cfg: OpenClawConfig;
|
||||
sessionKey?: string;
|
||||
agentId: string;
|
||||
loadFreshSessionEntry: FreshTelegramSessionEntryLoader;
|
||||
}): TelegramReasoningLevel {
|
||||
const configDefault = resolveTelegramConfigReasoningDefault(params.cfg, params.agentId);
|
||||
if (!params.sessionKey) {
|
||||
return configDefault;
|
||||
}
|
||||
try {
|
||||
const { entry } = params.loadFreshSessionEntry(params.agentId, params.sessionKey);
|
||||
const level = entry?.reasoningLevel;
|
||||
return level === "on" || level === "stream" || level === "off" ? level : configDefault;
|
||||
} catch {
|
||||
return "off";
|
||||
}
|
||||
}
|
||||
|
||||
function resolveTelegramMirroredTranscriptText(
|
||||
payload: TelegramTranscriptMirrorPayload,
|
||||
): string | null {
|
||||
const mediaUrls = payload.mediaUrls?.filter((url) => url.trim()) ?? [];
|
||||
if (mediaUrls.length > 0) {
|
||||
return mediaUrls
|
||||
.map((url) => {
|
||||
const pathname = url.split("#")[0]?.split("?")[0] ?? url;
|
||||
const base = path.basename(pathname);
|
||||
return base && base !== "." && base !== "/" ? base : "media";
|
||||
})
|
||||
.join(", ");
|
||||
}
|
||||
return payload.text?.trim() || null;
|
||||
}
|
||||
|
||||
function resolveTelegramScopedTranscriptSession(params: {
|
||||
agentId: string;
|
||||
loadFreshSessionEntry: FreshTelegramSessionEntryLoader;
|
||||
sessionKey: string;
|
||||
}): TelegramScopedTranscriptSession | undefined {
|
||||
const { entry, storePath } = params.loadFreshSessionEntry(params.agentId, params.sessionKey);
|
||||
const sessionId = entry?.sessionId?.trim();
|
||||
return sessionId ? { sessionId, storePath } : undefined;
|
||||
}
|
||||
|
||||
export async function mirrorTelegramAssistantReplyToTranscript(params: {
|
||||
cfg: OpenClawConfig;
|
||||
idempotencyKey: string;
|
||||
loadFreshSessionEntry: FreshTelegramSessionEntryLoader;
|
||||
route: TelegramMessageContext["route"];
|
||||
sessionKey: string;
|
||||
payload: TelegramTranscriptMirrorPayload;
|
||||
}) {
|
||||
const text = resolveTelegramMirroredTranscriptText(params.payload);
|
||||
if (!text) {
|
||||
return;
|
||||
}
|
||||
const session = resolveTelegramScopedTranscriptSession({
|
||||
agentId: params.route.agentId,
|
||||
loadFreshSessionEntry: params.loadFreshSessionEntry,
|
||||
sessionKey: params.sessionKey,
|
||||
});
|
||||
if (!session) {
|
||||
return;
|
||||
}
|
||||
const appended = await appendAssistantMirrorMessageByIdentity({
|
||||
agentId: params.route.agentId,
|
||||
config: params.cfg,
|
||||
idempotencyKey: params.idempotencyKey,
|
||||
deliveryMirror: { kind: "channel-final", sourceMessageId: params.idempotencyKey },
|
||||
sessionId: session.sessionId,
|
||||
sessionKey: params.sessionKey,
|
||||
storePath: session.storePath,
|
||||
text,
|
||||
});
|
||||
if (!appended.ok && appended.code !== "session-rebound") {
|
||||
logVerbose(`telegram transcript mirror append failed: ${appended.reason}`);
|
||||
}
|
||||
}
|
||||
|
||||
export function createCurrentTurnTranscriptFinalResolver(params: {
|
||||
agentId: string;
|
||||
dispatchStartedAt: number;
|
||||
loadFreshSessionEntry: FreshTelegramSessionEntryLoader;
|
||||
sessionKey?: string;
|
||||
}): () => Promise<CurrentTurnTranscriptFinal | undefined> {
|
||||
return async () => {
|
||||
if (!params.sessionKey) {
|
||||
return undefined;
|
||||
}
|
||||
try {
|
||||
const { entry, storePath } = params.loadFreshSessionEntry(params.agentId, params.sessionKey);
|
||||
if (!entry?.sessionId) {
|
||||
return undefined;
|
||||
}
|
||||
const latest = await readLatestAssistantTextByIdentity({
|
||||
agentId: params.agentId,
|
||||
sessionId: entry.sessionId,
|
||||
sessionKey: params.sessionKey,
|
||||
storePath,
|
||||
});
|
||||
if (!latest?.timestamp || latest.timestamp < params.dispatchStartedAt) {
|
||||
return undefined;
|
||||
}
|
||||
return { ...(latest.id ? { messageId: latest.id } : {}), text: latest.text };
|
||||
} catch (err) {
|
||||
logVerbose(`telegram transcript final candidate lookup failed: ${formatErrorMessage(err)}`);
|
||||
return undefined;
|
||||
}
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,91 @@
|
||||
// Telegram plugin module owns dispatch status-reaction finalization.
|
||||
import {
|
||||
DEFAULT_TIMING,
|
||||
logAckFailure,
|
||||
removeAckReactionAfterReply,
|
||||
} from "openclaw/plugin-sdk/channel-feedback";
|
||||
import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts";
|
||||
import { logVerbose, sleepWithAbort } from "openclaw/plugin-sdk/runtime-env";
|
||||
import type { TelegramMessageContext } from "./bot-message-context.js";
|
||||
|
||||
export function createTelegramDispatchStatus(params: {
|
||||
cfg: OpenClawConfig;
|
||||
context: TelegramMessageContext;
|
||||
}) {
|
||||
const { context } = params;
|
||||
const controller =
|
||||
context.ctxPayload.InboundEventKind === "room_event" ? null : context.statusReactionController;
|
||||
const timing = { ...DEFAULT_TIMING, ...params.cfg.messages?.statusReactions?.timing };
|
||||
|
||||
const clear = async () => {
|
||||
if (!context.msg.message_id || !context.reactionApi) {
|
||||
return;
|
||||
}
|
||||
await context.reactionApi(context.chatId, context.msg.message_id, []);
|
||||
};
|
||||
|
||||
const finalize = async (final: { outcome: "done" | "error"; hasFinalResponse: boolean }) => {
|
||||
if (!controller) {
|
||||
return;
|
||||
}
|
||||
if (final.outcome === "done") {
|
||||
await controller.setDone();
|
||||
if (context.removeAckAfterReply) {
|
||||
await sleepWithAbort(timing.doneHoldMs);
|
||||
await clear();
|
||||
} else {
|
||||
await controller.restoreInitial();
|
||||
}
|
||||
return;
|
||||
}
|
||||
await controller.setError();
|
||||
if (final.hasFinalResponse) {
|
||||
if (context.removeAckAfterReply) {
|
||||
await sleepWithAbort(timing.errorHoldMs);
|
||||
await clear();
|
||||
} else {
|
||||
await controller.restoreInitial();
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (context.removeAckAfterReply) {
|
||||
await sleepWithAbort(timing.errorHoldMs);
|
||||
}
|
||||
await controller.restoreInitial();
|
||||
};
|
||||
|
||||
const removeAck = () => {
|
||||
removeAckReactionAfterReply({
|
||||
removeAfterReply: context.removeAckAfterReply,
|
||||
ackReactionPromise: context.ackReactionPromise,
|
||||
ackReactionValue: context.ackReactionPromise ? "ack" : null,
|
||||
remove: () =>
|
||||
(
|
||||
context.reactionApi?.(context.chatId, context.msg.message_id ?? 0, []) ??
|
||||
Promise.resolve()
|
||||
).then(() => {}),
|
||||
onError: (err) => {
|
||||
if (!context.msg.message_id) {
|
||||
return;
|
||||
}
|
||||
logAckFailure({
|
||||
log: logVerbose,
|
||||
channel: "telegram",
|
||||
target: `${context.chatId}/${context.msg.message_id}`,
|
||||
error: err,
|
||||
});
|
||||
},
|
||||
});
|
||||
};
|
||||
|
||||
const finalizeInBackground = (
|
||||
final: { outcome: "done" | "error"; hasFinalResponse: boolean },
|
||||
label: string,
|
||||
) => {
|
||||
void finalize(final).catch((err: unknown) => {
|
||||
logVerbose(`telegram: status reaction ${label} failed: ${String(err)}`);
|
||||
});
|
||||
};
|
||||
|
||||
return { controller, finalizeInBackground, removeAck };
|
||||
}
|
||||
@@ -0,0 +1,250 @@
|
||||
import { logTypingFailure } from "openclaw/plugin-sdk/channel-feedback";
|
||||
import { runChannelInboundEvent } from "openclaw/plugin-sdk/channel-inbound";
|
||||
// Telegram plugin module wires inbound turn execution to Telegram delivery controllers.
|
||||
import {
|
||||
createChannelMessageReplyPipeline,
|
||||
resolveChannelStreamingPreviewToolProgress,
|
||||
} from "openclaw/plugin-sdk/channel-outbound";
|
||||
import type { OpenClawConfig, TelegramAccountConfig } from "openclaw/plugin-sdk/config-contracts";
|
||||
import { isFastModeAutoProgressPayload } from "openclaw/plugin-sdk/reply-payload";
|
||||
import { logVerbose } from "openclaw/plugin-sdk/runtime-env";
|
||||
import type { TelegramBotDeps } from "./bot-deps.js";
|
||||
import type { TelegramMessageContext } from "./bot-message-context.js";
|
||||
import type { TelegramDeliveryController } from "./bot-message-dispatch-delivery.js";
|
||||
import type { TelegramDraftController } from "./bot-message-dispatch-draft.js";
|
||||
import type { TelegramReplyFenceController } from "./bot-message-dispatch-fence.js";
|
||||
import type { TelegramProgressController } from "./bot-message-dispatch-progress.js";
|
||||
import type { TelegramReplyDelivery } from "./bot-message-dispatch-reply.js";
|
||||
import type { TelegramDispatchTurnState } from "./bot-message-dispatch.types.js";
|
||||
import type { TelegramStreamMode } from "./bot/types.js";
|
||||
import { beginTelegramInboundEventDeliveryCorrelation } from "./inbound-event-delivery.js";
|
||||
|
||||
const TELEGRAM_MAX_CONSECUTIVE_TYPING_FAILURES = 5;
|
||||
|
||||
export async function runTelegramDispatchTurn(params: {
|
||||
cfg: OpenClawConfig;
|
||||
context: TelegramMessageContext;
|
||||
delivery: TelegramDeliveryController;
|
||||
draft: TelegramDraftController;
|
||||
fence: TelegramReplyFenceController;
|
||||
progress: TelegramProgressController;
|
||||
reply: TelegramReplyDelivery;
|
||||
state: TelegramDispatchTurnState;
|
||||
statusReactionController: TelegramMessageContext["statusReactionController"];
|
||||
streamMode: TelegramStreamMode;
|
||||
telegramCfg: TelegramAccountConfig;
|
||||
telegramDeps: TelegramBotDeps;
|
||||
}) {
|
||||
const { context } = params;
|
||||
const isRoomEvent = context.ctxPayload.InboundEventKind === "room_event";
|
||||
const beginDeliveryCorrelation = () =>
|
||||
beginTelegramInboundEventDeliveryCorrelation(
|
||||
context.ctxPayload.SessionKey,
|
||||
{
|
||||
outboundTo: context.historyKey || String(context.chatId),
|
||||
outboundAccountId: context.route.accountId,
|
||||
markInboundEventDelivered: params.delivery.markDelivered,
|
||||
},
|
||||
{ inboundEventKind: context.ctxPayload.InboundEventKind },
|
||||
);
|
||||
const endDeliveryCorrelation = beginDeliveryCorrelation();
|
||||
let splitReasoningOnNextStream = false;
|
||||
|
||||
try {
|
||||
const { onModelSelected, ...replyPipeline } = (
|
||||
params.telegramDeps.createChannelMessageReplyPipeline ?? createChannelMessageReplyPipeline
|
||||
)({
|
||||
cfg: params.cfg,
|
||||
agentId: context.route.agentId,
|
||||
channel: "telegram",
|
||||
accountId: context.route.accountId,
|
||||
typing: {
|
||||
start: context.sendTyping,
|
||||
maxConsecutiveFailures: TELEGRAM_MAX_CONSECUTIVE_TYPING_FAILURES,
|
||||
onStartError: (err) => {
|
||||
logTypingFailure({
|
||||
log: logVerbose,
|
||||
channel: "telegram",
|
||||
target: String(context.chatId),
|
||||
error: err,
|
||||
});
|
||||
},
|
||||
},
|
||||
});
|
||||
const turnResult = await runChannelInboundEvent({
|
||||
channel: "telegram",
|
||||
accountId: context.route.accountId,
|
||||
raw: context,
|
||||
adapter: {
|
||||
ingest: () => ({
|
||||
id: context.ctxPayload.MessageSid ?? `${context.chatId}:${Date.now()}`,
|
||||
timestamp:
|
||||
typeof context.ctxPayload.Timestamp === "number"
|
||||
? context.ctxPayload.Timestamp
|
||||
: undefined,
|
||||
rawText: context.ctxPayload.RawBody ?? "",
|
||||
textForAgent: context.ctxPayload.BodyForAgent,
|
||||
textForCommands: context.ctxPayload.CommandBody,
|
||||
raw: context,
|
||||
}),
|
||||
resolveTurn: () => ({
|
||||
channel: "telegram",
|
||||
accountId: context.route.accountId,
|
||||
routeSessionKey: context.route.sessionKey,
|
||||
storePath: context.turn.storePath,
|
||||
ctxPayload: context.ctxPayload,
|
||||
recordInboundSession: context.turn.recordInboundSession,
|
||||
record: context.turn.record,
|
||||
runDispatch: () =>
|
||||
params.telegramDeps.dispatchReplyWithBufferedBlockDispatcher({
|
||||
ctx: context.ctxPayload,
|
||||
cfg: params.cfg,
|
||||
dispatcherOptions: {
|
||||
...replyPipeline,
|
||||
beforeDeliver: async (payload) => payload,
|
||||
onBeforeDeliverCancelled: params.reply.onBeforeDeliverCancelled,
|
||||
deliver: params.reply.deliver,
|
||||
onSkip: params.reply.onSkip,
|
||||
onError: params.reply.onError,
|
||||
},
|
||||
replyOptions: {
|
||||
skillFilter: context.skillFilter,
|
||||
disableBlockStreaming: params.draft.disableBlockStreaming,
|
||||
abortSignal: params.fence.abortSignal,
|
||||
onTurnAdopted: params.fence.adoptTurn,
|
||||
sourceReplyDeliveryMode: isRoomEvent ? "message_tool_only" : undefined,
|
||||
queuedDeliveryCorrelations: isRoomEvent
|
||||
? [{ begin: beginDeliveryCorrelation }]
|
||||
: undefined,
|
||||
queuedFollowupLifecycle: params.fence.queuedFollowupLifecycle,
|
||||
suppressTyping: isRoomEvent,
|
||||
onPartialReply:
|
||||
params.draft.answerLane.stream || params.draft.reasoningLane.stream
|
||||
? (payload) =>
|
||||
params.draft.enqueueEvent(async () => {
|
||||
await params.draft.ingestDraftLaneSegments(payload);
|
||||
})
|
||||
: undefined,
|
||||
onBlockReplyQueued: params.draft.answerLane.stream
|
||||
? (payload, blockContext) =>
|
||||
params.draft.enqueueEvent(async () => {
|
||||
await params.draft.prepareQueuedAnswerBlock(payload, blockContext);
|
||||
})
|
||||
: undefined,
|
||||
onReasoningStream: params.draft.reasoningLane.stream
|
||||
? (payload) =>
|
||||
params.draft.enqueueEvent(async () => {
|
||||
if (splitReasoningOnNextStream) {
|
||||
params.draft.repositionLaneForNewMessage(params.draft.reasoningLane);
|
||||
splitReasoningOnNextStream = false;
|
||||
}
|
||||
await params.draft.ingestDraftLaneSegments(payload, true);
|
||||
})
|
||||
: params.draft.streamReasoningInProgressDraft
|
||||
? (payload) =>
|
||||
params.draft.enqueueEvent(async () => {
|
||||
await params.progress.pushReasoningProgress(payload);
|
||||
})
|
||||
: undefined,
|
||||
onReasoningProgress: params.draft.answerLane.stream
|
||||
? (payload) =>
|
||||
params.draft.enqueueEvent(async () => {
|
||||
await params.progress.pushThinkingTokenProgress(payload.progressTokens);
|
||||
})
|
||||
: undefined,
|
||||
onAssistantMessageStart: params.draft.answerLane.stream
|
||||
? () =>
|
||||
params.draft.enqueueEvent(async () => {
|
||||
params.reply.reasoningStepState.resetForNextStep();
|
||||
params.progress.setFinalAnswerDelivered(false);
|
||||
if (params.streamMode !== "progress") {
|
||||
params.progress.reset();
|
||||
}
|
||||
if (params.draft.answerLane.finalized) {
|
||||
await params.draft.rotateLaneForNewMessage(params.draft.answerLane);
|
||||
params.draft.setRotateWhenQueuedBlocksSettle(false);
|
||||
} else if (
|
||||
params.draft.answerLane.hasStreamedMessage &&
|
||||
!params.draft.isAnswerToolProgressOnly()
|
||||
) {
|
||||
params.draft.setRotateWhenQueuedBlocksSettle(true);
|
||||
}
|
||||
})
|
||||
: undefined,
|
||||
onReasoningEnd: params.draft.reasoningLane.stream
|
||||
? () =>
|
||||
params.draft.enqueueEvent(async () => {
|
||||
params.progress.closeReasoningBurst();
|
||||
splitReasoningOnNextStream = params.draft.reasoningLane.hasStreamedMessage;
|
||||
params.progress.reset();
|
||||
})
|
||||
: () => params.progress.closeReasoningBurst(),
|
||||
suppressDefaultToolProgressMessages:
|
||||
!params.draft.streamDeliveryEnabled || Boolean(params.draft.answerLane.stream),
|
||||
forceToolResultProgress:
|
||||
params.streamMode === "progress" &&
|
||||
resolveChannelStreamingPreviewToolProgress(params.telegramCfg),
|
||||
allowProgressCallbacksWhenSourceDeliverySuppressed:
|
||||
!isRoomEvent && Boolean(params.draft.answerLane.stream),
|
||||
onVerboseProgressVisibility: (isActive) => {
|
||||
params.progress.setVerboseProgressActive(isActive);
|
||||
},
|
||||
commentaryProgressEnabled:
|
||||
params.streamMode === "progress"
|
||||
? params.progress.commentaryProgressEnabled
|
||||
: undefined,
|
||||
progressPreambleEnabled: params.progress.progressPreambleEnabled,
|
||||
reasoningPayloadsEnabled: params.draft.durableReasoningPayloadsEnabled,
|
||||
onToolStart: params.progress.handleToolStart,
|
||||
onItemEvent: params.progress.handleItemEvent,
|
||||
onPlanUpdate: params.progress.handlePlanUpdate,
|
||||
onApprovalEvent: params.progress.handleApprovalEvent,
|
||||
onToolResult: async (payload) => {
|
||||
const text = payload.text?.trim();
|
||||
if (!text) {
|
||||
return;
|
||||
}
|
||||
const updatedDraft = await params.progress.pushToolProgress(text, {
|
||||
startImmediately: true,
|
||||
});
|
||||
if (
|
||||
!updatedDraft &&
|
||||
isFastModeAutoProgressPayload(payload) &&
|
||||
!params.progress.canPushToolProgress()
|
||||
) {
|
||||
await params.delivery.sendPayload(payload);
|
||||
}
|
||||
},
|
||||
onCommandOutput: params.progress.handleCommandOutput,
|
||||
onPatchSummary: params.progress.handlePatchSummary,
|
||||
onCompactionStart: params.statusReactionController
|
||||
? async () => {
|
||||
await params.statusReactionController?.setCompacting();
|
||||
}
|
||||
: undefined,
|
||||
onCompactionEnd: params.statusReactionController
|
||||
? async () => {
|
||||
params.statusReactionController?.cancelPending();
|
||||
await params.statusReactionController?.setThinking();
|
||||
}
|
||||
: undefined,
|
||||
onModelSelected,
|
||||
},
|
||||
}),
|
||||
}),
|
||||
},
|
||||
});
|
||||
if (!turnResult.dispatched) {
|
||||
return false;
|
||||
}
|
||||
params.state.queuedFinal = turnResult.dispatchResult.queuedFinal;
|
||||
if ((turnResult.dispatchResult.counts?.final ?? 0) > 0) {
|
||||
params.progress.markSawFinal();
|
||||
}
|
||||
params.state.suppressSilentReplyFallback =
|
||||
turnResult.dispatchResult.sourceReplyDeliveryMode === "message_tool_only";
|
||||
return true;
|
||||
} finally {
|
||||
endDeliveryCorrelation();
|
||||
}
|
||||
}
|
||||
@@ -563,13 +563,14 @@ describe("dispatchTelegramMessage draft streaming", () => {
|
||||
onTurnDeferred?: Parameters<typeof dispatchTelegramMessage>[0]["onTurnDeferred"];
|
||||
onTurnAbandoned?: Parameters<typeof dispatchTelegramMessage>[0]["onTurnAbandoned"];
|
||||
turnAbortSignal?: Parameters<typeof dispatchTelegramMessage>[0]["turnAbortSignal"];
|
||||
runtime?: Parameters<typeof dispatchTelegramMessage>[0]["runtime"];
|
||||
}) {
|
||||
const bot = params.bot ?? createBot();
|
||||
return await dispatchTelegramMessage({
|
||||
context: params.context,
|
||||
bot,
|
||||
cfg: params.cfg ?? {},
|
||||
runtime: createRuntime(),
|
||||
runtime: params.runtime ?? createRuntime(),
|
||||
replyToMode: params.replyToMode ?? "first",
|
||||
streamMode: params.streamMode ?? "partial",
|
||||
textLimit: params.textLimit ?? 4096,
|
||||
@@ -7056,6 +7057,68 @@ describe("dispatchTelegramMessage draft streaming", () => {
|
||||
await sidePromise;
|
||||
});
|
||||
|
||||
it("does not acquire reply-fence ownership when draft initialization fails", async () => {
|
||||
const sessionKey = "agent:main:telegram:direct:draft-init-failure";
|
||||
createTelegramDraftStream.mockImplementationOnce(() => {
|
||||
throw new Error("draft initialization failed");
|
||||
});
|
||||
|
||||
await expect(
|
||||
dispatchWithContext({
|
||||
context: createContext({
|
||||
ctxPayload: {
|
||||
SessionKey: sessionKey,
|
||||
ChatType: "direct",
|
||||
} as TelegramMessageContext["ctxPayload"],
|
||||
}),
|
||||
}),
|
||||
).rejects.toThrow("draft initialization failed");
|
||||
|
||||
const { supersedeTelegramReplyFence } = await import("./telegram-reply-fence.js");
|
||||
expect(supersedeTelegramReplyFence(sessionKey)).toBe(false);
|
||||
});
|
||||
|
||||
it("cleans delivery correlation when reply-pipeline initialization fails", async () => {
|
||||
const sessionKey = "agent:main:telegram:direct:pipeline-init-failure";
|
||||
const statusReactionController = createStatusReactionController();
|
||||
const reactionApi = vi.fn(async () => undefined);
|
||||
const runtime = createRuntime();
|
||||
runtime.error = vi.fn(() => {
|
||||
notifyTelegramInboundEventOutboundSuccess({
|
||||
sessionKey,
|
||||
to: "123",
|
||||
accountId: "default",
|
||||
});
|
||||
});
|
||||
createChannelMessageReplyPipeline.mockImplementationOnce(() => {
|
||||
throw new Error("pipeline initialization failed");
|
||||
});
|
||||
|
||||
await dispatchWithContext({
|
||||
context: createContext({
|
||||
ctxPayload: {
|
||||
SessionKey: sessionKey,
|
||||
ChatType: "direct",
|
||||
} as TelegramMessageContext["ctxPayload"],
|
||||
statusReactionController: statusReactionController as never,
|
||||
reactionApi,
|
||||
removeAckAfterReply: true,
|
||||
}),
|
||||
cfg: {
|
||||
messages: {
|
||||
statusReactions: {
|
||||
timing: { errorHoldMs: 0 },
|
||||
},
|
||||
},
|
||||
},
|
||||
runtime,
|
||||
suppressFailureFallback: true,
|
||||
});
|
||||
|
||||
await vi.waitFor(() => expect(statusReactionController.restoreInitial).toHaveBeenCalled());
|
||||
expect(reactionApi).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("releases fence abort authority at turn adoption", async () => {
|
||||
const historyKey = "telegram:group:-100123";
|
||||
const groupHistories = new Map([[historyKey, []]]);
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,69 @@
|
||||
// Telegram plugin module defines message-dispatch contracts.
|
||||
import type { Bot } from "grammy";
|
||||
import type {
|
||||
OpenClawConfig,
|
||||
ReplyToMode,
|
||||
TelegramAccountConfig,
|
||||
} from "openclaw/plugin-sdk/config-contracts";
|
||||
import type { ReplyPayload } from "openclaw/plugin-sdk/reply-payload";
|
||||
import type { RuntimeEnv } from "openclaw/plugin-sdk/runtime-env";
|
||||
import type { TelegramBotDeps } from "./bot-deps.js";
|
||||
import type { TelegramMessageContext } from "./bot-message-context.js";
|
||||
import type { SessionEntry } from "./bot-message-dispatch.runtime.js";
|
||||
import type { TelegramBotOptions } from "./bot.types.js";
|
||||
import type { TelegramStreamMode } from "./bot/types.js";
|
||||
|
||||
export type DispatchTelegramMessageParams = {
|
||||
context: TelegramMessageContext;
|
||||
bot: Bot;
|
||||
cfg: OpenClawConfig;
|
||||
runtime: RuntimeEnv;
|
||||
replyToMode: ReplyToMode;
|
||||
streamMode: TelegramStreamMode;
|
||||
textLimit: number;
|
||||
telegramCfg: TelegramAccountConfig;
|
||||
telegramDeps?: TelegramBotDeps;
|
||||
opts: Pick<TelegramBotOptions, "token" | "mediaMaxMb">;
|
||||
retryDispatchErrors?: boolean;
|
||||
suppressFailureFallback?: boolean;
|
||||
/** Fires after recovery-relevant session/run state is durably persisted. */
|
||||
onTurnAdopted?: () => void | Promise<void>;
|
||||
/** Marks a queued follow-up whose adoption will happen at reply-lane admission. */
|
||||
onTurnDeferred?: () => void;
|
||||
/** Releases a deferred turn that completed without ever owning the reply lane. */
|
||||
onTurnAbandoned?: () => void;
|
||||
/** Cancels queued/model work when ingress ownership fails before adoption. */
|
||||
turnAbortSignal?: AbortSignal;
|
||||
};
|
||||
|
||||
export type TelegramDispatchResult =
|
||||
| { kind: "completed" }
|
||||
| { kind: "failed-retryable"; error: unknown };
|
||||
|
||||
export type TelegramReasoningLevel = "off" | "on" | "stream";
|
||||
export type TelegramTranscriptMirrorPayload = { text?: string; mediaUrls?: string[] };
|
||||
export type CurrentTurnTranscriptFinal = { messageId?: string; text: string };
|
||||
export type TelegramScopedTranscriptSession = { sessionId: string; storePath: string };
|
||||
|
||||
export type FreshTelegramSessionEntryLoader = ((
|
||||
agentId: string,
|
||||
sessionKey: string,
|
||||
) => {
|
||||
storePath: string;
|
||||
entry?: SessionEntry;
|
||||
}) & {
|
||||
clear: () => void;
|
||||
};
|
||||
|
||||
export type TelegramAnswerBlockDelivery = {
|
||||
payload: ReplyPayload;
|
||||
text: string;
|
||||
buttons: import("./button-types.js").TelegramInlineButtons | undefined;
|
||||
};
|
||||
|
||||
export type TelegramDispatchTurnState = {
|
||||
queuedFinal: boolean;
|
||||
suppressSilentReplyFallback: boolean;
|
||||
hadErrorReplyFailureOrSkip: boolean;
|
||||
dispatchError?: unknown;
|
||||
};
|
||||
@@ -21,6 +21,7 @@ const migratedMessageTurnFiles = [
|
||||
"extensions/slack/src/monitor/message-handler/prepare.ts",
|
||||
"extensions/telegram/src/bot-message-context.body.ts",
|
||||
"extensions/telegram/src/bot-message-context.session.ts",
|
||||
"extensions/telegram/src/bot-message-dispatch-context.ts",
|
||||
"extensions/telegram/src/bot-message-dispatch.ts",
|
||||
"extensions/whatsapp/src/auto-reply/monitor/group-gating.ts",
|
||||
"extensions/zalouser/src/monitor.ts",
|
||||
@@ -37,7 +38,7 @@ const historyWindowFiles = [
|
||||
"extensions/qqbot/src/bridge/sdk-adapter.ts",
|
||||
"extensions/signal/src/monitor/event-handler.ts",
|
||||
"extensions/slack/src/monitor/message-handler/prepare.ts",
|
||||
"extensions/telegram/src/bot-message-dispatch.ts",
|
||||
"extensions/telegram/src/bot-message-dispatch-context.ts",
|
||||
"extensions/telegram/src/group-history-window.ts",
|
||||
"extensions/whatsapp/src/auto-reply/monitor/group-gating.ts",
|
||||
"extensions/zalouser/src/monitor.ts",
|
||||
|
||||
Reference in New Issue
Block a user