feat(cli): output_schema for sim + batch render archive; review nits (v1.5)

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 `<output>/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(

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
-/// `# <version>` 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<Vec<String>> {
-    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<Vec<String>> {
     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<Vec<String>> {
+    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<PathBuf> {
+/// 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<Vec<String>> {
     let mut dirs: Vec<PathBuf> = 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<String, TableSchema> {
     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<String, Vec<u8>>) -> BTreeMap<String, TableSchema> {
+    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", &params, &est), ColumnRole::Chain);
-        assert_eq!(classify("sweep", &params, &est), ColumnRole::Iteration);
-        assert_eq!(classify("draw", &params, &est), ColumnRole::Iteration);
-        assert_eq!(classify("beta", &params, &est), ColumnRole::ParamEstimated);
-        assert_eq!(classify("gamma", &params, &est), ColumnRole::ParamFixed);
-        assert_eq!(classify("N0", &params, &est), ColumnRole::ParamFixed);
-        // neither axis nor parameter → diagnostic
-        assert_eq!(classify("log_posterior", &params, &est), ColumnRole::Diagnostic);
-        assert_eq!(classify("n_divergent", &params, &est), ColumnRole::Diagnostic);
+        let diag = ColumnRole::Diagnostic;
+        // fit columns (default = diagnostic)
+        assert_eq!(classify("chain", &params, &est, diag), ColumnRole::Chain);
+        assert_eq!(classify("sweep", &params, &est, diag), ColumnRole::Iteration);
+        assert_eq!(classify("draw", &params, &est, diag), ColumnRole::Iteration);
+        assert_eq!(classify("beta", &params, &est, diag), ColumnRole::ParamEstimated);
+        assert_eq!(classify("gamma", &params, &est, diag), ColumnRole::ParamFixed);
+        assert_eq!(classify("N0", &params, &est, diag), ColumnRole::ParamFixed);
+        assert_eq!(classify("log_posterior", &params, &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<String, Vec<u8>> = 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]

rust/crates/cli/src/resolve.rs

@@ -274,6 +274,11 @@ pub struct RecordMeta {
     pub children: BTreeMap<String, Vec<ContentHash>>,
     pub source_paths: Vec<String>,
     pub label: Option<String>,
+    /// 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<String, runid::record::TableSchema>,
 }
 
 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 {

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/<stream>.tsv`.
     Predictive,