// Shared reporting math + ingest transforms for the dashboard worker.

//

// Every formula here is a faithful port of the Python definitions in

// scripts/oakbench/stats.py and scripts/oakbench/reporting.py. Parity is

// enforced by tests/fixtures/reporting_golden.json, asserted from BOTH sides

// (tests/test_reporting_golden.py and cloudflare/reporting_parity.test.mjs).

// Null means unmeasured (ADR-0002): a null input NEVER becomes 0.

// --- stats.py ports -------------------------------------------------------

/** Port of oakbench.stats.percentile_nearest (nearest-rank, empty -> null). */

export function percentileNearest(values, percentile) {

  const ordered = [...values].map(Number).sort((a, b) => a - b);

  if (ordered.length === 0) {

    return null;

  }

  const rank = Math.ceil((percentile / 100.0) * ordered.length);

  const index = Math.min(Math.max(rank, 1), ordered.length) - 1;

  return ordered[index];

}

/** Port of oakbench.reporting.median (statistics.median, empty -> null). */

export function median(values) {

  const ordered = [...values].map(Number).sort((a, b) => a - b);

  const n = ordered.length;

  if (n === 0) {

    return null;

  }

  if (n % 2 === 1) {

    return ordered[(n - 1) / 2];

  }

  return (ordered[n / 2 - 1] + ordered[n / 2]) / 2;

}

export function mean(values) {

  if (values.length === 0) {

    return null;

  }

  let total = 0;

  for (const value of values) {

    total += Number(value);

  }

  return total / values.length;

}

// --- reporting.py ports ---------------------------------------------------

/**

 * Port of oakbench.reporting.delta_pct.

 * POSITIVE means value is LOWER/BETTER than base. base == 0 -> 0.0 (guard).

 */

export function deltaPct(base, value) {

  if (base === null || base === undefined || value === null || value === undefined) {

    return null;

  }

  if (base === 0) {

    return 0.0;

  }

  return ((base - value) / base) * 100.0;

}

/**

 * Port of oakbench.reporting.slower_pct.

 * POSITIVE means value is HIGHER/WORSE than baseline; non-positive or

 * unmeasured baseline -> null.

 */

export function slowerPct(baseline, value) {

  if (baseline === null || baseline === undefined || value === null || value === undefined) {

    return null;

  }

  if (baseline <= 0) {

    return null;

  }

  return ((value - baseline) / baseline) * 100.0;

}

/** Port of oakbench.reporting.delta_ms (null when either side unmeasured). */

export function deltaMs(baseline, value) {

  if (baseline === null || baseline === undefined || value === null || value === undefined) {

    return null;

  }

  return value - baseline;

}

// --- rows.py accessor ports -----------------------------------------------

export function numberOrNone(value) {

  if (value === null || value === undefined) {

    return null;

  }

  const parsed = Number(value);

  return Number.isFinite(parsed) ? parsed : null;

}

/** Port of oakbench.rows.row_returncode (missing/null -> failure, never crash). */

export function rowReturncode(row) {

  const value = row.returncode;

  if (value === null || value === undefined) {

    return 1;

  }

  const parsed = Number(value);

  return Number.isInteger(parsed) ? parsed : 1;

}

const SKIP_RETURNCODE = 77;

const SUBOP_SUFFIXES = [".add", ".commit"];

// Explicit separator for every composite map key. Plain concatenation is

// collision-prone ("a" + "bc" === "ab" + "c"); ASCII unit separator cannot

// appear in subject/scenario/operation strings produced by the harness.

const KEY_SEP = "\u001f";

/**

 * Deterministic ingest identity for a row.

 *

 * Platform-lane rows (scripts/oakbench/rows.py LANE_REQUIRED_KEYS["platform"])

 * legitimately carry NO subject field — their identity is platform/driver

 * (e.g. "github/cli"). Rows with neither subject nor platform have no subject

 * identity and return null; callers must skip them from subject-keyed tables

 * explicitly and count them in an ingest report (never silently).

 */

