From f34fd5b567b20f975de585ce670569476e2e9f42 Mon Sep 17 00:00:00 2001 From: Logan Nye <87274608+logannye@users.noreply.github.com> Date: Tue, 2 Jun 2026 13:47:23 -0700 Subject: [PATCH] =?UTF-8?q?feat(pack):=20plan=20--fleet=20=E2=80=94=20turn?= =?UTF-8?q?=20the=20predicted=20peak=20into=20a=20placement=20decision?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The predicted peak is computed from the index header alone (no run, no read I/O), is a conservative upper bound (sound since the Act-1 soundness fix), and is additive — so a scheduler can SUM predicted peaks across co-located jobs and PROVE a node fits before launching a byte. No emergent-peak caller (GATK, DeepVariant) can: their peak is only known after a possible OOM-kill, so every co-location is a gamble. - src/call/pack.rs: a pure, deterministic first-fit-decreasing bin-packer over predicted peaks. `Packed` is a proof — every node's summed peak <= capacity; a job larger than a node, or a set that needs more than --nodes, is NoFit. - `rosalind pack --jobs --node-mb N [--nodes M]`: read each job's index header → predicted peak → pack onto nodes, or refuse (exit 3). Human or --json. - `rosalind plan --index --json`: the predicted-peak breakdown as one-line JSON for a workflow engine to read. - README "Pack a fleet: prediction → placement" + roadmap. Tests: 5 packer unit tests (incl. the proof property + determinism) + a pack integration test (fit / refuse exit 3 / json) + plan --json. Full suite green; rustc 0 warnings; additions clippy-clean. Co-Authored-By: Claude Opus 4.8 (1M context) --- README.md | 19 +++++ src/call/mod.rs | 2 + src/call/pack.rs | 194 +++++++++++++++++++++++++++++++++++++++++++ src/main.rs | 208 +++++++++++++++++++++++++++++++++++++++++++++-- tests/pack.rs | 125 ++++++++++++++++++++++++++++ 5 files changed, 542 insertions(+), 6 deletions(-) create mode 100644 src/call/pack.rs create mode 100644 tests/pack.rs diff --git a/README.md b/README.md index e4ae97e..a9eb174 100644 --- a/README.md +++ b/README.md @@ -131,6 +131,24 @@ Drop the Rosalind budget Action (this repo's [`action.yml`](action.yml)) into an It runs `plan` (predicts the peak), then `variants --index --enforce` (honors the budget), and uploads the BLAKE3 receipt as a build artifact. This is the one thing a `--max-mem` flag on another caller can't give you: a portable, declarative, **verifiable** memory budget that fails a stranger's build loudly — the contract, enforced where your pipeline already lives. (Available once a release is published; see Quickstart.) +## Pack a fleet: prediction → placement + +Because the predicted peak is computed from the **index header alone** — no run, no read I/O, milliseconds — and is a conservative, *additive* upper bound, you can turn the prediction into a scheduling decision: **prove** that N calling jobs co-located on one node will all fit, before launching a single read. + +```sh +# jobs.tsv: one job per line — [\t[\t]] +rosalind pack --jobs jobs.tsv --node-mb 64000 # pack onto 64 GB nodes +rosalind pack --jobs jobs.tsv --node-mb 64000 --nodes 4 # …or refuse (exit 3) if it needs more than 4 +``` + +``` +pack: 37 job(s) → 3 node(s) of 64000 MiB — every node proven within capacity + node 0: 61920 / 64000 MiB [sampleA.idx, sampleB.idx, …] + ... +``` + +`rosalind plan --index ref.idx --budget-mb 64000 --json` emits the same predicted peak as one line of JSON for a workflow engine to read. This is what an emergent-peak caller (GATK, DeepVariant) structurally cannot do: their peak is only known *after* a possible OOM-kill, so every co-location is a gamble. Here, `Packed` is a proof — each node's summed predicted peak is `≤` its capacity, established up front, and the schedule is deterministic. + ## A reproducible feature substrate for ML The same bounded streaming engine that calls variants can emit **per-locus features** instead — one tabular row per callable position, ready for a model: @@ -158,6 +176,7 @@ The core primitive is a streaming, CIGAR-aware pileup column stream; variant cal - **Phase B (done):** streaming gzip/bgzf input; a multi-contig FM-index over the concatenated genome with `(contig, position)` resolution; a build-once, memory-mapped, byte-reproducible persisted index (`rosalind index`/`locate`); zero-copy reference access from the index; and **bounded whole-genome germline calling over a sorted BAM** (`rosalind variants --index`) with a realized-memory receipt. - **Phase C (done):** memory as a *verifiable contract* — `rosalind plan` (a checkable envelope before you commit), `--enforce` (honor-or-refuse: refuse up front / fail loud, never a silent OOM-kill), and `rosalind verify`. See [CONTRACT.md](CONTRACT.md). - **Hardening & reach (done):** unbiased depth-cap downsampling (no silent variant drops) and a CI-enforced memory gate; **measured** germline detection accuracy ([Accuracy](#accuracy)); the **`rosalind features`** reproducible ML feature substrate; and a one-command adoption on-ramp — prebuilt binaries (`install.sh`, with checksum verification) plus the **Rosalind budget GitHub Action** (`action.yml`, used as `logannye/rosalind@v0.1.0`) that enforces the contract in *your* CI. +- **Fleet scheduling (done):** [prediction → placement](#pack-a-fleet-prediction--placement) — `rosalind pack` proves a co-location of N calling jobs fits a node before launching a byte (predicted peaks are additive and read from the index header); `plan --index --json` for a scheduler to read. - **Phase D (research):** sublinear-space index construction — the `~√t` space/time knob across the full curve — extending the contract to the index *build* step (today's build is O(reference)). The headline space-complexity bet; see [`docs/OPEN_PROBLEMS.md`](docs/OPEN_PROBLEMS.md). - **Later:** the aligner over the persisted multi-contig index (`align --index`, whole-genome alignment); germline indels and richer read QC; deterministic multithreading; a Python/tensor binding over the pileup stream. diff --git a/src/call/mod.rs b/src/call/mod.rs index 9e091e5..20fc0dd 100644 --- a/src/call/mod.rs +++ b/src/call/mod.rs @@ -4,6 +4,7 @@ pub mod features; pub mod germline; +pub mod pack; pub mod pipeline; pub mod plan; pub mod somatic; @@ -14,6 +15,7 @@ pub use features::{ stream_features_region, stream_features_whole_genome, write_feature_header, write_feature_row, }; pub use germline::call_germline; +pub use pack::{first_fit_decreasing, NodeAssignment, PackJob, PackOutcome}; pub use pipeline::{ call_germline_region, call_germline_region_streaming, call_germline_region_tracked, call_somatic_region, diff --git a/src/call/pack.rs b/src/call/pack.rs new file mode 100644 index 0000000..b4dd8ba --- /dev/null +++ b/src/call/pack.rs @@ -0,0 +1,194 @@ +//! Bin-packing over PREDICTED memory peaks — the contract's prediction turned +//! into a *placement* decision. +//! +//! Each job's peak is a conservative upper bound computed from the index header +//! alone (no run, no read I/O — milliseconds), and peaks are additive. So a +//! scheduler can SUM predicted peaks across co-located jobs and *prove* a node +//! fits before launching a byte. No incumbent caller can: GATK/DeepVariant peaks +//! are emergent and only known after a possible OOM-kill, so every co-location is +//! a gamble. Here, `Packed(...)` is a proof — every node's summed peak is `<=` +//! its capacity, established up front. + +/// A job to place: a label and its predicted peak RSS, in bytes. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct PackJob { + /// Human-readable label (e.g. the index path or sample id). + pub label: String, + /// Predicted peak RSS (a conservative upper bound), in bytes. + pub predicted_peak_bytes: u64, +} + +/// One node's assignment: which jobs land on it and their summed predicted peak +/// (which is `<=` the node capacity — the fit, proven). +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct NodeAssignment { + /// Node index (0-based). + pub node: usize, + /// Labels of the jobs placed on this node. + pub job_labels: Vec, + /// Summed predicted peak of those jobs, in bytes. + pub used_bytes: u64, +} + +/// The outcome of a packing attempt. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum PackOutcome { + /// Every job placed. Each `NodeAssignment.used_bytes <= node_capacity_bytes`, + /// so the node is *proven* to fit before any job runs. + Packed(Vec), + /// No safe packing exists under the constraints. + NoFit { + /// Why no packing was found (an actionable message). + reason: String, + }, +} + +/// Pack `jobs` onto nodes of `node_capacity_bytes` via first-fit-decreasing +/// (largest job first → tighter packing, fewer nodes). With `max_nodes = Some(m)` +/// the pack refuses when the jobs cannot fit within `m` nodes; with `None` it +/// opens as many nodes as needed and reports the count. A single job whose +/// predicted peak exceeds a whole node is an immediate `NoFit` — it can run +/// nowhere. Deterministic: ties break by label, so the same jobs always pack the +/// same way (the contract's reproducibility extends to the schedule). +pub fn first_fit_decreasing( + jobs: &[PackJob], + node_capacity_bytes: u64, + max_nodes: Option, +) -> PackOutcome { + // A job larger than a whole node can never be placed — surface it explicitly + // rather than spinning up unbounded nodes. + if let Some(j) = jobs + .iter() + .find(|j| j.predicted_peak_bytes > node_capacity_bytes) + { + return PackOutcome::NoFit { + reason: format!( + "job '{}' predicted peak {} B exceeds the node capacity {} B — it cannot run on any node; \ + use a larger --node-mb or lower --max-depth", + j.label, j.predicted_peak_bytes, node_capacity_bytes + ), + }; + } + + // FFD: descending peak, ties by label (deterministic). + let mut order: Vec<&PackJob> = jobs.iter().collect(); + order.sort_by(|a, b| { + b.predicted_peak_bytes + .cmp(&a.predicted_peak_bytes) + .then_with(|| a.label.cmp(&b.label)) + }); + + let mut nodes: Vec = Vec::new(); + for job in order { + // Place in the first node that still has room (first-fit). + let slot = nodes + .iter_mut() + .find(|n| n.used_bytes + job.predicted_peak_bytes <= node_capacity_bytes); + match slot { + Some(n) => { + n.used_bytes += job.predicted_peak_bytes; + n.job_labels.push(job.label.clone()); + } + None => { + if let Some(max) = max_nodes { + if nodes.len() >= max { + return PackOutcome::NoFit { + reason: format!( + "jobs do not fit in {max} node(s) of {} B each; need more nodes or a larger --node-mb", + node_capacity_bytes + ), + }; + } + } + nodes.push(NodeAssignment { + node: nodes.len(), + job_labels: vec![job.label.clone()], + used_bytes: job.predicted_peak_bytes, + }); + } + } + } + PackOutcome::Packed(nodes) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn job(label: &str, peak: u64) -> PackJob { + PackJob { + label: label.to_string(), + predicted_peak_bytes: peak, + } + } + + #[test] + fn packs_all_jobs_and_every_node_is_proven_within_capacity() { + let jobs = vec![job("a", 30), job("b", 40), job("c", 50), job("d", 20)]; + let cap = 100; + let PackOutcome::Packed(nodes) = first_fit_decreasing(&jobs, cap, None) else { + panic!("should pack"); + }; + // THE proof property: every node's summed peak is within capacity. + for n in &nodes { + assert!( + n.used_bytes <= cap, + "node {} sums {} > capacity {cap}", + n.node, + n.used_bytes + ); + } + // Every job placed exactly once. + let mut placed: Vec<&String> = nodes.iter().flat_map(|n| &n.job_labels).collect(); + placed.sort(); + assert_eq!(placed, vec!["a", "b", "c", "d"]); + // FFD on {50,40,30,20} into 100 → 2 nodes (50+40, 30+20). + assert_eq!(nodes.len(), 2, "FFD should use 2 nodes: {nodes:?}"); + } + + #[test] + fn refuses_a_job_larger_than_a_node() { + let jobs = vec![job("ok", 50), job("toobig", 150)]; + match first_fit_decreasing(&jobs, 100, None) { + PackOutcome::NoFit { reason } => assert!( + reason.contains("toobig") && reason.contains("cannot run on any node"), + "unexpected reason: {reason}" + ), + other => panic!("a job bigger than a node must NoFit: {other:?}"), + } + } + + #[test] + fn refuses_when_jobs_exceed_the_node_budget() { + // Three 60 B jobs into 100 B nodes need 3 nodes; cap at 2 → NoFit. + let jobs = vec![job("a", 60), job("b", 60), job("c", 60)]; + match first_fit_decreasing(&jobs, 100, Some(2)) { + PackOutcome::NoFit { reason } => assert!(reason.contains("2 node"), "reason: {reason}"), + other => panic!("should not fit in 2 nodes: {other:?}"), + } + // With 3 nodes allowed, it fits. + assert!(matches!( + first_fit_decreasing(&jobs, 100, Some(3)), + PackOutcome::Packed(_) + )); + } + + #[test] + fn is_deterministic_regardless_of_input_order() { + let a = vec![job("x", 40), job("y", 40), job("z", 40)]; + let b = vec![job("z", 40), job("x", 40), job("y", 40)]; + assert_eq!( + first_fit_decreasing(&a, 100, None), + first_fit_decreasing(&b, 100, None), + "packing must be order-independent (deterministic schedule)" + ); + } + + #[test] + fn empty_jobs_pack_to_zero_nodes() { + assert_eq!( + first_fit_decreasing(&[], 100, None), + PackOutcome::Packed(vec![]) + ); + } +} diff --git a/src/main.rs b/src/main.rs index 2fa59c9..d591668 100644 --- a/src/main.rs +++ b/src/main.rs @@ -275,6 +275,35 @@ enum Commands { /// Declared memory budget (MiB) to check feasibility against. #[arg(long)] budget_mb: Option, + /// Emit the predicted-peak breakdown as one-line JSON (for a scheduler/CI + /// to read), instead of the human-readable table. `--index` only. + #[arg(long)] + json: bool, + }, + /// Pack many bounded `variants` jobs onto fixed-size nodes by their PREDICTED + /// peaks — prove a co-location fits before launching a byte. Each job's peak + /// is read from its index header (no run); peaks are additive, so the sum is a + /// conservative bound a scheduler can refuse on. Exit 3 if no packing fits. + Pack { + /// A jobs file: one job per line, `[\t[\t]]` + /// (TSV or whitespace; blank lines and `#` comments ignored). + #[arg(long)] + jobs: PathBuf, + /// Per-node memory capacity, in MiB (the RAM each node can give a co-located batch). + #[arg(long)] + node_mb: u64, + /// Cap on the number of nodes; refuse (exit 3) if the jobs need more. Unset = as many as needed. + #[arg(long)] + nodes: Option, + /// Default max active depth for jobs that do not specify one. + #[arg(long, default_value_t = 1000)] + max_depth: u32, + /// Default max read length for jobs that do not specify one. + #[arg(long, default_value_t = 250)] + max_read_len: u32, + /// Emit the schedule as JSON instead of the human-readable plan. + #[arg(long)] + json: bool, }, /// Re-check a reproducibility receipt without re-running: re-hash its inputs /// and outputs and confirm the realized peak landed within the budget. @@ -465,7 +494,16 @@ fn main() -> Result<()> { max_depth, max_read_len, budget_mb, - } => run_plan(index, reference, max_depth, max_read_len, budget_mb)?, + json, + } => run_plan(index, reference, max_depth, max_read_len, budget_mb, json)?, + Commands::Pack { + jobs, + node_mb, + nodes, + max_depth, + max_read_len, + json, + } => run_pack(jobs, node_mb, nodes, max_depth, max_read_len, json)?, Commands::Verify { manifest, budget_mb, @@ -615,8 +653,9 @@ fn run_plan( max_depth: u32, max_read_len: u32, budget_mb: Option, + json: bool, ) -> Result<()> { - use rosalind::call::plan::render_variants_plan; + use rosalind::call::plan::{predicted_peak_rss_bytes, render_variants_plan}; use rosalind::genomics::IndexReader; if let Some(index_path) = index { @@ -631,10 +670,41 @@ fn run_plan( // Measure the process baseline now (binary + libs + index mmap header); // the per-contig reference decode + active set are modeled on top. let baseline = peak_rss_bytes(); - print!( - "{}", - render_variants_plan(largest, max_depth, max_read_len, baseline, budget_mb) - ); + if json { + // Machine-readable: the predicted-peak fields a scheduler/CI reads. + let predicted = predicted_peak_rss_bytes(largest, max_depth, max_read_len, baseline); + let verdict = match budget_mb { + Some(mb) => { + if MemoryBudget::from_mb(mb).admits(predicted) { + "fits" + } else { + "refuse" + } + } + None => "none", + }; + let budget_field = budget_mb + .map(|mb| mb.to_string()) + .unwrap_or_else(|| "null".to_string()); + println!( + "{{\"index\":\"{}\",\"predicted_peak_rss_bytes\":{},\"baseline_rss_bytes\":{},\ + \"largest_contig_len\":{},\"max_depth\":{},\"max_read_len\":{},\ + \"budget_mb\":{},\"verdict\":\"{}\"}}", + index_path.display(), + predicted, + baseline, + largest, + max_depth, + max_read_len, + budget_field, + verdict, + ); + } else { + print!( + "{}", + render_variants_plan(largest, max_depth, max_read_len, baseline, budget_mb) + ); + } } else { let reference = reference.expect("clap guarantees one of --index/--reference"); let fasta_reader = open_input(&reference) @@ -657,6 +727,132 @@ fn run_plan( Ok(()) } +/// Pack many bounded `variants` jobs onto fixed-size nodes by their PREDICTED +/// peaks — each read from the job's index header (no run, no read I/O). Peaks are +/// conservative upper bounds and additive, so the printed schedule PROVES every +/// node fits before a single job launches — the contract turned into a placement +/// decision. Exits 3 when no safe packing exists. +fn run_pack( + jobs_path: PathBuf, + node_mb: u64, + nodes: Option, + default_max_depth: u32, + default_max_read_len: u32, + json: bool, +) -> Result<()> { + use rosalind::call::plan::predicted_peak_rss_bytes; + use rosalind::call::{first_fit_decreasing, PackJob, PackOutcome}; + use rosalind::genomics::IndexReader; + + let text = std::fs::read_to_string(&jobs_path) + .with_context(|| format!("failed to read jobs file {}", jobs_path.display()))?; + + // A nominal per-process baseline (binary + libs + index header): each + // co-located job runs in its own process with roughly this floor. Measured + // once and applied per job, so the per-job predicted peaks are comparable. + let baseline = peak_rss_bytes(); + + let mut pack_jobs: Vec = Vec::new(); + for (lineno, raw) in text.lines().enumerate() { + let line = raw.trim(); + if line.is_empty() || line.starts_with('#') { + continue; + } + let mut fields = line.split_whitespace(); + let index_path = fields.next().expect("a non-empty line has a first field"); + let max_depth = match fields.next() { + Some(s) => s.parse::().with_context(|| { + format!( + "jobs file {}:{}: invalid max_depth '{s}'", + jobs_path.display(), + lineno + 1 + ) + })?, + None => default_max_depth, + }; + let max_read_len = match fields.next() { + Some(s) => s.parse::().with_context(|| { + format!( + "jobs file {}:{}: invalid max_read_len '{s}'", + jobs_path.display(), + lineno + 1 + ) + })?, + None => default_max_read_len, + }; + let loaded = IndexReader::open(std::path::Path::new(index_path)).with_context(|| { + format!( + "jobs file {}:{}: failed to open index {index_path}", + jobs_path.display(), + lineno + 1 + ) + })?; + let largest = loaded + .contigs() + .iter() + .map(|c| c.length as u64) + .max() + .unwrap_or(0); + let predicted = predicted_peak_rss_bytes(largest, max_depth, max_read_len, baseline); + pack_jobs.push(PackJob { + label: index_path.to_string(), + predicted_peak_bytes: predicted, + }); + } + + if pack_jobs.is_empty() { + bail!("jobs file {} has no jobs", jobs_path.display()); + } + + let node_cap = node_mb.saturating_mul(1 << 20); + const MIB: u64 = 1 << 20; + match first_fit_decreasing(&pack_jobs, node_cap, nodes) { + PackOutcome::Packed(assignments) => { + if json { + let mut s = format!("{{\"node_mb\":{node_mb},\"nodes\":["); + for (i, n) in assignments.iter().enumerate() { + if i > 0 { + s.push(','); + } + let labels = n + .job_labels + .iter() + .map(|l| format!("\"{l}\"")) + .collect::>() + .join(","); + s.push_str(&format!( + "{{\"node\":{},\"used_bytes\":{},\"jobs\":[{}]}}", + n.node, n.used_bytes, labels + )); + } + s.push_str("]}"); + println!("{s}"); + } else { + println!( + "pack: {} job(s) → {} node(s) of {} MiB — every node proven within capacity", + pack_jobs.len(), + assignments.len(), + node_mb + ); + for n in &assignments { + println!( + " node {}: {} / {} MiB [{}]", + n.node, + n.used_bytes / MIB, + node_mb, + n.job_labels.join(", ") + ); + } + } + Ok(()) + } + PackOutcome::NoFit { reason } => { + eprintln!("pack: REFUSE — {reason}"); + std::process::exit(3); + } + } +} + /// Re-check a reproducibility receipt without re-running: parse it, re-hash each /// listed input/output and confirm the digests match, and confirm the recorded /// realized peak RSS landed within the budget (supplied, or recorded in the diff --git a/tests/pack.rs b/tests/pack.rs new file mode 100644 index 0000000..e2a3b3c --- /dev/null +++ b/tests/pack.rs @@ -0,0 +1,125 @@ +//! `rosalind pack` — the contract's prediction turned into a placement decision. +//! Each job's peak is read from its index header (no run); the schedule proves +//! every node fits before any job launches, or refuses (exit 3). + +use std::path::PathBuf; +use std::process::Command; +use std::sync::atomic::{AtomicU64, Ordering}; + +fn bin() -> &'static str { + env!("CARGO_BIN_EXE_rosalind") +} + +fn unique_dir(prefix: &str) -> PathBuf { + static COUNTER: AtomicU64 = AtomicU64::new(0); + let n = COUNTER.fetch_add(1, Ordering::Relaxed); + let nanos = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos(); + let d = std::env::temp_dir().join(format!("{prefix}-{nanos}-{n}")); + std::fs::create_dir_all(&d).unwrap(); + d +} + +fn build_index(dir: &std::path::Path, name: &str) -> PathBuf { + let fa = dir.join(format!("{name}.fa")); + // A small single contig — enough for a real index + a real predicted peak. + std::fs::write(&fa, format!(">{name}\n{}\n", "ACGTACGTAC".repeat(40))).unwrap(); + let idx = dir.join(format!("{name}.idx")); + let out = Command::new(bin()) + .args(["index", "--reference"]) + .arg(&fa) + .arg("--output") + .arg(&idx) + .output() + .unwrap(); + assert!(out.status.success(), "index build failed: {out:?}"); + idx +} + +#[test] +fn pack_proves_a_co_location_fits_and_refuses_when_it_cannot() { + let dir = unique_dir("rosalind-pack"); + let a = build_index(&dir, "a"); + let b = build_index(&dir, "b"); + let jobs = dir.join("jobs.tsv"); + std::fs::write( + &jobs, + format!( + "{}\n# a comment line\n\n{}\t500\t150\n", + a.display(), + b.display() + ), + ) + .unwrap(); + + // A generous node fits both jobs on one node, proven up front. + let out = Command::new(bin()) + .args(["pack", "--jobs"]) + .arg(&jobs) + .args(["--node-mb", "512"]) + .output() + .unwrap(); + assert!(out.status.success(), "generous node should pack: {out:?}"); + let stdout = String::from_utf8_lossy(&out.stdout); + assert!( + stdout.contains("every node proven within capacity"), + "missing proof line: {stdout}" + ); + + // JSON form is machine-readable. + let outj = Command::new(bin()) + .args(["pack", "--jobs"]) + .arg(&jobs) + .args(["--node-mb", "512", "--json"]) + .output() + .unwrap(); + assert!(outj.status.success()); + let j = String::from_utf8_lossy(&outj.stdout); + assert!( + j.contains("\"node_mb\":512") && j.contains("\"nodes\":["), + "unexpected json: {j}" + ); + + // A tiny node cannot hold even one job (predicted peak ≫ 4 MiB) → exit 3. + let refuse = Command::new(bin()) + .args(["pack", "--jobs"]) + .arg(&jobs) + .args(["--node-mb", "4"]) + .output() + .unwrap(); + assert_eq!( + refuse.status.code(), + Some(3), + "an impossible packing must exit 3: {refuse:?}" + ); + assert!( + String::from_utf8_lossy(&refuse.stderr).contains("REFUSE"), + "missing REFUSE message" + ); + + std::fs::remove_dir_all(&dir).ok(); +} + +#[test] +fn plan_json_emits_machine_readable_prediction() { + let dir = unique_dir("rosalind-planjson"); + let idx = build_index(&dir, "p"); + let out = Command::new(bin()) + .args(["plan", "--index"]) + .arg(&idx) + .args(["--budget-mb", "512", "--json"]) + .output() + .unwrap(); + assert!(out.status.success()); + let j = String::from_utf8_lossy(&out.stdout); + for needle in [ + "\"predicted_peak_rss_bytes\":", + "\"largest_contig_len\":", + "\"verdict\":\"fits\"", + ] { + assert!(j.contains(needle), "plan --json missing {needle}: {j}"); + } + std::fs::remove_dir_all(&dir).ok(); +}