//// A rollout with the noise removed.//// 91-95% of a real rollout is `custom` / `custom_message` entries written by pi// extensions on every turn — state snapshots that are inert to the transcript// and thrown away on every read (see bench/rollout-census.bench.ts). Paging the// transcript therefore means scanning 40-145 MB to find a few hundred messages,// and no amount of seeking helps because the messages are interleaved// throughout rather than clustered at the end.//// So keep a second copy containing only the lines that matter. It is a plain// `.jsonl` in the same format, which is the point: `readTailRegion` runs over it// unchanged, and cursors stay opaque byte offsets — just into a file that is// twenty times smaller.//// Both files are append-only, so the sidecar is extended rather than rebuilt as// the session grows, and a cursor handed out for an earlier page stays valid.// import { appendFileSync, createReadStream, mkdirSync, statSync, writeFileSync } from "node:fs";import path from "node:path";import {  evictIfCrowded,  rolloutCache,  rolloutCacheFilePath,  statRollout,} from "./rollout-cache"; const SIDECAR_KIND = "transcript"; /** * Mirrors `isInertEvent` in sessions-store. Checked as a byte prefix on the raw * line so the common case never pays for `JSON.parse`. */const INERT_PREFIXES = ['{"type":"custom"', '{"type":"custom_message"']; function lineIsInert(line: string): boolean {  for (const prefix of INERT_PREFIXES) {    if (line.startsWith(prefix)) return true;  }  // Backstop for logs whose key order differs from what pi writes today.  if (!line.includes('"custom')) return false;  try {    const type = (JSON.parse(line) as { type?: string }).type;    return type === "custom" || type === "custom_message";  } catch {    return false;  }} type SidecarState = {  /** Source size when the sidecar was last extended. */  sourceSize: number;  sourceMtimeMs: number;  /** Offset in the SOURCE just past the last complete line copied. */  scannedBytes: number;  /** Opening bytes of the source, to notice a rewrite rather than an append. */  head: string;}; const state = rolloutCache<SidecarState>("transcript-state"); const HEAD_FINGERPRINT_BYTES = 512; async function readHead(filepath: string): Promise<string> {  const chunks: Buffer[] = [];  const stream = createReadStream(filepath, { start: 0, end: HEAD_FINGERPRINT_BYTES - 1 });  for await (const chunk of stream) chunks.push(chunk as Buffer);  return Buffer.concat(chunks).toString("utf-8");} /** * Copy every complete non-inert line from `start` onward onto the sidecar. * Returns the source offset just past the last complete line consumed — a * partial trailing line is left for the next call, exactly as the usage scan * does, because a rollout is appended to while it is being read. */async function appendFrom(source: string, sidecar: string, start: number): Promise<number> {  let consumedBytes = start;  let pending = "";  let batch: string[] = [];   const flush = () => {    if (batch.length === 0) return;    appendFileSync(sidecar, `${batch.join("\n")}\n`, "utf-8");    batch = [];  };   const stream = createReadStream(source, { start, encoding: "utf-8" });  for await (const chunk of stream) {    pending += chunk as string;    let lineStart = 0;    let newline = pending.indexOf("\n", lineStart);    while (newline !== -1) {      const line = pending.slice(lineStart, newline);      consumedBytes += Buffer.byteLength(line, "utf-8") + 1;      if (line && !lineIsInert(line)) batch.push(line);      lineStart = newline + 1;      newline = pending.indexOf("\n", lineStart);    }    pending = pending.slice(lineStart);    // Bounded so a multi-GB rollout never buffers its whole transcript.    if (batch.length >= 2048) flush();  }  flush();  return consumedBytes;} export type TranscriptSource = { filepath: string; size: number }; /** * The file to page the transcript from: the sidecar when one can be built, * otherwise the rollout itself. * * Never throws and never blocks correctness — every failure path returns the * original rollout, which reads identically, just slower. */export async function transcriptSource(filepath: string): Promise<TranscriptSource> {  const original = (): TranscriptSource => ({    filepath,    size: statRollout(filepath)?.size ?? 0,  });   const stat = statRollout(filepath);  if (!stat) return original();   try {    const sidecar = rolloutCacheFilePath(SIDECAR_KIND, filepath, ".jsonl");    const head = await readHead(filepath);    const previous = state.readStale(filepath);     const sidecarSize = (() => {      try {        return statSync(sidecar).size;      } catch {        return -1;      }    })();     // Extend only when this is the same file, grown, and the sidecar we built    // for it is still there. Anything else means starting over.    const resumable =      previous !== undefined &&      previous.head === head &&      stat.size >= previous.scannedBytes &&      previous.scannedBytes > 0 &&      sidecarSize >= 0;     if (resumable && previous.sourceSize === stat.size && previous.sourceMtimeMs === stat.mtimeMs) {      return { filepath: sidecar, size: sidecarSize };    }     mkdirSync(path.dirname(sidecar), { recursive: true });    if (!resumable) writeFileSync(sidecar, "", "utf-8");     const scannedBytes = await appendFrom(filepath, sidecar, resumable ? previous.scannedBytes : 0);    state.write(filepath, stat, {      sourceSize: stat.size,      sourceMtimeMs: stat.mtimeMs,      scannedBytes,      head,    });    // Sidecars are ~5% of their rollout but there is one per session ever    // opened, so they need the same bound the envelopes get.    evictIfCrowded(path.dirname(sidecar), ".jsonl");     return { filepath: sidecar, size: statSync(sidecar).size };  } catch {    return original();  }}