// aidlc-usage.ts - the single token-usage + cost extraction seam.
//
// One place owns: the model rate table (framework-default public list prices,
// overridable), the Claude transcript readers (main + sibling sub-agent files),
// the pure cost math, and the durable per-file-cursor ledger. Everything else
// (audit rollup fields, metrics magnitude lines, the statusline segment)
// CONSUMES this module and never re-parses a transcript itself.
//
// HARNESS SCOPE. The transcript reader is Claude-Code-format-specific: only the
// Claude harness wires a producer (the PreToolUse + PostToolUse fold hook and
// the Stop-hook flush). On Kiro / Codex / opencode no producer is wired, so the
// ledger is never written and every consumer here degrades silently to no-data:
// the statusline renders no cost segment, and the audit rollup adds no fields.
// See docs/reference/06-hooks-and-tools.md.
//
// ROBUSTNESS CONTRACT. Nothing here throws on malformed or missing input. A
// half-written last JSONL line is normal for a live transcript, so a bad line
// is skipped silently; an absent or corrupt file yields [] / a fresh empty
// ledger. Unknown model => tokens recorded, cost `null` - never a fabricated
// number.
//
// The verified transcript facts this module encodes (separate sub-agent files,
// .meta.json sidecars, per-token cache costs on the converse API, converse/ +
// region model prefixes, and the per-file cursor requirement - uuids collide
// across concurrent sub-agent files, so `(sourceFile, uuid)` is the only unique
// key) are documented in docs/reference/06-hooks-and-tools.md.

import { dlopen, ptr } from "bun:ffi";
import { createHash } from "node:crypto";
import {
  closeSync,
  existsSync,
  fstatSync,
  mkdirSync,
  openSync,
  readSync,
  readdirSync,
  readFileSync,
} from "node:fs";
import { basename, dirname, join } from "node:path";
import {
  auditLockIdentity,
  intentUuidForSelection,
  modelRatesPath,
  readSessionIntentUuid,
  resolveProjectFlag,
  resolveWorkflowSelection,
  sessionsDir,
  withAuditLock,
  writeFileAtomic,
} from "./aidlc-lib.ts";

// ===========================================================================
// Task 1 - Rate table + pure cost math (no transcript I/O)
// ===========================================================================

// A per-million-token price row. `input`/`output` price ordinary prompt and
// completion tokens; `cacheWrite5m`/`cacheWrite1h` price cache CREATION at the
// two ephemeral TTLs; `cacheRead` prices a cache HIT.
export type PriceRow = {
  input: number;
  output: number;
  cacheWrite5m: number;
  cacheWrite1h: number;
  cacheRead: number;
};

// FRAMEWORK-DEFAULT rates, USD / 1e6 tokens. These are PUBLIC Anthropic list
// prices, shipped as DEFAULTS only, and are the pure dev-checkout fallback used
// when no shipped `tools/data/model-rates.json` is present (an authored core/
// tree carries the JSON, so an installed harness always reads that). The cache
// fields follow the standard multipliers: cacheWrite5m = 1.25x input,
// cacheWrite1h = 2x input, cacheRead = 0.1x input.
//
// GENERATION-DISCRETE keys (a row per model GENERATION, NOT one per family):
// verified real sessions mix generations, and a family-collapse silently
// misprices them onto whatever the "current" row happens to be. Key off the
// generation token, not the account/region variant.
//
// To OVERRIDE: set AIDLC_MODEL_RATES to a rates file (same shape as
// tools/data/model-rates.json), or edit the shipped model-rates.json in your
// install. The override layers ON TOP of these defaults - a partial file only
// changes the models it names; an unknown model stays tokens-with-null-cost.
export const DEFAULT_RATES: Record<string, PriceRow> = {
  "opus-5": { input: 5.0, output: 25.0, cacheWrite5m: 6.25, cacheWrite1h: 10.0, cacheRead: 0.5 },
  "opus-4-8": { input: 5.0, output: 25.0, cacheWrite5m: 6.25, cacheWrite1h: 10.0, cacheRead: 0.5 },
  "opus-4-7": { input: 5.0, output: 25.0, cacheWrite5m: 6.25, cacheWrite1h: 10.0, cacheRead: 0.5 },
  "opus-4-6": { input: 5.0, output: 25.0, cacheWrite5m: 6.25, cacheWrite1h: 10.0, cacheRead: 0.5 },
  "sonnet-5": { input: 3.0, output: 15.0, cacheWrite5m: 3.75, cacheWrite1h: 6.0, cacheRead: 0.3 },
  "sonnet-4-6": { input: 3.0, output: 15.0, cacheWrite5m: 3.75, cacheWrite1h: 6.0, cacheRead: 0.3 },
  "haiku-4-5": { input: 1.0, output: 5.0, cacheWrite5m: 1.25, cacheWrite1h: 2.0, cacheRead: 0.1 },
  "fable-5": { input: 10.0, output: 50.0, cacheWrite5m: 12.5, cacheWrite1h: 20.0, cacheRead: 1.0 },
};

// Validate one parsed JSON object into a PriceRow, or null when it is not a
// well-formed row (any missing/non-finite field). Keeps a malformed override
// file from poisoning a real rate with NaN - the offending row is dropped and
// the default (or nothing) stands.
function coercePriceRow(raw: unknown): PriceRow | null {
  if (!raw || typeof raw !== "object") return null;
  const o = raw as Record<string, unknown>;
  const num = (v: unknown): number | null =>
    typeof v === "number" && Number.isFinite(v) ? v : null;
  const input = num(o.input);
  const output = num(o.output);
  const cacheWrite5m = num(o.cacheWrite5m);
  const cacheWrite1h = num(o.cacheWrite1h);
  const cacheRead = num(o.cacheRead);
  if (
    input === null ||
    output === null ||
    cacheWrite5m === null ||
    cacheWrite1h === null ||
    cacheRead === null
  ) {
    return null;
  }
  return { input, output, cacheWrite5m, cacheWrite1h, cacheRead };
}

// Read a rates file's `rates` map into overlay rows. Missing/corrupt/ill-shaped
// file => {} (no overlay). Never throws.
function readRatesFile(path: string): Record<string, PriceRow> {
  let raw: string;
  try {
    raw = readFileSync(path, "utf-8");
  } catch {
    return {};
  }
  try {
    const parsed = JSON.parse(raw) as { rates?: unknown };
    const rates = parsed?.rates;
    if (!rates || typeof rates !== "object") return {};
    const out: Record<string, PriceRow> = {};
    for (const [key, value] of Object.entries(rates as Record<string, unknown>)) {
      const row = coercePriceRow(value);
      if (row) out[key] = row;
    }
    return out;
  } catch {
    return {};
  }
}

// The usage-tracking kill switch. Set AIDLC_DISABLE_USAGE_TRACKING=1 to stop
// the ledger producer from writing and every consumer from reading: the fold
// hooks and Stop-hook flush become no-ops, the statusline renders no cost
// segment, and STAGE_COMPLETED / WORKFLOW_COMPLETED add no rollup fields.
// Follows the AIDLC_DISABLE_* hook-flag convention (exact string "1"). Read at
// call time, never cached - each hook/tool is a fresh process, and tests
// toggle it mid-process.
export function usageTrackingDisabled(): boolean {
  return resolveProjectFlag("AIDLC_DISABLE_USAGE_TRACKING") === "1";
}

let _rates: Record<string, PriceRow> | null = null;

// The effective rate table, built in three layers (each overlays the previous,
// per-model, so a partial file only changes the models it names):
//   1. DEFAULT_RATES        - hardcoded public list prices (the dev-checkout floor)
//   2. tools/data/model-rates.json - the shipped framework default (an install edits this)
//   3. $AIDLC_MODEL_RATES   - a user/project-supplied override file
// Cached at first use - a per-line computeCost never re-reads the files. Robust:
// a missing or malformed source contributes nothing and the layers below stand.
export function loadRates(): Record<string, PriceRow> {
  if (_rates !== null) return _rates;
  const merged: Record<string, PriceRow> = { ...DEFAULT_RATES };
  // Layer 2: the shipped default file (present in an installed harness; absent
  // in a dev checkout's core/, where DEFAULT_RATES is the only source).
  try {
    for (const [k, v] of Object.entries(readRatesFile(modelRatesPath()))) merged[k] = v;
  } catch {
    /* no shipped file / unresolvable data dir - DEFAULT_RATES stands */
  }
  // Layer 3: the AIDLC_MODEL_RATES override file, on top of everything.
  const override = process.env.AIDLC_MODEL_RATES;
  if (override) {
    for (const [k, v] of Object.entries(readRatesFile(override))) merged[k] = v;
  }
  _rates = merged;
  return _rates;
}

// TEST SEAM: drop the cached rate table so a test that mutates AIDLC_MODEL_RATES
// mid-process re-reads it. Not used at runtime (each tool/hook is a fresh
// process). Pure.
export function _resetRatesCacheForTest(): void {
  _rates = null;
}

