Compare commits

..
8 Commits
Author SHA1 Message Date
root 0b946ffaa4 Merge remote-tracking branch 'gitea/master'
# Conflicts:
#	lib/bridge.js
#	src/bridge.ts
2026-09-17 11:24:47 +08:00
root 2c50a35795 feat(ask): 键盘常态提供 ✏️ 输入自定义答案 按钮
- 每个带选项的提问键盘固定追加第三行 ✏️ 输入自定义答案
- 点击后该题进入待文本状态(textWanted),丢弃该题已点选的按钮项
- 下一条文本消息自动填充最先待文本的题;带 1:/2: 前缀的作答同样可清除待文本状态
- 已实测:✏️ → 发送 111 → custom 答案正确回流
2026-09-17 11:21:30 +08:00
root 373abf586d chore: 移除误加的杂物文件 2026-09-17 11:15:25 +08:00
rootandShirobana Agent Bot <tg-bridge> da06485d64 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
dsh-agent 4039dec03c feat(config): 渲染默认 rich(用户要求 2026-09-14)
Config.rendering 默认 'html'→'rich';显式 'html'/'compat' 仍可强制兼容模式。
profile cordis.patch.yml 的 rendering 同步改为 rich(patch 会覆盖插件默认值,
两处都改才生效)。旧客户端显示'不支持'时用 /rich off 按聊天切回。
2026-09-16 10:42:45 +08:00
dsh-agent 5fa2827ed6 feat(tg): 去掉 '> Thinking…' 思考提示流水
用户要求:不再发送思考指示(assistant/chunk 的 reasoning 分支整段移除,
连带 thinkingSessions 状态与相关清理)。tool/call、tool/result、todo/write、
turn/end 的流水保持不变。
2026-09-14 19:16:28 +08:00
dsh-agent 9a45bde25a test: 兜底 DSH_HOME,杜绝用例把 fixture 写进真实绑定文件
tests/bridge.test.ts 有若干用例直接构造 bridge 并 bind,未设
DSH_TELEGRAM_BINDINGS_FILE,绑定落盘走 DSH_HOME 默认值 → 覆盖
/root/.dsh/telegram-channel-bindings.json(2026-09-14 实际发生,覆盖掉用户
真实绑定)。模块顶部把 DSH_HOME 指向 /tmp 兜底,显式 override 的用例不受影响。
2026-09-14 19:07:08 +08:00
dsh-agent 9f03c6c748 fix(ask_user): dsh 0.1.5 走 user-questions/request waterfall
0.1.5 移除了 userQuestions.provider,改为 waterfall 事件注册(与 approval/request 同形);
同时把 2026-09-12 只在编译产物上做过的修复回植到 src,避免源码/产物漂移。
2026-09-14 19:03:08 +08:00
11 changed files with 610 additions and 312 deletions
+15 -15
View File
@@ -43,8 +43,9 @@ export declare class TelegramBridge {
private hookTimer; private hookTimer;
private readonly pendingApprovalsTG; private readonly pendingApprovalsTG;
private disposeApprovalHook; private disposeApprovalHook;
/** sessionId → thinking indicator state (one notice per reasoning phase) */ /** Prototype-ask interceptor state (see installAskInterceptor). */
private readonly thinkingSessions; private askInstalled;
private askRestore;
/** callId → tool name (tool/result failure notices) */ /** callId → tool name (tool/result failure notices) */
private readonly callNames; private readonly callNames;
/** sessionId → latest todo snapshot (/mission) */ /** 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). */ /** Retry an outbound call with capped linear backoff (500ms, 1s, 2s… max 4s). */
private withRetry; private withRetry;
/** /**
* Send a notice; a leading `> ` keeps the quote look via a plain-text `> ` * Send a notice; a leading `> ` marks the body for quote rendering as a real
* prefix. NB: NOT `<blockquote>` HTML — older Telegram clients cannot render * `<blockquote>` (parse_mode HTML) — every current Telegram client draws the
* the blockquote entity at all and show the whole message as “not supported”. * accent-colored left bar. Old-client compatibility is no longer required,
* Falls back to plain text when the send fails — a notice is never lost. * but a failed HTML send still falls back to plain text — a notice is never lost.
*/ */
private deliverNotice; private deliverNotice;
/** Serialized per-chat notice chain — bursts can't race into Telegram 429s. */ /** Serialized per-chat notice chain — bursts can't race into Telegram 429s. */
@@ -128,18 +129,17 @@ export declare class TelegramBridge {
private startTypingHeartbeat; private startTypingHeartbeat;
private stopTypingHeartbeat; private stopTypingHeartbeat;
private stopAllHeartbeats; private stopAllHeartbeats;
private userQuestions; private installAskInterceptor;
/** private aroundUserQuestionAsk;
* 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 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 settleGuiSide;
private formatAskPending;
private handleTgAnswer; private handleTgAnswer;
/** Inline-keyboard tap on an ask message: toggle/record the choice, resolve when every question is answered. */
private handleAskCallback;
private onApprovalRequest; private onApprovalRequest;
private handleTgApproval; private handleTgApproval;
private stopBound; private stopBound;
+252 -139
View File
@@ -10,7 +10,7 @@ import { resolveApiProxy } from './apiproxy.js';
import { catalogFromLiveAgents, loadCatalog, truncateButton, visibleSessionsForWorkspace, workspacesWithVisibleSessions, } from './catalog.js'; import { catalogFromLiveAgents, loadCatalog, truncateButton, visibleSessionsForWorkspace, workspacesWithVisibleSessions, } from './catalog.js';
import { TelegramClient, } from './client.js'; import { TelegramClient, } from './client.js';
import { MSG, LAST_CB, parseCommand } from './commands.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 { splitRichMarkdown } from './rich-format.js';
import { formatLastTurn, loadLastTurn } from './history.js'; import { formatLastTurn, loadLastTurn } from './history.js';
import { describeAgent, displayLabel, workspaceName } from './label.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. */ /** Use eff: (not me:) — short prefix, no collision with other callbacks. */
const EFFORT_CB = 'eff:'; const EFFORT_CB = 'eff:';
const BACK_MODEL_CB = 'mb'; const BACK_MODEL_CB = 'mb';
/** ask:{questionIndex}:{optionIndex|ok} — inline-keyboard taps for ask_user_question. */
const ASK_CB = 'ask:';
const MAX_BUTTONS = 40; const MAX_BUTTONS = 40;
/** Marker stored on UserQuestionService.prototype for the ask interceptor. */
const kTgAskWrapped = Symbol.for('dsh-telegram-channel.askWrapped');
function lastContextKeyboard() { function lastContextKeyboard() {
return { return {
inline_keyboard: [[{ text: '查看上次对话', callback_data: LAST_CB }]], inline_keyboard: [[{ text: '查看上次对话', callback_data: LAST_CB }]],
}; };
} }
const defaultSleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms)); 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) { function contentToText(content) {
return content const parts = [];
.filter((block) => block.type === 'text') for (const block of content) {
.map((block) => block.text) if (block.type === 'text' && typeof block.text === 'string') {
.join(''); 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) { function isRichUnsupportedError(err) {
const message = err instanceof Error ? err.message : String(err); const message = err instanceof Error ? err.message : String(err);
@@ -111,8 +133,9 @@ export class TelegramBridge {
hookTimer; hookTimer;
pendingApprovalsTG = new Map(); pendingApprovalsTG = new Map();
disposeApprovalHook; disposeApprovalHook;
/** sessionId → thinking indicator state (one notice per reasoning phase) */ /** Prototype-ask interceptor state (see installAskInterceptor). */
thinkingSessions = new Map(); askInstalled = false;
askRestore;
/** callId → tool name (tool/result failure notices) */ /** callId → tool name (tool/result failure notices) */
callNames = new Map(); callNames = new Map();
/** sessionId → latest todo snapshot (/mission) */ /** sessionId → latest todo snapshot (/mission) */
@@ -151,7 +174,12 @@ export class TelegramBridge {
// Restore persisted bindings (survive hot reload / restart) and install // Restore persisted bindings (survive hot reload / restart) and install
// the TG answering hooks (ask_user dual-path + approval dual-path). // the TG answering hooks (ask_user dual-path + approval dual-path).
this.loadBindings(); this.loadBindings();
this.hookUserQuestions(); // 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.disposeApprovalHook = this.ctx.on('approval/request', (req, next) => this.onApprovalRequest(req, next)); this.disposeApprovalHook = this.ctx.on('approval/request', (req, next) => this.onApprovalRequest(req, next));
void this.client.setMyCommands([ void this.client.setMyCommands([
@@ -196,7 +224,6 @@ export class TelegramBridge {
this.noticeQueue.clear(); this.noticeQueue.clear();
this.pendingAsks.clear(); this.pendingAsks.clear();
this.pendingApprovalsTG.clear(); this.pendingApprovalsTG.clear();
this.thinkingSessions.clear();
this.callNames.clear(); this.callNames.clear();
this.lastTodos.clear(); this.lastTodos.clear();
for (const controller of this.compactAborts.values()) for (const controller of this.compactAborts.values())
@@ -210,6 +237,9 @@ export class TelegramBridge {
clearTimeout(this.hookTimer); clearTimeout(this.hookTimer);
this.hookTimer = undefined; this.hookTimer = undefined;
} }
this.askRestore?.();
this.askRestore = undefined;
this.askInstalled = false;
this.disposeApprovalHook?.(); this.disposeApprovalHook?.();
this.disposeApprovalHook = undefined; this.disposeApprovalHook = undefined;
if (this.pollPromise) { if (this.pollPromise) {
@@ -340,6 +370,10 @@ export class TelegramBridge {
} }
const data = cq.data ?? ''; const data = cq.data ?? '';
const picker = this.pickers.get(String(chatId)); const picker = this.pickers.get(String(chatId));
if (data.startsWith(ASK_CB)) {
await this.handleAskCallback(chatId, cq);
return;
}
if (data === LAST_CB) { if (data === LAST_CB) {
await this.client.answerCallbackQuery(cq.id); await this.client.answerCallbackQuery(cq.id);
await this.sendLastTurn(chatId); await this.sendLastTurn(chatId);
@@ -1082,6 +1116,8 @@ export class TelegramBridge {
]); ]);
} }
async onSessionEvent(session, event) { async onSessionEvent(session, event) {
if (!this.askInstalled)
this.installAskInterceptor();
const id = String(session.id); const id = String(session.id);
const targets = [...this.bindings.values()].filter((b) => b.sessionId === id); const targets = [...this.bindings.values()].filter((b) => b.sessionId === id);
if (targets.length === 0) if (targets.length === 0)
@@ -1096,7 +1132,6 @@ export class TelegramBridge {
} }
if (event.type === 'turn/end') { if (event.type === 'turn/end') {
this.busySessions.delete(id); this.busySessions.delete(id);
this.thinkingSessions.delete(id);
for (const b of targets) for (const b of targets)
this.stopTypingHeartbeat(b.chatId); this.stopTypingHeartbeat(b.chatId);
// Abnormal endings (API failure / abort / crash-recovery) must be // Abnormal endings (API failure / abort / crash-recovery) must be
@@ -1124,7 +1159,6 @@ export class TelegramBridge {
const name = event.data?.name ?? 'tool'; const name = event.data?.name ?? 'tool';
if (name === 'ask_user_question') if (name === 'ask_user_question')
return; return;
this.thinkingSessions.set(id, false);
if (event.data?.callId !== undefined) { if (event.data?.callId !== undefined) {
if (this.callNames.size > 400) { if (this.callNames.size > 400) {
const first = this.callNames.keys().next().value; const first = this.callNames.keys().next().value;
@@ -1142,25 +1176,11 @@ export class TelegramBridge {
} }
if (event.type === 'assistant/message') { if (event.type === 'assistant/message') {
const text = contentToText(event.data.message.content); const text = contentToText(event.data.message.content);
this.thinkingSessions.set(id, false);
if (!text) if (!text)
return; return;
await Promise.all(targets.map((b) => this.deliver(b.chatId, text))); await Promise.all(targets.map((b) => this.deliver(b.chatId, text)));
return; 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') { if (event.type === 'tool/result') {
// Only failures — successful results are too noisy for the phone. // Only failures — successful results are too noisy for the phone.
const data = event.data; const data = event.data;
@@ -1299,17 +1319,25 @@ export class TelegramBridge {
throw lastErr; throw lastErr;
} }
/** /**
* Send a notice; a leading `> ` keeps the quote look via a plain-text `> ` * Send a notice; a leading `> ` marks the body for quote rendering as a real
* prefix. NB: NOT `<blockquote>` HTML — older Telegram clients cannot render * `<blockquote>` (parse_mode HTML) — every current Telegram client draws the
* the blockquote entity at all and show the whole message as “not supported”. * accent-colored left bar. Old-client compatibility is no longer required,
* Falls back to plain text when the send fails — a notice is never lost. * but a failed HTML send still falls back to plain text — a notice is never lost.
*/ */
async deliverNotice(chatId, text) { async deliverNotice(chatId, text) {
const quoted = text.startsWith('> '); const quoted = text.startsWith('> ');
const body = quoted ? text.slice(2) : text; 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 { try {
await this.withRetry(() => this.client.sendMessage(chatId, quoted ? `> ${body}` : body)); await this.withRetry(() => this.client.sendMessage(chatId, body));
return;
} }
catch (err) { catch (err) {
this.ctx.logger.error(this.redact(err)); this.ctx.logger.error(this.redact(err));
@@ -1343,87 +1371,62 @@ export class TelegramBridge {
clearInterval(h); clearInterval(h);
this.heartbeats.clear(); this.heartbeats.clear();
} }
// ── ask_user_question: TG answering via dual-path race with the UI ── // ── ask_user_question: TG answering via a prototype interceptor ──
userQuestions() { //
const ctx = this.ctx; // dsh 0.1.5-rc.1 routes the 'user-questions/request' waterfall through a
if (typeof ctx.get === 'function') { // per-fiber events registry: plain ctx.on listeners in other fibers (this
try { // plugin included) never see the dispatch, so the waterfall hook cannot
const service = ctx.get('userQuestions'); // mirror questions to TG. The class is shared across instances though, so
if (service) // wrapping UserQuestionService.prototype.ask intercepts EVERY ask with its
return service; // agent + questions, regardless of fiber. The wrap races the TG answer
} // against the original ask (GUI path), preserving the old dual-path
catch { // semantics. Restore-on-stop keeps hot reloads from stacking wrappers.
// Continue to the plain-object fallback used by tests and simple hosts. installAskInterceptor() {
} if (this.askInstalled)
} return;
let svc;
try { try {
return Object.prototype.hasOwnProperty.call(ctx, 'userQuestions') svc = this.ctx.get('userQuestions');
? ctx.userQuestions
: undefined;
} }
catch { catch {
return undefined; return; // service not ready yet — retried lazily from onSessionEvent
} }
} if (!svc)
/** return;
* Wrap the UI provider's ask() so Telegram gets a parallel answer path. const proto = svc.constructor.prototype;
* `Promise.race` decides; the UI path is untouched. The TG promise NEVER const prev = proto[kTgAskWrapped];
* settles when there is no bound chat — race would kill the UI's window if (prev?.bridge === this) {
* with that early rejection. this.askInstalled = true;
*/ this.askRestore = prev.restore;
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; return;
} }
this.hookTimer = setTimeout(() => this.hookUserQuestions(attempt + 1), 2000); 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;
}
aroundUserQuestionAsk(service, request, orig) {
const tg = this.registerTgAsk(request);
let gui;
try {
gui = Promise.resolve().then(() => orig.call(service, request));
}
catch (err) {
tg.reject(err);
throw err;
}
void gui.then(() => this.settleGuiSide(request), () => this.settleGuiSide(request));
return Promise.race([tg.promise, gui]);
} }
registerTgAsk(request) { registerTgAsk(request) {
let resolveFn = () => { }; let resolveFn = () => { };
@@ -1453,16 +1456,83 @@ export class TelegramBridge {
for (const [chatId, binding] of this.bindings) { for (const [chatId, binding] of this.bindings) {
if (binding.sessionId !== sessionId) if (binding.sessionId !== sessionId)
continue; continue;
this.pendingAsks.set(String(chatId), { req: request, sessionId, questions, ask, at: Date.now() }); const pending = {
this.enqueueNotice(Number(chatId), this.formatAskPending(questions)); req: request,
sessionId,
questions,
ask,
at: Date.now(),
kbSel: new Map(),
textAnswers: new Map(),
textWanted: new Set(),
};
this.pendingAsks.set(String(chatId), pending);
void this.deliverAskPrompt(Number(chatId), pending);
asked = true; asked = true;
} }
if (asked)
return { promise, reject: rejectFn };
} }
// No bound TG chat: never settle — the UI remains the only answer path. // No bound TG chat: never settle — the UI remains the only answer path.
return { promise, reject: rejectFn }; 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}`,
}]),
// Standing third choice: free-text answer, always available.
[{ text: '✏️ 输入自定义答案', callback_data: `${ASK_CB}${i}:custom` }],
...(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) { settleGuiSide(request) {
for (const [chatId, p] of [...this.pendingAsks]) { for (const [chatId, p] of [...this.pendingAsks]) {
if (p.req !== request) if (p.req !== request)
@@ -1473,25 +1543,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) { async handleTgAnswer(chatId, pending, text) {
const { questions, ask } = pending; const { questions } = pending;
let parts; let parts;
if (questions.length === 1) { if (questions.length === 1) {
parts = [[0, text.trim()]]; parts = [[0, text.trim()]];
@@ -1507,11 +1560,17 @@ export class TelegramBridge {
parts.push([null, line]); parts.push([null, line]);
} }
if (parts.some(([idx]) => idx === null)) { if (parts.some(([idx]) => idx === null)) {
await this.enqueueNotice(chatId, `共 ${questions.length} 个问题,请每行用 1:/2: 前缀分别作答`); // ✏️ free-text route: a question awaiting text absorbs the whole message.
return; const target = [...pending.textWanted].sort((a, b) => a - b)[0];
if (target !== undefined) {
parts = [[target, text.trim()]];
}
else {
await this.enqueueNotice(chatId, `共 ${questions.length} 个问题,请每行用 1:/2: 前缀分别作答`);
return;
}
} }
} }
const answers = [];
for (const [idx, raw] of parts) { for (const [idx, raw] of parts) {
if (idx === null || idx < 0 || idx >= questions.length) { if (idx === null || idx < 0 || idx >= questions.length) {
await this.enqueueNotice(chatId, `题号 ${(idx ?? 0) + 1} 不存在(共 ${questions.length} 题)`); await this.enqueueNotice(chatId, `题号 ${(idx ?? 0) + 1} 不存在(共 ${questions.length} 题)`);
@@ -1534,19 +1593,73 @@ export class TelegramBridge {
} }
selected.push(opts[i].label); selected.push(opts[i].label);
} }
answers.push({ id: q.id, selected }); pending.textAnswers.set(idx, { id: q.id, selected });
} }
else { else {
answers.push({ id: q.id, selected: [], custom: cleaned }); pending.textAnswers.set(idx, { id: q.id, selected: [], custom: cleaned });
}
pending.textWanted.delete(idx);
}
// 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 === 'custom') {
// Free-text path: discard button selections for this question and wait
// for the user's next text message (routed via textWanted).
pending.kbSel.delete(qi);
pending.textWanted.add(qi);
await this.client.answerCallbackQuery(cq.id, '请直接发送你的答案文本');
return;
}
if (action === 'ok') {
if (!pending.kbSel.get(qi)?.size) {
await this.client.answerCallbackQuery(cq.id, '请先点选选项');
return;
} }
} }
ask.answered = true; else {
for (const [cid, p] of this.pendingAsks) { const j = Number(action);
if (p.ask === ask) if (!Number.isInteger(j) || j < 0 || j >= opts.length) {
this.pendingAsks.delete(cid); 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 ── // ── approval: TG answering via dual-path race on the request waterfall ──
async onApprovalRequest(req, next) { async onApprovalRequest(req, next) {
+8 -2
View File
@@ -37,12 +37,17 @@ export class TelegramClient {
redact(message) { redact(message) {
return message.split(this.token).join('***'); 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}`; const url = `${this.baseUrl}/bot${this.token}/${method}`;
try { try {
const init = { const init = {
method: 'POST', method: 'POST',
headers: { 'Content-Type': 'application/json' }, 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) { if (body !== undefined) {
init.body = JSON.stringify(body); init.body = JSON.stringify(body);
@@ -70,7 +75,8 @@ export class TelegramClient {
if (offset !== undefined) { if (offset !== undefined) {
body.offset = offset; 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) { async sendMessage(chatId, text, parseMode, replyMarkup) {
const body = { chat_id: chatId, text }; const body = { chat_id: chatId, text };
+25 -1
View File
@@ -12,6 +12,30 @@ function inlineToHtml(text) {
html = html.replace(/\*\*([^*\n]+)\*\*/g, '<b>$1</b>'); html = html.replace(/\*\*([^*\n]+)\*\*/g, '<b>$1</b>');
return html; return html;
} }
/** Convert `> …` line groups into `<blockquote>` (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(`<blockquote>${inlineToHtml(quote.join('\n'))}</blockquote>`);
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) { export function markdownToHtml(text) {
const parts = text.split(/```[\w-]*/); const parts = text.split(/```[\w-]*/);
if (parts.length % 2 === 0) if (parts.length % 2 === 0)
@@ -23,7 +47,7 @@ export function markdownToHtml(text) {
html += `<pre>${escapeHtml(code)}</pre>`; html += `<pre>${escapeHtml(code)}</pre>`;
} }
else { else {
html += inlineToHtml(parts[i]); html += blocksToHtml(parts[i]);
} }
} }
return html; return html;
+2 -2
View File
@@ -9,8 +9,8 @@ export interface TelegramChannelConfig {
allowAllUsers?: boolean; allowAllUsers?: boolean;
maxMessageLength?: number; maxMessageLength?: number;
pollingTimeoutSec?: number; pollingTimeoutSec?: number;
/** 'rich' = Telegram native Rich Messages (needs a recent client); /** 默认 'rich' = Telegram native Rich Messages(需较新客户端);
* anything else (default) = HTML compatibility mode that renders on all clients. */ * 显式 'html'/'compat' 等其它值 = 兼容模式,所有客户端可显示。 */
rendering?: string; rendering?: string;
} }
export declare const Config: Schema<TelegramChannelConfig>; export declare const Config: Schema<TelegramChannelConfig>;
+1 -1
View File
@@ -9,7 +9,7 @@ export const Config = Schema.object({
allowAllUsers: Schema.boolean().default(false), allowAllUsers: Schema.boolean().default(false),
maxMessageLength: Schema.number().default(4096), maxMessageLength: Schema.number().default(4096),
pollingTimeoutSec: Schema.number().default(30), pollingTimeoutSec: Schema.number().default(30),
rendering: Schema.string().default('html'), rendering: Schema.string().default('rich'),
}); });
function resolveAllowedUserIds(config) { function resolveAllowedUserIds(config) {
if (config.allowedUserIds && config.allowedUserIds.length > 0) { if (config.allowedUserIds && config.allowedUserIds.length > 0) {
+266 -146
View File
@@ -6,7 +6,7 @@ import { dirname, join } from 'node:path'
import type { Context } from '@deepseek-ai/cordis' import type { Context } from '@deepseek-ai/cordis'
import type { Agent } from '@deepseek-ai/dsh-agent' import type { Agent } from '@deepseek-ai/dsh-agent'
import { createUserMessage } from '@deepseek-ai/dsh-llm/message' 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 { SessionId, type SessionEvent } from '@deepseek-ai/dsh-session/types'
import { isAuthorized } from './auth.js' import { isAuthorized } from './auth.js'
import { resolveApiProxy } from './apiproxy.js' import { resolveApiProxy } from './apiproxy.js'
@@ -28,7 +28,7 @@ import {
type TelegramUpdate, type TelegramUpdate,
} from './client.js' } from './client.js'
import { MSG, LAST_CB, parseCommand } from './commands.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 { splitRichMarkdown } from './rich-format.js'
import { formatLastTurn, loadLastTurn } from './history.js' import { formatLastTurn, loadLastTurn } from './history.js'
import { describeAgent, displayLabel, workspaceName } from './label.js' import { describeAgent, displayLabel, workspaceName } from './label.js'
@@ -97,6 +97,12 @@ interface TgPendingAsk {
questions: TgQuestionItem[] questions: TgQuestionItem[]
ask: TgAnswerHandle ask: TgAnswerHandle
at: number at: number
/** question index → option indices chosen via inline-keyboard taps. */
kbSel: Map<number, Set<number>>
/** question index → answer parsed from a text reply (letters or custom). */
textAnswers: Map<number, { id?: string; selected: string[]; custom?: string }>
/** questions the user chose to answer with free text (✏️ button): the next text reply fills them. */
textWanted: Set<number>
} }
/** chatId → pending approval request mirrored to Telegram. */ /** chatId → pending approval request mirrored to Telegram. */
@@ -105,15 +111,6 @@ interface TgPendingApproval {
ask: TgAnswerHandle 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<unknown>
__tgHooked?: boolean
__tgRealAsk?: (request: unknown) => Promise<unknown>
}
}
/** Loose view of the approval request passed through the `approval/request` waterfall hook. */ /** Loose view of the approval request passed through the `approval/request` waterfall hook. */
interface ApprovalRequestLike { interface ApprovalRequestLike {
agent?: Agent agent?: Agent
@@ -194,7 +191,11 @@ const MODEL_CB = 'mdl:'
/** Use eff: (not me:) — short prefix, no collision with other callbacks. */ /** Use eff: (not me:) — short prefix, no collision with other callbacks. */
const EFFORT_CB = 'eff:' const EFFORT_CB = 'eff:'
const BACK_MODEL_CB = 'mb' const BACK_MODEL_CB = 'mb'
/** ask:{questionIndex}:{optionIndex|ok} — inline-keyboard taps for ask_user_question. */
const ASK_CB = 'ask:'
const MAX_BUTTONS = 40 const MAX_BUTTONS = 40
/** Marker stored on UserQuestionService.prototype for the ask interceptor. */
const kTgAskWrapped = Symbol.for('dsh-telegram-channel.askWrapped')
function lastContextKeyboard(): InlineKeyboardMarkup { function lastContextKeyboard(): InlineKeyboardMarkup {
return { return {
@@ -204,11 +205,28 @@ function lastContextKeyboard(): InlineKeyboardMarkup {
const defaultSleep = (ms: number) => new Promise<void>((resolve) => setTimeout(resolve, ms)) const defaultSleep = (ms: number) => new Promise<void>((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 { function contentToText(content: readonly ContentBlock[]): string {
return content const parts: string[] = []
.filter((block): block is TextBlock => block.type === 'text') for (const block of content) {
.map((block) => block.text) if (block.type === 'text' && typeof block.text === 'string') {
.join('') 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 { function isRichUnsupportedError(err: unknown): boolean {
@@ -252,8 +270,9 @@ export class TelegramBridge {
private hookTimer: ReturnType<typeof setTimeout> | undefined private hookTimer: ReturnType<typeof setTimeout> | undefined
private readonly pendingApprovalsTG = new Map<string, TgPendingApproval>() private readonly pendingApprovalsTG = new Map<string, TgPendingApproval>()
private disposeApprovalHook: (() => void) | undefined private disposeApprovalHook: (() => void) | undefined
/** sessionId → thinking indicator state (one notice per reasoning phase) */ /** Prototype-ask interceptor state (see installAskInterceptor). */
private readonly thinkingSessions = new Map<string, boolean>() private askInstalled = false
private askRestore: (() => void) | undefined
/** callId → tool name (tool/result failure notices) */ /** callId → tool name (tool/result failure notices) */
private readonly callNames = new Map<string, string>() private readonly callNames = new Map<string, string>()
/** sessionId → latest todo snapshot (/mission) */ /** sessionId → latest todo snapshot (/mission) */
@@ -295,7 +314,12 @@ export class TelegramBridge {
// Restore persisted bindings (survive hot reload / restart) and install // Restore persisted bindings (survive hot reload / restart) and install
// the TG answering hooks (ask_user dual-path + approval dual-path). // the TG answering hooks (ask_user dual-path + approval dual-path).
this.loadBindings() this.loadBindings()
this.hookUserQuestions() // 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.disposeApprovalHook = (this.ctx as unknown as { this.disposeApprovalHook = (this.ctx as unknown as {
on: (event: string, handler: (req: ApprovalRequestLike, next: () => Promise<string>) => Promise<string>) => () => void on: (event: string, handler: (req: ApprovalRequestLike, next: () => Promise<string>) => Promise<string>) => () => void
@@ -343,7 +367,6 @@ export class TelegramBridge {
this.noticeQueue.clear() this.noticeQueue.clear()
this.pendingAsks.clear() this.pendingAsks.clear()
this.pendingApprovalsTG.clear() this.pendingApprovalsTG.clear()
this.thinkingSessions.clear()
this.callNames.clear() this.callNames.clear()
this.lastTodos.clear() this.lastTodos.clear()
for (const controller of this.compactAborts.values()) controller.abort() for (const controller of this.compactAborts.values()) controller.abort()
@@ -355,6 +378,9 @@ export class TelegramBridge {
clearTimeout(this.hookTimer) clearTimeout(this.hookTimer)
this.hookTimer = undefined this.hookTimer = undefined
} }
this.askRestore?.()
this.askRestore = undefined
this.askInstalled = false
this.disposeApprovalHook?.() this.disposeApprovalHook?.()
this.disposeApprovalHook = undefined this.disposeApprovalHook = undefined
if (this.pollPromise) { if (this.pollPromise) {
@@ -488,6 +514,10 @@ export class TelegramBridge {
const data = cq.data ?? '' const data = cq.data ?? ''
const picker = this.pickers.get(String(chatId)) const picker = this.pickers.get(String(chatId))
if (data.startsWith(ASK_CB)) {
await this.handleAskCallback(chatId, cq)
return
}
if (data === LAST_CB) { if (data === LAST_CB) {
await this.client.answerCallbackQuery(cq.id) await this.client.answerCallbackQuery(cq.id)
await this.sendLastTurn(chatId) await this.sendLastTurn(chatId)
@@ -1255,6 +1285,7 @@ export class TelegramBridge {
} }
private async onSessionEvent(session: SessionLike, event: SessionEvent): Promise<void> { private async onSessionEvent(session: SessionLike, event: SessionEvent): Promise<void> {
if (!this.askInstalled) this.installAskInterceptor()
const id = String(session.id) const id = String(session.id)
const targets = [...this.bindings.values()].filter((b) => b.sessionId === id) const targets = [...this.bindings.values()].filter((b) => b.sessionId === id)
if (targets.length === 0) return if (targets.length === 0) return
@@ -1269,7 +1300,6 @@ export class TelegramBridge {
if (event.type === 'turn/end') { if (event.type === 'turn/end') {
this.busySessions.delete(id) this.busySessions.delete(id)
this.thinkingSessions.delete(id)
for (const b of targets) this.stopTypingHeartbeat(b.chatId) for (const b of targets) this.stopTypingHeartbeat(b.chatId)
// Abnormal endings (API failure / abort / crash-recovery) must be // Abnormal endings (API failure / abort / crash-recovery) must be
// visible on the phone — previously they ended in silence. // visible on the phone — previously they ended in silence.
@@ -1294,7 +1324,6 @@ export class TelegramBridge {
// Progress summary; ask_user is owned by the provider hook (avoids duplicates). // Progress summary; ask_user is owned by the provider hook (avoids duplicates).
const name = event.data?.name ?? 'tool' const name = event.data?.name ?? 'tool'
if (name === 'ask_user_question') return if (name === 'ask_user_question') return
this.thinkingSessions.set(id, false)
if (event.data?.callId !== undefined) { if (event.data?.callId !== undefined) {
if (this.callNames.size > 400) { if (this.callNames.size > 400) {
const first = this.callNames.keys().next().value const first = this.callNames.keys().next().value
@@ -1312,26 +1341,11 @@ export class TelegramBridge {
if (event.type === 'assistant/message') { if (event.type === 'assistant/message') {
const text = contentToText(event.data.message.content) const text = contentToText(event.data.message.content)
this.thinkingSessions.set(id, false)
if (!text) return if (!text) return
await Promise.all(targets.map((b) => this.deliver(b.chatId, text))) await Promise.all(targets.map((b) => this.deliver(b.chatId, text)))
return 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') { if (event.type === 'tool/result') {
// Only failures — successful results are too noisy for the phone. // 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 } } const data = event.data as { error?: { name?: string; code?: string }; message?: { tool_use_id?: string; callId?: string } }
@@ -1471,17 +1485,28 @@ export class TelegramBridge {
} }
/** /**
* Send a notice; a leading `> ` keeps the quote look via a plain-text `> ` * Send a notice; a leading `> ` marks the body for quote rendering as a real
* prefix. NB: NOT `<blockquote>` HTML — older Telegram clients cannot render * `<blockquote>` (parse_mode HTML) — every current Telegram client draws the
* the blockquote entity at all and show the whole message as “not supported”. * accent-colored left bar. Old-client compatibility is no longer required,
* Falls back to plain text when the send fails — a notice is never lost. * but a failed HTML send still falls back to plain text — a notice is never lost.
*/ */
private async deliverNotice(chatId: number, text: string): Promise<void> { private async deliverNotice(chatId: number, text: string): Promise<void> {
const quoted = text.startsWith('> ') const quoted = text.startsWith('> ')
const body = quoted ? text.slice(2) : text 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 { try {
await this.withRetry(() => this.client.sendMessage(chatId, quoted ? `> ${body}` : body)) await this.withRetry(() => this.client.sendMessage(chatId, body))
return
} catch (err) { } catch (err) {
this.ctx.logger.error(this.redact(err)) this.ctx.logger.error(this.redact(err))
} }
@@ -1518,86 +1543,65 @@ export class TelegramBridge {
this.heartbeats.clear() 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 { private installAskInterceptor(): void {
const ctx = this.ctx as Context & { if (this.askInstalled) return
get?: (name: string, strict?: boolean) => unknown let svc: { constructor: { prototype: Record<string | symbol, unknown> } } | undefined
userQuestions?: UserQuestionsServiceLike
}
if (typeof ctx.get === 'function') {
try {
const service = ctx.get('userQuestions') as UserQuestionsServiceLike | undefined
if (service) return service
} catch {
// Continue to the plain-object fallback used by tests and simple hosts.
}
}
try { try {
return Object.prototype.hasOwnProperty.call(ctx, 'userQuestions') svc = (this.ctx as unknown as { get: (name: string) => unknown }).get('userQuestions') as
? ctx.userQuestions | { constructor: { prototype: Record<string | symbol, unknown> } }
: undefined | undefined
} catch { } catch {
return undefined 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
* Wrap the UI provider's ask() so Telegram gets a parallel answer path. if (prev?.bridge === this) {
* `Promise.race` decides; the UI path is untouched. The TG promise NEVER this.askInstalled = true
* settles when there is no bound chat — race would kill the UI's window this.askRestore = prev.restore
* 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<unknown> {
const tg = self.registerTgAsk(request)
let gui: Promise<unknown>
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<unknown>
provider.__tgRealAsk = realAsk
const self = this
provider.ask = function (request: unknown): Promise<unknown> {
const tg = self.registerTgAsk(request)
let gui: Promise<unknown>
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 return
} }
this.hookTimer = setTimeout(() => this.hookUserQuestions(attempt + 1), 2000) prev?.restore() // stale wrapper from a disposed bridge instance
const orig = proto.ask as (this: unknown, request: unknown) => Promise<unknown>
const bridge = this
const wrapped = function (this: unknown, request: unknown): Promise<unknown> {
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
}
private aroundUserQuestionAsk(
service: unknown,
request: unknown,
orig: (this: unknown, request: unknown) => Promise<unknown>,
): Promise<unknown> {
const tg = this.registerTgAsk(request)
let gui: Promise<unknown>
try {
gui = Promise.resolve().then(() => orig.call(service, request))
} catch (err) {
tg.reject(err as Error)
throw err
}
void gui.then(() => this.settleGuiSide(request), () => this.settleGuiSide(request))
return Promise.race([tg.promise, gui])
} }
private registerTgAsk(request: unknown): { promise: Promise<unknown>; reject: (err: Error) => void } { private registerTgAsk(request: unknown): { promise: Promise<unknown>; reject: (err: Error) => void } {
@@ -1626,16 +1630,91 @@ export class TelegramBridge {
let asked = false let asked = false
for (const [chatId, binding] of this.bindings) { for (const [chatId, binding] of this.bindings) {
if (binding.sessionId !== sessionId) continue if (binding.sessionId !== sessionId) continue
this.pendingAsks.set(String(chatId), { req: request, sessionId, questions, ask, at: Date.now() }) const pending: TgPendingAsk = {
this.enqueueNotice(Number(chatId), this.formatAskPending(questions)) req: request,
sessionId,
questions,
ask,
at: Date.now(),
kbSel: new Map(),
textAnswers: new Map(),
textWanted: new Set(),
}
this.pendingAsks.set(String(chatId), pending)
void this.deliverAskPrompt(Number(chatId), pending)
asked = true asked = true
} }
if (asked) return { promise, reject: rejectFn }
} }
// No bound TG chat: never settle — the UI remains the only answer path. // No bound TG chat: never settle — the UI remains the only answer path.
return { promise, reject: rejectFn } 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<void> {
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}`,
}]),
// Standing third choice: free-text answer, always available.
[{ text: '✏️ 输入自定义答案', callback_data: `${ASK_CB}${i}:custom` }],
...(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 { private settleGuiSide(request: unknown): void {
for (const [chatId, p] of [...this.pendingAsks]) { for (const [chatId, p] of [...this.pendingAsks]) {
if (p.req !== request) continue if (p.req !== request) continue
@@ -1646,24 +1725,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<void> { private async handleTgAnswer(chatId: number, pending: TgPendingAsk, text: string): Promise<void> {
const { questions, ask } = pending const { questions } = pending
let parts: Array<[number | null, string]> let parts: Array<[number | null, string]>
if (questions.length === 1) { if (questions.length === 1) {
parts = [[0, text.trim()]] parts = [[0, text.trim()]]
@@ -1676,11 +1739,16 @@ export class TelegramBridge {
else parts.push([null, line]) else parts.push([null, line])
} }
if (parts.some(([idx]) => idx === null)) { if (parts.some(([idx]) => idx === null)) {
await this.enqueueNotice(chatId, `共 ${questions.length} 个问题,请每行用 1:/2: 前缀分别作答`) // ✏️ free-text route: a question awaiting text absorbs the whole message.
return const target = [...pending.textWanted].sort((a, b) => a - b)[0]
if (target !== undefined) {
parts = [[target, text.trim()]]
} else {
await this.enqueueNotice(chatId, `共 ${questions.length} 个问题,请每行用 1:/2: 前缀分别作答`)
return
}
} }
} }
const answers: Array<{ id?: string; selected: string[]; custom?: string }> = []
for (const [idx, raw] of parts) { for (const [idx, raw] of parts) {
if (idx === null || idx < 0 || idx >= questions.length) { if (idx === null || idx < 0 || idx >= questions.length) {
await this.enqueueNotice(chatId, `题号 ${(idx ?? 0) + 1} 不存在(共 ${questions.length} 题)`) await this.enqueueNotice(chatId, `题号 ${(idx ?? 0) + 1} 不存在(共 ${questions.length} 题)`)
@@ -1703,17 +1771,69 @@ export class TelegramBridge {
} }
selected.push(opts[i]!.label) selected.push(opts[i]!.label)
} }
answers.push({ id: q.id, selected }) pending.textAnswers.set(idx, { id: q.id, selected })
} else { } else {
answers.push({ id: q.id, selected: [], custom: cleaned }) pending.textAnswers.set(idx, { id: q.id, selected: [], custom: cleaned })
}
pending.textWanted.delete(idx)
}
// 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<void> {
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 === 'custom') {
// Free-text path: discard button selections for this question and wait
// for the user's next text message (routed via textWanted).
pending.kbSel.delete(qi)
pending.textWanted.add(qi)
await this.client.answerCallbackQuery(cq.id, '请直接发送你的答案文本')
return
}
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<number>()
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}`)
} }
} }
ask.answered = true if (!this.finalizeAskIfComplete(chatId, pending) && action === 'ok') {
for (const [cid, p] of this.pendingAsks) { await this.client.answerCallbackQuery(cq.id, '已提交本题选择,请继续作答其他问题')
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 ── // ── approval: TG answering via dual-path race on the request waterfall ──
+8 -2
View File
@@ -153,12 +153,17 @@ export class TelegramClient implements TelegramClientLike {
return message.split(this.token).join('***') return message.split(this.token).join('***')
} }
private async call<T>(method: string, body?: Record<string, unknown>): Promise<T> { private async call<T>(method: string, body?: Record<string, unknown>, timeoutMs = 15_000): Promise<T> {
const url = `${this.baseUrl}/bot${this.token}/${method}` const url = `${this.baseUrl}/bot${this.token}/${method}`
try { try {
const init: RequestInit = { const init: RequestInit = {
method: 'POST', method: 'POST',
headers: { 'Content-Type': 'application/json' }, 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) { if (body !== undefined) {
init.body = JSON.stringify(body) init.body = JSON.stringify(body)
@@ -187,7 +192,8 @@ export class TelegramClient implements TelegramClientLike {
if (offset !== undefined) { if (offset !== undefined) {
body.offset = offset body.offset = offset
} }
return this.call<TelegramUpdate[]>('getUpdates', body) // Long-poll: client-side abort must outlive the server-side hold.
return this.call<TelegramUpdate[]>('getUpdates', body, this.pollingTimeoutSec * 1000 + 10_000)
} }
async sendMessage( async sendMessage(
+23 -1
View File
@@ -14,6 +14,28 @@ function inlineToHtml(text: string): string {
return html return html
} }
/** Convert `> …` line groups into `<blockquote>` (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(`<blockquote>${inlineToHtml(quote.join('\n'))}</blockquote>`)
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 { export function markdownToHtml(text: string): string {
const parts = text.split(/```[\w-]*/) const parts = text.split(/```[\w-]*/)
if (parts.length % 2 === 0) return inlineToHtml(text) 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$/, '') const code = parts[i]!.replace(/^\n/, '').replace(/\n$/, '')
html += `<pre>${escapeHtml(code)}</pre>` html += `<pre>${escapeHtml(code)}</pre>`
} else { } else {
html += inlineToHtml(parts[i]!) html += blocksToHtml(parts[i]!)
} }
} }
return html return html
+3 -3
View File
@@ -12,8 +12,8 @@ export interface TelegramChannelConfig {
allowAllUsers?: boolean allowAllUsers?: boolean
maxMessageLength?: number maxMessageLength?: number
pollingTimeoutSec?: number pollingTimeoutSec?: number
/** 'rich' = Telegram native Rich Messages (needs a recent client); /** 默认 'rich' = Telegram native Rich Messages(需较新客户端);
* anything else (default) = HTML compatibility mode that renders on all clients. */ * 显式 'html'/'compat' 等其它值 = 兼容模式,所有客户端可显示。 */
rendering?: string rendering?: string
} }
@@ -23,7 +23,7 @@ export const Config: Schema<TelegramChannelConfig> = Schema.object({
allowAllUsers: Schema.boolean().default(false), allowAllUsers: Schema.boolean().default(false),
maxMessageLength: Schema.number().default(4096), maxMessageLength: Schema.number().default(4096),
pollingTimeoutSec: Schema.number().default(30), pollingTimeoutSec: Schema.number().default(30),
rendering: Schema.string().default('html'), rendering: Schema.string().default('rich'),
}) })
function resolveAllowedUserIds(config: TelegramChannelConfig): number[] { function resolveAllowedUserIds(config: TelegramChannelConfig): number[] {
+7
View File
@@ -7,6 +7,13 @@ import { TelegramBridge } from '../src/bridge.ts'
import type { InlineKeyboardMarkup, TelegramClientLike, TelegramUpdate } from '../src/client.ts' import type { InlineKeyboardMarkup, TelegramClientLike, TelegramUpdate } from '../src/client.ts'
import { BIND_CB_PREFIX, LAST_CB, MSG } from '../src/commands.ts' import { BIND_CB_PREFIX, LAST_CB, MSG } from '../src/commands.ts'
// 防呆(2026-09-14 事故):binding 落盘路径 = DSH_HOME/telegram-channel-bindings.json。
// 少数用例直接构造 bridge 并 bind,未设 DSH_TELEGRAM_BINDINGS_FILE,就把 fixture
// (chatId 10 / live-img-dl)写进了真实文件、覆盖掉用户真实绑定。
// 这里把 DSH_HOME 指向临时目录兜底:显式 override 的用例仍走自己的临时文件,
// 漏设的用例只会写进临时 HOME。
process.env.DSH_HOME = '/tmp/dsh-tg-test-home'
type SentMessage = { type SentMessage = {
chatId: number chatId: number
text: string text: string