start_command/

execution_store.rs

//! Execution Store - Dual storage (.lino text + .links binary) for command execution records

use crate::lino_value_json::{json_to_lino_value, lino_value_to_json};
use crate::store_decode_cache::{decode_store_content, DecodeSource};
use chrono::Utc;
use lino_objects_codec::{decode, encode, LinoValue};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::HashMap;
use std::env;
use std::fs;
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use uuid::Uuid;

/// Default application folder name
const DEFAULT_APP_FOLDER_NAME: &str = ".start-command";
/// Lino database file name
const LINO_DB_FILE: &str = "executions.lino";
/// Links database file name
const LINKS_DB_FILE: &str = "executions.links";
/// Lock file name
const LOCK_FILE: &str = "executions.lock";
/// Decoded records as JSON, for the next process (issue #243)
pub const DECODED_COPY_FILE: &str = "executions.decoded.rust.json";
/// Lock timeout in milliseconds
const LOCK_TIMEOUT_MS: u64 = 30000;

/// Execution status
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum ExecutionStatus {
    Executing,
    Executed,
}

impl ExecutionStatus {
    pub fn as_str(&self) -> &'static str {
        match self {
            ExecutionStatus::Executing => "executing",
            ExecutionStatus::Executed => "executed",
        }
    }
}

impl std::fmt::Display for ExecutionStatus {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.as_str())
    }
}

/// Command Execution Record
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ExecutionRecord {
    pub uuid: String,
    pub pid: Option<u32>,
    pub status: ExecutionStatus,
    pub exit_code: Option<i32>,
    pub command: String,
    pub log_path: String,
    pub start_time: String,
    pub end_time: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub attempt: Option<crate::execution_attempt::ExecutionAttempt>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub attempt_history: Vec<Value>,
    /// UUID of the execution this one continues (`--continues`, issue #231).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub continues: Option<String>,
    /// Where `end_time` came from (issue #170.2): `docker-finished-at`,
    /// `log-footer` or `observed-at`. Without it a real finish time and a
    /// timestamp taken when someone happened to run `--status` are
    /// indistinguishable.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub end_time_source: Option<String>,
    /// When the end of the execution was *observed*, for the cases where no
    /// real finish time exists (issue #170.2).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub observed_at: Option<String>,
    /// When `cleanup_stale()` gave up on a record. Deliberately not an
    /// `end_time`: cleanup time is not a finish time (issue #170.3).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub stale_detected_at: Option<String>,
    /// `docker inspect .State.StartedAt` of the container that ran the command.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub container_started_at: Option<String>,
    /// `removed` once the container is known to be gone (issue #225), with
    /// when that was observed; `options.containerId` is kept as history.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub container_state: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub container_removed_observed_at: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub oom_killed: Option<bool>,
    /// cgroup v2 memory counters the detached watcher sampled while the
    /// container ran (issue #182). Unlike the sticky `oom_killed`, they are
    /// counted for this run only.
    #[serde(
        default,
        skip_serializing_if = "Option::is_none",
        deserialize_with = "crate::cgroup_memory::deserialize_cgroup_memory"
    )]
    pub cgroup_memory: Option<crate::cgroup_memory::CgroupMemory>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub cpu_penalty: Option<Value>,
    /// Hint explaining an otherwise opaque exit code (issue #162).
    /// Derived from the log tail on read; never a stored verdict.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub exit_reason: Option<String>,
    /// Memory-exhaustion observation derived from the log tail (issue #165).
    /// A runtime that aborts on its own heap limit never trips `oom_killed`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub memory_exhausted: Option<bool>,
    /// The log line that carried the memory-exhaustion evidence (issue #165).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub memory_exhausted_reason: Option<String>,
    pub working_directory: String,
    pub shell: String,
    pub platform: String,
    #[serde(default)]
    pub options: HashMap<String, Value>,
}