export function effectiveSubject(row) {

  if (row.subject !== null && row.subject !== undefined) {

    return String(row.subject);

  }

  if (row.platform !== null && row.platform !== undefined) {

    return `${row.platform}/${row.driver}`;

  }

  return null;

}

/**

 * Subject kind matching effectiveSubject: explicit subject_kind when present;

 * for subject-less platform rows the kind is the platform name. Null when the

 * row carries neither (null means unmeasured, ADR-0002).

 */

export function effectiveSubjectKind(row) {

  if (row.subject_kind !== null && row.subject_kind !== undefined) {

    return String(row.subject_kind);

  }

  if (

    (row.subject === null || row.subject === undefined) &&

    row.platform !== null &&

    row.platform !== undefined

  ) {

    return String(row.platform);

  }

  return null;

}

function isObject(value) {

  return value !== null && typeof value === "object" && !Array.isArray(value);

}

/** Port of oakbench.rows.token_total (best-available, provider first). */

export function tokenTotal(row) {

  const tokenMetrics = isObject(row.token_metrics) ? row.token_metrics : null;

  if (tokenMetrics) {

    const total = numberOrNone(tokenMetrics.total_tokens_reported);

    if (total !== null) {

      return total;

    }

    const parts = [

      numberOrNone(tokenMetrics.input_tokens_reported),

      numberOrNone(tokenMetrics.output_tokens_reported),

      numberOrNone(tokenMetrics.reasoning_tokens_reported),

    ];

    if (parts.some((part) => part !== null)) {

      return parts.reduce((sum, part) => sum + (part || 0), 0);

    }

  }

  for (const key of ["estimated_tokens_total", "token_count", "total_tokens"]) {

    const value = numberOrNone(row[key]);

    if (value !== null) {

      return value;

    }

  }

  if (isObject(row.token_estimate)) {

    const value = numberOrNone(row.token_estimate.total);

    if (value !== null) {

      return value;

    }

  }

  if (isObject(row.tokens)) {

    for (const key of ["total", "total_tokens", "provider_total", "estimated_total"]) {

      const value = numberOrNone(row.tokens[key]);

      if (value !== null) {

        return value;

      }

    }

    if (isObject(row.tokens.provider_reported)) {

      const value = numberOrNone(row.tokens.provider_reported.total);

      if (value !== null) {

        return value;

      }

    }

  }

  if (tokenMetrics) {

    let chars = numberOrNone(tokenMetrics.transcript_chars);

    if (chars === null) {

      const inputChars = numberOrNone(tokenMetrics.normalized_input_chars);

      const outputChars = numberOrNone(tokenMetrics.normalized_output_chars);

      if (inputChars !== null || outputChars !== null) {

        chars = (inputChars || 0) + (outputChars || 0);

      }

    }

    if (chars !== null) {

      return Math.max(1, Math.ceil(chars / 4.0));

    }

  }

  return null;

}

/** Port of oakbench.rows.tool_calls. */

export function toolCalls(row) {

  for (const key of ["tool_call_count", "total_tool_calls"]) {

    const value = numberOrNone(row[key]);

    if (value !== null) {

      return value;

    }

  }

  if (isObject(row.tool_calls)) {

    for (const key of ["total", "all"]) {

      const value = numberOrNone(row.tool_calls[key]);

      if (value !== null) {

        return value;

      }

    }

  }

  if (isObject(row.tool_call_metrics)) {

    const value = numberOrNone(row.tool_call_metrics.tool_calls_total);

    if (value !== null) {

      return value;

    }

  }

  return null;

}

/** Port of oakbench.rows.output_bytes. */

export function outputBytes(row) {

  const raw = numberOrNone(row.raw_output_bytes);

  if (raw !== null) {

    return raw;

  }

  let stdout = numberOrNone(row.stdout_bytes);

  let stderr = numberOrNone(row.stderr_bytes);

  if (stdout !== null || stderr !== null) {

    return (stdout || 0) + (stderr || 0);

  }

  if (isObject(row.token_estimate)) {

    stdout = numberOrNone(row.token_estimate.stdout_bytes);

    stderr = numberOrNone(row.token_estimate.stderr_bytes);

    if (stdout !== null || stderr !== null) {

      return (stdout || 0) + (stderr || 0);

    }

  }

  return null;

}