// Bare family aliases with NO generation token. Matched only on an EXACT
// residual equality (never as a substring), so `opus` resolves but `opus-6`
// (a would-be new generation) does NOT match and stays null. These are a
// defensive convenience for hand-typed / test inputs; the wire form always
// carries a generation token. `opus`/`sonnet`/`haiku` map to the generation
// this harness ships as its default model set.
const BARE_ALIASES: Record<string, string> = {
  opus: "opus-4-8",
  sonnet: "sonnet-4-6",
  haiku: "haiku-4-5",
  fable: "fable-5",
};

// Map a transcript `message.model` to a rate-table key, or null if unknown.
//
// Handles the real Bedrock forms verified in transcripts:
//   converse/us.anthropic.claude-opus-4-8   (main thread)
//   us.anthropic.claude-opus-4-8            (some sub-agents)
//   claude-haiku-4-5-20251001               (sub-agent, dated, no region)
//   converse/au.anthropic.claude-haiku-4-5-20251001-v1:0
//   global.anthropic.claude-opus-4-8[1m]    (this harness's settings alias)
//
// UNKNOWN-GENERATION POLICY: a Claude model whose GENERATION is not in the rate
// table (e.g. a future `opus-6`, or an `opus-4-9` we haven't priced) returns
// `null` => the caller records the tokens but withholds cost. An honest
// "unknown" (made visible by the audit's `Cost USD: null`) beats a
// confidently-wrong number from an old generation's rate. `<synthetic>`, empty,
// malformed provider/model shapes, and non-Claude models => null too. Generation
// keys come from the EFFECTIVE rate table, so an AIDLC_MODEL_RATES override can
// add a new generation without a source-code matcher change.
export function normalizeModel(modelId: string): string | null {
  if (!modelId || typeof modelId !== "string") return null;
  let s = modelId.trim().toLowerCase();
  if (BARE_ALIASES[s]) return BARE_ALIASES[s];
  // 1. Drop a leading `converse/` provider tag. MUST run before step 2 - the
  //    region wildcard is anchored at the string start.
  const hadConversePrefix = s.startsWith("converse/");
  if (hadConversePrefix) s = s.slice("converse/".length);
  // 2. Drop an inference-profile region prefix + `anthropic.`. WILDCARDED:
  //    `<region>` is any `[a-z0-9-]+` token, so a new region (eu/apac/global/
  //    ... or one not yet seen) never silently breaks normalization. The
  //    `\.anthropic\.` anchor keeps the wildcard from eating a region-less model
  //    (bare `claude-opus-4-8` has no `.anthropic.`, so it is untouched here).
  const provider = s.match(/^(?:[a-z0-9-]+\.)?anthropic\./);
  if (provider) {
    s = s.slice(provider[0].length);
  } else if (hadConversePrefix) {
    return null;
  }
  // 3. Accept only a real Anthropic/Claude shape or one of the documented bare
  // family aliases handled above. Exact rate keys are not wire model IDs:
  // accepting them after `converse/` or `anthropic.` would price malformed
  // provider shapes such as `converse/opus-4-8`.
  const rates = loadRates();
  if (!s.startsWith("claude-")) return null;
  s = s.slice("claude-".length);

  // 4. Match a generation key only at a token boundary. Keys are longest-first
  // so an override containing related keys resolves the most specific one.
  for (const key of Object.keys(rates).sort((a, b) => b.length - a.length)) {
    if (s === key || s.startsWith(`${key}-`) || s.startsWith(`${key}[`)) {
      return key;
    }
  }
  // Unknown generation / non-Claude => null (tokens recorded, cost withheld).
  return null;
}

// The token magnitudes we price. Cache creation is split by ephemeral TTL;
// cache reads are a single bucket.
export type TokenCounts = {
  input: number;
  output: number;
  cacheCreate5m: number;
  cacheCreate1h: number;
  cacheRead: number;
};

// Price a token bundle for a model. Unknown model => `{ usd: null, model: null }`
// (tokens are preserved by the caller; only the cost is withheld). Cache reads
// price at `cacheRead`; cache creation splits across the 5m / 1h rows.
export function computeCost(
  counts: TokenCounts,
  modelId: string,
): { usd: number | null; model: string | null } {
  const key = normalizeModel(modelId);
  if (!key) return { usd: null, model: null };
  const r = loadRates()[key];
  if (!r) return { usd: null, model: null };
  const usd =
    (counts.input / 1e6) * r.input +
    (counts.output / 1e6) * r.output +
    (counts.cacheCreate5m / 1e6) * r.cacheWrite5m +
    (counts.cacheCreate1h / 1e6) * r.cacheWrite1h +
    (counts.cacheRead / 1e6) * r.cacheRead;
  return { usd, model: key };
}

// One priced assistant turn. `sourceKey` identifies the file the row came from
// (`"main"` or `"agent-<agentId>"`) and is the PER-FILE cursor key in Task 3.
// `agentId` is filled during extraction (Task 2) for sub-agent rows; `agentType`
// is filled during attribution (Task 4).
export type UsageRow = {
  uuid: string;
  // The Claude `message.id`. Claude Code splits ONE llm call (one message.id)
  // across MULTIPLE JSONL lines - one per content block (thinking/text/tool_use)
  // - stamping the SAME `message.usage` on every line. `msgId` lets
  // dedupeByMessageId collapse a contiguous run of same-id lines into ONE row so
  // the shared usage is counted ONCE, not 2-3x. `""` when a line carries no
  // `message.id` (hand-authored fixtures, older transcripts): an id-less row is
  // NEVER merged - each such line stays its own row.
  msgId: string;
  timestamp: string;
  model: string;
  isSidechain: boolean;
  sourceKey: string;
  agentId: string | null;
  agentType: string | null;
  counts: TokenCounts;
  usd: number | null;
};

// ===========================================================================
// Task 2 - Transcript extraction (Claude main + sub-agent files)
// ===========================================================================

// Parse one JSONL object's `message.usage` into a TokenCounts. The 5m/1h split
// lives under `cache_creation`; when that object is absent we treat the flat
// `cache_creation_input_tokens` total as 5m (its pre-split form).
function countsFromUsage(usage: Record<string, unknown>): TokenCounts {
  const num = (v: unknown): number => (typeof v === "number" && Number.isFinite(v) ? v : 0);
  const cc = (usage.cache_creation ?? {}) as Record<string, unknown>;
  const e5 = num(cc.ephemeral_5m_input_tokens);
  const e1 = num(cc.ephemeral_1h_input_tokens);
  const flat = num(usage.cache_creation_input_tokens);
  // Trust the nested 5m/1h split only when it actually accounts for cache
  // creation. Bedrock's converse API ships the split present-but-zeroed with
  // the real total in the flat cache_creation_input_tokens field; native
  // Anthropic always has flat == e5 + e1, so this is a no-op there. When the
  // split is absent or zeroed we fall back to the flat total, attributing it
  // to 5m: the harness only ever issues 5m-TTL cache writes, so a Bedrock
  // creation is a 5m write in practice. (A hypothetical 1h Bedrock write
  // would be under-priced at the 5m rate - but that beats the old behavior,
  // which dropped the flat value entirely and priced the write at zero.)
  const splitSum = e5 + e1;
  return {
    input: num(usage.input_tokens),
    output: num(usage.output_tokens),
    cacheCreate5m: splitSum > 0 ? e5 : flat,
    cacheCreate1h: splitSum > 0 ? e1 : 0,
    cacheRead: num(usage.cache_read_input_tokens),
  };
}

// Parse ONE JSONL line into a UsageRow, or null when it is not a priced
// assistant turn (non-assistant, no usage, malformed). Pure - the shared per-
// line parse used by both the whole-file reader (readClaudeTranscript) and the
// offset-aware fold reader. `sourceKey`/`fallbackAgentId` supply the file
// identity when the line does not carry its own `agentId`.
function parseTranscriptLine(
  line: string,
  sourceKey: string,
  fallbackAgentId: string | null,
): UsageRow | null {
  const trimmed = line.trim();
  if (!trimmed) return null;
  let obj: Record<string, unknown>;
  try {
    obj = JSON.parse(trimmed) as Record<string, unknown>;
  } catch {
    return null; // half-written / malformed line - skip, don't throw
  }
  const message = obj.message as Record<string, unknown> | undefined;
  if (!message) return null;
  if (message.role !== "assistant") return null;
  const usage = message.usage as Record<string, unknown> | undefined;
  if (!usage || typeof usage !== "object") return null;
  const counts = countsFromUsage(usage);
  const model = typeof message.model === "string" ? message.model : "";
  const agentId =
    typeof obj.agentId === "string" ? obj.agentId : fallbackAgentId;
  return {
    uuid: typeof obj.uuid === "string" ? obj.uuid : "",
    msgId: typeof message.id === "string" ? message.id : "",
    timestamp: typeof obj.timestamp === "string" ? obj.timestamp : "",
    model,
    isSidechain: obj.isSidechain === true,
    sourceKey,
    agentId,
    agentType: null,
    counts,
    usd: computeCost(counts, model).usd,
  };
}