impl ExecutionRecord {
    /// Create a new execution record
    pub fn new(command: &str) -> Self {
        let now = Utc::now();
        ExecutionRecord {
            uuid: Uuid::new_v4().to_string(),
            pid: None,
            status: ExecutionStatus::Executing,
            exit_code: None,
            command: command.to_string(),
            log_path: String::new(),
            start_time: now.to_rfc3339(),
            end_time: None,
            attempt: None,
            attempt_history: Vec::new(),
            continues: None,
            end_time_source: None,
            observed_at: None,
            stale_detected_at: None,
            container_started_at: None,
            container_state: None,
            container_removed_observed_at: None,
            oom_killed: None,
            cgroup_memory: None,
            cpu_penalty: None,
            exit_reason: None,
            memory_exhausted: None,
            memory_exhausted_reason: None,
            working_directory: env::current_dir()
                .map(|p| p.to_string_lossy().to_string())
                .unwrap_or_default(),
            shell: env::var("SHELL").unwrap_or_else(|_| "/bin/sh".to_string()),
            platform: std::env::consts::OS.to_string(),
            options: HashMap::new(),
        }
    }

    /// Create a new execution record with options
    pub fn with_options(options: ExecutionRecordOptions) -> Self {
        let mut record = Self::new(&options.command);
        if let Some(uuid) = options.uuid {
            record.uuid = uuid;
        }
        if let Some(pid) = options.pid {
            record.pid = Some(pid);
        }
        if let Some(status) = options.status {
            record.status = status;
        }
        if let Some(exit_code) = options.exit_code {
            record.exit_code = Some(exit_code);
        }
        if let Some(log_path) = options.log_path {
            record.log_path = log_path;
        }
        if let Some(start_time) = options.start_time {
            record.start_time = start_time;
        }
        if let Some(end_time) = options.end_time {
            record.end_time = Some(end_time);
        }
        if let Some(oom_killed) = options.oom_killed {
            record.oom_killed = Some(oom_killed);
        }
        if let Some(working_directory) = options.working_directory {
            record.working_directory = working_directory;
        }
        if let Some(shell) = options.shell {
            record.shell = shell;
        }
        if let Some(platform) = options.platform {
            record.platform = platform;
        }
        if let Some(opts) = options.options {
            record.options = opts;
        }
        record
    }

    /// Mark execution as completed
    pub fn complete(&mut self, exit_code: i32) {
        self.status = ExecutionStatus::Executed;
        self.exit_code = Some(exit_code);
        self.end_time = Some(Utc::now().to_rfc3339());
    }

    /// Convert to JSON Value
    pub fn to_json(&self) -> Value {
        serde_json::to_value(self).unwrap_or(Value::Null)
    }

    /// Create from JSON Value
    pub fn from_json(value: &Value) -> Option<Self> {
        serde_json::from_value(value.clone()).ok()
    }
}

/// Options for creating an execution record
#[derive(Debug, Default)]
pub struct ExecutionRecordOptions {
    pub uuid: Option<String>,
    pub command: String,
    pub pid: Option<u32>,
    pub status: Option<ExecutionStatus>,
    pub exit_code: Option<i32>,
    pub log_path: Option<String>,
    pub start_time: Option<String>,
    pub end_time: Option<String>,
    pub oom_killed: Option<bool>,
    pub working_directory: Option<String>,
    pub shell: Option<String>,
    pub platform: Option<String>,
    pub options: Option<HashMap<String, Value>>,
}

pub use crate::store_lock::LockManager;

/// Quote text as a clink reference: `"..."` with inner quotes doubled.
/// clink 4.x rejects backslash escapes inside references.
pub fn clink_reference(text: &str) -> String {
    format!("\"{}\"", text.replace('"', "\"\""))
}

/// Check if clink is installed
pub fn is_clink_installed() -> bool {
    Command::new("clink")
        .arg("--version")
        .stdout(Stdio::null())
        .stderr(Stdio::null())
        .status()
        .map(|s| s.success())
        .unwrap_or(false)
}

/// Get the default application folder path
pub fn get_default_app_folder() -> PathBuf {
    if let Ok(custom) = env::var("START_APP_FOLDER") {
        return PathBuf::from(custom);
    }

    let home = env::var("HOME")
        .or_else(|_| env::var("USERPROFILE"))
        .unwrap_or_else(|_| ".".to_string());
    PathBuf::from(home).join(DEFAULT_APP_FOLDER_NAME)
}

