cu29_export/

lib.rs

//! Log export helpers for Copper applications.
//!
//! This crate serves two related use cases:
//!
//! - Rust logreader binaries built with [`run_cli`]
//! - optional Python-facing iterators for offline analysis of structured logs,
//!   runtime lifecycle records, and app-specific CopperLists
//!
//! Python support here is intentionally offline. It reads data that Copper has
//! already recorded. That is very different from putting Python on the runtime
//! execution path, and it does not compromise realtime behavior during robot
//! execution.
//!
//! For runtime Python task prototyping, see `cu-python-task` instead.

mod fsck;
pub mod logstats;
#[doc(hidden)]
pub mod pgs;
mod runs;

#[cfg(feature = "self-describing-logs")]
pub mod catalog;
#[cfg(feature = "self-describing-logs")]
mod catalog_cli;
#[cfg(all(feature = "self-describing-logs", feature = "python"))]
mod catalog_python;
#[cfg(all(test, feature = "self-describing-logs"))]
mod catalog_tests;
#[cfg(feature = "self-describing-logs")]
mod value_export;

#[cfg(feature = "mcap")]
pub mod mcap_export;

#[cfg(feature = "mcap")]
pub mod serde_to_jsonschema;

use bincode::Decode;
use bincode::config::standard;
use bincode::decode_from_std_read;
use bincode::error::DecodeError;
use clap::{Parser, Subcommand, ValueEnum};
use cu29::UnifiedLogType;
use cu29::prelude::*;
use cu29_intern_strs::read_interned_strings;
use fsck::check;
#[cfg(feature = "mcap")]
use indicatif::{ProgressBar, ProgressDrawTarget, ProgressStyle};
use logstats::{compute_logstats, write_logstats};
use serde::Serialize;
use std::collections::BTreeMap;
use std::fmt::Write as _;
use std::fmt::{Display, Formatter};
#[cfg(feature = "mcap")]
use std::io::IsTerminal;
use std::io::Read;
use std::path::{Path, PathBuf};

#[cfg(feature = "mcap")]
pub use mcap_export::{McapExportStats, export_to_mcap, export_to_mcap_with_schemas, mcap_info};

#[cfg(feature = "mcap")]
#[allow(deprecated)]
pub use cu29::prelude::PayloadSchemas;

#[cfg(feature = "mcap")]
pub use serde_to_jsonschema::trace_type_to_jsonschema;

/// Registers the typed CopperList decoder used by the generic Python iterator.
///
/// Applications normally call this indirectly through
/// [`copperlist_iterator_unified_typed_py`].
#[cfg(feature = "python")]
pub use python::register_copperlist_python_type;

/// Creates a Python CopperList iterator for a specific CopperList tuple type.
///
/// This is intended for app-specific Python modules that know their generated
/// CopperList type at compile time. The helper registers the decoder and returns
/// an iterator object that yields Python objects built from the recorded
/// CopperLists.
#[cfg(feature = "python")]
pub fn copperlist_iterator_unified_typed_py<P>(
    unified_src_path: &str,
    py: pyo3::Python<'_>,
) -> pyo3::PyResult<pyo3::Py<pyo3::PyAny>>
where
    P: CopperListTuple + 'static,
{
    let _ = cu29::logcodec::seed_effective_config_from_log::<P>(Path::new(unified_src_path))
        .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))?;
    register_copperlist_python_type::<P>()
        .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))?;
    let iter = python::copperlist_iterator_unified(unified_src_path)?;
    pyo3::Py::new(py, iter).map(|obj| obj.into())
}

/// Creates a Python `RuntimeLifecycleRecord` iterator from a unified log.
///
/// This is useful for offline analysis scripts that need to inspect mission
/// starts, stops, faults, and related runtime events.
#[cfg(feature = "python")]
pub fn runtime_lifecycle_iterator_unified_py(
    unified_src_path: &str,
    py: pyo3::Python<'_>,
) -> pyo3::PyResult<pyo3::Py<pyo3::PyAny>> {
    let iter = python::runtime_lifecycle_iterator_unified(unified_src_path)?;
    pyo3::Py::new(py, iter).map(|obj| obj.into())
}
#[derive(Copy, Clone, PartialEq, Eq, PartialOrd, Ord, ValueEnum)]
pub enum ExportFormat {
    Json,
    Csv,
    Jsonl,
}

impl Display for ExportFormat {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        match self {
            ExportFormat::Json => write!(f, "json"),
            ExportFormat::Csv => write!(f, "csv"),
            ExportFormat::Jsonl => write!(f, "jsonl"),
        }
    }
}

/// Source of the CopperList payload decoder (experimental catalog support).
#[cfg(feature = "self-describing-logs")]
#[derive(Clone, Copy, PartialEq, Eq, ValueEnum)]
pub enum CopperListDecoder {
    Typed,
    Catalog,
}

/// Complete catalog dump or human inspection format (experimental).
#[cfg(feature = "self-describing-logs")]
#[derive(Clone, Copy, PartialEq, Eq, ValueEnum)]
pub enum CatalogFormat {
    Human,
    Ron,
    Json,
}

/// This is a generator for a main function to build a log extractor.
#[derive(Parser)]
#[command(author, version, about)]
pub struct LogReaderCli {
    /// The base path is the name with no _0 _1 et the end.
    /// for example for toto_0.copper, toto_1.copper ... the base name is toto.copper
    pub unifiedlog_base: PathBuf,

    /// Select the recorded run by its zero-based index from list-runs.
    #[arg(long, global = true)]
    pub run: Option<usize>,

    /// Colorize human output on terminals, always, or never.
    #[arg(long, global = true, value_enum, default_value_t = clap::ColorChoice::Auto)]
    pub color: clap::ColorChoice,

    #[command(subcommand)]
    pub command: Command,
}

#[derive(Subcommand)]
pub enum Command {
    /// List recorded runs announced by Instantiated lifecycle records.
    ListRuns,
    /// Inspect the embedded payload catalog.
    #[cfg(feature = "self-describing-logs")]
    Catalog {
        #[arg(short, long, value_enum, default_value = "human")]
        export_format: CatalogFormat,
    },
    /// Extract logs
    ExtractTextLog { log_index: PathBuf },
    /// Extract copperlists
    ExtractCopperlists {
        #[arg(short, long, default_value_t = ExportFormat::Json)]
        export_format: ExportFormat,
        /// Select the compiled application decoder or embedded catalog.
        #[cfg(feature = "self-describing-logs")]
        #[arg(long, value_enum, default_value = "typed")]
        decoder: CopperListDecoder,
    },
    /// Check the log and dump info about it.
    Fsck {
        #[arg(short, long, action = clap::ArgAction::Count)]
        verbose: u8,
        /// Decode and print RuntimeLifecycle events.
        #[arg(long)]
        dump_runtime_lifecycle: bool,
        /// Decode every recorded CopperList using the embedded catalog.
        #[arg(long)]
        deep: bool,
    },
    /// Export log statistics to JSON for offline DAG rendering.
    LogStats {
        /// Output JSON file path
        #[arg(short, long, default_value = "cu29_logstats.json")]
        output: PathBuf,
        /// Config override; defaults to the selected run's recorded configuration.
        #[arg(long)]
        config: Option<PathBuf>,
        /// Mission id override; defaults to the mission recorded in the log
        #[arg(long)]
        mission: Option<String>,
        /// Comma-separated Cargo features used by conditional config fragments
        #[arg(long, value_delimiter = ',')]
        features: Vec<String>,
    },
    /// Build ranked schedules from this recorded run.
    OptimizeSchedule {
        /// Scheduling requirements and allowed CPUs in RON format.
        #[arg(long)]
        contract: PathBuf,
        /// Directory for the profile, plans, prepared configs, and report.
        #[arg(short, long, default_value = "schedule")]
        output: PathBuf,
        /// Maximum number of distinct candidate plans to emit.
        #[arg(long, default_value_t = 3)]
        candidates: usize,
        /// Candidate measurement as NAME=LOG_BASE; may be repeated.
        #[arg(long = "measure", value_name = "NAME=LOG_BASE")]
        measurements: Vec<String>,
    },
    /// Export copperlists to MCAP format (requires 'mcap' feature)
    #[cfg(feature = "mcap")]
    ExportMcap {
        /// Output MCAP file path
        #[arg(short, long)]
        output: PathBuf,
        /// Force progress bar even when stderr is not a TTY
        #[arg(long)]
        progress: bool,
        /// Suppress the progress bar
        #[arg(long)]
        quiet: bool,
    },
    /// Inspect an MCAP file and dump metadata, schemas, and stats (requires 'mcap' feature)
    #[cfg(feature = "mcap")]
    McapInfo {
        /// Path to the MCAP file to inspect
        mcap_file: PathBuf,
        /// Show full schema content
        #[arg(short, long)]
        schemas: bool,
        /// Show sample messages (first N messages per channel)
        #[arg(short = 'n', long, default_value_t = 0)]
        sample_messages: usize,
    },
}

