fix(core): keep usage queries off the engine event loop

docs/webui.md

@@ -64,6 +64,8 @@ World 留在原位照常运行，名称旁换成重启进程的入口;「待生
 根 URL 不带 hash 时跳转到 `live`；未知路由不显示页面内容，侧栏仍可导航。
 存储项按归属分页:Core 的在运行诊断的「数据」子页,一键清空也在那里;World、Persona 与 Memory 页
 各有「数据」页签,列本页声明的项。配置组同理:Core 的在运行诊断的「配置」子页,其余在声明方自己的页。
+用量账本在存储清单中显示文件大小。完整条数与费用按需在工作线程聚合，相同的并发查询共用一次读取，
+不同查询串行读取；引擎停止时取消未结束的查询。损坏行保留在账本中，并通过 `ledger.error` 报告。
 工具表是装配好的整份,只挂在 Persona 页的「工具表」页签上。
 World 页与 Persona 页的 manifest 带 `undeclared`:运行中出现、不在定义自描述里的工具名与事件 type。
 

scripts/publish-package.ts

@@ -27,6 +27,7 @@ export const PUBLISHED_DEPENDENCIES = [
   'react-dom',
   'react-textarea-autosize',
   'tar-stream',
+  'tsx',
   'ws',
 ] as const;
 

src/bot.ts

@@ -33,7 +33,6 @@ import { RESERVED_FRAME_NAMES } from './core/loop.ts';
 import type { ResponseClient } from './core/generation.ts';
 import { acquireInstanceLock, type InstanceLock } from './core/instance-lock.ts';
 import { assembleSystemSegments, envPromptOverridePath, envPromptTemplateSource, renderWorldEnvPrompt, type EnvPromptDirs, type EnvPromptOrigin } from './core/prefix.ts';
-import { aggregateUsage } from './core/cost.ts';
 import { createUsageRates } from './core/usage-rates.ts';
 import { DeadlineError, nowIso, withDeadline, readTextFile } from './core/util.ts';
 import { closeRun } from './core/run.ts';