/** Port of oakbench.rows.output_truncated. */

export function outputTruncated(row) {

  if (row.output_truncated || row.stdout_truncated || row.stderr_truncated) {

    return true;

  }

  if (isObject(row.token_estimate)) {

    return Boolean(row.token_estimate.stdout_truncated || row.token_estimate.stderr_truncated);

  }

  return false;

}

// --- Ingest step 1: rows -> measurements -----------------------------------

/**

 * Project JSONL rows into measurements-table records.

 *

 * Pure field projection: null/missing stays null (NEVER coerced to 0).

 * Sub-operation rows (operation ending .add/.commit) are marked

 * is_canonical = 0, matching oakbench.baseline_book.SUBOP_SUFFIXES.

 *

 * Subject identity uses effectiveSubject/effectiveSubjectKind so platform

 * rows (no subject field; identity platform/driver) group correctly. Rows

 * with no subject identity at all (no subject AND no platform) are skipped

 * from this subject-keyed projection explicitly: when a `report` object is

 * passed, its skipped_no_subject counter is incremented per skipped row so

 * the ingest response surfaces them (never silent).

 *

 * `subjectIds` optionally maps subjectKey(row) -> subject_id; when absent the

 * effective subject name is used so the transform stays pure and testable.

 */

export function rowsToMeasurements(rows, runKey, now, subjectIds = null, report = null) {

  const measurements = [];

  for (const row of rows) {

    const subject = effectiveSubject(row);

    if (subject === null) {

      if (report) {

        report.skipped_no_subject = (report.skipped_no_subject ?? 0) + 1;

      }

      continue;

    }

    const operation = String(row.operation ?? "");

    const key = subjectKey(row);

    const kind = effectiveSubjectKind(row);

    measurements.push({

      run_key: runKey,

      subject_id: (subjectIds && subjectIds.get(key)) || subject,

      subject,

      subject_kind: kind,

      scenario: String(row.scenario ?? ""),

      operation,

      run_index: numberOrNone(row.run) ?? 0,

      is_canonical: SUBOP_SUFFIXES.some((suffix) => operation.endsWith(suffix)) ? 0 : 1,

      elapsed_ms: numberOrNone(row.elapsed_ms),

      estimated_tokens_total: numberOrNone(row.estimated_tokens_total),

      estimated_tokens_input: numberOrNone(row.estimated_tokens_input),

      estimated_tokens_output: numberOrNone(row.estimated_tokens_output),

      token_source: row.token_source === null || row.token_source === undefined ? null : String(row.token_source),

      raw_output_bytes: numberOrNone(row.raw_output_bytes),

      stdout_bytes: numberOrNone(row.stdout_bytes),

      stderr_bytes: numberOrNone(row.stderr_bytes),

      output_truncated: outputTruncated(row) ? 1 : 0,

      tool_call_count: numberOrNone(row.tool_call_count),

      vcs_tool_call_count: numberOrNone(row.vcs_tool_call_count),

      terminal_tool_call_count: numberOrNone(row.terminal_tool_call_count),

      returncode: rowReturncode(row),

      command_json: row.command === undefined || row.command === null ? null : JSON.stringify(row.command),

      stderr: typeof row.stderr === "string" && row.stderr ? row.stderr.slice(-4000) : null,

      row_json: JSON.stringify(row),

      created_at: now,

      // schema_v2 additive columns (null when the lane did not record them).

      runner_id: row.runner_id === null || row.runner_id === undefined ? null : String(row.runner_id),

      runner_class: row.runner_class === null || row.runner_class === undefined ? null : String(row.runner_class),

      load_tier: row.load_tier === null || row.load_tier === undefined ? null : String(row.load_tier),

      netshape_profile:

        row.netshape_profile === null || row.netshape_profile === undefined ? null : String(row.netshape_profile),

      // Best-available derived metrics (oakbench.rows accessors); used by

      // computeAggregates, not persisted as columns.

      tokens_best: tokenTotal(row),

      output_bytes_best: outputBytes(row),

      tool_calls_best: toolCalls(row),

    });

  }

  return measurements;

}

