/** * Telegram queue core contracts and pure planning helpers * Zones: telegram queue, pi agent lifecycle, scheduling * Owns queue item contracts, lane admission, pure queue mutations, and dispatch planning */ import { isVoiceTurn } from "./voice.ts"; // --- Queue Items --- export interface QueuedAttachment { path: string; fileName: string; } export interface TelegramPromptTextContent { type: "text"; text: string; } export interface TelegramPromptImageContent { type: "image"; data: string; mimeType: string; } export type TelegramPromptContent = TelegramPromptTextContent | TelegramPromptImageContent; export type TelegramQueueItemKind = "prompt" | "control"; export type TelegramQueueLane = "control" | "priority" | "default"; export type TelegramQueueAdmissionMode = "control-queue" | "priority-queue" | "default-queue"; export interface TelegramQueueLaneContract { lane: TelegramQueueLane; admissionMode: TelegramQueueAdmissionMode; dispatchRank: number; allowedKinds: readonly TelegramQueueItemKind[]; } export const TELEGRAM_QUEUE_LANE_CONTRACTS: readonly TelegramQueueLaneContract[] = [ // Control lane intentionally accepts both direct controls and resume prompts. // Model-switch continuations need prompt semantics but must run before queued user work. // Do not admit ordinary user prompts here without an explicit control-flow reason. { lane: "control", admissionMode: "control-queue", dispatchRank: 0, allowedKinds: ["control", "prompt"], }, { lane: "priority", admissionMode: "priority-queue", dispatchRank: 1, allowedKinds: ["prompt"], }, { lane: "default", admissionMode: "default-queue", dispatchRank: 2, allowedKinds: ["prompt"], }, ] as const; export interface TelegramQueueTarget { chatId: number; threadId?: number; } export interface TelegramTransportStamp { profile: string; generation: string; } export interface TelegramTransportStampRuntime { getStamp(): TelegramTransportStamp; isActive(stamp: TelegramTransportStamp | undefined): boolean; } export interface TelegramQueueItemBase { kind: TelegramQueueItemKind; chatId: number; target?: TelegramQueueTarget; transportStamp?: TelegramTransportStamp; replyToMessageId: number; guestQueryId?: string; queueOrder: number; queueLane: TelegramQueueLane; laneOrder: number; statusSummary: string; } export interface PendingTelegramTurn extends TelegramQueueItemBase { kind: "prompt"; sourceMessageIds: number[]; queuedAttachments: QueuedAttachment[]; content: TelegramPromptContent[]; historyText: string; priorityEmoji?: string; /** Turn should preferably be delivered as voice (mirror mode + user sent voice) */ voiceReplyPreferred?: boolean; /** Turn must be delivered as voice (voice mode) */ voiceReplyRequired?: boolean; } export interface PendingTelegramControlItem< TContext = unknown, > extends TelegramQueueItemBase { kind: "control"; controlType: "status" | "model"; execute: (ctx: TContext) => Promise; } export type TelegramQueueItem = PendingTelegramTurn | PendingTelegramControlItem; export interface TelegramQueueStore { getQueuedItems: () => TelegramQueueItem[]; setQueuedItems: (items: TelegramQueueItem[]) => void; } export interface TelegramQueueStateStore< TContext = unknown, > extends TelegramQueueStore { hasQueuedItems: () => boolean; } export interface TelegramActiveTurnStore< TTurn extends PendingTelegramTurn = PendingTelegramTurn, > { get: () => TTurn | undefined; has: () => boolean; set: (turn: TTurn) => void; clear: () => void; getChatId: () => number | undefined; getTarget: () => TelegramQueueTarget | undefined; getReplyToMessageId: () => number | undefined; getGuestQueryId: () => string | undefined; getSourceMessageIds: () => number[] | undefined; } export interface TelegramDispatchGuardState { compactionInProgress: boolean; hasActiveTelegramTurn: boolean; hasPendingTelegramDispatch: boolean; isIdle: boolean; hasPendingMessages: boolean; } export function getTelegramQueueLaneContract( lane: TelegramQueueLane, ): TelegramQueueLaneContract { const contract = TELEGRAM_QUEUE_LANE_CONTRACTS.find( (entry) => entry.lane === lane, ); if (!contract) throw new Error(`Unknown Telegram queue lane: ${lane}`); return contract; } export function getTelegramQueueItemAdmissionMode( item: Pick, ): TelegramQueueAdmissionMode { return getTelegramQueueLaneContract(item.queueLane).admissionMode; } export function isTelegramQueueItemAdmissionValid( item: Pick, ): boolean { return getTelegramQueueLaneContract(item.queueLane).allowedKinds.includes( item.kind, ); } export function assertTelegramQueueItemAdmissionValid( item: Pick, ): void { if (isTelegramQueueItemAdmissionValid(item)) return; throw new Error( `Invalid Telegram queue admission: ${item.kind} item cannot use ${item.queueLane} lane`, ); } function getTelegramQueueLaneRank(lane: TelegramQueueLane): number { return getTelegramQueueLaneContract(lane).dispatchRank; } export function isPendingTelegramTurn( item: TelegramQueueItem, ): item is PendingTelegramTurn { return item.kind === "prompt"; } export function createTelegramQueueStore( initialItems: TelegramQueueItem[] = [], ): TelegramQueueStateStore { let queuedItems = initialItems; return { getQueuedItems: () => queuedItems, setQueuedItems: (items) => { queuedItems = items; }, hasQueuedItems: () => queuedItems.length > 0, }; } export function createTelegramTransportStampRuntime(deps: { getProfileName(): string | undefined; getBotToken(): string | undefined; }): TelegramTransportStampRuntime { let profile: string | undefined; let botToken: string | undefined; let generation = 0; const getStamp = function (): TelegramTransportStamp { const nextProfile = deps.getProfileName() ?? "default"; const nextBotToken = deps.getBotToken(); if (nextProfile !== profile || nextBotToken !== botToken) { profile = nextProfile; botToken = nextBotToken; generation += 1; } return { profile: nextProfile, generation: String(generation) }; }; return { getStamp, isActive(stamp) { if (!stamp) return false; const current = getStamp(); return ( stamp.profile === current.profile && stamp.generation === current.generation ); }, }; } export function createTelegramTransportStampedQueueStore( store: TelegramQueueStateStore, getTransportStamp: () => TelegramTransportStamp, ): TelegramQueueStateStore { return { getQueuedItems: store.getQueuedItems, hasQueuedItems: store.hasQueuedItems, setQueuedItems(items) { const stamp = getTransportStamp(); store.setQueuedItems( items.map((item) => item.transportStamp ? item : { ...item, transportStamp: stamp }, ), ); }, }; } export function createTelegramQueueItemCountGetter( store: Pick, "getQueuedItems">, ): () => number { return () => { return store.getQueuedItems().length; }; } export function createTelegramActiveTurnStore< TTurn extends PendingTelegramTurn = PendingTelegramTurn, >(): TelegramActiveTurnStore { let activeTurn: TTurn | undefined; return { get: () => activeTurn, has: () => !!activeTurn, set: (turn) => { activeTurn = { ...turn }; }, clear: () => { activeTurn = undefined; }, getChatId: () => activeTurn?.chatId, getTarget: () => activeTurn?.target ? { ...activeTurn.target } : undefined, getReplyToMessageId: () => activeTurn?.replyToMessageId, getGuestQueryId: () => activeTurn?.guestQueryId, getSourceMessageIds: () => activeTurn?.sourceMessageIds, }; } // --- Queue Mutations --- export function partitionTelegramQueueItemsForHistory( items: TelegramQueueItem[], ): { historyTurns: PendingTelegramTurn[]; remainingItems: TelegramQueueItem[]; } { const historyTurns: PendingTelegramTurn[] = []; const remainingItems: TelegramQueueItem[] = []; for (const item of items) { if (isPendingTelegramTurn(item)) { historyTurns.push(item); continue; } remainingItems.push(item); } return { historyTurns, remainingItems }; } export function planTelegramPromptEnqueue( items: TelegramQueueItem[], foldQueuedPromptsIntoHistory: boolean, ): { historyTurns: PendingTelegramTurn[]; remainingItems: TelegramQueueItem[]; } { if (!foldQueuedPromptsIntoHistory) { return { historyTurns: [], remainingItems: items }; } return partitionTelegramQueueItemsForHistory(items); } export function appendTelegramQueueItem< TContext = unknown, TItem extends TelegramQueueItem = TelegramQueueItem, >( items: TelegramQueueItem[], item: TItem, ): TelegramQueueItem[] { assertTelegramQueueItemAdmissionValid(item); return [...items, item]; } function getTelegramPromptTextSignature(item: PendingTelegramTurn): string { return item.content .filter( (entry): entry is TelegramPromptTextContent => entry.type === "text", ) .map((entry) => entry.text) .join("\n"); } function isDuplicateTelegramPromptTurn( left: PendingTelegramTurn, right: PendingTelegramTurn, ): boolean { return ( left.chatId === right.chatId && left.target?.threadId === right.target?.threadId && left.replyToMessageId === right.replyToMessageId && getTelegramPromptTextSignature(left) === getTelegramPromptTextSignature(right) ); } export function appendTelegramPromptTurnOnce( items: TelegramQueueItem[], turn: PendingTelegramTurn, ): { items: TelegramQueueItem[]; appended: boolean } { assertTelegramQueueItemAdmissionValid(turn); const duplicate = items.some( (item) => isPendingTelegramTurn(item) && isDuplicateTelegramPromptTurn(item, turn), ); if (duplicate) return { items, appended: false }; return { items: [...items, turn], appended: true }; } export function compareTelegramQueueItems( left: TelegramQueueItem, right: TelegramQueueItem, ): number { assertTelegramQueueItemAdmissionValid(left); assertTelegramQueueItemAdmissionValid(right); const laneRankDelta = getTelegramQueueLaneRank(left.queueLane) - getTelegramQueueLaneRank(right.queueLane); if (laneRankDelta !== 0) return laneRankDelta; if (left.laneOrder !== right.laneOrder) { return left.laneOrder - right.laneOrder; } return left.queueOrder - right.queueOrder; } export interface TelegramQueueMessageScope { chatId?: number; threadId?: number; } function isTelegramQueueItemInMessageScope( item: TelegramQueueItem, scope: TelegramQueueMessageScope | undefined, ): boolean { if (!scope) return true; if (typeof scope.chatId === "number" && item.chatId !== scope.chatId) { return false; } if (typeof scope.threadId === "number") { return item.target?.threadId === scope.threadId; } return true; } export function removeTelegramQueueItemsByMessageIds( items: TelegramQueueItem[], messageIds: number[], scope?: TelegramQueueMessageScope, ): { items: TelegramQueueItem[]; removedCount: number } { if (messageIds.length === 0 || items.length === 0) { return { items, removedCount: 0 }; } const deletedMessageIds = new Set(messageIds); const nextItems = items.filter((item) => { if ( !isPendingTelegramTurn(item) || !isTelegramQueueItemInMessageScope(item, scope) ) return true; return !item.sourceMessageIds.some((messageId) => deletedMessageIds.has(messageId), ); }); return { items: nextItems, removedCount: items.length - nextItems.length, }; } export function clearTelegramQueuePromptPriority( items: TelegramQueueItem[], messageId: number, scope?: TelegramQueueMessageScope, ): { items: TelegramQueueItem[]; changed: boolean } { let changed = false; const nextItems = items.map((item) => { if ( !isPendingTelegramTurn(item) || !isTelegramQueueItemInMessageScope(item, scope) || !item.sourceMessageIds.includes(messageId) || item.queueLane !== "priority" ) { return item; } changed = true; return { ...item, queueLane: "default" as const, laneOrder: item.queueOrder, priorityEmoji: undefined, }; }); return { items: nextItems, changed }; } export function prioritizeTelegramQueuePrompt( items: TelegramQueueItem[], messageId: number, laneOrder: number, priorityEmoji = "⚡", scope?: TelegramQueueMessageScope, ): { items: TelegramQueueItem[]; changed: boolean } { let changed = false; const nextItems = items.map((item) => { if ( !isPendingTelegramTurn(item) || !isTelegramQueueItemInMessageScope(item, scope) || !item.sourceMessageIds.includes(messageId) ) { return item; } changed = true; return { ...item, queueLane: "priority" as const, laneOrder, priorityEmoji, }; }); return { items: nextItems, changed }; } export function consumeDispatchedTelegramPrompt( items: TelegramQueueItem[], hasPendingDispatch: boolean, ): { activeTurn?: PendingTelegramTurn; remainingItems: TelegramQueueItem[]; } { if (!hasPendingDispatch) { return { activeTurn: undefined, remainingItems: items }; } const nextItem = items[0]; if (!nextItem || !isPendingTelegramTurn(nextItem)) { return { activeTurn: undefined, remainingItems: items }; } return { activeTurn: nextItem, remainingItems: items.slice(1) }; } export function formatQueuedTelegramItemsStatus( items: TelegramQueueItem[], ): string { return items.length === 0 ? "" : ` +${items.length}`; } export function truncateTelegramQueueSummary( text: string, maxWords = 5, maxLength = 40, ): string { const normalized = text.replace(/\s+/g, " ").trim(); if (!normalized) return ""; const words = normalized.split(" "); let summary = words.slice(0, maxWords).join(" "); if (summary.length === 0) summary = normalized; if (summary.length > maxLength) { summary = summary.slice(0, maxLength).trimEnd(); } return summary.length < normalized.length || words.length > maxWords ? `${summary}…` : summary; } export function canDispatchTelegramTurnState( state: TelegramDispatchGuardState, ): boolean { return ( !state.compactionInProgress && !state.hasActiveTelegramTurn && !state.hasPendingTelegramDispatch && state.isIdle && !state.hasPendingMessages ); } export interface TelegramDispatchReadinessDeps { isCompactionInProgress: () => boolean; hasActiveTurn: () => boolean; hasDispatchPending: () => boolean; isIdle: (ctx: TContext) => boolean; hasPendingMessages: (ctx: TContext) => boolean; } export function createTelegramDispatchReadinessChecker( deps: TelegramDispatchReadinessDeps, ): (ctx: TContext) => boolean { return (ctx) => canDispatchTelegramTurnState({ compactionInProgress: deps.isCompactionInProgress(), hasActiveTelegramTurn: deps.hasActiveTurn(), hasPendingTelegramDispatch: deps.hasDispatchPending(), isIdle: deps.isIdle(ctx), hasPendingMessages: deps.hasPendingMessages(ctx), }); } export function buildPendingTelegramControlItem(options: { chatId: number; target?: TelegramQueueTarget; replyToMessageId: number; controlType: PendingTelegramControlItem["controlType"]; queueOrder: number; laneOrder: number; statusSummary: string; execute: PendingTelegramControlItem["execute"]; }): PendingTelegramControlItem { return { kind: "control", controlType: options.controlType, chatId: options.chatId, ...(options.target ? { target: options.target } : {}), replyToMessageId: options.replyToMessageId, queueOrder: options.queueOrder, queueLane: "control", laneOrder: options.laneOrder, statusSummary: options.statusSummary, execute: options.execute, }; } export interface TelegramControlItemBuilderDeps { allocateItemOrder: () => number; allocateControlOrder: () => number; } export function createTelegramControlItemBuilder( deps: TelegramControlItemBuilderDeps, ): (options: { chatId: number; target?: TelegramQueueTarget; replyToMessageId: number; controlType: PendingTelegramControlItem["controlType"]; statusSummary: string; execute: PendingTelegramControlItem["execute"]; }) => PendingTelegramControlItem { return (options) => buildPendingTelegramControlItem({ ...options, queueOrder: deps.allocateItemOrder(), laneOrder: deps.allocateControlOrder(), }); } // --- Dispatch Planning --- export type TelegramQueueDispatchAction = | { kind: "none"; remainingItems: TelegramQueueItem[] } | { kind: "control"; item: PendingTelegramControlItem; remainingItems: TelegramQueueItem[]; } | { kind: "prompt"; item: PendingTelegramTurn; remainingItems: TelegramQueueItem[]; }; export function planNextTelegramQueueAction( items: TelegramQueueItem[], canDispatch: boolean, ): TelegramQueueDispatchAction { if (!canDispatch || items.length === 0) { return { kind: "none", remainingItems: items }; } const [firstItem, ...remainingItems] = items; if (!firstItem) { return { kind: "none", remainingItems: items }; } assertTelegramQueueItemAdmissionValid(firstItem); if (isPendingTelegramTurn(firstItem)) { return { kind: "prompt", item: firstItem, remainingItems: items }; } return { kind: "control", item: firstItem, remainingItems }; } export function shouldDispatchAfterTelegramAgentEnd(options: { hasTurn: boolean; stopReason?: string; foldQueuedPromptsIntoHistory: boolean; }): boolean { if (!options.hasTurn) return true; if (options.stopReason === "aborted") { return !options.foldQueuedPromptsIntoHistory; } return true; } // --- Agent Runtime --- export interface TelegramAgentStartPlan { activeTurn?: PendingTelegramTurn; remainingItems: TelegramQueueItem[]; shouldResetPendingModelSwitch: boolean; shouldResetToolExecutions: boolean; shouldClearDispatchPending: boolean; shouldClearAbortHistory: boolean; } export interface TelegramAgentStartRuntimeDeps< TTurn extends PendingTelegramTurn, TContext = unknown, > { queuedItems: TelegramQueueItem[]; hasPendingDispatch: boolean; hasActiveTurn: boolean; resetToolExecutions: () => void; resetPendingModelSwitch: () => void; setQueuedItems: (items: TelegramQueueItem[]) => void; clearDispatchPending: () => void; setFoldQueuedPromptsIntoHistory: (fold: boolean) => void; setActiveTurn: (turn: TTurn) => void; createPreviewState: () => void; startTypingLoop: () => void; updateStatus: () => void; } export interface TelegramAgentStartHookRuntimeDeps< TTurn extends PendingTelegramTurn, TContext = unknown, > { setAbortHandler: (ctx: TContext) => void; getQueuedItems: () => TelegramQueueItem[]; hasPendingDispatch: () => boolean; hasActiveTurn: () => boolean; resetToolExecutions: () => void; resetPendingModelSwitch: () => void; setQueuedItems: (items: TelegramQueueItem[]) => void; clearDispatchPending: () => void; setFoldQueuedPromptsIntoHistory: (fold: boolean) => void; setActiveTurn: (turn: TTurn) => void; createPreviewState: () => void; startTypingLoop: (ctx: TContext) => void; updateStatus: (ctx: TContext) => void; } export type TelegramAgentStartHookEvent = unknown; export interface TelegramToolExecutionRuntimeDeps { hasActiveTurn: () => boolean; getActiveToolExecutions: () => number; setActiveToolExecutions: (count: number) => void; } export interface TelegramToolExecutionEndRuntimeDeps extends TelegramToolExecutionRuntimeDeps { triggerPendingModelSwitchAbort: () => void; } export interface TelegramToolExecutionHookRuntimeDeps< TContext, > extends TelegramToolExecutionRuntimeDeps { triggerPendingModelSwitchAbort: (ctx: TContext) => unknown; } export type TelegramToolExecutionHookEvent = unknown; export function buildTelegramAgentStartPlan(options: { queuedItems: TelegramQueueItem[]; hasPendingDispatch: boolean; hasActiveTurn: boolean; }): TelegramAgentStartPlan { if (options.hasActiveTurn || !options.hasPendingDispatch) { return { activeTurn: undefined, remainingItems: options.queuedItems, shouldResetPendingModelSwitch: true, shouldResetToolExecutions: true, shouldClearDispatchPending: options.hasPendingDispatch, shouldClearAbortHistory: !options.hasActiveTurn && !options.hasPendingDispatch, }; } const nextDispatch = consumeDispatchedTelegramPrompt( options.queuedItems, options.hasPendingDispatch, ); return { activeTurn: nextDispatch.activeTurn, remainingItems: nextDispatch.remainingItems, shouldResetPendingModelSwitch: true, shouldResetToolExecutions: true, shouldClearDispatchPending: options.hasPendingDispatch, shouldClearAbortHistory: false, }; } export function handleTelegramAgentStartRuntime< TTurn extends PendingTelegramTurn, TContext = unknown, >(deps: TelegramAgentStartRuntimeDeps): void { const startPlan = buildTelegramAgentStartPlan({ queuedItems: deps.queuedItems, hasPendingDispatch: deps.hasPendingDispatch, hasActiveTurn: deps.hasActiveTurn, }); if (startPlan.shouldResetToolExecutions) deps.resetToolExecutions(); if (startPlan.shouldResetPendingModelSwitch) deps.resetPendingModelSwitch(); if (startPlan.shouldClearAbortHistory) { deps.setFoldQueuedPromptsIntoHistory(false); } deps.setQueuedItems(startPlan.remainingItems); if (startPlan.shouldClearDispatchPending) deps.clearDispatchPending(); if (startPlan.activeTurn) { deps.setActiveTurn(startPlan.activeTurn as TTurn); deps.createPreviewState(); deps.startTypingLoop(); } deps.updateStatus(); } export function createTelegramAgentStartHook< TTurn extends PendingTelegramTurn, TContext = unknown, >(deps: TelegramAgentStartHookRuntimeDeps) { return async ( _event: TelegramAgentStartHookEvent, ctx: TContext, ): Promise => { deps.setAbortHandler(ctx); handleTelegramAgentStartRuntime({ queuedItems: deps.getQueuedItems(), hasPendingDispatch: deps.hasPendingDispatch(), hasActiveTurn: deps.hasActiveTurn(), resetToolExecutions: deps.resetToolExecutions, resetPendingModelSwitch: deps.resetPendingModelSwitch, setQueuedItems: deps.setQueuedItems, clearDispatchPending: deps.clearDispatchPending, setFoldQueuedPromptsIntoHistory: deps.setFoldQueuedPromptsIntoHistory, setActiveTurn: deps.setActiveTurn, createPreviewState: deps.createPreviewState, startTypingLoop: () => deps.startTypingLoop(ctx), updateStatus: () => deps.updateStatus(ctx), }); }; } export function getNextTelegramToolExecutionCount(options: { hasActiveTurn: boolean; currentCount: number; event: "start" | "end"; }): number { if (!options.hasActiveTurn) return options.currentCount; if (options.event === "start") { return options.currentCount + 1; } return Math.max(0, options.currentCount - 1); } export function handleTelegramToolExecutionStartRuntime( deps: TelegramToolExecutionRuntimeDeps, ): void { deps.setActiveToolExecutions( getNextTelegramToolExecutionCount({ hasActiveTurn: deps.hasActiveTurn(), currentCount: deps.getActiveToolExecutions(), event: "start", }), ); } export function handleTelegramToolExecutionEndRuntime( deps: TelegramToolExecutionEndRuntimeDeps, ): void { const hasActiveTurn = deps.hasActiveTurn(); deps.setActiveToolExecutions( getNextTelegramToolExecutionCount({ hasActiveTurn, currentCount: deps.getActiveToolExecutions(), event: "end", }), ); if (hasActiveTurn) deps.triggerPendingModelSwitchAbort(); } export type TelegramAgentLifecycleHooksRuntimeDeps< TTurn extends PendingTelegramTurn, TContext, TMessage, TReplyMarkup = unknown, > = TelegramAgentStartHookRuntimeDeps & TelegramAgentEndHookRuntimeDeps & TelegramToolExecutionHookRuntimeDeps; export function createTelegramAgentLifecycleHooks< TTurn extends PendingTelegramTurn, TContext, TMessage, TReplyMarkup = unknown, >( deps: TelegramAgentLifecycleHooksRuntimeDeps< TTurn, TContext, TMessage, TReplyMarkup >, ) { return { onAgentStart: createTelegramAgentStartHook(deps), onAgentEnd: createTelegramAgentEndHook< TTurn, TContext, TMessage, TReplyMarkup >(deps), ...createTelegramToolExecutionHooks(deps), }; } export function createTelegramToolExecutionHooks( deps: TelegramToolExecutionHookRuntimeDeps, ) { return { onToolExecutionStart: (): void => { handleTelegramToolExecutionStartRuntime(deps); }, onToolExecutionEnd: ( _event: TelegramToolExecutionHookEvent, ctx: TContext, ): void => { handleTelegramToolExecutionEndRuntime({ hasActiveTurn: deps.hasActiveTurn, getActiveToolExecutions: deps.getActiveToolExecutions, setActiveToolExecutions: deps.setActiveToolExecutions, triggerPendingModelSwitchAbort: () => { deps.triggerPendingModelSwitchAbort(ctx); }, }); }, }; } // --- Agent End Lifecycle --- export interface TelegramAgentEndPlan { kind: "no-turn" | "aborted" | "error" | "text" | "attachments-only" | "empty"; shouldClearPreview: boolean; shouldDispatchNext: boolean; shouldSendErrorMessage: boolean; shouldSendAttachmentNotice: boolean; } export interface TelegramAgentEndAssistantResult { text?: string; stopReason?: string; errorMessage?: string; } export interface TelegramAgentEndOutboundVoiceReply { text: string; lang?: string; rate?: string; } export interface TelegramAgentEndOutboundReplyPlan { markdown: string; replyMarkup?: TReplyMarkup; voiceText?: string; voiceReplies?: TelegramAgentEndOutboundVoiceReply[]; lang?: string; rate?: string; } export interface TelegramAgentEndRuntimeDeps< TTurn extends PendingTelegramTurn, TReplyMarkup = unknown, > { turn: TTurn | undefined; assistant: TelegramAgentEndAssistantResult; foldQueuedPromptsIntoHistory: boolean; resetRuntimeState: () => void; isSessionActive?: () => boolean; isTurnTransportActive?: (turn: TTurn) => boolean; waitForTypingIdle?: () => Promise; updateStatus: () => void; dispatchNextQueuedTelegramTurn: () => void; scheduleActiveTurnDelivery?: (task: () => Promise) => void; clearPreview: ( chatId: number, options?: { target?: TelegramQueueTarget }, ) => Promise; setPreviewPendingText: (text: string) => void; finalizeMarkdownPreview: ( chatId: number, markdown: string, replyToMessageId: number, options?: { replyMarkup?: TReplyMarkup; target?: TelegramQueueTarget }, ) => Promise; sendMarkdownReply: ( chatId: number, replyToMessageId: number | undefined, markdown: string, options?: { replyMarkup?: TReplyMarkup; target?: TelegramQueueTarget }, ) => Promise; sendTextReply: ( chatId: number, replyToMessageId: number, text: string, options?: { target?: TelegramQueueTarget }, ) => Promise; sendQueuedAttachments: (turn: TTurn) => Promise; answerGuestQuery?: ( guestQueryId: string, text?: string, options?: { parseMode?: string }, ) => Promise; sendGuestReply?: (guestQueryId: string, markdown: string) => Promise; sendGuestAttachment?: ( turn: TTurn, attachment: QueuedAttachment, caption?: string, ) => Promise; sendGuestVoiceReply?: ( turn: TTurn, plan: TelegramAgentEndOutboundReplyPlan, caption?: string, ) => Promise; planOutboundReply?: ( markdown: string, ) => TelegramAgentEndOutboundReplyPlan; sendOutboundReplyArtifacts?: ( turn: TTurn, plan: TelegramAgentEndOutboundReplyPlan, options?: { replyToPrompt?: boolean }, ) => Promise; getDefaultChatId?: () => number | undefined; getDefaultTarget?: () => TelegramQueueTarget | undefined; isProactivePushEnabled?: () => boolean; canSendProactivePush?: () => boolean; recordRuntimeEvent?: ( category: string, error: unknown, details?: Record, ) => void; } export interface TelegramAgentEndHookRuntimeDeps< TTurn extends PendingTelegramTurn, TContext, TMessage, TReplyMarkup = unknown, > { getActiveTurn: () => TTurn | undefined; loadConfig?: () => Promise; extractAssistant: ( messages: readonly TMessage[], ) => TelegramAgentEndAssistantResult; getFoldQueuedPromptsIntoHistory: () => boolean; resetRuntimeState: () => void; isSessionActive?: (ctx: TContext) => boolean; isTurnTransportActive?: (turn: TTurn) => boolean; waitForTypingIdle?: () => Promise; updateStatus: (ctx: TContext) => void; dispatchNextQueuedTelegramTurn: (ctx: TContext) => void; requestDeferredDispatchNextQueuedTelegramTurn: ( dispatch: (ctx: TContext) => void, ) => void; scheduleActiveTurnDelivery?: TelegramAgentEndRuntimeDeps< TTurn, TReplyMarkup >["scheduleActiveTurnDelivery"]; clearPreview: TelegramAgentEndRuntimeDeps< TTurn, TReplyMarkup >["clearPreview"]; setPreviewPendingText: (text: string) => void; finalizeMarkdownPreview: TelegramAgentEndRuntimeDeps< TTurn, TReplyMarkup >["finalizeMarkdownPreview"]; sendMarkdownReply: TelegramAgentEndRuntimeDeps< TTurn, TReplyMarkup >["sendMarkdownReply"]; sendTextReply: TelegramAgentEndRuntimeDeps["sendTextReply"]; sendQueuedAttachments: (turn: TTurn) => Promise; answerGuestQuery?: TelegramAgentEndRuntimeDeps["answerGuestQuery"]; sendGuestReply?: TelegramAgentEndRuntimeDeps["sendGuestReply"]; sendGuestAttachment?: TelegramAgentEndRuntimeDeps["sendGuestAttachment"]; sendGuestVoiceReply?: TelegramAgentEndRuntimeDeps["sendGuestVoiceReply"]; planOutboundReply?: TelegramAgentEndRuntimeDeps< TTurn, TReplyMarkup >["planOutboundReply"]; sendOutboundReplyArtifacts?: TelegramAgentEndRuntimeDeps["sendOutboundReplyArtifacts"]; getDefaultChatId?: TelegramAgentEndRuntimeDeps["getDefaultChatId"]; getDefaultTarget?: TelegramAgentEndRuntimeDeps["getDefaultTarget"]; isProactivePushEnabled?: TelegramAgentEndRuntimeDeps["isProactivePushEnabled"]; canSendProactivePush?: (ctx: TContext) => boolean; recordRuntimeEvent?: TelegramAgentEndRuntimeDeps["recordRuntimeEvent"]; } export interface TelegramAgentEndHookEvent { messages: readonly TMessage[]; } export function buildTelegramAgentEndPlan(options: { hasTurn: boolean; stopReason?: string; hasFinalText: boolean; hasQueuedAttachments: boolean; foldQueuedPromptsIntoHistory: boolean; }): TelegramAgentEndPlan { const shouldDispatchNext = shouldDispatchAfterTelegramAgentEnd({ hasTurn: options.hasTurn, stopReason: options.stopReason, foldQueuedPromptsIntoHistory: options.foldQueuedPromptsIntoHistory, }); if (!options.hasTurn) { return { kind: "no-turn", shouldClearPreview: false, shouldDispatchNext, shouldSendErrorMessage: false, shouldSendAttachmentNotice: false, }; } if (options.stopReason === "aborted") { return { kind: "aborted", shouldClearPreview: true, shouldDispatchNext, shouldSendErrorMessage: false, shouldSendAttachmentNotice: false, }; } if (options.stopReason === "error") { return { kind: "error", shouldClearPreview: true, shouldDispatchNext, shouldSendErrorMessage: true, shouldSendAttachmentNotice: false, }; } if (options.hasFinalText) { return { kind: "text", shouldClearPreview: false, shouldDispatchNext, shouldSendErrorMessage: false, shouldSendAttachmentNotice: false, }; } if (options.hasQueuedAttachments) { return { kind: "attachments-only", shouldClearPreview: true, shouldDispatchNext, shouldSendErrorMessage: false, shouldSendAttachmentNotice: true, }; } return { kind: "empty", shouldClearPreview: true, shouldDispatchNext, shouldSendErrorMessage: false, shouldSendAttachmentNotice: false, }; } export function createTelegramAgentEndHook< TTurn extends PendingTelegramTurn, TContext, TMessage, TReplyMarkup = unknown, >( deps: TelegramAgentEndHookRuntimeDeps< TTurn, TContext, TMessage, TReplyMarkup >, ) { return async ( event: TelegramAgentEndHookEvent, ctx: TContext, ): Promise => { await deps.loadConfig?.(); if (deps.isSessionActive && !deps.isSessionActive(ctx)) return; const turn = deps.getActiveTurn(); const proactiveEnabled = deps.isProactivePushEnabled?.() ?? false; const canProactivePush = deps.canSendProactivePush?.(ctx) ?? false; await handleTelegramAgentEndRuntime({ turn, assistant: turn || proactiveEnabled ? deps.extractAssistant(event.messages) : {}, foldQueuedPromptsIntoHistory: deps.getFoldQueuedPromptsIntoHistory(), resetRuntimeState: deps.resetRuntimeState, isSessionActive: () => deps.isSessionActive?.(ctx) ?? true, isTurnTransportActive: deps.isTurnTransportActive, waitForTypingIdle: deps.waitForTypingIdle, updateStatus: () => deps.updateStatus(ctx), dispatchNextQueuedTelegramTurn: () => { deps.requestDeferredDispatchNextQueuedTelegramTurn( deps.dispatchNextQueuedTelegramTurn, ); }, scheduleActiveTurnDelivery: deps.scheduleActiveTurnDelivery ? (task) => deps.scheduleActiveTurnDelivery?.(async () => { if (deps.isSessionActive?.(ctx) === false) return; await task(); }) : undefined, clearPreview: deps.clearPreview, setPreviewPendingText: deps.setPreviewPendingText, finalizeMarkdownPreview: deps.finalizeMarkdownPreview, sendMarkdownReply: deps.sendMarkdownReply, sendTextReply: deps.sendTextReply, sendQueuedAttachments: deps.sendQueuedAttachments, answerGuestQuery: deps.answerGuestQuery, sendGuestReply: deps.sendGuestReply, sendGuestAttachment: deps.sendGuestAttachment, sendGuestVoiceReply: deps.sendGuestVoiceReply, planOutboundReply: deps.planOutboundReply, sendOutboundReplyArtifacts: deps.sendOutboundReplyArtifacts, getDefaultChatId: deps.getDefaultChatId, getDefaultTarget: deps.getDefaultTarget, isProactivePushEnabled: deps.isProactivePushEnabled, canSendProactivePush: () => canProactivePush, recordRuntimeEvent: deps.recordRuntimeEvent, }); }; } export async function handleTelegramAgentEndRuntime< TTurn extends PendingTelegramTurn, TReplyMarkup = unknown, >(deps: TelegramAgentEndRuntimeDeps): Promise { const { turn, assistant } = deps; const rawFinalText = assistant.text; let outboundReply = rawFinalText ? deps.planOutboundReply?.(rawFinalText) : undefined; // Preserve the planned reply so voice-fallback can use stripped markdown + replyMarkup const plannedReply = outboundReply; // Transparent voice interception: when the turn is voice-tagged and the agent // did not explicitly use markup, we automatically // convert the whole response to voice. const voiceInterceptionGuard = turn && isVoiceTurn(turn) && rawFinalText?.trim() && deps.planOutboundReply && (!outboundReply || (!outboundReply.voiceText && !outboundReply.voiceReplies?.length)); if (voiceInterceptionGuard) { const voiceText = plannedReply !== undefined ? plannedReply.markdown?.trim() || "" : (rawFinalText ?? ""); outboundReply = outboundReply ? { ...outboundReply, voiceText, markdown: "" } : { markdown: "", voiceText }; } const finalText = outboundReply ? outboundReply.markdown : rawFinalText; const hasOutboundArtifacts = !!outboundReply?.voiceText || !!outboundReply?.voiceReplies?.length; const replyMarkup = outboundReply?.replyMarkup; const isDeliveryActive = (): boolean => deps.isSessionActive?.() !== false && (!turn || deps.isTurnTransportActive?.(turn) !== false); if (!isDeliveryActive()) { deps.resetRuntimeState(); deps.updateStatus(); deps.dispatchNextQueuedTelegramTurn(); return; } deps.resetRuntimeState(); await deps.waitForTypingIdle?.(); if (!isDeliveryActive()) { deps.updateStatus(); deps.dispatchNextQueuedTelegramTurn(); return; } deps.updateStatus(); const endPlan = buildTelegramAgentEndPlan({ hasTurn: !!turn, stopReason: assistant.stopReason, hasFinalText: !!finalText || hasOutboundArtifacts, hasQueuedAttachments: (turn?.queuedAttachments.length ?? 0) > 0, foldQueuedPromptsIntoHistory: deps.foldQueuedPromptsIntoHistory, }); if (!turn) { const proactiveEnabled = deps.isProactivePushEnabled?.() ?? false; const canProactivePush = deps.canSendProactivePush?.() ?? false; if (proactiveEnabled && finalText && !assistant.errorMessage) { if (canProactivePush) { const defaultTarget = deps.getDefaultTarget?.(); const defaultChatId = defaultTarget?.chatId ?? deps.getDefaultChatId?.(); if (defaultChatId !== undefined) { try { await deps.sendMarkdownReply(defaultChatId, undefined, finalText, { target: defaultTarget, }); } catch (error) { deps.recordRuntimeEvent?.("proactive-push", error, { chatId: defaultChatId, threadId: defaultTarget?.threadId, }); } } } else { deps.recordRuntimeEvent?.( "proactive-push", new Error( "Proactive push skipped because this instance does not own Telegram polling.", ), { phase: "ownership" }, ); } } if (endPlan.shouldDispatchNext) deps.dispatchNextQueuedTelegramTurn(); return; } if (turn.guestQueryId) { if (assistant.errorMessage) { await deps.answerGuestQuery?.( turn.guestQueryId, "Telegram bridge: Pi failed while processing the request.", ); if (endPlan.shouldDispatchNext) deps.dispatchNextQueuedTelegramTurn(); return; } const [guestAttachment] = turn.queuedAttachments; if (guestAttachment && deps.sendGuestAttachment) { try { await deps.sendGuestAttachment( turn, guestAttachment, finalText || undefined, ); } catch (error) { deps.recordRuntimeEvent?.("delivery", error, { phase: "guest-attachment", guestQueryId: turn.guestQueryId, }); } } else if ( outboundReply && (outboundReply.voiceText || outboundReply.voiceReplies?.length) && deps.sendGuestVoiceReply ) { try { await deps.sendGuestVoiceReply( turn, outboundReply, finalText || undefined, ); } catch (error) { deps.recordRuntimeEvent?.("delivery", error, { phase: "guest-voice", guestQueryId: turn.guestQueryId, }); } } else if (finalText) { if (deps.sendGuestReply) { await deps.sendGuestReply(turn.guestQueryId, finalText); } else { await deps.answerGuestQuery?.(turn.guestQueryId, finalText); } } if (!isDeliveryActive()) return; if (endPlan.shouldDispatchNext) deps.dispatchNextQueuedTelegramTurn(); return; } if (endPlan.shouldClearPreview) { await deps.clearPreview(turn.chatId, { target: turn.target }); } if (!isDeliveryActive()) return; if (endPlan.shouldSendErrorMessage) { await deps.sendTextReply( turn.chatId, turn.replyToMessageId, assistant.errorMessage || "Telegram bridge: Pi failed while processing the request.", { target: turn.target }, ); if (!isDeliveryActive()) return; if (endPlan.shouldDispatchNext) deps.dispatchNextQueuedTelegramTurn(); return; } const deliverActiveTurn = async () => { if (!isDeliveryActive()) return; if (finalText) deps.setPreviewPendingText(finalText); if (!finalText && hasOutboundArtifacts) await deps.clearPreview(turn.chatId, { target: turn.target }); if (!isDeliveryActive()) return; if (endPlan.kind === "text" && finalText) { try { const finalized = await deps.finalizeMarkdownPreview( turn.chatId, finalText, turn.replyToMessageId, { replyMarkup, target: turn.target }, ); if (!isDeliveryActive()) return; if (!finalized) { await deps.clearPreview(turn.chatId, { target: turn.target }); if (!isDeliveryActive()) return; await deps.sendMarkdownReply( turn.chatId, turn.replyToMessageId, finalText, { replyMarkup, target: turn.target }, ); } } catch (error) { deps.recordRuntimeEvent?.("delivery", error, { phase: "final-text", chatId: turn.chatId, replyToMessageId: turn.replyToMessageId, }); } } if (!isDeliveryActive()) return; if (outboundReply && deps.sendOutboundReplyArtifacts) { try { await deps.sendOutboundReplyArtifacts(turn, outboundReply, { replyToPrompt: !finalText, }); if (!isDeliveryActive()) return; } catch (error) { deps.recordRuntimeEvent?.("delivery", error, { phase: "voice-artifacts", chatId: turn.chatId, }); // Fallback to planned text when voice delivery fails and text wasn't already delivered if (!isDeliveryActive()) return; if (rawFinalText?.trim() && !finalText && hasOutboundArtifacts) { try { const fallbackMarkdown = plannedReply?.markdown || outboundReply?.voiceText || rawFinalText; await deps.sendMarkdownReply( turn.chatId, turn.replyToMessageId, fallbackMarkdown, plannedReply?.replyMarkup || turn.target ? { replyMarkup: plannedReply?.replyMarkup, target: turn.target, } : undefined, ); } catch (fallbackError) { deps.recordRuntimeEvent?.("delivery", fallbackError, { phase: "voice-fallback-text", chatId: turn.chatId, }); } } } } if (!isDeliveryActive()) return; if (endPlan.shouldSendAttachmentNotice) { await deps.sendTextReply( turn.chatId, turn.replyToMessageId, "Attached requested file(s).", { target: turn.target }, ); } if (!isDeliveryActive()) return; await deps.sendQueuedAttachments(turn); if (!isDeliveryActive()) return; if (endPlan.shouldDispatchNext) deps.dispatchNextQueuedTelegramTurn(); }; if ( deps.scheduleActiveTurnDelivery && (endPlan.kind === "text" || endPlan.kind === "attachments-only") ) { deps.scheduleActiveTurnDelivery(deliverActiveTurn); return; } await deliverActiveTurn(); } // --- Session Runtime --- export interface TelegramSessionStartState { currentTelegramModel: TModel | undefined; activeTelegramToolExecutions: number; pendingTelegramModelSwitch: undefined; nextQueuedTelegramItemOrder: number; nextQueuedTelegramControlOrder: number; telegramTurnDispatchPending: boolean; compactionInProgress: boolean; } export interface TelegramSessionShutdownState { queuedTelegramItems: TQueueItem[]; nextQueuedTelegramItemOrder: number; nextQueuedTelegramControlOrder: number; nextPriorityReactionOrder: number; currentTelegramModel: undefined; activeTelegramToolExecutions: number; pendingTelegramModelSwitch: undefined; telegramTurnDispatchPending: boolean; compactionInProgress: boolean; foldQueuedPromptsIntoHistory: boolean; } export interface TelegramSessionRuntimeCounterState { nextQueuedTelegramItemOrder?: number; nextQueuedTelegramControlOrder?: number; nextPriorityReactionOrder?: number; } export interface TelegramSessionRuntimeFlagState { activeTelegramToolExecutions?: number; telegramTurnDispatchPending?: boolean; compactionInProgress?: boolean; foldQueuedPromptsIntoHistory?: boolean; } export interface TelegramSessionStateApplierDeps { setQueuedItems: (items: TQueueItem[]) => void; setCurrentModel: (model: TModel | undefined) => void; setPendingModelSwitch: (selection: undefined) => void; syncCounters: (state: TelegramSessionRuntimeCounterState) => void; syncFlags: (state: TelegramSessionRuntimeFlagState) => void; } export interface TelegramSessionStateApplier { applyStartState: (state: TelegramSessionStartState) => void; applyShutdownState: (state: TelegramSessionShutdownState) => void; } export interface TelegramSessionStartRuntimeDeps { ctx: TContext; currentModel: TModel | undefined; loadConfig: () => Promise; isSessionActive?: () => boolean; applyState: (state: TelegramSessionStartState) => void; bindDeferredDispatchContext?: (ctx: TContext) => void; prepareTempDir: () => Promise; updateStatus: () => void; } export interface TelegramSessionShutdownRuntimeDeps { isSessionActive?: () => boolean; unbindDeferredDispatchContext?: () => void; applyState: (state: TelegramSessionShutdownState) => void; clearPendingMediaGroups: () => void; clearModelMenuState: () => void; getActiveTurnChatId: () => number | undefined; getActiveTurnTarget?: () => TelegramQueueTarget | undefined; clearPreview: ( chatId: number, options?: { target?: TelegramQueueTarget }, ) => Promise; previewShutdownTimeoutMs?: number; clearActiveTurn: () => void; clearAbort: () => void; stopPolling: () => Promise; } export interface TelegramSessionLifecycleHookRuntimeDeps< TContext, TQueueItem, TModel = unknown, > extends TelegramRuntimeEventRecorderPort { getCurrentModel: (ctx: TContext) => TModel | undefined; loadConfig: () => Promise; applySessionStartState: (state: TelegramSessionStartState) => void; bindDeferredDispatchContext?: (ctx: TContext) => void; prepareTempDir: () => Promise; updateStatus: (ctx: TContext) => void; isSessionActive?: (ctx: TContext) => boolean; unbindDeferredDispatchContext?: () => void; applySessionShutdownState: ( state: TelegramSessionShutdownState, ) => void; clearPendingMediaGroups: () => void; clearModelMenuState: () => void; getActiveTurnChatId: () => number | undefined; getActiveTurnTarget?: () => TelegramQueueTarget | undefined; clearPreview: ( chatId: number, options?: { target?: TelegramQueueTarget }, ) => Promise; previewShutdownTimeoutMs?: number; clearActiveTurn: () => void; clearAbort: () => void; stopPolling: () => Promise; } export type TelegramSessionLifecycleHookEvent = unknown; export function createTelegramSessionStateApplier( deps: TelegramSessionStateApplierDeps, ): TelegramSessionStateApplier { return { applyStartState: (state) => { deps.setCurrentModel(state.currentTelegramModel); deps.setPendingModelSwitch(state.pendingTelegramModelSwitch); deps.syncCounters(state); deps.syncFlags(state); }, applyShutdownState: (state) => { deps.setQueuedItems(state.queuedTelegramItems); deps.syncCounters(state); deps.syncFlags(state); deps.setCurrentModel(state.currentTelegramModel); deps.setPendingModelSwitch(state.pendingTelegramModelSwitch); }, }; } export interface TelegramQueueMutationRuntimeDeps< TContext, > extends TelegramQueueStore { ctx: TContext; getNextPriorityReactionOrder?: () => number; incrementNextPriorityReactionOrder?: () => void; updateStatus: (ctx: TContext) => void; } export interface TelegramQueueMutationControllerDeps< TContext, > extends TelegramQueueStore { getNextPriorityReactionOrder?: () => number; incrementNextPriorityReactionOrder?: () => void; updateStatus: (ctx: TContext) => void; } export interface TelegramQueueMutationController { append: (item: TelegramQueueItem, ctx: TContext) => void; reorder: (ctx: TContext) => void; clear: (ctx: TContext) => number; removeByMessageIds: ( messageIds: number[], ctx: TContext, scope?: TelegramQueueMessageScope, ) => number; clearPriorityByMessageId: ( messageId: number, ctx: TContext, scope?: TelegramQueueMessageScope, ) => boolean; prioritizeByMessageId: ( messageId: number, ctx: TContext, priorityEmoji?: string, scope?: TelegramQueueMessageScope, ) => boolean; } export interface TelegramControlQueueControllerDeps { appendControlItem: ( item: PendingTelegramControlItem, ctx: TContext, ) => void; dispatchNextQueuedTelegramTurn: (ctx: TContext) => void; } export interface TelegramControlQueueController { enqueue: (item: PendingTelegramControlItem, ctx: TContext) => void; } export interface TelegramPromptEnqueueRuntimeDeps< TMessage, TContext = unknown, > extends TelegramQueueStore { getFoldQueuedPromptsIntoHistory: () => boolean; setFoldQueuedPromptsIntoHistory: (fold: boolean) => void; createTurn: ( messages: TMessage[], historyTurns: PendingTelegramTurn[], ) => Promise; updateStatus: () => void; dispatchNextQueuedTelegramTurn: () => void; } export interface TelegramPromptEnqueueControllerDeps< TMessage, TContext = unknown, > extends TelegramQueueStore { getFoldQueuedPromptsIntoHistory: () => boolean; setFoldQueuedPromptsIntoHistory: (fold: boolean) => void; createTurn: ( messages: TMessage[], historyTurns: PendingTelegramTurn[], ctx: TContext, ) => Promise; updateStatus: (ctx: TContext) => void; dispatchNextQueuedTelegramTurn: (ctx: TContext) => void; } export interface TelegramPromptEnqueueController { enqueue: (messages: TMessage[], ctx: TContext) => Promise; } function isTelegramStaleContextError(error: unknown): boolean { return ( error instanceof Error && (error.message.includes("stale after session") || error.message.includes("stale ctx")) ); } export function buildTelegramSessionStartState( currentModel: TModel | undefined, ): TelegramSessionStartState { return { currentTelegramModel: currentModel, activeTelegramToolExecutions: 0, pendingTelegramModelSwitch: undefined, nextQueuedTelegramItemOrder: 0, nextQueuedTelegramControlOrder: 0, telegramTurnDispatchPending: false, compactionInProgress: false, }; } export function buildTelegramSessionShutdownState< TQueueItem, >(): TelegramSessionShutdownState { return { queuedTelegramItems: [], nextQueuedTelegramItemOrder: 0, nextQueuedTelegramControlOrder: 0, nextPriorityReactionOrder: 0, currentTelegramModel: undefined, activeTelegramToolExecutions: 0, pendingTelegramModelSwitch: undefined, telegramTurnDispatchPending: false, compactionInProgress: false, foldQueuedPromptsIntoHistory: false, }; } export async function startTelegramSessionRuntime( deps: TelegramSessionStartRuntimeDeps, ): Promise { await deps.loadConfig(); if (deps.isSessionActive?.() === false) return; deps.applyState(buildTelegramSessionStartState(deps.currentModel)); await deps.prepareTempDir(); if (deps.isSessionActive?.() === false) return; try { deps.bindDeferredDispatchContext?.(deps.ctx); } catch (error) { if (!isTelegramStaleContextError(error)) throw error; } deps.updateStatus(); } export async function shutdownTelegramSessionRuntime( deps: TelegramSessionShutdownRuntimeDeps, ): Promise { if (deps.isSessionActive?.() === false) return; deps.unbindDeferredDispatchContext?.(); await deps.stopPolling(); if (deps.isSessionActive?.() === false) return; deps.applyState(buildTelegramSessionShutdownState()); deps.clearPendingMediaGroups(); deps.clearModelMenuState(); const activeTurnChatId = deps.getActiveTurnChatId(); if (activeTurnChatId !== undefined) { const target = deps.getActiveTurnTarget?.(); const previewTimeoutMs = deps.previewShutdownTimeoutMs ?? 1000; let timeout: ReturnType | undefined; await Promise.race([ deps.clearPreview(activeTurnChatId, target ? { target } : undefined), new Promise((resolve) => { timeout = setTimeout(resolve, previewTimeoutMs); timeout.unref?.(); }), ]).finally(() => { if (timeout) clearTimeout(timeout); }); if (deps.isSessionActive?.() === false) return; } deps.clearActiveTurn(); deps.clearAbort(); } export type TelegramSessionLifecycleRuntimeDeps< TContext, TQueueItem, TModel = unknown, > = Omit< TelegramSessionLifecycleHookRuntimeDeps, "applySessionStartState" | "applySessionShutdownState" > & TelegramSessionStateApplierDeps; export function createTelegramSessionLifecycleRuntime< TContext, TQueueItem, TModel = unknown, >(deps: TelegramSessionLifecycleRuntimeDeps) { const stateApplier = createTelegramSessionStateApplier({ setQueuedItems: deps.setQueuedItems, setCurrentModel: deps.setCurrentModel, setPendingModelSwitch: deps.setPendingModelSwitch, syncCounters: deps.syncCounters, syncFlags: deps.syncFlags, }); return createTelegramSessionLifecycleHooks({ getCurrentModel: deps.getCurrentModel, loadConfig: deps.loadConfig, applySessionStartState: stateApplier.applyStartState, bindDeferredDispatchContext: deps.bindDeferredDispatchContext, prepareTempDir: deps.prepareTempDir, updateStatus: deps.updateStatus, isSessionActive: deps.isSessionActive, unbindDeferredDispatchContext: deps.unbindDeferredDispatchContext, applySessionShutdownState: stateApplier.applyShutdownState, clearPendingMediaGroups: deps.clearPendingMediaGroups, clearModelMenuState: deps.clearModelMenuState, getActiveTurnChatId: deps.getActiveTurnChatId, getActiveTurnTarget: deps.getActiveTurnTarget, clearPreview: deps.clearPreview, clearActiveTurn: deps.clearActiveTurn, clearAbort: deps.clearAbort, stopPolling: deps.stopPolling, recordRuntimeEvent: deps.recordRuntimeEvent, }); } export function createTelegramSessionLifecycleHooks< TContext, TQueueItem, TModel = unknown, >(deps: TelegramSessionLifecycleHookRuntimeDeps) { return { onSessionStart: async ( _event: TelegramSessionLifecycleHookEvent, ctx: TContext, ): Promise => { try { await startTelegramSessionRuntime({ ctx, currentModel: deps.getCurrentModel(ctx), loadConfig: deps.loadConfig, isSessionActive: () => deps.isSessionActive?.(ctx) ?? true, applyState: deps.applySessionStartState, bindDeferredDispatchContext: deps.bindDeferredDispatchContext, prepareTempDir: deps.prepareTempDir, updateStatus: () => deps.updateStatus(ctx), }); } catch (error) { deps.recordRuntimeEvent?.("session", error, { phase: "start" }); throw error; } }, onSessionShutdown: async ( _event?: TelegramSessionLifecycleHookEvent, ctx?: TContext, ): Promise => { try { await shutdownTelegramSessionRuntime({ isSessionActive: () => ctx === undefined ? true : (deps.isSessionActive?.(ctx) ?? true), unbindDeferredDispatchContext: deps.unbindDeferredDispatchContext, applyState: deps.applySessionShutdownState, clearPendingMediaGroups: deps.clearPendingMediaGroups, clearModelMenuState: deps.clearModelMenuState, getActiveTurnChatId: deps.getActiveTurnChatId, getActiveTurnTarget: deps.getActiveTurnTarget, clearPreview: deps.clearPreview, previewShutdownTimeoutMs: deps.previewShutdownTimeoutMs, clearActiveTurn: deps.clearActiveTurn, clearAbort: deps.clearAbort, stopPolling: deps.stopPolling, }); } catch (error) { deps.recordRuntimeEvent?.("session", error, { phase: "shutdown" }); throw error; } }, }; } export function createTelegramQueueMutationController( deps: TelegramQueueMutationControllerDeps, ): TelegramQueueMutationController { const buildRuntimeDeps = ( ctx: TContext, ): TelegramQueueMutationRuntimeDeps => ({ ...deps, ctx, }); return { append: (item, ctx) => appendTelegramQueueItemRuntime(item, buildRuntimeDeps(ctx)), reorder: (ctx) => reorderTelegramQueueItemsRuntime(buildRuntimeDeps(ctx)), clear: (ctx) => clearTelegramQueueItemsRuntime(buildRuntimeDeps(ctx)), removeByMessageIds: (messageIds, ctx, scope) => removeTelegramQueueItemsByMessageIdsRuntime( messageIds, buildRuntimeDeps(ctx), scope, ), clearPriorityByMessageId: (messageId, ctx, scope) => clearTelegramQueuePromptPriorityRuntime( messageId, buildRuntimeDeps(ctx), scope, ), prioritizeByMessageId: (messageId, ctx, priorityEmoji, scope) => prioritizeTelegramQueuePromptRuntime( messageId, buildRuntimeDeps(ctx), priorityEmoji, scope, ), }; } function appendTelegramQueueItemRuntime( item: TelegramQueueItem, deps: TelegramQueueMutationRuntimeDeps, ): void { deps.setQueuedItems(appendTelegramQueueItem(deps.getQueuedItems(), item)); reorderTelegramQueueItemsRuntime(deps); } export function reorderTelegramQueueItemsRuntime( deps: TelegramQueueMutationRuntimeDeps, ): void { deps.setQueuedItems( [...deps.getQueuedItems()].sort(compareTelegramQueueItems), ); try { deps.updateStatus(deps.ctx); } catch (error) { if (!isTelegramStaleContextError(error)) throw error; } } export function clearTelegramQueueItemsRuntime( deps: TelegramQueueMutationRuntimeDeps, ): number { const removedCount = deps.getQueuedItems().length; if (removedCount === 0) return 0; deps.setQueuedItems([]); try { deps.updateStatus(deps.ctx); } catch (error) { if (!isTelegramStaleContextError(error)) throw error; } return removedCount; } export function removeTelegramQueueItemsByMessageIdsRuntime( messageIds: number[], deps: TelegramQueueMutationRuntimeDeps, scope?: TelegramQueueMessageScope, ): number { const { items, removedCount } = removeTelegramQueueItemsByMessageIds( deps.getQueuedItems(), messageIds, scope, ); if (removedCount === 0) return 0; deps.setQueuedItems(items); try { deps.updateStatus(deps.ctx); } catch (error) { if (!isTelegramStaleContextError(error)) throw error; } return removedCount; } export function clearTelegramQueuePromptPriorityRuntime( messageId: number, deps: TelegramQueueMutationRuntimeDeps, scope?: TelegramQueueMessageScope, ): boolean { const { changed, items } = clearTelegramQueuePromptPriority( deps.getQueuedItems(), messageId, scope, ); if (!changed) return false; deps.setQueuedItems(items); reorderTelegramQueueItemsRuntime(deps); return true; } export function prioritizeTelegramQueuePromptRuntime( messageId: number, deps: TelegramQueueMutationRuntimeDeps, priorityEmoji?: string, scope?: TelegramQueueMessageScope, ): boolean { const nextPriorityReactionOrder = deps.getNextPriorityReactionOrder?.(); if (nextPriorityReactionOrder === undefined) return false; const { changed, items } = prioritizeTelegramQueuePrompt( deps.getQueuedItems(), messageId, nextPriorityReactionOrder, priorityEmoji, scope, ); if (!changed) return false; deps.setQueuedItems(items); deps.incrementNextPriorityReactionOrder?.(); reorderTelegramQueueItemsRuntime(deps); return true; } export async function enqueueTelegramPromptTurnRuntime< TMessage, TContext = unknown, >( messages: TMessage[], deps: TelegramPromptEnqueueRuntimeDeps, ): Promise { const enqueuePlan = planTelegramPromptEnqueue( deps.getQueuedItems(), deps.getFoldQueuedPromptsIntoHistory(), ); deps.setFoldQueuedPromptsIntoHistory(false); const turn = await deps.createTurn(messages, enqueuePlan.historyTurns); deps.setQueuedItems( appendTelegramQueueItem(enqueuePlan.remainingItems, turn), ); deps.updateStatus(); deps.dispatchNextQueuedTelegramTurn(); } export function createTelegramPromptEnqueueController< TMessage, TContext = unknown, >( deps: TelegramPromptEnqueueControllerDeps, ): TelegramPromptEnqueueController { return { enqueue: (messages, ctx) => enqueueTelegramPromptTurnRuntime(messages, { ...deps, createTurn: (nextMessages, historyTurns) => deps.createTurn(nextMessages, historyTurns, ctx), updateStatus: () => deps.updateStatus(ctx), dispatchNextQueuedTelegramTurn: () => deps.dispatchNextQueuedTelegramTurn(ctx), }), }; } export function createTelegramControlQueueController( deps: TelegramControlQueueControllerDeps, ): TelegramControlQueueController { return { enqueue: (item, ctx) => { deps.appendControlItem(item, ctx); deps.dispatchNextQueuedTelegramTurn(ctx); }, }; } // --- Control Runtime --- function getTelegramQueueErrorMessage(error: unknown): string { return error instanceof Error ? error.message : String(error); } export interface TelegramRuntimeEventRecorderPort { recordRuntimeEvent?: ( category: string, error: unknown, details?: Record, ) => void; } export interface TelegramControlRuntimeDeps< TContext, > extends TelegramRuntimeEventRecorderPort { ctx: TContext; sendTextReply: ( chatId: number, replyToMessageId: number, text: string, options?: { target?: TelegramQueueTarget }, ) => Promise; onSettled: () => void; } export async function executeTelegramControlItemRuntime( item: PendingTelegramControlItem, deps: TelegramControlRuntimeDeps, ): Promise { try { await item.execute(deps.ctx); } catch (error) { const message = getTelegramQueueErrorMessage(error); deps.recordRuntimeEvent?.("control", error, { controlType: item.controlType, chatId: item.chatId, replyToMessageId: item.replyToMessageId, }); await deps.sendTextReply( item.chatId, item.replyToMessageId, `Telegram control action failed: ${message}`, { target: item.target }, ); } finally { deps.onSettled(); } } // --- Deferred Dispatch Runtime --- export interface TelegramDeferredQueueDispatchRuntimeDeps extends TelegramRuntimeEventRecorderPort { delayMs?: number; setTimer?: ( callback: () => void, ms: number, ) => ReturnType; clearTimer?: (timer: ReturnType) => void; } export interface TelegramDeferredQueueDispatchRuntime { bind: (ctx: TContext) => void; unbind: () => void; isBound: () => boolean; request: (dispatchNextQueuedTelegramTurn: (ctx: TContext) => void) => void; } export function createTelegramDeferredQueueDispatchRuntime( deps: TelegramDeferredQueueDispatchRuntimeDeps = {}, ): TelegramDeferredQueueDispatchRuntime { let boundContext: TContext | undefined; let generation = 0; const timers = new Set>(); const delayMs = deps.delayMs ?? 0; const setTimer = deps.setTimer ?? ((callback: () => void, ms: number): ReturnType => setTimeout(callback, ms)); const clearTimer = deps.clearTimer ?? ((timer: ReturnType): void => clearTimeout(timer)); const clearTimers = (): void => { for (const timer of timers) clearTimer(timer); timers.clear(); }; return { bind: (ctx) => { boundContext = ctx; generation += 1; }, unbind: () => { boundContext = undefined; generation += 1; clearTimers(); }, isBound: () => boundContext !== undefined, request: (dispatchNextQueuedTelegramTurn) => { if (boundContext === undefined) return; const scheduledGeneration = generation; let timer: ReturnType; timer = setTimer(() => { timers.delete(timer); if (generation !== scheduledGeneration || boundContext === undefined) return; dispatchNextQueuedTelegramTurn(boundContext); }, delayMs); timer.unref?.(); timers.add(timer); }, }; } // --- Dispatch Watchdog Runtime --- export interface TelegramQueueDispatchWatchdogRuntime { start: (ctx: TContext) => void; stop: () => void; poke: () => void; } export interface TelegramQueueDispatchWatchdogRuntimeDeps< TContext = unknown, > extends TelegramRuntimeEventRecorderPort { hasQueuedItems: () => boolean; dispatchNextQueuedTelegramTurn: (ctx: TContext) => void; intervalMs?: number; setInterval?: ( callback: () => void, ms: number, ) => ReturnType; clearInterval?: (timer: ReturnType) => void; } export function createTelegramQueueDispatchWatchdogRuntime( deps: TelegramQueueDispatchWatchdogRuntimeDeps, ): TelegramQueueDispatchWatchdogRuntime { const intervalMs = deps.intervalMs ?? 1000; const setIntervalFn: NonNullable< TelegramQueueDispatchWatchdogRuntimeDeps["setInterval"] > = deps.setInterval ?? ((callback, ms) => setInterval(callback, ms)); const clearIntervalFn: NonNullable< TelegramQueueDispatchWatchdogRuntimeDeps["clearInterval"] > = deps.clearInterval ?? ((timer) => clearInterval(timer)); let ctx: TContext | undefined; let interval: ReturnType | undefined; let dispatchInFlight = false; const tick = (): void => { if (ctx === undefined || dispatchInFlight || !deps.hasQueuedItems()) return; dispatchInFlight = true; try { deps.dispatchNextQueuedTelegramTurn(ctx); } catch (error) { deps.recordRuntimeEvent?.("dispatch", error, { phase: "queue-watchdog", }); } finally { dispatchInFlight = false; } }; const stop = (): void => { ctx = undefined; if (!interval) return; clearIntervalFn(interval); interval = undefined; }; return { start: (nextCtx) => { ctx = nextCtx; if (!interval) { const nextInterval = setIntervalFn(tick, intervalMs); interval = nextInterval; nextInterval.unref?.(); } tick(); }, stop, poke: tick, }; } // --- Dispatch Runtime --- export interface TelegramPromptDeliveryOptions { deliverAs: "followUp"; } export const TELEGRAM_PROMPT_FOLLOW_UP_DELIVERY = { deliverAs: "followUp", } as const satisfies TelegramPromptDeliveryOptions; export interface TelegramDispatchRuntimeDeps { executeControlItem: ( item: Extract< TelegramQueueDispatchAction, { kind: "control" } >["item"], ) => void; onPromptDispatchStart: (chatId: number) => void; sendUserMessage: ( content: Extract< TelegramQueueDispatchAction, { kind: "prompt" } >["item"]["content"], options?: TelegramPromptDeliveryOptions, ) => void; onPromptDispatchFailure: (message: string) => void; onIdle: () => void; } export interface TelegramQueueDispatchControllerDeps< TContext = unknown, > extends TelegramRuntimeEventRecorderPort { getQueuedItems: () => TelegramQueueItem[]; setQueuedItems: (items: TelegramQueueItem[]) => void; canDispatch: (ctx: TContext) => boolean; hasDispatchContext?: () => boolean; updateStatus: (ctx: TContext, error?: string) => void; sendTextReply: TelegramControlRuntimeDeps["sendTextReply"]; onPromptDispatchStart: (ctx: TContext, chatId: number) => void; sendUserMessage: TelegramDispatchRuntimeDeps["sendUserMessage"]; onPromptDispatchFailure: (ctx: TContext, message: string) => void; isQueueItemTransportActive?: (item: TelegramQueueItem) => boolean; } export interface TelegramQueueDispatchController { dispatchNext: (ctx: TContext) => void; } export function executeTelegramQueueDispatchPlan( plan: TelegramQueueDispatchAction, deps: TelegramDispatchRuntimeDeps, ): void { if (plan.kind === "none") { deps.onIdle(); return; } if (plan.kind === "control") { deps.executeControlItem(plan.item); return; } deps.onPromptDispatchStart(plan.item.chatId); try { deps.sendUserMessage(plan.item.content); } catch (error) { const message = getTelegramQueueErrorMessage(error); deps.onPromptDispatchFailure(message); } } export type TelegramQueueDispatchRuntimeDeps = Omit< TelegramQueueDispatchControllerDeps, "canDispatch" > & TelegramDispatchReadinessDeps; export function createTelegramQueueDispatchRuntime( deps: TelegramQueueDispatchRuntimeDeps, ): TelegramQueueDispatchController { return createTelegramQueueDispatchController({ getQueuedItems: deps.getQueuedItems, setQueuedItems: deps.setQueuedItems, canDispatch: createTelegramDispatchReadinessChecker({ isCompactionInProgress: deps.isCompactionInProgress, hasActiveTurn: deps.hasActiveTurn, hasDispatchPending: deps.hasDispatchPending, isIdle: deps.isIdle, hasPendingMessages: deps.hasPendingMessages, }), hasDispatchContext: deps.hasDispatchContext, updateStatus: deps.updateStatus, sendTextReply: deps.sendTextReply, onPromptDispatchStart: deps.onPromptDispatchStart, sendUserMessage: deps.sendUserMessage, onPromptDispatchFailure: deps.onPromptDispatchFailure, isQueueItemTransportActive: deps.isQueueItemTransportActive, recordRuntimeEvent: deps.recordRuntimeEvent, }); } export function createTelegramQueueDispatchController( deps: TelegramQueueDispatchControllerDeps, ): TelegramQueueDispatchController { let controlDispatchPending = false; const controller: TelegramQueueDispatchController = { dispatchNext: (ctx) => { if (deps.hasDispatchContext && !deps.hasDispatchContext()) return; if (controlDispatchPending) { deps.updateStatus(ctx); return; } const queuedItems = deps.getQueuedItems(); const activeItems = deps.isQueueItemTransportActive ? queuedItems.filter(deps.isQueueItemTransportActive) : queuedItems; if (activeItems.length !== queuedItems.length) { deps.setQueuedItems(activeItems); deps.recordRuntimeEvent?.( "dispatch", new Error( "Dropped queue work from an inactive Telegram transport generation.", ), { phase: "transport-generation" }, ); } const dispatchPlan = planNextTelegramQueueAction( activeItems, deps.canDispatch(ctx), ); if (dispatchPlan.kind !== "none") { deps.setQueuedItems(dispatchPlan.remainingItems); } executeTelegramQueueDispatchPlan(dispatchPlan, { executeControlItem: (item) => { controlDispatchPending = true; deps.updateStatus(ctx); void executeTelegramControlItemRuntime(item, { ctx, sendTextReply: deps.sendTextReply, recordRuntimeEvent: deps.recordRuntimeEvent, onSettled: () => { controlDispatchPending = false; if (deps.hasDispatchContext && !deps.hasDispatchContext()) return; deps.updateStatus(ctx); controller.dispatchNext(ctx); }, }); }, onPromptDispatchStart: (chatId) => { deps.onPromptDispatchStart(ctx, chatId); }, sendUserMessage: deps.sendUserMessage, onPromptDispatchFailure: (message) => { deps.onPromptDispatchFailure(ctx, message); }, onIdle: () => { deps.updateStatus(ctx); }, }); }, }; return controller; }