// covers: function:dedupeByMessageId
// Collapse the split-line overcount (the CRITICAL bug): Claude Code writes one
// llm call (one `message.id`) as MULTIPLE contiguous JSONL lines - one per
// content block (thinking/text/tool_use). Left alone that inflates input ~2x and
// output ~2.6x. Verified invariants this relies on: (a) split lines of one
// message.id are always CONTIGUOUS; (b) the group's usage must be taken ONCE.
//
// TWO transcript styles exist (both handled by taking the group's usage as the
// MAXIMUM-usage line of the run):
//   OLD-style: every split line repeats the SAME usage => max == any line.
//   NEW-style: leading lines (thinking/text) carry usage=0; the REAL usage
//     appears on the LAST line(s) of the run (identical when several trail) =>
//     max == that real-usage line, so the zeros never mask the real count.
// The representative is the max-usage row (ties => the LAST such row, so its
// uuid/timestamp is the real end-of-turn line and the per-file cursor advances
// correctly). On OLD-style all lines tie so this is the last line, matching the
// prior behaviour exactly.
//
// FALLBACK: a row with an empty `msgId` (no `message.id` - hand-authored
// fixtures, older transcripts) is NEVER merged; each such row passes through
// as-is, preserving one-row-per-line behaviour. A `seen` Set guards against a
// (non-contiguous) reappearance of an already-emitted non-empty msgId so a
// second run of the same id can never double-count. Pure.
export function dedupeByMessageId(rows: UsageRow[]): UsageRow[] {
  const out: UsageRow[] = [];
  const seen = new Set<string>();
  let i = 0;
  while (i < rows.length) {
    const row = rows[i];
    if (!row.msgId) {
      // Id-less line - never merged, always its own row.
      out.push(row);
      i++;
      continue;
    }
    if (seen.has(row.msgId)) {
      // A non-contiguous reappearance of an already-collapsed id - drop it (its
      // usage was already counted once). Defensive; verified runs are contiguous.
      i++;
      continue;
    }
    // Consume the contiguous run sharing this msgId; take the group's usage ONCE
    // from the MAX-usage line (ties => last, the real end-of-turn line).
    let j = i;
    while (j + 1 < rows.length && rows[j + 1].msgId === row.msgId) j++;
    seen.add(row.msgId);
    out.push(representativeOfRun(rows, i, j));
    i = j + 1;
  }
  return out;
}

// The representative of a contiguous same-msgId run `rows[start..end]`: the row
// with the greatest total usage magnitude, breaking ties toward the LAST such
// row. This is what makes both transcript styles collapse losslessly - NEW-style
// leading usage=0 lines lose to the trailing real-usage line, OLD-style ties
// resolve to the last line (the real end-of-turn uuid/timestamp). Pure.
function representativeOfRun(rows: UsageRow[], start: number, end: number): UsageRow {
  let best = rows[start];
  let bestMag = usageMagnitude(best.counts);
  for (let k = start + 1; k <= end; k++) {
    const mag = usageMagnitude(rows[k].counts);
    if (mag >= bestMag) {
      // `>=` breaks ties toward the later row (real end-of-turn line).
      best = rows[k];
      bestMag = mag;
    }
  }
  return best;
}

// Total token magnitude of a bundle - the tie-break/selection key for a split
// run's representative. Sums every priced magnitude so a line carrying real
// usage always outranks a leading usage=0 line. Pure.
function usageMagnitude(c: TokenCounts): number {
  return c.input + c.output + c.cacheCreate5m + c.cacheCreate1h + c.cacheRead;
}

// Read a single Claude JSONL transcript into priced UsageRows. `opts.sourceKey`
// and `opts.agentId` default to a main-thread read; readClaudeSession supplies
// the sub-agent identity. Emits one row per assistant LLM CALL (message.id) -
// contiguous split lines sharing a message.id are collapsed via
// dedupeByMessageId so the shared usage is counted once. Malformed lines (incl.
// a half-written trailing line) are skipped silently - never throws on a bad
// line or a missing file.
export function readClaudeTranscript(
  path: string,
  opts?: { sourceKey?: string; agentId?: string | null },
): UsageRow[] {
  const sourceKey = opts?.sourceKey ?? "main";
  const fallbackAgentId = opts?.agentId ?? null;
  let raw: string;
  try {
    raw = readFileSync(path, "utf-8");
  } catch {
    return [];
  }
  const rows: UsageRow[] = [];
  for (const line of raw.split("\n")) {
    const row = parseTranscriptLine(line, sourceKey, fallbackAgentId);
    if (row) rows.push(row);
  }
  return dedupeByMessageId(rows);
}

// Resolve the sibling `subagents/` directory for a main transcript. Layout:
// main is `<projects>/<slug>/<session>.jsonl`; sub-agents are
// `<projects>/<slug>/<session>/subagents/agent-<agentId>.jsonl`.
export function subagentDir(mainTranscriptPath: string): string {
  const dir = dirname(mainTranscriptPath);
  const session = basename(mainTranscriptPath).replace(/\.jsonl$/, "");
  return join(dir, session, "subagents");
}

// Read one `agent-<id>.meta.json` sidecar; return its parsed object or null.
function readMetaSidecar(jsonlPath: string): { agentType?: string } | null {
  const metaPath = jsonlPath.replace(/\.jsonl$/, ".meta.json");
  if (!existsSync(metaPath)) return null;
  try {
    return JSON.parse(readFileSync(metaPath, "utf-8")) as { agentType?: string };
  } catch {
    return null;
  }
}

// Read a full Claude session: the main transcript plus every sibling
// `subagents/agent-*.jsonl`. Main rows carry `agentId: null`,
// `sourceKey: "main"`; sub-agent rows carry their file's `agentId` and
// `sourceKey: "agent-<agentId>"`. Each sub-agent's `.meta.json` sidecar is read
// for its `agentType`, and agent attribution (Task 4) is applied before
// returning. A missing `subagents/` dir => just the main rows. Never throws.
export function readClaudeSession(mainPath: string): UsageRow[] {
  const rows: UsageRow[] = readClaudeTranscript(mainPath, {
    sourceKey: "main",
    agentId: null,
  });
  const metaByAgentId: Record<string, string> = {};
  const subDir = subagentDir(mainPath);
  let entries: string[] = [];
  try {
    if (existsSync(subDir)) {
      entries = readdirSync(subDir).filter(
        (f) => f.startsWith("agent-") && f.endsWith(".jsonl"),
      );
    }
  } catch {
    entries = [];
  }
  for (const file of entries) {
    const agentId = file.replace(/^agent-/, "").replace(/\.jsonl$/, "");
    const full = join(subDir, file);
    const meta = readMetaSidecar(full);
    if (meta && typeof meta.agentType === "string" && meta.agentType) {
      metaByAgentId[agentId] = meta.agentType;
    }
    rows.push(
      ...readClaudeTranscript(full, {
        sourceKey: `agent-${agentId}`,
        agentId,
      }),
    );
  }
  return attributeAgents(rows, metaByAgentId);
}

// Sniff the transcript format from its first non-empty line and dispatch.
// Claude lines carry `message`/`uuid` => readClaudeSession. Any other shape
// (empty, non-Claude) => []. Only the Claude Code transcript format is
// supported - Kiro/Codex/opencode wire no producer, so their transcripts (if
// any) are never read here.
export function readTranscript(path: string): UsageRow[] {
  let raw: string;
  try {
    raw = readFileSync(path, "utf-8");
  } catch {
    return [];
  }
  for (const line of raw.split("\n")) {
    const trimmed = line.trim();
    if (!trimmed) continue;
    let obj: Record<string, unknown>;
    try {
      obj = JSON.parse(trimmed) as Record<string, unknown>;
    } catch {
      continue;
    }
    if ("message" in obj || "uuid" in obj) {
      return readClaudeSession(path);
    }
    // Sniffed a line we don't recognise - keep looking at subsequent lines.
  }
  return [];
}

// ===========================================================================
// Task 4 - Agent-type attribution (declared before Task 3's fold so
// readClaudeSession can call it)
// ===========================================================================

// Attribute each row to an agentType. Main-thread rows => "main". Sub-agent rows
// => the sidecar `agentType` for their agentId, falling back to "subagent" when
// no sidecar/field exists (never null, never dropped). Pure.
export function attributeAgents(
  rows: UsageRow[],
  metaByAgentId: Record<string, string>,
): UsageRow[] {
  for (const row of rows) {
    if (row.sourceKey === "main") {
      row.agentType = "main";
    } else {
      const id = row.agentId ?? "";
      row.agentType = metaByAgentId[id] ?? "subagent";
    }
  }
  return rows;
}

