fix(ask_user): dsh 0.1.5 走 user-questions/request waterfall

0.1.5 移除了 userQuestions.provider,改为 waterfall 事件注册(与 approval/request 同形);
同时把 2026-09-12 只在编译产物上做过的修复回植到 src,避免源码/产物漂移。
This commit is contained in:
dsh-agent
2026-09-14 19:03:08 +08:00
parent f3f6a3ebaf
commit 9f03c6c748
3 changed files with 45 additions and 111 deletions
+7 -5
View File
@@ -43,6 +43,8 @@ export declare class TelegramBridge {
private hookTimer; private hookTimer;
private readonly pendingApprovalsTG; private readonly pendingApprovalsTG;
private disposeApprovalHook; private disposeApprovalHook;
/** dsh 0.1.5 起 ask 走 'user-questions/request' waterfall(不再有 userQuestions.provider)。 */
private disposeUserQuestionHook;
/** sessionId → thinking indicator state (one notice per reasoning phase) */ /** sessionId → thinking indicator state (one notice per reasoning phase) */
private readonly thinkingSessions; private readonly thinkingSessions;
/** callId → tool name (tool/result failure notices) */ /** callId → tool name (tool/result failure notices) */
@@ -130,12 +132,12 @@ export declare class TelegramBridge {
private stopAllHeartbeats; private stopAllHeartbeats;
private userQuestions; private userQuestions;
/** /**
* Wrap the UI provider's ask() so Telegram gets a parallel answer path. * dsh 0.1.5 移除了 `userQuestions.provider`:ask 现在走 'user-questions/request'
* `Promise.race` decides; the UI path is untouched. The TG promise NEVER * waterfall(与 approval/request 同形),启动时注册一次即可,无需轮询 provider。
* settles when there is no bound chat — race would kill the UI's window * 让 TG 答案与 next()(UI 转发)赛跑;未绑定聊天时 TG promise 永不 settle,
* with that early rejection. * 于是 race 完全跟随 next()。
*/ */
private hookUserQuestions; private onUserQuestionRequest;
private registerTgAsk; private registerTgAsk;
private settleGuiSide; private settleGuiSide;
private formatAskPending; private formatAskPending;
+14 -50
View File
@@ -111,6 +111,8 @@ export class TelegramBridge {
hookTimer; hookTimer;
pendingApprovalsTG = new Map(); pendingApprovalsTG = new Map();
disposeApprovalHook; disposeApprovalHook;
/** dsh 0.1.5 起 ask 走 'user-questions/request' waterfall(不再有 userQuestions.provider)。 */
disposeUserQuestionHook;
/** sessionId → thinking indicator state (one notice per reasoning phase) */ /** sessionId → thinking indicator state (one notice per reasoning phase) */
thinkingSessions = new Map(); thinkingSessions = new Map();
/** callId → tool name (tool/result failure notices) */ /** callId → tool name (tool/result failure notices) */
@@ -151,7 +153,8 @@ 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(); this.disposeUserQuestionHook?.();
this.disposeUserQuestionHook = this.ctx.on('user-questions/request', (req, next) => this.onUserQuestionRequest(req, next));
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([
@@ -210,6 +213,8 @@ export class TelegramBridge {
clearTimeout(this.hookTimer); clearTimeout(this.hookTimer);
this.hookTimer = undefined; this.hookTimer = undefined;
} }
this.disposeUserQuestionHook?.();
this.disposeUserQuestionHook = undefined;
this.disposeApprovalHook?.(); this.disposeApprovalHook?.();
this.disposeApprovalHook = undefined; this.disposeApprovalHook = undefined;
if (this.pollPromise) { if (this.pollPromise) {
@@ -1348,64 +1353,23 @@ export class TelegramBridge {
return this.ctx.userQuestions; return this.ctx.userQuestions;
} }
/** /**
* Wrap the UI provider's ask() so Telegram gets a parallel answer path. * dsh 0.1.5 移除了 `userQuestions.provider`:ask 现在走 'user-questions/request'
* `Promise.race` decides; the UI path is untouched. The TG promise NEVER * waterfall(与 approval/request 同形),启动时注册一次即可,无需轮询 provider。
* settles when there is no bound chat — race would kill the UI's window * 让 TG 答案与 next()(UI 转发)赛跑;未绑定聊天时 TG promise 永不 settle,
* with that early rejection. * 于是 race 完全跟随 next()。
*/ */
hookUserQuestions(attempt = 0) { onUserQuestionRequest(req, next) {
try { const tg = this.registerTgAsk(req);
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; let gui;
try { try {
gui = realAsk(request); gui = Promise.resolve().then(next);
} }
catch (err) { catch (err) {
tg.reject(err); tg.reject(err);
throw err; throw err;
} }
void gui.then(() => self.settleGuiSide(request), () => self.settleGuiSide(request)); void gui.then(() => this.settleGuiSide(req), () => this.settleGuiSide(req));
return Promise.race([tg.promise, gui]); return Promise.race([tg.promise, gui]);
};
return;
}
const realAsk = provider.ask.bind(provider);
provider.__tgRealAsk = realAsk;
const self = this;
provider.ask = function (request) {
const tg = self.registerTgAsk(request);
let gui;
try {
gui = realAsk(request);
}
catch (err) {
tg.reject(err);
throw err;
}
void gui.then(() => self.settleGuiSide(request), () => self.settleGuiSide(request));
return Promise.race([tg.promise, gui]);
};
provider.__tgHooked = true;
this.ctx.logger.info('dsh-telegram-channel: ask_user TG answering hook installed');
return;
}
}
catch (err) {
if (attempt === 0)
this.ctx.logger.warn(`dsh-telegram-channel: userQuestions hook deferred: ${this.redact(err)}`);
}
if (attempt >= 30) {
this.ctx.logger.warn('dsh-telegram-channel: user-questions provider never appeared; TG answering disabled');
return;
}
this.hookTimer = setTimeout(() => this.hookUserQuestions(attempt + 1), 2000);
} }
registerTgAsk(request) { registerTgAsk(request) {
let resolveFn = () => { }; let resolveFn = () => { };
+16 -48
View File
@@ -252,6 +252,8 @@ 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
/** dsh 0.1.5 起 ask 走 'user-questions/request' waterfall(不再有 userQuestions.provider)。 */
private disposeUserQuestionHook: (() => void) | undefined
/** sessionId → thinking indicator state (one notice per reasoning phase) */ /** sessionId → thinking indicator state (one notice per reasoning phase) */
private readonly thinkingSessions = new Map<string, boolean>() private readonly thinkingSessions = new Map<string, boolean>()
/** callId → tool name (tool/result failure notices) */ /** callId → tool name (tool/result failure notices) */
@@ -295,7 +297,10 @@ 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() this.disposeUserQuestionHook?.()
this.disposeUserQuestionHook = (this.ctx as unknown as {
on: (event: string, handler: (req: unknown, next: () => Promise<unknown>) => Promise<unknown>) => () => void
}).on('user-questions/request', (req, next) => this.onUserQuestionRequest(req, next))
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
@@ -355,6 +360,8 @@ export class TelegramBridge {
clearTimeout(this.hookTimer) clearTimeout(this.hookTimer)
this.hookTimer = undefined this.hookTimer = undefined
} }
this.disposeUserQuestionHook?.()
this.disposeUserQuestionHook = undefined
this.disposeApprovalHook?.() this.disposeApprovalHook?.()
this.disposeApprovalHook = undefined this.disposeApprovalHook = undefined
if (this.pollPromise) { if (this.pollPromise) {
@@ -1525,62 +1532,23 @@ export class TelegramBridge {
} }
/** /**
* Wrap the UI provider's ask() so Telegram gets a parallel answer path. * dsh 0.1.5 移除了 `userQuestions.provider`:ask 现在走 'user-questions/request'
* `Promise.race` decides; the UI path is untouched. The TG promise NEVER * waterfall(与 approval/request 同形),启动时注册一次即可,无需轮询 provider。
* settles when there is no bound chat — race would kill the UI's window * 让 TG 答案与 next()(UI 转发)赛跑;未绑定聊天时 TG promise 永不 settle,
* with that early rejection. * 于是 race 完全跟随 next()。
*/ */
private hookUserQuestions(attempt = 0): void { private onUserQuestionRequest(req: unknown, next: () => Promise<unknown>): Promise<unknown> {
try { const tg = this.registerTgAsk(req)
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> let gui: Promise<unknown>
try { try {
gui = realAsk(request) gui = Promise.resolve().then(next)
} catch (err) { } catch (err) {
tg.reject(err as Error) tg.reject(err as Error)
throw err throw err
} }
void gui.then(() => self.settleGuiSide(request), () => self.settleGuiSide(request)) void gui.then(() => this.settleGuiSide(req), () => this.settleGuiSide(req))
return Promise.race([tg.promise, gui]) 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
}
this.hookTimer = setTimeout(() => this.hookUserQuestions(attempt + 1), 2000)
}
private registerTgAsk(request: unknown): { promise: Promise<unknown>; reject: (err: Error) => void } { private registerTgAsk(request: unknown): { promise: Promise<unknown>; reject: (err: Error) => void } {
let resolveFn: (v: unknown) => void = () => {} let resolveFn: (v: unknown) => void = () => {}