rpki/src/tools/rpki_artifact_metrics.rs

use std::collections::{BTreeMap, BTreeSet};
use std::fs;
use std::io::{Read, Write};
use std::net::{TcpListener, TcpStream};
use std::path::{Path, PathBuf};
use std::sync::{Arc, RwLock};
use std::thread;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use crate::ccr::decode_content_info;
use crate::cir::decode_cir;
use serde::Serialize;
use serde_json::{Value, json};
use sha2::{Digest, Sha256};

const LARGE_PP_OBJECT_THRESHOLDS: &[u64] = &[10, 50, 100, 500, 1000, 5000, 10000, 50000];
const PP_SYNC_SECONDS_BUCKETS: &[f64] = &[0.1, 0.5, 1.0, 5.0, 10.0, 30.0, 60.0, 120.0, 300.0];
const RTR_COUNT_TYPES: &[&str] = &["total", "vrp", "router_key", "aspa"];

#[derive(Clone, Debug, PartialEq, Eq)]
struct Args {
    run_root: PathBuf,
    listen: String,
    poll_secs: u64,
    instance: String,
    once: bool,
    out_metrics: Option<PathBuf>,
    out_status: Option<PathBuf>,
}

fn usage() -> &'static str {
    "Usage: rpki_artifact_metrics --run-root <path> [--listen <addr:port>] [--poll-secs <n>] [--instance <name>] [--once] [--out-metrics <path>] [--out-status <path>]"
}

pub fn main_entry() -> Result<(), String> {
    real_main()
}

fn real_main() -> Result<(), String> {
    let args = parse_args(&std::env::args().collect::<Vec<_>>())?;
    if args.once {
        let snapshot = scan_run_root(&args.run_root, &args.instance)?;
        let metrics = render_metrics(&snapshot);
        let status = render_status_json(&snapshot)?;
        if let Some(path) = args.out_metrics.as_ref() {
            write_file(path, metrics.as_bytes())?;
        } else {
            print!("{metrics}");
        }
        if let Some(path) = args.out_status.as_ref() {
            write_file(path, status.as_bytes())?;
        }
        return Ok(());
    }

    let shared = Arc::new(RwLock::new(scan_run_root(&args.run_root, &args.instance)?));
    let scanner = Arc::clone(&shared);
    let run_root = args.run_root.clone();
    let instance = args.instance.clone();
    let poll_secs = args.poll_secs.max(1);
    thread::spawn(move || {
        loop {
            thread::sleep(Duration::from_secs(poll_secs));
            let previous_snapshot = scanner.read().expect("metrics lock poisoned").clone();
            let next =
                match scan_run_root_incremental(&run_root, &instance, Some(&previous_snapshot)) {
                    Ok(snapshot) => snapshot,
                    Err(err) => {
                        let mut previous = scanner.write().expect("metrics lock poisoned");
                        previous
                            .service
                            .parse_errors
                            .push(format!("scan failed: {err}"));
                        previous.service.last_reload_success = false;
                        previous.service.last_scan_timestamp_seconds = unix_now_seconds();
                        continue;
                    }
                };
            *scanner.write().expect("metrics lock poisoned") = next;
        }
    });

    serve_http(&args.listen, shared)
}

fn parse_args(argv: &[String]) -> Result<Args, String> {
    let mut run_root = None;
    let mut listen = "127.0.0.1:9556".to_string();
    let mut poll_secs = 10u64;
    let mut instance = "ours-rp".to_string();
    let mut once = false;
    let mut out_metrics = None;
    let mut out_status = None;
    let mut index = 1usize;
    while index < argv.len() {
        match argv[index].as_str() {
            "--run-root" => {
                index += 1;
                run_root = Some(PathBuf::from(value_at(argv, index, "--run-root")?));
            }
            "--listen" => {
                index += 1;
                listen = value_at(argv, index, "--listen")?.to_string();
            }
            "--poll-secs" => {
                index += 1;
                let value = value_at(argv, index, "--poll-secs")?;
                poll_secs = value
                    .parse::<u64>()
                    .map_err(|_| format!("invalid --poll-secs: {value}"))?;
            }
            "--instance" => {
                index += 1;
                instance = value_at(argv, index, "--instance")?.to_string();
            }
            "--once" => once = true,
            "--out-metrics" => {
                index += 1;
                out_metrics = Some(PathBuf::from(value_at(argv, index, "--out-metrics")?));
            }
            "--out-status" => {
                index += 1;
                out_status = Some(PathBuf::from(value_at(argv, index, "--out-status")?));
            }
            "-h" | "--help" => return Err(usage().to_string()),
            other => return Err(format!("unknown argument: {other}\n{}", usage())),
        }
        index += 1;
    }
    Ok(Args {
        run_root: run_root.ok_or_else(|| format!("--run-root is required\n{}", usage()))?,
        listen,
        poll_secs,
        instance,
        once,
        out_metrics,
        out_status,
    })
}

fn value_at<'a>(argv: &'a [String], index: usize, flag: &str) -> Result<&'a str, String> {
    argv.get(index)
        .map(|s| s.as_str())
        .ok_or_else(|| format!("{flag} requires a value"))
}

#[derive(Clone, Debug, Default, Serialize)]
#[serde(rename_all = "camelCase")]
struct MetricsSnapshot {
    instance: String,
    service: ServiceMetrics,
    runs: RunScanSummary,
    latest_run: Option<LatestRunMetrics>,
    cumulative: CumulativeMetrics,
    repo_stats: Vec<RepoMetrics>,
    object_counts: BTreeMap<(String, String), u64>,
    large_pp_counts: BTreeMap<u64, u64>,
    pp_sync_histograms: BTreeMap<String, Histogram>,
    top_repos_by_sync_duration: Vec<TopRepo>,
    top_pp_by_object_count: Vec<TopPublicationPoint>,
    top_pp_by_sync_duration: Vec<TopPublicationPoint>,
    cir: Option<CirMetrics>,
    ccr: Option<CcrMetrics>,
    rtr: RtrMetrics,
}

#[derive(Clone, Debug, Default, Serialize)]
#[serde(rename_all = "camelCase")]
struct ServiceMetrics {
    last_scan_timestamp_seconds: f64,
    last_scan_duration_seconds: f64,
    last_reload_success: bool,
    parse_errors: Vec<String>,
    run_root: String,
    runs_root: String,
}

#[derive(Clone, Debug, Default, Serialize)]
#[serde(rename_all = "camelCase")]
struct RtrMetrics {
    enabled: bool,
    report_dir: Option<String>,
    parse_errors: Vec<String>,
    parse_errors_total: u64,
    files: BTreeMap<String, RtrFileScan>,
    source: Option<RtrSourceMetrics>,
    runtime: Option<RtrRuntimeMetrics>,
    clients: Option<RtrClientsMetrics>,
}

impl RtrMetrics {
    fn preferred_service(&self) -> Option<&RtrServiceMetrics> {
        self.runtime
            .as_ref()
            .and_then(|runtime| runtime.service.as_ref())
            .or_else(|| {
                self.clients
                    .as_ref()
                    .and_then(|clients| clients.service.as_ref())
            })
    }
}

#[derive(Clone, Debug, Default, Serialize)]
#[serde(rename_all = "camelCase")]
struct RtrFileScan {
    seen: u64,
    latest_path: Option<String>,
    latest_generated_at: Option<String>,
    latest_timestamp_seconds: Option<f64>,
}

#[derive(Clone, Debug, Default, Serialize)]
#[serde(rename_all = "camelCase")]
struct RtrSourceMetrics {
    file_path: String,
    generated_at: Option<String>,
    generated_at_seconds: Option<f64>,
    phase: Option<String>,
    refresh_status: Option<String>,
    refresh_changed: Option<bool>,
    refresh_duration_seconds: Option<f64>,
    refresh_consecutive_failures: Option<u64>,
    refresh_last_success_at: Option<String>,
    refresh_last_success_seconds: Option<f64>,
    ccr_file_size_bytes: Option<u64>,
    data_quality: BTreeMap<String, BTreeMap<String, u64>>,
    slurm_filters: BTreeMap<String, u64>,
    slurm_assertions: BTreeMap<String, u64>,
    cache_availability: Option<String>,
    cache_versions: Vec<RtrCacheVersion>,
}

#[derive(Clone, Debug, Default, Serialize)]
#[serde(rename_all = "camelCase")]
struct RtrCacheVersion {
    version: u64,
    serial: Option<u64>,
    snapshot: BTreeMap<String, u64>,
    latest_delta_announced: BTreeMap<String, u64>,
    latest_delta_withdrawn: BTreeMap<String, u64>,
    delta_window_length: Option<u64>,
}

#[derive(Clone, Debug, Default, Serialize)]
#[serde(rename_all = "camelCase")]
struct RtrRuntimeMetrics {
    file_path: String,
    generated_at: Option<String>,
    generated_at_seconds: Option<f64>,
    phase: Option<String>,
    process_rss_bytes: Option<u64>,
    service: Option<RtrServiceMetrics>,
}

#[derive(Clone, Debug, Default, Serialize)]
#[serde(rename_all = "camelCase")]
struct RtrClientsMetrics {
    file_path: String,
    generated_at: Option<String>,
    generated_at_seconds: Option<f64>,
    phase: Option<String>,
    service: Option<RtrServiceMetrics>,
}

#[derive(Clone, Debug, Default, Serialize)]
#[serde(rename_all = "camelCase")]
struct RtrServiceMetrics {
    started_at: Option<String>,
    uptime_seconds: Option<u64>,
    active_connections: Option<u64>,
    connections_by_transport: BTreeMap<String, u64>,
    session_listeners: Option<u64>,
    max_connections: Option<u64>,
    connection_utilization: Option<f64>,
}

#[derive(Clone, Debug, Default, Serialize)]
#[serde(rename_all = "camelCase")]
struct RunScanSummary {
    known: u64,
    success: u64,
    failed: u64,
    partial: u64,
    consecutive_failures: u64,
}

#[derive(Clone, Debug, Default, Serialize)]
#[serde(rename_all = "camelCase")]
struct CumulativeMetrics {
    completed_success_total: u64,
    completed_failed_total: u64,
    observed_duration_seconds_sum: f64,
    observed_duration_seconds_count: u64,
    observed_download_bytes_total: u64,
}

#[derive(Clone, Debug, Default, Serialize)]
#[serde(rename_all = "camelCase")]
struct LatestRunMetrics {
    run_seq: u64,
    run_id: String,
    run_dir: String,
    status: String,
    sync_mode: String,
    snapshot_reason: Option<String>,
    started_at: Option<String>,
    finished_at: Option<String>,
    start_timestamp_seconds: Option<f64>,
    finish_timestamp_seconds: Option<f64>,
    wall_seconds: f64,
    user_cpu_seconds: Option<f64>,
    system_cpu_seconds: Option<f64>,
    cpu_percent: Option<f64>,
    max_rss_bytes: Option<u64>,
    exit_code: Option<i64>,
    vrps: u64,
    vrps_unique: Option<u64>,
    vaps: u64,
    publication_points: u64,
    warnings: u64,
    tree_instances_processed: Option<u64>,
    tree_instances_failed: Option<u64>,
    stage_seconds: BTreeMap<String, f64>,
    repo_sync_phase: BTreeMap<String, CountDuration>,
    repo_terminal_state: BTreeMap<String, CountDuration>,
    download_events: Option<u64>,
    download_bytes: Option<u64>,
    artifact_sizes: BTreeMap<String, u64>,
    state_path_sizes: BTreeMap<String, PathSize>,
}

#[derive(Clone, Debug, Default, Serialize)]
#[serde(rename_all = "camelCase")]
struct CountDuration {
    count: u64,
    duration_seconds_total: f64,
}

#[derive(Clone, Debug, Default, Serialize)]
#[serde(rename_all = "camelCase")]
struct PathSize {
    total_size_bytes: u64,
    file_count: u64,
    dir_count: u64,
}

#[derive(Clone, Debug, Default, Serialize)]
#[serde(rename_all = "camelCase")]
struct RepoMetrics {
    repo_id: String,
    uri: String,
    host: String,
    transport: String,
    publication_points: u64,
    sync_success: bool,
    download_bytes: u64,
    duration_seconds_sum: f64,
    duration_seconds_max: f64,
    duration_seconds_avg: f64,
    phase_counts: BTreeMap<String, u64>,
    terminal_state_counts: BTreeMap<String, u64>,
    object_counts: BTreeMap<String, u64>,
}

#[derive(Clone, Debug, Default, Serialize)]
#[serde(rename_all = "camelCase")]
struct TopRepo {
    rank: usize,
    repo_id: String,
    uri: String,
    host: String,
    transport: String,
    duration_ms_max: u64,
    duration_ms_sum: u64,
    publication_points: u64,
}

#[derive(Clone, Debug, Default, Serialize)]
#[serde(rename_all = "camelCase")]
struct TopPublicationPoint {
    rank: usize,
    pp_id: String,
    repo_id: String,
    uri: String,
    repo_uri: String,
    host: String,
    transport: String,
    object_count: u64,
    sync_duration_ms: u64,
    terminal_state: String,
    phase: String,
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct Histogram {
    buckets: Vec<f64>,
    counts: Vec<u64>,
    sum: f64,
    count: u64,
}

impl Default for Histogram {
    fn default() -> Self {
        Self {
            buckets: Vec::new(),
            counts: Vec::new(),
            sum: 0.0,
            count: 0,
        }
    }
}

impl Histogram {
    fn new(buckets: &[f64]) -> Self {
        Self {
            buckets: buckets.to_vec(),
            counts: vec![0; buckets.len() + 1],
            sum: 0.0,
            count: 0,
        }
    }

    fn observe(&mut self, value: f64) {
        self.sum += value;
        self.count += 1;
        let mut placed = false;
        for (index, bucket) in self.buckets.iter().enumerate() {
            if value <= *bucket {
                self.counts[index] += 1;
                placed = true;
                break;
            }
        }
        if !placed {
            let last = self.counts.len() - 1;
            self.counts[last] += 1;
        }
    }

