fix(ask): prototype 拦截器替代 waterfall hook + TG 原生按钮作答

根因：dsh 0.1.5-rc.1 把 'user-questions/request' waterfall 分发到 per-fiber events 注册表，其他 fiber 的 ctx.on 监听（含本插件）永远收不到分发（实测 root/global/agent 作用域监听全部落空）。修复：包装 UserQuestionService.prototype.ask（类跨实例共享），无论哪个 fiber/实例发起的 ask 都能拦截到 agent+questions；TG 答案与原 ask 竞速，保留 dual-path 语义；stop 时恢复原型，热重载不叠加。功能（应用户要求）： - 提问前显式发送预告文本消息 - 选项改用 TG 原生 inline keyboard 按钮（单选即点即答，多选带提交键） - 文本作答保留并与按钮作答合并，逐题累计、齐题自动提交 - 附带修复：Telegram API 调用加 15s 客户端超时（代理挂死会卡死通知队列） Co-authored-by: Shirobana Agent Bot <tg-bridge>
2026-09-17 11:15:16 +08:00
parent 4039dec03c
commit da06485d64
8 changed files with 547 additions and 152 deletions
@@ -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 `<blockquote>` 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
+     * `<blockquote>` (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, `<blockquote>${escapeHtml(body)}</blockquote>`, '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) {