Skip to main content

start_command/
execution_store.rs

1//! Execution Store - Dual storage (.lino text + .links binary) for command execution records
2
3use crate::lino_value_json::{json_to_lino_value, lino_value_to_json};
4use chrono::Utc;
5use lino_objects_codec::{decode, encode, LinoValue};
6use serde::{Deserialize, Serialize};
7use serde_json::Value;
8use std::collections::HashMap;
9use std::env;
10use std::fs;
11use std::path::{Path, PathBuf};
12use std::process::{Command, Stdio};
13use uuid::Uuid;
14
15/// Default application folder name
16const DEFAULT_APP_FOLDER_NAME: &str = ".start-command";
17/// Lino database file name
18const LINO_DB_FILE: &str = "executions.lino";
19/// Links database file name
20const LINKS_DB_FILE: &str = "executions.links";
21/// Lock file name
22const LOCK_FILE: &str = "executions.lock";
23/// Lock timeout in milliseconds
24const LOCK_TIMEOUT_MS: u64 = 30000;
25
26/// Execution status
27#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
28#[serde(rename_all = "lowercase")]
29pub enum ExecutionStatus {
30    Executing,
31    Executed,
32}
33
34impl ExecutionStatus {
35    pub fn as_str(&self) -> &'static str {
36        match self {
37            ExecutionStatus::Executing => "executing",
38            ExecutionStatus::Executed => "executed",
39        }
40    }
41}
42
43impl std::fmt::Display for ExecutionStatus {
44    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
45        write!(f, "{}", self.as_str())
46    }
47}
48
49/// Command Execution Record
50#[derive(Debug, Clone, Serialize, Deserialize)]
51#[serde(rename_all = "camelCase")]
52pub struct ExecutionRecord {
53    pub uuid: String,
54    pub pid: Option<u32>,
55    pub status: ExecutionStatus,
56    pub exit_code: Option<i32>,
57    pub command: String,
58    pub log_path: String,
59    pub start_time: String,
60    pub end_time: Option<String>,
61    #[serde(default, skip_serializing_if = "Option::is_none")]
62    pub attempt: Option<crate::execution_attempt::ExecutionAttempt>,
63    #[serde(default, skip_serializing_if = "Vec::is_empty")]
64    pub attempt_history: Vec<Value>,
65    /// Where `end_time` came from (issue #170.2): `docker-finished-at`,
66    /// `log-footer` or `observed-at`. Without it a real finish time and a
67    /// timestamp taken when someone happened to run `--status` are
68    /// indistinguishable.
69    #[serde(default, skip_serializing_if = "Option::is_none")]
70    pub end_time_source: Option<String>,
71    /// When the end of the execution was *observed*, for the cases where no
72    /// real finish time exists (issue #170.2).
73    #[serde(default, skip_serializing_if = "Option::is_none")]
74    pub observed_at: Option<String>,
75    /// When `cleanup_stale()` gave up on a record. Deliberately not an
76    /// `end_time`: cleanup time is not a finish time (issue #170.3).
77    #[serde(default, skip_serializing_if = "Option::is_none")]
78    pub stale_detected_at: Option<String>,
79    /// `docker inspect .State.StartedAt` of the container that ran the command.
80    #[serde(default, skip_serializing_if = "Option::is_none")]
81    pub container_started_at: Option<String>,
82    #[serde(default, skip_serializing_if = "Option::is_none")]
83    pub oom_killed: Option<bool>,
84    /// cgroup v2 memory counters the detached watcher sampled while the
85    /// container ran (issue #182). Unlike the sticky `oom_killed`, they are
86    /// counted for this run only.
87    #[serde(
88        default,
89        skip_serializing_if = "Option::is_none",
90        deserialize_with = "crate::cgroup_memory::deserialize_cgroup_memory"
91    )]
92    pub cgroup_memory: Option<crate::cgroup_memory::CgroupMemory>,
93    #[serde(default, skip_serializing_if = "Option::is_none")]
94    pub cpu_penalty: Option<Value>,
95    /// Hint explaining an otherwise opaque exit code (issue #162).
96    /// Derived from the log tail on read; never a stored verdict.
97    #[serde(default, skip_serializing_if = "Option::is_none")]
98    pub exit_reason: Option<String>,
99    /// Memory-exhaustion observation derived from the log tail (issue #165).
100    /// A runtime that aborts on its own heap limit never trips `oom_killed`.
101    #[serde(default, skip_serializing_if = "Option::is_none")]
102    pub memory_exhausted: Option<bool>,
103    /// The log line that carried the memory-exhaustion evidence (issue #165).
104    #[serde(default, skip_serializing_if = "Option::is_none")]
105    pub memory_exhausted_reason: Option<String>,
106    pub working_directory: String,
107    pub shell: String,
108    pub platform: String,
109    #[serde(default)]
110    pub options: HashMap<String, Value>,
111}
112
113impl ExecutionRecord {
114    /// Create a new execution record
115    pub fn new(command: &str) -> Self {
116        let now = Utc::now();
117        ExecutionRecord {
118            uuid: Uuid::new_v4().to_string(),
119            pid: None,
120            status: ExecutionStatus::Executing,
121            exit_code: None,
122            command: command.to_string(),
123            log_path: String::new(),
124            start_time: now.to_rfc3339(),
125            end_time: None,
126            attempt: None,
127            attempt_history: Vec::new(),
128            end_time_source: None,
129            observed_at: None,
130            stale_detected_at: None,
131            container_started_at: None,
132            oom_killed: None,
133            cgroup_memory: None,
134            cpu_penalty: None,
135            exit_reason: None,
136            memory_exhausted: None,
137            memory_exhausted_reason: None,
138            working_directory: env::current_dir()
139                .map(|p| p.to_string_lossy().to_string())
140                .unwrap_or_default(),
141            shell: env::var("SHELL").unwrap_or_else(|_| "/bin/sh".to_string()),
142            platform: std::env::consts::OS.to_string(),
143            options: HashMap::new(),
144        }
145    }
146
147    /// Create a new execution record with options
148    pub fn with_options(options: ExecutionRecordOptions) -> Self {
149        let mut record = Self::new(&options.command);
150        if let Some(uuid) = options.uuid {
151            record.uuid = uuid;
152        }
153        if let Some(pid) = options.pid {
154            record.pid = Some(pid);
155        }
156        if let Some(status) = options.status {
157            record.status = status;
158        }
159        if let Some(exit_code) = options.exit_code {
160            record.exit_code = Some(exit_code);
161        }
162        if let Some(log_path) = options.log_path {
163            record.log_path = log_path;
164        }
165        if let Some(start_time) = options.start_time {
166            record.start_time = start_time;
167        }
168        if let Some(end_time) = options.end_time {
169            record.end_time = Some(end_time);
170        }
171        if let Some(oom_killed) = options.oom_killed {
172            record.oom_killed = Some(oom_killed);
173        }
174        if let Some(working_directory) = options.working_directory {
175            record.working_directory = working_directory;
176        }
177        if let Some(shell) = options.shell {
178            record.shell = shell;
179        }
180        if let Some(platform) = options.platform {
181            record.platform = platform;
182        }
183        if let Some(opts) = options.options {
184            record.options = opts;
185        }
186        record
187    }
188
189    /// Mark execution as completed
190    pub fn complete(&mut self, exit_code: i32) {
191        self.status = ExecutionStatus::Executed;
192        self.exit_code = Some(exit_code);
193        self.end_time = Some(Utc::now().to_rfc3339());
194    }
195
196    /// Convert to JSON Value
197    pub fn to_json(&self) -> Value {
198        serde_json::to_value(self).unwrap_or(Value::Null)
199    }
200
201    /// Create from JSON Value
202    pub fn from_json(value: &Value) -> Option<Self> {
203        serde_json::from_value(value.clone()).ok()
204    }
205}
206
207/// Options for creating an execution record
208#[derive(Debug, Default)]
209pub struct ExecutionRecordOptions {
210    pub uuid: Option<String>,
211    pub command: String,
212    pub pid: Option<u32>,
213    pub status: Option<ExecutionStatus>,
214    pub exit_code: Option<i32>,
215    pub log_path: Option<String>,
216    pub start_time: Option<String>,
217    pub end_time: Option<String>,
218    pub oom_killed: Option<bool>,
219    pub working_directory: Option<String>,
220    pub shell: Option<String>,
221    pub platform: Option<String>,
222    pub options: Option<HashMap<String, Value>>,
223}
224
225pub use crate::store_lock::LockManager;
226
227/// Check if clink is installed
228pub fn is_clink_installed() -> bool {
229    Command::new("clink")
230        .arg("--version")
231        .stdout(Stdio::null())
232        .stderr(Stdio::null())
233        .status()
234        .map(|s| s.success())
235        .unwrap_or(false)
236}
237
238/// Get the default application folder path
239pub fn get_default_app_folder() -> PathBuf {
240    if let Ok(custom) = env::var("START_APP_FOLDER") {
241        return PathBuf::from(custom);
242    }
243
244    let home = env::var("HOME")
245        .or_else(|_| env::var("USERPROFILE"))
246        .unwrap_or_else(|_| ".".to_string());
247    PathBuf::from(home).join(DEFAULT_APP_FOLDER_NAME)
248}
249
250/// ExecutionStore - Main store class for managing execution records
251#[derive(Clone)]
252pub struct ExecutionStore {
253    app_folder: PathBuf,
254    lino_db_path: PathBuf,
255    links_db_path: PathBuf,
256    lock_file_path: PathBuf,
257    use_links: bool,
258    verbose: bool,
259}
260
261/// Options for creating an ExecutionStore
262#[derive(Debug, Default)]
263pub struct ExecutionStoreOptions {
264    pub app_folder: Option<PathBuf>,
265    pub use_links: Option<bool>,
266    pub verbose: bool,
267}
268
269impl ExecutionStore {
270    /// Create a new ExecutionStore with default options
271    pub fn new() -> Self {
272        Self::with_options(ExecutionStoreOptions::default())
273    }
274
275    /// Create a new ExecutionStore with options
276    pub fn with_options(options: ExecutionStoreOptions) -> Self {
277        let app_folder = options.app_folder.unwrap_or_else(get_default_app_folder);
278        let lino_db_path = app_folder.join(LINO_DB_FILE);
279        let links_db_path = app_folder.join(LINKS_DB_FILE);
280        let lock_file_path = app_folder.join(LOCK_FILE);
281        let use_links = options.use_links.unwrap_or_else(is_clink_installed);
282
283        // Ensure app folder exists
284        let _ = fs::create_dir_all(&app_folder);
285
286        ExecutionStore {
287            app_folder,
288            lino_db_path,
289            links_db_path,
290            lock_file_path,
291            use_links,
292            verbose: options.verbose,
293        }
294    }
295
296    /// Log verbose message
297    fn log(&self, message: &str) {
298        if self.verbose {
299            println!("[ExecutionStore] {}", message);
300        }
301    }
302
303    /// Read all execution records from lino file
304    pub fn read_lino_records(&self) -> Vec<ExecutionRecord> {
305        if !self.lino_db_path.exists() {
306            return Vec::new();
307        }
308
309        match fs::read_to_string(&self.lino_db_path) {
310            Ok(content) => {
311                if content.trim().is_empty() {
312                    return Vec::new();
313                }
314
315                match decode(&content) {
316                    Ok(data) => {
317                        if let LinoValue::Array(arr) = data {
318                            arr.iter()
319                                .map(lino_value_to_json)
320                                .filter_map(|value| ExecutionRecord::from_json(&value))
321                                .collect()
322                        } else {
323                            Vec::new()
324                        }
325                    }
326                    Err(e) => {
327                        self.log(&format!("Error decoding lino records: {}", e));
328                        Vec::new()
329                    }
330                }
331            }
332            Err(e) => {
333                self.log(&format!("Error reading lino records: {}", e));
334                Vec::new()
335            }
336        }
337    }
338
339    /// Write execution records to lino file
340    fn write_lino_records(&self, records: &[ExecutionRecord]) -> std::io::Result<()> {
341        let data: Vec<LinoValue> = records
342            .iter()
343            .map(|record| json_to_lino_value(&record.to_json()))
344            .collect();
345        let content = encode(&LinoValue::Array(data));
346        crate::atomic_write::atomic_write(&self.lino_db_path, content.as_bytes())?;
347        self.log(&format!("Wrote {} records to lino file", records.len()));
348        Ok(())
349    }
350
351    /// Build clink query for creating/updating a record
352    fn build_clink_create_query(&self, record: &ExecutionRecord) -> String {
353        let obj = record.to_json();
354        let mut links = Vec::new();
355
356        // Create main record link
357        links.push(format!(
358            "({}: ExecutionRecord {})",
359            record.uuid, record.uuid
360        ));
361
362        // Create property links
363        if let Value::Object(map) = obj {
364            for (key, value) in map {
365                let escaped_value = match value {
366                    Value::Object(_) | Value::Array(_) => {
367                        serde_json::to_string(&value).unwrap_or_default()
368                    }
369                    Value::String(s) => s,
370                    Value::Null => "null".to_string(),
371                    other => other.to_string(),
372                };
373                links.push(format!(
374                    "({}.{}: {} \"{}\")",
375                    record.uuid,
376                    key,
377                    key,
378                    escaped_value.replace('"', "\\\"")
379                ));
380            }
381        }
382
383        format!("() (({})))", links.join(") ("))
384    }
385
386    /// Execute clink command
387    fn exec_clink(&self, query: &str) -> Result<String, String> {
388        match Command::new("clink")
389            .arg(query)
390            .arg("--db")
391            .arg(&self.links_db_path)
392            .output()
393        {
394            Ok(output) => {
395                if output.status.success() {
396                    Ok(String::from_utf8_lossy(&output.stdout).to_string())
397                } else {
398                    let stderr = String::from_utf8_lossy(&output.stderr).to_string();
399                    self.log(&format!("Clink error: {}", stderr));
400                    Err(stderr)
401                }
402            }
403            Err(e) => {
404                self.log(&format!("Clink execution error: {}", e));
405                Err(e.to_string())
406            }
407        }
408    }
409
410    /// Write a record to the links database using clink
411    fn write_links_record(&self, record: &ExecutionRecord) -> bool {
412        if !self.use_links {
413            return false;
414        }
415
416        let query = self.build_clink_create_query(record);
417        match self.exec_clink(&query) {
418            Ok(_) => {
419                self.log(&format!("Wrote record {} to links database", record.uuid));
420                true
421            }
422            Err(_) => false,
423        }
424    }
425
426    /// Delete a record from links database
427    fn delete_links_record(&self, uuid: &str) -> bool {
428        if !self.use_links {
429            return false;
430        }
431
432        let query = format!("(($id: {} $any)) ()", uuid);
433        self.exec_clink(&query).is_ok()
434    }
435
436    /// Save an execution record (creates or updates)
437    pub fn save(&self, record: &ExecutionRecord) -> Result<(), String> {
438        self.save_reserved(record, None)
439    }
440
441    pub fn reserve_launch(
442        &self,
443        record: &ExecutionRecord,
444        previous: &ExecutionRecord,
445    ) -> Result<(), String> {
446        self.save_reserved(record, Some(previous))
447    }
448
449    fn save_reserved(
450        &self,
451        record: &ExecutionRecord,
452        previous: Option<&ExecutionRecord>,
453    ) -> Result<(), String> {
454        let mut lock = LockManager::new(self.lock_file_path.clone());
455
456        if !lock.acquire(LOCK_TIMEOUT_MS) {
457            return Err("Failed to acquire lock for database write".to_string());
458        }
459
460        // Read existing records
461        let mut records = self.read_lino_records();
462
463        // Find existing record index
464        let existing_index = records.iter().position(|r| r.uuid == record.uuid);
465        if let Some(previous) = previous {
466            let current = existing_index
467                .and_then(|i| records.get(i))
468                .ok_or("Execution disappeared before launch reservation")?;
469            if crate::launch_owner::has_active_launch(current)
470                || current.attempt.as_ref().map(|a| a.number)
471                    != previous.attempt.as_ref().map(|a| a.number)
472            {
473                return Err("A different resume already reserved this execution".into());
474            }
475        }
476
477        if let Some(idx) = existing_index {
478            // Update existing record
479            records[idx] = record.clone();
480        } else {
481            // Add new record
482            records.push(record.clone());
483        }
484
485        // Write to lino file
486        self.write_lino_records(&records)
487            .map_err(|e| e.to_string())?;
488
489        // Also write to links database if available
490        if self.use_links {
491            self.write_links_record(record);
492        }
493
494        Ok(())
495    }
496
497    /// Merge attempt metadata under the lock without replacing a watcher's terminal state.
498    pub fn patch_attempt(
499        &self,
500        uuid: &str,
501        number: u64,
502        fields: Value,
503    ) -> Result<Option<ExecutionRecord>, String> {
504        let mut lock = LockManager::new(self.lock_file_path.clone());
505        if !lock.acquire(LOCK_TIMEOUT_MS) {
506            return Err("Failed to acquire lock for attempt metadata".into());
507        }
508        let mut records = self.read_lino_records();
509        let Some(current) = records
510            .iter_mut()
511            .find(|r| r.uuid == uuid && r.attempt.as_ref().is_some_and(|a| a.number == number))
512        else {
513            return Ok(None);
514        };
515        let mut value =
516            serde_json::to_value(current.attempt.as_ref().unwrap()).map_err(|e| e.to_string())?;
517        if let Some(fields) = fields.as_object() {
518            for (key, field) in fields {
519                value[key] = field.clone();
520            }
521        }
522        current.attempt = Some(serde_json::from_value(value).map_err(|e| e.to_string())?);
523        let updated = current.clone();
524        self.write_lino_records(&records)
525            .map_err(|e| e.to_string())?;
526        if self.use_links {
527            self.write_links_record(&updated);
528        }
529        Ok(Some(updated))
530    }
531
532    /// Get an execution record by UUID or session name (falls back to options.sessionName)
533    pub fn get(&self, identifier: &str) -> Option<ExecutionRecord> {
534        let records = self.read_lino_records();
535        records
536            .iter()
537            .find(|r| r.uuid == identifier)
538            .cloned()
539            .or_else(|| {
540                records
541                    .iter()
542                    .find(|r| {
543                        r.options.get("sessionName").and_then(|v| v.as_str()) == Some(identifier)
544                    })
545                    .cloned()
546            })
547            .or_else(|| {
548                // Finally, honor session names this record used before it was
549                // resumed, so one logical session stays addressable across
550                // restarts (issue #162).
551                records.into_iter().find(|r| {
552                    r.options
553                        .get("sessionNameHistory")
554                        .and_then(|v| v.as_array())
555                        .is_some_and(|history| {
556                            history.iter().any(|name| name.as_str() == Some(identifier))
557                        })
558                })
559            })
560    }
561
562    /// Get all execution records
563    pub fn get_all(&self) -> Vec<ExecutionRecord> {
564        self.read_lino_records()
565    }
566    /// Get records filtered by status
567    pub fn get_by_status(&self, status: ExecutionStatus) -> Vec<ExecutionRecord> {
568        self.read_lino_records()
569            .into_iter()
570            .filter(|r| r.status == status)
571            .collect()
572    }
573    /// Get currently executing commands
574    pub fn get_executing(&self) -> Vec<ExecutionRecord> {
575        self.get_by_status(ExecutionStatus::Executing)
576    }
577    /// Get recently executed commands (sorted by start_time descending)
578    pub fn get_recent(&self, limit: usize) -> Vec<ExecutionRecord> {
579        let mut records = self.read_lino_records();
580        records.sort_by(|a, b| b.start_time.cmp(&a.start_time));
581        records.truncate(limit);
582        records
583    }
584    /// Clean up stale "executing" records (processes no longer running or aged out)
585    pub fn cleanup_stale(&self, options: CleanupOptions) -> CleanupResult {
586        let max_age_ms = options.max_age_ms.unwrap_or(24 * 60 * 60 * 1000);
587        let dry_run = options.dry_run;
588        let mut result = CleanupResult {
589            cleaned: 0,
590            records: Vec::new(),
591            errors: Vec::new(),
592        };
593
594        let records = self.read_lino_records();
595        let mut stale_records: Vec<ExecutionRecord> = Vec::new();
596
597        for record in records
598            .iter()
599            .filter(|r| r.status == ExecutionStatus::Executing)
600        {
601            if crate::launch_owner::has_active_launch(record) {
602                continue;
603            }
604            let mut is_stale = false;
605
606            // Check if process is still running (Unix only, same platform)
607            #[cfg(unix)]
608            if let Some(pid) = record.pid {
609                if record.platform == std::env::consts::OS
610                    && unsafe { libc::kill(pid as i32, 0) } != 0
611                {
612                    is_stale = true;
613                    self.log(&format!("Stale: {} (PID {} gone)", record.uuid, pid));
614                }
615            }
616
617            // Check age
618            if !is_stale {
619                if let Ok(st) = chrono::DateTime::parse_from_rfc3339(&record.start_time) {
620                    let age_ms =
621                        (chrono::Utc::now() - st.with_timezone(&chrono::Utc)).num_milliseconds();
622                    if age_ms > max_age_ms as i64 {
623                        is_stale = true;
624                        self.log(&format!(
625                            "Stale: {} ({}min old)",
626                            record.uuid,
627                            age_ms / 60000
628                        ));
629                    }
630                }
631            }
632
633            if is_stale {
634                stale_records.push(record.clone());
635            }
636        }
637
638        result.records = stale_records.clone();
639
640        if !dry_run && !stale_records.is_empty() {
641            let mut lock = LockManager::new(self.lock_file_path.clone());
642            if !lock.acquire(LOCK_TIMEOUT_MS) {
643                result.errors.push("Failed to acquire lock".to_string());
644                return result;
645            }
646
647            let mut current = self.read_lino_records();
648            for stale in &stale_records {
649                if let Some(i) = current.iter().position(|r| r.uuid == stale.uuid) {
650                    current[i].status = ExecutionStatus::Executed;
651                    current[i].exit_code = Some(-1);
652                    // Cleanup time is not a finish time: the execution ended at
653                    // an unknown moment before this one. Record when it was
654                    // detected and leave `end_time` empty (issue #170.3).
655                    current[i].end_time = None;
656                    current[i].stale_detected_at = Some(chrono::Utc::now().to_rfc3339());
657                    result.cleaned += 1;
658                }
659            }
660
661            if let Err(e) = self.write_lino_records(&current) {
662                result.errors.push(format!("Cleanup error: {}", e));
663            } else {
664                self.log(&format!("Cleaned {} stale records", result.cleaned));
665            }
666        } else if dry_run {
667            result.cleaned = stale_records.len();
668        }
669
670        result
671    }
672
673    /// Delete an execution record
674    pub fn delete(&self, uuid: &str) -> Result<bool, String> {
675        let mut lock = LockManager::new(self.lock_file_path.clone());
676
677        if !lock.acquire(LOCK_TIMEOUT_MS) {
678            return Err("Failed to acquire lock for database write".to_string());
679        }
680
681        let records = self.read_lino_records();
682        let filtered: Vec<_> = records.iter().filter(|r| r.uuid != uuid).cloned().collect();
683
684        if filtered.len() == records.len() {
685            return Ok(false); // Record not found
686        }
687
688        self.write_lino_records(&filtered)
689            .map_err(|e| e.to_string())?;
690
691        // Also delete from links database
692        if self.use_links {
693            self.delete_links_record(uuid);
694        }
695
696        Ok(true)
697    }
698
699    /// Delete all records
700    pub fn clear(&self) -> Result<(), String> {
701        let mut lock = LockManager::new(self.lock_file_path.clone());
702
703        if !lock.acquire(LOCK_TIMEOUT_MS) {
704            return Err("Failed to acquire lock for database write".to_string());
705        }
706
707        self.write_lino_records(&[]).map_err(|e| e.to_string())?;
708
709        // Clear links database by removing the file
710        if self.use_links && self.links_db_path.exists() {
711            let _ = fs::remove_file(&self.links_db_path);
712        }
713
714        Ok(())
715    }
716
717    /// Verify that both databases have consistent data
718    pub fn verify_consistency(&self) -> ConsistencyResult {
719        let mut result = ConsistencyResult {
720            consistent: true,
721            lino_count: 0,
722            links_count: 0,
723            errors: Vec::new(),
724        };
725
726        // Read lino records
727        let lino_records = self.read_lino_records();
728        result.lino_count = lino_records.len();
729
730        if !self.use_links {
731            result
732                .errors
733                .push("clink not installed - links database not available".to_string());
734            return result;
735        }
736
737        // Query links database for all ExecutionRecord links
738        match self.exec_clink("((($id: ExecutionRecord $uuid)) (($id: ExecutionRecord $uuid)))") {
739            Ok(output) => {
740                // Count unique UUIDs in the output
741                let re = regex::Regex::new(r"ExecutionRecord\s+([a-f0-9-]{36})").unwrap();
742                let uuids: std::collections::HashSet<_> = re
743                    .captures_iter(&output)
744                    .filter_map(|c| c.get(1).map(|m| m.as_str().to_lowercase()))
745                    .collect();
746                result.links_count = uuids.len();
747
748                // Check if counts match
749                if result.lino_count != result.links_count {
750                    result.consistent = false;
751                    result.errors.push(format!(
752                        "Record count mismatch: lino={}, links={}",
753                        result.lino_count, result.links_count
754                    ));
755                }
756
757                // Verify each lino record exists in links
758                for record in &lino_records {
759                    if !uuids.contains(&record.uuid.to_lowercase()) {
760                        result.consistent = false;
761                        result.errors.push(format!(
762                            "Record {} missing from links database",
763                            record.uuid
764                        ));
765                    }
766                }
767            }
768            Err(e) => {
769                result
770                    .errors
771                    .push(format!("Failed to query links database: {}", e));
772                result.consistent = false;
773            }
774        }
775
776        result
777    }
778
779    /// Get database statistics
780    pub fn get_stats(&self) -> ExecutionStats {
781        let records = self.read_lino_records();
782        let executing = records
783            .iter()
784            .filter(|r| r.status == ExecutionStatus::Executing)
785            .count();
786        let executed = records
787            .iter()
788            .filter(|r| r.status == ExecutionStatus::Executed)
789            .count();
790        let successful = records
791            .iter()
792            .filter(|r| r.status == ExecutionStatus::Executed && r.exit_code == Some(0))
793            .count();
794        let failed = records
795            .iter()
796            .filter(|r| {
797                r.status == ExecutionStatus::Executed
798                    && r.exit_code.map(|c| c != 0).unwrap_or(false)
799            })
800            .count();
801
802        ExecutionStats {
803            total: records.len(),
804            executing,
805            executed,
806            successful,
807            failed,
808            clink_available: self.use_links,
809            lino_db_path: self.lino_db_path.to_string_lossy().to_string(),
810            links_db_path: self.links_db_path.to_string_lossy().to_string(),
811        }
812    }
813
814    /// Get the app folder path
815    pub fn app_folder(&self) -> &Path {
816        &self.app_folder
817    }
818}
819
820impl Default for ExecutionStore {
821    fn default() -> Self {
822        Self::new()
823    }
824}
825
826/// Result of consistency verification
827#[derive(Debug)]
828pub struct ConsistencyResult {
829    pub consistent: bool,
830    pub lino_count: usize,
831    pub links_count: usize,
832    pub errors: Vec<String>,
833}
834
835/// Options for cleanup_stale: max_age_ms (default 24h), dry_run
836#[derive(Debug, Default)]
837pub struct CleanupOptions {
838    pub max_age_ms: Option<u64>,
839    pub dry_run: bool,
840}
841
842/// Result of cleanup_stale: cleaned count, records, errors
843#[derive(Debug)]
844pub struct CleanupResult {
845    pub cleaned: usize,
846    pub records: Vec<ExecutionRecord>,
847    pub errors: Vec<String>,
848}
849
850/// Execution statistics
851#[derive(Debug)]
852pub struct ExecutionStats {
853    pub total: usize,
854    pub executing: usize,
855    pub executed: usize,
856    pub successful: usize,
857    pub failed: usize,
858    pub clink_available: bool,
859    pub lino_db_path: String,
860    pub links_db_path: String,
861}
862
863#[cfg(test)]
864#[path = "execution_store_cases.rs"]
865mod tests;