// Fallback join: build `agentId -> subagent_type` from a parent transcript's
// lines. A `Task`/`Agent` tool_use carries an `id` (tool_use_id) and
// `input.subagent_type`; the matching `toolUseResult` line carries `agentId`
// for that same tool_use_id. Used ONLY for agentIds absent from the sidecar map
// (the sidecar `agentType` is present in ~100% of sampled sidecars; `toolUseId`
// only ~86%). Never throws.
export function buildAgentTypeMapFromParent(
  parentLines: string[],
): Record<string, string> {
  const typeByToolUseId: Record<string, string> = {};
  const map: Record<string, string> = {};
  for (const line of parentLines) {
    const trimmed = line.trim();
    if (!trimmed) continue;
    let obj: Record<string, unknown>;
    try {
      obj = JSON.parse(trimmed) as Record<string, unknown>;
    } catch {
      continue;
    }
    // (a) Collect subagent_type per tool_use id from assistant tool_use blocks.
    const message = obj.message as Record<string, unknown> | undefined;
    const content = message?.content;
    if (Array.isArray(content)) {
      for (const block of content) {
        const b = block as Record<string, unknown>;
        if (
          b.type === "tool_use" &&
          (b.name === "Task" || b.name === "Agent") &&
          typeof b.id === "string"
        ) {
          const input = (b.input ?? {}) as Record<string, unknown>;
          if (typeof input.subagent_type === "string") {
            typeByToolUseId[b.id] = input.subagent_type;
          }
        }
      }
    }
    // (b) Join a toolUseResult's agentId back to the tool_use's subagent_type.
    const tur = obj.toolUseResult as Record<string, unknown> | undefined;
    const toolUseId =
      typeof obj.tool_use_id === "string"
        ? obj.tool_use_id
        : typeof tur?.tool_use_id === "string"
          ? (tur.tool_use_id as string)
          : undefined;
    const agentId =
      typeof tur?.agentId === "string" ? (tur.agentId as string) : undefined;
    if (toolUseId && agentId && typeByToolUseId[toolUseId]) {
      map[agentId] = typeByToolUseId[toolUseId];
    }
  }
  return map;
}

// ===========================================================================
// Task 3 - Per-file-cursor ledger
// ===========================================================================

// Summed token magnitudes + total USD for a bucket.
export type Totals = {
  tokens: TokenCounts;
  usd: number;
};

// A per-stage bucket. `totals` is the flat stage sum; `byModel`/`byAgent` are
// the stage's OWN per-model / per-agent sub-splits - stage-scoped, NOT the
// global session buckets. This is what lets STAGE_COMPLETED show a By Model line
// that agrees with its stage-scoped Cost USD (a global By Model would sum every
// stage and contradict a single-stage cost).
export type StageBucket = {
  totals: Totals;
  byModel: Record<string, Totals>;
  byAgent: Record<string, Totals>;
};

// One aggregate at a particular ownership boundary. The top-level ledger keeps
// a workspace aggregate for diagnostics, while `workflows[workflow].sessions`
// provides the authoritative workflow/session intersection.
export type UsageAggregate = {
  totals: Totals;
  byStage: Record<string, StageBucket>;
  byModel: Record<string, Totals>;
  byAgent: Record<string, Totals>;
};

export type WorkflowUsage = UsageAggregate & {
  sessions: Record<string, UsageAggregate>;
};

// The current on-disk ledger schema version. BUMP THIS whenever the token
// COUNTING semantics change so old, differently-counted totals are discarded
// rather than re-added onto. v2 is the first schema whose totals were produced
// by the HOLDBACK fold (a group is counted once, when provably complete); every
// pre-v2 ledger was produced by the buggy per-split-line counter (input ~2x,
// output ~2.6x inflated), so its totals are unreliable and loadLedger resets
// them (see the migration in loadLedger). v3 adds session/intent ownership and
// pending-group attribution; v2's workspace-only totals cannot be partitioned
// retrospectively, so they are rebuilt too.
const CURRENT_SCHEMA_VERSION = 3 as const;

// The durable rollup. `cursors` is keyed by SOURCE FILE identity - the OFFSET
// FOLD (foldFileIntoLedger) keys each cursor by the transcript FILE PATH (the
// main transcript's path; each sub-agent file's full path), which is unique
// ACROSS sessions so a cumulative workspace ledger never reuses one session's
// byteOffset against a different session's file. The row-based updateLedger -
// which has no file path - keys by the row's `sourceKey` (`"main"` /
// `"agent-<agentId>"`) instead; the two schemes coexist without collision
// because a file path never equals `"main"`. Verified: uuids collide across
// concurrent sub-agent files, so a global-uuid cursor would drop real turns or
// count the 0-token broadcast copies - per-file cursors are the whole mechanism.
// `byStage` carries stage-scoped sub-splits (StageBucket); top-level totals and
// splits span the workspace and are diagnostic only.
export type LedgerCursor = {
  lastUuid: string;
  lastTimestamp: string;
  // Byte position in the source file already folded. The offset-aware fold
  // reads only bytes `[byteOffset, size)` each call.
  byteOffset?: number;
  // The `message.id` of the last folded group (diagnostics only).
  lastMessageId?: string;
  // A non-flush fold retains the last message group. Its ownership is captured
  // NOW, before a lifecycle tool can advance state, and reused when the group
  // is eventually folded.
  pending?: {
    byteOffset: number;
    messageId: string;
    stageSlug: string | null;
    sessionKey: string;
    workflowKey: string;
  };
};

export type Ledger = UsageAggregate & {
  // Schema version of the totals below. Absent/`< CURRENT_SCHEMA_VERSION` on
  // disk => produced by an older, differently-counted fold => loadLedger
  // discards it and rebuilds from the transcript.
  schemaVersion: number;
  cursors: Record<string, LedgerCursor>;
  workflows: Record<string, WorkflowUsage>;
};

function emptyTokenCounts(): TokenCounts {
  return { input: 0, output: 0, cacheCreate5m: 0, cacheCreate1h: 0, cacheRead: 0 };
}

function emptyTotals(): Totals {
  return { tokens: emptyTokenCounts(), usd: 0 };
}

function emptyStageBucket(): StageBucket {
  return { totals: emptyTotals(), byModel: {}, byAgent: {} };
}

function emptyUsageAggregate(): UsageAggregate {
  return { totals: emptyTotals(), byStage: {}, byModel: {}, byAgent: {} };
}

function emptyLedger(): Ledger {
  return {
    schemaVersion: CURRENT_SCHEMA_VERSION,
    cursors: {},
    ...emptyUsageAggregate(),
    workflows: {},
  };
}

// The gitignored runtime ledger path: `aidlc/.aidlc-sessions/usage-ledger.json`.
export function ledgerPath(projectDir: string): string {
  return join(sessionsDir(projectDir), "usage-ledger.json");
}

export type UsageContext = {
  sessionId?: string;
  sessionKey?: string;
  workflowKey?: string;
};

// The transcript path is the stable session identity available to both the fold
// hook and statusline. A caller without a transcript can supply an explicit
// session key (the row-based test/utility seam) or fall back to session_id.
export function sessionUsageKey(
  transcriptPath?: string,
  sessionId?: string,
): string {
  const transcript = transcriptPath?.trim();
  if (transcript) return `transcript:${transcript}`;
  const session = safeSessionSegment(sessionId ?? "");
  return session ? `session:${session}` : "session:unknown";
}

// Resolve the active intent to a stable UUID when possible. Legacy/orphan
// records fall back to their space + record-dir identity.
export function intentUsageKey(
  projectDir: string,
  sessionId?: string,
): string {
  try {
    if (sessionId) {
      const stamped = readSessionIntentUuid(projectDir, sessionId);
      if (stamped) return `intent:${stamped}`;
    }
    const selection = resolveWorkflowSelection(projectDir, { sessionId });
    const uuid = intentUuidForSelection(projectDir, selection);
    if (uuid) return `intent:${uuid}`;
    return `record:${selection.space}/${selection.intent ?? "legacy"}`;
  } catch {
    return "record:default/legacy";
  }
}

// Coerce a possibly-old-shape byStage map into the StageBucket shape. An older
// ledger stored a flat Totals per stage; treat that as `{ totals, byModel:{},
// byAgent:{} }`. A missing sub-map on a new-shape entry becomes {}. Never throws.
function normalizeByStage(
  raw: Record<string, unknown> | undefined,
): Record<string, StageBucket> {
  const out: Record<string, StageBucket> = {};
  if (!raw || typeof raw !== "object") return out;
  for (const [slug, val] of Object.entries(raw)) {
    const v = (val ?? {}) as Record<string, unknown>;
    if (v.totals && typeof v.totals === "object") {
      // New shape (StageBucket) - tolerate missing sub-maps.
      out[slug] = {
        totals: v.totals as Totals,
        byModel: (v.byModel as Record<string, Totals>) ?? {},
        byAgent: (v.byAgent as Record<string, Totals>) ?? {},
      };
    } else if (v.tokens && typeof v.tokens === "object") {
      // Old shape (flat Totals) - wrap it, no per-model/agent history available.
      out[slug] = { totals: v as unknown as Totals, byModel: {}, byAgent: {} };
    } else {
      out[slug] = emptyStageBucket();
    }
  }
  return out;
}

function normalizeUsageAggregate(raw: unknown): UsageAggregate {
  if (!raw || typeof raw !== "object") return emptyUsageAggregate();
  const value = raw as Partial<UsageAggregate> & {
    byStage?: Record<string, unknown>;
  };
  return {
    totals: value.totals ?? emptyTotals(),
    byStage: normalizeByStage(value.byStage),
    byModel: value.byModel ?? {},
    byAgent: value.byAgent ?? {},
  };
}