/// Standalone catalog CLI entrypoint used by cu29-logextract.
#[cfg(feature = "self-describing-logs")]
#[doc(hidden)]
pub fn run_catalog_cli() -> CuResult<()> {
    catalog_cli::run_cli()
}

fn print_runs(runs: &[runs::RecordedRun]) {
    for run in runs {
        println!(
            "Run {}: mission={} app={} instance_id={} started={} shutdown={}",
            run.index,
            run.missions.join(","),
            run.stack
                .as_ref()
                .map_or("unknown", |stack| stack.app_name.as_str()),
            run.stack.as_ref().map_or_else(
                || "unknown".to_string(),
                |stack| stack.instance_id.to_string()
            ),
            run.started_at
                .map_or_else(|| "unknown".to_string(), |time| time.to_string()),
            run.shutdown_completed,
        );
    }
}

fn write_json_pretty<T: Serialize + ?Sized>(value: &T) -> CuResult<()> {
    serde_json::to_writer_pretty(std::io::stdout(), value)
        .map_err(|e| CuError::new_with_cause("Failed to write JSON output", e))
}

fn write_json<T: Serialize + ?Sized>(value: &T) -> CuResult<()> {
    serde_json::to_writer(std::io::stdout(), value)
        .map_err(|e| CuError::new_with_cause("Failed to write JSON output", e))
}

fn build_read_logger(unifiedlog_base: &Path) -> CuResult<UnifiedLoggerRead> {
    let logger = UnifiedLoggerBuilder::new()
        .file_base_name(unifiedlog_base)
        .build()
        .map_err(|e| CuError::new_with_cause("Failed to create logger", e))?;
    match logger {
        UnifiedLogger::Read(dl) => Ok(dl),
        UnifiedLogger::Write(_) => Err(CuError::from(
            "Expected read-only unified logger in export CLI",
        )),
    }
}

/// This is a generator for a main function to build a log extractor.
/// It depends on the specific type of the CopperList payload that is determined at compile time from the configuration.
///
#[cfg(feature = "mcap")]
pub fn run_cli<P>() -> CuResult<()>
where
    P: CopperListTuple + CuPayloadRawBytes + 'static,
{
    #[cfg(feature = "python")]
    let _ = python::register_copperlist_python_type::<P>();

    run_cli_inner::<P>()
}

/// This is a generator for a main function to build a log extractor.
/// It depends on the specific type of the CopperList payload that is determined at compile time from the configuration.
#[cfg(not(feature = "mcap"))]
pub fn run_cli<P>() -> CuResult<()>
where
    P: CopperListTuple + CuPayloadRawBytes + 'static,
{
    #[cfg(feature = "python")]
    let _ = python::register_copperlist_python_type::<P>();

    run_cli_inner::<P>()
}

fn run_cli_inner<P>() -> CuResult<()>
where
    P: CopperListTuple + CuPayloadRawBytes + 'static,
{
    run_cli_with_args::<P>(LogReaderCli::parse())
}

fn run_cli_with_args<P>(args: LogReaderCli) -> CuResult<()>
where
    P: CopperListTuple + CuPayloadRawBytes + 'static,
{
    let unifiedlog_base = args.unifiedlog_base;
    #[cfg(feature = "mcap")]
    if let Command::McapInfo {
        mcap_file,
        schemas,
        sample_messages,
    } = &args.command
    {
        return mcap_info(mcap_file, *schemas, *sample_messages);
    }

    let runs = runs::discover(&unifiedlog_base)?;
    if matches!(args.command, Command::ListRuns) {
        print_runs(&runs);
        return Ok(());
    }
    let run = runs::select(&runs, args.run)?;
    #[cfg(feature = "self-describing-logs")]
    let needs_typed_config = !matches!(
        args.command,
        Command::Catalog { .. }
            | Command::ExtractCopperlists {
                decoder: CopperListDecoder::Catalog,
                ..
            }
            | Command::Fsck { deep: true, .. }
    );
    #[cfg(not(feature = "self-describing-logs"))]
    let needs_typed_config = true;
    if needs_typed_config && let Some(config) = &run.config {
        cu29::logcodec::set_effective_config_ron::<P>(config);
    }
    let mut dl = run.reader(&unifiedlog_base)?;

    match args.command {
        Command::ListRuns => unreachable!("run listing handled before selection"),
        Command::ExtractTextLog { log_index } => {
            let reader = dl.stream(UnifiedLogType::StructuredLogLine);
            textlog_dump(reader, &log_index)?;
        }
        #[cfg(feature = "self-describing-logs")]
        Command::Catalog { export_format } => {
            catalog_cli::dump_catalog(
                run,
                &unifiedlog_base,
                export_format,
                args.color,
                &mut std::io::stdout().lock(),
            )?;
        }
        Command::ExtractCopperlists {
            export_format,
            #[cfg(feature = "self-describing-logs")]
            decoder,
        } => {
            #[cfg(feature = "self-describing-logs")]
            if decoder == CopperListDecoder::Catalog {
                return catalog_cli::extract(
                    &unifiedlog_base,
                    Some(run.index),
                    export_format,
                    &mut std::io::stdout().lock(),
                );
            }
            eprintln!("Extracting copperlists with format: {export_format}");
            let mut reader = dl.stream(UnifiedLogType::CopperList);
            let iter = copperlists_reader::<P>(&mut reader);

            match export_format {
                ExportFormat::Json => {
                    for entry in iter {
                        write_json_pretty(&entry)?;
                    }
                }
                ExportFormat::Jsonl => {
                    for entry in iter {
                        write_json(&entry)?;
                        println!();
                    }
                }
                ExportFormat::Csv => {
                    let mut first = true;
                    for origin in P::get_all_task_ids() {
                        if !first {
                            print!(", ");
                        } else {
                            print!("id, ");
                        }
                        print!("{origin}_time, {origin}_tov, {origin},");
                        first = false;
                    }
                    println!();
                    for entry in iter {
                        let mut first = true;
                        for msg in entry.cumsgs() {
                            if let Some(payload) = msg.payload() {
                                if !first {
                                    print!(", ");
                                } else {
                                    print!("{}, ", entry.id);
                                }
                                let metadata = msg.metadata();
                                print!("{}, {}, ", metadata.process_time(), msg.tov());
                                write_json(payload)?; // TODO: escape for CSV
                                first = false;
                            }
                        }
                        println!();
                    }
                }
            }
        }
        Command::Fsck {
            verbose,
            dump_runtime_lifecycle,
            deep,
        } => {
            if deep {
                #[cfg(feature = "self-describing-logs")]
                return catalog_cli::deep_check(
                    run,
                    &unifiedlog_base,
                    verbose,
                    dump_runtime_lifecycle,
                );
                #[cfg(not(feature = "self-describing-logs"))]
                return Err("Deep validation requires the self-describing-logs feature".into());
            }
            check::<P>(&mut dl, verbose, dump_runtime_lifecycle)?;
        }
        Command::LogStats {
            output,
            config,
            mission,
            features,
        } => {
            run_logstats::<P>(run, dl, output, config, mission, &features)?;
        }
        Command::OptimizeSchedule {
            contract,
            output,
            candidates,
            measurements,
        } => {
            run_schedule_optimizer::<P>(run, dl, &contract, &output, candidates, &measurements)?;
        }
        #[cfg(feature = "mcap")]
        Command::ExportMcap {
            output,
            progress,
            quiet,
        } => {
            println!("Exporting copperlists to MCAP format: {}", output.display());

            let show_progress = should_show_progress(progress, quiet);
            let total_bytes = if show_progress {
                Some(run.copperlist_bytes)
            } else {
                None
            };

            let reader = dl.stream(UnifiedLogType::CopperList);

            // Export to MCAP with schemas derived from generated output metadata.
            let stats = if let Some(total_bytes) = total_bytes {
                let progress_bar = make_progress_bar(total_bytes);
                let reader = ProgressReader::new(reader, progress_bar.clone());
                let result = export_to_mcap_impl::<P>(reader, &output);
                progress_bar.finish_and_clear();
                result?
            } else {
                export_to_mcap_impl::<P>(reader, &output)?
            };
            println!("{stats}");
        }
        #[cfg(feature = "mcap")]
        Command::McapInfo {
            mcap_file,
            schemas,
            sample_messages,
        } => {
            mcap_info(&mcap_file, schemas, sample_messages)?;
        }
    }

    Ok(())
}

fn run_logstats<P>(
    run: &runs::RecordedRun,
    dl: runs::RunReader,
    output: PathBuf,
    config: Option<PathBuf>,
    mission: Option<String>,
    features: &[String],
) -> CuResult<()>
where
    P: CopperListTuple + CuPayloadRawBytes,
{
    let cfg = if config.is_none() && run.config.is_some() {
        CuConfig::deserialize_ron(run.config.as_deref().unwrap()).map_err(|e| {
            CuError::new_with_cause("Failed to read the selected run's configuration", e)
        })?
    } else {
        let config = config.unwrap_or_else(|| PathBuf::from("copperconfig.ron"));
        let config_path = config
            .to_str()
            .ok_or_else(|| CuError::from("Config path is not valid UTF-8"))?;
        let feature_refs = features.iter().map(String::as_str).collect::<Vec<_>>();
        cu29::config::read_configuration_with_features(config_path, &feature_refs)
            .map_err(|e| CuError::new_with_cause("Failed to read configuration", e))?
    };
    let mission = resolve_logstats_mission(&cfg, mission, run.missions.first().cloned())?;
    let reader = dl.stream(UnifiedLogType::CopperList);
    let stats = compute_logstats::<P>(reader, &cfg, mission.as_deref())?;
    write_logstats(&stats, &output)
}

