use crate::commands::{print_banner, resolve_workflow, samples};
use anyhow::{Context, Result};
use colored::Colorize;
use oxo_flow_core::config::WorkflowConfig;
use oxo_flow_core::config_impact::{ConfigChangeReport, config_value_string};
use oxo_flow_core::dag::WorkflowDag;
use oxo_flow_core::executor::{
CheckpointState, ExecutorConfig, LocalExecutor, WorkdirLock, output_tail_bytes_from_env,
};
use oxo_flow_core::rule::{Rule, parse_duration_secs};
use oxo_flow_core::wildcard;
use std::collections::{HashMap, HashSet};
use std::io::Write as _;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use tokio::sync::Mutex;
fn resolve_default_targets(config: &WorkflowConfig, target: &[String]) -> Vec<String> {
if !target.is_empty() {
return target.to_vec();
}
let marked: Vec<String> = config
.rules
.iter()
.filter(|r| r.target)
.map(|r| r.name.clone())
.collect();
if marked.is_empty() {
target.to_vec()
} else {
marked
}
}
fn render_benchmark_json(b: &oxo_flow_core::executor::checkpoint::BenchmarkRecord) -> String {
serde_json::json!({
"rule": b.rule,
"wall_time_secs": b.wall_time_secs,
"max_memory_mb": b.max_memory_mb,
"memory_limit_mb": b.memory_limit_mb,
"cpu_seconds": b.cpu_seconds,
"retries": b.retries,
})
.to_string()
}
#[cfg(test)]
mod default_target_tests {
use super::*;
#[test]
fn resolve_default_targets_prefers_explicit_selection() {
let config = WorkflowConfig::parse(
r#"
[workflow]
name = "t"
[[rules]]
name = "a"
target = true
shell = "true"
[[rules]]
name = "b"
shell = "true"
"#,
)
.unwrap();
assert_eq!(
resolve_default_targets(&config, &["b".to_string()]),
vec!["b".to_string()]
);
assert_eq!(resolve_default_targets(&config, &[]), vec!["a".to_string()]);
assert_eq!(
resolve_default_targets(&config, &["m_rule".to_string()]),
vec!["m_rule".to_string()]
);
}
#[test]
fn resolve_default_targets_full_dag_without_marks() {
let config = WorkflowConfig::parse(
r#"
[workflow]
name = "t"
[[rules]]
name = "a"
shell = "true"
"#,
)
.unwrap();
assert!(resolve_default_targets(&config, &[]).is_empty());
}
#[test]
fn render_benchmark_json_carries_sampled_fields() {
let b = oxo_flow_core::executor::checkpoint::BenchmarkRecord {
rule: "fastqc".into(),
wall_time_secs: 2.75,
max_memory_mb: Some(512),
memory_limit_mb: Some(1024),
cpu_seconds: Some(2.1),
retries: 0,
recorded_as: None,
};
let json = render_benchmark_json(&b);
assert!(json.contains(r#""rule":"fastqc""#), "{json}");
assert!(json.contains(r#""wall_time_secs":2.75"#), "{json}");
assert!(json.contains(r#""max_memory_mb":512"#), "{json}");
}
}
pub(crate) fn config_placeholder_values(
config: &HashMap<String, toml::Value>,
) -> HashMap<String, String> {
oxo_flow_core::config_impact::config_placeholder_map(config)
}
type SharedRunLog = Arc<std::sync::Mutex<Option<crate::logging::RunLogGuard>>>;
fn write_run_log_line(run_log: &SharedRunLog, text: &str) {
let mut slot = run_log
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
if let Some(log) = slot.as_mut() {
let line = format!("{}\n", text.trim_start());
if let Err(e) = log.write_all(line.as_bytes()) {
tracing::debug!(error = %e, "narrative line could not be written to the run log");
}
}
}
fn progress_narrate(msg: std::fmt::Arguments<'_>, run_log: &SharedRunLog) {
if !crate::commands::is_quiet() {
eprintln!("{msg}");
}
write_run_log_line(run_log, &strip_ansi(&msg.to_string()));
}
fn diagnostic_narrate(msg: std::fmt::Arguments<'_>, run_log: Option<&SharedRunLog>) {
eprintln!("{msg}");
if let Some(run_log) = run_log {
write_run_log_line(run_log, &strip_ansi(&msg.to_string()));
}
}
fn narrate_source_link_report(
report: &crate::commands::workdir_sources::SourceLinkReport,
run_log: Option<&SharedRunLog>,
) {
if let Some(note) = &report.skipped_note {
diagnostic_narrate(format_args!("workdir source linking: {}", note), run_log);
return;
}
if report.is_empty() {
return;
}
if !report.linked.is_empty() {
diagnostic_narrate(
format_args!(
"workdir source linking: created {} symlink(s) into the workdir (repo stays the single source of truth): {}",
report.linked.len(),
report.linked.join(", ")
),
run_log,
);
}
if !report.sidecars.is_empty() {
diagnostic_narrate(
format_args!(
"workdir source linking: linked {} tool-convention index sidecar(s) next to linked file(s) (bwa/samtools/picard auto-discovery — not spelled out in any shell command): {}",
report.sidecars.len(),
report.sidecars.join(", ")
),
run_log,
);
}
if !report.converted.is_empty() {
diagnostic_narrate(
format_args!(
"workdir source linking: replaced {} byte-identical workdir duplicate(s) with symlink(s) into the repo: {}",
report.converted.len(),
report.converted.join(", ")
),
run_log,
);
}
for path in &report.conflicts {
diagnostic_narrate(
format_args!(
"{} workdir source linking: '{}' differs from the workflow-repo file — keeping the workdir copy for this run (remove the copy or update the repo to link it)",
"Warning:".yellow(),
path
),
run_log,
);
}
if !report.undeclared.is_empty() {
diagnostic_narrate(
format_args!(
"{} workdir source linking: shell command(s) reference workflow-repo file(s) that no rule declares as inputs — they were NOT linked (declaring the path as a rule input links it automatically):\n {}",
"Note:".yellow(),
report.undeclared.join("\n ")
),
run_log,
);
}
for err in &report.errors {
diagnostic_narrate(
format_args!("{} workdir source linking: {}", "Warning:".yellow(), err),
run_log,
);
}
}
const SPAWN_WATCHDOG_TICK_SECS: u64 = 30;
const DEFAULT_SPAWN_WATCHDOG_SECS: u64 = 600;
const SPAWN_WATCHDOG_ENV: &str = "OXO_FLOW_SPAWN_WATCHDOG_SECS";
const DISK_PRESSURE_TICK_SECS: u64 = 60;
const DISK_HOLD_ABORT_ROUNDS: u32 = 2;
const DISK_TICK_ENV: &str = "OXO_FLOW_DISK_TICK_SECS";
const DISK_HOLD_ROUNDS_ENV: &str = "OXO_FLOW_DISK_HOLD_ROUNDS";
fn disk_tick_secs() -> u64 {
parse_disk_tick_secs(std::env::var(DISK_TICK_ENV).ok().as_deref())
}
fn parse_disk_tick_secs(raw: Option<&str>) -> u64 {
raw.and_then(|v| v.parse::<u64>().ok())
.filter(|&s| s > 0)
.unwrap_or(DISK_PRESSURE_TICK_SECS)
}
fn disk_hold_abort_rounds() -> u32 {
parse_disk_hold_abort_rounds(std::env::var(DISK_HOLD_ROUNDS_ENV).ok().as_deref())
}
fn parse_disk_hold_abort_rounds(raw: Option<&str>) -> u32 {
raw.and_then(|v| v.parse::<u32>().ok())
.filter(|&s| s > 0)
.unwrap_or(DISK_HOLD_ABORT_ROUNDS)
}
fn spawn_watchdog_threshold() -> std::time::Duration {
parse_spawn_watchdog_threshold(std::env::var(SPAWN_WATCHDOG_ENV).ok().as_deref())
}
fn parse_spawn_watchdog_threshold(raw: Option<&str>) -> std::time::Duration {
raw.and_then(|v| v.parse::<u64>().ok())
.filter(|&s| s > 0)
.map(std::time::Duration::from_secs)
.unwrap_or(std::time::Duration::from_secs(DEFAULT_SPAWN_WATCHDOG_SECS))
}
fn strip_ansi(text: &str) -> String {
let bytes = text.as_bytes();
let mut plain = String::with_capacity(text.len());
let mut i = 0;
while i < bytes.len() {
if bytes[i] == 0x1b {
if i + 1 < bytes.len() && bytes[i + 1] == b'[' {
i += 2;
while i < bytes.len() && !(0x40..=0x7e).contains(&bytes[i]) {
i += 1;
}
if i < bytes.len() {
i += 1;
}
} else {
i += 1;
}
} else {
let start = i;
while i < bytes.len() && bytes[i] != 0x1b {
i += 1;
}
plain.push_str(&text[start..i]);
}
}
plain
}
fn emit_execution_event(event: oxo_flow_core::executor::ExecutionEvent) {
if std::env::var_os("OXO_FLOW_STDERR_ALREADY_REDIRECTED").is_none() {
tracing::info!(target: "execution_event", "{}", event.to_json_log());
}
}
#[derive(Debug, Clone, Copy)]
pub struct InterruptSignal {
pub name: &'static str,
pub signum: i32,
}
impl InterruptSignal {
pub fn exit_code(&self) -> i32 {
128 + self.signum
}
}
async fn wait_for_termination_signal() -> Option<InterruptSignal> {
#[cfg(unix)]
{
use tokio::signal::unix::{SignalKind, signal};
let mut sigint = signal(SignalKind::interrupt()).ok()?;
let mut sigterm = signal(SignalKind::terminate()).ok()?;
let mut sighup = signal(SignalKind::hangup()).ok()?;
tokio::select! {
_ = sigint.recv() => Some(InterruptSignal { name: "SIGINT", signum: 2 }),
_ = sigterm.recv() => Some(InterruptSignal { name: "SIGTERM", signum: 15 }),
_ = sighup.recv() => Some(InterruptSignal { name: "SIGHUP", signum: 1 }),
}
}
#[cfg(not(unix))]
{
tokio::signal::ctrl_c().await.ok()?;
Some(InterruptSignal {
name: "SIGINT",
signum: 2,
})
}
}
#[allow(clippy::too_many_arguments)] async fn terminate_on_signal(
signal: InterruptSignal,
executor: &LocalExecutor,
join_set: &mut tokio::task::JoinSet<(
String,
oxo_flow_core::executor::JobStatus,
oxo_flow_core::executor::JobRecord,
)>,
in_flight: impl Iterator<Item = &String>,
checkpoint: &Arc<Mutex<CheckpointState>>,
checkpoint_path: &Path,
json_summary: &mut RunJsonSummary,
config: &WorkflowConfig,
) -> ! {
eprintln!(
"\n{} {} received — stopping in-flight rules and saving the checkpoint",
"Interrupted:".yellow().bold(),
signal.name
);
executor.clear_resource_waiters().await;
let snapshot = executor.active_pids();
tokio::task::spawn_blocking(move || {
for (rule, pid) in snapshot {
if let Err(e) = oxo_flow_core::executor::timeout::kill_process_tree(pid) {
tracing::warn!(rule = %rule, pid, error = %e, "failed to signal in-flight rule on {}", signal.name);
}
}
})
.await
.ok();
join_set.abort_all();
let late = executor.active_pids();
if !late.is_empty() {
tokio::task::spawn_blocking(move || {
for (rule, pid) in late {
if let Err(e) = oxo_flow_core::executor::timeout::kill_process_tree(pid) {
tracing::warn!(rule = %rule, pid, error = %e, "failed to signal late-spawned rule on {}", signal.name);
}
}
})
.await
.ok();
}
let mut ck = checkpoint.lock().await;
for rule in in_flight {
let record = oxo_flow_core::executor::JobRecord {
signal: Some(15), rule: rule.clone(),
status: oxo_flow_core::executor::JobStatus::Failed,
enqueued_at: None,
started_at: None,
finished_at: Some(chrono::Utc::now()),
exit_code: None,
stdout: None,
stderr: None,
command: None,
retries: 0,
skip_reason: Some(format!("interrupted by {}", signal.name)),
max_rss_mb: None,
cpu_seconds: None,
caption: None,
};
ck.record_run(&record, output_tail_bytes_from_env());
ck.mark_failed(rule);
}
if let Err(e) = ck.save_to_file(checkpoint_path) {
tracing::warn!(error = %e, "failed to save checkpoint after interruption");
}
drop(ck);
json_summary.emit("failed", &RunCounts::default(), vec![]);
notify_webhook(
config,
oxo_flow_core::webhook::WebhookEvent::WorkflowFailed,
None,
)
.await;
std::process::exit(signal.exit_code());
}
async fn reclaim_rule_outputs(
candidate: &oxo_flow_core::scheduler::ReclaimCandidate,
workdir: &std::path::Path,
checkpoint: &Arc<Mutex<CheckpointState>>,
checkpoint_path: &Path,
run_log: &SharedRunLog,
) -> anyhow::Result<usize> {
let mut paths = Vec::new();
{
let mut ck = checkpoint.lock().await;
for path in &candidate.paths {
let resolved = workdir.join(path);
if !resolved.exists() {
continue;
}
if let Some(sha) = ck.checksums.remove(path) {
ck.record_cleaned_checksum(path, sha);
}
paths.push(path.clone());
}
if paths.is_empty() {
return Ok(0);
}
ck.tombstones.insert(candidate.rule.clone(), paths.clone());
if let Err(e) = ck.save_to_file_async(checkpoint_path).await {
ck.tombstones.remove(&candidate.rule);
for path in &paths {
if let Some(sha) = ck.cleaned_checksums.remove(path) {
ck.checksums.insert(path.clone(), sha);
}
}
return Err(anyhow::anyhow!(
"persisting tombstone failed: {e} — outputs kept"
));
}
}
let mut unlinked = 0usize;
for path in &paths {
let resolved = workdir.join(path);
let ok = match std::fs::symlink_metadata(&resolved) {
Ok(md) if md.file_type().is_symlink() => std::fs::remove_file(&resolved).is_ok(),
Ok(md) if md.is_dir() => std::fs::remove_dir_all(&resolved).is_ok(),
Ok(_) => std::fs::remove_file(&resolved).is_ok(),
Err(_) => false,
};
if ok {
unlinked += 1;
} else {
diagnostic_narrate(
format_args!(
" {} reclaim unlink failed for '{}' path '{path}'",
"!".yellow(),
candidate.rule
),
Some(run_log),
);
}
}
diagnostic_narrate(
format_args!(
" {} reclaimed temporary outputs of '{}' ({} file(s)) — regenerated on demand via cascade-up",
"⊘".dimmed(),
candidate.rule,
unlinked
),
Some(run_log),
);
Ok(unlinked)
}
#[allow(clippy::too_many_arguments)]
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()
}
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"
);
}
}
}
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)] 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);
}
fn absolutize(path: &Path) -> Result<PathBuf> {
if path.is_absolute() {
Ok(path.to_path_buf())
} else {
Ok(std::env::current_dir()?.join(path))
}
}
fn normalize_dots(path: PathBuf) -> PathBuf {
let mut out = PathBuf::with_capacity(path.as_os_str().len());
for comp in path.components() {
match comp {
std::path::Component::CurDir => {}
std::path::Component::ParentDir => {
out.push(comp);
}
other => out.push(other),
}
}
if out.as_os_str().is_empty() {
PathBuf::from(".")
} else {
out
}
}
const RUN_FLAG_NAMES: &[&str] = &[
"jobs",
"keep-going",
"workdir",
"target",
"module",
"retry",
"timeout",
"log-file",
"resume-failed",
"profile",
"max-threads",
"max-memory",
"max-submitted",
"skip-env-setup",
"skip-ref-build",
"cache-dir",
"provenance",
"bundle",
"yes",
"arg",
"ai-recover",
"ai-max-retries",
"samples",
"rerun",
"no-report-snapshot",
"background",
"json",
"quiet",
"verbose",
"no-color",
"help",
"version",
];
const RUN_SHORT_FLAGS: &[&str] = &["j", "k", "d", "t", "r", "h", "V"];
const DRY_RUN_SHARED_FLAGS: &[&str] = &["cache-dir"];
const REMOVED_RUN_FLAGS: &[(&str, &str)] = &[("sample", "--samples")];
fn validate_config_value(
name: &str,
value: &str,
def: &oxo_flow_core::config::ConfigDef,
from_cli: bool,
) -> anyhow::Result<()> {
if let Some(ref choices) = def.choices
&& !choices.iter().any(|c| c == value)
{
anyhow::bail!(
"invalid value '{}' for config '{}'. Allowed: [{}]\n {}",
value,
name,
choices.join(", "),
def.help.as_deref().unwrap_or("")
);
}
if let Some(ref type_) = def.type_ {
match type_.as_str() {
"int" => {
value.parse::<i64>().map_err(|_| {
anyhow::anyhow!(
"config '{}' expects an integer, got '{}'\n {}",
name,
value,
def.help.as_deref().unwrap_or("")
)
})?;
}
"float" => {
value.parse::<f64>().map_err(|_| {
anyhow::anyhow!(
"config '{}' expects a float, got '{}'\n {}",
name,
value,
def.help.as_deref().unwrap_or("")
)
})?;
}
"bool" => {
if value != "true" && value != "false" {
anyhow::bail!(
"config '{}' expects a boolean (true/false), got '{}'\n {}",
name,
value,
def.help.as_deref().unwrap_or("")
);
}
}
"path" => {
if value.is_empty() && from_cli {
return Ok(());
}
if value.is_empty() {
anyhow::bail!(
"config '{}' expects a path, got empty string\n {}",
name,
def.help.as_deref().unwrap_or("")
);
}
if def.must_exist {
let p = std::path::Path::new(value);
if !p.exists() {
anyhow::bail!(
"config '{}' path does not exist: '{}'\n {}",
name,
value,
def.help.as_deref().unwrap_or("")
);
}
}
}
_ => { }
}
}
if let Some(ref range) = def.range
&& let Some((min_str, max_str)) = range.split_once("..")
{
let min: f64 = min_str.trim().parse().map_err(|_| {
anyhow::anyhow!("invalid range min '{}' for config '{}'", min_str, name)
})?;
let max: f64 = max_str.trim().parse().map_err(|_| {
anyhow::anyhow!("invalid range max '{}' for config '{}'", max_str, name)
})?;
let val: f64 = value.parse().map_err(|_| {
anyhow::anyhow!(
"config '{}' has range '{range}' but value '{}' is not numeric\n {}",
name,
value,
def.help.as_deref().unwrap_or("")
)
})?;
if val < min || val > max {
anyhow::bail!(
"config '{}' value {} is outside range {}..{}\n {}",
name,
val,
min,
max,
def.help.as_deref().unwrap_or("")
);
}
}
Ok(())
}
fn cleanup_cache_dir(cache_dir: &std::path::Path, max_age_days: u64) -> usize {
let max_age = std::time::Duration::from_secs(max_age_days * 24 * 3600);
let now = std::time::SystemTime::now();
let mut removed = 0;
let mut stack = vec![cache_dir.to_path_buf()];
while let Some(dir) = stack.pop() {
let Ok(entries) = std::fs::read_dir(&dir) else {
continue;
};
for entry in entries.flatten() {
let path = entry.path();
let is_file = path.is_file();
if !is_file && !path.is_dir() {
continue;
}
let Ok(metadata) = std::fs::metadata(&path) else {
continue;
};
let Ok(modified) = metadata.modified() else {
continue;
};
let Ok(age) = now.duration_since(modified) else {
continue;
};
if age <= max_age {
if !is_file {
stack.push(path);
}
continue;
}
let ok = if is_file {
std::fs::remove_file(&path).is_ok()
} else {
std::fs::remove_dir_all(&path).is_ok()
};
if ok {
removed += 1;
}
}
}
removed
}
fn ai_attempts(cli_max_retries: Option<u32>) -> u32 {
cli_max_retries.unwrap_or(1).max(1)
}
fn failure_exit_code(failure_reason: &str) -> i32 {
failure_reason
.lines()
.next()
.and_then(|line| line.split("exit code ").nth(1))
.and_then(|rest| {
rest.split_whitespace()
.next()
.and_then(|tok| tok.parse().ok())
})
.unwrap_or(-1)
}
#[allow(clippy::too_many_arguments)]
pub(crate) fn parse_cli_overrides(
cli_args: Vec<String>,
declared_config_keys: &std::collections::HashSet<String>,
command: &str,
) -> anyhow::Result<std::collections::HashMap<String, String>> {
let mut cli_arg_values: std::collections::HashMap<String, String> =
std::collections::HashMap::new();
let mut cli_args_iter = cli_args.into_iter().peekable();
while let Some(arg_str) = cli_args_iter.next() {
let swallowed_flag = arg_str
.strip_prefix("--")
.map(|flag| flag.split('=').next().unwrap_or(flag))
.filter(|name| RUN_FLAG_NAMES.contains(name))
.map(|name| format!("--{name}"))
.or_else(|| {
arg_str
.strip_prefix('-')
.filter(|rest| !rest.starts_with('-'))
.map(|flag| flag.split('=').next().unwrap_or(flag))
.filter(|name| RUN_SHORT_FLAGS.contains(name))
.map(|name| format!("-{name}"))
});
if let Some(flag) = swallowed_flag {
let flag_name = flag.trim_start_matches('-');
let is_run_flag = RUN_FLAG_NAMES.contains(&flag_name)
|| (flag_name.len() == 2 && RUN_SHORT_FLAGS.contains(&&flag_name[1..]));
let hint = if command == "run" {
format!(
"'{flag}' is a command flag, not a config override.\n \
Command flags must come before KEY=VALUE overrides, e.g.:\n \
oxo-flow run <workflow.oxoflow> --json min_quality=30\n \
For a config key that itself starts with dashes, use --arg KEY=VALUE\n \
(also placed before positional overrides)."
)
} else if is_run_flag {
let shared = command == "dry-run" && DRY_RUN_SHARED_FLAGS.contains(&flag_name);
if shared {
format!(
"'{flag}' is a command flag, not a config override.\n \
Command flags must come before KEY=VALUE overrides, e.g.:\n \
oxo-flow {command} <workflow.oxoflow> --cache-dir .env-cache min_quality=30"
)
} else {
format!(
"'{flag}' is a run flag and is not supported by {command}.\n \
Run flags only work on `oxo-flow run`; {command} accepts its own \
flags before KEY=VALUE overrides, e.g.:\n \
oxo-flow {command} <workflow.oxoflow> --target my_rule min_quality=30"
)
}
} else {
format!(
"'{flag}' is not supported by {command} and is not a config override.\n \
Config overrides take KEY=VALUE, e.g.:\n \
oxo-flow {command} <workflow.oxoflow> min_quality=30"
)
};
anyhow::bail!("{hint}");
}
let (mut k, v) = if let Some(eq) = arg_str.find('=') {
let k = arg_str[..eq].trim_start_matches('-').to_string();
(k, arg_str[eq + 1..].to_string())
} else if let Some(k) = arg_str.strip_prefix("--") {
let space_form_key = if declared_config_keys.contains(k) {
Some(k.to_string())
} else {
canonicalize_override_key(k, declared_config_keys)
};
if let Some(k) = space_form_key {
if let Some(next) = cli_args_iter.peek()
&& next.starts_with("--")
{
anyhow::bail!(
"config flag '--{k}' expects a value, but the next argument '{next}' \
is a command flag.\n Use the '=' form (--{k}=<value>) or place the \
override after the command flags as {k}=<value>."
);
}
let v = cli_args_iter.next().with_context(|| {
format!(
"invalid config flag: '{arg_str}' — expected --KEY=VALUE or --KEY VALUE"
)
})?;
(k.to_string(), v)
} else {
let rename = REMOVED_RUN_FLAGS
.iter()
.find(|(removed, _)| *removed == k)
.map(|(removed, replacement)| {
format!("\n '--{removed}' was renamed to {replacement}.")
})
.unwrap_or_default();
anyhow::bail!(
"unknown argument '{arg_str}' — did you mean KEY=VALUE overrides?{rename}\n \
Config overrides take KEY=VALUE (e.g. threads=8) or --KEY=VALUE, e.g.:\n \
oxo-flow {command} <workflow.oxoflow> threads=8; \
for a config key that itself starts with dashes, use --arg KEY=VALUE"
);
}
} else {
anyhow::bail!(
"invalid config value format: '{arg_str}' — expected KEY=VALUE, --KEY=VALUE, or --KEY VALUE"
);
};
if k.is_empty() {
anyhow::bail!("invalid config value format: '{arg_str}' — KEY must be non-empty");
}
if !declared_config_keys.contains(&k)
&& let Some(canonical) = canonicalize_override_key(&k, declared_config_keys)
{
k = canonical;
}
cli_arg_values.insert(k, v);
}
Ok(cli_arg_values)
}
fn canonicalize_override_key(
key: &str,
declared_config_keys: &std::collections::HashSet<String>,
) -> Option<String> {
if !key.contains('-') {
return None;
}
let normalized = key.replace('-', "_");
if declared_config_keys.contains(&normalized) {
return Some(normalized);
}
let matches: Vec<&String> = declared_config_keys
.iter()
.filter(|d| d.replace('-', "_") == normalized)
.collect();
match matches.as_slice() {
[only] => Some((*only).clone()),
_ => None,
}
}
pub(crate) fn apply_cli_overrides(
config: &mut WorkflowConfig,
cli_arg_values: &std::collections::HashMap<String, String>,
) -> anyhow::Result<()> {
for (name, cfg_def) in &config.config_meta {
let effective_value = if let Some(val) = cli_arg_values.get(name) {
validate_config_value(name, val, cfg_def, true)?;
config
.config
.insert(name.clone(), toml::Value::String(val.clone()));
val.clone()
} else if let Some(ref default) = cfg_def.default {
validate_config_value(name, default, cfg_def, false)?;
config
.config
.entry(name.clone())
.or_insert_with(|| toml::Value::String(default.clone()));
default.clone()
} else if cfg_def.required {
let help_suffix = cfg_def
.help
.as_deref()
.map(|h| format!("\n {h}"))
.unwrap_or_default();
let display_name = if cfg_def.sensitive {
format!("{name} (sensitive)")
} else {
name.clone()
};
anyhow::bail!(
"required config '{display_name}' not set. Use --{name} <value>{help_suffix}"
);
} else {
continue;
};
if cfg_def.sensitive {
tracing::info!("config '{}' = **** (sensitive)", name);
} else {
tracing::debug!("config '{}' = '{}'", name, effective_value);
}
}
for (k, v) in cli_arg_values {
config
.config
.insert(k.clone(), toml::Value::String(v.clone()));
}
Ok(())
}
pub(crate) fn warn_unknown_override_keys(
overrides: &HashMap<String, String>,
declared: &HashSet<String>,
workflow: &Path,
) {
let unknown: Vec<&String> = overrides
.keys()
.filter(|key| !declared.contains(*key))
.collect();
if unknown.is_empty() {
return;
}
let text = std::fs::read_to_string(workflow).unwrap_or_default();
for key in unknown {
if text.contains(&format!("config.{key}")) {
continue;
}
let hint = closest_declared_key(key, declared)
.map(|candidate| format!(" — did you mean '{candidate}'?"))
.unwrap_or_default();
eprintln!(
"{} override '{key}' is not a declared [config] key and the workflow never \
references {{config.{key}}}{hint} It changes nothing.",
"Warning:".bold().yellow(),
);
}
}
fn closest_declared_key<'a>(key: &str, declared: &'a HashSet<String>) -> Option<&'a str> {
let lowered = key.to_ascii_lowercase();
declared
.iter()
.map(|candidate| {
(
edit_distance(&lowered, &candidate.to_ascii_lowercase(), 2),
candidate,
)
})
.filter(|(distance, _)| *distance <= 2)
.min_by_key(|(distance, candidate)| (*distance, candidate.as_str()))
.map(|(_, candidate)| candidate.as_str())
}
fn edit_distance(a: &str, b: &str, cap: usize) -> usize {
let a: Vec<char> = a.chars().collect();
let b: Vec<char> = b.chars().collect();
if a.len().abs_diff(b.len()) > cap {
return cap + 1;
}
let mut prev: Vec<usize> = (0..=b.len()).collect();
let mut cur = vec![0usize; b.len() + 1];
for (i, ca) in a.iter().enumerate() {
cur[0] = i + 1;
for (j, cb) in b.iter().enumerate() {
let cost = usize::from(ca != cb);
cur[j + 1] = (prev[j + 1] + 1).min(cur[j] + 1).min(prev[j] + cost);
}
std::mem::swap(&mut prev, &mut cur);
}
prev[b.len()]
}
fn quote_shell_path(path: &str) -> String {
format!("'{}'", path.replace('\'', "'\\''"))
}
fn substitute_source_placeholder(build_cmd: &str, expanded_source: &str) -> String {
build_cmd.replace("{source}", "e_shell_path(expanded_source))
}
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(", ")
}
}
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()
}
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 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 rule.scatter.as_ref().is_some_and(|scatter| {
if !scatter.values.is_empty() {
return false;
}
match scatter.values_from.as_deref() {
None => true,
Some(values_from) => config
.resolve_config_list(values_from)
.is_some_and(|values| values.is_empty()),
}
}) {
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;
}
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
}
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()
}
#[allow(clippy::too_many_arguments)]
pub async fn run_command(
workflow: Option<PathBuf>,
jobs: usize,
keep_going: bool,
workdir: Option<PathBuf>,
log_file: Option<PathBuf>,
target: Vec<String>,
module: Vec<String>,
retry: u32,
timeout: String,
resume_failed: bool,
profile: Option<String>,
max_threads: u32,
max_memory: u64,
skip_env_setup: bool,
skip_ref_build: bool,
cache_dir: Option<PathBuf>,
provenance: bool,
json: bool,
cli_args: Vec<String>,
ai_recover: bool,
ai_max_retries: Option<u32>,
samples_filter: Vec<String>,
rerun: bool,
no_report_snapshot: bool,
max_submitted: Option<usize>,
checkpoint_override: Option<PathBuf>,
) -> Result<()> {
print_banner();
let mut json_summary = RunJsonSummary::new(json, workflow.as_deref());
let _background_pid_file = crate::background::BackgroundPidFileGuard::from_env();
let jobs = jobs.max(1);
let (workflow, from_repo) = match workflow {
Some(w) => {
let text = w.to_string_lossy();
match crate::commands::pull::classify_run_source(&text) {
Some(crate::commands::pull::RunSource::Repo { url, git_ref }) => {
let cache = crate::commands::pull::repo_cache_dir(&url, git_ref.as_deref())?;
let wf = crate::commands::pull::checkout_repo_workflow(
&url,
git_ref.as_deref(),
&cache,
)
.await?;
(wf, true)
}
None => (w, false),
}
}
None => (resolve_workflow(None)?, false),
};
json_summary.set_workflow(&workflow);
let workflow_dir = normalize_dots(absolutize(oxo_flow_core::parent_dir(&workflow))?);
let workdir_default: PathBuf = if from_repo {
std::env::current_dir().context("cannot determine current directory")?
} else {
workflow_dir.clone()
};
let workdir_effective =
normalize_dots(absolutize(workdir.as_ref().unwrap_or(&workdir_default))?);
let _workdir_lock =
WorkdirLock::acquire(&workdir_effective).map_err(|e| match e.suggestion() {
Some(s) => anyhow::anyhow!("{e}\n hint: {s}"),
None => anyhow::anyhow!("{e}"),
})?;
let discovery_fallback =
if workflow.starts_with(std::env::temp_dir()) && workdir_effective != workflow_dir {
Some(workdir_effective.as_path())
} else {
None
};
let mut config =
WorkflowConfig::from_file_with_discovery_fallback(&workflow, discovery_fallback)
.with_context(|| format!("failed to parse {}", workflow.display()))?;
let declared_config_keys: std::collections::HashSet<String> =
config.config.keys().cloned().collect();
let cli_arg_values = parse_cli_overrides(cli_args, &declared_config_keys, "run")?;
apply_cli_overrides(&mut config, &cli_arg_values)?;
warn_unknown_override_keys(&cli_arg_values, &declared_config_keys, &workflow);
if !samples_filter.is_empty()
&& let Some(readiness) = samples::apply_samples_filter(
&mut config,
&samples_filter,
true,
workdir.as_deref().unwrap_or(&workdir_default),
)?
{
samples::print_readiness_section(&readiness);
}
if let Some(ref profile_name) = profile {
crate::commands::run_preview::merge_profile(&mut config, profile_name, &workflow_dir)?;
}
config.apply_defaults();
if let Some(effective_workdir) = workdir.as_ref() {
config.runtime_workdir = Some(normalize_dots(absolutize(effective_workdir)?));
}
config
.expand_wildcards()
.context("failed to expand wildcard rules")?;
let e005 = undefined_config_findings(&config);
if !e005.is_empty() {
json_summary.emit("failed", &RunCounts::default(), vec![]);
return Err(anyhow::anyhow!(
"workflow references undefined config variable(s) — fix before running:\n {}",
e005.join("\n ")
));
}
let mut dag = WorkflowDag::from_rules_with_config(
&config.rules,
&config_placeholder_values(&config.config),
)
.context("failed to build workflow DAG")?;
let mut target = target;
for m in &module {
match config.module_closure(m) {
Some(names) => target.extend(names),
None => {
json_summary.emit("failed", &RunCounts::default(), vec![]);
return Err(anyhow::anyhow!(
"unknown module '{m}' — known modules: {}",
known_modules_hint(&config.module_rules)
));
}
}
}
let sensitive_values = sensitive_values_of(&config);
let workflow_abs = absolutize(&workflow)?;
let workflow_git_sha =
oxo_flow_core::executor::checkpoint::CheckpointState::workflow_git_sha(&workflow_abs);
let effective_workdir_log = workdir.as_ref().unwrap_or(&workdir_default);
let run_log_path = match &log_file {
Some(p) if p.is_absolute() => p.clone(),
Some(p) => effective_workdir_log.join(p),
None => effective_workdir_log.join(".oxo-flow/logs/oxo-flow.log"),
};
let run_log_header = format!(
"oxo-flow run log\nstarted_at: {}\noxo-flow: v{}\ncommand: {}\nworkflow: {}\nworkflow_name: {}\nworkflow_version: {}\ngit_sha: {}\nworkdir: {}\n\n",
chrono::Local::now().to_rfc3339(),
env!("CARGO_PKG_VERSION"),
oxo_flow_core::executor::process::mask_sensitive(
&std::env::args().collect::<Vec<_>>().join(" "),
&sensitive_values,
),
workflow_abs.display(),
config.workflow.name,
config.workflow.version,
workflow_git_sha
.as_deref()
.unwrap_or("(not inside a git repository)"),
effective_workdir_log.display(),
);
let run_log: SharedRunLog = Arc::new(std::sync::Mutex::new(
if std::env::var_os("OXO_FLOW_STDERR_ALREADY_REDIRECTED").is_some() {
eprintln!("{run_log_header}");
None
} else {
match crate::logging::activate_run_log(&run_log_path, &run_log_header) {
Ok(guard) => Some(guard),
Err(e) => {
tracing::warn!(error = %e, path = %run_log_path.display(), "failed to open run log; continuing without file logging");
None
}
}
},
));
tracing::info!(
workflow = %config.workflow.name,
version = %config.workflow.version,
git_sha = workflow_git_sha.as_deref().unwrap_or("(none)"),
workdir = %effective_workdir_log.display(),
"workflow run started (log: {})",
run_log_path.display()
);
let when_false_rules: std::collections::HashSet<String> = if target.is_empty() {
std::collections::HashSet::new()
} else {
let wildcard_values: HashMap<String, String> = config_placeholder_values(&config.config);
config
.rules
.iter()
.filter(|rule| {
crate::commands::run_preview::when_condition_false(rule, &config, &wildcard_values)
})
.map(|rule| rule.name.clone())
.collect()
};
let target = resolve_default_targets(&config, &target);
let mut order = if target.is_empty() {
dag.execution_order()?
} else {
let target_refs: Vec<&str> = target.iter().map(String::as_str).collect();
let (filtered_order, skipped_targets) = dag
.execution_order_for_targets_skipping_with_source_check(
&target_refs,
&when_false_rules,
&|path| run_workdir_check(Some(workdir.as_ref().unwrap_or(&workdir_default)), path),
)
.with_context(|| "failed to resolve target rules")?;
for skipped in &skipped_targets {
print_target_skipped_note(skipped, &when_false_rules, Some(&run_log));
}
if filtered_order.is_empty() {
json_summary.emit("failed", &RunCounts::default(), vec![]);
return Err(anyhow::anyhow!(
"all requested targets are when-gated false — nothing to run"
));
}
filtered_order
};
let link_report = crate::commands::workdir_sources::link_workflow_sources(
&config,
&dag,
&order,
&workflow_dir,
workdir.as_ref().unwrap_or(&workdir_default),
&config_placeholder_values(&config.config),
);
narrate_source_link_report(&link_report, Some(&run_log));
let missing_sources = missing_source_findings(
&config,
&dag,
&order,
workdir.as_ref().unwrap_or(&workdir_default),
&workflow_dir,
&config_placeholder_values(&config.config),
);
if !missing_sources.is_empty() {
json_summary.emit("failed", &RunCounts::default(), vec![]);
return Err(anyhow::anyhow!(
"workflow is missing required source input(s) — no rule produces them:\n {}",
missing_sources.join("\n ")
));
}
emit_execution_event(oxo_flow_core::executor::ExecutionEvent::WorkflowStarted {
workflow_name: config.workflow.name.clone(),
total_rules: order.len(),
});
notify_webhook(
&config,
oxo_flow_core::webhook::WebhookEvent::WorkflowStarted,
None,
)
.await;
if !crate::commands::is_quiet() {
eprintln!(
"{} {} rules in execution order",
"DAG:".bold().green(),
order.len()
);
for (i, rule_name) in order.iter().enumerate() {
eprintln!(" {}. {}", i + 1, rule_name);
}
}
let checkpoint_path = checkpoint_override.unwrap_or_else(|| {
workdir
.as_ref()
.unwrap_or(&workdir_default)
.join(".oxo-flow/checkpoint.json")
});
let stale_asides = oxo_flow_core::executor::output_invalidation::cleanup_stale_failed_asides(
workdir.as_ref().unwrap_or(&workdir_default),
oxo_flow_core::executor::output_invalidation::OXOX_FAILED_RETENTION_DAYS,
);
if stale_asides > 0 {
tracing::info!(
count = stale_asides,
"removed stale .oxo-failed aside files (retention)"
);
}
let mut loaded_checkpoint = if checkpoint_path.exists() {
match CheckpointState::load_from_file(&checkpoint_path) {
Ok(ck) => ck,
Err(e) => {
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
));
}
}
} else {
CheckpointState::default()
};
adopt_checkpoint_for_workflow(&mut loaded_checkpoint, &workflow_abs, Some(&run_log));
let stale_running: Vec<String> = loaded_checkpoint.running_rules().iter().cloned().collect();
if !stale_running.is_empty() {
for rule in &stale_running {
loaded_checkpoint.clear_running(rule);
}
diagnostic_narrate(
format_args!(
" {} checkpoint lists {} rule(s) as running from a previous run \
({}): they died with it and will re-run.",
"!".yellow(),
stale_running.len(),
stale_running.join(", ")
),
Some(&run_log),
);
if let Err(e) = loaded_checkpoint.save_to_file(&checkpoint_path) {
tracing::warn!("Failed to persist cleaned checkpoint: {e}");
}
}
let checkpoint: Arc<Mutex<CheckpointState>> = Arc::new(Mutex::new(loaded_checkpoint));
let sensitive_keys: std::collections::HashSet<String> = config
.config_meta
.iter()
.filter(|(_, def)| def.sensitive)
.map(|(key, _)| key.clone())
.collect();
{
let ck = checkpoint.lock().await;
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() {
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) = if old.starts_with("sha256:") || new.starts_with("sha256:")
{
("****".to_string(), "****".to_string())
} else {
(old.clone(), new)
};
eprintln!(" {key}: {old} → {new}");
}
None => eprintln!(" {key}: (removed from this invocation)"),
}
}
}
}
{
let mut ck = checkpoint.lock().await;
ck.set_workflow_path(&workflow_abs);
if let Some(sha) = &workflow_git_sha {
ck.set_workflow_git_sha(sha.clone());
}
ck.set_workdir(&absolutize(workdir.as_deref().unwrap_or(&workdir_default))?);
}
let run_workdir = absolutize(workdir.as_deref().unwrap_or(&workdir_default))?;
let old_snapshot = {
let ck = checkpoint.lock().await;
ck.config_snapshot.clone()
};
let (change_report, mut force_rules, completed_in_run) = {
let mut ck = checkpoint.lock().await;
let report = oxo_flow_core::config_impact::detect_config_changes_with_replay(
&mut ck,
&config.rules,
&dag,
&config.config,
&sensitive_keys,
&config.workflow.interpreter_map,
config.defaults.shell_prelude.as_deref(),
Some(&workflow_dir),
Some(&run_workdir),
Some(&config),
);
let force_rules: std::collections::HashSet<String> =
report.invalidated.iter().cloned().collect();
let completed_in_run = order
.iter()
.filter(|name| ck.completed_rules.contains(*name))
.count();
if let Err(e) = ck.save_to_file(&checkpoint_path) {
tracing::warn!(error = %e, "failed to save checkpoint after config-change detection");
}
(report, force_rules, completed_in_run)
};
print_config_change_summary(
&change_report,
&old_snapshot,
&config,
&sensitive_keys,
&order,
completed_in_run,
rerun,
Some(&run_log),
);
if !change_report.sample_selection_exempt.is_empty() {
diagnostic_narrate(
format_args!(
" {} skipped {} rule(s) whose definition 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(),
change_report.sample_selection_exempt.len(),
change_report.sample_selection_exempt.join(", ")
),
Some(&run_log),
);
}
let wildcard_values: Arc<HashMap<String, String>> =
Arc::new(config_placeholder_values(&config.config));
let workdir_actual = Arc::new(workdir.as_ref().unwrap_or(&workdir_default).clone());
for (label, path) in [
("workdir", workdir_actual.as_path()),
("temp dir", std::env::temp_dir().as_path()),
] {
if let Some(free_kb) = free_kilobytes(path) {
let free_gb = free_kb as f64 / 1024.0 / 1024.0;
if free_gb < 1.0 {
tracing::warn!(
label,
path = %path.display(),
free_gb = format!("{free_gb:.1}"),
"less than 1 GB free disk space — the run may fail with ENOSPC mid-way"
);
}
}
}
{
let mut ck = checkpoint.lock().await;
let (mismatched, missing_inputs, baselined, sample_selection_driven) =
crate::commands::run_preview::detect_input_manifest_invalidations(
&mut ck,
&config,
&dag,
&order,
workdir_actual.as_ref(),
&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 {} rule(s) 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(),
names.len(),
names.join(", ")
),
Some(&run_log),
);
}
if !missing_inputs.is_empty() {
let upstream = crate::commands::run_preview::cascade_up(&mut ck, &dag, &missing_inputs);
force_rules.extend(upstream.iter().cloned());
if upstream.is_empty() {
let lines = missing_source_lines(
&missing_inputs,
&config,
workdir_actual.as_ref(),
&wildcard_values,
);
diagnostic_narrate(
format_args!(
" {} required input(s) missing and cannot be regenerated in this run — affected rule(s):\n{}",
"⚠".yellow(),
lines.join("\n")
),
Some(&run_log),
);
} else {
diagnostic_narrate(
format_args!(
" {} missing intermediate inputs — re-running {} producer rule(s): {}",
"↻".yellow(),
upstream.len(),
upstream.join(", ")
),
Some(&run_log),
);
}
}
if !mismatched.is_empty() {
let invalidated = crate::commands::run_preview::invalidate_with_downstream(
&mut ck,
&dag,
&mismatched,
);
force_rules.extend(invalidated.iter().cloned());
let prior_failures = !ck.failed_rules.is_empty();
diagnostic_narrate(
format_args!(
" {} input changes invalidated {} rule(s): {}{}",
"↻".yellow(),
invalidated.len(),
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 {
""
}
),
Some(&run_log),
);
}
if baselined > 0 {
diagnostic_narrate(
format_args!(
" Note: checkpoint predates input tracking: recorded baseline input manifests for {} completed rule(s); future input changes will invalidate them automatically",
baselined
),
Some(&run_log),
);
}
if (!mismatched.is_empty() || baselined > 0)
&& let Err(e) = ck.save_to_file(&checkpoint_path)
{
tracing::warn!(error = %e, "failed to save checkpoint after input-manifest detection");
}
}
let tombstone_keep: std::collections::HashSet<String> = {
let ck = checkpoint.lock().await;
ck.tombstones
.keys()
.filter(|rule| {
dag.dependents(rule)
.map(|deps| deps.iter().all(|d| ck.completed_rules.contains(d)))
.unwrap_or(true)
})
.cloned()
.collect()
};
if !rerun {
let will_run_seeds: std::collections::HashSet<String> = {
let ck = checkpoint.lock().await;
order
.iter()
.filter(|name| {
!ck.completed_rules.contains(*name)
|| (!tombstone_keep.contains(*name)
&& !config.get_rule(name).is_some_and(|rule| {
crate::commands::run_preview::when_condition_false(
rule,
&config,
&wildcard_values,
)
})
&& config.get_rule(name).is_some_and(|rule| {
!crate::commands::run_preview::rule_outputs_exist(
rule,
workdir_actual.as_ref(),
&wildcard_values,
)
}))
})
.cloned()
.collect()
};
let upstream = {
let mut ck = checkpoint.lock().await;
crate::commands::run_preview::cascade_up_tombstoned(&mut ck, &dag, &will_run_seeds)
};
if !upstream.is_empty() {
force_rules.extend(upstream.iter().cloned());
diagnostic_narrate(
format_args!(
" {} temporary outputs needed again — re-running {} producer rule(s): {}",
"↻".yellow(),
upstream.len(),
upstream.join(", ")
),
Some(&run_log),
);
}
}
let is_tty = std::io::IsTerminal::is_terminal(&std::io::stderr());
let stderr_redirected = std::env::var_os("OXO_FLOW_STDERR_ALREADY_REDIRECTED").is_some();
let progress = indicatif::ProgressBar::new(order.len() as u64);
progress.set_style(
indicatif::ProgressStyle::default_bar()
.template(
"{spinner:.green} [{elapsed_precise}] [{bar:40.cyan/blue}] {pos}/{len} ETA:{eta} ({msg})",
)?
.progress_chars("#>-"),
);
if !is_tty || stderr_redirected {
progress.set_draw_target(indicatif::ProgressDrawTarget::hidden());
}
let timeout_secs: u64 = if timeout == "0" {
0
} else if let Ok(n) = timeout.parse::<u64>() {
n
} else {
parse_duration_secs(&timeout).ok_or_else(|| {
anyhow::anyhow!(
"invalid --timeout value '{timeout}'.\n \
Expected seconds (e.g. 3600) or a duration like 1h / 30m / 2d."
)
})?
};
if profile.is_some()
&& let Some(cluster_profile) = config.cluster.clone()
{
let summary = crate::commands::run_cluster::run_on_cluster(
&cluster_profile,
crate::commands::run_cluster::ClusterRunArgs {
config: &config,
dag: &dag,
order: &order,
checkpoint: &checkpoint,
checkpoint_path: &checkpoint_path,
workdir: workdir_actual.as_ref(),
wildcard_values: wildcard_values.as_ref(),
sensitive_keys: &sensitive_keys,
sensitive_values: &sensitive_values,
force_rules: &force_rules,
max_submitted,
rerun,
resume_failed,
cache_dir: cache_dir.clone(),
skip_env_setup,
ai_recover,
},
)
.await?;
json_summary.emit(
if summary.is_success() {
"completed"
} else {
"failed"
},
&RunCounts {
succeeded: summary.succeeded,
skipped: summary.skipped,
failed: summary.failed,
non_required_failed: summary.non_required_failed,
blocked: 0,
},
vec![],
);
notify_webhook(
&config,
if summary.is_success() {
oxo_flow_core::webhook::WebhookEvent::WorkflowCompleted
} else {
oxo_flow_core::webhook::WebhookEvent::WorkflowFailed
},
Some((summary.succeeded, summary.failed, summary.skipped)),
)
.await;
if !summary.is_success() {
return Err(anyhow::anyhow!("workflow execution failed"));
}
if !no_report_snapshot {
let ck = checkpoint.lock().await;
if let Err(e) =
crate::commands::output::snapshot_report(&workflow, &workdir_actual, &ck)
{
eprintln!(" {} Report snapshot failed: {e}", "⚠".yellow());
}
}
return Ok(());
}
let exec_config = ExecutorConfig {
max_jobs: jobs,
dry_run: false,
workdir: match workdir.as_ref() {
Some(p) => normalize_dots(absolutize(p)?),
None => workdir_default.clone(),
},
sensitive_values: sensitive_values.clone(),
shell_prelude: config.defaults.shell_prelude.clone(),
keep_going,
retry_count: retry,
timeout: if timeout_secs > 0 {
Some(std::time::Duration::from_secs(timeout_secs))
} else {
None
},
max_threads: if max_threads > 0 {
Some(max_threads)
} else {
None
},
max_memory_mb: if max_memory > 0 {
Some(max_memory)
} else {
None
},
force_rerun: rerun,
force_rules: force_rules.clone(),
resource_groups: config
.resource_groups
.iter()
.map(|(k, v)| (k.clone(), v.max))
.collect(),
skip_env_setup,
cache_dir: cache_dir.clone(),
interpreter_map: config.workflow.interpreter_map.clone(),
wildcard_constraints: config.wildcard_constraints.clone(),
base_dir: Some(workflow_dir.clone()),
workflow_root: if *workdir_effective != workflow_dir {
Some(workflow_dir.clone())
} else {
None
},
storage_resolver: crate::commands::run_preview::storage_resolver(),
output_tail_bytes: output_tail_bytes_from_env(),
checkpoint: Some(checkpoint.clone()),
};
let scheduled: Vec<&oxo_flow_core::rule::Rule> = order
.iter()
.filter_map(|name| config.get_rule(name))
.collect();
let breaches = oxo_flow_core::scheduler::check_budget_feasibility(
&scheduled,
exec_config.max_threads,
exec_config.max_memory_mb,
);
if !breaches.is_empty() {
progress.finish_and_clear();
let detail = breaches
.iter()
.map(|b| format!(" - {b}"))
.collect::<Vec<_>>()
.join("\n");
json_summary.emit("failed", &RunCounts::default(), vec![]);
return Err(anyhow::anyhow!(
"resource budget too small for {} rule(s); no rules were run:\n{}",
breaches.len(),
detail
));
}
let disk_warnings = oxo_flow_core::scheduler::validate_disk_requirements(
&scheduled.iter().map(|r| (*r).clone()).collect::<Vec<_>>(),
workdir_actual.as_ref(),
);
for warning in &disk_warnings {
eprintln!(" {} {}", "Warning:".bold().yellow(), warning);
}
if config
.engine
.as_ref()
.is_some_and(|e| e.min_free_disk.is_some())
{
#[cfg(not(unix))]
{
eprintln!(
" {} [engine] min_free_disk is configured but free-space monitoring is not supported on this platform — the monitor will stay inactive.",
"Warning:".bold().yellow()
);
}
#[cfg(unix)]
{
if let Some((min_mb, _reclaim_mb)) =
config.engine.as_ref().and_then(|e| e.disk_thresholds())
&& let Some(start_mb) =
oxo_flow_core::scheduler::check_available_disk_mb(workdir_actual.as_ref())
&& start_mb < min_mb
{
eprintln!(
" {} workdir starts below [engine] min_free_disk: {}MB free, threshold {}MB — the run will hold dispatches until space is freed.",
"Warning:".bold().yellow(),
start_mb,
min_mb
);
}
}
}
let pending_envs: Vec<(String, oxo_flow_core::rule::EnvironmentSpec)> = {
let ck = checkpoint.lock().await;
let preview = crate::commands::run_preview::preview_run_plan(
&ck,
&config,
&dag,
&order,
workdir_actual.as_ref(),
&wildcard_values,
&sensitive_keys,
&config.workflow.interpreter_map,
&checkpoint_path,
rerun,
resume_failed,
);
preview
.plan
.iter()
.filter(|entry| !entry.status.is_skip())
.filter_map(|entry| {
config.get_rule(&entry.name).map(|rule| {
(
entry.name.clone(),
rule.environment.resolve_relative_paths(Some(&workflow_dir)),
)
})
})
.filter(|(_, env)| !env.is_empty())
.collect()
};
let env_failures = backend_preflight_failures(
&pending_envs,
&oxo_flow_core::environment::EnvironmentResolver::new(),
);
if !env_failures.is_empty() {
progress.finish_and_clear();
let detail = env_failures
.iter()
.map(|f| format!(" - {f}"))
.collect::<Vec<_>>()
.join("\n");
json_summary.emit("failed", &RunCounts::default(), vec![]);
return Err(anyhow::anyhow!(
"{} environment backend(s) required by pending rules are unavailable; no rules were run:\n{}",
env_failures.len(),
detail
));
}
let env_cache_dir = exec_config
.cache_dir
.clone()
.unwrap_or_else(|| workdir_actual.as_ref().join(".oxo-flow").join("env-cache"));
let executor = Arc::new(LocalExecutor::new(exec_config));
let success_count = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let fail_count = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let provenance_recorded = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let non_required_fail_count = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let skipped_count = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let failed_rules_set: Arc<Mutex<std::collections::HashSet<String>>> =
Arc::new(Mutex::new(std::collections::HashSet::new()));
let skipped_no_output: Arc<Mutex<std::collections::HashSet<String>>> =
Arc::new(Mutex::new(std::collections::HashSet::new()));
let failures: Arc<Mutex<Vec<(String, String)>>> = Arc::new(Mutex::new(Vec::new()));
let blocked: Arc<Mutex<Vec<(String, String)>>> = Arc::new(Mutex::new(Vec::new()));
if resume_failed && checkpoint_path.exists() {
let mut ck = checkpoint.lock().await;
let failed_count = ck.failed_rules.len();
let completed_count = ck.completed_rules.len();
ck.failed_rules.clear();
eprintln!(
"{} Resuming {} completed, re-running {} failed rules",
"Resume:".bold().cyan(),
completed_count,
failed_count
);
}
if !skip_ref_build && !config.references.is_empty() {
let ref_workdir = workdir.as_ref().unwrap_or(&workdir_default);
let ref_checkpoint = checkpoint_path
.parent()
.unwrap_or(Path::new("."))
.join("reference-checkpoint.json");
let legacy_ref_checkpoint = ref_checkpoint
.parent()
.unwrap_or(Path::new("."))
.join(".oxo-flow/reference-checkpoint.json");
if !ref_checkpoint.exists() && legacy_ref_checkpoint.exists() {
let _ = std::fs::rename(&legacy_ref_checkpoint, &ref_checkpoint);
if let Some(parent) = legacy_ref_checkpoint.parent() {
let _ = std::fs::remove_dir(parent);
}
}
let mut ref_state: HashMap<String, RefRecord> = if ref_checkpoint.exists() {
let raw = std::fs::read_to_string(&ref_checkpoint).unwrap_or_default();
serde_json::from_str::<HashMap<String, RefRecord>>(&raw)
.ok()
.or_else(|| {
serde_json::from_str::<HashMap<String, String>>(&raw)
.ok()
.map(|m| {
m.into_iter()
.map(|(k, v)| {
(
k,
RefRecord {
fingerprint: v,
content_sig: None,
source_path: None,
},
)
})
.collect()
})
})
.or_else(|| {
serde_json::from_str::<Vec<String>>(&raw)
.map(|names| {
names
.into_iter()
.map(|n| {
(
n,
RefRecord {
fingerprint: String::new(),
content_sig: None,
source_path: None,
},
)
})
.collect::<HashMap<String, RefRecord>>()
})
.ok()
})
.unwrap_or_default()
} else {
HashMap::new()
};
let mut rebuilt_outputs: Vec<String> = Vec::new();
for ref_def in &config.references {
let output_path = oxo_flow_core::executor::checkpoint::expand_config_in_path(
&ref_def.output,
&wildcard_values,
);
let output_full = ref_workdir.join(&output_path);
let resolved_source = ref_def.source.as_ref().map(|source| {
let expanded = oxo_flow_core::executor::checkpoint::expand_config_in_path(
source,
&wildcard_values,
);
let path = std::path::Path::new(&expanded);
if path.is_absolute() {
path.to_path_buf()
} else {
workflow_dir.join(path)
}
});
let current_fp = oxo_flow_core::config_impact::reference_fingerprint(
ref_def,
&config.config,
resolved_source.as_deref(),
);
let stored: Option<RefRecord> = ref_state.get(&ref_def.name).cloned();
let current_sig = resolved_source.as_deref().and_then(source_content_sig);
let source_newer = resolved_source.as_deref().is_some_and(|p| {
p.exists() && oxo_flow_core::executor::checkpoint::file_is_newer(p, &output_full)
});
let rebuild_reason = if !output_full.exists() {
Some("output missing")
} else if stored.as_ref().is_some_and(|r| r.fingerprint.is_empty()) {
ref_state.insert(
ref_def.name.clone(),
RefRecord {
fingerprint: current_fp.clone(),
content_sig: current_sig.clone(),
source_path: ref_def.source.clone(),
},
);
if let Some(parent) = ref_checkpoint.parent() {
let _ = std::fs::create_dir_all(parent);
}
let _ = std::fs::write(
&ref_checkpoint,
serde_json::to_string(&ref_state).unwrap_or_default(),
);
None
} else if source_newer {
Some("source is newer than output")
} else if stored.as_ref().is_some_and(|r| r.fingerprint != current_fp) {
let pure_path_migration = (|| {
let rec = stored.as_ref()?;
let old_path_str = rec.source_path.as_deref()?;
let old_def = oxo_flow_core::config::ReferenceDef {
source: Some(old_path_str.to_string()),
..ref_def.clone()
};
let fp_at_old_path = oxo_flow_core::config_impact::reference_fingerprint(
&old_def,
&config.config,
Some(ref_workdir.join(old_path_str)).as_deref(),
);
if fp_at_old_path != rec.fingerprint {
return None; }
let cur = current_sig.as_deref()?;
if !cur.contains("|hash:") {
return None; }
(rec.content_sig.as_deref() == Some(cur)).then_some(())
})()
.is_some();
if pure_path_migration {
tracing::info!(
reference = %ref_def.name,
"reference source moved to a new path with identical content — keeping the existing artifact"
);
ref_state.insert(
ref_def.name.clone(),
RefRecord {
fingerprint: current_fp.clone(),
content_sig: current_sig.clone(),
source_path: ref_def.source.clone(),
},
);
if let Some(parent) = ref_checkpoint.parent() {
let _ = std::fs::create_dir_all(parent);
}
let _ = std::fs::write(
&ref_checkpoint,
serde_json::to_string(&ref_state).unwrap_or_default(),
);
None
} else {
Some("definition, source content, or referenced config changed")
}
} else {
None
};
if let Some(reason) = rebuild_reason {
let source_rendered: Option<String> = ref_def.source.as_ref().map(|source| {
let expanded = oxo_flow_core::executor::checkpoint::expand_config_in_path(
source,
&wildcard_values,
);
if std::path::Path::new(&expanded).is_absolute() || workflow_dir == *ref_workdir
{
expanded
} else {
workflow_dir.join(expanded).to_string_lossy().into_owned()
}
});
let synthetic_rule = oxo_flow_core::rule::Rule {
name: format!("ref:{}", ref_def.name),
input: source_rendered
.as_deref()
.map(|s| vec![s.to_string()].into())
.unwrap_or_default(),
output: vec![output_path.clone()].into(),
..Default::default()
};
let mut build_cmd = oxo_flow_core::executor::process::render_shell_command(
&ref_def.build,
&synthetic_rule,
&wildcard_values,
oxo_flow_core::scheduler::detect_system_limits(),
);
if let Some(expanded) = source_rendered.as_deref() {
build_cmd = substitute_source_placeholder(&build_cmd, expanded);
}
build_cmd = config.defaults.apply_shell_prelude(&build_cmd);
if let Some(ref env) = ref_def.environment
&& !env.is_empty()
{
let resolver = oxo_flow_core::environment::EnvironmentResolver::with_cache_dir(
&env_cache_dir,
);
let resolved_env = env.resolve_relative_paths(Some(&workflow_dir));
let key = resolver.cache_key(&resolved_env);
if !resolver.cache_is_ready(&key).await {
let setup = resolver.setup_command(&resolved_env)?;
let out = tokio::process::Command::new("sh")
.arg("-c")
.arg(&setup)
.current_dir(ref_workdir)
.output()
.await;
match out {
Ok(o) if o.status.success() => {
resolver.cache_mark_ready(&key).await;
}
Ok(o) => {
let stderr = String::from_utf8_lossy(&o.stderr).into_owned();
json_summary.emit("failed", &RunCounts::default(), vec![]);
return Err(anyhow::anyhow!(
"failed to set up the environment for reference '{}': {}",
ref_def.name,
stderr.trim()
));
}
Err(e) => {
json_summary.emit("failed", &RunCounts::default(), vec![]);
return Err(anyhow::anyhow!(
"failed to run the environment setup for reference '{}': {e}",
ref_def.name
));
}
}
}
build_cmd =
resolver.wrap_command(&build_cmd, &resolved_env, None, ref_workdir)?;
}
eprintln!(
" {} Building {}: {} ({})",
"⚙".cyan().bold(),
ref_def.name,
ref_def.description.as_deref().unwrap_or(&ref_def.output),
reason
);
let spawn_ref_build = |shell: &str| {
std::process::Command::new(shell)
.arg("-c")
.arg(&build_cmd)
.current_dir(ref_workdir)
.status()
};
let status = match spawn_ref_build("bash") {
Err(e) if e.kind() == std::io::ErrorKind::NotFound => spawn_ref_build("sh"),
other => other,
};
match status {
Ok(s) if s.success() => {
let post_fp = oxo_flow_core::config_impact::reference_fingerprint(
ref_def,
&config.config,
resolved_source.as_deref(),
);
ref_state.insert(
ref_def.name.clone(),
RefRecord {
fingerprint: post_fp,
content_sig: resolved_source
.as_deref()
.and_then(source_content_sig),
source_path: ref_def.source.clone(),
},
);
rebuilt_outputs.push(output_path.clone());
if let Some(parent) = ref_checkpoint.parent() {
let _ = std::fs::create_dir_all(parent);
}
let _ = std::fs::write(
&ref_checkpoint,
serde_json::to_string(&ref_state).unwrap_or_default(),
);
eprintln!(" {} {} built", "✓".green(), ref_def.name);
}
Ok(s) => {
anyhow::bail!(
"failed to build reference '{}' (exit {}). Command: {}",
ref_def.name,
s.code().unwrap_or(-1),
oxo_flow_core::executor::process::mask_sensitive(
&build_cmd,
&sensitive_values
)
);
}
Err(e) => {
anyhow::bail!("failed to run build command for '{}': {}", ref_def.name, e);
}
}
} else if stored.is_none() {
ref_state.insert(
ref_def.name.clone(),
RefRecord {
fingerprint: current_fp,
content_sig: current_sig,
source_path: ref_def.source.clone(),
},
);
if let Some(parent) = ref_checkpoint.parent() {
let _ = std::fs::create_dir_all(parent);
}
let _ = std::fs::write(
&ref_checkpoint,
serde_json::to_string(&ref_state).unwrap_or_default(),
);
}
}
if !rebuilt_outputs.is_empty() {
let mut consumers: HashSet<String> = HashSet::new();
for rule in &config.rules {
let inputs: Vec<String> = rule
.input
.to_vec()
.into_iter()
.map(|i| {
oxo_flow_core::executor::checkpoint::expand_config_in_path(
&i,
&wildcard_values,
)
})
.collect();
if inputs.iter().any(|input| {
!input.contains('{') && rebuilt_outputs.iter().any(|rebuilt| rebuilt == input)
}) {
consumers.insert(rule.name.clone());
}
}
if !consumers.is_empty() {
let mut ck = checkpoint.lock().await;
let invalidated = crate::commands::run_preview::invalidate_with_downstream(
&mut ck, &dag, &consumers,
);
if let Err(e) = ck.save_to_file(&checkpoint_path) {
tracing::warn!(error = %e, "failed to save checkpoint after reference rebuild");
}
eprintln!(
" {} reference rebuild invalidated {} rule(s): {}",
"↻".yellow(),
invalidated.len(),
invalidated.join(", ")
);
}
}
}
{
let ck = checkpoint.lock().await;
if !ck.reentries.is_empty() {
let valid: std::collections::HashSet<String> = config
.rules
.iter()
.filter(|rule| {
ck.is_completed(&rule.name)
&& !rerun
&& (tombstone_keep.contains(&rule.name)
|| crate::commands::run_preview::rule_outputs_exist(
rule,
workdir_actual.as_ref(),
&wildcard_values,
))
})
.map(|r| r.name.clone())
.collect();
let replayed =
oxo_flow_core::reentry::replay_valid_reentries(&mut config, &ck.reentries, &valid)?;
tracing::info!(count = replayed.len(), "replayed checkpoint re-entries");
dag = WorkflowDag::from_rules_with_config(
&config.rules,
&config_placeholder_values(&config.config),
)
.context("failed to rebuild workflow DAG after re-entry replay")?;
order = if target.is_empty() {
dag.execution_order()?
} else {
let target_refs: Vec<&str> = target.iter().map(String::as_str).collect();
let (filtered, skipped_targets) = dag
.execution_order_for_targets_skipping_with_source_check(
&target_refs,
&when_false_rules,
&|path| {
run_workdir_check(
Some(workdir.as_ref().unwrap_or(&workdir_default)),
path,
)
},
)
.with_context(|| "failed to resolve target rules")?;
for skipped in &skipped_targets {
print_target_skipped_note(skipped, &when_false_rules, Some(&run_log));
}
filtered
};
}
}
{
let ck = checkpoint.lock().await;
if !ck.output_pattern_domains.is_empty() && !rerun {
config.discovered_output_patterns = ck.output_pattern_domains.clone();
let all_producers_complete = ck.output_pattern_domains.keys().all(|template| {
config.rules.iter().all(|r| {
r.output_pattern.is_none()
|| (config.expansion_templates.get(&r.name).map(String::as_str)
!= Some(template.as_str())
&& r.name != *template)
|| ck.completed_rules.contains(&r.name)
})
});
let replayed = if all_producers_complete {
config.expand_output_pattern_consumers()?
} else {
tracing::info!(
"deferred output_pattern consumers stay pending: a producer instance \
is not completed yet — the runtime discovery pass will instantiate them"
);
Vec::new()
};
tracing::info!(
count = replayed.len(),
"replayed runtime-discovered output_pattern domains"
);
if !replayed.is_empty() {
dag = WorkflowDag::from_rules_with_config(
&config.rules,
&config_placeholder_values(&config.config),
)
.context("failed to rebuild workflow DAG after output_pattern replay")?;
order = if target.is_empty() {
dag.execution_order()?
} else {
let target_refs: Vec<&str> = target.iter().map(String::as_str).collect();
let (filtered, skipped_targets) = dag
.execution_order_for_targets_skipping_with_source_check(
&target_refs,
&when_false_rules,
&|path| {
run_workdir_check(
Some(workdir.as_ref().unwrap_or(&workdir_default)),
path,
)
},
)
.with_context(|| "failed to resolve target rules")?;
for skipped in &skipped_targets {
print_target_skipped_note(skipped, &when_false_rules, Some(&run_log));
}
filtered
};
}
}
}
let rule_names: Vec<String> = order.to_vec();
let rule_name_refs: Vec<&str> = rule_names.iter().map(String::as_str).collect();
let mut sched = oxo_flow_core::scheduler::SchedulerState::new(&rule_name_refs);
let mut order_set: std::collections::HashSet<String> = order.iter().cloned().collect();
let run_started = std::time::Instant::now();
let semaphore = Arc::new(tokio::sync::Semaphore::new(jobs.max(1)));
let mut join_set = tokio::task::JoinSet::new();
let mut submitted: std::collections::HashSet<String> = std::collections::HashSet::new();
let mut task_rule: std::collections::HashMap<tokio::task::Id, String> =
std::collections::HashMap::new();
{
let ck = checkpoint.lock().await;
for rule_name in &order {
if ck.is_running(rule_name) {
continue;
}
if ck.is_completed(rule_name)
&& !rerun
&& order_set.contains(rule_name.as_str())
&& !submitted.contains(rule_name.as_str())
{
let outputs_ok = tombstone_keep.contains(rule_name)
|| config
.get_rule(rule_name)
.map(|rule| {
crate::commands::run_preview::rule_outputs_exist(
rule,
workdir_actual.as_ref(),
&wildcard_values,
)
})
.unwrap_or(true);
if outputs_ok {
submitted.insert(rule_name.clone());
sched.mark_completed(oxo_flow_core::executor::JobRecord {
signal: None,
rule: rule_name.clone(),
status: oxo_flow_core::executor::JobStatus::Success,
enqueued_at: None,
started_at: None,
finished_at: None,
exit_code: Some(0),
stdout: None,
stderr: None,
command: None,
retries: 0,
skip_reason: None,
max_rss_mb: None,
cpu_seconds: None,
caption: config.get_rule(rule_name).and_then(|r| {
oxo_flow_core::executor::process::rule_report_caption(
r,
workdir_actual.as_ref(),
)
}),
});
skipped_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
if !is_tty {
diagnostic_narrate(
format_args!(" {} {} (already completed)", "⊝".dimmed(), rule_name),
Some(&run_log),
);
}
progress.inc(1);
}
}
}
}
let priority_map: std::collections::HashMap<String, i32> = config
.rules
.iter()
.map(|r| (r.name.clone(), r.priority))
.collect();
const AGING_STEP: i32 = 1;
let mut waited_rounds: std::collections::HashMap<String, i32> = Default::default();
let mut running_since: std::collections::HashMap<String, std::time::Instant> =
Default::default();
let mut watchdog_reported: std::collections::HashSet<String> = std::collections::HashSet::new();
let watchdog_threshold = spawn_watchdog_threshold();
let mut watchdog_tick =
tokio::time::interval(std::time::Duration::from_secs(SPAWN_WATCHDOG_TICK_SECS));
let disk_thresholds = config.engine.as_ref().and_then(|e| e.disk_thresholds());
let mut disk_monitor = oxo_flow_core::scheduler::DiskPressureMonitor::new(
disk_thresholds.map_or(0, |t| t.0),
disk_thresholds.map_or(0, |t| t.1),
);
let mut disk_tick = tokio::time::interval(std::time::Duration::from_secs(disk_tick_secs()));
let mut disk_hold_rounds: u32 = 0;
let disk_hold_abort_rounds = disk_hold_abort_rounds();
let (signal_tx, mut signal_rx) = tokio::sync::watch::channel(None::<InterruptSignal>);
tokio::spawn(async move {
if let Some(signal) = wait_for_termination_signal().await
&& signal_tx.send(Some(signal)).is_err()
{
std::process::exit(signal.exit_code());
}
});
loop {
if disk_thresholds.is_some() {
let mut n = |args: std::fmt::Arguments<'_>| {
diagnostic_narrate(args, Some(&run_log));
};
let level = disk_monitor.observe(workdir_actual.as_ref(), &mut n);
let hold = level >= oxo_flow_core::scheduler::DiskPressureLevel::Hold;
if level >= oxo_flow_core::scheduler::DiskPressureLevel::Reclaim {
let wildcards = wildcard_values.clone();
let candidates = {
let ck = checkpoint.lock().await;
oxo_flow_core::scheduler::reclaim_candidates(
&config.rules,
&dag,
&ck,
workdir_actual.as_ref(),
&wildcards,
&config.expansion_templates,
)
};
for candidate in candidates {
if let Err(reclaim_err) = reclaim_rule_outputs(
&candidate,
workdir_actual.as_ref(),
&checkpoint,
&checkpoint_path,
&run_log,
)
.await
{
diagnostic_narrate(
format_args!(
" {} reclaim of '{}' failed: {reclaim_err}",
"!".yellow(),
candidate.rule
),
Some(&run_log),
);
}
}
}
if hold && !sched.is_complete() && sched.running_count() == 0 && join_set.is_empty() {
if disk_hold_rounds >= disk_hold_abort_rounds {
{
let ck = checkpoint.lock().await;
ck.save_to_file_async(&checkpoint_path).await.ok();
}
progress.finish_and_clear();
json_summary.emit("failed", &RunCounts::default(), vec![]);
anyhow::bail!(
"disk pressure: {}MB free is below min_free_disk ({}MB) with no rules \
in flight and nothing left to reclaim — the run cannot make progress. \
Free space manually (delete intermediates or run `oxo-flow clean`), \
then resume from the checkpoint, or lower [engine] min_free_disk.",
oxo_flow_core::scheduler::check_available_disk_mb(workdir_actual.as_ref())
.unwrap_or(0),
disk_thresholds.map_or(0, |t| t.0),
);
}
disk_hold_rounds += 1;
tokio::time::sleep(std::time::Duration::from_secs(disk_tick_secs())).await;
continue;
}
disk_hold_rounds = 0;
}
if !sched.is_complete() && sched.running_count() == 0 && join_set.is_empty() {
sched.check_deadlock(&dag)?;
}
let ready = age_ready_list(
sched.ready_rules(&dag)?,
&order_set,
&submitted,
&priority_map,
&waited_rounds,
);
let available = if disk_thresholds.is_some()
&& disk_monitor.last_level() >= oxo_flow_core::scheduler::DiskPressureLevel::Hold
{
0
} else {
jobs.saturating_sub(sched.running_count())
};
let to_submit: Vec<String> = ready.iter().take(available).cloned().collect();
for name in ready.iter().skip(available) {
*waited_rounds.entry(name.clone()).or_insert(0) += AGING_STEP;
}
for rule_name in &to_submit {
let rule_is_optional = config
.get_rule(rule_name)
.is_some_and(|r| r.optional.is_optional());
let blocked_by = if rule_is_optional {
None
} else {
let frs = failed_rules_set.lock().await;
let sno = skipped_no_output.lock().await;
dag.dependencies(rule_name).ok().and_then(|deps| {
deps.into_iter()
.find(|d| frs.contains(d.as_str()) || sno.contains(d.as_str()))
})
};
if let Some(dep) = blocked_by {
skipped_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let dep_skipped = {
let sno = skipped_no_output.lock().await;
sno.contains(dep.as_str())
};
if dep_skipped {
let mut sno = skipped_no_output.lock().await;
sno.insert(rule_name.clone());
} else {
let mut frs = failed_rules_set.lock().await;
frs.insert(rule_name.clone());
}
blocked.lock().await.push((rule_name.clone(), dep));
submitted.insert(rule_name.clone());
sched.mark_completed(oxo_flow_core::executor::JobRecord {
signal: None,
rule: rule_name.clone(),
status: oxo_flow_core::executor::JobStatus::Skipped,
enqueued_at: None,
started_at: None,
finished_at: None,
exit_code: None,
stdout: None,
stderr: None,
command: None,
retries: 0,
skip_reason: Some(
if dep_skipped {
"blocked by skipped upstream dependency"
} else {
"blocked by failed upstream dependency"
}
.into(),
),
max_rss_mb: None,
cpu_seconds: None,
caption: config.get_rule(rule_name).and_then(|r| {
oxo_flow_core::executor::process::rule_report_caption(
r,
workdir_actual.as_ref(),
)
}),
});
if !is_tty {
eprintln!(
" {} {} (blocked by {} dependency)",
"⊘".yellow(),
rule_name,
if dep_skipped { "skipped" } else { "failed" }
);
}
progress.inc(1);
continue;
}
submitted.insert(rule_name.clone());
sched.mark_running(rule_name);
{
let mut ck = checkpoint.lock().await;
ck.mark_running(rule_name);
if let Err(e) = ck.save_to_file_async(&checkpoint_path).await {
tracing::warn!("Failed to save checkpoint (running mark): {e}");
}
}
running_since.insert(rule_name.clone(), std::time::Instant::now());
let Some(rule) = config.get_rule(rule_name).cloned() else {
anyhow::bail!("rule '{rule_name}' not found in workflow");
};
let instance_bindings = config.instance_bindings(rule_name);
let rule_name = rule_name.clone();
let executor = executor.clone();
let checkpoint = checkpoint.clone();
let checkpoint_path = checkpoint_path.clone();
let failed_rules_set = failed_rules_set.clone();
let skipped_no_output = skipped_no_output.clone();
let failures = failures.clone();
let success_count = success_count.clone();
let fail_count = fail_count.clone();
let non_required_fail_count = non_required_fail_count.clone();
let sensitive_values = sensitive_values.clone();
let skipped_count = skipped_count.clone();
let wildcard_values = wildcard_values.clone();
let provenance_recorded = provenance_recorded.clone();
let workdir_actual = workdir_actual.clone();
let semaphore = semaphore.clone();
let progress = progress.clone();
progress.set_message(format!("executing {}", rule_name));
if !is_tty {
progress_narrate(
format_args!(" {} {}", "Running:".bold().cyan(), rule_name),
&run_log,
);
}
emit_execution_event(oxo_flow_core::executor::ExecutionEvent::RuleStarted {
rule: rule_name.clone(),
command: None,
});
let typed_config = config.config.clone();
let rule_caption = config.get_rule(&rule_name).and_then(|r| {
oxo_flow_core::executor::process::rule_report_caption(r, workdir_actual.as_ref())
});
let when_gated_off = rule
.when
.as_deref()
.is_some_and(|when| {
!when.contains('{')
&& !oxo_flow_core::executor::process::evaluate_condition_with_wildcards_and_base_dir(
when,
&config.config,
&wildcard_values,
Some(workdir_actual.as_ref().as_path()),
)
});
let instance_bindings = instance_bindings.clone();
let task_rule_name = rule_name.clone();
let task_run_log = Arc::clone(&run_log);
let handle = join_set.spawn(async move {
let _permit = semaphore.acquire().await;
let input_manifest = if when_gated_off {
None
} else {
match oxo_flow_core::executor::checkpoint::snapshot_input_manifest_async(
&rule,
workdir_actual.as_ref(),
&wildcard_values,
&crate::commands::run_preview::storage_resolver(),
)
.await
{
Ok(manifest) => manifest,
Err(e) => {
let msg = format!("could not snapshot inputs: {e}");
diagnostic_narrate(
format_args!(
" {} could not snapshot inputs of rule '{}' — {} (no manifest recorded; later runs may re-execute it)",
"⚠".yellow(),
rule_name,
oxo_flow_core::executor::process::mask_sensitive(
&msg,
&sensitive_values,
)
),
Some(&task_run_log),
);
None
}
}
};
let result = executor
.execute_rule_with_bindings(&rule, &wildcard_values, &typed_config, &instance_bindings)
.await;
match result {
Ok(mut record) => {
let duration = record
.finished_at
.and_then(|f| record.started_at.map(|s| f.signed_duration_since(s)))
.map(|d| d.num_milliseconds() as f64 / 1000.0)
.unwrap_or(0.0);
if record.status == oxo_flow_core::executor::JobStatus::Success {
success_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
emit_execution_event(oxo_flow_core::executor::ExecutionEvent::RuleCompleted {
rule: rule_name.clone(),
status: oxo_flow_core::executor::JobStatus::Success,
duration_ms: (duration * 1000.0) as u64,
});
if !is_tty {
if record.skip_reason.as_deref()
== Some("restored from content cache")
{
progress_narrate(
format_args!(
" {} {} (restored from content cache)",
"↺".cyan(),
rule_name,
),
&task_run_log,
);
} else {
progress_narrate(
format_args!(" {} {} ({:.1}s)", "✓".green(), rule_name, duration),
&task_run_log,
);
}
}
if let Some(log_pattern) = &rule.log {
let expanded = oxo_flow_core::executor::checkpoint::expand_config_in_path(
log_pattern,
&wildcard_values,
);
if !expanded.contains('{') {
let log_path = workdir_actual.as_ref().join(&expanded);
let log_empty = if log_path.is_dir() {
std::fs::read_dir(&log_path)
.map(|mut entries| entries.next().is_none())
.unwrap_or(false)
} else if log_path.is_file() {
std::fs::metadata(&log_path)
.map(|m| m.len() == 0)
.unwrap_or(false)
} else {
false
};
let tails_empty = record
.stdout
.as_deref()
.map(str::trim)
.unwrap_or("")
.is_empty()
&& record
.stderr
.as_deref()
.map(str::trim)
.unwrap_or("")
.is_empty();
if log_empty && tails_empty {
diagnostic_narrate(
format_args!(
" {} rule '{}' succeeded but its declared log '{}' is empty — the tool likely writes its own log elsewhere (declare that path in `log`, or check the tool's default log location); captured stdout/stderr tails are persisted in the checkpoint job records regardless",
"ℹ".dimmed(),
rule_name,
log_path.display()
),
Some(&task_run_log),
);
}
}
}
let benchmark =
oxo_flow_core::executor::checkpoint::BenchmarkRecord {
rule: rule_name.clone(),
wall_time_secs: duration,
max_memory_mb: record.max_rss_mb,
memory_limit_mb: rule
.effective_memory()
.and_then(oxo_flow_core::scheduler::parse_memory_mb),
cpu_seconds: record.cpu_seconds,
retries: record.retries,
recorded_as: None,
};
if let Some(bench_decl) = &rule.benchmark {
let bench_path = oxo_flow_core::executor::checkpoint::expand_config_in_path(bench_decl, &wildcard_values);
if !bench_path.contains('{') {
let rendered = render_benchmark_json(&benchmark);
let dest = workdir_actual.as_ref().join(&bench_path);
if let Some(parent) = dest.parent() {
let _ = std::fs::create_dir_all(parent);
}
if let Err(e) = std::fs::write(&dest, rendered) {
tracing::warn!(
rule = %rule_name,
path = %dest.display(),
error = %e,
"failed to write benchmark file"
);
}
}
}
let post_result =
oxo_flow_core::executor::checkpoint::snapshot_input_manifest_async(
&rule,
workdir_actual.as_ref(),
&wildcard_values,
&crate::commands::run_preview::storage_resolver(),
)
.await;
let post_manifest = match post_result {
Ok(manifest) => manifest,
Err(e) => {
let msg = format!("could not re-snapshot inputs: {e}");
diagnostic_narrate(
format_args!(
" {} could not re-snapshot inputs of rule '{}' — {} (mid-run input changes cannot be detected for it)",
"⚠".yellow(),
rule_name,
oxo_flow_core::executor::process::mask_sensitive(
&msg,
&sensitive_values,
)
),
Some(&task_run_log),
);
None
}
};
if let (Some(pre), Some(post)) = (&input_manifest, &post_manifest) {
let changes =
oxo_flow_core::executor::checkpoint::manifest_changes(
pre, post,
);
if !changes.is_empty() {
let note = format!(
"\n[oxo-flow] WARNING: an input changed while this rule was running ({}); outputs may be inconsistent — the next run will re-execute this rule",
changes.join(", ")
);
tracing::warn!(
rule = %rule_name,
changes = ?changes,
"input changed while the rule was running; outputs may be inconsistent"
);
if let Some(ref mut stderr) = record.stderr {
stderr.push_str(&oxo_flow_core::executor::process::mask_sensitive(
¬e,
&sensitive_values,
));
} else {
record.stderr = Some(
oxo_flow_core::executor::process::mask_sensitive(
¬e,
&sensitive_values,
),
);
}
}
}
let mut ck = checkpoint.lock().await;
ck.record_run(&record, output_tail_bytes_from_env());
ck.mark_completed(&rule_name, benchmark);
ck.tombstones.remove(&rule_name);
if let Some(ref manifest) = input_manifest {
ck.record_input_manifest(&rule_name, manifest.clone());
}
if provenance {
let mut recorded_here = 0usize;
for output in &rule.output {
let expanded = oxo_flow_core::executor::checkpoint::expand_config_in_path(output, &wildcard_values);
if expanded.contains('{') {
continue; }
let output_path = workdir_actual.join(&expanded);
if output_path.exists()
&& let Ok(checksum) =
oxo_flow_core::executor::checkpoint::compute_file_checksum(&output_path)
{
ck.record_checksum(&expanded, checksum);
recorded_here += 1;
}
}
provenance_recorded.fetch_add(recorded_here, std::sync::atomic::Ordering::Relaxed);
}
if let Err(e) = ck.save_to_file_async(&checkpoint_path).await {
tracing::warn!("Failed to save checkpoint: {e}");
}
(rule_name, oxo_flow_core::executor::JobStatus::Success, record)
} else if record.status == oxo_flow_core::executor::JobStatus::Skipped {
skipped_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
if record.skip_reason.as_deref() != Some("outputs up-to-date") {
let mut ck = checkpoint.lock().await;
ck.clear_running(&rule_name);
if let Err(e) = ck.save_to_file_async(&checkpoint_path).await {
tracing::warn!("Failed to save checkpoint (skip): {e}");
}
}
if record.skip_reason.as_deref() == Some("optional inputs missing") {
let mut sno = skipped_no_output.lock().await;
sno.insert(rule_name.clone());
}
if record.skip_reason.as_deref() == Some("outputs up-to-date") {
let mut ck = checkpoint.lock().await;
if !ck.benchmarks.contains_key(&rule_name) {
let benchmark =
oxo_flow_core::executor::checkpoint::BenchmarkRecord {
rule: rule_name.clone(),
wall_time_secs: 0.0,
max_memory_mb: None,
memory_limit_mb: rule
.effective_memory()
.and_then(oxo_flow_core::scheduler::parse_memory_mb),
cpu_seconds: None,
retries: 0,
recorded_as: Some("outputs up-to-date".to_string()),
};
ck.mark_completed(&rule_name, benchmark);
} else {
ck.mark_completed_quiet(&rule_name);
}
if let Err(e) = ck.save_to_file_async(&checkpoint_path).await {
tracing::warn!("Failed to save checkpoint: {e}");
}
}
if !is_tty {
let reason = record
.skip_reason
.as_deref()
.unwrap_or("skipped");
eprintln!(" {} {} ({reason})", "⊝".dimmed(), rule_name);
}
(rule_name, oxo_flow_core::executor::JobStatus::Skipped, record)
} else {
if rule.required {
fail_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
} else {
non_required_fail_count
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
}
{
let mut frs = failed_rules_set.lock().await;
frs.insert(rule_name.clone());
}
let mut ck = checkpoint.lock().await;
ck.record_run(&record, output_tail_bytes_from_env());
ck.mark_failed(&rule_name);
if let Err(e) = ck.save_to_file_async(&checkpoint_path).await {
tracing::warn!("Failed to save checkpoint: {e}");
}
let mut err_msg = format!("rule '{}' failed", rule_name);
if let Some(ref stderr) = record.stderr {
let trimmed = stderr.trim();
if !trimmed.is_empty() {
err_msg.push_str(&format!("\nstderr: {}", trimmed));
}
}
if let Some(code) = record.exit_code {
err_msg.push_str(&format!("\nexit code: {}", code));
}
progress_narrate(
format_args!(" {} {}", "✗".red(), err_msg),
&task_run_log,
);
{
let mut reason = String::new();
if let Some(code) = record.exit_code {
reason.push_str(&format!("exit code {}", code));
}
if let Some(ref stderr) = record.stderr
&& let Some(last) = stderr
.trim()
.lines()
.next_back()
.filter(|l| !l.is_empty())
{
if !reason.is_empty() {
reason.push_str(" — ");
}
reason.push_str(last);
}
if reason.is_empty() {
reason.push_str("failed");
}
if record.retries > 0 {
reason.push_str(&format!(
" (after {} retr{})",
record.retries,
if record.retries == 1 { "y" } else { "ies" }
));
}
if !rule.required {
reason.push_str(" (non-required)");
}
let mut f = failures.lock().await;
f.push((rule_name.clone(), reason));
}
(
rule_name,
record.status, record,
)
}
}
Err(e) => {
if rule.required {
fail_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
} else {
non_required_fail_count
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
}
{
let mut frs = failed_rules_set.lock().await;
frs.insert(rule_name.clone());
}
let record = oxo_flow_core::executor::JobRecord {
signal: None,
rule: rule_name.clone(),
status: oxo_flow_core::executor::JobStatus::Failed,
enqueued_at: None,
started_at: None,
finished_at: None,
exit_code: Some(-1),
stdout: None,
stderr: Some(oxo_flow_core::executor::process::mask_sensitive(
&e.to_string(),
&sensitive_values,
)),
command: None,
retries: 0,
skip_reason: None,
max_rss_mb: None,
cpu_seconds: None,
caption: rule_caption,
};
emit_execution_event(oxo_flow_core::executor::ExecutionEvent::RuleCompleted {
rule: rule_name.clone(),
status: oxo_flow_core::executor::JobStatus::Failed,
duration_ms: 0,
});
let mut ck = checkpoint.lock().await;
ck.record_run(&record, output_tail_bytes_from_env());
ck.mark_failed(&rule_name);
if let Err(e) = ck.save_to_file_async(&checkpoint_path).await {
tracing::warn!("Failed to save checkpoint: {e}");
}
if !keep_going {
progress_narrate(
format_args!(
" {} rule '{}' failed: {}",
"✗".red(),
rule_name,
e
),
&task_run_log,
);
}
let mut f = failures.lock().await;
let reason = if rule.required {
e.to_string()
} else {
format!("{} (non-required)", e)
};
f.push((rule_name.clone(), reason));
(rule_name, oxo_flow_core::executor::JobStatus::Failed, record)
}
}
});
task_rule.insert(handle.id(), task_rule_name);
}
if join_set.is_empty() {
break;
}
let join_result = tokio::select! {
biased;
result = join_set.join_next_with_id() => Some(result),
received = async {
match signal_rx.changed().await {
Ok(()) => *signal_rx.borrow_and_update(),
Err(_) => None,
}
} => match received {
Some(signal) => {
terminate_on_signal(
signal,
&executor,
&mut join_set,
task_rule.values(),
&checkpoint,
&checkpoint_path,
&mut json_summary,
&config,
)
.await
}
None => None,
},
_ = watchdog_tick.tick() => {
let live_rules: std::collections::HashSet<String> = executor
.active_pids()
.into_iter()
.map(|(rule, _)| rule)
.collect();
let queued: std::collections::HashSet<String> =
executor.resource_waiters().await.into_iter().collect();
let n = |args: std::fmt::Arguments<'_>| {
diagnostic_narrate(args, Some(&run_log));
};
for rule in sched.running_rules() {
if live_rules.contains(&rule) || watchdog_reported.contains(&rule) {
continue;
}
let Some(started) = running_since.get(&rule) else {
continue;
};
if started.elapsed() < watchdog_threshold {
continue;
}
watchdog_reported.insert(rule.clone());
let running_secs = started.elapsed().as_secs();
if queued.contains(&rule) {
n(format_args!(
"{} rule '{rule}' has been Running for {running_secs}s while \
queued in the resource pool — benign wait, the scheduler \
reports blockage details every 60s (issue #847). The engine \
will keep waiting.",
"Watchdog:".yellow().bold()
));
} else {
n(format_args!(
"{} rule '{rule}' has been Running for {running_secs}s but never \
spawned a child process — likely causes: environment setup \
still in progress, or a lost scheduler wakeup (issue #685). \
The engine will keep waiting; if the rule never starts, \
capture the run log and file an issue.",
"Watchdog ERROR:".yellow().bold()
));
tracing::error!(
rule = %rule,
running_secs,
"rule has been Running for over {}s but never spawned a child \
process (issue #685)",
watchdog_threshold.as_secs()
);
}
}
None
}
_ = disk_tick.tick() => {
if disk_thresholds.is_some() {
let mut n = |args: std::fmt::Arguments<'_>| {
diagnostic_narrate(args, Some(&run_log));
};
disk_monitor.observe(workdir_actual.as_ref(), &mut n);
}
None
}
};
let Some(join_result) = join_result else {
continue;
};
let (completed_rule, status, record) = match join_result {
Some(Ok((id, v))) => {
task_rule.remove(&id);
v
}
Some(Err(e)) => {
if e.is_panic() {
tracing::error!("Task panicked: {e}");
}
executor.clear_resource_waiters().await;
fail_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let Some(rule_name) = task_rule.remove(&e.id()) else {
json_summary.emit(
"failed",
&RunCounts {
succeeded: success_count.load(std::sync::atomic::Ordering::Relaxed),
skipped: skipped_count.load(std::sync::atomic::Ordering::Relaxed),
failed: fail_count.load(std::sync::atomic::Ordering::Relaxed),
non_required_failed: non_required_fail_count
.load(std::sync::atomic::Ordering::Relaxed),
blocked: blocked.lock().await.len(),
},
vec![],
);
return Err(anyhow::anyhow!(
"internal error: task {} has no recorded rule",
e.id()
));
};
{
let mut frs = failed_rules_set.lock().await;
frs.insert(rule_name.clone());
}
let record = oxo_flow_core::executor::JobRecord {
signal: None,
rule: rule_name.clone(),
status: oxo_flow_core::executor::JobStatus::Failed,
enqueued_at: None,
started_at: None,
finished_at: None,
exit_code: Some(-1),
stdout: None,
stderr: Some(oxo_flow_core::executor::process::mask_sensitive(
&e.to_string(),
&sensitive_values,
)),
command: None,
retries: 0,
skip_reason: None,
max_rss_mb: None,
cpu_seconds: None,
caption: config.get_rule(&rule_name).and_then(|r| {
oxo_flow_core::executor::process::rule_report_caption(
r,
workdir_actual.as_ref(),
)
}),
};
let mut ck = checkpoint.lock().await;
ck.record_run(&record, output_tail_bytes_from_env());
ck.mark_failed(&rule_name);
if let Err(save_err) = ck.save_to_file_async(&checkpoint_path).await {
tracing::warn!("Failed to save checkpoint: {save_err}");
}
let mut f = failures.lock().await;
f.push((rule_name.clone(), format!("task panicked: {e}")));
(
rule_name,
oxo_flow_core::executor::JobStatus::Failed,
record,
)
}
None => break,
};
sched.mark_completed(record);
running_since.remove(&completed_rule);
progress.inc(1);
if status == oxo_flow_core::executor::JobStatus::Success
&& let Some(cp_rule) = config.get_rule(&completed_rule).cloned()
&& cp_rule.checkpoint
&& let Err(e) = process_reentry(
&mut config,
&cp_rule,
workdir_actual.as_ref(),
&wildcard_values,
&mut sched,
&mut order,
&mut order_set,
&mut dag,
&checkpoint,
)
.await
{
tracing::error!(rule = %completed_rule, error = %e, "checkpoint re-entry failed");
let record = oxo_flow_core::executor::JobRecord {
signal: None,
rule: completed_rule.clone(),
status: oxo_flow_core::executor::JobStatus::Failed,
enqueued_at: None,
started_at: None,
finished_at: None,
exit_code: Some(1),
stdout: None,
stderr: None,
command: None,
retries: 0,
skip_reason: Some(format!("re-entry manifest: {e}")),
max_rss_mb: None,
cpu_seconds: None,
caption: config.get_rule(&completed_rule).and_then(|r| {
oxo_flow_core::executor::process::rule_report_caption(
r,
workdir_actual.as_ref(),
)
}),
};
{
let mut frs = failed_rules_set.lock().await;
frs.insert(completed_rule.clone());
}
{
let mut ck = checkpoint.lock().await;
ck.record_run(&record, output_tail_bytes_from_env());
ck.mark_failed(&completed_rule);
if let Err(save_err) = ck.save_to_file_async(&checkpoint_path).await {
tracing::warn!("Failed to save checkpoint: {save_err}");
}
}
{
let mut f = failures.lock().await;
let reason = if cp_rule.required {
format!("re-entry manifest: {e}")
} else {
format!("re-entry manifest: {e} (non-required)")
};
f.push((completed_rule.clone(), reason));
}
if cp_rule.required {
fail_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
} else {
non_required_fail_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
}
sched.mark_completed(record);
}
if status == oxo_flow_core::executor::JobStatus::Success
&& config
.get_rule(&completed_rule)
.is_some_and(|r| r.output_pattern.is_some())
&& let Err(e) = process_output_pattern_discovery(
&mut config,
&completed_rule,
workdir_actual.as_ref(),
&mut sched,
&mut order,
&mut order_set,
&mut dag,
&checkpoint,
&checkpoint_path,
)
.await
{
tracing::error!(rule = %completed_rule, error = %e, "output_pattern discovery failed");
let record = oxo_flow_core::executor::JobRecord {
signal: None,
rule: completed_rule.clone(),
status: oxo_flow_core::executor::JobStatus::Failed,
enqueued_at: None,
started_at: None,
finished_at: None,
exit_code: Some(1),
stdout: None,
stderr: None,
command: None,
retries: 0,
skip_reason: Some(format!("output_pattern discovery: {e}")),
max_rss_mb: None,
cpu_seconds: None,
caption: config.get_rule(&completed_rule).and_then(|r| {
oxo_flow_core::executor::process::rule_report_caption(
r,
workdir_actual.as_ref(),
)
}),
};
{
let mut frs = failed_rules_set.lock().await;
frs.insert(completed_rule.clone());
}
{
let mut ck = checkpoint.lock().await;
ck.record_run(&record, output_tail_bytes_from_env());
ck.mark_failed(&completed_rule);
if let Err(save_err) = ck.save_to_file_async(&checkpoint_path).await {
tracing::warn!("Failed to save checkpoint: {save_err}");
}
}
{
let mut f = failures.lock().await;
f.push((
completed_rule.clone(),
format!("output_pattern discovery: {e}"),
));
}
fail_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
sched.mark_completed(record);
}
if (status == oxo_flow_core::executor::JobStatus::Failed
|| status == oxo_flow_core::executor::JobStatus::TimedOut
|| status == oxo_flow_core::executor::JobStatus::Cancelled)
&& let Some(producer_template) = config.output_pattern_template_of(&completed_rule)
{
let consumers = config.pending_output_pattern_consumers_of(&producer_template);
if !consumers.is_empty() {
eprintln!(
"{} {} not instantiated — output_pattern producer '{}' failed",
"Warning:".bold().yellow(),
consumers.join(", "),
producer_template
);
let mut f = failures.lock().await;
for c in consumers {
f.push((
c,
format!(
"not instantiated — output_pattern producer '{}' failed",
producer_template
),
));
}
}
}
if status == oxo_flow_core::executor::JobStatus::Skipped
&& let Some(producer_template) = config.output_pattern_template_of(&completed_rule)
{
let consumers = config.pending_output_pattern_consumers_of(&producer_template);
if !consumers.is_empty() {
eprintln!(
"{} {} not instantiated — output_pattern producer '{}' was skipped",
"Warning:".bold().yellow(),
consumers.join(", "),
producer_template
);
tracing::warn!(
producer = %producer_template,
consumers = ?consumers,
"skipped output_pattern producer leaves deferred consumers uninstantiated"
);
}
}
let fc = fail_count.load(std::sync::atomic::Ordering::Relaxed);
if fc > 0
&& !keep_going
&& (status == oxo_flow_core::executor::JobStatus::Failed
|| status == oxo_flow_core::executor::JobStatus::TimedOut
|| status == oxo_flow_core::executor::JobStatus::Cancelled)
{
progress.finish_and_clear();
executor.clear_resource_waiters().await;
let snapshot = executor.active_pids();
let mut abort_killed: std::collections::HashSet<String> =
snapshot.iter().map(|(r, _)| r.clone()).collect();
tokio::task::spawn_blocking(move || {
for (rule_name, pid) in snapshot {
if let Err(e) = oxo_flow_core::executor::timeout::kill_process_tree(pid) {
tracing::warn!(
rule = %rule_name,
pid,
error = %e,
"failed to signal in-flight rule process during abort"
);
}
}
})
.await
.ok();
join_set.abort_all();
let late = executor.active_pids();
abort_killed.extend(late.iter().map(|(r, _)| r.clone()));
if !late.is_empty() {
tokio::task::spawn_blocking(move || {
for (rule_name, pid) in late {
if let Err(e) = oxo_flow_core::executor::timeout::kill_process_tree(pid) {
tracing::warn!(
rule = %rule_name,
pid,
error = %e,
"failed to signal late-spawned rule process during abort"
);
}
}
})
.await
.ok();
}
{
let verb = if status == oxo_flow_core::executor::JobStatus::Cancelled {
"cancelled"
} else {
"failed"
};
let killed: Vec<String> = task_rule
.values()
.filter(|name| *name != &completed_rule)
.cloned()
.collect();
if !killed.is_empty() {
let already_failed = failed_rules_set.lock().await.clone();
let mut ck = checkpoint.lock().await;
for rule_name in &killed {
if ck.is_completed(rule_name) {
continue;
}
if already_failed.contains(rule_name) && !abort_killed.contains(rule_name) {
continue;
}
if already_failed.contains(rule_name) {
let skip_reason = format!(
"run aborted before this rule finished — required rule \
'{completed_rule}' {verb}"
);
if ck.demote_failed_to_cancelled(rule_name, skip_reason) {
let required =
config.get_rule(rule_name).is_some_and(|r| r.required);
if required {
fail_count.fetch_sub(1, std::sync::atomic::Ordering::Relaxed);
} else {
non_required_fail_count
.fetch_sub(1, std::sync::atomic::Ordering::Relaxed);
}
failures.lock().await.retain(|(n, _)| n != rule_name);
if !is_tty {
diagnostic_narrate(
format_args!(
" {} {} (run aborted — required rule '{}' {})",
"⊝".dimmed(),
rule_name,
completed_rule,
verb
),
Some(&run_log),
);
}
}
continue;
}
let record = oxo_flow_core::executor::JobRecord {
signal: None,
rule: rule_name.clone(),
status: oxo_flow_core::executor::JobStatus::Cancelled,
enqueued_at: None,
started_at: None,
finished_at: Some(chrono::Utc::now()),
exit_code: None,
stdout: None,
stderr: None,
command: None,
retries: 0,
skip_reason: Some(format!(
"run aborted before this rule finished — required rule \
'{completed_rule}' {verb}"
)),
max_rss_mb: None,
cpu_seconds: None,
caption: None,
};
ck.record_run(&record, output_tail_bytes_from_env());
if !is_tty {
diagnostic_narrate(
format_args!(
" {} {} (run aborted — required rule '{}' {})",
"⊝".dimmed(),
rule_name,
completed_rule,
verb
),
Some(&run_log),
);
}
}
if let Err(e) = ck.save_to_file_async(&checkpoint_path).await {
tracing::warn!(error = %e, "failed to save checkpoint after abort");
}
}
}
let should_recover =
ai_recover || crate::commands::ai_template::should_use_ai(Some(&workflow), false);
if should_recover {
let failures_guard = failures.lock().await;
let failure = failures_guard
.iter()
.find(|(name, _)| name == &completed_rule)
.or_else(|| failures_guard.first());
if let Some((rule, error)) = failure
&& let Some(provider) =
crate::commands::ai_template::try_resolve_ai(Some(&workflow), true)
{
let attempts = ai_attempts(ai_max_retries);
let mut result: Result<crate::commands::ai_recover::DiagnoseResult> =
Err(anyhow::anyhow!("AI recovery did not run"));
for attempt in 1..=attempts {
let exit_code = failure_exit_code(error);
result = crate::commands::ai_recover::diagnose_failure(
&workflow, rule, exit_code, error, &provider,
)
.await;
if result.is_ok() || attempt == attempts {
break;
}
if let Err(e) = &result {
eprintln!(
" AI diagnosis attempt {attempt}/{attempts} failed: {e} — retrying"
);
}
}
match result {
Ok(diag) => {
if let Some(ref toml) = diag.modified_toml {
if !diag.safe_to_auto_apply {
eprintln!(
" AI fix not auto-applied: flagged unsafe by the \
safety assessment. Review the suggestion manually."
);
} else if let Err(e) = oxo_flow_core::WorkflowConfig::parse(toml) {
eprintln!(
" AI fix not auto-applied: generated TOML failed \
validation ({e}). Review the suggestion manually."
);
} else if let Err(e) = crate::commands::ai_recover::apply_fix(
&workflow, toml, "recovery",
) {
eprintln!(" Failed to apply AI fix: {e}");
}
}
}
Err(e) => eprintln!(" AI diagnosis failed: {e}"),
}
}
}
let ck = checkpoint.lock().await;
json_summary.emit(
"failed",
&RunCounts {
succeeded: success_count.load(std::sync::atomic::Ordering::Relaxed),
skipped: skipped_count.load(std::sync::atomic::Ordering::Relaxed),
failed: fail_count.load(std::sync::atomic::Ordering::Relaxed),
non_required_failed: non_required_fail_count
.load(std::sync::atomic::Ordering::Relaxed),
blocked: blocked.lock().await.len(),
},
rule_resource_rows(&ck),
);
run_terminal_hook(
&config,
&workdir_actual,
success_count.load(std::sync::atomic::Ordering::Relaxed),
fail_count.load(std::sync::atomic::Ordering::Relaxed),
skipped_count.load(std::sync::atomic::Ordering::Relaxed),
)
.await;
notify_webhook(
&config,
oxo_flow_core::webhook::WebhookEvent::WorkflowFailed,
Some((
success_count.load(std::sync::atomic::Ordering::Relaxed),
fail_count.load(std::sync::atomic::Ordering::Relaxed),
skipped_count.load(std::sync::atomic::Ordering::Relaxed),
)),
)
.await;
return Err(anyhow::anyhow!("workflow execution failed"));
}
let nrf = non_required_fail_count.load(std::sync::atomic::Ordering::Relaxed);
if (fc > 0 && keep_going) || nrf > 0 {
let frs = failed_rules_set.lock().await;
let mut worklist: Vec<String> = Vec::new();
for rule_name in &rule_names {
if submitted.contains(rule_name) {
continue;
}
let deps = dag.dependencies(rule_name).unwrap_or_default();
if deps.iter().any(|d| frs.contains(d.as_str())) {
worklist.push(rule_name.to_string());
}
}
drop(frs);
let mut idx = 0;
while idx < worklist.len() {
let name = &worklist[idx];
if submitted.contains(name.as_str()) {
idx += 1;
continue;
}
submitted.insert(name.clone());
skipped_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
{
let mut frs = failed_rules_set.lock().await;
frs.insert(name.clone());
}
sched.mark_completed(oxo_flow_core::executor::JobRecord {
signal: None,
rule: name.clone(),
status: oxo_flow_core::executor::JobStatus::Skipped,
enqueued_at: None,
started_at: None,
finished_at: None,
exit_code: None,
stdout: None,
stderr: None,
command: None,
retries: 0,
skip_reason: Some("blocked by failed upstream dependency".into()),
max_rss_mb: None,
cpu_seconds: None,
caption: config.get_rule(name).and_then(|r| {
oxo_flow_core::executor::process::rule_report_caption(
r,
workdir_actual.as_ref(),
)
}),
});
progress.inc(1);
if !is_tty {
eprintln!(" {} {} (blocked by failed dependency)", "⊘".yellow(), name);
}
{
let frs = failed_rules_set.lock().await;
for other in &rule_names {
if submitted.contains(other) || worklist.contains(other) {
continue;
}
let deps = dag.dependencies(other).unwrap_or_default();
if deps.iter().any(|d| frs.contains(d.as_str())) {
worklist.push(other.to_string());
}
}
}
idx += 1;
}
{
let frs = failed_rules_set.lock().await;
for name in &worklist {
let deps = dag.dependencies(name).unwrap_or_default();
if let Some(dep) = deps.into_iter().find(|d| frs.contains(d.as_str())) {
blocked.lock().await.push((name.clone(), dep));
}
}
}
}
}
drop(signal_rx);
let success_count = success_count.load(std::sync::atomic::Ordering::Relaxed);
let fail_count = fail_count.load(std::sync::atomic::Ordering::Relaxed);
let non_required_fail_count =
non_required_fail_count.load(std::sync::atomic::Ordering::Relaxed);
let skipped_count = skipped_count.load(std::sync::atomic::Ordering::Relaxed);
let mut checkpoint = checkpoint.lock().await;
let failures = failures.lock().await;
let blocked = blocked.lock().await;
progress.finish_and_clear();
const DEFAULT_CACHE_MAX_AGE_DAYS: u64 = 90;
let cache_max_age_days: u64 = config
.config
.get("cache_max_age_days")
.and_then(toml::Value::as_integer)
.filter(|d| *d >= 0)
.map(|d| d as u64)
.unwrap_or(DEFAULT_CACHE_MAX_AGE_DAYS);
if cache_max_age_days > 0
&& let Some(ref cache_dir) = cache_dir
{
let removed = cleanup_cache_dir(cache_dir, cache_max_age_days);
if removed > 0 {
tracing::info!(
removed,
cache = %cache_dir.display(),
"aged environment-cache files removed"
);
}
}
emit_execution_event(oxo_flow_core::executor::ExecutionEvent::WorkflowCompleted {
total_duration_ms: run_started.elapsed().as_millis() as u64,
succeeded: success_count,
failed: fail_count,
skipped: skipped_count,
});
if non_required_fail_count > 0 {
progress_narrate(
format_args!(
"\n{} {} succeeded, {} skipped, {} failed, {} non-required failed",
"Done:".bold(),
success_count,
skipped_count,
fail_count,
non_required_fail_count
),
&run_log,
);
} else {
progress_narrate(
format_args!(
"\n{} {} succeeded, {} skipped, {} failed",
"Done:".bold(),
success_count,
skipped_count,
fail_count
),
&run_log,
);
}
if !no_report_snapshot
&& let Err(e) =
crate::commands::output::snapshot_report(&workflow, &workdir_actual, &checkpoint)
{
eprintln!(" {} Report snapshot failed: {e}", "⚠".yellow());
}
if provenance {
let recorded = provenance_recorded.load(std::sync::atomic::Ordering::Relaxed);
if recorded == 0 {
eprintln!(
" {} --provenance recorded 0 checksums — no literal outputs completed \
(templated or wildcard outputs are tracked per instance only after fan-out); \
'provenance verify' will find nothing to check",
"⚠".yellow()
);
}
}
run_terminal_hook(
&config,
&workdir_actual,
success_count,
fail_count,
skipped_count,
)
.await;
notify_webhook(
&config,
if fail_count > 0 {
oxo_flow_core::webhook::WebhookEvent::WorkflowFailed
} else {
oxo_flow_core::webhook::WebhookEvent::WorkflowCompleted
},
Some((success_count, fail_count, skipped_count)),
)
.await;
if !failures.is_empty() {
eprintln!("\n{}", "Failed rules:".bold().red());
for (name, reason) in failures.iter() {
eprintln!(" {} {} — {}", "✗".red(), name.bold(), reason);
}
}
if !blocked.is_empty() {
eprintln!("\n{}", "Blocked rules (did not run):".bold().yellow());
for (name, dep) in blocked.iter() {
eprintln!(" {} {} — depends on '{}'", "⊘".yellow(), name.bold(), dep);
}
}
if success_count > 0 {
let workdir_actual = workdir.as_ref().unwrap_or(&workdir_default);
let mut missing_outputs = Vec::new();
let mut verified = 0usize;
let mut total_size: u64 = 0;
for rule_name in &order {
if checkpoint.is_completed(rule_name)
&& let Some(rule) = config.get_rule(rule_name)
{
for output in &rule.output {
if !output.contains('{') {
let expanded = oxo_flow_core::executor::checkpoint::expand_config_in_path(
output,
&wildcard_values,
);
let resolved = workdir_actual.join(&expanded);
if resolved.exists() {
verified += 1;
if let Ok(meta) = std::fs::metadata(&resolved) {
total_size += meta.len();
}
} else if !workdir_actual.join(output).exists() {
missing_outputs.push(format!(" {}: {}", rule_name, output));
} else {
verified += 1;
}
}
}
}
}
if verified > 0 {
let size_str = if total_size > 1_073_741_824 {
format!("{:.1}GB", total_size as f64 / 1_073_741_824.0)
} else if total_size > 1_048_576 {
format!("{:.1}MB", total_size as f64 / 1_048_576.0)
} else {
format!("{}B", total_size)
};
eprintln!(
"{} {} output files verified ({} total)",
"✓".green(),
verified,
size_str
);
}
if !missing_outputs.is_empty() {
eprintln!(
"{} {} output file(s) were not found:",
"⚠".yellow(),
missing_outputs.len()
);
for m in &missing_outputs {
eprintln!("{}", m.dimmed());
}
}
}
if fail_count == 0 {
let workdir_actual = workdir.as_ref().unwrap_or(&workdir_default);
for rule in &config.rules {
if rule.cleanup_chunks && checkpoint.is_completed(&rule.name) {
let mut migrated: Vec<String> = Vec::new();
for chunk in rule.input.iter() {
if let Some(sha) = checkpoint.checksums.remove(&chunk.to_string()) {
checkpoint.record_cleaned_checksum(&chunk.to_string(), sha);
migrated.push(chunk.to_string());
}
}
if !migrated.is_empty()
&& let Err(e) = checkpoint.save_to_file(&checkpoint_path)
{
tracing::warn!(
error = %e,
"failed to persist chunk migration — keeping chunk files"
);
for path in &migrated {
if let Some(sha) = checkpoint.cleaned_checksums.remove(path) {
checkpoint.checksums.insert(path.clone(), sha);
}
}
continue;
}
oxo_flow_core::executor::checkpoint::cleanup_transform_chunks(rule, workdir_actual)
.await;
}
}
let mut new_tombstones: Vec<(String, Vec<String>)> = Vec::new();
for rule in &config.rules {
if !rule.temporary || !checkpoint.is_completed(&rule.name) {
continue;
}
if checkpoint
.tombstones
.get(&rule.name)
.is_some_and(|paths| paths.iter().all(|p| !workdir_actual.join(p).exists()))
{
continue;
}
let dependents = dag.dependents(&rule.name).unwrap_or_default();
let mut consumers: HashSet<String> = dependents.iter().cloned().collect();
if let Some(pattern) = &rule.output_pattern {
let expanded_pattern = oxo_flow_core::executor::checkpoint::expand_config_in_path(
pattern,
&wildcard_values,
);
if let Ok(re) = wildcard::pattern_to_regex(&expanded_pattern) {
for other in &config.rules {
if other.name == rule.name {
continue;
}
for input in other.input.iter() {
let expanded =
oxo_flow_core::executor::checkpoint::expand_config_in_path(
input,
&wildcard_values,
);
if !expanded.contains('{') && re.is_match(&expanded) {
consumers.insert(other.name.clone());
break;
}
}
}
}
}
if consumers.is_empty() {
eprintln!(
" {} temporary rule '{}' is a leaf — outputs kept",
"ℹ".dimmed(),
rule.name
);
continue;
}
if !consumers.iter().all(|d| checkpoint.is_completed(d)) {
let pending: Vec<&str> = consumers
.iter()
.filter(|d| !checkpoint.is_completed(d))
.map(String::as_str)
.collect();
diagnostic_narrate(
format_args!(
" {} temporary rule '{}' kept: dependents not complete yet ({})",
"!".yellow(),
rule.name,
pending.join(", ")
),
Some(&run_log),
);
continue;
}
let mut deleted = Vec::new();
let mut skipped: Vec<String> = Vec::new();
let template = config.output_pattern_template_of(&rule.name);
let domains = template
.as_deref()
.and_then(|t| checkpoint.output_pattern_domains.get(t))
.cloned()
.unwrap_or_default();
for output in rule.output.iter().chain(rule.output_pattern.iter()) {
let expanded = oxo_flow_core::executor::checkpoint::expand_config_in_path(
output,
&wildcard_values,
);
if !expanded.contains('{') {
if oxo_flow_core::executor::output_invalidation::is_protected_path(
rule,
&wildcard_values,
&expanded,
) {
continue;
}
deleted.push(expanded);
continue;
}
let mut resolved_any = false;
for combo in &domains {
let instance = wildcard::expand_pattern(&expanded, combo)
.unwrap_or_else(|_| expanded.clone());
if instance.contains('{') {
continue;
}
resolved_any = true;
if oxo_flow_core::executor::output_invalidation::is_protected_path(
rule,
&wildcard_values,
&instance,
) {
continue;
}
if !deleted.contains(&instance) {
deleted.push(instance);
}
}
if !resolved_any {
if domains.is_empty() {
skipped.push(format!(
"{output} (no recorded instance bindings — legacy checkpoint or non-output_pattern fan-out; use `oxo-flow clean`)"
));
} else {
skipped.push(format!(
"{expanded} (wildcard did not resolve against recorded combos)"
));
}
}
}
if !skipped.is_empty() {
diagnostic_narrate(
format_args!(
" {} temporary wildcard outputs of '{}' left in place: instance paths could not be expanded: {}",
"!".yellow(),
rule.name,
skipped.join("; ")
),
Some(&run_log),
);
}
if !deleted.is_empty() {
new_tombstones.push((rule.name.clone(), deleted));
}
}
if !new_tombstones.is_empty() {
let mut rollback: Vec<String> = Vec::new();
for (rule, paths) in &new_tombstones {
checkpoint.tombstones.insert(rule.clone(), paths.clone());
for path in paths {
if let Some(sha) = checkpoint.checksums.remove(path) {
checkpoint.record_cleaned_checksum(path, sha);
rollback.push(path.clone());
}
}
}
if let Err(e) = checkpoint.save_to_file(&checkpoint_path) {
tracing::warn!(
error = %e,
"failed to persist tombstone migration — keeping temporary outputs"
);
for (rule, _) in &new_tombstones {
checkpoint.tombstones.remove(rule);
}
for path in &rollback {
if let Some(sha) = checkpoint.cleaned_checksums.remove(path) {
checkpoint.checksums.insert(path.clone(), sha);
}
}
} else {
for (rule, paths) in &new_tombstones {
for expanded in paths {
let resolved = workdir_actual.join(expanded);
let _ = match std::fs::symlink_metadata(&resolved) {
Ok(md) if md.file_type().is_symlink() => {
std::fs::remove_file(&resolved)
}
Ok(md) if md.is_dir() => std::fs::remove_dir_all(&resolved),
Ok(_) => std::fs::remove_file(&resolved),
Err(_) => Ok(()),
};
}
eprintln!(
" {} temporary outputs deleted for '{}' ({} file(s), regenerated on demand)",
"⊘".dimmed(),
rule,
paths.len()
);
}
}
}
}
if !samples_filter.is_empty() {
let pilot_count = oxo_flow_core::scientific_preflight::count_samples(&config);
let total_count = WorkflowConfig::from_file(&workflow)
.map(|c| oxo_flow_core::scientific_preflight::count_samples(&c))
.unwrap_or(pilot_count);
let elapsed = run_started.elapsed();
let elapsed_secs = elapsed.as_secs_f64();
let per_sample = if pilot_count > 0 {
elapsed_secs / pilot_count as f64
} else {
0.0
};
let projected_secs = per_sample * total_count as f64;
eprintln!("\n{}", "Pilot summary:".bold().cyan());
eprintln!(
" Samples: {}/{} (pilot) | Wall time: {:.1}s | Per-sample: {:.1}s | Projected full run: ~{:.0}s",
pilot_count, total_count, elapsed_secs, per_sample, projected_secs
);
let scientific =
oxo_flow_core::scientific_preflight::analyze_scientific_constraints(&config);
if !scientific.is_empty() {
type WarningGroup<'a> = (
(&'a str, &'a str, &'a str),
Vec<&'a oxo_flow_core::scientific_preflight::ScientificWarning>,
);
let mut distinct: Vec<WarningGroup> = Vec::new();
for w in &scientific {
let key = (w.code.as_str(), w.message.as_str(), w.suggestion.as_str());
match distinct.iter_mut().find(|(k, _)| *k == key) {
Some((_, group)) => group.push(w),
None => distinct.push((key, vec![w])),
}
}
eprintln!(
" {} distinct scientific preflight finding(s) ({} instance(s) across sample rules) — see 'oxo-flow dry-run --samples ...'",
distinct.len(),
scientific.len()
);
for (key, group) in &distinct {
eprintln!(" ⚠ [{}] {} (×{})", key.0, group[0].rule, group.len());
}
}
if let Some(provider) = crate::commands::ai_template::try_resolve_ai(Some(&workflow), false)
{
let warnings_text = scientific
.iter()
.map(|w| format!("- [{}] {}: {}", w.code, w.rule, w.message))
.collect::<Vec<_>>()
.join("\n");
let prompt = format!(
"A pilot run of {} sample(s) out of {} completed in {:.1}s ({} succeeded, {} failed).\n\
Per-sample wall time: {:.1}s; projected full run: ~{:.0}s.\n\
Scientific preflight findings:\n{}\n\n\
Write a short pilot report (3-5 sentences): whether the pilot results are\n\
healthy, whether scaling up is advisable, and what to fix first.",
pilot_count,
total_count,
elapsed_secs,
success_count,
fail_count,
per_sample,
projected_secs,
if warnings_text.is_empty() {
"none"
} else {
&warnings_text
}
);
match provider
.chat(
"You are a bioinformatics workflow consultant summarizing a pilot run.",
&prompt,
)
.await
{
Ok(response) => {
eprintln!();
eprintln!("{}", "Pilot report (AI):".bold().green());
eprintln!("{response}");
}
Err(e) => eprintln!(" AI pilot report failed: {e}"),
}
}
}
json_summary.emit(
if fail_count > 0 {
"failed"
} else {
"completed"
},
&RunCounts {
succeeded: success_count,
skipped: skipped_count,
failed: fail_count,
non_required_failed: non_required_fail_count,
blocked: blocked.len(),
},
rule_resource_rows(&checkpoint),
);
if fail_count > 0 {
return Err(anyhow::anyhow!("workflow execution failed"));
}
Ok(())
}
async fn run_terminal_hook(
config: &WorkflowConfig,
workdir: &std::path::Path,
succeeded: usize,
failed: usize,
skipped: usize,
) {
let hook_cmd = if failed > 0 {
config.workflow.on_error.as_deref()
} else {
config.workflow.on_complete.as_deref()
};
let Some(hook_cmd) = hook_cmd else {
return;
};
let mut values: std::collections::HashMap<String, String> = config
.config
.iter()
.map(|(k, v)| (format!("config.{k}"), config_value_string(v)))
.collect();
values.insert("succeeded".to_string(), succeeded.to_string());
values.insert("failed".to_string(), failed.to_string());
values.insert("skipped".to_string(), skipped.to_string());
values.insert("workdir".to_string(), workdir.display().to_string());
let rendered = oxo_flow_core::executor::process::expand_placeholders(hook_cmd, &values);
oxo_flow_core::executor::process::run_workflow_hook(
&rendered,
workdir,
config.defaults.shell_prelude.as_deref(),
)
.await;
}
async fn notify_webhook(
config: &WorkflowConfig,
event: oxo_flow_core::webhook::WebhookEvent,
counts: Option<(usize, usize, usize)>,
) {
let Some(ref webhook) = config.webhook else {
return;
};
let (succeeded, failed, skipped) = counts.unwrap_or_default();
let payload = oxo_flow_core::webhook::WebhookPayload {
event,
workflow_name: config.workflow.name.clone(),
timestamp: chrono::Utc::now().to_rfc3339(),
data: oxo_flow_core::webhook::WebhookData {
total_rules: None,
succeeded: Some(succeeded),
failed: Some(failed),
skipped: Some(skipped),
duration_ms: None,
rule: None,
exit_code: None,
error: if failed > 0 {
Some(format!("{failed} rule(s) failed"))
} else {
None
},
},
version: env!("CARGO_PKG_VERSION").to_string(),
};
let client = oxo_flow_core::webhook::WebhookClient::new(webhook.clone());
if let Err(e) = client.send(&payload).await {
tracing::warn!(error = %e, event = %event, "webhook notification failed (run continues)");
}
}
#[derive(Default)]
struct RunCounts {
succeeded: usize,
skipped: usize,
failed: usize,
non_required_failed: usize,
blocked: usize,
}
struct RunJsonSummary {
json: bool,
workflow: PathBuf,
emitted: bool,
}
impl RunJsonSummary {
fn new(json: bool, workflow: Option<&Path>) -> Self {
Self {
json,
workflow: workflow.unwrap_or_else(|| Path::new("")).to_path_buf(),
emitted: false,
}
}
fn set_workflow(&mut self, workflow: &Path) {
self.workflow = workflow.to_path_buf();
}
fn emit(&mut self, status: &str, counts: &RunCounts, resources: Vec<serde_json::Value>) {
if !self.json || self.emitted {
return;
}
self.emitted = true;
let output = serde_json::json!({
"command": "run",
"status": status,
"workflow": self.workflow.to_string_lossy(),
"results": serde_json::json!({
"succeeded": counts.succeeded,
"skipped": counts.skipped,
"failed": counts.failed,
"non_required_failed": counts.non_required_failed,
"blocked": counts.blocked,
}),
"resources": resources,
});
println!("{}", serde_json::to_string_pretty(&output).unwrap());
}
}
impl Drop for RunJsonSummary {
fn drop(&mut self) {
self.emit("failed", &RunCounts::default(), vec![]);
}
}
fn rule_resource_rows(
checkpoint: &oxo_flow_core::executor::checkpoint::CheckpointState,
) -> Vec<serde_json::Value> {
let mut names: Vec<&String> = checkpoint.benchmarks.keys().collect();
names.sort_unstable();
names
.into_iter()
.map(|name| {
let b = &checkpoint.benchmarks[name];
let status = if checkpoint.completed_rules.contains(name) {
"completed"
} else if checkpoint.failed_rules.contains(name) {
if checkpoint.when_verdicts.get(name) == Some(&false) {
"skipped_by_when"
} else {
"failed"
}
} else {
"running"
};
serde_json::json!({
"rule": name,
"status": status,
"wall_time_secs": b.wall_time_secs,
"recorded_as": b.recorded_as,
"peak_rss_mb": b.max_memory_mb,
"cpu_seconds": b.cpu_seconds,
"retries": b.retries,
})
})
.collect()
}
#[derive(serde::Serialize, serde::Deserialize, Clone, Default)]
struct RefRecord {
fingerprint: String,
content_sig: Option<String>,
source_path: Option<String>,
}
fn source_content_sig(path: &std::path::Path) -> Option<String> {
let md = std::fs::metadata(path).ok()?;
if !md.is_file() {
return None;
}
let mut sig = format!("size:{}", md.len());
if let Some(hash) = oxo_flow_core::executor::checkpoint::content_hash_if_small(path, &md) {
sig.push_str(&format!("|hash:{hash}"));
}
Some(sig)
}
fn free_kilobytes(path: &std::path::Path) -> Option<u64> {
let out = std::process::Command::new("df")
.args(["-Pk", path.to_str()?])
.output()
.ok()?;
if !out.status.success() {
return None;
}
let stdout = String::from_utf8_lossy(&out.stdout);
let line = stdout.lines().next_back()?;
line.split_whitespace().nth(3)?.parse().ok()
}
fn age_ready_list(
mut ready: Vec<String>,
order_set: &std::collections::HashSet<String>,
submitted: &std::collections::HashSet<String>,
priority_map: &std::collections::HashMap<String, i32>,
waited: &std::collections::HashMap<String, i32>,
) -> Vec<String> {
ready.retain(|name| order_set.contains(name) && !submitted.contains(name));
ready.sort_by(|a, b| {
let pa = priority_map.get(a).copied().unwrap_or(0) + waited.get(a).copied().unwrap_or(0);
let pb = priority_map.get(b).copied().unwrap_or(0) + waited.get(b).copied().unwrap_or(0);
pb.cmp(&pa).then_with(|| a.cmp(b))
});
ready
}
fn suggest_jobs(config: &WorkflowConfig, groups: &[Vec<String>], system_threads: u32) -> u32 {
let system_threads = system_threads.max(1);
let wave_capacity = |names: &[String]| -> u32 {
let max_threads = names
.iter()
.filter_map(|n| config.get_rule(n))
.map(|r| r.effective_threads())
.max()
.unwrap_or(1)
.max(1);
(system_threads / max_threads)
.min(names.len() as u32)
.max(1)
};
groups
.iter()
.map(|g| wave_capacity(g))
.max()
.unwrap_or(1)
.clamp(1, system_threads)
}
#[allow(clippy::too_many_arguments)]
pub async fn dry_run_command(
workflow: Option<PathBuf>,
target: Vec<String>,
module: Vec<String>,
verbose: bool,
json: bool,
ai: bool,
ai_max_retries: Option<u32>,
samples_filter: Vec<String>,
workdir: Option<PathBuf>,
profile: Option<String>,
skip_ref_build: bool,
_cache_dir: Option<PathBuf>,
cli_args: Vec<String>,
rerun: bool,
resume_failed: bool,
) -> Result<()> {
print_banner();
let workflow = resolve_workflow(workflow)?;
let workflow_dir = oxo_flow_core::parent_dir(&workflow).to_path_buf();
let base_dir = workdir.as_deref().unwrap_or(&workflow_dir);
let discovery_fallback =
if workflow.starts_with(std::env::temp_dir()) && base_dir != workflow_dir {
Some(base_dir)
} else {
None
};
let mut config =
WorkflowConfig::from_file_with_discovery_fallback(&workflow, discovery_fallback)
.with_context(|| format!("failed to parse {}", workflow.display()))?;
let declared_config_keys: std::collections::HashSet<String> =
config.config.keys().cloned().collect();
let cli_arg_values = parse_cli_overrides(cli_args, &declared_config_keys, "dry-run")?;
apply_cli_overrides(&mut config, &cli_arg_values)?;
warn_unknown_override_keys(&cli_arg_values, &declared_config_keys, &workflow);
let ready_report = if !samples_filter.is_empty() {
samples::apply_samples_filter(&mut config, &samples_filter, false, base_dir)?
} else {
None
};
if let Some(ref profile_name) = profile {
crate::commands::run_preview::merge_profile(&mut config, profile_name, &workflow_dir)?;
}
let scientific = oxo_flow_core::scientific_preflight::analyze_scientific_constraints(&config);
let scientific_context = if scientific.is_empty() {
String::new()
} else {
let mut text = String::new();
for w in &scientific {
text.push_str(&format!(
"- [{}] {}: {}\n Fix: {}\n",
w.code, w.rule, w.message, w.suggestion
));
}
text
};
if !json
&& let Some(provider) = crate::commands::ai_template::try_resolve_ai(Some(&workflow), ai)
{
let attempts = ai_attempts(ai_max_retries);
let mut result = Err(anyhow::anyhow!("AI analysis did not run"));
for attempt in 1..=attempts {
result = crate::commands::ai_check::analyze_workflow(
&workflow,
&provider,
"dry-run",
&scientific_context,
)
.await;
if result.is_ok() || attempt == attempts {
break;
}
if let Err(e) = &result {
eprintln!(" AI analysis attempt {attempt}/{attempts} failed: {e} — retrying");
}
}
result?;
println!();
}
if !scientific.is_empty() {
eprintln!("\n{}", "Scientific preflight:".bold().yellow());
for w in &scientific {
eprintln!(" ⚠ [{}] {}: {}", w.code, w.rule, w.message);
eprintln!(" {} {}", "→".bold(), w.suggestion);
}
}
config.apply_defaults();
config
.expand_wildcards()
.context("failed to expand wildcard rules")?;
let e005 = undefined_config_findings(&config);
if !e005.is_empty() {
return Err(anyhow::anyhow!(
"workflow references undefined config variable(s) — fix before running:\n {}",
e005.join("\n ")
));
}
let mut dag = WorkflowDag::from_rules_with_config(
&config.rules,
&config_placeholder_values(&config.config),
)
.context("failed to build workflow DAG")?;
let mut target = target;
for m in &module {
match config.module_closure(m) {
Some(names) => target.extend(names),
None => {
return Err(anyhow::anyhow!(
"unknown module '{m}' — known modules: {}",
known_modules_hint(&config.module_rules)
));
}
}
}
let target = resolve_default_targets(&config, &target);
let dry_wildcard_values: HashMap<String, String> = config_placeholder_values(&config.config);
let when_false_rules: std::collections::HashSet<String> = if target.is_empty() {
std::collections::HashSet::new()
} else {
config
.rules
.iter()
.filter(|rule| {
crate::commands::run_preview::when_condition_false(
rule,
&config,
&dry_wildcard_values,
)
})
.map(|rule| rule.name.clone())
.collect()
};
let mut order = if target.is_empty() {
let order_set: std::collections::HashSet<String> =
dag.execution_order()?.into_iter().collect();
let mut ordered: Vec<String> = Vec::new();
for group in dag.parallel_groups()? {
let mut level: Vec<String> = group
.into_iter()
.filter(|name| order_set.contains(name))
.collect();
level.sort();
ordered.extend(level);
}
ordered
} else {
let target_refs: Vec<&str> = target.iter().map(String::as_str).collect();
let (filtered, skipped_targets) = dag
.execution_order_for_targets_skipping_with_source_check(
&target_refs,
&when_false_rules,
&|path| run_workdir_check(Some(base_dir), path),
)
.with_context(|| "failed to resolve target rules")?;
for skipped in &skipped_targets {
print_target_skipped_note(skipped, &when_false_rules, None);
}
filtered
};
let mut wildcard_values: HashMap<String, String> = HashMap::new();
for (key, value) in &config.config {
let string_val = match value {
toml::Value::String(s) => s.clone(),
other => other.to_string(),
};
wildcard_values.insert(format!("config.{key}"), string_val);
}
let checkpoint_path = base_dir.join(".oxo-flow/checkpoint.json");
let mut checkpoint_state =
match oxo_flow_core::executor::checkpoint::CheckpointState::load_from_file(&checkpoint_path)
{
Ok(ck) => ck,
Err(_) => {
eprintln!(
"{} no checkpoint at {} — treating every rule as never completed",
"⚠".yellow(),
checkpoint_path.display()
);
oxo_flow_core::executor::checkpoint::CheckpointState::default()
}
};
adopt_checkpoint_for_workflow(&mut checkpoint_state, &workflow, None);
let sensitive_keys: std::collections::HashSet<String> = config
.config_meta
.iter()
.filter(|(_, def)| def.sensitive)
.map(|(key, _)| key.clone())
.collect();
let sensitive_values = sensitive_values_of(&config);
let interpreter_map = config.workflow.interpreter_map.clone();
let mut preview = crate::commands::run_preview::preview_run_plan(
&checkpoint_state,
&config,
&dag,
&order,
base_dir,
&wildcard_values,
&sensitive_keys,
&interpreter_map,
&checkpoint_path,
rerun,
resume_failed,
);
if !checkpoint_state.reentries.is_empty() {
use crate::commands::run_preview::RuleStatus;
let valid: std::collections::HashSet<String> = preview
.plan
.iter()
.filter(|p| p.status == RuleStatus::Skipped)
.map(|p| p.name.clone())
.collect();
let replayed = oxo_flow_core::reentry::replay_valid_reentries(
&mut config,
&checkpoint_state.reentries,
&valid,
)?;
if !replayed.is_empty() {
tracing::info!(
count = replayed.len(),
"dry-run replayed checkpoint re-entries"
);
dag = WorkflowDag::from_rules_with_config(
&config.rules,
&config_placeholder_values(&config.config),
)
.context("failed to rebuild workflow DAG after re-entry replay")?;
let new_order: Vec<String> = if target.is_empty() {
let order_set: std::collections::HashSet<String> =
dag.execution_order()?.into_iter().collect();
let mut ordered: Vec<String> = Vec::new();
for group in dag.parallel_groups()? {
let mut level: Vec<String> = group
.into_iter()
.filter(|n| order_set.contains(n))
.collect();
level.sort();
ordered.extend(level);
}
ordered
} else {
let target_refs: Vec<&str> = target.iter().map(String::as_str).collect();
let (filtered, _) = dag
.execution_order_for_targets_skipping_with_source_check(
&target_refs,
&when_false_rules,
&|path| run_workdir_check(Some(base_dir), path),
)
.with_context(|| "failed to resolve target rules")?;
filtered
};
let old_order_len = order.len();
order = new_order;
if order.len() != old_order_len {
eprintln!(
" {} re-entry replay: plan grew from {} to {} rules",
"Replay:".bold().cyan(),
old_order_len,
order.len()
);
}
preview = crate::commands::run_preview::preview_run_plan(
&checkpoint_state,
&config,
&dag,
&order,
base_dir,
&wildcard_values,
&sensitive_keys,
&interpreter_map,
&checkpoint_path,
rerun,
resume_failed,
);
}
}
let env_manifest_issues: Vec<(String, String)> = {
let mut issues = Vec::new();
for entry in preview.plan.iter().filter(|entry| !entry.status.is_skip()) {
let Some(rule) = config.get_rule(&entry.name) else {
continue;
};
let resolved = rule.environment.resolve_relative_paths(Some(&workflow_dir));
if let Some(spec) = resolved.pixi
&& oxo_flow_core::environment::validate_pixi_manifest(&spec).is_err()
{
issues.push((entry.name.clone(), spec));
}
}
issues
};
{
let p = &preview;
let modified = p
.checkpoint_modified
.map(|t| {
chrono::DateTime::<chrono::Local>::from(t)
.format("%Y-%m-%d %H:%M:%S")
.to_string()
})
.unwrap_or_else(|| "unknown time".to_string());
eprintln!(
"{} {} (modified {})",
"Checkpoint:".bold().cyan(),
p.checkpoint_path.display(),
modified.dimmed()
);
let will_run = p.plan.len() - p.will_skip;
eprintln!(
"{} would run: {} | skip: {} | completed: {} (DAG size: {})",
"Plan:".bold().yellow(),
will_run.to_string().green(),
p.will_skip.to_string().dimmed(),
p.completed_total.to_string().cyan(),
order.len()
);
for chain in &p.cascade_chains {
eprintln!(" {} {}", "rerun cascade:".yellow(), chain.join(" → "));
}
}
let reference_builds: Vec<String> = if skip_ref_build {
Vec::new()
} else {
config
.references
.iter()
.filter(|r| !base_dir.join(&r.output).exists())
.map(|r| r.name.clone())
.collect()
};
if !reference_builds.is_empty() {
eprintln!(
"{} {} reference build(s) would run: {}",
"References:".bold().cyan(),
reference_builds.len(),
reference_builds.join(", ")
);
eprintln!(
" {} pass --skip-ref-build to assume they are pre-built.",
"ℹ".dimmed()
);
}
let readiness_report = match ready_report {
Some(report) => Some(report),
None => Some(oxo_flow_core::readiness::compute_readiness(
&config, base_dir,
)),
};
if let Some(report) = &readiness_report {
samples::print_readiness_section(report);
}
for (i, rule_name) in order.iter().enumerate() {
let rule = config
.get_rule(rule_name)
.ok_or_else(|| anyhow::anyhow!("rule '{}' not found", rule_name))?;
let status_text = preview
.plan
.iter()
.find(|r| r.name == *rule_name)
.map(|r| match &r.status {
crate::commands::run_preview::RuleStatus::NeverCompleted => {
format!("{}", "[run: never completed]".bold())
}
crate::commands::run_preview::RuleStatus::ConfigInvalidated => {
format!("{}", "[run: config changed]".bold())
}
crate::commands::run_preview::RuleStatus::InputInvalidated => {
format!("{}", "[run: input changed]".bold().yellow())
}
crate::commands::run_preview::RuleStatus::OutputsMissing => {
format!("{}", "[run: outputs missing]".bold().yellow())
}
crate::commands::run_preview::RuleStatus::Cascaded { from } => {
format!(
"{}",
format!("[rerun: downstream of {from}]").bold().yellow()
)
}
crate::commands::run_preview::RuleStatus::Skipped => {
format!("{}", "[skip: up to date]".dimmed())
}
crate::commands::run_preview::RuleStatus::SkippedByWhen => {
format!("{}", "[skip: when condition false]".dimmed())
}
crate::commands::run_preview::RuleStatus::CascadedUpstream { from } => {
format!("{}", format!("[rerun: upstream of {from}]").bold().yellow())
}
crate::commands::run_preview::RuleStatus::Forced => {
format!("{}", "[run: forced (--rerun)]".bold())
}
crate::commands::run_preview::RuleStatus::SkippedFresh => {
format!("{}", "[skip: outputs up-to-date]".dimmed())
}
})
.unwrap_or_default();
eprintln!(" {}. {} {}", i + 1, rule_name.bold().cyan(), status_text);
let threads = rule.effective_threads();
eprintln!(" threads={}", threads);
if !rule.environment.is_empty() {
eprintln!(" env={}", rule.environment.kind());
}
if let Some(ref mem) = rule.effective_memory() {
eprintln!(" memory={}", mem);
}
if rule.checkpoint {
eprintln!(" checkpoint=true");
}
if !rule.output.is_empty() {
let expanded_outputs: Vec<String> = rule
.output
.iter()
.map(|o| {
oxo_flow_core::executor::checkpoint::expand_config_in_path(o, &wildcard_values)
})
.collect();
eprintln!(" outputs: {:?}", expanded_outputs);
}
if let Some(ref cmd) = rule.shell {
let expanded = oxo_flow_core::executor::process::render_shell_command(
cmd,
rule,
&wildcard_values,
oxo_flow_core::scheduler::detect_system_limits(),
);
let expanded =
oxo_flow_core::executor::process::mask_sensitive(&expanded, &sensitive_values);
let blocked_label = if let Some((_category, description)) =
oxo_flow_core::format::shell_blocking_pattern(&expanded)
{
match oxo_flow_core::executor::validate_shell_safety_in_workdir(&expanded, base_dir)
{
Err(e) => Some(e.to_string()),
Ok(()) => None,
}
.map(|_| format!("(blocked: E011 — {description})"))
} else {
None
};
if let Some(label) = blocked_label {
eprintln!(" command: {} {}", expanded, label.red().bold());
} else {
eprintln!(" command: {}", expanded);
}
}
for inp in &rule.input {
let s = inp.to_string();
if !s.contains('{') && !s.contains('*') && !s.starts_with('/') {
let exists = base_dir.join(&s).exists();
let icon = if exists { "✓" } else { "✗" };
eprintln!(" input {}: {}", icon, s);
}
}
if verbose {
}
}
if !env_manifest_issues.is_empty() {
eprintln!(
"\n {} {} pixi manifest(s) declared by pending rule(s) do not exist:",
"✗".red().bold(),
env_manifest_issues.len()
);
for (rule, spec) in &env_manifest_issues {
eprintln!(" - {rule}: {spec} (resolved relative to the workflow directory)");
}
eprintln!(
" {} create the manifest at the resolved path or correct the rule's `pixi` path",
"hint:".yellow().bold()
);
}
let total_threads: u32 = config.rules.iter().map(|r| r.effective_threads()).sum();
let max_threads: u32 = config
.rules
.iter()
.map(|r| r.effective_threads())
.max()
.unwrap_or(1);
let memory_values: Vec<&str> = config
.rules
.iter()
.filter_map(|r| r.effective_memory())
.collect();
eprintln!();
eprintln!(
"{} {} rules, total {} threads declared, max {} threads/rule",
"Summary:".bold(),
order.len(),
total_threads,
max_threads,
);
if !memory_values.is_empty() {
eprintln!(
" {} rule(s) with memory requirements",
memory_values.len()
);
}
if !config.sample_groups.is_empty() {
eprintln!(
" {} sample group(s), {} pair(s)",
config.sample_groups.len(),
config.pairs.len()
);
}
let system_threads = std::thread::available_parallelism()
.map(|n| n.get() as u32)
.unwrap_or(4);
let groups = dag.parallel_groups().unwrap_or_default();
let suggested_jobs = suggest_jobs(&config, &groups, system_threads).to_string();
eprintln!(
"\n{} oxo-flow run {} -j {}",
"To execute:".bold().cyan(),
workflow.display(),
suggested_jobs
);
if json {
let order_list = order.clone();
let rule_list: Vec<serde_json::Value> = order
.iter()
.filter_map(|name| config.get_rule(name))
.map(|r| {
serde_json::json!({
"name": r.name,
"threads": r.effective_threads(),
"environment": r.environment.kind(),
"memory": r.effective_memory(),
"checkpoint": r.checkpoint,
})
})
.collect();
let samples_block = readiness_report.as_ref().map(|report| {
let ready_names: Vec<&str> = report.ready.iter().map(|s| s.name.as_str()).collect();
let waiting: Vec<serde_json::Value> = report
.waiting
.iter()
.map(|s| {
serde_json::json!({
"name": s.name,
"missing": s.missing,
})
})
.collect();
serde_json::json!({
"total": report.total,
"ready": report.ready.len(),
"waiting_count": report.waiting.len(),
"ready_names": ready_names,
"waiting": waiting,
"missing_global": report.missing_global,
})
});
let checkpoint_block = Some(&preview).map(|p| {
let plan: Vec<serde_json::Value> = p
.plan
.iter()
.map(|r| {
let (status, cascaded_from) = match &r.status {
crate::commands::run_preview::RuleStatus::NeverCompleted => {
("run-never-completed", None)
}
crate::commands::run_preview::RuleStatus::ConfigInvalidated => {
("run-config-changed", None)
}
crate::commands::run_preview::RuleStatus::InputInvalidated => {
("run-input-changed", None)
}
crate::commands::run_preview::RuleStatus::OutputsMissing => {
("run-outputs-missing", None)
}
crate::commands::run_preview::RuleStatus::Cascaded { from } => {
("run-cascaded", Some(from.clone()))
}
crate::commands::run_preview::RuleStatus::Skipped => ("skip", None),
crate::commands::run_preview::RuleStatus::SkippedByWhen => {
("skip-when-condition", None)
}
crate::commands::run_preview::RuleStatus::CascadedUpstream { from } => {
("run-cascaded-upstream", Some(from.clone()))
}
crate::commands::run_preview::RuleStatus::Forced => ("run-forced", None),
crate::commands::run_preview::RuleStatus::SkippedFresh => {
("skip-fresh", None)
}
};
serde_json::json!({
"name": r.name,
"status": status,
"cascaded_from": cascaded_from,
})
})
.collect();
serde_json::json!({
"path": p.checkpoint_path.display().to_string(),
"modified": p.checkpoint_modified.map(|t| {
chrono::DateTime::<chrono::Local>::from(t)
.format("%Y-%m-%d %H:%M:%S")
.to_string()
}),
"completed_total": p.completed_total,
"summary": {
"will_run": p.plan.len() - p.will_skip,
"will_skip": p.will_skip,
"protected_outside": p.protected_outside,
},
"plan": plan,
"cascade_chains": p.cascade_chains,
})
});
let plan_entries: Vec<serde_json::Value> = order
.iter()
.filter_map(|name| {
let rule = config.get_rule(name)?;
let (status, reason, cascaded_from) = match preview
.plan
.iter()
.find(|r| r.name == *name)
.map(|r| &r.status)
{
Some(crate::commands::run_preview::RuleStatus::NeverCompleted) => {
("run", "never completed".to_string(), None)
}
Some(crate::commands::run_preview::RuleStatus::ConfigInvalidated) => {
("run", "config changed".to_string(), None)
}
Some(crate::commands::run_preview::RuleStatus::InputInvalidated) => {
("run", "input changed".to_string(), None)
}
Some(crate::commands::run_preview::RuleStatus::OutputsMissing) => {
("run", "outputs missing".to_string(), None)
}
Some(crate::commands::run_preview::RuleStatus::Cascaded { from }) => (
"rerun",
format!("downstream of {from}"),
Some(from.clone()),
),
Some(crate::commands::run_preview::RuleStatus::Skipped) => {
("skip", "up to date".to_string(), None)
}
Some(crate::commands::run_preview::RuleStatus::SkippedByWhen) => {
("skip", "when condition false".to_string(), None)
}
Some(crate::commands::run_preview::RuleStatus::CascadedUpstream { from }) => (
"rerun",
format!("upstream of {from}"),
Some(from.clone()),
),
Some(crate::commands::run_preview::RuleStatus::Forced) => {
("run", "forced (--rerun)".to_string(), None)
}
Some(crate::commands::run_preview::RuleStatus::SkippedFresh) => {
("skip", "outputs up-to-date".to_string(), None)
}
None => ("unknown", "unknown".to_string(), None),
};
let expanded_shell = rule.shell.as_ref().map(|cmd| {
oxo_flow_core::executor::process::render_shell_command(
cmd,
rule,
&wildcard_values,
oxo_flow_core::scheduler::detect_system_limits(),
)
});
let inputs_expanded: Vec<String> = rule
.input
.to_vec()
.iter()
.map(|p| {
oxo_flow_core::executor::checkpoint::expand_config_in_path(
p,
&wildcard_values,
)
})
.collect();
let outputs_expanded: Vec<String> = rule
.output
.to_vec()
.iter()
.map(|p| {
oxo_flow_core::executor::checkpoint::expand_config_in_path(
p,
&wildcard_values,
)
})
.collect();
Some(serde_json::json!({
"name": rule.name,
"status": status,
"reason": reason,
"cascaded_from": cascaded_from,
"threads": rule.effective_threads(),
"memory": rule.effective_memory(),
"environment": rule.environment.kind(),
"command": expanded_shell,
"inputs": serde_json::to_value(&rule.input).unwrap_or(serde_json::Value::Null),
"outputs": serde_json::to_value(&rule.output).unwrap_or(serde_json::Value::Null),
"inputs_expanded": inputs_expanded,
"outputs_expanded": outputs_expanded,
}))
})
.collect();
let output = serde_json::json!({
"command": "dry-run",
"schema_version": 1,
"workflow": workflow.display().to_string(),
"total_rules": order.len(),
"execution_order": order_list,
"rules": rule_list,
"plan": plan_entries,
"summary": {
"total_threads": total_threads,
"max_threads_per_rule": max_threads,
"memory_rules": memory_values.len(),
"would_execute": preview.plan.len() - preview.will_skip,
"will_skip": preview.will_skip,
"total_rules": order.len(),
},
"sample_groups": config
.sample_groups
.iter()
.map(|g| {
serde_json::json!({
"name": g.name,
"samples": g.samples,
})
})
.collect::<Vec<_>>(),
"pairs": config
.pairs
.iter()
.map(|p| {
serde_json::json!({
"pair_id": p.pair_id,
"experiment": p.experiment,
"control": p.control,
})
})
.collect::<Vec<_>>(),
"suggested_jobs": suggested_jobs,
"samples": samples_block,
"checkpoint_preview": checkpoint_block,
"reentry": {
"recorded": checkpoint_state.reentries,
"possible": config
.rules
.iter()
.filter(|r| r.checkpoint)
.map(|r| serde_json::json!({
"rule": r.name,
"manifest": r.checkpoint_manifest,
}))
.collect::<Vec<_>>(),
},
"profile": profile,
"reference_builds": reference_builds,
"env_manifest_issues": env_manifest_issues
.iter()
.map(|(rule, spec)| serde_json::json!({"rule": rule, "spec": spec}))
.collect::<Vec<_>>(),
});
println!("{}", serde_json::to_string_pretty(&output).unwrap());
}
if !env_manifest_issues.is_empty() {
anyhow::bail!(
"{} pixi manifest(s) declared by pending rule(s) do not exist — \
create the manifest at the resolved path or correct the rule's \
`pixi` path (resolved relative to the workflow directory)",
env_manifest_issues.len()
);
}
Ok(())
}
pub async fn debug_command(workflow: PathBuf, rule_name: Option<String>, ai: bool) -> Result<()> {
print_banner();
let mut config = WorkflowConfig::from_file(&workflow)
.with_context(|| format!("failed to parse {}", workflow.display()))?;
config.apply_defaults();
config
.expand_wildcards()
.context("failed to expand wildcard rules")?;
if let Some(provider) = crate::commands::ai_template::try_resolve_ai(Some(&workflow), ai) {
let mut debug_wildcards = std::collections::HashMap::new();
for (key, value) in &config.config {
let string_val = match value {
toml::Value::String(s) => s.clone(),
other => other.to_string(),
};
debug_wildcards.insert(format!("config.{key}"), string_val);
}
let expanded: Vec<String> = config
.rules
.iter()
.map(|r| {
let shell_cmd = r.shell.as_deref().unwrap_or("(no shell command)");
let shell = oxo_flow_core::executor::process::render_shell_command(
shell_cmd,
r,
&debug_wildcards,
oxo_flow_core::scheduler::detect_system_limits(),
);
format!(
"## {}\nthreads={}, memory={}, env={:?}\n```bash\n{}\n```",
r.name,
r.resources.threads,
r.resources.memory.as_deref().unwrap_or("1G"),
r.environment,
shell
)
})
.collect();
let joined = expanded.join("\n\n");
let system = r#"## Role
You are a bioinformatics pipeline debugger. Explain each expanded shell command in plain language,
flagging potential issues: parameter misuse, resource mismatch, missing required flags, or
logical errors. Output format per rule:
**rule_name**: <what this command does — 1 sentence>
⚠ <any issues found — 1 line per issue, or "No issues" if clean>
"#
.to_string();
let user = format!(
"## Workflow: {}\n\n## Expanded Commands\n\n{joined}\n\n## Task\nExplain each command and flag issues.",
workflow.display()
);
println!("{}", " Analyzing commands...".bold().cyan());
match provider.chat(&system, &user).await {
Ok(response) => {
println!("\n{}\n{response}", "AI Analysis".bold().green().underline());
return Ok(());
}
Err(e) => eprintln!(" AI analysis failed: {e}"),
}
}
let dag = WorkflowDag::from_rules_with_config(
&config.rules,
&config_placeholder_values(&config.config),
)
.context("failed to build workflow DAG")?;
let rules_to_show: Vec<&oxo_flow_core::rule::Rule> = if let Some(ref name) = rule_name {
match config.rules.iter().find(|r| r.name == *name) {
Some(r) => vec![r],
None => {
eprintln!("{} rule '{}' not found", "error:".bold().red(), name);
return Err(anyhow::anyhow!("rule not found"));
}
}
} else {
config.rules.iter().collect()
};
eprintln!(
"{} Debugging {} rules",
"Debug:".bold().cyan(),
rules_to_show.len()
);
let mut wildcard_values: HashMap<String, String> = HashMap::new();
for (key, value) in &config.config {
let string_val = match value {
toml::Value::String(s) => s.clone(),
other => other.to_string(),
};
wildcard_values.insert(format!("config.{key}"), string_val);
}
for rule in &rules_to_show {
eprintln!("{}", format!("── Rule: {} ──", rule.name).bold().cyan());
if let Some(ref desc) = rule.description {
eprintln!(" {} {}", "Description:".dimmed(), desc);
}
if !rule.output.is_empty() {
let expanded_outputs: Vec<String> = rule
.output
.iter()
.map(|o| {
oxo_flow_core::executor::checkpoint::expand_config_in_path(o, &wildcard_values)
})
.collect();
eprintln!(" {} {:?}", "Outputs:".dimmed(), expanded_outputs);
}
if let Some(ref cmd) = rule.shell {
let expanded = oxo_flow_core::executor::process::render_shell_command(
cmd,
rule,
&wildcard_values,
oxo_flow_core::scheduler::detect_system_limits(),
);
eprintln!(" {} {}", "Shell (expanded):".dimmed(), expanded);
}
if let Ok(deps) = dag.dependencies(&rule.name)
&& !deps.is_empty()
{
let mut deps = deps;
deps.sort();
deps.dedup();
eprintln!(" {} {:?}", "Dependencies:".dimmed(), deps);
}
eprintln!();
}
Ok(())
}
const DEFAULT_CHECKPOINT: &str = ".oxo-flow/checkpoint.json";
const SNAPSHOT_STALE_SECS: i64 = 300;
#[allow(clippy::too_many_arguments)]
async fn process_reentry(
config: &mut WorkflowConfig,
rule: &Rule,
workdir: &std::path::Path,
wildcard_values: &std::collections::HashMap<String, String>,
sched: &mut oxo_flow_core::scheduler::SchedulerState,
order: &mut Vec<String>,
order_set: &mut std::collections::HashSet<String>,
dag: &mut WorkflowDag,
checkpoint: &Arc<tokio::sync::Mutex<CheckpointState>>,
) -> anyhow::Result<()> {
let manifest_path = rule.checkpoint_manifest.as_ref().ok_or_else(|| {
anyhow::anyhow!("checkpoint rule '{}' has no checkpoint_manifest", rule.name)
})?;
let expanded =
oxo_flow_core::executor::checkpoint::expand_config_in_path(manifest_path, wildcard_values);
let full = workdir.join(&expanded);
let content = std::fs::read_to_string(&full).map_err(|e| {
anyhow::anyhow!(
"checkpoint rule '{}' produced no readable manifest at '{}': {e}",
rule.name,
full.display()
)
})?;
let (group, samples, pairs) = oxo_flow_core::reentry::parse_manifest(&content)
.map_err(|e| anyhow::anyhow!("checkpoint rule '{}': {e}", rule.name))?;
if samples.is_empty() && pairs.is_empty() {
return Ok(()); }
let new_names =
oxo_flow_core::reentry::apply_reentry(config, group.as_deref(), &samples, &pairs)?;
if new_names.is_empty() {
return Ok(()); }
let round = {
let mut ck = checkpoint.lock().await;
let round = ck.reentries.iter().map(|r| r.round).max().unwrap_or(0) + 1;
if round > oxo_flow_core::reentry::MAX_REENTRY_ROUNDS {
anyhow::bail!(
"re-entry round cap ({}) exceeded — checkpoint rule '{}' keeps discovering new values; this is a workflow bug",
oxo_flow_core::reentry::MAX_REENTRY_ROUNDS,
rule.name
);
}
ck.record_reentry(oxo_flow_core::reentry::ReentryRecord {
round,
rule: rule.name.clone(),
group,
samples,
pairs,
});
round
};
for name in &new_names {
sched.add_rule(name);
order_set.insert(name.clone());
order.push(name.clone());
}
*dag = WorkflowDag::from_rules_with_config(
&config.rules,
&config_placeholder_values(&config.config),
)
.context("failed to rebuild workflow DAG after re-entry")?;
tracing::info!(
rule = %rule.name,
round = round,
new_instances = ?new_names,
"checkpoint re-entry expanded the plan"
);
Ok(())
}
#[allow(clippy::too_many_arguments)]
async fn process_output_pattern_discovery(
config: &mut WorkflowConfig,
instance_name: &str,
workdir: &std::path::Path,
sched: &mut oxo_flow_core::scheduler::SchedulerState,
order: &mut Vec<String>,
order_set: &mut std::collections::HashSet<String>,
dag: &mut WorkflowDag,
checkpoint: &Arc<tokio::sync::Mutex<CheckpointState>>,
checkpoint_path: &std::path::Path,
) -> anyhow::Result<Vec<String>> {
let Some(template) = config.output_pattern_template_of(instance_name) else {
return Ok(Vec::new());
};
let Some(instance) = config.get_rule(instance_name).cloned() else {
return Ok(Vec::new());
};
let combos = if config.output_pattern_needs_discovery(&instance) {
config.discover_output_pattern_files(&instance, workdir)?
} else {
let bindings = config.instance_bindings(instance_name);
if bindings.is_empty()
|| config
.pending_output_pattern_consumers_of(&template)
.is_empty()
{
return Ok(Vec::new());
}
vec![bindings]
};
if combos.is_empty() {
eprintln!(
"{} output_pattern '{}' of rule '{}' matched no files — \
downstream consumers stay uninstantiated",
"Warning:".bold().yellow(),
instance.output_pattern.as_deref().unwrap_or_default(),
instance_name
);
tracing::warn!(
rule = instance_name,
pattern = ?instance.output_pattern,
"output_pattern discovery matched no files"
);
return Ok(Vec::new());
}
config.contribute_output_pattern_domain(&template, combos.clone());
{
let mut ck = checkpoint.lock().await;
ck.record_output_pattern_domain(&template, combos);
if let Err(save_err) = ck.save_to_file(checkpoint_path) {
tracing::warn!("Failed to save checkpoint: {save_err}");
}
}
let belongs_to = |r: &Rule| {
r.output_pattern.is_some()
&& (r.name == template
|| config.expansion_templates.get(&r.name).map(String::as_str)
== Some(template.as_str()))
};
let outstanding = config.rules.iter().any(|r| {
belongs_to(r) && sched.status(&r.name) != Some(oxo_flow_core::executor::JobStatus::Success)
});
if outstanding {
return Ok(Vec::new());
}
let new_names = config.expand_output_pattern_consumers()?;
if new_names.is_empty() {
return Ok(Vec::new());
}
for name in &new_names {
sched.add_rule(name);
order_set.insert(name.clone());
order.push(name.clone());
}
*dag = WorkflowDag::from_rules_with_config(
&config.rules,
&config_placeholder_values(&config.config),
)
.context("failed to rebuild workflow DAG after output_pattern fan-out")?;
tracing::info!(
producer = template,
new_instances = ?new_names,
"output_pattern discovery expanded the plan"
);
Ok(new_names)
}
fn rule_timings(state: &CheckpointState) -> (Vec<(&str, f64)>, f64) {
let mut timings: Vec<(&str, f64)> = state
.benchmarks
.iter()
.filter(|(_, b)| b.recorded_as.is_none())
.map(|(rule, bench)| (rule.as_str(), bench.wall_time_secs))
.collect();
timings.sort_by(|a, b| b.1.total_cmp(&a.1));
let total = timings.iter().map(|(_, secs)| secs).sum();
(timings, total)
}
fn print_target_skipped_note(
skipped: &str,
when_false_rules: &std::collections::HashSet<String>,
run_log: Option<&SharedRunLog>,
) {
let detail = if when_false_rules.contains(skipped) {
format!(
"target '{skipped}' is when-gated false — it never runs; \
removed from the execution set (its upstream was pruned too)"
)
} else {
format!(
"target '{skipped}' is unreachable — its upstream is when-gated \
false; removed from the execution set"
)
};
diagnostic_narrate(format_args!("{} {detail}", "Note:".yellow()), run_log);
}
fn run_workdir_check(workdir: Option<&std::path::Path>, path: &str) -> bool {
let p = std::path::Path::new(path);
if p.is_absolute() {
p.exists()
} else {
workdir
.unwrap_or_else(|| std::path::Path::new("."))
.join(p)
.exists()
}
}
#[allow(clippy::type_complexity)]
fn compute_staleness_reasons(
state: &CheckpointState,
checkpoint_path: &std::path::Path,
) -> std::collections::BTreeMap<String, (String, String)> {
use crate::commands::run_preview::RuleStatus;
let mut reasons = std::collections::BTreeMap::new();
if state.completed_rules.is_empty() {
return reasons;
}
let Some(workflow_path) = state.workflow_path.as_ref() else {
return reasons;
};
let workflow = std::path::PathBuf::from(workflow_path);
if !workflow.exists() {
return reasons;
}
let Ok(config) = WorkflowConfig::from_file(&workflow) else {
return reasons;
};
let workflow_dir = oxo_flow_core::parent_dir(&workflow).to_path_buf();
let Ok(dag) = WorkflowDag::from_rules_with_config(
&config.rules,
&config_placeholder_values(&config.config),
) else {
return reasons;
};
let Ok(raw_order) = dag.execution_order() else {
return reasons;
};
let order_set: std::collections::HashSet<String> = raw_order.into_iter().collect();
let mut order: Vec<String> = Vec::new();
let Ok(groups) = dag.parallel_groups() else {
return reasons;
};
for group in groups {
let mut level: Vec<String> = group
.into_iter()
.filter(|n| order_set.contains(n))
.collect();
level.sort();
order.extend(level);
}
let wildcard_values = config_placeholder_values(&config.config);
let sensitive_keys: std::collections::HashSet<String> = config
.config_meta
.iter()
.filter(|(_, def)| def.sensitive)
.map(|(key, _)| key.clone())
.collect();
let interpreter_map = config.workflow.interpreter_map.clone();
let mut adopted = state.clone();
crate::commands::run::adopt_checkpoint_for_workflow(&mut adopted, &workflow, None);
let preview = crate::commands::run_preview::preview_run_plan(
&adopted,
&config,
&dag,
&order,
&workflow_dir,
&wildcard_values,
&sensitive_keys,
&interpreter_map,
checkpoint_path,
false,
false,
);
for entry in &preview.plan {
if !state.completed_rules.contains(&entry.name) {
continue;
}
let (reason, detail) = match entry.status {
RuleStatus::Skipped | RuleStatus::SkippedByWhen | RuleStatus::SkippedFresh => {
continue;
}
RuleStatus::NeverCompleted if !adopted.completed_rules.contains(&entry.name) => (
"checkpoint reset".to_string(),
"completed credit was invalidated (foreign workflow adoption)".to_string(),
),
RuleStatus::NeverCompleted => (
"output missing".to_string(),
"completion record present but the executor would re-run — outputs are gone or failed cleanup removed them".to_string(),
),
RuleStatus::ConfigInvalidated => (
"config changed".to_string(),
"an effective config value differs from the recorded snapshot".to_string(),
),
RuleStatus::InputInvalidated => (
"input changed".to_string(),
"recorded input manifest (paths/size/mtime) no longer matches disk".to_string(),
),
RuleStatus::OutputsMissing => (
"output missing".to_string(),
"recorded output is gone from the workdir".to_string(),
),
RuleStatus::Cascaded { from: ref f1 } => (
"cascaded".to_string(),
format!("upstream rule '{f1}' will re-run"),
),
RuleStatus::CascadedUpstream { from: ref f2 } => (
"missing input".to_string(),
format!(
"a downstream rule needs an intermediate that is gone; producer '{f2}' re-runs first"
),
),
RuleStatus::Forced => ("forced".to_string(), "--rerun was requested".to_string()),
};
reasons.insert(entry.name.clone(), (reason, detail));
}
reasons
}
fn reconcile_failed_rules<'a>(
failed_rules: &'a std::collections::BTreeSet<String>,
when_verdicts: &'a std::collections::BTreeMap<String, bool>,
) -> (Vec<&'a str>, Vec<&'a str>) {
let mut real_failed = Vec::new();
let mut skipped_when_false = Vec::new();
for rule in failed_rules {
if when_verdicts.get(rule) == Some(&false) {
skipped_when_false.push(rule.as_str());
} else {
real_failed.push(rule.as_str());
}
}
(real_failed, skipped_when_false)
}
pub async fn handle_status(
checkpoint: Option<PathBuf>,
workdir: Option<PathBuf>,
json: bool,
timing: bool,
limit: usize,
) -> Result<()> {
print_banner();
if let Some(path) = &checkpoint
&& path.extension().is_some_and(|ext| ext == "oxoflow")
{
eprintln!(
"{} '{}' appears to be a workflow file, not a checkpoint.",
"Warning:".bold().yellow(),
path.display()
);
eprintln!(
" The 'status' command expects a checkpoint file (e.g., .oxo-flow/checkpoint.json)."
);
eprintln!(
" Run 'oxo-flow run {}' first to generate a checkpoint.",
path.display()
);
anyhow::bail!("Cannot read workflow file as checkpoint");
}
let checkpoint_path = match (&checkpoint, &workdir) {
(Some(p), _) => p.clone(),
(None, Some(dir)) => dir.join(".oxo-flow/checkpoint.json"),
(None, None) => PathBuf::from(DEFAULT_CHECKPOINT),
};
let state = CheckpointState::load_from_file(&checkpoint_path).with_context(|| {
format!(
"failed to load checkpoint from '{}'.\n \
Check that the file exists and is a valid checkpoint (JSON format).\n \
Checkpoint files are generated automatically by 'oxo-flow run' in .oxo-flow/checkpoint.json.",
checkpoint_path.display()
)
})?;
let checkpoint_mtime = std::fs::metadata(&checkpoint_path)
.and_then(|m| m.modified())
.ok()
.map(|t| {
let t: chrono::DateTime<chrono::Utc> = t.into();
t
});
let snapshot_age_secs = checkpoint_mtime.as_ref().map(|m| {
chrono::Utc::now()
.signed_duration_since(*m)
.num_seconds()
.max(0)
});
if snapshot_age_secs.is_some_and(|age| age > SNAPSHOT_STALE_SECS) {
tracing::warn!(
"checkpoint snapshot is {}s old (threshold {}s)",
snapshot_age_secs.unwrap(),
SNAPSHOT_STALE_SECS
);
}
let mut completed: Vec<&str> = state.completed_rules.iter().map(String::as_str).collect();
completed.sort_unstable();
let (failed, when_gated_off) =
reconcile_failed_rules(&state.failed_rules, &state.when_verdicts);
let staleness = compute_staleness_reasons(&state, &checkpoint_path);
let staleness_json: Option<serde_json::Map<String, serde_json::Value>> = if staleness.is_empty()
{
None
} else {
Some(
staleness
.iter()
.map(|(rule, (reason, detail))| {
(
rule.clone(),
serde_json::json!({
"status": reason,
"detail": detail,
}),
)
})
.collect(),
)
};
if json {
let mut output = serde_json::json!({
"command": "status",
"checkpoint": checkpoint_path.display().to_string(),
"workflow": state.workflow_path,
"completed": completed,
"failed": failed,
});
if let Some(mtime) = checkpoint_mtime {
output["checkpoint_mtime"] =
serde_json::json!(mtime.to_rfc3339_opts(chrono::SecondsFormat::Secs, true));
}
if let Some(age) = snapshot_age_secs {
output["snapshot_age_secs"] = serde_json::json!(age);
}
if !when_gated_off.is_empty() {
output["skipped_by_when"] = serde_json::json!(when_gated_off);
}
if let Some(map) = staleness_json {
output["staleness"] = serde_json::Value::Object(map);
}
if timing {
let (timings, total) = rule_timings(&state);
let timings_map: serde_json::Map<String, serde_json::Value> = timings
.into_iter()
.map(|(rule, secs)| (rule.to_string(), serde_json::json!(secs)))
.collect();
output["timings"] = serde_json::Value::Object(timings_map);
output["total_time_secs"] = serde_json::json!(total);
let memory_map: serde_json::Map<String, serde_json::Value> = state
.benchmarks
.iter()
.filter(|(_, b)| b.max_memory_mb.is_some())
.map(|(rule, b)| {
(
rule.to_string(),
serde_json::json!({
"max_memory_mb": b.max_memory_mb,
"memory_limit_mb": b.memory_limit_mb,
}),
)
})
.collect();
if !memory_map.is_empty() {
output["memory"] = serde_json::Value::Object(memory_map);
}
}
println!("{}", serde_json::to_string_pretty(&output)?);
return Ok(());
}
eprintln!(
"{} Status for checkpoint: {}",
"Status:".bold().cyan(),
checkpoint_path.display()
);
if let (Some(mtime), Some(age)) = (checkpoint_mtime, snapshot_age_secs) {
let written = mtime.with_timezone(&chrono::Local).format("%H:%M:%S");
if age > SNAPSHOT_STALE_SECS {
eprintln!(
" {} Snapshot is {}s old (checkpoint written {} local) — flush lag, not necessarily stalled progress",
"Warning:".bold().yellow(),
age,
written
);
} else {
eprintln!(
" State as of {}s ago (checkpoint written {} local)",
age, written
);
}
}
eprintln!(" Completed: {}", completed.len());
eprintln!(" Failed: {}", failed.len());
if !when_gated_off.is_empty() {
eprintln!(" Skipped (when-gated off): {}", when_gated_off.len());
}
if timing {
let (timings, total) = rule_timings(&state);
if !timings.is_empty() {
eprintln!(
"\n{} (top {}, total {:.1}s)",
"Rule timings:".bold().green(),
limit.min(timings.len()),
total
);
for (rule, secs) in timings.iter().take(limit) {
let mem = state.benchmarks.get(*rule).and_then(|b| {
b.max_memory_mb.map(|m| match b.memory_limit_mb {
Some(l) if l > 0 => {
if m * 100 >= l * 80 {
format!(" peak {m}/{l} MiB ⚠")
} else {
format!(" peak {m}/{l} MiB")
}
}
_ => format!(" peak {m} MiB"),
})
});
match mem {
Some(mem) => eprintln!(" {} {} ({:.1}s){}", "✓".green(), rule, secs, mem),
None => eprintln!(" {} {} ({:.1}s)", "✓".green(), rule, secs),
}
}
}
} else {
if !staleness.is_empty() {
eprintln!("\n{}", "Staleness reasons (as of now):".bold().yellow());
eprintln!(
" {}",
"why each completed rule would re-run under the current workflow+config".dimmed()
);
for rule in &completed {
match staleness.get(*rule) {
Some((reason, detail)) => {
eprintln!(
" {} {} — {} ({})",
"✓".green(),
rule,
reason,
detail.dimmed()
);
}
None => eprintln!(" {} {} — up to date", "✓".green(), rule),
}
}
}
if !completed.is_empty() {
eprintln!("\n{}", "Completed rules:".bold().green());
for rule in &completed {
eprintln!(" {} {}", "✓".green(), rule);
}
}
if !failed.is_empty() {
eprintln!("\n{}", "Failed rules:".bold().red());
for rule in &failed {
eprintln!(" {} {}", "✗".red(), rule);
}
}
if !when_gated_off.is_empty() {
eprintln!("\n{}", "Skipped (when condition false):".bold().yellow());
for rule in &when_gated_off {
eprintln!(
" {} {} — {}",
"⊘".yellow(),
rule,
"condition evaluated to false".dimmed()
);
}
}
}
Ok(())
}
#[allow(clippy::too_many_arguments)]
pub async fn resume_command(
checkpoint: PathBuf,
jobs: usize,
ai_recover: bool,
ai_max_retries: Option<u32>,
keep_going: bool,
timeout: String,
workdir: Option<PathBuf>,
no_report_snapshot: bool,
) -> Result<()> {
print_banner();
let state = CheckpointState::load_from_file(&checkpoint)
.with_context(|| {
format!(
"failed to load checkpoint from '{}'.\n \
Check that the file exists and is a valid checkpoint (JSON format).",
checkpoint.display()
)
})
.map_err(|e| {
if checkpoint.extension().is_some_and(|ext| ext == "oxoflow") {
anyhow::anyhow!(
"{e}\n '{}' appears to be a workflow file, not a checkpoint.\n \
The 'resume' command expects a checkpoint file (e.g., .oxo-flow/checkpoint.json).",
checkpoint.display()
)
} else {
e
}
})?;
let workflow_path = match &state.workflow_path {
Some(p) => PathBuf::from(p),
None => {
anyhow::bail!(
"Checkpoint does not contain a workflow reference.\n \
The checkpoint at '{}' was generated by an older version of oxo-flow.\n \
To resume manually, run: oxo-flow run <workflow.oxoflow>\n \
(oxo-flow run automatically resumes from the checkpoint)",
checkpoint.display()
);
}
};
if !workflow_path.exists() {
anyhow::bail!(
"Workflow file '{}' referenced by checkpoint no longer exists.\n \
The workflow may have been moved or deleted.",
workflow_path.display()
);
}
let completed = state.completed_rules.len();
let (failed, skipped_when_false) =
reconcile_failed_rules(&state.failed_rules, &state.when_verdicts);
let failed = failed.len();
eprintln!(
"{} Resuming workflow '{}'",
"Resume:".bold().cyan(),
workflow_path.display()
);
eprintln!(" Checkpoint: {}", checkpoint.display());
let remaining = WorkflowConfig::from_file(&workflow_path)
.map(|cfg| cfg.rules.len().saturating_sub(completed))
.unwrap_or(failed);
eprintln!(" State: {completed} completed, {failed} failed, {remaining} remaining");
if !skipped_when_false.is_empty() {
eprintln!(
" Note: {} rule(s) were skipped by their `when` condition (recorded false); the gate is re-judged on this resume.",
skipped_when_false.len()
);
}
if !state.running_rules().is_empty() {
eprintln!(
" Note: {} rule(s) were still running when the previous run died; they will re-run.",
state.running_rules().len()
);
}
if completed == 0 && failed == 0 {
eprintln!(
" {} No rules have been executed yet. Use 'oxo-flow run' instead.",
"Note:".yellow()
);
return Ok(());
}
eprintln!();
eprintln!(
"{} Launching executor with {} parallel job(s)...",
"Info:".bold().cyan(),
jobs
);
let effective_workdir = workdir.or_else(|| state.workdir.clone().map(PathBuf::from));
eprintln!(
" Workdir: {}",
effective_workdir
.as_deref()
.map(|p| p.display().to_string())
.unwrap_or_else(|| "(workflow directory)".to_string())
);
run_command(
Some(workflow_path),
jobs,
keep_going, effective_workdir, None, Vec::new(), Vec::new(), 0, timeout, false, None, 0, 0, false, true, None, false, false, Vec::new(), ai_recover,
ai_max_retries,
Vec::new(), false, no_report_snapshot,
None, Some(checkpoint), )
.await
}
#[derive(Debug, PartialEq)]
struct BackendPreflightFailure {
kind: String,
message: String,
example_rule: String,
spec: String,
count: usize,
}
impl std::fmt::Display for BackendPreflightFailure {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
"{kind}: {message} — required by {count} pending rule(s), e.g. `{example_rule}` (environment: {spec}); {hint}",
kind = self.kind,
message = self.message,
count = self.count,
example_rule = self.example_rule,
spec = self.spec,
hint = backend_install_hint(self),
)
}
}
fn backend_install_hint(failure: &BackendPreflightFailure) -> &'static str {
if failure.kind == "pixi" && failure.message.contains("not found") {
return "create the manifest or correct the rule's `pixi` path (resolved relative to the workflow directory)";
}
match failure.kind.as_str() {
"pixi" => "install pixi (https://pixi.sh) and ensure it is on PATH",
"mamba" => "install mamba or micromamba and ensure it is on PATH",
"conda" => "install conda and ensure it is on PATH",
"docker" => "install/start Docker and ensure `docker` is on PATH",
"singularity" => "install Singularity/Apptainer and ensure it is on PATH",
"modules" => "ensure environment modules (modulecmd) is on PATH",
_ => "make the environment backend available on PATH",
}
}
fn primary_env_spec(env: &oxo_flow_core::rule::EnvironmentSpec) -> String {
env.mamba
.as_deref()
.or(env.conda.as_deref())
.or(env.pixi.as_deref())
.or(env.docker.as_deref())
.or(env.singularity.as_deref())
.or(env.venv.as_deref())
.map(str::to_string)
.unwrap_or_else(|| env.modules.join(","))
}
fn backend_preflight_failures(
pending: &[(String, oxo_flow_core::rule::EnvironmentSpec)],
resolver: &oxo_flow_core::environment::EnvironmentResolver,
) -> Vec<BackendPreflightFailure> {
let mut spec_slot: std::collections::HashMap<String, Option<usize>> =
std::collections::HashMap::new();
let mut msg_slot: std::collections::HashMap<(String, String), usize> =
std::collections::HashMap::new();
let mut out: Vec<BackendPreflightFailure> = Vec::new();
for (rule_name, env) in pending {
let key = resolver.cache_key(env);
match spec_slot.get(&key) {
Some(Some(slot)) => {
out[*slot].count += 1;
continue;
}
Some(None) => continue, None => {}
}
if let Err(err) = resolver.validate_spec(env) {
let (kind, message) = match &err {
oxo_flow_core::OxoFlowError::Environment { kind, message } => {
(kind.clone(), message.clone())
}
other => ("environment".to_string(), other.to_string()),
};
let slot = match msg_slot.get(&(kind.clone(), message.clone())) {
Some(&slot) => {
out[slot].count += 1;
slot
}
None => {
out.push(BackendPreflightFailure {
kind: kind.clone(),
message: message.clone(),
example_rule: rule_name.clone(),
spec: primary_env_spec(env),
count: 1,
});
msg_slot.insert((kind, message), out.len() - 1);
out.len() - 1
}
};
spec_slot.insert(key, Some(slot));
} else {
spec_slot.insert(key, None);
}
}
out
}
#[cfg(test)]
mod tests {
use super::{
age_ready_list, ai_attempts, apply_cli_overrides, cleanup_cache_dir, closest_declared_key,
config_placeholder_values, failure_exit_code, known_modules_hint, missing_source_findings,
missing_source_lines, parse_cli_overrides, parse_disk_hold_abort_rounds,
parse_disk_tick_secs, parse_spawn_watchdog_threshold, reconcile_failed_rules,
substitute_source_placeholder, suggest_jobs,
};
use oxo_flow_core::dag::WorkflowDag;
use std::collections::{HashMap, HashSet};
#[test]
fn reconcile_failed_rules_separates_real_failures_from_when_false_skips() {
let failed: std::collections::BTreeSet<String> = ["align", "qc_S2", "trim"]
.into_iter()
.map(str::to_string)
.collect();
let verdicts: std::collections::BTreeMap<String, bool> =
[("qc_S2".to_string(), false)].into_iter().collect();
let (real_failed, skipped) = reconcile_failed_rules(&failed, &verdicts);
assert_eq!(real_failed, vec!["align", "trim"]);
assert_eq!(skipped, vec!["qc_S2"]);
}
#[test]
fn reconcile_failed_rules_keeps_failures_without_a_recorded_verdict() {
let failed: std::collections::BTreeSet<String> =
["a".to_string(), "b".to_string()].into_iter().collect();
let verdicts: std::collections::BTreeMap<String, bool> =
[("a".to_string(), true)].into_iter().collect();
let (real_failed, skipped) = reconcile_failed_rules(&failed, &verdicts);
assert_eq!(real_failed, vec!["a", "b"]);
assert!(skipped.is_empty());
}
#[test]
fn parse_disk_env_knobs_fall_back_to_defaults() {
assert_eq!(parse_disk_tick_secs(None), 60);
assert_eq!(parse_disk_tick_secs(Some("")), 60);
assert_eq!(parse_disk_tick_secs(Some("abc")), 60);
assert_eq!(parse_disk_tick_secs(Some("0")), 60);
assert_eq!(parse_disk_tick_secs(Some("-5")), 60);
assert_eq!(parse_disk_hold_abort_rounds(None), 2);
assert_eq!(parse_disk_hold_abort_rounds(Some("xyz")), 2);
assert_eq!(parse_disk_hold_abort_rounds(Some("0")), 2);
}
#[test]
fn parse_disk_env_knobs_honor_valid_values() {
assert_eq!(parse_disk_tick_secs(Some("1")), 1);
assert_eq!(parse_disk_tick_secs(Some("300")), 300);
assert_eq!(parse_disk_hold_abort_rounds(Some("1")), 1);
assert_eq!(parse_disk_hold_abort_rounds(Some("10")), 10);
}
#[test]
fn snapshot_age_is_derived_from_checkpoint_mtime() {
let dir = std::env::temp_dir().join(format!(
"oxoflow-status-age-{}-{}",
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos()
));
std::fs::create_dir_all(&dir).unwrap();
let path = dir.join("checkpoint.json");
std::fs::write(&path, "{}").unwrap();
let mtime_now: chrono::DateTime<chrono::Utc> =
std::fs::metadata(&path).unwrap().modified().unwrap().into();
let age_now = chrono::Utc::now()
.signed_duration_since(mtime_now)
.num_seconds()
.max(0);
assert!(age_now < 60, "fresh file reported {age_now}s old");
let backdated = chrono::Utc::now() - chrono::Duration::seconds(600);
let age_old = chrono::Utc::now()
.signed_duration_since(backdated)
.num_seconds()
.max(0);
assert_eq!(age_old, 600);
assert!(age_old > super::SNAPSHOT_STALE_SECS);
std::fs::remove_dir_all(&dir).ok();
}
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_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 suggest_jobs_evaluates_each_wave_against_its_own_heaviest_rule() {
let mut toml = String::from("[workflow]\nname = \"t\"\nversion = \"0.1\"\n");
for i in 0..20 {
toml.push_str(&format!(
"\n[[rules]]\nname = \"a{i}\"\noutput = [\"a{i}.txt\"]\nshell = \"true\"\n\n[rules.resources]\nthreads = 1\n"
));
}
for i in 0..2 {
toml.push_str(&format!(
"\n[[rules]]\nname = \"b{i}\"\noutput = [\"b{i}.txt\"]\nshell = \"true\"\n\n[rules.resources]\nthreads = 16\n"
));
}
for i in 0..20 {
toml.push_str(&format!(
"\n[[rules]]\nname = \"c{i}\"\noutput = [\"c{i}.txt\"]\nshell = \"true\"\n\n[rules.resources]\nthreads = 1\n"
));
}
let config = config_with_rules(&toml);
let groups = vec![
(0..20).map(|i| format!("a{i}")).collect::<Vec<_>>(),
(0..2).map(|i| format!("b{i}")).collect::<Vec<_>>(),
(0..20).map(|i| format!("c{i}")).collect::<Vec<_>>(),
];
assert_eq!(suggest_jobs(&config, &groups, 64), 20);
assert_ne!(suggest_jobs(&config, &groups, 64), 4);
}
#[test]
fn suggest_jobs_clamps_to_the_wave_width_and_system_threads() {
let mut toml = String::from("[workflow]\nname = \"t\"\nversion = \"0.1\"\n");
for i in 0..5 {
toml.push_str(&format!(
"\n[[rules]]\nname = \"h{i}\"\noutput = [\"h{i}.txt\"]\nshell = \"true\"\n\n[rules.resources]\nthreads = 16\n"
));
}
let config = config_with_rules(&toml);
let groups = vec![(0..5).map(|i| format!("h{i}")).collect::<Vec<_>>()];
assert_eq!(suggest_jobs(&config, &groups, 4), 1);
let mut toml = String::from("[workflow]\nname = \"t\"\nversion = \"0.1\"\n");
for i in 0..20 {
toml.push_str(&format!(
"\n[[rules]]\nname = \"l{i}\"\noutput = [\"l{i}.txt\"]\nshell = \"true\"\n\n[rules.resources]\nthreads = 1\n"
));
}
let config = config_with_rules(&toml);
let groups = vec![(0..20).map(|i| format!("l{i}")).collect::<Vec<_>>()];
assert_eq!(suggest_jobs(&config, &groups, 64), 20);
}
#[test]
fn suggest_jobs_defaults_to_one_without_rules() {
let config = config_with_rules("[workflow]\nname = \"t\"\nversion = \"0.1\"\n");
assert_eq!(suggest_jobs(&config, &[], 64), 1);
assert_eq!(suggest_jobs(&config, &[vec![]], 64), 1);
}
#[test]
fn cleanup_cache_dir_is_recursive() {
let dir = tempfile::tempdir().unwrap();
let nested = dir.path().join("sub").join("deeper");
std::fs::create_dir_all(&nested).unwrap();
std::fs::write(dir.path().join("old.txt"), "x").unwrap();
std::fs::write(nested.join("old-too.txt"), "y").unwrap();
let removed = cleanup_cache_dir(dir.path(), 0);
assert_eq!(removed, 2, "file + one nested directory removed");
assert!(!dir.path().join("old.txt").exists());
assert!(!dir.path().join("sub").exists());
}
fn declared(keys: &[&str]) -> HashSet<String> {
keys.iter().map(|k| k.to_string()).collect()
}
fn map_of(pairs: &[(&str, i32)]) -> std::collections::HashMap<String, i32> {
pairs.iter().map(|(k, v)| (k.to_string(), *v)).collect()
}
fn set_of(names: &[&str]) -> HashSet<String> {
names.iter().map(|n| n.to_string()).collect()
}
#[test]
fn age_ready_list_orders_by_declared_priority_then_name() {
let priority = map_of(&[("high", 20), ("low", 10)]);
let ready = age_ready_list(
vec!["low".into(), "high".into()],
&set_of(&["high", "low"]),
&HashSet::new(),
&priority,
&std::collections::HashMap::new(),
);
assert_eq!(ready, vec!["high", "low"]);
}
#[test]
fn age_ready_list_aged_rule_overtakes_higher_declared_priority() {
let priority = map_of(&[("merge", 20), ("dump", 10)]);
let waited = map_of(&[("dump", 15)]);
let ready = age_ready_list(
vec!["merge".into(), "dump".into()],
&set_of(&["merge", "dump"]),
&HashSet::new(),
&priority,
&waited,
);
assert_eq!(ready, vec!["dump", "merge"]);
}
#[test]
fn age_ready_list_drops_submitted_and_out_of_scope_rules() {
let priority = map_of(&[("a", 5), ("b", 5), ("c", 5)]);
let submitted = set_of(&["b"]);
let ready = age_ready_list(
vec!["c".into(), "b".into(), "a".into()],
&set_of(&["a", "b", "c"]),
&submitted,
&priority,
&std::collections::HashMap::new(),
);
assert_eq!(ready, vec!["a", "c"]);
}
#[test]
fn age_ready_list_treats_missing_priority_and_waits_as_zero() {
let ready = age_ready_list(
vec!["z".into(), "a".into()],
&set_of(&["a", "z"]),
&HashSet::new(),
&std::collections::HashMap::new(),
&std::collections::HashMap::new(),
);
assert_eq!(ready, vec!["a", "z"]);
}
#[test]
fn accepts_equals_forms_for_any_key() {
let map = parse_cli_overrides(
vec![
"threads=8".to_string(),
"--mode=protein".to_string(),
"--new_key=1".to_string(), ],
&declared(&["threads", "mode"]),
"run",
)
.unwrap();
assert_eq!(map["threads"], "8");
assert_eq!(map["mode"], "protein");
assert_eq!(map["new_key"], "1");
}
#[test]
fn accepts_space_form_for_declared_keys() {
let map = parse_cli_overrides(
vec!["--min_quality".to_string(), "45".to_string()],
&declared(&["min_quality"]),
"run",
)
.unwrap();
assert_eq!(map["min_quality"], "45");
}
#[test]
fn canonicalizes_hyphen_spelled_override_to_declared_key() {
let map = parse_cli_overrides(
vec![
"--api-token=sk-OVERRIDE-123".to_string(),
"min-quality=45".to_string(),
],
&declared(&["api_token", "min_quality"]),
"run",
)
.unwrap();
assert_eq!(map["api_token"], "sk-OVERRIDE-123");
assert_eq!(map["min_quality"], "45");
assert!(!map.contains_key("api-token"), "{map:?}");
assert!(!map.contains_key("min-quality"), "{map:?}");
}
#[test]
fn canonicalizes_space_form_for_hyphen_spelled_declared_key() {
let map = parse_cli_overrides(
vec!["--api-token".to_string(), "sk-OVERRIDE-123".to_string()],
&declared(&["api_token"]),
"run",
)
.unwrap();
assert_eq!(map["api_token"], "sk-OVERRIDE-123");
}
#[test]
fn canonicalization_wins_on_exact_match_and_stays_silent_on_ambiguity() {
let map = parse_cli_overrides(
vec!["--sub-mode=2".to_string()],
&declared(&["sub_mode"]),
"run",
)
.unwrap();
assert_eq!(map["sub_mode"], "2");
let map = parse_cli_overrides(
vec!["--a-b=1".to_string()],
&declared(&["a_b", "a-b"]),
"run",
)
.unwrap();
assert_eq!(map["a-b"], "1");
let map = parse_cli_overrides(
vec!["--totally-unknown=7".to_string()],
&declared(&["min_quality"]),
"run",
)
.unwrap();
assert_eq!(map["totally-unknown"], "7");
let err = parse_cli_overrides(
vec!["--totally-unknown".to_string(), "7".to_string()],
&declared(&["min_quality"]),
"run",
)
.unwrap_err();
assert!(format!("{err}").contains("unknown argument"), "{err:?}");
}
#[test]
fn apply_cli_overrides_puts_override_on_declared_sensitive_key() {
let mut config = config_with_rules(
r#"
[workflow]
name = "t"
version = "1.0"
[config]
api_token = { default = "sk-DEFAULT-abc", sensitive = true, help = "API token" }
[[rules]]
name = "r"
shell = "true"
"#,
);
let overrides = parse_cli_overrides(
vec!["--api-token=sk-OVERRIDE-123".to_string()],
&declared(&["api_token"]),
"run",
)
.unwrap();
apply_cli_overrides(&mut config, &overrides).unwrap();
assert_eq!(config.config["api_token"].as_str(), Some("sk-OVERRIDE-123"));
}
#[test]
fn rejects_unknown_dash_token() {
let err = parse_cli_overrides(
vec!["--config".to_string(), "config/x.toml".to_string()],
&declared(&["min_quality"]),
"run",
)
.unwrap_err();
let msg = format!("{err}");
assert!(msg.contains("unknown argument"), "{msg}");
assert!(msg.contains("--config"), "{msg}");
assert!(msg.contains("KEY=VALUE"), "{msg}");
}
#[test]
fn rejects_known_flag_after_positional_with_flag_error() {
let err = parse_cli_overrides(
vec!["threads=8".to_string(), "--json".to_string()],
&declared(&["threads"]),
"run",
)
.unwrap_err();
let msg = format!("{err}");
assert!(msg.contains("command flag"), "{msg}");
assert!(msg.contains("--json"), "{msg}");
}
#[test]
fn dry_run_treats_shared_run_flags_as_misordered_not_unsupported() {
let err = parse_cli_overrides(
vec!["threads=8".to_string(), "--cache-dir=x".to_string()],
&declared(&["threads"]),
"dry-run",
)
.unwrap_err();
let msg = format!("{err}");
assert!(msg.contains("command flag"), "{msg}");
assert!(msg.contains("must come before"), "{msg}");
let err = parse_cli_overrides(
vec!["threads=8".to_string(), "--provenance".to_string()],
&declared(&["threads"]),
"dry-run",
)
.unwrap_err();
let msg = format!("{err}");
assert!(msg.contains("not supported by dry-run"), "{msg}");
}
#[test]
fn rejects_space_form_for_undeclared_key_even_via_arg() {
let err = parse_cli_overrides(
vec!["--config".to_string(), "x".to_string()],
&declared(&[]),
"run",
)
.unwrap_err();
assert!(format!("{err}").contains("unknown argument"));
}
#[test]
fn rejects_declared_key_without_value() {
let err = parse_cli_overrides(
vec!["--min_quality".to_string()],
&declared(&["min_quality"]),
"run",
)
.unwrap_err();
assert!(format!("{err}").contains("invalid config flag"), "{err:?}");
}
#[test]
fn rejects_space_form_that_would_swallow_a_command_flag() {
let err = parse_cli_overrides(
vec!["--mode".to_string(), "--json".to_string()],
&declared(&["mode"]),
"run",
)
.unwrap_err();
let msg = format!("{err}");
assert!(msg.contains("--mode"), "{msg}");
assert!(msg.contains("--json"), "{msg}");
assert!(msg.contains("--mode=<value>"), "{msg}");
}
#[test]
fn ai_attempts_defaults_to_one_and_never_zero() {
assert_eq!(ai_attempts(None), 1);
assert_eq!(ai_attempts(Some(0)), 1);
assert_eq!(ai_attempts(Some(3)), 3);
}
#[test]
fn failure_exit_code_parses_recorded_reasons() {
assert_eq!(
failure_exit_code("exit code 3 — ERROR: unknown reference genome hg999"),
3
);
assert_eq!(failure_exit_code("exit code 137"), 137);
assert_eq!(failure_exit_code("exit code 137\nstderr: ..."), 137);
assert_eq!(failure_exit_code("task panicked: boom"), -1);
assert_eq!(failure_exit_code(""), -1);
assert_eq!(failure_exit_code("exit code notanumber"), -1);
}
#[test]
fn closest_declared_key_names_a_typo_but_not_an_unrelated_key() {
let keys = declared(&["sample_name", "threads", "min_quality"]);
assert_eq!(
closest_declared_key("samplename", &keys),
Some("sample_name")
);
assert_eq!(closest_declared_key("threds", &keys), Some("threads"));
assert_eq!(closest_declared_key("completely_different", &keys), None);
}
#[test]
fn rejects_bare_non_key_value_positional() {
let err =
parse_cli_overrides(vec!["naked".to_string()], &declared(&[]), "run").unwrap_err();
assert!(format!("{err}").contains("KEY=VALUE"), "{err:?}");
}
#[test]
fn accepts_empty_value_overrides() {
let map = parse_cli_overrides(
vec![
"gene_bed=".to_string(),
"--chrom_sizes".to_string(),
String::new(),
"--transcript_fasta=".to_string(),
],
&declared(&["gene_bed", "chrom_sizes", "transcript_fasta"]),
"run",
)
.unwrap();
assert_eq!(map["gene_bed"], "");
assert_eq!(map["chrom_sizes"], "");
assert_eq!(map["transcript_fasta"], "");
}
#[test]
fn rejects_empty_key_but_accepts_empty_value() {
let err = parse_cli_overrides(vec!["=v".to_string()], &declared(&[]), "run").unwrap_err();
assert!(
format!("{err}").contains("KEY must be non-empty"),
"{err:?}"
);
}
#[test]
fn dry_run_flag_error_is_command_aware() {
let err = parse_cli_overrides(
vec!["-j".to_string(), "12".to_string()],
&declared(&[]),
"dry-run",
)
.unwrap_err();
let msg = format!("{err}");
assert!(msg.contains("not supported by dry-run"), "{msg}");
assert!(
!msg.contains("command flag, not a config override"),
"{msg}"
);
}
#[test]
fn run_flag_error_under_run_keeps_ordering_hint() {
let err = parse_cli_overrides(
vec!["threads=8".to_string(), "-j".to_string()],
&declared(&["threads"]),
"run",
)
.unwrap_err();
let msg = format!("{err}");
assert!(msg.contains("command flag"), "{msg}");
assert!(msg.contains("before KEY=VALUE"), "{msg}");
}
#[test]
fn unknown_token_under_dry_run_stays_command_aware() {
let err = parse_cli_overrides(
vec!["--config".to_string(), "x".to_string()],
&declared(&[]),
"dry-run",
)
.unwrap_err();
let msg = format!("{err}");
assert!(msg.contains("unknown argument"), "{msg}");
assert!(msg.contains("dry-run"), "{msg}");
}
#[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");
}
#[test]
fn dry_run_headline_reports_run_skip_completed_not_dag_size() {
use assert_cmd::Command;
let wf_toml = r#"
[workflow]
name = "headline"
[[rules]]
name = "step"
input = []
output = ["out.txt"]
shell = "touch out.txt"
"#;
let dir = tempfile::tempdir().unwrap();
let wf = dir.path().join("wf.oxoflow");
std::fs::write(&wf, wf_toml).unwrap();
let assert = Command::cargo_bin("oxo-flow")
.unwrap()
.args(["dry-run", wf.to_str().unwrap()])
.env("NO_COLOR", "1")
.assert()
.success();
let stderr = String::from_utf8_lossy(&assert.get_output().stderr).to_string();
assert!(
stderr.contains("Plan: would run: 1 | skip: 0 | completed: 0 (DAG size: 1)"),
"headline must lead with would-run/skip/completed, got: {stderr}"
);
assert!(
!stderr.contains("rules would execute"),
"the old misleading headline must be gone, got: {stderr}"
);
}
#[test]
fn status_accepts_workdir_like_run() {
use assert_cmd::Command;
let wf_toml = r#"
[workflow]
name = "status-workdir"
[[rules]]
name = "step"
input = []
output = ["out.txt"]
shell = "touch out.txt"
"#;
let dir = tempfile::tempdir().unwrap();
let wf = dir.path().join("wf.oxoflow");
std::fs::write(&wf, wf_toml).unwrap();
let run_dir = dir.path().join("run1");
Command::cargo_bin("oxo-flow")
.unwrap()
.args([
"run",
wf.to_str().unwrap(),
"--workdir",
run_dir.to_str().unwrap(),
"-j",
"1",
])
.env("NO_COLOR", "1")
.assert()
.success();
let assert = Command::cargo_bin("oxo-flow")
.unwrap()
.args(["status", "--workdir", run_dir.to_str().unwrap(), "--json"])
.env("NO_COLOR", "1")
.assert()
.success();
let stdout = String::from_utf8_lossy(&assert.get_output().stdout).to_string();
let parsed: serde_json::Value =
serde_json::from_str(&stdout).expect("status --json must emit JSON");
assert_eq!(parsed["completed"], serde_json::json!(["step"]));
}
#[test]
fn status_reports_staleness_reasons_for_a_stale_completed_rule() {
use assert_cmd::Command;
let wf_toml = r#"
[workflow]
name = "stale-reasons"
[[rules]]
name = "producer"
input = []
output = ["stage1.txt"]
shell = "printf 'v1' > stage1.txt"
[[rules]]
name = "consumer"
input = ["stage1.txt"]
output = ["stage2.txt"]
shell = "cp stage1.txt stage2.txt"
"#;
let dir = tempfile::tempdir().unwrap();
let wf = dir.path().join("wf.oxoflow");
std::fs::write(&wf, wf_toml).unwrap();
let run_dir = dir.path().join("run1");
Command::cargo_bin("oxo-flow")
.unwrap()
.args([
"run",
wf.to_str().unwrap(),
"--workdir",
run_dir.to_str().unwrap(),
"-j",
"1",
])
.env("NO_COLOR", "1")
.assert()
.success();
std::fs::remove_file(run_dir.join("stage1.txt")).unwrap();
let assert = Command::cargo_bin("oxo-flow")
.unwrap()
.args(["status", "--workdir", run_dir.to_str().unwrap(), "--json"])
.env("NO_COLOR", "1")
.assert()
.success();
let stdout = String::from_utf8_lossy(&assert.get_output().stdout).to_string();
let parsed: serde_json::Value = serde_json::from_str(&stdout).unwrap();
let staleness = parsed["staleness"].as_object().expect("staleness map");
let producer = &staleness["producer"];
assert_eq!(producer["status"], "missing input");
let consumer = &staleness["consumer"];
assert_eq!(consumer["status"], "input changed");
assert!(
consumer["detail"].as_str().unwrap().contains("manifest"),
"input-changed detail must explain the manifest mismatch, got: {consumer}"
);
}
#[test]
fn spawn_watchdog_threshold_parsing() {
assert_eq!(
parse_spawn_watchdog_threshold(None),
std::time::Duration::from_secs(600)
);
assert_eq!(
parse_spawn_watchdog_threshold(Some("abc")),
std::time::Duration::from_secs(600)
);
assert_eq!(
parse_spawn_watchdog_threshold(Some("0")),
std::time::Duration::from_secs(600)
);
assert_eq!(
parse_spawn_watchdog_threshold(Some("120")),
std::time::Duration::from_secs(120)
);
}
fn pixi_env(manifest: &str) -> oxo_flow_core::rule::EnvironmentSpec {
oxo_flow_core::rule::EnvironmentSpec {
pixi: Some(manifest.to_string()),
..Default::default()
}
}
#[test]
fn backend_preflight_passes_empty_and_empty_specs() {
let resolver = oxo_flow_core::environment::EnvironmentResolver::new();
assert!(super::backend_preflight_failures(&[], &resolver).is_empty());
let no_env = vec![("qc".to_string(), Default::default())];
assert!(super::backend_preflight_failures(&no_env, &resolver).is_empty());
}
#[test]
fn backend_preflight_reports_missing_pixi_manifest_once_per_unique_spec() {
let resolver = oxo_flow_core::environment::EnvironmentResolver::new();
let pending = vec![
("trim_S1".to_string(), pixi_env("envs/absent/pixi.toml")),
("trim_S2".to_string(), pixi_env("envs/absent/pixi.toml")),
];
let failures = super::backend_preflight_failures(&pending, &resolver);
assert_eq!(
failures.len(),
1,
"identical specs must group: {failures:?}"
);
let f = &failures[0];
assert_eq!(f.kind, "pixi");
assert_eq!(f.count, 2);
assert_eq!(f.example_rule, "trim_S1");
assert_eq!(f.spec, "envs/absent/pixi.toml");
let rendered = f.to_string();
assert!(
rendered.contains("required by 2 pending rule(s)"),
"the message must carry the cohort size: {rendered}"
);
}
#[test]
fn backend_install_hint_splits_manifest_not_found_from_missing_binary() {
let make = |message: &str| super::BackendPreflightFailure {
kind: "pixi".to_string(),
message: message.to_string(),
example_rule: "trim_S1".to_string(),
spec: "envs/absent/pixi.toml".to_string(),
count: 2,
};
let manifest = super::backend_install_hint(&make(
"pixi manifest 'envs/absent/pixi.toml' not found — the rule declares it via `pixi = \"envs/absent/pixi.toml\"` (resolved relative to the workflow directory)",
));
assert_eq!(
manifest,
"create the manifest or correct the rule's `pixi` path (resolved relative to the workflow directory)"
);
let binary = super::backend_install_hint(&make("pixi is not installed or not in PATH"));
assert_eq!(
binary,
"install pixi (https://pixi.sh) and ensure it is on PATH"
);
assert!(!"pixi is not installed or not in PATH".contains("not found"));
}
#[test]
fn backend_preflight_missing_backend_binary_still_advises_install() {
let resolver = oxo_flow_core::environment::EnvironmentResolver::new();
let pending = vec![("trim_S1".to_string(), pixi_env("envs/absent/pixi.toml"))];
let failures = super::backend_preflight_failures(&pending, &resolver);
let rendered = failures[0].to_string();
if failures[0].message.contains("not found") {
assert!(
rendered.contains("create the manifest"),
"manifest mode must carry the file advice: {rendered}"
);
} else {
assert!(
rendered.contains("install pixi"),
"binary mode must keep the install advice: {rendered}"
);
}
}
}