/// ExecutionStore - Main store class for managing execution records
#[derive(Clone)]
pub struct ExecutionStore {
    app_folder: PathBuf,
    lino_db_path: PathBuf,
    links_db_path: PathBuf,
    lock_file_path: PathBuf,
    use_links: bool,
    verbose: bool,
}

/// Options for creating an ExecutionStore
#[derive(Debug, Default)]
pub struct ExecutionStoreOptions {
    pub app_folder: Option<PathBuf>,
    pub use_links: Option<bool>,
    pub verbose: bool,
}

impl ExecutionStore {
    /// Create a new ExecutionStore with default options
    pub fn new() -> Self {
        Self::with_options(ExecutionStoreOptions::default())
    }

    /// Create a new ExecutionStore with options
    pub fn with_options(options: ExecutionStoreOptions) -> Self {
        let app_folder = options.app_folder.unwrap_or_else(get_default_app_folder);
        let lino_db_path = app_folder.join(LINO_DB_FILE);
        let links_db_path = app_folder.join(LINKS_DB_FILE);
        let lock_file_path = app_folder.join(LOCK_FILE);
        let use_links = options.use_links.unwrap_or_else(is_clink_installed);

        // Ensure app folder exists
        let _ = fs::create_dir_all(&app_folder);

        ExecutionStore {
            app_folder,
            lino_db_path,
            links_db_path,
            lock_file_path,
            use_links,
            verbose: options.verbose,
        }
    }

    /// Log verbose message
    fn log(&self, message: &str) {
        if self.verbose {
            println!("[ExecutionStore] {}", message);
        }
    }

    /// Read all execution records from lino file
    pub fn read_lino_records(&self) -> Vec<ExecutionRecord> {
        if !self.lino_db_path.exists() {
            return Vec::new();
        }

        match fs::read_to_string(&self.lino_db_path) {
            Ok(content) => {
                if content.trim().is_empty() {
                    return Vec::new();
                }

                // Decoded once per content, not once per call (issue #243).
                let decoded = decode_store_content(
                    &self.lino_db_path,
                    &content,
                    |content| {
                        decode(content).map(|data| match data {
                            LinoValue::Array(arr) => arr.iter().map(lino_value_to_json).collect(),
                            _ => Vec::new(),
                        })
                    },
                    Some(&self.app_folder.join(DECODED_COPY_FILE)),
                );
                match decoded {
                    Ok((records, source)) => {
                        self.log(&match source {
                            DecodeSource::Memory => {
                                "Reused the decoded store: the file is unchanged".to_string()
                            }
                            DecodeSource::Copy => {
                                "Read the decoded copy of the store: the file is unchanged"
                                    .to_string()
                            }
                            DecodeSource::Decode => {
                                format!("Decoded the store ({} characters)", content.len())
                            }
                        });
                        records
                    }
                    Err(e) => {
                        self.log(&format!("Error decoding lino records: {}", e));
                        Vec::new()
                    }
                }
            }
            Err(e) => {
                self.log(&format!("Error reading lino records: {}", e));
                Vec::new()
            }
        }
    }

    /// Write execution records to lino file
    fn write_lino_records(&self, records: &[ExecutionRecord]) -> std::io::Result<()> {
        let data: Vec<LinoValue> = records
            .iter()
            .map(|record| json_to_lino_value(&record.to_json()))
            .collect();
        let content = encode(&LinoValue::Array(data));
        crate::atomic_write::atomic_write(&self.lino_db_path, content.as_bytes())?;
        self.log(&format!("Wrote {} records to lino file", records.len()));
        Ok(())
    }

    /// Convert a record's properties to clink links notation.
    ///
    /// The record is one link `(uuid: ExecutionRecord uuid)` plus one link per
    /// property, `(uuid.key: uuid "key=value")`. Property links have the record
    /// as their source, so records that share a value stay distinct links
    /// (doublets are deduplicated by source and target), and the `key=` prefix
    /// keeps a number from being read as a link address (issue #253).
    pub fn record_to_links_notation(&self, record: &ExecutionRecord) -> String {
        let mut links = Vec::new();

        if let Value::Object(map) = record.to_json() {
            for (key, value) in map {
                let text = match value {
                    Value::Null => continue,
                    Value::Object(_) | Value::Array(_) => {
                        serde_json::to_string(&value).unwrap_or_default()
                    }
                    Value::String(s) => s,
                    other => other.to_string(),
                };
                links.push(format!(
                    "({}.{}: {} {})",
                    record.uuid,
                    key,
                    record.uuid,
                    clink_reference(&format!("{}={}", key, text))
                ));
            }
        }

        links.join(" ")
    }