fn run_schedule_optimizer<P>(
    run: &runs::RecordedRun,
    dl: runs::RunReader,
    contract_path: &Path,
    output: &Path,
    candidate_count: usize,
    measurement_args: &[String],
) -> CuResult<()>
where
    P: CopperListTuple + CuPayloadRawBytes,
{
    let config_ron = run.config.as_deref().ok_or_else(|| {
        CuError::from(
            "The selected run has no recorded configuration; profile-guided scheduling requires a runtime lifecycle record",
        )
    })?;
    let config = CuConfig::deserialize_ron(config_ron).map_err(|error| {
        CuError::new_with_cause("Failed to read the recorded configuration", error)
    })?;
    let mission = resolve_schedule_mission(&config, &run.missions)?;
    let mission_id = mission
        .as_deref()
        .unwrap_or(cu29::config::DEFAULT_MISSION_ID);
    let contract = cu29::planner::CuContract::read(contract_path)?;
    let profile = pgs::compute_profile::<P>(
        dl.stream(UnifiedLogType::CopperList),
        &config,
        mission.as_deref(),
        &contract,
    )?;
    let candidates = cu29::planner::propose(&cu29::planner::ProposeRequest {
        config: &config,
        mission: mission_id,
        contract: &contract,
        profile: &profile,
        candidates: candidate_count,
    })?;

    std::fs::create_dir_all(output).map_err(|error| {
        CuError::new_with_cause("Could not create schedule output directory", error)
    })?;
    contract.write(&output.join("contract.ron"))?;
    profile.write(&output.join("profile.ron"))?;

    let mut predictions = BTreeMap::new();
    for (index, candidate) in candidates.iter().enumerate() {
        let name = format!("plan-{}", index + 1);
        candidate
            .plan
            .write(&output.join(format!("{name}.plan.ron")))?;
        let mut prepared = config.clone();
        cu29::planner::ExplicitSchedule::new(candidate.plan.clone()).apply(&mut prepared)?;
        write_text(
            &output.join(format!("{name}.config.ron")),
            prepared.serialize_ron()?,
            "candidate configuration",
        )?;
        predictions.insert(name, candidate.prediction.clone());
    }
    let predictions_ron =
        ron::ser::to_string_pretty(&predictions, ron::ser::PrettyConfig::default()).map_err(
            |error| CuError::new_with_cause("Could not serialize schedule predictions", error),
        )?;
    write_text(
        &output.join("predictions.ron"),
        predictions_ron,
        "schedule predictions",
    )?;

    let mut report = prediction_report(&candidates, mission_id);
    if !measurement_args.is_empty() {
        let mut measured = BTreeMap::from([("baseline".to_string(), profile)]);
        for argument in measurement_args {
            let (name, log_base) = parse_measurement(argument)?;
            if measured.contains_key(name) {
                return Err(CuError::from(format!(
                    "Schedule measurement name '{name}' is repeated"
                )));
            }
            let measured_runs = runs::discover(log_base)?;
            let measured_run = runs::select(&measured_runs, None)?;
            let measured_config_ron = measured_run.config.as_deref().ok_or_else(|| {
                CuError::from(format!(
                    "Measurement '{name}' has no recorded configuration"
                ))
            })?;
            let measured_config =
                CuConfig::deserialize_ron(measured_config_ron).map_err(|error| {
                    CuError::from(format!(
                        "Failed to read measurement '{name}' configuration: {error}"
                    ))
                })?;
            let measured_mission =
                resolve_schedule_mission(&measured_config, &measured_run.missions)?;
            let measured_reader = measured_run.reader(log_base)?;
            let measured_profile = pgs::compute_profile::<P>(
                measured_reader.stream(UnifiedLogType::CopperList),
                &measured_config,
                measured_mission.as_deref(),
                &contract,
            )?;
            measured.insert(name.to_string(), measured_profile);
        }
        let score = cu29::planner::CuScoreTable::new(&contract, &predictions, &measured)?;
        write_text(
            &output.join("measurements.ron"),
            score.serialize_ron()?,
            "schedule measurements",
        )?;
        writeln!(report, "\nMeasured ranking:\n{score}")
            .map_err(|_| CuError::from("Could not format schedule report"))?;
        if let Some(best) = score.best() {
            let selected = output.join("selected.plan.ron");
            if best.candidate == "baseline" {
                cu29::planner::CuPlan::from_config(&config)?.write(&selected)?;
            } else if let Some(index) = candidate_index(&best.candidate) {
                let candidate = candidates.get(index).ok_or_else(|| {
                    CuError::from(format!(
                        "Measurement '{}' does not name an emitted candidate",
                        best.candidate
                    ))
                })?;
                candidate.plan.write(&selected)?;
            }
        }
    }
    write_text(&output.join("report.txt"), &report, "schedule report")?;
    print!("{report}");
    println!("Artifacts: {}", output.display());
    Ok(())
}

fn write_text(path: &Path, contents: impl AsRef<[u8]>, kind: &str) -> CuResult<()> {
    std::fs::write(path, contents)
        .map_err(|error| CuError::from(format!("Could not write {kind}: {error}")))
}

fn parse_measurement(argument: &str) -> CuResult<(&str, &Path)> {
    let (name, path) = argument.split_once('=').ok_or_else(|| {
        CuError::from(format!(
            "Invalid measurement '{argument}'; expected NAME=LOG_BASE"
        ))
    })?;
    if name.is_empty() || path.is_empty() || name == "baseline" {
        return Err(CuError::from(format!(
            "Invalid measurement '{argument}'; use a non-empty name other than 'baseline'"
        )));
    }
    Ok((name, Path::new(path)))
}

fn candidate_index(name: &str) -> Option<usize> {
    name.split('/')
        .next()?
        .strip_prefix("plan-")?
        .parse::<usize>()
        .ok()?
        .checked_sub(1)
}

fn prediction_report(candidates: &[cu29::planner::CuCandidate], mission_id: &str) -> String {
    let mut report = String::from("Predicted schedule ranking\n");
    for (index, candidate) in candidates.iter().enumerate() {
        let mission = candidate.plan.missions.get(mission_id);
        let shape = mission.map_or_else(
            || "unknown".to_string(),
            |mission| {
                format!(
                    "{} CopperList(s)/cycle, {} in flight, {} worker(s)",
                    mission.copperlists_per_cycle,
                    mission.max_in_flight,
                    mission.workers.len()
                )
            },
        );
        let _ = writeln!(
            report,
            "{}. plan-{} score {:?} ({shape})",
            index + 1,
            index + 1,
            candidate.prediction.score
        );
        for (chain, prediction) in &candidate.prediction.chains {
            let _ = writeln!(
                report,
                "   {chain}: {:.2} ms / {:.2} ms ({:.1}%)",
                prediction.latency_ns as f64 / 1e6,
                prediction.deadline_ns as f64 / 1e6,
                prediction.ratio * 100.0
            );
        }
        for (source, rate) in &candidate.prediction.sources {
            let _ = writeln!(report, "   {source}: {:.1}% delivered", rate * 100.0);
        }
        for (worker, prediction) in &candidate.prediction.workers {
            let response = prediction.response_ns.map_or_else(
                || "unbounded".to_string(),
                |ns| format!("{:.2} ms", ns as f64 / 1e6),
            );
            let _ = writeln!(
                report,
                "   {worker}: CPU {}, {:.1}% load, {:.1}% rate, {response} response",
                prediction.cpu,
                prediction.load * 100.0,
                prediction.rate * 100.0,
            );
        }
    }
    report
}

fn resolve_schedule_mission(config: &CuConfig, logged: &[String]) -> CuResult<Option<String>> {
    if logged.len() > 1 {
        return Err(CuError::from(format!(
            "Profile-guided scheduling needs a run containing one mission; this run contains {}",
            logged.join(", ")
        )));
    }
    resolve_logstats_mission(config, None, logged.first().cloned())
}

fn resolve_logstats_mission(
    config: &CuConfig,
    requested: Option<String>,
    logged: Option<String>,
) -> CuResult<Option<String>> {
    let mission = requested.or(logged);
    let cu29::config::ConfigGraphs::Missions(graphs) = &config.graphs else {
        return Ok(None);
    };
    if let Some(mission) = mission {
        if !graphs.contains_key(&mission) {
            return Err(CuError::from(format!(
                "Mission '{mission}' is absent from the statistics configuration"
            )));
        }
        return Ok(Some(mission));
    }
    if graphs.contains_key("default") {
        return Ok(Some("default".to_string()));
    }
    Ok(graphs.keys().min().cloned())
}

