start_command/
launch_identity.rs

1//! What a new launch may do with its execution id (issue #229).
2//!
3//! A launch with an explicit `--session-id` that is already tracked used to
4//! replace the record and truncate the log. It is now refused unless
5//! `--append-attempt` is given, which keeps the record, archives the previous
6//! attempt into `attemptHistory`, and appends the new attempt to the same log
7//! after a lifecycle header (the layout `--resume` already writes).
8//!
9//! Two cases keep the earlier behaviour:
10//! - the next level of a stacked isolation, which receives the same
11//!   `--session-id` from its parent and runs with START_EXECUTION_ID already
12//!   set to it (`is_nested_launch_of`);
13//! - a generated UUID, which cannot collide.
14use crate::args_parser::WrapperOptions;
15use crate::execution_attempt::{
16    archive_attempt, create_attempt, current_attempt_number, next_attempt_number, ExecutionAttempt,
17};
18use crate::execution_env::{
19    build_execution_env, inherited_attempt, is_nested_launch_of, prefix_with_execution_env,
20    ExecutionEnv, ATTEMPT_ENV, EXECUTION_ID_ENV,
21};
22use crate::execution_identity::find_record;
23use crate::execution_store::{ExecutionRecord, ExecutionStatus, ExecutionStore};
24use crate::isolation::isolation_log::{append_log_file, write_log_file};
25use crate::launch_owner::{has_active_launch, process_alive};
26use chrono::Utc;
27use serde_json::{json, Value};
28use std::fs;
29use std::path::{Path, PathBuf};
30
31const LOG_HEADER: &str = "=== Start Command Log ===";
32
33/// How a launch uses its execution id, decided before anything is printed.
34#[derive(Debug, Clone)]
35pub struct LaunchPlan {
36    /// Append to the existing log as the next attempt.
37    pub append: bool,
38    /// Stored record the launch appends an attempt to.
39    pub previous: Option<ExecutionRecord>,
40    pub attempt_number: u64,
41    pub log_path: PathBuf,
42    /// Next level of a stacked isolation of the same execution.
43    pub nested: bool,
44    /// The id was checked, so saving must re-check it under the store lock.
45    pub checked: bool,
46    /// UUID of the execution this one continues (`--continues`, #231).
47    pub continues: Option<String>,
48}
49
50impl LaunchPlan {
51    /// A launch with a generated id, or without tracking.
52    pub fn fresh(log_path: PathBuf) -> Self {
53        LaunchPlan {
54            append: false,
55            previous: None,
56            attempt_number: 1,
57            log_path,
58            nested: false,
59            checked: false,
60            continues: None,
61        }
62    }
63}
64
65/// Why saving a launch failed.
66#[derive(Debug)]
67pub enum ClaimError {
68    /// Another launch took the id: the launch must not run.
69    Refused(String),
70    /// Tracking failed; the launch may still run.
71    Store(String),
72}
73
74pub fn already_exists_message(uuid: &str) -> String {
75    format!(
76        "execution {} already exists; use $ --resume {} -- <command> to continue it, or pass --append-attempt",
77        uuid, uuid
78    )
79}
80
81/// True while the stored execution may still be running.
82pub fn is_execution_live(record: &ExecutionRecord) -> bool {
83    if has_active_launch(record) {
84        return true;
85    }
86    if record.status != ExecutionStatus::Executing {
87        return false;
88    }
89    if record
90        .options
91        .get("isolated")
92        .and_then(Value::as_str)
93        .is_some()
94    {
95        return crate::session_probe::probe_session(
96            record,
97            &crate::execution_control::SystemCommandRunner,
98        )
99        .alive;
100    }
101    record.pid.is_some_and(|pid| process_alive(u64::from(pid)))
102}
103
104/// Attempts already in a log written without tracking, from its headers.
105pub fn count_logged_attempts(log_path: &Path) -> u64 {
106    fs::read_to_string(log_path)
107        .map(|content| content.matches(LOG_HEADER).count() as u64)
108        .unwrap_or(0)
109}
110
111fn log_size(log_path: &Path) -> u64 {
112    fs::metadata(log_path).map(|m| m.len()).unwrap_or(0)
113}
114
115/// Inputs of `preflight_launch`.
116pub struct PreflightParams<'a> {
117    pub store: Option<&'a ExecutionStore>,
118    pub session_id: &'a str,
119    pub explicit_session_id: bool,
120    pub append_attempt: bool,
121    pub log_path: PathBuf,
122    /// START_EXECUTION_ID and START_ATTEMPT of this process.
123    pub env_execution_id: Option<String>,
124    pub env_attempt: Option<String>,
125}
126
127/// Decide how a launch uses its execution id; `Err` when it must not run.
128pub fn preflight_launch(params: PreflightParams<'_>) -> Result<LaunchPlan, String> {
129    let session_id = params.session_id;
130    let fresh = LaunchPlan::fresh(params.log_path.clone());
131    if !params.explicit_session_id {
132        return Ok(fresh);
133    }
134    if is_nested_launch_of(session_id, params.env_execution_id.as_deref()) {
135        return Ok(LaunchPlan {
136            nested: true,
137            attempt_number: inherited_attempt(params.env_attempt.as_deref()),
138            ..fresh
139        });
140    }
141    let existing = params.store.and_then(|store| store.get(session_id));
142    if let Some(existing) = existing {
143        if existing.uuid != session_id {
144            return Err(format!(
145                "execution id {} is already used as a name of execution {}; pick another --session-id",
146                session_id, existing.uuid
147            ));
148        }
149        if !params.append_attempt {
150            return Err(already_exists_message(session_id));
151        }
152        if is_execution_live(&existing) {
153            return Err(format!(
154                "execution {} is still running; wait for it to finish or use $ --stop {} before --append-attempt",
155                session_id, session_id
156            ));
157        }
158        let log_path = if existing.log_path.is_empty() {
159            params.log_path
160        } else {
161            PathBuf::from(&existing.log_path)
162        };
163        return Ok(LaunchPlan {
164            append: true,
165            attempt_number: next_attempt_number(&existing),
166            previous: Some(existing),
167            log_path,
168            checked: true,
169            ..fresh
170        });
171    }
172    // Untracked (or deleted) executions are recognised by their log file.
173    if params.log_path.exists() {
174        if !params.append_attempt {
175            return Err(already_exists_message(session_id));
176        }
177        return Ok(LaunchPlan {
178            append: true,
179            attempt_number: count_logged_attempts(&params.log_path) + 1,
180            checked: true,
181            ..fresh
182        });
183    }
184    Ok(LaunchPlan {
185        checked: true,
186        ..fresh
187    })
188}
189
190/// Preflight of the CLI: how this launch uses its id (#229) and which
191/// execution it continues (#231).
192pub fn prepare_launch(
193    store: Option<&ExecutionStore>,
194    options: &WrapperOptions,
195    session_id: &str,
196    log_path: PathBuf,
197) -> Result<LaunchPlan, String> {
198    let mut launch = preflight_launch(PreflightParams {
199        store,
200        session_id,
201        explicit_session_id: options.session_id.is_some(),
202        append_attempt: options.append_attempt,
203        log_path,
204        env_execution_id: std::env::var(EXECUTION_ID_ENV).ok(),
205        env_attempt: std::env::var(ATTEMPT_ENV).ok(),
206    })?;
207    if let Some(identifier) = &options.continues {
208        launch.continues = Some(crate::execution_chain::resolve_continues(
209            store, identifier, session_id,
210        )?);
211    }
212    Ok(launch)
213}
214
215fn option_str(record: &ExecutionRecord, key: &str) -> Option<String> {
216    record
217        .options
218        .get(key)
219        .and_then(Value::as_str)
220        .map(str::to_string)
221}
222
223/// Turn the record a launch built into the next attempt of `previous`,
224/// keeping the identity, history and log of the execution.
225pub fn merge_appended_attempt(record: &mut ExecutionRecord, previous: &ExecutionRecord) {
226    let mut prior = previous.clone();
227    if prior.attempt.is_none() && current_attempt_number(&prior) == 1 {
228        // A launch writes a fresh log, so its first attempt starts at offset 0.
229        prior.attempt = Some(ExecutionAttempt {
230            number: 1,
231            started_at: prior.start_time.clone(),
232            log_offset: Some(0),
233            last_output_at: None,
234            launch_accepted_at: None,
235            watcher_attached_at: None,
236            mode: String::new(),
237            previous_session_name: None,
238            session_name: None,
239            watcher_error: None,
240        });
241    }
242    let session_name = option_str(record, "sessionName");
243    let mut attempt = create_attempt(
244        &prior,
245        "append-attempt",
246        session_name.as_deref().unwrap_or(""),
247    );
248    attempt.session_name = session_name.clone();
249    archive_attempt(&mut prior);
250    let mut history: Vec<String> = prior
251        .options
252        .get("sessionNameHistory")
253        .and_then(Value::as_array)
254        .map(|items| {
255            items
256                .iter()
257                .filter_map(Value::as_str)
258                .map(str::to_string)
259                .collect()
260        })
261        .unwrap_or_default();
262    if let Some(old) = option_str(&prior, "sessionName") {
263        if Some(&old) != session_name.as_ref() {
264            history.push(old);
265        }
266    }
267    let mut unique: Vec<String> = Vec::new();
268    for name in history {
269        if !unique.contains(&name) {
270            unique.push(name);
271        }
272    }
273    record.attempt = Some(attempt);
274    record.attempt_history = prior.attempt_history.clone();
275    record.start_time = prior.start_time.clone();
276    if !prior.log_path.is_empty() {
277        record.log_path = prior.log_path.clone();
278    }
279    if record.continues.is_none() {
280        record.continues = prior.continues.clone();
281    }
282    if !unique.is_empty() {
283        record
284            .options
285            .insert("sessionNameHistory".into(), json!(unique));
286    }
287    for key in ["aliases", "name", "resumeCount"] {
288        if let Some(value) = prior.options.get(key).filter(|v| !is_empty_value(v)) {
289            record.options.insert(key.into(), value.clone());
290        }
291    }
292}
293
294fn is_empty_value(value: &Value) -> bool {
295    match value {
296        Value::Null => true,
297        Value::Bool(b) => !b,
298        Value::String(s) => s.is_empty(),
299        Value::Number(n) => n.as_f64() == Some(0.0),
300        _ => false,
301    }
302}
303
304/// Apply the preflight to the record a launch built: the continued UUID and,
305/// for `--append-attempt`, the identity and history of the stored execution.
306pub fn apply_launch(record: &mut ExecutionRecord, launch: &LaunchPlan) {
307    if let Some(continues) = &launch.continues {
308        record.continues = Some(continues.clone());
309    }
310    if let Some(previous) = &launch.previous {
311        merge_appended_attempt(record, previous);
312    } else if launch.attempt_number > 1 {
313        // A nested level or an untracked log: number the attempt all the same.
314        record.attempt = Some(ExecutionAttempt {
315            number: launch.attempt_number,
316            started_at: Utc::now().to_rfc3339(),
317            log_offset: Some(if launch.append {
318                log_size(&launch.log_path)
319            } else {
320                0
321            }),
322            last_output_at: None,
323            launch_accepted_at: None,
324            watcher_attached_at: None,
325            mode: if launch.nested {
326                "nested".into()
327            } else {
328                "append-attempt".into()
329            },
330            previous_session_name: None,
331            session_name: option_str(record, "sessionName"),
332            watcher_error: None,
333        });
334    }
335}
336
337/// Persist the launch's record, refusing if another launch claimed the id in
338/// the meantime: a new id must still be free, an appended attempt must still
339/// follow the attempt the preflight saw.
340pub fn claim_launch(
341    store: &ExecutionStore,
342    record: &ExecutionRecord,
343    launch: &LaunchPlan,
344) -> Result<(), ClaimError> {
345    if !launch.checked {
346        return store.save(record).map_err(ClaimError::Store);
347    }
348    let mut refused: Option<String> = None;
349    let result = store.with_lock(|records| {
350        let index = records.iter().position(|r| r.uuid == record.uuid);
351        if let Some(previous) = &launch.previous {
352            let attempt_of = |r: &ExecutionRecord| r.attempt.as_ref().map_or(0, |a| a.number);
353            let claimable = index.is_some_and(|i| {
354                !has_active_launch(&records[i]) && attempt_of(&records[i]) == attempt_of(previous)
355            });
356            if !claimable {
357                let message = format!("execution {} was claimed by another launch", record.uuid);
358                refused = Some(message.clone());
359                return Err(message);
360            }
361            records[index.unwrap()] = record.clone();
362        } else {
363            if find_record(records, &record.uuid).is_some() {
364                let message = already_exists_message(&record.uuid);
365                refused = Some(message.clone());
366                return Err(message);
367            }
368            records.push(record.clone());
369        }
370        Ok(Some(record.clone()))
371    });
372    match (result, refused) {
373        (Ok(_), _) => Ok(()),
374        (Err(_), Some(message)) => Err(ClaimError::Refused(message)),
375        (Err(error), None) => Err(ClaimError::Store(error)),
376    }
377}
378
379/// Header a launch writes into its log: unchanged for a fresh log; for an
380/// appended attempt, preceded by a lifecycle line in the format `--resume`
381/// writes, so the attempts stay apart.
382pub fn launch_log_content(
383    content: &str,
384    launch: &LaunchPlan,
385    record: Option<&ExecutionRecord>,
386) -> String {
387    if !launch.append {
388        return content.to_string();
389    }
390    let mut lifecycle = json!({
391        "event": "attempt-appended",
392        "at": Utc::now().to_rfc3339(),
393        "uuid": record.map(|r| r.uuid.clone()),
394        "attemptNumber": launch.attempt_number,
395    });
396    if let Some(Value::Object(attempt)) = record
397        .and_then(|r| r.attempt.as_ref())
398        .and_then(|a| serde_json::to_value(a).ok())
399    {
400        if let Value::Object(map) = &mut lifecycle {
401            map.extend(attempt);
402        }
403    }
404    format!("\n[Start Command Lifecycle] {}\n{}", lifecycle, content)
405}
406
407/// Write the launch header: a fresh log, or, for an appended attempt, a
408/// lifecycle line and the header after the earlier attempts.
409pub fn write_launch_log(
410    log_path: &PathBuf,
411    content: &str,
412    launch: &LaunchPlan,
413    record: Option<&ExecutionRecord>,
414) -> bool {
415    let text = launch_log_content(content, launch, record);
416    if launch.append {
417        append_log_file(log_path, &text)
418    } else {
419        write_log_file(log_path, &text)
420    }
421}
422
423/// START_* variables of a run without a backend session (direct, or
424/// `--isolated-user` alone): the session name is empty.
425pub fn direct_execution_env(session_id: &str, launch: &LaunchPlan) -> ExecutionEnv {
426    build_execution_env(Some(session_id), Some(""), launch.attempt_number)
427}
428
429/// Command of an `--isolated-user` run without a backend: `sudo` resets the
430/// environment, so the variables are exported by the command itself.
431pub fn isolated_user_command(command: &str, session_id: &str, launch: &LaunchPlan) -> String {
432    prefix_with_execution_env(command, &direct_execution_env(session_id, launch))
433}
start_command/launch_identity.rs

start_command/
launch_identity.rs
