beam_core/
workflow_cold_scan.rs

1use std::path::Path;
2
3use anyhow::{Context, Result};
4use serde::{Deserialize, Serialize};
5
6use crate::{
7    BeamPaths, RunChatBinding, WorkflowDefinition, read_run_snapshot,
8    workflow_snapshot::{RunSnapshotDTO, RunStatus},
9};
10
11#[derive(Debug, Clone, Serialize, Deserialize)]
12pub struct ColdWorkflowRun {
13    pub run_id: String,
14    pub def: WorkflowDefinition,
15    pub snapshot: RunSnapshotDTO,
16    pub binding: RunChatBinding,
17}
18
19#[derive(Debug, Clone, Default)]
20pub struct ColdScanStats {
21    pub discovered: usize,
22    pub skipped: Vec<String>,
23}
24
25pub async fn scan_cold_workflow_runs(
26    paths: &BeamPaths,
27    owner_lark_app_id: &str,
28) -> Result<(Vec<ColdWorkflowRun>, ColdScanStats)> {
29    let runs_dir = paths.workflow_runs_dir();
30    if !runs_dir.exists() {
31        return Ok((vec![], ColdScanStats::default()));
32    }
33
34    let mut runs = Vec::new();
35    let mut stats = ColdScanStats::default();
36    let dir_entries = std::fs::read_dir(&runs_dir).context("failed to read workflow runs dir")?;
37
38    for entry in dir_entries {
39        let entry = entry.context("failed to read dir entry")?;
40        if !entry.file_type().map(|t| t.is_dir()).unwrap_or(false) {
41            continue;
42        }
43        let run_id = entry.file_name().to_string_lossy().to_string();
44        let run_dir = runs_dir.join(&run_id);
45
46        match scan_single_run(&run_dir, &run_id, owner_lark_app_id).await {
47            Ok(Some(run)) => {
48                stats.discovered += 1;
49                runs.push(run);
50            }
51            Ok(None) => {}
52            Err(reason) => {
53                stats.skipped.push(format!("{}: {}", run_id, reason));
54            }
55        }
56    }
57
58    Ok((runs, stats))
59}
60
61async fn scan_single_run(
62    run_dir: &Path,
63    run_id: &str,
64    owner_lark_app_id: &str,
65) -> Result<Option<ColdWorkflowRun>> {
66    let binding_path = run_dir.join("chat-binding.json");
67    if !binding_path.exists() {
68        return Ok(None);
69    }
70
71    let binding: RunChatBinding = serde_json::from_str(
72        &std::fs::read_to_string(&binding_path)
73            .with_context(|| "failed to read chat-binding.json")?,
74    )
75    .with_context(|| "invalid chat-binding.json")?;
76
77    if binding.lark_app_id != owner_lark_app_id {
78        return Ok(None);
79    }
80
81    let def_path = run_dir.join("workflow.json");
82    if !def_path.exists() {
83        anyhow::bail!("missing workflow.json");
84    }
85
86    let raw_def =
87        std::fs::read_to_string(&def_path).with_context(|| "failed to read workflow.json")?;
88    let def: WorkflowDefinition =
89        serde_json::from_str(&raw_def).with_context(|| "invalid workflow.json")?;
90
91    let Some(snapshot) = read_run_snapshot(run_dir)
92        .await
93        .with_context(|| "failed to read snapshot")?
94    else {
95        anyhow::bail!("empty event log");
96    };
97
98    let terminal = matches!(
99        snapshot.run.status,
100        RunStatus::Succeeded | RunStatus::Failed | RunStatus::Cancelled
101    );
102    if terminal {
103        return Ok(None);
104    }
105
106    Ok(Some(ColdWorkflowRun {
107        run_id: run_id.to_string(),
108        def,
109        snapshot,
110        binding,
111    }))
112}
113
114#[cfg(test)]
115mod tests {
116    use super::*;
117    use crate::BeamPaths;
118    use std::{fs, path::PathBuf};
119
120    fn temp_data_root() -> PathBuf {
121        std::env::temp_dir().join(format!("beam-coldscan-{}", uuid::Uuid::new_v4()))
122    }
123
124    #[tokio::test]
125    async fn scan_empty_runs_dir_returns_nothing() {
126        let root = temp_data_root();
127        let paths = BeamPaths::from_root(&root);
128        fs::create_dir_all(paths.workflow_runs_dir()).unwrap();
129
130        let (runs, stats) = scan_cold_workflow_runs(&paths, "app-1").await.unwrap();
131        assert!(runs.is_empty());
132        assert_eq!(stats.discovered, 0);
133
134        let _ = fs::remove_dir_all(root);
135    }
136
137    #[tokio::test]
138    async fn scan_silently_skips_runs_without_chat_binding() {
139        let root = temp_data_root();
140        let paths = BeamPaths::from_root(&root);
141        let run_dir = paths.workflow_run_dir("run-1");
142        fs::create_dir_all(&run_dir).unwrap();
143
144        let (runs, _stats) = scan_cold_workflow_runs(&paths, "app-1").await.unwrap();
145        assert!(runs.is_empty());
146
147        let _ = fs::remove_dir_all(root);
148    }
149}
beam_core/workflow_cold_scan.rs

beam_core/
workflow_cold_scan.rs
