[METRICS-M3] Prove prompt request session and PR usage attribution

apps/server/src/usage/UsageService.ts

@@ -61,7 +61,7 @@ import {
   pruneScanCache,
   type ScanCache,
 } from "./usageScanCache.ts";
-import type { UsageRecord } from "./usageTranscripts.ts";
+import { usageEventOccurrenceBaseKey, type UsageRecord } from "./usageTranscripts.ts";
 
 const LITELLM_RATES_URL =
   "https://raw.githubusercontent.com/BerriAI/litellm/main/model_prices_and_context_window.json";
@@ -101,7 +101,6 @@ const encodeRatesCache = Schema.encodeEffect(
 const ScanCacheJson = Schema.fromJsonString(Schema.Unknown as unknown as Schema.Codec<unknown>);
 const decodeScanCacheFile = Schema.decodeUnknownEffect(ScanCacheJson);
 const encodeScanCacheFile = Schema.encodeEffect(ScanCacheJson);
-const encodeUsageRecordKey = Schema.encodeSync(ScanCacheJson);
 const CachedSource = Schema.Struct({ dir: Schema.String, volumeId: Schema.String });
 const decodeCachedSources = Schema.decodeUnknownOption(
   Schema.Struct({ sources: Schema.Record(Schema.String, CachedSource) }),
@@ -394,8 +393,14 @@ export const make = Effect.gen(function* () {
       const cached = fileCache.get(filePath);
       // Provider is part of the identity: if both providers were ever pointed
       // at one directory, a hit parsed by the other parser must not be reused.
+      // The cache format is part of it too: a legacy entry erased native ids and
+      // measurement presence, and a predecessor v4 entry never asserted numeric
+      // completeness. An unchanged file must still cold re-parse once to enrich
+      // either, instead of serving the erased or unasserted row forever.
       if (
         cached &&
+        cached.identity === "declared" &&
+        cached.qualityMetadata === "declared" &&
         cached.size === size &&
         cached.mtimeMs === mtimeMs &&
         cached.provider === provider
@@ -406,9 +411,16 @@ export const make = Effect.gen(function* () {
       }
 
       // Only a strictly grown file may resume. Same size with a new mtime, or
-      // a shrunken file, means rewritten content; re-parse it whole.
+      // a shrunken file, means rewritten content; re-parse it whole. A legacy
+      // or predecessor entry is also re-parsed whole: resuming would keep
+      // serving id-less or completeness-less records and the enrichment would
+      // never happen.
       const resumeFrom =
-        cached !== undefined && cached.provider === provider && size > cached.size
+        cached !== undefined &&
+        cached.provider === provider &&
+        cached.identity === "declared" &&
+        cached.qualityMetadata === "declared" &&
+        size > cached.size
           ? cached.position
           : undefined;
 
@@ -436,6 +448,8 @@ export const make = Effect.gen(function* () {
         records,
         tailRecords,
         position: parsed.position,
+        identity: "declared",
+        qualityMetadata: "declared",
       });
       cacheDirty = true;
       return tailRecords.length === 0 ? records : [...records, ...tailRecords];
@@ -572,6 +586,11 @@ export const make = Effect.gen(function* () {
       }
       let scannedFiles = 0;
       let skippedFiles = 0;
+      // A usage container with no recognised token field (Claude `usage: {}`)
+      // or one whose fields are all present-but-invalid parses to a record but
+      // measured nothing; surface it rather than letting it read as a measured
+      // zero.
+      let malformedRecords = 0;
       // Distinct per directory. Buckets carry per-cell session counts, but a
       // session spans days and models, so clients total this figure instead.
       const sessionIds = new Set<string>();
@@ -584,17 +603,13 @@ export const make = Effect.gen(function* () {
         scannedFiles += 1;
         const codexEventOccurrences = new Map<string, number>();
         for (const record of file.records) {
+          if (record.measurement === "empty" || record.measurement === "invalid")
+            malformedRecords += 1;
           let usageRecord = record;
           if (record.provider === "codex" && record.sessionId.length > 0) {
             // Match moved rollout copies without collapsing repeated equal events
             // within one rollout (timestamps can have only second precision).
-            const key = encodeUsageRecordKey([
-              record.provider,
-              record.sessionId,
-              record.timestampMs,
-              record.model,
-              record.totals,
-            ]);
+            const key = usageEventOccurrenceBaseKey(record);
             const occurrence = (codexEventOccurrences.get(key) ?? 0) + 1;
             codexEventOccurrences.set(key, occurrence);
             usageRecord = { ...record, dedupeKey: key + ":" + occurrence };
@@ -613,7 +628,7 @@ export const make = Effect.gen(function* () {
         status: files === null && scannedFiles === 0 ? "missing" : "ok",
         scannedFiles,
         skippedFiles,
-        malformedRecords: 0,
+        malformedRecords,
         distinctSessions: sessionIds.size,
         message: files === null ? "No transcript directory on this environment." : null,
       });