    /// Build the create query with the same notation as the property helper.
    pub fn build_clink_create_query(&self, record: &ExecutionRecord) -> String {
        format!(
            "() (({}: ExecutionRecord {}) {})",
            record.uuid,
            record.uuid,
            self.record_to_links_notation(record)
        )
    }

    /// Execute clink command, appending `extra_args` after `--db <path>`
    fn exec_clink(&self, query: &str, extra_args: &[&str]) -> Result<String, String> {
        match Command::new("clink")
            .arg(query)
            .arg("--db")
            .arg(&self.links_db_path)
            .args(extra_args)
            .output()
        {
            Ok(output) => {
                if output.status.success() {
                    Ok(String::from_utf8_lossy(&output.stdout).to_string())
                } else {
                    let stderr = String::from_utf8_lossy(&output.stderr).to_string();
                    self.log(&format!("Clink error: {}", stderr));
                    Err(stderr)
                }
            }
            Err(e) => {
                self.log(&format!("Clink execution error: {}", e));
                Err(e.to_string())
            }
        }
    }

    /// Write a record to the links database using clink; `is_new` skips
    /// dropping the previous links
    fn write_links_record(&self, record: &ExecutionRecord, is_new: bool) -> bool {
        if !self.use_links {
            return false;
        }

        // Replace, not merge: drop the previous property links first. clink
        // reports an error when the record has no links yet, which is harmless.
        if !is_new {
            let _ = self.exec_clink(&format!("(($id: {} $value)) ()", record.uuid), &[]);
        }
        let query = self.build_clink_create_query(record);
        match self.exec_clink(&query, &["--auto-create-missing-references"]) {
            Ok(_) => {
                self.log(&format!("Wrote record {} to links database", record.uuid));
                true
            }
            Err(_) => false,
        }
    }

    /// Delete a record from links database
    fn delete_links_record(&self, uuid: &str) -> bool {
        if !self.use_links {
            return false;
        }

        let _ = self.exec_clink(&format!("(($id: {} $value)) ()", uuid), &[]);
        let query = format!("(({}: ExecutionRecord {})) ()", uuid, uuid);
        self.exec_clink(&query, &[]).is_ok()
    }

    /// Save an execution record (creates or updates)
    pub fn save(&self, record: &ExecutionRecord) -> Result<(), String> {
        self.save_reserved(record, None)
    }

    pub fn reserve_launch(
        &self,
        record: &ExecutionRecord,
        previous: &ExecutionRecord,
    ) -> Result<(), String> {
        self.save_reserved(record, Some(previous))
    }

    fn save_reserved(
        &self,
        record: &ExecutionRecord,
        previous: Option<&ExecutionRecord>,
    ) -> Result<(), String> {
        let mut lock = LockManager::new(self.lock_file_path.clone());

        if !lock.acquire(LOCK_TIMEOUT_MS) {
            return Err("Failed to acquire lock for database write".to_string());
        }

        // Read existing records
        let mut records = self.read_lino_records();

        // Find existing record index
        let existing_index = records.iter().position(|r| r.uuid == record.uuid);
        if let Some(previous) = previous {
            let current = existing_index
                .and_then(|i| records.get(i))
                .ok_or("Execution disappeared before launch reservation")?;
            if crate::launch_owner::has_active_launch(current)
                || current.attempt.as_ref().map(|a| a.number)
                    != previous.attempt.as_ref().map(|a| a.number)
            {
                return Err("A different resume already reserved this execution".into());
            }
        }

        if let Some(idx) = existing_index {
            // Update existing record
            records[idx] = record.clone();
        } else {
            // Add new record
            records.push(record.clone());
        }

        // Write to lino file
        self.write_lino_records(&records)
            .map_err(|e| e.to_string())?;

        // Also write to links database if available
        if self.use_links {
            self.write_links_record(record, existing_index.is_none());
        }

        Ok(())
    }