@@ -250,7 +249,7 @@ function deriveStorage<C extends CoreConfig>(core: Core<C>, dataDir: string, lan
       kind: 'disk',
       location: 'data/usage.jsonl',
       note: s.usage.note,
-      stat: () => s.usage.stat(core.usageLog.count(), size(dataDir, 'usage.jsonl')),
+      stat: () => size(dataDir, 'usage.jsonl'),
       clear: () => s.usage.cleared(core.usageLog.clear()),
     },
     {
@@ -1043,7 +1042,7 @@ export function createBot<C extends CoreConfig>(
       usage: {
         aggregate: async (opts) => {
           const { rates, sources } = await liveRates.resolve();
-          return { ...aggregateUsage(core.usageLog.readAll(), { ...opts, timezone: core.config.timezone, rates }), rateSources: sources };
+          return { ...await core.usageLog.aggregate({ ...opts, timezone: core.config.timezone, rates }), rateSources: sources };
         },
         status: () => core.usageLog.status(),
       },

src/core/core.ts

@@ -765,6 +765,7 @@ export class Core<C extends CoreConfig = CoreConfig> {
     failures.push(...moduleFailures);
     // 运行中卸载的 World 各自的 stop() 有期限,结果已记日志
     await Promise.all(this.failedStops);
+    this.usageLog.stopQueries();
     this.usageLog.flush();
     const ledger = this.usageLog.status();
     if (ledger.pending) failures.push({ worldId: 'core.usage', detail: `${ledger.pending} usage records remain unwritten: ${ledger.error}` });

src/core/usage-log.ts

@@ -2,6 +2,8 @@
 import { appendFileSync, closeSync, existsSync, mkdirSync, openSync, readSync, writeFileSync } from 'node:fs';
 import { dirname } from 'node:path';
 import type { UsageRecord } from './types.ts';
+import type { UsageAggregate, UsageAggregateOptions } from './cost.ts';
+import { queryUsage } from './usage-query.ts';
 
 /** The ledger is read in chunks so its size is not bounded by V8's maximum string length. */
 const READ_CHUNK_BYTES = 4 * 1024 * 1024;
@@ -31,6 +33,9 @@ export class UsageLog {
   private pending: UsageRecord[] = [];
   private error: string | null = null;
   private reconcile = false;
+  private queryAbort = new AbortController();
+  private queryQueue: Promise<unknown> = Promise.resolve();
+  private queries = new Map<string, Promise<UsageAggregate>>();
   constructor(private readonly file: string, private readonly report: (message: string) => void = console.error) {}
 
   append(rec: UsageRecord): void {
@@ -83,15 +88,42 @@ export class UsageLog {
     return out;
   }
 
-  readAll(): UsageRecord[] {
+  readAll(options: { strict?: boolean } = {}): UsageRecord[] {
     let records: UsageRecord[] = [];
-    try { records = this.readDisk(); } catch (error) { this.note(error); }
+    try { records = this.readDisk(); } catch (error) {
+      this.note(error);
+      if (options.strict) throw error;
+    }
     const ids = new Set(records.map(record => record.recordId ?? record.attempt?.id));
     return [...records, ...this.pending.filter(record => !ids.has(record.recordId))];
   }
 
   count(): number { return this.readAll().length; }
 
+  /** Concurrent identical queries share one scan; different queries scan serially. */
+  aggregate(options: UsageAggregateOptions): Promise<UsageAggregate> {
+    const key = JSON.stringify(options);
+    const existing = this.queries.get(key);
+    if (existing) return existing;
+    const query = this.queryQueue.then(async () => {
+      const result = await queryUsage({ file: this.file, pending: structuredClone(this.pending), options }, this.queryAbort.signal);
+      if (result.error) {
+        if (result.error !== this.error) this.report(result.error);
+        this.error = result.error;
+      }
+      return result.aggregate;
+    }).catch(error => {
+      if (!this.queryAbort.signal.aborted) this.note(error);
+      throw error;
+    });
+    this.queries.set(key, query);
+    this.queryQueue = query.then(() => { this.queries.delete(key); }, () => { this.queries.delete(key); });
+    return query;
+  }
+
+  /** Pending WebUI scans must not delay engine shutdown. The ledger remains intact. */
+  stopQueries(): void { this.queryAbort.abort(new Error('Usage queries stopped')); }
+
   clear(): number {
     const count = this.count();
     mkdirSync(dirname(this.file), { recursive: true });

src/core/usage-query-worker.mjs

@@ -0,0 +1,22 @@
+import { parentPort, workerData } from 'node:worker_threads';
+import { tsImport } from 'tsx/esm/api';
+
+try {
+  const { UsageLog } = await tsImport('./usage-log.ts', import.meta.url);
+  const { aggregateUsage } = await tsImport('./cost.ts', import.meta.url);
+  const errors = [];
+  const log = new UsageLog(workerData.file, error => errors.push(error));
+  const records = log.readAll({ strict: true });
+  const ids = new Set(records.map(record => record.recordId ?? record.attempt?.id));
+  for (const record of workerData.pending) {
+    const id = record.recordId ?? record.attempt?.id;
+    if (id && ids.has(id)) continue;
+    if (id) ids.add(id);
+    records.push(record);
+  }
+  parentPort.postMessage({ aggregate: aggregateUsage(records, workerData.options), error: errors.at(-1) ?? null });
+} catch (error) {
+  parentPort.postMessage({ failure: String(error) });
+} finally {
+  parentPort.close();
+}

src/core/usage-query.ts

@@ -0,0 +1,35 @@
+/** Usage scans and aggregation run outside the engine event loop. */
+import { Worker } from 'node:worker_threads';
+import type { UsageRecord } from './types.ts';
+import type { UsageAggregate, UsageAggregateOptions } from './cost.ts';
+
+export interface UsageQueryInput {
+  file: string;
+  pending: UsageRecord[];
+  options: UsageAggregateOptions;
+}
+
+export interface UsageQueryResult {
+  aggregate: UsageAggregate;
+  error: string | null;
+}
+
+export function queryUsage(input: UsageQueryInput, signal: AbortSignal): Promise<UsageQueryResult> {
+  if (signal.aborted) return Promise.reject(signal.reason);
+  return new Promise((resolve, reject) => {
+    // The bootstrap installs its own TS loader; it must not inherit host/test preloads.
+    const worker = new Worker(new URL('./usage-query-worker.mjs', import.meta.url), { workerData: input, execArgv: [] });
+    const abort = () => { void worker.terminate(); reject(signal.reason); };
+    signal.addEventListener('abort', abort, { once: true });
+    worker.once('message', (message: UsageQueryResult | { failure: string }) => {
+      if ('failure' in message) reject(new Error(message.failure));
+      else resolve(message);
+    });
+    worker.once('error', reject);
+    worker.once('messageerror', reject);
+    worker.once('exit', code => {
+      signal.removeEventListener('abort', abort);
+      reject(new Error(`Usage query worker exited before returning a result (${code})`));
+    });
+  });
+}

tests/core/usage-query.test.ts

@@ -0,0 +1,82 @@
+import { afterEach, beforeEach, describe, expect, it } from 'vitest';
+import { appendFileSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { UsageLog } from '../../src/core/usage-log.ts';
+import { aggregateUsage } from '../../src/core/cost.ts';
+import type { UsageRecord } from '../../src/core/types.ts';
+
+let dir: string;
+let file: string;
+const logs: UsageLog[] = [];
+const row = (id: string): UsageRecord => ({ version: 2, recordId: id, ts: '2026-01-02T12:00:00Z',
+  sessionId: 'test', role: 'main', label: '测试', model: 'test-model', promptTokens: 100,
+  completionTokens: 10, cacheHitTokens: 50, cacheMissTokens: 50, reasoningTokens: 3 });
+const options = { bucket: 'day' as const, now: new Date('2026-01-02T13:00:00Z') };
+function log(path = file, report = (_message: string) => {}) {
+  const value = new UsageLog(path, report); logs.push(value); return value;
+}
+beforeEach(() => { dir = mkdtempSync(join(tmpdir(), 'usage-query-')); file = join(dir, 'usage.jsonl'); });
+afterEach(() => { for (const value of logs.splice(0)) value.stopQueries(); rmSync(dir, { recursive: true, force: true }); });
+
+describe('UsageLog.aggregate', () => {
+  it('matches the existing accounting, preserves corrupt lines and deduplicates record ids', async () => {
+    const records = [row('a'), row('b')];
+    const raw = records.map(record => JSON.stringify(record)).join('\n') + '\nbroken\n' + JSON.stringify(records[0]) + '\n';
+    writeFileSync(file, raw);
+    const errors: string[] = [];
+    const usage = log(file, error => errors.push(error));
+    expect(await usage.aggregate(options)).toEqual(aggregateUsage(records, options));
+    expect(errors).toHaveLength(1);
+    expect(errors[0]).toContain('1 damaged ledger lines');
+    expect(readFileSync(file, 'utf8')).toBe(raw);
+  });
+
+  it('leaves the main event loop available while a large query is pending', async () => {
+    const record = JSON.stringify({ ...row('large'), label: '资料'.repeat(1200) });
+    writeFileSync(file, (record + '\n').repeat(2200));
+    const usage = log();
+    let settled = false;
+    const query = usage.aggregate(options).finally(() => { settled = true; });
+    await new Promise<void>(resolve => setImmediate(resolve));
+    expect(settled).toBe(false);
+    usage.append(row('new'));
+    expect((await query).totals.calls).toBeGreaterThanOrEqual(1);
+    expect((await usage.aggregate(options)).totals.calls).toBe(2);
+  });
+
+  it('shares simultaneous identical scans and refreshes after an append', async () => {
+    const usage = log(); usage.append(row('a'));
+    const first = usage.aggregate(options);
+    expect(usage.aggregate(options)).toBe(first);
+    expect((await first).totals.calls).toBe(1);
+    usage.append(row('b'));
+    expect((await usage.aggregate(options)).totals.calls).toBe(2);
+  });
+
+  it('includes queued appends and does not count an already persisted retry twice', async () => {
+    const parent = join(dir, 'blocked'); writeFileSync(parent, 'occupied');
+    const target = join(parent, 'usage.jsonl');
+    const usage = log(target); usage.append(row('a')); usage.append(row('b'));
+    rmSync(parent); mkdirSync(parent);
+    appendFileSync(target, JSON.stringify(row('a')) + '\n');
+    expect((await usage.aggregate(options)).totals.calls).toBe(2);
+    expect(usage.status().pending).toBe(2);
+  });
+
+  it('reports read failures instead of returning an empty successful report', async () => {
+    mkdirSync(file);
+    await expect(log().aggregate(options)).rejects.toThrow();
+  });
+
+  it('cancels both active and queued scans when the engine stops', async () => {
+    const usage = log(); usage.append(row('a'));
+    const first = usage.aggregate(options);
+    const second = usage.aggregate({ ...options, bucket: 'hour' });
+    const results = Promise.allSettled([first, second]);
+    await new Promise<void>(resolve => setImmediate(resolve));
+    usage.stopQueries();
+    expect((await results).map(result => result.status)).toEqual(['rejected', 'rejected']);
+    expect(readFileSync(file, 'utf8')).toContain('"recordId":"a"');
+  });
+});

tests/web/persona-usage.test.ts

@@ -4,7 +4,6 @@ import { tmpdir } from 'node:os';
 import { join } from 'node:path';
 import { WebApp } from '../../src/web/server.ts';
 import { UsageLog } from '../../src/core/usage-log.ts';
-import { aggregateUsage } from '../../src/core/cost.ts';
 import { nullLogger } from '../../src/core/util.ts';
 import type { UsageRecord } from '../../src/core/types.ts';
 import type { BotConfig } from '../../bots/corti-soulmate/assemble.ts';
@@ -54,7 +53,7 @@ beforeAll(async () => {
     memoryDir,
     dataDir,
     getStatus: () => ({}),
-    usage: { aggregate: (opts) => aggregateUsage(usage.readAll(), opts) },
+    usage: { aggregate: (opts) => usage.aggregate(opts) },
     config: {
       groups: () => configGroups.map((group) => ({ group, values: readGroupValues(cfg, group) })),
       set: (groupId, values) => {