lean_ctx/core/workflow/
store.rs

1use crate::core::workflow::types::WorkflowRun;
2use std::path::PathBuf;
3
4/// Stale threshold: workflows inactive for over 30 minutes are auto-cleared on load.
5const STALE_MINUTES: i64 = 30;
6
7/// TTL for expired workflow files (24 hours).
8pub const WORKFLOW_TTL_SECS: u64 = 24 * 60 * 60;
9
10fn workflows_dir() -> Option<PathBuf> {
11    crate::core::data_dir::lean_ctx_data_dir()
12        .ok()
13        .map(|d| d.join("workflows"))
14}
15
16fn workflow_path_for_agent(agent_id: Option<&str>) -> Option<PathBuf> {
17    let dir = workflows_dir()?;
18    let filename = match agent_id {
19        Some(id) if !id.trim().is_empty() => {
20            let safe_id: String = id
21                .chars()
22                .map(|c| {
23                    if c.is_alphanumeric() || c == '-' || c == '_' {
24                        c
25                    } else {
26                        '_'
27                    }
28                })
29                .collect();
30            format!("workflow-{safe_id}.json")
31        }
32        _ => "active.json".to_string(),
33    };
34    Some(dir.join(filename))
35}
36
37pub fn load_active() -> Result<Option<WorkflowRun>, String> {
38    load_active_for_agent(None)
39}
40
41pub fn load_active_for_agent(agent_id: Option<&str>) -> Result<Option<WorkflowRun>, String> {
42    let Some(path) = workflow_path_for_agent(agent_id) else {
43        return Ok(None);
44    };
45    let content = match std::fs::read_to_string(&path) {
46        Ok(c) => c,
47        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
48            // Backward compat: if agent-scoped file missing, try legacy active.json (read-only migration)
49            if agent_id.is_some()
50                && let Some(legacy) = workflow_path_for_agent(None)
51                && let Ok(lc) = std::fs::read_to_string(&legacy)
52            {
53                let run: WorkflowRun = serde_json::from_str(&lc)
54                    .map_err(|e| format!("Invalid legacy workflow JSON: {e}"))?;
55                let elapsed = chrono::Utc::now()
56                    .signed_duration_since(run.updated_at)
57                    .num_minutes();
58                if elapsed <= STALE_MINUTES && run.current != "done" {
59                    return Ok(Some(run));
60                }
61            }
62            return Ok(None);
63        }
64        Err(e) => return Err(format!("read {}: {e}", path.display())),
65    };
66    let run: WorkflowRun =
67        serde_json::from_str(&content).map_err(|e| format!("Invalid workflow JSON: {e}"))?;
68
69    let elapsed = chrono::Utc::now()
70        .signed_duration_since(run.updated_at)
71        .num_minutes();
72    if elapsed > STALE_MINUTES || run.current == "done" {
73        let _ = std::fs::remove_file(&path);
74        return Ok(None);
75    }
76    Ok(Some(run))
77}
78
79pub fn save_active(run: &WorkflowRun) -> Result<(), String> {
80    save_active_for_agent(run, None)
81}
82
83pub fn save_active_for_agent(run: &WorkflowRun, agent_id: Option<&str>) -> Result<(), String> {
84    let Some(path) = workflow_path_for_agent(agent_id) else {
85        return Err("No home directory available".to_string());
86    };
87    if let Some(parent) = path.parent() {
88        std::fs::create_dir_all(parent).map_err(|e| format!("mkdir failed: {e}"))?;
89    }
90    let json = serde_json::to_string_pretty(run).map_err(|e| format!("serialize failed: {e}"))?;
91    let tmp = path.with_extension("tmp");
92    std::fs::write(&tmp, json).map_err(|e| format!("write failed: {e}"))?;
93    std::fs::rename(&tmp, &path).map_err(|e| format!("rename failed: {e}"))?;
94    Ok(())
95}
96
97pub fn clear_active() -> Result<(), String> {
98    clear_active_for_agent(None)
99}
100
101pub fn clear_active_for_agent(agent_id: Option<&str>) -> Result<(), String> {
102    let Some(path) = workflow_path_for_agent(agent_id) else {
103        return Ok(());
104    };
105    match std::fs::remove_file(&path) {
106        Ok(()) => Ok(()),
107        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
108        Err(e) => Err(format!("remove {}: {e}", path.display())),
109    }
110}
111
112/// Remove workflow files older than `WORKFLOW_TTL_SECS`.
113pub fn cleanup_expired() -> (u32, u64) {
114    let Some(dir) = workflows_dir() else {
115        return (0, 0);
116    };
117    let Ok(entries) = std::fs::read_dir(&dir) else {
118        return (0, 0);
119    };
120    let now = std::time::SystemTime::now();
121    let mut removed = 0u32;
122    let mut freed = 0u64;
123
124    for entry in entries.flatten() {
125        let path = entry.path();
126        if !path.is_file() {
127            continue;
128        }
129        let ext = path.extension().and_then(|e| e.to_str());
130        if ext != Some("json") {
131            continue;
132        }
133        let Ok(meta) = std::fs::metadata(&path) else {
134            continue;
135        };
136        let age = meta
137            .modified()
138            .ok()
139            .and_then(|m| now.duration_since(m).ok())
140            .map_or(0, |d| d.as_secs());
141        if age > WORKFLOW_TTL_SECS {
142            freed += meta.len();
143            let _ = std::fs::remove_file(&path);
144            removed += 1;
145        }
146    }
147    (removed, freed)
148}
lean_ctx/core/workflow/store.rs

lean_ctx/core/workflow/
store.rs