    fn cumulative_counts(&self) -> Vec<u64> {
        let mut out = Vec::with_capacity(self.counts.len());
        let mut running = 0u64;
        for count in &self.counts {
            running += *count;
            out.push(running);
        }
        out
    }
}

#[derive(Clone, Debug, Default, Serialize)]
#[serde(rename_all = "camelCase")]
struct CirMetrics {
    version: u32,
    objects: u64,
    fresh_objects: u64,
    cached_objects: u64,
    trust_anchors: u64,
    rejected_objects: u64,
    fresh_rejected_objects: u64,
    cached_rejected_objects: u64,
    object_list_sha256: String,
    fresh_object_list_sha256: String,
    cached_object_list_sha256: String,
    reject_list_sha256: String,
    fresh_reject_list_sha256: String,
    cached_reject_list_sha256: String,
    objects_by_type: BTreeMap<String, u64>,
    fresh_objects_by_type: BTreeMap<String, u64>,
    cached_objects_by_type: BTreeMap<String, u64>,
    rejected_objects_by_type: BTreeMap<String, u64>,
    fresh_rejected_objects_by_type: BTreeMap<String, u64>,
    cached_rejected_objects_by_type: BTreeMap<String, u64>,
}

#[derive(Clone, Debug, Default, Serialize)]
#[serde(rename_all = "camelCase")]
struct CcrMetrics {
    version: u32,
    state_present: BTreeMap<String, bool>,
    state_items: BTreeMap<String, u64>,
    state_digests: BTreeMap<String, String>,
}

#[derive(Clone, Debug)]
struct RunRecord {
    path: PathBuf,
    status: String,
    summary: Option<Value>,
    meta: Option<Value>,
}

fn scan_run_root(input_root: &Path, instance: &str) -> Result<MetricsSnapshot, String> {
    let rtr_report_dir = configured_rtr_report_dir();
    scan_run_root_incremental_with_rtr_dir(input_root, instance, None, rtr_report_dir.as_deref())
}

fn scan_run_root_incremental(
    input_root: &Path,
    instance: &str,
    previous: Option<&MetricsSnapshot>,
) -> Result<MetricsSnapshot, String> {
    let rtr_report_dir = configured_rtr_report_dir();
    scan_run_root_incremental_with_rtr_dir(
        input_root,
        instance,
        previous,
        rtr_report_dir.as_deref(),
    )
}

fn scan_run_root_incremental_with_rtr_dir(
    input_root: &Path,
    instance: &str,
    previous: Option<&MetricsSnapshot>,
    rtr_report_dir: Option<&Path>,
) -> Result<MetricsSnapshot, String> {
    let started = Instant::now();
    let runs_root = resolve_runs_root(input_root);
    let mut snapshot = MetricsSnapshot {
        instance: instance.to_string(),
        service: ServiceMetrics {
            run_root: input_root.display().to_string(),
            runs_root: runs_root.display().to_string(),
            ..ServiceMetrics::default()
        },
        ..MetricsSnapshot::default()
    };

    let records = collect_run_records(&runs_root, &mut snapshot.service.parse_errors)?;
    snapshot.runs.known = records.len() as u64;
    for record in &records {
        match record.status.as_str() {
            "success" => snapshot.runs.success += 1,
            "failed" | "spawn_failed" => snapshot.runs.failed += 1,
            _ => snapshot.runs.partial += 1,
        }
    }
    snapshot.runs.consecutive_failures = consecutive_failures(&records);
    snapshot.cumulative.completed_success_total = snapshot.runs.success;
    snapshot.cumulative.completed_failed_total = snapshot.runs.failed;

    for record in records.iter().filter(|record| record.status == "success") {
        if let Some(summary) = record.summary.as_ref() {
            let wall_seconds = json_u64(summary, &["wallMs"]).unwrap_or(0) as f64 / 1000.0;
            snapshot.cumulative.observed_duration_seconds_sum += wall_seconds;
            snapshot.cumulative.observed_duration_seconds_count += 1;
            if let Some(bytes) = json_u64(summary, &["stageTiming", "download_bytes_total"]) {
                snapshot.cumulative.observed_download_bytes_total = snapshot
                    .cumulative
                    .observed_download_bytes_total
                    .saturating_add(bytes);
            }
        }
    }

    if let Some(latest) = records
        .iter()
        .rev()
        .find(|record| record.status == "success")
    {
        if can_reuse_latest_metrics(latest, previous) {
            reuse_latest_artifact_metrics(previous.expect("checked previous"), &mut snapshot);
        } else {
            build_latest_metrics(latest, &mut snapshot);
            let _ = crate::memory_telemetry::malloc_trim_probe();
        }
    }

    snapshot.rtr = scan_rtr_reports(rtr_report_dir, previous.map(|snapshot| &snapshot.rtr));
    snapshot.service.last_scan_timestamp_seconds = unix_now_seconds();
    snapshot.service.last_scan_duration_seconds = started.elapsed().as_secs_f64();
    snapshot.service.last_reload_success =
        snapshot.service.parse_errors.is_empty() && snapshot.rtr.parse_errors.is_empty();
    Ok(snapshot)
}

fn can_reuse_latest_metrics(record: &RunRecord, previous: Option<&MetricsSnapshot>) -> bool {
    let Some(previous) = previous else {
        return false;
    };
    let Some(previous_latest) = previous.latest_run.as_ref() else {
        return false;
    };
    if previous_latest.run_dir != record.path.display().to_string() {
        return false;
    }
    if previous_latest.status != record.status {
        return false;
    }
    let summary = record.summary.as_ref();
    let meta = record.meta.as_ref();
    let finished_at = summary
        .and_then(|v| json_str(v, &["finishedAtRfc3339Utc"]))
        .or_else(|| meta.and_then(|v| json_str(v, &["completed_at_rfc3339_utc"])));
    let wall_seconds = summary.and_then(|v| json_u64(v, &["wallMs"])).unwrap_or(0) as f64 / 1000.0;
    previous_latest.finished_at.as_deref() == finished_at
        && (previous_latest.wall_seconds - wall_seconds).abs() < f64::EPSILON
}

fn reuse_latest_artifact_metrics(previous: &MetricsSnapshot, snapshot: &mut MetricsSnapshot) {
    snapshot.latest_run = previous.latest_run.clone();
    snapshot.repo_stats = previous.repo_stats.clone();
    snapshot.object_counts = previous.object_counts.clone();
    snapshot.large_pp_counts = previous.large_pp_counts.clone();
    snapshot.pp_sync_histograms = previous.pp_sync_histograms.clone();
    snapshot.top_repos_by_sync_duration = previous.top_repos_by_sync_duration.clone();
    snapshot.top_pp_by_object_count = previous.top_pp_by_object_count.clone();
    snapshot.top_pp_by_sync_duration = previous.top_pp_by_sync_duration.clone();
    snapshot.cir = previous.cir.clone();
    snapshot.ccr = previous.ccr.clone();
}

fn resolve_runs_root(input_root: &Path) -> PathBuf {
    let runs = input_root.join("runs");
    if runs.is_dir() {
        runs
    } else {
        input_root.to_path_buf()
    }
}

fn configured_rtr_report_dir() -> Option<PathBuf> {
    std::env::var("RPKI_METRICS_RTR_REPORT_DIR")
        .ok()
        .map(|value| value.trim().to_string())
        .filter(|value| !value.is_empty())
        .map(PathBuf::from)
}

fn scan_rtr_reports(report_dir: Option<&Path>, previous: Option<&RtrMetrics>) -> RtrMetrics {
    let mut metrics = RtrMetrics {
        enabled: report_dir.is_some(),
        report_dir: report_dir.map(|path| path.display().to_string()),
        parse_errors_total: previous.map(|prev| prev.parse_errors_total).unwrap_or(0),
        ..RtrMetrics::default()
    };

    let Some(report_dir) = report_dir else {
        return metrics;
    };

    if !report_dir.is_dir() {
        metrics.parse_errors.push(format!(
            "rtr report dir is not a directory: {}",
            report_dir.display()
        ));
        metrics.parse_errors_total = metrics
            .parse_errors_total
            .saturating_add(metrics.parse_errors.len() as u64);
        return metrics;
    }

    metrics.files.insert(
        "source".to_string(),
        scan_rtr_kind(
            report_dir,
            &["rtr-source-*.json"],
            &mut metrics.parse_errors,
        ),
    );
    metrics.files.insert(
        "runtime".to_string(),
        scan_rtr_kind(
            report_dir,
            &["rtr-runtime-*.json"],
            &mut metrics.parse_errors,
        ),
    );
    metrics.files.insert(
        "clients".to_string(),
        scan_rtr_kind(
            report_dir,
            &["rtr-clients-*.json", "rtr-client-*.json"],
            &mut metrics.parse_errors,
        ),
    );

    if let Some(file) = metrics
        .files
        .get("source")
        .and_then(|scan| scan.latest_path.as_deref())
    {
        if let Some(value) = read_rtr_json(&PathBuf::from(file), &mut metrics.parse_errors) {
            metrics.source = Some(parse_rtr_source_metrics(&PathBuf::from(file), &value));
            if let Some(source) = metrics.source.as_ref() {
                if let Some(scan) = metrics.files.get_mut("source") {
                    scan.latest_generated_at = source.generated_at.clone();
                    scan.latest_timestamp_seconds = source.generated_at_seconds;
                }
            }
        }
    }

    if let Some(file) = metrics
        .files
        .get("runtime")
        .and_then(|scan| scan.latest_path.as_deref())
    {
        if let Some(value) = read_rtr_json(&PathBuf::from(file), &mut metrics.parse_errors) {
            metrics.runtime = Some(parse_rtr_runtime_metrics(&PathBuf::from(file), &value));
            if let Some(runtime) = metrics.runtime.as_ref() {
                if let Some(scan) = metrics.files.get_mut("runtime") {
                    scan.latest_generated_at = runtime.generated_at.clone();
                    scan.latest_timestamp_seconds = runtime.generated_at_seconds;
                }
            }
        }
    }

    if let Some(file) = metrics
        .files
        .get("clients")
        .and_then(|scan| scan.latest_path.as_deref())
    {
        if let Some(value) = read_rtr_json(&PathBuf::from(file), &mut metrics.parse_errors) {
            metrics.clients = Some(parse_rtr_clients_metrics(&PathBuf::from(file), &value));
            if let Some(clients) = metrics.clients.as_ref() {
                if let Some(scan) = metrics.files.get_mut("clients") {
                    scan.latest_generated_at = clients.generated_at.clone();
                    scan.latest_timestamp_seconds = clients.generated_at_seconds;
                }
            }
        }
    }

    metrics.parse_errors_total = metrics
        .parse_errors_total
        .saturating_add(metrics.parse_errors.len() as u64);
    metrics
}

fn scan_rtr_kind(report_dir: &Path, patterns: &[&str], errors: &mut Vec<String>) -> RtrFileScan {
    let mut names = BTreeSet::new();
    for pattern in patterns {
        match glob_rtr_pattern(report_dir, pattern) {
            Ok(found) => {
                for path in found {
                    names.insert(path);
                }
            }
            Err(err) => errors.push(err),
        }
    }
    let latest_path = names.iter().next_back().cloned();
    RtrFileScan {
        seen: names.len() as u64,
        latest_path: latest_path.map(|path| path.display().to_string()),
        latest_generated_at: None,
        latest_timestamp_seconds: None,
    }
}

fn glob_rtr_pattern(report_dir: &Path, pattern: &str) -> Result<Vec<PathBuf>, String> {
    let Some((prefix, suffix)) = pattern.split_once('*') else {
        return Err(format!("unsupported RTR report pattern: {pattern}"));
    };
    let mut matches = Vec::new();
    let entries = fs::read_dir(report_dir)
        .map_err(|e| format!("read rtr report dir failed: {}: {e}", report_dir.display()))?;
    for entry in entries {
        let entry = entry.map_err(|e| format!("read rtr report entry failed: {e}"))?;
        let path = entry.path();
        if !path.is_file() {
            continue;
        }
        let Some(name) = path.file_name().and_then(|name| name.to_str()) else {
            continue;
        };
        if name.starts_with(prefix) && name.ends_with(suffix) {
            matches.push(path);
        }
    }
    matches.sort();
    Ok(matches)
}

fn read_rtr_json(path: &Path, errors: &mut Vec<String>) -> Option<Value> {
    let Ok(bytes) = fs::read(path) else {
        errors.push(format!("read rtr report failed: {}", path.display()));
        return None;
    };
    match serde_json::from_slice::<Value>(&bytes) {
        Ok(value) => Some(value),
        Err(err) => {
            errors.push(format!(
                "parse rtr report failed: {}: {err}",
                path.display()
            ));
            None
        }
    }
}

fn parse_rtr_source_metrics(path: &Path, value: &Value) -> RtrSourceMetrics {
    let generated_at = json_str(value, &["generated_at"]).map(|value| value.to_string());
    let generated_at_seconds = generated_at.as_deref().and_then(parse_rfc3339_to_unix);
    let mut data_quality = BTreeMap::new();
    for stage in ["ccr_input", "invalid", "before_slurm", "after_slurm"] {
        if let Some(stage_metrics) =
            json_counts_map(value.pointer(&format!("/data_quality/{stage}")))
        {
            data_quality.insert(stage.to_string(), stage_metrics);
        }
    }

    let cache_versions = value
        .pointer("/cache/versions")
        .and_then(|value| value.as_array())
        .map(|versions| {
            versions
                .iter()
                .map(|version| RtrCacheVersion {
                    version: json_u64(version, &["version"]).unwrap_or(0),
                    serial: json_u64(version, &["serial"]),
                    snapshot: json_counts_map(version.get("snapshot")).unwrap_or_default(),
                    latest_delta_announced: json_counts_map(
                        version.pointer("/latest_delta/announced"),
                    )
                    .unwrap_or_default(),
                    latest_delta_withdrawn: json_counts_map(
                        version.pointer("/latest_delta/withdrawn"),
                    )
                    .unwrap_or_default(),
                    delta_window_length: json_u64(version, &["delta_window", "length"]),
                })
                .collect::<Vec<_>>()
        })
        .unwrap_or_default();

    RtrSourceMetrics {
        file_path: path.display().to_string(),
        generated_at,
        generated_at_seconds,
        phase: json_str(value, &["phase"]).map(|value| value.to_string()),
        refresh_status: json_str(value, &["refresh", "status"]).map(|value| value.to_string()),
        refresh_changed: json_bool(value, &["refresh", "changed"]),
        refresh_duration_seconds: json_u64(value, &["refresh", "duration_ms"])
            .map(|value| value as f64 / 1000.0),
        refresh_consecutive_failures: json_u64(value, &["refresh", "consecutive_failures"]),
        refresh_last_success_at: json_str(value, &["refresh", "last_success_at"])
            .map(|value| value.to_string()),
        refresh_last_success_seconds: json_str(value, &["refresh", "last_success_at"])
            .and_then(parse_rfc3339_to_unix),
        ccr_file_size_bytes: json_u64(value, &["source", "ccr_file_size_bytes"]),
        data_quality,
        slurm_filters: json_counts_map(value.pointer("/data_quality/slurm_filters"))
            .unwrap_or_default(),
        slurm_assertions: json_counts_map(value.pointer("/data_quality/slurm_assertions"))
            .unwrap_or_default(),
        cache_availability: json_str(value, &["cache", "availability"])
            .map(|value| value.to_string()),
        cache_versions,
    }
}

fn parse_rtr_runtime_metrics(path: &Path, value: &Value) -> RtrRuntimeMetrics {
    let generated_at = json_str(value, &["generated_at"]).map(|value| value.to_string());
    let generated_at_seconds = generated_at.as_deref().and_then(parse_rfc3339_to_unix);
    RtrRuntimeMetrics {
        file_path: path.display().to_string(),
        generated_at,
        generated_at_seconds,
        phase: json_str(value, &["phase"]).map(|value| value.to_string()),
        process_rss_bytes: json_u64(value, &["process", "rss_mib"])
            .map(|value| value.saturating_mul(1024 * 1024)),
        service: parse_rtr_service_metrics(value),
    }
}

fn parse_rtr_clients_metrics(path: &Path, value: &Value) -> RtrClientsMetrics {
    let generated_at = json_str(value, &["generated_at"]).map(|value| value.to_string());
    let generated_at_seconds = generated_at.as_deref().and_then(parse_rfc3339_to_unix);
    RtrClientsMetrics {
        file_path: path.display().to_string(),
        generated_at,
        generated_at_seconds,
        phase: json_str(value, &["phase"]).map(|value| value.to_string()),
        service: parse_rtr_service_metrics(value),
    }
}

fn parse_rtr_service_metrics(value: &Value) -> Option<RtrServiceMetrics> {
    let service = value.get("service")?;
    Some(RtrServiceMetrics {
        started_at: json_str(service, &["started_at"]).map(|value| value.to_string()),
        uptime_seconds: json_u64(service, &["uptime_seconds"]),
        active_connections: json_u64(service, &["active_connections"]),
        connections_by_transport: json_counts_map(service.get("connections_by_transport"))
            .unwrap_or_default(),
        session_listeners: json_u64(service, &["session_listeners"]),
        max_connections: json_u64(service, &["max_connections"]),
        connection_utilization: json_f64(service, &["connection_utilization"]),
    })
}

fn json_counts_map(value: Option<&Value>) -> Option<BTreeMap<String, u64>> {
    let mut out = BTreeMap::new();
    let object = value?.as_object()?;
    for (key, value) in object {
        if let Some(count) = value.as_u64() {
            out.insert(key.clone(), count);
        }
    }
    Some(out)
}

fn collect_run_records(
    runs_root: &Path,
    errors: &mut Vec<String>,
) -> Result<Vec<RunRecord>, String> {
    let mut records = Vec::new();
    if !runs_root.is_dir() {
        return Err(format!(
            "runs root is not a directory: {}",
            runs_root.display()
        ));
    }
    let entries = fs::read_dir(runs_root)
        .map_err(|e| format!("read runs root failed: {}: {e}", runs_root.display()))?;
    for entry in entries {
        let entry = entry.map_err(|e| format!("read runs entry failed: {e}"))?;
        let path = entry.path();
        if !path.is_dir() {
            continue;
        }
        let Some(name) = path.file_name().and_then(|name| name.to_str()) else {
            continue;
        };
        if !name.starts_with("run_") {
            continue;
        }
        let summary = read_json_optional(&path.join("run-summary.json"), errors);
        let meta = read_json_optional(&path.join("run-meta.json"), errors);
        let status = classify_run_status(&summary, &meta, &path);
        records.push(RunRecord {
            path,
            status,
            summary,
            meta,
        });
    }
    records.sort_by(|left, right| left.path.cmp(&right.path));
    Ok(records)
}

fn classify_run_status(summary: &Option<Value>, meta: &Option<Value>, path: &Path) -> String {
    let summary_status = summary.as_ref().and_then(|v| json_str(v, &["status"]));
    let meta_status = meta.as_ref().and_then(|v| json_str(v, &["status"]));
    if summary_status == Some("success") && meta_status == Some("success") {
        return "success".to_string();
    }
    if matches!(summary_status, Some("failed" | "spawn_failed"))
        || matches!(meta_status, Some("failed" | "spawn_failed"))
    {
        return "failed".to_string();
    }
    if path.join("run-summary.json").exists() || path.join("run-meta.json").exists() {
        "partial".to_string()
    } else {
        "missing_metadata".to_string()
    }
}

fn consecutive_failures(records: &[RunRecord]) -> u64 {
    let mut count = 0u64;
    for record in records.iter().rev() {
        if record.status == "success" {
            break;
        }
        count += 1;
    }
    count
}

fn read_json_optional(path: &Path, errors: &mut Vec<String>) -> Option<Value> {
    if !path.exists() {
        return None;
    }
    match fs::read(path)
        .ok()
        .and_then(|bytes| serde_json::from_slice::<Value>(&bytes).ok())
    {
        Some(value) => Some(value),
        None => {
            errors.push(format!("parse json failed: {}", path.display()));
            None
        }
    }
}

fn build_latest_metrics(record: &RunRecord, snapshot: &mut MetricsSnapshot) {
    let summary = record.summary.as_ref();
    let meta = record.meta.as_ref();
    let run_seq = summary
        .and_then(|v| json_u64(v, &["runSeq"]))
        .or_else(|| meta.and_then(|v| json_u64(v, &["run_index"])))
        .unwrap_or_else(|| run_index_from_path(&record.path).unwrap_or(0));
    let run_id = summary
        .and_then(|v| json_str(v, &["runId"]))
        .or_else(|| meta.and_then(|v| json_str(v, &["run_id"])))
        .unwrap_or_else(|| {
            record
                .path
                .file_name()
                .and_then(|n| n.to_str())
                .unwrap_or("unknown")
        })
        .to_string();
    let sync_mode = meta
        .and_then(|v| json_str(v, &["sync_mode"]))
        .unwrap_or("unknown")
        .to_string();
    let snapshot_reason = meta
        .and_then(|v| json_str(v, &["snapshot_reason"]))
        .map(|s| s.to_string());
    let started_at = summary
        .and_then(|v| json_str(v, &["startedAtRfc3339Utc"]))
        .or_else(|| meta.and_then(|v| json_str(v, &["started_at_rfc3339_utc"])))
        .map(|s| s.to_string());
    let finished_at = summary
        .and_then(|v| json_str(v, &["finishedAtRfc3339Utc"]))
        .or_else(|| meta.and_then(|v| json_str(v, &["completed_at_rfc3339_utc"])))
        .map(|s| s.to_string());
    let wall_seconds = summary.and_then(|v| json_u64(v, &["wallMs"])).unwrap_or(0) as f64 / 1000.0;

    let mut latest = LatestRunMetrics {
        run_seq,
        run_id,
        run_dir: record.path.display().to_string(),
        status: record.status.clone(),
        sync_mode,
        snapshot_reason,
        started_at: started_at.clone(),
        finished_at: finished_at.clone(),
        start_timestamp_seconds: started_at.as_deref().and_then(parse_rfc3339_to_unix),
        finish_timestamp_seconds: finished_at.as_deref().and_then(parse_rfc3339_to_unix),
        wall_seconds,
        user_cpu_seconds: summary.and_then(|v| json_f64(v, &["processMetrics", "userSeconds"])),
        system_cpu_seconds: summary.and_then(|v| json_f64(v, &["processMetrics", "systemSeconds"])),
        cpu_percent: summary.and_then(|v| json_f64(v, &["processMetrics", "cpuPercent"])),
        max_rss_bytes: summary
            .and_then(|v| json_u64(v, &["processMetrics", "maxRssKb"]))
            .map(|kb| kb.saturating_mul(1024)),
        exit_code: summary.and_then(|v| json_i64(v, &["exitCode"])),
        ..LatestRunMetrics::default()
    };

    if let Some(summary) = summary {
        latest.vrps = json_u64(summary, &["reportCounts", "vrps"]).unwrap_or(0);
        latest.vaps = json_u64(summary, &["reportCounts", "aspas"]).unwrap_or(0);
        latest.publication_points =
            json_u64(summary, &["reportCounts", "publicationPoints"]).unwrap_or(0);
        latest.warnings = json_u64(summary, &["reportCounts", "warnings"]).unwrap_or(0);
        latest.tree_instances_processed =
            json_u64(summary, &["reportCounts", "treeInstancesProcessed"]);
        latest.tree_instances_failed = json_u64(summary, &["reportCounts", "treeInstancesFailed"]);
        latest.stage_seconds = extract_stage_seconds(summary.get("stageTiming"));
        latest.repo_sync_phase =
            extract_count_duration_map(summary.pointer("/repoSyncStats/by_phase"));
        latest.repo_terminal_state =
            extract_count_duration_map(summary.pointer("/repoSyncStats/by_terminal_state"));
        latest.download_events = json_u64(summary, &["stageTiming", "download_event_count"]);
        latest.download_bytes = json_u64(summary, &["stageTiming", "download_bytes_total"]);
        latest.artifact_sizes = extract_artifact_sizes(summary.get("artifacts"));
        latest.state_path_sizes = extract_path_sizes(summary.get("pathStats"));
    }

    parse_report(&record.path.join("report.json"), snapshot, &mut latest);
    latest.vrps_unique = count_vrp_csv_unique_keys_opt(
        &record.path.join("vrps.csv"),
        &mut snapshot.service.parse_errors,
    );
    parse_cir(&record.path.join("input.cir"), snapshot);
    parse_ccr(&record.path.join("result.ccr"), snapshot);
    snapshot.latest_run = Some(latest);
}

fn count_vrp_csv_unique_keys_opt(path: &Path, errors: &mut Vec<String>) -> Option<u64> {
    if !path.exists() {
        return None;
    }
    match count_vrp_csv_unique_keys(path) {
        Ok(count) => Some(count),
        Err(err) => {
            errors.push(err);
            None
        }
    }
}

fn count_vrp_csv_unique_keys(path: &Path) -> Result<u64, String> {
    let content = fs::read_to_string(path)
        .map_err(|e| format!("read VRP CSV failed: {}: {e}", path.display()))?;
    let mut unique = BTreeSet::new();
    for (index, line) in data_csv_lines(&content).enumerate() {
        if index == 0 {
            continue;
        }
        let columns = split_csv_simple(line);
        if columns.len() < 3 {
            return Err(format!(
                "invalid VRP CSV row in {}: expected at least 3 columns, got {}",
                path.display(),
                columns.len()
            ));
        }
        unique.insert((
            columns[0].to_string(),
            columns[1].to_string(),
            columns[2].to_string(),
        ));
    }
    Ok(unique.len() as u64)
}

fn data_csv_lines(content: &str) -> impl Iterator<Item = &str> {
    content
        .lines()
        .map(str::trim)
        .filter(|line| !line.is_empty())
        .filter(|line| !line.starts_with('#'))
}

fn split_csv_simple(line: &str) -> Vec<&str> {
    line.split(',').map(str::trim).collect()
}

fn parse_report(path: &Path, snapshot: &mut MetricsSnapshot, latest: &mut LatestRunMetrics) {
    if !path.exists() {
        return;
    }
    let Ok(bytes) = fs::read(path) else {
        snapshot
            .service
            .parse_errors
            .push(format!("read report.json failed: {}", path.display()));
        return;
    };
    let Ok(report) = serde_json::from_slice::<Value>(&bytes) else {
        snapshot
            .service
            .parse_errors
            .push(format!("parse report.json failed: {}", path.display()));
        return;
    };
    if latest.vrps == 0 {
        latest.vrps = report
            .get("vrps")
            .and_then(|v| v.as_array())
            .map(|a| a.len() as u64)
            .unwrap_or(0);
    }
    if latest.vaps == 0 {
        latest.vaps = report
            .get("aspas")
            .and_then(|v| v.as_array())
            .map(|a| a.len() as u64)
            .unwrap_or(0);
    }
    latest.warnings = latest.warnings.max(
        report
            .pointer("/tree/warnings")
            .and_then(|v| v.as_array())
            .map(|a| a.len() as u64)
            .unwrap_or(0),
    );
    if let Some(processed) = json_u64(&report, &["tree", "instances_processed"]) {
        latest.tree_instances_processed = Some(processed);
    }
    if let Some(failed) = json_u64(&report, &["tree", "instances_failed"]) {
        latest.tree_instances_failed = Some(failed);
    }
    if latest.repo_sync_phase.is_empty() {
        latest.repo_sync_phase =
            extract_count_duration_map(report.pointer("/repo_sync_stats/by_phase"));
    }
    if latest.repo_terminal_state.is_empty() {
        latest.repo_terminal_state =
            extract_count_duration_map(report.pointer("/repo_sync_stats/by_terminal_state"));
    }
    if let Some(pps) = report.get("publication_points").and_then(|v| v.as_array()) {
        latest.publication_points = pps.len() as u64;
        extract_publication_point_metrics(pps, report.get("downloads"), snapshot);
    }
}

fn extract_publication_point_metrics(
    pps: &[Value],
    downloads: Option<&Value>,
    snapshot: &mut MetricsSnapshot,
) {
    let mut repos: BTreeMap<String, RepoMetrics> = BTreeMap::new();
    let mut pp_by_object_count = Vec::<TopPublicationPoint>::new();
    let mut pp_by_sync_duration = Vec::<TopPublicationPoint>::new();
    let mut large_pp_counts = BTreeMap::<u64, u64>::new();
    let mut pp_sync_histograms = BTreeMap::<String, Histogram>::new();
    let mut object_counts = BTreeMap::<(String, String), u64>::new();

    for pp in pps {
        let pp_uri = json_str(pp, &["publication_point_rsync_uri"])
            .or_else(|| json_str(pp, &["manifest_rsync_uri"]))
            .or_else(|| json_str(pp, &["rsync_base_uri"]))
            .unwrap_or("unknown");
        let repo_uri = json_str(pp, &["rrdp_notification_uri"])
            .or_else(|| json_str(pp, &["rsync_base_uri"]))
            .or_else(|| json_str(pp, &["publication_point_rsync_uri"]))
            .unwrap_or(pp_uri);
        let repo_id = short_sha256(repo_uri);
        let pp_id = short_sha256(pp_uri);
        let host = uri_host(repo_uri);
        let transport = json_str(pp, &["repo_sync_source"])
            .or_else(|| json_str(pp, &["source"]))
            .map(normalize_transport)
            .unwrap_or_else(|| infer_transport(repo_uri));
        let duration_ms = json_u64(pp, &["repo_sync_duration_ms"]).unwrap_or(0);
        let duration_seconds = duration_ms as f64 / 1000.0;
        let phase = json_str(pp, &["repo_sync_phase"])
            .unwrap_or("unknown")
            .to_string();
        let terminal_state = json_str(pp, &["repo_terminal_state"])
            .unwrap_or("unknown")
            .to_string();
        let object_count = pp
            .get("objects")
            .and_then(|v| v.as_array())
            .map(|a| a.len() as u64)
            .unwrap_or(0);

        let repo = repos.entry(repo_id.clone()).or_insert_with(|| RepoMetrics {
            repo_id: repo_id.clone(),
            uri: repo_uri.to_string(),
            host: host.clone(),
            transport: transport.clone(),
            sync_success: true,
            ..RepoMetrics::default()
        });
        repo.publication_points += 1;
        repo.duration_seconds_sum += duration_seconds;
        repo.duration_seconds_max = repo.duration_seconds_max.max(duration_seconds);
        if !is_success_terminal_state(&terminal_state) {
            repo.sync_success = false;
        }
        *repo.phase_counts.entry(phase.clone()).or_default() += 1;
        *repo
            .terminal_state_counts
            .entry(terminal_state.clone())
            .or_default() += 1;

        for threshold in LARGE_PP_OBJECT_THRESHOLDS {
            if object_count > *threshold {
                *large_pp_counts.entry(*threshold).or_default() += 1;
            }
        }
        pp_sync_histograms
            .entry(transport.clone())
            .or_insert_with(|| Histogram::new(PP_SYNC_SECONDS_BUCKETS))
            .observe(duration_seconds);

        if let Some(objects) = pp.get("objects").and_then(|v| v.as_array()) {
            for object in objects {
                let kind = json_str(object, &["kind"]).unwrap_or("unknown").to_string();
                let result = json_str(object, &["result"])
                    .unwrap_or("unknown")
                    .to_string();
                *object_counts.entry((kind.clone(), result)).or_default() += 1;
                *repo.object_counts.entry(kind).or_default() += 1;
            }
        }

        let top = TopPublicationPoint {
            rank: 0,
            pp_id,
            repo_id,
            uri: pp_uri.to_string(),
            repo_uri: repo_uri.to_string(),
            host,
            transport,
            object_count,
            sync_duration_ms: duration_ms,
            terminal_state,
            phase,
        };
        pp_by_object_count.push(top.clone());
        pp_by_sync_duration.push(top);
    }

    let mut repo_stats = repos.into_values().collect::<Vec<_>>();
    assign_download_bytes_to_repos(&mut repo_stats, downloads);
    for repo in &mut repo_stats {
        if repo.publication_points > 0 {
            repo.duration_seconds_avg = repo.duration_seconds_sum / repo.publication_points as f64;
        }
    }
    let mut top_repos = repo_stats
        .iter()
        .map(|repo| TopRepo {
            rank: 0,
            repo_id: repo.repo_id.clone(),
            uri: repo.uri.clone(),
            host: repo.host.clone(),
            transport: repo.transport.clone(),
            duration_ms_max: (repo.duration_seconds_max * 1000.0).round() as u64,
            duration_ms_sum: (repo.duration_seconds_sum * 1000.0).round() as u64,
            publication_points: repo.publication_points,
        })
        .collect::<Vec<_>>();
    top_repos.sort_by(|a, b| b.duration_ms_max.cmp(&a.duration_ms_max));
    top_repos.truncate(20);
    for (index, item) in top_repos.iter_mut().enumerate() {
        item.rank = index + 1;
    }

    pp_by_object_count.sort_by(|a, b| b.object_count.cmp(&a.object_count));
    pp_by_object_count.truncate(20);
    for (index, item) in pp_by_object_count.iter_mut().enumerate() {
        item.rank = index + 1;
    }
    pp_by_sync_duration.sort_by(|a, b| b.sync_duration_ms.cmp(&a.sync_duration_ms));
    pp_by_sync_duration.truncate(20);
    for (index, item) in pp_by_sync_duration.iter_mut().enumerate() {
        item.rank = index + 1;
    }

    snapshot.repo_stats = repo_stats;
    snapshot.object_counts = object_counts;
    snapshot.large_pp_counts = large_pp_counts;
    snapshot.pp_sync_histograms = pp_sync_histograms;
    snapshot.top_repos_by_sync_duration = top_repos;
    snapshot.top_pp_by_object_count = pp_by_object_count;
    snapshot.top_pp_by_sync_duration = pp_by_sync_duration;
}

fn assign_download_bytes_to_repos(repos: &mut [RepoMetrics], downloads: Option<&Value>) {
    let Some(downloads) = downloads.and_then(|v| v.as_array()) else {
        return;
    };
    for download in downloads {
        let Some(uri) = json_str(download, &["uri"]) else {
            continue;
        };
        let bytes = json_u64(download, &["bytes"]).unwrap_or(0);
        if bytes == 0 {
            continue;
        }
        if let Some(index) = find_repo_for_download(repos, uri) {
            repos[index].download_bytes = repos[index].download_bytes.saturating_add(bytes);
        }
    }
}

fn find_repo_for_download(repos: &[RepoMetrics], uri: &str) -> Option<usize> {
    if let Some(index) = repos.iter().position(|repo| repo.uri == uri) {
        return Some(index);
    }
    if uri.starts_with("rsync://") {
        return repos
            .iter()
            .enumerate()
            .filter(|(_, repo)| uri.starts_with(&repo.uri))
            .max_by_key(|(_, repo)| repo.uri.len())
            .map(|(index, _)| index);
    }

    let uri_host = uri_host(uri);
    let mut candidates = repos
        .iter()
        .enumerate()
        .filter(|(_, repo)| repo.host == uri_host && repo.uri.starts_with("http"))
        .collect::<Vec<_>>();
    if candidates.len() == 1 {
        return Some(candidates[0].0);
    }
    candidates.sort_by_key(|(_, repo)| common_prefix_len(&repo.uri, uri));
    candidates
        .last()
        .and_then(|(index, repo)| (common_prefix_len(&repo.uri, uri) > 0).then_some(*index))
}

fn is_success_terminal_state(state: &str) -> bool {
    matches!(state, "fresh" | "cached" | "reused" | "valid")
}

fn parse_cir(path: &Path, snapshot: &mut MetricsSnapshot) {
    if !path.exists() {
        return;
    }
    match fs::read(path)
        .map_err(|e| e.to_string())
        .and_then(|bytes| decode_cir(&bytes).map_err(|e| e.to_string()))
    {
        Ok(cir) => {
            let mut objects_by_type = BTreeMap::new();
            for object in cir.validated_objects() {
                *objects_by_type
                    .entry(object_type_from_uri(&object.rsync_uri))
                    .or_default() += 1;
            }
            let mut fresh_objects_by_type = BTreeMap::new();
            for object in &cir.fresh_validated_objects {
                *fresh_objects_by_type
                    .entry(object_type_from_uri(&object.rsync_uri))
                    .or_default() += 1;
            }
            let mut cached_objects_by_type = BTreeMap::new();
            for object in &cir.cached_validated_objects {
                *cached_objects_by_type
                    .entry(object_type_from_uri(&object.rsync_uri))
                    .or_default() += 1;
            }
            let mut rejected_objects_by_type = BTreeMap::new();
            for object in cir.rejected_objects() {
                *rejected_objects_by_type
                    .entry(object_type_from_uri(&object.object_uri))
                    .or_default() += 1;
            }
            let mut fresh_rejected_objects_by_type = BTreeMap::new();
            for object in &cir.fresh_rejected_objects {
                *fresh_rejected_objects_by_type
                    .entry(object_type_from_uri(&object.object_uri))
                    .or_default() += 1;
            }
            let mut cached_rejected_objects_by_type = BTreeMap::new();
            for object in &cir.cached_rejected_objects {
                *cached_rejected_objects_by_type
                    .entry(object_type_from_uri(&object.object_uri))
                    .or_default() += 1;
            }
            snapshot.cir = Some(CirMetrics {
                version: cir.version,
                objects: cir.validated_object_count() as u64,
                fresh_objects: cir.fresh_validated_objects.len() as u64,
                cached_objects: cir.cached_validated_objects.len() as u64,
                trust_anchors: cir.trust_anchors.len() as u64,
                rejected_objects: cir.rejected_object_count() as u64,
                fresh_rejected_objects: cir.fresh_rejected_objects.len() as u64,
                cached_rejected_objects: cir.cached_rejected_objects.len() as u64,
                object_list_sha256: hex::encode(&cir.object_list_sha256),
                fresh_object_list_sha256: hex::encode(&cir.fresh_object_list_sha256),
                cached_object_list_sha256: hex::encode(&cir.cached_object_list_sha256),
                reject_list_sha256: hex::encode(&cir.reject_list_sha256),
                fresh_reject_list_sha256: hex::encode(&cir.fresh_reject_list_sha256),
                cached_reject_list_sha256: hex::encode(&cir.cached_reject_list_sha256),
                objects_by_type,
                fresh_objects_by_type,
                cached_objects_by_type,
                rejected_objects_by_type,
                fresh_rejected_objects_by_type,
                cached_rejected_objects_by_type,
            });
        }
        Err(err) => snapshot
            .service
            .parse_errors
            .push(format!("decode CIR failed: {}: {err}", path.display())),
    }
}

fn parse_ccr(path: &Path, snapshot: &mut MetricsSnapshot) {
    if !path.exists() {
        return;
    }
    match fs::read(path)
        .map_err(|e| e.to_string())
        .and_then(|bytes| decode_content_info(&bytes).map_err(|e| e.to_string()))
    {
        Ok(ccr) => {
            let content = ccr.content;
            let mut state_present = BTreeMap::new();
            let mut state_items = BTreeMap::new();
            let mut state_digests = BTreeMap::new();
            if let Some(state) = content.mfts.as_ref() {
                state_present.insert("mfts".to_string(), true);
                state_items.insert("mfts".to_string(), state.mis.len() as u64);
                state_digests.insert("mfts".to_string(), hex::encode(&state.hash));
            } else {
                state_present.insert("mfts".to_string(), false);
            }
            if let Some(state) = content.vrps.as_ref() {
                state_present.insert("vrps".to_string(), true);
                state_items.insert("vrps".to_string(), state.rps.len() as u64);
                state_digests.insert("vrps".to_string(), hex::encode(&state.hash));
            } else {
                state_present.insert("vrps".to_string(), false);
            }
            if let Some(state) = content.vaps.as_ref() {
                state_present.insert("vaps".to_string(), true);
                state_items.insert("vaps".to_string(), state.aps.len() as u64);
                state_digests.insert("vaps".to_string(), hex::encode(&state.hash));
            } else {
                state_present.insert("vaps".to_string(), false);
            }
            if let Some(state) = content.tas.as_ref() {
                state_present.insert("tas".to_string(), true);
                state_items.insert("tas".to_string(), state.skis.len() as u64);
                state_digests.insert("tas".to_string(), hex::encode(&state.hash));
            } else {
                state_present.insert("tas".to_string(), false);
            }
            if let Some(state) = content.rks.as_ref() {
                state_present.insert("rks".to_string(), true);
                state_items.insert("rks".to_string(), state.rksets.len() as u64);
                state_digests.insert("rks".to_string(), hex::encode(&state.hash));
            } else {
                state_present.insert("rks".to_string(), false);
            }
            snapshot.ccr = Some(CcrMetrics {
                version: content.version,
                state_present,
                state_items,
                state_digests,
            });
        }
        Err(err) => snapshot
            .service
            .parse_errors
            .push(format!("decode CCR failed: {}: {err}", path.display())),
    }
}

fn render_metrics(snapshot: &MetricsSnapshot) -> String {
    let mut out = String::new();
    let mut writer = PromWriter::new(&mut out);
    let instance = snapshot.instance.as_str();

    writer.gauge(
        "ours_rp_metrics_service_up",
        "Artifact metrics service is up",
        &[label("instance", instance)],
        1.0,
    );
    writer.gauge(
        "ours_rp_metrics_service_last_scan_timestamp_seconds",
        "Unix timestamp of the last artifact scan",
        &[label("instance", instance)],
        snapshot.service.last_scan_timestamp_seconds,
    );
    writer.gauge(
        "ours_rp_metrics_service_last_scan_duration_seconds",
        "Duration of the last artifact scan",
        &[label("instance", instance)],
        snapshot.service.last_scan_duration_seconds,
    );
    writer.gauge(
        "ours_rp_metrics_service_last_reload_success",
        "Whether the last artifact reload had no parse errors",
        &[label("instance", instance)],
        bool_value(snapshot.service.last_reload_success),
    );
    writer.gauge(
        "ours_rp_metrics_service_parse_errors",
        "Current parse error count",
        &[label("instance", instance)],
        snapshot.service.parse_errors.len() as f64,
    );
    writer.gauge(
        "ours_rp_metrics_service_known_runs",
        "Known run directories by status",
        &[label("instance", instance), label("status", "success")],
        snapshot.runs.success as f64,
    );
    writer.gauge(
        "ours_rp_metrics_service_known_runs",
        "Known run directories by status",
        &[label("instance", instance), label("status", "failed")],
        snapshot.runs.failed as f64,
    );
    writer.gauge(
        "ours_rp_metrics_service_known_runs",
        "Known run directories by status",
        &[label("instance", instance), label("status", "partial")],
        snapshot.runs.partial as f64,
    );

    writer.counter(
        "ours_rp_run_completed_total",
        "Completed runs observed by the artifact metrics service",
        &[label("instance", instance), label("status", "success")],
        snapshot.cumulative.completed_success_total as f64,
    );
    writer.counter(
        "ours_rp_run_completed_total",
        "Completed runs observed by the artifact metrics service",
        &[label("instance", instance), label("status", "failed")],
        snapshot.cumulative.completed_failed_total as f64,
    );
    writer.counter(
        "ours_rp_run_observed_duration_seconds_sum",
        "Observed wall duration sum for successful runs",
        &[label("instance", instance)],
        snapshot.cumulative.observed_duration_seconds_sum,
    );
    writer.counter(
        "ours_rp_run_observed_duration_seconds_count",
        "Observed wall duration count for successful runs",
        &[label("instance", instance)],
        snapshot.cumulative.observed_duration_seconds_count as f64,
    );
    writer.counter(
        "ours_rp_run_observed_download_bytes_total",
        "Observed download bytes across successful runs",
        &[label("instance", instance)],
        snapshot.cumulative.observed_download_bytes_total as f64,
    );
    writer.gauge(
        "ours_rp_run_consecutive_failures",
        "Consecutive non-success runs at the end of the run list",
        &[label("instance", instance)],
        snapshot.runs.consecutive_failures as f64,
    );

    if let Some(latest) = snapshot.latest_run.as_ref() {
        render_latest_metrics(&mut writer, instance, latest);
    }
    render_repo_metrics(&mut writer, instance, &snapshot.repo_stats);
    render_failed_repo_metrics(&mut writer, instance, &snapshot.repo_stats);
    render_top_repo_metrics(&mut writer, instance, &snapshot.top_repos_by_sync_duration);
    render_object_metrics(&mut writer, instance, &snapshot.object_counts);
    render_large_pp_metrics(&mut writer, instance, &snapshot.large_pp_counts);
    render_top_publication_point_metrics(&mut writer, instance, &snapshot.top_pp_by_object_count);
    for (transport, histogram) in &snapshot.pp_sync_histograms {
        writer.histogram(
            "ours_rp_publication_point_sync_duration_seconds",
            "Distribution of sync duration per publication point",
            &[label("instance", instance), label("transport", transport)],
            histogram,
        );
    }
    if let Some(cir) = snapshot.cir.as_ref() {
        render_cir_metrics(&mut writer, instance, cir);
    }
    if let Some(ccr) = snapshot.ccr.as_ref() {
        render_ccr_metrics(&mut writer, instance, ccr);
    }
    render_rtr_metrics(&mut writer, instance, snapshot);
    out
}

fn render_rtr_metrics(writer: &mut PromWriter<'_>, instance: &str, snapshot: &MetricsSnapshot) {
    let rtr = &snapshot.rtr;
    let scan_now = snapshot.service.last_scan_timestamp_seconds;

    writer.gauge(
        "ours_rp_rtr_metrics_enabled",
        "Whether RTR report collection is enabled by environment variable",
        &[label("instance", instance)],
        bool_value(rtr.enabled),
    );
    writer.counter(
        "ours_rp_rtr_metrics_parse_errors_total",
        "Total RTR report parse errors observed by the artifact metrics service",
        &[label("instance", instance)],
        rtr.parse_errors_total as f64,
    );
    writer.gauge(
        "ours_rp_rtr_metrics_parse_errors",
        "Current RTR report parse error count in the latest scan",
        &[label("instance", instance)],
        rtr.parse_errors.len() as f64,
    );

    for kind in ["source", "runtime", "clients"] {
        let scan = rtr.files.get(kind).cloned().unwrap_or_default();
        writer.gauge(
            "ours_rp_rtr_report_files_seen",
            "RTR report files seen by kind",
            &[label("instance", instance), label("kind", kind)],
            scan.seen as f64,
        );
        if let Some(timestamp) = scan.latest_timestamp_seconds {
            writer.gauge(
                "ours_rp_rtr_report_latest_timestamp_seconds",
                "Latest RTR report generated timestamp by kind",
                &[label("instance", instance), label("kind", kind)],
                timestamp,
            );
            writer.gauge(
                "ours_rp_rtr_report_age_seconds",
                "Age of the latest RTR report by kind at last scan time",
                &[label("instance", instance), label("kind", kind)],
                (scan_now - timestamp).max(0.0),
            );
        }
    }

    if let Some(source) = rtr.source.as_ref() {
        if let Some(status) = source.refresh_status.as_deref() {
            writer.gauge(
                "ours_rp_rtr_source_refresh_status",
                "Current RTR source refresh status",
                &[label("instance", instance), label("status", status)],
                1.0,
            );
        }
        if let Some(value) = source.refresh_changed {
            writer.gauge(
                "ours_rp_rtr_source_refresh_changed",
                "Whether the latest RTR source refresh changed data",
                &[label("instance", instance)],
                bool_value(value),
            );
        }
        if let Some(value) = source.refresh_duration_seconds {
            writer.gauge(
                "ours_rp_rtr_source_refresh_duration_seconds",
                "Latest RTR source refresh duration",
                &[label("instance", instance)],
                value,
            );
        }
        if let Some(value) = source.refresh_consecutive_failures {
            writer.gauge(
                "ours_rp_rtr_source_refresh_consecutive_failures",
                "Consecutive RTR source refresh failures",
                &[label("instance", instance)],
                value as f64,
            );
        }
        if let Some(value) = source.refresh_last_success_seconds {
            writer.gauge(
                "ours_rp_rtr_source_last_success_age_seconds",
                "Age of the last successful RTR source refresh at last scan time",
                &[label("instance", instance)],
                (scan_now - value).max(0.0),
            );
        }
        if let Some(value) = source.generated_at_seconds {
            writer.gauge(
                "ours_rp_rtr_source_report_age_seconds",
                "Age of the latest RTR source report at last scan time",
                &[label("instance", instance)],
                (scan_now - value).max(0.0),
            );
        }
        if let Some(value) = source.ccr_file_size_bytes {
            writer.gauge(
                "ours_rp_rtr_source_ccr_file_size_bytes",
                "RTR source CCR input file size",
                &[label("instance", instance)],
                value as f64,
            );
        }
        for (stage, counts) in &source.data_quality {
            for (kind, count) in counts {
                writer.gauge(
                    "ours_rp_rtr_data_quality_items",
                    "RTR data quality item counts by stage and type",
                    &[
                        label("instance", instance),
                        label("stage", stage),
                        label("type", kind),
                    ],
                    *count as f64,
                );
            }
        }
        for (kind, count) in &source.slurm_filters {
            writer.gauge(
                "ours_rp_rtr_slurm_filters",
                "RTR SLURM filtered item counts by type",
                &[label("instance", instance), label("type", kind)],
                *count as f64,
            );
        }
        for (kind, count) in &source.slurm_assertions {
            writer.gauge(
                "ours_rp_rtr_slurm_assertions",
                "RTR SLURM asserted item counts by type",
                &[label("instance", instance), label("type", kind)],
                *count as f64,
            );
        }
        if let Some(availability) = source.cache_availability.as_deref() {
            writer.gauge(
                "ours_rp_rtr_cache_ready",
                "Whether RTR cache availability is ready",
                &[label("instance", instance)],
                bool_value(availability == "ready"),
            );
        }
        for version in &source.cache_versions {
            let version_label = version.version.to_string();
            if let Some(serial) = version.serial {
                writer.gauge(
                    "ours_rp_rtr_cache_version_serial",
                    "RTR cache version serial by version",
                    &[
                        label("instance", instance),
                        label("version", &version_label),
                    ],
                    serial as f64,
                );
            }
            if let Some(length) = version.delta_window_length {
                writer.gauge(
                    "ours_rp_rtr_cache_delta_window_length",
                    "RTR cache delta window length by version",
                    &[
                        label("instance", instance),
                        label("version", &version_label),
                    ],
                    length as f64,
                );
            }
            for kind in RTR_COUNT_TYPES {
                writer.gauge(
                    "ours_rp_rtr_cache_snapshot_items",
                    "RTR cache snapshot item counts by version and type",
                    &[
                        label("instance", instance),
                        label("version", &version_label),
                        label("type", kind),
                    ],
                    version.snapshot.get(*kind).copied().unwrap_or(0) as f64,
                );
            }
            for (direction, counts) in [
                ("announced", &version.latest_delta_announced),
                ("withdrawn", &version.latest_delta_withdrawn),
            ] {
                for kind in RTR_COUNT_TYPES {
                    writer.gauge(
                        "ours_rp_rtr_cache_delta_items",
                        "RTR cache latest delta item counts by version, direction and type",
                        &[
                            label("instance", instance),
                            label("version", &version_label),
                            label("direction", direction),
                            label("type", kind),
                        ],
                        counts.get(*kind).copied().unwrap_or(0) as f64,
                    );
                }
            }
        }
    }

    if let Some(runtime) = rtr.runtime.as_ref() {
        if let Some(value) = runtime.process_rss_bytes {
            writer.gauge(
                "ours_rp_rtr_process_rss_bytes",
                "RTR process resident set size",
                &[label("instance", instance)],
                value as f64,
            );
        }
    }
    if let Some(runtime_age) = rtr
        .runtime
        .as_ref()
        .and_then(|runtime| runtime.generated_at_seconds)
        .map(|timestamp| (scan_now - timestamp).max(0.0))
    {
        writer.gauge(
            "ours_rp_rtr_runtime_report_age_seconds",
            "Age of the latest RTR runtime report at last scan time",
            &[label("instance", instance)],
            runtime_age,
        );
    }
    if let Some(clients_age) = rtr
        .clients
        .as_ref()
        .and_then(|clients| clients.generated_at_seconds)
        .map(|timestamp| (scan_now - timestamp).max(0.0))
    {
        writer.gauge(
            "ours_rp_rtr_clients_report_age_seconds",
            "Age of the latest RTR clients report at last scan time",
            &[label("instance", instance)],
            clients_age,
        );
    }

    if let Some(service) = rtr.preferred_service() {
        if let Some(value) = service.uptime_seconds {
            writer.gauge(
                "ours_rp_rtr_service_uptime_seconds",
                "RTR service uptime",
                &[label("instance", instance)],
                value as f64,
            );
        }
        if let Some(value) = service.active_connections {
            writer.gauge(
                "ours_rp_rtr_active_connections",
                "RTR active connections",
                &[label("instance", instance)],
                value as f64,
            );
        }
        for (transport, count) in &service.connections_by_transport {
            writer.gauge(
                "ours_rp_rtr_connections",
                "RTR connections by transport",
                &[label("instance", instance), label("transport", transport)],
                *count as f64,
            );
        }
        if let Some(value) = service.session_listeners {
            writer.gauge(
                "ours_rp_rtr_session_listeners",
                "RTR session listeners",
                &[label("instance", instance)],
                value as f64,
            );
        }
        if let Some(value) = service.max_connections {
            writer.gauge(
                "ours_rp_rtr_max_connections",
                "RTR maximum configured connections",
                &[label("instance", instance)],
                value as f64,
            );
        }
        if let Some(value) = service.connection_utilization {
            writer.gauge(
                "ours_rp_rtr_connection_utilization",
                "RTR connection utilization ratio",
                &[label("instance", instance)],
                value,
            );
        }
    }
}

fn render_latest_metrics(writer: &mut PromWriter<'_>, instance: &str, latest: &LatestRunMetrics) {
    writer.gauge(
        "ours_rp_run_sequence",
        "Latest successful run sequence",
        &[label("instance", instance)],
        latest.run_seq as f64,
    );
    writer.gauge(
        "ours_rp_run_success",
        "Whether the latest selected run is successful",
        &[label("instance", instance)],
        bool_value(latest.status == "success"),
    );
    writer.gauge(
        "ours_rp_run_sync_mode",
        "Latest run sync mode state",
        &[
            label("instance", instance),
            label("sync_mode", &latest.sync_mode),
        ],
        1.0,
    );
    if let Some(ts) = latest.start_timestamp_seconds {
        writer.gauge(
            "ours_rp_run_start_timestamp_seconds",
            "Latest run start timestamp",
            &[label("instance", instance)],
            ts,
        );
    }
    if let Some(ts) = latest.finish_timestamp_seconds {
        writer.gauge(
            "ours_rp_run_finish_timestamp_seconds",
            "Latest run finish timestamp",
            &[label("instance", instance)],
            ts,
        );
    }
    writer.gauge(
        "ours_rp_run_duration_seconds",
        "Latest run wall duration",
        &[label("instance", instance)],
        latest.wall_seconds,
    );
    if let Some(value) = latest.user_cpu_seconds {
        writer.gauge(
            "ours_rp_run_user_cpu_seconds",
            "Latest run user CPU seconds",
            &[label("instance", instance)],
            value,
        );
    }
    if let Some(value) = latest.system_cpu_seconds {
        writer.gauge(
            "ours_rp_run_system_cpu_seconds",
            "Latest run system CPU seconds",
            &[label("instance", instance)],
            value,
        );
    }
    if let Some(value) = latest.cpu_percent {
        writer.gauge(
            "ours_rp_run_cpu_percent",
            "Latest run CPU percent from GNU time",
            &[label("instance", instance)],
            value,
        );
    }
    if let Some(value) = latest.max_rss_bytes {
        writer.gauge(
            "ours_rp_run_max_rss_bytes",
            "Latest run maximum resident set size",
            &[label("instance", instance)],
            value as f64,
        );
    }
    if let Some(value) = latest.exit_code {
        writer.gauge(
            "ours_rp_run_exit_code",
            "Latest run exit code",
            &[label("instance", instance)],
            value as f64,
        );
    }
    writer.gauge(
        "ours_rp_vrps",
        "Latest run VRP count",
        &[label("instance", instance), label("kind", "total")],
        latest.vrps as f64,
    );
    if let Some(value) = latest.vrps_unique {
        writer.gauge(
            "ours_rp_vrps",
            "Latest run VRP count",
            &[label("instance", instance), label("kind", "unique")],
            value as f64,
        );
    }
    writer.gauge(
        "ours_rp_vaps",
        "Latest run VAP/ASPA count",
        &[label("instance", instance), label("kind", "total")],
        latest.vaps as f64,
    );
    writer.gauge(
        "ours_rp_publication_points",
        "Latest run publication point count",
        &[label("instance", instance)],
        latest.publication_points as f64,
    );
    writer.gauge(
        "ours_rp_warnings",
        "Latest run warning count",
        &[label("instance", instance)],
        latest.warnings as f64,
    );
    if let Some(value) = latest.tree_instances_processed {
        writer.gauge(
            "ours_rp_tree_instances",
            "Latest run tree instances by state",
            &[label("instance", instance), label("state", "processed")],
            value as f64,
        );
    }
    if let Some(value) = latest.tree_instances_failed {
        writer.gauge(
            "ours_rp_tree_instances",
            "Latest run tree instances by state",
            &[label("instance", instance), label("state", "failed")],
            value as f64,
        );
    }
    for (stage, value) in &latest.stage_seconds {
        writer.gauge(
            "ours_rp_run_stage_duration_seconds",
            "Latest run stage duration",
            &[label("instance", instance), label("stage", stage)],
            *value,
        );
    }
    for (phase, stat) in &latest.repo_sync_phase {
        writer.gauge(
            "ours_rp_repo_sync_phase_count",
            "Publication points by repo sync phase",
            &[label("instance", instance), label("phase", phase)],
            stat.count as f64,
        );
        writer.gauge(
            "ours_rp_repo_sync_phase_duration_seconds_total",
            "Repo sync phase cumulative duration in latest run",
            &[label("instance", instance), label("phase", phase)],
            stat.duration_seconds_total,
        );
    }
    for (state, stat) in &latest.repo_terminal_state {
        writer.gauge(
            "ours_rp_repo_terminal_state_count",
            "Publication points by terminal state",
            &[label("instance", instance), label("terminal_state", state)],
            stat.count as f64,
        );
        writer.gauge(
            "ours_rp_repo_terminal_state_duration_seconds_total",
            "Terminal state cumulative duration in latest run",
            &[label("instance", instance), label("terminal_state", state)],
            stat.duration_seconds_total,
        );
    }
    if let Some(value) = latest.download_events {
        writer.gauge(
            "ours_rp_download_events",
            "Latest run download event count",
            &[label("instance", instance)],
            value as f64,
        );
    }
    if let Some(value) = latest.download_bytes {
        writer.gauge(
            "ours_rp_download_bytes",
            "Latest run download bytes",
            &[label("instance", instance)],
            value as f64,
        );
    }
    for (artifact, size) in &latest.artifact_sizes {
        writer.gauge(
            "ours_rp_artifact_size_bytes",
            "Latest run artifact size",
            &[label("instance", instance), label("artifact", artifact)],
            *size as f64,
        );
    }
    for (path, stat) in &latest.state_path_sizes {
        writer.gauge(
            "ours_rp_state_path_size_bytes",
            "State path size",
            &[label("instance", instance), label("path", path)],
            stat.total_size_bytes as f64,
        );
        writer.gauge(
            "ours_rp_state_path_files",
            "State path file count",
            &[label("instance", instance), label("path", path)],
            stat.file_count as f64,
        );
        writer.gauge(
            "ours_rp_state_db_size_bytes",
            "Persistent state database size from latest run pathStats",
            &[label("instance", instance), label("db", path)],
            stat.total_size_bytes as f64,
        );
        writer.gauge(
            "ours_rp_state_db_files",
            "Persistent state database file count from latest run pathStats",
            &[label("instance", instance), label("db", path)],
            stat.file_count as f64,
        );
        writer.gauge(
            "ours_rp_state_db_dirs",
            "Persistent state database directory count from latest run pathStats",
            &[label("instance", instance), label("db", path)],
            stat.dir_count as f64,
        );
    }
}

fn render_repo_metrics(writer: &mut PromWriter<'_>, instance: &str, repos: &[RepoMetrics]) {
    for repo in repos {
        let base = [
            label("instance", instance),
            label("repo_id", &repo.repo_id),
            label("host", &repo.host),
            label("uri", &repo.uri),
            label("transport", &repo.transport),
        ];
        writer.gauge("ours_rp_repository_info", "Repository metadata", &base, 1.0);
        writer.gauge(
            "ours_rp_repository_publication_points",
            "Publication points per repository",
            &base,
            repo.publication_points as f64,
        );
        writer.gauge(
            "ours_rp_repository_sync_success",
            "Whether repository sync is successful in the latest run",
            &base,
            bool_value(repo.sync_success),
        );
        writer.gauge(
            "ours_rp_repository_download_bytes",
            "Repository download bytes attributed from latest run download events",
            &base,
            repo.download_bytes as f64,
        );
        for (stat, value) in [
            ("sum", repo.duration_seconds_sum),
            ("max", repo.duration_seconds_max),
            ("avg", repo.duration_seconds_avg),
        ] {
            let labels = [
                label("instance", instance),
                label("repo_id", &repo.repo_id),
                label("host", &repo.host),
                label("transport", &repo.transport),
                label("stat", stat),
            ];
            writer.gauge(
                "ours_rp_repository_sync_duration_seconds",
                "Repository sync duration summary",
                &labels,
                value,
            );
        }
        for (phase, count) in &repo.phase_counts {
            let labels = [
                label("instance", instance),
                label("repo_id", &repo.repo_id),
                label("host", &repo.host),
                label("phase", phase),
            ];
            writer.gauge(
                "ours_rp_repository_sync_phase_publication_points",
                "Repository publication points by sync phase",
                &labels,
                *count as f64,
            );
        }
        for (state, count) in &repo.terminal_state_counts {
            let labels = [
                label("instance", instance),
                label("repo_id", &repo.repo_id),
                label("host", &repo.host),
                label("terminal_state", state),
            ];
            writer.gauge(
                "ours_rp_repository_terminal_state_publication_points",
                "Repository publication points by terminal state",
                &labels,
                *count as f64,
            );
        }
        for kind in ["manifest", "crl", "certificate", "roa", "aspa"] {
            let labels = [
                label("instance", instance),
                label("repo_id", &repo.repo_id),
                label("host", &repo.host),
                label("uri", &repo.uri),
                label("object_type", kind),
            ];
            writer.gauge(
                "ours_rp_repository_objects_by_type",
                "Repository object count by object type from latest run report",
                &labels,
                repo.object_counts.get(kind).copied().unwrap_or(0) as f64,
            );
        }
    }
}

fn render_failed_repo_metrics(writer: &mut PromWriter<'_>, instance: &str, repos: &[RepoMetrics]) {
    for repo in repos {
        if repo.phase_counts.contains_key("rrdp_failed_rsync_failed") {
            writer.gauge(
                "ours_rp_rrdp_rsync_failed_repository_duration_seconds",
                "Repositories whose RRDP and rsync sync both failed; value is max sync duration when available",
                &[
                    label("instance", instance),
                    label("repo_id", &repo.repo_id),
                    label("host", &repo.host),
                    label("phase", "rrdp_failed_rsync_failed"),
                    label("transport", &repo.transport),
                    label("uri", &repo.uri),
                ],
                repo.duration_seconds_max,
            );
        }
    }
}

fn render_top_repo_metrics(writer: &mut PromWriter<'_>, instance: &str, repos: &[TopRepo]) {
    for repo in repos {
        writer.gauge(
            "ours_rp_top_repository_sync_duration_seconds",
            "Top repositories by max sync duration in latest run",
            &[
                label("instance", instance),
                label("rank", &repo.rank.to_string()),
                label("repo_id", &repo.repo_id),
                label("host", &repo.host),
                label("transport", &repo.transport),
                label("publication_points", &repo.publication_points.to_string()),
                label("uri", &repo.uri),
            ],
            repo.duration_ms_max as f64 / 1000.0,
        );
    }
}

fn render_object_metrics(
    writer: &mut PromWriter<'_>,
    instance: &str,
    counts: &BTreeMap<(String, String), u64>,
) {
    for ((object_type, result), count) in counts {
        writer.gauge(
            "ours_rp_objects",
            "Latest run audited objects by type and result",
            &[
                label("instance", instance),
                label("object_type", object_type),
                label("result", result),
            ],
            *count as f64,
        );
    }
}

fn render_top_publication_point_metrics(
    writer: &mut PromWriter<'_>,
    instance: &str,
    publication_points: &[TopPublicationPoint],
) {
    for publication_point in publication_points {
        writer.gauge(
            "ours_rp_top_publication_point_object_count",
            "Top publication points by object count in latest run",
            &[
                label("instance", instance),
                label("rank", &publication_point.rank.to_string()),
                label("pp_id", &publication_point.pp_id),
                label("repo_id", &publication_point.repo_id),
                label("host", &publication_point.host),
                label("transport", &publication_point.transport),
                label("terminal_state", &publication_point.terminal_state),
                label("phase", &publication_point.phase),
                label("uri", &publication_point.uri),
            ],
            publication_point.object_count as f64,
        );
    }
}

fn render_large_pp_metrics(
    writer: &mut PromWriter<'_>,
    instance: &str,
    counts: &BTreeMap<u64, u64>,
) {
    for threshold in LARGE_PP_OBJECT_THRESHOLDS {
        writer.gauge(
            "ours_rp_large_publication_points",
            "Publication points with object count greater than threshold",
            &[
                label("instance", instance),
                label("object_count_gt", &threshold.to_string()),
            ],
            counts.get(threshold).copied().unwrap_or(0) as f64,
        );
    }
}

fn render_cir_metrics(writer: &mut PromWriter<'_>, instance: &str, cir: &CirMetrics) {
    writer.gauge(
        "ours_rp_cir_version",
        "CIR version",
        &[label("instance", instance)],
        cir.version as f64,
    );
    writer.gauge(
        "ours_rp_cir_objects",
        "CIR object count",
        &[label("instance", instance)],
        cir.objects as f64,
    );
    for (source, count) in [("fresh", cir.fresh_objects), ("cached", cir.cached_objects)] {
        writer.gauge(
            "ours_rp_cir_objects_by_source",
            "CIR object count by validation input source",
            &[label("instance", instance), label("source", source)],
            count as f64,
        );
    }
    writer.gauge(
        "ours_rp_cir_trust_anchors",
        "CIR trust anchor count",
        &[label("instance", instance)],
        cir.trust_anchors as f64,
    );
    writer.gauge(
        "ours_rp_cir_rejected_objects",
        "CIR rejected object count",
        &[label("instance", instance)],
        cir.rejected_objects as f64,
    );
    for (source, count) in [
        ("fresh", cir.fresh_rejected_objects),
        ("cached", cir.cached_rejected_objects),
    ] {
        writer.gauge(
            "ours_rp_cir_rejected_objects_by_source",
            "CIR rejected object count by validation input source",
            &[label("instance", instance), label("source", source)],
            count as f64,
        );
    }
    for (source, digest) in [
        ("merged", cir.object_list_sha256.as_str()),
        ("fresh", cir.fresh_object_list_sha256.as_str()),
        ("cached", cir.cached_object_list_sha256.as_str()),
    ] {
        writer.gauge(
            "ours_rp_cir_object_list_digest_present",
            "CIR object list digest is present by validation input source",
            &[label("instance", instance), label("source", source)],
            if digest.len() == 64 { 1.0 } else { 0.0 },
        );
    }
    writer.gauge(
        "ours_rp_cir_reject_list_digest_present",
        "CIR reject list digest is present",
        &[label("instance", instance)],
        if cir.reject_list_sha256.len() == 64 {
            1.0
        } else {
            0.0
        },
    );
    for (source, digest) in [
        ("merged", cir.reject_list_sha256.as_str()),
        ("fresh", cir.fresh_reject_list_sha256.as_str()),
        ("cached", cir.cached_reject_list_sha256.as_str()),
    ] {
        writer.gauge(
            "ours_rp_cir_reject_list_digest_present_by_source",
            "CIR reject list digest is present by validation input source",
            &[label("instance", instance), label("source", source)],
            if digest.len() == 64 { 1.0 } else { 0.0 },
        );
    }
    for (object_type, count) in &cir.objects_by_type {
        writer.gauge(
            "ours_rp_cir_objects_by_type",
            "CIR object count by file type",
            &[
                label("instance", instance),
                label("object_type", object_type),
            ],
            *count as f64,
        );
    }
    for (source, counts) in [
        ("fresh", &cir.fresh_objects_by_type),
        ("cached", &cir.cached_objects_by_type),
    ] {
        for (object_type, count) in counts {
            writer.gauge(
                "ours_rp_cir_objects_by_source_type",
                "CIR object count by validation input source and file type",
                &[
                    label("instance", instance),
                    label("source", source),
                    label("object_type", object_type),
                ],
                *count as f64,
            );
        }
    }
    for (object_type, count) in &cir.rejected_objects_by_type {
        writer.gauge(
            "ours_rp_cir_rejected_objects_by_type",
            "CIR rejected object count by file type",
            &[
                label("instance", instance),
                label("object_type", object_type),
            ],
            *count as f64,
        );
    }
    for (source, counts) in [
        ("fresh", &cir.fresh_rejected_objects_by_type),
        ("cached", &cir.cached_rejected_objects_by_type),
    ] {
        for (object_type, count) in counts {
            writer.gauge(
                "ours_rp_cir_rejected_objects_by_source_type",
                "CIR rejected object count by validation input source and file type",
                &[
                    label("instance", instance),
                    label("source", source),
                    label("object_type", object_type),
                ],
                *count as f64,
            );
        }
    }
}

fn render_ccr_metrics(writer: &mut PromWriter<'_>, instance: &str, ccr: &CcrMetrics) {
    writer.gauge(
        "ours_rp_ccr_version",
        "CCR version",
        &[label("instance", instance)],
        ccr.version as f64,
    );
    for (state, present) in &ccr.state_present {
        writer.gauge(
            "ours_rp_ccr_state_present",
            "CCR state presence",
            &[label("instance", instance), label("state", state)],
            bool_value(*present),
        );
    }
    for (state, count) in &ccr.state_items {
        writer.gauge(
            "ours_rp_ccr_state_items",
            "CCR state item count",
            &[label("instance", instance), label("state", state)],
            *count as f64,
        );
    }
    for state in ccr.state_digests.keys() {
        writer.gauge(
            "ours_rp_ccr_state_digest_present",
            "CCR state digest presence",
            &[label("instance", instance), label("state", state)],
            1.0,
        );
    }
}

fn render_status_json(snapshot: &MetricsSnapshot) -> Result<String, String> {
    serde_json::to_string_pretty(&json!({
        "schemaVersion": 1,
        "generatedBy": "rpki_artifact_metrics",
        "instance": snapshot.instance,
        "service": snapshot.service,
        "runs": snapshot.runs,
        "latestRun": snapshot.latest_run,
        "cir": snapshot.cir,
        "ccr": snapshot.ccr,
        "rtr": snapshot.rtr,
        "topRepositoriesBySyncDuration": snapshot.top_repos_by_sync_duration,
        "topPublicationPointsByObjectCount": snapshot.top_pp_by_object_count,
        "topPublicationPointsBySyncDuration": snapshot.top_pp_by_sync_duration,
    }))
    .map_err(|e| e.to_string())
}

struct PromWriter<'a> {
    out: &'a mut String,
    emitted_headers: BTreeSet<String>,
}

