start_command/
execution_resume_all.rs

1//! `--resume-all`: re-attach or reconcile every execution still marked running.
2//!
3//! When the supervisor host restarts, the detached docker completion watcher
4//! (`docker logs -f ... ; docker inspect ... ; footer`) dies with it. The
5//! container keeps running but nothing streams its output into the session log
6//! any more, and nothing will ever write the exit footer — the record stays
7//! `executing` forever. `--resume-all` repairs that state:
8//!
9//! - reattached: a live docker container gets a fresh completion watcher.
10//! - running: a live screen/tmux session needs nothing (its logging is
11//!   in-session), it is only reported.
12//! - reconciled: the session is gone, so the record is finalized from the same
13//!   evidence `--status` uses (docker exit code, log footer).
14//! - unknown: the backend cannot be probed locally (ssh), left untouched.
15//!
16//! Commands are never silently restarted here: resuming actual work is an
17//! explicit, per-session decision made with `--resume`.
18
19use serde_json::json;
20
21use crate::execution_control::{CommandRunner, SystemCommandRunner};
22use crate::execution_resume::{ResumeHooks, SystemResumeHooks};
23use crate::execution_store::{ExecutionRecord, ExecutionStatus, ExecutionStore};
24use crate::output_blocks::escape_for_links_notation;
25use crate::session_probe::{probe_session, SessionProbe, SessionState};
26
27/// Outcomes reported for each execution that was still marked running.
28#[derive(Debug, Clone, Copy, PartialEq, Eq)]
29pub enum ResumeAllAction {
30    Reattached,
31    Running,
32    Reconciled,
33    Unknown,
34}
35
36impl ResumeAllAction {
37    pub fn as_str(self) -> &'static str {
38        match self {
39            ResumeAllAction::Reattached => "reattached",
40            ResumeAllAction::Running => "running",
41            ResumeAllAction::Reconciled => "reconciled",
42            ResumeAllAction::Unknown => "unknown",
43        }
44    }
45}
46
47/// One line of the `--resume-all` report.
48#[derive(Debug, Clone, PartialEq, Eq)]
49pub struct ResumeAllEntry {
50    pub uuid: String,
51    pub backend: Option<String>,
52    pub session_name: Option<String>,
53    pub state: SessionState,
54    pub action: ResumeAllAction,
55    pub exit_code: Option<i32>,
56    pub message: String,
57}
58
59/// Result of `resume_all_executions`.
60pub struct ExecutionResumeAllResult {
61    pub success: bool,
62    pub output: Option<String>,
63    pub error: Option<String>,
64}
65
66fn describe_record(
67    record: &ExecutionRecord,
68    probe: &SessionProbe,
69    action: ResumeAllAction,
70    message: String,
71    exit_code: Option<i32>,
72) -> ResumeAllEntry {
73    ResumeAllEntry {
74        uuid: record.uuid.clone(),
75        backend: probe.backend.clone().or_else(|| {
76            record
77                .options
78                .get("isolated")
79                .and_then(|value| value.as_str())
80                .map(str::to_string)
81        }),
82        session_name: probe.session_name.clone().or_else(|| {
83            record
84                .options
85                .get("sessionName")
86                .and_then(|value| value.as_str())
87                .map(str::to_string)
88        }),
89        state: probe.state,
90        action,
91        exit_code,
92        message,
93    }
94}
95
96fn format_as_links_notation(entries: &[ResumeAllEntry]) -> String {
97    let mut lines = vec![
98        "executionResumeAll".to_string(),
99        format!("  count {}", entries.len()),
100    ];
101    for entry in entries {
102        lines.push("  execution".to_string());
103        lines.push(format!(
104            "    uuid {}",
105            escape_for_links_notation(&entry.uuid)
106        ));
107        lines.push(format!(
108            "    backend {}",
109            escape_for_links_notation(entry.backend.as_deref().unwrap_or(""))
110        ));
111        lines.push(format!(
112            "    sessionName {}",
113            escape_for_links_notation(entry.session_name.as_deref().unwrap_or(""))
114        ));
115        lines.push(format!(
116            "    state {}",
117            escape_for_links_notation(entry.state.as_str())
118        ));
119        lines.push(format!(
120            "    action {}",
121            escape_for_links_notation(entry.action.as_str())
122        ));
123        if let Some(exit_code) = entry.exit_code {
124            lines.push(format!("    exitCode {}", exit_code));
125        }
126        lines.push(format!(
127            "    message {}",
128            escape_for_links_notation(&entry.message)
129        ));
130    }
131    lines.join("\n")
132}
133
134fn format_as_text(entries: &[ResumeAllEntry]) -> String {
135    if entries.is_empty() {
136        return "No executions are currently marked as running.".to_string();
137    }
138    let mut lines = vec![
139        format!("Executions still marked running: {}", entries.len()),
140        String::new(),
141    ];
142    for entry in entries {
143        lines.push(format!(
144            "{}  {}",
145            entry.action.as_str().to_uppercase(),
146            entry.uuid
147        ));
148        lines.push(format!(
149            "  Backend:      {}",
150            entry.backend.as_deref().unwrap_or("")
151        ));
152        lines.push(format!(
153            "  Session Name: {}",
154            entry.session_name.as_deref().unwrap_or("")
155        ));
156        lines.push(format!("  State:        {}", entry.state.as_str()));
157        lines.push(format!("  {}", entry.message));
158        lines.push(String::new());
159    }
160    lines.join("\n").trim_end().to_string()
161}
162
163fn format_as_json(entries: &[ResumeAllEntry]) -> String {
164    let executions: Vec<_> = entries
165        .iter()
166        .map(|entry| {
167            json!({
168                "uuid": entry.uuid,
169                "backend": entry.backend,
170                "sessionName": entry.session_name,
171                "state": entry.state.as_str(),
172                "action": entry.action.as_str(),
173                "exitCode": entry.exit_code,
174                "message": entry.message,
175            })
176        })
177        .collect();
178    serde_json::to_string_pretty(&json!({
179        "count": entries.len(),
180        "executions": executions,
181    }))
182    .unwrap_or_else(|_| "{}".to_string())
183}
184
185/// Format a `--resume-all` report in the requested output format.
186pub fn format_resume_all(entries: &[ResumeAllEntry], output_format: Option<&str>) -> String {
187    match output_format {
188        Some("json") => format_as_json(entries),
189        Some("text") => format_as_text(entries),
190        _ => format_as_links_notation(entries),
191    }
192}
193
194/// Re-attach or reconcile every execution still marked running.
195pub fn resume_all_executions(
196    store: Option<&ExecutionStore>,
197    output_format: Option<&str>,
198) -> ExecutionResumeAllResult {
199    resume_all_executions_with(
200        store,
201        output_format,
202        &SystemCommandRunner,
203        &SystemResumeHooks,
204    )
205}
206
207/// Re-attach or reconcile with an injectable command runner and hooks.
208pub fn resume_all_executions_with<R: CommandRunner, H: ResumeHooks>(
209    store: Option<&ExecutionStore>,
210    output_format: Option<&str>,
211    runner: &R,
212    hooks: &H,
213) -> ExecutionResumeAllResult {
214    let Some(store) = store else {
215        return ExecutionResumeAllResult {
216            success: false,
217            output: None,
218            error: Some("Execution tracking is disabled.".to_string()),
219        };
220    };
221
222    let mut entries: Vec<ResumeAllEntry> = Vec::new();
223
224    for record in store.get_executing() {
225        let probe = probe_session(&record, runner);
226        if crate::launch_owner::has_active_launch(&record) {
227            entries.push(describe_record(
228                &record,
229                &probe,
230                ResumeAllAction::Running,
231                "Launch is reserved; no watcher or duplicate launch was started.".into(),
232                None,
233            ));
234            continue;
235        }
236        let session_name = probe.session_name.clone().unwrap_or_default();
237
238        if probe.alive && probe.backend.as_deref() == Some("docker") {
239            hooks.start_watcher(&session_name, &record);
240            entries.push(describe_record(
241                &record,
242                &probe,
243                ResumeAllAction::Reattached,
244                format!(
245                    "Re-attached completion watcher to running container: {}",
246                    session_name
247                ),
248                None,
249            ));
250            continue;
251        }
252
253        if probe.alive {
254            entries.push(describe_record(
255                &record,
256                &probe,
257                ResumeAllAction::Running,
258                format!("Session is still running: {}", session_name),
259                None,
260            ));
261            continue;
262        }
263
264        if probe.state == SessionState::Unknown {
265            entries.push(describe_record(
266                &record,
267                &probe,
268                ResumeAllAction::Unknown,
269                format!(
270                    "Session liveness cannot be probed locally: {}",
271                    if session_name.is_empty() {
272                        record.uuid.clone()
273                    } else {
274                        session_name.clone()
275                    }
276                ),
277                None,
278            ));
279            continue;
280        }
281
282        let reconciled = hooks.reconcile(&record);
283        if reconciled.status == ExecutionStatus::Executed {
284            let exit_code = reconciled.exit_code;
285            let _ = store.save(&reconciled);
286            entries.push(describe_record(
287                &record,
288                &probe,
289                ResumeAllAction::Reconciled,
290                format!(
291                    "Session ended while unsupervised; record finalized. Use `$ --resume {}` to continue it.",
292                    record.uuid
293                ),
294                exit_code,
295            ));
296            continue;
297        }
298
299        entries.push(describe_record(
300            &record,
301            &probe,
302            ResumeAllAction::Unknown,
303            format!(
304                "Session is not running but no terminal result could be resolved: {}",
305                if session_name.is_empty() {
306                    record.uuid.clone()
307                } else {
308                    session_name.clone()
309                }
310            ),
311            None,
312        ));
313    }
314
315    ExecutionResumeAllResult {
316        success: true,
317        output: Some(format_resume_all(&entries, output_format)),
318        error: None,
319    }
320}
321
322#[cfg(test)]
323#[path = "execution_resume_all_cases.rs"]
324mod tests;
start_command/execution_resume_all.rs

start_command/
execution_resume_all.rs