// --- Ingest step 2: subjects upsert ----------------------------------------

/**

 * Identity key for a subject within a run: (effective subject, effective

 * subject_kind, binary_sha256). Platform rows (no subject field) key on

 * their platform/driver identity. Null when the row has no subject identity

 * at all (no subject AND no platform) so callers skip it explicitly.

 */

export function subjectKey(row) {

  const subject = effectiveSubject(row);

  if (subject === null) {

    return null;

  }

  const kind = effectiveSubjectKind(row);

  return [

    subject,

    kind === null ? "" : kind,

    row.binary_sha256 === null || row.binary_sha256 === undefined ? "" : String(row.binary_sha256),

  ].join(KEY_SEP);

}

/**

 * Build run-scoped subject upsert records, keyed (effective subject,

 * effective subject_kind, binary_sha256), enriched from the first row's

 * subject_details map. Platform rows (no subject field) are identified by

 * platform/driver via effectiveSubject so they ingest like any other subject.

 * Returns { records, idsByKey, skipped }: idsByKey maps subjectKey ->

 * subject_id; skipped counts rows with no subject identity at all (no

 * subject AND no platform), which are excluded from subject-keyed tables.

 */

export async function upsertSubjects(rows, runKey, subjectDetails = {}, subjectVersions = {}) {

  const records = [];

  const idsByKey = new Map();

  let skipped = 0;

  for (const row of rows) {

    const name = effectiveSubject(row);

    if (name === null) {

      skipped += 1;

      continue;

    }

    const key = subjectKey(row);

    if (idsByKey.has(key)) {

      continue;

    }

    const details = isObject(subjectDetails[name]) ? subjectDetails[name] : {};

    const binarySha = row.binary_sha256 ?? details.binary_sha256 ?? null;

    const subjectId = await stableId(`${runKey}${KEY_SEP}${key}`);

    idsByKey.set(key, subjectId);

    records.push({

      subject_id: subjectId,

      run_key: runKey,

      name,

      kind: String(effectiveSubjectKind(row) ?? details.kind ?? "unknown"),

      label: details.label === undefined ? null : details.label,

      version: subjectVersions[name] ?? details.version ?? null,

      binary_path: details.bin === undefined ? null : details.bin,

      metadata_json: JSON.stringify({ binary_sha256: binarySha, role: details.role ?? null }),

    });

  }

  return { records, idsByKey, skipped };

}

// --- Ingest step 3: aggregates ---------------------------------------------

/**

 * Per (run_key, subject_id, scenario, operation) aggregates.

 *

 * Latency percentiles use nearest-rank (percentileNearest) over the

 * elapsed_ms of SUCCESSFUL (returncode 0) measurements only, exactly matching

 * oakbench.stats.percentile_nearest semantics including empty -> null.

 * Skip rows (returncode 77) are excluded from sample/failure accounting.

 */

// Tail honesty thresholds, mirroring scripts/benchmark_stats.py

// TAIL_MIN_SAMPLES = {"p95": 20, "p99": 100} (itself a mirror of

// oakbench.reporting.TAIL_PERCENTILE_MIN_SAMPLES): below 1/(1-p) samples a

// nearest-rank tail percentile is just the max relabelled, so claiming it

// would overstate precision. Gated percentiles are null (unmeasured,

// ADR-0002), NEVER the lone sample. p50/p90 stay ungated to match the Python

// reporting behavior (benchmark_stats gates only p95/p99).

export const TAIL_MIN_SAMPLES = { p95: 20, p99: 100 };

