/**
 * Several records, one roll-up, and every contributor's gaps still visible.
 *
 * Everything else in this repository assumes one operator with files on disk.
 * `--by-source` and `owners` divide a bill somebody already collected;
 * `fleetRollup` compares services whose logs one machine could open. None of
 * them answers the question a team actually has: **four people measured four
 * things, and nobody wants to email logs around.**
 *
 * So this module merges *documents*, not logs. Each contributor runs
 * `trazum profile --json` wherever their traffic already is, and hands over a
 * profile document — which carries no prompt text, no completion text, no
 * session keys and no credentials, and never has. The roll-up is a format and
 * a merge rather than a service: the transport is somebody else's problem,
 * deliberately, because a tool whose argument is that it reads your bill
 * without uploading it cannot also be the place everybody's bill is uploaded.
 *
 * **The merge is the easy half.** The half worth building carefully is what
 * *cannot* be merged, and this module refuses in four different ways:
 *
 * - **Findings that need the records.** Percentile shapes, conversation
 *   growth, repeated turns, truncation retries — every one of them is computed
 *   from individual calls, and a summary of a summary cannot reproduce them.
 *   They are named, with the contributors that had them, rather than dropped.
 * - **A day's dearest label.** Each contributor states its own; the merged
 *   answer needs per-label-per-day figures no document carries. Where two
 *   contributors share a day the answer is `null` with the reason attached,
 *   not the louder contributor's.
 * - **Overlap between contributors.** Two people exporting the same traffic
 *   double the bill, and nothing here can see it: the roll-up never sees a raw
 *   line, so the duplicate detection a single profile does is structurally out
 *   of reach. Every roll-up of more than one contributor says so.
 * - **Each contributor's own blind spots.** Unreadable lines, unpriced calls,
 *   a log with no clock. Summing them into one figure would say "3% of this
 *   roll-up is unpriced" when the truth is "one of your four machines is 90%
 *   unpriced and the other three are clean" — the same averaging-away this arc
 *   exists to refuse. They stay per contributor.
 *
 * No I/O: the caller hands over text it read, so this stays browser-safe and
 * the CLI keeps its monopoly on the filesystem.
 */

import { conform } from './conform.js';
import type { OutcomeTally } from './outcome.js';
import type { FieldCoverage, UsageBreakdown } from './usage.js';

/** A document somebody handed over, under the name it should answer to. */
export interface RollupInput {
  /** How this contributor is named in the roll-up. The caller's choice. */
  name: string;
  /** The profile document, as text — parsed and checked here, never trusted. */
  text: string;
}

/**
 * One contributor's own gap, kept whole.
 *
 * `usd` and `calls` are `null` where the kind does not have one, never `0`: a
 * gap with no money attached and a gap that cost nothing are different
 * statements, and the second one is a measurement.
 */
export interface ContributorGap {
  kind:
    /** Lines this contributor's parser could not read at all. */
    | 'unreadable-lines'
    /** Calls whose model the price catalogue does not know. */
    | 'unpriced-calls'
    /** No record carried a timestamp, so this contributor is in no day. */
    | 'no-clock'
    /** Some records carried a timestamp and some did not. */
    | 'partial-clock'
    /** No record carried a session. */
    | 'no-sessions'
    /** No record carried a label. */
    | 'no-labels'
    /** Duplicate lines this contributor found inside its own log. */
    | 'duplicate-lines'
    /** Days inside the window it asked for on which it recorded nothing. */
    | 'silent-days';
  detail: string;
  usd: number | null;
  calls: number | null;
}

/**
 * A stretch inside a claimed window that recorded nothing at all.
 *
 * Contiguous runs rather than a list of dates: a contributor claiming a year
 * and recording three days of traffic produces one entry per gap instead of
 * three hundred and sixty-two strings, and the reader can still see exactly
 * which days are missing.
 */
export interface SilentRun {
  /** First silent day, `YYYY-MM-DD` UTC. */
  from: string;
  /** Last silent day, inclusive. */
  to: string;
  days: number;
}

export interface RollupContributor {
  name: string;
  /**
   * The roll-up this contributor arrived through, or null when it was handed
   * over directly.
   *
   * Contributors are **flattened, never collapsed**: a roll-up of three
   * roll-ups lists twelve machines rather than three, because collapsing them
   * would average twelve sets of gaps into three and that is the averaging
   * this whole arc exists to refuse.
   */
  via: string | null;
  totalUsd: number;
  calls: number;
  /** The period this contributor's log covers, or null when it carried no clock. */
  span: { fromMs: number; toMs: number; calls: number } | null;
  /** The same span in days, or null. Stated, never extrapolated from. */
  spanDays: number | null;
  /**
   * The window this contributor **asked for**, when it filtered by one.
   *
   * A claim, not a measurement, and the distinction is the point. `span` says
   * what the records showed; this says what was gone looking for. A log whose
   * latest record is the 5th may be a log of a quiet week or a log that
   * stopped being written on the 5th, and only a claim can tell those apart —
   * so a contributor that made none gets `null` here and the roll-up says that
   * out loud rather than reading its span as a period.
   *
   * The window is half-open, `[sinceMs, untilMs)`, as `profileUsage` applies
   * it.
   */
  claimed: { sinceMs: number | null; untilMs: number | null } | null;
  /**
   * Days inside a fully bounded claim on which this contributor recorded
   * nothing, and how many there are.
   *
   * **Named rather than interpolated**, the way a year report names its
   * missing months. Whether a silent stretch is a quiet week or a broken
   * export is the reader's to know; that it is silent is this tool's to say,
   * and a roll-up that folded it into a smaller total would be wrong by an
   * unknown amount in the flattering direction.
   *
   * Null when there is nothing to measure against: no claim, a claim with only
   * one end, or a claim too long to enumerate.
   */
  silence: { runs: SilentRun[]; days: number } | null;
  /**
   * Records the contributor's own window could not place, because they carried
   * no clock — the honesty cost of filtering by one.
   *
   * Null when there was no window, never 0: zero would say a window excluded
   * nothing, and no window is a different statement.
   */
  undatedExcluded: number | null;
  gaps: ContributorGap[];
}