#[derive(Clone, Debug)]
struct Label<'a> {
    key: &'a str,
    value: &'a str,
}

fn label<'a>(key: &'a str, value: &'a str) -> Label<'a> {
    Label { key, value }
}

impl<'a> PromWriter<'a> {
    fn new(out: &'a mut String) -> Self {
        Self {
            out,
            emitted_headers: BTreeSet::new(),
        }
    }

    fn gauge(&mut self, name: &str, help: &str, labels: &[Label<'_>], value: f64) {
        self.metric("gauge", name, help, labels, value);
    }

    fn counter(&mut self, name: &str, help: &str, labels: &[Label<'_>], value: f64) {
        self.metric("counter", name, help, labels, value);
    }

    fn metric(
        &mut self,
        metric_type: &str,
        name: &str,
        help: &str,
        labels: &[Label<'_>],
        value: f64,
    ) {
        self.header(name, help, metric_type);
        self.out.push_str(name);
        write_labels(self.out, labels);
        self.out.push(' ');
        self.out.push_str(&format_prom_value(value));
        self.out.push('\n');
    }

    fn histogram(
        &mut self,
        name: &str,
        help: &str,
        base_labels: &[Label<'_>],
        histogram: &Histogram,
    ) {
        self.header(name, help, "histogram");
        let cumulative = histogram.cumulative_counts();
        for (index, count) in cumulative.iter().enumerate() {
            let le = if index < histogram.buckets.len() {
                format_prom_value(histogram.buckets[index])
            } else {
                "+Inf".to_string()
            };
            let mut labels = base_labels.to_vec();
            labels.push(label("le", &le));
            self.out.push_str(name);
            self.out.push_str("_bucket");
            write_labels(self.out, &labels);
            self.out.push(' ');
            self.out.push_str(&count.to_string());
            self.out.push('\n');
        }
        self.out.push_str(name);
        self.out.push_str("_sum");
        write_labels(self.out, base_labels);
        self.out.push(' ');
        self.out.push_str(&format_prom_value(histogram.sum));
        self.out.push('\n');
        self.out.push_str(name);
        self.out.push_str("_count");
        write_labels(self.out, base_labels);
        self.out.push(' ');
        self.out.push_str(&histogram.count.to_string());
        self.out.push('\n');
    }

    fn header(&mut self, name: &str, help: &str, metric_type: &str) {
        if self.emitted_headers.insert(name.to_string()) {
            self.out.push_str("# HELP ");
            self.out.push_str(name);
            self.out.push(' ');
            self.out.push_str(&escape_help(help));
            self.out.push('\n');
            self.out.push_str("# TYPE ");
            self.out.push_str(name);
            self.out.push(' ');
            self.out.push_str(metric_type);
            self.out.push('\n');
        }
    }
}

fn write_labels(out: &mut String, labels: &[Label<'_>]) {
    if labels.is_empty() {
        return;
    }
    out.push('{');
    for (index, label) in labels.iter().enumerate() {
        if index > 0 {
            out.push(',');
        }
        out.push_str(label.key);
        out.push_str("=\"");
        out.push_str(&escape_label(label.value));
        out.push('"');
    }
    out.push('}');
}

fn serve_http(listen: &str, shared: Arc<RwLock<MetricsSnapshot>>) -> Result<(), String> {
    let listener = TcpListener::bind(listen).map_err(|e| format!("bind failed: {listen}: {e}"))?;
    for stream in listener.incoming() {
        match stream {
            Ok(mut stream) => {
                let snapshot = shared.read().expect("metrics lock poisoned").clone();
                if let Err(err) = handle_http_stream(&mut stream, &snapshot) {
                    eprintln!("http request failed: {err}");
                }
            }
            Err(err) => eprintln!("accept failed: {err}"),
        }
    }
    Ok(())
}

fn handle_http_stream(stream: &mut TcpStream, snapshot: &MetricsSnapshot) -> Result<(), String> {
    let mut buf = [0u8; 4096];
    let len = stream.read(&mut buf).map_err(|e| e.to_string())?;
    let req = String::from_utf8_lossy(&buf[..len]);
    let path = req
        .lines()
        .next()
        .and_then(|line| line.split_whitespace().nth(1))
        .unwrap_or("/");
    match path {
        "/metrics" => write_http_response(
            stream,
            "200 OK",
            "text/plain; version=0.0.4",
            &render_metrics(snapshot),
        ),
        "/status" => write_http_response(
            stream,
            "200 OK",
            "application/json",
            &render_status_json(snapshot)?,
        ),
        "/healthz" => write_http_response(stream, "200 OK", "text/plain", "ok\n"),
        _ => write_http_response(stream, "404 Not Found", "text/plain", "not found\n"),
    }
}

fn write_http_response(
    stream: &mut TcpStream,
    status: &str,
    content_type: &str,
    body: &str,
) -> Result<(), String> {
    let header = format!(
        "HTTP/1.1 {status}\r\nContent-Type: {content_type}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
        body.as_bytes().len()
    );
    stream
        .write_all(header.as_bytes())
        .map_err(|e| e.to_string())?;
    stream.write_all(body.as_bytes()).map_err(|e| e.to_string())
}

fn write_file(path: &Path, bytes: &[u8]) -> Result<(), String> {
    if let Some(parent) = path.parent() {
        fs::create_dir_all(parent)
            .map_err(|e| format!("create parent failed: {}: {e}", parent.display()))?;
    }
    fs::write(path, bytes).map_err(|e| format!("write failed: {}: {e}", path.display()))
}

fn extract_stage_seconds(value: Option<&Value>) -> BTreeMap<String, f64> {
    let mut out = BTreeMap::new();
    let Some(value) = value else {
        return out;
    };
    let mapping = [
        ("validation_ms", "validation"),
        ("report_build_ms", "report_build"),
        ("report_write_ms", "report_write"),
        ("ccr_build_ms", "ccr_build"),
        ("ccr_write_ms", "ccr_write"),
        ("compare_view_build_ms", "compare_view_build"),
        ("compare_view_write_ms", "compare_view_write"),
        ("cir_build_cir_ms", "cir_build"),
        ("cir_write_cir_ms", "cir_write"),
        ("cir_total_ms", "cir_total"),
        ("total_ms", "total"),
        ("repo_sync_ms_total", "repo_sync_total"),
        ("rrdp_download_ms_total", "rrdp_download_total"),
        ("rsync_download_ms_total", "rsync_download_total"),
    ];
    for (field, stage) in mapping {
        if let Some(ms) = json_u64(value, &[field]) {
            out.insert(stage.to_string(), ms as f64 / 1000.0);
        }
    }
    out
}

fn extract_count_duration_map(value: Option<&Value>) -> BTreeMap<String, CountDuration> {
    let mut out = BTreeMap::new();
    let Some(object) = value.and_then(|v| v.as_object()) else {
        return out;
    };
    for (key, value) in object {
        out.insert(
            key.clone(),
            CountDuration {
                count: json_u64(value, &["count"]).unwrap_or(0),
                duration_seconds_total: json_u64(value, &["duration_ms_total"]).unwrap_or(0) as f64
                    / 1000.0,
            },
        );
    }
    out
}

fn extract_artifact_sizes(value: Option<&Value>) -> BTreeMap<String, u64> {
    let mut out = BTreeMap::new();
    for item in value.and_then(|v| v.as_array()).into_iter().flatten() {
        let artifact = json_str(item, &["type"])
            .or_else(|| {
                json_str(item, &["path"])
                    .and_then(|path| Path::new(path).file_name().and_then(|name| name.to_str()))
            })
            .unwrap_or("unknown");
        let size = json_u64(item, &["sizeBytes"])
            .or_else(|| json_u64(item, &["size"]))
            .unwrap_or(0);
        *out.entry(artifact.to_string()).or_default() += size;
    }
    out
}

fn extract_path_sizes(value: Option<&Value>) -> BTreeMap<String, PathSize> {
    let mut out = BTreeMap::new();
    for item in value.and_then(|v| v.as_array()).into_iter().flatten() {
        let label = json_str(item, &["label"]).unwrap_or("unknown").to_string();
        out.insert(
            label,
            PathSize {
                total_size_bytes: json_u64(item, &["totalSizeBytes"]).unwrap_or(0),
                file_count: json_u64(item, &["fileCount"]).unwrap_or(0),
                dir_count: json_u64(item, &["dirCount"]).unwrap_or(0),
            },
        );
    }
    out
}

fn run_index_from_path(path: &Path) -> Option<u64> {
    path.file_name()
        .and_then(|name| name.to_str())
        .and_then(|name| name.strip_prefix("run_"))
        .and_then(|value| value.parse::<u64>().ok())
}

fn json_str<'a>(value: &'a Value, path: &[&str]) -> Option<&'a str> {
    let mut current = value;
    for key in path {
        current = current.get(*key)?;
    }
    current.as_str()
}

fn json_u64(value: &Value, path: &[&str]) -> Option<u64> {
    let mut current = value;
    for key in path {
        current = current.get(*key)?;
    }
    current.as_u64()
}

fn json_i64(value: &Value, path: &[&str]) -> Option<i64> {
    let mut current = value;
    for key in path {
        current = current.get(*key)?;
    }
    current.as_i64()
}

fn json_f64(value: &Value, path: &[&str]) -> Option<f64> {
    let mut current = value;
    for key in path {
        current = current.get(*key)?;
    }
    current
        .as_f64()
        .or_else(|| current.as_u64().map(|v| v as f64))
}

fn json_bool(value: &Value, path: &[&str]) -> Option<bool> {
    let mut current = value;
    for key in path {
        current = current.get(*key)?;
    }
    current.as_bool()
}

fn parse_rfc3339_to_unix(value: &str) -> Option<f64> {
    time::OffsetDateTime::parse(value, &time::format_description::well_known::Rfc3339)
        .ok()
        .map(|dt| dt.unix_timestamp() as f64)
}

fn unix_now_seconds() -> f64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs_f64())
        .unwrap_or(0.0)
}

fn bool_value(value: bool) -> f64 {
    if value { 1.0 } else { 0.0 }
}

fn normalize_transport(value: &str) -> String {
    let lower = value.to_ascii_lowercase();
    if lower.contains("rrdp") || lower.contains("https") {
        "rrdp".to_string()
    } else if lower.contains("rsync") {
        "rsync".to_string()
    } else {
        lower
    }
}

fn infer_transport(uri: &str) -> String {
    if uri.starts_with("http://") || uri.starts_with("https://") {
        "rrdp".to_string()
    } else if uri.starts_with("rsync://") {
        "rsync".to_string()
    } else {
        "unknown".to_string()
    }
}

fn uri_host(uri: &str) -> String {
    let without_scheme = uri.split_once("://").map(|(_, rest)| rest).unwrap_or(uri);
    without_scheme
        .split('/')
        .next()
        .filter(|s| !s.is_empty())
        .unwrap_or("unknown")
        .to_string()
}

fn object_type_from_uri(uri: &str) -> String {
    let lower = uri.to_ascii_lowercase();
    for (suffix, kind) in [
        (".mft", "manifest"),
        (".crl", "crl"),
        (".cer", "certificate"),
        (".roa", "roa"),
        (".asa", "aspa"),
        (".gbr", "gbr"),
    ] {
        if lower.ends_with(suffix) {
            return kind.to_string();
        }
    }
    "other".to_string()
}

fn short_sha256(value: &str) -> String {
    let digest = Sha256::digest(value.as_bytes());
    hex::encode(&digest[..6])
}

fn common_prefix_len(left: &str, right: &str) -> usize {
    left.bytes()
        .zip(right.bytes())
        .take_while(|(l, r)| l == r)
        .count()
}

fn format_prom_value(value: f64) -> String {
    if value.is_infinite() && value.is_sign_positive() {
        "+Inf".to_string()
    } else if value.fract() == 0.0 {
        format!("{value:.0}")
    } else {
        format!("{value:.6}")
            .trim_end_matches('0')
            .trim_end_matches('.')
            .to_string()
    }
}

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

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

#[cfg(test)]
mod tests {
    use super::*;
    use crate::ccr::model::CCR_VERSION_V0;
    use crate::ccr::{
        CcrContentInfo, CcrDigestAlgorithm, RpkiCanonicalCacheRepresentation, TrustAnchorState,
        encode_content_info,
    };
    use crate::cir::{
        CanonicalInputRepresentation, CirObject, CirRejectedObject, CirTrustAnchor, encode_cir,
        sha256,
    };
    use tempfile::TempDir;

