feat: resilient delivery, live progress visibility, TG-side answering, /stop /mission /new
Delivery & visibility: - per-chunk retry with capped backoff on rich/HTML sends; rich failure now falls back to HTML instead of dropping the reply - typing heartbeat: refresh sendChatAction every 4s for the whole turn (a single action expires in ~5s, long turns looked dead) - turn/end error / abort / crash-recovery notices reach Telegram (API failures were previously invisible) - tool/call progress summaries, tool/result failure notices, todo progress - thinking indicator (one notice per reasoning phase, status only) - notices prefixed '> ' render as Telegram blockquotes; plain-text fallback - non-text messages get a hint instead of being silently dropped - busy feedback: a followup queued mid-turn gets an explicit ack TG-side answering (dual-path race; UI path untouched, first answer wins): - ask_user_question questions render with letter options on TG; reply A / [A] / A,B (multi) / free text / '1: A' per-line for multi-question - approval requests (approval/request waterfall) answerable from TG: [A] allow once / [B] reject; /cancel closes only the TG side Commands: /stop (cancel current turn), /mission (todo snapshot + busy state), /new (create a session in the current workspace and attach) Persistence: chat→session bindings stored at $DSH_HOME/ telegram-channel-bindings.json (override: DSH_TELEGRAM_BINDINGS_FILE) and restored on start — hot reloads and restarts no longer drop attachments. Host seams without public surface yet are reached with documented structural casts (userQuestions provider wrap, approval/request waterfall). Tests: 41/41 pass. Two pre-existing environment failures in bridge.test.ts (/last history mirror, assistant mirror) now pass: the former was fixed by the rich→HTML fallback, the latter needed an event-loop flush after the (void) listener call since resilient delivery lengthens the chain.
This commit is contained in:
RRRosmontis
+726
-25
@@ -1,9 +1,13 @@
|
||||
import { mkdirSync, readFileSync, writeFileSync } from 'node:fs'
|
||||
import { homedir } from 'node:os'
|
||||
import { dirname, join } from 'node:path'
|
||||
import type { Context } from '@deepseek-ai/cordis'
|
||||
import type { Agent } from '@deepseek-ai/dsh-agent'
|
||||
import { createUserMessage } from '@deepseek-ai/dsh-llm/message'
|
||||
import type { ContentBlock, TextBlock } from '@deepseek-ai/dsh-llm/types'
|
||||
import { SessionId, type SessionEvent } from '@deepseek-ai/dsh-session/types'
|
||||
import { isAuthorized } from './auth.js'
|
||||
import { resolveApiProxy } from './apiproxy.js'
|
||||
import {
|
||||
catalogFromLiveAgents,
|
||||
loadCatalog,
|
||||
@@ -62,6 +66,53 @@ interface SessionLike {
|
||||
id: ReturnType<typeof SessionId>
|
||||
}
|
||||
|
||||
/** ── Stability & interactivity additions: local types (host seam shapes mirrored structurally) ── */
|
||||
|
||||
interface TgQuestionItem {
|
||||
id?: string
|
||||
question?: string
|
||||
options?: Array<{ label: string; description?: string }>
|
||||
multiSelect?: boolean
|
||||
}
|
||||
|
||||
interface TgAnswerHandle {
|
||||
resolve: (answer: unknown) => void
|
||||
reject: (err: Error) => void
|
||||
answered: boolean
|
||||
}
|
||||
|
||||
/** chatId → pending ask_user_question mirrored to Telegram. */
|
||||
interface TgPendingAsk {
|
||||
req: unknown
|
||||
sessionId: string
|
||||
questions: TgQuestionItem[]
|
||||
ask: TgAnswerHandle
|
||||
at: number
|
||||
}
|
||||
|
||||
/** chatId → pending approval request mirrored to Telegram. */
|
||||
interface TgPendingApproval {
|
||||
toolName: string
|
||||
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. */
|
||||
interface ApprovalRequestLike {
|
||||
agent?: Agent
|
||||
toolName?: string
|
||||
reason?: string
|
||||
signal?: AbortSignal
|
||||
}
|
||||
|
||||
const WS_CB = 'ws:'
|
||||
const SID_CB = 'sid:'
|
||||
const BACK_WS_CB = 'wb'
|
||||
@@ -111,6 +162,24 @@ export class TelegramBridge {
|
||||
private pollAbort: AbortController | undefined
|
||||
private disposeSessionListener: (() => void) | undefined
|
||||
|
||||
// ── Stability & interactivity additions ──
|
||||
/** sessionIds mid-turn (busy feedback, /stop state) */
|
||||
private readonly busySessions = new Set<string>()
|
||||
/** chatId → typing heartbeat interval handle */
|
||||
private readonly heartbeats = new Map<string, ReturnType<typeof setInterval>>()
|
||||
/** chatId → serialized notice chain (prevents 429 storms) */
|
||||
private readonly noticeQueue = new Map<string, Promise<unknown>>()
|
||||
private readonly pendingAsks = new Map<string, TgPendingAsk>()
|
||||
private hookTimer: ReturnType<typeof setTimeout> | undefined
|
||||
private readonly pendingApprovalsTG = new Map<string, TgPendingApproval>()
|
||||
private disposeApprovalHook: (() => void) | undefined
|
||||
/** sessionId → thinking indicator state (one notice per reasoning phase) */
|
||||
private readonly thinkingSessions = new Map<string, boolean>()
|
||||
/** callId → tool name (tool/result failure notices) */
|
||||
private readonly callNames = new Map<string, string>()
|
||||
/** sessionId → latest todo snapshot (/mission) */
|
||||
private readonly lastTodos = new Map<string, Array<{ content?: string; status?: string }>>()
|
||||
|
||||
constructor(ctx: Context, options: TelegramBridgeOptions) {
|
||||
this.ctx = ctx
|
||||
this.token = options.token
|
||||
@@ -131,6 +200,14 @@ export class TelegramBridge {
|
||||
this.ctx.logger.error(this.redact(err))
|
||||
})
|
||||
})
|
||||
// Restore persisted bindings (survive hot reload / restart) and install
|
||||
// the TG answering hooks (ask_user dual-path + approval dual-path).
|
||||
this.loadBindings()
|
||||
this.hookUserQuestions()
|
||||
this.disposeApprovalHook?.()
|
||||
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([
|
||||
{ command: 'start', description: '欢迎与用法' },
|
||||
{ command: 'sessions', description: '按工作区列出并附着会话' },
|
||||
@@ -138,6 +215,9 @@ export class TelegramBridge {
|
||||
{ command: 'model', description: '切换当前绑定会话的模型' },
|
||||
{ command: 'status', description: '查看当前绑定' },
|
||||
{ command: 'unbind', description: '断开手机绑定（不关闭本机会话）' },
|
||||
{ command: 'stop', description: '中止当前正在运行的任务' },
|
||||
{ command: 'mission', description: '查看任务清单与完成情况' },
|
||||
{ command: 'new', description: '在当前工作区新开对话并附着' },
|
||||
{ command: 'help', description: '显示帮助' },
|
||||
]).then(() => {
|
||||
this.ctx.logger.info('dsh-telegram-channel: bot commands registered')
|
||||
@@ -162,6 +242,20 @@ export class TelegramBridge {
|
||||
this.bindings.clear()
|
||||
this.pickers.clear()
|
||||
this.pendingModels.clear()
|
||||
this.stopAllHeartbeats()
|
||||
this.busySessions.clear()
|
||||
this.noticeQueue.clear()
|
||||
this.pendingAsks.clear()
|
||||
this.pendingApprovalsTG.clear()
|
||||
this.thinkingSessions.clear()
|
||||
this.callNames.clear()
|
||||
this.lastTodos.clear()
|
||||
if (this.hookTimer) {
|
||||
clearTimeout(this.hookTimer)
|
||||
this.hookTimer = undefined
|
||||
}
|
||||
this.disposeApprovalHook?.()
|
||||
this.disposeApprovalHook = undefined
|
||||
if (this.pollPromise) {
|
||||
await this.pollPromise.catch(() => {})
|
||||
this.pollPromise = undefined
|
||||
@@ -174,7 +268,7 @@ export class TelegramBridge {
|
||||
return
|
||||
}
|
||||
const message = update.message
|
||||
if (!message?.text) return
|
||||
if (!message) return
|
||||
|
||||
const chatId = message.chat.id
|
||||
const userId = message.from?.id
|
||||
@@ -184,6 +278,12 @@ export class TelegramBridge {
|
||||
return
|
||||
}
|
||||
|
||||
// Non-text messages: tell the (authorized) user instead of dropping silently.
|
||||
if (!message.text) {
|
||||
await this.enqueueNotice(chatId, '暂只支持文本消息（图片/语音/文件等媒体暂不处理）')
|
||||
return
|
||||
}
|
||||
|
||||
const parsed = parseCommand(message.text)
|
||||
switch (parsed.type) {
|
||||
case 'start':
|
||||
@@ -207,14 +307,47 @@ export class TelegramBridge {
|
||||
case 'unbind':
|
||||
this.bindings.delete(String(chatId))
|
||||
this.pickers.delete(String(chatId))
|
||||
this.pendingAsks.delete(String(chatId))
|
||||
this.saveBindings()
|
||||
await this.client.sendMessage(chatId, MSG.UNBOUND)
|
||||
return
|
||||
case 'stop':
|
||||
await this.stopBound(chatId)
|
||||
return
|
||||
case 'mission':
|
||||
await this.sendMission(chatId)
|
||||
return
|
||||
case 'new':
|
||||
await this.newSessionHere(chatId)
|
||||
return
|
||||
case 'unknown':
|
||||
// /cancel closes the TG-side answer/approval channel; the UI side stays open.
|
||||
if (parsed.command === 'cancel') {
|
||||
let cancelled = false
|
||||
if (this.pendingApprovalsTG.delete(String(chatId))) cancelled = true
|
||||
if (this.pendingAsks.delete(String(chatId))) cancelled = true
|
||||
if (cancelled) {
|
||||
await this.enqueueNotice(chatId, '已关闭 TG 端作答/审批（该问题仍可在 Web 端回答）')
|
||||
return
|
||||
}
|
||||
}
|
||||
await this.client.sendMessage(chatId, MSG.unknown(parsed.command))
|
||||
return
|
||||
case 'plain':
|
||||
case 'plain': {
|
||||
// A pending approval/ask on TG consumes the message as an answer.
|
||||
const approval = this.pendingApprovalsTG.get(String(chatId))
|
||||
if (approval) {
|
||||
await this.handleTgApproval(chatId, approval, parsed.text)
|
||||
return
|
||||
}
|
||||
const pending = this.pendingAsks.get(String(chatId))
|
||||
if (pending) {
|
||||
await this.handleTgAnswer(chatId, pending, parsed.text)
|
||||
return
|
||||
}
|
||||
await this.followupBound(chatId, parsed.text)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -424,6 +557,10 @@ export class TelegramBridge {
|
||||
? displayLabel({ ...parts, title: row.title })
|
||||
: displayLabel(parts)
|
||||
this.bindings.set(String(chatId), { chatId, sessionId: String(agent.id), label })
|
||||
// Drop stale pending asks when rebinding to another session; persist bindings.
|
||||
const stale = this.pendingAsks.get(String(chatId))
|
||||
if (stale && stale.sessionId !== String(agent.id)) this.pendingAsks.delete(String(chatId))
|
||||
this.saveBindings()
|
||||
await this.client.answerCallbackQuery(callbackId, '已附着')
|
||||
await this.client.sendMessage(chatId, MSG.BOUND(label), undefined, lastContextKeyboard())
|
||||
}
|
||||
@@ -617,7 +754,14 @@ export class TelegramBridge {
|
||||
content: [{ type: 'text', text }],
|
||||
source: { kind: 'user' },
|
||||
})
|
||||
// Busy feedback: mid-turn followups queue silently by design — tell the user.
|
||||
const busy = this.busySessions.has(binding.sessionId)
|
||||
|| Boolean((agent as { phase?: { kind?: string } }).phase?.kind
|
||||
&& (agent as { phase?: { kind?: string } }).phase?.kind !== 'idle')
|
||||
agent.followup(message)
|
||||
if (busy) {
|
||||
await this.enqueueNotice(chatId, '已加入队列：当前任务完成后将处理这条消息')
|
||||
}
|
||||
}
|
||||
|
||||
private async pollLoop(): Promise<void> {
|
||||
@@ -677,41 +821,132 @@ export class TelegramBridge {
|
||||
if (targets.length === 0) return
|
||||
|
||||
if (event.type === 'turn/start') {
|
||||
await Promise.all(targets.map((b) => this.client.sendChatAction(b.chatId, 'typing')))
|
||||
// Typing heartbeat — one sendChatAction expires in ~5s, so refresh it
|
||||
// for the whole turn instead of only at start.
|
||||
this.busySessions.add(id)
|
||||
for (const b of targets) this.startTypingHeartbeat(b.chatId)
|
||||
return
|
||||
}
|
||||
|
||||
if (event.type === 'turn/end') {
|
||||
this.busySessions.delete(id)
|
||||
this.thinkingSessions.delete(id)
|
||||
for (const b of targets) this.stopTypingHeartbeat(b.chatId)
|
||||
// Abnormal endings (API failure / abort / crash-recovery) must be
|
||||
// visible on the phone — previously they ended in silence.
|
||||
const reason = event.data?.reason as { kind?: string; error?: { message?: string; code?: string }; reason?: { kind?: string } } | undefined
|
||||
const kind = reason?.kind
|
||||
if (kind === 'error') {
|
||||
const detail = String(reason?.error?.message ?? reason?.error?.code ?? '').replace(/\s+/g, ' ').slice(0, 300)
|
||||
for (const b of targets)
|
||||
this.enqueueNotice(b.chatId, `本轮异常结束（API/内部错误）${detail ? `：${detail}` : ''}`)
|
||||
} else if (kind === 'aborted') {
|
||||
const manual = reason?.reason?.kind === 'user'
|
||||
for (const b of targets)
|
||||
this.enqueueNotice(b.chatId, manual ? '本轮已被手动中止' : '本轮已被中断')
|
||||
} else if (kind === 'interrupted') {
|
||||
for (const b of targets)
|
||||
this.enqueueNotice(b.chatId, '会话曾被异常退出（崩溃恢复），本轮被标记中断')
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
if (event.type === 'tool/call') {
|
||||
// Progress summary; ask_user is owned by the provider hook (avoids duplicates).
|
||||
const name = event.data?.name ?? 'tool'
|
||||
if (name === 'ask_user_question') return
|
||||
this.thinkingSessions.set(id, false)
|
||||
if (event.data?.callId !== undefined) {
|
||||
if (this.callNames.size > 400) {
|
||||
const first = this.callNames.keys().next().value
|
||||
if (first !== undefined) this.callNames.delete(first)
|
||||
}
|
||||
this.callNames.set(String(event.data.callId), String(name))
|
||||
}
|
||||
const args = typeof event.data?.arguments === 'string' ? event.data.arguments : ''
|
||||
const brief = args.replace(/\s+/g, ' ').trim()
|
||||
const shown = brief.length > 60 ? `${brief.slice(0, 60)}…` : brief
|
||||
for (const b of targets)
|
||||
this.enqueueNotice(b.chatId, `> ${name}${shown ? ` ${shown}` : ''}`)
|
||||
return
|
||||
}
|
||||
|
||||
if (event.type === 'assistant/message') {
|
||||
const text = contentToText(event.data.message.content)
|
||||
this.thinkingSessions.set(id, false)
|
||||
if (!text) return
|
||||
await Promise.all(targets.map((b) => this.deliver(b.chatId, text)))
|
||||
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') {
|
||||
// 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 } }
|
||||
if (data?.error) {
|
||||
const callKey = String(data.message?.tool_use_id ?? data.message?.callId ?? '')
|
||||
const name = this.callNames.get(callKey) ?? 'tool'
|
||||
for (const b of targets)
|
||||
this.enqueueNotice(b.chatId, `> 工具 ${name} 失败：${String(data.error.name ?? 'Error')}${data.error.code ? `/${String(data.error.code)}` : ''}`)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
if (event.type === 'todo/write') {
|
||||
const todos = Array.isArray(event.data?.todos) ? event.data.todos : []
|
||||
this.lastTodos.set(id, todos as Array<{ content?: string; status?: string }>)
|
||||
if (todos.length === 0) return
|
||||
const done = todos.filter((t) => t?.status === 'completed').length
|
||||
const current = todos.find((t) => t?.status === 'in_progress') ?? todos.find((t) => t?.status === 'pending')
|
||||
const cur = current ? String(current.content ?? '').slice(0, 50) : ''
|
||||
for (const b of targets)
|
||||
this.enqueueNotice(b.chatId, `> 进度 ${done}/${todos.length}${cur ? `：${cur}` : ''}`)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
private async deliver(chatId: number, markdown: string): Promise<void> {
|
||||
if (this.renderingMode === 'html') {
|
||||
await this.deliverHtml(chatId, markdown)
|
||||
return
|
||||
}
|
||||
|
||||
let sentAny = false
|
||||
try {
|
||||
const chunks = splitRichMarkdown(markdown)
|
||||
for (const chunk of chunks) {
|
||||
await this.client.sendRichMessage(chatId, chunk)
|
||||
sentAny = true
|
||||
}
|
||||
} catch (err) {
|
||||
if (!sentAny && isRichUnsupportedError(err)) {
|
||||
this.ctx.logger.warn(
|
||||
'dsh-telegram-channel: Rich Message API unavailable, falling back to HTML rendering',
|
||||
)
|
||||
this.renderingMode = 'html'
|
||||
await this.deliverHtml(chatId, markdown)
|
||||
if (this.renderingMode !== 'html') {
|
||||
let sentAny = false
|
||||
try {
|
||||
const chunks = splitRichMarkdown(markdown)
|
||||
for (const chunk of chunks) {
|
||||
try {
|
||||
// Per-chunk retry: transient network blips must not drop a reply.
|
||||
await this.withRetry(() => this.client.sendRichMessage(chatId, chunk))
|
||||
sentAny = true
|
||||
} catch (err) {
|
||||
if (!sentAny && isRichUnsupportedError(err)) {
|
||||
this.ctx.logger.warn(
|
||||
'dsh-telegram-channel: Rich Message API unavailable, falling back to HTML rendering',
|
||||
)
|
||||
this.renderingMode = 'html'
|
||||
await this.deliverHtml(chatId, markdown)
|
||||
return
|
||||
}
|
||||
throw err
|
||||
}
|
||||
}
|
||||
return
|
||||
} catch (err) {
|
||||
// Persistent rich failure → HTML fallback instead of dropping the message.
|
||||
this.ctx.logger.warn(`dsh-telegram-channel: rich deliver failed (${this.redact(err)}), falling back to HTML`)
|
||||
}
|
||||
this.ctx.logger.error(this.redact(err))
|
||||
}
|
||||
await this.deliverHtml(chatId, markdown)
|
||||
}
|
||||
|
||||
private async deliverHtml(chatId: number, markdown: string): Promise<void> {
|
||||
@@ -719,10 +954,10 @@ export class TelegramBridge {
|
||||
for (const chunk of chunks) {
|
||||
const html = markdownToHtml(chunk)
|
||||
try {
|
||||
await this.client.sendMessage(chatId, html, 'HTML')
|
||||
await this.withRetry(() => this.client.sendMessage(chatId, html, 'HTML'))
|
||||
} catch {
|
||||
try {
|
||||
await this.client.sendMessage(chatId, chunk)
|
||||
await this.withRetry(() => this.client.sendMessage(chatId, chunk))
|
||||
} catch (err) {
|
||||
this.ctx.logger.error(this.redact(err))
|
||||
}
|
||||
@@ -730,6 +965,472 @@ export class TelegramBridge {
|
||||
}
|
||||
}
|
||||
|
||||
// ── Stability & interactivity helpers ──
|
||||
|
||||
/** Retry an outbound call with capped linear backoff (500ms, 1s, 2s… max 4s). */
|
||||
private async withRetry<T>(fn: () => Promise<T>, tries = 3): Promise<T> {
|
||||
let lastErr: unknown
|
||||
for (let attempt = 1; attempt <= tries; attempt++) {
|
||||
try {
|
||||
return await fn()
|
||||
} catch (err) {
|
||||
lastErr = err
|
||||
if (attempt < tries) await this.sleep(Math.min(500 * 2 ** (attempt - 1), 4000))
|
||||
}
|
||||
}
|
||||
throw lastErr
|
||||
}
|
||||
|
||||
private escHtml(s: string): string {
|
||||
return String(s)
|
||||
.replace(/&/g, '&')
|
||||
.replace(/</g, '<')
|
||||
.replace(/>/g, '>')
|
||||
}
|
||||
|
||||
/**
|
||||
* Send a notice; a leading `> ` renders as a Telegram blockquote (HTML).
|
||||
* Falls back to plain text when the HTML send fails — a notice is never lost.
|
||||
*/
|
||||
private async deliverNotice(chatId: number, text: string): Promise<void> {
|
||||
const quoted = text.startsWith('> ')
|
||||
const body = quoted ? text.slice(2) : text
|
||||
try {
|
||||
await this.withRetry(() => {
|
||||
if (!quoted) return this.client.sendMessage(chatId, body)
|
||||
return this.client.sendMessage(chatId, `<blockquote>${this.escHtml(body)}</blockquote>`, 'HTML')
|
||||
})
|
||||
return
|
||||
} catch {
|
||||
try {
|
||||
await this.client.sendMessage(chatId, quoted ? `> ${body}` : body)
|
||||
} catch (err) {
|
||||
this.ctx.logger.error(this.redact(err))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Serialized per-chat notice chain — bursts can't race into Telegram 429s. */
|
||||
private enqueueNotice(chatId: number, text: string): Promise<void> {
|
||||
const key = String(chatId)
|
||||
const prev = this.noticeQueue.get(key) ?? Promise.resolve()
|
||||
const next = prev.then(() => this.deliverNotice(chatId, text)).catch(() => {})
|
||||
this.noticeQueue.set(key, next)
|
||||
return next
|
||||
}
|
||||
|
||||
private startTypingHeartbeat(chatId: number): void {
|
||||
this.stopTypingHeartbeat(chatId)
|
||||
const send = (): void => {
|
||||
void this.client.sendChatAction(chatId, 'typing').catch(() => {})
|
||||
}
|
||||
send()
|
||||
this.heartbeats.set(String(chatId), setInterval(send, 4000))
|
||||
}
|
||||
|
||||
private stopTypingHeartbeat(chatId: number): void {
|
||||
const h = this.heartbeats.get(String(chatId))
|
||||
if (h) {
|
||||
clearInterval(h)
|
||||
this.heartbeats.delete(String(chatId))
|
||||
}
|
||||
}
|
||||
|
||||
private stopAllHeartbeats(): void {
|
||||
for (const h of this.heartbeats.values()) clearInterval(h)
|
||||
this.heartbeats.clear()
|
||||
}
|
||||
|
||||
// ── ask_user_question: TG answering via dual-path race with the UI ──
|
||||
|
||||
private userQuestions(): UserQuestionsServiceLike | undefined {
|
||||
return (this.ctx as unknown as { userQuestions?: UserQuestionsServiceLike }).userQuestions
|
||||
}
|
||||
|
||||
/**
|
||||
* 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(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
|
||||
}
|
||||
this.hookTimer = setTimeout(() => this.hookUserQuestions(attempt + 1), 2000)
|
||||
}
|
||||
|
||||
private registerTgAsk(request: unknown): { promise: Promise<unknown>; reject: (err: Error) => void } {
|
||||
let resolveFn: (v: unknown) => void = () => {}
|
||||
let rejectFn: (err: Error) => void = () => {}
|
||||
const promise = new Promise<unknown>((resolve, reject) => {
|
||||
resolveFn = resolve
|
||||
rejectFn = reject
|
||||
})
|
||||
const ask: TgAnswerHandle = { resolve: resolveFn, reject: rejectFn, answered: false }
|
||||
const req = request as { agent?: Agent; questions?: TgQuestionItem[]; signal?: AbortSignal }
|
||||
const sessionId = req.agent?.id !== undefined ? String(req.agent.id) : undefined
|
||||
const questions = Array.isArray(req.questions) ? req.questions : []
|
||||
const onAbort = (): void => {
|
||||
for (const [chatId, p] of this.pendingAsks) {
|
||||
if (p.ask === ask) this.pendingAsks.delete(chatId)
|
||||
}
|
||||
rejectFn(new Error('ask aborted'))
|
||||
}
|
||||
if (req.signal?.aborted) {
|
||||
rejectFn(new Error('ask aborted'))
|
||||
return { promise, reject: rejectFn }
|
||||
}
|
||||
req.signal?.addEventListener('abort', onAbort, { once: true })
|
||||
if (sessionId !== undefined && questions.length > 0) {
|
||||
let asked = false
|
||||
for (const [chatId, binding] of this.bindings) {
|
||||
if (binding.sessionId !== sessionId) continue
|
||||
this.pendingAsks.set(String(chatId), { req: request, sessionId, questions, ask, at: Date.now() })
|
||||
this.enqueueNotice(Number(chatId), this.formatAskPending(questions))
|
||||
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 }
|
||||
}
|
||||
|
||||
private settleGuiSide(request: unknown): void {
|
||||
for (const [chatId, p] of [...this.pendingAsks]) {
|
||||
if (p.req !== request) continue
|
||||
this.pendingAsks.delete(chatId)
|
||||
if (!p.ask.answered) {
|
||||
this.enqueueNotice(Number(chatId), '该问题已在 Web 端作答，TG 作答通道关闭')
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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> {
|
||||
const { questions, ask } = pending
|
||||
let parts: Array<[number | null, string]>
|
||||
if (questions.length === 1) {
|
||||
parts = [[0, text.trim()]]
|
||||
} else {
|
||||
parts = []
|
||||
const lines = text.split(/\n|；|;/).map((s) => s.trim()).filter(Boolean)
|
||||
for (const line of lines) {
|
||||
const m = line.match(/^([0-9]+)\s*[:：.、)]\s*(.+)$/)
|
||||
if (m) parts.push([Number(m[1]) - 1, m[2]!.trim()])
|
||||
else parts.push([null, line])
|
||||
}
|
||||
if (parts.some(([idx]) => idx === null)) {
|
||||
await this.enqueueNotice(chatId, `共 ${questions.length} 个问题，请每行用 1:/2: 前缀分别作答`)
|
||||
return
|
||||
}
|
||||
}
|
||||
const answers: Array<{ id?: string; selected: string[]; custom?: string }> = []
|
||||
for (const [idx, raw] of parts) {
|
||||
if (idx === null || idx < 0 || idx >= questions.length) {
|
||||
await this.enqueueNotice(chatId, `题号 ${(idx ?? 0) + 1} 不存在（共 ${questions.length} 题）`)
|
||||
return
|
||||
}
|
||||
const q = questions[idx]!
|
||||
const cleaned = raw.replace(/^\[+/, '').replace(/\]+$/, '').trim()
|
||||
const letters = cleaned.split(/[,，、\s]+/).map((s) => s.trim()).filter(Boolean)
|
||||
const opts = Array.isArray(q.options) ? q.options : []
|
||||
const isLetterChoice = letters.length > 0
|
||||
&& letters.every((s) => /^[a-zA-Z]$/.test(s))
|
||||
&& opts.length > 0
|
||||
if (isLetterChoice) {
|
||||
const selected: string[] = []
|
||||
for (const s of letters) {
|
||||
const i = s.toUpperCase().charCodeAt(0) - 65
|
||||
if (i < 0 || i >= opts.length) {
|
||||
await this.enqueueNotice(chatId, `选项 ${s.toUpperCase()} 不存在（范围 A-${String.fromCharCode(64 + opts.length)}），请重答或 /cancel`)
|
||||
return
|
||||
}
|
||||
selected.push(opts[i]!.label)
|
||||
}
|
||||
answers.push({ id: q.id, selected })
|
||||
} else {
|
||||
answers.push({ id: q.id, selected: [], custom: cleaned })
|
||||
}
|
||||
}
|
||||
ask.answered = true
|
||||
for (const [cid, p] of this.pendingAsks) {
|
||||
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 ──
|
||||
|
||||
private async onApprovalRequest(
|
||||
req: ApprovalRequestLike,
|
||||
next: () => Promise<string>,
|
||||
): Promise<string> {
|
||||
const sessionId = req?.agent?.id !== undefined ? String(req.agent.id) : undefined
|
||||
const targets = sessionId !== undefined
|
||||
? [...this.bindings.values()].filter((b) => b.sessionId === sessionId).map((b) => String(b.chatId))
|
||||
: []
|
||||
let resolveFn: (v: unknown) => void = () => {}
|
||||
let rejectFn: (err: Error) => void = () => {}
|
||||
const promise = new Promise<unknown>((resolve, reject) => {
|
||||
resolveFn = resolve
|
||||
rejectFn = reject
|
||||
})
|
||||
const ask: TgAnswerHandle = { resolve: resolveFn, reject: rejectFn, answered: false }
|
||||
const onAbort = (): void => {
|
||||
for (const [chatId, p] of this.pendingApprovalsTG) {
|
||||
if (p.ask === ask) this.pendingApprovalsTG.delete(chatId)
|
||||
}
|
||||
rejectFn(new Error('approval withdrawn'))
|
||||
}
|
||||
if (req?.signal?.aborted) {
|
||||
rejectFn(new Error('approval withdrawn'))
|
||||
return next()
|
||||
}
|
||||
req?.signal?.addEventListener('abort', onAbort, { once: true })
|
||||
if (targets.length > 0) {
|
||||
const toolName = String(req?.toolName ?? 'tool')
|
||||
const reason = req?.reason ? String(req.reason).slice(0, 200) : ''
|
||||
for (const chatId of targets) {
|
||||
this.pendingApprovalsTG.set(chatId, { toolName, ask })
|
||||
this.enqueueNotice(Number(chatId), `权限审批：${toolName}${reason ? `\n(${reason})` : ''}\n回复 [A] 允许一次 / [B] 拒绝（/cancel 关闭 TG 审批）`)
|
||||
}
|
||||
}
|
||||
// No bound TG chat: promise never settles — the UI remains the only answer path.
|
||||
const gui = next()
|
||||
void gui.then(() => {
|
||||
for (const [chatId, p] of [...this.pendingApprovalsTG]) {
|
||||
if (p.ask !== ask) continue
|
||||
this.pendingApprovalsTG.delete(chatId)
|
||||
if (!p.ask.answered) {
|
||||
p.ask.reject(new Error('superseded-by-gui'))
|
||||
this.enqueueNotice(Number(chatId), '该审批已在 Web 端处理')
|
||||
}
|
||||
}
|
||||
}, () => {
|
||||
for (const [chatId, p] of [...this.pendingApprovalsTG]) {
|
||||
if (p.ask === ask) this.pendingApprovalsTG.delete(chatId)
|
||||
}
|
||||
})
|
||||
return Promise.race([promise, gui]) as Promise<string>
|
||||
}
|
||||
|
||||
private async handleTgApproval(chatId: number, approval: TgPendingApproval, text: string): Promise<void> {
|
||||
const cleaned = text.trim().replace(/^\[+/, '').replace(/\]+$/, '').toUpperCase()
|
||||
if (cleaned === 'A' || cleaned === 'B') {
|
||||
approval.ask.answered = true
|
||||
for (const [cid, p] of this.pendingApprovalsTG) {
|
||||
if (p.ask === approval.ask) this.pendingApprovalsTG.delete(cid)
|
||||
}
|
||||
if (cleaned === 'A') {
|
||||
approval.ask.resolve('allowed-once')
|
||||
await this.enqueueNotice(chatId, `已允许 ${approval.toolName} 执行一次`)
|
||||
} else {
|
||||
approval.ask.resolve('rejected')
|
||||
await this.enqueueNotice(chatId, '已拒绝该工具执行')
|
||||
}
|
||||
return
|
||||
}
|
||||
await this.enqueueNotice(chatId, '请回复 [A] 允许一次 或 [B] 拒绝（/cancel 关闭）')
|
||||
}
|
||||
|
||||
// ── /stop /mission /new commands ──
|
||||
|
||||
private async stopBound(chatId: number): Promise<void> {
|
||||
const binding = this.bindings.get(String(chatId))
|
||||
if (!binding) {
|
||||
await this.client.sendMessage(chatId, MSG.NEED_BIND)
|
||||
return
|
||||
}
|
||||
const agent = this.findLiveAgent(binding.sessionId)
|
||||
if (!agent) {
|
||||
await this.client.sendMessage(chatId, '当前会话没有正在运行的任务')
|
||||
return
|
||||
}
|
||||
try {
|
||||
agent.cancel({ kind: 'user' })
|
||||
} catch (err) {
|
||||
this.ctx.logger.warn(`dsh-telegram-channel: stop failed: ${this.redact(err)}`)
|
||||
}
|
||||
await this.enqueueNotice(chatId, '已请求中止当前轮…')
|
||||
}
|
||||
|
||||
private async sendMission(chatId: number): Promise<void> {
|
||||
const binding = this.bindings.get(String(chatId))
|
||||
if (!binding) {
|
||||
await this.client.sendMessage(chatId, MSG.NEED_BIND)
|
||||
return
|
||||
}
|
||||
const todos = this.lastTodos.get(binding.sessionId) ?? []
|
||||
const lines = ['任务清单']
|
||||
if (todos.length === 0) {
|
||||
lines.push('（暂无任务记录，agent 使用 todo 工具后这里会显示）')
|
||||
} else {
|
||||
const mark: Record<string, string> = { completed: '✓', in_progress: '⟳', pending: '·' }
|
||||
todos.forEach((t, i) => lines.push(`${mark[t?.status ?? ''] ?? '·'} ${i + 1}. ${String(t?.content ?? '')}`))
|
||||
const done = todos.filter((t) => t?.status === 'completed').length
|
||||
lines.push(`—— ${done}/${todos.length} 已完成`)
|
||||
}
|
||||
lines.push(this.busySessions.has(binding.sessionId) ? '（当前任务：运行中，/stop 可中止）' : '（当前空闲）')
|
||||
await this.client.sendMessage(chatId, lines.join('\n'))
|
||||
}
|
||||
|
||||
/** /new: create a session in the bound session's workspace and attach to it. */
|
||||
private async newSessionHere(chatId: number): Promise<void> {
|
||||
const binding = this.bindings.get(String(chatId))
|
||||
if (!binding) {
|
||||
await this.client.sendMessage(chatId, MSG.NEED_BIND)
|
||||
return
|
||||
}
|
||||
let workspaceId: string | undefined
|
||||
let fallbackCwd: string | undefined
|
||||
try {
|
||||
const catalog = await this.resolveCatalog()
|
||||
for (const ws of catalog?.workspaces ?? []) {
|
||||
if ((ws.sessionIds ?? []).includes(binding.sessionId)) {
|
||||
workspaceId = ws.id
|
||||
break
|
||||
}
|
||||
}
|
||||
if (workspaceId === undefined)
|
||||
fallbackCwd = catalog?.sessionsById?.get(binding.sessionId)?.cwd
|
||||
} catch {
|
||||
// Catalog unavailable → fall back to the host default workspace.
|
||||
}
|
||||
const proxy = resolveApiProxy(this.ctx)
|
||||
if (!proxy?.sessions?.create) {
|
||||
await this.client.sendMessage(chatId, '当前宿主不支持会话创建接口')
|
||||
return
|
||||
}
|
||||
let res: { result?: { ok?: boolean; value?: { sessionId?: string }; error?: { message?: string } } } | undefined
|
||||
try {
|
||||
const payload = workspaceId !== undefined ? { workspaceId } : (fallbackCwd ? { cwd: fallbackCwd } : {})
|
||||
res = await proxy.sessions.create({
|
||||
rpcId: `tg-${Date.now()}-${Math.random().toString(36).slice(2)}`,
|
||||
payload,
|
||||
}) as typeof res
|
||||
} catch (err) {
|
||||
await this.client.sendMessage(chatId, `创建失败：${this.redact(err)}`)
|
||||
return
|
||||
}
|
||||
const newId = res?.result?.ok ? res.result.value?.sessionId : undefined
|
||||
if (!newId) {
|
||||
const detail = res?.result?.error?.message ?? JSON.stringify(res?.result ?? res)
|
||||
await this.client.sendMessage(chatId, `创建失败：${String(detail).slice(0, 200)}`)
|
||||
return
|
||||
}
|
||||
const stale = this.pendingAsks.get(String(chatId))
|
||||
if (stale && stale.sessionId !== String(newId)) this.pendingAsks.delete(String(chatId))
|
||||
this.bindings.set(String(chatId), { chatId, sessionId: String(newId), label: binding.label })
|
||||
this.saveBindings()
|
||||
const where = workspaceId !== undefined ? '当前工作区' : (fallbackCwd ? `目录 ${fallbackCwd}` : '默认工作区')
|
||||
await this.client.sendMessage(chatId, `已在${where}新开对话并附着：\n${String(newId)}\n直接发消息即可开始`)
|
||||
}
|
||||
|
||||
// ── Binding persistence (survives hot reload / restart) ──
|
||||
|
||||
private bindingsPath(): string {
|
||||
// DSH_TELEGRAM_BINDINGS_FILE overrides the location (tests must isolate
|
||||
// from the real profile data — parallel test files share this file).
|
||||
const override = process.env.DSH_TELEGRAM_BINDINGS_FILE
|
||||
if (override) return override
|
||||
const home = process.env.DSH_HOME ?? join(homedir(), '.dsh')
|
||||
return join(home, 'telegram-channel-bindings.json')
|
||||
}
|
||||
|
||||
private saveBindings(): void {
|
||||
try {
|
||||
const obj = Object.fromEntries(this.bindings)
|
||||
mkdirSync(dirname(this.bindingsPath()), { recursive: true })
|
||||
writeFileSync(this.bindingsPath(), JSON.stringify(obj, null, 2))
|
||||
} catch (err) {
|
||||
this.ctx.logger.warn(`dsh-telegram-channel: saveBindings failed: ${this.redact(err)}`)
|
||||
}
|
||||
}
|
||||
|
||||
private loadBindings(): void {
|
||||
try {
|
||||
const raw = readFileSync(this.bindingsPath(), 'utf8')
|
||||
const obj = JSON.parse(raw) as Record<string, { chatId?: number; sessionId?: string; label?: string }>
|
||||
let n = 0
|
||||
for (const [chatId, b] of Object.entries(obj ?? {})) {
|
||||
if (b && b.sessionId) {
|
||||
this.bindings.set(String(chatId), {
|
||||
chatId: b.chatId ?? Number(chatId),
|
||||
sessionId: String(b.sessionId),
|
||||
label: b.label ?? String(b.sessionId),
|
||||
})
|
||||
n += 1
|
||||
}
|
||||
}
|
||||
if (n > 0)
|
||||
this.ctx.logger.info(`dsh-telegram-channel: restored ${n} binding(s) from disk`)
|
||||
} catch {
|
||||
// First run: no file yet.
|
||||
}
|
||||
}
|
||||
|
||||
private redact(value: unknown): string {
|
||||
const message = value instanceof Error ? value.message : String(value)
|
||||
return message.split(this.token).join('***')
|
||||
|
||||
Reference in New Issue
Block a user