fix(approval): prototype 拦截器替代 approval/request waterfall hook

与 ask_user 同一根因:'approval/request' waterfall 分发在 per-fiber events
注册表,本插件 ctx.on 监听收不到。改为包装 ApprovalService.prototype.
request(类跨实例/fiber 共享),TG 审批答案与原请求竞速,保留 dual-path
语义;stop 时恢复原型。已验证运行态拦截器标记就位。
This commit is contained in:
root
2026-09-17 11:47:48 +08:00
parent 566932e0ae
commit f49ac5d3c0
3 changed files with 113 additions and 23 deletions
+5 -2
View File
@@ -42,10 +42,12 @@ export declare class TelegramBridge {
private readonly pendingAsks; private readonly pendingAsks;
private hookTimer; private hookTimer;
private readonly pendingApprovalsTG; private readonly pendingApprovalsTG;
private disposeApprovalHook;
/** Prototype-ask interceptor state (see installAskInterceptor). */ /** Prototype-ask interceptor state (see installAskInterceptor). */
private askInstalled; private askInstalled;
private askRestore; private askRestore;
/** Prototype-approval interceptor state (see installApprovalInterceptor). */
private approvalInstalled;
private approvalRestore;
/** 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) */
@@ -140,7 +142,8 @@ export declare class TelegramBridge {
private handleTgAnswer; private handleTgAnswer;
/** Inline-keyboard tap on an ask message: toggle/record the choice, resolve when every question is answered. */ /** Inline-keyboard tap on an ask message: toggle/record the choice, resolve when every question is answered. */
private handleAskCallback; private handleAskCallback;
private onApprovalRequest; private installApprovalInterceptor;
private aroundApprovalRequest;
private handleTgApproval; private handleTgApproval;
private stopBound; private stopBound;
private sendMission; private sendMission;
+54 -9
View File
@@ -65,6 +65,8 @@ const ASK_CB = 'ask:';
const MAX_BUTTONS = 40; const MAX_BUTTONS = 40;
/** Marker stored on UserQuestionService.prototype for the ask interceptor. */ /** Marker stored on UserQuestionService.prototype for the ask interceptor. */
const kTgAskWrapped = Symbol.for('dsh-telegram-channel.askWrapped'); const kTgAskWrapped = Symbol.for('dsh-telegram-channel.askWrapped');
/** Marker stored on ApprovalService.prototype for the approval interceptor. */
const kTgApprovalWrapped = Symbol.for('dsh-telegram-channel.approvalWrapped');
function lastContextKeyboard() { function lastContextKeyboard() {
return { return {
inline_keyboard: [[{ text: '查看上次对话', callback_data: LAST_CB }]], inline_keyboard: [[{ text: '查看上次对话', callback_data: LAST_CB }]],
@@ -132,10 +134,12 @@ export class TelegramBridge {
pendingAsks = new Map(); pendingAsks = new Map();
hookTimer; hookTimer;
pendingApprovalsTG = new Map(); pendingApprovalsTG = new Map();
disposeApprovalHook;
/** Prototype-ask interceptor state (see installAskInterceptor). */ /** Prototype-ask interceptor state (see installAskInterceptor). */
askInstalled = false; askInstalled = false;
askRestore; askRestore;
/** Prototype-approval interceptor state (see installApprovalInterceptor). */
approvalInstalled = false;
approvalRestore;
/** 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) */
@@ -180,8 +184,7 @@ export class TelegramBridge {
// per-fiber events registry in dsh 0.1.5-rc.1), so the prototype wrap — // 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. // which sees EVERY ask regardless of fiber/instance — is the reliable seam.
this.installAskInterceptor(); this.installAskInterceptor();
this.disposeApprovalHook?.(); this.installApprovalInterceptor();
this.disposeApprovalHook = this.ctx.on('approval/request', (req, next) => this.onApprovalRequest(req, next));
void this.client.setMyCommands([ void this.client.setMyCommands([
{ command: 'start', description: '欢迎与用法' }, { command: 'start', description: '欢迎与用法' },
{ command: 'sessions', description: '按工作区列出并附着会话' }, { command: 'sessions', description: '按工作区列出并附着会话' },
@@ -240,8 +243,9 @@ export class TelegramBridge {
this.askRestore?.(); this.askRestore?.();
this.askRestore = undefined; this.askRestore = undefined;
this.askInstalled = false; this.askInstalled = false;
this.disposeApprovalHook?.(); this.approvalRestore?.();
this.disposeApprovalHook = undefined; this.approvalRestore = undefined;
this.approvalInstalled = false;
if (this.pollPromise) { if (this.pollPromise) {
await this.pollPromise.catch(() => { }); await this.pollPromise.catch(() => { });
this.pollPromise = undefined; this.pollPromise = undefined;
@@ -1118,6 +1122,8 @@ export class TelegramBridge {
async onSessionEvent(session, event) { async onSessionEvent(session, event) {
if (!this.askInstalled) if (!this.askInstalled)
this.installAskInterceptor(); this.installAskInterceptor();
if (!this.approvalInstalled)
this.installApprovalInterceptor();
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)
@@ -1661,8 +1667,47 @@ export class TelegramBridge {
await this.client.answerCallbackQuery(cq.id, '已提交本题选择,请继续作答其他问题'); await this.client.answerCallbackQuery(cq.id, '已提交本题选择,请继续作答其他问题');
} }
} }
// ── approval: TG answering via dual-path race on the request waterfall ── // ── approval: TG answering via a prototype interceptor ──
async onApprovalRequest(req, next) { //
// Same root cause and same cure as ask_user: the 'approval/request'
// waterfall dispatches on a per-fiber events registry that this plugin's
// ctx.on never sees (dsh 0.1.5-rc.1), so ApprovalService.prototype.request
// is wrapped instead — the class is shared across instances and fibers.
installApprovalInterceptor() {
if (this.approvalInstalled)
return;
let svc;
try {
svc = this.ctx.get('approval');
}
catch {
return; // service not ready yet — retried lazily from onSessionEvent
}
if (!svc)
return;
const proto = svc.constructor.prototype;
const prev = proto[kTgApprovalWrapped];
if (prev?.bridge === this) {
this.approvalInstalled = true;
this.approvalRestore = prev.restore;
return;
}
prev?.restore(); // stale wrapper from a disposed bridge instance
const orig = proto.request;
const bridge = this;
const wrapped = function (req) {
return bridge.aroundApprovalRequest(this, req, orig);
};
proto.request = wrapped;
const restore = () => {
if (proto.request === wrapped)
proto.request = orig;
};
proto[kTgApprovalWrapped] = { restore, bridge: this };
this.approvalRestore = restore;
this.approvalInstalled = true;
}
aroundApprovalRequest(service, req, orig) {
const sessionId = req?.agent?.id !== undefined ? String(req.agent.id) : undefined; const sessionId = req?.agent?.id !== undefined ? String(req.agent.id) : undefined;
const targets = sessionId !== undefined const targets = sessionId !== undefined
? [...this.bindings.values()].filter((b) => b.sessionId === sessionId).map((b) => String(b.chatId)) ? [...this.bindings.values()].filter((b) => b.sessionId === sessionId).map((b) => String(b.chatId))
@@ -1683,7 +1728,7 @@ export class TelegramBridge {
}; };
if (req?.signal?.aborted) { if (req?.signal?.aborted) {
rejectFn(new Error('approval withdrawn')); rejectFn(new Error('approval withdrawn'));
return next(); return orig.call(service, req);
} }
req?.signal?.addEventListener('abort', onAbort, { once: true }); req?.signal?.addEventListener('abort', onAbort, { once: true });
if (targets.length > 0) { if (targets.length > 0) {
@@ -1695,7 +1740,7 @@ export class TelegramBridge {
} }
} }
// No bound TG chat: promise never settles — the UI remains the only answer path. // No bound TG chat: promise never settles — the UI remains the only answer path.
const gui = next(); const gui = Promise.resolve().then(() => orig.call(service, req));
void gui.then(() => { void gui.then(() => {
for (const [chatId, p] of [...this.pendingApprovalsTG]) { for (const [chatId, p] of [...this.pendingApprovalsTG]) {
if (p.ask !== ask) if (p.ask !== ask)
+54 -12
View File
@@ -196,6 +196,8 @@ const ASK_CB = 'ask:'
const MAX_BUTTONS = 40 const MAX_BUTTONS = 40
/** Marker stored on UserQuestionService.prototype for the ask interceptor. */ /** Marker stored on UserQuestionService.prototype for the ask interceptor. */
const kTgAskWrapped = Symbol.for('dsh-telegram-channel.askWrapped') const kTgAskWrapped = Symbol.for('dsh-telegram-channel.askWrapped')
/** Marker stored on ApprovalService.prototype for the approval interceptor. */
const kTgApprovalWrapped = Symbol.for('dsh-telegram-channel.approvalWrapped')
function lastContextKeyboard(): InlineKeyboardMarkup { function lastContextKeyboard(): InlineKeyboardMarkup {
return { return {
@@ -269,10 +271,12 @@ export class TelegramBridge {
private readonly pendingAsks = new Map<string, TgPendingAsk>() private readonly pendingAsks = new Map<string, TgPendingAsk>()
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
/** Prototype-ask interceptor state (see installAskInterceptor). */ /** Prototype-ask interceptor state (see installAskInterceptor). */
private askInstalled = false private askInstalled = false
private askRestore: (() => void) | undefined private askRestore: (() => void) | undefined
/** Prototype-approval interceptor state (see installApprovalInterceptor). */
private approvalInstalled = false
private approvalRestore: (() => 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) */
@@ -320,10 +324,7 @@ export class TelegramBridge {
// per-fiber events registry in dsh 0.1.5-rc.1), so the prototype wrap — // 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. // which sees EVERY ask regardless of fiber/instance — is the reliable seam.
this.installAskInterceptor() this.installAskInterceptor()
this.disposeApprovalHook?.() this.installApprovalInterceptor()
this.disposeApprovalHook = (this.ctx as unknown as {
on: (event: string, handler: (req: ApprovalRequestLike, next: () => Promise<string>) => Promise<string>) => () => void
}).on('approval/request', (req, next) => this.onApprovalRequest(req, next))
void this.client.setMyCommands([ void this.client.setMyCommands([
{ command: 'start', description: '欢迎与用法' }, { command: 'start', description: '欢迎与用法' },
{ command: 'sessions', description: '按工作区列出并附着会话' }, { command: 'sessions', description: '按工作区列出并附着会话' },
@@ -381,8 +382,9 @@ export class TelegramBridge {
this.askRestore?.() this.askRestore?.()
this.askRestore = undefined this.askRestore = undefined
this.askInstalled = false this.askInstalled = false
this.disposeApprovalHook?.() this.approvalRestore?.()
this.disposeApprovalHook = undefined this.approvalRestore = undefined
this.approvalInstalled = false
if (this.pollPromise) { if (this.pollPromise) {
await this.pollPromise.catch(() => {}) await this.pollPromise.catch(() => {})
this.pollPromise = undefined this.pollPromise = undefined
@@ -1286,6 +1288,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() if (!this.askInstalled) this.installAskInterceptor()
if (!this.approvalInstalled) this.installApprovalInterceptor()
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
@@ -1836,11 +1839,50 @@ export class TelegramBridge {
} }
} }
// ── approval: TG answering via dual-path race on the request waterfall ── // ── approval: TG answering via a prototype interceptor ──
//
// Same root cause and same cure as ask_user: the 'approval/request'
// waterfall dispatches on a per-fiber events registry that this plugin's
// ctx.on never sees (dsh 0.1.5-rc.1), so ApprovalService.prototype.request
// is wrapped instead — the class is shared across instances and fibers.
private async onApprovalRequest( private installApprovalInterceptor(): void {
if (this.approvalInstalled) return
let svc: { constructor: { prototype: Record<string | symbol, unknown> } } | undefined
try {
svc = (this.ctx as unknown as { get: (name: string) => unknown }).get('approval') as
| { constructor: { prototype: Record<string | symbol, unknown> } }
| undefined
} catch {
return // service not ready yet — retried lazily from onSessionEvent
}
if (!svc) return
const proto = svc.constructor.prototype
const prev = proto[kTgApprovalWrapped] as { restore: () => void; bridge: unknown } | undefined
if (prev?.bridge === this) {
this.approvalInstalled = true
this.approvalRestore = prev.restore
return
}
prev?.restore() // stale wrapper from a disposed bridge instance
const orig = proto.request as (this: unknown, req: unknown) => Promise<string>
const bridge = this
const wrapped = function (this: unknown, req: unknown): Promise<string> {
return bridge.aroundApprovalRequest(this, req as ApprovalRequestLike, orig as (this: unknown, req: ApprovalRequestLike) => Promise<string>)
}
proto.request = wrapped
const restore = (): void => {
if (proto.request === wrapped) proto.request = orig
}
proto[kTgApprovalWrapped] = { restore, bridge: this }
this.approvalRestore = restore
this.approvalInstalled = true
}
private aroundApprovalRequest(
service: unknown,
req: ApprovalRequestLike, req: ApprovalRequestLike,
next: () => Promise<string>, orig: (this: unknown, req: ApprovalRequestLike) => Promise<string>,
): Promise<string> { ): Promise<string> {
const sessionId = req?.agent?.id !== undefined ? String(req.agent.id) : undefined const sessionId = req?.agent?.id !== undefined ? String(req.agent.id) : undefined
const targets = sessionId !== undefined const targets = sessionId !== undefined
@@ -1861,7 +1903,7 @@ export class TelegramBridge {
} }
if (req?.signal?.aborted) { if (req?.signal?.aborted) {
rejectFn(new Error('approval withdrawn')) rejectFn(new Error('approval withdrawn'))
return next() return orig.call(service, req)
} }
req?.signal?.addEventListener('abort', onAbort, { once: true }) req?.signal?.addEventListener('abort', onAbort, { once: true })
if (targets.length > 0) { if (targets.length > 0) {
@@ -1873,7 +1915,7 @@ export class TelegramBridge {
} }
} }
// No bound TG chat: promise never settles — the UI remains the only answer path. // No bound TG chat: promise never settles — the UI remains the only answer path.
const gui = next() const gui = Promise.resolve().then(() => orig.call(service, req))
void gui.then(() => { void gui.then(() => {
for (const [chatId, p] of [...this.pendingApprovalsTG]) { for (const [chatId, p] of [...this.pendingApprovalsTG]) {
if (p.ask !== ask) continue if (p.ask !== ask) continue