fix(usage): Claude subagent output is no longer undercounted

apps/server/src/provider/Drivers/claudeUsage.ts

@@ -13,9 +13,9 @@ import {
 import * as HostProcess from "@t3tools/shared/HostProcess";
 
 /**
- * T3 Code writes one record per assistant *content block*, and every one of those records repeats
- * the same complete `usage` object for the parent message. Summing them overcounts by roughly
- * 2.4x on a real workload, so the caller must drop repeats by `dedupeKey` and keep the first.
+ * Claude Code repeats a message's `dedupeKey` for each assistant content block.
+ * Subagent blocks can progress from stream-start to final usage, so callers
+ * must count each key once using the maximum of each token channel.
  */
 export function parseClaudeLine(line: string): UsageRecord | null {
   let parsed: unknown;

apps/server/src/usage/UsageService.test.ts

@@ -61,6 +61,42 @@ function claudeLine(id: number, outputTokens: number, model = "claude-fable-5"):
   })}\n`;
 }
 
+/** One content-block line of message `msg_1`; Claude Code writes several per message. */
+function claudeCopy(usage: Readonly<Record<string, number>>): string {
+  return `${JSON.stringify({
+    type: "assistant",
+    timestamp: "2026-08-01T10:00:00Z",
+    requestId: "req_1",
+    sessionId: "session-1",
+    message: { id: "msg_1", model: "claude-fable-5", usage },
+  })}\n`;
+}
+
+/** The two lines of one subagent message from #20246: its stream-start copy, then its final copy. */
+const PARTIAL_USAGE = {
+  input_tokens: 2,
+  cache_creation_input_tokens: 1097,
+  cache_read_input_tokens: 67934,
+  output_tokens: 3,
+};
+const COMPLETE_USAGE = { ...PARTIAL_USAGE, output_tokens: 264 };
+const COPY_RATES = {
+  "claude-fable-5": {
+    input_cost_per_token: 1e-5,
+    output_cost_per_token: 5e-5,
+    cache_read_input_token_cost: 1e-6,
+    cache_creation_input_token_cost: 1.25e-5,
+  },
+};
+const COMPLETE_COST_USD = 2 * 1e-5 + 67934 * 1e-6 + 1097 * 1.25e-5 + 264 * 5e-5;
+const COMPLETE_TOTALS = {
+  uncachedInputTokens: 2,
+  cachedInputTokens: 67934,
+  cacheCreationTokens: 1097,
+  outputTokens: 264,
+  reasoningTokens: 0,
+};
+
 const WINDOW: UsageSummaryInput = {
   timeZone: "UTC",
   sinceDay: UsageDay.make("2026-07-31"),
@@ -1447,6 +1483,65 @@ describe("UsageService", () => {
     }).pipe(Effect.scoped),
   );
 
+  it.live(
+    "reconciles appended copies through tail replacement, newline completion and restart",
+    () =>
+      Effect.gen(function* () {
+        const { transcript, settings, home } = yield* setup;
+        const partial = claudeCopy(PARTIAL_USAGE);
+        const complete = claudeCopy(COMPLETE_USAGE);
+        yield* Effect.promise(() => NodeFSP.writeFile(transcript, partial));
+        yield* Effect.gen(function* () {
+          const first = yield* UsageService.make;
+          assert.strictEqual(totalOutputTokens(yield* first.readSummary(WINDOW)), 3);
+          yield* first.awaitPersisted;
+
+          yield* Effect.promise(() => NodeFSP.appendFile(transcript, complete.trimEnd()));
+          const resumed = yield* UsageService.make;
+          const tail = yield* resumed.readSummary(WINDOW);
+          assert.deepStrictEqual(tail.buckets[0]?.totals, COMPLETE_TOTALS);
+          assert.strictEqual(tail.buckets[0]?.records, 1);
+          assert.closeTo(tail.buckets[0]?.costUsd ?? 0, COMPLETE_COST_USD, 1e-12);
+          yield* resumed.awaitPersisted;
+          const restarted = yield* UsageService.make;
+          assert.deepStrictEqual((yield* restarted.readSummary(WINDOW)).buckets, tail.buckets);
+
+          // Grow the file without changing its settled prefix: the old tail must be discarded.
+          yield* Effect.promise(() =>
+            NodeFSP.writeFile(transcript, partial + claudeLine(2, 7) + claudeLine(3, 5)),
+          );
+          const replaced = yield* restarted.readSummary(WINDOW);
+          assert.strictEqual(totalOutputTokens(replaced), 3 + 7 + 5);
+          assert.strictEqual(replaced.buckets[0]?.records, 3);
+
+          yield* Effect.promise(() => NodeFSP.appendFile(transcript, complete.trimEnd()));
+          const grown = yield* restarted.readSummary(WINDOW);
+          assert.deepStrictEqual(grown.buckets[0]?.totals, {
+            ...COMPLETE_TOTALS,
+            uncachedInputTokens: 22,
+            outputTokens: 264 + 7 + 5,
+          });
+          assert.strictEqual(grown.buckets[0]?.records, 3);
+          yield* Effect.promise(() => NodeFSP.appendFile(transcript, "\n"));
+          assert.deepStrictEqual((yield* restarted.readSummary(WINDOW)).buckets, grown.buckets);
+          yield* restarted.awaitPersisted;
+          assert.deepStrictEqual(
+            (yield* (yield* UsageService.make).readSummary(WINDOW)).buckets,
+            grown.buckets,
+          );
+        }).pipe(
+          Effect.provide(
+            layerService({
+              prefix: "usage-service-copy-lifecycle-test",
+              home,
+              settings,
+              ratesDocument: COPY_RATES,
+            }),
+          ),
+        );
+      }).pipe(Effect.scoped),
+  );
+
   it.live(
     "keeps large-record totals and costs exact through append, dedupe, restart and cleanup",
     () =>
@@ -1540,7 +1635,7 @@ describe("UsageService", () => {
 
         yield* Effect.gen(function* () {
           const { stateDir } = yield* ServerConfig.ServerConfig;
-          const cachePath = NodePath.join(stateDir, "usage-scan-cache-v5.json");
+          const cachePath = NodePath.join(stateDir, "usage-scan-cache-v6.json");
           const legacyPath = NodePath.join(stateDir, "usage-scan-cache.json");
           const first = yield* UsageService.make;
           yield* first.readSummary(WINDOW);
@@ -1591,6 +1686,64 @@ describe("UsageService", () => {
       }).pipe(Effect.scoped),
   );
 
+  it.live(
+    "upgrades a v5 cache: re-parses live partial copies, keeps deleted transcripts, leaves v5 intact",
+    () =>
+      Effect.gen(function* () {
+        const { transcript, settings, home } = yield* setup;
+        const deleted = NodePath.join(NodePath.dirname(transcript), "deleted.jsonl");
+        yield* Effect.promise(async () => {
+          await NodeFSP.writeFile(
+            transcript,
+            claudeCopy(PARTIAL_USAGE) + claudeCopy(COMPLETE_USAGE),
+          );
+          await NodeFSP.writeFile(deleted, claudeLine(2, 7));
+        });
+
+        yield* Effect.gen(function* () {
+          const { stateDir } = yield* ServerConfig.ServerConfig;
+          const cachePath = NodePath.join(stateDir, "usage-scan-cache-v6.json");
+          const v5Path = NodePath.join(stateDir, "usage-scan-cache-v5.json");
+          const legacyPath = NodePath.join(stateDir, "usage-scan-cache.json");
+          const first = yield* UsageService.make;
+          yield* first.readSummary(WINDOW);
+          yield* first.awaitPersisted;
+
+          // A v5 server kept the first copy, so its row for msg_1 holds the partial count.
+          const v5 = yield* Effect.promise(async () => {
+            const document = decodeUnknownJsonString(await NodeFSP.readFile(cachePath, "utf8")) as {
+              files: Record<string, { r: unknown[][] }>;
+            };
+            const entry = document.files[transcript]!;
+            entry.r = entry.r.map((row) => [...row.slice(0, 6), 3, ...row.slice(7)]);
+            const text = encodeUnknownJsonString({ ...document, version: 5 });
+            await NodeFSP.writeFile(v5Path, text);
+            await NodeFSP.writeFile(
+              legacyPath,
+              encodeUnknownJsonString({ ...document, version: 4, files: {} }),
+            );
+            await NodeFSP.rm(cachePath);
+            await NodeFSP.rm(deleted);
+            return text;
+          });
+
+          const upgraded = yield* UsageService.make;
+          const summary = yield* upgraded.readSummary(WINDOW);
+          assert.strictEqual(totalOutputTokens(summary), 264 + 7);
+          assert.strictEqual(summary.buckets[0]?.records, 2);
+          yield* upgraded.awaitPersisted;
+          yield* Effect.promise(() => NodeFSP.rm(transcript));
+          assert.deepStrictEqual(
+            (yield* (yield* UsageService.make).readSummary(WINDOW)).buckets,
+            summary.buckets,
+          );
+          assert.strictEqual(yield* Effect.promise(() => NodeFSP.readFile(v5Path, "utf8")), v5);
+        }).pipe(
+          Effect.provide(layerService({ prefix: "usage-service-v5-upgrade-test", home, settings })),
+        );
+      }).pipe(Effect.scoped),
+  );
+
   it.live("preserves saved tokens, costs and sessions after transcript cleanup and restart", () =>
     Effect.gen(function* () {
       const { transcript, settings, home } = yield* setup;

apps/server/src/usage/UsageService.ts

@@ -50,8 +50,8 @@ import {
 import {
   decodeScanCache,
   dedupeWithinFile,
-  LEGACY_SCAN_CACHE_FILE_NAME,
   makeScanCacheWriter,
+  PREVIOUS_SCAN_CACHE_FILE_NAMES,
   pruneScanCache,
   SCAN_CACHE_FILE_NAME,
   type CachedFile,
@@ -204,7 +204,9 @@ export const make = Effect.gen(function* () {
 
   const ratesCachePath = path.join(config.stateDir, "usage-model-rates.json");
   const scanCachePath = path.join(config.stateDir, SCAN_CACHE_FILE_NAME);
-  const legacyScanCachePath = path.join(config.stateDir, LEGACY_SCAN_CACHE_FILE_NAME);
+  const previousScanCachePaths = PREVIOUS_SCAN_CACHE_FILE_NAMES.map((name) =>
+    path.join(config.stateDir, name),
+  );
   const writeCacheFile = (filePath: string, contents: string) =>
     writeFileStringAtomically({ filePath, contents }).pipe(
       Effect.provideService(FileSystem.FileSystem, fileSystem),
@@ -392,8 +394,9 @@ export const make = Effect.gen(function* () {
           Effect.catchCause(() => Effect.succeed(null)),
         );
       let document = yield* readDocument(scanCachePath);
-      if (document === null) {
-        document = yield* readDocument(legacyScanCachePath);
+      for (const previousPath of previousScanCachePaths) {
+        if (document !== null) break;
+        document = yield* readDocument(previousPath);
         cacheDirty = document !== null;
       }
       if (document === null) return;
@@ -449,7 +452,7 @@ export const make = Effect.gen(function* () {
   /**
    * Reads run concurrently, and the caller stores updates in walk order rather
    * than completion order: saved records of deleted transcripts aggregate in
-   * cache order, where the first copy of a duplicate wins.
+   * cache order, where the first copy of a duplicate decides its bucket.
    */
   const readFileRecords = (
     filePath: string,
@@ -499,9 +502,8 @@ export const make = Effect.gen(function* () {
         };
 
       const base = parsed.resumed && cached !== undefined ? cached.records : [];
-      const seen = new Set<string>();
-      const records = dedupeWithinFile([...base, ...parsed.records], seen);
-      const tailRecords = dedupeWithinFile(parsed.tailRecords, seen);
+      const records = dedupeWithinFile([...base, ...parsed.records]);
+      const tailRecords = parsed.tailRecords;
 
       return {
         records: tailRecords.length === 0 ? records : [...records, ...tailRecords],
@@ -618,7 +620,7 @@ export const make = Effect.gen(function* () {
 
     // Independent sources scan together. Transcript directories go one at a
     // time, so open files stay at `TRANSCRIPT_READ_CONCURRENCY`. The result
-    // keeps this order, since aggregation keeps the first copy of a duplicate.
+    // keeps this order, since the first copy decides a duplicate's bucket.
     const [transcripts, scanDirs] = yield* Effect.all(
       [Effect.forEach(dirs, (dir) => scanTranscriptDir(dir, windowStartMs)), scans],
       { concurrency: "unbounded" },

apps/server/src/usage/usageAggregation.test.ts

@@ -98,7 +98,7 @@ describe("UsageAggregator", () => {
     expect(bucket).not.toHaveProperty("ultrafastCostUsd");
   });
 
-  it("keeps only the first record for a repeated dedupe key", () => {
+  it("counts identical copies of a dedupe key once", () => {
     const result = aggregate([
       record({ dedupeKey: "msg_1:" }),
       record({ dedupeKey: "msg_1:" }),
@@ -111,6 +111,85 @@ describe("UsageAggregator", () => {
     expect(result.buckets[0]?.totals.outputTokens).toBe(50);
   });
 
+  it("merges partial and zeroed copies in either order without moving the first copy's bucket", () => {
+    const partial = record({
+      dedupeKey: "msg_1:",
+      speed: "fast",
+      totals: { ...record().totals, cachedInputTokens: 1200, outputTokens: 3, reasoningTokens: 5 },
+    });
+    const complete = record({
+      dedupeKey: "msg_1:",
+      model: "claude-fable-5[1m]",
+      timestampMs: Date.parse("2026-08-08T12:00:00Z"),
+      sessionId: "session-fork",
+      totals: {
+        uncachedInputTokens: 2,
+        cachedInputTokens: 1000,
+        cacheCreationTokens: 40,
+        outputTokens: 264,
+        reasoningTokens: 135,
+      },
+    });
+    const zero = {
+      ...partial,
+      totals: {
+        uncachedInputTokens: 0,
+        cachedInputTokens: 0,
+        cacheCreationTokens: 0,
+        outputTokens: 0,
+        reasoningTokens: 0,
+      },
+    };
+    const totals = { ...complete.totals, uncachedInputTokens: 100, cachedInputTokens: 1200 };
+
+    for (const copies of [
+      [zero, partial, complete, complete],
+      [complete, partial, zero, zero],
+    ]) {
+      const aggregator = new UsageAggregator({
+        timeZone: "UTC",
+        sinceDay: "2026-08-01",
+        untilDay: "2026-08-31",
+        rates,
+      });
+      copies.forEach((copy, index) => aggregator.add(copy, String(index) + ".jsonl"));
+      const result = aggregator.finish();
+      const expected = aggregate([{ ...copies[0]!, totals }]).buckets[0];
+
+      expect(result.duplicatesDropped).toBe(3);
+      expect(result.buckets).toEqual([{ ...expected, sourcePath: "0.jsonl" }]);
+    }
+  });
+
+  it("replaces cost-source counts and removes the last category breakdown when copies report a cost", () => {
+    const zero = record({
+      dedupeKey: "msg_1:",
+      totals: {
+        uncachedInputTokens: 0,
+        cachedInputTokens: 0,
+        cacheCreationTokens: 0,
+        outputTokens: 0,
+        reasoningTokens: 0,
+      },
+    });
+    const unpriced = record({ provider: "grok", model: "unknown", dedupeKey: "msg_2:" });
+    const result = aggregate([
+      zero,
+      unpriced,
+      { ...zero, reportedCostUsd: 1 },
+      { ...zero, reportedCostUsd: 2 },
+      { ...zero, reportedCostUsd: 1 },
+      { ...unpriced, reportedCostUsd: 3 },
+    ]);
+
+    expect(result.buckets).toEqual(
+      aggregate([
+        { ...zero, reportedCostUsd: 2 },
+        { ...unpriced, reportedCostUsd: 3 },
+      ]).buckets,
+    );
+  });
+
   it("still sums records that carry no dedupe key", () => {
     const result = aggregate([record(), record()]);
 
@@ -228,8 +307,12 @@ describe("UsageAggregator", () => {
     expect(result.buckets[0]?.costSource).toBe("providerReported");
   });
 
-  it("drops records outside the window", () => {
-    const result = aggregate([record({ timestampMs: Date.parse("2026-07-01T12:00:00Z") })]);
+  it("keeps a key outside the window when only its later copy falls inside", () => {
+    const result = aggregate([
+      record({ dedupeKey: "msg_1:", timestampMs: Date.parse("2026-07-01T12:00:00Z") }),
+      record({ dedupeKey: "msg_1:", totals: { ...record().totals, outputTokens: 500 } }),
+    ]);
+    expect(result.duplicatesDropped).toBe(1);
 
     expect(result.outOfWindow).toBe(1);
     expect(result.buckets).toHaveLength(0);