feat: /status 通用状态显示 + /compact 手动压缩;HTML 兼容渲染默认化

- /status:绑定会话/会话 ID/工作区/当前模型/思考强度/上下文长度/
  首 token 平均/输出速率/输入输出 tokens —— 与 Web 底部统计条同源
  同算法(sessionProjections: sessionStats/tokenUsage/contextPressure),
  投影插件缺失时逐组优雅降级;冷会话不强制 resume
- /compact:复用 ctx.compaction.compactNow 手动压缩绑定会话历史,
  异步执行不阻塞轮询,完成/失败均中文通知,同会话去重,
  桥停止/热重载自动中止进行中的压缩
- 全部用户可见文案不使用 emoji
- rendering 默认改 html(兼容优先)+ userQuestions/approval 双通道等
  既有修改一并合入(此前已上线确认)
- tests:+6 覆盖 /status 全字段与 /compact 各分支(51 全绿)
This commit is contained in:
RRRosmontis
2026-09-05 17:24:41 +08:00
parent 20898d5d21
commit d712fb3494
15 changed files with 1695 additions and 72 deletions
+347 -30
View File
@@ -28,13 +28,19 @@ import { MSG, LAST_CB, parseCommand } from './commands.js'
import { markdownToHtml, splitMessage } from './format.js'
import { splitRichMarkdown } from './rich-format.js'
import { formatLastTurn, loadLastTurn } from './history.js'
import { describeAgent, displayLabel } from './label.js'
import { describeAgent, displayLabel, workspaceName } from './label.js'
import {
formatModel,
loadSessionModels,
selectSessionModel,
type ModelOption,
} from './models.js'
import {
formatStatusText,
type StatusInfo,
type StatusRuntime,
type StatusWorkspace,
} from './status.js'
export interface TelegramBridgeOptions {
token: string
@@ -113,6 +119,38 @@ interface ApprovalRequestLike {
signal?: AbortSignal
}
/** ── /status 与 /compact 需要的宿主缝隙(结构镜像,避免引入额外依赖)── */
/** 会话投影注册表(只读使用 snapshot):对齐 Web 底部统计条的数据源。 */
interface ProjectionRegistryLike {
snapshot?: (session: unknown) => { asOfSeq: number; values: Record<string, unknown> } | undefined
}
/** 手动压缩引擎(结构镜像 ctx.compaction.compactNow)。 */
interface CompactionEngineLike {
compactNow?: (
agent: unknown,
signal: AbortSignal,
sourceCommandId?: string,
) => Promise<{
shadowedSeqs: number[]
shadowedTokenCount: number
summarySeq: number
} | null>
}
/** 手动压缩的预期失败类(ManualCompactionError.code 的运行时字符串)。 */
type CompactFailureCode = 'busy' | 'cancelled' | 'changed' | 'summary' | 'commit' | 'persistence'
function compactErrorCode(err: unknown): CompactFailureCode | undefined {
const code = (err as { code?: unknown } | undefined)?.code
if (code === 'busy' || code === 'cancelled' || code === 'changed'
|| code === 'summary' || code === 'commit' || code === 'persistence') {
return code
}
return undefined
}
const WS_CB = 'ws:'
const SID_CB = 'sid:'
const BACK_WS_CB = 'wb'
@@ -150,7 +188,12 @@ export class TelegramBridge {
private readonly client: TelegramClientLike
private readonly sleep: (ms: number) => Promise<void>
private readonly maxMessageLength: number
/** Global rendering mode: follows config; flipped to 'html' permanently when the
* Rich Message API itself is unavailable (a server-wide fact, not per-chat). */
private renderingMode: 'rich' | 'html'
/** Per-chat rendering override (persisted): rich messages need a recent Telegram
* client — old ones (e.g. 10.x) cannot render them and show “unsupported”. */
private readonly renderPrefs = new Map<string, 'rich' | 'html'>()
private readonly bindings = new Map<string, Binding>()
private readonly pickers = new Map<string, PickerState>()
@@ -179,6 +222,10 @@ export class TelegramBridge {
private readonly callNames = new Map<string, string>()
/** sessionId → latest todo snapshot (/mission) */
private readonly lastTodos = new Map<string, Array<{ content?: string; status?: string }>>()
/** sessionId → in-flight manual compaction (dedup /compact) */
private readonly compacting = new Set<string>()
/** sessionId → compaction abort controller (cancelled on bridge stop) */
private readonly compactAborts = new Map<string, AbortController>()
constructor(ctx: Context, options: TelegramBridgeOptions) {
this.ctx = ctx
@@ -190,7 +237,14 @@ export class TelegramBridge {
})
this.sleep = options.sleep ?? defaultSleep
this.maxMessageLength = options.maxMessageLength ?? 4096
this.renderingMode = options.rendering === 'html' ? 'html' : 'rich'
// Compat-first default: only an explicit `rendering: 'rich'` opts into rich
// messages, because old Telegram clients (10.x) cannot render them at all.
this.renderingMode = options.rendering === 'rich' ? 'rich' : 'html'
}
/** Effective rendering mode for a chat: per-chat override, else the global mode. */
private modeFor(chatId: number): 'rich' | 'html' {
return this.renderPrefs.get(String(chatId)) ?? this.renderingMode
}
start(): void {
@@ -213,7 +267,9 @@ export class TelegramBridge {
{ command: 'sessions', description: '按工作区列出并附着会话' },
{ command: 'last', description: '查看上次问答(续接上下文)' },
{ command: 'model', description: '切换当前绑定会话的模型' },
{ command: 'status', description: '查看当前绑定' },
{ command: 'status', description: '会话状态:模型/上下文/统计(首token/速率/tokens)' },
{ command: 'compact', description: '压缩当前会话历史(缩短上下文,空闲时执行)' },
{ command: 'rich', description: '渲染:on 富文本 / off HTML 兼容(本聊天,需新版客户端)' },
{ command: 'unbind', description: '断开手机绑定(不关闭本机会话)' },
{ command: 'stop', description: '中止当前正在运行的任务' },
{ command: 'mission', description: '查看任务清单与完成情况' },
@@ -241,6 +297,7 @@ export class TelegramBridge {
this.disposeSessionListener = undefined
// Never dispose host agents — only clear remote bindings.
this.bindings.clear()
this.renderPrefs.clear()
this.pickers.clear()
this.pendingModels.clear()
this.stopAllHeartbeats()
@@ -251,6 +308,9 @@ export class TelegramBridge {
this.thinkingSessions.clear()
this.callNames.clear()
this.lastTodos.clear()
for (const controller of this.compactAborts.values()) controller.abort()
this.compactAborts.clear()
this.compacting.clear()
if (this.hookTimer) {
clearTimeout(this.hookTimer)
this.hookTimer = undefined
@@ -305,6 +365,12 @@ export class TelegramBridge {
case 'status':
await this.sendStatus(chatId)
return
case 'compact':
await this.requestCompact(chatId)
return
case 'rich':
await this.handleRichCommand(chatId, parsed.arg)
return
case 'unbind':
this.bindings.delete(String(chatId))
this.pickers.delete(String(chatId))
@@ -732,12 +798,215 @@ export class TelegramBridge {
await this.client.sendMessage(chatId, MSG.STATUS_NONE)
return
}
const stillLive = this.findLiveAgent(binding.sessionId)
if (!stillLive) {
await this.client.sendMessage(chatId, MSG.STATUS_BOUND_COLD(binding.label))
const info: StatusInfo = {
label: binding.label,
sessionId: binding.sessionId,
}
// 冷会话:不强制 resume(只读查询),给出可用的绑定/工作区信息即可。
const agent = this.findLiveAgent(binding.sessionId)
if (!agent) {
info.live = false
info.workspace = await this.workspaceOf(binding.sessionId, undefined)
await this.client.sendMessage(chatId, formatStatusText(info))
return
}
await this.client.sendMessage(chatId, MSG.STATUS_BOUND(binding.label))
info.live = true
info.workspace = await this.workspaceOf(binding.sessionId, agent)
try {
const snap = await loadSessionModels(this.ctx, binding.sessionId)
if (snap.current) {
info.model = {
provider: snap.current.provider,
model: snap.current.model,
reasoningEffort: snap.current.reasoningEffort,
}
}
} catch (err) {
this.ctx.logger.warn(`dsh-telegram-channel: /status model read failed: ${this.redact(err)}`)
}
info.runtime = this.readRuntimeValues(agent)
await this.client.sendMessage(chatId, formatStatusText(info))
}
/** 按 sessionId 解析工作区:apiProxy 目录优先,冷/热会话均可用;退化为 agent cwd。 */
private async workspaceOf(sessionId: string, agent: Agent | undefined): Promise<StatusWorkspace | undefined> {
try {
const catalog = await this.resolveCatalog()
const ws = (catalog?.workspaces ?? []).find((w) => (w.sessionIds ?? []).includes(sessionId))
if (ws) {
return { title: ws.title || undefined, path: ws.path || undefined }
}
const row = catalog?.sessionsById?.get(sessionId)
if (row?.cwd) {
return { title: workspaceName(row.cwd) || undefined, path: row.cwd }
}
} catch {
// 目录不可用 → 继续走 agent 侧兜底
}
if (agent) {
const parts = describeAgent(agent, 0, this.ctx)
if (parts.cwd) {
return { title: parts.workspace || undefined, path: parts.cwd }
}
}
return undefined
}
/** 从 sessionProjections 快照读 Web 底部同款统计(sessionStats/tokenUsage/contextPressure)。 */
private readRuntimeValues(agent: Agent): StatusRuntime | undefined {
const registry = this.serviceOf<ProjectionRegistryLike>('sessionProjections')
let values: Record<string, unknown> | undefined
if (registry?.snapshot) {
try {
values = registry.snapshot(agent.session)?.values
} catch (err) {
this.ctx.logger.warn(`dsh-telegram-channel: projection snapshot failed: ${this.redact(err)}`)
}
}
if (!values) return undefined
const num = (v: unknown): number | undefined =>
typeof v === 'number' && Number.isFinite(v) ? v : undefined
const stats = (values['sessionStats'] ?? {}) as Record<string, unknown>
const usage = (values['tokenUsage'] ?? {}) as Record<string, unknown>
const pressure = (values['contextPressure'] ?? {}) as Record<string, unknown>
const runtime: StatusRuntime = {
turns: num(stats.turns),
steps: num(stats.steps),
llmMs: num(stats.llmMs),
toolMs: num(stats.toolMs),
ttftMs: num(stats.ttftMs),
ttftSteps: num(stats.ttftSteps),
decodeMs: num(stats.decodeMs),
decodeTokens: num(stats.decodeTokens),
uncachedInputTokens: num(usage.uncachedInputTokens),
cacheReadTokens: num(usage.cacheReadTokens),
cacheWriteTokens: num(usage.cacheWriteTokens),
outputTokens: num(usage.outputTokens),
contextUsed: num(pressure.projectedTokens) ?? num(pressure.pressureTokens),
contextWindow: num(pressure.contextWindow),
}
// 完全无数据时视为无统计块。
const hasAny = Object.values(runtime).some((v) => v !== undefined)
return hasAny ? runtime : undefined
}
/** 按名称解析宿主服务(Cordis 需 ctx.get;mock/plain ctx 走自有属性兜底)。 */
private serviceOf<T>(name: string): T | undefined {
const c = this.ctx as { get?: (name: string, strict?: boolean) => unknown }
if (typeof c.get === 'function') {
try {
const via = c.get(name, false)
if (via) return via as T
} catch {
// 服务未装配或当前 ctx 不可解析
}
}
try {
if (Object.prototype.hasOwnProperty.call(this.ctx, name)) {
return (this.ctx as unknown as Record<string, unknown>)[name] as T
}
} catch {
// proxy ctx:无 inject 直接读属性会抛错,忽略
}
return undefined
}
// ── /compact:手动压缩当前绑定会话(需会话空闲)──
private async requestCompact(chatId: number): Promise<void> {
const binding = this.bindings.get(String(chatId))
if (!binding) {
await this.client.sendMessage(chatId, MSG.NEED_BIND)
return
}
const engine = this.serviceOf<CompactionEngineLike>('compaction')
if (!engine?.compactNow) {
await this.client.sendMessage(chatId, MSG.COMPACT_UNAVAILABLE)
return
}
if (this.compacting.has(binding.sessionId)) {
await this.client.sendMessage(chatId, MSG.COMPACT_INFLIGHT)
return
}
const agent = await this.ensureLiveAgent(binding.sessionId)
if (!agent) {
this.bindings.delete(String(chatId))
await this.client.sendMessage(chatId, MSG.GONE)
return
}
const controller = new AbortController()
let operation: Promise<{ shadowedSeqs: number[]; shadowedTokenCount: number; summarySeq: number } | null>
try {
// compactNow 在会话忙时同步抛 busy(runMaintenance 同步判定)。
operation = engine.compactNow(agent as never, controller.signal)
} catch (err) {
const code = compactErrorCode(err)
if (code === 'busy') {
await this.client.sendMessage(chatId, MSG.COMPACT_BUSY)
return
}
this.ctx.logger.warn(`dsh-telegram-channel: /compact rejected: ${this.redact(err)}`)
await this.client.sendMessage(chatId, MSG.COMPACT_FAILED(this.redact(err)))
return
}
this.compacting.add(binding.sessionId)
this.compactAborts.set(binding.sessionId, controller)
await this.client.sendMessage(chatId, MSG.COMPACT_STARTED)
void this.runCompaction(chatId, binding.sessionId, operation, controller).catch((err) => {
this.ctx.logger.error(`dsh-telegram-channel: /compact runner crashed: ${this.redact(err)}`)
})
}
/** 等待压缩落定并汇报结果(不阻塞轮询循环)。 */
private async runCompaction(
chatId: number,
sessionId: string,
operation: Promise<{ shadowedSeqs: number[]; shadowedTokenCount: number; summarySeq: number } | null>,
controller: AbortController,
): Promise<void> {
try {
const result = await operation
if (result === null) {
await this.enqueueNotice(chatId, MSG.COMPACT_NOTHING)
} else {
await this.enqueueNotice(chatId, MSG.COMPACT_DONE(result.shadowedSeqs.length, result.shadowedTokenCount))
}
} catch (err) {
if (controller.signal.aborted) {
// 桥停止/热重载导致的取消:静默,不打扰(本实例正在退出)。
this.ctx.logger.info('dsh-telegram-channel: compaction aborted by bridge stop')
return
}
const code = compactErrorCode(err)
switch (code) {
case 'cancelled':
await this.enqueueNotice(chatId, MSG.COMPACT_CANCELLED)
return
case 'changed':
await this.enqueueNotice(chatId, MSG.COMPACT_CHANGED)
return
case 'summary':
await this.enqueueNotice(chatId, MSG.COMPACT_SUMMARY_FAILED)
return
case 'commit':
await this.enqueueNotice(chatId, MSG.COMPACT_COMMIT_FAILED)
return
case 'persistence':
await this.enqueueNotice(chatId, MSG.COMPACT_PERSIST_FAILED)
return
case 'busy':
await this.enqueueNotice(chatId, MSG.COMPACT_BUSY)
return
default:
this.ctx.logger.warn(`dsh-telegram-channel: /compact failed: ${this.redact(err)}`)
await this.enqueueNotice(chatId, MSG.COMPACT_FAILED(this.redact(err)))
}
} finally {
this.compacting.delete(sessionId)
this.compactAborts.delete(sessionId)
}
}
private async followupBound(chatId: number, text: string): Promise<void> {
@@ -920,8 +1189,47 @@ export class TelegramBridge {
}
}
// ── /rich: per-chat rendering mode (persisted) ──
//
// Rich messages (Bot API sendRichMessage) only render on recent Telegram
// clients; old ones (10.x) show the message as “not supported”. The bot cannot
// detect the recipient's client version, so the choice is per-chat and manual,
// remembered forever. HTML mode is the compat default and works everywhere.
private async handleRichCommand(chatId: number, arg?: string): Promise<void> {
const key = String(chatId)
const setAndReply = async (mode: 'rich' | 'html'): Promise<void> => {
this.renderPrefs.set(key, mode)
this.saveBindings()
await this.client.sendMessage(chatId, MSG.RICH_SET(mode))
}
switch (arg) {
case 'on':
case 'rich':
await setAndReply('rich')
return
case 'off':
case 'html':
case 'compat':
await setAndReply('html')
return
case 'auto':
case 'default':
case 'reset':
this.renderPrefs.delete(key)
this.saveBindings()
await this.client.sendMessage(chatId, MSG.RICH_STATE(this.renderingMode))
return
case undefined:
await this.client.sendMessage(chatId, MSG.RICH_STATE(this.modeFor(chatId)))
return
default:
await this.client.sendMessage(chatId, MSG.RICH_USAGE)
}
}
private async deliver(chatId: number, markdown: string): Promise<void> {
if (this.renderingMode !== 'html') {
if (this.modeFor(chatId) !== 'html') {
let sentAny = false
try {
const chunks = splitRichMarkdown(markdown)
@@ -936,6 +1244,12 @@ export class TelegramBridge {
'dsh-telegram-channel: Rich Message API unavailable, falling back to HTML rendering',
)
this.renderingMode = 'html'
// Rich API being gone is a server-wide fact: per-chat rich overrides
// must not bypass the global downgrade (modeFor prefers the override).
for (const [key, mode] of this.renderPrefs) {
if (mode === 'rich') this.renderPrefs.delete(key)
}
this.saveBindings()
await this.deliverHtml(chatId, markdown)
return
}
@@ -988,32 +1302,20 @@ export class TelegramBridge {
throw lastErr
}
private escHtml(s: string): string {
return String(s)
.replace(/&/g, '&amp;')
.replace(/</g, '&lt;')
.replace(/>/g, '&gt;')
}
/**
* 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.
* Send a notice; a leading `> ` keeps the quote look via a plain-text `> `
* prefix. NB: NOT `<blockquote>` HTML — older Telegram clients cannot render
* the blockquote entity at all and show the whole message as “not supported”.
* Falls back to plain text when the send fails — a notice is never lost.
*/
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')
})
await this.withRetry(() => this.client.sendMessage(chatId, quoted ? `> ${body}` : body))
return
} catch {
try {
await this.client.sendMessage(chatId, quoted ? `> ${body}` : body)
} catch (err) {
this.ctx.logger.error(this.redact(err))
}
} catch (err) {
this.ctx.logger.error(this.redact(err))
}
}
@@ -1408,7 +1710,10 @@ export class TelegramBridge {
private saveBindings(): void {
try {
const obj = Object.fromEntries(this.bindings)
const obj = {
bindings: Object.fromEntries(this.bindings),
renderPrefs: Object.fromEntries(this.renderPrefs),
}
mkdirSync(dirname(this.bindingsPath()), { recursive: true })
writeFileSync(this.bindingsPath(), JSON.stringify(obj, null, 2))
} catch (err) {
@@ -1419,9 +1724,16 @@ export class TelegramBridge {
private loadBindings(): void {
try {
const raw = readFileSync(this.bindingsPath(), 'utf8')
const obj = JSON.parse(raw) as Record<string, { chatId?: number; sessionId?: string; label?: string }>
const parsed = JSON.parse(raw) as unknown
const data = parsed as {
bindings?: Record<string, { chatId?: number; sessionId?: string; label?: string }>
renderPrefs?: Record<string, string>
}
// Legacy files (pre-renderPrefs) stored the bindings flat at the top level.
const flat = parsed as Record<string, { chatId?: number; sessionId?: string; label?: string }>
const bindingEntries = data.bindings ?? flat
let n = 0
for (const [chatId, b] of Object.entries(obj ?? {})) {
for (const [chatId, b] of Object.entries(bindingEntries ?? {})) {
if (b && b.sessionId) {
this.bindings.set(String(chatId), {
chatId: b.chatId ?? Number(chatId),
@@ -1431,6 +1743,11 @@ export class TelegramBridge {
n += 1
}
}
for (const [chatId, mode] of Object.entries(data.renderPrefs ?? {})) {
if (mode === 'rich' || mode === 'html') {
this.renderPrefs.set(String(chatId), mode)
}
}
if (n > 0)
this.ctx.logger.info(`dsh-telegram-channel: restored ${n} binding(s) from disk`)
} catch {