/** A finding that exists per contributor and does not roll up. */
export interface UnmergedFinding {
  finding: string;
  because: string;
  /** The contributors that had one, so the reader knows where to go and look. */
  presentIn: string[];
}

/**
 * What a roll-up cannot say about itself.
 *
 * String codes rather than prose, so a consumer can branch on them and the
 * renderings can carry the sentences. `annual-record` established the shape.
 */
export type RollupCaveat =
  /** More than one contributor, so overlap between them is unmeasurable. */
  | 'overlap-invisible'
  /** Contributors cover meaningfully different periods. */
  | 'mismatched-spans'
  /** Some contributor carried no clock at all. */
  | 'contributor-without-clock'
  /** A day drew from more than one contributor, so its dearest label is unknown. */
  | 'day-top-label-unknown'
  /** Two contributions were the same document. */
  | 'identical-contributions'
  /** A contribution was handed over and not merged. */
  | 'contribution-rejected'
  /** A contribution carried a numeric field this version cannot classify. */
  | 'unknown-fields-dropped'
  /** A contributor stated no window, so its span is all that is known of it. */
  | 'no-claimed-period'
  /** A contributor claimed days on which it recorded nothing. */
  | 'silence-inside-a-claim'
  /** A contributor claimed one end of a window and not the other. */
  | 'claim-not-bounded'
  /** A claim was too long to enumerate day by day, and was not. */
  | 'claim-too-long-to-enumerate'
  /** A contributor name appears more than once, so its money may be counted twice. */
  | 'contributor-named-twice';

export interface RollupDay {
  /** `YYYY-MM-DD`, UTC — the contributors' own bucketing, never re-derived. */
  day: string;
  usd: number;
  calls: number;
  /** How many contributors saw traffic on this day. */
  contributors: number;
  byModel: Array<{ model: string; usd: number; calls: number }>;
  /**
   * The dearest label of the day, or **null** when more than one contributor
   * covered it.
   *
   * A profile knows its own day's dearest label; the merged answer needs each
   * contributor's per-label-per-day spend, which no document carries. Picking
   * the larger of two contributors' answers is what a helpful implementation
   * would do, and it is wrong whenever a runner-up in both adds up to more
   * than either winner.
   */
  topLabel: string | null;
  topLabelUsd: number | null;
}

export interface RollupDocument {
  schemaVersion: 1;
  contributors: RollupContributor[];
  /**
   * Handed over and not merged, each with why. Never dropped in silence.
   *
   * `via` names the roll-up a rejection arrived through, when it came from
   * one. A rejection that stopped travelling at a nesting boundary would mean
   * a broken export could be made to disappear by adding a layer, which is the
   * one thing a format built out of other people's measurements must not
   * allow.
   */
  rejected: Array<{ name: string; via: string | null; because: string }>;
  /**
   * Contributions that were the same document, grouped, and what the repeats
   * added to the total.
   *
   * Merged rather than discarded, and stated rather than repaired — the rule a
   * single profile already applies to duplicate lines. Whether it is one export
   * handed over twice or two machines that genuinely produced identical
   * documents is the reader's to know, and this tool does not decide it by
   * throwing money away.
   *
   * The comparison is over the whole text, not a hash of it: a hash collision
   * would report a duplicate that is not one, and this figure exists to make
   * somebody distrust a total.
   */
  identicalContributions: { groups: string[][]; usd: number };
  total: UsageBreakdown;
  unpriced: UsageBreakdown;
  unpricedModels: string[];
  byLabel: Array<{ label: string; breakdown: UsageBreakdown }>;
  byModel: Array<{ model: string; breakdown: UsageBreakdown }>;
  byLabelAndModel: Array<{ label: string; model: string; breakdown: UsageBreakdown }>;
  spendByDay: RollupDay[];
  /** Earliest start to latest end, over contributors that carried a clock. */
  span: { fromMs: number; toMs: number; calls: number } | null;
  /**
   * Earliest claimed start to latest claimed end, over contributors that
   * stated a fully bounded window.
   *
   * **Kept apart from `span`, deliberately.** One is what the records showed
   * and the other is what somebody went looking for, and a roll-up that merged
   * them would answer "what period does this cover" with a number that is half
   * measurement and half intention. Null when no contributor claimed one.
   */
  claimedSpan: { fromMs: number; toMs: number; contributors: number } | null;
  fieldCoverage: FieldCoverage;
  outcomeTally: OutcomeTally;
  /** Summed **within-contributor** duplicates. Overlap between them is elsewhere. */
  duplicateLines: { count: number; usd: number };
  /**
   * Contributor names that appear more than once, across nesting.
   *
   * Handing over both a roll-up and one of the machines inside it counts that
   * machine's money twice, and unlike the identical-document check this one can
   * see it — the documents differ, but the name is the same. **Named, never
   * subtracted**: two machines genuinely called `api.json` in two teams is
   * possible, and deciding which case this is by removing money would be the
   * repair this tool never makes.
   */
  repeatedContributors: string[];
  notMerged: UnmergedFinding[];
  cannotSay: RollupCaveat[];
}

