start_command/
execution_attempt.rs

1//! Attempt boundaries and evidence for executions sharing one UUID/log.
2use chrono::{DateTime, Utc};
3use serde::{Deserialize, Serialize};
4use serde_json::{json, Value};
5use std::fs;
6use std::io::Read;
7use std::path::PathBuf;
8
9use crate::execution_store::ExecutionRecord;
10use crate::isolation::isolation_log::{append_log_file, read_log_tail_from};
11
12#[derive(Debug, Clone, Serialize, Deserialize)]
13#[serde(rename_all = "camelCase")]
14pub struct ExecutionAttempt {
15    pub number: u64,
16    pub started_at: String,
17    pub log_offset: Option<u64>,
18    pub last_output_at: Option<String>,
19    pub launch_accepted_at: Option<String>,
20    pub watcher_attached_at: Option<String>,
21    pub mode: String,
22    pub previous_session_name: Option<String>,
23    pub session_name: Option<String>,
24    #[serde(default, skip_serializing_if = "Option::is_none")]
25    pub watcher_error: Option<String>,
26}
27
28/// Number of the record's current attempt: 1 until it is resumed.
29pub fn current_attempt_number(record: &ExecutionRecord) -> u64 {
30    let counter = |key| record.options.get(key).and_then(Value::as_u64).unwrap_or(0);
31    record
32        .attempt
33        .as_ref()
34        .map(|a| a.number)
35        .filter(|n| *n > 0)
36        .unwrap_or(counter("resumeCount") + counter("recoveryAttempts") + 1)
37}
38
39/// Number the next attempt of `record` gets (`START_ATTEMPT`, issue #230).
40pub fn next_attempt_number(record: &ExecutionRecord) -> u64 {
41    current_attempt_number(record) + 1
42}
43
44pub fn create_attempt(
45    record: &ExecutionRecord,
46    mode: &str,
47    session_name: &str,
48) -> ExecutionAttempt {
49    let log_offset = if record.log_path.is_empty() {
50        None
51    } else {
52        match fs::metadata(&record.log_path) {
53            Ok(metadata) => Some(metadata.len()),
54            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Some(0),
55            Err(_) => None,
56        }
57    };
58    ExecutionAttempt {
59        number: next_attempt_number(record),
60        started_at: Utc::now().to_rfc3339(),
61        log_offset,
62        last_output_at: None,
63        launch_accepted_at: None,
64        watcher_attached_at: None,
65        mode: mode.into(),
66        previous_session_name: record
67            .options
68            .get("sessionName")
69            .and_then(Value::as_str)
70            .map(str::to_string),
71        session_name: Some(session_name.into()),
72        watcher_error: None,
73    }
74}
75
76pub fn archive_attempt(record: &mut ExecutionRecord) {
77    read_attempt_activity(record);
78    let value = serde_json::to_value(&*record).unwrap_or_default();
79    let mut snapshot = record.attempt.as_ref().and_then(|a| serde_json::to_value(a).ok())
80        .unwrap_or_else(|| json!({ "number": current_attempt_number(record),
81            "startedAt": record.options.get("resumedAt").cloned().unwrap_or(json!(record.start_time)), "logOffset": null }));
82    for key in [
83        "command",
84        "status",
85        "exitCode",
86        "endTime",
87        "endTimeSource",
88        "observedAt",
89        "staleDetectedAt",
90        "containerStartedAt",
91        "containerState",
92        "containerRemovedObservedAt",
93        "exitReason",
94        "oomKilled",
95        "memoryExhausted",
96        "memoryExhaustedReason",
97        "cgroupMemory",
98    ] {
99        if let Some(field) = value.get(key) {
100            snapshot[key] = field.clone();
101        }
102    }
103    for key in ["sessionName", "containerId", "containerError"] {
104        snapshot[key] = record.options.get(key).cloned().unwrap_or(Value::Null);
105    }
106    if let Some(evidence) = record.options.get("exitEvidence") {
107        snapshot["exitEvidence"] = evidence.clone();
108    }
109    record.attempt_history.push(snapshot);
110    record.end_time_source = None;
111    record.observed_at = None;
112    record.stale_detected_at = None;
113    record.container_started_at = None;
114    record.container_state = None;
115    record.container_removed_observed_at = None;
116    record.exit_reason = None;
117    record.oom_killed = None;
118    record.memory_exhausted = None;
119    record.memory_exhausted_reason = None;
120    record.cgroup_memory = None;
121    record.options.remove("containerError");
122}
123
124pub fn append_lifecycle(record: &ExecutionRecord, event: &str, details: Value) {
125    if let Some(attempt) = &record.attempt {
126        if record.log_path.is_empty() {
127            return;
128        }
129        let mut entry = serde_json::to_value(attempt).unwrap_or_default();
130        entry["event"] = json!(event);
131        entry["at"] = json!(Utc::now().to_rfc3339());
132        entry["uuid"] = json!(record.uuid);
133        entry["attemptNumber"] = json!(attempt.number);
134        if let Some(details) = details.as_object() {
135            for (key, value) in details {
136                entry[key] = value.clone();
137            }
138        }
139        append_log_file(
140            &PathBuf::from(&record.log_path),
141            &format!("\n[Start Command Lifecycle] {}\n", entry),
142        );
143    }
144}
145
146pub fn read_attempt_log_tail(record: &ExecutionRecord, bytes: u64) -> Option<String> {
147    let offset = match &record.attempt {
148        Some(attempt) => attempt.log_offset?,
149        None => 0,
150    };
151    read_log_tail_from(&record.log_path, bytes, offset)
152}
153
154pub fn activity_path(log_path: &str, number: u64) -> String {
155    format!("{}.attempt-{}.activity", log_path, number)
156}
157
158pub fn read_attempt_activity(record: &mut ExecutionRecord) {
159    let Some(attempt) = record.attempt.as_mut() else {
160        return;
161    };
162    if record.log_path.is_empty() {
163        return;
164    }
165    let Ok(file) = fs::File::open(activity_path(&record.log_path, attempt.number)) else {
166        return;
167    };
168    let mut text = String::new();
169    if file.take(128).read_to_string(&mut text).is_err() {
170        return;
171    }
172    let time = DateTime::parse_from_rfc3339(text.trim());
173    let since = DateTime::parse_from_rfc3339(&attempt.started_at);
174    if let (Ok(time), Ok(since)) = (time, since) {
175        if time >= since {
176            attempt.last_output_at = Some(text.trim().into());
177        }
178    }
179}
start_command/execution_attempt.rs

start_command/
execution_attempt.rs
