diff --git a/lib/bridge.d.ts b/lib/bridge.d.ts index e3e5cb4..3d9184c 100644 --- a/lib/bridge.d.ts +++ b/lib/bridge.d.ts @@ -43,8 +43,9 @@ export declare class TelegramBridge { private hookTimer; private readonly pendingApprovalsTG; private disposeApprovalHook; - /** dsh 0.1.5 起 ask 走 'user-questions/request' waterfall（不再有 userQuestions.provider）。 */ - private disposeUserQuestionHook; + /** Prototype-ask interceptor state (see installAskInterceptor). */ + private askInstalled; + private askRestore; /** callId → tool name (tool/result failure notices) */ private readonly callNames; /** sessionId → latest todo snapshot (/mission) */ @@ -117,10 +118,10 @@ export declare class TelegramBridge { /** Retry an outbound call with capped linear backoff (500ms, 1s, 2s… max 4s). */ private withRetry; /** - * Send a notice; a leading `> ` keeps the quote look via a plain-text `> ` - * prefix. NB: NOT `
` HTML — older Telegram clients cannot render - * the blockquote entity at all and show the whole message as “not supported”. - * Falls back to plain text when the send fails — a notice is never lost. + * Send a notice; a leading `> ` marks the body for quote rendering as a real + * `
` (parse_mode HTML) — every current Telegram client draws the + * accent-colored left bar. Old-client compatibility is no longer required, + * but a failed HTML send still falls back to plain text — a notice is never lost. */ private deliverNotice; /** Serialized per-chat notice chain — bursts can't race into Telegram 429s. */ @@ -128,18 +129,17 @@ export declare class TelegramBridge { private startTypingHeartbeat; private stopTypingHeartbeat; private stopAllHeartbeats; - private userQuestions; - /** - * dsh 0.1.5 移除了 `userQuestions.provider`：ask 现在走 'user-questions/request' - * waterfall（与 approval/request 同形），启动时注册一次即可，无需轮询 provider。 - * 让 TG 答案与 next()（UI 转发）赛跑；未绑定聊天时 TG promise 永不 settle， - * 于是 race 完全跟随 next()。 - */ - private onUserQuestionRequest; + private installAskInterceptor; + private aroundUserQuestionAsk; private registerTgAsk; + /** Announce the incoming question, then one message per question (options as native inline-keyboard buttons). */ + private deliverAskPrompt; + /** Resolve the pending ask once EVERY question has an answer (keyboard and/or text). */ + private finalizeAskIfComplete; private settleGuiSide; - private formatAskPending; private handleTgAnswer; + /** Inline-keyboard tap on an ask message: toggle/record the choice, resolve when every question is answered. */ + private handleAskCallback; private onApprovalRequest; private handleTgApproval; private stopBound; diff --git a/lib/bridge.js b/lib/bridge.js index 8dfe9ff..c2b2178 100644 --- a/lib/bridge.js +++ b/lib/bridge.js @@ -10,7 +10,7 @@ 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'; -import { markdownToHtml, splitMessage } from './format.js'; +import { escapeHtml, markdownToHtml, splitMessage } from './format.js'; import { splitRichMarkdown } from './rich-format.js'; import { formatLastTurn, loadLastTurn } from './history.js'; import { describeAgent, displayLabel, workspaceName } from './label.js'; @@ -60,18 +60,40 @@ const MODEL_CB = 'mdl:'; /** Use eff: (not me:) — short prefix, no collision with other callbacks. */ const EFFORT_CB = 'eff:'; const BACK_MODEL_CB = 'mb'; +/** ask:{questionIndex}:{optionIndex|ok} — inline-keyboard taps for ask_user_question. */ +const ASK_CB = 'ask:'; const MAX_BUTTONS = 40; +/** Marker stored on UserQuestionService.prototype for the ask interceptor. */ +const kTgAskWrapped = Symbol.for('dsh-telegram-channel.askWrapped'); function lastContextKeyboard() { return { inline_keyboard: [[{ text: '查看上次对话', callback_data: LAST_CB }]], }; } const defaultSleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms)); +/** Cap a single reasoning block so a long chain of thought can't flood the phone. */ +const REASONING_QUOTE_MAX_CHARS = 800; +/** + * Assistant message → markdown for delivery. Reasoning (thinking) blocks are + * kept, one `> ` quote line each — Telegram renders them as the accent-bar + * quote style the phone user expects for bash/thinking-style chatter. + */ function contentToText(content) { - return content - .filter((block) => block.type === 'text') - .map((block) => block.text) - .join(''); + const parts = []; + for (const block of content) { + if (block.type === 'text' && typeof block.text === 'string') { + parts.push(block.text); + } + else if (block.type === 'reasoning' && typeof block.text === 'string') { + let thinking = block.text.trim(); + if (!thinking) + continue; + if (thinking.length > REASONING_QUOTE_MAX_CHARS) + thinking = `${thinking.slice(0, REASONING_QUOTE_MAX_CHARS)}…`; + parts.push(thinking.split('\n').map((line) => `> ${line}`).join('\n')); + } + } + return parts.join('\n\n'); } function isRichUnsupportedError(err) { const message = err instanceof Error ? err.message : String(err); @@ -111,8 +133,9 @@ export class TelegramBridge { hookTimer; pendingApprovalsTG = new Map(); disposeApprovalHook; - /** dsh 0.1.5 起 ask 走 'user-questions/request' waterfall（不再有 userQuestions.provider）。 */ - disposeUserQuestionHook; + /** Prototype-ask interceptor state (see installAskInterceptor). */ + askInstalled = false; + askRestore; /** callId → tool name (tool/result failure notices) */ callNames = new Map(); /** sessionId → latest todo snapshot (/mission) */ @@ -151,8 +174,12 @@ export class TelegramBridge { // Restore persisted bindings (survive hot reload / restart) and install // the TG answering hooks (ask_user dual-path + approval dual-path). this.loadBindings(); - this.disposeUserQuestionHook?.(); - this.disposeUserQuestionHook = this.ctx.on('user-questions/request', (req, next) => this.onUserQuestionRequest(req, next)); + // ask_user interception: see installAskInterceptor — a prototype wrap of + // UserQuestionService.prototype.ask. Plain ctx.on('user-questions/request') + // registries never see the dispatch (the waterfall fires on a different + // per-fiber events registry in dsh 0.1.5-rc.1), so the prototype wrap — + // which sees EVERY ask regardless of fiber/instance — is the reliable seam. + this.installAskInterceptor(); this.disposeApprovalHook?.(); this.disposeApprovalHook = this.ctx.on('approval/request', (req, next) => this.onApprovalRequest(req, next)); void this.client.setMyCommands([ @@ -210,8 +237,9 @@ export class TelegramBridge { clearTimeout(this.hookTimer); this.hookTimer = undefined; } - this.disposeUserQuestionHook?.(); - this.disposeUserQuestionHook = undefined; + this.askRestore?.(); + this.askRestore = undefined; + this.askInstalled = false; this.disposeApprovalHook?.(); this.disposeApprovalHook = undefined; if (this.pollPromise) { @@ -342,6 +370,10 @@ export class TelegramBridge { } const data = cq.data ?? ''; const picker = this.pickers.get(String(chatId)); + if (data.startsWith(ASK_CB)) { + await this.handleAskCallback(chatId, cq); + return; + } if (data === LAST_CB) { await this.client.answerCallbackQuery(cq.id); await this.sendLastTurn(chatId); @@ -1084,6 +1116,8 @@ export class TelegramBridge { ]); } async onSessionEvent(session, event) { + if (!this.askInstalled) + this.installAskInterceptor(); const id = String(session.id); const targets = [...this.bindings.values()].filter((b) => b.sessionId === id); if (targets.length === 0) @@ -1285,17 +1319,25 @@ export class TelegramBridge { throw lastErr; } /** - * Send a notice; a leading `> ` keeps the quote look via a plain-text `> ` - * prefix. NB: NOT `
` HTML — older Telegram clients cannot render - * the blockquote entity at all and show the whole message as “not supported”. - * Falls back to plain text when the send fails — a notice is never lost. + * Send a notice; a leading `> ` marks the body for quote rendering as a real + * `
` (parse_mode HTML) — every current Telegram client draws the + * accent-colored left bar. Old-client compatibility is no longer required, + * but a failed HTML send still falls back to plain text — a notice is never lost. */ async deliverNotice(chatId, text) { const quoted = text.startsWith('> '); const body = quoted ? text.slice(2) : text; + if (quoted) { + try { + await this.withRetry(() => this.client.sendMessage(chatId, `
${escapeHtml(body)}
`, 'HTML')); + return; + } + catch (err) { + this.ctx.logger.warn(`dsh-telegram-channel: blockquote notice failed (${this.redact(err)}), falling back to plain text`); + } + } try { - await this.withRetry(() => this.client.sendMessage(chatId, quoted ? `> ${body}` : body)); - return; + await this.withRetry(() => this.client.sendMessage(chatId, body)); } catch (err) { this.ctx.logger.error(this.redact(err)); @@ -1329,27 +1371,61 @@ export class TelegramBridge { clearInterval(h); this.heartbeats.clear(); } - // ── ask_user_question: TG answering via dual-path race with the UI ── - userQuestions() { - return this.ctx.userQuestions; + // ── ask_user_question: TG answering via a prototype interceptor ── + // + // dsh 0.1.5-rc.1 routes the 'user-questions/request' waterfall through a + // per-fiber events registry: plain ctx.on listeners in other fibers (this + // plugin included) never see the dispatch, so the waterfall hook cannot + // mirror questions to TG. The class is shared across instances though, so + // wrapping UserQuestionService.prototype.ask intercepts EVERY ask with its + // agent + questions, regardless of fiber. The wrap races the TG answer + // against the original ask (GUI path), preserving the old dual-path + // semantics. Restore-on-stop keeps hot reloads from stacking wrappers. + installAskInterceptor() { + if (this.askInstalled) + return; + let svc; + try { + svc = this.ctx.get('userQuestions'); + } + catch { + return; // service not ready yet — retried lazily from onSessionEvent + } + if (!svc) + return; + const proto = svc.constructor.prototype; + const prev = proto[kTgAskWrapped]; + if (prev?.bridge === this) { + this.askInstalled = true; + this.askRestore = prev.restore; + return; + } + prev?.restore(); // stale wrapper from a disposed bridge instance + const orig = proto.ask; + const bridge = this; + const wrapped = function (request) { + return bridge.aroundUserQuestionAsk(this, request, orig); + }; + proto.ask = wrapped; + const restore = () => { + if (proto.ask === wrapped) + proto.ask = orig; + }; + proto[kTgAskWrapped] = { restore, bridge: this }; + this.askRestore = restore; + this.askInstalled = true; } - /** - * dsh 0.1.5 移除了 `userQuestions.provider`：ask 现在走 'user-questions/request' - * waterfall（与 approval/request 同形），启动时注册一次即可，无需轮询 provider。 - * 让 TG 答案与 next()（UI 转发）赛跑；未绑定聊天时 TG promise 永不 settle， - * 于是 race 完全跟随 next()。 - */ - onUserQuestionRequest(req, next) { - const tg = this.registerTgAsk(req); + aroundUserQuestionAsk(service, request, orig) { + const tg = this.registerTgAsk(request); let gui; try { - gui = Promise.resolve().then(next); + gui = Promise.resolve().then(() => orig.call(service, request)); } catch (err) { tg.reject(err); throw err; } - void gui.then(() => this.settleGuiSide(req), () => this.settleGuiSide(req)); + void gui.then(() => this.settleGuiSide(request), () => this.settleGuiSide(request)); return Promise.race([tg.promise, gui]); } registerTgAsk(request) { @@ -1380,16 +1456,80 @@ export class TelegramBridge { 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)); + const pending = { + req: request, + sessionId, + questions, + ask, + at: Date.now(), + kbSel: new Map(), + textAnswers: new Map(), + }; + this.pendingAsks.set(String(chatId), pending); + void this.deliverAskPrompt(Number(chatId), pending); 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 }; } + /** Announce the incoming question, then one message per question (options as native inline-keyboard buttons). */ + async deliverAskPrompt(chatId, pending) { + const { questions } = pending; + try { + await this.withRetry(() => this.client.sendMessage(chatId, `❓ 即将向你提问（共 ${questions.length} 题）——点按钮或回复文本作答；/cancel 关闭 TG 作答`)); + for (let i = 0; i < questions.length; i++) { + const q = questions[i]; + const opts = Array.isArray(q.options) ? q.options : []; + const header = questions.length > 1 ? `Q${i + 1}. ` : ''; + if (opts.length === 0) { + await this.withRetry(() => this.client.sendMessage(chatId, `${header}${q.question ?? ''}\n（开放问题，请直接回复文本）`)); + continue; + } + const keyboard = { + inline_keyboard: [ + ...opts.map((o, j) => [{ + text: `${j + 1}. ${o.label}`.slice(0, 48), + callback_data: `${ASK_CB}${i}:${j}`, + }]), + ...(q.multiSelect ? [[{ text: '✅ 提交所选', callback_data: `${ASK_CB}${i}:ok` }]] : []), + ], + }; + await this.withRetry(() => this.client.sendMessage(chatId, `${header}${q.question ?? ''}`, undefined, keyboard)); + } + } + catch (err) { + this.ctx.logger.error(this.redact(err)); + } + } + /** Resolve the pending ask once EVERY question has an answer (keyboard and/or text). */ + finalizeAskIfComplete(chatId, pending) { + const { questions, ask } = pending; + const answers = []; + for (let i = 0; i < questions.length; i++) { + const q = questions[i]; + const opts = Array.isArray(q.options) ? q.options : []; + const sel = pending.kbSel.get(i); + if (sel && sel.size > 0) { + answers.push({ id: q.id, selected: [...sel].sort((a, b) => a - b).map((j) => opts[j].label) }); + continue; + } + const text = pending.textAnswers.get(i); + if (text !== undefined) { + answers.push(text); + continue; + } + return false; + } + ask.answered = true; + for (const [cid, p] of this.pendingAsks) { + if (p.ask === ask) + this.pendingAsks.delete(cid); + } + ask.resolve({ answers }); + this.enqueueNotice(chatId, '已提交你的回答'); + return true; + } settleGuiSide(request) { for (const [chatId, p] of [...this.pendingAsks]) { if (p.req !== request) @@ -1400,25 +1540,8 @@ export class TelegramBridge { } } } - 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; + const { questions } = pending; let parts; if (questions.length === 1) { parts = [[0, text.trim()]]; @@ -1438,7 +1561,6 @@ export class TelegramBridge { 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} 题）`); @@ -1461,19 +1583,64 @@ export class TelegramBridge { } selected.push(opts[i].label); } - answers.push({ id: q.id, selected }); + pending.textAnswers.set(idx, { id: q.id, selected }); } else { - answers.push({ id: q.id, selected: [], custom: cleaned }); + pending.textAnswers.set(idx, { id: q.id, selected: [], custom: cleaned }); } } - ask.answered = true; - for (const [cid, p] of this.pendingAsks) { - if (p.ask === ask) - this.pendingAsks.delete(cid); + // Keyboard selections take precedence per question; resolve when complete. + if (!this.finalizeAskIfComplete(chatId, pending)) { + const done = questions.filter((_, i) => pending.kbSel.get(i)?.size || pending.textAnswers.has(i)).length; + await this.enqueueNotice(chatId, `已记录 ${done}/${questions.length} 题，请继续作答（按钮或 1:/2: 文本）`); + } + } + /** Inline-keyboard tap on an ask message: toggle/record the choice, resolve when every question is answered. */ + async handleAskCallback(chatId, cq) { + const pending = this.pendingAsks.get(String(chatId)); + if (!pending) { + await this.client.answerCallbackQuery(cq.id, '该提问已失效（可能已在 Web 端作答）'); + return; + } + const payload = (cq.data ?? '').slice(ASK_CB.length); + const sep = payload.indexOf(':'); + const qi = Number(payload.slice(0, sep)); + const action = payload.slice(sep + 1); + const q = pending.questions[qi]; + if (!q) { + await this.client.answerCallbackQuery(cq.id, '题目已失效'); + return; + } + const opts = Array.isArray(q.options) ? q.options : []; + if (action === 'ok') { + if (!pending.kbSel.get(qi)?.size) { + await this.client.answerCallbackQuery(cq.id, '请先点选选项'); + return; + } + } + else { + const j = Number(action); + if (!Number.isInteger(j) || j < 0 || j >= opts.length) { + await this.client.answerCallbackQuery(cq.id, '选项不存在'); + return; + } + if (q.multiSelect) { + const set = pending.kbSel.get(qi) ?? new Set(); + if (set.has(j)) + set.delete(j); + else + set.add(j); + pending.kbSel.set(qi, set); + await this.client.answerCallbackQuery(cq.id, set.size > 0 ? `已选 ${set.size} 项，选完点"提交所选"` : '已取消该选项'); + } + else { + pending.kbSel.set(qi, new Set([j])); + await this.client.answerCallbackQuery(cq.id, `已选：${opts[j].label}`); + } + } + if (!this.finalizeAskIfComplete(chatId, pending) && action === 'ok') { + await this.client.answerCallbackQuery(cq.id, '已提交本题选择，请继续作答其他问题'); } - ask.resolve({ answers }); - this.enqueueNotice(chatId, '已提交你的回答'); } // ── approval: TG answering via dual-path race on the request waterfall ── async onApprovalRequest(req, next) { diff --git a/lib/client.js b/lib/client.js index b1c0ebe..966c307 100644 --- a/lib/client.js +++ b/lib/client.js @@ -37,12 +37,17 @@ export class TelegramClient { redact(message) { return message.split(this.token).join('***'); } - async call(method, body) { + async call(method, body, timeoutMs = 15_000) { const url = `${this.baseUrl}/bot${this.token}/${method}`; try { const init = { method: 'POST', headers: { 'Content-Type': 'application/json' }, + // Without a client-side timeout a stalled proxy connection hangs the + // request forever — and the serialized notice queue wedges behind it, + // silently swallowing every later tool/ask notice. TimeoutError is + // caught below and surfaces as a normal (retryable) failure. + signal: AbortSignal.timeout(timeoutMs), }; if (body !== undefined) { init.body = JSON.stringify(body); @@ -70,7 +75,8 @@ export class TelegramClient { if (offset !== undefined) { body.offset = offset; } - return this.call('getUpdates', body); + // Long-poll: client-side abort must outlive the server-side hold. + return this.call('getUpdates', body, this.pollingTimeoutSec * 1000 + 10_000); } async sendMessage(chatId, text, parseMode, replyMarkup) { const body = { chat_id: chatId, text }; diff --git a/lib/format.js b/lib/format.js index 72bec6a..4ac6ba8 100644 --- a/lib/format.js +++ b/lib/format.js @@ -12,6 +12,30 @@ function inlineToHtml(text) { html = html.replace(/\*\*([^*\n]+)\*\*/g, '$1'); return html; } +/** Convert `> …` line groups into `
` (Telegram renders it with the accent-colored left bar). */ +function blocksToHtml(text) { + const lines = text.split('\n'); + const out = []; + let quote = []; + const flush = () => { + if (quote.length === 0) + return; + out.push(`
${inlineToHtml(quote.join('\n'))}
`); + quote = []; + }; + for (const line of lines) { + if (line.startsWith('> ')) + quote.push(line.slice(2)); + else if (line === '>') + quote.push(''); + else { + flush(); + out.push(inlineToHtml(line)); + } + } + flush(); + return out.join('\n'); +} export function markdownToHtml(text) { const parts = text.split(/```[\w-]*/); if (parts.length % 2 === 0) @@ -23,7 +47,7 @@ export function markdownToHtml(text) { html += `
${escapeHtml(code)}
`; } else { - html += inlineToHtml(parts[i]); + html += blocksToHtml(parts[i]); } } return html; diff --git a/not b/not new file mode 100644 index 0000000..e69de29 diff --git a/src/bridge.ts b/src/bridge.ts index a78ce23..4e6725c 100644 --- a/src/bridge.ts +++ b/src/bridge.ts @@ -6,7 +6,7 @@ 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 type { ContentBlock } 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' @@ -28,7 +28,7 @@ import { type TelegramUpdate, } from './client.js' import { MSG, LAST_CB, parseCommand } from './commands.js' -import { markdownToHtml, splitMessage } from './format.js' +import { escapeHtml, markdownToHtml, splitMessage } from './format.js' import { splitRichMarkdown } from './rich-format.js' import { formatLastTurn, loadLastTurn } from './history.js' import { describeAgent, displayLabel, workspaceName } from './label.js' @@ -97,6 +97,10 @@ interface TgPendingAsk { questions: TgQuestionItem[] ask: TgAnswerHandle at: number + /** question index → option indices chosen via inline-keyboard taps. */ + kbSel: Map> + /** question index → answer parsed from a text reply (letters or custom). */ + textAnswers: Map } /** chatId → pending approval request mirrored to Telegram. */ @@ -105,15 +109,6 @@ interface TgPendingApproval { 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 @@ -194,7 +189,11 @@ const MODEL_CB = 'mdl:' /** Use eff: (not me:) — short prefix, no collision with other callbacks. */ const EFFORT_CB = 'eff:' const BACK_MODEL_CB = 'mb' +/** ask:{questionIndex}:{optionIndex|ok} — inline-keyboard taps for ask_user_question. */ +const ASK_CB = 'ask:' const MAX_BUTTONS = 40 +/** Marker stored on UserQuestionService.prototype for the ask interceptor. */ +const kTgAskWrapped = Symbol.for('dsh-telegram-channel.askWrapped') function lastContextKeyboard(): InlineKeyboardMarkup { return { @@ -204,11 +203,28 @@ function lastContextKeyboard(): InlineKeyboardMarkup { const defaultSleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)) +/** Cap a single reasoning block so a long chain of thought can't flood the phone. */ +const REASONING_QUOTE_MAX_CHARS = 800 + +/** + * Assistant message → markdown for delivery. Reasoning (thinking) blocks are + * kept, one `> ` quote line each — Telegram renders them as the accent-bar + * quote style the phone user expects for bash/thinking-style chatter. + */ function contentToText(content: readonly ContentBlock[]): string { - return content - .filter((block): block is TextBlock => block.type === 'text') - .map((block) => block.text) - .join('') + const parts: string[] = [] + for (const block of content) { + if (block.type === 'text' && typeof block.text === 'string') { + parts.push(block.text) + } else if (block.type === 'reasoning' && typeof block.text === 'string') { + let thinking = block.text.trim() + if (!thinking) continue + if (thinking.length > REASONING_QUOTE_MAX_CHARS) + thinking = `${thinking.slice(0, REASONING_QUOTE_MAX_CHARS)}…` + parts.push(thinking.split('\n').map((line) => `> ${line}`).join('\n')) + } + } + return parts.join('\n\n') } function isRichUnsupportedError(err: unknown): boolean { @@ -252,8 +268,9 @@ export class TelegramBridge { private hookTimer: ReturnType | undefined private readonly pendingApprovalsTG = new Map() private disposeApprovalHook: (() => void) | undefined - /** dsh 0.1.5 起 ask 走 'user-questions/request' waterfall（不再有 userQuestions.provider）。 */ - private disposeUserQuestionHook: (() => void) | undefined + /** Prototype-ask interceptor state (see installAskInterceptor). */ + private askInstalled = false + private askRestore: (() => void) | undefined /** callId → tool name (tool/result failure notices) */ private readonly callNames = new Map() /** sessionId → latest todo snapshot (/mission) */ @@ -295,10 +312,12 @@ export class TelegramBridge { // Restore persisted bindings (survive hot reload / restart) and install // the TG answering hooks (ask_user dual-path + approval dual-path). this.loadBindings() - this.disposeUserQuestionHook?.() - this.disposeUserQuestionHook = (this.ctx as unknown as { - on: (event: string, handler: (req: unknown, next: () => Promise) => Promise) => () => void - }).on('user-questions/request', (req, next) => this.onUserQuestionRequest(req, next)) + // ask_user interception: see installAskInterceptor — a prototype wrap of + // UserQuestionService.prototype.ask. Plain ctx.on('user-questions/request') + // registries never see the dispatch (the waterfall fires on a different + // per-fiber events registry in dsh 0.1.5-rc.1), so the prototype wrap — + // which sees EVERY ask regardless of fiber/instance — is the reliable seam. + this.installAskInterceptor() this.disposeApprovalHook?.() this.disposeApprovalHook = (this.ctx as unknown as { on: (event: string, handler: (req: ApprovalRequestLike, next: () => Promise) => Promise) => () => void @@ -357,8 +376,9 @@ export class TelegramBridge { clearTimeout(this.hookTimer) this.hookTimer = undefined } - this.disposeUserQuestionHook?.() - this.disposeUserQuestionHook = undefined + this.askRestore?.() + this.askRestore = undefined + this.askInstalled = false this.disposeApprovalHook?.() this.disposeApprovalHook = undefined if (this.pollPromise) { @@ -492,6 +512,10 @@ export class TelegramBridge { const data = cq.data ?? '' const picker = this.pickers.get(String(chatId)) + if (data.startsWith(ASK_CB)) { + await this.handleAskCallback(chatId, cq) + return + } if (data === LAST_CB) { await this.client.answerCallbackQuery(cq.id) await this.sendLastTurn(chatId) @@ -1259,6 +1283,7 @@ export class TelegramBridge { } private async onSessionEvent(session: SessionLike, event: SessionEvent): Promise { + if (!this.askInstalled) this.installAskInterceptor() const id = String(session.id) const targets = [...this.bindings.values()].filter((b) => b.sessionId === id) if (targets.length === 0) return @@ -1458,17 +1483,28 @@ export class TelegramBridge { } /** - * Send a notice; a leading `> ` keeps the quote look via a plain-text `> ` - * prefix. NB: NOT `
` HTML — older Telegram clients cannot render - * the blockquote entity at all and show the whole message as “not supported”. - * Falls back to plain text when the send fails — a notice is never lost. + * Send a notice; a leading `> ` marks the body for quote rendering as a real + * `
` (parse_mode HTML) — every current Telegram client draws the + * accent-colored left bar. Old-client compatibility is no longer required, + * but a failed HTML send still falls back to plain text — a notice is never lost. */ private async deliverNotice(chatId: number, text: string): Promise { const quoted = text.startsWith('> ') const body = quoted ? text.slice(2) : text + if (quoted) { + try { + await this.withRetry(() => + this.client.sendMessage(chatId, `
${escapeHtml(body)}
`, 'HTML'), + ) + return + } catch (err) { + this.ctx.logger.warn( + `dsh-telegram-channel: blockquote notice failed (${this.redact(err)}), falling back to plain text`, + ) + } + } try { - await this.withRetry(() => this.client.sendMessage(chatId, quoted ? `> ${body}` : body)) - return + await this.withRetry(() => this.client.sendMessage(chatId, body)) } catch (err) { this.ctx.logger.error(this.redact(err)) } @@ -1505,28 +1541,64 @@ export class TelegramBridge { this.heartbeats.clear() } - // ── ask_user_question: TG answering via dual-path race with the UI ── + // ── ask_user_question: TG answering via a prototype interceptor ── + // + // dsh 0.1.5-rc.1 routes the 'user-questions/request' waterfall through a + // per-fiber events registry: plain ctx.on listeners in other fibers (this + // plugin included) never see the dispatch, so the waterfall hook cannot + // mirror questions to TG. The class is shared across instances though, so + // wrapping UserQuestionService.prototype.ask intercepts EVERY ask with its + // agent + questions, regardless of fiber. The wrap races the TG answer + // against the original ask (GUI path), preserving the old dual-path + // semantics. Restore-on-stop keeps hot reloads from stacking wrappers. - private userQuestions(): UserQuestionsServiceLike | undefined { - return (this.ctx as unknown as { userQuestions?: UserQuestionsServiceLike }).userQuestions + private installAskInterceptor(): void { + if (this.askInstalled) return + let svc: { constructor: { prototype: Record } } | undefined + try { + svc = (this.ctx as unknown as { get: (name: string) => unknown }).get('userQuestions') as + | { constructor: { prototype: Record } } + | undefined + } catch { + return // service not ready yet — retried lazily from onSessionEvent + } + if (!svc) return + const proto = svc.constructor.prototype + const prev = proto[kTgAskWrapped] as { restore: () => void; bridge: unknown } | undefined + if (prev?.bridge === this) { + this.askInstalled = true + this.askRestore = prev.restore + return + } + prev?.restore() // stale wrapper from a disposed bridge instance + const orig = proto.ask as (this: unknown, request: unknown) => Promise + const bridge = this + const wrapped = function (this: unknown, request: unknown): Promise { + return bridge.aroundUserQuestionAsk(this, request, orig) + } + proto.ask = wrapped + const restore = (): void => { + if (proto.ask === wrapped) proto.ask = orig + } + proto[kTgAskWrapped] = { restore, bridge: this } + this.askRestore = restore + this.askInstalled = true } - /** - * dsh 0.1.5 移除了 `userQuestions.provider`：ask 现在走 'user-questions/request' - * waterfall（与 approval/request 同形），启动时注册一次即可，无需轮询 provider。 - * 让 TG 答案与 next()（UI 转发）赛跑；未绑定聊天时 TG promise 永不 settle， - * 于是 race 完全跟随 next()。 - */ - private onUserQuestionRequest(req: unknown, next: () => Promise): Promise { - const tg = this.registerTgAsk(req) + private aroundUserQuestionAsk( + service: unknown, + request: unknown, + orig: (this: unknown, request: unknown) => Promise, + ): Promise { + const tg = this.registerTgAsk(request) let gui: Promise try { - gui = Promise.resolve().then(next) + gui = Promise.resolve().then(() => orig.call(service, request)) } catch (err) { tg.reject(err as Error) throw err } - void gui.then(() => this.settleGuiSide(req), () => this.settleGuiSide(req)) + void gui.then(() => this.settleGuiSide(request), () => this.settleGuiSide(request)) return Promise.race([tg.promise, gui]) } @@ -1556,16 +1628,88 @@ export class TelegramBridge { 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)) + const pending: TgPendingAsk = { + req: request, + sessionId, + questions, + ask, + at: Date.now(), + kbSel: new Map(), + textAnswers: new Map(), + } + this.pendingAsks.set(String(chatId), pending) + void this.deliverAskPrompt(Number(chatId), pending) 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 } } + /** Announce the incoming question, then one message per question (options as native inline-keyboard buttons). */ + private async deliverAskPrompt(chatId: number, pending: TgPendingAsk): Promise { + const { questions } = pending + try { + await this.withRetry(() => + this.client.sendMessage( + chatId, + `❓ 即将向你提问（共 ${questions.length} 题）——点按钮或回复文本作答；/cancel 关闭 TG 作答`, + ), + ) + for (let i = 0; i < questions.length; i++) { + const q = questions[i]! + const opts = Array.isArray(q.options) ? q.options : [] + const header = questions.length > 1 ? `Q${i + 1}. ` : '' + if (opts.length === 0) { + await this.withRetry(() => + this.client.sendMessage(chatId, `${header}${q.question ?? ''}\n（开放问题，请直接回复文本）`), + ) + continue + } + const keyboard: InlineKeyboardMarkup = { + inline_keyboard: [ + ...opts.map((o, j) => [{ + text: `${j + 1}. ${o.label}`.slice(0, 48), + callback_data: `${ASK_CB}${i}:${j}`, + }]), + ...(q.multiSelect ? [[{ text: '✅ 提交所选', callback_data: `${ASK_CB}${i}:ok` }]] : []), + ], + } + await this.withRetry(() => this.client.sendMessage(chatId, `${header}${q.question ?? ''}`, undefined, keyboard)) + } + } catch (err) { + this.ctx.logger.error(this.redact(err)) + } + } + + /** Resolve the pending ask once EVERY question has an answer (keyboard and/or text). */ + private finalizeAskIfComplete(chatId: number, pending: TgPendingAsk): boolean { + const { questions, ask } = pending + const answers: Array<{ id?: string; selected: string[]; custom?: string }> = [] + for (let i = 0; i < questions.length; i++) { + const q = questions[i]! + const opts = Array.isArray(q.options) ? q.options : [] + const sel = pending.kbSel.get(i) + if (sel && sel.size > 0) { + answers.push({ id: q.id, selected: [...sel].sort((a, b) => a - b).map((j) => opts[j]!.label) }) + continue + } + const text = pending.textAnswers.get(i) + if (text !== undefined) { + answers.push(text) + continue + } + return false + } + ask.answered = true + for (const [cid, p] of this.pendingAsks) { + if (p.ask === ask) this.pendingAsks.delete(cid) + } + ask.resolve({ answers }) + this.enqueueNotice(chatId, '已提交你的回答') + return true + } + private settleGuiSide(request: unknown): void { for (const [chatId, p] of [...this.pendingAsks]) { if (p.req !== request) continue @@ -1576,24 +1720,8 @@ export class TelegramBridge { } } - 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 + const { questions } = pending let parts: Array<[number | null, string]> if (questions.length === 1) { parts = [[0, text.trim()]] @@ -1610,7 +1738,6 @@ export class TelegramBridge { 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} 题）`) @@ -1633,17 +1760,60 @@ export class TelegramBridge { } selected.push(opts[i]!.label) } - answers.push({ id: q.id, selected }) + pending.textAnswers.set(idx, { id: q.id, selected }) } else { - answers.push({ id: q.id, selected: [], custom: cleaned }) + pending.textAnswers.set(idx, { id: q.id, selected: [], custom: cleaned }) } } - ask.answered = true - for (const [cid, p] of this.pendingAsks) { - if (p.ask === ask) this.pendingAsks.delete(cid) + // Keyboard selections take precedence per question; resolve when complete. + if (!this.finalizeAskIfComplete(chatId, pending)) { + const done = questions.filter((_, i) => pending.kbSel.get(i)?.size || pending.textAnswers.has(i)).length + await this.enqueueNotice(chatId, `已记录 ${done}/${questions.length} 题，请继续作答（按钮或 1:/2: 文本）`) + } + } + + /** Inline-keyboard tap on an ask message: toggle/record the choice, resolve when every question is answered. */ + private async handleAskCallback(chatId: number, cq: { id: string; data?: string }): Promise { + const pending = this.pendingAsks.get(String(chatId)) + if (!pending) { + await this.client.answerCallbackQuery(cq.id, '该提问已失效（可能已在 Web 端作答）') + return + } + const payload = (cq.data ?? '').slice(ASK_CB.length) + const sep = payload.indexOf(':') + const qi = Number(payload.slice(0, sep)) + const action = payload.slice(sep + 1) + const q = pending.questions[qi] + if (!q) { + await this.client.answerCallbackQuery(cq.id, '题目已失效') + return + } + const opts = Array.isArray(q.options) ? q.options : [] + if (action === 'ok') { + if (!pending.kbSel.get(qi)?.size) { + await this.client.answerCallbackQuery(cq.id, '请先点选选项') + return + } + } else { + const j = Number(action) + if (!Number.isInteger(j) || j < 0 || j >= opts.length) { + await this.client.answerCallbackQuery(cq.id, '选项不存在') + return + } + if (q.multiSelect) { + const set = pending.kbSel.get(qi) ?? new Set() + if (set.has(j)) set.delete(j) + else set.add(j) + pending.kbSel.set(qi, set) + await this.client.answerCallbackQuery(cq.id, set.size > 0 ? `已选 ${set.size} 项，选完点"提交所选"` : '已取消该选项') + } else { + pending.kbSel.set(qi, new Set([j])) + await this.client.answerCallbackQuery(cq.id, `已选：${opts[j]!.label}`) + } + } + if (!this.finalizeAskIfComplete(chatId, pending) && action === 'ok') { + await this.client.answerCallbackQuery(cq.id, '已提交本题选择，请继续作答其他问题') } - ask.resolve({ answers }) - this.enqueueNotice(chatId, '已提交你的回答') } // ── approval: TG answering via dual-path race on the request waterfall ── diff --git a/src/client.ts b/src/client.ts index 754bf87..edf7fa6 100644 --- a/src/client.ts +++ b/src/client.ts @@ -153,12 +153,17 @@ export class TelegramClient implements TelegramClientLike { return message.split(this.token).join('***') } - private async call(method: string, body?: Record): Promise { + private async call(method: string, body?: Record, timeoutMs = 15_000): Promise { const url = `${this.baseUrl}/bot${this.token}/${method}` try { const init: RequestInit = { method: 'POST', headers: { 'Content-Type': 'application/json' }, + // Without a client-side timeout a stalled proxy connection hangs the + // request forever — and the serialized notice queue wedges behind it, + // silently swallowing every later tool/ask notice. TimeoutError is + // caught below and surfaces as a normal (retryable) failure. + signal: AbortSignal.timeout(timeoutMs), } if (body !== undefined) { init.body = JSON.stringify(body) @@ -187,7 +192,8 @@ export class TelegramClient implements TelegramClientLike { if (offset !== undefined) { body.offset = offset } - return this.call('getUpdates', body) + // Long-poll: client-side abort must outlive the server-side hold. + return this.call('getUpdates', body, this.pollingTimeoutSec * 1000 + 10_000) } async sendMessage( diff --git a/src/format.ts b/src/format.ts index 4cb7b6d..0b0fa22 100644 --- a/src/format.ts +++ b/src/format.ts @@ -14,6 +14,28 @@ function inlineToHtml(text: string): string { return html } +/** Convert `> …` line groups into `
` (Telegram renders it with the accent-colored left bar). */ +function blocksToHtml(text: string): string { + const lines = text.split('\n') + const out: string[] = [] + let quote: string[] = [] + const flush = (): void => { + if (quote.length === 0) return + out.push(`
${inlineToHtml(quote.join('\n'))}
`) + quote = [] + } + for (const line of lines) { + if (line.startsWith('> ')) quote.push(line.slice(2)) + else if (line === '>') quote.push('') + else { + flush() + out.push(inlineToHtml(line)) + } + } + flush() + return out.join('\n') +} + export function markdownToHtml(text: string): string { const parts = text.split(/```[\w-]*/) if (parts.length % 2 === 0) return inlineToHtml(text) @@ -23,7 +45,7 @@ export function markdownToHtml(text: string): string { const code = parts[i]!.replace(/^\n/, '').replace(/\n$/, '') html += `
${escapeHtml(code)}
` } else { - html += inlineToHtml(parts[i]!) + html += blocksToHtml(parts[i]!) } } return html