function normalizeAggregateMap(
  raw: Record<string, unknown> | undefined,
): Record<string, UsageAggregate> {
  const out: Record<string, UsageAggregate> = {};
  if (!raw || typeof raw !== "object") return out;
  for (const [key, value] of Object.entries(raw)) {
    out[key] = normalizeUsageAggregate(value);
  }
  return out;
}

function normalizeWorkflowMap(
  raw: Record<string, unknown> | undefined,
): Record<string, WorkflowUsage> {
  const out: Record<string, WorkflowUsage> = {};
  if (!raw || typeof raw !== "object") return out;
  for (const [key, value] of Object.entries(raw)) {
    const aggregate = normalizeUsageAggregate(value);
    const sessions =
      value && typeof value === "object"
        ? normalizeAggregateMap(
            (value as { sessions?: Record<string, unknown> }).sessions,
          )
        : {};
    out[key] = { ...aggregate, sessions };
  }
  return out;
}

// Whether an on-disk ledger's cursors are all pre-v2 shaped (belt-and-suspenders
// migration signal). A current fold cursor ALWAYS carries a numeric
// `byteOffset`; a
// pre-v2 ledger's cursors carry only `lastUuid`/`lastMessageId`. If ANY cursor
// lacks a byteOffset the totals below it were produced (at least partly) by the
// legacy per-line counter, so we treat the whole ledger as pre-v2. An empty
// cursors map is NOT a downgrade signal (a fresh ledger has none yet); the
// schemaVersion check is authoritative there.
function cursorsLackByteOffset(
  cursors: Record<string, unknown> | undefined,
): boolean {
  if (!cursors || typeof cursors !== "object") return false;
  for (const c of Object.values(cursors)) {
    if (
      c === null ||
      typeof c !== "object" ||
      typeof (c as { byteOffset?: unknown }).byteOffset !== "number"
    ) {
      return true;
    }
  }
  return false;
}

// Load the ledger, or a fresh empty one on missing/corrupt/invalid-shape input
// (rebuild-on-corruption). Never throws, logs nothing.
//
// MIGRATION. A ledger whose on-disk `schemaVersion` is missing or
// `< CURRENT_SCHEMA_VERSION` - or (belt-and-suspenders) whose cursors lack a
// numeric byteOffset - had its totals produced by the buggy per-split-line
// counter. Those totals are UNRELIABLE and must not be re-added onto, so we
// DISCARD the whole file and return a fresh current-schema ledger; the next fold
// rebuilds totals/byStage/byModel/byAgent/cursors cleanly from the
// transcript(s). This resets, rather than migrates, because the old numbers
// cannot be corrected in place.
export function loadLedger(projectDir: string): Ledger {
  let raw: string;
  try {
    raw = readFileSync(ledgerPath(projectDir), "utf-8");
  } catch {
    return emptyLedger();
  }
  try {
    const parsed = JSON.parse(raw) as Partial<Ledger> & {
      byStage?: Record<string, unknown>;
      schemaVersion?: unknown;
      cursors?: Record<string, unknown>;
      workflows?: Record<string, unknown>;
    };
    if (!parsed || typeof parsed !== "object") return emptyLedger();
    // Migration reset: an older ledger cannot supply the current ownership and
    // counting guarantees, so discard and rebuild rather than folding onto it.
    const onDiskVersion =
      typeof parsed.schemaVersion === "number" ? parsed.schemaVersion : 0;
    if (
      onDiskVersion < CURRENT_SCHEMA_VERSION ||
      cursorsLackByteOffset(parsed.cursors)
    ) {
      return emptyLedger();
    }
    // Merge onto an empty ledger so a partial/old file can't crash a consumer.
    return {
      schemaVersion: CURRENT_SCHEMA_VERSION,
      cursors: (parsed.cursors as Record<string, LedgerCursor>) ?? {},
      totals: parsed.totals ?? emptyTotals(),
      byStage: normalizeByStage(parsed.byStage),
      byModel: parsed.byModel ?? {},
      byAgent: parsed.byAgent ?? {},
      workflows: normalizeWorkflowMap(parsed.workflows),
    };
  } catch {
    return emptyLedger();
  }
}

// Fold one row's counts + usd into a Totals accumulator (mutates + returns it).
function addInto(t: Totals, row: UsageRow): Totals {
  t.tokens.input += row.counts.input;
  t.tokens.output += row.counts.output;
  t.tokens.cacheCreate5m += row.counts.cacheCreate5m;
  t.tokens.cacheCreate1h += row.counts.cacheCreate1h;
  t.tokens.cacheRead += row.counts.cacheRead;
  if (typeof row.usd === "number" && Number.isFinite(row.usd)) t.usd += row.usd;
  return t;
}

// The bucket key for byModel: the normalized rate-table key when known, else the
// raw model string, else "unknown". Keeps unknown-model tokens visible.
function modelBucketKey(row: UsageRow): string {
  return normalizeModel(row.model) ?? row.model ?? "unknown";
}

function foldRowIntoAggregate(
  aggregate: UsageAggregate,
  row: UsageRow,
  stageSlug: string | null,
): void {
  addInto(aggregate.totals, row);
  const mk = modelBucketKey(row);
  const ak = row.agentType ?? "subagent";
  aggregate.byModel[mk] = addInto(aggregate.byModel[mk] ?? emptyTotals(), row);
  aggregate.byAgent[ak] = addInto(aggregate.byAgent[ak] ?? emptyTotals(), row);
  if (stageSlug) {
    const stage = aggregate.byStage[stageSlug] ?? emptyStageBucket();
    addInto(stage.totals, row);
    // Stage-scoped sub-splits mirror the aggregate folds so By Model / By Agent
    // on STAGE_COMPLETED agree with the stage's own Cost USD.
    stage.byModel[mk] = addInto(stage.byModel[mk] ?? emptyTotals(), row);
    stage.byAgent[ak] = addInto(stage.byAgent[ak] ?? emptyTotals(), row);
    aggregate.byStage[stageSlug] = stage;
  }
}

// Fold one row into the workspace diagnostic aggregate and the authoritative
// session + intent aggregates.
function foldRowIntoLedger(
  ledger: Ledger,
  row: UsageRow,
  stageSlug: string | null,
  sessionKey: string,
  workflowKey: string,
): void {
  foldRowIntoAggregate(ledger, row, stageSlug);
  const workflow = ledger.workflows[workflowKey] ?? {
    ...emptyUsageAggregate(),
    sessions: {},
  };
  foldRowIntoAggregate(workflow, row, stageSlug);
  const session = workflow.sessions[sessionKey] ?? emptyUsageAggregate();
  foldRowIntoAggregate(session, row, stageSlug);
  workflow.sessions[sessionKey] = session;
  ledger.workflows[workflowKey] = workflow;
}

const USAGE_LOCK_INTENT = "__usage-ledger__";
const USAGE_LOCK_SPACE = "__runtime__";
// The directory lock's stale-reaper restore gap can admit two holders on Win32.
// A kernel mutex closes that gap and transfers ownership after an abandoned owner.
const WIN32_USAGE_MUTEX =
  process.platform === "win32"
    ? dlopen("kernel32.dll", {
        CreateMutexW: { args: ["ptr", "i32", "ptr"], returns: "ptr" },
        WaitForSingleObject: { args: ["ptr", "u32"], returns: "u32" },
        ReleaseMutex: { args: ["ptr"], returns: "i32" },
        CloseHandle: { args: ["ptr"], returns: "i32" },
      })
    : null;

const WAIT_OBJECT_0 = 0;
const WAIT_ABANDONED = 0x80;
const USAGE_MUTEX_WAIT_MS = 5000;

function withUsageLedgerLock(projectDir: string, fn: () => Ledger): Ledger {
  if (WIN32_USAGE_MUTEX !== null) {
    const identity = auditLockIdentity(
      projectDir,
      USAGE_LOCK_INTENT,
      USAGE_LOCK_SPACE,
    );
    const hash = createHash("sha256").update(identity).digest("hex").slice(0, 32);
    const name = Buffer.from(`Global\\aidlc-usage-${hash}\0`, "utf16le");
    const handle = WIN32_USAGE_MUTEX.symbols.CreateMutexW(null, 0, ptr(name));
    if (handle === null) {
      throw new Error("Failed to create the Windows usage-ledger mutex");
    }
    const waitResult = WIN32_USAGE_MUTEX.symbols.WaitForSingleObject(
      handle,
      USAGE_MUTEX_WAIT_MS,
    );
    if (waitResult !== WAIT_OBJECT_0 && waitResult !== WAIT_ABANDONED) {
      WIN32_USAGE_MUTEX.symbols.CloseHandle(handle);
      throw new Error(
        `Failed to acquire the Windows usage-ledger mutex: ${waitResult}`,
      );
    }
    try {
      return fn();
    } finally {
      WIN32_USAGE_MUTEX.symbols.ReleaseMutex(handle);
      WIN32_USAGE_MUTEX.symbols.CloseHandle(handle);
    }
  }
  return withAuditLock(
    projectDir,
    fn,
    USAGE_LOCK_INTENT,
    USAGE_LOCK_SPACE,
    200,
    25,
  );
}

