jiangchen
3205aaea76
挖到真源：事件流里 assistant/chunk 带 chunk.type='usage'
{inputTokens,outputTokens,cacheReadTokens,reasoningTokens}
从此算出真实指标（不再依赖 tokenMeter 的估算）：
缓存命中率 = cacheRead/(input+cacheRead) 实测 0.9976（与 App 状态栏 99% 一致）
总量 = output + input + cacheRead 实测 88.99M
tok/s = outputTokens/该步耗时 实测 37.3
上下文占用 = 最后一步的 input+cacheRead / 窗口（不是历史累加） 实测 56.3%
插件与镜像桥两侧都接上了；镜像桥改为跟随插件认定的当前会话。
另：autoOpen（默认开）——DSH 启动后 1.5s 若浮窗没在跑就自动拉起（用户选定）。
253 lines
10 KiB
JavaScript
253 lines
10 KiB
JavaScript
/**
|
||
* mirror-bridge.mjs — 独立镜像桥（不依赖 DSH 插件重启）
|
||
*
|
||
* 为什么不走插件：桌面壳（DSH Desktop）的 HTTP 网关对未认证请求一律 403，浮窗是独立进程拿不到
|
||
* cookie；而插件要重启 DSH 才会加载新版。所以这里直接读宿主自己的落盘数据：
|
||
*
|
||
* %DSH_HOME%/storages/session_projcache.json ← 每个会话的投影：标题 / 轮步 / 令牌 / 上下文压力
|
||
* %DSH_HOME%/sessions/<proj>/<sid>/session.jsonl.zstd ← 会话事件流（多帧 zstd）→ 消息行
|
||
*
|
||
* 产物：%DSH_HOME%/float-window/snapshot.json（与插件写的同一份 schema），浮窗每秒读它渲染。
|
||
* 只有当插件没在写（mtime 超过 3s）时才写，避免两边打架。
|
||
*
|
||
* 用法：node tools/mirror-bridge.mjs [--home <DSH_HOME>] [--interval 1200] [--once]
|
||
*/
|
||
import fs from 'node:fs'
|
||
import path from 'node:path'
|
||
import os from 'node:os'
|
||
import zlib from 'node:zlib'
|
||
|
||
const argv = process.argv.slice(2)
|
||
const arg = (name, def) => {
|
||
const i = argv.indexOf(name)
|
||
return i >= 0 && argv[i + 1] ? argv[i + 1] : def
|
||
}
|
||
const HOME = arg('--home', process.env.DSH_HOME || path.join(os.homedir(), '.dsh'))
|
||
const INTERVAL = Number(arg('--interval', '1200'))
|
||
const ONCE = argv.includes('--once')
|
||
const STATE_DIR = path.join(HOME, 'float-window')
|
||
const SNAPSHOT = path.join(STATE_DIR, 'snapshot.json')
|
||
const SNAPSHOT_BRIDGE = path.join(STATE_DIR, 'snapshot-bridge.json')
|
||
const WATCH = path.join(STATE_DIR, 'watch.json')
|
||
|
||
const log = (...a) => { if (!argv.includes('--quiet')) console.log('[bridge]', ...a) }
|
||
|
||
const ZSTD_MAGIC = Buffer.from([0x28, 0xb5, 0x2f, 0xfd])
|
||
|
||
/** DSH 往同一个 .zstd 文件追加多个帧，zstdDecompressSync 只吃第一帧 → 按魔数切开逐帧解 */
|
||
function decompressFrames(buf) {
|
||
let out = ''
|
||
let idx = buf.indexOf(ZSTD_MAGIC)
|
||
while (idx !== -1) {
|
||
const next = buf.indexOf(ZSTD_MAGIC, idx + 4)
|
||
const end = next === -1 ? buf.length : next
|
||
try { out += zlib.zstdDecompressSync(buf.subarray(idx, end)).toString('utf8') } catch { }
|
||
idx = next
|
||
}
|
||
return out
|
||
}
|
||
|
||
function readJson(file) {
|
||
try { return JSON.parse(fs.readFileSync(file, 'utf8')) } catch { return null }
|
||
}
|
||
|
||
function listSessionLogs() {
|
||
const root = path.join(HOME, 'sessions')
|
||
const out = []
|
||
if (!fs.existsSync(root)) return out
|
||
for (const proj of fs.readdirSync(root)) {
|
||
const pdir = path.join(root, proj)
|
||
let isDir = false
|
||
try { isDir = fs.statSync(pdir).isDirectory() } catch { }
|
||
if (!isDir) continue
|
||
for (const sid of fs.readdirSync(pdir)) {
|
||
const f = path.join(pdir, sid, 'session.jsonl.zstd')
|
||
try { out.push({ file: f, sid, mtime: fs.statSync(f).mtimeMs }) } catch { }
|
||
}
|
||
}
|
||
return out.sort((a, b) => b.mtime - a.mtime)
|
||
}
|
||
|
||
function textOf(content) {
|
||
if (typeof content === 'string') return content
|
||
if (Array.isArray(content)) return content.map((c) => (typeof c === 'string' ? c : (c?.text ?? ''))).join('')
|
||
return ''
|
||
}
|
||
const clip = (s, n = 4000) => (s.length > n ? s.slice(0, n) + '…' : s)
|
||
|
||
/** 事件流 → 浮窗消息行 + 实时指标（轮/步精确；token 用字符估算，标 ≈） */
|
||
function rowsFromLog(text) {
|
||
const rows = []
|
||
const titles = new Map()
|
||
let turns = 0
|
||
let steps = 0
|
||
let outChars = 0
|
||
let userChars = 0
|
||
let contextWindow = null
|
||
let streamStart = 0
|
||
let streamChars = 0
|
||
let lastChunkAt = 0
|
||
let outTokens = 0
|
||
let inTokens = 0
|
||
let cacheRead = 0
|
||
let stepStart = 0
|
||
let decodeMs = 0
|
||
let lastTps = null
|
||
let lastPromptTokens = 0 // 最后一步的输入(=未缓存+命中缓存)：这才是"当前上下文占用"
|
||
const est = (chars) => Math.round(chars / 3.2) // 中英混排的粗略 token 估算
|
||
|
||
for (const line of text.split('\n')) {
|
||
if (!line) continue
|
||
let e
|
||
try { e = JSON.parse(line) } catch { continue }
|
||
const type = String(e?.type || '')
|
||
const d = e?.data || e || {}
|
||
const t = e?.time || 0
|
||
if (type === 'turn/start') turns++
|
||
else if (type === 'step/start') { steps++; stepStart = t || Date.now() }
|
||
else if (type === 'request/context' && d.contextWindow) contextWindow = d.contextWindow
|
||
else if (type === 'session/title' && d.title) titles.set('title', String(d.title))
|
||
else if (type === 'user/message') {
|
||
if (d.source?.kind && d.source.kind !== 'user') continue
|
||
const s = textOf(d.content)
|
||
if (s) { userChars += s.length; rows.push({ role: 'user', text: clip(s) }) }
|
||
} else if (type === 'assistant/message') {
|
||
const s = textOf(d.content ?? d.message?.content)
|
||
if (s) { outChars += s.length; rows.push({ role: 'assistant', text: clip(s) }) }
|
||
} else if (type === 'assistant/chunk') {
|
||
const c = d.chunk || {}
|
||
const kind = String(c.type || '')
|
||
if (kind === 'usage' && c.usage) {
|
||
// 真实用量：{inputTokens,outputTokens,cacheReadTokens,reasoningTokens}
|
||
const u = c.usage
|
||
inTokens += u.inputTokens || 0
|
||
cacheRead += u.cacheReadTokens || 0
|
||
outTokens += u.outputTokens || 0
|
||
lastPromptTokens = (u.inputTokens || 0) + (u.cacheReadTokens || 0)
|
||
const ms = stepStart ? (t || Date.now()) - stepStart : 0
|
||
if (u.outputTokens && ms > 0) { decodeMs += ms; lastTps = u.outputTokens / (ms / 1000) }
|
||
continue
|
||
}
|
||
const piece = c.text ?? c.delta ?? (kind.includes('text') ? c.content : '')
|
||
if (piece) {
|
||
const now = t || Date.now()
|
||
if (!streamStart || now - lastChunkAt > 8000) { streamStart = now; streamChars = 0 }
|
||
lastChunkAt = now
|
||
streamChars += String(piece).length
|
||
outChars += String(piece).length
|
||
const last = rows[rows.length - 1]
|
||
if (last && last.role === 'assistant' && last.streaming) last.text += piece
|
||
else rows.push({ role: 'assistant', text: String(piece), streaming: true })
|
||
} else if (kind === 'finish' && c.reason?.kind === 'error') {
|
||
rows.push({ role: 'assistant', text: '⚠️ ' + String(c.reason.failure?.message || c.reason.failure || 'error').slice(0, 300) })
|
||
}
|
||
} else if (type === 'tool/call') {
|
||
rows.push({ role: 'tool', text: `${d.name || 'tool'}(${JSON.stringify(d.args ?? '').slice(0, 120)})` })
|
||
}
|
||
}
|
||
for (const r of rows) delete r.streaming
|
||
const streaming = lastChunkAt && Date.now() - lastChunkAt < 8000
|
||
// tok/s：优先用真实 outputTokens/步耗时；否则退化成"流式字符估算"
|
||
const tps = lastTps ?? (streaming && streamStart ? est(streamChars) / Math.max(0.5, (lastChunkAt - streamStart) / 1000) : null)
|
||
const promptTokens = inTokens + cacheRead
|
||
const realUsage = promptTokens + outTokens > 0
|
||
return {
|
||
rows, titles, turns, steps, streaming,
|
||
metrics: {
|
||
turns,
|
||
steps,
|
||
tps,
|
||
// 有真实用量就用真实值；否则退化成字符估算（界面会带 ≈）
|
||
tokens: realUsage ? outTokens + promptTokens : (outChars ? est(outChars + userChars) : null),
|
||
cacheHit: promptTokens > 0 ? cacheRead / promptTokens : null,
|
||
contextUsed: contextWindow ? Math.min(1, (realUsage ? lastPromptTokens : est(userChars + outChars)) / contextWindow) : null,
|
||
},
|
||
}
|
||
}
|
||
|
||
function buildSnapshot() {
|
||
const cache = readJson(path.join(HOME, 'storages', 'session_projcache.json'))
|
||
const table = cache?.tables?.sessions || {}
|
||
const watch = readJson(WATCH)
|
||
const logs = listSessionLogs()
|
||
|
||
// 会话清单：以**日志**为主（实时 mtime），投影缓存只用于补标题/历史统计
|
||
const entries = []
|
||
for (const l of logs) {
|
||
const rows = table[l.sid]?.rows || table[l.sid.replace(/^session-/, '')]?.rows || {}
|
||
const stats = rows.sessionStats?.val || {}
|
||
entries.push({
|
||
id: l.sid,
|
||
title: rows.title?.val || null,
|
||
cacheTurns: stats.turns ?? null,
|
||
cacheSteps: stats.steps ?? null,
|
||
mtime: l.mtime,
|
||
log: l,
|
||
})
|
||
}
|
||
|
||
// 当前会话：watch.json 指定 > 插件自己认定的当前会话 > 日志最新
|
||
let active = null
|
||
if (watch?.sessionId) active = entries.find((e) => e.id === watch.sessionId || e.id.replace(/^session-/, '') === watch.sessionId) || null
|
||
if (!active) {
|
||
// 插件写的 snapshot.json 带 sessionId：跟它走，指标缺口才能对上同一个会话
|
||
const plug = readJson(SNAPSHOT)
|
||
if (plug?.sessionId && !plug.source) {
|
||
active = entries.find((e) => e.id === plug.sessionId || e.id.replace(/^session-/, '') === plug.sessionId) || null
|
||
}
|
||
}
|
||
if (!active) active = entries.sort((a, b) => b.mtime - a.mtime)[0] || null
|
||
if (!active) return null
|
||
|
||
let rows = []
|
||
let title = active.title
|
||
let metrics = { turns: active.cacheTurns, steps: active.cacheSteps, tps: null, tokens: null, cacheHit: null, contextUsed: null }
|
||
let streaming = false
|
||
try {
|
||
const parsed = rowsFromLog(decompressFrames(fs.readFileSync(active.log.file)))
|
||
rows = parsed.rows.slice(-40)
|
||
title = title || parsed.titles.get('title') || null
|
||
streaming = parsed.streaming
|
||
// 轮/步取实时计数（缓存的那份是周期性落盘的，可能滞后）
|
||
metrics = {
|
||
...parsed.metrics,
|
||
turns: Math.max(parsed.metrics.turns, active.cacheTurns || 0),
|
||
steps: Math.max(parsed.metrics.steps, active.cacheSteps || 0),
|
||
}
|
||
} catch (e) { log('log read failed:', String(e?.message || e)) }
|
||
|
||
return {
|
||
ok: true,
|
||
at: Date.now(),
|
||
source: 'mirror-bridge',
|
||
approx: true, // token/上下文是字符估算，界面上会带 ≈
|
||
streaming,
|
||
sessionId: active.id,
|
||
pinned: watch?.sessionId || null,
|
||
title,
|
||
messages: rows,
|
||
metrics,
|
||
sessions: entries.slice(0, 20).map((e) => ({ id: e.id, title: e.title, turns: e.cacheTurns || 0, steps: e.cacheSteps || 0, messages: 0, active: e.id === active.id })),
|
||
}
|
||
}
|
||
|
||
function writeSnapshot() {
|
||
try {
|
||
const snap = buildSnapshot()
|
||
if (!snap) return
|
||
fs.mkdirSync(STATE_DIR, { recursive: true })
|
||
// 兜底文件总是写：插件缺的字段（token/上下文估算、标题）由浮窗用它补齐
|
||
fs.writeFileSync(SNAPSHOT_BRIDGE, JSON.stringify(snap))
|
||
// 插件在写就不抢 snapshot.json：mtime 3 秒内视为它的
|
||
try {
|
||
const st = fs.statSync(SNAPSHOT)
|
||
if (Date.now() - st.mtimeMs < 3000 && readJson(SNAPSHOT)?.source !== 'mirror-bridge') return
|
||
} catch { }
|
||
fs.writeFileSync(SNAPSHOT, JSON.stringify(snap))
|
||
log(`snapshot: ${snap.sessionId} "${(snap.title || '').slice(0, 24)}" turns=${snap.metrics.turns} steps=${snap.metrics.steps} rows=${snap.messages.length} ctx=${snap.metrics.contextUsed == null ? '-' : Math.round(snap.metrics.contextUsed * 100) + '%'}`)
|
||
} catch (e) { log('write failed:', String(e?.message || e)) }
|
||
}
|
||
|
||
writeSnapshot()
|
||
if (!ONCE) setInterval(writeSnapshot, INTERVAL)
|