fix(codex): read Zstandard-compressed rollouts

src/codex-fork-replay.ts

@@ -2,7 +2,7 @@ import { stat } from 'fs/promises'
 import { basename, dirname } from 'path'
 
 import { readSessionLines } from './fs-utils.js'
-import { walkRolloutFiles } from './launcher-homes.js'
+import { rolloutStem, walkRolloutFiles } from './launcher-homes.js'
 
 /** Maximum timestamp gap between copied records in Codex's fork replay burst. */
 export const CODEX_FORK_REPLAY_MAX_GAP_MS = 1000
@@ -84,15 +84,15 @@ export function codexReplayUsageIdentity(info: { total_token_usage?: CodexUsageC
 
 export const codexReplayResponseIdentity = (responseId: string): string => `response:${responseId}`
 
-// Rollout names are `rollout-YYYY-MM-DDTHH-MM-SS-<id>.jsonl`.
+// Rollout names are `rollout-YYYY-MM-DDTHH-MM-SS-<id>.jsonl` (or `.jsonl.zst`).
 const ROLLOUT_ID_OFFSET = 'rollout-YYYY-MM-DDTHH-MM-SS-'.length
 const rolloutIndexes = new Map<string, Map<string, string | null>>()
 
 function findRollout(home: string, id: string): string | undefined {
   let index = rolloutIndexes.get(home)
   if (!index?.has(id)) {
     index = new Map()
-    walkRolloutFiles(home, path => index!.set(basename(path).slice(ROLLOUT_ID_OFFSET, -'.jsonl'.length), path))
+    walkRolloutFiles(home, path => index!.set(rolloutStem(basename(path)).slice(ROLLOUT_ID_OFFSET), path))
     if (!index.has(id)) index.set(id, null)
     rolloutIndexes.set(home, index)
   }

src/codex-throughput.ts

@@ -1,6 +1,7 @@
 import { open, stat } from 'node:fs/promises'
 import { StringDecoder } from 'node:string_decoder'
 
+import { createTextReadStream } from './fs-utils.js'
 import { billableOutputTokens } from './models.js'
 import { isCodexForkReplay, startCodexForkReplay, type CodexForkReplayState } from './codex-fork-replay.js'
 
@@ -448,10 +449,55 @@ export class CodexThroughputReader {
     if (state.points.length > 10000) state.points.splice(0, state.points.length - 10000)
   }
 
+  private consume(buffer: Buffer): void {
+    let chunk = this.decoder.write(buffer)
+    while (chunk.length > 0) {
+      const newlineIndex = chunk.search(/\r?\n/)
+      const segment = newlineIndex >= 0 ? chunk.slice(0, newlineIndex) : chunk
+      this.pending += segment
+      this.scanDurationSegment(segment)
+      if (newlineIndex < 0) break
+      const newlineLength = chunk[newlineIndex] === '\r' ? 2 : 1
+      const line = this.pending
+      const durationOverride = this.pendingDurationMs
+      this.pending = ''
+      this.pendingDurationMs = undefined
+      this.scanDepth = 0
+      this.scanPayloadDepth = undefined
+      this.scanInString = false
+      this.scanEscape = false
+      this.scanString = ''
+      this.scanLastString = ''
+      this.scanAwaitingColon = false
+      this.scanCurrentKey = undefined
+      this.scanCapture = undefined
+      this.processLine(line, durationOverride)
+      chunk = chunk.slice(newlineIndex + newlineLength)
+    }
+    if (this.pending.length > MAX_PENDING_LINE_CHARS) {
+      const body = this.pending.startsWith(TRUNCATION_MARKER)
+        ? this.pending.slice(TRUNCATION_MARKER.length)
+        : this.pending
+      this.pending = TRUNCATION_MARKER + body.slice(0, 256 * 1024) + body.slice(-256 * 1024)
+    }
+  }
+
   async update(filePath: string, limit = 10, finalize = false): Promise<CodexThroughputPoint[]> {
     const info = await stat(filePath)
     if (info.size < this.offset) this.reset()
-    const bytesToRead = info.size - this.offset
+    // A compressed rollout is immutable and its byte offsets are not text
+    // offsets, so it is decoded whole, once.
+    const compressed = filePath.endsWith('.zst')
+    if (compressed && this.offset === 0) {
+      try {
+        for await (const buffer of createTextReadStream(filePath)) this.consume(buffer as Buffer)
+      } catch (err) {
+        this.reset()
+        throw err
+      }
+      this.offset = info.size
+    }
+    const bytesToRead = compressed ? 0 : info.size - this.offset
     if (bytesToRead > 0) {
       const file = await open(filePath, 'r')
       try {
@@ -462,36 +508,7 @@ export class CodexThroughputReader {
           if (bytesRead === 0) break
           position += bytesRead
           this.offset = position
-          let chunk = this.decoder.write(buffer.subarray(0, bytesRead))
-          while (chunk.length > 0) {
-            const newlineIndex = chunk.search(/\r?\n/)
-            const segment = newlineIndex >= 0 ? chunk.slice(0, newlineIndex) : chunk
-            this.pending += segment
-            this.scanDurationSegment(segment)
-            if (newlineIndex < 0) break
-            const newlineLength = chunk[newlineIndex] === '\r' ? 2 : 1
-            const line = this.pending
-            const durationOverride = this.pendingDurationMs
-            this.pending = ''
-            this.pendingDurationMs = undefined
-            this.scanDepth = 0
-            this.scanPayloadDepth = undefined
-            this.scanInString = false
-            this.scanEscape = false
-            this.scanString = ''
-            this.scanLastString = ''
-            this.scanAwaitingColon = false
-            this.scanCurrentKey = undefined
-            this.scanCapture = undefined
-            this.processLine(line, durationOverride)
-            chunk = chunk.slice(newlineIndex + newlineLength)
-          }
-          if (this.pending.length > MAX_PENDING_LINE_CHARS) {
-            const body = this.pending.startsWith(TRUNCATION_MARKER)
-              ? this.pending.slice(TRUNCATION_MARKER.length)
-              : this.pending
-            this.pending = TRUNCATION_MARKER + body.slice(0, 256 * 1024) + body.slice(-256 * 1024)
-          }
+          this.consume(buffer.subarray(0, bytesRead))
         }
       } finally {
         await file.close()

src/context-tree-codex.ts

@@ -3,7 +3,8 @@ import { existsSync } from 'fs'
 import { basename, join } from 'path'
 import { homedir } from 'os'
 
-import { readSessionLines } from './fs-utils.js'
+import { readSessionLines, readTextHeadSync } from './fs-utils.js'
+import { isRolloutFileName } from './launcher-homes.js'
 import { estimateTokensFromChars } from './token-estimate.js'
 import {
   add,
@@ -243,7 +244,7 @@ export async function buildCodexContextTree(session: SessionRef): Promise<Contex
   }
 }
 
-const ROLLOUT_RE = /^rollout-.{19}-(.+)\.jsonl$/
+const ROLLOUT_RE = /^rollout-.{19}-(.+)\.jsonl(?:\.zst)?$/
 
 // Mirrors the CODEX_HOME handling of providers/codex.ts.
 function codexSessionsRoot(): string {
@@ -262,9 +263,12 @@ async function listRolloutFiles(): Promise<RolloutFile[]> {
     return []
   }
   const rollouts: RolloutFile[] = []
+  const all = new Set(files)
   for (const rel of files) {
-    const match = ROLLOUT_RE.exec(basename(rel))
-    if (match) rollouts.push({ filePath: join(root, rel), sessionId: match[1] })
+    if (rel.endsWith('.zst') && all.has(rel.slice(0, -4))) continue
+    const name = basename(rel)
+    const match = ROLLOUT_RE.exec(name)
+    if (match && isRolloutFileName(name)) rollouts.push({ filePath: join(root, rel), sessionId: match[1] })
   }
   return rollouts
 }
@@ -299,7 +303,9 @@ export async function findCodexSession(idPrefix: string): Promise<SessionRef | n
 async function readCodexHeadInfo(ref: SessionRef): Promise<{ project: string; title: string }> {
   let chunk: string
   try {
-    chunk = await readChunk(ref.filePath, 0, 262_144)
+    chunk = ref.filePath.endsWith('.zst')
+      ? readTextHeadSync(ref.filePath, 262_144).toString('utf-8')
+      : await readChunk(ref.filePath, 0, 262_144)
   } catch {
     return { project: '', title: '' }
   }

src/daily-cache.ts

@@ -361,7 +361,11 @@ import type { DateRange, ProjectSummary } from './types.js'
 // copilotCredits, and a chat session that also ran on the agent host is left
 // to it. That twin drops calls, so copilot joins
 // PENDING_REDERIVE_PROVIDER_VERSIONS at 75.
-export const DAILY_CACHE_VERSION = 75
+// v76: Codex reads the Zstandard-compressed rollouts (`rollout-*.jsonl.zst`)
+// Codex moves cold sessions into, which were skipped, so days settled after
+// that move miss their Codex usage. Calls and cost only rise, so no
+// PENDING_REDERIVE_PROVIDER_VERSIONS entry is needed.
+export const DAILY_CACHE_VERSION = 76
 const MIN_SUPPORTED_VERSION = 28
 
 /// Providers whose per-day CALL COUNT means something different at

src/fs-utils.ts

@@ -1,5 +1,7 @@
 import { readFile, stat } from 'fs/promises'
-import { readFileSync, statSync, createReadStream } from 'fs'
+import { readFileSync, statSync, createReadStream, openSync, readSync, closeSync } from 'fs'
+import { pipeline, type Readable } from 'stream'
+import zlib from 'zlib'
 
 // Hard cap well below V8's 512 MB string limit. Callers that need line-by-line
 // processing should use readSessionLines(), which avoids materializing the
@@ -23,6 +25,58 @@ function warn(msg: string): void {
   if (verbose()) process.stderr.write(`codeburn: ${msg}\n`)
 }
 
+// zstd landed in node:zlib in 22.15 / 23.8; the package floor is 22.13, so the
+// exports are optional and callers check zstdSupported before handing over a
+// `.zst` path.
+const zstd = zlib as unknown as {
+  createZstdDecompress?: typeof zlib.createZstdDecompress
+  zstdDecompressSync?: typeof zlib.zstdDecompressSync
+}
+export const zstdSupported = !!zstd.createZstdDecompress && !!zstd.zstdDecompressSync
+const MAX_ZSTD_HEAD_DECODED_BYTES = 64 * 1024 * 1024
+
+function requireZstd(filePath: string): void {
+  if (!zstdSupported) throw new Error(`${filePath} is zstd-compressed; reading it needs Node 22.15+`)
+}
+
+/** Reads a file's text, decompressing a `.zst` file on the fly. */
+export function createTextReadStream(filePath: string, options?: { start?: number; end?: number }): Readable {
+  if (!filePath.endsWith('.zst')) return createReadStream(filePath, options)
+  requireZstd(filePath)
+  const raw = createReadStream(filePath, options)
+  // pipeline, not pipe: a read error has to fail the decoder too, or its
+  // consumer waits forever.
+  return pipeline(raw, zstd.createZstdDecompress!(), () => {})
+}
+
+/** Up to `maxBytes` of a file's text from its start; for a `.zst` file only the compressed head is read. */
+export function readTextHeadSync(filePath: string, maxBytes: number): Buffer {
+  const fd = openSync(filePath, 'r')
+  let head: Buffer
+  try {
+    const buf = Buffer.alloc(maxBytes)
+    head = buf.subarray(0, readSync(fd, buf, 0, maxBytes, 0))
+  } finally {
+    closeSync(fd)
+  }
+  if (!filePath.endsWith('.zst') || head.length === 0) return head
+  requireZstd(filePath)
+  // ZSTD_e_continue decodes a cut-off frame as far as it goes instead of
+  // rejecting it as truncated; a head that inflates past the cap is retried
+  // with less input.
+  for (;;) {
+    try {
+      return zstd.zstdDecompressSync!(head, {
+        finishFlush: zlib.constants.ZSTD_e_continue,
+        maxOutputLength: MAX_ZSTD_HEAD_DECODED_BYTES,
+      }).subarray(0, maxBytes)
+    } catch (err) {
+      if ((err as NodeJS.ErrnoException).code !== 'ERR_BUFFER_TOO_LARGE' || head.length < 2) throw err
+      head = head.subarray(0, head.length >> 1)
+    }
+  }
+}
+
 // Always surfaced (not verbose-gated): dropping an entire session file silently
 // understates reported usage with no signal, so oversize skips use this.
 function notice(msg: string): void {
@@ -116,7 +170,7 @@ export async function* readSessionLines(
     return
   }
 
-  const stream = createReadStream(
+  const stream = createTextReadStream(
     filePath,
     options.startByteOffset !== undefined ? { start: options.startByteOffset } : undefined,
   )

src/launcher-homes.ts

@@ -1,7 +1,9 @@
-import { closeSync, existsSync, openSync, readdirSync, readSync, realpathSync } from 'fs'
+import { existsSync, readdirSync, realpathSync } from 'fs'
 import { homedir } from 'os'
 import { join, resolve, sep } from 'path'
 
+import { readTextHeadSync, zstdSupported } from './fs-utils.js'
+
 export type LauncherNote = {
   name: string
   path: string
@@ -49,33 +51,47 @@ export const FIRST_LINE_READ_CAP = 1024 * 1024
  *  Missing / unreadable files return undefined — callers must not drop those. */
 export function rolloutFileSessionId(filePath: string): string | undefined {
   try {
-    const fd = openSync(filePath, 'r')
-    try {
-      const buf = Buffer.alloc(FIRST_LINE_READ_CAP)
-      const n = readSync(fd, buf, 0, buf.length, 0)
-      if (n === 0) return undefined
-      const slice = buf.subarray(0, n)
-      const nl = slice.indexOf(0x0a)
-      // Hit the 1 MiB cap without a newline: incomplete, not a parseable id.
-      if (nl === -1 && n === FIRST_LINE_READ_CAP) return undefined
-      const line = (nl === -1 ? slice : slice.subarray(0, nl)).toString('utf8')
-      if (!line) return undefined
-      const parsed: unknown = JSON.parse(line)
-      if (!parsed || typeof parsed !== 'object') return undefined
-      const payload = (parsed as { payload?: { session_id?: unknown } }).payload
-      const id = payload?.session_id
-      return typeof id === 'string' && id.length > 0 ? id : undefined
-    } finally {
-      closeSync(fd)
-    }
+    const slice = readTextHeadSync(filePath, FIRST_LINE_READ_CAP)
+    if (slice.length === 0) return undefined
+    const nl = slice.indexOf(0x0a)
+    // Hit the 1 MiB cap without a newline: incomplete, not a parseable id.
+    if (nl === -1 && slice.length === FIRST_LINE_READ_CAP) return undefined
+    const line = (nl === -1 ? slice : slice.subarray(0, nl)).toString('utf8')
+    if (!line) return undefined
+    const parsed: unknown = JSON.parse(line)
+    if (!parsed || typeof parsed !== 'object') return undefined
+    const payload = (parsed as { payload?: { session_id?: unknown } }).payload
+    const id = payload?.session_id
+    return typeof id === 'string' && id.length > 0 ? id : undefined
   } catch {
     return undefined
   }
 }
 
+let zstdSkipNoticeSent = false
+
+/** `rollout-*.jsonl`, or the `.jsonl.zst` form Codex compresses cold rollouts
+ *  into, when this Node can decompress it. */
+export function isRolloutFileName(name: string): boolean {
+  if (!name.startsWith('rollout-')) return false
+  if (name.endsWith('.jsonl')) return true
+  if (!name.endsWith('.jsonl.zst')) return false
+  if (zstdSupported) return true
+  if (!zstdSkipNoticeSent) {
+    zstdSkipNoticeSent = true
+    process.stderr.write('codeburn: codex: skipping compressed rollouts (.jsonl.zst); counting them needs Node 22.15+ (node:zlib zstd)\n')
+  }
+  return false
+}
+
+/** Rollout filename without `.jsonl` / `.jsonl.zst`. */
+export function rolloutStem(name: string): string {
+  return name.replace(/\.jsonl(\.zst)?$/, '')
+}
+
 export function walkRolloutFiles(codexDir: string, visit: (filePath: string) => void): void {
   const take = (dir: string, file: string) => {
-    if (file.startsWith('rollout-') && file.endsWith('.jsonl')) visit(join(dir, file))
+    if (isRolloutFileName(file)) visit(join(dir, file))
   }
   const sessionsDir = join(codexDir, 'sessions')
   try {

src/parser.ts

@@ -3811,7 +3811,9 @@ export async function parseProviderSources(
       if (await readCachedCodexResults(source.path)) continue
       workerJobs.push({ kind: 'codex', source })
       workerPaths.add(source.path)
-      pendingBytes += fp.sizeBytes
+      // The worker memory budget scales with decoded size; real compressed
+      // rollouts decode to 2.3-4x their size on disk.
+      pendingBytes += source.path.endsWith('.zst') ? fp.sizeBytes * 4 : fp.sizeBytes
     }
   }
   const decision = workerJobs.length > 0