    /// Merge attempt metadata under the lock without replacing a watcher's terminal state.
    pub fn patch_attempt(
        &self,
        uuid: &str,
        number: u64,
        fields: Value,
    ) -> Result<Option<ExecutionRecord>, String> {
        let mut lock = LockManager::new(self.lock_file_path.clone());
        if !lock.acquire(LOCK_TIMEOUT_MS) {
            return Err("Failed to acquire lock for attempt metadata".into());
        }
        let mut records = self.read_lino_records();
        let Some(current) = records
            .iter_mut()
            .find(|r| r.uuid == uuid && r.attempt.as_ref().is_some_and(|a| a.number == number))
        else {
            return Ok(None);
        };
        let mut value =
            serde_json::to_value(current.attempt.as_ref().unwrap()).map_err(|e| e.to_string())?;
        if let Some(fields) = fields.as_object() {
            for (key, field) in fields {
                value[key] = field.clone();
            }
        }
        current.attempt = Some(serde_json::from_value(value).map_err(|e| e.to_string())?);
        let updated = current.clone();
        self.write_lino_records(&records)
            .map_err(|e| e.to_string())?;
        if self.use_links {
            self.write_links_record(&updated, false);
        }
        Ok(Some(updated))
    }

    /// Record that a finished execution's container no longer exists (issue
    /// #225). Applied under the lock, and only while the record is still
    /// finished, so a concurrent resume is never undone.
    pub fn mark_container_removed(
        &self,
        uuid: &str,
        observed_at: &str,
    ) -> Result<Option<ExecutionRecord>, String> {
        let mut lock = LockManager::new(self.lock_file_path.clone());
        if !lock.acquire(LOCK_TIMEOUT_MS) {
            return Err("Failed to acquire lock for container state".into());
        }
        let mut records = self.read_lino_records();
        let Some(current) = records.iter_mut().find(|r| {
            r.uuid == uuid
                && r.status == ExecutionStatus::Executed
                && r.container_state.as_deref() != Some("removed")
        }) else {
            return Ok(None);
        };
        current.container_state = Some("removed".to_string());
        current.container_removed_observed_at = Some(observed_at.to_string());
        let updated = current.clone();
        self.write_lino_records(&records)
            .map_err(|e| e.to_string())?;
        if self.use_links {
            self.write_links_record(&updated, false);
        }
        Ok(Some(updated))
    }

    /// Get an execution record by UUID, session name, earlier session name
    /// (issue #162), display name or alias (issue #231).
    pub fn get(&self, identifier: &str) -> Option<ExecutionRecord> {
        crate::execution_identity::find_record(&self.read_lino_records(), identifier).cloned()
    }

    /// Run `mutate` on the stored records under the database lock and persist
    /// them when it returns a changed record (issues #229, #231).
    pub fn with_lock<F>(&self, mutate: F) -> Result<Option<ExecutionRecord>, String>
    where
        F: FnOnce(&mut Vec<ExecutionRecord>) -> Result<Option<ExecutionRecord>, String>,
    {
        let mut lock = LockManager::new(self.lock_file_path.clone());
        if !lock.acquire(LOCK_TIMEOUT_MS) {
            return Err("Failed to acquire lock for database write".into());
        }
        let mut records = self.read_lino_records();
        let changed = mutate(&mut records)?;
        if let Some(record) = &changed {
            self.write_lino_records(&records)
                .map_err(|e| e.to_string())?;
            if self.use_links {
                self.write_links_record(record, false);
            }
        }
        Ok(changed)
    }

