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
28pub fn create_attempt(
29    record: &ExecutionRecord,
30    mode: &str,
31    session_name: &str,
32) -> ExecutionAttempt {
33    let counter = |key| record.options.get(key).and_then(Value::as_u64).unwrap_or(0);
34    let previous = record
35        .attempt
36        .as_ref()
37        .map(|a| a.number)
38        .unwrap_or(counter("resumeCount") + counter("recoveryAttempts") + 1);
39    let log_offset = if record.log_path.is_empty() {
40        None
41    } else {
42        match fs::metadata(&record.log_path) {
43            Ok(metadata) => Some(metadata.len()),
44            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Some(0),
45            Err(_) => None,
46        }
47    };
48    ExecutionAttempt {
49        number: previous + 1,
50        started_at: Utc::now().to_rfc3339(),
51        log_offset,
52        last_output_at: None,
53        launch_accepted_at: None,
54        watcher_attached_at: None,
55        mode: mode.into(),
56        previous_session_name: record
57            .options
58            .get("sessionName")
59            .and_then(Value::as_str)
60            .map(str::to_string),
61        session_name: Some(session_name.into()),
62        watcher_error: None,
63    }
64}
65
66pub fn archive_attempt(record: &mut ExecutionRecord) {
67    read_attempt_activity(record);
68    let value = serde_json::to_value(&*record).unwrap_or_default();
69    let mut snapshot = record.attempt.as_ref().and_then(|a| serde_json::to_value(a).ok())
70        .unwrap_or_else(|| json!({ "number": record.options.get("resumeCount").and_then(Value::as_u64).unwrap_or(0)
71            + record.options.get("recoveryAttempts").and_then(Value::as_u64).unwrap_or(0) + 1,
72            "startedAt": record.options.get("resumedAt").cloned().unwrap_or(json!(record.start_time)), "logOffset": null }));
73    for key in [
74        "command",
75        "status",
76        "exitCode",
77        "endTime",
78        "endTimeSource",
79        "observedAt",
80        "staleDetectedAt",
81        "containerStartedAt",
82        "containerState",
83        "containerRemovedObservedAt",
84        "exitReason",
85        "oomKilled",
86        "memoryExhausted",
87        "memoryExhaustedReason",
88        "cgroupMemory",
89    ] {
90        if let Some(field) = value.get(key) {
91            snapshot[key] = field.clone();
92        }
93    }
94    for key in ["sessionName", "containerId", "containerError"] {
95        snapshot[key] = record.options.get(key).cloned().unwrap_or(Value::Null);
96    }
97    if let Some(evidence) = record.options.get("exitEvidence") {
98        snapshot["exitEvidence"] = evidence.clone();
99    }
100    record.attempt_history.push(snapshot);
101    record.end_time_source = None;
102    record.observed_at = None;
103    record.stale_detected_at = None;
104    record.container_started_at = None;
105    record.container_state = None;
106    record.container_removed_observed_at = None;
107    record.exit_reason = None;
108    record.oom_killed = None;
109    record.memory_exhausted = None;
110    record.memory_exhausted_reason = None;
111    record.cgroup_memory = None;
112    record.options.remove("containerError");
113}
114
115pub fn append_lifecycle(record: &ExecutionRecord, event: &str, details: Value) {
116    if let Some(attempt) = &record.attempt {
117        if record.log_path.is_empty() {
118            return;
119        }
120        let mut entry = serde_json::to_value(attempt).unwrap_or_default();
121        entry["event"] = json!(event);
122        entry["at"] = json!(Utc::now().to_rfc3339());
123        entry["uuid"] = json!(record.uuid);
124        entry["attemptNumber"] = json!(attempt.number);
125        if let Some(details) = details.as_object() {
126            for (key, value) in details {
127                entry[key] = value.clone();
128            }
129        }
130        append_log_file(
131            &PathBuf::from(&record.log_path),
132            &format!("\n[Start Command Lifecycle] {}\n", entry),
133        );
134    }
135}
136
137pub fn read_attempt_log_tail(record: &ExecutionRecord, bytes: u64) -> Option<String> {
138    let offset = match &record.attempt {
139        Some(attempt) => attempt.log_offset?,
140        None => 0,
141    };
142    read_log_tail_from(&record.log_path, bytes, offset)
143}
144
145pub fn activity_path(log_path: &str, number: u64) -> String {
146    format!("{}.attempt-{}.activity", log_path, number)
147}
148
149pub fn read_attempt_activity(record: &mut ExecutionRecord) {
150    let Some(attempt) = record.attempt.as_mut() else {
151        return;
152    };
153    if record.log_path.is_empty() {
154        return;
155    }
156    let Ok(file) = fs::File::open(activity_path(&record.log_path, attempt.number)) else {
157        return;
158    };
159    let mut text = String::new();
160    if file.take(128).read_to_string(&mut text).is_err() {
161        return;
162    }
163    let time = DateTime::parse_from_rfc3339(text.trim());
164    let since = DateTime::parse_from_rfc3339(&attempt.started_at);
165    if let (Ok(time), Ok(since)) = (time, since) {
166        if time >= since {
167            attempt.last_output_at = Some(text.trim().into());
168        }
169    }
170}
start_command/execution_attempt.rs

start_command/
execution_attempt.rs