export function computeAggregates(measurements, profile, now) {

  const groups = new Map();

  for (const m of measurements) {

    const key = [m.run_key, m.subject_id, m.scenario, m.operation].join(KEY_SEP);

    if (!groups.has(key)) {

      groups.set(key, []);

    }

    groups.get(key).push(m);

  }

  const aggregates = [];

  for (const group of groups.values()) {

    const first = group[0];

    const successes = group.filter((m) => m.returncode === 0);

    const failures = group.filter((m) => m.returncode !== 0 && m.returncode !== SKIP_RETURNCODE);

    const sampleCount = successes.length + failures.length;

    const elapsed = successes.map((m) => m.elapsed_ms).filter((v) => v !== null);

    const tokens = successes.map((m) => m.tokens_best).filter((v) => v !== null && v !== undefined);

    const bytes = successes.map((m) => m.output_bytes_best).filter((v) => v !== null && v !== undefined);

    const calls = successes.map((m) => m.tool_calls_best).filter((v) => v !== null && v !== undefined);

    aggregates.push({

      run_key: first.run_key,

      subject_id: first.subject_id,

      subject: first.subject,

      subject_kind: first.subject_kind,

      profile,

      scenario: first.scenario,

      operation: first.operation,

      is_canonical: first.is_canonical,

      sample_count: sampleCount,

      failure_count: failures.length,

      failure_rate: sampleCount > 0 ? failures.length / sampleCount : null,

      n: elapsed.length,

      p50_ms: median(elapsed),

      p90_ms: percentileNearest(elapsed, 90),

      p95_ms: elapsed.length >= TAIL_MIN_SAMPLES.p95 ? percentileNearest(elapsed, 95) : null,

      p99_ms: elapsed.length >= TAIL_MIN_SAMPLES.p99 ? percentileNearest(elapsed, 99) : null,

      mean_ms: mean(elapsed),

      min_ms: elapsed.length ? Math.min(...elapsed) : null,

      max_ms: elapsed.length ? Math.max(...elapsed) : null,

      avg_tokens: mean(tokens),

      avg_output_bytes: mean(bytes),

      truncated_count: group.filter((m) => m.output_truncated === 1).length,

      avg_tool_calls: mean(calls),

      updated_at: now,

    });

  }

  return aggregates;

}

// --- Ingest step 4: comparisons --------------------------------------------

/**

 * Target-vs-baseline comparison rows: every non-git subject aggregate is

 * compared against EVERY git-kind subject aggregate for the same

 * run/scenario/operation (e.g. stock git AND git_fsmonitor each yield a

 * comparison record, distinguished by baseline_subject_id). Ingest holds no

 * notion of "the" baseline: selecting which git variant to report against is

 * a reporting/targets concern (hardest-fair-comparator doctrine). Output is

 * sorted on (run_key, scenario, operation, target_subject_id,

 * baseline_subject_id) so results are deterministic regardless of input

 * row order.

 *

 * p50_ratio is target/baseline (null when baseline is unmeasured or 0 — a

 * zero baseline makes the ratio undefined, mirroring scorecard UNDECIDABLE).

 * p50_delta_pct_better uses deltaPct (positive = better) including its

 * base == 0 -> 0.0 guard; *_slower_pct uses slowerPct (positive = worse).

 *

 * is_regression is 1 when EITHER p50 or p90 breaches (slower_pct >

 * thresholdPct AND delta_ms > minDeltaMs, evaluated per dimension);

 * regression_dimensions lists the breaching dimensions ("p50"/"p90"),

 * empty when none.

 */

