kache 0.28.0

use anyhow::Result;
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use std::path::{Path, PathBuf};

use crate::cli::format_duration_ms;
use crate::config::Config;
use crate::daemon::{TransferDirection, TransferEvent};
use crate::events::{self, BuildEvent, EventResult};
use crate::since::SinceWindow;

// ── Data Model ──────────────────────────────────────────────────────────────

/// `gc_stats.json`: the last GC run, whichever driver ran it. Files written
/// by earlier versions can carry a `totals` object, which is ignored.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct GcStatsPersisted {
    pub last_run: String,
    pub entries_evicted: usize,
    pub bytes_freed: u64,
    #[serde(default)]
    pub disk_bytes_reclaimed: u64,
    pub blobs_removed: usize,
    pub duration_ms: u64,
    /// `daemon` (its sweep, or `kache gc` routed through it), `auto` (the
    /// detached worker a wrapper spawns when the store outgrows its limit) or
    /// `manual` (`kache gc` with no daemon). Empty in files written before the
    /// field existed.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub source: String,
    #[serde(default)]
    pub entries_pinned: usize,
    #[serde(default)]
    pub entries_unreclaimable: usize,
    /// Bytes the size pass found held by target directories (#1206).
    #[serde(default)]
    pub bytes_held: u64,
    #[serde(default)]
    pub entries_failed: usize,
    #[serde(default)]
    pub entries_locked: usize,
    #[serde(default)]
    pub entries_busy_snapshot: usize,
    #[serde(default)]
    pub entries_recent_prefiltered: usize,
    /// Kept because the remote delivered them within the import pin (#1008).
    #[serde(default)]
    pub entries_import_pinned: usize,
    /// Time the run spent in its eviction writes, busy waits included.
    #[serde(default)]
    pub evict_write_ms: u64,
    /// Stale key lock files the run unlinked. Absent, like the two below,
    /// when the run did no housekeeping.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub key_locks_removed: Option<usize>,
    /// Key lock files left in `store/` after the run.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub key_locks_remaining: Option<usize>,
    /// Input predictions deleted as unused.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub predictions_pruned: Option<usize>,
    /// File hash memo rows deleted as not written for a month.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub file_hashes_pruned: Option<usize>,
}

pub(crate) const GC_STATS_FILE: &str = "gc_stats.json";

/// Read `cache_dir/gc_stats.json`, or `None` when it is absent. A file that
/// does not parse also reads as `None`, with a warning, so a damaged record
/// does not pass silently.
pub(crate) fn read_gc_stats(cache_dir: &Path) -> Option<GcStatsPersisted> {
    let path = cache_dir.join(GC_STATS_FILE);
    let content = std::fs::read_to_string(&path).ok()?;
    match serde_json::from_str(&content) {
        Ok(stats) => Some(stats),
        Err(e) => {
            tracing::warn!(
                "{} does not parse ({e}); the next GC run replaces it",
                path.display()
            );
            None
        }
    }
}

/// Schema of one line in `cache_dir/telemetry/gc-runs.jsonl`.
pub const GC_RUN_RECORD_SCHEMA: u32 = 2;

/// One GC run in `telemetry/gc-runs.jsonl`, written only with session
/// recording on. `gc_stats.json` keeps the latest run; this history is what
/// a backend adds up across runs and drivers.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct GcRunRecord {
    pub ts: String,
    pub schema: u32,
    /// `daemon`, `auto` or `manual`, as in `gc_stats.json`.
    pub source: String,
    pub entries_evicted: usize,
    pub bytes_freed: u64,
    pub disk_bytes_reclaimed: u64,
    pub blobs_removed: usize,
    pub entries_failed: usize,
    /// The failed evictions that lost the index write lock.
    pub entries_locked: usize,
    /// Failed read-to-write upgrades on stale WAL snapshots.
    #[serde(default)]
    pub entries_busy_snapshot: usize,
    /// Recent candidates skipped before reading metadata or opening a txn.
    #[serde(default)]
    pub entries_recent_prefiltered: usize,
    /// Kept because the remote delivered them within the import pin (#1008).
    #[serde(default)]
    pub entries_import_pinned: usize,
    pub entries_pinned: usize,
    pub entries_unreclaimable: usize,
    #[serde(default)]
    pub bytes_held: u64,
    pub duration_ms: u64,
    /// Time spent in eviction writes, busy waits included.
    #[serde(default)]
    pub evict_write_ms: u64,
    /// Housekeeping counts, as in `gc_stats.json`; absent when none ran.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub key_locks_removed: Option<usize>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub key_locks_remaining: Option<usize>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub predictions_pruned: Option<usize>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub file_hashes_pruned: Option<usize>,
}

impl GcRunRecord {
    fn new(source: &str, stats: &crate::store::GcStats) -> Self {
        Self {
            ts: Utc::now().to_rfc3339(),
            schema: GC_RUN_RECORD_SCHEMA,
            source: source.to_string(),
            entries_evicted: stats.entries_evicted,
            bytes_freed: stats.bytes_freed,
            disk_bytes_reclaimed: stats.disk_bytes_reclaimed,
            blobs_removed: stats.blobs_removed,
            entries_failed: stats.entries_failed,
            entries_locked: stats.entries_locked,
            entries_busy_snapshot: stats.entries_busy_snapshot,
            entries_recent_prefiltered: stats.entries_recent_prefiltered,
            entries_import_pinned: stats.entries_import_pinned,
            entries_pinned: stats.entries_pinned,
            entries_unreclaimable: stats.entries_unreclaimable,
            bytes_held: stats.bytes_held,
            duration_ms: stats.duration_ms,
            evict_write_ms: stats.evict_write_ms,
            key_locks_removed: stats.housekeeping.map(|h| h.key_locks_removed),
            key_locks_remaining: stats.housekeeping.map(|h| h.key_locks_remaining),
            predictions_pruned: stats.housekeeping.map(|h| h.predictions_pruned),
            file_hashes_pruned: stats.housekeeping.map(|h| h.file_hashes_pruned),
        }
    }
}

/// The machine-level GC history, beside `sessions.jsonl` in the cache dir.
pub(crate) fn gc_runs_log_path(cache_dir: &Path) -> PathBuf {
    cache_dir.join("telemetry").join("gc-runs.jsonl")
}

/// Record one finished GC run: `gc_stats.json` always, and one line in
/// `telemetry/gc-runs.jsonl` when session recording is on, rotated like the
/// other logs. Callers hold `gc.lock`.
pub fn record_gc_run(config: &Config, source: &str, stats: &crate::store::GcStats) -> Result<()> {
    write_last_gc_run(&config.cache_dir, source, stats)?;
    if !config.record_sessions {
        return Ok(());
    }
    let path = gc_runs_log_path(&config.cache_dir);
    crate::events::append_json_line(&path, &GcRunRecord::new(source, stats))?;
    crate::events::rotate_if_needed(
        &path,
        config.event_log_max_size,
        config.event_log_keep_lines,
    )
}

/// Replace `gc_stats.json` with this run.
pub(crate) fn write_last_gc_run(
    cache_dir: &Path,
    source: &str,
    stats: &crate::store::GcStats,
) -> Result<()> {
    let persisted = GcStatsPersisted {
        last_run: Utc::now().to_rfc3339(),
        entries_evicted: stats.entries_evicted,
        bytes_freed: stats.bytes_freed,
        disk_bytes_reclaimed: stats.disk_bytes_reclaimed,
        blobs_removed: stats.blobs_removed,
        duration_ms: stats.duration_ms,
        source: source.to_string(),
        entries_pinned: stats.entries_pinned,
        entries_unreclaimable: stats.entries_unreclaimable,
        bytes_held: stats.bytes_held,
        entries_failed: stats.entries_failed,
        entries_locked: stats.entries_locked,
        entries_busy_snapshot: stats.entries_busy_snapshot,
        entries_recent_prefiltered: stats.entries_recent_prefiltered,
        entries_import_pinned: stats.entries_import_pinned,
        evict_write_ms: stats.evict_write_ms,
        key_locks_removed: stats.housekeeping.map(|h| h.key_locks_removed),
        key_locks_remaining: stats.housekeeping.map(|h| h.key_locks_remaining),
        predictions_pruned: stats.housekeeping.map(|h| h.predictions_pruned),
        file_hashes_pruned: stats.housekeeping.map(|h| h.file_hashes_pruned),
    };
    let json = serde_json::to_string_pretty(&persisted)?;
    kache_store::atomic::atomic_replace(&cache_dir.join(GC_STATS_FILE), json.as_bytes())
}

/// Schema of one line in `cache_dir/telemetry/sessions.jsonl`.
pub const SESSION_RECORD_SCHEMA: u32 = 1;

/// One build session's report, kept at machine level. `kache report` reads
/// events from the runtime dir, which CI deletes with the job; this line is
/// what survives, in the cache dir every job on the host shares.
#[derive(Debug, Serialize, Deserialize)]
pub struct SessionRecord {
    pub ts: String,
    pub schema: u32,
    pub kache_version: String,
    /// The report window as its heading shows it: `24h`, `build session`.
    pub window: String,
    /// First 16 hex digits of blake3(root), never the path itself: enough to
    /// tell builds of one tree apart across jobs without recording where the
    /// tree lives.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub root_hash: Option<String>,
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub session_id: String,
    pub summary: ReportSummary,
    pub timing: TimingBreakdown,
    pub machine: SessionMachine,
}

/// The host when the session was recorded: the context that separates a slow
/// cache from a busy machine.
#[derive(Debug, Default, Serialize, Deserialize)]
pub struct SessionMachine {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub load_1m: Option<f64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub cpus: Option<u32>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub index_bytes: Option<u64>,
    pub store_max: u64,
}

impl SessionRecord {
    pub fn from_report(report: &BuildReport, machine: SessionMachine) -> Self {
        let meta = &report.meta;
        let window = meta.window_label();
        let (root, session_id) = match &meta.session {
            Some(session) => (Some(session.root.clone()), session.session_id.clone()),
            None => (meta.root_filter.clone(), String::new()),
        };
        let root_hash = root
            .filter(|root| !root.is_empty())
            .map(|root| blake3::hash(root.as_bytes()).to_hex().as_str()[..16].to_string());
        Self {
            ts: Utc::now().to_rfc3339(),
            schema: SESSION_RECORD_SCHEMA,
            kache_version: meta.kache_version.clone(),
            window,
            root_hash,
            session_id,
            summary: report.summary.clone(),
            timing: report.timing.clone(),
            machine,
        }
    }
}

/// GC summary included in build reports when GC ran recently.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct GcSummary {
    pub last_run: String,
    pub entries_evicted: usize,
    pub bytes_freed: u64,
    pub disk_bytes_reclaimed: u64,
    pub shared_bytes_retained: u64,
    pub blobs_removed: usize,
}

/// Schema version of the machine-readable JSON build report.
///
/// Incremented when a breaking change is made to the report's JSON structure
/// or code semantics. Additive fields (such as new optional/default fields or
/// non-breaking telemetry metrics) do not increment this version.
pub const REPORT_SCHEMA_VERSION: u32 = 1;

#[derive(Debug, Serialize, Deserialize)]
pub struct BuildReport {
    pub schema_version: u32,
    pub meta: ReportMeta,
    pub summary: ReportSummary,
    #[serde(default)]
    pub timeline: ReportTimeline,
    pub timing: TimingBreakdown,
    pub storage: StorageBreakdown,
    pub network: Option<NetworkAnalysis>,
    pub prefetch: PrefetchAnalysis,
    pub top_misses: Vec<CrateDetail>,
    pub top_hits: Vec<CrateDetail>,
    pub all_events: Vec<CrateDetail>,
    #[serde(default, rename = "traceEvents", alias = "trace_events")]
    pub trace_events: Vec<TraceEvent>,
    #[serde(
        default = "default_trace_display_time_unit",
        rename = "displayTimeUnit"
    )]
    pub display_time_unit: String,
    #[serde(default)]
    pub bypass: BypassAnalysis,
    pub errors_detail: Vec<ErrorDetail>,
    pub suggestions: Vec<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub gc: Option<GcSummary>,
}

#[derive(Debug, Serialize, Deserialize)]
pub struct ReportMeta {
    pub kache_version: String,
    pub generated_at: String,
    /// The window in whole hours, rounded down (0 for a sub-hour window).
    /// Kept for consumers that predate `since_secs`.
    pub since_hours: u64,
    /// The window in seconds (kunobi-ninja/kache#897). Authoritative.
    #[serde(default)]
    pub since_secs: u64,
    /// The window as requested: `15m`, `2h`, `24h`. Empty in reports written
    /// before this field existed; render through [`ReportMeta::window_label`].
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub since: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub root_filter: Option<String>,
    /// Present when the report selects one activity session instead of a window.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub session: Option<ReportSession>,
}

#[derive(Debug, Serialize, Deserialize)]
pub struct ReportSession {
    pub root: String,
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub session_id: String,
    /// Local and older events lack session IDs; group their activity by idle gap.
    pub inferred: bool,
    pub inactivity_secs: u64,
}

impl ReportSession {
    fn description(&self) -> String {
        let identity = if self.inferred {
            format!(
                "inferred from activity ({}s idle gap)",
                self.inactivity_secs
            )
        } else {
            format!("recorded {}", self.session_id)
        };
        format!(
            "Session: {identity}; root: {}. May include multiple Cargo commands. Only retained, completed compiler events are included. Store inventory covers the whole cache.",
            self.root
        )
    }
}

impl ReportMeta {
    /// The window for headings: `since` when present, else the whole-hour
    /// value an older report carried.
    pub fn window_label(&self) -> String {
        if self.session.is_some() {
            return "build session".to_string();
        }
        if self.since.is_empty() {
            format!("{}h", self.since_hours)
        } else {
            self.since.clone()
        }
    }
}

