use super::*;
pub(crate) fn sensitive_values_of(config: &WorkflowConfig) -> Vec<String> {
config
.config_meta
.iter()
.filter(|(_, def)| def.sensitive)
.filter_map(|(key, _)| config.config.get(key))
.map(oxo_flow_core::config_impact::config_value_string)
.collect()
}
pub(crate) fn sensitive_keys_of(config: &WorkflowConfig) -> std::collections::HashSet<String> {
config
.config_meta
.iter()
.filter(|(_, def)| def.sensitive)
.map(|(key, _)| key.clone())
.collect()
}
pub(crate) fn adopt_checkpoint_for_workflow(
ck: &mut CheckpointState,
workflow: &Path,
run_log: Option<&SharedRunLog>,
) {
match ck.foreign_workflow(workflow) {
Some(true) => {
diagnostic_narrate(
format_args!(
" {} checkpoint belongs to a different workflow — ignoring its \
completed/failed records so nothing is reused (every rule re-executes)",
"Note:".yellow()
),
run_log,
);
ck.invalidate_reuse_records();
}
Some(false) => {}
None => {
tracing::debug!(
"checkpoint predates workflow-identity tracking; reuse checks stay \
path/mtime/fingerprint based"
);
}
}
}
#[derive(Debug, Clone, Copy)]
pub(crate) enum CheckpointLoadMode {
Execute,
ReadOnlyPreview,
}
pub(crate) fn load_checkpoint_for_command(
checkpoint_path: &Path,
workflow: &Path,
mode: CheckpointLoadMode,
run_log: Option<&SharedRunLog>,
) -> anyhow::Result<(CheckpointState, Vec<String>)> {
let mut state = if checkpoint_path.exists() {
match CheckpointState::load_from_file(checkpoint_path) {
Ok(ck) => ck,
Err(e) => match mode {
CheckpointLoadMode::Execute => {
return Err(anyhow::anyhow!(
"checkpoint at {} is corrupt and cannot be loaded ({}); \
refusing to start — every completed rule would silently re-run. \
Fix or remove the file, then retry.",
checkpoint_path.display(),
e
));
}
CheckpointLoadMode::ReadOnlyPreview => {
eprintln!(
"{} checkpoint at {} is corrupt and cannot be loaded ({}) — \
planning as if every rule never completed; fix or remove the \
file for an accurate plan",
"⚠".yellow(),
checkpoint_path.display(),
e
);
CheckpointState::default()
}
},
}
} else {
if matches!(mode, CheckpointLoadMode::ReadOnlyPreview) {
eprintln!(
"{} no checkpoint at {} — treating every rule as never completed",
"⚠".yellow(),
checkpoint_path.display()
);
}
CheckpointState::default()
};
match mode {
CheckpointLoadMode::Execute => {
adopt_checkpoint_for_workflow(&mut state, workflow, run_log);
}
CheckpointLoadMode::ReadOnlyPreview => {
adopt_checkpoint_for_workflow(&mut state, workflow, None);
}
}
let stale_running: Vec<String> = state.running_rules().iter().cloned().collect();
if !stale_running.is_empty() {
for rule in &stale_running {
state.clear_running(rule);
}
if matches!(mode, CheckpointLoadMode::Execute) {
if let Err(e) = state.save_to_file(checkpoint_path) {
tracing::warn!("Failed to persist cleaned checkpoint: {e}");
}
}
}
Ok((state, stale_running))
}
pub(crate) fn config_snapshot_drift(
ck: &CheckpointState,
config: &WorkflowConfig,
sensitive_keys: &HashSet<String>,
) -> Vec<String> {
let drifted: Vec<String> = ck
.config_snapshot
.keys()
.filter(|key| {
!oxo_flow_core::config_impact::is_engine_injected_key(key)
&& match config.config.get(key.as_str()) {
Some(value) => {
let new = oxo_flow_core::config_impact::snapshot_value(
value,
sensitive_keys.contains(key.as_str()),
);
&new != ck.config_snapshot.get(*key).unwrap()
}
None => true,
}
})
.cloned()
.collect();
if drifted.is_empty() || ck.completed_rules.is_empty() {
return Vec::new();
}
eprintln!(
"{} effective config differs from the original run's snapshot — \
plan-time gated instances (pair/sample `when`, `{{meta.*}}`) may \
drift from the recorded run. Re-pass the original overrides to \
resume the same instance set:",
"Config drift:".bold().yellow()
);
for key in &drifted {
match config.config.get(key.as_str()) {
Some(value) => {
let new = oxo_flow_core::config_impact::snapshot_value(
value,
sensitive_keys.contains(key.as_str()),
);
let old = ck.config_snapshot.get(key.as_str()).unwrap();
let (old, new) = mask_snapshot_pair(old, &new);
eprintln!(" {key}: {old} → {new}");
}
None => eprintln!(" {key}: (removed from this invocation)"),
}
}
drifted
}
fn mask_snapshot_pair(old: &str, new: &str) -> (String, String) {
if old.starts_with("sha256:") || new.starts_with("sha256:") {
("****".to_string(), "****".to_string())
} else {
(old.to_string(), new.to_string())
}
}
pub(crate) struct InputManifestPhase {
pub force_rules: Vec<String>,
pub protected_drift_lines: Vec<String>,
pub protected_drift: bool,
}
#[allow(clippy::too_many_arguments)] pub(crate) fn apply_input_manifest_phase(
ck: &mut CheckpointState,
config: &WorkflowConfig,
dag: &WorkflowDag,
order: &[String],
workdir: &Path,
wildcard_values: &HashMap<String, String>,
checkpoint_path: &Path,
run_log: Option<&SharedRunLog>,
) -> InputManifestPhase {
let mut force_rules: Vec<String> = Vec::new();
let (
mismatched,
missing_inputs,
baselined,
sample_selection_driven,
input_diffs,
symlink_notices,
) = crate::commands::run_preview::detect_input_manifest_invalidations(
ck,
config,
dag,
order,
workdir,
wildcard_values,
);
if !sample_selection_driven.is_empty() {
let names: Vec<&str> = sample_selection_driven.iter().map(String::as_str).collect();
diagnostic_narrate(
format_args!(
" {} skipped {} whose inputs only changed with the --samples selection: {} — outputs still cover the previous run's full sample set; use --rerun to regenerate with the new selection",
"⚠".yellow(),
oxo_flow_core::plural(names.len(), "rule", "rules"),
names.join(", ")
),
run_log,
);
}
if !missing_inputs.is_empty() {
let upstream = crate::commands::run_preview::cascade_up(ck, dag, &missing_inputs);
force_rules.extend(upstream.iter().cloned());
if upstream.is_empty() {
let lines = missing_source_lines(&missing_inputs, config, workdir, wildcard_values);
diagnostic_narrate(
format_args!(
" {} {} missing and cannot be regenerated in this run — affected rule{}:\n{}",
"⚠".yellow(),
oxo_flow_core::plural(
missing_inputs.len(),
"required input",
"required inputs"
),
if missing_inputs.len() == 1 { ":" } else { "s:" },
lines.join("\n")
),
run_log,
);
} else {
diagnostic_narrate(
format_args!(
" {} missing intermediate inputs — re-running {}: {}",
"↻".yellow(),
oxo_flow_core::plural(upstream.len(), "producer rule", "producer rules"),
upstream.join(", ")
),
run_log,
);
}
}
if !mismatched.is_empty() {
let invalidated =
crate::commands::run_preview::invalidate_with_downstream(ck, dag, &mismatched);
force_rules.extend(invalidated.iter().cloned());
let prior_failures = !ck.failed_rules.is_empty();
diagnostic_narrate(
format_args!(
" {} input changes invalidated {}: {}{}",
"↻".yellow(),
oxo_flow_core::plural(invalidated.len(), "rule", "rules"),
invalidated.join(", "),
if prior_failures {
" — the previous run failed after some rules wrote into shared input dirs (logs/ QC dirs); their inputs changed on disk, so affected rules re-run. This is expected invalidation, not lost credit"
} else {
""
}
),
run_log,
);
let mut diff_names: Vec<&String> = mismatched.iter().collect();
diff_names.sort();
for name in diff_names {
if let Some(lines) = input_diffs.get(name) {
for line in lines {
diagnostic_narrate(format_args!(" {} {}", "·".yellow(), line), run_log);
}
}
}
}
if !symlink_notices.is_empty() {
for line in &symlink_notices {
diagnostic_narrate(format_args!(" {} {}", "ℹ".cyan(), line), run_log);
}
}
if baselined > 0 {
diagnostic_narrate(
format_args!(
" Note: checkpoint predates input tracking: recorded baseline input manifests for {} completed; future input changes will invalidate them automatically",
oxo_flow_core::plural(baselined, "rule", "rules")
),
run_log,
);
}
let (protected_drift_lines, protected_baselined, protected_warnings) =
crate::commands::run_preview::detect_protected_tree_drift(ck, config, workdir);
let protected_drift = !protected_drift_lines.is_empty();
for line in &protected_warnings {
diagnostic_narrate(format_args!(" {} {}", "⚠".yellow(), line), run_log);
}
if protected_baselined > 0 {
diagnostic_narrate(
format_args!(
" Note: {} — future changes will be flagged by plan/dry-run/run",
oxo_flow_core::plural(
protected_baselined,
"baseline recorded for protected tree",
"baselines recorded for protected trees"
)
),
run_log,
);
}
if protected_drift {
diagnostic_narrate(
format_args!(
" {} PROTECTED TREE DRIFT detected in {} file(s):",
"✗".red().bold(),
protected_drift_lines.len()
),
run_log,
);
for line in &protected_drift_lines {
diagnostic_narrate(format_args!(" {} {}", "·".red(), line), run_log);
}
}
if (!mismatched.is_empty() || baselined > 0 || protected_baselined > 0 || protected_drift)
&& let Err(e) = ck.save_to_file(checkpoint_path)
{
tracing::warn!(
error = %e,
"failed to save checkpoint after input-manifest/protected-tree detection"
);
}
InputManifestPhase {
force_rules,
protected_drift_lines,
protected_drift,
}
}
pub(crate) fn print_truncated_list(label: &str, names: &[String]) -> String {
let shown: Vec<&str> = names.iter().take(3).map(String::as_str).collect();
let extra = names.len().saturating_sub(3);
let suffix = if extra > 0 {
format!(", … (+{extra} more)")
} else {
String::new()
};
format!(" {label} {}{}", shown.join(", "), suffix)
}
#[allow(clippy::too_many_arguments)] pub(crate) fn print_config_change_summary(
report: &ConfigChangeReport,
old_snapshot: &std::collections::BTreeMap<String, String>,
config: &WorkflowConfig,
sensitive_keys: &HashSet<String>,
order: &[String],
completed_in_run: usize,
rerun: bool,
run_log: Option<&SharedRunLog>,
) {
if rerun {
return;
}
if report.is_legacy {
diagnostic_narrate(
format_args!(
" {} checkpoint predates config tracking: recorded a baseline snapshot; \
future config changes will invalidate affected rules automatically",
"Note:".yellow()
),
run_log,
);
return;
}
if report.changed_keys.is_empty()
&& report.added_keys.is_empty()
&& report.removed_keys.is_empty()
&& report.fingerprint_mismatches.is_empty()
&& report.when_flip_invalidated.is_empty()
{
return;
}
let mut lines: Vec<String> = vec!["Config change:".bold().cyan().to_string()];
for key in &report.changed_keys {
if sensitive_keys.contains(key) {
lines.push(format!(" {key}: **** → ****"));
} else {
let old = old_snapshot.get(key).map(String::as_str).unwrap_or("?");
let new = config
.config
.get(key)
.map(config_value_string)
.unwrap_or_else(|| "?".to_string());
lines.push(format!(" {key}: {old} → {new}"));
}
}
for key in &report.added_keys {
lines.push(format!(" {key}: (new key)"));
}
for key in &report.removed_keys {
lines.push(format!(" {key}: (removed)"));
}
if !report.fingerprint_mismatches.is_empty() {
lines.push(print_truncated_list(
"rule definition changed:",
&report.fingerprint_mismatches,
));
}
if !report.when_flip_invalidated.is_empty() {
lines.push(print_truncated_list(
"when-condition flipped:",
&report.when_flip_invalidated,
));
}
if !report.when_gate_exempt.is_empty() {
lines.push(print_truncated_list(
"when-condition unchanged, reused:",
&report.when_gate_exempt,
));
}
let order_set: HashSet<&str> = order.iter().map(String::as_str).collect();
let rerun_this_run = report
.invalidated
.iter()
.filter(|name| order_set.contains(name.as_str()))
.count();
lines.push(format!(
" → invalidated {} ({} directly affected), re-running {}/{} this run, skipping {}",
report.invalidated.len(),
report.directly_affected.len(),
rerun_this_run,
order.len(),
completed_in_run,
));
diagnostic_narrate(format_args!("{}", lines.join("\n")), run_log);
}
pub(crate) fn quote_shell_path(path: &str) -> String {
format!("'{}'", path.replace('\'', "'\\''"))
}
pub(crate) fn substitute_source_placeholder(build_cmd: &str, expanded_source: &str) -> String {
build_cmd.replace("{source}", "e_shell_path(expanded_source))
}
pub(crate) fn known_modules_hint(
module_rules: &std::collections::HashMap<String, Vec<String>>,
) -> String {
if module_rules.is_empty() {
"(none — no [[include]] modules)".to_string()
} else {
module_rules
.keys()
.map(|k| k.as_str())
.collect::<Vec<_>>()
.join(", ")
}
}
pub(crate) fn module_groups_for_rules<'a>(
names: &[&'a str],
module_rules: &'a HashMap<String, Vec<String>>,
) -> Option<Vec<(Option<&'a str>, Vec<&'a str>)>> {
if module_rules.is_empty() {
return None;
}
let module_members: Vec<(&str, &str)> = module_rules
.iter()
.flat_map(|(module, members)| {
members
.iter()
.map(move |member| (member.as_str(), module.as_str()))
})
.collect();
let mut groups: Vec<(Option<&str>, Vec<&str>)> = Vec::new();
let mut group_index: HashMap<Option<&str>, usize> = HashMap::new();
for &name in names {
let group = if let Some((prefix, _)) = name.split_once("::") {
Some(prefix)
} else {
oxo_flow_core::dag::owning_module(name, &module_members)
};
let idx = *group_index.entry(group).or_insert_with(|| {
groups.push((group, Vec::new()));
groups.len() - 1
});
groups[idx].1.push(name);
}
Some(groups)
}
pub(crate) fn run_group_title(group: Option<&str>) -> String {
match group {
Some(module) => oxo_flow_core::stage::module_display(module),
None => "Main workflow".to_string(),
}
}
pub(crate) fn undefined_config_findings(config: &WorkflowConfig) -> Vec<String> {
config
.rules
.iter()
.flat_map(|rule| oxo_flow_core::format::undefined_config_refs(rule, config))
.map(|d| {
format!(
"rule '{}': {} ({})",
d.rule.as_deref().unwrap_or("<unknown>"),
d.message,
d.suggestion.unwrap_or_default()
)
})
.collect()
}
pub(crate) fn missing_source_findings(
config: &WorkflowConfig,
dag: &WorkflowDag,
order: &[String],
workdir: &Path,
workflow_dir: &Path,
wildcard_values: &HashMap<String, String>,
) -> Vec<String> {
let reference_outputs: HashSet<String> = config
.references
.iter()
.map(|r| {
oxo_flow_core::executor::checkpoint::expand_config_in_path(&r.output, wildcard_values)
})
.collect();
let sample_search_dir = config.workflow.sample_pattern.as_ref().and_then(|pattern| {
let expanded =
oxo_flow_core::executor::checkpoint::expand_config_in_path(pattern, wildcard_values);
let path = Path::new(&expanded);
if path.is_absolute() {
path.parent().map(|p| p.to_path_buf())
} else {
let joined = workflow_dir.join(path);
Some(
joined
.parent()
.map(|p| p.to_path_buf())
.unwrap_or_else(|| workflow_dir.to_path_buf()),
)
}
});
let when_skipped: HashSet<String> = config
.when_skipped_outputs
.iter()
.map(|recorded| {
oxo_flow_core::executor::checkpoint::expand_config_in_path(recorded, wildcard_values)
})
.collect();
let mut findings = Vec::new();
for name in order {
let Some(rule) = config.get_rule(name) else {
continue;
};
if crate::commands::run_preview::when_condition_false(rule, config, wildcard_values) {
continue;
}
if config.scatter_fan_out_is_empty(rule) {
continue;
}
if rule.optional.is_optional() {
continue;
}
let missing = oxo_flow_core::executor::checkpoint::missing_input_patterns(
rule,
workdir,
wildcard_values,
);
for pattern in missing {
if reference_outputs.contains(&pattern) {
continue;
}
if dag.producer_of(&pattern).is_some() {
continue;
}
if when_skipped.contains(&pattern) {
continue;
}
let absolute = Path::new(&pattern).is_absolute();
if !absolute
&& sample_search_dir
.as_ref()
.is_some_and(|dir| dir.join(&pattern).exists())
{
continue;
}
findings.push(format!(
"rule '{name}': input '{pattern}' does not exist and no rule produces it"
));
}
}
findings
}
pub(crate) fn missing_source_lines(
missing_inputs: &HashSet<String>,
config: &WorkflowConfig,
workdir: &Path,
wildcard_values: &HashMap<String, String>,
) -> Vec<String> {
let mut names: Vec<&String> = missing_inputs.iter().collect();
names.sort();
names
.into_iter()
.map(|name| {
let patterns = config
.get_rule(name)
.map(|rule| {
oxo_flow_core::executor::checkpoint::missing_input_patterns(
rule,
workdir,
wildcard_values,
)
})
.unwrap_or_default();
let listing = if patterns.is_empty() {
"(unresolvable)".to_string()
} else {
patterns.join(", ")
};
format!(" {name} — missing: {listing}")
})
.collect()
}
#[cfg(test)]
mod tests {
use super::*;
use oxo_flow_core::dag::WorkflowDag;
use std::collections::{HashMap, HashSet};
fn workflow_config_with(toml_body: &str) -> WorkflowConfig {
let toml = format!(
r#"
[workflow]
name = "t"
version = "1.0"
{toml_body}
"#
);
WorkflowConfig::parse(&toml).expect("test workflow config parses")
}
#[test]
fn snapshot_drift_names_changed_and_removed_keys_only() {
let mut ck = CheckpointState::default();
ck.completed_rules.insert("gen".to_string());
ck.config_snapshot
.insert("threads".to_string(), "4".to_string());
ck.config_snapshot
.insert("same".to_string(), "keep".to_string());
ck.config_snapshot
.insert("samples_list".to_string(), "old".to_string());
ck.config_snapshot
.insert("gone".to_string(), "x".to_string());
let config = workflow_config_with(
r#"
[config]
threads = "8"
same = "keep"
samples_list = "new"
"#,
);
let sensitive = std::collections::HashSet::new();
let drifted = config_snapshot_drift(&ck, &config, &sensitive);
assert_eq!(drifted, vec!["gone".to_string(), "threads".to_string()]);
}
#[test]
fn snapshot_drift_stays_silent_without_completed_rules() {
let ck = CheckpointState::default();
let config = workflow_config_with("[config]\nthreads = \"8\"\n");
let sensitive = std::collections::HashSet::new();
assert!(config_snapshot_drift(&ck, &config, &sensitive).is_empty());
}
#[test]
fn snapshot_pair_hash_is_masked_on_either_side() {
assert_eq!(
mask_snapshot_pair("sha256:abc", "sha256:def"),
("****".to_string(), "****".to_string())
);
assert_eq!(
mask_snapshot_pair("plain", "sha256:def"),
("****".to_string(), "****".to_string())
);
assert_eq!(
mask_snapshot_pair("plain", "other"),
("plain".to_string(), "other".to_string())
);
}
fn corrupt_checkpoint(dir: &std::path::Path) -> std::path::PathBuf {
let path = dir.join(".oxo-flow");
std::fs::create_dir_all(&path).unwrap();
let file = path.join("checkpoint.json");
std::fs::write(&file, "{ not json at all").unwrap();
file
}
#[test]
fn execute_mode_refuses_to_start_on_a_corrupt_checkpoint() {
let tmp = tempfile::tempdir().unwrap();
let path = corrupt_checkpoint(tmp.path());
let err = load_checkpoint_for_command(
&path,
std::path::Path::new("t.oxoflow"),
CheckpointLoadMode::Execute,
None,
)
.unwrap_err();
assert!(format!("{err}").contains("refusing to start"), "{err}");
}
#[test]
fn preview_mode_degrades_on_corruption_and_stays_read_only() {
let tmp = tempfile::tempdir().unwrap();
let path = corrupt_checkpoint(tmp.path());
let before = std::fs::read_to_string(&path).unwrap();
let (state, stale) = load_checkpoint_for_command(
&path,
std::path::Path::new("t.oxoflow"),
CheckpointLoadMode::ReadOnlyPreview,
None,
)
.unwrap();
assert!(state.completed_rules.is_empty());
assert!(stale.is_empty());
assert_eq!(std::fs::read_to_string(&path).unwrap(), before);
}
#[test]
fn stale_running_entries_are_cleared_once_per_mode() {
for mode in [
CheckpointLoadMode::Execute,
CheckpointLoadMode::ReadOnlyPreview,
] {
let tmp = tempfile::tempdir().unwrap();
let dir = tmp.path().join(".oxo-flow");
std::fs::create_dir_all(&dir).unwrap();
let path = dir.join("checkpoint.json");
let mut seeded = CheckpointState::default();
seeded.mark_running("gen");
seeded.save_to_file(&path).unwrap();
let (state, stale) =
load_checkpoint_for_command(&path, std::path::Path::new("t.oxoflow"), mode, None)
.unwrap();
assert_eq!(stale, vec!["gen".to_string()]);
assert!(state.running_rules().is_empty());
let on_disk = CheckpointState::load_from_file(&path).unwrap();
match mode {
CheckpointLoadMode::Execute => {
assert!(on_disk.running_rules().is_empty());
}
CheckpointLoadMode::ReadOnlyPreview => {
assert!(on_disk.running_rules().contains("gen"));
}
}
}
}
fn config_with_rules(rules_toml: &str) -> oxo_flow_core::config::WorkflowConfig {
oxo_flow_core::config::WorkflowConfig::parse(rules_toml).unwrap()
}
fn source_gate_fixture(
dir: &std::path::Path,
) -> (
oxo_flow_core::config::WorkflowConfig,
WorkflowDag,
Vec<String>,
HashMap<String, String>,
) {
let toml = r#"
[workflow]
name = "t"
version = "1.0"
[config]
threshold = 5
empty = []
[[rules]]
name = "gen"
input = []
output = ["out.txt"]
shell = "echo x > {output[0]}"
[[rules]]
name = "use"
input = ["out.txt", "ext.txt"]
output = ["final.txt"]
shell = "cat {input} > {output[0]}"
[[rules]]
name = "gated"
input = ["gated_missing.txt"]
output = ["gated.txt"]
when = "config.threshold >= 10"
shell = "true"
[[rules]]
name = "scatter_zero"
input = ["scatter_missing.txt"]
output = ["scatter.txt"]
scatter = { variable = "chunk", values_from = "config.empty" }
shell = "true"
[[rules]]
name = "opt"
input = ["optional_missing.txt"]
output = ["opt.txt"]
optional = true
shell = "true"
"#;
let mut config = config_with_rules(toml);
config.apply_defaults();
let dag = WorkflowDag::from_rules_with_config(
&config.rules,
&config_placeholder_values(&config.config),
)
.unwrap();
let order = dag.execution_order().unwrap();
let wildcard_values = config_placeholder_values(&config.config);
std::fs::create_dir_all(dir).unwrap();
(config, dag, order, wildcard_values)
}
fn source_gate_call(
config: &oxo_flow_core::config::WorkflowConfig,
dag: &WorkflowDag,
order: &[String],
workdir: &std::path::Path,
workflow_dir: &std::path::Path,
wildcard_values: &HashMap<String, String>,
) -> Vec<String> {
missing_source_findings(config, dag, order, workdir, workflow_dir, wildcard_values)
}
#[test]
fn missing_source_gate_flags_only_unproduced_required_inputs() {
let dir = tempfile::tempdir().unwrap();
let (config, dag, order, wildcard_values) = source_gate_fixture(dir.path());
let findings = source_gate_call(
&config,
&dag,
&order,
dir.path(),
dir.path(),
&wildcard_values,
);
assert_eq!(
findings,
vec!["rule 'use': input 'ext.txt' does not exist and no rule produces it".to_string()],
"'out.txt' has a producer (gen) and belongs to the cascade path; when-false, \
zero-fan-out scatter, and optional rules are all skipped"
);
}
#[test]
fn missing_source_gate_passes_when_sources_exist() {
let dir = tempfile::tempdir().unwrap();
let (config, dag, order, wildcard_values) = source_gate_fixture(dir.path());
std::fs::write(dir.path().join("ext.txt"), b"data").unwrap();
let findings = source_gate_call(
&config,
&dag,
&order,
dir.path(),
dir.path(),
&wildcard_values,
);
assert!(findings.is_empty(), "{findings:?}");
}
#[test]
fn missing_source_gate_silent_with_full_checkpoint_is_the_regressed_scenario() {
let dir = tempfile::tempdir().unwrap();
let (config, dag, order, wildcard_values) = source_gate_fixture(dir.path());
std::fs::write(dir.path().join("out.txt"), b"stale").unwrap();
std::fs::write(dir.path().join("final.txt"), b"stale").unwrap();
let findings = source_gate_call(
&config,
&dag,
&order,
dir.path(),
dir.path(),
&wildcard_values,
);
assert!(!findings.is_empty(), "missing ext.txt must gate the run");
}
#[test]
fn missing_source_gate_skips_reference_outputs() {
let dir = tempfile::tempdir().unwrap();
let toml = r#"
[workflow]
name = "refgate"
[[references]]
name = "idx"
source = "genome.fa"
output = "genome.idx"
build = "cat genome.fa > genome.idx"
[[rules]]
name = "use_ref"
input = ["genome.idx"]
output = ["result.txt"]
shell = "cat {input} > {output}"
"#;
let mut config = config_with_rules(toml);
config.apply_defaults();
let dag = WorkflowDag::from_rules_with_config(
&config.rules,
&config_placeholder_values(&config.config),
)
.unwrap();
let order = dag.execution_order().unwrap();
let wildcard_values = config_placeholder_values(&config.config);
std::fs::write(dir.path().join("genome.fa"), b"ACGT").unwrap();
let findings = source_gate_call(
&config,
&dag,
&order,
dir.path(),
dir.path(),
&wildcard_values,
);
assert!(
findings.is_empty(),
"reference-built output must not gate the run: {findings:?}"
);
}
#[test]
fn missing_source_gate_accepts_sample_pattern_dir_inputs() {
let dir = tempfile::tempdir().unwrap(); let data = tempfile::tempdir().unwrap(); let toml = format!(
r#"
[workflow]
name = "samplegate"
sample_pattern = "{root}/data/{{{{sample}}}}.fq"
[[rules]]
name = "qc"
input = ["{{{{sample}}}}.fq"]
output = ["qc_{{{{sample}}}}.txt"]
shell = "cat {{{{input}}}} > {{{{output}}}}"
"#,
root = data.path().display(),
);
let mut config = config_with_rules(&toml);
config.apply_defaults();
std::fs::create_dir_all(data.path().join("data")).unwrap();
std::fs::write(data.path().join("data").join("S1.fq"), b"read").unwrap();
let dag = WorkflowDag::from_rules_with_config(
&config.rules,
&config_placeholder_values(&config.config),
)
.unwrap();
let order = dag.execution_order().unwrap();
let wildcard_values = config_placeholder_values(&config.config);
let findings = source_gate_call(
&config,
&dag,
&order,
dir.path(),
dir.path(),
&wildcard_values,
);
assert!(
findings.is_empty(),
"input present in the sample_pattern dir must not gate: {findings:?}"
);
}
#[test]
fn missing_source_gate_skips_input_of_when_false_shared_output_producer() {
let dir = tempfile::tempdir().unwrap();
let toml = r#"
[workflow]
name = "sharedout"
version = "1.0"
[config]
mode = "tumoronly"
[[rules]]
name = "consumer"
input = ["shared.txt"]
output = ["final.txt"]
shell = "cat {input} > {output[0]}"
"#;
let mut config = config_with_rules(toml);
config.apply_defaults();
config.when_skipped_outputs.insert("shared.txt".to_string());
let wildcard_values = config_placeholder_values(&config.config);
let dag = WorkflowDag::from_rules_with_config(&config.rules, &wildcard_values).unwrap();
assert!(
dag.producer_of("shared.txt").is_none(),
"fixture precondition: the vetoed side must leave shared.txt producerless"
);
let order = dag.execution_order().unwrap();
let findings = source_gate_call(
&config,
&dag,
&order,
dir.path(),
dir.path(),
&wildcard_values,
);
assert!(
findings.is_empty(),
"the gated-off producer's input must not gate the satisfiable consumer: {findings:?}"
);
}
#[test]
fn missing_source_gate_skips_pattern_recorded_in_when_skipped_outputs() {
let dir = tempfile::tempdir().unwrap();
let toml = r#"
[workflow]
name = "skipped"
version = "1.0"
[config]
mode = "off"
[[rules]]
name = "consumer"
input = ["morphed.txt"]
output = ["final.txt"]
shell = "cat {input} > {output[0]}"
"#;
let mut config = config_with_rules(toml);
config.apply_defaults();
config
.when_skipped_outputs
.insert("morphed.txt".to_string());
let wildcard_values = config_placeholder_values(&config.config);
let dag = WorkflowDag::from_rules_with_config(&config.rules, &wildcard_values).unwrap();
assert!(
dag.producer_of("morphed.txt").is_none(),
"fixture precondition: the vetoed producer must leave morphed.txt producerless"
);
let order = dag.execution_order().unwrap();
let findings = source_gate_call(
&config,
&dag,
&order,
dir.path(),
dir.path(),
&wildcard_values,
);
assert!(
findings.is_empty(),
"a when_skipped_outputs pattern must not gate its consumers: {findings:?}"
);
}
#[test]
fn missing_source_gate_expands_config_tokens_in_when_skipped_outputs() {
let dir = tempfile::tempdir().unwrap();
let toml = r#"
[workflow]
name = "cfgtoken"
version = "1.0"
[config]
genome = "hg38"
[[rules]]
name = "consumer"
input = ["ref_hg38.fai"]
output = ["final.txt"]
shell = "cat {input} > {output[0]}"
"#;
let mut config = config_with_rules(toml);
config.apply_defaults();
config
.when_skipped_outputs
.insert("ref_{config.genome}.fai".to_string());
let wildcard_values = config_placeholder_values(&config.config);
let dag = WorkflowDag::from_rules_with_config(&config.rules, &wildcard_values).unwrap();
let order = dag.execution_order().unwrap();
let findings = source_gate_call(
&config,
&dag,
&order,
dir.path(),
dir.path(),
&wildcard_values,
);
assert!(
findings.is_empty(),
"a raw {{config.*}} recorded entry must match after compare-time expansion: {findings:?}"
);
}
#[test]
fn missing_source_lines_lists_rule_names_and_patterns() {
let dir = tempfile::tempdir().unwrap();
let (config, _dag, _order, wildcard_values) = source_gate_fixture(dir.path());
let mut missing = HashSet::new();
missing.insert("use".to_string());
let lines = missing_source_lines(&missing, &config, dir.path(), &wildcard_values);
assert_eq!(lines.len(), 1);
assert!(lines[0].contains("use"), "{:?}", lines[0]);
assert!(lines[0].contains("ext.txt"), "{:?}", lines[0]);
}
#[test]
fn source_placeholder_quotes_paths_with_spaces() {
let rendered =
substitute_source_placeholder("STAR --genomeDir {source}", "refs/genome data/hg38");
assert_eq!(rendered, "STAR --genomeDir 'refs/genome data/hg38'");
}
#[test]
fn source_placeholder_escapes_embedded_quotes() {
let rendered = substitute_source_placeholder("--in {source}", "dir/we'ird");
assert_eq!(rendered, "--in 'dir/we'\\''ird'");
}
#[test]
fn source_placeholder_without_placeholder_is_unchanged() {
let rendered = substitute_source_placeholder("echo built", "ignored");
assert_eq!(rendered, "echo built");
}
#[test]
fn known_modules_hint_names_the_empty_case() {
let empty: std::collections::HashMap<String, Vec<String>> =
std::collections::HashMap::new();
assert_eq!(
known_modules_hint(&empty),
"(none — no [[include]] modules)"
);
let mut map: std::collections::HashMap<String, Vec<String>> =
std::collections::HashMap::new();
map.insert("qc".to_string(), vec!["qc".to_string()]);
assert_eq!(known_modules_hint(&map), "qc");
}
}