import { createUserMessage } from '@deepseek-ai/dsh-llm/message'; import { SessionId } from '@deepseek-ai/dsh-session/types'; import { isAuthorized } from './auth.js'; import { catalogFromLiveAgents, loadCatalog, truncateButton, visibleSessionsForWorkspace, workspacesWithVisibleSessions, } from './catalog.js'; import { TelegramClient, } from './client.js'; import { MSG, LAST_CB, parseCommand } from './commands.js'; import { markdownToHtml, splitMessage } from './format.js'; import { formatLastTurn, loadLastTurn } from './history.js'; import { describeAgent, displayLabel } from './label.js'; import { formatModel, loadSessionModels, selectSessionModel, } from './models.js'; const WS_CB = 'ws:'; const SID_CB = 'sid:'; const BACK_WS_CB = 'wb'; const MODEL_CB = 'mdl:'; const EFFORT_CB = 'me:'; const BACK_MODEL_CB = 'mb'; const MAX_BUTTONS = 40; function lastContextKeyboard() { return { inline_keyboard: [[{ text: '查看上次对话', callback_data: LAST_CB }]], }; } const defaultSleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms)); function contentToText(content) { return content .filter((block) => block.type === 'text') .map((block) => block.text) .join(''); } export class TelegramBridge { ctx; token; allowedUserIds; allowAllUsers; client; sleep; maxMessageLength; bindings = new Map(); pickers = new Map(); polling = false; offset; pollPromise; pollAbort; disposeSessionListener; constructor(ctx, options) { this.ctx = ctx; this.token = options.token; this.allowedUserIds = options.allowedUserIds; this.allowAllUsers = options.allowAllUsers; this.client = options.client ?? new TelegramClient(options.token, { pollingTimeoutSec: options.pollingTimeoutSec ?? 30, }); this.sleep = options.sleep ?? defaultSleep; this.maxMessageLength = options.maxMessageLength ?? 4096; } start() { this.disposeSessionListener?.(); this.disposeSessionListener = this.ctx.on('session/event', (session, event) => { void this.onSessionEvent(session, event).catch((err) => { this.ctx.logger.error(this.redact(err)); }); }); void this.client.setMyCommands([ { command: 'start', description: '欢迎与用法' }, { command: 'sessions', description: '按工作区列出并附着会话' }, { command: 'last', description: '查看上次问答(续接上下文)' }, { command: 'model', description: '切换当前绑定会话的模型' }, { command: 'status', description: '查看当前绑定' }, { command: 'unbind', description: '断开手机绑定(不关闭本机会话)' }, { command: 'help', description: '显示帮助' }, ]).then(() => { this.ctx.logger.info('dsh-telegram-channel: bot commands registered'); }).catch((err) => { this.ctx.logger.warn(`dsh-telegram-channel: setMyCommands failed: ${this.redact(err)}`); }); if (!this.polling) { this.polling = true; this.pollAbort = new AbortController(); this.ctx.logger.info('dsh-telegram-channel: long-polling started'); this.pollPromise = this.pollLoop(); } } async stop() { this.polling = false; this.pollAbort?.abort(); this.pollAbort = undefined; this.disposeSessionListener?.(); this.disposeSessionListener = undefined; // Never dispose host agents — only clear remote bindings. this.bindings.clear(); this.pickers.clear(); if (this.pollPromise) { await this.pollPromise.catch(() => { }); this.pollPromise = undefined; } } async processUpdate(update) { if (update.callback_query) { await this.handleCallback(update); return; } const message = update.message; if (!message?.text) return; const chatId = message.chat.id; const userId = message.from?.id; if (!isAuthorized({ allowAllUsers: this.allowAllUsers, allowedUserIds: this.allowedUserIds, userId })) { await this.client.sendMessage(chatId, MSG.DENIED); return; } const parsed = parseCommand(message.text); switch (parsed.type) { case 'start': await this.client.sendMessage(chatId, MSG.WELCOME); return; case 'help': await this.client.sendMessage(chatId, MSG.HELP); return; case 'sessions': await this.sendWorkspacePicker(chatId); return; case 'last': await this.sendLastTurn(chatId); return; case 'model': await this.sendModelPicker(chatId); return; case 'status': await this.sendStatus(chatId); return; case 'unbind': this.bindings.delete(String(chatId)); this.pickers.delete(String(chatId)); await this.client.sendMessage(chatId, MSG.UNBOUND); return; case 'unknown': await this.client.sendMessage(chatId, MSG.unknown(parsed.command)); return; case 'plain': await this.followupBound(chatId, parsed.text); return; } } async handleCallback(update) { const cq = update.callback_query; const userId = cq.from.id; const chatId = cq.message?.chat.id; if (chatId === undefined) { await this.client.answerCallbackQuery(cq.id); return; } if (!isAuthorized({ allowAllUsers: this.allowAllUsers, allowedUserIds: this.allowedUserIds, userId })) { await this.client.answerCallbackQuery(cq.id, MSG.DENIED); await this.client.sendMessage(chatId, MSG.DENIED); return; } const data = cq.data ?? ''; const picker = this.pickers.get(String(chatId)); if (data === LAST_CB) { await this.client.answerCallbackQuery(cq.id); await this.sendLastTurn(chatId); return; } if (data === BACK_WS_CB) { await this.client.answerCallbackQuery(cq.id); await this.sendWorkspacePicker(chatId, picker?.catalog); return; } if (data === BACK_MODEL_CB) { await this.client.answerCallbackQuery(cq.id); await this.sendModelPicker(chatId); return; } if (data.startsWith(WS_CB)) { const index = Number(data.slice(WS_CB.length)); await this.client.answerCallbackQuery(cq.id); await this.sendSessionPicker(chatId, index); return; } if (data.startsWith(SID_CB)) { const index = Number(data.slice(SID_CB.length)); const row = picker?.sessions[index]; if (!row) { await this.client.answerCallbackQuery(cq.id, '会话已过期'); await this.client.sendMessage(chatId, MSG.PICKER_STALE); return; } await this.bindSession(chatId, cq.id, row); return; } if (data.startsWith(MODEL_CB)) { const index = Number(data.slice(MODEL_CB.length)); const option = picker?.models?.[index]; if (!option) { await this.client.answerCallbackQuery(cq.id, '列表已过期'); await this.client.sendMessage(chatId, MSG.PICKER_STALE); return; } if (option.efforts && option.efforts.length > 0) { if (picker) picker.pendingModel = option; await this.client.answerCallbackQuery(cq.id); await this.sendEffortPicker(chatId, option); return; } await this.applyModel(chatId, cq.id, option); return; } if (data.startsWith(EFFORT_CB)) { const index = Number(data.slice(EFFORT_CB.length)); const pending = picker?.pendingModel; const effort = pending?.efforts?.[index]; if (!pending || !effort) { await this.client.answerCallbackQuery(cq.id, '列表已过期'); await this.client.sendMessage(chatId, MSG.PICKER_STALE); return; } await this.applyModel(chatId, cq.id, pending, effort.id); return; } // Legacy bind: callbacks (older messages / tests) if (data.startsWith('bind:')) { const sessionId = data.slice('bind:'.length); await this.bindSession(chatId, cq.id, { sessionId, title: sessionId, blank: false, running: true, updatedAt: 0, }); return; } await this.client.answerCallbackQuery(cq.id); } async resolveCatalog() { try { const fromApi = await loadCatalog(this.ctx); if (fromApi) return fromApi; } catch (err) { this.ctx.logger.warn(`dsh-telegram-channel: catalog via apiProxy failed: ${this.redact(err)}`); } return catalogFromLiveAgents(this.liveAgents(), this.ctx); } async sendWorkspacePicker(chatId, existing) { const catalog = existing ?? await this.resolveCatalog(); const workspaces = workspacesWithVisibleSessions(catalog); if (workspaces.length === 0) { this.pickers.delete(String(chatId)); await this.client.sendMessage(chatId, MSG.NO_SESSIONS); return; } const shown = workspaces.slice(0, MAX_BUTTONS); this.pickers.set(String(chatId), { workspaces: shown, sessions: [], catalog }); const keyboard = { inline_keyboard: shown.map((ws, i) => ([{ text: truncateButton(`${i + 1}. ${ws.title}`), callback_data: `${WS_CB}${i}`, }])), }; const body = [ catalog.complete ? `选择工作区(共 ${workspaces.length} 个,与 Web 对齐,已排除归档):` : `选择工作区(共 ${workspaces.length} 个)⚠️ 仅运行中的会话(apiProxy 未就绪,完整列表需插件 ≥0.3.2 并重启 dsh web):`, '', ...shown.map((ws, i) => { const n = visibleSessionsForWorkspace(catalog, ws).length; return `${i + 1}. ${ws.title}\n ${ws.path}\n 会话:${n}`; }), workspaces.length > MAX_BUTTONS ? `\n仅显示前 ${MAX_BUTTONS} 个工作区。` : '', '', '点下方按钮进入该工作区的会话列表。', ].filter(Boolean).join('\n'); await this.client.sendMessage(chatId, body, undefined, keyboard); } async sendSessionPicker(chatId, workspaceIndex) { const picker = this.pickers.get(String(chatId)); const catalog = picker?.catalog ?? await this.resolveCatalog(); const workspaces = picker?.workspaces?.length ? picker.workspaces : workspacesWithVisibleSessions(catalog); const workspace = workspaces[workspaceIndex]; if (!workspace) { await this.client.sendMessage(chatId, MSG.PICKER_STALE); return; } const sessions = visibleSessionsForWorkspace(catalog, workspace) .slice() .sort((a, b) => b.updatedAt - a.updatedAt); if (sessions.length === 0) { await this.client.sendMessage(chatId, MSG.NO_SESSIONS_IN_WS(workspace.title)); return; } const shown = sessions.slice(0, MAX_BUTTONS); this.pickers.set(String(chatId), { workspaces, sessions: shown, catalog }); const keyboard = { inline_keyboard: [ ...shown.map((row, i) => ([{ text: truncateButton(`${i + 1}. ${row.title}${row.running ? '' : ' · 冷'}`), callback_data: `${SID_CB}${i}`, }])), [{ text: '← 返回工作区', callback_data: BACK_WS_CB }], ], }; const body = [ `工作区:${workspace.title}`, workspace.path, '', `选择会话(共 ${sessions.length} 个):`, '', ...shown.map((row, i) => { const mark = row.running ? '运行中' : '未附着'; return `${i + 1}. ${row.title}\n ${mark} · …${row.sessionId.slice(-12)}`; }), sessions.length > MAX_BUTTONS ? `\n仅显示前 ${MAX_BUTTONS} 个会话。` : '', '', '点下方按钮附着;冷会话会自动 resume(不关闭 Web)。', ].filter(Boolean).join('\n'); await this.client.sendMessage(chatId, body, undefined, keyboard); } async bindSession(chatId, callbackId, row) { const agent = await this.ensureLiveAgent(row.sessionId); if (!agent) { await this.client.answerCallbackQuery(callbackId, '无法附着'); await this.client.sendMessage(chatId, MSG.RESUME_FAILED); return; } const parts = describeAgent(agent, 0, this.ctx); const label = row.title && row.title !== row.sessionId ? displayLabel({ ...parts, title: row.title }) : displayLabel(parts); this.bindings.set(String(chatId), { chatId, sessionId: String(agent.id), label }); await this.client.answerCallbackQuery(callbackId, '已附着'); await this.client.sendMessage(chatId, MSG.BOUND(label), undefined, lastContextKeyboard()); } async sendLastTurn(chatId) { const binding = this.bindings.get(String(chatId)); if (!binding) { await this.client.sendMessage(chatId, MSG.NEED_BIND); return; } const agent = await this.ensureLiveAgent(binding.sessionId); try { const turn = await loadLastTurn(this.ctx, binding.sessionId, agent); const text = formatLastTurn(turn); await this.deliver(chatId, text); } catch (err) { this.ctx.logger.warn(`dsh-telegram-channel: /last failed: ${this.redact(err)}`); await this.client.sendMessage(chatId, MSG.LAST_FAILED); } } async sendModelPicker(chatId) { const binding = this.bindings.get(String(chatId)); if (!binding) { await this.client.sendMessage(chatId, MSG.NEED_BIND); return; } const agent = await this.ensureLiveAgent(binding.sessionId); if (!agent) { this.bindings.delete(String(chatId)); await this.client.sendMessage(chatId, MSG.GONE); return; } try { const snap = await loadSessionModels(this.ctx, binding.sessionId); if (!snap.routable) { await this.client.sendMessage(chatId, MSG.MODEL_UNROUTABLE(formatModel(snap.current))); return; } if (snap.options.length === 0) { await this.client.sendMessage(chatId, MSG.MODEL_EMPTY(formatModel(snap.current))); return; } const shown = snap.options.slice(0, MAX_BUTTONS); const prev = this.pickers.get(String(chatId)); this.pickers.set(String(chatId), { workspaces: prev?.workspaces ?? [], sessions: prev?.sessions ?? [], catalog: prev?.catalog, models: shown, }); const keyboard = { inline_keyboard: shown.map((opt, i) => ([{ text: truncateButton(`${opt.label}${opt.provider === snap.current.provider && opt.model === snap.current.model ? ' ✓' : ''}`), callback_data: `${MODEL_CB}${i}`, }])), }; const body = [ `当前模型:${formatModel(snap.current)}`, `会话:${binding.label}`, '', '选择新模型(下一回合生效):', ].join('\n'); await this.client.sendMessage(chatId, body, undefined, keyboard); } catch (err) { this.ctx.logger.warn(`dsh-telegram-channel: /model failed: ${this.redact(err)}`); const detail = err instanceof Error ? err.message : String(err); await this.client.sendMessage(chatId, MSG.MODEL_UNAVAILABLE(detail)); } } async sendEffortPicker(chatId, option) { const efforts = option.efforts ?? []; const keyboard = { inline_keyboard: [ ...efforts.map((e, i) => ([{ text: truncateButton(e.name || e.id), callback_data: `${EFFORT_CB}${i}`, }])), [{ text: '← 返回模型列表', callback_data: BACK_MODEL_CB }], ], }; await this.client.sendMessage(chatId, `已选 ${option.label}\n请选择 reasoning effort:`, undefined, keyboard); } async applyModel(chatId, callbackId, option, reasoningEffort) { const binding = this.bindings.get(String(chatId)); if (!binding) { await this.client.answerCallbackQuery(callbackId, '未绑定'); await this.client.sendMessage(chatId, MSG.NEED_BIND); return; } try { const selected = await selectSessionModel(this.ctx, binding.sessionId, { provider: option.provider, model: option.model, reasoningEffort, }); await this.client.answerCallbackQuery(callbackId, '已切换'); await this.client.sendMessage(chatId, MSG.MODEL_SET(formatModel(selected))); } catch (err) { this.ctx.logger.warn(`dsh-telegram-channel: selectModel failed: ${this.redact(err)}`); await this.client.answerCallbackQuery(callbackId, '切换失败'); await this.client.sendMessage(chatId, MSG.MODEL_FAILED); } } liveAgents() { const agents = this.ctx.agents; if (typeof agents.roots === 'function') { const roots = agents.roots(); if (roots.length > 0) return roots; } if (typeof agents.list === 'function') return agents.list(); return []; } findLiveAgent(sessionId) { const agents = this.ctx.agents; if (typeof agents.get === 'function') { try { const found = agents.get(SessionId(sessionId)); if (found) return found; } catch { // fall through to list scan } } return this.liveAgents().find((a) => String(a.id) === sessionId); } /** Resume cold sessions when needed; never dispose the returned handle. */ async ensureLiveAgent(sessionId) { const live = this.findLiveAgent(sessionId); if (live) return live; const agents = this.ctx.agents; if (typeof agents.resume !== 'function') return undefined; try { const handle = await agents.resume({ resumeSessionId: SessionId(sessionId) }); return handle.agent; } catch (err) { this.ctx.logger.warn(`dsh-telegram-channel: resume failed for ${sessionId}: ${this.redact(err)}`); return undefined; } } async sendStatus(chatId) { const binding = this.bindings.get(String(chatId)); if (!binding) { await this.client.sendMessage(chatId, MSG.STATUS_NONE); return; } const stillLive = this.findLiveAgent(binding.sessionId); if (!stillLive) { await this.client.sendMessage(chatId, MSG.STATUS_BOUND_COLD(binding.label)); return; } await this.client.sendMessage(chatId, MSG.STATUS_BOUND(binding.label)); } async followupBound(chatId, text) { const binding = this.bindings.get(String(chatId)); if (!binding) { await this.client.sendMessage(chatId, MSG.NEED_BIND); return; } const agent = await this.ensureLiveAgent(binding.sessionId); if (!agent) { this.bindings.delete(String(chatId)); await this.client.sendMessage(chatId, MSG.GONE); return; } const message = createUserMessage({ content: [{ type: 'text', text }], source: { kind: 'user' }, }); agent.followup(message); } async pollLoop() { const signal = this.pollAbort?.signal; let errorCount = 0; while (this.polling) { try { const updates = await this.client.getUpdates(this.offset); if (!this.polling) break; errorCount = 0; if (updates.length === 0) { await this.interruptibleDelay(50, signal); continue; } for (const update of updates) { if (!this.polling) break; await this.processUpdate(update); this.offset = update.update_id + 1; } } catch (err) { if (!this.polling) break; errorCount += 1; this.ctx.logger.error(this.redact(err)); await this.interruptibleSleep(Math.min(1000 * errorCount, 10_000), signal); } } } interruptibleDelay(ms, signal) { if (signal?.aborted || !this.polling) return Promise.resolve(); return new Promise((resolve) => { const timer = setTimeout(resolve, ms); signal?.addEventListener('abort', () => { clearTimeout(timer); resolve(); }, { once: true }); }); } async interruptibleSleep(ms, signal) { if (signal?.aborted || !this.polling) return; await Promise.race([ this.sleep(ms), new Promise((resolve) => { if (signal?.aborted) { resolve(); return; } signal?.addEventListener('abort', () => resolve(), { once: true }); }), ]); } async onSessionEvent(session, event) { const id = String(session.id); const targets = [...this.bindings.values()].filter((b) => b.sessionId === id); if (targets.length === 0) return; if (event.type === 'turn/start') { await Promise.all(targets.map((b) => this.client.sendChatAction(b.chatId, 'typing'))); return; } if (event.type === 'assistant/message') { const text = contentToText(event.data.message.content); if (!text) return; await Promise.all(targets.map((b) => this.deliver(b.chatId, text))); } } async deliver(chatId, markdown) { const chunks = splitMessage(markdown, this.maxMessageLength); for (const chunk of chunks) { const html = markdownToHtml(chunk); try { await this.client.sendMessage(chatId, html, 'HTML'); } catch { try { await this.client.sendMessage(chatId, chunk); } catch (err) { this.ctx.logger.error(this.redact(err)); } } } } redact(value) { const message = value instanceof Error ? value.message : String(value); return message.split(this.token).join('***'); } }