    /// Get all execution records
    pub fn get_all(&self) -> Vec<ExecutionRecord> {
        self.read_lino_records()
    }
    /// Get records filtered by status
    pub fn get_by_status(&self, status: ExecutionStatus) -> Vec<ExecutionRecord> {
        self.read_lino_records()
            .into_iter()
            .filter(|r| r.status == status)
            .collect()
    }
    /// Get currently executing commands
    pub fn get_executing(&self) -> Vec<ExecutionRecord> {
        self.get_by_status(ExecutionStatus::Executing)
    }
    /// Get recently executed commands (sorted by start_time descending)
    pub fn get_recent(&self, limit: usize) -> Vec<ExecutionRecord> {
        let mut records = self.read_lino_records();
        records.sort_by(|a, b| b.start_time.cmp(&a.start_time));
        records.truncate(limit);
        records
    }
    /// Clean up stale "executing" records (processes no longer running or aged out)
    pub fn cleanup_stale(&self, options: CleanupOptions) -> CleanupResult {
        let max_age_ms = options.max_age_ms.unwrap_or(24 * 60 * 60 * 1000);
        let dry_run = options.dry_run;
        let mut result = CleanupResult {
            cleaned: 0,
            records: Vec::new(),
            errors: Vec::new(),
        };

        let records = self.read_lino_records();
        let mut stale_records: Vec<ExecutionRecord> = Vec::new();

        for record in records
            .iter()
            .filter(|r| r.status == ExecutionStatus::Executing)
        {
            if crate::launch_owner::has_active_launch(record) {
                continue;
            }
            let mut is_stale = false;

            // Check if process is still running (Unix only, same platform)
            #[cfg(unix)]
            if let Some(pid) = record.pid {
                if record.platform == std::env::consts::OS
                    && unsafe { libc::kill(pid as i32, 0) } != 0
                {
                    is_stale = true;
                    self.log(&format!("Stale: {} (PID {} gone)", record.uuid, pid));
                }
            }

            // Check age
            if !is_stale {
                if let Ok(st) = chrono::DateTime::parse_from_rfc3339(&record.start_time) {
                    let age_ms =
                        (chrono::Utc::now() - st.with_timezone(&chrono::Utc)).num_milliseconds();
                    if age_ms > max_age_ms as i64 {
                        is_stale = true;
                        self.log(&format!(
                            "Stale: {} ({}min old)",
                            record.uuid,
                            age_ms / 60000
                        ));
                    }
                }
            }

            if is_stale {
                stale_records.push(record.clone());
            }
        }

        result.records = stale_records.clone();

        if !dry_run && !stale_records.is_empty() {
            let mut lock = LockManager::new(self.lock_file_path.clone());
            if !lock.acquire(LOCK_TIMEOUT_MS) {
                result.errors.push("Failed to acquire lock".to_string());
                return result;
            }

            let mut current = self.read_lino_records();
            for stale in &stale_records {
                if let Some(i) = current.iter().position(|r| r.uuid == stale.uuid) {
                    current[i].status = ExecutionStatus::Executed;
                    current[i].exit_code = Some(-1);
                    // Cleanup time is not a finish time: the execution ended at
                    // an unknown moment before this one. Record when it was
                    // detected and leave `end_time` empty (issue #170.3).
                    current[i].end_time = None;
                    current[i].stale_detected_at = Some(chrono::Utc::now().to_rfc3339());
                    result.cleaned += 1;
                }
            }

            if let Err(e) = self.write_lino_records(&current) {
                result.errors.push(format!("Cleanup error: {}", e));
            } else {
                self.log(&format!("Cleaned {} stale records", result.cleaned));
            }
        } else if dry_run {
            result.cleaned = stale_records.len();
        }

        result
    }

    /// Delete an execution record
    pub fn delete(&self, uuid: &str) -> Result<bool, String> {
        let mut lock = LockManager::new(self.lock_file_path.clone());

        if !lock.acquire(LOCK_TIMEOUT_MS) {
            return Err("Failed to acquire lock for database write".to_string());
        }

        let records = self.read_lino_records();
        let filtered: Vec<_> = records.iter().filter(|r| r.uuid != uuid).cloned().collect();

        if filtered.len() == records.len() {
            return Ok(false); // Record not found
        }

        self.write_lino_records(&filtered)
            .map_err(|e| e.to_string())?;

        // Also delete from links database
        if self.use_links {
            self.delete_links_record(uuid);
        }

        Ok(true)
    }

    /// Delete all records
    pub fn clear(&self) -> Result<(), String> {
        let mut lock = LockManager::new(self.lock_file_path.clone());

        if !lock.acquire(LOCK_TIMEOUT_MS) {
            return Err("Failed to acquire lock for database write".to_string());
        }

        self.write_lino_records(&[]).map_err(|e| e.to_string())?;

        // Clear links database by removing the file
        if self.use_links && self.links_db_path.exists() {
            let _ = fs::remove_file(&self.links_db_path);
        }

        Ok(())
    }

