diff --git a/rust/crates/cli/src/batch.rs b/rust/crates/cli/src/batch.rs index 24d81256..9417d17a 100644 --- a/rust/crates/cli/src/batch.rs +++ b/rust/crates/cli/src/batch.rs @@ -593,6 +593,19 @@ pub fn cmd_batch_run(a: &crate::args::BatchArgs) { eprintln!("warning: could not write model.ir.json: {}", e); }); + // Archive the model's display render beside the IR so a viewer (camdl-watch) + // can show the model's math without recompiling. Best-effort — a render + // failure (e.g. the model was given as compiled IR, not source) warns and + // skips, never aborts the run. + match crate::util::render_model_json(std::path::Path::new(&model_path)) { + Ok(json) => { + if let Err(e) = std::fs::write(format!("{}/model.render.json", output_dir), &json) { + eprintln!("warning: could not write model.render.json: {}", e); + } + } + Err(e) => eprintln!("warning: could not render model for archive: {}", e), + } + // Copy any boundary GeoJSON into the output tree as a sibling artifact // (a map viewer reads `/geo/boundaries.geojson`). if let Some(ref geo_src) = exp.config.geo { @@ -1178,7 +1191,7 @@ impl crate::engine::RunSink for CasSink { run_id: rt.run_id, display_inputs: serde_json::Value::Null, }; - let meta = crate::resolve::RecordMeta::new( + let mut meta = crate::resolve::RecordMeta::new( ir::IR_VERSION.trim(), self.model_path.clone(), self.label.clone()) .with_deps(self.fit_dep.clone()) .with_children(children); @@ -1216,6 +1229,10 @@ impl crate::engine::RunSink for CasSink { let bytes = sim::reactive::format_reactive_log(firings).into_bytes(); artifacts.insert("reactive_log.tsv", bytes); } + // Declare the tabular outputs' column schema in run.json — classify the + // in-memory artifact headers (they're committed atomically, not yet on + // disk). Recorded, not hashed. + meta.output_schema = crate::output_schema::sim_output_schema(&artifacts.files); let root = self.root(); let store = runid::FsCasStore::new(&root); let dest = match crate::resolve::begin_resolved_write( diff --git a/rust/crates/cli/src/output_schema.rs b/rust/crates/cli/src/output_schema.rs index 6f8e11f3..2e75b12b 100644 --- a/rust/crates/cli/src/output_schema.rs +++ b/rust/crates/cli/src/output_schema.rs @@ -12,55 +12,81 @@ use std::path::{Path, PathBuf}; use runid::record::{ColumnRole, ColumnSpec, TableRole, TableSchema}; -/// Classify one column name into its semantic role. `params` is every model -/// parameter name; `estimated` is the sampled subset. A column that is neither a -/// chain key, an iteration axis, nor a model parameter is a sampler diagnostic -/// (`loglik`, `log_posterior`, `tree_depth`, `accepted`, …). -fn classify(name: &str, params: &HashSet<&str>, estimated: &HashSet<&str>) -> ColumnRole { +/// Classify one column name into its semantic role, using the model's parameter +/// sets and a table-specific `default` for columns matching no reserved name or +/// prefix. `params` is every model parameter name; `estimated` is the sampled +/// subset. +/// +/// The reserved vocabulary (`chain`, the iteration/time axes, `flow_`/`inc_` +/// prefixes) is assumed not to collide with model parameter names — safe because +/// camdl parameters are epidemiological (`beta`, `gamma`), never sampler or +/// trajectory column names. A colliding parameter would take its reserved role +/// rather than a parameter role, but that is unreachable for a real model. +fn classify( + name: &str, + params: &HashSet<&str>, + estimated: &HashSet<&str>, + default: ColumnRole, +) -> ColumnRole { match name { "chain" => ColumnRole::Chain, - "sweep" | "step" | "draw" | "iteration" => ColumnRole::Iteration, + "replicate" => ColumnRole::Replicate, + "scenario" => ColumnRole::Scenario, + "t" | "time" | "date" => ColumnRole::Time, + "sweep" | "step" | "draw" | "iteration" | "point_id" => ColumnRole::Iteration, + n if n.starts_with("flow_") => ColumnRole::Flow, + n if n.starts_with("inc_") => ColumnRole::Incidence, n if estimated.contains(n) => ColumnRole::ParamEstimated, n if params.contains(n) => ColumnRole::ParamFixed, - _ => ColumnRole::Diagnostic, + _ => default, } } -/// The tab-separated column names of `path`'s header, skipping a leading -/// `# ` comment line if present. `None` when the file is absent or -/// unreadable — the schema is best-effort provenance, never a hard dependency. -fn header(path: &Path) -> Option> { - let text = std::fs::read_to_string(path).ok()?; +/// Column names of a TSV header — the first non-`#` line, tab-split. `None` when +/// there is no such line. +fn header_cols(text: &str) -> Option> { let line = text.lines().find(|l| !l.starts_with('#'))?; Some(line.split('\t').map(str::to_string).collect()) } -fn table(cols: &[String], role: TableRole, params: &HashSet<&str>, estimated: &HashSet<&str>) -> TableSchema { +/// The header columns of the file at `path` — best-effort (`None` when absent or +/// unreadable; the schema is provenance, never a hard dependency). +fn header(path: &Path) -> Option> { + header_cols(&std::fs::read_to_string(path).ok()?) +} + +fn table( + cols: &[String], + role: TableRole, + params: &HashSet<&str>, + estimated: &HashSet<&str>, + default: ColumnRole, +) -> TableSchema { TableSchema { role, columns: cols .iter() - .map(|name| ColumnSpec { name: name.clone(), role: classify(name, params, estimated) }) + .map(|name| ColumnSpec { name: name.clone(), role: classify(name, params, estimated, default) }) .collect(), } } -/// The lowest-numbered `chain_*` directory under `leaf`, if any — the per-chain -/// trace schema is identical across chains, so one is read and keyed by `{n}`. -fn first_chain_dir(leaf: &Path) -> Option { +/// The header of `fname` in the first `chain_*` directory that has a readable +/// one. The per-chain schema is identical across chains, so any chain suffices — +/// reading the first *readable* one (not merely the lexicographically-first +/// directory) keeps the entry present when an early chain crashed or is absent. +fn first_chain_header(leaf: &Path, fname: &str) -> Option> { let mut dirs: Vec = std::fs::read_dir(leaf) .ok()? .flatten() .map(|e| e.path()) .filter(|p| { p.is_dir() - && p.file_name() - .and_then(|n| n.to_str()) - .is_some_and(|n| n.starts_with("chain_")) + && p.file_name().and_then(|n| n.to_str()).is_some_and(|n| n.starts_with("chain_")) }) .collect(); dirs.sort(); - dirs.into_iter().next() + dirs.iter().find_map(|d| header(&d.join(fname))) } /// Build the `output_schema` for a completed fit stage by classifying the actual @@ -73,19 +99,47 @@ pub fn fit_output_schema( estimated: &HashSet<&str>, ) -> BTreeMap { let mut out = BTreeMap::new(); + let diag = ColumnRole::Diagnostic; // draws.tsv — the thinned posterior cloud (Bayesian methods; if2 writes none). if let Some(cols) = header(&leaf.join("draws.tsv")) { - out.insert("draws.tsv".to_string(), table(&cols, TableRole::PosteriorCloud, all_params, estimated)); + out.insert( + "draws.tsv".to_string(), + table(&cols, TableRole::PosteriorCloud, all_params, estimated, diag), + ); } - // Per-chain trace — one entry keyed by the `{n}` wildcard, read from the - // first chain directory. `trace.tsv` (pgas/pmmh/mh/nuts) or - // `parameter_traces.tsv` (if2). - if let Some(chain1) = first_chain_dir(leaf) { - for fname in ["trace.tsv", "parameter_traces.tsv"] { - if let Some(cols) = header(&chain1.join(fname)) { - out.insert(format!("chain_{{n}}/{fname}"), table(&cols, TableRole::Trace, all_params, estimated)); + // Per-chain trace — one entry per trace filename, keyed by `{n}`, read from + // the first chain that has it. `trace.tsv` (pgas/pmmh/mh/nuts) or + // `parameter_traces.tsv` (if2 / nlopt). + for fname in ["trace.tsv", "parameter_traces.tsv"] { + if let Some(cols) = first_chain_header(leaf, fname) { + out.insert( + format!("chain_{{n}}/{fname}"), + table(&cols, TableRole::Trace, all_params, estimated, diag), + ); + } + } + out +} + +/// Build the `output_schema` for a simulation leaf by classifying its in-memory +/// artifact headers (the sim commits atomically, so the files are not yet on +/// disk). `traj.tsv`/`ensemble.tsv` are trajectories — a column matching no +/// reserved name is a compartment `state`; other artifacts are skipped for now. +pub fn sim_output_schema(files: &BTreeMap>) -> BTreeMap { + let none: HashSet<&str> = HashSet::new(); + let mut out = BTreeMap::new(); + for (name, bytes) in files { + let role_default = match name.as_str() { + "traj.tsv" | "ensemble.tsv" => Some((TableRole::Trajectory, ColumnRole::State)), + _ => None, + }; + if let Some((role, default)) = role_default { + if let Ok(text) = std::str::from_utf8(bytes) { + if let Some(cols) = header_cols(text) { + out.insert(name.clone(), table(&cols, role, &none, &none, default)); + } } } } @@ -101,17 +155,40 @@ mod tests { } #[test] - fn classify_axes_params_and_diagnostics() { + fn classify_axes_params_and_defaults() { let (params, est) = sets(&["beta", "gamma", "N0"], &["beta"]); - assert_eq!(classify("chain", ¶ms, &est), ColumnRole::Chain); - assert_eq!(classify("sweep", ¶ms, &est), ColumnRole::Iteration); - assert_eq!(classify("draw", ¶ms, &est), ColumnRole::Iteration); - assert_eq!(classify("beta", ¶ms, &est), ColumnRole::ParamEstimated); - assert_eq!(classify("gamma", ¶ms, &est), ColumnRole::ParamFixed); - assert_eq!(classify("N0", ¶ms, &est), ColumnRole::ParamFixed); - // neither axis nor parameter → diagnostic - assert_eq!(classify("log_posterior", ¶ms, &est), ColumnRole::Diagnostic); - assert_eq!(classify("n_divergent", ¶ms, &est), ColumnRole::Diagnostic); + let diag = ColumnRole::Diagnostic; + // fit columns (default = diagnostic) + assert_eq!(classify("chain", ¶ms, &est, diag), ColumnRole::Chain); + assert_eq!(classify("sweep", ¶ms, &est, diag), ColumnRole::Iteration); + assert_eq!(classify("draw", ¶ms, &est, diag), ColumnRole::Iteration); + assert_eq!(classify("beta", ¶ms, &est, diag), ColumnRole::ParamEstimated); + assert_eq!(classify("gamma", ¶ms, &est, diag), ColumnRole::ParamFixed); + assert_eq!(classify("N0", ¶ms, &est, diag), ColumnRole::ParamFixed); + assert_eq!(classify("log_posterior", ¶ms, &est, diag), ColumnRole::Diagnostic); + // trajectory columns (no params; default = state) + let none: HashSet<&str> = HashSet::new(); + let st = ColumnRole::State; + assert_eq!(classify("t", &none, &none, st), ColumnRole::Time); + assert_eq!(classify("date", &none, &none, st), ColumnRole::Time); + assert_eq!(classify("flow_infection", &none, &none, st), ColumnRole::Flow); + assert_eq!(classify("inc_cases", &none, &none, st), ColumnRole::Incidence); + assert_eq!(classify("replicate", &none, &none, st), ColumnRole::Replicate); + assert_eq!(classify("S", &none, &none, st), ColumnRole::State); // compartment → default + } + + #[test] + fn sim_schema_classifies_trajectory_header() { + let mut files: BTreeMap> = BTreeMap::new(); + files.insert("traj.tsv".to_string(), b"t\tS\tI\tR\tflow_infection\n0\t99\t1\t0\t0\n".to_vec()); + files.insert("event_log.tsv".to_string(), b"anything\n".to_vec()); // not a trajectory + let schema = sim_output_schema(&files); + assert!(!schema.contains_key("event_log.tsv"), "non-trajectory artifact skipped"); + let traj = &schema["traj.tsv"]; + assert_eq!(traj.role, TableRole::Trajectory); + assert_eq!(traj.columns[0].role, ColumnRole::Time); // t — the x-axis + assert_eq!(traj.columns[1].role, ColumnRole::State); // S (compartment) + assert_eq!(traj.columns[4].role, ColumnRole::Flow); // flow_infection } #[test] diff --git a/rust/crates/cli/src/resolve.rs b/rust/crates/cli/src/resolve.rs index 12472a44..c448d86f 100644 --- a/rust/crates/cli/src/resolve.rs +++ b/rust/crates/cli/src/resolve.rs @@ -274,6 +274,11 @@ pub struct RecordMeta { pub children: BTreeMap>, pub source_paths: Vec, pub label: Option, + /// Column schema for the leaf's tabular outputs, attached to the record at + /// build time — used by atomic-commit paths (e.g. `sim`, whose files are + /// known in memory before commit); the streaming fit path sets it at + /// finalize instead. + pub output_schema: BTreeMap, } impl RecordMeta { @@ -290,6 +295,7 @@ impl RecordMeta { children: BTreeMap::new(), source_paths: vec![model_path.into()], label, + output_schema: BTreeMap::new(), } } @@ -366,7 +372,7 @@ fn build_record(resolved: &ResolvedArtifact, meta: &RecordMeta, status: RunStatu deps: meta.deps.clone(), status, artifacts: Default::default(), - output_schema: Default::default(), + output_schema: meta.output_schema.clone(), children: meta.children.clone(), inputs: resolved.display_inputs.clone(), provenance: Provenance { diff --git a/rust/crates/runid/src/record.rs b/rust/crates/runid/src/record.rs index 0ca0b80a..8713c5d5 100644 --- a/rust/crates/runid/src/record.rs +++ b/rust/crates/runid/src/record.rs @@ -107,7 +107,8 @@ pub enum TableRole { Observation, /// The thinned posterior-draws cloud: `draws.tsv`. PosteriorCloud, - /// The full per-chain sampler trace: `chain_N/trace.tsv`. + /// The full per-chain trace of a sampler or optimizer: `chain_N/trace.tsv` + /// (MCMC) or `chain_N/parameter_traces.tsv` (if2 / nlopt). Trace, /// Predicted-vs-observed bands: `predictive/.tsv`. Predictive,