/// Helper function for MCAP export.
///
/// Uses generated per-slot payload type metadata to derive schemas.
#[cfg(feature = "mcap")]
fn export_to_mcap_impl<P>(src: impl Read, output: &Path) -> CuResult<McapExportStats>
where
    P: CopperListTuple,
{
    mcap_export::export_to_mcap::<P, _>(src, output)
}

#[cfg(feature = "mcap")]
struct ProgressReader<R> {
    inner: R,
    progress: ProgressBar,
}

#[cfg(feature = "mcap")]
impl<R> ProgressReader<R> {
    fn new(inner: R, progress: ProgressBar) -> Self {
        Self { inner, progress }
    }
}

#[cfg(feature = "mcap")]
impl<R: Read> Read for ProgressReader<R> {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        let read = self.inner.read(buf)?;
        if read > 0 {
            self.progress.inc(read as u64);
        }
        Ok(read)
    }
}

#[cfg(feature = "mcap")]
fn make_progress_bar(total_bytes: u64) -> ProgressBar {
    let progress_bar = ProgressBar::new(total_bytes);
    progress_bar.set_draw_target(ProgressDrawTarget::stderr_with_hz(5));

    let style = ProgressStyle::with_template(
        "[{elapsed_precise}] {bar:40} {bytes}/{total_bytes} ({bytes_per_sec}, ETA {eta})",
    )
    .unwrap_or_else(|_| ProgressStyle::default_bar());

    progress_bar.set_style(style.progress_chars("=>-"));
    progress_bar
}

#[cfg(feature = "mcap")]
fn should_show_progress(force_progress: bool, quiet: bool) -> bool {
    !quiet && (force_progress || std::io::stderr().is_terminal())
}

fn read_next_entry<T: Decode<()>>(src: &mut impl Read) -> Option<T> {
    let entry = decode_from_std_read::<T, _, _>(src, standard());
    match entry {
        Ok(entry) => Some(entry),
        Err(DecodeError::UnexpectedEnd { .. }) => None,
        Err(DecodeError::Io { inner, additional }) => {
            if inner.kind() == std::io::ErrorKind::UnexpectedEof {
                None
            } else {
                println!("Error {inner:?} additional:{additional}");
                None
            }
        }
        Err(e) => {
            println!("Error {e:?}");
            None
        }
    }
}

/// Reads received archive provenance, inclusive source gaps and verified recovery points.
pub fn stream_continuity_reader(
    mut src: impl Read,
) -> impl Iterator<Item = cu29::continuity::StreamContinuityRecord> {
    std::iter::from_fn(move || read_next_entry(&mut src))
}

/// Extracts the copper lists from a binary representation.
/// P is the Payload determined by the configuration of the application.
pub fn copperlists_reader<P: CopperListTuple>(
    mut src: impl Read,
) -> impl Iterator<Item = CopperList<P>> {
    std::iter::from_fn(move || read_next_entry::<CopperList<P>>(&mut src))
}

/// Extracts the keyframes from the log.
pub fn keyframes_reader(mut src: impl Read) -> impl Iterator<Item = KeyFrame> {
    std::iter::from_fn(move || read_next_entry::<KeyFrame>(&mut src))
}

/// Extracts the runtime lifecycle records from the log.
pub fn runtime_lifecycle_reader(
    mut src: impl Read,
) -> impl Iterator<Item = RuntimeLifecycleRecord> {
    std::iter::from_fn(move || read_next_entry::<RuntimeLifecycleRecord>(&mut src))
}

/// Returns the first mission announced by the runtime lifecycle section, if any.
pub fn unified_log_mission(unifiedlog_base: &Path) -> CuResult<Option<String>> {
    let mut dl = build_read_logger(unifiedlog_base)?;
    let Some(metadata) = dl.application_metadata()? else {
        return Ok(None);
    };
    loop {
        let header = dl.raw_skip_section()?;
        if header.entry_type == UnifiedLogType::LastEntry {
            return Ok(None);
        }
        if header.context.run_id != 0 {
            return metadata
                .missions
                .get(header.context.mission_index as usize)
                .cloned()
                .map(Some)
                .ok_or(CuError::from("Invalid mission index"));
        }
    }
}

/// Ensures the unified log was recorded for the expected mission.
pub fn assert_unified_log_mission(unifiedlog_base: &Path, expected_mission: &str) -> CuResult<()> {
    match unified_log_mission(unifiedlog_base)? {
        Some(actual_mission) if actual_mission == expected_mission => Ok(()),
        Some(actual_mission) => Err(CuError::from(format!(
            "Mission mismatch: expected '{expected_mission}', found '{actual_mission}'"
        ))),
        None => Err(CuError::from(format!(
            "No MissionStarted runtime lifecycle event found while validating expected mission '{expected_mission}'"
        ))),
    }
}

pub fn structlog_reader(mut src: impl Read) -> impl Iterator<Item = CuResult<CuLogEntry>> {
    std::iter::from_fn(move || {
        let entry = decode_from_std_read::<CuLogEntry, _, _>(&mut src, standard());

        match entry {
            Err(DecodeError::UnexpectedEnd { .. }) => None,
            Err(DecodeError::Io {
                inner,
                additional: _,
            }) => {
                if inner.kind() == std::io::ErrorKind::UnexpectedEof {
                    None
                } else {
                    Some(Err(CuError::new_with_cause("Error reading log", inner)))
                }
            }
            Err(e) => Some(Err(CuError::new_with_cause("Error reading log", e))),
            Ok(entry) => {
                if entry.msg_index == 0 {
                    None
                } else {
                    Some(Ok(entry))
                }
            }
        }
    })
}

/// Full dump of the copper structured log from its binary representation.
/// This rebuilds a textual log.
/// src: the source of the log data
/// index: the path to the index file (containing the interned strings constructed at build time)
pub fn textlog_dump(src: impl Read, index: &Path) -> CuResult<()> {
    let all_strings = read_interned_strings(index).map_err(|e| {
        CuError::new_with_cause(
            "Failed to read interned strings from index",
            std::io::Error::other(e),
        )
    })?;

    for result in structlog_reader(src) {
        let entry = result?;
        match rebuild_logline(&all_strings, &entry) {
            Ok(line) => println!("{line}"),
            Err(e) => println!("Failed to rebuild log line: {e:?}"),
        }
    }

    Ok(())
}

// Only compiled for users opting into the Python interface.
#[cfg(feature = "python")]
mod python {
    use bincode::config::standard;
    use bincode::decode_from_std_read;
    use bincode::error::DecodeError;
    use cu29::bevy_reflect::{PartialReflect, ReflectRef, VariantType};
    use cu29::prelude::*;
    use cu29_intern_strs::read_interned_strings;
    use pyo3::exceptions::{PyIOError, PyRuntimeError};
    use pyo3::prelude::*;
    use pyo3::types::{PyDelta, PyDict, PyList};
    use std::io::Read;
    use std::path::Path;
    use std::sync::OnceLock;