export function computeComparisons(aggregates, options = {}) {

  const thresholdPct = options.thresholdPct ?? 10.0;

  const minDeltaMs = options.minDeltaMs ?? 5.0;

  const baselinesByGroup = new Map();

  for (const agg of aggregates) {

    if (agg.subject_kind === "git") {

      const key = [agg.run_key, agg.scenario, agg.operation].join(KEY_SEP);

      if (!baselinesByGroup.has(key)) {

        baselinesByGroup.set(key, []);

      }

      baselinesByGroup.get(key).push(agg);

    }

  }

  const comparisons = [];

  for (const target of aggregates) {

    if (target.subject_kind === "git") {

      continue;

    }

    const bases = baselinesByGroup.get([target.run_key, target.scenario, target.operation].join(KEY_SEP)) || [];

    for (const base of bases) {

      if (base.subject_id === target.subject_id) {

        continue;

      }

      const p50SlowerPct = slowerPct(base.p50_ms, target.p50_ms);

      const p50DeltaMs = deltaMs(base.p50_ms, target.p50_ms);

      const p90SlowerPct = slowerPct(base.p90_ms, target.p90_ms);

      const p90DeltaMs = deltaMs(base.p90_ms, target.p90_ms);

      // A regression in EITHER p50 or p90 flags the comparison: agents pay

      // tail latency, so a p90-only blowup must not hide behind a flat p50.

      // Each dimension breaches when slower_pct > thresholdPct AND

      // delta_ms > minDeltaMs (nulls — unmeasured — never breach).

      const breaches = (slower, delta) =>

        slower !== null && delta !== null && slower > thresholdPct && delta > minDeltaMs;

      const regressionDimensions = [];

      if (breaches(p50SlowerPct, p50DeltaMs)) {

        regressionDimensions.push("p50");

      }

      if (breaches(p90SlowerPct, p90DeltaMs)) {

        regressionDimensions.push("p90");

      }

      const isRegression = regressionDimensions.length > 0 ? 1 : 0;

      comparisons.push({

        run_key: target.run_key,

        target_subject_id: target.subject_id,

        baseline_subject_id: base.subject_id,

        baseline_subject: base.subject,

        baseline_kind: "git",

        profile: target.profile,

        scenario: target.scenario,

        operation: target.operation,

        p50_delta_ms: p50DeltaMs,

        p50_slower_pct: p50SlowerPct,

        p50_ratio:

          base.p50_ms === null || base.p50_ms === 0 || target.p50_ms === null ? null : target.p50_ms / base.p50_ms,

        p50_delta_pct_better: deltaPct(base.p50_ms, target.p50_ms),

        p90_delta_ms: p90DeltaMs,

        p90_slower_pct: p90SlowerPct,

        p99_delta_ms: deltaMs(base.p99_ms, target.p99_ms),

        p99_slower_pct: slowerPct(base.p99_ms, target.p99_ms),

        avg_token_delta: deltaMs(base.avg_tokens, target.avg_tokens),

        token_higher_pct: slowerPct(base.avg_tokens, target.avg_tokens),

        avg_output_byte_delta: deltaMs(base.avg_output_bytes, target.avg_output_bytes),

        output_byte_higher_pct: slowerPct(base.avg_output_bytes, target.avg_output_bytes),

        avg_tool_call_delta: deltaMs(base.avg_tool_calls, target.avg_tool_calls),

        tool_call_higher_pct: slowerPct(base.avg_tool_calls, target.avg_tool_calls),

        failure_rate_delta_pp:

          target.failure_rate === null || target.failure_rate === undefined ||

          base.failure_rate === null || base.failure_rate === undefined

            ? null

            : (target.failure_rate - base.failure_rate) * 100.0,

        threshold_pct: thresholdPct,

        min_delta_ms: minDeltaMs,

        is_regression: isRegression,

        // Additive: which dimensions breached ("p50"/"p90"); empty when none.

        regression_dimensions: regressionDimensions,

        state: isRegression ? "regression" : "ok",

        severity_score: isRegression

          ? Math.max(

              regressionDimensions.includes("p50") ? p50SlowerPct : 0,

              regressionDimensions.includes("p90") ? p90SlowerPct : 0,

            )

          : 0,

      });

    }

  }

  comparisons.sort((a, b) => {

    const left = [a.run_key, a.scenario, a.operation, a.target_subject_id, a.baseline_subject_id].join(KEY_SEP);

    const right = [b.run_key, b.scenario, b.operation, b.target_subject_id, b.baseline_subject_id].join(KEY_SEP);

    return left < right ? -1 : left > right ? 1 : 0;

  });

  return comparisons;

}

// --- Scorecard + trends -----------------------------------------------------

const VERDICT_PRIORITY = ["RED", "INVALID", "UNDECIDABLE", "AMBER", "UNMEASURED", "GREEN"];

