beam_core/
workflow_run.rs

1use std::collections::BTreeMap;
2use std::fs;
3use std::path::Path;
4
5use anyhow::{Context, Result};
6use serde::{Deserialize, Serialize};
7use sha2::{Digest, Sha256};
8
9use crate::{BeamPaths, EventDraft, EventLog, WorkflowActor, parse_workflow_definition};
10
11#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
12#[serde(rename_all = "camelCase")]
13pub struct RunChatBinding {
14    pub chat_id: String,
15    pub lark_app_id: String,
16}
17
18#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
19#[serde(rename_all = "camelCase")]
20pub struct WorkflowOutputRef {
21    pub output_hash: String,
22    pub output_path: String,
23    pub output_bytes: usize,
24    pub output_schema_version: u32,
25    #[serde(default, skip_serializing_if = "Option::is_none")]
26    pub content_type: Option<String>,
27}
28
29#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
30pub struct WorkflowRunBootstrap {
31    pub run_id: String,
32    pub workflow_id: String,
33    pub revision_id: String,
34    pub input_ref: WorkflowOutputRef,
35}
36
37#[derive(Debug, Clone)]
38pub struct BootstrapWorkflowRunInput<'a> {
39    pub run_id: &'a str,
40    pub workflow_json: &'a str,
41    pub expected_workflow_id: Option<&'a str>,
42    pub params: &'a BTreeMap<String, String>,
43    pub initiator: &'a str,
44    pub chat_binding: Option<RunChatBinding>,
45}
46
47pub fn mint_workflow_run_id(workflow_id: &str, now_ms: u64) -> String {
48    let safe = workflow_id
49        .chars()
50        .map(|ch| match ch {
51            'a'..='z' | 'A'..='Z' | '0'..='9' | '.' | '_' | '-' => ch,
52            _ => '_',
53        })
54        .collect::<String>();
55    format!("{}-{}", safe, now_ms)
56}
57
58pub fn bootstrap_workflow_run(
59    paths: &BeamPaths,
60    input: BootstrapWorkflowRunInput<'_>,
61) -> Result<WorkflowRunBootstrap> {
62    let workflow = parse_workflow_definition(input.workflow_json)?;
63    let workflow_id = workflow.workflow_id.clone();
64    if let Some(expected) = input.expected_workflow_id {
65        if expected != workflow_id {
66            anyhow::bail!(
67                "workflowId mismatch: requested={} file={}",
68                expected,
69                workflow_id
70            );
71        }
72    }
73
74    let run_dir = paths.workflow_run_dir(input.run_id);
75    fs::create_dir_all(run_dir.join("blobs"))?;
76    fs::write(run_dir.join("workflow.json"), input.workflow_json)?;
77    if let Some(binding) = input.chat_binding {
78        fs::write(
79            run_dir.join("chat-binding.json"),
80            serde_json::to_vec_pretty(&binding)?,
81        )?;
82    }
83
84    let params_json = serde_json::to_vec(input.params)?;
85    let params_hash = sha256_hex(&params_json);
86    let input_path = run_dir.join("blobs").join(&params_hash);
87    fs::write(&input_path, &params_json)?;
88    let input_ref = WorkflowOutputRef {
89        output_hash: format!("sha256:{}", params_hash),
90        output_path: input_path.display().to_string(),
91        output_bytes: params_json.len(),
92        output_schema_version: 1,
93        content_type: Some("application/json".to_string()),
94    };
95
96    let mut log = EventLog::new(input.run_id.to_string(), paths.workflow_runs_dir())?;
97    let revision_id = sha256_hex(&serde_json::to_vec(&workflow)?);
98    let _run_created = log.append(EventDraft {
99        event_type: "runCreated".to_string(),
100        actor: WorkflowActor::System,
101        payload: serde_json::json!({
102            "workflowId": workflow_id,
103            "revisionId": revision_id,
104            "inputRef": input_ref,
105            "initiator": input.initiator,
106        }),
107        timestamp: None,
108        payload_hash: None,
109    })?;
110    let _run_started = log.append(EventDraft {
111        event_type: "runStarted".to_string(),
112        actor: WorkflowActor::Scheduler,
113        payload: serde_json::json!({}),
114        timestamp: None,
115        payload_hash: None,
116    })?;
117
118    Ok(WorkflowRunBootstrap {
119        run_id: input.run_id.to_string(),
120        workflow_id,
121        revision_id,
122        input_ref,
123    })
124}
125
126fn sha256_hex(bytes: &[u8]) -> String {
127    let mut hasher = Sha256::new();
128    hasher.update(bytes);
129    format!("{:x}", hasher.finalize())
130}
131
132pub fn read_workflow_definition_from_path(path: &Path) -> Result<String> {
133    Ok(fs::read_to_string(path).with_context(|| format!("读取 {} 失败", path.display()))?)
134}
135
136#[cfg(test)]
137mod tests {
138    use super::*;
139    use std::collections::BTreeMap;
140    use std::time::{SystemTime, UNIX_EPOCH};
141
142    fn temp_paths(label: &str) -> BeamPaths {
143        let nanos = SystemTime::now()
144            .duration_since(UNIX_EPOCH)
145            .unwrap_or_default()
146            .as_nanos();
147        BeamPaths::from_root(std::env::temp_dir().join(format!(
148            "beam-workflow-run-{label}-{nanos}-{}",
149            std::process::id()
150        )))
151    }
152
153    #[test]
154    fn mint_workflow_run_id_sanitizes_input() {
155        let run_id = mint_workflow_run_id("flow/a:b", 123);
156        assert!(run_id.starts_with("flow_a_b-123"));
157        assert!(!run_id.contains('/'));
158        assert!(!run_id.contains(':'));
159    }
160
161    #[test]
162    fn bootstrap_workflow_run_writes_snapshot_and_events() {
163        let paths = temp_paths("bootstrap");
164        let params = BTreeMap::from([(String::from("foo"), String::from("bar"))]);
165        let result = bootstrap_workflow_run(
166            &paths,
167            BootstrapWorkflowRunInput {
168                run_id: "run-1",
169                workflow_json: r#"{"workflowId":"flow-a","version":1,"nodes":{"node-a":{"type":"subagent","bot":"bot-a","prompt":"hi"}}}"#,
170                expected_workflow_id: Some("flow-a"),
171                params: &params,
172                initiator: "cli",
173                chat_binding: Some(RunChatBinding {
174                    chat_id: "chat-1".to_string(),
175                    lark_app_id: "app-1".to_string(),
176                }),
177            },
178        )
179        .expect("bootstrap");
180        assert_eq!(result.run_id, "run-1");
181        assert_eq!(result.workflow_id, "flow-a");
182        assert!(
183            paths
184                .workflow_run_dir("run-1")
185                .join("workflow.json")
186                .exists()
187        );
188        assert!(
189            paths
190                .workflow_run_dir("run-1")
191                .join("chat-binding.json")
192                .exists()
193        );
194        assert!(paths.workflow_run_dir("run-1").join("blobs").exists());
195        let log = EventLog::new("run-1", paths.workflow_runs_dir()).expect("log");
196        let events = log.read_all().expect("events");
197        assert_eq!(events.len(), 2);
198        assert_eq!(events[0].event_type, "runCreated");
199        assert_eq!(events[1].event_type, "runStarted");
200        let _ = std::fs::remove_dir_all(paths.root());
201    }
202
203    #[test]
204    fn bootstrap_workflow_run_hashes_canonical_definition_bytes() {
205        let params = BTreeMap::new();
206        let raw_a = r#"{"workflowId":"flow-a","version":1,"nodes":{"node-a":{"type":"subagent","bot":"bot-a","prompt":"hi","workingDir":"/tmp/demo"}}}"#;
207        let raw_b = r#"
208        {
209            "nodes": {
210                "node-a": {
211                    "prompt": "hi",
212                    "type": "subagent",
213                    "bot": "bot-a",
214                    "workingDir": "/tmp/demo"
215                }
216            },
217            "version": 1,
218            "workflowId": "flow-a"
219        }
220        "#;
221
222        let paths_a = temp_paths("bootstrap-canonical-a");
223        let paths_b = temp_paths("bootstrap-canonical-b");
224        let rev_a = bootstrap_workflow_run(
225            &paths_a,
226            BootstrapWorkflowRunInput {
227                run_id: "run-a",
228                workflow_json: raw_a,
229                expected_workflow_id: Some("flow-a"),
230                params: &params,
231                initiator: "cli",
232                chat_binding: None,
233            },
234        )
235        .expect("bootstrap a")
236        .revision_id;
237        let rev_b = bootstrap_workflow_run(
238            &paths_b,
239            BootstrapWorkflowRunInput {
240                run_id: "run-b",
241                workflow_json: raw_b,
242                expected_workflow_id: Some("flow-a"),
243                params: &params,
244                initiator: "cli",
245                chat_binding: None,
246            },
247        )
248        .expect("bootstrap b")
249        .revision_id;
250        assert_eq!(rev_a, rev_b);
251        let _ = std::fs::remove_dir_all(paths_a.root());
252        let _ = std::fs::remove_dir_all(paths_b.root());
253    }
254}
beam_core/workflow_run.rs

beam_core/
workflow_run.rs
