1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
//! `resume` command execution: loads checkpoint state and re-enters the run loop.
use super::*;
use crate::commands::print_banner;
use anyhow::{Context, Result};
use colored::Colorize;
use oxo_flow_core::config::WorkflowConfig;
use oxo_flow_core::executor::checkpoint::CheckpointState;
use std::path::PathBuf;
#[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<()> {
// resume does not produce structured JSON output
print_banner();
// Load checkpoint state
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| {
// Same friendly hint `status` gives for the identical mistake
// (audit #276 P4-4): passing a workflow file where a checkpoint
// belongs should name the confusion, not just the JSON parse error.
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
}
})?;
// Get workflow path from checkpoint
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();
// Issue #690: don't count stale when-gated entries as failures in the
// resume banner either — they are skips, and the executor will re-judge
// their gate on this resume anyway.
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());
// Remaining = the rules this resume will actually execute: everything
// the checkpoint has not completed (failed rules are retried). The
// checkpoint records only what ran, so the workflow supplies the total
// (its template count — wildcards expand later); if it cannot be read
// here, `run_command` reports that error with full context a moment
// later and the failed count is the fallback. The old arithmetic
// reduced to `failed` (audit finding).
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: {} {} skipped by their `when` condition (recorded false); the gate is re-judged on this resume.",
skipped_when_false.len(),
if skipped_when_false.len() == 1 {
"rule was"
} else {
"rules were"
}
);
}
// Issue #685: entries persisting in the running set mean the prior run
// died mid-flight; run_command clears them (with a warning) on start.
if !state.running_rules().is_empty() {
eprintln!(
" Note: {} {} still running when the previous run died; they will re-run.",
state.running_rules().len(),
if state.running_rules().len() == 1 {
"rule was"
} else {
"rules were"
}
);
}
if completed == 0 && failed == 0 {
eprintln!(
" {} No rules have been executed yet. Use 'oxo-flow run' instead.",
"Note:".yellow()
);
return Ok(());
}
// Re-run the workflow — the checkpoint with completed rules will cause
// already-finished rules to be skipped automatically
eprintln!();
eprintln!(
"{} Launching executor with {}...",
"Info:".bold().cyan(),
oxo_flow_core::plural(jobs, "parallel job", "parallel jobs")
);
// Workdir precedence (issue #68): explicit --workdir > the directory the
// original run recorded in the checkpoint > the workflow's directory.
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(RunRequest {
workflow: Some(workflow_path),
jobs,
keep_going,
// recorded by the original run
workdir: effective_workdir,
log_file: None,
target: Vec::new(),
module: Vec::new(),
retry: 0,
timeout,
// user can re-run with --resume-failed in 'run'
resume_failed: false,
profile: None,
max_threads: 0,
max_memory: 0,
skip_env_setup: false,
// resume: refs already built
skip_ref_build: true,
cache_dir: None,
// checkpoint already has checksums
provenance: false,
// resume defaults to human-readable
json: false,
// resume reuses checkpoint state
cli_args: Vec::new(),
ai_recover,
ai_max_retries,
// resume restores checkpoint state as-is
samples_filter: Vec::new(),
// resume skips completed rules by design
rerun: false,
no_report_snapshot,
// cluster queue cap — resume keeps the profile's
max_submitted: None,
// #538: the user's checkpoint path wins over the workdir-derived
// default — the banner's "Resuming… N completed" must be the truth.
checkpoint_override: Some(checkpoint),
// resume honors the protection gate
allow_protected_drift: false,
// resume never re-derives a from-rule closure: it continues the
// recorded checkpoint's pending set exactly as-is.
from_rule: None,
// resume re-solves degraded environments like a fresh run would
strict_envs: false,
})
.await
}