    type CopperListDecodeFn =
        for<'py> fn(&mut Box<dyn Read + Send + Sync>, Python<'py>) -> Option<PyResult<Py<PyAny>>>;
    static COPPERLIST_DECODER: OnceLock<CopperListDecodeFn> = OnceLock::new();

    /// Iterator over structured Copper log entries.
    #[pyclass]
    pub struct PyLogIterator {
        reader: Box<dyn Read + Send + Sync>,
    }

    /// Iterator over application-specific CopperLists decoded into Python values.
    #[pyclass]
    pub struct PyCopperListIterator {
        reader: Box<dyn Read + Send + Sync>,
        decode_next: CopperListDecodeFn,
    }

    /// Iterator over runtime lifecycle records stored in a unified log.
    #[pyclass]
    pub struct PyRuntimeLifecycleIterator {
        reader: Box<dyn Read + Send + Sync>,
    }

    /// Helper wrapper used when reflected unit-bearing values are exposed to Python.
    #[pyclass(get_all)]
    pub struct PyUnitValue {
        pub value: f64,
        pub unit: String,
    }

    /// Register the Python decoder for one concrete CopperList tuple type.
    ///
    /// App-specific extension modules call this once before constructing a
    /// `PyCopperListIterator`.
    pub fn register_copperlist_python_type<P>() -> CuResult<()>
    where
        P: CopperListTuple,
    {
        if COPPERLIST_DECODER.get().is_none() {
            COPPERLIST_DECODER
                .set(decode_next_copperlist::<P>)
                .map_err(|_| CuError::from("Failed to register CopperList Python decoder"))?;
        }
        Ok(())
    }
    #[pymethods]
    impl PyLogIterator {
        fn __iter__(slf: PyRefMut<Self>) -> PyRefMut<Self> {
            slf
        }

        fn __next__(mut slf: PyRefMut<Self>) -> Option<PyResult<PyCuLogEntry>> {
            match decode_from_std_read::<CuLogEntry, _, _>(&mut slf.reader, standard()) {
                Ok(entry) => {
                    if entry.msg_index == 0 {
                        None
                    } else {
                        Some(Ok(PyCuLogEntry { inner: entry }))
                    }
                }
                Err(DecodeError::UnexpectedEnd { .. }) => None,
                Err(DecodeError::Io { inner, .. })
                    if inner.kind() == std::io::ErrorKind::UnexpectedEof =>
                {
                    None
                }
                Err(e) => Some(Err(PyIOError::new_err(e.to_string()))),
            }
        }
    }

    #[pymethods]
    impl PyCopperListIterator {
        fn __iter__(slf: PyRefMut<Self>) -> PyRefMut<Self> {
            slf
        }

        fn __next__(mut slf: PyRefMut<Self>, py: Python<'_>) -> Option<PyResult<Py<PyAny>>> {
            (slf.decode_next)(&mut slf.reader, py)
        }
    }

    #[pymethods]
    impl PyRuntimeLifecycleIterator {
        fn __iter__(slf: PyRefMut<Self>) -> PyRefMut<Self> {
            slf
        }

        fn __next__(mut slf: PyRefMut<Self>, py: Python<'_>) -> Option<PyResult<Py<PyAny>>> {
            let entry = super::read_next_entry::<RuntimeLifecycleRecord>(&mut slf.reader)?;
            Some(runtime_lifecycle_record_to_py(&entry, py))
        }
    }
    /// Create an iterator over structured log entries from a bare structured log file.
    ///
    /// This is the non-unified-log path used by standalone structured log setups.
    /// The function returns the iterator together with the interned string table
    /// needed to format each message.
    #[pyfunction]
    pub fn struct_log_iterator_bare(
        bare_struct_src_path: &str,
        index_path: &str,
    ) -> PyResult<(PyLogIterator, Vec<String>)> {
        let file = std::fs::File::open(bare_struct_src_path)
            .map_err(|e| PyIOError::new_err(e.to_string()))?;
        let all_strings = read_interned_strings(Path::new(index_path))
            .map_err(|e| PyIOError::new_err(e.to_string()))?;
        Ok((
            PyLogIterator {
                reader: Box::new(file),
            },
            all_strings,
        ))
    }
    /// Create an iterator over structured log entries from a unified log file.
    ///
    /// The function returns the iterator together with the interned string table
    /// needed to rebuild the text messages.
    #[pyfunction]
    pub fn struct_log_iterator_unified(
        unified_src_path: &str,
        index_path: &str,
    ) -> PyResult<(PyLogIterator, Vec<String>)> {
        let all_strings = read_interned_strings(Path::new(index_path))
            .map_err(|e| PyIOError::new_err(e.to_string()))?;

        let logger = UnifiedLoggerBuilder::new()
            .file_base_name(Path::new(unified_src_path))
            .build()
            .map_err(|e| PyIOError::new_err(e.to_string()))?;
        let dl = match logger {
            UnifiedLogger::Read(dl) => dl,
            UnifiedLogger::Write(_) => {
                return Err(PyIOError::new_err(
                    "Expected read-only unified logger for Python export",
                ));
            }
        };

        let reader = UnifiedLoggerIOReader::new(dl, UnifiedLogType::StructuredLogLine);
        Ok((
            PyLogIterator {
                reader: Box::new(reader),
            },
            all_strings,
        ))
    }

    /// Create an iterator over CopperLists from a unified log file.
    ///
    /// The concrete CopperList tuple type must be registered from Rust first with
    /// `register_copperlist_python_type::<P>()`.
    #[pyfunction]
    pub fn copperlist_iterator_unified(unified_src_path: &str) -> PyResult<PyCopperListIterator> {
        let decode_next = *COPPERLIST_DECODER.get().ok_or_else(|| {
            PyRuntimeError::new_err(
                "CopperList decoder is not registered. \
Call register_copperlist_python_type::<P>() from Rust before using this function.",
            )
        })?;

        let logger = UnifiedLoggerBuilder::new()
            .file_base_name(Path::new(unified_src_path))
            .build()
            .map_err(|e| PyIOError::new_err(e.to_string()))?;
        let dl = match logger {
            UnifiedLogger::Read(dl) => dl,
            UnifiedLogger::Write(_) => {
                return Err(PyIOError::new_err(
                    "Expected read-only unified logger for Python export",
                ));
            }
        };

        let reader = UnifiedLoggerIOReader::new(dl, UnifiedLogType::CopperList);
        Ok(PyCopperListIterator {
            reader: Box::new(reader),
            decode_next,
        })
    }

    /// Create an iterator over runtime lifecycle records from a unified log file.
    #[pyfunction]
    pub fn runtime_lifecycle_iterator_unified(
        unified_src_path: &str,
    ) -> PyResult<PyRuntimeLifecycleIterator> {
        let logger = UnifiedLoggerBuilder::new()
            .file_base_name(Path::new(unified_src_path))
            .build()
            .map_err(|e| PyIOError::new_err(e.to_string()))?;
        let dl = match logger {
            UnifiedLogger::Read(dl) => dl,
            UnifiedLogger::Write(_) => {
                return Err(PyIOError::new_err(
                    "Expected read-only unified logger for Python export",
                ));
            }
        };

        let reader = UnifiedLoggerIOReader::new(dl, UnifiedLogType::RuntimeLifecycle);
        Ok(PyRuntimeLifecycleIterator {
            reader: Box::new(reader),
        })
    }
    /// Python wrapper for [`CuLogEntry`].
    #[pyclass]
    pub struct PyCuLogEntry {
        pub inner: CuLogEntry,
    }

    #[pymethods]
    impl PyCuLogEntry {
        /// Return the timestamp of the log entry as a `datetime.timedelta`.
        pub fn ts<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyDelta>> {
            let nanoseconds: u64 = self.inner.time.into();

            // Convert nanoseconds to seconds and microseconds
            let days = (nanoseconds / 86_400_000_000_000) as i32;
            let seconds = (nanoseconds / 1_000_000_000) as i32;
            let microseconds = ((nanoseconds % 1_000_000_000) / 1_000) as i32;

            PyDelta::new(py, days, seconds, microseconds, false)
        }

        /// Return the index of the message format string in the interned string table.
        pub fn msg_index(&self) -> u32 {
            self.inner.msg_index
        }

        /// Return the CopperList id captured for this log entry, if any.
        pub fn culistid(&self) -> Option<u64> {
            self.inner.origin.culistid
        }

        /// Return the runtime component id captured for this log entry, if any.
        pub fn component_id(&self) -> Option<u32> {
            self.inner.origin.component_id
        }

        /// Return the task index captured for this log entry, if any.
        pub fn task_index(&self) -> Option<u32> {
            self.inner.origin.task_index
        }

        /// Return the indexes of the parameter names in the interned string table.
        pub fn paramname_indexes(&self) -> Vec<u32> {
            self.inner.paramname_indexes.iter().copied().collect()
        }

        /// Return the structured parameters carried by this log line.
        pub fn params(&self, py: Python<'_>) -> PyResult<Vec<Py<PyAny>>> {
            self.inner
                .params
                .iter()
                .map(|value| value_to_py(value, py))
                .collect()
        }
    }

    /// This needs to match the name of the generated '.so'
    #[pymodule(name = "libcu29_export")]
    fn cu29_export(m: &Bound<'_, PyModule>) -> PyResult<()> {
        #[cfg(feature = "self-describing-logs")]
        crate::catalog_python::add_functions(m)?;
        m.add_class::<PyCuLogEntry>()?;
        m.add_class::<PyLogIterator>()?;
        m.add_class::<PyCopperListIterator>()?;
        m.add_class::<PyRuntimeLifecycleIterator>()?;
        m.add_class::<PyUnitValue>()?;
        m.add_function(wrap_pyfunction!(struct_log_iterator_bare, m)?)?;
        m.add_function(wrap_pyfunction!(struct_log_iterator_unified, m)?)?;
        m.add_function(wrap_pyfunction!(copperlist_iterator_unified, m)?)?;
        m.add_function(wrap_pyfunction!(runtime_lifecycle_iterator_unified, m)?)?;
        Ok(())
    }

    fn decode_next_copperlist<P>(
        reader: &mut Box<dyn Read + Send + Sync>,
        py: Python<'_>,
    ) -> Option<PyResult<Py<PyAny>>>
    where
        P: CopperListTuple,
    {
        let entry = super::read_next_entry::<CopperList<P>>(reader)?;
        Some(copperlist_to_py::<P>(&entry, py))
    }

    fn copperlist_to_py<P>(entry: &CopperList<P>, py: Python<'_>) -> PyResult<Py<PyAny>>
    where
        P: CopperListTuple,
    {
        let task_ids = P::get_all_task_ids();
        let root = PyDict::new(py);
        root.set_item("id", entry.id)?;

        let mut messages: Vec<Py<PyAny>> = Vec::new();
        for (idx, msg) in entry.cumsgs().into_iter().enumerate() {
            let message = PyDict::new(py);
            message.set_item("task_id", task_ids.get(idx).copied().unwrap_or("unknown"))?;
            message.set_item("tov", tov_to_py(msg.tov(), py)?)?;
            message.set_item("metadata", metadata_to_py(msg.metadata(), py)?)?;
            match msg.payload_reflect() {
                Some(payload) => message.set_item(
                    "payload",
                    partial_reflect_to_py(payload.as_partial_reflect(), py)?,
                )?,
                None => message.set_item("payload", py.None())?,
            }
            messages.push(dict_to_namespace(message, py)?);
        }

        root.set_item("messages", PyList::new(py, messages)?)?;
        dict_to_namespace(root, py)
    }

    fn runtime_lifecycle_record_to_py(
        entry: &RuntimeLifecycleRecord,
        py: Python<'_>,
    ) -> PyResult<Py<PyAny>> {
        let root = PyDict::new(py);
        root.set_item("timestamp_ns", entry.timestamp.as_nanos())?;
        root.set_item("event", runtime_lifecycle_event_to_py(&entry.event, py)?)?;
        dict_to_namespace(root, py)
    }

    fn runtime_lifecycle_event_to_py(
        event: &RuntimeLifecycleEvent,
        py: Python<'_>,
    ) -> PyResult<Py<PyAny>> {
        let root = PyDict::new(py);
        match event {
            RuntimeLifecycleEvent::Instantiated { config_source } => {
                root.set_item("kind", "instantiated")?;
                root.set_item("config_source", runtime_config_source_to_py(config_source))?;
            }
            RuntimeLifecycleEvent::MissionStarted => {
                root.set_item("kind", "mission_started")?;
            }
            RuntimeLifecycleEvent::MissionStopped { reason } => {
                root.set_item("kind", "mission_stopped")?;
                root.set_item("reason", format!("{reason:?}"))?;
            }
            RuntimeLifecycleEvent::LifecycleFailed { operation, error } => {
                root.set_item("kind", "lifecycle_failed")?;
                root.set_item("operation", format!("{operation:?}"))?;
                root.set_item("error", error)?;
            }
            RuntimeLifecycleEvent::Panic {
                message,
                file,
                line,
                column,
            } => {
                root.set_item("kind", "panic")?;
                root.set_item("message", message)?;
                root.set_item("file", file)?;
                root.set_item("line", line)?;
                root.set_item("column", column)?;
            }
            RuntimeLifecycleEvent::ShutdownCompleted => {
                root.set_item("kind", "shutdown_completed")?;
            }
            #[allow(unreachable_patterns)] // dependency features may enable extended records
            event => {
                if let Some(record) = event.clock_sync_record() {
                    root.set_item("kind", "clock_sync")?;
                    root.set_item("culistid", record.culistid)?;
                    let snapshot = serde_json::to_string(&record.snapshot).map_err(|error| {
                        pyo3::exceptions::PyValueError::new_err(error.to_string())
                    })?;
                    root.set_item(
                        "snapshot",
                        py.import("json")?.getattr("loads")?.call1((snapshot,))?,
                    )?;
                }
            }
        }

        dict_to_namespace(root, py)
    }

    fn runtime_config_source_to_py(source: &RuntimeLifecycleConfigSource) -> &'static str {
        match source {
            RuntimeLifecycleConfigSource::ProgrammaticOverride => "programmatic_override",
            RuntimeLifecycleConfigSource::ExternalFile => "external_file",
            RuntimeLifecycleConfigSource::BundledDefault => "bundled_default",
        }
    }

    fn metadata_to_py(metadata: &dyn CuMsgMetadataTrait, py: Python<'_>) -> PyResult<Py<PyAny>> {
        let process = metadata.process_time();
        let start: Option<CuTime> = process.start.into();
        let end: Option<CuTime> = process.end.into();

        let process_time = PyDict::new(py);
        process_time.set_item("start_ns", start.map(|t| t.as_nanos()))?;
        process_time.set_item("end_ns", end.map(|t| t.as_nanos()))?;

        let metadata_py = PyDict::new(py);
        metadata_py.set_item("process_time", dict_to_namespace(process_time, py)?)?;
        metadata_py.set_item("status_txt", metadata.status_txt().0.to_string())?;
        if let Some(origin) = metadata.origin() {
            let origin_py = PyDict::new(py);
            origin_py.set_item("subsystem_code", origin.subsystem_code)?;
            origin_py.set_item("instance_id", origin.instance_id)?;
            origin_py.set_item("cl_id", origin.cl_id)?;
            metadata_py.set_item("origin", dict_to_namespace(origin_py, py)?)?;
        } else {
            metadata_py.set_item("origin", py.None())?;
        }
        dict_to_namespace(metadata_py, py)
    }

    fn tov_to_py(tov: Tov, py: Python<'_>) -> PyResult<Py<PyAny>> {
        let tov_py = PyDict::new(py);
        match tov {
            Tov::None => {
                tov_py.set_item("kind", "none")?;
            }
            Tov::Time(t) => {
                tov_py.set_item("kind", "time")?;
                tov_py.set_item("time_ns", t.as_nanos())?;
            }
            Tov::Range(r) => {
                tov_py.set_item("kind", "range")?;
                tov_py.set_item("start_ns", r.start.as_nanos())?;
                tov_py.set_item("end_ns", r.end.as_nanos())?;
            }
        }
        dict_to_namespace(tov_py, py)
    }

    fn partial_reflect_to_py(value: &dyn PartialReflect, py: Python<'_>) -> PyResult<Py<PyAny>> {
        #[allow(unreachable_patterns)]
        match value.reflect_ref() {
            ReflectRef::Struct(s) => struct_to_py(s, py),
            ReflectRef::TupleStruct(ts) => tuple_struct_to_py(ts, py),
            ReflectRef::Tuple(t) => tuple_to_py(t, py),
            ReflectRef::List(list) => list_to_py(list, py),
            ReflectRef::Array(array) => array_to_py(array, py),
            ReflectRef::Map(map) => map_to_py(map, py),
            ReflectRef::Set(set) => set_to_py(set, py),
            ReflectRef::Enum(e) => enum_to_py(e, py),
            ReflectRef::Opaque(opaque) => opaque_to_py(opaque, py),
            _ => Ok(py.None()),
        }
    }

    fn struct_to_py(value: &dyn cu29::bevy_reflect::Struct, py: Python<'_>) -> PyResult<Py<PyAny>> {
        let dict = PyDict::new(py);
        for idx in 0..value.field_len() {
            if let Some(field) = value.field_at(idx) {
                let name = value
                    .name_at(idx)
                    .map(str::to_owned)
                    .unwrap_or_else(|| format!("field_{idx}"));
                dict.set_item(name, partial_reflect_to_py(field, py)?)?;
            }
        }

        if let Some(unit) = unit_abbrev_for_type_path(value.reflect_type_path())
            && let Some(raw_value) = dict.get_item("value")?
        {
            if let Ok(v) = raw_value.extract::<f64>() {
                let unit_value = PyUnitValue {
                    value: v,
                    unit: unit.to_string(),
                };
                return Ok(Py::new(py, unit_value)?.into());
            }
            if let Ok(v) = raw_value.extract::<f32>() {
                let unit_value = PyUnitValue {
                    value: v as f64,
                    unit: unit.to_string(),
                };
                return Ok(Py::new(py, unit_value)?.into());
            }
        }

        dict_to_namespace(dict, py)
    }

    fn tuple_struct_to_py(
        value: &dyn cu29::bevy_reflect::TupleStruct,
        py: Python<'_>,
    ) -> PyResult<Py<PyAny>> {
        let mut fields = Vec::with_capacity(value.field_len());
        for idx in 0..value.field_len() {
            if let Some(field) = value.field(idx) {
                fields.push(partial_reflect_to_py(field, py)?);
            } else {
                fields.push(py.None());
            }
        }
        Ok(PyList::new(py, fields)?.into_pyobject(py)?.into())
    }

    fn tuple_to_py(value: &dyn cu29::bevy_reflect::Tuple, py: Python<'_>) -> PyResult<Py<PyAny>> {
        let mut fields = Vec::with_capacity(value.field_len());
        for idx in 0..value.field_len() {
            if let Some(field) = value.field(idx) {
                fields.push(partial_reflect_to_py(field, py)?);
            } else {
                fields.push(py.None());
            }
        }
        Ok(PyList::new(py, fields)?.into_pyobject(py)?.into())
    }

    fn list_to_py(value: &dyn cu29::bevy_reflect::List, py: Python<'_>) -> PyResult<Py<PyAny>> {
        let mut items = Vec::with_capacity(value.len());
        for item in value.iter() {
            items.push(partial_reflect_to_py(item, py)?);
        }
        Ok(PyList::new(py, items)?.into_pyobject(py)?.into())
    }

    fn array_to_py(value: &dyn cu29::bevy_reflect::Array, py: Python<'_>) -> PyResult<Py<PyAny>> {
        let mut items = Vec::with_capacity(value.len());
        for item in value.iter() {
            items.push(partial_reflect_to_py(item, py)?);
        }
        Ok(PyList::new(py, items)?.into_pyobject(py)?.into())
    }

    fn map_to_py(value: &dyn cu29::bevy_reflect::Map, py: Python<'_>) -> PyResult<Py<PyAny>> {
        let dict = PyDict::new(py);
        for (key, val) in value.iter() {
            let key_str = reflect_key_to_string(key);
            dict.set_item(key_str, partial_reflect_to_py(val, py)?)?;
        }
        Ok(dict.into_pyobject(py)?.into())
    }

    fn set_to_py(value: &dyn cu29::bevy_reflect::Set, py: Python<'_>) -> PyResult<Py<PyAny>> {
        let mut items = Vec::with_capacity(value.len());
        for item in value.iter() {
            items.push(partial_reflect_to_py(item, py)?);
        }
        Ok(PyList::new(py, items)?.into_pyobject(py)?.into())
    }

    fn enum_to_py(value: &dyn cu29::bevy_reflect::Enum, py: Python<'_>) -> PyResult<Py<PyAny>> {
        let dict = PyDict::new(py);
        dict.set_item("variant", value.variant_name())?;

        match value.variant_type() {
            VariantType::Unit => {}
            VariantType::Tuple => {
                let mut fields = Vec::with_capacity(value.field_len());
                for idx in 0..value.field_len() {
                    if let Some(field) = value.field_at(idx) {
                        fields.push(partial_reflect_to_py(field, py)?);
                    } else {
                        fields.push(py.None());
                    }
                }
                dict.set_item("fields", PyList::new(py, fields)?)?;
            }
            VariantType::Struct => {
                let fields = PyDict::new(py);
                for idx in 0..value.field_len() {
                    if let Some(field) = value.field_at(idx) {
                        let name = value
                            .name_at(idx)
                            .map(str::to_owned)
                            .unwrap_or_else(|| format!("field_{idx}"));
                        fields.set_item(name, partial_reflect_to_py(field, py)?)?;
                    }
                }
                dict.set_item("fields", fields)?;
            }
        }

        dict_to_namespace(dict, py)
    }

    fn dict_to_namespace(dict: Bound<'_, PyDict>, py: Python<'_>) -> PyResult<Py<PyAny>> {
        let types = py.import("types")?;
        let namespace_ctor = types.getattr("SimpleNamespace")?;
        let namespace = namespace_ctor.call((), Some(&dict))?;
        Ok(namespace.into())
    }

    fn reflect_key_to_string(value: &dyn PartialReflect) -> String {
        if let Some(v) = value.try_downcast_ref::<String>() {
            return v.clone();
        }
        if let Some(v) = value.try_downcast_ref::<&'static str>() {
            return (*v).to_string();
        }
        if let Some(v) = value.try_downcast_ref::<char>() {
            return v.to_string();
        }
        if let Some(v) = value.try_downcast_ref::<bool>() {
            return v.to_string();
        }
        if let Some(v) = value.try_downcast_ref::<u64>() {
            return v.to_string();
        }
        if let Some(v) = value.try_downcast_ref::<i64>() {
            return v.to_string();
        }
        if let Some(v) = value.try_downcast_ref::<usize>() {
            return v.to_string();
        }
        if let Some(v) = value.try_downcast_ref::<isize>() {
            return v.to_string();
        }
        format!("{value:?}")
    }

    fn unit_abbrev_for_type_path(type_path: &str) -> Option<&'static str> {
        match type_path.rsplit("::").next()? {
            "Acceleration" => Some("m/s^2"),
            "Angle" => Some("rad"),
            "AngularVelocity" => Some("rad/s"),
            "ElectricPotential" => Some("V"),
            "Length" => Some("m"),
            "MagneticFluxDensity" => Some("T"),
            "Pressure" => Some("Pa"),
            "Ratio" => Some("1"),
            "ThermodynamicTemperature" => Some("K"),
            "Time" => Some("s"),
            "Velocity" => Some("m/s"),
            _ => None,
        }
    }

    fn opaque_to_py(value: &dyn PartialReflect, py: Python<'_>) -> PyResult<Py<PyAny>> {
        macro_rules! downcast_copy {
            ($ty:ty) => {
                if let Some(v) = value.try_downcast_ref::<$ty>() {
                    return Ok(v.into_pyobject(py)?.to_owned().into());
                }
            };
        }

        downcast_copy!(bool);
        downcast_copy!(u8);
        downcast_copy!(u16);
        downcast_copy!(u32);
        downcast_copy!(u64);
        downcast_copy!(u128);
        downcast_copy!(usize);
        downcast_copy!(i8);
        downcast_copy!(i16);
        downcast_copy!(i32);
        downcast_copy!(i64);
        downcast_copy!(i128);
        downcast_copy!(isize);
        downcast_copy!(f32);
        downcast_copy!(f64);
        downcast_copy!(char);

        if let Some(v) = value.try_downcast_ref::<String>() {
            return Ok(v.into_pyobject(py)?.into());
        }
        if let Some(v) = value.try_downcast_ref::<&'static str>() {
            return Ok(v.into_pyobject(py)?.into());
        }
        if let Some(v) = value.try_downcast_ref::<Vec<u8>>() {
            return Ok(v.into_pyobject(py)?.into());
        }

        let fallback = format!("{value:?}");
        Ok(fallback.into_pyobject(py)?.into())
    }
    pub(super) fn value_to_py(value: &cu29::prelude::Value, py: Python<'_>) -> PyResult<Py<PyAny>> {
        match value {
            Value::String(s) => Ok(s.into_pyobject(py)?.into()),
            Value::U64(u) => Ok(u.into_pyobject(py)?.into()),
            Value::U128(u) => Ok(u.into_pyobject(py)?.into()),
            Value::I64(i) => Ok(i.into_pyobject(py)?.into()),
            Value::I128(i) => Ok(i.into_pyobject(py)?.into()),
            Value::F64(f) => Ok(f.into_pyobject(py)?.into()),
            Value::Bool(b) => Ok(b.into_pyobject(py)?.to_owned().into()),
            Value::CuTime(t) => Ok(t.0.into_pyobject(py)?.into()),
            Value::Bytes(b) => Ok(b.into_pyobject(py)?.into()),
            Value::Char(c) => Ok(c.into_pyobject(py)?.into()),
            Value::I8(i) => Ok(i.into_pyobject(py)?.into()),
            Value::U8(u) => Ok(u.into_pyobject(py)?.into()),
            Value::I16(i) => Ok(i.into_pyobject(py)?.into()),
            Value::U16(u) => Ok(u.into_pyobject(py)?.into()),
            Value::I32(i) => Ok(i.into_pyobject(py)?.into()),
            Value::U32(u) => Ok(u.into_pyobject(py)?.into()),
            Value::Map(m) => {
                let dict = PyDict::new(py);
                for (k, v) in m.iter() {
                    dict.set_item(value_to_py(k, py)?, value_to_py(v, py)?)?;
                }
                Ok(dict.into_pyobject(py)?.into())
            }
            Value::F32(f) => Ok(f.into_pyobject(py)?.into()),
            Value::Option(o) => match o.as_ref() {
                Some(value) => value_to_py(value, py),
                None => Ok(py.None()),
            },
            Value::Unit => Ok(py.None()),
            Value::Newtype(v) => value_to_py(v, py),
            Value::Seq(s) => {
                let items: Vec<Py<PyAny>> = s
                    .iter()
                    .map(|value| value_to_py(value, py))
                    .collect::<PyResult<_>>()?;
                let list = PyList::new(py, items)?;
                Ok(list.into_pyobject(py)?.into())
            }
        }
    }

    #[cfg(test)]
    mod tests {
        use super::*;

        #[derive(Debug, Default, bincode::Encode, bincode::Decode, serde::Serialize)]
        struct EmptyMessages;

        impl ErasedCuStampedDataSet for EmptyMessages {
            fn cumsgs(&self) -> Vec<&dyn ErasedCuStampedData> {
                Vec::new()
            }
        }

        impl MatchingTasks for EmptyMessages {
            fn get_all_task_ids() -> &'static [&'static str] {
                &[]
            }
        }

        #[test]
        fn copperlist_python_export_omits_runtime_state() {
            Python::initialize();
            Python::attach(|py| {
                let list = CopperList::new(7, EmptyMessages);
                let value = copperlist_to_py(&list, py).unwrap();
                assert!(!value.bind(py).hasattr("state").unwrap());
                assert_eq!(
                    value
                        .bind(py)
                        .getattr("id")
                        .unwrap()
                        .extract::<u64>()
                        .unwrap(),
                    7
                );
            });
        }

        #[test]
        fn value_to_py_preserves_128_bit_integers() {
            Python::initialize();
            Python::attach(|py| {
                let u128_value = u128::from(u64::MAX) + 99;
                let u128_py = value_to_py(&Value::U128(u128_value), py).unwrap();
                assert_eq!(u128_py.bind(py).extract::<u128>().unwrap(), u128_value);

                let i128_value = i128::from(i64::MIN) - 99;
                let i128_py = value_to_py(&Value::I128(i128_value), py).unwrap();
                assert_eq!(i128_py.bind(py).extract::<i128>().unwrap(), i128_value);
            });
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use bincode::{Decode, Encode, encode_into_slice};
    use serde::Deserialize;
    use std::env;
    use std::fs;
    use std::io::Cursor;
    use std::path::PathBuf;
    use std::sync::{Arc, Mutex};
    use tempfile::{TempDir, tempdir};

    fn copy_stringindex_to_temp(tmpdir: &TempDir) -> PathBuf {
        // Build a minimal index on the fly so tests don't depend on build-time artifacts.
        let fake_out_dir = tmpdir.path().join("build").join("out").join("dir");
        fs::create_dir_all(&fake_out_dir).unwrap();
        // SAFETY: Tests run single-threaded here and we only read the variable after setting it.
        unsafe {
            env::set_var("LOG_INDEX_DIR", &fake_out_dir);
        }

        // Provide entries for the message indexes used in this test module.
        let _ = cu29_intern_strs::intern_string("unused to start counter");
        let _ = cu29_intern_strs::intern_string("Just a String {}");
        let _ = cu29_intern_strs::intern_string("Just a String (low level) {}");
        let _ = cu29_intern_strs::intern_string("Just a String (end to end) {}");

        let index_dir = cu29_intern_strs::default_log_index_dir();
        cu29_intern_strs::read_interned_strings(&index_dir).unwrap();
        index_dir
    }

    #[test]
    fn test_extract_low_level_cu29_log() {
        let temp_dir = TempDir::new().unwrap();
        let temp_path = copy_stringindex_to_temp(&temp_dir);
        let entry = CuLogEntry::new(3, CuLogLevel::Info);
        let bytes = bincode::encode_to_vec(&entry, standard()).unwrap();
        let reader = Cursor::new(bytes.as_slice());
        textlog_dump(reader, temp_path.as_path()).unwrap();
    }

    #[test]
    fn end_to_end_datalogger_and_structlog_test() {
        let dir = tempdir().expect("Failed to create temp dir");
        let path = dir
            .path()
            .join("end_to_end_datalogger_and_structlog_test.copper");
        {
            // Write a couple log entries
            let UnifiedLogger::Write(logger) = UnifiedLoggerBuilder::new()
                .write(true)
                .create(true)
                .file_base_name(&path)
                .preallocated_size(100000)
                .build()
                .expect("Failed to create logger")
            else {
                panic!("Failed to create logger")
            };
            let data_logger = Arc::new(Mutex::new(logger));
            let stream = stream_write(data_logger.clone(), UnifiedLogType::StructuredLogLine, 1024)
                .expect("Failed to create stream");
            let rt = LoggerRuntime::init(RobotClock::default(), stream, None::<NullLog>);

            let mut entry = CuLogEntry::new(4, CuLogLevel::Info); // this is a "Just a String {}" log line
            entry.add_param(0, Value::String("Parameter for the log line".into()));
            log(&mut entry).expect("Failed to log");
            let mut entry = CuLogEntry::new(2, CuLogLevel::Info); // this is a "Just a String {}" log line
            entry.add_param(0, Value::String("Parameter for the log line".into()));
            log(&mut entry).expect("Failed to log");

            // everything is dropped here
            drop(rt);
        }
        // Read back the log
        let UnifiedLogger::Read(logger) = UnifiedLoggerBuilder::new()
            .file_base_name(
                &dir.path()
                    .join("end_to_end_datalogger_and_structlog_test.copper"),
            )
            .build()
            .expect("Failed to create logger")
        else {
            panic!("Failed to create logger")
        };
        let reader = UnifiedLoggerIOReader::new(logger, UnifiedLogType::StructuredLogLine);
        let temp_dir = TempDir::new().unwrap();
        textlog_dump(
            reader,
            Path::new(copy_stringindex_to_temp(&temp_dir).as_path()),
        )
        .expect("Failed to dump log");
    }

    // This is normally generated at compile time in CuPayload.
    #[derive(Debug, PartialEq, Clone, Copy, Serialize, Deserialize, Encode, Decode, Default)]
    struct MyMsgs((u8, i32, f32));

    impl ErasedCuStampedDataSet for MyMsgs {
        fn cumsgs(&self) -> Vec<&dyn ErasedCuStampedData> {
            Vec::new()
        }
    }

    impl MatchingTasks for MyMsgs {
        fn get_all_task_ids() -> &'static [&'static str] {
            &[]
        }
    }

    /// Checks if we can recover the copper lists from a binary representation.
    #[test]
    fn test_copperlists_dump() {
        let mut data = vec![0u8; 10000];
        let mypls: [MyMsgs; 4] = [
            MyMsgs((1, 2, 3.0)),
            MyMsgs((2, 3, 4.0)),
            MyMsgs((3, 4, 5.0)),
            MyMsgs((4, 5, 6.0)),
        ];

        let mut offset: usize = 0;
        for pl in mypls.iter() {
            let cl = CopperList::<MyMsgs>::new(1, *pl);
            offset +=
                encode_into_slice(&cl, &mut data.as_mut_slice()[offset..], standard()).unwrap();
        }

        let reader = Cursor::new(data);

        let mut iter = copperlists_reader::<MyMsgs>(reader);
        assert_eq!(iter.next().unwrap().msgs, MyMsgs((1, 2, 3.0)));
        assert_eq!(iter.next().unwrap().msgs, MyMsgs((2, 3, 4.0)));
        assert_eq!(iter.next().unwrap().msgs, MyMsgs((3, 4, 5.0)));
        assert_eq!(iter.next().unwrap().msgs, MyMsgs((4, 5, 6.0)));
    }

    #[test]
    fn runtime_lifecycle_reader_retains_timestamps_and_events() {
        let records = vec![
            RuntimeLifecycleRecord {
                timestamp: CuTime::default(),
                event: RuntimeLifecycleEvent::Instantiated {
                    config_source: RuntimeLifecycleConfigSource::BundledDefault,
                },
            },
            RuntimeLifecycleRecord {
                timestamp: CuTime::from_nanos(42),
                event: RuntimeLifecycleEvent::MissionStarted,
            },
        ];
        let mut bytes = Vec::new();
        for record in &records {
            bytes.extend(bincode::encode_to_vec(record, standard()).unwrap());
        }

        assert_eq!(
            runtime_lifecycle_reader(Cursor::new(bytes)).collect::<Vec<_>>(),
            records
        );
    }

    #[test]
    fn logstats_mission_defaults_to_log_then_named_default_then_first() {
        let config = CuConfig::deserialize_ron(
            r#"(
                missions: [(id: "zeta"), (id: "default"), (id: "alpha")],
                tasks: [],
                cnx: [],
            )"#,
        )
        .unwrap();
        assert_eq!(
            resolve_logstats_mission(&config, None, Some("zeta".to_string()))
                .unwrap()
                .as_deref(),
            Some("zeta")
        );
        assert!(resolve_logstats_mission(&config, None, Some("missing".to_string())).is_err());
        assert_eq!(
            resolve_logstats_mission(&config, Some("alpha".to_string()), None)
                .unwrap()
                .as_deref(),
            Some("alpha")
        );

        let config = CuConfig::deserialize_ron(
            r#"(
                missions: [(id: "zeta"), (id: "alpha")],
                tasks: [],
                cnx: [],
            )"#,
        )
        .unwrap();
        assert_eq!(
            resolve_logstats_mission(&config, None, None)
                .unwrap()
                .as_deref(),
            Some("alpha")
        );
    }

    #[test]
    fn schedule_optimizer_cli_is_explicit_and_measurements_are_validated() {
        let args = LogReaderCli::try_parse_from([
            "logreader",
            "robot.copper",
            "optimize-schedule",
            "--contract",
            "schedule.ron",
            "--output",
            "target/pgs",
            "--candidates",
            "2",
            "--measure",
            "plan-1=logs/plan-1.copper",
        ])
        .unwrap();
        let Command::OptimizeSchedule {
            contract,
            output,
            candidates,
            measurements,
        } = args.command
        else {
            panic!("expected optimize-schedule command");
        };
        assert_eq!(contract, PathBuf::from("schedule.ron"));
        assert_eq!(output, PathBuf::from("target/pgs"));
        assert_eq!(candidates, 2);
        assert_eq!(measurements, ["plan-1=logs/plan-1.copper"]);
        assert_eq!(
            parse_measurement(&measurements[0]).unwrap(),
            ("plan-1", Path::new("logs/plan-1.copper"))
        );
        assert!(parse_measurement("plan-1").is_err());
        assert!(parse_measurement("baseline=other.copper").is_err());
        assert_eq!(candidate_index("plan-2/round-3"), Some(1));
        assert_eq!(candidate_index("baseline"), None);

        let config = CuConfig::deserialize_ron(
            r#"(
                missions: [(id: "drive"), (id: "park")],
                tasks: [],
                cnx: [],
            )"#,
        )
        .unwrap();
        assert!(resolve_schedule_mission(&config, &["drive".into(), "park".into()]).is_err());
        assert_eq!(
            resolve_schedule_mission(&config, &["park".into()])
                .unwrap()
                .as_deref(),
            Some("park")
        );
    }
}