/**
 * Every numeric field of a breakdown, and how two of them combine.
 *
 * Listed rather than inferred, because both mistakes here are silent. A field
 * left out of both lists would vanish from every merged breakdown — a finding
 * present in each contribution and absent from the roll-up. And
 * `maxCallInputTokens` summed would report a fleet whose largest call is the
 * sum of four machines' largest calls, which is a number no call ever had, in
 * the direction that makes a context window look tight.
 *
 * `rollup.test.js` derives the field names from `usage.ts` and fails the build
 * when one is in neither list, so the next field added upstream cannot be
 * quietly dropped here.
 */
const BREAKDOWN_SUM = [
  'calls',
  'inputTokens',
  'cacheReadTokens',
  'cacheWriteTokens',
  'cacheWrite5mTokens',
  'cacheWrite1hTokens',
  'outputTokens',
  'assumedWriteTtlCalls',
  'inputUsd',
  'cacheReadUsd',
  'cacheWriteUsd',
  'outputUsd',
  'totalUsd',
  'cachedTokensAtInputRateUsd',
  'cacheWriteUsdIfAssumed1h',
  'truncatedCalls',
  'truncatedOutputUsd',
  'stopReasonCalls',
] as const;

/** Fields whose combination is a maximum, because a sum would invent a call. */
const BREAKDOWN_MAX = ['maxCallInputTokens'] as const;

/** Every counter of a coverage tally. All of them sum. */
const COVERAGE_FIELDS = [
  'label',
  'session',
  'outcome',
  'ts',
  'stopReason',
  'cacheTtl',
  'cacheWrites',
  'parsed',
] as const;

const emptyBreakdown = (): UsageBreakdown => {
  const out: Record<string, number> = {};
  for (const field of BREAKDOWN_SUM) out[field] = 0;
  for (const field of BREAKDOWN_MAX) out[field] = 0;
  return out as unknown as UsageBreakdown;
};

const emptyCoverage = (): FieldCoverage => {
  const out: Record<string, number> = {};
  for (const field of COVERAGE_FIELDS) out[field] = 0;
  return out as unknown as FieldCoverage;
};

const numberAt = (source: Record<string, unknown>, field: string): number => {
  const value = source[field];
  return typeof value === 'number' && Number.isFinite(value) ? value : 0;
};

const asString = (value: unknown): string | null =>
  typeof value === 'string' && value !== '' ? value : null;

const isRecord = (value: unknown): value is Record<string, unknown> =>
  typeof value === 'object' && value !== null && !Array.isArray(value);

/**
 * Adds one contribution's breakdown into an accumulator.
 *
 * Returns the numeric fields it did not recognise, so a document from a newer
 * Trazum is reported rather than half-merged. Guessing that an unknown number
 * is additive is how a ratio becomes four times itself.
 */
function addBreakdown(into: UsageBreakdown, from: unknown): string[] {
  if (!isRecord(from)) return [];
  const target = into as unknown as Record<string, number>;
  for (const field of BREAKDOWN_SUM) target[field] = (target[field] ?? 0) + numberAt(from, field);
  for (const field of BREAKDOWN_MAX) {
    target[field] = Math.max(target[field] ?? 0, numberAt(from, field));
  }
  const known = new Set<string>([...BREAKDOWN_SUM, ...BREAKDOWN_MAX]);
  return Object.keys(from).filter((key) => !known.has(key) && typeof from[key] === 'number');
}

/** One merged slice, keyed by whatever identifies it. */
interface KeyedSlice {
  label: string | null;
  model: string | null;
  breakdown: UsageBreakdown;
}

/**
 * Merges keyed breakdowns — labels, models, or the pair — into one map.
 *
 * The identity travels in the value rather than being parsed back out of the
 * key: a label may contain any character, including whatever separator a key
 * would join on, and a pair key split on a separator the label also uses
 * silently merges two different workloads.
 */
function mergeKeyed(
  into: Map<string, KeyedSlice>,
  rows: unknown,
  identityOf: (row: Record<string, unknown>) => { label: string | null; model: string | null } | null,
): void {
  if (!Array.isArray(rows)) return;
  for (const row of rows) {
    if (!isRecord(row)) continue;
    const identity = identityOf(row);
    if (identity === null) continue;
    const key = JSON.stringify([identity.label, identity.model]);
    const slice = into.get(key) ?? { ...identity, breakdown: emptyBreakdown() };
    addBreakdown(slice.breakdown, row.breakdown);
    into.set(key, slice);
  }
}

/** Largest bill first, the order somebody would act in. */
const byMoney = <T>(rows: T[], usd: (row: T) => number): T[] =>
  [...rows].sort((a, b) => usd(b) - usd(a));

const DAY_MS = 86_400_000;

/** UTC `YYYY-MM-DD` for an instant — the bucketing every day figure here uses. */
const dayOf = (ms: number): string => new Date(ms).toISOString().slice(0, 10);

/** Midnight UTC of the day an instant falls in. */
const midnightOf = (ms: number): number => Date.parse(`${dayOf(ms)}T00:00:00.000Z`);