// Incrementally fold new rows into the ledger and advance the per-file cursors.
//
// 1. Group rows by sourceKey (file identity).
// 2. Rows are already in file order from extraction - do NOT re-sort across
//    files by timestamp (timestamps collide across concurrent sub-agents).
// 3. If a group has a cursor, drop rows up to AND INCLUDING lastUuid (append-only
//    files => everything after the checkpoint is new). No cursor => all new.
// 4. Fold survivors into totals / byModel / byAgent, and byStage[stageSlug] when
//    a stage is supplied.
// 5. Advance cursors[sourceKey] to each group's last folded row.
// 6. Write atomically.
//
// Duplicate-turn guard: a broadcast turn appears in several files with
// output_tokens:0 in the copies; folding stays correct because each file is
// counted ONLY against its own cursor. Do NOT dedupe by uuid across files - that
// would drop the one real-token copy.
export function updateLedger(
  projectDir: string,
  rows: UsageRow[],
  stageSlug: string | null,
  context: UsageContext = {},
): Ledger {
  return withUsageLedgerLock(projectDir, () => {
    const ledger = loadLedger(projectDir);
    const sessionKey =
      context.sessionKey ?? sessionUsageKey(undefined, context.sessionId);
    const workflowKey =
      context.workflowKey ?? intentUsageKey(projectDir, context.sessionId);

    // 1. Group by sourceKey, preserving file order within each group.
    const groups = new Map<string, UsageRow[]>();
    for (const row of rows) {
      const arr = groups.get(row.sourceKey) ?? [];
      arr.push(row);
      groups.set(row.sourceKey, arr);
    }

    for (const [sourceKey, rawGroupRows] of groups) {
      // Collapse split-line overcount WITHIN this file's group (a caller may pass
      // raw split rows). Per-group so the dedup never crosses files - a broadcast
      // turn sharing an id across concurrent sub-agent files must still count once
      // per file (the per-file-cursor invariant). Reuses the same pure helper the
      // whole-file reader uses.
      const groupRows = dedupeByMessageId(rawGroupRows);
      const cursor = ledger.cursors[sourceKey];
      let fresh = groupRows;
      if (cursor?.lastUuid) {
        const idx = groupRows.findIndex((r) => r.uuid === cursor.lastUuid);
        // Everything AFTER the checkpoint row is new. If the checkpoint uuid isn't
        // in this batch (e.g. the batch is entirely older, or the file was
        // truncated), keep all rows - a re-fold is harmless per idempotency.
        fresh = idx >= 0 ? groupRows.slice(idx + 1) : groupRows;
      }
      if (fresh.length === 0) continue;

      for (const row of fresh) {
        foldRowIntoLedger(ledger, row, stageSlug, sessionKey, workflowKey);
      }

      const last = fresh[fresh.length - 1];
      // Preserve any byteOffset the file-fold layer set for this source - the
      // row-based API advances uuid/timestamp/msgId but does not track bytes.
      // ALWAYS write a numeric byteOffset (default 0) so this API produces a
      // cursor-complete current-schema ledger. This is safe alongside the fold
      // layer because the two keyspaces are disjoint: updateLedger keys by
      // row.sourceKey (`"main"`/`"agent-<id>"`) while foldFileIntoLedger keys by
      // FILE PATH, so a 0 here is never read by a fold.
      const prevOffset = ledger.cursors[sourceKey]?.byteOffset ?? 0;
      ledger.cursors[sourceKey] = {
        lastUuid: last.uuid,
        lastTimestamp: last.timestamp,
        lastMessageId:
          last.msgId || ledger.cursors[sourceKey]?.lastMessageId || "",
        byteOffset: prevOffset,
      };
    }

    // 6. Persist atomically (mkdir the sessions dir first).
    try {
      mkdirSync(sessionsDir(projectDir), { recursive: true });
    } catch {
      /* dir may already exist */
    }
    writeFileAtomic(ledgerPath(projectDir), JSON.stringify(ledger, null, 2));
    return ledger;
  });
}

// ===========================================================================
// Task 5 - Audit per-stage rollup fields (ledger read only, NO transcript I/O)
// ===========================================================================

// Compact token count for audit field values: 1234 -> "1.2k", 3_400_000 ->
// "3.4M", sub-1000 -> the integer, 0/negative/non-finite -> "0". Mirrors
// aidlc-statusline.ts fmtTokens exactly (kept a LOCAL copy - usage.ts must not
// import a hook, and the metrics reader un-abbreviates this same form). One
// decimal at k/M scale, trailing ".0" trimmed.
function fmtTokensCompact(n: number): string {
  if (!Number.isFinite(n) || n <= 0) return "0";
  const trim = (s: string): string => s.replace(/\.0$/, "");
  if (n >= 1e6) return `${trim((n / 1e6).toFixed(1))}M`;
  if (n >= 1e3) return `${trim((n / 1e3).toFixed(1))}k`;
  return String(Math.round(n));
}

// Format a per-key COST breakdown for the `By Agent` line - `k=1.23; k2=0.45`,
// USD to 2dp, only buckets with a positive cost. Empty -> "". (Agent buckets
// can't classify unknown-vs-zero at agent granularity, so zero-cost agents are
// simply omitted here; their TOKENS still surface via the token-breakdown
// fields below, so no agent is hidden.)
function formatByCost(buckets: Record<string, Totals>): string {
  const parts: string[] = [];
  for (const [key, t] of Object.entries(buckets)) {
    if (t.usd > 0) parts.push(`${key}=${t.usd.toFixed(2)}`);
  }
  return parts.join("; ");
}

// Format the per-MODEL cost breakdown with null-visibility: a known
// (rate-keyed) model prints its cost `opus-4-8=1.23` (even `=0.00`); an
// unknown-model bucket prints `<rawModel>=null` so an unpriceable slice is
// VISIBLE, never silently dropped. Empty map -> "".
function formatModelCost(buckets: Record<string, Totals>): string {
  const rates = loadRates();
  const parts: string[] = [];
  for (const [key, t] of Object.entries(buckets)) {
    parts.push(key in rates ? `${key}=${t.usd.toFixed(2)}` : `${key}=null`);
  }
  return parts.join("; ");
}

// Format a per-key TOKEN breakdown - `k=in/out/cacheRead/cacheWrite; k2=...`
// with each count in compact form (`12.3k/4.1k/2k/79.6k`). The quad order is
// input / output / cacheRead / cacheWrite, documented in
// docs/reference/06-hooks-and-tools.md. cacheWrite = cacheCreate5m +
// cacheCreate1h - both TTLs summed into one bucket, mirroring the single Cache
// Read bucket (the harness issues 5m-TTL writes in practice, but summing is
// correct for either TTL and matches computeCost, which prices both). A bucket
// whose four counts are all zero is skipped. Empty -> "". This is what makes the
// four token counts visible per model AND per agent, not just cost. The metrics
// reader parses this exact shape (un-abbreviating the compact values) into
// per-model / per-agent token magnitude lines.
function formatByTokens(buckets: Record<string, Totals>): string {
  const parts: string[] = [];
  for (const [key, t] of Object.entries(buckets)) {
    const { input, output, cacheRead } = t.tokens;
    const cacheWrite = t.tokens.cacheCreate5m + t.tokens.cacheCreate1h;
    if (input <= 0 && output <= 0 && cacheRead <= 0 && cacheWrite <= 0) continue;
    parts.push(
      `${key}=${fmtTokensCompact(input)}/${fmtTokensCompact(output)}/${fmtTokensCompact(cacheRead)}/${fmtTokensCompact(cacheWrite)}`,
    );
  }
  return parts.join("; ");
}

// Whether a stage bucket carries any priceable (known-model) sub-bucket.
function hasKnownModel(byModel: Record<string, Totals>): boolean {
  const rates = loadRates();
  return Object.keys(byModel).some((k) => k in rates);
}

// Whether a stage recorded any tokens at all.
function hasAnyTokens(t: Totals): boolean {
  const c = t.tokens;
  return (
    c.input > 0 ||
    c.output > 0 ||
    c.cacheCreate5m > 0 ||
    c.cacheCreate1h > 0 ||
    c.cacheRead > 0
  );
}

