fix(usage): 修正缓存输入重复计数与旧基线迁移

src/core/cost-calculator.ts

@@ -66,6 +66,34 @@ function pickNum(obj: any, keys: readonly string[]): number {
   return 0;
 }
 
+interface PartitionedInputTokens {
+  rawInputTokens: number;
+  inputTokens: number;
+  cacheReadTokens: number;
+  cacheCreateTokens: number;
+}
+
+/** Partition a provider's cache-inclusive input total into mutually exclusive
+ *  accounting buckets. Cache read consumes the raw total first, then cache
+ *  creation consumes only the remainder; malformed counters cannot make the
+ *  three buckets exceed the provider's raw input total. */
+function partitionInclusiveInputTokens(
+  rawInput: number,
+  reportedCacheRead: number,
+  reportedCacheCreate: number,
+): PartitionedInputTokens {
+  const rawInputTokens = Math.max(0, rawInput);
+  const cacheReadTokens = Math.min(rawInputTokens, Math.max(0, reportedCacheRead));
+  const afterCacheRead = rawInputTokens - cacheReadTokens;
+  const cacheCreateTokens = Math.min(afterCacheRead, Math.max(0, reportedCacheCreate));
+  return {
+    rawInputTokens,
+    inputTokens: afterCacheRead - cacheCreateTokens,
+    cacheReadTokens,
+    cacheCreateTokens,
+  };
+}
+
 function extractNativeUsage(entry: any): { usage: any; model?: string } | null {
   const candidates = [
     { usage: entry?.message?.usage, model: entry?.message?.model },
@@ -91,16 +119,23 @@ function extractCodexTokenCountUsage(entry: any): SessionTokenUsage | null {
   if (entry?.type !== 'event_msg' || entry?.payload?.type !== 'token_count') return null;
   const u = entry.payload?.info?.total_token_usage;
   if (!u || typeof u !== 'object') return null;
-  const inputTokens = pickNum(u, ['input_tokens', 'inputTokens']);
+  // Codex-compatible token_count snapshots define input_tokens as the whole
+  // prompt-side total: cached tokens are a subset, not an additional bucket.
+  // Keep the raw total in `in` for the dashboard, while the accounting fields
+  // are mutually exclusive so ledger consumers can safely sum them.
   const outputTokens = pickNum(u, ['output_tokens', 'outputTokens']);
-  const cacheReadTokens = pickNum(u, ['cached_input_tokens', 'cachedInputTokens']);
+  const { rawInputTokens, inputTokens, cacheReadTokens, cacheCreateTokens } = partitionInclusiveInputTokens(
+    pickNum(u, ['input_tokens', 'inputTokens']),
+    pickNum(u, ['cached_input_tokens', 'cachedInputTokens', 'cache_read_input_tokens', 'cacheReadInputTokens']),
+    pickNum(u, ['cache_creation_input_tokens', 'cacheCreationInputTokens', 'cache_write_input_tokens', 'cacheWriteInputTokens']),
+  );
   return {
-    in: inputTokens,
+    in: rawInputTokens,
     out: outputTokens,
     inputTokens,
     outputTokens,
     cacheReadTokens,
-    cacheCreateTokens: 0,
+    cacheCreateTokens,
     model: '',
     turns: 0,
   };
@@ -467,6 +502,7 @@ export function readSessionTokenUsageFile(path: string, kind: CachedUsageKind, o
 }
 
 function readTokenUsageFromAidenCheckpoint(path: string): SessionTokenUsage | null {
+  let rawInputTokens = 0;
   let inputTokens = 0;
   let outputTokens = 0;
   let cacheReadTokens = 0;
@@ -484,16 +520,20 @@ function readTokenUsageFromAidenCheckpoint(path: string): SessionTokenUsage | nu
       if (msg?.type === 'human' || msg?.type === 'tool') continue;
       const u = msg?.usage_metadata ?? msg?.usage;
       if (!u || typeof u !== 'object') continue;
-      const input = pickNum(u, ['input_tokens', 'inputTokens', 'prompt_tokens', 'promptTokens']);
+      const rawInput = pickNum(u, ['input_tokens', 'inputTokens', 'prompt_tokens', 'promptTokens']);
       const output = pickNum(u, ['output_tokens', 'outputTokens', 'completion_tokens', 'completionTokens']);
-      inputTokens += input;
-      outputTokens += output;
-      cacheReadTokens +=
+      const reportedCacheRead =
         pickNum(u?.input_token_details, ['cache_read', 'cached_tokens', 'cacheRead']) +
         pickNum(u?.input_tokens_details, ['cache_read', 'cached_tokens', 'cacheRead']);
-      cacheCreateTokens +=
+      const reportedCacheCreate =
         pickNum(u?.input_token_details, ['cache_creation', 'cache_write', 'cacheCreate']) +
         pickNum(u?.input_tokens_details, ['cache_creation', 'cache_write', 'cacheCreate']);
+      const partitioned = partitionInclusiveInputTokens(rawInput, reportedCacheRead, reportedCacheCreate);
+      rawInputTokens += partitioned.rawInputTokens;
+      inputTokens += partitioned.inputTokens;
+      outputTokens += output;
+      cacheReadTokens += partitioned.cacheReadTokens;
+      cacheCreateTokens += partitioned.cacheCreateTokens;
       if (!model && typeof msg?.response_metadata?.model_name === 'string') model = msg.response_metadata.model_name;
       turns++;
     }
@@ -504,7 +544,7 @@ function readTokenUsageFromAidenCheckpoint(path: string): SessionTokenUsage | nu
 
   if (turns === 0) return null;
   return {
-    in: inputTokens,
+    in: rawInputTokens,
     out: outputTokens,
     inputTokens,
     outputTokens,

src/services/usage-ledger.ts

@@ -23,8 +23,12 @@ import { logger } from '../utils/logger.js';
 import { getSessionTokenUsage, type SessionTokenUsage } from '../core/cost-calculator.js';
 import type { DaemonSession } from '../core/types.js';
 
+export type InputTokenSemantics = 'includes_cache' | 'uncached';
+
 export interface UsageLedgerRecord {
-  v: 1;
+  v: 2;
+  /** `inputTokens` / `totalInputTokens` exclude both cache buckets. */
+  inputTokenSemantics: 'uncached';
   /** 'ownership' marks a zero-delta marker written at session spawn so
    *  consumers can exclude the session from native parsers BEFORE its first
    *  positive delta lands; absent for normal usage records. */
@@ -78,14 +82,17 @@ interface SessionBaseline {
   outputTokens: number;
   cacheReadTokens: number;
   cacheCreateTokens: number;
+  /** Missing on v1 state/ledger baselines. For Codex/TraeX/Aiden those legacy
+   *  baselines used includes_cache prompt input and must be migrated once. */
+  inputTokenSemantics?: InputTokenSemantics;
   recordedAt: string;
   /** Bumped on every baseline reset (shrink, explicit anchor) so identical
    *  totals transitions in different epochs get distinct recordIds. */
   epoch?: number;
 }
 
 interface LedgerState {
-  v: 1;
+  v: 2;
   sessions: { [sessionId: string]: SessionBaseline };
 }
 
@@ -110,6 +117,43 @@ function finiteNum(v: unknown): number {
   return typeof v === 'number' && Number.isFinite(v) ? v : 0;
 }
 
+function inputTokenSemantics(v: unknown): InputTokenSemantics | undefined {
+  return v === 'includes_cache' || v === 'uncached' ? v : undefined;
+}
+
+/** Convert any persisted baseline into the producer's current mutually
+ *  exclusive accounting contract before comparing candidates or diffing.
+ *  v1 Codex/TraeX/Aiden baselines had no marker and stored raw includes_cache input;
+ *  other v1 producers already stored uncached input and remain unchanged. */
+function normalizeBaseline(
+  baseline: SessionBaseline | undefined | null,
+  cliId: string | undefined,
+): SessionBaseline | undefined {
+  if (!baseline) return undefined;
+  const semantics = inputTokenSemantics(baseline.inputTokenSemantics);
+  const semanticsPresent = Object.prototype.hasOwnProperty.call(baseline, 'inputTokenSemantics');
+  const includedCache = semantics === 'includes_cache'
+    || (!semanticsPresent && (cliId === 'codex' || cliId === 'traex' || cliId === 'aiden'));
+  if (!includedCache) {
+    return { ...baseline, inputTokenSemantics: 'uncached' };
+  }
+
+  // Mirror the native parser's bounded partitioning. Persisted legacy cache
+  // counters can be inconsistent; each bucket is capped by the remaining raw
+  // prompt total so the migrated buckets still sum exactly to that total.
+  const rawInputTokens = Math.max(0, baseline.inputTokens);
+  const cacheReadTokens = Math.min(rawInputTokens, Math.max(0, baseline.cacheReadTokens));
+  const afterCacheRead = rawInputTokens - cacheReadTokens;
+  const cacheCreateTokens = Math.min(afterCacheRead, Math.max(0, baseline.cacheCreateTokens));
+  return {
+    ...baseline,
+    inputTokens: afterCacheRead - cacheCreateTokens,
+    cacheReadTokens,
+    cacheCreateTokens,
+    inputTokenSemantics: 'uncached',
+  };
+}
+
 /** Reconstruct the newest baseline for a session from the ledger files
  *  themselves (newest file first; last matching line in a file is newest).
  *  This is the crash-recovery source of truth: a record that reached the
@@ -142,11 +186,13 @@ function baselineFromLedger(dir: string, sessionId: string): SessionBaseline | n
       } catch { /* skip malformed lines */ }
     }
     if (latest) {
+      const semanticsPresent = Object.prototype.hasOwnProperty.call(latest, 'inputTokenSemantics');
       return {
         inputTokens: finiteNum(latest.totalInputTokens),
         outputTokens: finiteNum(latest.totalOutputTokens),
         cacheReadTokens: finiteNum(latest.totalCacheReadTokens),
         cacheCreateTokens: finiteNum(latest.totalCacheCreateTokens),
+        ...(semanticsPresent ? { inputTokenSemantics: latest.inputTokenSemantics } : {}),
         recordedAt: typeof latest.ts === 'string' ? latest.ts : new Date(0).toISOString(),
         epoch: finiteNum(latest.epoch),
       };
@@ -173,15 +219,16 @@ function resolveBaseline(
   dir: string,
   larkAppId: string | undefined,
   sessionId: string,
+  cliId: string | undefined,
   stateBaseline: SessionBaseline | undefined,
 ): SessionBaseline | undefined {
   const key = baselineMemoryKey(larkAppId, sessionId);
   let remembered = sessionBaselineMemory.get(key);
   if (remembered === undefined) {
-    remembered = baselineFromLedger(dir, sessionId);
+    remembered = normalizeBaseline(baselineFromLedger(dir, sessionId), cliId) ?? null;
     sessionBaselineMemory.set(key, remembered);
   }
-  return newerBaseline(stateBaseline, remembered);
+  return newerBaseline(normalizeBaseline(stateBaseline, cliId), normalizeBaseline(remembered, cliId));
 }
 
 /** Baselines for sessions idle longer than this are pruned from state.json. */
@@ -203,10 +250,10 @@ function loadState(dir: string, larkAppId?: string): LedgerState {
   try {
     const parsed = JSON.parse(readFileSync(statePath(dir, larkAppId), 'utf8'));
     if (parsed && typeof parsed === 'object' && parsed.sessions && typeof parsed.sessions === 'object') {
-      return { v: 1, sessions: parsed.sessions };
+      return { v: 2, sessions: parsed.sessions };
     }
   } catch { /* first run or corrupt state — start fresh */ }
-  return { v: 1, sessions: {} };
+  return { v: 2, sessions: {} };
 }
 
 function saveState(dir: string, larkAppId: string | undefined, state: LedgerState, now: Date): void {
@@ -267,7 +314,7 @@ export function recordSessionUsage(args: RecordSessionUsageArgs): UsageLedgerRec
     mkdirSync(dir, { recursive: true });
 
     const state = loadState(dir, args.larkAppId);
-    const prev = resolveBaseline(dir, args.larkAppId, args.sessionId, state.sessions[args.sessionId]);
+    const prev = resolveBaseline(dir, args.larkAppId, args.sessionId, args.cliId, state.sessions[args.sessionId]);
     const cur = args.usage;
     const prevEpoch = prev?.epoch ?? 0;
 
@@ -281,6 +328,7 @@ export function recordSessionUsage(args: RecordSessionUsageArgs): UsageLedgerRec
       outputTokens: cur.outputTokens,
       cacheReadTokens: cur.cacheReadTokens,
       cacheCreateTokens: cur.cacheCreateTokens,
+      inputTokenSemantics: 'uncached',
       recordedAt: now.toISOString(),
       epoch: prevEpoch,
     };
@@ -300,7 +348,8 @@ export function recordSessionUsage(args: RecordSessionUsageArgs): UsageLedgerRec
     }
 
     const record: UsageLedgerRecord = {
-      v: 1,
+      v: 2,
+      inputTokenSemantics: 'uncached',
       recordId: deterministicRecordId(args.sessionId, prevEpoch, prev, cur),
       ts: now.toISOString(),
       epoch: prevEpoch,
@@ -353,12 +402,13 @@ export function anchorSessionUsage(args: RecordSessionUsageArgs): void {
     mkdirSync(dir, { recursive: true });
 
     const state = loadState(dir, args.larkAppId);
-    const prev = resolveBaseline(dir, args.larkAppId, args.sessionId, state.sessions[args.sessionId]);
+    const prev = resolveBaseline(dir, args.larkAppId, args.sessionId, args.cliId, state.sessions[args.sessionId]);
     const baseline: SessionBaseline = {
       inputTokens: args.usage.inputTokens,
       outputTokens: args.usage.outputTokens,
       cacheReadTokens: args.usage.cacheReadTokens,
       cacheCreateTokens: args.usage.cacheCreateTokens,
+      inputTokenSemantics: 'uncached',
       recordedAt: now.toISOString(),
       // Anchors start a new epoch: transitions after a re-anchor must never
       // collide with recordIds from before it.
@@ -464,7 +514,8 @@ export function recordSessionOwnership(args: RecordSessionOwnershipArgs): UsageL
     mkdirSync(dir, { recursive: true });
 
     const record: UsageLedgerRecord = {
-      v: 1,
+      v: 2,
+      inputTokenSemantics: 'uncached',
       kind: 'ownership',
       recordId,
       ts: now.toISOString(),
@@ -530,7 +581,7 @@ export function reconcileUsageForDaemonSession(ds: DaemonSession, opts?: DaemonS
     if (!args.usage) return null;
     const dir = opts?.ledgerDir ?? defaultLedgerDir();
     const state = loadState(dir, args.larkAppId);
-    if (resolveBaseline(dir, args.larkAppId, args.sessionId, state.sessions[args.sessionId])) {
+    if (resolveBaseline(dir, args.larkAppId, args.sessionId, args.cliId, state.sessions[args.sessionId])) {
       return recordSessionUsage({ ...args, usage: args.usage, ...opts });
     }
     anchorSessionUsage({ ...args, usage: args.usage, ...opts });

test/cost-calculator-cache.test.ts

@@ -41,10 +41,13 @@ function claudeLine(id: string | null, input: number, output: number): string {
   });
 }
 
-function codexCountLine(input: number, output: number): string {
+function codexCountLine(input: number, output: number, cacheRead = 0): string {
   return JSON.stringify({
     type: 'event_msg',
-    payload: { type: 'token_count', info: { total_token_usage: { input_tokens: input, output_tokens: output } } },
+    payload: {
+      type: 'token_count',
+      info: { total_token_usage: { input_tokens: input, output_tokens: output, cached_input_tokens: cacheRead } },
+    },
   });
 }
 
@@ -177,13 +180,23 @@ describe('readSessionTokenUsageFile caching', () => {
 
   it('keeps codex cumulative semantics across incremental reads', () => {
     const p = join(dir, 'rollout.jsonl');
-    writeFileSync(p, `${codexCountLine(100, 20)}\n`);
-    expect(readSessionTokenUsageFile(p, 'codex')).toMatchObject({ in: 100, out: 20 });
+    writeFileSync(p, `${codexCountLine(100, 20, 40)}\n`);
+    expect(readSessionTokenUsageFile(p, 'codex')).toMatchObject({
+      in: 100,
+      inputTokens: 60,
+      cacheReadTokens: 40,
+      out: 20,
+    });
 
     now += 20_000;
-    appendFileSync(p, `${codexCountLine(150, 30)}\n`);
+    appendFileSync(p, `${codexCountLine(150, 30, 60)}\n`);
     // Latest cumulative snapshot wins — not 100+150.
-    expect(readSessionTokenUsageFile(p, 'codex')).toMatchObject({ in: 150, out: 30 });
+    expect(readSessionTokenUsageFile(p, 'codex')).toMatchObject({
+      in: 150,
+      inputTokens: 90,
+      cacheReadTokens: 60,
+      out: 30,
+    });
   });
 
   it('skips oversized transcripts instead of scanning them from byte zero', () => {