/**
 * The longest claim this module will walk day by day: ten years.
 *
 * A bound rather than trust, because these documents come from elsewhere. A
 * contribution claiming `untilMs: 1e15` is a malformed document, not a team
 * with a long memory, and enumerating it would be thirty million iterations
 * inside a merge somebody ran on four files. The claim is kept and reported;
 * only the enumeration is refused, with a caveat saying so — a refusal never
 * arrives bare.
 */
const MAX_CLAIM_DAYS = 3660;

/**
 * The stretches of a claimed window on which nothing was recorded.
 *
 * Returns `null` when the claim cannot be walked: unbounded on either end, out
 * of order, or longer than the bound above. The window is half-open, so the
 * last claimed day is the one containing `untilMs - 1` — an `until` of
 * midnight claims up to the previous day and not that day, which is how the
 * profile's own filter reads it.
 */
function silenceIn(
  claim: { sinceMs: number | null; untilMs: number | null },
  daysWithTraffic: Set<string>,
): { runs: SilentRun[]; days: number } | null {
  const { sinceMs, untilMs } = claim;
  if (sinceMs === null || untilMs === null) return null;
  if (!Number.isFinite(sinceMs) || !Number.isFinite(untilMs)) return null;
  if (untilMs <= sinceMs) return null;
  const first = midnightOf(sinceMs);
  const last = midnightOf(untilMs - 1);
  const span = Math.round((last - first) / DAY_MS) + 1;
  if (span > MAX_CLAIM_DAYS) return null;

  const runs: SilentRun[] = [];
  let open: { from: string; to: string; days: number } | null = null;
  let days = 0;
  for (let at = first; at <= last; at += DAY_MS) {
    const day = dayOf(at);
    if (daysWithTraffic.has(day)) {
      if (open !== null) runs.push(open);
      open = null;
      continue;
    }
    days += 1;
    if (open === null) open = { from: day, to: day, days: 1 };
    else {
      open.to = day;
      open.days += 1;
    }
  }
  if (open !== null) runs.push(open);
  return { runs, days };
}

/** The findings a document carries per call, which no summary can reconstruct. */
const PER_RECORD_FINDINGS: Array<{ field: string; finding: string; because: string }> = [
  {
    field: 'conversations',
    finding: 'conversation growth',
    because:
      'it is measured over the turns of one session, and a document carries the growth rather than the turns',
  },
  {
    field: 'inputShapes',
    finding: 'input shape — the median and p95 call',
    because:
      'a percentile of two contributors is not a percentile of their percentiles, and the calls behind them are in neither document',
  },
  {
    field: 'outputShapes',
    finding: 'output concentration',
    because:
      'it asks what share of the calls holds what share of the output spend, and both shares need the individual calls',
  },
  {
    field: 'repeatedTurns',
    finding: 'repeated turns — the shape of a retry loop',
    because:
      'it compares consecutive calls in one session seconds apart, and consecutive calls are what a summary has already collapsed',
  },
  {
    field: 'truncationRetries',
    finding: 'truncation followed by a retry',
    because: 'it pairs a truncated answer with the call that followed it, and a document carries neither call',
  },
];

/**
 * Merges profile documents into one roll-up.
 *
 * Every input is checked against the `profile` contract before a figure of it
 * is used. A document that does not conform is **rejected with its reason and
 * not merged** — a roll-up that quietly skipped a malformed contribution would
 * report a total missing one machine's entire bill, and the reader would have
 * no way to tell that from that machine having spent nothing.
 */