// Compact usage summaries for STAGE_COMPLETED / WORKFLOW_COMPLETED, read from
// the authoritative workflow aggregate - no transcript re-parse or time-slicing.
//
// THREE cost states are distinguished:
//   (a) no usage data for the stage      => {} (no fields; caller stays clean)
//   (b) usage recorded but UNPRICEABLE   => `Cost USD: null` (all unknown-model)
//   (c) priced                           => `Cost USD: 1.23`
// A mixed known+unknown stage prices the KNOWN portion (state c) and the
// `By Model` line shows the unknown slice as `<model>=null` - no fabricated cost.
//
// `Tokens By Model` / `Tokens By Agent` carry the four token counts
// (in/out/cacheRead/cacheWrite) per model and per agent, so the breakdowns are
// token-aware, not cost-only.
function aggregateUsageAuditFields(
  stage: Pick<UsageAggregate, "totals" | "byModel" | "byAgent">,
): Record<string, string> {
  const t = stage.totals;
  const fields: Record<string, string> = {
    "Tokens In": String(t.tokens.input),
    "Tokens Out": String(t.tokens.output),
    "Cache Read": String(t.tokens.cacheRead),
    "Cache Write": String(t.tokens.cacheCreate5m + t.tokens.cacheCreate1h),
  };
  // Cost USD - three-state. Priced => 2dp; unpriceable-but-used => "null";
  // genuinely no usage => omitted.
  if (t.usd > 0 || hasKnownModel(stage.byModel)) {
    fields["Cost USD"] = t.usd.toFixed(2);
  } else if (hasAnyTokens(t)) {
    fields["Cost USD"] = "null"; // recorded usage, all unknown-model => unpriceable
  }
  // Stage-scoped breakdowns - read the stage's OWN sub-maps, never the global
  // byModel/byAgent (those sum every stage and would contradict Cost USD).
  const byModel = formatModelCost(stage.byModel);
  if (byModel) fields["By Model"] = byModel;
  const byAgent = formatByCost(stage.byAgent);
  if (byAgent) fields["By Agent"] = byAgent;
  // Token counts grouped by model and by agent (in/out/cacheRead/cacheWrite).
  const tokModel = formatByTokens(stage.byModel);
  if (tokModel) fields["Tokens By Model"] = tokModel;
  const tokAgent = formatByTokens(stage.byAgent);
  if (tokAgent) fields["Tokens By Agent"] = tokAgent;
  return fields;
}

export function stageUsageAuditFields(
  projectDir: string,
  stageSlug: string,
  workflowKey: string = intentUsageKey(projectDir),
): Record<string, string> {
  if (usageTrackingDisabled()) return {};
  const stage =
    loadLedger(projectDir).workflows[workflowKey]?.byStage[stageSlug];
  return stage ? aggregateUsageAuditFields(stage) : {};
}

export function workflowUsageAuditFields(
  projectDir: string,
  workflowKey: string = intentUsageKey(projectDir),
): Record<string, string> {
  if (usageTrackingDisabled()) return {};
  const workflow = loadLedger(projectDir).workflows[workflowKey];
  return workflow && hasAnyTokens(workflow.totals)
    ? aggregateUsageAuditFields(workflow)
    : {};
}

export function sessionUsageAggregate(
  projectDir: string,
  transcriptPath?: string,
  workflowKey?: string,
  sessionId?: string,
): UsageAggregate | null {
  if (usageTrackingDisabled()) return null;
  const resolvedTranscript =
    transcriptPath ?? (sessionId ? readCurrentTranscriptPath(projectDir, sessionId) : null);
  const sessionKey = sessionUsageKey(resolvedTranscript ?? undefined, sessionId);
  const resolvedWorkflowKey =
    workflowKey ?? intentUsageKey(projectDir, sessionId);
  return (
    loadLedger(projectDir).workflows[resolvedWorkflowKey]?.sessions[sessionKey] ??
    null
  );
}

// ===========================================================================
// Task 6 - Persisted transcript path + runtime ledger producer (usage.ts half)
// ===========================================================================

// Normalise a session id to a safe filename segment (mirrors aidlc-lib's
// sessionRecordPath sanitisation) so a host-supplied id can't escape the dir.
function safeSessionSegment(sessionId: string): string {
  return sessionId.replace(/[^A-Za-z0-9._-]+/g, "-").replace(/^-+|-+$/g, "");
}

// Atomically persist the live transcript path so transcript-blind CLI tools
// (state, statusline) can find it. Writes both a session-scoped
// `<sessionId>.transcript` (safe for concurrent sessions) AND a
// `current.transcript` convenience pointer for readers that lack a session id.
// Best-effort: swallows write errors (never breaks a hook).
export function writeCurrentTranscriptPath(
  projectDir: string,
  sessionId: string,
  transcriptPath: string,
): void {
  // Only usage consumers read these pointers, so the kill switch covers them.
  if (usageTrackingDisabled()) return;
  if (!transcriptPath) return;
  try {
    mkdirSync(sessionsDir(projectDir), { recursive: true });
  } catch {
    /* dir may already exist */
  }
  const seg = safeSessionSegment(sessionId);
  try {
    if (seg) {
      writeFileAtomic(
        join(sessionsDir(projectDir), `${seg}.transcript`),
        transcriptPath,
      );
    }
    writeFileAtomic(
      join(sessionsDir(projectDir), "current.transcript"),
      transcriptPath,
    );
  } catch {
    /* best-effort */
  }
}

// Read back the persisted transcript path. With a sessionId, reads the
// session-scoped file; without one, falls back to `current.transcript`. Returns
// null if absent/empty. Never throws.
export function readCurrentTranscriptPath(
  projectDir: string,
  sessionId?: string,
): string | null {
  const candidates: string[] = [];
  if (sessionId) {
    const seg = safeSessionSegment(sessionId);
    if (seg) candidates.push(join(sessionsDir(projectDir), `${seg}.transcript`));
  } else {
    candidates.push(join(sessionsDir(projectDir), "current.transcript"));
  }
  for (const path of candidates) {
    try {
      const raw = readFileSync(path, "utf-8").trim();
      if (raw) return raw;
    } catch {
      /* try next candidate */
    }
  }
  return null;
}

// The result of an offset-aware chunk read: the COMPLETE (newline-terminated)
// lines newly available since `byteOffset`, the ABSOLUTE byte start position of
// each of those lines (parallel to `lines`, so the caller can rewind the offset
// to where a held-back group begins), and the byte position up to which those
// complete lines run. A trailing partial line (no final `\n` - a half-written
// append) is DROPPED and left for the next fold. `reset` signals a
// truncation/rotation was detected (size < byteOffset) so the caller resets the
// per-file checkpoint too. `null` means nothing to do (file gone, or
// size === byteOffset with no new bytes).
type ChunkRead = {
  lines: string[];
  lineByteStarts: number[];
  newByteOffset: number;
  reset: boolean;
  trailingPartial: boolean;
};

// Read only bytes `[byteOffset, size)` of a file, BYTE-accurately (UTF-8 multi-
// byte safe - offsets are byte positions, never char indices). Returns the
// complete lines in that window, each line's absolute byte start, and the
// advanced byte offset. Never throws.
function readChunkFromOffset(
  path: string,
  byteOffset: number,
  flush: boolean,
): ChunkRead | null {
  let fd: number;
  try {
    fd = openSync(path, "r");
  } catch {
    return null; // file absent / unreadable
  }
  try {
    const size = fstatSync(fd).size;
    let reset = false;
    let start = byteOffset;
    if (size === start) return null; // no new bytes - zero parse
    if (size < start) {
      // Truncation / rotation - start over from the top.
      reset = true;
      start = 0;
    }
    const len = size - start;
    const buf = Buffer.allocUnsafe(len);
    let read = 0;
    while (read < len) {
      const n = readSync(fd, buf, read, len - read, start + read);
      if (n <= 0) break;
      read += n;
    }
    const chunk = buf.subarray(0, read);
    // Find the LAST newline byte in the chunk. Everything up to and including it
    // is complete lines; any bytes after it are a partial trailing write we drop
    // until the next fold completes the line.
    const lastNl = chunk.lastIndexOf(0x0a); // '\n'
    // Split the complete region on newline bytes, recording each line's ABSOLUTE
    // byte start (start + relative offset). Byte-accurate: we scan the raw buffer
    // rather than string-splitting so a multi-byte char never skews an offset.
    const lines: string[] = [];
    const lineByteStarts: number[] = [];
    let lineStart = 0;
    for (let idx = 0; idx <= lastNl; idx++) {
      if (chunk[idx] === 0x0a) {
        lines.push(chunk.subarray(lineStart, idx).toString("utf-8"));
        lineByteStarts.push(start + lineStart);
        lineStart = idx + 1;
      }
    }
    let newByteOffset = lastNl >= 0 ? start + lastNl + 1 : start;
    let trailingPartial = lineStart < chunk.length;
    // A final flush may observe a fully-written JSON object whose writer never
    // appended `\n`. Parse only syntactically complete JSON at EOF; malformed
    // trailing bytes keep the old offset so a later append can complete them.
    if (flush && trailingPartial) {
      const trailing = chunk.subarray(lineStart).toString("utf-8");
      try {
        JSON.parse(trailing);
        lines.push(trailing);
        lineByteStarts.push(start + lineStart);
        newByteOffset = start + chunk.length;
        trailingPartial = false;
      } catch {
        // partial final JSON - retain it for a later fold
      }
    }
    return { lines, lineByteStarts, newByteOffset, reset, trailingPartial };
  } catch {
    return null;
  } finally {
    try {
      closeSync(fd);
    } catch {
      /* best-effort */
    }
  }
}

