roder_dynamic_workflows/
store.rs

1use std::fs::{self, OpenOptions};
2use std::io::{BufRead, BufReader, Write};
3use std::path::{Path, PathBuf};
4
5use crate::host_api::WorkflowCheckpoint;
6use crate::model::{WorkflowRuntimeError, WorkflowRuntimeErrorKind, WorkflowRuntimeResult};
7use roder_api::subagents::SubagentResult;
8use serde::{Deserialize, Serialize};
9use time::OffsetDateTime;
10
11#[derive(Debug, Clone)]
12pub struct WorkflowCheckpointStore {
13    root: PathBuf,
14}
15
16impl WorkflowCheckpointStore {
17    pub fn new(root: impl Into<PathBuf>) -> Self {
18        Self { root: root.into() }
19    }
20
21    pub fn append_checkpoint(
22        &self,
23        run_id: &str,
24        checkpoint: &WorkflowCheckpoint,
25    ) -> WorkflowRuntimeResult<()> {
26        let dir = self.run_dir(run_id);
27        fs::create_dir_all(&dir).map_err(store_error)?;
28        let path = dir.join("checkpoints.jsonl");
29        let mut file = OpenOptions::new()
30            .create(true)
31            .append(true)
32            .open(path)
33            .map_err(store_error)?;
34        let line = serde_json::to_string(checkpoint).map_err(|err| {
35            WorkflowRuntimeError::new(
36                WorkflowRuntimeErrorKind::Store,
37                format!("checkpoint is not serializable: {err}"),
38            )
39        })?;
40        writeln!(file, "{line}").map_err(store_error)?;
41        Ok(())
42    }
43
44    pub fn read_checkpoints(&self, run_id: &str) -> WorkflowRuntimeResult<Vec<WorkflowCheckpoint>> {
45        let path = self.run_dir(run_id).join("checkpoints.jsonl");
46        if !path.exists() {
47            return Ok(Vec::new());
48        }
49
50        let file = fs::File::open(path).map_err(store_error)?;
51        let reader = BufReader::new(file);
52        let mut checkpoints = Vec::new();
53        for line in reader.lines() {
54            let line = line.map_err(store_error)?;
55            if line.trim().is_empty() {
56                continue;
57            }
58            let checkpoint = serde_json::from_str(&line).map_err(|err| {
59                WorkflowRuntimeError::new(
60                    WorkflowRuntimeErrorKind::Store,
61                    format!("invalid checkpoint record: {err}"),
62                )
63            })?;
64            checkpoints.push(checkpoint);
65        }
66        Ok(checkpoints)
67    }
68
69    pub fn append_agent_result(
70        &self,
71        run_id: &str,
72        result: &WorkflowCachedAgentResult,
73    ) -> WorkflowRuntimeResult<()> {
74        let dir = self.run_dir(run_id);
75        fs::create_dir_all(&dir).map_err(store_error)?;
76        let path = dir.join("agent-results.jsonl");
77        let mut file = OpenOptions::new()
78            .create(true)
79            .append(true)
80            .open(path)
81            .map_err(store_error)?;
82        let line = serde_json::to_string(result).map_err(|err| {
83            WorkflowRuntimeError::new(
84                WorkflowRuntimeErrorKind::Store,
85                format!("agent result is not serializable: {err}"),
86            )
87        })?;
88        writeln!(file, "{line}").map_err(store_error)?;
89        Ok(())
90    }
91
92    pub fn read_agent_results(
93        &self,
94        run_id: &str,
95    ) -> WorkflowRuntimeResult<Vec<WorkflowCachedAgentResult>> {
96        let path = self.run_dir(run_id).join("agent-results.jsonl");
97        if !path.exists() {
98            return Ok(Vec::new());
99        }
100
101        let file = fs::File::open(path).map_err(store_error)?;
102        let reader = BufReader::new(file);
103        let mut results = Vec::new();
104        for line in reader.lines() {
105            let line = line.map_err(store_error)?;
106            if line.trim().is_empty() {
107                continue;
108            }
109            let result = serde_json::from_str(&line).map_err(|err| {
110                WorkflowRuntimeError::new(
111                    WorkflowRuntimeErrorKind::Store,
112                    format!("invalid agent result record: {err}"),
113                )
114            })?;
115            results.push(result);
116        }
117        Ok(results)
118    }
119
120    pub fn find_agent_result(
121        &self,
122        run_id: &str,
123        key: &WorkflowAgentCacheKey,
124    ) -> WorkflowRuntimeResult<Option<WorkflowCachedAgentResult>> {
125        Ok(self
126            .read_agent_results(run_id)?
127            .into_iter()
128            .rev()
129            .find(|record| &record.key == key))
130    }
131
132    pub fn invalidate_agent_results(
133        &self,
134        run_id: &str,
135        agent_id: &str,
136    ) -> WorkflowRuntimeResult<usize> {
137        let path = self.run_dir(run_id).join("agent-results.jsonl");
138        if !path.exists() {
139            return Ok(0);
140        }
141
142        let results = self.read_agent_results(run_id)?;
143        let original_len = results.len();
144        let retained = results
145            .into_iter()
146            .filter(|record| record.key.agent_id != agent_id)
147            .collect::<Vec<_>>();
148        let removed = original_len.saturating_sub(retained.len());
149        let mut file = OpenOptions::new()
150            .write(true)
151            .truncate(true)
152            .open(path)
153            .map_err(store_error)?;
154        for result in retained {
155            let line = serde_json::to_string(&result).map_err(|err| {
156                WorkflowRuntimeError::new(
157                    WorkflowRuntimeErrorKind::Store,
158                    format!("agent result is not serializable: {err}"),
159                )
160            })?;
161            writeln!(file, "{line}").map_err(store_error)?;
162        }
163        Ok(removed)
164    }
165
166    fn run_dir(&self, run_id: &str) -> PathBuf {
167        self.root
168            .join("dynamic-workflows")
169            .join("runs")
170            .join(run_id)
171    }
172
173    pub fn root(&self) -> &Path {
174        &self.root
175    }
176}
177
178#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Hash)]
179#[serde(rename_all = "camelCase")]
180pub struct WorkflowAgentCacheKey {
181    pub run_id: String,
182    pub phase_id: String,
183    pub agent_id: String,
184    pub prompt_hash: String,
185    #[serde(default, skip_serializing_if = "Option::is_none")]
186    pub model: Option<String>,
187    #[serde(default)]
188    pub tool_scope: Vec<String>,
189}
190
191#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
192#[serde(rename_all = "camelCase")]
193pub struct WorkflowCachedAgentResult {
194    pub key: WorkflowAgentCacheKey,
195    pub result: SubagentResult,
196    #[serde(with = "time::serde::rfc3339")]
197    pub completed_at: OffsetDateTime,
198}
199
200fn store_error(error: std::io::Error) -> WorkflowRuntimeError {
201    WorkflowRuntimeError::new(WorkflowRuntimeErrorKind::Store, error.to_string())
202}
roder_dynamic_workflows/store.rs

roder_dynamic_workflows/
store.rs