export function rollUp(inputs: RollupInput[]): RollupDocument {
  const contributors: RollupContributor[] = [];
  const rejected: Array<{ name: string; via: string | null; because: string }> = [];
  const cannotSay = new Set<RollupCaveat>();

  const total = emptyBreakdown();
  const unpriced = emptyBreakdown();
  const unpricedModels = new Set<string>();
  const labels = new Map<string, KeyedSlice>();
  const models = new Map<string, KeyedSlice>();
  const pairs = new Map<string, KeyedSlice>();
  const coverage = emptyCoverage();
  const outcomeValues = new Map<string, { calls: number; usd: number }>();
  const outcome = { recorded: 0, parsed: 0, unrecordedUsd: 0 };
  const duplicateLines = { count: 0, usd: 0 };
  const unknownFields = new Set<string>();
  const days = new Map<
    string,
    {
      usd: number;
      calls: number;
      contributors: number;
      models: Map<string, { usd: number; calls: number }>;
      from: Set<string>;
      topLabel: string | null;
      topLabelUsd: number | null;
    }
  >();
  const findingsPresentIn = new Map<string, string[]>();
  /** Findings a nested roll-up already refused to merge, carried through. */
  const nestedFindings = new Map<string, { because: string; presentIn: string[] }>();
  /** Identical-document groups a nested roll-up already found. */
  const nestedIdentical: string[][] = [];
  /** The text of each accepted contribution, for the identical-document check. */
  const seenText = new Map<string, string[]>();
  let identicalUsd = 0;
  let claimFrom: number | null = null;
  let claimTo: number | null = null;
  let claimants = 0;
  let spanFrom: number | null = null;
  let spanTo: number | null = null;
  let spanCalls = 0;

  for (const input of inputs) {
    /**
     * A profile, or a roll-up somebody else already made.
     *
     * Detected rather than forced, because a roll-up is a contribution too:
     * three teams roll up their own machines, and the organisation rolls up
     * the three. Every summable part of a roll-up carries the same field names
     * as a profile — that is what makes the nesting arithmetic free — and the
     * parts that are *not* summable are exactly the ones this loop has to
     * carry through by hand rather than lose a layer at a time.
     */
    const checked = conform(input.text);
    const nested = checked.contract === 'roll-up';
    if (!checked.conforms || (checked.contract !== 'profile' && !nested)) {
      const first = checked.problems[0];
      rejected.push({
        name: input.name,
        via: null,
        because:
          first === undefined
            ? (checked.because ?? 'it is neither a profile nor a roll-up')
            : `${first.at}: ${first.detail}`,
      });
      cannotSay.add('contribution-rejected');
      continue;
    }

    // `conform` parsed it once already and reported nothing, so this cannot
    // throw — but a parse whose failure mode is a crash is a parse worth
    // guarding, and the reject path above is already the right answer.
    let parsed: unknown;
    try {
      parsed = JSON.parse(input.text.trim());
    } catch {
      rejected.push({ name: input.name, via: null, because: 'it is not valid JSON' });
      cannotSay.add('contribution-rejected');
      continue;
    }
    if (!isRecord(parsed)) {
      rejected.push({ name: input.name, via: null, because: 'a document must be a JSON object' });
      cannotSay.add('contribution-rejected');
      continue;
    }
    const doc = parsed;

    const text = input.text.trim();
    const sameAs = seenText.get(text);
    if (sameAs === undefined) {
      seenText.set(text, [input.name]);
    } else {
      sameAs.push(input.name);
      cannotSay.add('identical-contributions');
      identicalUsd += isRecord(doc.total) ? numberAt(doc.total, 'totalUsd') : 0;
    }

    for (const field of addBreakdown(total, doc.total)) unknownFields.add(field);
    addBreakdown(unpriced, doc.unpriced);
    if (Array.isArray(doc.unpricedModels)) {
      for (const model of doc.unpricedModels) {
        const name = asString(model);
        if (name !== null) unpricedModels.add(name);
      }
    }

    mergeKeyed(labels, doc.byLabel, (row) => {
      const label = asString(row.label);
      return label === null ? null : { label, model: null };
    });
    mergeKeyed(models, doc.byModel, (row) => {
      const model = asString(row.model);
      return model === null ? null : { label: null, model };
    });
    mergeKeyed(pairs, doc.byLabelAndModel, (row) => {
      const label = asString(row.label);
      const model = asString(row.model);
      return label === null || model === null ? null : { label, model };
    });

    if (isRecord(doc.fieldCoverage)) {
      const target = coverage as unknown as Record<string, number>;
      for (const field of COVERAGE_FIELDS) {
        target[field] = (target[field] ?? 0) + numberAt(doc.fieldCoverage, field);
      }
    }

    if (isRecord(doc.outcomeTally)) {
      const tally = doc.outcomeTally;
      outcome.recorded += numberAt(tally, 'recorded');
      outcome.parsed += numberAt(tally, 'parsed');
      outcome.unrecordedUsd += numberAt(tally, 'unrecordedUsd');
      if (Array.isArray(tally.byValue)) {
        for (const row of tally.byValue) {
          if (!isRecord(row)) continue;
          const value = asString(row.value);
          if (value === null) continue;
          const current = outcomeValues.get(value) ?? { calls: 0, usd: 0 };
          current.calls += numberAt(row, 'calls');
          current.usd += numberAt(row, 'usd');
          outcomeValues.set(value, current);
        }
      }
    }

    if (isRecord(doc.duplicateLines)) {
      duplicateLines.count += numberAt(doc.duplicateLines, 'count');
      duplicateLines.usd += numberAt(doc.duplicateLines, 'usd');
    }

    if (Array.isArray(doc.spendByDay)) {
      for (const row of doc.spendByDay) {
        if (!isRecord(row)) continue;
        const day = asString(row.day);
        if (day === null) continue;
        const bucket = days.get(day) ?? {
          usd: 0,
          calls: 0,
          contributors: 0,
          models: new Map<string, { usd: number; calls: number }>(),
          from: new Set<string>(),
          topLabel: null,
          topLabelUsd: null,
        };
        bucket.usd += numberAt(row, 'usd');
        bucket.calls += numberAt(row, 'calls');
        bucket.from.add(input.name);
        // How many *machines* saw this day, not how many documents mentioned
        // it: a nested roll-up already knows its own count, and counting it as
        // one would report twelve machines as three.
        bucket.contributors += Math.max(1, numberAt(row, 'contributors') || 1);
        if (bucket.from.size === 1) {
          bucket.topLabel = asString(row.topLabel);
          bucket.topLabelUsd = bucket.topLabel === null ? null : numberAt(row, 'topLabelUsd');
        } else {
          // A second contributor on this day puts the merged answer out of
          // reach, and the first contributor's answer stops being the
          // roll-up's answer.
          bucket.topLabel = null;
          bucket.topLabelUsd = null;
          cannotSay.add('day-top-label-unknown');
        }
        if (Array.isArray(row.byModel)) {
          for (const entry of row.byModel) {
            if (!isRecord(entry)) continue;
            const name = asString(entry.model);
            if (name === null) continue;
            const current = bucket.models.get(name) ?? { usd: 0, calls: 0 };
            current.usd += numberAt(entry, 'usd');
            current.calls += numberAt(entry, 'calls');
            bucket.models.set(name, current);
          }
        }
        days.set(day, bucket);
      }
    }

    for (const entry of PER_RECORD_FINDINGS) {
      const value = doc[entry.field];
      if (Array.isArray(value) && value.length > 0) {
        const list = findingsPresentIn.get(entry.field) ?? [];
        list.push(input.name);
        findingsPresentIn.set(entry.field, list);
      }
    }

    /**
     * A roll-up somebody else already made: carry through what does not sum.
     *
     * Everything above this point merged already, because a roll-up names its
     * summable parts exactly the way a profile does. What is left is the half
     * that would quietly disappear a layer at a time — and each of these is a
     * refusal that must survive nesting or the format is worse than no format:
     *
     * - **Contributors are flattened, not collapsed.** Twelve machines stay
     *   twelve machines with twelve sets of gaps.
     * - **Rejections travel, with the roll-up they came through.** Otherwise a
     *   machine whose document did not conform could be made to disappear by
     *   adding a layer.
     * - **Caveats travel.** An inner roll-up that could not see overlap does
     *   not become an outer roll-up that could.
     * - **Findings that did not roll up inside do not roll up outside.**
     */
    if (nested) {
      if (Array.isArray(doc.contributors)) {
        for (const row of doc.contributors) {
          if (!isRecord(row)) continue;
          const name = asString(row.name);
          if (name === null) continue;
          const innerSpan = isRecord(row.span)
            ? {
                fromMs: numberAt(row.span, 'fromMs'),
                toMs: numberAt(row.span, 'toMs'),
                calls: numberAt(row.span, 'calls'),
              }
            : null;
          contributors.push({
            name,
            // The roll-up it arrived through — never overwritten when it is
            // already set, so a third layer says which roll-up handed it over
            // rather than claiming the machine came straight from there.
            via: asString(row.via) ?? input.name,
            totalUsd: numberAt(row, 'totalUsd'),
            calls: numberAt(row, 'calls'),
            span: innerSpan,
            spanDays: typeof row.spanDays === 'number' ? row.spanDays : null,
            claimed: isRecord(row.claimed)
              ? {
                  sinceMs: typeof row.claimed.sinceMs === 'number' ? row.claimed.sinceMs : null,
                  untilMs: typeof row.claimed.untilMs === 'number' ? row.claimed.untilMs : null,
                }
              : null,
            silence: isRecord(row.silence)
              ? {
                  runs: Array.isArray(row.silence.runs) ? (row.silence.runs as SilentRun[]) : [],
                  days: numberAt(row.silence, 'days'),
                }
              : null,
            undatedExcluded:
              typeof row.undatedExcluded === 'number' ? row.undatedExcluded : null,
            gaps: Array.isArray(row.gaps) ? (row.gaps as ContributorGap[]) : [],
          });
        }
      }

      if (Array.isArray(doc.rejected)) {
        for (const row of doc.rejected) {
          if (!isRecord(row)) continue;
          const name = asString(row.name);
          if (name === null) continue;
          rejected.push({
            name,
            via: asString(row.via) ?? input.name,
            because: asString(row.because) ?? 'no reason travelled with it',
          });
          cannotSay.add('contribution-rejected');
        }
      }

      if (Array.isArray(doc.cannotSay)) {
        for (const code of doc.cannotSay) {
          const caveat = asString(code);
          if (caveat !== null) cannotSay.add(caveat as RollupCaveat);
        }
      }

      if (Array.isArray(doc.notMerged)) {
        for (const row of doc.notMerged) {
          if (!isRecord(row)) continue;
          const finding = asString(row.finding);
          if (finding === null) continue;
          const list = nestedFindings.get(finding) ?? {
            because: asString(row.because) ?? '',
            presentIn: [] as string[],
          };
          if (Array.isArray(row.presentIn)) {
            for (const who of row.presentIn) {
              const named = asString(who);
              if (named !== null && !list.presentIn.includes(named)) list.presentIn.push(named);
            }
          }
          nestedFindings.set(finding, list);
        }
      }

      if (isRecord(doc.identicalContributions)) {
        identicalUsd += numberAt(doc.identicalContributions, 'usd');
        if (Array.isArray(doc.identicalContributions.groups)) {
          for (const group of doc.identicalContributions.groups) {
            if (Array.isArray(group)) {
              nestedIdentical.push(group.map((name) => String(name)));
            }
          }
        }
      }

      if (isRecord(doc.claimedSpan)) {
        const from = numberAt(doc.claimedSpan, 'fromMs');
        const to = numberAt(doc.claimedSpan, 'toMs');
        if (from > 0 && to > from) {
          claimFrom = claimFrom === null ? from : Math.min(claimFrom, from);
          claimTo = claimTo === null ? to : Math.max(claimTo, to);
          claimants += Math.max(1, numberAt(doc.claimedSpan, 'contributors') || 1);
        }
      }

      if (isRecord(doc.span)) {
        const from = numberAt(doc.span, 'fromMs');
        const to = numberAt(doc.span, 'toMs');
        spanFrom = spanFrom === null ? from : Math.min(spanFrom, from);
        spanTo = spanTo === null ? to : Math.max(spanTo, to);
        spanCalls += numberAt(doc.span, 'calls');
      }

      continue;
    }

    // --- this contributor, as it will be listed ---------------------------
    const totals = isRecord(doc.total) ? doc.total : {};
    const ownUnpriced = isRecord(doc.unpriced) ? doc.unpriced : {};
    const ownCoverage = isRecord(doc.fieldCoverage) ? doc.fieldCoverage : {};
    const span = isRecord(doc.span)
      ? {
          fromMs: numberAt(doc.span, 'fromMs'),
          toMs: numberAt(doc.span, 'toMs'),
          calls: numberAt(doc.span, 'calls'),
        }
      : null;

    if (span !== null) {
      spanFrom = spanFrom === null ? span.fromMs : Math.min(spanFrom, span.fromMs);
      spanTo = spanTo === null ? span.toMs : Math.max(spanTo, span.toMs);
      spanCalls += span.calls;
    }

    const gaps: ContributorGap[] = [];
    const skipped = Array.isArray(doc.skippedLines) ? doc.skippedLines.length : 0;
    if (skipped > 0) {
      gaps.push({
        kind: 'unreadable-lines',
        // The positions are deliberately absent: a line number is an offset
        // into a file only this contributor has, and a merged list of them
        // points at nothing.
        detail: `${skipped} line${skipped === 1 ? '' : 's'} of this contributor's log could not be read`,
        usd: null,
        calls: null,
      });
    }
    const unpricedCalls = numberAt(ownUnpriced, 'calls');
    if (unpricedCalls > 0) {
      gaps.push({
        kind: 'unpriced-calls',
        detail: `${unpricedCalls} call${unpricedCalls === 1 ? '' : 's'} ran on a model the catalogue does not price`,
        usd: null,
        calls: unpricedCalls,
      });
    }
    const parsedRecords = numberAt(ownCoverage, 'parsed');
    const dated = numberAt(ownCoverage, 'ts');
    if (span === null) {
      gaps.push({
        kind: 'no-clock',
        detail: 'no record carried a timestamp, so none of this contributor is in any day',
        usd: null,
        calls: null,
      });
      cannotSay.add('contributor-without-clock');
    } else if (parsedRecords > 0 && dated < parsedRecords) {
      gaps.push({
        kind: 'partial-clock',
        detail: `${parsedRecords - dated} of ${parsedRecords} records carried no timestamp`,
        usd: null,
        calls: parsedRecords - dated,
      });
    }
    if (parsedRecords > 0 && numberAt(ownCoverage, 'session') === 0) {
      gaps.push({
        kind: 'no-sessions',
        detail: 'no record carried a session, so this contributor brings no conversation findings',
        usd: null,
        calls: null,
      });
    }
    if (parsedRecords > 0 && numberAt(ownCoverage, 'label') === 0) {
      gaps.push({
        kind: 'no-labels',
        detail:
          'no record carried a label, so this contributor is unlabelled in every per-workload figure',
        usd: null,
        calls: null,
      });
    }
    if (isRecord(doc.duplicateLines)) {
      const count = numberAt(doc.duplicateLines, 'count');
      if (count > 0) {
        gaps.push({
          kind: 'duplicate-lines',
          detail: `${count} line${count === 1 ? '' : 's'} of this contributor's log repeated an earlier line exactly`,
          usd: numberAt(doc.duplicateLines, 'usd'),
          calls: count,
        });
      }
    }

    /**
     * The window this contributor asked for, if it asked for one.
     *
     * `timeWindow` is present in a profile document only when the profile was
     * run with `--since` or `--until`, so its absence is the ordinary case and
     * is an answer rather than a defect: nobody claimed a period, and the span
     * is all that is known.
     */
    const window = isRecord(doc.timeWindow) ? doc.timeWindow : null;
    const claimed =
      window === null
        ? null
        : {
            sinceMs: typeof window.sinceMs === 'number' ? window.sinceMs : null,
            untilMs: typeof window.untilMs === 'number' ? window.untilMs : null,
          };

    if (claimed === null) {
      cannotSay.add('no-claimed-period');
    } else if (claimed.sinceMs === null || claimed.untilMs === null) {
      cannotSay.add('claim-not-bounded');
    }

    /** The days this contributor itself put spend on — its own bucketing. */
    const ownDays = new Set<string>();
    if (Array.isArray(doc.spendByDay)) {
      for (const row of doc.spendByDay) {
        if (!isRecord(row)) continue;
        const day = asString(row.day);
        if (day !== null) ownDays.add(day);
      }
    }

    const silence = claimed === null ? null : silenceIn(claimed, ownDays);
    if (
      claimed !== null &&
      claimed.sinceMs !== null &&
      claimed.untilMs !== null &&
      silence === null
    ) {
      // Bounded, in order, and still not walked: it was longer than the bound.
      cannotSay.add('claim-too-long-to-enumerate');
    }
    if (silence !== null && silence.days > 0) {
      cannotSay.add('silence-inside-a-claim');
      gaps.push({
        kind: 'silent-days',
        detail: `${silence.days} day${silence.days === 1 ? '' : 's'} inside the window this contributor asked for recorded nothing`,
        usd: null,
        calls: null,
      });
    }

    if (claimed?.sinceMs != null && claimed.untilMs != null) {
      claimFrom = claimFrom === null ? claimed.sinceMs : Math.min(claimFrom, claimed.sinceMs);
      claimTo = claimTo === null ? claimed.untilMs : Math.max(claimTo, claimed.untilMs);
      claimants += 1;
    }

    contributors.push({
      name: input.name,
      via: null,
      totalUsd: numberAt(totals, 'totalUsd'),
      calls: numberAt(totals, 'calls'),
      span,
      spanDays: span === null ? null : (span.toMs - span.fromMs) / 86_400_000,
      claimed,
      silence,
      undatedExcluded: window === null ? null : numberAt(window, 'undatedExcluded'),
      gaps,
    });
  }

  if (contributors.length > 1) cannotSay.add('overlap-invisible');
  if (unknownFields.size > 0) cannotSay.add('unknown-fields-dropped');

  /**
   * Mismatched spans, on the rule `fleetRollup` already uses: more than a day
   * apart in length, or one contributor with a clock beside one without. A
   * share of a sum stays valid either way; reading it as a comparison of
   * *rates* is the mistake this names.
   */
  const spanLengths = contributors.map((contributor) => contributor.spanDays);
  const known = spanLengths.filter((days_): days_ is number => days_ !== null);
  if (
    (spanLengths.some((length) => length === null) && known.length > 0) ||
    (known.length > 1 && Math.max(...known) - Math.min(...known) > 1)
  ) {
    cannotSay.add('mismatched-spans');
  }

  /**
   * Contributor names that turn up more than once.
   *
   * Handing over both a roll-up and one of the machines inside it counts that
   * machine's money twice — and unlike the identical-document check, this one
   * can see it: the two documents differ, but the name does not. Named and
   * never subtracted, because two teams genuinely running `api.json` is
   * possible and removing money to decide between them is the repair this tool
   * does not make.
   */
  const timesSeen = new Map<string, number>();
  for (const contributor of contributors) {
    timesSeen.set(contributor.name, (timesSeen.get(contributor.name) ?? 0) + 1);
  }
  const repeatedContributors = [...timesSeen.entries()]
    .filter(([, times]) => times > 1)
    .map(([name]) => name)
    .sort();
  if (repeatedContributors.length > 0) cannotSay.add('contributor-named-twice');

  const notMerged: UnmergedFinding[] = PER_RECORD_FINDINGS.filter((entry) =>
    findingsPresentIn.has(entry.field),
  ).map((entry) => ({
    finding: entry.finding,
    because: entry.because,
    presentIn: findingsPresentIn.get(entry.field) ?? [],
  }));

  if (cannotSay.has('day-top-label-unknown')) {
    const shared = [...days.values()].filter((bucket) => bucket.from.size > 1);
    notMerged.push({
      finding: "a day's dearest label",
      because:
        'each contributor knows its own, and the merged answer needs the per-label-per-day spend that no document carries',
      presentIn: [...new Set(shared.flatMap((bucket) => [...bucket.from]))].sort(),
    });
  }

  // Findings an inner roll-up already refused to merge do not become mergeable
  // by being handed on. Merged by their own text so the same finding from two
  // roll-ups is one entry naming both sets of contributors.
  for (const [finding, entry] of nestedFindings) {
    const existing = notMerged.find((row) => row.finding === finding);
    if (existing === undefined) {
      notMerged.push({ finding, because: entry.because, presentIn: [...entry.presentIn].sort() });
      continue;
    }
    for (const who of entry.presentIn) {
      if (!existing.presentIn.includes(who)) existing.presentIn.push(who);
    }
    existing.presentIn.sort();
  }

  if (unknownFields.size > 0) {
    notMerged.push({
      finding: `numeric fields this version cannot combine: ${[...unknownFields].sort().join(', ')}`,
      because:
        'a field added after this roll-up was written may be a sum, a maximum or a ratio, and combining it the wrong way is worse than leaving it out',
      presentIn: [],
    });
  }

  return {
    schemaVersion: 1,
    contributors,
    rejected,
    identicalContributions: {
      groups: [...[...seenText.values()].filter((names) => names.length > 1), ...nestedIdentical],
      usd: identicalUsd,
    },
    total,
    unpriced,
    unpricedModels: [...unpricedModels].sort(),
    byLabel: byMoney(
      [...labels.values()].map((slice) => ({ label: slice.label ?? '', breakdown: slice.breakdown })),
      (row) => row.breakdown.totalUsd,
    ),
    byModel: byMoney(
      [...models.values()].map((slice) => ({ model: slice.model ?? '', breakdown: slice.breakdown })),
      (row) => row.breakdown.totalUsd,
    ),
    byLabelAndModel: byMoney(
      [...pairs.values()].map((slice) => ({
        label: slice.label ?? '',
        model: slice.model ?? '',
        breakdown: slice.breakdown,
      })),
      (row) => row.breakdown.totalUsd,
    ),
    spendByDay: [...days.entries()]
      .sort((a, b) => (a[0] < b[0] ? -1 : 1))
      .map(([day, bucket]) => ({
        day,
        usd: bucket.usd,
        calls: bucket.calls,
        contributors: bucket.contributors,
        byModel: byMoney(
          [...bucket.models.entries()].map(([model, figures]) => ({ model, ...figures })),
          (row) => row.usd,
        ),
        topLabel: bucket.topLabel,
        topLabelUsd: bucket.topLabelUsd,
      })),
    span:
      spanFrom === null || spanTo === null ? null : { fromMs: spanFrom, toMs: spanTo, calls: spanCalls },
    claimedSpan:
      claimFrom === null || claimTo === null
        ? null
        : { fromMs: claimFrom, toMs: claimTo, contributors: claimants },
    fieldCoverage: coverage,
    outcomeTally: {
      byValue: byMoney(
        [...outcomeValues.entries()].map(([value, figures]) => ({ value, ...figures })),
        (row) => row.usd,
      ),
      recorded: outcome.recorded,
      parsed: outcome.parsed,
      unrecordedUsd: outcome.unrecordedUsd,
    },
    duplicateLines,
    repeatedContributors,
    notMerged,
    cannotSay: [...cannotSay].sort(),
  };
}