    /// Verify that both databases have consistent data
    pub fn verify_consistency(&self) -> ConsistencyResult {
        let mut result = ConsistencyResult {
            consistent: true,
            lino_count: 0,
            links_count: 0,
            errors: Vec::new(),
        };

        // Read lino records
        let lino_records = self.read_lino_records();
        result.lino_count = lino_records.len();

        if !self.use_links {
            result
                .errors
                .push("clink not installed - links database not available".to_string());
            return result;
        }

        // Query links database for all ExecutionRecord links
        // Without --changes clink prints nothing for a read-only match
        match self.exec_clink(
            "((($id: ExecutionRecord $uuid)) (($id: ExecutionRecord $uuid)))",
            &["--changes"],
        ) {
            Ok(output) => {
                // Count unique UUIDs in the output
                let re = regex::Regex::new(r"ExecutionRecord\s+([a-f0-9-]{36})").unwrap();
                let uuids: std::collections::HashSet<_> = re
                    .captures_iter(&output)
                    .filter_map(|c| c.get(1).map(|m| m.as_str().to_lowercase()))
                    .collect();
                result.links_count = uuids.len();

                // Check if counts match
                if result.lino_count != result.links_count {
                    result.consistent = false;
                    result.errors.push(format!(
                        "Record count mismatch: lino={}, links={}",
                        result.lino_count, result.links_count
                    ));
                }

                // Verify each lino record exists in links
                for record in &lino_records {
                    if !uuids.contains(&record.uuid.to_lowercase()) {
                        result.consistent = false;
                        result.errors.push(format!(
                            "Record {} missing from links database",
                            record.uuid
                        ));
                    }
                }
            }
            Err(e) => {
                result
                    .errors
                    .push(format!("Failed to query links database: {}", e));
                result.consistent = false;
            }
        }

        result
    }

    /// Get database statistics
    pub fn get_stats(&self) -> ExecutionStats {
        let records = self.read_lino_records();
        let executing = records
            .iter()
            .filter(|r| r.status == ExecutionStatus::Executing)
            .count();
        let executed = records
            .iter()
            .filter(|r| r.status == ExecutionStatus::Executed)
            .count();
        let successful = records
            .iter()
            .filter(|r| r.status == ExecutionStatus::Executed && r.exit_code == Some(0))
            .count();
        let failed = records
            .iter()
            .filter(|r| {
                r.status == ExecutionStatus::Executed
                    && r.exit_code.map(|c| c != 0).unwrap_or(false)
            })
            .count();

        ExecutionStats {
            total: records.len(),
            executing,
            executed,
            successful,
            failed,
            clink_available: self.use_links,
            lino_db_path: self.lino_db_path.to_string_lossy().to_string(),
            links_db_path: self.links_db_path.to_string_lossy().to_string(),
        }
    }

    /// Get the app folder path
    pub fn app_folder(&self) -> &Path {
        &self.app_folder
    }
}

impl Default for ExecutionStore {
    fn default() -> Self {
        Self::new()
    }
}

/// Result of consistency verification
#[derive(Debug)]
pub struct ConsistencyResult {
    pub consistent: bool,
    pub lino_count: usize,
    pub links_count: usize,
    pub errors: Vec<String>,
}

/// Options for cleanup_stale: max_age_ms (default 24h), dry_run
#[derive(Debug, Default)]
pub struct CleanupOptions {
    pub max_age_ms: Option<u64>,
    pub dry_run: bool,
}

/// Result of cleanup_stale: cleaned count, records, errors
#[derive(Debug)]
pub struct CleanupResult {
    pub cleaned: usize,
    pub records: Vec<ExecutionRecord>,
    pub errors: Vec<String>,
}

/// Execution statistics
#[derive(Debug)]
pub struct ExecutionStats {
    pub total: usize,
    pub executing: usize,
    pub executed: usize,
    pub successful: usize,
    pub failed: usize,
    pub clink_available: bool,
    pub lino_db_path: String,
    pub links_db_path: String,
}

#[cfg(test)]
#[path = "execution_store_cases.rs"]
mod tests;