    #[test]
    fn parse_args_accepts_once_outputs() {
        let args = parse_args(&[
            "rpki_artifact_metrics".to_string(),
            "--run-root".to_string(),
            "root".to_string(),
            "--once".to_string(),
            "--out-metrics".to_string(),
            "metrics.prom".to_string(),
            "--out-status".to_string(),
            "status.json".to_string(),
        ])
        .expect("parse");
        assert!(args.once);
        assert_eq!(args.run_root, PathBuf::from("root"));
        assert_eq!(args.out_metrics.as_deref(), Some(Path::new("metrics.prom")));
    }

    #[test]
    fn scan_fixture_exports_repo_pp_cir_and_ccr_metrics() {
        let td = TempDir::new().expect("tempdir");
        let run = td.path().join("runs/run_0001");
        fs::create_dir_all(&run).expect("create run");
        fs::write(
            run.join("run-meta.json"),
            r#"{"status":"success","run_index":1,"run_id":"run_0001","sync_mode":"snapshot","snapshot_reason":"first_run","started_at_rfc3339_utc":"2026-05-25T00:00:00Z","completed_at_rfc3339_utc":"2026-05-25T00:00:10Z"}"#,
        )
        .expect("meta");
        fs::write(
            run.join("run-summary.json"),
            r#"{"runSeq":1,"runId":"run_0001","runDir":"RUN","startedAtRfc3339Utc":"2026-05-25T00:00:00Z","finishedAtRfc3339Utc":"2026-05-25T00:00:10Z","wallMs":10000,"status":"success","exitCode":0,"processMetrics":{"userSeconds":2.5,"systemSeconds":1.5,"cpuPercent":40,"maxRssKb":1000},"stageTiming":{"validation_ms":7000,"total_ms":9000,"download_event_count":2,"download_bytes_total":1234},"reportCounts":{"vrps":3,"aspas":1,"publicationPoints":2,"warnings":0},"repoSyncStats":{"by_phase":{"rrdp_delta":{"count":2,"duration_ms_total":3000}},"by_terminal_state":{"fresh":{"count":2,"duration_ms_total":3000}}},"pathStats":[{"label":"work-db","totalSizeBytes":99,"fileCount":2,"dirCount":1}],"artifacts":[{"path":"report.json","sizeBytes":10}]}"#,
        )
        .expect("summary");
        fs::write(run.join("process-time.txt"), "time").expect("time");
        fs::write(run.join("stage-timing.json"), "{}").expect("stage");
        fs::write(
            run.join("vrps.csv"),
            "ASN,IP Prefix,Max Length,Trust Anchor,Expires\n\
             AS64496,192.0.2.0/24,24,ta,2026-05-25T01:00:00Z\n\
             AS64496,192.0.2.0/24,24,ta,2026-05-25T01:00:00Z\n\
             AS64497,2001:db8::/32,48,ta,2026-05-25T01:00:00Z\n",
        )
        .expect("vrps");
        fs::write(
            run.join("report.json"),
            r#"{"tree":{"instances_processed":2,"instances_failed":0,"warnings":[]},"vrps":[{},{},{}],"aspas":[{}],"downloads":[{"kind":"rrdp_notification","uri":"https://repo.example/notify.xml","success":true,"duration_ms":100,"bytes":111},{"kind":"rrdp_delta","uri":"https://repo.example/session/1/delta.xml","success":true,"duration_ms":200,"bytes":222}],"publication_points":[{"rsync_base_uri":"rsync://repo.example/a/","manifest_rsync_uri":"rsync://repo.example/a/a.mft","publication_point_rsync_uri":"rsync://repo.example/a/","rrdp_notification_uri":"https://repo.example/notify.xml","repo_sync_source":"rrdp","repo_sync_phase":"rrdp_delta","repo_sync_duration_ms":1000,"repo_terminal_state":"fresh","objects":[{"kind":"roa","result":"ok"},{"kind":"manifest","result":"ok"}]},{"rsync_base_uri":"rsync://repo.example/b/","manifest_rsync_uri":"rsync://repo.example/b/b.mft","publication_point_rsync_uri":"rsync://repo.example/b/","rrdp_notification_uri":"https://repo.example/notify.xml","repo_sync_source":"rrdp","repo_sync_phase":"rrdp_delta","repo_sync_duration_ms":2000,"repo_terminal_state":"fresh","objects":[{"kind":"roa","result":"ok"}]}],"repo_sync_stats":{"publication_points_total":2,"by_phase":{"rrdp_delta":{"count":2,"duration_ms_total":3000}},"by_terminal_state":{"fresh":{"count":2,"duration_ms_total":3000}}}}"#,
        )
        .expect("report");
        fs::write(run.join("input.cir"), sample_cir()).expect("cir");
        fs::write(run.join("result.ccr"), sample_ccr()).expect("ccr");

        let snapshot = scan_run_root(td.path(), "test").expect("scan");
        assert_eq!(snapshot.runs.success, 1);
        assert_eq!(snapshot.repo_stats.len(), 1);
        assert!(snapshot.repo_stats[0].sync_success);
        assert_eq!(snapshot.repo_stats[0].download_bytes, 333);
        assert_eq!(snapshot.repo_stats[0].object_counts["roa"], 2);
        assert_eq!(snapshot.repo_stats[0].object_counts["manifest"], 1);
        assert_eq!(snapshot.top_pp_by_object_count[0].object_count, 2);
        assert_eq!(snapshot.cir.as_ref().unwrap().objects, 2);
        assert_eq!(snapshot.ccr.as_ref().unwrap().state_items["tas"], 1);
        let metrics = render_metrics(&snapshot);
        assert!(metrics.contains("ours_rp_repository_info"));
        assert!(metrics.contains("ours_rp_repository_sync_success"));
        assert!(metrics.contains("ours_rp_repository_download_bytes"));
        assert!(metrics.contains("ours_rp_repository_objects_by_type"));
        assert!(metrics.contains(r#"ours_rp_repository_objects_by_type{instance="test",repo_id="#));
        assert!(metrics.contains(r#"object_type="roa"} 2"#));
        assert!(metrics.contains(r#"object_type="manifest"} 1"#));
        assert!(metrics.contains("ours_rp_large_publication_points"));
        assert!(metrics.contains("ours_rp_cir_objects"));
        assert!(metrics.contains("ours_rp_cir_objects_by_source"));
        assert!(metrics.contains("ours_rp_cir_rejected_objects_by_source"));
        assert!(metrics.contains("ours_rp_cir_objects_by_source_type"));
        assert!(metrics.contains("ours_rp_cir_rejected_objects_by_source_type"));
        assert!(metrics.contains("ours_rp_cir_object_list_digest_present"));
        assert!(metrics.contains("ours_rp_cir_reject_list_digest_present_by_source"));
        assert!(metrics.contains("ours_rp_ccr_state_items"));
        assert!(
            metrics.contains(r#"ours_rp_state_db_size_bytes{instance="test",db="work-db"} 99"#)
        );
        assert!(metrics.contains(r#"ours_rp_state_db_files{instance="test",db="work-db"} 2"#));
        assert!(metrics.contains(r#"ours_rp_state_db_dirs{instance="test",db="work-db"} 1"#));
        assert!(
            metrics.contains(r#"ours_rp_cir_objects_by_source{instance="test",source="fresh"} 2"#)
        );
        assert!(
            metrics.contains(r#"ours_rp_cir_objects_by_source{instance="test",source="cached"} 0"#)
        );
        assert!(metrics.contains(
            r#"ours_rp_cir_rejected_objects_by_source{instance="test",source="fresh"} 1"#
        ));
        assert!(metrics.contains(r#"ours_rp_vrps{instance="test",kind="total"} 3"#));
        assert!(metrics.contains(r#"ours_rp_vrps{instance="test",kind="unique"} 2"#));
        let status = render_status_json(&snapshot).expect("status");
        assert!(status.contains("topPublicationPointsByObjectCount"));
        assert!(status.contains(r#""vrpsUnique": 2"#));
        assert!(status.contains(r#""freshObjects": 2"#));
        assert!(status.contains(r#""cachedObjects": 0"#));
    }

    #[test]
    fn partial_run_does_not_become_latest_success() {
        let td = TempDir::new().expect("tempdir");
        let run = td.path().join("runs/run_0001");
        fs::create_dir_all(&run).expect("create run");
        fs::write(run.join("run-meta.json"), r#"{"status":"running"}"#).expect("meta");
        let snapshot = scan_run_root(td.path(), "test").expect("scan");
        assert_eq!(snapshot.runs.partial, 1);
        assert!(snapshot.latest_run.is_none());
    }

    #[test]
    fn scan_fixture_exports_rtr_metrics_when_env_enabled() {
        let td = TempDir::new().expect("tempdir");
        let runs = td.path().join("runs");
        fs::create_dir_all(&runs).expect("create runs");
        let rtr = td.path().join("rtr");
        fs::create_dir_all(&rtr).expect("create rtr");

        fs::write(
            rtr.join("rtr-source-20260625142349411798736.json"),
            r#"{
  "schema_version": 2,
  "generated_at": "2026-06-25T14:23:49.411798736+08:00",
  "phase": "refresh_skipped_unchanged",
  "source": { "ccr_file_size_bytes": 21763268 },
  "refresh": {
    "last_success_at": "2026-06-25T14:23:49.411053688+08:00",
    "status": "success",
    "changed": false,
    "duration_ms": 33,
    "consecutive_failures": 0
  },
  "data_quality": {
    "ccr_input": { "total": 833492, "vrp": 832054, "router_key": 0, "aspa": 1438 },
    "invalid": { "total": 0, "vrp": 0, "router_key": 0, "aspa": 0 },
    "before_slurm": { "total": 833492, "vrp": 832054, "router_key": 0, "aspa": 1438 },
    "after_slurm": { "total": 828563, "vrp": 827126, "router_key": 0, "aspa": 1437 },
    "slurm_filters": { "prefix": 1, "router_key": 0, "aspa": 1 },
    "slurm_assertions": { "prefix": 0, "router_key": 0, "aspa": 0 }
  },
  "cache": {
    "availability": "ready",
    "versions": [
      {
        "version": 2,
        "serial": 9,
        "snapshot": { "total": 828563, "vrp": 827126, "router_key": 0, "aspa": 1437 },
        "latest_delta": {
          "announced": { "total": 1, "vrp": 1, "router_key": 0, "aspa": 0 },
          "withdrawn": { "total": 1, "vrp": 1, "router_key": 0, "aspa": 0 }
        },
        "delta_window": { "length": 4 }
      }
    ]
  }
}"#,
        )
        .expect("source");

        fs::write(
            rtr.join("rtr-runtime-20260625142610957948453.json"),
            r#"{
  "schema_version": 2,
  "generated_at": "2026-06-25T14:26:10.957948453+08:00",
  "phase": "runtime_periodic",
  "service": {
    "started_at": "2026-06-24T11:51:08.599427589+08:00",
    "uptime_seconds": 95714,
    "active_connections": 0,
    "connections_by_transport": { "tcp": 0, "tls": 0, "ssh": 0 },
    "session_listeners": 0,
    "max_connections": 512,
    "connection_utilization": 0.0
  },
  "process": { "rss_mib": 1285 }
}"#,
        )
        .expect("runtime");

        fs::write(
            rtr.join("rtr-clients-20260624115123282266076.json"),
            r#"{
  "schema_version": 2,
  "generated_at": "2026-06-24T11:51:23.282266076+08:00",
  "phase": "startup",
  "service": {
    "started_at": "2026-06-24T11:51:08.599427589+08:00",
    "uptime_seconds": 14,
    "active_connections": 0,
    "connections_by_transport": { "tcp": 0, "tls": 0, "ssh": 0 },
    "session_listeners": 0,
    "max_connections": 512,
    "connection_utilization": 0.0
  }
}"#,
        )
        .expect("clients");

        let snapshot = scan_run_root_incremental_with_rtr_dir(td.path(), "test", None, Some(&rtr))
            .expect("scan");
        assert!(snapshot.rtr.enabled);
        assert_eq!(snapshot.rtr.files["source"].seen, 1);
        assert_eq!(
            snapshot
                .rtr
                .source
                .as_ref()
                .unwrap()
                .refresh_status
                .as_deref(),
            Some("success")
        );
        assert_eq!(
            snapshot.rtr.source.as_ref().unwrap().data_quality["after_slurm"]["aspa"],
            1437
        );
        assert_eq!(
            snapshot.rtr.runtime.as_ref().unwrap().process_rss_bytes,
            Some(1285 * 1024 * 1024)
        );
        assert_eq!(
            snapshot
                .rtr
                .preferred_service()
                .and_then(|service| service.max_connections),
            Some(512)
        );

        let metrics = render_metrics(&snapshot);
        assert!(metrics.contains("ours_rp_rtr_metrics_enabled"));
        assert!(metrics.contains(r#"ours_rp_rtr_cache_ready{instance="test"} 1"#));
        assert!(metrics.contains(
            r#"ours_rp_rtr_data_quality_items{instance="test",stage="after_slurm",type="aspa"} 1437"#
        ));
        assert!(metrics.contains(
            r#"ours_rp_rtr_cache_snapshot_items{instance="test",version="2",type="vrp"} 827126"#
        ));
        assert!(metrics.contains(
            r#"ours_rp_rtr_cache_delta_items{instance="test",version="2",direction="withdrawn",type="total"} 1"#
        ));
        assert!(metrics.contains(r#"ours_rp_rtr_process_rss_bytes{instance="test"} 1347420160"#));
        assert!(metrics.contains(r#"ours_rp_rtr_max_connections{instance="test"} 512"#));
        assert!(metrics.contains(r#"ours_rp_rtr_connections{instance="test",transport="ssh"} 0"#));

        let status = render_status_json(&snapshot).expect("status");
        assert!(status.contains(r#""enabled": true"#));
        assert!(status.contains(r#""processRssBytes": 1347420160"#));
    }

    #[test]
    fn rtr_bad_json_and_missing_fields_do_not_crash() {
        let td = TempDir::new().expect("tempdir");
        let runs = td.path().join("runs");
        fs::create_dir_all(&runs).expect("create runs");
        let rtr = td.path().join("rtr");
        fs::create_dir_all(&rtr).expect("create rtr");

        fs::write(rtr.join("rtr-source-20260625000000000000000.json"), "{bad").expect("bad");
        fs::write(
            rtr.join("rtr-runtime-20260625000000000000000.json"),
            r#"{"generated_at":"2026-06-25T14:26:10+08:00","process":{}}"#,
        )
        .expect("runtime");
        fs::write(
            rtr.join("rtr-clients-20260625000000000000000.json"),
            r#"{"generated_at":"2026-06-25T14:26:10+08:00"}"#,
        )
        .expect("clients");

        let snapshot = scan_run_root_incremental_with_rtr_dir(td.path(), "test", None, Some(&rtr))
            .expect("scan");
        assert!(snapshot.rtr.enabled);
        assert_eq!(snapshot.rtr.files["source"].seen, 1);
        assert!(snapshot.rtr.source.is_none());
        assert!(snapshot.rtr.runtime.is_some());
        assert!(snapshot.rtr.clients.is_some());
        assert!(!snapshot.rtr.parse_errors.is_empty());

        let metrics = render_metrics(&snapshot);
        assert!(metrics.contains("ours_rp_rtr_metrics_parse_errors_total"));
        assert!(metrics.contains(r#"ours_rp_rtr_metrics_enabled{instance="test"} 1"#));
    }

    fn sample_cir() -> Vec<u8> {
        let rejected = vec![CirRejectedObject {
            object_uri: "rsync://repo.example/a/bad.roa".to_string(),
            reason: Some("bad".to_string()),
        }];
        let cir = CanonicalInputRepresentation::new_v4(
            time::OffsetDateTime::parse(
                "2026-05-25T00:00:00Z",
                &time::format_description::well_known::Rfc3339,
            )
            .unwrap(),
            vec![
                CirObject {
                    rsync_uri: "rsync://repo.example/a/a.roa".to_string(),
                    sha256: vec![1; 32],
                },
                CirObject {
                    rsync_uri: "rsync://repo.example/a/bad.roa".to_string(),
                    sha256: vec![2; 32],
                },
            ],
            Vec::new(),
            vec![CirTrustAnchor {
                ta_rsync_uri: "rsync://repo.example/ta.cer".to_string(),
                tal_uri: "https://tal.example/tal.tal".to_string(),
                tal_bytes: b"rsync://repo.example/ta.cer\n\nAQID\n".to_vec(),
                ta_certificate_der: b"ta".to_vec(),
                ta_certificate_sha256: sha256(b"ta"),
            }],
            rejected,
            Vec::new(),
        );
        encode_cir(&cir).expect("encode cir")
    }

    fn sample_ccr() -> Vec<u8> {
        let ci = CcrContentInfo::new(RpkiCanonicalCacheRepresentation {
            version: CCR_VERSION_V0,
            hash_alg: CcrDigestAlgorithm::Sha256,
            produced_at: time::OffsetDateTime::parse(
                "2026-05-25T00:00:00Z",
                &time::format_description::well_known::Rfc3339,
            )
            .unwrap(),
            mfts: None,
            vrps: None,
            vaps: None,
            tas: Some(TrustAnchorState {
                skis: vec![vec![1; 20]],
                hash: vec![2; 32],
            }),
            rks: None,
        });
        encode_content_info(&ci).expect("encode ccr")
    }
}