/**

 * Summarize scorecard cells (the verdict dicts produced by

 * oakbench.scorecard.evaluate_targets) into dashboard headline counts.

 */

export function scorecardFromCells(cells) {

  const counts = {};

  for (const cell of cells || []) {

    const verdict = String(cell.verdict ?? "UNMEASURED");

    counts[verdict] = (counts[verdict] || 0) + 1;

  }

  let overall = "UNMEASURED";

  for (const verdict of VERDICT_PRIORITY) {

    if (counts[verdict]) {

      overall = verdict;

      break;

    }

  }

  return { total: (cells || []).length, counts, overall };

}

const TREND_METRICS = {

  p50: "p50_ms",

  median: "p50_ms",

  p50_ms: "p50_ms",

  p90: "p90_ms",

  p90_ms: "p90_ms",

  p95: "p95_ms",

  p95_ms: "p95_ms",

  p99: "p99_ms",

  p99_ms: "p99_ms",

  mean: "mean_ms",

  mean_ms: "mean_ms",

  avg_tokens: "avg_tokens",

  avg_output_bytes: "avg_output_bytes",

  avg_tool_calls: "avg_tool_calls",

  failure_rate: "failure_rate",

};

/**

 * Build a per-subject time series from aggregate rows joined with run

 * timestamps: rows need {timestamp_utc, run_key, subject, scenario,

 * operation, <metric columns>}. Unknown metric -> null (caller 400s).

 */

export function trendSeries(rows, { scenario, operation, metric = "p50" } = {}) {

  const column = TREND_METRICS[metric];

  if (!column) {

    return null;

  }

  const yes = (row) =>

    (scenario === undefined || scenario === null || row.scenario === scenario) &&

    (operation === undefined || operation === null || row.operation === operation);

  const bySubject = new Map();

  const sorted = rows.filter(yes).sort((a, b) => String(a.timestamp_utc).localeCompare(String(b.timestamp_utc)));

  for (const row of sorted) {

    const subject = String(row.subject ?? row.name ?? "");

    if (!bySubject.has(subject)) {

      bySubject.set(subject, []);

    }

    bySubject.get(subject).push({

      t: row.timestamp_utc,

      run_key: row.run_key,

      value: row[column] === undefined ? null : row[column],

    });

  }

  return {

    scenario: scenario ?? null,

    operation: operation ?? null,

    metric: column,

    series: [...bySubject.entries()].map(([subject, points]) => ({ subject, points })),

  };

}

// --- Webhook signature ------------------------------------------------------

/**

 * Verify a GitHub X-Hub-Signature-256 header ("sha256=<hex>") against the raw

 * request body using HMAC-SHA256 (crypto.subtle; works in Workers and node).

 */

export async function verifySignature(secret, body, signatureHeader) {

  if (!secret || typeof signatureHeader !== "string" || !signatureHeader.startsWith("sha256=")) {

    return false;

  }

  const encoder = new TextEncoder();

  const key = await crypto.subtle.importKey(

    "raw",

    encoder.encode(secret),

    { name: "HMAC", hash: "SHA-256" },

    false,

    ["sign"],

  );

  const digest = await crypto.subtle.sign("HMAC", key, encoder.encode(body));

  const expected = [...new Uint8Array(digest)].map((byte) => byte.toString(16).padStart(2, "0")).join("");

  const provided = signatureHeader.slice("sha256=".length).toLowerCase();

  if (provided.length !== expected.length) {

    return false;

  }

  let diff = 0;

  for (let index = 0; index < expected.length; index += 1) {

    diff |= expected.charCodeAt(index) ^ provided.charCodeAt(index);

  }

  return diff === 0;

}

// --- shared hashing ---------------------------------------------------------

export async function sha256Hex(text) {

  const bytes = new TextEncoder().encode(text);

  const digest = await crypto.subtle.digest("SHA-256", bytes);

  return [...new Uint8Array(digest)].map((byte) => byte.toString(16).padStart(2, "0")).join("");

}

export async function stableId(text) {

  return (await sha256Hex(text)).slice(0, 24);

}