start_command/
execution_chain.rs

1//! Continuation chains (issue #231): a launch with `--continues <id>` stores
2//! the UUID it continues, so a supervisor that relaunches fresh instead of
3//! `--resume`-ing in place can still treat every execution of one task as a
4//! whole. `--status` prints the chain and `--upload-log` uploads one file with
5//! every execution of it, oldest first.
6use crate::execution_identity::{execution_aliases, has_user_names};
7use crate::execution_store::{ExecutionRecord, ExecutionStore};
8use serde_json::{json, Value};
9use std::fs;
10use std::io::Write;
11use std::path::PathBuf;
12
13fn start_millis(record: &ExecutionRecord) -> i64 {
14    chrono::DateTime::parse_from_rfc3339(&record.start_time)
15        .map(|t| t.timestamp_millis())
16        .unwrap_or(i64::MIN)
17}
18
19/// Records of the chain `record` belongs to, oldest first. Backward links
20/// come from `continues`; forward, the most recently started record that
21/// continues the current one is followed. Cycles stop the walk.
22pub fn build_chain(records: &[ExecutionRecord], record: &ExecutionRecord) -> Vec<ExecutionRecord> {
23    let mut seen = vec![record.uuid.clone()];
24    let mut chain = vec![record.clone()];
25    let mut current = record.clone();
26    while let Some(previous_id) = current.continues.clone() {
27        let Some(previous) = records.iter().find(|r| r.uuid == previous_id) else {
28            break;
29        };
30        if seen.contains(&previous.uuid) {
31            break;
32        }
33        seen.push(previous.uuid.clone());
34        chain.insert(0, previous.clone());
35        current = previous.clone();
36    }
37    let mut current = record.clone();
38    loop {
39        let next = records
40            .iter()
41            .filter(|r| r.continues.as_deref() == Some(&current.uuid) && !seen.contains(&r.uuid))
42            .max_by_key(|r| start_millis(r));
43        let Some(next) = next else {
44            break;
45        };
46        seen.push(next.uuid.clone());
47        chain.push(next.clone());
48        current = next.clone();
49    }
50    chain
51}
52
53/// Resolve `--continues <id>` for a new launch to the UUID it stores.
54pub fn resolve_continues(
55    store: Option<&ExecutionStore>,
56    identifier: &str,
57    session_id: &str,
58) -> Result<String, String> {
59    let store = store.ok_or("--continues requires execution tracking.")?;
60    let target = store.get(identifier).ok_or_else(|| {
61        format!(
62            "No execution found with UUID or session name: {}",
63            identifier
64        )
65    })?;
66    if target.uuid == session_id {
67        return Err(format!("Execution {} cannot continue itself", session_id));
68    }
69    Ok(target.uuid)
70}
71
72/// Fields `--status` reports right after `uuid`: the names the execution
73/// answers to (`aliases`, once `--alias`/`--rename` gave it one) and its
74/// continuation `chain`, when it has more than one member.
75pub fn status_identity(
76    record: &ExecutionRecord,
77    store: Option<&ExecutionStore>,
78) -> Vec<(String, Value)> {
79    let mut fields = Vec::new();
80    if has_user_names(record) {
81        fields.push(("aliases".to_string(), json!(execution_aliases(record))));
82    }
83    if let Some(store) = store {
84        let records = store.get_all();
85        if record.continues.is_some()
86            || records
87                .iter()
88                .any(|r| r.continues.as_deref() == Some(&record.uuid))
89        {
90            let chain: Vec<String> = build_chain(&records, record)
91                .into_iter()
92                .map(|r| r.uuid)
93                .collect();
94            if chain.len() > 1 {
95                fields.push(("chain".to_string(), json!(chain)));
96            }
97        }
98    }
99    fields
100}
101
102fn joined(fields: &[(String, Value)], key: &str, separator: &str) -> Option<String> {
103    let (_, value) = fields.iter().find(|(k, _)| k == key)?;
104    let items: Vec<&str> = value.as_array()?.iter().filter_map(Value::as_str).collect();
105    Some(items.join(separator))
106}
107
108/// Text-format lines for the identity fields of a status record.
109pub fn identity_text_lines(record: &ExecutionRecord, fields: &[(String, Value)]) -> Vec<String> {
110    let mut lines = Vec::new();
111    if let Some(aliases) = joined(fields, "aliases", ", ") {
112        lines.push(format!("Aliases:           {}", aliases));
113    }
114    if let Some(continues) = &record.continues {
115        lines.push(format!("Continues:         {}", continues));
116    }
117    if let Some(chain) = joined(fields, "chain", " -> ") {
118        lines.push(format!("Chain:             {}", chain));
119    }
120    lines
121}
122
123/// A temporary combined log, removed when dropped.
124pub struct ChainLog {
125    dir: PathBuf,
126    pub path: PathBuf,
127}
128
129impl Drop for ChainLog {
130    fn drop(&mut self) {
131        let _ = fs::remove_dir_all(&self.dir);
132    }
133}
134
135/// Concatenate the logs of a chain into one temporary file, oldest first,
136/// each preceded by a lifecycle header in the format `--resume` writes.
137pub fn write_chain_log(chain: &[ExecutionRecord]) -> std::io::Result<ChainLog> {
138    let dir = std::env::temp_dir().join(format!("start-chain-log-{}", uuid::Uuid::new_v4()));
139    fs::create_dir(&dir)?;
140    let last = chain.last().map(|r| r.uuid.as_str()).unwrap_or("chain");
141    let log = ChainLog {
142        path: dir.join(format!("{}.chain.log", last)),
143        dir,
144    };
145    let mut file = fs::File::create(&log.path)?;
146    for (index, record) in chain.iter().enumerate() {
147        let log_path = (!record.log_path.is_empty()).then_some(&record.log_path);
148        let header = json!({
149            "event": "continuation",
150            "uuid": record.uuid,
151            "chainIndex": index + 1,
152            "chainLength": chain.len(),
153            "continues": record.continues,
154            "command": record.command,
155            "startTime": record.start_time,
156            "logPath": log_path,
157        });
158        write!(file, "\n[Start Command Lifecycle] {}\n", header)?;
159        match log_path.map(fs::read) {
160            Some(Ok(content)) => file.write_all(&content)?,
161            _ => writeln!(file, "Log file not found: {}", record.log_path)?,
162        }
163    }
164    Ok(log)
165}
166
167/// Combined log `--upload-log` uploads for `record` when it belongs to a
168/// chain of more than one execution; `None` for a single execution.
169pub fn prepare_chain_upload(
170    store: &ExecutionStore,
171    record: &ExecutionRecord,
172) -> Result<Option<ChainLog>, String> {
173    let chain = build_chain(&store.get_all(), record);
174    if chain.len() < 2 {
175        return Ok(None);
176    }
177    write_chain_log(&chain)
178        .map(Some)
179        .map_err(|e| format!("Could not join chain logs: {}", e))
180}
start_command/execution_chain.rs

start_command/
execution_chain.rs