fn default_trace_display_time_unit() -> String {
    "ms".to_string()
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ReportSummary {
    pub hit_rate_pct: f64,
    pub weighted_hit_rate_pct: Option<f64>,
    pub time_saved_ms: u64,
    pub total_crates: usize,
    pub local_hits: usize,
    pub prefetch_hits: usize,
    pub remote_hits: usize,
    #[serde(default)]
    pub dups: usize,
    pub misses: usize,
    pub errors: usize,
    #[serde(default)]
    pub passthroughs: usize,
    /// Query / probe invocations (`--print`, `-vV`, `cc -###`) that ran
    /// uncached. Counted separately from `passthroughs` because a probe is
    /// not a compilation — lumping it in inflates the "couldn't cache"
    /// signal (a clean build does many probes, zero real passthroughs).
    #[serde(default)]
    pub probes: usize,
    #[serde(default)]
    pub skipped: usize,
    /// Compiles that ran and produced outputs kache then failed to store
    /// (kunobi-ninja/kache#629). Still counted in `misses` — the compiler did
    /// run — but broken out because these are the misses that repeat on every
    /// build instead of warming up.
    #[serde(default)]
    pub store_failures: usize,
    #[serde(default)]
    pub fallbacks: usize,
    pub total_duration_ms: u64,
    /// Percentage of total compile time avoided by cache: time_saved / (time_saved + miss_compile_time).
    #[serde(default)]
    pub cache_efficiency_pct: f64,
}

#[derive(Debug, Default, Serialize, Deserialize)]
pub struct ReportTimeline {
    #[serde(default)]
    pub start_time: Option<String>,
    #[serde(default)]
    pub end_time: Option<String>,
    #[serde(default)]
    pub start_unix_ms: Option<i64>,
    #[serde(default)]
    pub end_unix_ms: Option<i64>,
    pub duration_ms: u64,
    pub event_count: usize,
    pub cacheable_count: usize,
    pub hit_count: usize,
    pub compiled_count: usize,
    pub passthrough_count: usize,
    /// Query / probe invocations, split out of `passthrough_count` (see
    /// [`ReportSummary::probes`]).
    #[serde(default)]
    pub probe_count: usize,
    pub skipped_count: usize,
    pub error_count: usize,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TimingBreakdown {
    pub hit_time_ms: u64,
    pub miss_time_ms: u64,
    pub avg_hit_ms: f64,
    pub avg_miss_ms: f64,
    pub avg_hit_overhead_ms: f64,
    pub miss_compile_time_ms: u64,
    #[serde(default)]
    pub avg_key_ms: f64,
    #[serde(default)]
    pub avg_lookup_ms: f64,
    #[serde(default)]
    pub avg_restore_ms: f64,
    #[serde(default)]
    pub avg_store_ms: f64,
    #[serde(default)]
    pub total_key_ms: u64,
    #[serde(default)]
    pub total_lookup_ms: u64,
    #[serde(default)]
    pub total_restore_ms: u64,
    #[serde(default)]
    pub total_store_ms: u64,
    /// Process start to wrapper entry, summed over cacheable crates. Zero
    /// for events written before schema 17.
    #[serde(default)]
    pub total_startup_ms: u64,
    #[serde(default)]
    pub avg_startup_ms: f64,
    /// The rustc dep-info pre-pass: inside `total_key_ms`, split out because
    /// it is a full extra compiler start per invocation. Average is per run.
    #[serde(default)]
    pub total_dep_info_ms: u64,
    #[serde(default)]
    pub dep_info_runs: u64,
    #[serde(default)]
    pub avg_dep_info_ms: f64,
    /// Sampled verifications where a predicted input closure disagreed with
    /// the dep-info pre-pass. Zero unless `KACHE_VERIFY_INPUT_PREDICTIONS`
    /// asked for the comparison; the pre-pass answer always won regardless.
    #[serde(default)]
    pub prediction_mismatches: u64,
    /// Scheduler wait: flight join plus permit acquisition, kept apart so a
    /// saturated pool and a shared flight stay distinguishable.
    #[serde(default)]
    pub total_wait_ms: u64,
    #[serde(default)]
    pub total_flight_wait_ms: u64,
    #[serde(default)]
    pub total_permit_wait_ms: u64,
    #[serde(default)]
    pub avg_wait_ms: f64,
    /// Wrapper overhead no measured phase accounts for: overhead minus
    /// startup, key, lookup, wait, restore and store, clamped at zero per
    /// event. Overhead excludes the compile on a compiled crate.
    #[serde(default)]
    pub total_unattributed_ms: u64,
    #[serde(default)]
    pub avg_unattributed_ms: f64,
}

/// Mean of `total_ms` over `count` to one decimal; zero for an empty count.
fn avg_ms(total_ms: u64, count: u64) -> f64 {
    if count == 0 {
        return 0.0;
    }
    (total_ms as f64 / count as f64 * 10.0).round() / 10.0
}

/// `part` as a percentage of `total` to one decimal; zero for an empty total.
fn pct_of(part: u64, total: u64) -> f64 {
    if total == 0 {
        return 0.0;
    }
    (part as f64 / total as f64 * 1000.0).round() / 10.0
}

/// kache's storage efficiency — both halves of it.
///
/// **Restore side** — how cache-hit restores landed on disk: kache prefers
/// a CoW reflink (physically zero-copy *and* write-isolated), falling back
/// to a hardlink, then a full copy.
///
/// **Store side** — content-addressed dedup: an artifact whose content is
/// already in the store is referenced, not stored a second time. And how a
/// new blob entered the store: a CoW reflink first (shares blocks with the
/// build's own output), then a hardlink for immutable kinds when CoW is
/// unavailable (shared inode), then a full copy.
#[derive(Debug, Serialize, Deserialize)]
pub struct StorageBreakdown {
    #[serde(default)]
    pub store_entries: usize,
    #[serde(default)]
    pub stores: Vec<crate::store_view::StoreSummary>,
    /// Bytes restored by CoW reflink — zero physical copy.
    pub reflinked_bytes: u64,
    /// Bytes restored by hardlink — zero physical copy, shared inode.
    pub hardlinked_bytes: u64,
    /// Bytes restored by a full physical copy.
    pub copied_bytes: u64,
    /// Total bytes restored from cache (reflinked + hardlinked + copied).
    pub restored_bytes: u64,
    /// Share of restored bytes that cost no physical copy (reflink + hardlink).
    pub zero_copy_pct: f64,
    /// Bytes ingested into the store by a CoW reflink (shares blocks with the
    /// build's output — the store is not a second physical copy of these).
    pub store_reflinked_bytes: u64,
    /// Bytes ingested into the store by a hardlink (shares an inode with the
    /// build's output — zero-copy on filesystems without CoW).
    #[serde(default)]
    pub store_hardlinked_bytes: u64,
    /// Bytes ingested into the store by a full physical copy (a real second
    /// copy — the fallback when neither reflink nor hardlink is available).
    pub store_copied_bytes: u64,
    /// Copy-fallback reasons (#835), still as bytes. All `serde(default)` so
    /// old reports without them still parse; new reports always write them.
    #[serde(default)]
    pub store_copy_cross_device_bytes: u64,
    #[serde(default)]
    pub store_copy_permission_bytes: u64,
    #[serde(default)]
    pub store_copy_ineligible_bytes: u64,
    #[serde(default)]
    pub store_copy_other_bytes: u64,
    #[serde(default)]
    pub restore_copy_cross_device_bytes: u64,
    #[serde(default)]
    pub restore_copy_permission_bytes: u64,
    #[serde(default)]
    pub restore_copy_exclusive_bytes: u64,
    #[serde(default)]
    pub restore_copy_other_bytes: u64,
    /// Unique content-addressed blobs in the local store.
    pub store_blobs: u64,
    /// Sum of every cache entry's logical size — what the store would
    /// occupy with no dedup.
    pub logical_bytes: u64,
    /// Physical bytes the unique blobs occupy on disk.
    pub blob_bytes: u64,
    /// Bytes saved by content-addressed dedup (`logical_bytes - blob_bytes`).
    pub dedup_saved_bytes: u64,
    /// Whether indexed unique blob bytes fit within the logical bytes held by
    /// live entries. False means the store index needs repair; in that state a
    /// dedup saving cannot be reported truthfully.
    #[serde(default = "default_true")]
    pub accounting_consistent: bool,
}

fn default_true() -> bool {
    true
}

fn blob_accounting_consistent(logical_bytes: u64, blob_bytes: u64) -> bool {
    blob_bytes <= logical_bytes
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum StorageAccountingState {
    Absent,
    Consistent,
    Inconsistent,
}

fn storage_accounting_state(storage: &StorageBreakdown) -> StorageAccountingState {
    match (
        storage.logical_bytes,
        storage.blob_bytes,
        storage.accounting_consistent,
    ) {
        (0, 0, _) => StorageAccountingState::Absent,
        (_, _, true) => StorageAccountingState::Consistent,
        (_, _, false) => StorageAccountingState::Inconsistent,
    }
}

#[derive(Debug, Serialize, Deserialize)]
pub struct NetworkAnalysis {
    pub bytes_up: u64,
    pub bytes_down: u64,
    pub uploads_ok: usize,
    pub uploads_failed: usize,
    pub downloads_ok: usize,
    pub downloads_failed: usize,
    pub avg_download_ms: f64,
    pub p95_download_ms: u64,
    pub max_download_ms: u64,
    /// Legacy cumulative end-to-end service-time rate. This divides bytes by
    /// the sum of per-download durations; it is not an observed wall-clock rate.
    pub throughput_mbps: f64,
    /// Bytes represented by download events carrying exact v3 wall timestamps.
    #[serde(default)]
    pub observed_bytes_down: u64,
    /// Number of successful downloads represented in the observed wall span.
    #[serde(default)]
    pub observed_downloads: usize,
    /// Span from the earliest timed download start to the latest timed download
    /// finish. It includes overlap and local import work.
    #[serde(default)]
    pub observed_span_ms: u64,
    /// Bytes per observed wall span, accounting for concurrent downloads.
    #[serde(default)]
    pub observed_throughput_mbps: f64,
    /// Peak overlap among successful downloads carrying exact wall timestamps.
    #[serde(default)]
    pub max_concurrent_downloads: usize,
    /// The remote backend CONFIGURED WHEN THIS REPORT WAS GENERATED: `"s3"`,
    /// `"filesystem"`, or `""` when no remote is configured.
    ///
    /// The surrounding field NAMES are HTTP-shaped for backward compatibility
    /// (`network_throughput_mbps`, `total_get_requests`), but a filesystem remote
    /// populates them with local file reads, so a CI consumer otherwise cannot
    /// tell a network regression from a fast local mount.
    ///
    /// Not per-transfer: transfer events do not record which backend served them,
    /// so a report whose window spans a backend switch carries a single label that
    /// does not describe all of its transfers. Hence the explicit name — treat it
    /// as "how the remote is configured now", not "what produced these bytes".
    #[serde(default)]
    pub configured_backend: String,
    /// Legacy cumulative open+read service-time rate.
    pub network_throughput_mbps: f64,
    /// Legacy cumulative response-body service-time rate.
    #[serde(default)]
    pub body_throughput_mbps: f64,
    /// Largest cumulative phase for successful downloads, derived from raw phase totals.
    #[serde(default)]
    pub dominant_download_phase: String,
    #[serde(default)]
    pub dominant_download_phase_ms: u64,
    #[serde(default)]
    pub dominant_download_phase_pct: f64,
    /// Time spent waiting for response headers across GET requests.
    #[serde(default)]
    pub total_request_ms: u64,
    /// Time spent reading response bodies across GET requests.
    #[serde(default)]
    pub total_body_ms: u64,
    /// Time spent waiting for remote-operation concurrency permits.
    #[serde(default)]
    pub total_semaphore_wait_ms: u64,
    /// Time spent on HEAD/existence checks before downloads.
    #[serde(default)]
    pub total_head_ms: u64,
    /// Total number of GET requests issued for successful downloads.
    #[serde(default)]
    pub total_get_requests: u32,
    /// Compression ratio (original / compressed). 0 if no data.
    pub compression_ratio: f64,
    /// Total original (uncompressed) bytes downloaded.
    pub original_bytes_down: u64,
    /// Total time spent in zstd decompression (ms).
    pub total_decompress_ms: u64,
    /// Total time spent extracting downloaded archives (ms).
    #[serde(default)]
    pub total_extract_ms: u64,
    /// Disk I/O time for downloads (directly measured when available), ms.
    pub total_disk_io_ms: u64,
    /// Total time spent importing downloaded entries into SQLite.
    #[serde(default)]
    pub total_import_ms: u64,
    /// Total time spent waiting for the SQLite store lock before import.
    #[serde(default)]
    pub total_import_lock_wait_ms: u64,
    /// Total upload compression time (ms).
    #[serde(default)]
    pub total_compression_ms: u64,
    /// Total upload HEAD check time (ms).
    #[serde(default)]
    pub total_head_checks_ms: u64,
    /// Number of v2 blobs that were already local (dedup savings).
    pub blobs_skipped: u32,
    /// Total v2 blobs across all downloads.
    pub blobs_total: u32,
    #[serde(default)]
    pub v1_downloads: usize,
    #[serde(default)]
    pub v2_downloads: usize,
    #[serde(default)]
    pub v3_downloads: usize,
    #[serde(default)]
    pub unknown_format_downloads: usize,
    pub slowest_downloads: Vec<TransferDetail>,
}

#[derive(Debug, Serialize, Deserialize)]
pub struct PrefetchAnalysis {
    pub prefetch_hits: usize,
    pub total_hits: usize,
    pub contribution_pct: f64,
}

#[derive(Debug, Default, Serialize, Deserialize)]
pub struct BypassAnalysis {
    pub passthroughs: usize,
    /// Query / probe invocations, excluded from `passthroughs` (see
    /// [`ReportSummary::probes`]).
    #[serde(default)]
    pub probes: usize,
    pub skipped: usize,
    pub fallbacks: usize,
    pub direct_passthroughs: usize,
    pub reasons: Vec<BypassReason>,
    pub slowest: Vec<BypassDetail>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BypassReason {
    pub result: String,
    pub route: String,
    pub reason: String,
    pub count: usize,
    pub failures: usize,
    pub max_elapsed_ms: u64,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BypassDetail {
    pub crate_name: String,
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub root: String,
    pub result: String,
    pub route: String,
    pub reason: String,
    #[serde(default)]
    pub start_time: String,
    #[serde(default)]
    pub end_time: String,
    #[serde(default)]
    pub start_unix_ms: i64,
    #[serde(default)]
    pub end_unix_ms: i64,
    pub elapsed_ms: u64,
    pub exit_code: Option<i32>,
    pub timestamp: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub fallback_attempt: Option<crate::fallback::Attempt>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TraceEvent {
    pub name: String,
    pub cat: String,
    pub ph: String,
    /// Chrome trace timestamp in microseconds.
    pub ts: i64,
    /// Chrome trace duration in microseconds.
    pub dur: u64,
    pub pid: u32,
    pub tid: u32,
    /// Perfetto color hint keyed to the result (hit / miss / passthrough …), so
    /// the timeline is legible at a glance. Top-level chrome-trace field;
    /// omitted when absent (#456).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub cname: Option<String>,
    pub args: TraceArgs,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TraceArgs {
    pub crate_name: String,
    pub root: String,
    pub result: String,
    pub route: String,
    pub reason: String,
    pub cache_key: String,
    pub elapsed_ms: u64,
    pub compile_time_ms: u64,
    pub overhead_ms: u64,
    pub size: u64,
    pub compiler_runs: u32,
    pub preprocessor_runs: u32,
    pub probe_runs: u32,
    pub store_output_blobs: u32,
    pub store_duplicate_blobs: u32,
    pub store_new_blobs: u32,
    /// Wrapper phases (ms), the numbers the nested phase slices are drawn
    /// from. `dep_info_ms` is inside `key_ms`; `wait_ms` is flight plus
    /// permit; `unattributed_ms` is the overhead no phase accounts for.
    #[serde(default)]
    pub startup_ms: u64,
    #[serde(default)]
    pub key_ms: u64,
    #[serde(default)]
    pub dep_info_ms: u64,
    #[serde(default)]
    pub dep_info_runs: u32,
    #[serde(default)]
    pub lookup_ms: u64,
    #[serde(default)]
    pub wait_ms: u64,
    #[serde(default)]
    pub flight_wait_ms: u64,
    #[serde(default)]
    pub permit_wait_ms: u64,
    #[serde(default)]
    pub restore_ms: u64,
    #[serde(default)]
    pub store_ms: u64,
    #[serde(default)]
    pub unattributed_ms: u64,
    pub exit_code: Option<i32>,
}

/// One wrapper phase, drawn as an `X` slice nested inside its crate's slice
/// on the same lane. Emitted by the trace format only; the JSON report's
/// `traceEvents` stay one slice per crate so a consumer summing `dur` does
/// not double count.
#[derive(Debug, Clone, Serialize, PartialEq)]
pub struct TracePhaseEvent {
    pub name: String,
    pub cat: String,
    pub ph: String,
    /// Chrome trace timestamp in microseconds.
    pub ts: i64,
    /// Chrome trace duration in microseconds.
    pub dur: u64,
    pub pid: u32,
    pub tid: u32,
    pub args: TracePhaseArgs,
}

#[derive(Debug, Clone, Serialize, PartialEq)]
pub struct TracePhaseArgs {
    pub crate_name: String,
    pub result: String,
    pub phase: String,
    /// The recorded phase time. `dur` can be shorter when the slice was
    /// clamped to its parent.
    pub phase_ms: u64,
    /// The slice this one nests in: the crate slice's name, or `key` for
    /// `dep-info`.
    pub parent: String,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CrateDetail {
    pub crate_name: String,
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub root: String,
    pub result: String,
    #[serde(default)]
    pub start_time: String,
    #[serde(default)]
    pub end_time: String,
    #[serde(default)]
    pub start_unix_ms: i64,
    #[serde(default)]
    pub end_unix_ms: i64,
    pub elapsed_ms: u64,
    pub compile_time_ms: u64,
    pub overhead_ms: u64,
    pub size: u64,
    pub cache_key: String,
    #[serde(default)]
    pub store_output_blobs: u32,
    #[serde(default)]
    pub store_duplicate_blobs: u32,
    #[serde(default)]
    pub store_new_blobs: u32,
    /// Times kache spawned the underlying compiler (0 on a hit, 1 on a
    /// dup/miss). Deterministic; the e2e harness asserts on it.
    #[serde(default)]
    pub compiler_runs: u32,
    /// Times kache spawned the preprocessor (`cc -E`) — once per C/C++
    /// compile for the cache key, 0 for rustc.
    #[serde(default)]
    pub preprocessor_runs: u32,
    /// Times kache spawned a compiler probe (`cc --version` / `cc -###`).
    /// Memoized on disk — one per build per flag set, 0 once warm.
    #[serde(default)]
    pub probe_runs: u32,
    /// Times kache spawned the rustc dep-info pre-pass for this event.
    /// `0` on a warm predicted hit. Deterministic; the e2e harness asserts
    /// on it.
    #[serde(default)]
    pub dep_info_runs: u32,
    /// Sampled verifications this event's key computation ran where the
    /// prediction disagreed with the pre-pass. Zero unless verification was
    /// asked for; the e2e harness asserts on it.
    #[serde(default)]
    pub prediction_mismatches: u32,
    /// Why this compile's outputs were not cached, when `Store::put` failed
    /// (kunobi-ninja/kache#629). Empty on every normal outcome; when set, this
    /// row is a miss that will recur on every build until the cause is fixed.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub store_error: String,
    #[serde(default)]
    pub store_handed_off: bool,
    #[serde(default)]
    pub daemon_store_ms: u64,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TransferDetail {
    pub crate_name: String,
    pub direction: String,
    #[serde(default)]
    pub format: String,
    #[serde(default)]
    pub cache_key: String,
    #[serde(default)]
    pub object_key: String,
    pub compressed_bytes: u64,
    #[serde(default)]
    pub started_at_unix_ms: u64,
    #[serde(default)]
    pub finished_at_unix_ms: u64,
    pub elapsed_ms: u64,
    #[serde(default)]
    pub network_ms: u64,
    #[serde(default)]
    pub semaphore_wait_ms: u64,
    #[serde(default)]
    pub head_ms: u64,
    #[serde(default)]
    pub request_ms: u64,
    #[serde(default)]
    pub body_ms: u64,
    #[serde(default)]
    pub decompress_ms: u64,
    #[serde(default)]
    pub extract_ms: u64,
    #[serde(default)]
    pub disk_io_ms: u64,
    #[serde(default)]
    pub import_lock_wait_ms: u64,
    #[serde(default)]
    pub import_ms: u64,
    #[serde(default)]
    pub request_count: u32,
    #[serde(default)]
    pub blobs_skipped: u32,
    #[serde(default)]
    pub blobs_total: u32,
    pub throughput_mbps: f64,
    pub ok: bool,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ErrorDetail {
    pub crate_name: String,
    pub cache_key: String,
    pub timestamp: String,
}

// ── Report Generation ───────────────────────────────────────────────────────

#[derive(Debug, Clone, Default)]
pub struct ReportFilter {
    pub root: Option<PathBuf>,
    pub last_build: bool,
}

pub fn generate_report(config: &Config, window: SinceWindow, top: usize) -> Result<BuildReport> {
    generate_report_with_filter(config, window, top, &ReportFilter::default())
}

pub fn generate_report_with_filter(
    config: &Config,
    window: SinceWindow,
    top: usize,
    filter: &ReportFilter,
) -> Result<BuildReport> {
    let now = Utc::now();
    let since = window.cutoff(now);
    let mut build_events = if filter.last_build {
        events::read_events(&config.event_log_path())?
    } else {
        events::read_events_since(&config.event_log_path(), since)?
    };
    let mut root_filter = filter.root.as_deref().map(normalize_filter_root);
    if let Some(root) = root_filter.as_deref() {
        build_events.retain(|event| event_matches_root(event, root));
    }
    let session = if filter.last_build {
        let session = select_last_build(&mut build_events)?;
        root_filter = Some(session.root.clone());
        Some(session)
    } else {
        None
    };
    let window = if filter.last_build {
        let start = build_events.iter().map(event_start).min().unwrap();
        SinceWindow::from_secs(now.signed_duration_since(start).num_seconds().max(0) as u64)
    } else {
        window
    };
    let since_ts = window.cutoff_unix_secs(now);
    let transfers = if root_filter.is_some() {
        Vec::new()
    } else {
        events::read_transfers_since(&config.transfer_log_path(), since_ts).unwrap_or_default()
    };

    let stats = events::compute_stats(&build_events);
    let total_compiled = stats.dups + stats.misses;
    let total_cacheable =
        stats.local_hits + stats.prefetch_hits + stats.remote_hits + total_compiled;
    let total_hits = stats.local_hits + stats.prefetch_hits + stats.remote_hits;
    let bypass = build_bypass_analysis(&build_events, top);
    let timeline = build_report_timeline(&build_events);
    let trace_events = build_trace_events(&build_events);

    let hit_rate = if total_cacheable > 0 {
        (total_hits as f64 / total_cacheable as f64) * 100.0
    } else {
        0.0
    };

    let total_compile = stats.hit_compile_time_ms + stats.miss_compile_time_ms;
    let weighted = if total_compile > 0 {
        Some((stats.hit_compile_time_ms as f64 / total_compile as f64) * 100.0)
    } else {
        None
    };

    // Build CrateDetail list
    let all_events: Vec<CrateDetail> = build_events
        .iter()
        .filter(|e| !matches!(e.result, EventResult::Skipped | EventResult::Passthrough))
        .map(to_crate_detail)
        .collect();

    let mut misses: Vec<CrateDetail> = build_events
        .iter()
        .filter(|e| matches!(e.result, EventResult::Dup | EventResult::Miss))
        .map(to_crate_detail)
        .collect();
    misses.sort_by_key(|entry| std::cmp::Reverse(entry.compile_time_ms));

    let mut hits: Vec<CrateDetail> = build_events
        .iter()
        .filter(|e| {
            matches!(
                e.result,
                EventResult::LocalHit | EventResult::PrefetchHit | EventResult::RemoteHit
            )
        })
        .map(to_crate_detail)
        .collect();
    hits.sort_by_key(|entry| std::cmp::Reverse(entry.compile_time_ms));

    let errors_detail: Vec<ErrorDetail> = build_events
        .iter()
        .filter(|e| matches!(e.result, EventResult::Error))
        .map(|e| ErrorDetail {
            crate_name: e.crate_name.clone(),
            cache_key: e.cache_key.clone(),
            timestamp: e.ts.to_rfc3339(),
        })
        .collect();

    // Timing
    let avg_hit_ms = if total_hits > 0 {
        stats.hit_elapsed_ms as f64 / total_hits as f64
    } else {
        0.0
    };
    let avg_miss_ms = if total_compiled > 0 {
        stats.miss_elapsed_ms as f64 / total_compiled as f64
    } else {
        0.0
    };
    let avg_hit_overhead = if total_hits > 0 && stats.hit_compile_time_ms > 0 {
        let overhead = stats.hit_elapsed_ms.saturating_sub(0); // hit_elapsed_ms IS the overhead for hits
        overhead as f64 / total_hits as f64
    } else {
        avg_hit_ms
    };

    // Remote transfers
    let network = if transfers.is_empty() {
        None
    } else {
        let mut analysis = build_network_analysis(&transfers, top);
        analysis.configured_backend = config
            .remote
            .as_ref()
            .map(|remote| remote.backend_kind().to_string())
            .unwrap_or_default();
        Some(analysis)
    };

    // Prefetch
    let prefetch = PrefetchAnalysis {
        prefetch_hits: stats.prefetch_hits,
        total_hits,
        contribution_pct: if total_hits > 0 {
            (stats.prefetch_hits as f64 / total_hits as f64) * 100.0
        } else {
            0.0
        },
    };

    // Suggestions
    let mut suggestions = generate_suggestions(
        &stats,
        &prefetch,
        &network,
        root_filter.is_some(),
        &misses,
        total_cacheable,
        total_hits,
    );
    if let Some(cut) = events::session_cut_by_rotation(&config.event_log_path())
        && build_events.iter().any(|event| event.session_id == cut)
    {
        suggestions.insert(0, rotation_cut_notice(config.event_log_max_size));
    }

    // Storage: restore side — how cache-hit restores landed on disk
    // (reflink / hardlink / copy). Store side — content-addressed dedup,
    // queried from the blob store. Both are deterministic byte counts,
    // independent of machine speed. A missing/unreadable store degrades
    // to zeroed dedup stats rather than failing the whole report.
    let restored_bytes = stats.reflinked_bytes + stats.hardlinked_bytes + stats.copied_bytes;
    let inventory = crate::store_view::read(config, false, "name").unwrap_or_default();
    let blob_stats = inventory.blob_stats;
    let stores = inventory.stores;
    let accounting_consistent = stores.iter().all(|s| {
        s.error.is_none()
            && s.blob_stats
                .as_ref()
                .is_none_or(|b| blob_accounting_consistent(b.total_logical_size, b.total_blob_size))
    });
    for store in &stores {
        if let Some(error) = &store.error {
            suggestions.push(format!(
                "Store {} unavailable: {error}",
                store.path.display()
            ));
        }
    }
    let storage = StorageBreakdown {
        store_entries: inventory.entry_count,
        stores,
        reflinked_bytes: stats.reflinked_bytes,
        hardlinked_bytes: stats.hardlinked_bytes,
        copied_bytes: stats.copied_bytes,
        restored_bytes,
        zero_copy_pct: if restored_bytes > 0 {
            let zero_copy = stats.reflinked_bytes + stats.hardlinked_bytes;
            let pct = zero_copy as f64 / restored_bytes as f64 * 100.0;
            (pct * 10.0).round() / 10.0
        } else {
            0.0
        },
        store_reflinked_bytes: stats.store_reflinked_bytes,
        store_hardlinked_bytes: stats.store_hardlinked_bytes,
        store_copied_bytes: stats.store_copied_bytes,
        store_copy_cross_device_bytes: stats.store_copy_cross_device_bytes,
        store_copy_permission_bytes: stats.store_copy_permission_bytes,
        store_copy_ineligible_bytes: stats.store_copy_ineligible_bytes,
        store_copy_other_bytes: stats.store_copy_other_bytes,
        restore_copy_cross_device_bytes: stats.restore_copy_cross_device_bytes,
        restore_copy_permission_bytes: stats.restore_copy_permission_bytes,
        restore_copy_exclusive_bytes: stats.restore_copy_exclusive_bytes,
        restore_copy_other_bytes: stats.restore_copy_other_bytes,
        store_blobs: blob_stats.total_blobs as u64,
        logical_bytes: blob_stats.total_logical_size,
        blob_bytes: blob_stats.total_blob_size,
        dedup_saved_bytes: blob_stats.savings,
        accounting_consistent,
    };

    Ok(BuildReport {
        schema_version: REPORT_SCHEMA_VERSION,
        meta: ReportMeta {
            kache_version: crate::VERSION.to_string(),
            generated_at: Utc::now().to_rfc3339(),
            since_hours: window.hours(),
            since_secs: window.secs(),
            since: window.label(),
            root_filter: root_filter.clone(),
            session,
        },
        summary: ReportSummary {
            hit_rate_pct: (hit_rate * 10.0).round() / 10.0,
            weighted_hit_rate_pct: weighted.map(|w| (w * 10.0).round() / 10.0),
            time_saved_ms: stats.hit_compile_time_ms,
            total_crates: total_cacheable,
            local_hits: stats.local_hits,
            prefetch_hits: stats.prefetch_hits,
            remote_hits: stats.remote_hits,
            dups: stats.dups,
            misses: stats.misses,
            errors: stats.errors,
            passthroughs: bypass.passthroughs,
            probes: bypass.probes,
            skipped: bypass.skipped,
            store_failures: stats.store_failures,
            fallbacks: bypass.fallbacks,
            total_duration_ms: stats.total_elapsed_ms,
            cache_efficiency_pct: {
                let denom = stats.hit_compile_time_ms + stats.miss_compile_time_ms;
                if denom > 0 {
                    let raw = (stats.hit_compile_time_ms as f64 / denom as f64) * 100.0;
                    (raw * 10.0).round() / 10.0
                } else {
                    0.0
                }
            },
        },
        timeline,
        timing: TimingBreakdown {
            hit_time_ms: stats.hit_elapsed_ms,
            miss_time_ms: stats.miss_elapsed_ms,
            avg_hit_ms: (avg_hit_ms * 10.0).round() / 10.0,
            avg_miss_ms: (avg_miss_ms * 10.0).round() / 10.0,
            avg_hit_overhead_ms: (avg_hit_overhead * 10.0).round() / 10.0,
            miss_compile_time_ms: stats.miss_compile_time_ms,
            avg_key_ms: if total_cacheable > 0 {
                (stats.total_key_ms as f64 / total_cacheable as f64 * 10.0).round() / 10.0
            } else {
                0.0
            },
            avg_lookup_ms: if total_cacheable > 0 {
                (stats.total_lookup_ms as f64 / total_cacheable as f64 * 10.0).round() / 10.0
            } else {
                0.0
            },
            avg_restore_ms: if total_hits > 0 {
                (stats.total_restore_ms as f64 / total_hits as f64 * 10.0).round() / 10.0
            } else {
                0.0
            },
            avg_store_ms: if total_compiled > 0 {
                (stats.total_store_ms as f64 / total_compiled as f64 * 10.0).round() / 10.0
            } else {
                0.0
            },
            total_key_ms: stats.total_key_ms,
            total_lookup_ms: stats.total_lookup_ms,
            total_restore_ms: stats.total_restore_ms,
            total_store_ms: stats.total_store_ms,
            total_startup_ms: stats.total_startup_ms,
            avg_startup_ms: avg_ms(stats.total_startup_ms, total_cacheable as u64),
            total_dep_info_ms: stats.total_dep_info_ms,
            dep_info_runs: stats.total_dep_info_runs,
            prediction_mismatches: stats.total_prediction_mismatches,
            avg_dep_info_ms: avg_ms(stats.total_dep_info_ms, stats.total_dep_info_runs),
            total_wait_ms: stats.total_flight_wait_ms + stats.total_permit_wait_ms,
            total_flight_wait_ms: stats.total_flight_wait_ms,
            total_permit_wait_ms: stats.total_permit_wait_ms,
            avg_wait_ms: avg_ms(
                stats.total_flight_wait_ms + stats.total_permit_wait_ms,
                total_cacheable as u64,
            ),
            total_unattributed_ms: stats.total_unattributed_ms,
            avg_unattributed_ms: avg_ms(stats.total_unattributed_ms, total_cacheable as u64),
        },
        storage,
        network,
        prefetch,
        top_misses: misses.into_iter().take(top).collect(),
        top_hits: hits.into_iter().take(top).collect(),
        all_events,
        trace_events,
        display_time_unit: default_trace_display_time_unit(),
        bypass,
        errors_detail,
        suggestions,
        gc: if filter.last_build {
            None
        } else {
            load_gc_summary_all(config, since)
        },
    })
}

/// Select by completion timestamp, not append order: wrappers finish in parallel.
/// Session IDs belong to a root. Events without IDs can only support an inferred
/// activity group; they cannot identify individual Cargo invocations.
fn select_last_build(events: &mut Vec<BuildEvent>) -> Result<ReportSession> {
    let latest = events
        .iter()
        .max_by_key(|event| event.ts)
        .ok_or_else(|| anyhow::anyhow!("No recorded compiler events for the selected root(s)."))?;
    anyhow::ensure!(
        !latest.root.is_empty(),
        "The latest compiler event has no recorded root; use --root to select a known build tree."
    );
    let session = ReportSession {
        root: latest.root.clone(),
        session_id: latest.session_id.clone(),
        inferred: latest.session_id.is_empty(),
        inactivity_secs: crate::wrapper::BUILD_SESSION_SECS,
    };
    events.retain(|event| event.root == session.root);
    if session.inferred {
        events.sort_by_key(|event| event.ts);
        let mut start = event_start(events.last().unwrap());
        let mut first = events.len();
        for (index, event) in events.iter().enumerate().rev() {
            if !event.session_id.is_empty()
                || start.signed_duration_since(event.ts)
                    >= chrono::Duration::seconds(session.inactivity_secs as i64)
            {
                break;
            }
            first = index;
            start = start.min(event_start(event));
        }
        events.drain(..first);
    } else {
        events.retain(|event| event.session_id == session.session_id);
    }
    Ok(session)
}

fn normalize_filter_root(root: &Path) -> String {
    let abs = if root.is_absolute() {
        root.to_path_buf()
    } else {
        std::env::current_dir()
            .unwrap_or_else(|_| PathBuf::from("."))
            .join(root)
    };
    std::fs::canonicalize(&abs)
        .unwrap_or(abs)
        .to_string_lossy()
        .into_owned()
}

fn event_matches_root(event: &BuildEvent, root: &str) -> bool {
    if event.root.is_empty() {
        return false;
    }
    event.root == root
        || event
            .root
            .strip_prefix(root)
            .is_some_and(|suffix| suffix.starts_with(std::path::MAIN_SEPARATOR))
}

fn load_gc_summary_all(config: &Config, cutoff: DateTime<Utc>) -> Option<GcSummary> {
    std::iter::once(config.clone())
        .chain(crate::store_view::shard_configs(config))
        .filter_map(|store| load_gc_summary(&store.cache_dir, cutoff))
        .reduce(|mut total, next| {
            // load_gc_summary has already validated both timestamps.
            total.last_run = [total.last_run, next.last_run]
                .into_iter()
                .max_by_key(|time| DateTime::parse_from_rfc3339(time).unwrap())
                .unwrap();
            total.entries_evicted += next.entries_evicted;
            total.bytes_freed += next.bytes_freed;
            total.disk_bytes_reclaimed += next.disk_bytes_reclaimed;
            total.shared_bytes_retained += next.shared_bytes_retained;
            total.blobs_removed += next.blobs_removed;
            total
        })
}

/// Load GC stats from gc_stats.json if GC ran at or after `cutoff`.
fn load_gc_summary(cache_dir: &std::path::Path, cutoff: DateTime<Utc>) -> Option<GcSummary> {
    let path = cache_dir.join("gc_stats.json");
    let content = std::fs::read_to_string(&path).ok()?;
    let persisted: GcStatsPersisted = serde_json::from_str(&content).ok()?;

    // Only include if GC ran within the report window
    let last_run = chrono::DateTime::parse_from_rfc3339(&persisted.last_run).ok()?;
    if last_run < cutoff {
        return None;
    }

    Some(GcSummary {
        last_run: persisted.last_run,
        entries_evicted: persisted.entries_evicted,
        bytes_freed: persisted.bytes_freed,
        disk_bytes_reclaimed: persisted.disk_bytes_reclaimed,
        shared_bytes_retained: persisted
            .bytes_freed
            .saturating_sub(persisted.disk_bytes_reclaimed),
        blobs_removed: persisted.blobs_removed,
    })
}

fn build_report_timeline(events: &[BuildEvent]) -> ReportTimeline {
    let Some(first) = events.first() else {
        return ReportTimeline::default();
    };

    let mut start = event_start(first);
    let mut end = first.ts;
    let mut cacheable_count = 0;
    let mut hit_count = 0;
    let mut compiled_count = 0;
    let mut passthrough_count = 0;
    let mut probe_count = 0;
    let mut skipped_count = 0;
    let mut error_count = 0;

    for event in events {
        start = start.min(event_start(event));
        end = end.max(event.ts);
        match event.result {
            EventResult::LocalHit | EventResult::PrefetchHit | EventResult::RemoteHit => {
                cacheable_count += 1;
                hit_count += 1;
            }
            EventResult::Dup | EventResult::Miss => {
                cacheable_count += 1;
                compiled_count += 1;
            }
            EventResult::Error => {
                error_count += 1;
            }
            // A probe / query is a passthrough event but not a compile —
            // counted on its own so `passthrough_count` means refusals.
            EventResult::Passthrough if is_probe_passthrough(event) => {
                probe_count += 1;
            }
            EventResult::Passthrough => {
                passthrough_count += 1;
            }
            EventResult::Skipped => {
                skipped_count += 1;
            }
        }
    }

    let duration_ms = end
        .signed_duration_since(start)
        .num_milliseconds()
        .try_into()
        .unwrap_or(0);

    ReportTimeline {
        start_time: Some(start.to_rfc3339()),
        end_time: Some(end.to_rfc3339()),
        start_unix_ms: Some(start.timestamp_millis()),
        end_unix_ms: Some(end.timestamp_millis()),
        duration_ms,
        event_count: events.len(),
        cacheable_count,
        hit_count,
        compiled_count,
        passthrough_count,
        probe_count,
        skipped_count,
        error_count,
    }
}

fn build_trace_events(events: &[BuildEvent]) -> Vec<TraceEvent> {
    let lanes = assign_trace_lanes(events);
    events
        .iter()
        .enumerate()
        .map(|(index, event)| to_trace_event(event, lanes[index]))
        .collect()
}

/// Assign each event to a worker lane by greedy interval packing on start time:
/// an event takes the lowest-numbered lane whose previous slice has already
/// ended, so concurrent compiles land on distinct lanes and the lane count is
/// the build's peak concurrency — a compact, readable timeline instead of one
/// lane per event. Deterministic (ties broken by event index), and the small
/// `0..N` lane ids never leak OS thread ids (#456).
fn assign_trace_lanes(events: &[BuildEvent]) -> Vec<u32> {
    let mut order: Vec<usize> = (0..events.len()).collect();
    order.sort_by_key(|&i| (event_start(&events[i]).timestamp_micros(), i));

    let mut lane_free_at: Vec<i64> = Vec::new();
    let mut lanes = vec![0u32; events.len()];
    for &i in &order {
        let start = event_start(&events[i]).timestamp_micros();
        let end = start.saturating_add(events[i].elapsed_ms.saturating_mul(1000) as i64);
        let lane = match lane_free_at.iter().position(|&free| free <= start) {
            Some(l) => {
                lane_free_at[l] = end;
                l
            }
            None => {
                lane_free_at.push(end);
                lane_free_at.len() - 1
            }
        };
        lanes[i] = u32::try_from(lane).unwrap_or(u32::MAX);
    }
    lanes
}

/// Slice label prefix + Perfetto color hint (`cname`) for a result, so hits,
/// misses, dups, and passthroughs are visually distinct in the timeline without
/// clicking into a slice (#456).
fn trace_result_style(result: EventResult) -> (&'static str, &'static str) {
    match result {
        EventResult::LocalHit | EventResult::PrefetchHit | EventResult::RemoteHit => {
            ("hit", "good")
        }
        EventResult::Miss => ("miss", "bad"),
        EventResult::Dup => ("dup", "olive"),
        EventResult::Passthrough => ("passthrough", "grey"),
        EventResult::Error => ("error", "terrible"),
        EventResult::Skipped => ("skipped", "grey"),
    }
}

fn to_trace_event(event: &BuildEvent, lane: u32) -> TraceEvent {
    let overhead_ms = event.overhead_ms();
    let start = event_start(event);
    let (label, cname) = trace_result_style(event.result);
    TraceEvent {
        name: format!("{label}: {}", event.crate_name),
        cat: "kache".to_string(),
        ph: "X".to_string(),
        ts: start.timestamp_micros(),
        dur: event.elapsed_ms.saturating_mul(1000),
        pid: 1,
        tid: lane,
        cname: Some(cname.to_string()),
        args: TraceArgs {
            crate_name: event.crate_name.clone(),
            root: event.root.clone(),
            result: event.result.to_string(),
            route: bypass_route(event).to_string(),
            reason: bypass_reason(event),
            cache_key: event.cache_key.clone(),
            elapsed_ms: event.elapsed_ms,
            compile_time_ms: event.compile_time_ms,
            overhead_ms,
            size: event.size,
            compiler_runs: event.compiler_runs,
            preprocessor_runs: event.preprocessor_runs,
            probe_runs: event.probe_runs,
            store_output_blobs: event.store_output_blobs,
            store_duplicate_blobs: event.store_duplicate_blobs,
            store_new_blobs: event.store_new_blobs,
            startup_ms: event.startup_ms,
            key_ms: event.key_ms,
            dep_info_ms: event.dep_info_ms,
            dep_info_runs: event.dep_info_runs,
            lookup_ms: event.lookup_ms,
            wait_ms: event.wait_ms(),
            flight_wait_ms: event.flight_wait_ms,
            permit_wait_ms: event.permit_wait_ms,
            restore_ms: event.restore_ms,
            store_ms: event.store_ms,
            unattributed_ms: event.unattributed_ms(),
            exit_code: event.exit_code,
        },
    }
}

const TRACE_PHASE_CATEGORY: &str = "kache.phase";

/// Lays phase slices along a parent slice's lane, clamped to the parent.
struct PhaseLane<'a> {
    parent: &'a TraceEvent,
    end_us: i64,
    cursor_us: i64,
    slices: Vec<TracePhaseEvent>,
}

impl<'a> PhaseLane<'a> {
    fn new(parent: &'a TraceEvent) -> Self {
        let end_us = parent
            .ts
            .saturating_add(i64::try_from(parent.dur).unwrap_or(i64::MAX));
        Self {
            parent,
            end_us,
            cursor_us: parent.ts,
            slices: Vec::new(),
        }
    }

    /// The next phase in wrapper order: starts at the cursor and advances it.
    fn next(&mut self, phase: &str, ms: u64) {
        let start_us = self.cursor_us;
        if let Some(slice) = self.slice_at(phase, ms, &self.parent.name, start_us) {
            self.cursor_us = start_us.saturating_add(i64::try_from(slice.dur).unwrap_or(i64::MAX));
            self.slices.push(slice);
        }
    }

    /// A phase nested inside an earlier one: placed at that phase's start,
    /// leaving the cursor alone.
    fn nested(&mut self, phase: &str, ms: u64, parent_phase: &str, start_us: i64) {
        if let Some(slice) = self.slice_at(phase, ms, parent_phase, start_us) {
            self.slices.push(slice);
        }
    }

    fn slice_at(
        &self,
        phase: &str,
        ms: u64,
        parent_name: &str,
        start_us: i64,
    ) -> Option<TracePhaseEvent> {
        if ms == 0 || start_us >= self.end_us {
            return None;
        }
        let room_us = u64::try_from(self.end_us - start_us).unwrap_or(0);
        let dur = ms.saturating_mul(1000).min(room_us);
        Some(TracePhaseEvent {
            name: phase.to_string(),
            cat: TRACE_PHASE_CATEGORY.to_string(),
            ph: "X".to_string(),
            ts: start_us,
            dur,
            pid: self.parent.pid,
            tid: self.parent.tid,
            args: TracePhaseArgs {
                crate_name: self.parent.args.crate_name.clone(),
                result: self.parent.args.result.clone(),
                phase: phase.to_string(),
                phase_ms: ms,
                parent: parent_name.to_string(),
            },
        })
    }
}

/// Nested phase slices for one crate slice, in wrapper order: startup, key
/// (with the dep-info pre-pass as its child), lookup, wait, compile when this
/// process compiled, then restore or store.
///
/// The wrapper records durations, not start times, so offsets accumulate
/// from the parent's start in that order and a phase sits where it lands in
/// the sum. What is left at the end of the parent is the unattributed
/// remainder, visible as a gap. Every duration is a whole millisecond, so a
/// phase under 1 ms has no slice. Children never extend past the parent: a
/// phase set that sums past `elapsed_ms` is cut at the parent's end and the
/// phases after it are dropped.
fn trace_phase_events(parent: &TraceEvent) -> Vec<TracePhaseEvent> {
    let args = &parent.args;
    let mut lane = PhaseLane::new(parent);
    lane.next("startup", args.startup_ms);
    let key_start_us = lane.cursor_us;
    lane.next("key", args.key_ms);
    // The pre-pass runs inside key computation; where within it is not
    // recorded, so it is drawn at the key slice's start and never past it.
    lane.nested(
        "dep-info",
        args.dep_info_ms.min(args.key_ms),
        "key",
        key_start_us,
    );
    lane.next("lookup", args.lookup_ms);
    lane.next("wait", args.wait_ms);
    // Overhead equals elapsed on a hit (the stored compile cost was not spent
    // here), so this is non-zero only when this process ran the compiler.
    lane.next("compile", args.elapsed_ms.saturating_sub(args.overhead_ms));
    lane.next("restore", args.restore_ms);
    lane.next("store", args.store_ms);
    lane.slices
}

fn to_crate_detail(e: &BuildEvent) -> CrateDetail {
    let overhead = e.overhead_ms();
    let (start_time, end_time, start_unix_ms, end_unix_ms) = event_timeline(e);
    CrateDetail {
        crate_name: e.crate_name.clone(),
        root: e.root.clone(),
        result: e.result.to_string(),
        start_time,
        end_time,
        start_unix_ms,
        end_unix_ms,
        elapsed_ms: e.elapsed_ms,
        compile_time_ms: e.compile_time_ms,
        overhead_ms: overhead,
        size: e.size,
        cache_key: e.cache_key.clone(),
        store_output_blobs: e.store_output_blobs,
        store_duplicate_blobs: e.store_duplicate_blobs,
        store_new_blobs: e.store_new_blobs,
        compiler_runs: e.compiler_runs,
        preprocessor_runs: e.preprocessor_runs,
        probe_runs: e.probe_runs,
        dep_info_runs: e.dep_info_runs,
        prediction_mismatches: e.prediction_mismatches,
        store_error: e.store_error.clone(),
        store_handed_off: e.store_handed_off,
        daemon_store_ms: e.daemon_store_ms,
    }
}

fn build_bypass_analysis(events: &[BuildEvent], top: usize) -> BypassAnalysis {
    let mut details: Vec<BypassDetail> = events
        .iter()
        .filter(|event| {
            matches!(
                event.result,
                EventResult::Passthrough | EventResult::Skipped
            )
        })
        .map(to_bypass_detail)
        .collect();

    // A probe / query (`--print`, `-vV`, `cc -###`) is recorded as a
    // passthrough event but is NOT a compile kache failed to cache. Split
    // it into its own count so `passthroughs` stays the actionable signal.
    let probes = details
        .iter()
        .filter(|detail| {
            detail.result == "passthrough"
                && passthrough_category(&detail.reason) == "not-a-compile"
        })
        .count();
    let passthroughs = details
        .iter()
        .filter(|detail| {
            detail.result == "passthrough"
                && passthrough_category(&detail.reason) != "not-a-compile"
        })
        .count();
    let skipped = details
        .iter()
        .filter(|detail| detail.result == "skipped")
        .count();
    let fallbacks = details
        .iter()
        .filter(|detail| detail.route == "fallback")
        .count();
    let direct_passthroughs = details
        .iter()
        .filter(|detail| detail.route == "direct")
        .count();

    let mut grouped = std::collections::BTreeMap::<(String, String, String), BypassReason>::new();
    for detail in &details {
        let key = (
            detail.result.clone(),
            detail.route.clone(),
            detail.reason.clone(),
        );
        let entry = grouped.entry(key).or_insert_with(|| BypassReason {
            result: detail.result.clone(),
            route: detail.route.clone(),
            reason: detail.reason.clone(),
            count: 0,
            failures: 0,
            max_elapsed_ms: 0,
        });
        entry.count += 1;
        if detail.exit_code.is_some_and(|code| code != 0) {
            entry.failures += 1;
        }
        entry.max_elapsed_ms = entry.max_elapsed_ms.max(detail.elapsed_ms);
    }

    let mut reasons: Vec<BypassReason> = grouped.into_values().collect();
    reasons.sort_by_key(|reason| {
        (
            std::cmp::Reverse(reason.count),
            std::cmp::Reverse(reason.max_elapsed_ms),
            reason.reason.clone(),
        )
    });
    reasons.truncate(top);

    details.sort_by_key(|detail| std::cmp::Reverse(detail.elapsed_ms));
    details.truncate(top);

    BypassAnalysis {
        passthroughs,
        probes,
        skipped,
        fallbacks,
        direct_passthroughs,
        reasons,
        slowest: details,
    }
}

fn to_bypass_detail(e: &BuildEvent) -> BypassDetail {
    let (start_time, end_time, start_unix_ms, end_unix_ms) = event_timeline(e);
    BypassDetail {
        crate_name: e.crate_name.clone(),
        root: e.root.clone(),
        result: e.result.to_string(),
        route: bypass_route(e).to_string(),
        reason: bypass_reason(e),
        fallback_attempt: e.fallback_attempt.clone(),
        start_time,
        end_time,
        start_unix_ms,
        end_unix_ms,
        elapsed_ms: e.elapsed_ms,
        exit_code: e.exit_code,
        timestamp: e.ts.to_rfc3339(),
    }
}

fn bypass_detail_reason(detail: &BypassDetail) -> String {
    match &detail.fallback_attempt {
        Some(attempt) => format!(
            "{}; fallback `{}`: {}",
            detail.reason, attempt.wrapper, attempt.detail
        ),
        None => detail.reason.clone(),
    }
}

fn event_timeline(e: &BuildEvent) -> (String, String, i64, i64) {
    let start = event_start(e);
    (
        start.to_rfc3339(),
        e.ts.to_rfc3339(),
        start.timestamp_millis(),
        e.ts.timestamp_millis(),
    )
}

fn event_start(e: &BuildEvent) -> DateTime<Utc> {
    let elapsed_ms = i64::try_from(e.elapsed_ms).unwrap_or(i64::MAX);
    e.ts.checked_sub_signed(chrono::Duration::milliseconds(elapsed_ms))
        .unwrap_or(e.ts)
}

fn bypass_route(e: &BuildEvent) -> &'static str {
    match e.result {
        EventResult::Passthrough if e.fallback => "fallback",
        EventResult::Passthrough => "direct",
        EventResult::Skipped => "skipped",
        _ => "n/a",
    }
}

fn bypass_reason(e: &BuildEvent) -> String {
    let reason = e.passthrough_reason.trim();
    if reason.is_empty() {
        "unknown".to_string()
    } else {
        reason.to_string()
    }
}

/// Coarse category of a passthrough reason — the `category` half of the
/// structured `category|detail` string kache emits (see
/// `wrapper::refuse_reason_string`). `not-a-compile` marks a query / probe
/// (`--print`, `-vV`, `cc -###`); `unsupported` marks a real compile kache
/// refused to cache. Policy skips / legacy events carry no prefix, so the
/// whole reason is returned and won't match the probe category.
fn passthrough_category(reason: &str) -> &str {
    reason.split('|').next().unwrap_or("").trim()
}

/// Whether a passthrough event is a probe / query rather than a compile
/// kache failed to cache. Probes get their own count so `passthroughs`
/// stays the actionable "compiles we couldn't cache" signal — a probe is
/// not a compilation at all (see [`crate::compiler::RefuseReason::NotPrimary`]).
fn is_probe_passthrough(e: &BuildEvent) -> bool {
    matches!(e.result, EventResult::Passthrough)
        && passthrough_category(&e.passthrough_reason) == "not-a-compile"
}

fn build_network_analysis(transfers: &[TransferEvent], top: usize) -> NetworkAnalysis {
    let mut bytes_up = 0u64;
    let mut bytes_down = 0u64;
    let mut uploads_ok = 0usize;
    let mut uploads_failed = 0usize;
    let mut downloads_ok = 0usize;
    let mut downloads_failed = 0usize;
    let mut download_latencies: Vec<u64> = Vec::new();
    let mut total_download_bytes = 0u64;
    let mut total_download_ms = 0u64;
    let mut total_network_ms = 0u64;
    let mut total_request_ms = 0u64;
    let mut total_body_ms = 0u64;
    let mut total_semaphore_wait_ms = 0u64;
    let mut total_head_ms = 0u64;
    let mut total_get_requests = 0u32;
    let mut total_original_bytes = 0u64;
    let mut total_decompress_ms = 0u64;
    let mut total_extract_ms = 0u64;
    let mut total_disk_io_ms_measured = 0u64;
    let mut has_disk_io_measurement = false;
    let mut total_import_ms = 0u64;
    let mut total_import_lock_wait_ms = 0u64;
    let mut timed_import_ms = 0u64;
    let mut timed_import_lock_wait_ms = 0u64;
    let mut observed_intervals: Vec<(u64, u64, u64)> = Vec::new();
    let mut total_compression_ms = 0u64;
    let mut total_head_checks_ms = 0u64;
    let mut blobs_skipped = 0u32;
    let mut blobs_total = 0u32;
    let mut v1_downloads = 0usize;
    let mut v2_downloads = 0usize;
    let mut v3_downloads = 0usize;
    let mut unknown_format_downloads = 0usize;

    for t in transfers {
        if t.accounting.as_ref().is_some_and(|accounting| {
            accounting.operation == kache_core::timeline::PrefetchOperation::List
        }) {
            continue;
        }
        match t.direction {
            TransferDirection::Upload => {
                if t.ok {
                    uploads_ok += 1;
                    bytes_up += t.compressed_bytes;
                    total_compression_ms += t.compression_ms;
                    total_head_checks_ms += t.head_checks_ms;
                } else {
                    uploads_failed += 1;
                }
            }
            TransferDirection::Download => {
                if t.ok {
                    downloads_ok += 1;
                    bytes_down += t.compressed_bytes;
                    download_latencies.push(t.elapsed_ms);
                    total_download_bytes += t.compressed_bytes;
                    total_original_bytes += t.original_bytes;
                    total_decompress_ms += t.decompress_ms;
                    total_extract_ms += t.extract_ms;
                    total_import_ms += t.import_ms;
                    total_import_lock_wait_ms += t.import_lock_wait_ms;
                    if t.schema >= 3 {
                        timed_import_ms += t.import_ms;
                        timed_import_lock_wait_ms += t.import_lock_wait_ms;
                    }
                    total_semaphore_wait_ms += t.semaphore_wait_ms;
                    total_head_ms += t.head_ms;
                    total_request_ms += t.request_ms;
                    total_body_ms += t.body_ms;
                    total_get_requests += t.request_count;
                    if t.disk_io_ms > 0 {
                        total_disk_io_ms_measured += t.disk_io_ms;
                        has_disk_io_measurement = true;
                    }
                    blobs_skipped += t.blobs_skipped;
                    blobs_total += t.blobs_total;
                    match t.format.as_str() {
                        "v1" => v1_downloads += 1,
                        "v2" => v2_downloads += 1,
                        "v3" => v3_downloads += 1,
                        _ => unknown_format_downloads += 1,
                    }
                    total_download_ms += t.elapsed_ms;
                    if t.started_at_unix_ms > 0 && t.finished_at_unix_ms > t.started_at_unix_ms {
                        observed_intervals.push((
                            t.started_at_unix_ms,
                            t.finished_at_unix_ms,
                            t.compressed_bytes,
                        ));
                    }
                    // network_ms defaults to 0 for older log entries
                    total_network_ms += if t.network_ms > 0 {
                        t.network_ms
                    } else {
                        t.elapsed_ms
                    };
                } else if !matches!(t.outcome.as_str(), "not_found" | "cancelled" | "skipped") {
                    downloads_failed += 1;
                }
            }
        }
    }

    download_latencies.sort_unstable();

    let avg_download_ms = if !download_latencies.is_empty() {
        total_download_ms as f64 / download_latencies.len() as f64
    } else {
        0.0
    };

    let p95_download_ms = if !download_latencies.is_empty() {
        let idx = (download_latencies.len() * 95 / 100).min(download_latencies.len() - 1);
        download_latencies[idx]
    } else {
        0
    };

    let max_download_ms = download_latencies.last().copied().unwrap_or(0);

    // Cumulative end-to-end service-time rate. Summing durations serializes
    // concurrent downloads, so this must not be presented as wall-clock rate.
    let throughput_mbps = if total_download_ms > 0 {
        (total_download_bytes as f64 / (1024.0 * 1024.0)) / (total_download_ms as f64 / 1000.0)
    } else {
        0.0
    };

    let observed_downloads = observed_intervals.len();
    let observed_bytes_down = observed_intervals
        .iter()
        .map(|(_, _, bytes)| *bytes)
        .sum::<u64>();
    let observed_span_ms = observed_intervals
        .iter()
        .map(|(start, _, _)| *start)
        .min()
        .zip(observed_intervals.iter().map(|(_, end, _)| *end).max())
        .map(|(start, end)| end.saturating_sub(start))
        .unwrap_or(0);
    let observed_throughput_mbps = if observed_span_ms > 0 {
        (observed_bytes_down as f64 / (1024.0 * 1024.0)) / (observed_span_ms as f64 / 1000.0)
    } else {
        0.0
    };
    let mut interval_edges = Vec::with_capacity(observed_intervals.len() * 2);
    for (start, end, _) in &observed_intervals {
        interval_edges.push((*start, 1i32));
        interval_edges.push((*end, -1i32));
    }
    // At a shared timestamp, process finishes before starts so adjacent
    // half-open intervals do not count as overlapping.
    interval_edges.sort_unstable_by_key(|(at, delta)| (*at, *delta));
    let mut concurrent_downloads = 0i32;
    let mut max_concurrent_downloads = 0usize;
    for (_, delta) in interval_edges {
        concurrent_downloads += delta;
        max_concurrent_downloads =
            max_concurrent_downloads.max(concurrent_downloads.max(0) as usize);
    }

    // Remote-read throughput (request + body, excludes decompress/disk).
    let network_throughput_mbps = if total_network_ms > 0 {
        (total_download_bytes as f64 / (1024.0 * 1024.0)) / (total_network_ms as f64 / 1000.0)
    } else {
        0.0
    };

    // Body-only throughput isolates raw object-store transfer once bytes start flowing.
    let body_throughput_mbps = if total_body_ms > 0 {
        (total_download_bytes as f64 / (1024.0 * 1024.0)) / (total_body_ms as f64 / 1000.0)
    } else {
        0.0
    };

    // Slowest downloads
    let mut download_details: Vec<TransferDetail> = transfers
        .iter()
        .filter(|t| matches!(t.direction, TransferDirection::Download) && t.ok)
        .map(|t| {
            let tp = if t.elapsed_ms > 0 {
                (t.compressed_bytes as f64 / (1024.0 * 1024.0)) / (t.elapsed_ms as f64 / 1000.0)
            } else {
                0.0
            };
            TransferDetail {
                crate_name: t.crate_name.clone(),
                direction: "download".to_string(),
                format: t.format.clone(),
                cache_key: t.cache_key.clone(),
                object_key: t.object_key.clone(),
                compressed_bytes: t.compressed_bytes,
                started_at_unix_ms: t.started_at_unix_ms,
                finished_at_unix_ms: t.finished_at_unix_ms,
                elapsed_ms: t.elapsed_ms,
                network_ms: t.network_ms,
                semaphore_wait_ms: t.semaphore_wait_ms,
                head_ms: t.head_ms,
                request_ms: t.request_ms,
                body_ms: t.body_ms,
                decompress_ms: t.decompress_ms,
                extract_ms: t.extract_ms,
                disk_io_ms: t.disk_io_ms,
                import_lock_wait_ms: t.import_lock_wait_ms,
                import_ms: t.import_ms,
                request_count: t.request_count,
                blobs_skipped: t.blobs_skipped,
                blobs_total: t.blobs_total,
                throughput_mbps: (tp * 10.0).round() / 10.0,
                ok: t.ok,
            }
        })
        .collect();
    download_details.sort_by_key(|entry| std::cmp::Reverse(entry.elapsed_ms));

    let compression_ratio = if total_download_bytes > 0 && total_original_bytes > 0 {
        total_original_bytes as f64 / total_download_bytes as f64
    } else {
        0.0
    };

    // Disk I/O: use directly measured value when available, otherwise approximate
    let total_disk_io_ms = if has_disk_io_measurement {
        total_disk_io_ms_measured
    } else {
        total_download_ms.saturating_sub(
            total_network_ms
                + total_decompress_ms
                + total_extract_ms
                + timed_import_lock_wait_ms
                + timed_import_ms,
        )
    };
    let phase_totals = [
        ("wait", total_semaphore_wait_ms),
        ("HEAD", total_head_ms),
        ("request", total_request_ms),
        ("body", total_body_ms),
        ("decompress", total_decompress_ms),
        ("extract", total_extract_ms),
        ("import lock wait", total_import_lock_wait_ms),
        ("import", total_import_ms),
        ("disk", total_disk_io_ms),
    ];
    let phase_total_ms: u64 = phase_totals.iter().map(|(_, ms)| *ms).sum();
    let (dominant_phase, dominant_phase_ms) = phase_totals
        .iter()
        .copied()
        .max_by_key(|(_, ms)| *ms)
        .unwrap_or(("unknown", 0));
    let dominant_phase_pct = if phase_total_ms > 0 {
        dominant_phase_ms as f64 / phase_total_ms as f64 * 100.0
    } else {
        0.0
    };

    NetworkAnalysis {
        configured_backend: String::new(),
        bytes_up,
        bytes_down,
        uploads_ok,
        uploads_failed,
        downloads_ok,
        downloads_failed,
        avg_download_ms: (avg_download_ms * 10.0).round() / 10.0,
        p95_download_ms,
        max_download_ms,
        throughput_mbps: (throughput_mbps * 10.0).round() / 10.0,
        observed_bytes_down,
        observed_downloads,
        observed_span_ms,
        observed_throughput_mbps: (observed_throughput_mbps * 10.0).round() / 10.0,
        max_concurrent_downloads,
        network_throughput_mbps: (network_throughput_mbps * 10.0).round() / 10.0,
        body_throughput_mbps: (body_throughput_mbps * 10.0).round() / 10.0,
        dominant_download_phase: dominant_phase.to_string(),
        dominant_download_phase_ms: dominant_phase_ms,
        dominant_download_phase_pct: (dominant_phase_pct * 10.0).round() / 10.0,
        total_request_ms,
        total_body_ms,
        total_semaphore_wait_ms,
        total_head_ms,
        total_get_requests,
        compression_ratio: (compression_ratio * 10.0).round() / 10.0,
        original_bytes_down: total_original_bytes,
        total_decompress_ms,
        total_extract_ms,
        total_disk_io_ms,
        total_import_ms,
        total_import_lock_wait_ms,
        total_compression_ms,
        total_head_checks_ms,
        blobs_skipped,
        blobs_total,
        v1_downloads,
        v2_downloads,
        v3_downloads,
        unknown_format_downloads,
        slowest_downloads: download_details.into_iter().take(top).collect(),
    }
}

/// The report covers a build whose early events the event log rotated
/// away (kunobi-ninja/kache#1209).
fn rotation_cut_notice(max_size: u64) -> String {
    format!(
        "this build outgrew the event log ({}), so its earliest events were rotated away and \
         the counts above are incomplete; raise [cache] event_log_max_size",
        bytesize::ByteSize(max_size)
    )
}

fn generate_suggestions(
    stats: &events::EventStats,
    prefetch: &PrefetchAnalysis,
    network: &Option<NetworkAnalysis>,
    root_filtered: bool,
    top_misses: &[CrateDetail],
    total_cacheable: usize,
    total_hits: usize,
) -> Vec<String> {
    let mut suggestions = Vec::new();

    // Compiles kache could not store (kunobi-ninja/kache#629). First, because
    // it outranks every tuning hint below: an ordinary miss becomes a hit on the
    // next build, one of these misses forever. Named where possible — the
    // reasons are per-crate causes (a corrupt output, a full disk, store
    // permissions), not one global condition.
    if stats.store_failures > 0 {
        // `top_misses` is already truncated to the report's top-N, so it can
        // hold fewer failures than actually occurred. Name the one example it
        // gives, but take the count from `stats`, which sees every event.
        let example = top_misses.iter().find(|c| !c.store_error.is_empty());
        let detail = match example {
            Some(first) => format!(
                " — e.g. `{}`: {}{}",
                first.crate_name,
                first.store_error,
                if stats.store_failures > 1 {
                    format!(" ({} affected in total)", stats.store_failures)
                } else {
                    String::new()
                }
            ),
            None => String::new(),
        };
        suggestions.push(format!(
            "{} compile{} produced outputs kache failed to store, so {} miss on every build{}",
            stats.store_failures,
            if stats.store_failures == 1 { "" } else { "s" },
            if stats.store_failures == 1 {
                "it will"
            } else {
                "they will"
            },
            detail,
        ));
    }

    // High miss share
    let total_compiled = stats.dups + stats.misses;
    if total_cacheable > 0 && stats.miss_compile_time_ms > 0 {
        let miss_share = stats.miss_compile_time_ms as f64
            / (stats.miss_compile_time_ms + stats.hit_compile_time_ms) as f64
            * 100.0;
        if miss_share > 80.0 && total_compiled > 3 {
            let top_names: Vec<&str> = top_misses
                .iter()
                .take(3)
                .map(|c| c.crate_name.as_str())
                .collect();
            suggestions.push(format!(
                "{:.0}% of compile time spent on compiled cache-key misses — improve hit rate for {}",
                miss_share,
                if top_names.is_empty() {
                    "top compiled cache-key misses".to_string()
                } else {
                    top_names
                        .iter()
                        .map(|n| format!("`{n}`"))
                        .collect::<Vec<_>>()
                        .join(", ")
                },
            ));
        }
    }

    // High hit overhead
    if total_hits > 0 {
        let avg_overhead = stats.hit_elapsed_ms as f64 / total_hits as f64;
        if avg_overhead > 50.0 {
            suggestions.push(format!(
                "Average cache hit overhead is {:.0}ms — check disk I/O or consider faster storage",
                avg_overhead
            ));
        }
    }

    // Low prefetch contribution
    if prefetch.total_hits > 10 && prefetch.contribution_pct < 20.0 {
        suggestions.push(
            "Prefetch contributed <20% of hits — check namespace/shard configuration".to_string(),
        );
    }

    // Remote-transfer issues
    if let Some(net) = network {
        let total_downloads = net.downloads_ok + net.downloads_failed;
        if total_downloads > 0 {
            let fail_rate = net.downloads_failed as f64 / total_downloads as f64 * 100.0;
            if fail_rate > 10.0 {
                suggestions.push(format!(
                    "{:.0}% of downloads failed — check remote connectivity, paths, and credentials",
                    fail_rate
                ));
            }
        }
        if net.downloads_ok > 0 && net.total_get_requests > net.downloads_ok as u32 * 3 {
            suggestions.push(format!(
                "Downloads fan out to {:.1} remote reads per cache hit — check remote layout granularity or prefer pack-first downloads on CI",
                net.total_get_requests as f64 / net.downloads_ok as f64
            ));
        }
        if net.total_semaphore_wait_ms > 10_000 {
            suggestions.push(format!(
                "Aggregate remote semaphore wait totaled {} — tune concurrency only if the remote can absorb it",
                format_duration_ms(net.total_semaphore_wait_ms)
            ));
        }
        if net.total_request_ms > 30_000 && net.total_request_ms > net.total_body_ms {
            suggestions.push(format!(
                "Aggregate remote open/setup latency ({}) exceeds read/transfer time ({}) — check the remote path, storage latency, or read fan-out",
                format_duration_ms(net.total_request_ms),
                format_duration_ms(net.total_body_ms)
            ));
        }
        if net.total_extract_ms > 30_000 && net.total_extract_ms > net.total_body_ms {
            suggestions.push(format!(
                "Aggregate archive extract time ({}) exceeds read/transfer time ({}) — profile zstd/tar extraction and SQLite import separately",
                format_duration_ms(net.total_extract_ms),
                format_duration_ms(net.total_body_ms)
            ));
        }
    }

    if root_filtered {
        suggestions.push(
            "Remote transfer data omitted because transfer events are not root-scoped yet"
                .to_string(),
        );
    } else if network.is_none() {
        suggestions.push("No remote transfer data available for this session".to_string());
    }

    suggestions
}

// ── Output Formatters ───────────────────────────────────────────────────────

fn bypass_total(bypass: &BypassAnalysis) -> usize {
    // Probes included so the detail tables still render for a build whose
    // only uncached activity was query/probe invocations.
    bypass.passthroughs + bypass.probes + bypass.skipped
}

fn format_bypass_summary(bypass: &BypassAnalysis) -> String {
    let mut parts = Vec::new();
    if bypass.passthroughs > 0 {
        let mut part = format!("{} passthrough", bypass.passthroughs);
        if bypass.passthroughs != 1 {
            part.push('s');
        }
        if bypass.fallbacks > 0 {
            part.push_str(&format!(" ({} via fallback)", bypass.fallbacks));
        }
        parts.push(part);
    }
    if bypass.skipped > 0 {
        let mut part = format!("{} skipped", bypass.skipped);
        if bypass.skipped == 1 {
            part = "1 skipped".to_string();
        }
        parts.push(part);
    }
    if bypass.probes > 0 {
        // Probes are not refusals — label them so a clean build's probe
        // traffic doesn't read as a caching problem.
        parts.push(format!("{} probe{}", bypass.probes, plural(bypass.probes)));
    }
    if parts.is_empty() {
        "none".to_string()
    } else {
        parts.join(" / ")
    }
}

/// `""` for 1, `"s"` otherwise — for pluralizing count labels.
fn plural(n: usize) -> &'static str {
    if n == 1 { "" } else { "s" }
}

fn markdown_cell(value: &str) -> String {
    value.replace('\n', " ").replace('|', "\\|")
}

fn format_exit_code(exit_code: Option<i32>) -> String {
    exit_code
        .map(|code| code.to_string())
        .unwrap_or_else(|| "-".to_string())
}

fn cache_roi(report: &BuildReport) -> Option<f64> {
    if report.summary.time_saved_ms > 0 && report.timing.hit_time_ms > 0 {
        let raw = report.summary.time_saved_ms as f64 / report.timing.hit_time_ms as f64;
        Some((raw * 10.0).round() / 10.0)
    } else {
        None
    }
}

fn cache_overhead_summary(report: &BuildReport) -> Option<String> {
    let total_hits =
        report.summary.local_hits + report.summary.prefetch_hits + report.summary.remote_hits;
    if total_hits == 0 || report.timing.hit_time_ms == 0 {
        return None;
    }
    Some(format!(
        "{} aggregate, avg {:.0}ms/hit",
        format_duration_ms(report.timing.hit_time_ms),
        report.timing.avg_hit_ms
    ))
}

fn has_storage_data(storage: &StorageBreakdown) -> bool {
    storage.restored_bytes > 0
        || storage.logical_bytes > 0
        || storage.blob_bytes > 0
        || storage.dedup_saved_bytes > 0
        || storage.store_reflinked_bytes > 0
        || storage.store_hardlinked_bytes > 0
        || storage.store_copied_bytes > 0
        || storage.store_copy_cross_device_bytes > 0
        || storage.store_copy_permission_bytes > 0
        || storage.store_copy_ineligible_bytes > 0
        || storage.store_copy_other_bytes > 0
        || storage.restore_copy_cross_device_bytes > 0
        || storage.restore_copy_permission_bytes > 0
        || storage.restore_copy_exclusive_bytes > 0
        || storage.restore_copy_other_bytes > 0
}

fn push_storage_table(lines: &mut Vec<String>, storage: &StorageBreakdown) {
    lines.push("| Metric | Value |".to_string());
    lines.push("|---|---|".to_string());
    lines.push(format!(
        "| Stored entries (unique keys) | {} |",
        storage.store_entries
    ));
    if storage.restored_bytes > 0 {
        lines.push(format!(
            "| Restored bytes | {} total: {} reflink, {} hardlink, {} copied |",
            format_bytes(storage.restored_bytes),
            format_bytes(storage.reflinked_bytes),
            format_bytes(storage.hardlinked_bytes),
            format_bytes(storage.copied_bytes),
        ));
        lines.push(format!(
            "| Zero-copy restores | {:.1}% |",
            storage.zero_copy_pct
        ));
    }
    match storage_accounting_state(storage) {
        StorageAccountingState::Consistent => {
            lines.push(format!(
                "| Store footprint | {} logical -> {} blobs, {} dedup saved |",
                format_bytes(storage.logical_bytes),
                format_bytes(storage.blob_bytes),
                format_bytes(storage.dedup_saved_bytes),
            ));
            lines.push(format!("| Store blobs | {} |", storage.store_blobs));
        }
        StorageAccountingState::Inconsistent => {
            lines.push(format!(
                "| Store accounting | **Inconsistent:** {} logical entry bytes, {} indexed blob bytes; the store index needs repair |",
                format_bytes(storage.logical_bytes),
                format_bytes(storage.blob_bytes),
            ));
            lines.push(format!("| Store blobs | {} |", storage.store_blobs));
        }
        StorageAccountingState::Absent => {}
    }
    let ingested =
        storage.store_reflinked_bytes + storage.store_hardlinked_bytes + storage.store_copied_bytes;
    if ingested > 0 {
        // Reflinked and hardlinked ingest share storage with the build's own
        // output, so they add ~no physical disk; only copied ingest is a
        // genuine second copy.
        let zero_copy = storage.store_reflinked_bytes + storage.store_hardlinked_bytes;
        let zero_copy_pct = zero_copy as f64 / ingested as f64 * 100.0;
        lines.push(format!(
            "| Store ingest | {} reflinked (CoW), {} hardlinked, {} copied — {:.1}% shared with build output |",
            format_bytes(storage.store_reflinked_bytes),
            format_bytes(storage.store_hardlinked_bytes),
            format_bytes(storage.store_copied_bytes),
            (zero_copy_pct * 10.0).round() / 10.0,
        ));
    }
    let store_copy_reasons = copy_reason_bytes_total(
        storage.store_copy_cross_device_bytes,
        storage.store_copy_permission_bytes,
        storage.store_copy_ineligible_bytes,
        storage.store_copy_other_bytes,
    );
    if store_copy_reasons > 0 {
        lines.push(format!(
            "| Store copy reasons | {} cross-device (EXDEV), {} permission (EPERM), {} kind-ineligible, {} other |",
            format_bytes(storage.store_copy_cross_device_bytes),
            format_bytes(storage.store_copy_permission_bytes),
            format_bytes(storage.store_copy_ineligible_bytes),
            format_bytes(storage.store_copy_other_bytes),
        ));
    }
    let restore_copy_reasons = copy_reason_bytes_total(
        storage.restore_copy_cross_device_bytes,
        storage.restore_copy_permission_bytes,
        storage.restore_copy_exclusive_bytes,
        storage.restore_copy_other_bytes,
    );
    if restore_copy_reasons > 0 {
        lines.push(format!(
            "| Restore copy reasons | {} cross-device (EXDEV), {} permission (EPERM), {} exclusive-carrier, {} other |",
            format_bytes(storage.restore_copy_cross_device_bytes),
            format_bytes(storage.restore_copy_permission_bytes),
            format_bytes(storage.restore_copy_exclusive_bytes),
            format_bytes(storage.restore_copy_other_bytes),
        ));
    }
}

fn copy_reason_bytes_total(a: u64, b: u64, c: u64, d: u64) -> u64 {
    a + b + c + d
}

fn push_error_table(lines: &mut Vec<String>, errors: &[ErrorDetail]) {
    lines.push("| Crate | Time | Key |".to_string());
    lines.push("|---|---|---|".to_string());
    for err in errors.iter().take(10) {
        let key_short = if err.cache_key.len() > 12 {
            &err.cache_key[..12]
        } else {
            &err.cache_key
        };
        lines.push(format!(
            "| `{}` | {} | `{}` |",
            markdown_cell(&err.crate_name),
            err.timestamp,
            markdown_cell(key_short),
        ));
    }
    if errors.len() > 10 {
        lines.push(format!("| *... {} more* | | |", errors.len() - 10));
    }
}

fn push_bypass_tables(lines: &mut Vec<String>, bypass: &BypassAnalysis) {
    if !bypass.reasons.is_empty() {
        lines.push("| Result | Route | Reason | Count | Failures | Max time |".to_string());
        lines.push("|---|---|---|---:|---:|---:|".to_string());
        for reason in &bypass.reasons {
            lines.push(format!(
                "| {} | {} | {} | {} | {} | {} |",
                reason.result,
                reason.route,
                markdown_cell(&reason.reason),
                reason.count,
                reason.failures,
                format_duration_ms(reason.max_elapsed_ms),
            ));
        }
    }

    if !bypass.slowest.is_empty() {
        lines.push(String::new());
        lines.push("**Slowest bypassed invocations:**".to_string());
        lines.push(String::new());
        lines.push("| Crate | Result | Route | Time | Exit | Reason |".to_string());
        lines.push("|---|---|---|---:|---:|---|".to_string());
        for detail in &bypass.slowest {
            lines.push(format!(
                "| `{}` | {} | {} | {} | {} | {} |",
                markdown_cell(&detail.crate_name),
                detail.result,
                detail.route,
                format_duration_ms(detail.elapsed_ms),
                format_exit_code(detail.exit_code),
                markdown_cell(&bypass_detail_reason(detail)),
            ));
        }
    }
}

pub fn format_json(report: &BuildReport) -> Result<String> {
    Ok(serde_json::to_string_pretty(report)?)
}

pub fn format_trace_json(report: &BuildReport) -> Result<String> {
    #[derive(Serialize)]
    struct TraceOutput<'a> {
        #[serde(rename = "displayTimeUnit")]
        display_time_unit: &'a str,
        #[serde(rename = "traceEvents")]
        trace_events: Vec<serde_json::Value>,
        #[serde(skip_serializing_if = "Option::is_none")]
        session: Option<&'a ReportSession>,
    }

    // Lead with chrome-trace metadata events that name the process and each
    // worker lane, so Perfetto labels them "kache" / "worker N" instead of
    // "Process 1" / "Thread <n>" (#456).
    let mut trace_events: Vec<serde_json::Value> = Vec::new();
    trace_events.push(trace_metadata_event("process_name", 0, "kache"));
    let mut lanes: Vec<u32> = report.trace_events.iter().map(|e| e.tid).collect();
    lanes.sort_unstable();
    lanes.dedup();
    for tid in lanes {
        trace_events.push(trace_metadata_event(
            "thread_name",
            tid,
            &format!("worker {tid}"),
        ));
    }
    // Parent first, then its phases: Perfetto nests `X` slices by containment
    // and keeps file order for equal timestamps, so the crate slice must
    // precede the `startup` slice that starts with it.
    for event in &report.trace_events {
        trace_events.push(serde_json::to_value(event)?);
        for phase in trace_phase_events(event) {
            trace_events.push(serde_json::to_value(phase)?);
        }
    }

    Ok(serde_json::to_string_pretty(&TraceOutput {
        display_time_unit: &report.display_time_unit,
        trace_events,
        session: report.meta.session.as_ref(),
    })?)
}

/// A chrome-trace `M` (metadata) event used to label the process / worker lanes.
fn trace_metadata_event(name: &str, tid: u32, value: &str) -> serde_json::Value {
    serde_json::json!({
        "name": name,
        "ph": "M",
        "pid": 1,
        "tid": tid,
        "args": { "name": value },
    })
}

/// Backend-neutral label for the phase names retained in the JSON telemetry.
fn human_download_phase(phase: &str) -> &str {
    match phase {
        "wait" => "queue wait",
        "HEAD" => "existence check",
        "request" => "open/setup",
        "body" => "read/transfer",
        "disk" => "local disk",
        other => other,
    }
}

/// Per-phase rows of the Markdown timing table, as a share of `total_ms`
/// (tracked wrapper time). Dep-info is indented under key because it is
/// inside it; unattributed is what the phases leave of the overhead.
fn push_phase_rows(lines: &mut Vec<String>, t: &TimingBreakdown, total_ms: u64) {
    use crate::cli::format_duration_ms;

    let mut row = |label: String, ms: u64| {
        lines.push(format!(
            "| {label} | {} | {:.1}% |",
            format_duration_ms(ms),
            pct_of(ms, total_ms)
        ));
    };
    row("Startup".to_string(), t.total_startup_ms);
    row("Key computation".to_string(), t.total_key_ms);
    row(
        format!(
            "&nbsp;&nbsp;of which dep-info pre-pass ({} runs)",
            t.dep_info_runs
        ),
        t.total_dep_info_ms,
    );
    row("Lookup".to_string(), t.total_lookup_ms);
    row(
        format!(
            "Scheduler wait (flight {}, permit {})",
            format_duration_ms(t.total_flight_wait_ms),
            format_duration_ms(t.total_permit_wait_ms)
        ),
        t.total_wait_ms,
    );
    row("Restore".to_string(), t.total_restore_ms);
    row("Store".to_string(), t.total_store_ms);
    row("Unattributed".to_string(), t.total_unattributed_ms);
}

pub fn format_markdown(report: &BuildReport) -> String {
    use crate::cli::format_duration_ms;

    let mut lines = Vec::new();
    let s = &report.summary;
    let t = &report.timing;

    let total_hits = s.local_hits + s.prefetch_hits + s.remote_hits;
    let total_compiled = s.dups + s.misses;
    lines.push("### kache build report".to_string());
    lines.push(String::new());
    lines.push(format!(
        "**{:.1}% hit rate** — {}/{} cacheable crates from cache, {} compiled | **{} compile work avoided**",
        s.hit_rate_pct,
        total_hits,
        s.total_crates,
        total_compiled,
        format_duration_ms(s.time_saved_ms),
    ));
    lines.push(String::new());

    // Summary table
    lines.push("#### Summary".to_string());
    lines.push("| Metric | Value |".to_string());
    lines.push("|---|---|".to_string());
    lines.push(format!("| Window | last {} |", report.meta.window_label()));
    if let Some(session) = &report.meta.session {
        lines.push(format!(
            "| Scope | {} |",
            markdown_cell(&session.description())
        ));
    }
    lines.push(format!("| Hit rate (count) | {:.1}% |", s.hit_rate_pct));
    if let Some(w) = s.weighted_hit_rate_pct {
        lines.push(format!("| Hit rate (compile-cost weighted) | {:.1}% |", w));
    }
    lines.push(format!(
        "| Compile work avoided | {} aggregate |",
        format_duration_ms(s.time_saved_ms)
    ));
    if let Some(overhead) = cache_overhead_summary(report) {
        lines.push(format!("| Cache hit overhead | {} |", overhead));
    }
    if let Some(roi) = cache_roi(report) {
        lines.push(format!(
            "| Cache ROI | {:.1}x compile work per cache-hit overhead |",
            roi
        ));
    }
    if t.miss_compile_time_ms > 0 {
        lines.push(format!(
            "| Miss compile work | {} aggregate |",
            format_duration_ms(t.miss_compile_time_ms)
        ));
    }
    lines.push(format!("| Total crates | {} |", s.total_crates));
    lines.push(format!(
        "| Hits | {} (local: {}, prefetch: {}, remote: {}) |",
        total_hits, s.local_hits, s.prefetch_hits, s.remote_hits
    ));
    if s.dups > 0 {
        lines.push(format!("| Dups | {} |", s.dups));
    }
    lines.push(format!("| Misses | {} |", s.misses));
    if s.store_failures > 0 {
        lines.push(format!(
            "| Compiled but not cached | {} (will miss again) |",
            s.store_failures
        ));
    }
    if s.errors > 0 {
        lines.push(format!("| Errors | {} |", s.errors));
    }
    if s.passthroughs > 0 || s.skipped > 0 || s.probes > 0 {
        lines.push(format!(
            "| Passthroughs / skipped | {} |",
            format_bypass_summary(&report.bypass)
        ));
    }
    lines.push(String::new());

    // Timing table
    let total_ms = t.hit_time_ms + t.miss_time_ms;
    lines.push("#### Timing".to_string());
    lines.push("| Phase | Aggregate time | % of tracked wrapper time |".to_string());
    lines.push("|---|---|---|".to_string());
    let hit_pct = if total_ms > 0 {
        t.hit_time_ms as f64 / total_ms as f64 * 100.0
    } else {
        0.0
    };
    let miss_pct = if total_ms > 0 {
        t.miss_time_ms as f64 / total_ms as f64 * 100.0
    } else {
        0.0
    };
    lines.push(format!(
        "| Cache hits (wrapper overhead) | {} | {:.1}% |",
        format_duration_ms(t.hit_time_ms),
        hit_pct
    ));
    lines.push(format!(
        "| Compiles (wrapper total) | {} | {:.1}% |",
        format_duration_ms(t.miss_time_ms),
        miss_pct
    ));
    if s.total_crates > 0 {
        push_phase_rows(&mut lines, t, total_ms);
    }
    lines.push(String::new());

    // Remote-transfer table
    if let Some(net) = &report.network {
        lines.push("#### Remote transfer".to_string());
        lines.push("| Metric | Value |".to_string());
        lines.push("|---|---|".to_string());
        lines.push(format!(
            "| Downloaded | {} ({} crates) |",
            format_bytes(net.bytes_down),
            net.downloads_ok
        ));
        lines.push(format!(
            "| Uploaded | {} ({} crates) |",
            format_bytes(net.bytes_up),
            net.uploads_ok
        ));
        lines.push(format!(
            "| Avg download time | {:.0}ms |",
            net.avg_download_ms
        ));
        lines.push(format!("| P95 download time | {}ms |", net.p95_download_ms));
        if net.observed_span_ms > 0 {
            lines.push(format!(
                "| Observed wall-span throughput | {:.1} MB/s over {} ({} timed downloads, peak {} concurrent) |",
                net.observed_throughput_mbps,
                format_duration_ms(net.observed_span_ms),
                net.observed_downloads,
                net.max_concurrent_downloads
            ));
        } else if net.downloads_ok > 0 {
            lines.push(
                "| Observed wall-span throughput | unavailable (legacy transfer events) |"
                    .to_string(),
            );
        }
        lines.push(format!(
            "| Cumulative service-time rates | read {:.1} MB/s, open+read {:.1} MB/s, end-to-end {:.1} MB/s |",
            net.body_throughput_mbps, net.network_throughput_mbps, net.throughput_mbps
        ));
        if !net.dominant_download_phase.is_empty() && net.dominant_download_phase_ms > 0 {
            lines.push(format!(
                "| Dominant cumulative download phase | {} — {} ({:.1}%) |",
                human_download_phase(&net.dominant_download_phase),
                format_duration_ms(net.dominant_download_phase_ms),
                net.dominant_download_phase_pct
            ));
        }
        if net.compression_ratio > 0.0 {
            lines.push(format!(
                "| Compression ratio | {:.1}x ({} → {}) |",
                net.compression_ratio,
                format_bytes(net.original_bytes_down),
                format_bytes(net.bytes_down)
            ));
        }
        if net.total_semaphore_wait_ms > 0
            || net.total_head_ms > 0
            || net.total_decompress_ms > 0
            || net.total_extract_ms > 0
            || net.total_import_lock_wait_ms > 0
            || net.total_import_ms > 0
            || net.total_disk_io_ms > 0
        {
            lines.push(format!(
                "| Cumulative download phase time | queue wait {}ms, existence check {}ms, open/setup {}ms, read/transfer {}ms, decompress {}ms, extract {}ms, import lock wait {}ms, import execution {}ms, local disk I/O {}ms |",
                net.total_semaphore_wait_ms,
                net.total_head_ms,
                net.total_request_ms,
                net.total_body_ms,
                net.total_decompress_ms,
                net.total_extract_ms,
                net.total_import_lock_wait_ms,
                net.total_import_ms,
                net.total_disk_io_ms
            ));
        }
        if net.blobs_total > 0 {
            lines.push(format!(
                "| Blob dedup | {} / {} blobs already local ({:.0}% skipped) |",
                net.blobs_skipped,
                net.blobs_total,
                if net.blobs_total > 0 {
                    net.blobs_skipped as f64 / net.blobs_total as f64 * 100.0
                } else {
                    0.0
                }
            ));
        }
        if net.downloads_failed > 0 {
            lines.push(format!("| Failed downloads | {} |", net.downloads_failed));
        }
        if net.uploads_failed > 0 {
            lines.push(format!("| Failed uploads | {} |", net.uploads_failed));
        }
        lines.push(String::new());

        // Slowest downloads
        if !net.slowest_downloads.is_empty() {
            lines.push("#### Slowest Downloads".to_string());
            lines.push(
                "| Crate | Size | Time | Key | Wait/Check | Open/Read | Extract/Lock/Import |"
                    .to_string(),
            );
            lines.push("|---|---|---|---|---|---|---|".to_string());
            for d in &net.slowest_downloads {
                let key = if d.cache_key.is_empty() {
                    "?"
                } else {
                    &d.cache_key[..d.cache_key.len().min(12)]
                };
                lines.push(format!(
                    "| `{}` | {} | {}ms | `{}` | {}/{}ms | {}/{}ms | {}/{}/{}ms |",
                    d.crate_name,
                    format_bytes(d.compressed_bytes),
                    d.elapsed_ms,
                    key,
                    d.semaphore_wait_ms,
                    d.head_ms,
                    d.request_ms,
                    d.body_ms,
                    d.extract_ms.max(d.decompress_ms),
                    d.import_lock_wait_ms,
                    d.import_ms,
                ));
            }
            let repro_keys: Vec<_> = net
                .slowest_downloads
                .iter()
                .filter(|d| !d.object_key.is_empty())
                .take(3)
                .collect();
            if !repro_keys.is_empty() {
                lines.push(String::new());
                lines.push("Raw object keys for reproduction:".to_string());
                for d in repro_keys {
                    lines.push(format!("- `{}`: `{}`", d.crate_name, d.object_key));
                }
            }
            lines.push(String::new());
        }
    }

    if has_storage_data(&report.storage) {
        lines.push("#### Storage".to_string());
        push_storage_table(&mut lines, &report.storage);
        lines.push(String::new());
    }

    // Prefetch
    let p = &report.prefetch;
    lines.push("#### Prefetch".to_string());
    lines.push("| Metric | Value |".to_string());
    lines.push("|---|---|".to_string());
    lines.push(format!(
        "| Prefetch hits | {} / {} total hits |",
        p.prefetch_hits, p.total_hits
    ));
    lines.push(format!("| Contribution | {:.1}% |", p.contribution_pct));
    lines.push(String::new());

    if bypass_total(&report.bypass) > 0 {
        lines.push("#### Passthroughs & Skips".to_string());
        push_bypass_tables(&mut lines, &report.bypass);
        lines.push(String::new());
    }

    if !report.errors_detail.is_empty() {
        lines.push("#### Errors".to_string());
        push_error_table(&mut lines, &report.errors_detail);
        lines.push(String::new());
    }

    // Top compiled cache-key misses (dups + misses)
    if !report.top_misses.is_empty() {
        lines.push("#### Top Compiled Cache-Key Misses".to_string());
        lines.push("| Crate | Compile time | Size | Key |".to_string());
        lines.push("|---|---|---|---|".to_string());
        for c in &report.top_misses {
            let key_short = if c.cache_key.len() > 12 {
                &c.cache_key[..12]
            } else {
                &c.cache_key
            };
            lines.push(format!(
                "| `{}` | {} | {} | `{}` |",
                c.crate_name,
                format_duration_ms(c.compile_time_ms),
                format_bytes(c.size),
                key_short,
            ));
        }
        lines.push(String::new());
    }

    // Top cache hits
    if !report.top_hits.is_empty() {
        lines.push("#### Top Cache Hits (most expensive cached)".to_string());
        lines.push("| Crate | Compile cost | Size | Key |".to_string());
        lines.push("|---|---|---|---|".to_string());
        for c in &report.top_hits {
            let key_short = if c.cache_key.len() > 12 {
                &c.cache_key[..12]
            } else {
                &c.cache_key
            };
            lines.push(format!(
                "| `{}` | {} | {} | `{}` |",
                c.crate_name,
                format_duration_ms(c.compile_time_ms),
                format_bytes(c.size),
                key_short,
            ));
        }
        lines.push(String::new());
    }

    // Suggestions
    if !report.suggestions.is_empty() {
        lines.push("#### Suggestions".to_string());
        for s in &report.suggestions {
            lines.push(format!("- {s}"));
        }
        lines.push(String::new());
    }

    // GC
    if let Some(gc) = &report.gc {
        lines.push("#### GC".to_string());
        lines.push("| Metric | Value |".to_string());
        lines.push("|---|---|".to_string());
        lines.push(format!("| Last run | {} |", gc.last_run));
        lines.push(format!("| Entries evicted | {} |", gc.entries_evicted));
        lines.push(format!(
            "| Store bytes removed | {} |",
            format_bytes(gc.bytes_freed)
        ));
        lines.push(format!(
            "| Disk bytes reclaimed | {} |",
            format_bytes(gc.disk_bytes_reclaimed)
        ));
        lines.push(format!(
            "| Shared bytes retained | {} |",
            format_bytes(gc.shared_bytes_retained)
        ));
        lines.push(format!("| Blobs removed | {} |", gc.blobs_removed));
        lines.push(String::new());
    }

    lines.join("\n")
}

/// GitHub-optimized markdown: compact key metrics always visible, details in collapsible sections.
/// Designed to be posted directly as a PR comment by kache-action.
pub fn format_github(report: &BuildReport) -> String {
    use crate::cli::format_duration_ms;

    let mut lines = Vec::new();
    let s = &report.summary;
    let t = &report.timing;
    let total_hits = s.local_hits + s.prefetch_hits + s.remote_hits;
    let total_compiled = s.dups + s.misses;

    // Header
    lines.push("### kache build cache".to_string());
    lines.push(String::new());
    lines.push(format!(
        "**{:.1}%** hit rate — {}/{} cacheable crates from cache, {} compiled | **{} compile work avoided**",
        s.hit_rate_pct,
        total_hits,
        s.total_crates,
        total_compiled,
        format_duration_ms(s.time_saved_ms),
    ));
    lines.push(String::new());

    // ── Key metrics (always visible) ──
    lines.push("| | |".to_string());
    lines.push("|---|---|".to_string());
    lines.push(format!(
        "| **Window** | last {} |",
        report.meta.window_label()
    ));
    if let Some(session) = &report.meta.session {
        lines.push(format!(
            "| **Scope** | {} |",
            markdown_cell(&session.description())
        ));
    }
    lines.push(format!(
        "| **Crates** | {} cached / {} compiled / {} total |",
        total_hits, total_compiled, s.total_crates
    ));
    if s.dups > 0 {
        lines.push(format!(
            "| **Dups** | {} storage duplicates after compile |",
            s.dups
        ));
    }
    lines.push(format!(
        "| **Hit rate** | {:.1}% count{} |",
        s.hit_rate_pct,
        s.weighted_hit_rate_pct
            .map(|w| format!(" / {:.1}% by compile cost", w))
            .unwrap_or_default()
    ));
    lines.push(format!(
        "| **Compile work avoided** | {} aggregate |",
        format_duration_ms(s.time_saved_ms)
    ));
    if let Some(overhead) = cache_overhead_summary(report) {
        lines.push(format!("| **Cache hit overhead** | {} |", overhead));
    }
    if let Some(roi) = cache_roi(report) {
        lines.push(format!(
            "| **Cache ROI** | {:.1}x compile work per cache-hit overhead |",
            roi
        ));
    }
    if t.miss_compile_time_ms > 0 {
        lines.push(format!(
            "| **Miss compile work** | {} aggregate |",
            format_duration_ms(t.miss_compile_time_ms)
        ));
    }
    if s.store_failures > 0 {
        lines.push(format!(
            "| **Compiled but not cached** | {} (will miss again) |",
            s.store_failures
        ));
    }
    if s.errors > 0 {
        lines.push(format!("| **Errors** | {} |", s.errors));
    }
    if s.passthroughs > 0 || s.skipped > 0 || s.probes > 0 {
        lines.push(format!(
            "| **Passthroughs / skipped** | {} |",
            format_bypass_summary(&report.bypass)
        ));
    }

    // ── Suggestions (always visible — actionable) ──
    if !report.suggestions.is_empty() {
        lines.push(String::new());
        for sg in &report.suggestions {
            lines.push(format!("> {sg}"));
        }
    }

    // ── Top cache-key misses (collapsed) ──
    if !report.top_misses.is_empty() {
        lines.push(String::new());
        lines.push("<details>".to_string());
        lines.push(format!(
            "<summary><strong>Top compiled cache-key misses</strong> ({} compiled)</summary>",
            total_compiled
        ));
        lines.push(String::new());
        lines.push("| Crate | Compile time | Size |".to_string());
        lines.push("|-------|-------------|------|".to_string());
        for c in report.top_misses.iter().take(10) {
            lines.push(format!(
                "| `{}` | {} | {} |",
                c.crate_name,
                format_duration_ms(c.compile_time_ms),
                format_bytes(c.size),
            ));
        }
        if total_compiled > 10 {
            lines.push(format!("| *... {} more* | | |", total_compiled - 10));
        }
        lines.push(String::new());
        lines.push("</details>".to_string());
    }

    // ── Top hits (collapsed) ──
    if !report.top_hits.is_empty() {
        lines.push(String::new());
        lines.push("<details>".to_string());
        lines.push(format!(
            "<summary><strong>Expensive cache hits</strong> — top {} by avoided compile work</summary>",
            report.top_hits.len().min(10)
        ));
        lines.push(String::new());
        lines.push("| Crate | Avoided compile work | Size |".to_string());
        lines.push("|-------|----------------------|------|".to_string());
        for c in report.top_hits.iter().take(10) {
            lines.push(format!(
                "| `{}` | {} | {} |",
                markdown_cell(&c.crate_name),
                format_duration_ms(c.compile_time_ms),
                format_bytes(c.size),
            ));
        }
        lines.push(String::new());
        lines.push("</details>".to_string());
    }

    // ── Passthroughs / skipped (collapsed) ──
    if bypass_total(&report.bypass) > 0 {
        lines.push(String::new());
        lines.push("<details>".to_string());
        lines.push(format!(
            "<summary><strong>Passthroughs & skips</strong> — {}</summary>",
            format_bypass_summary(&report.bypass)
        ));
        lines.push(String::new());
        push_bypass_tables(&mut lines, &report.bypass);
        lines.push(String::new());
        lines.push("</details>".to_string());
    }

    // ── Errors (collapsed) ──
    if !report.errors_detail.is_empty() {
        lines.push(String::new());
        lines.push("<details>".to_string());
        lines.push(format!(
            "<summary><strong>Errors</strong> — {}</summary>",
            report.errors_detail.len()
        ));
        lines.push(String::new());
        push_error_table(&mut lines, &report.errors_detail);
        lines.push(String::new());
        lines.push("</details>".to_string());
    }

    // ── Remote transfer (collapsed) ──
    if let Some(net) = &report.network {
        let (summary_rate, summary_rate_label) = if net.observed_span_ms > 0 {
            (net.observed_throughput_mbps, "observed wall span")
        } else {
            (net.body_throughput_mbps, "cumulative read service")
        };

        lines.push(String::new());
        lines.push("<details>".to_string());
        let dominant_summary =
            if !net.dominant_download_phase.is_empty() && net.dominant_download_phase_ms > 0 {
                format!(
                    ", dominant cumulative {}",
                    human_download_phase(&net.dominant_download_phase)
                )
            } else {
                String::new()
            };
        lines.push(format!(
            "<summary><strong>Remote transfer</strong> — {} downloaded, {:.0} MB/s {}{}</summary>",
            format_bytes(net.bytes_down),
            summary_rate,
            summary_rate_label,
            dominant_summary
        ));
        lines.push(String::new());
        lines.push("| | |".to_string());
        lines.push("|---|---|".to_string());
        lines.push(format!(
            "| Downloaded | {} ({} crates) |",
            format_bytes(net.bytes_down),
            net.downloads_ok
        ));
        if net.uploads_ok > 0 || net.uploads_failed > 0 {
            lines.push(format!(
                "| Uploaded | {} ({} crates) |",
                format_bytes(net.bytes_up),
                net.uploads_ok
            ));
            if net.total_compression_ms > 0 || net.total_head_checks_ms > 0 {
                lines.push(format!(
                    "| Upload time split | compress {}ms + existence checks {}ms |",
                    net.total_compression_ms, net.total_head_checks_ms,
                ));
            }
        }
        lines.push(format!(
            "| Download time | avg {:.0}ms · p95 {}ms |",
            net.avg_download_ms, net.p95_download_ms
        ));
        if net.v1_downloads > 0
            || net.v2_downloads > 0
            || net.v3_downloads > 0
            || net.unknown_format_downloads > 0
        {
            lines.push(format!(
                "| Download format | v1 {} · v2 {} · v3 {} · unknown {} |",
                net.v1_downloads, net.v2_downloads, net.v3_downloads, net.unknown_format_downloads
            ));
        }
        if net.total_get_requests > 0 {
            let req_per_download = net.total_get_requests as f64 / net.downloads_ok.max(1) as f64;
            lines.push(format!(
                "| Read fan-out | {} reads total · {:.1} per download |",
                net.total_get_requests, req_per_download
            ));
        }
        if net.observed_span_ms > 0 {
            lines.push(format!(
                "| Observed wall-span throughput | {:.1} MB/s over {} · {} timed downloads · peak {} concurrent |",
                net.observed_throughput_mbps,
                format_duration_ms(net.observed_span_ms),
                net.observed_downloads,
                net.max_concurrent_downloads
            ));
        } else if net.downloads_ok > 0 {
            lines.push(
                "| Observed wall-span throughput | unavailable (legacy transfer events) |"
                    .to_string(),
            );
        }
        lines.push(format!(
            "| Cumulative service-time rates | {:.1} MB/s read · {:.1} MB/s open+read · {:.1} MB/s end-to-end |",
            net.body_throughput_mbps, net.network_throughput_mbps, net.throughput_mbps
        ));
        if !net.dominant_download_phase.is_empty() && net.dominant_download_phase_ms > 0 {
            lines.push(format!(
                "| Dominant cumulative download phase | {} — {} ({:.1}%) |",
                human_download_phase(&net.dominant_download_phase),
                format_duration_ms(net.dominant_download_phase_ms),
                net.dominant_download_phase_pct
            ));
        }
        if net.compression_ratio > 0.0 {
            lines.push(format!(
                "| Compression | {:.1}x ({} → {}) |",
                net.compression_ratio,
                format_bytes(net.original_bytes_down),
                format_bytes(net.bytes_down)
            ));
        }
        if net.total_semaphore_wait_ms > 0
            || net.total_head_ms > 0
            || net.total_decompress_ms > 0
            || net.total_extract_ms > 0
            || net.total_import_lock_wait_ms > 0
            || net.total_import_ms > 0
            || net.total_disk_io_ms > 0
        {
            lines.push(format!(
                "| Cumulative download phase time | queue wait {}ms · existence check {}ms · open/setup {}ms · read/transfer {}ms · decompress {}ms · extract {}ms · import lock wait {}ms · import execution {}ms · local disk {}ms |",
                net.total_semaphore_wait_ms,
                net.total_head_ms,
                net.total_request_ms,
                net.total_body_ms,
                net.total_decompress_ms,
                net.total_extract_ms,
                net.total_import_lock_wait_ms,
                net.total_import_ms,
                net.total_disk_io_ms
            ));
        }
        if net.blobs_total > 0 {
            let pct = if net.blobs_total > 0 {
                net.blobs_skipped as f64 / net.blobs_total as f64 * 100.0
            } else {
                0.0
            };
            lines.push(format!(
                "| Blob dedup | {}/{} already local ({:.0}% saved) |",
                net.blobs_skipped, net.blobs_total, pct
            ));
        }
        if net.downloads_failed > 0 {
            lines.push(format!("| Failed downloads | {} |", net.downloads_failed));
        }
        if net.uploads_failed > 0 {
            lines.push(format!("| Failed uploads | {} |", net.uploads_failed));
        }

        // Slowest downloads sub-table
        if !net.slowest_downloads.is_empty() {
            lines.push(String::new());
            lines.push("**Slowest downloads:**".to_string());
            lines.push(String::new());
            lines.push(
                "| Crate | Fmt | Size | Time | Reads | Key | Wait/Check | Open/Read | Extract/Lock/Import |"
                    .to_string(),
            );
            lines.push(
                "|-------|-----|------|------|------|-----|-----------|----------|----------------|"
                    .to_string(),
            );
            for d in net.slowest_downloads.iter().take(5) {
                let key = if d.cache_key.is_empty() {
                    "?"
                } else {
                    &d.cache_key[..d.cache_key.len().min(12)]
                };
                lines.push(format!(
                    "| `{}` | {} | {} | {}ms | {} | `{}` | {}/{}ms | {}/{}ms | {}/{}/{}ms |",
                    d.crate_name,
                    if d.format.is_empty() { "?" } else { &d.format },
                    format_bytes(d.compressed_bytes),
                    d.elapsed_ms,
                    d.request_count,
                    key,
                    d.semaphore_wait_ms,
                    d.head_ms,
                    d.request_ms,
                    d.body_ms,
                    d.extract_ms.max(d.decompress_ms),
                    d.import_lock_wait_ms,
                    d.import_ms,
                ));
            }
            let repro_keys: Vec<_> = net
                .slowest_downloads
                .iter()
                .filter(|d| !d.object_key.is_empty())
                .take(3)
                .collect();
            if !repro_keys.is_empty() {
                lines.push(String::new());
                lines.push("Raw object keys for reproduction:".to_string());
                for d in repro_keys {
                    lines.push(format!("- `{}`: `{}`", d.crate_name, d.object_key));
                }
            }
        }
        lines.push(String::new());
        lines.push("</details>".to_string());
    }

    // ── Storage (collapsed) ──
    if has_storage_data(&report.storage) {
        lines.push(String::new());
        lines.push("<details>".to_string());
        let storage_summary = if !report.storage.accounting_consistent {
            format!(
                "accounting inconsistent: {} logical, {} indexed blobs",
                format_bytes(report.storage.logical_bytes),
                format_bytes(report.storage.blob_bytes)
            )
        } else if report.storage.restored_bytes > 0 {
            format!(
                "{:.1}% zero-copy restores, {} restored",
                report.storage.zero_copy_pct,
                format_bytes(report.storage.restored_bytes)
            )
        } else {
            format!(
                "{} logical, {} blobs",
                format_bytes(report.storage.logical_bytes),
                format_bytes(report.storage.blob_bytes)
            )
        };
        lines.push(format!(
            "<summary><strong>Storage</strong> — {}</summary>",
            storage_summary
        ));
        lines.push(String::new());
        push_storage_table(&mut lines, &report.storage);
        lines.push(String::new());
        lines.push("</details>".to_string());
    }

    // ── Timing & Prefetch (collapsed) ──
    let total_ms = t.hit_time_ms + t.miss_time_ms;
    let p = &report.prefetch;
    if total_ms > 0 || p.total_hits > 0 {
        lines.push(String::new());
        lines.push("<details>".to_string());
        lines.push("<summary><strong>Timing & Prefetch</strong></summary>".to_string());
        lines.push(String::new());
        if total_ms > 0 {
            let hit_pct = t.hit_time_ms as f64 / total_ms as f64 * 100.0;
            let miss_pct = t.miss_time_ms as f64 / total_ms as f64 * 100.0;
            lines.push("| Phase | Aggregate time | % of tracked wrapper time |".to_string());
            lines.push("|-------|------|---|".to_string());
            lines.push(format!(
                "| Cache hits (wrapper overhead) | {} | {:.1}% |",
                format_duration_ms(t.hit_time_ms),
                hit_pct
            ));
            lines.push(format!(
                "| Compiles (wrapper total) | {} | {:.1}% |",
                format_duration_ms(t.miss_time_ms),
                miss_pct
            ));
        }
        // Per-crate timing breakdown
        if t.total_key_ms > 0 || t.total_lookup_ms > 0 || t.total_restore_ms > 0 {
            lines.push(format!(
                "| Hit overhead | avg {:.0}ms key + {:.0}ms lookup + {:.0}ms restore |",
                t.avg_key_ms, t.avg_lookup_ms, t.avg_restore_ms
            ));
        }
        if t.total_store_ms > 0 {
            lines.push(format!(
                "| Miss overhead | avg {:.0}ms key + {:.0}ms lookup + {:.0}ms store |",
                t.avg_key_ms, t.avg_lookup_ms, t.avg_store_ms
            ));
        }
        lines.push(format!(
            "| Startup | {} aggregate (avg {:.1}ms/crate) |",
            format_duration_ms(t.total_startup_ms),
            t.avg_startup_ms
        ));
        lines.push(format!(
            "| Dep-info pre-pass | {} runs, {} aggregate (avg {:.1}ms/run) |",
            t.dep_info_runs,
            format_duration_ms(t.total_dep_info_ms),
            t.avg_dep_info_ms
        ));
        lines.push(format!(
            "| Scheduler wait | {} aggregate (flight {}, permit {}) |",
            format_duration_ms(t.total_wait_ms),
            format_duration_ms(t.total_flight_wait_ms),
            format_duration_ms(t.total_permit_wait_ms)
        ));
        lines.push(format!(
            "| Unattributed | {} aggregate (avg {:.1}ms/crate) |",
            format_duration_ms(t.total_unattributed_ms),
            t.avg_unattributed_ms
        ));
        if p.total_hits > 0 {
            lines.push(String::new());
            lines.push(format!(
                "**Prefetch:** {}/{} hits ({:.1}%)",
                p.prefetch_hits, p.total_hits, p.contribution_pct
            ));
        }
        lines.push(String::new());
        lines.push("</details>".to_string());
    }

    // ── GC (collapsed, only if GC ran recently) ──
    if let Some(gc) = &report.gc {
        lines.push(String::new());
        lines.push("<details>".to_string());
        lines.push(format!(
            "<summary><strong>GC</strong> — {} entries evicted, {} removed from the store</summary>",
            gc.entries_evicted,
            format_bytes(gc.bytes_freed),
        ));
        lines.push(String::new());
        lines.push("| | |".to_string());
        lines.push("|---|---|".to_string());
        lines.push(format!("| Last run | {} |", gc.last_run));
        lines.push(format!("| Entries evicted | {} |", gc.entries_evicted));
        lines.push(format!(
            "| Store bytes removed | {} |",
            format_bytes(gc.bytes_freed)
        ));
        lines.push(format!(
            "| Disk bytes reclaimed | {} |",
            format_bytes(gc.disk_bytes_reclaimed)
        ));
        lines.push(format!(
            "| Shared bytes retained | {} |",
            format_bytes(gc.shared_bytes_retained)
        ));
        lines.push(format!("| Blobs removed | {} |", gc.blobs_removed));
        lines.push(String::new());
        lines.push("</details>".to_string());
    }

    lines.push(String::new());
    lines.push(format!(
        "*Posted by [kache-action](https://github.com/kunobi-ninja/kache-action) · kache v{} · last {}*",
        report.meta.kache_version,
        report.meta.window_label(),
    ));

    lines.join("\n")
}

pub fn format_text(report: &BuildReport) -> String {
    use crate::cli::format_duration_ms;

    let mut lines = Vec::new();
    let s = &report.summary;
    let t = &report.timing;
    let total_hits = s.local_hits + s.prefetch_hits + s.remote_hits;
    let total_compiled = s.dups + s.misses;

    lines.push(format!(
        "kache build report (last {})",
        report.meta.window_label()
    ));
    if let Some(session) = &report.meta.session {
        lines.push(format!("  {}", session.description()));
    }
    lines.push(format!(
        "  {:.1}% hit rate — {}/{} cacheable crates cached, {} compiled",
        s.hit_rate_pct, total_hits, s.total_crates, total_compiled,
    ));
    if s.dups > 0 {
        lines.push(format!(
            "  Dups: {} storage duplicates after compile",
            s.dups
        ));
    }
    if let Some(w) = s.weighted_hit_rate_pct {
        lines.push(format!("  {:.1}% by compile cost", w));
    }
    lines.push(format!(
        "  Compile work avoided: {} aggregate",
        format_duration_ms(s.time_saved_ms)
    ));
    if let Some(overhead) = cache_overhead_summary(report) {
        lines.push(format!("  Cache hit overhead: {overhead}"));
    }
    if let Some(roi) = cache_roi(report) {
        lines.push(format!(
            "  Cache ROI: {:.1}x compile work per cache-hit overhead",
            roi
        ));
    }
    if t.miss_compile_time_ms > 0 {
        lines.push(format!(
            "  Miss compile work: {} aggregate",
            format_duration_ms(t.miss_compile_time_ms)
        ));
    }
    if s.store_failures > 0 {
        lines.push(format!(
            "  Compiled but not cached: {} (will miss again next build)",
            s.store_failures
        ));
    }
    if s.errors > 0 {
        lines.push(format!("  Errors: {}", s.errors));
    }
    if s.passthroughs > 0 || s.skipped > 0 || s.probes > 0 {
        lines.push(format!(
            "  Passthroughs/skipped: {}",
            format_bypass_summary(&report.bypass)
        ));
    }
    lines.push(String::new());

    // Timing
    lines.push("Timing:".to_string());
    lines.push(format!(
        "  Hits overhead: {} aggregate (avg {:.0}ms/hit)",
        format_duration_ms(t.hit_time_ms),
        t.avg_hit_ms
    ));
    lines.push(format!(
        "  Compiles: {} (avg {:.0}ms/crate)",
        format_duration_ms(t.miss_time_ms),
        t.avg_miss_ms
    ));
    if t.total_key_ms > 0 || t.total_lookup_ms > 0 || t.total_restore_ms > 0 {
        lines.push(format!(
            "  Hit overhead: avg {:.0}ms key + {:.0}ms lookup + {:.0}ms restore",
            t.avg_key_ms, t.avg_lookup_ms, t.avg_restore_ms
        ));
    }
    if t.total_store_ms > 0 {
        lines.push(format!(
            "  Miss overhead: avg {:.0}ms key + {:.0}ms lookup + {:.0}ms store",
            t.avg_key_ms, t.avg_lookup_ms, t.avg_store_ms
        ));
    }
    if s.total_crates > 0 {
        lines.push(format!(
            "  Startup: {} aggregate (avg {:.1}ms/crate)",
            format_duration_ms(t.total_startup_ms),
            t.avg_startup_ms
        ));
        lines.push(format!(
            "  Dep-info pre-pass: {} runs, {} aggregate (avg {:.1}ms/run)",
            t.dep_info_runs,
            format_duration_ms(t.total_dep_info_ms),
            t.avg_dep_info_ms
        ));
        lines.push(format!(
            "  Scheduler wait: {} aggregate (flight {}, permit {})",
            format_duration_ms(t.total_wait_ms),
            format_duration_ms(t.total_flight_wait_ms),
            format_duration_ms(t.total_permit_wait_ms)
        ));
        lines.push(format!(
            "  Unattributed: {} aggregate (avg {:.1}ms/crate)",
            format_duration_ms(t.total_unattributed_ms),
            t.avg_unattributed_ms
        ));
    }
    lines.push(String::new());

    // Remote transfers
    if let Some(net) = &report.network {
        lines.push("Remote transfer:".to_string());
        lines.push(format!(
            "  Downloaded: {} ({} ok, {} failed)",
            format_bytes(net.bytes_down),
            net.downloads_ok,
            net.downloads_failed
        ));
        lines.push(format!(
            "  Uploaded: {} ({} ok, {} failed)",
            format_bytes(net.bytes_up),
            net.uploads_ok,
            net.uploads_failed
        ));
        lines.push(format!(
            "  Latency: avg {:.0}ms, p95 {}ms, max {}ms",
            net.avg_download_ms, net.p95_download_ms, net.max_download_ms
        ));
        if net.observed_span_ms > 0 {
            lines.push(format!(
                "  Observed wall-span throughput: {:.1} MB/s over {} ({} timed downloads, peak {} concurrent)",
                net.observed_throughput_mbps,
                format_duration_ms(net.observed_span_ms),
                net.observed_downloads,
                net.max_concurrent_downloads
            ));
        } else if net.downloads_ok > 0 {
            lines.push(
                "  Observed wall-span throughput: unavailable (legacy transfer events)".to_string(),
            );
        }
        lines.push(format!(
            "  Cumulative service-time rates: {:.1} MB/s read, {:.1} MB/s open+read, {:.1} MB/s end-to-end",
            net.body_throughput_mbps, net.network_throughput_mbps, net.throughput_mbps
        ));
        if !net.dominant_download_phase.is_empty() && net.dominant_download_phase_ms > 0 {
            lines.push(format!(
                "  Dominant cumulative phase: {} — {} ({:.1}%)",
                human_download_phase(&net.dominant_download_phase),
                format_duration_ms(net.dominant_download_phase_ms),
                net.dominant_download_phase_pct
            ));
        }
        if net.compression_ratio > 0.0 {
            lines.push(format!(
                "  Compression: {:.1}x ratio ({} → {})",
                net.compression_ratio,
                format_bytes(net.original_bytes_down),
                format_bytes(net.bytes_down)
            ));
        }
        if net.total_semaphore_wait_ms > 0
            || net.total_head_ms > 0
            || net.total_decompress_ms > 0
            || net.total_extract_ms > 0
            || net.total_import_lock_wait_ms > 0
            || net.total_import_ms > 0
            || net.total_disk_io_ms > 0
        {
            lines.push(format!(
                "  Cumulative phase time: queue wait {}ms, existence check {}ms, open/setup {}ms, read/transfer {}ms, decompress {}ms, extract {}ms, import lock wait {}ms, import execution {}ms, local disk I/O {}ms",
                net.total_semaphore_wait_ms,
                net.total_head_ms,
                net.total_request_ms,
                net.total_body_ms,
                net.total_decompress_ms,
                net.total_extract_ms,
                net.total_import_lock_wait_ms,
                net.total_import_ms,
                net.total_disk_io_ms
            ));
        }
        if net.blobs_total > 0 {
            lines.push(format!(
                "  Blob dedup: {}/{} already local ({:.0}% skipped)",
                net.blobs_skipped,
                net.blobs_total,
                net.blobs_skipped as f64 / net.blobs_total.max(1) as f64 * 100.0
            ));
        }
        lines.push(String::new());
    }

    if has_storage_data(&report.storage) {
        lines.push("Storage:".to_string());
        if report.storage.restored_bytes > 0 {
            lines.push(format!(
                "  Restored: {} ({:.1}% zero-copy, {} copied)",
                format_bytes(report.storage.restored_bytes),
                report.storage.zero_copy_pct,
                format_bytes(report.storage.copied_bytes)
            ));
        }
        match storage_accounting_state(&report.storage) {
            StorageAccountingState::Consistent => lines.push(format!(
                "  Store: {} logical -> {} blobs ({} dedup saved)",
                format_bytes(report.storage.logical_bytes),
                format_bytes(report.storage.blob_bytes),
                format_bytes(report.storage.dedup_saved_bytes)
            )),
            StorageAccountingState::Inconsistent => lines.push(format!(
                "  Store accounting inconsistent: {} logical entry bytes, {} indexed blob bytes; the store index needs repair",
                format_bytes(report.storage.logical_bytes),
                format_bytes(report.storage.blob_bytes)
            )),
            StorageAccountingState::Absent => {}
        }
        lines.push(String::new());
    }

    lines.push(format!(
        "Stored entries: {} unique keys",
        report.storage.store_entries
    ));

    // Prefetch
    lines.push(format!(
        "Prefetch: {} / {} hits ({:.1}%)",
        report.prefetch.prefetch_hits, report.prefetch.total_hits, report.prefetch.contribution_pct
    ));
    lines.push(String::new());

    if bypass_total(&report.bypass) > 0 {
        lines.push("Passthroughs/skips:".to_string());
        for reason in &report.bypass.reasons {
            lines.push(format!(
                "  {} via {}: {} ({} total, {} failed, max {})",
                reason.result,
                reason.route,
                reason.reason,
                reason.count,
                reason.failures,
                format_duration_ms(reason.max_elapsed_ms)
            ));
        }
        if !report.bypass.slowest.is_empty() {
            lines.push("  Slowest:".to_string());
            for detail in &report.bypass.slowest {
                lines.push(format!(
                    "    {} — {} via {}, {}, exit {}, {}",
                    detail.crate_name,
                    detail.result,
                    detail.route,
                    format_duration_ms(detail.elapsed_ms),
                    format_exit_code(detail.exit_code),
                    bypass_detail_reason(detail)
                ));
            }
        }
        lines.push(String::new());
    }

    // Top compiled cache-key misses
    if !report.top_misses.is_empty() {
        lines.push("Top compiled cache-key misses:".to_string());
        for c in &report.top_misses {
            // A row that failed to store is not a cold miss that warms up on the
            // next build; say so where the user is already looking (#629).
            let not_cached = if c.store_error.is_empty() {
                String::new()
            } else {
                format!("  [not cached: {}]", c.store_error)
            };
            lines.push(format!(
                "  {} — {} ({}){}",
                c.crate_name,
                format_duration_ms(c.compile_time_ms),
                format_bytes(c.size),
                not_cached,
            ));
        }
        lines.push(String::new());
    }

    if !report.top_hits.is_empty() {
        lines.push("Expensive cache hits:".to_string());
        for c in &report.top_hits {
            lines.push(format!(
                "  {} — {} avoided ({})",
                c.crate_name,
                format_duration_ms(c.compile_time_ms),
                format_bytes(c.size),
            ));
        }
        lines.push(String::new());
    }

    if !report.errors_detail.is_empty() {
        lines.push("Errors:".to_string());
        for err in report.errors_detail.iter().take(10) {
            lines.push(format!("  {} — {}", err.crate_name, err.timestamp));
        }
        lines.push(String::new());
    }

    // Suggestions
    if !report.suggestions.is_empty() {
        lines.push("Suggestions:".to_string());
        for s in &report.suggestions {
            lines.push(format!("  - {s}"));
        }
        lines.push(String::new());
    }

    // GC
    if let Some(gc) = &report.gc {
        lines.push("GC:".to_string());
        lines.push(format!("  Last run: {}", gc.last_run));
        lines.push(format!("  Entries evicted: {}", gc.entries_evicted));
        lines.push(format!(
            "  Store bytes removed: {}",
            format_bytes(gc.bytes_freed)
        ));
        lines.push(format!(
            "  Disk bytes reclaimed: {}",
            format_bytes(gc.disk_bytes_reclaimed)
        ));
        lines.push(format!(
            "  Shared bytes retained: {}",
            format_bytes(gc.shared_bytes_retained)
        ));
        lines.push(format!("  Blobs removed: {}", gc.blobs_removed));
        lines.push(String::new());
    }

    lines.join("\n")
}

pub fn format_bytes(bytes: u64) -> String {
    let b = bytes as f64;
    if b >= 1024.0 * 1024.0 * 1024.0 {
        format!("{:.1} GB", b / (1024.0 * 1024.0 * 1024.0))
    } else if b >= 1024.0 * 1024.0 {
        format!("{:.1} MB", b / (1024.0 * 1024.0))
    } else if b >= 1024.0 {
        format!("{:.1} KB", b / 1024.0)
    } else {
        format!("{bytes} B")
    }
}

// ── Tests ───────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests;