beam_core/
workflow_sidecar.rs

1use std::fs;
2use std::path::{Path, PathBuf};
3
4use anyhow::{Context, Result};
5use serde_json::Value;
6
7use crate::EventLog;
8
9pub async fn write_effect_input_sidecar(
10    log: &EventLog,
11    activity_id: &str,
12    attempt_id: &str,
13    input: &Value,
14) -> Result<PathBuf> {
15    let dir = effect_input_dir(log, activity_id, attempt_id);
16    fs::create_dir_all(&dir)?;
17    let path = dir.join("effect-input.json");
18    fs::write(&path, serde_json::to_vec_pretty(input)?)?;
19    Ok(path)
20}
21
22pub async fn load_effect_input_sidecar(
23    run_dir: &Path,
24    activity_id: &str,
25    attempt_id: &str,
26) -> Result<Option<Value>> {
27    let path = run_dir
28        .join("attempts")
29        .join(activity_id)
30        .join(attempt_id)
31        .join("effect-input.json");
32    let raw = match fs::read_to_string(&path) {
33        Ok(raw) => raw,
34        Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(None),
35        Err(err) => return Err(err.into()),
36    };
37    let parsed = serde_json::from_str::<Value>(&raw)
38        .with_context(|| format!("failed to parse {}", path.display()))?;
39    Ok(Some(parsed))
40}
41
42fn effect_input_dir(log: &EventLog, activity_id: &str, attempt_id: &str) -> PathBuf {
43    log.run_dir
44        .join("attempts")
45        .join(activity_id)
46        .join(attempt_id)
47}
48
49#[cfg(test)]
50mod tests {
51    use super::*;
52    use crate::BeamPaths;
53    use std::time::{SystemTime, UNIX_EPOCH};
54
55    fn temp_paths(label: &str) -> BeamPaths {
56        let nanos = SystemTime::now()
57            .duration_since(UNIX_EPOCH)
58            .unwrap_or_default()
59            .as_nanos();
60        BeamPaths::from_root(std::env::temp_dir().join(format!(
61            "beam-workflow-sidecar-{label}-{nanos}-{}",
62            std::process::id()
63        )))
64    }
65
66    #[tokio::test]
67    async fn effect_input_sidecar_round_trips() {
68        let paths = temp_paths("roundtrip");
69        let log = EventLog::new("run-1", paths.workflow_runs_dir()).unwrap();
70        let input = serde_json::json!({"chatId":"chat-1","content":"hello"});
71        let path = write_effect_input_sidecar(&log, "act-1", "act-1::att-1", &input)
72            .await
73            .expect("write");
74        assert!(path.exists());
75        let loaded = load_effect_input_sidecar(&log.run_dir, "act-1", "act-1::att-1")
76            .await
77            .expect("load")
78            .expect("some");
79        assert_eq!(loaded, input);
80        let _ = std::fs::remove_dir_all(paths.root());
81    }
82}
beam_core/workflow_sidecar.rs

beam_core/
workflow_sidecar.rs
