diff --git a/lib/apiproxy.d.ts b/lib/apiproxy.d.ts index 5b22e25..898e208 100644 --- a/lib/apiproxy.d.ts +++ b/lib/apiproxy.d.ts @@ -17,6 +17,11 @@ export type ApiProxySessions = { rpcId: string; payload: unknown; }) => Promise; + /** Create a session (used by /new; payload: workspaceId | cwd). */ + create?: (req: { + rpcId: string; + payload: unknown; + }) => Promise; }; export type ApiProxyLike = { workspace?: { diff --git a/lib/bridge.d.ts b/lib/bridge.d.ts index 0d6655a..6bd9684 100644 --- a/lib/bridge.d.ts +++ b/lib/bridge.d.ts @@ -28,6 +28,22 @@ export declare class TelegramBridge { private pollPromise; private pollAbort; private disposeSessionListener; + /** sessionIds mid-turn (busy feedback, /stop state) */ + private readonly busySessions; + /** chatId → typing heartbeat interval handle */ + private readonly heartbeats; + /** chatId → serialized notice chain (prevents 429 storms) */ + private readonly noticeQueue; + private readonly pendingAsks; + private hookTimer; + private readonly pendingApprovalsTG; + private disposeApprovalHook; + /** sessionId → thinking indicator state (one notice per reasoning phase) */ + private readonly thinkingSessions; + /** callId → tool name (tool/result failure notices) */ + private readonly callNames; + /** sessionId → latest todo snapshot (/mission) */ + private readonly lastTodos; constructor(ctx: Context, options: TelegramBridgeOptions); start(): void; stop(): Promise; @@ -53,5 +69,39 @@ export declare class TelegramBridge { private onSessionEvent; private deliver; private deliverHtml; + /** Retry an outbound call with capped linear backoff (500ms, 1s, 2s… max 4s). */ + private withRetry; + private escHtml; + /** + * Send a notice; a leading `> ` renders as a Telegram blockquote (HTML). + * Falls back to plain text when the HTML send fails — a notice is never lost. + */ + private deliverNotice; + /** Serialized per-chat notice chain — bursts can't race into Telegram 429s. */ + private enqueueNotice; + private startTypingHeartbeat; + private stopTypingHeartbeat; + private stopAllHeartbeats; + private userQuestions; + /** + * Wrap the UI provider's ask() so Telegram gets a parallel answer path. + * `Promise.race` decides; the UI path is untouched. The TG promise NEVER + * settles when there is no bound chat — race would kill the UI's window + * with that early rejection. + */ + private hookUserQuestions; + private registerTgAsk; + private settleGuiSide; + private formatAskPending; + private handleTgAnswer; + private onApprovalRequest; + private handleTgApproval; + private stopBound; + private sendMission; + /** /new: create a session in the bound session's workspace and attach to it. */ + private newSessionHere; + private bindingsPath; + private saveBindings; + private loadBindings; private redact; } diff --git a/lib/bridge.js b/lib/bridge.js index 177c25f..0f4463a 100644 --- a/lib/bridge.js +++ b/lib/bridge.js @@ -1,6 +1,10 @@ +import { mkdirSync, readFileSync, writeFileSync } from 'node:fs'; +import { homedir } from 'node:os'; +import { dirname, join } from 'node:path'; import { createUserMessage } from '@deepseek-ai/dsh-llm/message'; import { SessionId } from '@deepseek-ai/dsh-session/types'; import { isAuthorized } from './auth.js'; +import { resolveApiProxy } from './apiproxy.js'; import { catalogFromLiveAgents, loadCatalog, truncateButton, visibleSessionsForWorkspace, workspacesWithVisibleSessions, } from './catalog.js'; import { TelegramClient, } from './client.js'; import { MSG, LAST_CB, parseCommand } from './commands.js'; @@ -51,6 +55,23 @@ export class TelegramBridge { pollPromise; pollAbort; disposeSessionListener; + // ── Stability & interactivity additions ── + /** sessionIds mid-turn (busy feedback, /stop state) */ + busySessions = new Set(); + /** chatId → typing heartbeat interval handle */ + heartbeats = new Map(); + /** chatId → serialized notice chain (prevents 429 storms) */ + noticeQueue = new Map(); + pendingAsks = new Map(); + hookTimer; + pendingApprovalsTG = new Map(); + disposeApprovalHook; + /** sessionId → thinking indicator state (one notice per reasoning phase) */ + thinkingSessions = new Map(); + /** callId → tool name (tool/result failure notices) */ + callNames = new Map(); + /** sessionId → latest todo snapshot (/mission) */ + lastTodos = new Map(); constructor(ctx, options) { this.ctx = ctx; this.token = options.token; @@ -70,6 +91,12 @@ export class TelegramBridge { this.ctx.logger.error(this.redact(err)); }); }); + // Restore persisted bindings (survive hot reload / restart) and install + // the TG answering hooks (ask_user dual-path + approval dual-path). + this.loadBindings(); + this.hookUserQuestions(); + this.disposeApprovalHook?.(); + this.disposeApprovalHook = this.ctx.on('approval/request', (req, next) => this.onApprovalRequest(req, next)); void this.client.setMyCommands([ { command: 'start', description: '欢迎与用法' }, { command: 'sessions', description: '按工作区列出并附着会话' }, @@ -77,6 +104,9 @@ export class TelegramBridge { { command: 'model', description: '切换当前绑定会话的模型' }, { command: 'status', description: '查看当前绑定' }, { command: 'unbind', description: '断开手机绑定（不关闭本机会话）' }, + { command: 'stop', description: '中止当前正在运行的任务' }, + { command: 'mission', description: '查看任务清单与完成情况' }, + { command: 'new', description: '在当前工作区新开对话并附着' }, { command: 'help', description: '显示帮助' }, ]).then(() => { this.ctx.logger.info('dsh-telegram-channel: bot commands registered'); @@ -100,6 +130,20 @@ export class TelegramBridge { this.bindings.clear(); this.pickers.clear(); this.pendingModels.clear(); + this.stopAllHeartbeats(); + this.busySessions.clear(); + this.noticeQueue.clear(); + this.pendingAsks.clear(); + this.pendingApprovalsTG.clear(); + this.thinkingSessions.clear(); + this.callNames.clear(); + this.lastTodos.clear(); + if (this.hookTimer) { + clearTimeout(this.hookTimer); + this.hookTimer = undefined; + } + this.disposeApprovalHook?.(); + this.disposeApprovalHook = undefined; if (this.pollPromise) { await this.pollPromise.catch(() => { }); this.pollPromise = undefined; @@ -111,7 +155,7 @@ export class TelegramBridge { return; } const message = update.message; - if (!message?.text) + if (!message) return; const chatId = message.chat.id; const userId = message.from?.id; @@ -119,6 +163,11 @@ export class TelegramBridge { await this.client.sendMessage(chatId, MSG.DENIED); return; } + // Non-text messages: tell the (authorized) user instead of dropping silently. + if (!message.text) { + await this.enqueueNotice(chatId, '暂只支持文本消息（图片/语音/文件等媒体暂不处理）'); + return; + } const parsed = parseCommand(message.text); switch (parsed.type) { case 'start': @@ -142,14 +191,49 @@ export class TelegramBridge { case 'unbind': this.bindings.delete(String(chatId)); this.pickers.delete(String(chatId)); + this.pendingAsks.delete(String(chatId)); + this.saveBindings(); await this.client.sendMessage(chatId, MSG.UNBOUND); return; + case 'stop': + await this.stopBound(chatId); + return; + case 'mission': + await this.sendMission(chatId); + return; + case 'new': + await this.newSessionHere(chatId); + return; case 'unknown': + // /cancel closes the TG-side answer/approval channel; the UI side stays open. + if (parsed.command === 'cancel') { + let cancelled = false; + if (this.pendingApprovalsTG.delete(String(chatId))) + cancelled = true; + if (this.pendingAsks.delete(String(chatId))) + cancelled = true; + if (cancelled) { + await this.enqueueNotice(chatId, '已关闭 TG 端作答/审批（该问题仍可在 Web 端回答）'); + return; + } + } await this.client.sendMessage(chatId, MSG.unknown(parsed.command)); return; - case 'plain': + case 'plain': { + // A pending approval/ask on TG consumes the message as an answer. + const approval = this.pendingApprovalsTG.get(String(chatId)); + if (approval) { + await this.handleTgApproval(chatId, approval, parsed.text); + return; + } + const pending = this.pendingAsks.get(String(chatId)); + if (pending) { + await this.handleTgAnswer(chatId, pending, parsed.text); + return; + } await this.followupBound(chatId, parsed.text); return; + } } } async handleCallback(update) { @@ -354,6 +438,11 @@ export class TelegramBridge { ? displayLabel({ ...parts, title: row.title }) : displayLabel(parts); this.bindings.set(String(chatId), { chatId, sessionId: String(agent.id), label }); + // Drop stale pending asks when rebinding to another session; persist bindings. + const stale = this.pendingAsks.get(String(chatId)); + if (stale && stale.sessionId !== String(agent.id)) + this.pendingAsks.delete(String(chatId)); + this.saveBindings(); await this.client.answerCallbackQuery(callbackId, '已附着'); await this.client.sendMessage(chatId, MSG.BOUND(label), undefined, lastContextKeyboard()); } @@ -537,7 +626,14 @@ export class TelegramBridge { content: [{ type: 'text', text }], source: { kind: 'user' }, }); + // Busy feedback: mid-turn followups queue silently by design — tell the user. + const busy = this.busySessions.has(binding.sessionId) + || Boolean(agent.phase?.kind + && agent.phase?.kind !== 'idle'); agent.followup(message); + if (busy) { + await this.enqueueNotice(chatId, '已加入队列：当前任务完成后将处理这条消息'); + } } async pollLoop() { const signal = this.pollAbort?.signal; @@ -599,49 +695,144 @@ export class TelegramBridge { if (targets.length === 0) return; if (event.type === 'turn/start') { - await Promise.all(targets.map((b) => this.client.sendChatAction(b.chatId, 'typing'))); + // Typing heartbeat — one sendChatAction expires in ~5s, so refresh it + // for the whole turn instead of only at start. + this.busySessions.add(id); + for (const b of targets) + this.startTypingHeartbeat(b.chatId); + return; + } + if (event.type === 'turn/end') { + this.busySessions.delete(id); + this.thinkingSessions.delete(id); + for (const b of targets) + this.stopTypingHeartbeat(b.chatId); + // Abnormal endings (API failure / abort / crash-recovery) must be + // visible on the phone — previously they ended in silence. + const reason = event.data?.reason; + const kind = reason?.kind; + if (kind === 'error') { + const detail = String(reason?.error?.message ?? reason?.error?.code ?? '').replace(/\s+/g, ' ').slice(0, 300); + for (const b of targets) + this.enqueueNotice(b.chatId, `本轮异常结束（API/内部错误）${detail ? `：${detail}` : ''}`); + } + else if (kind === 'aborted') { + const manual = reason?.reason?.kind === 'user'; + for (const b of targets) + this.enqueueNotice(b.chatId, manual ? '本轮已被手动中止' : '本轮已被中断'); + } + else if (kind === 'interrupted') { + for (const b of targets) + this.enqueueNotice(b.chatId, '会话曾被异常退出（崩溃恢复），本轮被标记中断'); + } + return; + } + if (event.type === 'tool/call') { + // Progress summary; ask_user is owned by the provider hook (avoids duplicates). + const name = event.data?.name ?? 'tool'; + if (name === 'ask_user_question') + return; + this.thinkingSessions.set(id, false); + if (event.data?.callId !== undefined) { + if (this.callNames.size > 400) { + const first = this.callNames.keys().next().value; + if (first !== undefined) + this.callNames.delete(first); + } + this.callNames.set(String(event.data.callId), String(name)); + } + const args = typeof event.data?.arguments === 'string' ? event.data.arguments : ''; + const brief = args.replace(/\s+/g, ' ').trim(); + const shown = brief.length > 60 ? `${brief.slice(0, 60)}…` : brief; + for (const b of targets) + this.enqueueNotice(b.chatId, `> ${name}${shown ? ` ${shown}` : ''}`); return; } if (event.type === 'assistant/message') { const text = contentToText(event.data.message.content); + this.thinkingSessions.set(id, false); if (!text) return; await Promise.all(targets.map((b) => this.deliver(b.chatId, text))); + return; + } + if (event.type === 'assistant/chunk') { + // Thinking indicator: once per reasoning phase, status only (no content). + const chunk = event.data?.chunk; + const ctype = String(chunk?.type ?? ''); + const btype = String(chunk?.block?.type ?? ''); + const isReasoning = ctype.includes('reasoning') || btype.includes('reasoning') || btype.includes('think'); + if (isReasoning && this.thinkingSessions.get(id) !== true) { + this.thinkingSessions.set(id, true); + for (const b of targets) + this.enqueueNotice(b.chatId, '> Thinking…'); + } + return; + } + if (event.type === 'tool/result') { + // Only failures — successful results are too noisy for the phone. + const data = event.data; + if (data?.error) { + const callKey = String(data.message?.tool_use_id ?? data.message?.callId ?? ''); + const name = this.callNames.get(callKey) ?? 'tool'; + for (const b of targets) + this.enqueueNotice(b.chatId, `> 工具 ${name} 失败：${String(data.error.name ?? 'Error')}${data.error.code ? `/${String(data.error.code)}` : ''}`); + } + return; + } + if (event.type === 'todo/write') { + const todos = Array.isArray(event.data?.todos) ? event.data.todos : []; + this.lastTodos.set(id, todos); + if (todos.length === 0) + return; + const done = todos.filter((t) => t?.status === 'completed').length; + const current = todos.find((t) => t?.status === 'in_progress') ?? todos.find((t) => t?.status === 'pending'); + const cur = current ? String(current.content ?? '').slice(0, 50) : ''; + for (const b of targets) + this.enqueueNotice(b.chatId, `> 进度 ${done}/${todos.length}${cur ? `：${cur}` : ''}`); + return; } } async deliver(chatId, markdown) { - if (this.renderingMode === 'html') { - await this.deliverHtml(chatId, markdown); - return; - } - let sentAny = false; - try { - const chunks = splitRichMarkdown(markdown); - for (const chunk of chunks) { - await this.client.sendRichMessage(chatId, chunk); - sentAny = true; - } - } - catch (err) { - if (!sentAny && isRichUnsupportedError(err)) { - this.ctx.logger.warn('dsh-telegram-channel: Rich Message API unavailable, falling back to HTML rendering'); - this.renderingMode = 'html'; - await this.deliverHtml(chatId, markdown); + if (this.renderingMode !== 'html') { + let sentAny = false; + try { + const chunks = splitRichMarkdown(markdown); + for (const chunk of chunks) { + try { + // Per-chunk retry: transient network blips must not drop a reply. + await this.withRetry(() => this.client.sendRichMessage(chatId, chunk)); + sentAny = true; + } + catch (err) { + if (!sentAny && isRichUnsupportedError(err)) { + this.ctx.logger.warn('dsh-telegram-channel: Rich Message API unavailable, falling back to HTML rendering'); + this.renderingMode = 'html'; + await this.deliverHtml(chatId, markdown); + return; + } + throw err; + } + } return; } - this.ctx.logger.error(this.redact(err)); + catch (err) { + // Persistent rich failure → HTML fallback instead of dropping the message. + this.ctx.logger.warn(`dsh-telegram-channel: rich deliver failed (${this.redact(err)}), falling back to HTML`); + } } + await this.deliverHtml(chatId, markdown); } async deliverHtml(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'); + await this.withRetry(() => this.client.sendMessage(chatId, html, 'HTML')); } catch { try { - await this.client.sendMessage(chatId, chunk); + await this.withRetry(() => this.client.sendMessage(chatId, chunk)); } catch (err) { this.ctx.logger.error(this.redact(err)); @@ -649,6 +840,476 @@ export class TelegramBridge { } } } + // ── Stability & interactivity helpers ── + /** Retry an outbound call with capped linear backoff (500ms, 1s, 2s… max 4s). */ + async withRetry(fn, tries = 3) { + let lastErr; + for (let attempt = 1; attempt <= tries; attempt++) { + try { + return await fn(); + } + catch (err) { + lastErr = err; + if (attempt < tries) + await this.sleep(Math.min(500 * 2 ** (attempt - 1), 4000)); + } + } + throw lastErr; + } + escHtml(s) { + return String(s) + .replace(/&/g, '&') + .replace(//g, '>'); + } + /** + * Send a notice; a leading `> ` renders as a Telegram blockquote (HTML). + * Falls back to plain text when the HTML send fails — a notice is never lost. + */ + async deliverNotice(chatId, text) { + const quoted = text.startsWith('> '); + const body = quoted ? text.slice(2) : text; + try { + await this.withRetry(() => { + if (!quoted) + return this.client.sendMessage(chatId, body); + return this.client.sendMessage(chatId, `
${this.escHtml(body)}
`, 'HTML'); + }); + return; + } + catch { + try { + await this.client.sendMessage(chatId, quoted ? `> ${body}` : body); + } + catch (err) { + this.ctx.logger.error(this.redact(err)); + } + } + } + /** Serialized per-chat notice chain — bursts can't race into Telegram 429s. */ + enqueueNotice(chatId, text) { + const key = String(chatId); + const prev = this.noticeQueue.get(key) ?? Promise.resolve(); + const next = prev.then(() => this.deliverNotice(chatId, text)).catch(() => { }); + this.noticeQueue.set(key, next); + return next; + } + startTypingHeartbeat(chatId) { + this.stopTypingHeartbeat(chatId); + const send = () => { + void this.client.sendChatAction(chatId, 'typing').catch(() => { }); + }; + send(); + this.heartbeats.set(String(chatId), setInterval(send, 4000)); + } + stopTypingHeartbeat(chatId) { + const h = this.heartbeats.get(String(chatId)); + if (h) { + clearInterval(h); + this.heartbeats.delete(String(chatId)); + } + } + stopAllHeartbeats() { + for (const h of this.heartbeats.values()) + clearInterval(h); + this.heartbeats.clear(); + } + // ── ask_user_question: TG answering via dual-path race with the UI ── + userQuestions() { + return this.ctx.userQuestions; + } + /** + * Wrap the UI provider's ask() so Telegram gets a parallel answer path. + * `Promise.race` decides; the UI path is untouched. The TG promise NEVER + * settles when there is no bound chat — race would kill the UI's window + * with that early rejection. + */ + hookUserQuestions(attempt = 0) { + try { + const provider = this.userQuestions()?.provider; + if (provider) { + if (provider.__tgHooked && provider.__tgRealAsk) { + // Hot reload: re-point the wrapper at THIS bridge instance. + const realAsk = provider.__tgRealAsk; + const self = this; + provider.ask = function (request) { + const tg = self.registerTgAsk(request); + let gui; + try { + gui = realAsk(request); + } + catch (err) { + tg.reject(err); + throw err; + } + void gui.then(() => self.settleGuiSide(request), () => self.settleGuiSide(request)); + return Promise.race([tg.promise, gui]); + }; + return; + } + const realAsk = provider.ask.bind(provider); + provider.__tgRealAsk = realAsk; + const self = this; + provider.ask = function (request) { + const tg = self.registerTgAsk(request); + let gui; + try { + gui = realAsk(request); + } + catch (err) { + tg.reject(err); + throw err; + } + void gui.then(() => self.settleGuiSide(request), () => self.settleGuiSide(request)); + return Promise.race([tg.promise, gui]); + }; + provider.__tgHooked = true; + this.ctx.logger.info('dsh-telegram-channel: ask_user TG answering hook installed'); + return; + } + } + catch (err) { + if (attempt === 0) + this.ctx.logger.warn(`dsh-telegram-channel: userQuestions hook deferred: ${this.redact(err)}`); + } + if (attempt >= 30) { + this.ctx.logger.warn('dsh-telegram-channel: user-questions provider never appeared; TG answering disabled'); + return; + } + this.hookTimer = setTimeout(() => this.hookUserQuestions(attempt + 1), 2000); + } + registerTgAsk(request) { + let resolveFn = () => { }; + let rejectFn = () => { }; + const promise = new Promise((resolve, reject) => { + resolveFn = resolve; + rejectFn = reject; + }); + const ask = { resolve: resolveFn, reject: rejectFn, answered: false }; + const req = request; + const sessionId = req.agent?.id !== undefined ? String(req.agent.id) : undefined; + const questions = Array.isArray(req.questions) ? req.questions : []; + const onAbort = () => { + for (const [chatId, p] of this.pendingAsks) { + if (p.ask === ask) + this.pendingAsks.delete(chatId); + } + rejectFn(new Error('ask aborted')); + }; + if (req.signal?.aborted) { + rejectFn(new Error('ask aborted')); + return { promise, reject: rejectFn }; + } + req.signal?.addEventListener('abort', onAbort, { once: true }); + if (sessionId !== undefined && questions.length > 0) { + let asked = false; + for (const [chatId, binding] of this.bindings) { + if (binding.sessionId !== sessionId) + continue; + this.pendingAsks.set(String(chatId), { req: request, sessionId, questions, ask, at: Date.now() }); + this.enqueueNotice(Number(chatId), this.formatAskPending(questions)); + asked = true; + } + if (asked) + return { promise, reject: rejectFn }; + } + // No bound TG chat: never settle — the UI remains the only answer path. + return { promise, reject: rejectFn }; + } + settleGuiSide(request) { + for (const [chatId, p] of [...this.pendingAsks]) { + if (p.req !== request) + continue; + this.pendingAsks.delete(chatId); + if (!p.ask.answered) { + this.enqueueNotice(Number(chatId), '该问题已在 Web 端作答，TG 作答通道关闭'); + } + } + } + formatAskPending(questions) { + const lines = ['待回答（回复字母选项，或直接输入自定义答案；/cancel 关闭 TG 作答）']; + questions.forEach((q, i) => { + const prefix = questions.length > 1 ? `Q${i + 1}. ` : ''; + lines.push(`${prefix}${q.question ?? ''}`); + const opts = Array.isArray(q.options) ? q.options : []; + if (opts.length === 0) { + lines.push(' （开放问题，直接输入答案）'); + } + else { + opts.forEach((o, j) => lines.push(` [${String.fromCharCode(65 + j)}] ${o.label}${o.description ? ` — ${o.description}` : ''}`)); + if (q.multiSelect) + lines.push(' （可多选，如：A,B）'); + } + }); + return lines.join('\n'); + } + async handleTgAnswer(chatId, pending, text) { + const { questions, ask } = pending; + let parts; + if (questions.length === 1) { + parts = [[0, text.trim()]]; + } + else { + parts = []; + const lines = text.split(/\n|；|;/).map((s) => s.trim()).filter(Boolean); + for (const line of lines) { + const m = line.match(/^([0-9]+)\s*[:：.、)]\s*(.+)$/); + if (m) + parts.push([Number(m[1]) - 1, m[2].trim()]); + else + parts.push([null, line]); + } + if (parts.some(([idx]) => idx === null)) { + await this.enqueueNotice(chatId, `共 ${questions.length} 个问题，请每行用 1:/2: 前缀分别作答`); + return; + } + } + const answers = []; + for (const [idx, raw] of parts) { + if (idx === null || idx < 0 || idx >= questions.length) { + await this.enqueueNotice(chatId, `题号 ${(idx ?? 0) + 1} 不存在（共 ${questions.length} 题）`); + return; + } + const q = questions[idx]; + const cleaned = raw.replace(/^\[+/, '').replace(/\]+$/, '').trim(); + const letters = cleaned.split(/[,，、\s]+/).map((s) => s.trim()).filter(Boolean); + const opts = Array.isArray(q.options) ? q.options : []; + const isLetterChoice = letters.length > 0 + && letters.every((s) => /^[a-zA-Z]$/.test(s)) + && opts.length > 0; + if (isLetterChoice) { + const selected = []; + for (const s of letters) { + const i = s.toUpperCase().charCodeAt(0) - 65; + if (i < 0 || i >= opts.length) { + await this.enqueueNotice(chatId, `选项 ${s.toUpperCase()} 不存在（范围 A-${String.fromCharCode(64 + opts.length)}），请重答或 /cancel`); + return; + } + selected.push(opts[i].label); + } + answers.push({ id: q.id, selected }); + } + else { + answers.push({ id: q.id, selected: [], custom: cleaned }); + } + } + ask.answered = true; + for (const [cid, p] of this.pendingAsks) { + if (p.ask === ask) + this.pendingAsks.delete(cid); + } + ask.resolve({ answers }); + this.enqueueNotice(chatId, '已提交你的回答'); + } + // ── approval: TG answering via dual-path race on the request waterfall ── + async onApprovalRequest(req, next) { + const sessionId = req?.agent?.id !== undefined ? String(req.agent.id) : undefined; + const targets = sessionId !== undefined + ? [...this.bindings.values()].filter((b) => b.sessionId === sessionId).map((b) => String(b.chatId)) + : []; + let resolveFn = () => { }; + let rejectFn = () => { }; + const promise = new Promise((resolve, reject) => { + resolveFn = resolve; + rejectFn = reject; + }); + const ask = { resolve: resolveFn, reject: rejectFn, answered: false }; + const onAbort = () => { + for (const [chatId, p] of this.pendingApprovalsTG) { + if (p.ask === ask) + this.pendingApprovalsTG.delete(chatId); + } + rejectFn(new Error('approval withdrawn')); + }; + if (req?.signal?.aborted) { + rejectFn(new Error('approval withdrawn')); + return next(); + } + req?.signal?.addEventListener('abort', onAbort, { once: true }); + if (targets.length > 0) { + const toolName = String(req?.toolName ?? 'tool'); + const reason = req?.reason ? String(req.reason).slice(0, 200) : ''; + for (const chatId of targets) { + this.pendingApprovalsTG.set(chatId, { toolName, ask }); + this.enqueueNotice(Number(chatId), `权限审批：${toolName}${reason ? `\n(${reason})` : ''}\n回复 [A] 允许一次 / [B] 拒绝（/cancel 关闭 TG 审批）`); + } + } + // No bound TG chat: promise never settles — the UI remains the only answer path. + const gui = next(); + void gui.then(() => { + for (const [chatId, p] of [...this.pendingApprovalsTG]) { + if (p.ask !== ask) + continue; + this.pendingApprovalsTG.delete(chatId); + if (!p.ask.answered) { + p.ask.reject(new Error('superseded-by-gui')); + this.enqueueNotice(Number(chatId), '该审批已在 Web 端处理'); + } + } + }, () => { + for (const [chatId, p] of [...this.pendingApprovalsTG]) { + if (p.ask === ask) + this.pendingApprovalsTG.delete(chatId); + } + }); + return Promise.race([promise, gui]); + } + async handleTgApproval(chatId, approval, text) { + const cleaned = text.trim().replace(/^\[+/, '').replace(/\]+$/, '').toUpperCase(); + if (cleaned === 'A' || cleaned === 'B') { + approval.ask.answered = true; + for (const [cid, p] of this.pendingApprovalsTG) { + if (p.ask === approval.ask) + this.pendingApprovalsTG.delete(cid); + } + if (cleaned === 'A') { + approval.ask.resolve('allowed-once'); + await this.enqueueNotice(chatId, `已允许 ${approval.toolName} 执行一次`); + } + else { + approval.ask.resolve('rejected'); + await this.enqueueNotice(chatId, '已拒绝该工具执行'); + } + return; + } + await this.enqueueNotice(chatId, '请回复 [A] 允许一次 或 [B] 拒绝（/cancel 关闭）'); + } + // ── /stop /mission /new commands ── + async stopBound(chatId) { + const binding = this.bindings.get(String(chatId)); + if (!binding) { + await this.client.sendMessage(chatId, MSG.NEED_BIND); + return; + } + const agent = this.findLiveAgent(binding.sessionId); + if (!agent) { + await this.client.sendMessage(chatId, '当前会话没有正在运行的任务'); + return; + } + try { + agent.cancel({ kind: 'user' }); + } + catch (err) { + this.ctx.logger.warn(`dsh-telegram-channel: stop failed: ${this.redact(err)}`); + } + await this.enqueueNotice(chatId, '已请求中止当前轮…'); + } + async sendMission(chatId) { + const binding = this.bindings.get(String(chatId)); + if (!binding) { + await this.client.sendMessage(chatId, MSG.NEED_BIND); + return; + } + const todos = this.lastTodos.get(binding.sessionId) ?? []; + const lines = ['任务清单']; + if (todos.length === 0) { + lines.push('（暂无任务记录，agent 使用 todo 工具后这里会显示）'); + } + else { + const mark = { completed: '✓', in_progress: '⟳', pending: '·' }; + todos.forEach((t, i) => lines.push(`${mark[t?.status ?? ''] ?? '·'} ${i + 1}. ${String(t?.content ?? '')}`)); + const done = todos.filter((t) => t?.status === 'completed').length; + lines.push(`—— ${done}/${todos.length} 已完成`); + } + lines.push(this.busySessions.has(binding.sessionId) ? '（当前任务：运行中，/stop 可中止）' : '（当前空闲）'); + await this.client.sendMessage(chatId, lines.join('\n')); + } + /** /new: create a session in the bound session's workspace and attach to it. */ + async newSessionHere(chatId) { + const binding = this.bindings.get(String(chatId)); + if (!binding) { + await this.client.sendMessage(chatId, MSG.NEED_BIND); + return; + } + let workspaceId; + let fallbackCwd; + try { + const catalog = await this.resolveCatalog(); + for (const ws of catalog?.workspaces ?? []) { + if ((ws.sessionIds ?? []).includes(binding.sessionId)) { + workspaceId = ws.id; + break; + } + } + if (workspaceId === undefined) + fallbackCwd = catalog?.sessionsById?.get(binding.sessionId)?.cwd; + } + catch { + // Catalog unavailable → fall back to the host default workspace. + } + const proxy = resolveApiProxy(this.ctx); + if (!proxy?.sessions?.create) { + await this.client.sendMessage(chatId, '当前宿主不支持会话创建接口'); + return; + } + let res; + try { + const payload = workspaceId !== undefined ? { workspaceId } : (fallbackCwd ? { cwd: fallbackCwd } : {}); + res = await proxy.sessions.create({ + rpcId: `tg-${Date.now()}-${Math.random().toString(36).slice(2)}`, + payload, + }); + } + catch (err) { + await this.client.sendMessage(chatId, `创建失败：${this.redact(err)}`); + return; + } + const newId = res?.result?.ok ? res.result.value?.sessionId : undefined; + if (!newId) { + const detail = res?.result?.error?.message ?? JSON.stringify(res?.result ?? res); + await this.client.sendMessage(chatId, `创建失败：${String(detail).slice(0, 200)}`); + return; + } + const stale = this.pendingAsks.get(String(chatId)); + if (stale && stale.sessionId !== String(newId)) + this.pendingAsks.delete(String(chatId)); + this.bindings.set(String(chatId), { chatId, sessionId: String(newId), label: binding.label }); + this.saveBindings(); + const where = workspaceId !== undefined ? '当前工作区' : (fallbackCwd ? `目录 ${fallbackCwd}` : '默认工作区'); + await this.client.sendMessage(chatId, `已在${where}新开对话并附着：\n${String(newId)}\n直接发消息即可开始`); + } + // ── Binding persistence (survives hot reload / restart) ── + bindingsPath() { + // DSH_TELEGRAM_BINDINGS_FILE overrides the location (tests must isolate + // from the real profile data — parallel test files share this file). + const override = process.env.DSH_TELEGRAM_BINDINGS_FILE; + if (override) + return override; + const home = process.env.DSH_HOME ?? join(homedir(), '.dsh'); + return join(home, 'telegram-channel-bindings.json'); + } + saveBindings() { + try { + const obj = Object.fromEntries(this.bindings); + mkdirSync(dirname(this.bindingsPath()), { recursive: true }); + writeFileSync(this.bindingsPath(), JSON.stringify(obj, null, 2)); + } + catch (err) { + this.ctx.logger.warn(`dsh-telegram-channel: saveBindings failed: ${this.redact(err)}`); + } + } + loadBindings() { + try { + const raw = readFileSync(this.bindingsPath(), 'utf8'); + const obj = JSON.parse(raw); + let n = 0; + for (const [chatId, b] of Object.entries(obj ?? {})) { + if (b && b.sessionId) { + this.bindings.set(String(chatId), { + chatId: b.chatId ?? Number(chatId), + sessionId: String(b.sessionId), + label: b.label ?? String(b.sessionId), + }); + n += 1; + } + } + if (n > 0) + this.ctx.logger.info(`dsh-telegram-channel: restored ${n} binding(s) from disk`); + } + catch { + // First run: no file yet. + } + } redact(value) { const message = value instanceof Error ? value.message : String(value); return message.split(this.token).join('***'); diff --git a/lib/commands.d.ts b/lib/commands.d.ts index be24ea9..4631c20 100644 --- a/lib/commands.d.ts +++ b/lib/commands.d.ts @@ -44,6 +44,15 @@ export type ParsedCommand = { } | { type: 'unbind'; text: string; +} | { + type: 'stop'; + text: string; +} | { + type: 'mission'; + text: string; +} | { + type: 'new'; + text: string; } | { type: 'unknown'; command: string; diff --git a/lib/commands.js b/lib/commands.js index f1dedc4..f0d6648 100644 --- a/lib/commands.js +++ b/lib/commands.js @@ -86,6 +86,15 @@ export function parseCommand(text) { case '/unbind': case '/disconnect': return { type: 'unbind', text }; + case '/stop': + case '/halt': + return { type: 'stop', text }; + case '/mission': + case '/todos': + return { type: 'mission', text }; + case '/new': + case '/create': + return { type: 'new', text }; default: return { type: 'unknown', command, text }; } diff --git a/lib/index.d.ts b/lib/index.d.ts index 1636ab1..1b77e98 100644 --- a/lib/index.d.ts +++ b/lib/index.d.ts @@ -1,7 +1,7 @@ import type { Context } from '@deepseek-ai/cordis'; import Schema from '@deepseek-ai/schemastery'; export declare const name = "dsh-telegram-channel"; -/** agents: followup; apiProxy: workspace/session catalog + /model (Cordis requires inject to read ctx.apiProxy). */ +/** agents: followup; apiProxy: workspace/session catalog + /model; userQuestions: TG answer hook (Cordis requires inject). */ export declare const inject: string[]; export interface TelegramChannelConfig { token?: string; diff --git a/lib/index.js b/lib/index.js index d905698..05c9385 100644 --- a/lib/index.js +++ b/lib/index.js @@ -1,8 +1,8 @@ import Schema from '@deepseek-ai/schemastery'; import { TelegramBridge } from './bridge.js'; export const name = 'dsh-telegram-channel'; -/** agents: followup; apiProxy: workspace/session catalog + /model (Cordis requires inject to read ctx.apiProxy). */ -export const inject = ['agents', 'apiProxy']; +/** agents: followup; apiProxy: workspace/session catalog + /model; userQuestions: TG answer hook (Cordis requires inject). */ +export const inject = ['agents', 'apiProxy', 'userQuestions']; export const Config = Schema.object({ token: Schema.string().default(''), allowedUserIds: Schema.array(Schema.number()).default([]), diff --git a/src/apiproxy.ts b/src/apiproxy.ts index 1928d4f..3affb53 100644 --- a/src/apiproxy.ts +++ b/src/apiproxy.ts @@ -6,6 +6,8 @@ export type ApiProxySessions = { history?: (req: { rpcId: string; payload: unknown }) => Promise models?: (req: { rpcId: string; payload: unknown }) => Promise selectModel?: (req: { rpcId: string; payload: unknown }) => Promise + /** Create a session (used by /new; payload: workspaceId | cwd). */ + create?: (req: { rpcId: string; payload: unknown }) => Promise } export type ApiProxyLike = { diff --git a/src/bridge.ts b/src/bridge.ts index c835c0f..9e886ff 100644 --- a/src/bridge.ts +++ b/src/bridge.ts @@ -1,9 +1,13 @@ +import { mkdirSync, readFileSync, writeFileSync } from 'node:fs' +import { homedir } from 'node:os' +import { dirname, join } from 'node:path' import type { Context } from '@deepseek-ai/cordis' import type { Agent } from '@deepseek-ai/dsh-agent' import { createUserMessage } from '@deepseek-ai/dsh-llm/message' import type { ContentBlock, TextBlock } from '@deepseek-ai/dsh-llm/types' import { SessionId, type SessionEvent } from '@deepseek-ai/dsh-session/types' import { isAuthorized } from './auth.js' +import { resolveApiProxy } from './apiproxy.js' import { catalogFromLiveAgents, loadCatalog, @@ -62,6 +66,53 @@ interface SessionLike { id: ReturnType } +/** ── Stability & interactivity additions: local types (host seam shapes mirrored structurally) ── */ + +interface TgQuestionItem { + id?: string + question?: string + options?: Array<{ label: string; description?: string }> + multiSelect?: boolean +} + +interface TgAnswerHandle { + resolve: (answer: unknown) => void + reject: (err: Error) => void + answered: boolean +} + +/** chatId → pending ask_user_question mirrored to Telegram. */ +interface TgPendingAsk { + req: unknown + sessionId: string + questions: TgQuestionItem[] + ask: TgAnswerHandle + at: number +} + +/** chatId → pending approval request mirrored to Telegram. */ +interface TgPendingApproval { + toolName: string + ask: TgAnswerHandle +} + +/** Structural view of the user-questions seam (types live in the host's dsh-user-questions package). */ +interface UserQuestionsServiceLike { + provider?: { + ask: (request: unknown) => Promise + __tgHooked?: boolean + __tgRealAsk?: (request: unknown) => Promise + } +} + +/** Loose view of the approval request passed through the `approval/request` waterfall hook. */ +interface ApprovalRequestLike { + agent?: Agent + toolName?: string + reason?: string + signal?: AbortSignal +} + const WS_CB = 'ws:' const SID_CB = 'sid:' const BACK_WS_CB = 'wb' @@ -111,6 +162,24 @@ export class TelegramBridge { private pollAbort: AbortController | undefined private disposeSessionListener: (() => void) | undefined + // ── Stability & interactivity additions ── + /** sessionIds mid-turn (busy feedback, /stop state) */ + private readonly busySessions = new Set() + /** chatId → typing heartbeat interval handle */ + private readonly heartbeats = new Map>() + /** chatId → serialized notice chain (prevents 429 storms) */ + private readonly noticeQueue = new Map>() + private readonly pendingAsks = new Map() + private hookTimer: ReturnType | undefined + private readonly pendingApprovalsTG = new Map() + private disposeApprovalHook: (() => void) | undefined + /** sessionId → thinking indicator state (one notice per reasoning phase) */ + private readonly thinkingSessions = new Map() + /** callId → tool name (tool/result failure notices) */ + private readonly callNames = new Map() + /** sessionId → latest todo snapshot (/mission) */ + private readonly lastTodos = new Map>() + constructor(ctx: Context, options: TelegramBridgeOptions) { this.ctx = ctx this.token = options.token @@ -131,6 +200,14 @@ export class TelegramBridge { this.ctx.logger.error(this.redact(err)) }) }) + // Restore persisted bindings (survive hot reload / restart) and install + // the TG answering hooks (ask_user dual-path + approval dual-path). + this.loadBindings() + this.hookUserQuestions() + this.disposeApprovalHook?.() + this.disposeApprovalHook = (this.ctx as unknown as { + on: (event: string, handler: (req: ApprovalRequestLike, next: () => Promise) => Promise) => () => void + }).on('approval/request', (req, next) => this.onApprovalRequest(req, next)) void this.client.setMyCommands([ { command: 'start', description: '欢迎与用法' }, { command: 'sessions', description: '按工作区列出并附着会话' }, @@ -138,6 +215,9 @@ export class TelegramBridge { { command: 'model', description: '切换当前绑定会话的模型' }, { command: 'status', description: '查看当前绑定' }, { command: 'unbind', description: '断开手机绑定（不关闭本机会话）' }, + { command: 'stop', description: '中止当前正在运行的任务' }, + { command: 'mission', description: '查看任务清单与完成情况' }, + { command: 'new', description: '在当前工作区新开对话并附着' }, { command: 'help', description: '显示帮助' }, ]).then(() => { this.ctx.logger.info('dsh-telegram-channel: bot commands registered') @@ -162,6 +242,20 @@ export class TelegramBridge { this.bindings.clear() this.pickers.clear() this.pendingModels.clear() + this.stopAllHeartbeats() + this.busySessions.clear() + this.noticeQueue.clear() + this.pendingAsks.clear() + this.pendingApprovalsTG.clear() + this.thinkingSessions.clear() + this.callNames.clear() + this.lastTodos.clear() + if (this.hookTimer) { + clearTimeout(this.hookTimer) + this.hookTimer = undefined + } + this.disposeApprovalHook?.() + this.disposeApprovalHook = undefined if (this.pollPromise) { await this.pollPromise.catch(() => {}) this.pollPromise = undefined @@ -174,7 +268,7 @@ export class TelegramBridge { return } const message = update.message - if (!message?.text) return + if (!message) return const chatId = message.chat.id const userId = message.from?.id @@ -184,6 +278,12 @@ export class TelegramBridge { return } + // Non-text messages: tell the (authorized) user instead of dropping silently. + if (!message.text) { + await this.enqueueNotice(chatId, '暂只支持文本消息（图片/语音/文件等媒体暂不处理）') + return + } + const parsed = parseCommand(message.text) switch (parsed.type) { case 'start': @@ -207,14 +307,47 @@ export class TelegramBridge { case 'unbind': this.bindings.delete(String(chatId)) this.pickers.delete(String(chatId)) + this.pendingAsks.delete(String(chatId)) + this.saveBindings() await this.client.sendMessage(chatId, MSG.UNBOUND) return + case 'stop': + await this.stopBound(chatId) + return + case 'mission': + await this.sendMission(chatId) + return + case 'new': + await this.newSessionHere(chatId) + return case 'unknown': + // /cancel closes the TG-side answer/approval channel; the UI side stays open. + if (parsed.command === 'cancel') { + let cancelled = false + if (this.pendingApprovalsTG.delete(String(chatId))) cancelled = true + if (this.pendingAsks.delete(String(chatId))) cancelled = true + if (cancelled) { + await this.enqueueNotice(chatId, '已关闭 TG 端作答/审批（该问题仍可在 Web 端回答）') + return + } + } await this.client.sendMessage(chatId, MSG.unknown(parsed.command)) return - case 'plain': + case 'plain': { + // A pending approval/ask on TG consumes the message as an answer. + const approval = this.pendingApprovalsTG.get(String(chatId)) + if (approval) { + await this.handleTgApproval(chatId, approval, parsed.text) + return + } + const pending = this.pendingAsks.get(String(chatId)) + if (pending) { + await this.handleTgAnswer(chatId, pending, parsed.text) + return + } await this.followupBound(chatId, parsed.text) return + } } } @@ -424,6 +557,10 @@ export class TelegramBridge { ? displayLabel({ ...parts, title: row.title }) : displayLabel(parts) this.bindings.set(String(chatId), { chatId, sessionId: String(agent.id), label }) + // Drop stale pending asks when rebinding to another session; persist bindings. + const stale = this.pendingAsks.get(String(chatId)) + if (stale && stale.sessionId !== String(agent.id)) this.pendingAsks.delete(String(chatId)) + this.saveBindings() await this.client.answerCallbackQuery(callbackId, '已附着') await this.client.sendMessage(chatId, MSG.BOUND(label), undefined, lastContextKeyboard()) } @@ -617,7 +754,14 @@ export class TelegramBridge { content: [{ type: 'text', text }], source: { kind: 'user' }, }) + // Busy feedback: mid-turn followups queue silently by design — tell the user. + const busy = this.busySessions.has(binding.sessionId) + || Boolean((agent as { phase?: { kind?: string } }).phase?.kind + && (agent as { phase?: { kind?: string } }).phase?.kind !== 'idle') agent.followup(message) + if (busy) { + await this.enqueueNotice(chatId, '已加入队列：当前任务完成后将处理这条消息') + } } private async pollLoop(): Promise { @@ -677,41 +821,132 @@ export class TelegramBridge { if (targets.length === 0) return if (event.type === 'turn/start') { - await Promise.all(targets.map((b) => this.client.sendChatAction(b.chatId, 'typing'))) + // Typing heartbeat — one sendChatAction expires in ~5s, so refresh it + // for the whole turn instead of only at start. + this.busySessions.add(id) + for (const b of targets) this.startTypingHeartbeat(b.chatId) + return + } + + if (event.type === 'turn/end') { + this.busySessions.delete(id) + this.thinkingSessions.delete(id) + for (const b of targets) this.stopTypingHeartbeat(b.chatId) + // Abnormal endings (API failure / abort / crash-recovery) must be + // visible on the phone — previously they ended in silence. + const reason = event.data?.reason as { kind?: string; error?: { message?: string; code?: string }; reason?: { kind?: string } } | undefined + const kind = reason?.kind + if (kind === 'error') { + const detail = String(reason?.error?.message ?? reason?.error?.code ?? '').replace(/\s+/g, ' ').slice(0, 300) + for (const b of targets) + this.enqueueNotice(b.chatId, `本轮异常结束（API/内部错误）${detail ? `：${detail}` : ''}`) + } else if (kind === 'aborted') { + const manual = reason?.reason?.kind === 'user' + for (const b of targets) + this.enqueueNotice(b.chatId, manual ? '本轮已被手动中止' : '本轮已被中断') + } else if (kind === 'interrupted') { + for (const b of targets) + this.enqueueNotice(b.chatId, '会话曾被异常退出（崩溃恢复），本轮被标记中断') + } + return + } + + if (event.type === 'tool/call') { + // Progress summary; ask_user is owned by the provider hook (avoids duplicates). + const name = event.data?.name ?? 'tool' + if (name === 'ask_user_question') return + this.thinkingSessions.set(id, false) + if (event.data?.callId !== undefined) { + if (this.callNames.size > 400) { + const first = this.callNames.keys().next().value + if (first !== undefined) this.callNames.delete(first) + } + this.callNames.set(String(event.data.callId), String(name)) + } + const args = typeof event.data?.arguments === 'string' ? event.data.arguments : '' + const brief = args.replace(/\s+/g, ' ').trim() + const shown = brief.length > 60 ? `${brief.slice(0, 60)}…` : brief + for (const b of targets) + this.enqueueNotice(b.chatId, `> ${name}${shown ? ` ${shown}` : ''}`) return } if (event.type === 'assistant/message') { const text = contentToText(event.data.message.content) + this.thinkingSessions.set(id, false) if (!text) return await Promise.all(targets.map((b) => this.deliver(b.chatId, text))) + return + } + + if (event.type === 'assistant/chunk') { + // Thinking indicator: once per reasoning phase, status only (no content). + const chunk = event.data?.chunk as { type?: string; block?: { type?: string } } | undefined + const ctype = String(chunk?.type ?? '') + const btype = String(chunk?.block?.type ?? '') + const isReasoning = ctype.includes('reasoning') || btype.includes('reasoning') || btype.includes('think') + if (isReasoning && this.thinkingSessions.get(id) !== true) { + this.thinkingSessions.set(id, true) + for (const b of targets) + this.enqueueNotice(b.chatId, '> Thinking…') + } + return + } + + if (event.type === 'tool/result') { + // Only failures — successful results are too noisy for the phone. + const data = event.data as { error?: { name?: string; code?: string }; message?: { tool_use_id?: string; callId?: string } } + if (data?.error) { + const callKey = String(data.message?.tool_use_id ?? data.message?.callId ?? '') + const name = this.callNames.get(callKey) ?? 'tool' + for (const b of targets) + this.enqueueNotice(b.chatId, `> 工具 ${name} 失败：${String(data.error.name ?? 'Error')}${data.error.code ? `/${String(data.error.code)}` : ''}`) + } + return + } + + if (event.type === 'todo/write') { + const todos = Array.isArray(event.data?.todos) ? event.data.todos : [] + this.lastTodos.set(id, todos as Array<{ content?: string; status?: string }>) + if (todos.length === 0) return + const done = todos.filter((t) => t?.status === 'completed').length + const current = todos.find((t) => t?.status === 'in_progress') ?? todos.find((t) => t?.status === 'pending') + const cur = current ? String(current.content ?? '').slice(0, 50) : '' + for (const b of targets) + this.enqueueNotice(b.chatId, `> 进度 ${done}/${todos.length}${cur ? `：${cur}` : ''}`) + return } } private async deliver(chatId: number, markdown: string): Promise { - if (this.renderingMode === 'html') { - await this.deliverHtml(chatId, markdown) - return - } - - let sentAny = false - try { - const chunks = splitRichMarkdown(markdown) - for (const chunk of chunks) { - await this.client.sendRichMessage(chatId, chunk) - sentAny = true - } - } catch (err) { - if (!sentAny && isRichUnsupportedError(err)) { - this.ctx.logger.warn( - 'dsh-telegram-channel: Rich Message API unavailable, falling back to HTML rendering', - ) - this.renderingMode = 'html' - await this.deliverHtml(chatId, markdown) + if (this.renderingMode !== 'html') { + let sentAny = false + try { + const chunks = splitRichMarkdown(markdown) + for (const chunk of chunks) { + try { + // Per-chunk retry: transient network blips must not drop a reply. + await this.withRetry(() => this.client.sendRichMessage(chatId, chunk)) + sentAny = true + } catch (err) { + if (!sentAny && isRichUnsupportedError(err)) { + this.ctx.logger.warn( + 'dsh-telegram-channel: Rich Message API unavailable, falling back to HTML rendering', + ) + this.renderingMode = 'html' + await this.deliverHtml(chatId, markdown) + return + } + throw err + } + } return + } catch (err) { + // Persistent rich failure → HTML fallback instead of dropping the message. + this.ctx.logger.warn(`dsh-telegram-channel: rich deliver failed (${this.redact(err)}), falling back to HTML`) } - this.ctx.logger.error(this.redact(err)) } + await this.deliverHtml(chatId, markdown) } private async deliverHtml(chatId: number, markdown: string): Promise { @@ -719,10 +954,10 @@ export class TelegramBridge { for (const chunk of chunks) { const html = markdownToHtml(chunk) try { - await this.client.sendMessage(chatId, html, 'HTML') + await this.withRetry(() => this.client.sendMessage(chatId, html, 'HTML')) } catch { try { - await this.client.sendMessage(chatId, chunk) + await this.withRetry(() => this.client.sendMessage(chatId, chunk)) } catch (err) { this.ctx.logger.error(this.redact(err)) } @@ -730,6 +965,472 @@ export class TelegramBridge { } } + // ── Stability & interactivity helpers ── + + /** Retry an outbound call with capped linear backoff (500ms, 1s, 2s… max 4s). */ + private async withRetry(fn: () => Promise, tries = 3): Promise { + let lastErr: unknown + for (let attempt = 1; attempt <= tries; attempt++) { + try { + return await fn() + } catch (err) { + lastErr = err + if (attempt < tries) await this.sleep(Math.min(500 * 2 ** (attempt - 1), 4000)) + } + } + throw lastErr + } + + private escHtml(s: string): string { + return String(s) + .replace(/&/g, '&') + .replace(//g, '>') + } + + /** + * Send a notice; a leading `> ` renders as a Telegram blockquote (HTML). + * Falls back to plain text when the HTML send fails — a notice is never lost. + */ + private async deliverNotice(chatId: number, text: string): Promise { + const quoted = text.startsWith('> ') + const body = quoted ? text.slice(2) : text + try { + await this.withRetry(() => { + if (!quoted) return this.client.sendMessage(chatId, body) + return this.client.sendMessage(chatId, `
${this.escHtml(body)}
`, 'HTML') + }) + return + } catch { + try { + await this.client.sendMessage(chatId, quoted ? `> ${body}` : body) + } catch (err) { + this.ctx.logger.error(this.redact(err)) + } + } + } + + /** Serialized per-chat notice chain — bursts can't race into Telegram 429s. */ + private enqueueNotice(chatId: number, text: string): Promise { + const key = String(chatId) + const prev = this.noticeQueue.get(key) ?? Promise.resolve() + const next = prev.then(() => this.deliverNotice(chatId, text)).catch(() => {}) + this.noticeQueue.set(key, next) + return next + } + + private startTypingHeartbeat(chatId: number): void { + this.stopTypingHeartbeat(chatId) + const send = (): void => { + void this.client.sendChatAction(chatId, 'typing').catch(() => {}) + } + send() + this.heartbeats.set(String(chatId), setInterval(send, 4000)) + } + + private stopTypingHeartbeat(chatId: number): void { + const h = this.heartbeats.get(String(chatId)) + if (h) { + clearInterval(h) + this.heartbeats.delete(String(chatId)) + } + } + + private stopAllHeartbeats(): void { + for (const h of this.heartbeats.values()) clearInterval(h) + this.heartbeats.clear() + } + + // ── ask_user_question: TG answering via dual-path race with the UI ── + + private userQuestions(): UserQuestionsServiceLike | undefined { + return (this.ctx as unknown as { userQuestions?: UserQuestionsServiceLike }).userQuestions + } + + /** + * Wrap the UI provider's ask() so Telegram gets a parallel answer path. + * `Promise.race` decides; the UI path is untouched. The TG promise NEVER + * settles when there is no bound chat — race would kill the UI's window + * with that early rejection. + */ + private hookUserQuestions(attempt = 0): void { + try { + const provider = this.userQuestions()?.provider + if (provider) { + if (provider.__tgHooked && provider.__tgRealAsk) { + // Hot reload: re-point the wrapper at THIS bridge instance. + const realAsk = provider.__tgRealAsk + const self = this + provider.ask = function (request: unknown): Promise { + const tg = self.registerTgAsk(request) + let gui: Promise + try { + gui = realAsk(request) + } catch (err) { + tg.reject(err as Error) + throw err + } + void gui.then(() => self.settleGuiSide(request), () => self.settleGuiSide(request)) + return Promise.race([tg.promise, gui]) + } + return + } + const realAsk = provider.ask.bind(provider) as (request: unknown) => Promise + provider.__tgRealAsk = realAsk + const self = this + provider.ask = function (request: unknown): Promise { + const tg = self.registerTgAsk(request) + let gui: Promise + try { + gui = realAsk(request) + } catch (err) { + tg.reject(err as Error) + throw err + } + void gui.then(() => self.settleGuiSide(request), () => self.settleGuiSide(request)) + return Promise.race([tg.promise, gui]) + } + provider.__tgHooked = true + this.ctx.logger.info('dsh-telegram-channel: ask_user TG answering hook installed') + return + } + } catch (err) { + if (attempt === 0) + this.ctx.logger.warn(`dsh-telegram-channel: userQuestions hook deferred: ${this.redact(err)}`) + } + if (attempt >= 30) { + this.ctx.logger.warn('dsh-telegram-channel: user-questions provider never appeared; TG answering disabled') + return + } + this.hookTimer = setTimeout(() => this.hookUserQuestions(attempt + 1), 2000) + } + + private registerTgAsk(request: unknown): { promise: Promise; reject: (err: Error) => void } { + let resolveFn: (v: unknown) => void = () => {} + let rejectFn: (err: Error) => void = () => {} + const promise = new Promise((resolve, reject) => { + resolveFn = resolve + rejectFn = reject + }) + const ask: TgAnswerHandle = { resolve: resolveFn, reject: rejectFn, answered: false } + const req = request as { agent?: Agent; questions?: TgQuestionItem[]; signal?: AbortSignal } + const sessionId = req.agent?.id !== undefined ? String(req.agent.id) : undefined + const questions = Array.isArray(req.questions) ? req.questions : [] + const onAbort = (): void => { + for (const [chatId, p] of this.pendingAsks) { + if (p.ask === ask) this.pendingAsks.delete(chatId) + } + rejectFn(new Error('ask aborted')) + } + if (req.signal?.aborted) { + rejectFn(new Error('ask aborted')) + return { promise, reject: rejectFn } + } + req.signal?.addEventListener('abort', onAbort, { once: true }) + if (sessionId !== undefined && questions.length > 0) { + let asked = false + for (const [chatId, binding] of this.bindings) { + if (binding.sessionId !== sessionId) continue + this.pendingAsks.set(String(chatId), { req: request, sessionId, questions, ask, at: Date.now() }) + this.enqueueNotice(Number(chatId), this.formatAskPending(questions)) + asked = true + } + if (asked) return { promise, reject: rejectFn } + } + // No bound TG chat: never settle — the UI remains the only answer path. + return { promise, reject: rejectFn } + } + + private settleGuiSide(request: unknown): void { + for (const [chatId, p] of [...this.pendingAsks]) { + if (p.req !== request) continue + this.pendingAsks.delete(chatId) + if (!p.ask.answered) { + this.enqueueNotice(Number(chatId), '该问题已在 Web 端作答，TG 作答通道关闭') + } + } + } + + private formatAskPending(questions: TgQuestionItem[]): string { + const lines = ['待回答（回复字母选项，或直接输入自定义答案；/cancel 关闭 TG 作答）'] + questions.forEach((q, i) => { + const prefix = questions.length > 1 ? `Q${i + 1}. ` : '' + lines.push(`${prefix}${q.question ?? ''}`) + const opts = Array.isArray(q.options) ? q.options : [] + if (opts.length === 0) { + lines.push(' （开放问题，直接输入答案）') + } else { + opts.forEach((o, j) => lines.push(` [${String.fromCharCode(65 + j)}] ${o.label}${o.description ? ` — ${o.description}` : ''}`)) + if (q.multiSelect) lines.push(' （可多选，如：A,B）') + } + }) + return lines.join('\n') + } + + private async handleTgAnswer(chatId: number, pending: TgPendingAsk, text: string): Promise { + const { questions, ask } = pending + let parts: Array<[number | null, string]> + if (questions.length === 1) { + parts = [[0, text.trim()]] + } else { + parts = [] + const lines = text.split(/\n|；|;/).map((s) => s.trim()).filter(Boolean) + for (const line of lines) { + const m = line.match(/^([0-9]+)\s*[:：.、)]\s*(.+)$/) + if (m) parts.push([Number(m[1]) - 1, m[2]!.trim()]) + else parts.push([null, line]) + } + if (parts.some(([idx]) => idx === null)) { + await this.enqueueNotice(chatId, `共 ${questions.length} 个问题，请每行用 1:/2: 前缀分别作答`) + return + } + } + const answers: Array<{ id?: string; selected: string[]; custom?: string }> = [] + for (const [idx, raw] of parts) { + if (idx === null || idx < 0 || idx >= questions.length) { + await this.enqueueNotice(chatId, `题号 ${(idx ?? 0) + 1} 不存在（共 ${questions.length} 题）`) + return + } + const q = questions[idx]! + const cleaned = raw.replace(/^\[+/, '').replace(/\]+$/, '').trim() + const letters = cleaned.split(/[,，、\s]+/).map((s) => s.trim()).filter(Boolean) + const opts = Array.isArray(q.options) ? q.options : [] + const isLetterChoice = letters.length > 0 + && letters.every((s) => /^[a-zA-Z]$/.test(s)) + && opts.length > 0 + if (isLetterChoice) { + const selected: string[] = [] + for (const s of letters) { + const i = s.toUpperCase().charCodeAt(0) - 65 + if (i < 0 || i >= opts.length) { + await this.enqueueNotice(chatId, `选项 ${s.toUpperCase()} 不存在（范围 A-${String.fromCharCode(64 + opts.length)}），请重答或 /cancel`) + return + } + selected.push(opts[i]!.label) + } + answers.push({ id: q.id, selected }) + } else { + answers.push({ id: q.id, selected: [], custom: cleaned }) + } + } + ask.answered = true + for (const [cid, p] of this.pendingAsks) { + if (p.ask === ask) this.pendingAsks.delete(cid) + } + ask.resolve({ answers }) + this.enqueueNotice(chatId, '已提交你的回答') + } + + // ── approval: TG answering via dual-path race on the request waterfall ── + + private async onApprovalRequest( + req: ApprovalRequestLike, + next: () => Promise, + ): Promise { + const sessionId = req?.agent?.id !== undefined ? String(req.agent.id) : undefined + const targets = sessionId !== undefined + ? [...this.bindings.values()].filter((b) => b.sessionId === sessionId).map((b) => String(b.chatId)) + : [] + let resolveFn: (v: unknown) => void = () => {} + let rejectFn: (err: Error) => void = () => {} + const promise = new Promise((resolve, reject) => { + resolveFn = resolve + rejectFn = reject + }) + const ask: TgAnswerHandle = { resolve: resolveFn, reject: rejectFn, answered: false } + const onAbort = (): void => { + for (const [chatId, p] of this.pendingApprovalsTG) { + if (p.ask === ask) this.pendingApprovalsTG.delete(chatId) + } + rejectFn(new Error('approval withdrawn')) + } + if (req?.signal?.aborted) { + rejectFn(new Error('approval withdrawn')) + return next() + } + req?.signal?.addEventListener('abort', onAbort, { once: true }) + if (targets.length > 0) { + const toolName = String(req?.toolName ?? 'tool') + const reason = req?.reason ? String(req.reason).slice(0, 200) : '' + for (const chatId of targets) { + this.pendingApprovalsTG.set(chatId, { toolName, ask }) + this.enqueueNotice(Number(chatId), `权限审批：${toolName}${reason ? `\n(${reason})` : ''}\n回复 [A] 允许一次 / [B] 拒绝（/cancel 关闭 TG 审批）`) + } + } + // No bound TG chat: promise never settles — the UI remains the only answer path. + const gui = next() + void gui.then(() => { + for (const [chatId, p] of [...this.pendingApprovalsTG]) { + if (p.ask !== ask) continue + this.pendingApprovalsTG.delete(chatId) + if (!p.ask.answered) { + p.ask.reject(new Error('superseded-by-gui')) + this.enqueueNotice(Number(chatId), '该审批已在 Web 端处理') + } + } + }, () => { + for (const [chatId, p] of [...this.pendingApprovalsTG]) { + if (p.ask === ask) this.pendingApprovalsTG.delete(chatId) + } + }) + return Promise.race([promise, gui]) as Promise + } + + private async handleTgApproval(chatId: number, approval: TgPendingApproval, text: string): Promise { + const cleaned = text.trim().replace(/^\[+/, '').replace(/\]+$/, '').toUpperCase() + if (cleaned === 'A' || cleaned === 'B') { + approval.ask.answered = true + for (const [cid, p] of this.pendingApprovalsTG) { + if (p.ask === approval.ask) this.pendingApprovalsTG.delete(cid) + } + if (cleaned === 'A') { + approval.ask.resolve('allowed-once') + await this.enqueueNotice(chatId, `已允许 ${approval.toolName} 执行一次`) + } else { + approval.ask.resolve('rejected') + await this.enqueueNotice(chatId, '已拒绝该工具执行') + } + return + } + await this.enqueueNotice(chatId, '请回复 [A] 允许一次 或 [B] 拒绝（/cancel 关闭）') + } + + // ── /stop /mission /new commands ── + + private async stopBound(chatId: number): Promise { + const binding = this.bindings.get(String(chatId)) + if (!binding) { + await this.client.sendMessage(chatId, MSG.NEED_BIND) + return + } + const agent = this.findLiveAgent(binding.sessionId) + if (!agent) { + await this.client.sendMessage(chatId, '当前会话没有正在运行的任务') + return + } + try { + agent.cancel({ kind: 'user' }) + } catch (err) { + this.ctx.logger.warn(`dsh-telegram-channel: stop failed: ${this.redact(err)}`) + } + await this.enqueueNotice(chatId, '已请求中止当前轮…') + } + + private async sendMission(chatId: number): Promise { + const binding = this.bindings.get(String(chatId)) + if (!binding) { + await this.client.sendMessage(chatId, MSG.NEED_BIND) + return + } + const todos = this.lastTodos.get(binding.sessionId) ?? [] + const lines = ['任务清单'] + if (todos.length === 0) { + lines.push('（暂无任务记录，agent 使用 todo 工具后这里会显示）') + } else { + const mark: Record = { completed: '✓', in_progress: '⟳', pending: '·' } + todos.forEach((t, i) => lines.push(`${mark[t?.status ?? ''] ?? '·'} ${i + 1}. ${String(t?.content ?? '')}`)) + const done = todos.filter((t) => t?.status === 'completed').length + lines.push(`—— ${done}/${todos.length} 已完成`) + } + lines.push(this.busySessions.has(binding.sessionId) ? '（当前任务：运行中，/stop 可中止）' : '（当前空闲）') + await this.client.sendMessage(chatId, lines.join('\n')) + } + + /** /new: create a session in the bound session's workspace and attach to it. */ + private async newSessionHere(chatId: number): Promise { + const binding = this.bindings.get(String(chatId)) + if (!binding) { + await this.client.sendMessage(chatId, MSG.NEED_BIND) + return + } + let workspaceId: string | undefined + let fallbackCwd: string | undefined + try { + const catalog = await this.resolveCatalog() + for (const ws of catalog?.workspaces ?? []) { + if ((ws.sessionIds ?? []).includes(binding.sessionId)) { + workspaceId = ws.id + break + } + } + if (workspaceId === undefined) + fallbackCwd = catalog?.sessionsById?.get(binding.sessionId)?.cwd + } catch { + // Catalog unavailable → fall back to the host default workspace. + } + const proxy = resolveApiProxy(this.ctx) + if (!proxy?.sessions?.create) { + await this.client.sendMessage(chatId, '当前宿主不支持会话创建接口') + return + } + let res: { result?: { ok?: boolean; value?: { sessionId?: string }; error?: { message?: string } } } | undefined + try { + const payload = workspaceId !== undefined ? { workspaceId } : (fallbackCwd ? { cwd: fallbackCwd } : {}) + res = await proxy.sessions.create({ + rpcId: `tg-${Date.now()}-${Math.random().toString(36).slice(2)}`, + payload, + }) as typeof res + } catch (err) { + await this.client.sendMessage(chatId, `创建失败：${this.redact(err)}`) + return + } + const newId = res?.result?.ok ? res.result.value?.sessionId : undefined + if (!newId) { + const detail = res?.result?.error?.message ?? JSON.stringify(res?.result ?? res) + await this.client.sendMessage(chatId, `创建失败：${String(detail).slice(0, 200)}`) + return + } + const stale = this.pendingAsks.get(String(chatId)) + if (stale && stale.sessionId !== String(newId)) this.pendingAsks.delete(String(chatId)) + this.bindings.set(String(chatId), { chatId, sessionId: String(newId), label: binding.label }) + this.saveBindings() + const where = workspaceId !== undefined ? '当前工作区' : (fallbackCwd ? `目录 ${fallbackCwd}` : '默认工作区') + await this.client.sendMessage(chatId, `已在${where}新开对话并附着：\n${String(newId)}\n直接发消息即可开始`) + } + + // ── Binding persistence (survives hot reload / restart) ── + + private bindingsPath(): string { + // DSH_TELEGRAM_BINDINGS_FILE overrides the location (tests must isolate + // from the real profile data — parallel test files share this file). + const override = process.env.DSH_TELEGRAM_BINDINGS_FILE + if (override) return override + const home = process.env.DSH_HOME ?? join(homedir(), '.dsh') + return join(home, 'telegram-channel-bindings.json') + } + + private saveBindings(): void { + try { + const obj = Object.fromEntries(this.bindings) + mkdirSync(dirname(this.bindingsPath()), { recursive: true }) + writeFileSync(this.bindingsPath(), JSON.stringify(obj, null, 2)) + } catch (err) { + this.ctx.logger.warn(`dsh-telegram-channel: saveBindings failed: ${this.redact(err)}`) + } + } + + private loadBindings(): void { + try { + const raw = readFileSync(this.bindingsPath(), 'utf8') + const obj = JSON.parse(raw) as Record + let n = 0 + for (const [chatId, b] of Object.entries(obj ?? {})) { + if (b && b.sessionId) { + this.bindings.set(String(chatId), { + chatId: b.chatId ?? Number(chatId), + sessionId: String(b.sessionId), + label: b.label ?? String(b.sessionId), + }) + n += 1 + } + } + if (n > 0) + this.ctx.logger.info(`dsh-telegram-channel: restored ${n} binding(s) from disk`) + } catch { + // First run: no file yet. + } + } + private redact(value: unknown): string { const message = value instanceof Error ? value.message : String(value) return message.split(this.token).join('***') diff --git a/src/commands.ts b/src/commands.ts index ee80664..76d0aec 100644 --- a/src/commands.ts +++ b/src/commands.ts @@ -70,6 +70,9 @@ export type ParsedCommand = | { type: 'model'; text: string } | { type: 'status'; text: string } | { type: 'unbind'; text: string } + | { type: 'stop'; text: string } + | { type: 'mission'; text: string } + | { type: 'new'; text: string } | { type: 'unknown'; command: string; text: string } | { type: 'plain'; text: string } @@ -95,6 +98,15 @@ export function parseCommand(text: string): ParsedCommand { case '/unbind': case '/disconnect': return { type: 'unbind', text } + case '/stop': + case '/halt': + return { type: 'stop', text } + case '/mission': + case '/todos': + return { type: 'mission', text } + case '/new': + case '/create': + return { type: 'new', text } default: return { type: 'unknown', command, text } } diff --git a/src/index.ts b/src/index.ts index c1f4437..26ff6c9 100644 --- a/src/index.ts +++ b/src/index.ts @@ -3,8 +3,8 @@ import Schema from '@deepseek-ai/schemastery' import { TelegramBridge } from './bridge.js' export const name = 'dsh-telegram-channel' -/** agents: followup; apiProxy: workspace/session catalog + /model (Cordis requires inject to read ctx.apiProxy). */ -export const inject = ['agents', 'apiProxy'] +/** agents: followup; apiProxy: workspace/session catalog + /model; userQuestions: TG answer hook (Cordis requires inject). */ +export const inject = ['agents', 'apiProxy', 'userQuestions'] export interface TelegramChannelConfig { token?: string diff --git a/tests/bridge.test.ts b/tests/bridge.test.ts index e79ff69..cc00239 100644 --- a/tests/bridge.test.ts +++ b/tests/bridge.test.ts @@ -298,6 +298,10 @@ test('callback bind then plain text followups live agent; mirror assistant to ch data: { message: { content: [{ type: 'text', text: 'reply from agent' }] } }, }, ) + // Delivery is intentionally resilient now (per-chunk retry + rich→HTML + // fallback), which lengthens the async microtask chain past this listener's + // void return — flush the event loop before asserting. + await new Promise((r) => setTimeout(r, 0)) assert.ok(sent.some((m) => m.text.includes('reply from agent') || m.text.includes('reply'))) await bridge.stop()