// Offset-aware fold of ONE source file into the ledger. Reads only the new bytes
// since the file's cursor, parses+groups them, HOLDS BACK the last (not-yet-
// provably-complete) group, folds the rest via the shared accumulation path, and
// advances the per-file cursor (byteOffset + lastUuid/lastTimestamp). Mutates
// `ledger`.
//
// CURSOR KEY. The cursor is keyed by the transcript FILE PATH (`path`), NOT the
// constant `sourceKey`. The workspace ledger is cumulative across sessions and
// each session is a DIFFERENT file, so a constant `"main"` key made one
// session's byteOffset get read against another session's longer file, skipping
// its early turns. A file path is unique per session, so cursors never collide.
// `sourceKey` ("main"/"agent-<id>") is still stamped on every ROW for
// byModel/byAgent attribution - only the CURSOR key changed.
//
// HOLDBACK. Claude Code splits one llm call (one message.id) across contiguous
// JSONL lines; a mid-turn fold may land on a chunk boundary INSIDE such a group,
// before its real-usage line is written (new-style transcripts carry usage=0 on
// the leading lines and the real usage on the LAST line of the run). The fix:
// never count a group until it is provably COMPLETE. A group is complete only if
// a DIFFERENT-msgId group follows it in this chunk (so a new group has started,
// closing the prior one), OR the caller passed `flush=true` (the turn truly
// ended - the Stop hook). The LAST group in a non-flush chunk is held back: it
// is NOT folded, and byteOffset is rewound to the BYTE START of that group so
// its lines are re-read in full next fold. Because a held-back group is never
// counted until complete and is always re-read whole, incremental folding equals
// a single whole-file flush fold for BOTH transcript styles, regardless of where
// a chunk boundary falls (mid-group or mid-line).
function foldFileIntoLedger(
  ledger: Ledger,
  sourceKey: string,
  path: string,
  fallbackAgentId: string | null,
  metaByAgentId: Record<string, string>,
  stageSlug: string | null,
  sessionKey: string,
  workflowKey: string,
  flush: boolean,
): void {
  // The cursor is keyed by FILE PATH, unique across sessions.
  const cursorKey = path;
  const cursor = ledger.cursors[cursorKey];
  const prevOffset = cursor?.byteOffset ?? 0;
  const chunk = readChunkFromOffset(path, prevOffset, flush);
  if (chunk === null) return; // no new bytes / unreadable

  // Parse each complete line, keeping the parallel byte-start so a held-back
  // group can rewind the offset to where its FIRST line begins.
  const parsed: UsageRow[] = [];
  const parsedByteStarts: number[] = [];
  for (let i = 0; i < chunk.lines.length; i++) {
    const row = parseTranscriptLine(chunk.lines[i], sourceKey, fallbackAgentId);
    if (row) {
      parsed.push(row);
      parsedByteStarts.push(chunk.lineByteStarts[i]);
    }
  }

  // Group contiguous same-msgId runs (the split-line collapse), recording each
  // group's representative row (max-usage line of the run) and its first line's
  // byte start. Id-less rows are singleton groups (never merged). Within one
  // chunk, same-msgId lines are always contiguous (verified), so a contiguous
  // grouping matches dedupeByMessageId exactly here.
  const groups: { rep: UsageRow; byteStart: number }[] = [];
  {
    let i = 0;
    while (i < parsed.length) {
      const row = parsed[i];
      const byteStart = parsedByteStarts[i];
      if (!row.msgId) {
        groups.push({ rep: row, byteStart });
        i++;
        continue;
      }
      let j = i;
      while (j + 1 < parsed.length && parsed[j + 1].msgId === row.msgId) j++;
      groups.push({ rep: representativeOfRun(parsed, i, j), byteStart });
      i = j + 1;
    }
  }

  // HOLDBACK. Fold every complete group; on a non-flush fold the LAST group is
  // not provably complete => hold it back and rewind the offset to its start so
  // it is re-read whole next time. A flush closes it only when EOF is clean: a
  // malformed trailing fragment may be another split line for the same message.
  const closeLastGroup = flush && !chunk.trailingPartial;
  let foldCount = groups.length;
  let newByteOffset = chunk.newByteOffset;
  if (!closeLastGroup && groups.length > 0) {
    foldCount = groups.length - 1;
    newByteOffset = groups[groups.length - 1].byteStart;
  }
  const toFold = groups.slice(0, foldCount).map((g) => g.rep);

  // Attribute agent types BEFORE folding (main -> "main", sub -> sidecar type,
  // else "subagent"), reusing the shared attributor.
  attributeAgents(toFold, metaByAgentId);

  const newCursor = {
    lastUuid: cursor?.lastUuid ?? "",
    lastTimestamp: cursor?.lastTimestamp ?? "",
    lastMessageId: cursor?.lastMessageId ?? "",
    byteOffset: newByteOffset,
    pending: chunk.reset ? undefined : cursor?.pending,
  };

  const priorPending = chunk.reset ? undefined : cursor?.pending;
  for (let i = 0; i < toFold.length; i++) {
    const group = groups[i];
    const captured =
      i === 0 &&
      priorPending !== undefined &&
      priorPending.byteOffset === group.byteStart
        ? priorPending
        : null;
    foldRowIntoLedger(
      ledger,
      toFold[i],
      captured?.stageSlug ?? stageSlug,
      captured?.sessionKey ?? sessionKey,
      captured?.workflowKey ?? workflowKey,
    );
  }
  if (toFold.length > 0) {
    const last = toFold[toFold.length - 1];
    newCursor.lastUuid = last.uuid;
    newCursor.lastTimestamp = last.timestamp;
    if (last.msgId) newCursor.lastMessageId = last.msgId; // diagnostics only
  }
  if (!closeLastGroup && groups.length > 0) {
    const held = groups[groups.length - 1];
    newCursor.pending =
      priorPending?.byteOffset === held.byteStart
        ? priorPending
        : {
            byteOffset: held.byteStart,
            messageId: held.rep.msgId,
            stageSlug,
            sessionKey,
            workflowKey,
          };
  } else if (closeLastGroup) {
    newCursor.pending = undefined;
  }
  ledger.cursors[cursorKey] = newCursor;
}

// The runtime ledger PRODUCER: read the transcript (main + sub-agents) and fold
// its NEW bytes into the ledger under the current stage. This is the runtime
// caller wired from the PreToolUse, PostToolUse, and Stop hooks (Claude only).
// Offset-aware: each fold reads only bytes appended since the last fold per file
// (main + each sub-agent sidecar), so a per-tool-call fold never re-parses whole
// files. Fully guarded - never throws into a hook; on any failure returns the
// existing ledger unchanged and persists nothing.
//
// Fold modes. `holdback` retains every file's last group for PostToolUse;
// `seal-main` closes only the main group; `flush-all` closes every complete
// group at an engine boundary or Stop.
export type FoldMode = "holdback" | "seal-main" | "flush-all";

export function foldTranscriptIntoLedger(
  projectDir: string,
  transcriptPath: string,
  stageSlug: string | null,
  modeOrFlush: FoldMode | boolean = "holdback",
  context: UsageContext = {},
): Ledger {
  // Kill switch: the producer writes nothing (existing ledger left untouched).
  if (usageTrackingDisabled()) return emptyLedger();
  try {
    return withUsageLedgerLock(projectDir, () => {
      const mode: FoldMode =
        typeof modeOrFlush === "boolean"
          ? modeOrFlush
            ? "flush-all"
            : "holdback"
          : modeOrFlush;
      const sessionKey =
        context.sessionKey ??
        sessionUsageKey(transcriptPath, context.sessionId);
      const workflowKey =
        context.workflowKey ?? intentUsageKey(projectDir, context.sessionId);
      const ledger = loadLedger(projectDir);

      // Main transcript.
      foldFileIntoLedger(
        ledger,
        "main",
        transcriptPath,
        null,
        {},
        stageSlug,
        sessionKey,
        workflowKey,
        mode !== "holdback",
      );

      // Each sibling sub-agent file, with its sidecar agentType map (rebuilt each
      // fold - sidecars are tiny and the set can grow between folds).
      const subDir = subagentDir(transcriptPath);
      let entries: string[] = [];
      try {
        if (existsSync(subDir)) {
          entries = readdirSync(subDir).filter(
            (f) => f.startsWith("agent-") && f.endsWith(".jsonl"),
          );
        }
      } catch {
        entries = [];
      }
      for (const file of entries) {
        const agentId = file.replace(/^agent-/, "").replace(/\.jsonl$/, "");
        const full = join(subDir, file);
        const metaByAgentId: Record<string, string> = {};
        const meta = readMetaSidecar(full);
        if (meta && typeof meta.agentType === "string" && meta.agentType) {
          metaByAgentId[agentId] = meta.agentType;
        }
        foldFileIntoLedger(
          ledger,
          `agent-${agentId}`,
          full,
          agentId,
          metaByAgentId,
          stageSlug,
          sessionKey,
          workflowKey,
          mode === "flush-all",
        );
      }

      // Persist atomically (mkdir the sessions dir first).
      try {
        mkdirSync(sessionsDir(projectDir), { recursive: true });
      } catch {
        /* dir may already exist */
      }
      writeFileAtomic(ledgerPath(projectDir), JSON.stringify(ledger, null, 2));
      return ledger;
    });
  } catch {
    return loadLedger(projectDir);
  }
}