diff --git a/.gitignore b/.gitignore index 51ab688f..72202dce 100644 --- a/.gitignore +++ b/.gitignore @@ -31,3 +31,4 @@ fly.toml crates/core/target/ apps/ +third/ diff --git a/crates/core/src/export_type.rs b/crates/core/src/export_type.rs index c1e420f5..2bc35e00 100644 --- a/crates/core/src/export_type.rs +++ b/crates/core/src/export_type.rs @@ -29,7 +29,21 @@ mod private { seen.insert(name.clone()); let decl = (et.decl_fn)(&cfg); decls.push((name, decl)); + continue; } + let decl = (et.decl_fn)(&cfg); + let Some((_, prev_decl)) = decls.iter().find(|(n, _)| n == &name) else { + continue; + }; + if prev_decl == &decl { + continue; // identical duplicate registration — safe to ignore + } + panic!( + "TS export collision: two different Rust types resolve to the \ + type name `{name}` (flat single-file export requires globally \ + unique names). Registering module must rename one of them via \ + the export_types! registration or a distinct Rust type name." + ); } decls.sort_by(|a, b| a.0.cmp(&b.0)); decls diff --git a/crates/core/src/flows/engine.rs b/crates/core/src/flows/engine.rs new file mode 100644 index 00000000..749ee4b2 --- /dev/null +++ b/crates/core/src/flows/engine.rs @@ -0,0 +1,865 @@ +//! Serial execution core (execution-engine.md, contracts C4-adjacent). +//! +//! P1.2 skeleton: ready-queue + single-point state writes + Join/skip + whole +//! snapshot. Internal semantics for start/end/branch; `script`/`egress` run via +//! an injected [`NodeExecutor`] (wired in P1.5/1.6); await/resume in P2. +//! +//! Execution is serial (one node at a time). Fan-out runs every target that has +//! data; join waits for ALL incoming edges to be decided with at least one taken. + +use std::collections::{HashMap, VecDeque}; + +use async_trait::async_trait; +use serde::{Deserialize, Serialize}; +use serde_json::{Value, json}; + +use crate::errors::app_error::{AppError, AppResult}; + +use super::graph::{Edge, Graph, GraphNode}; +use super::nodes::{self, BranchConfig, EndConfig}; + +/// variable pool: ns -> name -> value (child fields live inside object values). +pub type Pool = HashMap>; + +pub const S_RUNNING: &str = "running"; +pub const S_WAITING: &str = "waiting"; +pub const S_SUCCESS: &str = "success"; +pub const S_FAILED: &str = "failed"; +pub const S_SKIPPED: &str = "skipped"; + +pub const N_SUCCESS: &str = "success"; +pub const N_SKIPPED: &str = "skipped"; +pub const N_IN_PROGRESS: &str = "in_progress"; +pub const N_WAITING: &str = "waiting"; +pub const N_FAILED: &str = "failed"; + +/// One node's per-attempt result (idempotency: success/skipped never re-run). +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +pub struct NodeState { + pub status: String, + #[serde(default)] + pub output: Option, + #[serde(default)] + pub error: Option, + #[serde(default)] + pub attempt: i64, +} + +/// `modifiers.retry` config. +#[derive(Debug, Clone, Deserialize, Default)] +pub struct RetryModifier { + #[serde(default)] + pub attempts: Option, + #[allow(dead_code)] + #[serde(default)] + pub backoff: Option, +} + +/// Orthogonal node modifiers (contracts C1.4). Retry is attempted in-process; +/// `continue_on_error` fails the node but lets the run pass through. +#[derive(Debug, Clone, Deserialize, Default)] +pub struct NodeModifiers { + #[serde(default)] + pub retry: Option, + #[serde(default)] + pub continue_on_error: bool, +} + +/// Edge verdict for readiness (join = all decided; skip = all skipped). +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub enum EdgeMark { + Taken, + Skipped, +} + +/// Whole runnable snapshot (durable; one row per instance). +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +pub struct Snapshot { + pub pool: Pool, + pub node_states: HashMap, + /// keyed `source|handle->target`. + pub edge_marks: HashMap, + pub status: String, + #[serde(default)] + pub error: Option, + #[serde(default)] + pub outputs: Option, + /// Nodes parked on `await` (resume completes the head of this list). + #[serde(default)] + pub waiting_nodes: Vec, +} + +fn edge_key(e: &Edge) -> String { + format!("{}|{}->{}", e.source, e.source_handle, e.target) +} + +impl Snapshot { + pub fn new() -> Self { + Self { + pool: HashMap::new(), + node_states: HashMap::new(), + edge_marks: HashMap::new(), + status: S_RUNNING.to_string(), + error: None, + outputs: None, + waiting_nodes: Vec::new(), + } + } +} + +/// Complete the head waiting node with `payload` and resume the run. +pub fn resume_snapshot(snap: &mut Snapshot, payload: Option) -> AppResult<()> { + let Some(node) = snap.waiting_nodes.first().cloned() else { + return Err(AppError::BadRequest("实例没有等待中的节点".into())); + }; + snap.waiting_nodes.retain(|n| *n != node); + let payload = payload.unwrap_or(Value::Null); + let st = snap.node_states.entry(node.clone()).or_default(); + st.status = N_SUCCESS.to_string(); + st.output = Some(payload.clone()); + st.attempt += 1; + let ns = snap.pool.entry(node.clone()).or_default(); + ns.insert("resume".to_string(), payload.clone()); + ns.insert("output".to_string(), payload); + snap.status = S_RUNNING.to_string(); + Ok(()) +} + +/// What an executable node produced. +#[derive(Debug)] +pub struct ExecOutcome { + pub output: Value, +} + +/// Async executor for action nodes (`script`/`egress`). Wired in P1.5/1.6. +#[async_trait] +pub trait NodeExecutor: Send + Sync { + async fn exec(&self, node: &GraphNode, input: Value) -> AppResult; +} + +/// Durability hook: persist the snapshot after each claim / node completion. +#[async_trait] +pub trait Persist: Send + Sync { + async fn persist(&self, snap: &Snapshot) -> AppResult<()>; +} + +/// No-op persistence (pure in-memory runs / tests). +pub struct NoopPersist; +#[async_trait] +impl Persist for NoopPersist { + async fn persist(&self, _snap: &Snapshot) -> AppResult<()> { + Ok(()) + } +} + +/// Run one full pass of the graph over the snapshot (serial, in-memory). +pub async fn run(graph: &Graph, snap: &mut Snapshot, exec: &dyn NodeExecutor) -> AppResult<()> { + run_persisted(graph, snap, exec, &NoopPersist).await +} + +/// Serial pass with a durability hook: [`Persist::persist`] is invoked right +/// after a node is claimed (`in_progress`, before its side effect runs) and +/// after each node completes — so a crash mid-node leaves a claimed state and +/// completed nodes never re-run on resume (A.3 claim-then-act, at-least-once). +pub async fn run_persisted( + graph: &Graph, + snap: &mut Snapshot, + exec: &dyn NodeExecutor, + persist: &dyn Persist, +) -> AppResult<()> { + if snap.status != S_RUNNING { + return Ok(()); + } + let mut queue: VecDeque = VecDeque::new(); + queue.push_back(graph.start.clone()); + + while let Some(id) = queue.pop_front() { + if snap.status != S_RUNNING { + break; + } + let Some(node) = graph.nodes.get(&id) else { + continue; + }; + // Idempotent resume: already decided nodes never re-run, but their + // downstream must still be (re)propagated from persisted state. + if let Some(st) = snap.node_states.get(&id) + && st.status == N_SUCCESS + { + resume_completed(graph, snap, &id, &mut queue)?; + continue; + } + if let Some(st) = snap.node_states.get(&id) + && st.status == N_SKIPPED + { + skip_node(graph, snap, &id, &mut queue)?; + continue; + } + + let attempt = snap.node_states.get(&id).map_or(1, |s| s.attempt + 1); + match node.data.kind.as_str() { + nodes::T_START => { + set_in_progress(snap, &id, attempt); + persist.persist(snap).await?; + mark_node_success(snap, &id, Value::Null); + fan_out_after_run(graph, snap, &id, &mut queue)?; + } + nodes::T_END => { + set_in_progress(snap, &id, attempt); + persist.persist(snap).await?; + let outputs = resolve_end_outputs(node, snap)?; + finish_success(snap, &id, outputs); + } + nodes::T_BRANCH => { + set_in_progress(snap, &id, attempt); + persist.persist(snap).await?; + let cfg: BranchConfig = serde_json::from_value(node.data.config.clone()) + .map_err(|e| AppError::BadRequest(format!("branch config: {e}")))?; + let handle = pick_branch(&cfg, &snap.pool)?; + mark_node_success(snap, &id, json!({"handle": handle})); + fan_out_after_branch(graph, snap, &id, Some(handle.as_str()), &mut queue)?; + } + nodes::T_SCRIPT | nodes::T_EGRESS => { + let mods: NodeModifiers = + serde_json::from_value(node.data.modifiers.clone()).unwrap_or_default(); + let attempts = mods + .retry + .as_ref() + .and_then(|r| r.attempts) + .unwrap_or(1) + .max(1); + let input = match resolve_inputs(&node.data.config, &snap.pool) { + Ok(v) => v, + Err(e) => { + fail(snap, &id, e.to_string()); + continue; + } + }; + let mut last_error: Option = None; + let mut outcome: Option = None; + for i in 1..=attempts { + // Claim before each attempt: persisted → crash mid-node + // re-runs at-least-once from this claim (A.3). + set_in_progress(snap, &id, i); + persist.persist(snap).await?; + match exec.exec(node, input.clone()).await { + Ok(o) => { + outcome = Some(o); + break; + } + Err(e) => { + last_error = Some(e.to_string()); + } + } + } + match outcome { + Some(o) => { + mark_node_success(snap, &id, o.output); + fan_out_after_run(graph, snap, &id, &mut queue)?; + } + None => { + let msg = last_error.unwrap_or_default(); + if mods.continue_on_error { + // Fail the node but pass the run through (downstream + // continues; its inputs still resolve upstream). + let st = snap.node_states.entry(id.clone()).or_default(); + st.status = N_FAILED.to_string(); + st.error = Some(json!({"message": msg})); + fan_out_after_run(graph, snap, &id, &mut queue)?; + } else { + fail(snap, &id, msg); + } + } + } + } + nodes::T_AWAIT => { + // Park the run: node marked waiting + recorded, instance -> + // waiting. Resume completes it (see `resume_snapshot`), then the + // engine's success-resume fan-out continues downstream. + let st = snap.node_states.entry(id.clone()).or_default(); + st.status = N_WAITING.to_string(); + if !snap.waiting_nodes.contains(&id) { + snap.waiting_nodes.push(id.clone()); + } + snap.status = S_WAITING.to_string(); + persist.persist(snap).await?; + break; + } + other => { + fail(snap, &id, format!("unsupported node type '{other}'")); + } + } + persist.persist(snap).await?; + } + if snap.status == S_RUNNING { + // queue drained without an end → treat as success (no outputs). + snap.status = S_SUCCESS.to_string(); + persist.persist(snap).await?; + } + Ok(()) +} + +fn set_in_progress(snap: &mut Snapshot, id: &str, attempt: i64) { + let st = snap.node_states.entry(id.to_string()).or_default(); + st.status = N_IN_PROGRESS.to_string(); + st.attempt = attempt; +} + +fn mark_node_success(snap: &mut Snapshot, id: &str, output: Value) { + let st = snap.node_states.entry(id.to_string()).or_default(); + st.status = N_SUCCESS.to_string(); + st.output = Some(output.clone()); + if output.is_object() || output.is_array() { + snap.pool + .entry(id.to_string()) + .or_default() + .insert("output".to_string(), output); + } +} + +fn finish_success(snap: &mut Snapshot, id: &str, outputs: Value) { + let st = snap.node_states.entry(id.to_string()).or_default(); + st.status = N_SUCCESS.to_string(); + st.output = Some(outputs.clone()); + snap.outputs = Some(outputs); + snap.status = S_SUCCESS.to_string(); +} + +fn fail(snap: &mut Snapshot, id: &str, msg: String) { + let st = snap.node_states.entry(id.to_string()).or_default(); + st.status = N_FAILED.to_string(); + st.error = Some(json!({"message": msg})); + snap.error = Some(json!({"node_id": id, "message": msg})); + snap.status = S_FAILED.to_string(); +} + +/// Non-branch node: every outgoing edge is taken; then targets become ready. +fn fan_out_after_run( + graph: &Graph, + snap: &mut Snapshot, + id: &str, + queue: &mut VecDeque, +) -> AppResult<()> { + let Some(idx) = graph.out_edges.get(id) else { + return Ok(()); + }; + for &ei in idx { + mark(graph, snap, ei, EdgeMark::Taken); + } + for &ei in idx { + consider_target(graph, snap, &graph.edges[ei].target, queue)?; + } + Ok(()) +} + +/// Branch node: only the chosen handle edges are taken; others skipped. +fn fan_out_after_branch( + graph: &Graph, + snap: &mut Snapshot, + id: &str, + handle: Option<&str>, + queue: &mut VecDeque, +) -> AppResult<()> { + let Some(idx) = graph.out_edges.get(id) else { + return Ok(()); + }; + for &ei in idx { + let edge = &graph.edges[ei]; + let taken = Some(edge.source_handle.as_str()) == handle; + mark( + graph, + snap, + ei, + if taken { + EdgeMark::Taken + } else { + EdgeMark::Skipped + }, + ); + } + for &ei in idx { + consider_target(graph, snap, &graph.edges[ei].target, queue)?; + } + Ok(()) +} + +fn mark(graph: &Graph, snap: &mut Snapshot, ei: usize, mark: EdgeMark) { + snap.edge_marks.insert(edge_key(&graph.edges[ei]), mark); +} + +/// After an incoming edge of `target` was decided: if all decided → ready +/// (any taken) or skipped (all skipped, propagated). +fn consider_target( + graph: &Graph, + snap: &mut Snapshot, + target: &str, + queue: &mut VecDeque, +) -> AppResult<()> { + let Some(in_edges) = graph.in_edges.get(target) else { + return Ok(()); + }; + let decided = in_edges + .iter() + .filter(|ei| snap.edge_marks.contains_key(&edge_key(&graph.edges[**ei]))) + .count(); + if decided < in_edges.len() { + return Ok(()); // still waiting on other branches (join) + } + let any_taken = in_edges.iter().any(|ei| { + matches!( + snap.edge_marks.get(&edge_key(&graph.edges[*ei])), + Some(EdgeMark::Taken) + ) + }); + if any_taken { + if !queue.contains(&target.to_string()) { + queue.push_back(target.to_string()); + } + } else { + skip_node(graph, snap, target, queue)?; + } + Ok(()) +} + +/// Mark a node skipped and propagate down (skip = its edges are Skipped). +fn skip_node( + graph: &Graph, + snap: &mut Snapshot, + id: &str, + queue: &mut VecDeque, +) -> AppResult<()> { + let st = snap.node_states.entry(id.to_string()).or_default(); + st.status = N_SKIPPED.to_string(); + let Some(idx) = graph.out_edges.get(id) else { + return Ok(()); + }; + for &ei in idx { + mark(graph, snap, ei, EdgeMark::Skipped); + } + for &ei in idx { + consider_target(graph, snap, &graph.edges[ei].target, queue)?; + } + Ok(()) +} + +/// Replay fan-out for a node already completed in a persisted snapshot (resume). +fn resume_completed( + graph: &Graph, + snap: &mut Snapshot, + id: &str, + queue: &mut VecDeque, +) -> AppResult<()> { + let Some(node) = graph.nodes.get(id) else { + return Ok(()); + }; + match node.data.kind.as_str() { + nodes::T_BRANCH => { + let handle = snap + .node_states + .get(id) + .and_then(|s| s.output.as_ref()) + .and_then(|o| o.get("handle")) + .and_then(Value::as_str) + .map(str::to_string); + fan_out_after_branch(graph, snap, id, handle.as_deref(), queue)?; + } + nodes::T_END => {} + _ => { + fan_out_after_run(graph, snap, id, queue)?; + } + } + Ok(()) +} + +// ── value resolution (literal/ref; expr → P1.7) ────────────────────────── + +fn resolve(raw: &Value, pool: &Pool) -> AppResult { + if let Some(arr) = raw.get("ref").and_then(Value::as_array) { + return resolve_ref(arr, pool); + } + if let Some(v) = raw.get("literal") { + return Ok(v.clone()); + } + if raw.get("expr").is_some() { + return Err(AppError::BadRequest("expr 求值尚未接线（P1.7）".into())); + } + Ok(raw.clone()) +} + +fn resolve_ref(sel: &[Value], pool: &Pool) -> AppResult { + if sel.len() < 2 { + return Err(AppError::BadRequest("ref 需至少 [namespace, name]".into())); + } + let ns = sel[0].as_str().unwrap_or_default(); + let name = sel[1].as_str().unwrap_or_default(); + let mut v = pool + .get(ns) + .and_then(|m| m.get(name)) + .cloned() + .ok_or_else(|| AppError::BadRequest(format!("ref 引用不存在: {ns}.{name}")))?; + for part in sel.iter().skip(2) { + let key = part + .as_str() + .ok_or_else(|| AppError::BadRequest("ref 子路径元素必须是字符串".into()))?; + v = v + .get(key) + .cloned() + .ok_or_else(|| AppError::BadRequest(format!("ref 子路径不存在: {key}")))?; + } + Ok(v) +} + +fn resolve_inputs(config: &Value, pool: &Pool) -> AppResult { + let mut out = serde_json::Map::new(); + if let Some(input) = config.get("input").and_then(Value::as_object) { + for (k, v) in input { + out.insert(k.clone(), resolve(v, pool)?); + } + } + Ok(Value::Object(out)) +} + +fn resolve_end_outputs(node: &GraphNode, snap: &Snapshot) -> AppResult { + let cfg: EndConfig = serde_json::from_value(node.data.config.clone()) + .map_err(|e| AppError::BadRequest(format!("end config: {e}")))?; + let mut out = serde_json::Map::new(); + for o in &cfg.outputs { + out.insert(o.name.clone(), resolve(&o.value, &snap.pool)?); + } + Ok(Value::Object(out)) +} + +fn pick_branch(cfg: &BranchConfig, pool: &Pool) -> AppResult { + for rule in &cfg.branches { + let matched = eval_condition(&rule.when, pool)?; + if matched { + return Ok(rule.handle.clone().unwrap_or_default()); + } + } + Ok(cfg.else_handle.clone().unwrap_or_default()) +} + +/// Structured condition `{op, var, value}` or a literal bool. Expression +/// strings (`{{#…#}} >= 3`) require the P1.7 evaluator. +fn eval_condition(when: &Value, pool: &Pool) -> AppResult { + if let Some(op) = when.get("op").and_then(Value::as_str) { + let left = resolve(when.get("var").unwrap_or(&Value::Null), pool)?; + let right = when.get("value").cloned().unwrap_or(Value::Null); + return eval_op(op, &left, &right); + } + if let Some(b) = when.as_bool() { + return Ok(b); + } + if let Some(s) = when.as_str() { + return super::expr::eval_bool(s, pool); + } + Ok(false) +} + +fn eval_op(op: &str, left: &Value, right: &Value) -> AppResult { + let num_cmp = |o: &str| -> Option { + let l = left.as_f64()?; + let r = right.as_f64()?; + Some(match o { + ">" => l > r, + ">=" => l >= r, + "<" => l < r, + "<=" => l <= r, + "==" => l == r, + "!=" => l != r, + _ => return None, + }) + }; + let r = match op { + "==" => equalish(left, right), + "!=" => !equalish(left, right), + "in" => right + .as_array() + .map(|a| a.iter().any(|x| equalish(x, left))) + .unwrap_or(false), + "contains" => left + .as_str() + .map(|s| s.contains(right.as_str().unwrap_or_default())) + .unwrap_or(false), + "starts_with" => left + .as_str() + .map(|s| s.starts_with(right.as_str().unwrap_or_default())) + .unwrap_or(false), + "ends_with" => left + .as_str() + .map(|s| s.ends_with(right.as_str().unwrap_or_default())) + .unwrap_or(false), + "and" | "or" | "not" => { + return Err(AppError::BadRequest(format!( + "组合条件 '{op}' 待 P1.7 递归支持" + ))); + } + _ => num_cmp(op).unwrap_or(false), + }; + Ok(r) +} + +fn equalish(a: &Value, b: &Value) -> bool { + match (a.as_f64(), b.as_f64()) { + (Some(x), Some(y)) => x == y, + _ => a == b, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + struct StubExec; + #[async_trait] + impl NodeExecutor for StubExec { + async fn exec(&self, _node: &GraphNode, _input: Value) -> AppResult { + Ok(ExecOutcome { + output: json!({"stub": true}), + }) + } + } + + /// Executor that fails the first `fail_until` calls, then succeeds. + struct FlakyExec { + calls: std::sync::atomic::AtomicUsize, + fail_until: usize, + } + #[async_trait] + impl NodeExecutor for FlakyExec { + async fn exec(&self, _node: &GraphNode, _input: Value) -> AppResult { + let n = self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + if n < self.fail_until { + Err(AppError::BadRequest("flaky boom".into())) + } else { + Ok(ExecOutcome { + output: json!({"stub": true}), + }) + } + } + } + + fn graph_of(def: Value) -> Graph { + super::super::graph::load_definition(&def).unwrap() + } + fn def(nodes: Value, edges: Value) -> Value { + json!({"name":"t","graph":{"nodes":nodes,"edges":edges}}) + } + + fn node(id: &str, kind: &str, config: Value) -> Value { + json!({"id": id, "data": {"type": kind, "config": config}}) + } + fn edge(s: &str, h: &str, t: &str) -> Value { + json!({"source": s, "sourceHandle": h, "target": t}) + } + + #[tokio::test] + async fn linear_end_outputs_resolved() { + let g = graph_of(def( + json!([ + node("start", "start", json!({})), + node( + "end", + "end", + json!({"outputs": [{"name": "answer", "value": {"ref": ["start", "msg"]}}]}) + ) + ]), + json!([edge("start", "out", "end")]), + )); + let mut snap = Snapshot::new(); + let mut start_input = HashMap::new(); + start_input.insert("msg".into(), json!("hi")); + snap.pool.insert("start".into(), start_input); + run(&g, &mut snap, &StubExec).await.unwrap(); + assert_eq!(snap.status, S_SUCCESS); + assert_eq!(snap.outputs.unwrap()["answer"], "hi"); + assert_eq!(snap.node_states["end"].status, N_SUCCESS); + } + + #[tokio::test] + async fn branch_false_skips_true_path() { + let g = graph_of(def( + json!([ + node("start", "start", json!({})), + node( + "br", + "branch", + json!({ + "branches": [{"id": "b1", "when": {"op": "==", "var": ["start", "msg"], "value": "hi"}, "handle": "true"}], + "else_handle": "false" + }) + ), + node("na", "end", json!({"outputs": []})), + node("nb", "end", json!({"outputs": []})) + ]), + json!([ + edge("start", "out", "br"), + edge("br", "true", "na"), + edge("br", "false", "nb") + ]), + )); + let mut snap = Snapshot::new(); + let mut start_input = HashMap::new(); + start_input.insert("msg".into(), json!("bye")); + snap.pool.insert("start".into(), start_input); + run(&g, &mut snap, &StubExec).await.unwrap(); + assert_eq!(snap.status, S_SUCCESS); + assert_eq!(snap.node_states["na"].status, N_SKIPPED); + assert_eq!(snap.node_states["nb"].status, N_SUCCESS); + } + + #[tokio::test] + async fn branch_expr_string_when() { + let g = graph_of(def( + json!([ + node("start", "start", json!({})), + node( + "br", + "branch", + json!({ + "branches": [{"id": "b1", "when": "{{#start.msg#}} == \"hi\"", "handle": "true"}], + "else_handle": "false" + }) + ), + node("na", "end", json!({"outputs": []})), + node("nb", "end", json!({"outputs": []})) + ]), + json!([ + edge("start", "out", "br"), + edge("br", "true", "na"), + edge("br", "false", "nb") + ]), + )); + let mut snap = Snapshot::new(); + let mut start_input = HashMap::new(); + start_input.insert("msg".into(), json!("hi")); + snap.pool.insert("start".into(), start_input); + run(&g, &mut snap, &StubExec).await.unwrap(); + assert_eq!(snap.status, S_SUCCESS); + assert_eq!(snap.node_states["na"].status, N_SUCCESS); + assert_eq!(snap.node_states["nb"].status, N_SKIPPED); + } + + #[tokio::test] + async fn join_waits_for_both_branches() { + let g = graph_of(def( + json!([ + node("start", "start", json!({})), + node("e1", "egress", json!({"client_key": "k", "op": "o"})), + node("e2", "egress", json!({"client_key": "k", "op": "o"})), + node( + "end", + "end", + json!({"outputs": [{"name": "v", "value": {"ref": ["e2", "output"]}}]}) + ) + ]), + json!([ + edge("start", "out", "e1"), + edge("start", "out", "e2"), + edge("e1", "out", "end"), + edge("e2", "out", "end") + ]), + )); + let mut snap = Snapshot::new(); + snap.pool.insert("start".into(), HashMap::new()); + run(&g, &mut snap, &StubExec).await.unwrap(); + assert_eq!(snap.status, S_SUCCESS); + assert_eq!(snap.node_states["e1"].status, N_SUCCESS); + assert_eq!(snap.node_states["e2"].status, N_SUCCESS); + assert_eq!(snap.outputs.unwrap()["v"]["stub"], true); + } + + fn node_m(id: &str, kind: &str, config: Value, mods: Value) -> Value { + json!({"id": id, "data": {"type": kind, "config": config, "modifiers": mods}}) + } + + #[tokio::test] + async fn retry_recovers_after_failure() { + let g = graph_of(def( + json!([ + node("start", "start", json!({})), + node_m( + "e1", + "egress", + json!({"client_key": "k", "op": "o"}), + json!({"retry": {"attempts": 3}}) + ), + node("end", "end", json!({})) + ]), + json!([edge("start", "out", "e1"), edge("e1", "out", "end")]), + )); + let mut snap = Snapshot::new(); + snap.pool.insert("start".into(), HashMap::new()); + let flaky = FlakyExec { + calls: Default::default(), + fail_until: 1, + }; + run(&g, &mut snap, &flaky).await.unwrap(); + assert_eq!(snap.status, S_SUCCESS); + assert_eq!(snap.node_states["e1"].status, N_SUCCESS); + assert_eq!(snap.node_states["e1"].attempt, 2, "one fail + one success"); + } + + #[tokio::test] + async fn continue_on_error_passes_through() { + let g = graph_of(def( + json!([ + node("start", "start", json!({})), + node_m( + "e1", + "egress", + json!({"client_key": "k", "op": "o"}), + json!({"continue_on_error": true}) + ), + node("end", "end", json!({"outputs": []})) + ]), + json!([edge("start", "out", "e1"), edge("e1", "out", "end")]), + )); + let mut snap = Snapshot::new(); + snap.pool.insert("start".into(), HashMap::new()); + let always_fail = FlakyExec { + calls: Default::default(), + fail_until: usize::MAX, + }; + run(&g, &mut snap, &always_fail).await.unwrap(); + assert_eq!( + snap.status, S_SUCCESS, + "run passes through on continue_on_error" + ); + assert_eq!(snap.node_states["e1"].status, N_FAILED); + assert_eq!(snap.node_states["end"].status, N_SUCCESS); + } + + #[tokio::test] + async fn await_parks_then_resume_continues() { + let g = graph_of(def( + json!([ + node("start", "start", json!({})), + node("gate", "await", json!({"kind": "human"})), + node( + "end", + "end", + json!({ + "outputs": [{"name": "ok", "value": {"ref": ["gate", "resume", "approved"]}}] + }) + ) + ]), + json!([edge("start", "out", "gate"), edge("gate", "out", "end")]), + )); + let mut snap = Snapshot::new(); + snap.pool.insert("start".into(), HashMap::new()); + run(&g, &mut snap, &StubExec).await.unwrap(); + assert_eq!(snap.status, S_WAITING); + assert_eq!(snap.node_states["gate"].status, N_WAITING); + assert!(snap.waiting_nodes.contains(&"gate".to_string())); + + // Resume: complete the gate with an approval payload. + resume_snapshot(&mut snap, Some(json!({"approved": true}))).unwrap(); + assert_eq!(snap.status, S_RUNNING); + run(&g, &mut snap, &StubExec).await.unwrap(); + assert_eq!(snap.status, S_SUCCESS); + assert_eq!(snap.node_states["gate"].status, N_SUCCESS); + assert_eq!(snap.node_states["end"].status, N_SUCCESS); + assert_eq!(snap.outputs.unwrap()["ok"], true); + } +} diff --git a/crates/core/src/flows/exec.rs b/crates/core/src/flows/exec.rs new file mode 100644 index 00000000..d18def7d --- /dev/null +++ b/crates/core/src/flows/exec.rs @@ -0,0 +1,184 @@ +//! Production node executors wired to the integration plane + plugin engines. +//! +//! [`FlowsExec`] implements [`super::engine::NodeExecutor`]: +//! - `egress` → `IntegrationPlane.call_api` (OAuth2/signing/rate-limit/log/trace) +//! - `script` → `PluginManager::run_inline_script_value` — runs the node code as +//! a one-shot inline **script** (not an installed plugin) on the JS engine, +//! reusing the same sandboxed host (`host.callApi` gated by an egress +//! allowlist derived from `host_permissions`). Zero state leakage: load → +//! call → unload per invocation. + +use std::sync::Arc; + +use async_trait::async_trait; +use serde_json::Value; + +use crate::errors::app_error::{AppError, AppResult}; +use crate::integration::IntegrationPlane; +use crate::plugins::{Permissions, PluginManager}; + +use super::engine::{ExecOutcome, NodeExecutor}; +use super::graph::GraphNode; +use super::nodes::{self, EgressConfig, ScriptConfig}; + +pub struct FlowsExec { + pub plane: Option>, + pub plugins: Option>, +} + +impl FlowsExec { + async fn run_script(&self, node: &GraphNode, input: Value) -> AppResult { + let cfg: ScriptConfig = serde_json::from_value(node.data.config.clone()) + .map_err(|e| AppError::BadRequest(format!("script config: {e}")))?; + if cfg.language != "js" { + return Err(AppError::BadRequest(format!( + "script language '{}' 暂未接入（v1 支持 js）", + cfg.language + ))); + } + let Some(plugins) = &self.plugins else { + return Err(AppError::Internal(anyhow::anyhow!( + "plugin manager disabled" + ))); + }; + + // Permissions derived from the node's declared host surface. egress = + // allowlist of api-clients (empty = denied); everything else denied by + // default. No raw http/fs/db unless declared. + let mut perms = Permissions::default(); + if let Some(hp) = &cfg.host_permissions + && let Some(list) = &hp.call_api + { + perms.egress = list.clone(); + } + if let Some(sb) = &cfg.sandbox { + perms.max_memory_mb = sb.memory_mb.filter(|&m| m > 0).map(|m| m as u32); + perms.timeout_ms = sb.timeout_ms.filter(|&t| t > 0).map(|t| t as u64); + } + + // Authoring contract (same as cron inline scripts): the code IS a full + // exported function. It receives the input as a JSON string and may + // return any value (engine stringifies it back). + if cfg.plugin_id.is_some() { + return Err(AppError::BadRequest( + "script: plugin_id 引用未接入（v1 用 code）".into(), + )); + } + if cfg.code.trim().is_empty() { + return Err(AppError::BadRequest("script: code 为空".into())); + } + // Unique id per invocation: no KV bleed across concurrent instances. + let id = format!("__wf__{}", crate::utils::id::new_snowflake_id()); + + let out = plugins + .run_inline_script_value("js", &id, &cfg.code, "main", &input, perms) + .await?; + Ok(ExecOutcome { output: out }) + } +} + +#[async_trait] +impl NodeExecutor for FlowsExec { + async fn exec(&self, node: &GraphNode, input: Value) -> AppResult { + match node.data.kind.as_str() { + nodes::T_SCRIPT => self.run_script(node, input).await, + nodes::T_EGRESS => { + let cfg: EgressConfig = serde_json::from_value(node.data.config.clone()) + .map_err(|e| AppError::BadRequest(format!("egress config: {e}")))?; + let Some(plane) = &self.plane else { + return Err(AppError::Internal(anyhow::anyhow!( + "integration plane disabled" + ))); + }; + let receipt = plane + .call_api(&cfg.client_key, &cfg.op, input) + .await + .map_err(|e| { + AppError::Internal(anyhow::anyhow!( + "egress '{}'.{} failed: {e}", + cfg.client_key, + cfg.op + )) + })?; + Ok(ExecOutcome { + output: receipt.output, + }) + } + other => Err(AppError::BadRequest(format!( + "FlowsExec 不支持节点 '{other}'" + ))), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::flows::graph::NodeData; + use serde_json::json; + + fn node(kind: &str, config: Value) -> GraphNode { + GraphNode { + id: "n1".into(), + data: NodeData { + kind: kind.into(), + version: 1, + title: "".into(), + desc: None, + config, + modifiers: Value::Null, + }, + } + } + + #[tokio::test] + async fn egress_without_plane_errors_clearly() { + let exec = FlowsExec { + plane: None, + plugins: None, + }; + let err = exec + .exec( + &node(nodes::T_EGRESS, json!({"client_key": "k", "op": "o"})), + json!({}), + ) + .await + .unwrap_err(); + assert!(err.to_string().contains("integration plane disabled")); + } + + #[tokio::test] + async fn script_without_plugins_errors_clearly() { + let exec = FlowsExec { + plane: None, + plugins: None, + }; + let err = exec + .exec( + &node( + nodes::T_SCRIPT, + json!({"language": "js", "code": "return 1"}), + ), + json!({}), + ) + .await + .unwrap_err(); + assert!(err.to_string().contains("plugin manager disabled")); + } + + #[tokio::test] + async fn script_non_js_rejected() { + let exec = FlowsExec { + plane: None, + plugins: None, + }; + let err = exec + .exec( + &node(nodes::T_SCRIPT, json!({"language": "rhai", "code": "1"})), + json!({}), + ) + .await + .unwrap_err(); + assert!(err.to_string().contains("暂未接入")); + } +} diff --git a/crates/core/src/flows/expr.rs b/crates/core/src/flows/expr.rs new file mode 100644 index 00000000..f2ca0c31 --- /dev/null +++ b/crates/core/src/flows/expr.rs @@ -0,0 +1,364 @@ +//! Minimal value/template & safe-expression evaluation (contracts.md C3). +//! +//! Pure-Rust, language-agnostic. v1 subset: +//! - refs `{{#ns.name.child#}}` (whole-string → typed value) +//! - template interpolation in strings (inline `{{#…#}}`) +//! - expressions: numbers / strings / booleans + `+ - * / % > >= < <= == != +//! && || ! ( )` (no free function calls — structured conditions or a script +//! node cover richer logic). + +use std::collections::HashMap; + +use serde_json::Value; + +use crate::errors::app_error::{AppError, AppResult}; + +use super::engine::Pool; + +fn resolve_ref_in_pool(pool: &Pool, sel: &[&str]) -> AppResult { + if sel.len() < 2 { + return Err(AppError::BadRequest("ref 需至少 [namespace, name]".into())); + } + let ns = sel[0]; + let name = sel[1]; + let mut v = pool + .get(ns) + .and_then(|m| m.get(name)) + .cloned() + .ok_or_else(|| AppError::BadRequest(format!("ref 引用不存在: {ns}.{name}")))?; + for part in &sel[2..] { + v = v + .get(part) + .cloned() + .ok_or_else(|| AppError::BadRequest(format!("ref 子路径不存在: {part}")))?; + } + Ok(v) +} + +/// Positions + inner selector of `{{#sel#}}` tokens in a string. +fn find_tokens(text: &str) -> Vec<(usize, usize, String)> { + let mut out = Vec::new(); + let mut i = 0; + while i < text.len() { + if text[i..].starts_with("{{#") + && let Some(rel) = text[i + 3..].find("#}}") + { + let inner = &text[i + 3..i + 3 + rel]; + out.push((i, i + 3 + rel + 3, inner.to_string())); + i += 3 + rel + 3; + continue; + } + i += 1; + } + out +} + +/// Resolve a string possibly containing `{{#sel#}}`: a single whole-string +/// token returns the typed value; otherwise tokens are interpolated as text. +pub fn resolve_text(text: &str, pool: &Pool) -> AppResult { + let tokens = find_tokens(text); + if tokens.is_empty() { + return Ok(Value::String(text.to_string())); + } + let whole = tokens.len() == 1 && tokens[0].0 == 0 && tokens[0].1 == text.len(); + if whole { + let sel: Vec<&str> = tokens[0].2.split('.').collect(); + return resolve_ref_in_pool(pool, &sel); + } + let mut out = String::with_capacity(text.len()); + let mut cursor = 0; + for (start, end, inner) in tokens { + out.push_str(&text[cursor..start]); + let sel: Vec<&str> = inner.split('.').collect(); + let v = resolve_ref_in_pool(pool, &sel)?; + out.push_str(&scalar_text(&v)); + cursor = end; + } + out.push_str(&text[cursor..]); + Ok(Value::String(out)) +} + +fn scalar_text(v: &Value) -> String { + match v { + Value::String(s) => s.clone(), + Value::Number(n) => n.to_string(), + Value::Bool(b) => b.to_string(), + Value::Null => String::new(), + other => serde_json::to_string(other).unwrap_or_default(), + } +} + +/// Evaluate an expression string to a boolean. `{{#…#}}` refs are resolved to +/// literals before parsing (used by branch `when`/`skip_if`/`retry_if`). +pub fn eval_bool(expr: &str, pool: &Pool) -> AppResult { + let mut normalized = String::new(); + let mut rest = expr; + while let Some(start) = rest.find("{{#") { + normalized.push_str(&rest[..start]); + let after = &rest[start + 3..]; + let Some(end_rel) = after.find("#}}") else { + return Err(AppError::BadRequest(format!("表达式含未闭合引用: {expr}"))); + }; + let sel: Vec<&str> = after[..end_rel].split('.').collect(); + let v = resolve_ref_in_pool(pool, &sel)?; + normalized.push_str(&literal_syntax(&v)?); + rest = &after[end_rel + 3..]; + } + normalized.push_str(rest); + Parser::new(&normalized).parse() +} + +fn literal_syntax(v: &Value) -> AppResult { + Ok(match v { + Value::Number(n) => n.to_string(), + Value::Bool(b) => b.to_string(), + Value::String(s) => format!("{s:?}"), + other => { + return Err(AppError::BadRequest(format!( + "表达式引用不支持该类型: {other}" + ))); + } + }) +} + +struct Parser<'a> { + s: &'a str, + b: &'a [u8], + pos: usize, +} + +impl<'a> Parser<'a> { + fn new(s: &'a str) -> Self { + Self { + s, + b: s.as_bytes(), + pos: 0, + } + } + fn peek(&self) -> Option { + self.b.get(self.pos).copied().map(char::from) + } + fn skip_ws(&mut self) { + while self.pos < self.b.len() && (self.b[self.pos] as char).is_whitespace() { + self.pos += 1; + } + } + fn eat(&mut self, c: char) -> bool { + self.skip_ws(); + if self.peek() == Some(c) { + self.pos += 1; + true + } else { + false + } + } + fn take_op(&mut self, op: &str) -> bool { + self.skip_ws(); + if self.s[self.pos..].starts_with(op) { + self.pos += op.len(); + true + } else { + false + } + } + fn parse(&mut self) -> AppResult { + let v = self.parse_or()?; + self.skip_ws(); + if self.pos != self.b.len() { + return Err(AppError::BadRequest(format!( + "表达式尾部有多余内容: {}", + &self.s[self.pos..] + ))); + } + Ok(v) + } + fn parse_or(&mut self) -> AppResult { + let mut v = self.parse_and()?; + while self.take_op("||") { + v = v || self.parse_and()?; + } + Ok(v) + } + fn parse_and(&mut self) -> AppResult { + let mut v = self.parse_cmp()?; + while self.take_op("&&") { + v = v && self.parse_cmp()?; + } + Ok(v) + } + fn parse_cmp(&mut self) -> AppResult { + let left = self.parse_arith()?; + self.skip_ws(); + let op = if self.take_op("==") { + Some("==") + } else if self.take_op("!=") { + Some("!=") + } else if self.take_op(">=") { + Some(">=") + } else if self.take_op("<=") { + Some("<=") + } else if self.take_op(">") { + Some(">") + } else if self.take_op("<") { + Some("<") + } else { + None + }; + let Some(op) = op else { + return Ok(self.as_bool(&left)); + }; + let right = self.parse_arith()?; + Ok(cmp(op, &left, &right)) + } + fn parse_arith(&mut self) -> AppResult { + let mut v = self.parse_unary()?; + loop { + self.skip_ws(); + let c = self.peek(); + let op = match c { + Some('+') => Some('+'), + Some('-') => Some('-'), + Some('*') => Some('*'), + Some('/') => Some('/'), + Some('%') => Some('%'), + _ => None, + }; + let Some(op) = op else { return Ok(v) }; + self.pos += 1; + let r = self.parse_unary()?; + let (a, b) = (self.num(&v)?, self.num(&r)?); + v = Value::from(match op { + '+' => a + b, + '-' => a - b, + '*' => a * b, + '/' => a / b, + '%' => a % b, + _ => unreachable!(), + }); + } + } + fn parse_unary(&mut self) -> AppResult { + self.skip_ws(); + if self.eat('!') { + let v = self.parse_unary()?; + return Ok(Value::Bool(!self.as_bool(&v))); + } + if self.eat('(') { + let v = self.parse_or()?; + if !self.eat(')') { + return Err(AppError::BadRequest("表达式缺 ')'".into())); + } + return Ok(Value::Bool(v)); + } + self.parse_primary() + } + fn parse_primary(&mut self) -> AppResult { + self.skip_ws(); + let start = self.pos; + while self.pos < self.b.len() { + let c = self.b[self.pos] as char; + if c.is_whitespace() + || matches!( + c, + ')' | '>' | '<' | '=' | '!' | '&' | '|' | '+' | '-' | '*' | '/' | '%' | '(' + ) + { + break; + } + self.pos += 1; + } + let tok = &self.s[start..self.pos]; + if tok.is_empty() { + return Err(AppError::BadRequest("表达式意外结束".into())); + } + if tok == "true" { + return Ok(Value::Bool(true)); + } + if tok == "false" { + return Ok(Value::Bool(false)); + } + if let Some(q) = tok.chars().next() + && (q == '"' || q == '\'') + { + let inner = tok[1..].strip_suffix(q).unwrap_or(&tok[1..]); + return Ok(Value::String(inner.replace("\\\"", "\""))); + } + tok.parse::() + .map(Value::from) + .map_err(|_| AppError::BadRequest(format!("表达式无法解析: {tok}"))) + } + fn num(&self, v: &Value) -> AppResult { + v.as_f64() + .ok_or_else(|| AppError::BadRequest(format!("表达式需要数字，遇到 {v}"))) + } + fn as_bool(&self, v: &Value) -> bool { + v.as_bool() + .unwrap_or_else(|| !v.is_null() && v.as_f64().unwrap_or(0.0) != 0.0) + } +} + +fn cmp(op: &str, l: &Value, r: &Value) -> bool { + match (l.as_f64(), r.as_f64()) { + (Some(a), Some(b)) => match op { + ">" => a > b, + ">=" => a >= b, + "<" => a < b, + "<=" => a <= b, + "==" => a == b, + "!=" => a != b, + _ => false, + }, + _ => match op { + "==" => l == r, + "!=" => l != r, + _ => false, + }, + } +} + +/// Keep HashMap referenced so future token-cache work compiles cleanly. +#[allow(dead_code)] +fn _unused(_: HashMap) {} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn pool_of(ns: &str, name: &str, v: Value) -> Pool { + let mut p = Pool::new(); + let mut m = HashMap::new(); + m.insert(name.to_string(), v); + p.insert(ns.to_string(), m); + p + } + + #[test] + fn text_whole_ref_is_typed() { + let p = pool_of("start", "n", json!(42)); + assert_eq!(resolve_text("{{#start.n#}}", &p).unwrap(), json!(42)); + } + + #[test] + fn text_interpolates_inline() { + let p = pool_of("start", "name", json!("alice")); + assert_eq!( + resolve_text("hi {{#start.name#}}!", &p).unwrap(), + json!("hi alice!") + ); + } + + #[test] + fn bool_expr_with_refs() { + let p = pool_of("classify", "level", json!(5)); + assert!(eval_bool("{{#classify.level#}} >= 3", &p).unwrap()); + assert!(!eval_bool("{{#classify.level#}} >= 3 && false", &p).unwrap()); + } + + #[test] + fn bool_expr_string_eq() { + let p = pool_of("start", "s", json!("hi")); + assert!(eval_bool("{{#start.s#}} == \"hi\"", &p).unwrap()); + assert!(eval_bool("{{#start.s#}} != \"x\"", &p).unwrap()); + } +} diff --git a/crates/core/src/flows/graph.rs b/crates/core/src/flows/graph.rs new file mode 100644 index 00000000..4ec6f498 --- /dev/null +++ b/crates/core/src/flows/graph.rs @@ -0,0 +1,252 @@ +//! Graph loading & structural validation (flow-graph.md). +//! +//! Parses a flow definition's `graph` (ReactFlow canvas JSON) into a `Graph`: +//! unique node ids, edge endpoints exist, exactly one start (no incoming edge), +//! per-node `data.type/version` validated through the node registry. +//! Builds out-edges + in-count indexes ready for the P1.2 scheduler. + +use std::collections::HashMap; + +use serde::Deserialize; +use serde_json::Value; + +use crate::errors::app_error::{AppError, AppResult}; + +use super::nodes; + +/// Per-node engine data (`nodes[].data`): engine reads type/version/config. +#[derive(Debug, Clone, Deserialize)] +pub struct NodeData { + #[serde(rename = "type")] + pub kind: String, + #[serde(default = "default_version")] + pub version: i64, + #[serde(default)] + pub title: String, + #[serde(default)] + pub desc: Option, + #[serde(default)] + pub config: Value, + #[serde(default)] + pub modifiers: Value, +} + +fn default_version() -> i64 { + 1 +} + +#[derive(Debug, Clone)] +pub struct GraphNode { + pub id: String, + pub data: NodeData, +} + +/// Connection (edge) — ports default to `out` / `in`. +#[derive(Debug, Clone)] +pub struct Edge { + pub source: String, + pub source_handle: String, + pub target: String, + pub target_handle: String, +} + +/// Loaded, validated graph ready for execution. +pub struct Graph { + pub nodes: HashMap, + pub edges: Vec, + /// Edges by source node id (fan-out; branch handles among them). + pub out_edges: HashMap>, + /// Edge indices by target node id (join = all resolved). + pub in_edges: HashMap>, + /// Incoming edge count by target node id (join = all resolved). + pub in_count: HashMap, + pub start: String, +} + +fn read_node(n: &Value) -> AppResult<(String, NodeData)> { + let id = n + .get("id") + .and_then(Value::as_str) + .ok_or_else(|| AppError::BadRequest("node missing 'id'".into()))?; + let data: NodeData = serde_json::from_value( + n.get("data") + .cloned() + .unwrap_or_else(|| serde_json::json!({})), + ) + .map_err(|e| AppError::BadRequest(format!("node '{id}' data invalid: {e}")))?; + Ok((id.to_string(), data)) +} + +fn read_edge(e: &Value) -> AppResult { + let source = e + .get("source") + .and_then(Value::as_str) + .ok_or_else(|| AppError::BadRequest("edge missing 'source'".into()))?; + let target = e + .get("target") + .and_then(Value::as_str) + .ok_or_else(|| AppError::BadRequest("edge missing 'target'".into()))?; + Ok(Edge { + source: source.to_string(), + source_handle: e + .get("sourceHandle") + .and_then(Value::as_str) + .unwrap_or(nodes::H_OUT) + .to_string(), + target: target.to_string(), + target_handle: e + .get("targetHandle") + .and_then(Value::as_str) + .unwrap_or(nodes::H_IN) + .to_string(), + }) +} + +/// Load + validate a graph object `{nodes: [...], edges: [...]}`. +/// +/// # Errors +/// +/// `BadRequest` on unknown node type/version, malformed config, duplicate ids, +/// dangling edge, or not exactly one start with no incoming edges. +pub fn load_graph(graph_value: &Value) -> AppResult { + let nodes_val = graph_value + .get("nodes") + .and_then(Value::as_array) + .ok_or_else(|| AppError::BadRequest("graph missing 'nodes[]'".into()))?; + let edges_val = graph_value + .get("edges") + .and_then(Value::as_array) + .ok_or_else(|| AppError::BadRequest("graph missing 'edges[]'".into()))?; + + let mut nodes = HashMap::new(); + let mut start: Option = None; + for n in nodes_val { + let (id, data) = read_node(n)?; + if data.kind == "custom-note" { + continue; // canvas-only note nodes are ignored by the engine. + } + if nodes + .insert( + id.clone(), + GraphNode { + id: id.clone(), + data: data.clone(), + }, + ) + .is_some() + { + return Err(AppError::BadRequest(format!("duplicate node id '{id}'"))); + } + nodes::validate_node(&data.kind, data.version, &data.config)?; + if data.kind == nodes::T_START { + if start.is_some() { + return Err(AppError::BadRequest("v1 只允许一个 start 节点".into())); + } + start = Some(id.clone()); + } + } + let Some(start_id) = start else { + return Err(AppError::BadRequest("graph 缺少 start 节点".into())); + }; + + let mut edges = Vec::new(); + let mut out_edges: HashMap> = HashMap::new(); + let mut in_edges: HashMap> = HashMap::new(); + let mut in_count: HashMap = HashMap::new(); + for (i, e) in edges_val.iter().enumerate() { + let edge = read_edge(e)?; + if !nodes.contains_key(&edge.source) || !nodes.contains_key(&edge.target) { + return Err(AppError::BadRequest(format!( + "edge #{i} 引用不存在节点: {} -> {}", + edge.source, edge.target + ))); + } + if edge.target == start_id { + return Err(AppError::BadRequest("start 节点不允许入边".into())); + } + let edge_idx = edges.len(); + out_edges + .entry(edge.source.clone()) + .or_default() + .push(edge_idx); + in_edges + .entry(edge.target.clone()) + .or_default() + .push(edge_idx); + *in_count.entry(edge.target.clone()).or_insert(0) += 1; + edges.push(edge); + } + + Ok(Graph { + nodes, + edges, + out_edges, + in_edges, + in_count, + start: start_id, + }) +} + +/// Convenience: load from a full flow definition (`{... graph: {...}}`). +pub fn load_definition(def: &Value) -> AppResult { + let graph = def + .get("graph") + .ok_or_else(|| AppError::BadRequest("flow definition missing 'graph'".into()))?; + load_graph(graph) +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn sample() -> Value { + json!({ + "name": "test", + "graph": { + "nodes": [ + { "id": "start", "data": { "type": "start", "config": {} } }, + { "id": "s1", "data": { "type": "script", "config": {"language": "js", "code": "return {}"} } }, + { "id": "e1", "data": { "type": "egress", "config": {"client_key": "llm", "op": "chat"} } }, + { "id": "end", "data": { "type": "end", "config": {} } } + ], + "edges": [ + { "source": "start", "target": "s1" }, + { "source": "s1", "target": "e1" }, + { "source": "e1", "target": "end" } + ] + } + }) + } + + #[test] + fn loads_valid_graph() { + let g = load_definition(&sample()).unwrap(); + assert_eq!(g.start, "start"); + assert_eq!(g.nodes.len(), 4); + assert_eq!(g.edges.len(), 3); + assert_eq!(g.in_count["end"], 1); + assert_eq!(g.out_edges["start"].len(), 1); + } + + #[test] + fn rejects_dangling_edge() { + let mut v = sample(); + v["graph"]["edges"][0]["target"] = json!("missing"); + assert!(load_definition(&v).is_err()); + } + + #[test] + fn rejects_unknown_node_type() { + let mut v = sample(); + v["graph"]["nodes"][1]["data"]["type"] = json!("nope"); + assert!(load_definition(&v).is_err()); + } + + #[test] + fn rejects_two_starts() { + let mut v = sample(); + v["graph"]["nodes"][3]["data"]["type"] = json!("start"); + assert!(load_definition(&v).is_err()); + } +} diff --git a/crates/core/src/flows/handler.rs b/crates/core/src/flows/handler.rs new file mode 100644 index 00000000..95ec2e21 --- /dev/null +++ b/crates/core/src/flows/handler.rs @@ -0,0 +1,409 @@ +//! Flow engine HTTP handlers (P1.10 minimal admin surface). +//! +//! Endpoints: +//! - POST /admin/flows create flow + first version (definition) +//! - GET /admin/flows list flows (default tenant) +//! - GET /admin/flows/{id} flow detail (incl. latest version definition) +//! - PUT /admin/flows/{id} update metadata + publish new version +//! - DELETE /admin/flows/{id} cascade-delete flow + versions/instances +//! - POST /admin/flows/{id}/run create + execute an instance (inputs) +//! - GET /admin/flows/instances/{id} instance status/outputs +//! - POST /admin/flows/instances/{id}/stop mark canceled +//! +//! cron/event triggers + draft/publish (explicit) land in P2. + +use axum::Json; +use axum::extract::{Path, State}; +use serde::{Deserialize, Serialize}; +use serde_json::{Value, json}; + +use crate::AppState; +use crate::db::Pool; +use crate::errors::app_error::{AppError, AppResult}; +use crate::errors::response::ApiResponse; +use crate::types::snowflake_id::SnowflakeId; +use crate::utils::tz::Timestamp; + +use super::exec::FlowsExec; +use super::graph; +use super::model::{self, Flow, FlowInstance, FlowVersion}; + +fn now() -> Timestamp { + crate::utils::tz::now_utc() +} + +#[cfg_attr(feature = "export-types", derive(ts_rs::TS))] +#[derive(Debug, Deserialize)] +pub struct CreateFlowReq { + pub name: String, + #[serde(default)] + pub description: Option, + #[cfg_attr(feature = "export-types", ts(type = "unknown"))] + pub definition: Value, +} + +#[cfg_attr(feature = "export-types", derive(ts_rs::TS))] +#[derive(Debug, Deserialize)] +pub struct RunFlowReq { + #[serde(default)] + #[cfg_attr(feature = "export-types", ts(type = "unknown"))] + pub inputs: Option, +} + +#[cfg_attr(feature = "export-types", derive(ts_rs::TS))] +#[derive(Debug, Deserialize)] +pub struct UpdateFlowReq { + #[serde(default)] + pub name: Option, + #[serde(default)] + pub description: Option, + #[serde(default)] + #[cfg_attr(feature = "export-types", ts(type = "unknown"))] + pub definition: Option, +} + +#[cfg_attr(feature = "export-types", derive(ts_rs::TS))] +#[derive(Debug, Serialize)] +pub struct FlowDetail { + pub id: SnowflakeId, + pub tenant_id: String, + pub name: String, + pub description: Option, + pub enabled: bool, + #[cfg_attr(feature = "export-types", ts(type = "unknown"))] + pub definition: Option, + #[cfg_attr(feature = "export-types", ts(type = "number"))] + pub version_number: Option, + pub created_at: Timestamp, + pub updated_at: Timestamp, +} + +pub fn routes( + registry: &mut crate::server::RouteRegistry, + config: &crate::config::app::AppConfig, +) -> axum::Router { + let restful = config.api_restful; + let r = axum::Router::new(); + let r = reg_route!( + r, + registry, + restful, + "/admin/flows", + post, + create_flow, + "flows", + "admin/flows" + ); + let r = reg_route!( + r, + registry, + restful, + "/admin/flows", + get, + list_flows, + "flows", + "admin/flows" + ); + let r = reg_route!( + r, + registry, + restful, + "/admin/flows/{id}", + get, + get_flow, + "flows", + "admin/flows" + ); + let r = reg_route!( + r, + registry, + restful, + "/admin/flows/{id}", + put, + update_flow, + "flows", + "admin/flows" + ); + let r = reg_route!( + r, + registry, + restful, + "/admin/flows/{id}", + delete, + delete_flow, + "flows", + "admin/flows" + ); + let r = reg_route!( + r, + registry, + restful, + "/admin/flows/{id}/run", + post, + run_flow, + "flows", + "admin/flows" + ); + let r = reg_route!( + r, + registry, + restful, + "/admin/flows/instances/{id}", + get, + get_instance, + "flows", + "admin/flows" + ); + let r = reg_route!( + r, + registry, + restful, + "/admin/flows/instances/{id}/stop", + post, + stop_instance, + "flows", + "admin/flows" + ); + let r = reg_route!( + r, + registry, + restful, + "/admin/flows/instances/{id}/resume", + post, + resume_instance, + "flows", + "admin/flows" + ); + let _ = restful; + r +} + +#[cfg_attr(feature = "export-types", derive(ts_rs::TS))] +#[derive(Debug, Deserialize)] +pub struct ResumeReq { + #[serde(default)] + #[cfg_attr(feature = "export-types", ts(type = "unknown"))] + pub payload: Option, +} + +async fn resume_instance( + State(state): State, + Path(id): Path, + Json(req): Json, +) -> AppResult> { + let instance_id = parse_id(id)?; + super::run::resume_instance(&state.pool, instance_id, req.payload).await?; + let done = model::find_instance_by_id(&state.pool, instance_id).await?; + Ok(ApiResponse::success( + serde_json::to_value(&done).unwrap_or_default(), + )) +} + +/// Create a flow with its first published version (definition validated). +async fn create_flow( + State(state): State, + Json(req): Json, +) -> AppResult> { + // Validate graph structure + node configs up front. + graph::load_definition(&req.definition)?; + + let flow_id = crate::utils::id::new_snowflake_id(); + let now = now(); + let flow = Flow { + id: flow_id, + tenant_id: crate::constants::DEFAULT_TENANT.to_string(), + name: req.name, + description: req.description, + enabled: true, + current_version: None, + extra: None, + created_at: now, + updated_at: now, + }; + model::insert_flow(&state.pool, &flow).await?; + + let version_id = crate::utils::id::new_snowflake_id(); + let version = FlowVersion { + id: version_id, + flow_id, + version_number: 1, + definition: req.definition, + created_by: None, + created_at: now, + }; + model::insert_flow_version(&state.pool, &version).await?; + model::set_flow_current_version(&state.pool, flow_id, version_id).await?; + + Ok(ApiResponse::success(json!({ + "flow_id": flow_id, + "flow_version_id": version_id + }))) +} + +async fn list_flows(State(state): State) -> AppResult> { + let flows = model::find_flows_by_tenant(&state.pool, crate::constants::DEFAULT_TENANT).await?; + Ok(ApiResponse::success( + serde_json::to_value(flows).unwrap_or_default(), + )) +} + +/// Flow detail incl. the latest published version's definition (for the editor). +async fn get_flow( + State(state): State, + Path(id): Path, +) -> AppResult> { + let flow_id = parse_id(id)?; + let flow = model::find_flow_by_id(&state.pool, flow_id).await?; + let latest = model::latest_version(&state.pool, flow_id).await?; + Ok(ApiResponse::success( + serde_json::to_value(&FlowDetail { + id: flow.id, + tenant_id: flow.tenant_id, + name: flow.name, + description: flow.description, + enabled: flow.enabled, + definition: latest.as_ref().map(|v| v.definition.clone()), + version_number: latest.map(|v| v.version_number), + created_at: flow.created_at, + updated_at: flow.updated_at, + }) + .map_err(|e| AppError::Internal(anyhow::anyhow!("serialize flow detail: {e}")))?, + )) +} + +/// Update metadata (name/description) and/or publish a new version (definition). +async fn update_flow( + State(state): State, + Path(id): Path, + Json(req): Json, +) -> AppResult> { + let flow_id = parse_id(id)?; + let flow = model::find_flow_by_id(&state.pool, flow_id).await?; + + let name = req.name.unwrap_or(flow.name); + let description = req.description.or(flow.description); + model::update_flow_meta(&state.pool, flow_id, &name, description.as_deref()).await?; + + if let Some(definition) = req.definition { + graph::load_definition(&definition)?; + let next = model::latest_version(&state.pool, flow_id) + .await? + .map(|v| v.version_number + 1) + .unwrap_or(1); + let version = FlowVersion { + id: crate::utils::id::new_snowflake_id(), + flow_id, + version_number: next, + definition, + created_by: None, + created_at: now(), + }; + model::insert_flow_version(&state.pool, &version).await?; + model::set_flow_current_version(&state.pool, flow_id, version.id).await?; + } + + let fresh = model::find_flow_by_id(&state.pool, flow_id).await?; + let latest = model::latest_version(&state.pool, flow_id).await?; + Ok(ApiResponse::success( + serde_json::to_value(&FlowDetail { + id: fresh.id, + tenant_id: fresh.tenant_id, + name: fresh.name, + description: fresh.description, + enabled: fresh.enabled, + definition: latest.as_ref().map(|v| v.definition.clone()), + version_number: latest.map(|v| v.version_number), + created_at: fresh.created_at, + updated_at: fresh.updated_at, + }) + .map_err(|e| AppError::Internal(anyhow::anyhow!("serialize flow detail: {e}")))?, + )) +} + +/// Cascade-delete a flow and all its versions/instances. +async fn delete_flow( + State(state): State, + Path(id): Path, +) -> AppResult> { + let flow_id = parse_id(id)?; + model::delete_flow(&state.pool, flow_id).await?; + Ok(ApiResponse::success(json!({"deleted": true}))) +} + +/// Create an instance from the flow's current version and execute it. +async fn run_flow( + State(state): State, + Path(id): Path, + Json(req): Json, +) -> AppResult> { + let flow_id = crate::types::snowflake_id::parse_id(&id)?; + let flow = model::find_flow_by_id(&state.pool, flow_id).await?; + let version = model::latest_version(&state.pool, flow_id) + .await? + .ok_or_else(|| AppError::not_found("flow_version"))?; + // Definitions validated at create; validate this snapshot too. + graph::load_definition(&version.definition)?; + + let instance_id = crate::utils::id::new_snowflake_id(); + let now = now(); + let instance = FlowInstance { + id: instance_id, + tenant_id: flow.tenant_id.clone(), + flow_id, + flow_version_id: version.id, + status: "running".into(), + has_exceptions: false, + trigger_kind: "api".into(), + trigger_payload: req.inputs, + inputs_summary: None, + outputs: None, + error: None, + started_by: None, + started_at: Some(now), + finished_at: None, + waiting_kind: None, + waiting_needed: None, + waiting_received: 0, + resume_until: None, + created_at: now, + }; + model::insert_flow_instance(&state.pool, &instance).await?; + + let exec = FlowsExec { + plane: state.integration.clone(), + plugins: Some(state.plugins.clone()), + }; + super::run::execute_instance(&state.pool, instance_id, &exec).await?; + + let done = model::find_instance_by_id(&state.pool, instance_id).await?; + Ok(ApiResponse::success( + serde_json::to_value(&done).unwrap_or_default(), + )) +} + +async fn get_instance( + State(state): State, + Path(id): Path, +) -> AppResult> { + let instance_id = parse_id(id)?; + let instance = model::find_instance_by_id(&state.pool, instance_id).await?; + Ok(ApiResponse::success( + serde_json::to_value(&instance).unwrap_or_default(), + )) +} + +async fn stop_instance( + State(state): State, + Path(id): Path, +) -> AppResult> { + let instance_id = parse_id(id)?; + model::finalize_instance(&state.pool, instance_id, "canceled", false, None, None).await?; + Ok(ApiResponse::success(json!({"ok": true}))) +} + +fn parse_id(s: String) -> AppResult { + crate::types::snowflake_id::parse_id(&s) +} + +#[allow(dead_code)] +async fn _noop(_: Pool) {} diff --git a/crates/core/src/flows/mod.rs b/crates/core/src/flows/mod.rs new file mode 100644 index 00000000..8c6f0884 --- /dev/null +++ b/crates/core/src/flows/mod.rs @@ -0,0 +1,51 @@ +//! Flow orchestration engine v2 (dev-docs/workflow). +//! +//! RaisFast 内置动态逻辑编排引擎：图 DAG + 串行执行 + 整棵快照 durable。 +//! 深度集成自家资产：host 脚本（四引擎裁剪 host）/ egress / vault+租户 / worker 队列 / +//! eventbus / CT。数据自建独立命名空间（不写 itg_* 表）。 +//! +//! 规划子模块（随 P0.5+ 依次落地，此处按文档地图预留）： +//! - `model` —— flow/flow_version/flow_instance/flow_instance_snapshot/flow_node_run +//! 模型 + repo（db-schema.md） +//! - `graph` —— 图定义加载/校验（flow-graph.md / contracts.md C1-C3） +//! - `nodes` —— 节点注册表 + 各节点实现（start/end/script/egress/branch/await） +//! - `engine` —— 串行执行器 + Join/skip + 快照推进 + 错误分层（execution-engine.md） +//! - `handler`—— admin/run/resume/SSE API（contracts.md C4） +//! +//! 设计文档：`dev-docs/workflow/`（README 为地图，contracts.md 为冻结契约，db-schema.md 为表）。 + +pub mod engine; +pub mod exec; +pub mod expr; +pub mod graph; +pub mod handler; +pub mod model; +pub mod nodes; +pub mod params; +pub mod run; + +#[cfg(feature = "export-types")] +crate::export_types!( + model::Flow, + model::FlowVersion, + model::FlowInstance, + nodes::StartParam, + nodes::StartConfig, + nodes::EndOutput, + nodes::EndConfig, + nodes::SandboxLimits, + nodes::HostPermissions, + nodes::ScriptConfig, + nodes::EgressConfig, + nodes::BranchRule, + nodes::BranchConfig, + nodes::AwaitConfig, + nodes::NodeConfigVariant, + nodes::ValueExpr, + nodes::NodeKind, + handler::CreateFlowReq, + handler::RunFlowReq, + handler::ResumeReq, + handler::UpdateFlowReq, + handler::FlowDetail, +); diff --git a/crates/core/src/flows/model/flow.rs b/crates/core/src/flows/model/flow.rs new file mode 100644 index 00000000..f9fba826 --- /dev/null +++ b/crates/core/src/flows/model/flow.rs @@ -0,0 +1,153 @@ +//! flow model + queries (dev-docs/workflow db-schema.md). +//! Flow row = workflow metadata; `current_version` → published snapshot. +use serde::Serialize; +use serde_json::Value; + +use crate::db::{DbDriver, Driver}; +use crate::errors::app_error::{AppError, AppResult}; +use crate::types::snowflake_id::SnowflakeId; +use crate::utils::tz::Timestamp; + +const FLOW_COLS: &str = "id, tenant_id, name, description, enabled, current_version, extra, \ + created_at, updated_at"; + +pub(crate) fn row_not_found(table: &str) -> AppError { + AppError::not_found(table) +} +#[cfg_attr(feature = "export-types", derive(ts_rs::TS))] +#[derive(Debug, Clone, Serialize, sqlx::FromRow)] +pub struct Flow { + pub id: SnowflakeId, + pub tenant_id: String, + pub name: String, + pub description: Option, + pub enabled: bool, + pub current_version: Option, + #[cfg_attr(feature = "export-types", ts(type = "unknown"))] + #[cfg_attr(feature = "export-types", ts(type = "unknown"))] + pub extra: Option, + pub created_at: Timestamp, + pub updated_at: Timestamp, +} + +pub async fn insert_flow(pool: &crate::db::Pool, f: &Flow) -> AppResult<()> { + let now = crate::utils::tz::now_utc(); + let sql = format!( + "INSERT INTO flow ({FLOW_COLS}) VALUES ({}, {}, {}, {}, {}, {}, {}, {}, {})", + Driver::ph(1), + Driver::ph(2), + Driver::ph(3), + Driver::ph(4), + Driver::ph(5), + Driver::ph(6), + Driver::ph(7), + Driver::ph(8), + Driver::ph(9) + ); + sqlx::query(crate::db::safe_sql(&sql)) + .bind(*f.id) + .bind(&f.tenant_id) + .bind(&f.name) + .bind(&f.description) + .bind(f.enabled) + .bind(f.current_version) + .bind(&f.extra) + .bind(now) + .bind(now) + .execute(pool) + .await?; + Ok(()) +} + +pub async fn find_flow_by_id(pool: &crate::db::Pool, id: SnowflakeId) -> AppResult { + let sql = format!("SELECT {FLOW_COLS} FROM flow WHERE id = {}", Driver::ph(1)); + sqlx::query_as::(crate::db::safe_sql(&sql)) + .bind(*id) + .fetch_optional(pool) + .await? + .ok_or_else(|| row_not_found("flow")) +} + +pub async fn find_flows_by_tenant(pool: &crate::db::Pool, tenant_id: &str) -> AppResult> { + let sql = format!( + "SELECT {FLOW_COLS} FROM flow WHERE tenant_id = {} ORDER BY id", + Driver::ph(1) + ); + Ok( + sqlx::query_as::(crate::db::safe_sql(&sql)) + .bind(tenant_id) + .fetch_all(pool) + .await?, + ) +} + +/// Update the mutable metadata columns of a flow row (name/description). +pub async fn update_flow_meta( + pool: &crate::db::Pool, + flow_id: SnowflakeId, + name: &str, + description: Option<&str>, +) -> AppResult<()> { + let sql = format!( + "UPDATE flow SET name = {}, description = {}, updated_at = {} WHERE id = {}", + Driver::ph(1), + Driver::ph(2), + Driver::ph(3), + Driver::ph(4) + ); + sqlx::query(crate::db::safe_sql(&sql)) + .bind(name) + .bind(description) + .bind(crate::utils::tz::now_utc()) + .bind(*flow_id) + .execute(pool) + .await?; + Ok(()) +} + +/// Point `flow.current_version` at the newly published version. +pub async fn set_flow_current_version( + pool: &crate::db::Pool, + flow_id: SnowflakeId, + version_id: SnowflakeId, +) -> AppResult<()> { + let sql = format!( + "UPDATE flow SET current_version = {}, updated_at = {} WHERE id = {}", + Driver::ph(1), + Driver::ph(2), + Driver::ph(3) + ); + sqlx::query(crate::db::safe_sql(&sql)) + .bind(*version_id) + .bind(crate::utils::tz::now_utc()) + .bind(*flow_id) + .execute(pool) + .await?; + Ok(()) +} + +/// Cascade-delete a flow and everything it owns: versions, instances plus +/// their durable snapshots and per-node run history. Cross-database safe +/// (subselect on SQLite/PG/MySQL). +pub async fn delete_flow(pool: &crate::db::Pool, flow_id: SnowflakeId) -> AppResult<()> { + let p1 = Driver::ph(1); + let cascade = [ + format!( + "DELETE FROM flow_instance_snapshot WHERE instance_id IN \ + (SELECT id FROM flow_instance WHERE flow_id = {p1})" + ), + format!( + "DELETE FROM flow_node_run WHERE instance_id IN (SELECT id FROM flow_instance WHERE flow_id = {p1})" + ), + format!("DELETE FROM flow_instance WHERE flow_id = {p1}"), + format!("DELETE FROM flow_version WHERE flow_id = {p1}"), + format!("DELETE FROM flow WHERE id = {p1}"), + ]; + for sql in cascade { + sqlx::query(crate::db::safe_sql(&sql)) + .bind(*flow_id) + .execute(pool) + .await?; + } + Ok(()) +} diff --git a/crates/core/src/flows/model/flow_instance.rs b/crates/core/src/flows/model/flow_instance.rs new file mode 100644 index 00000000..2acd6066 --- /dev/null +++ b/crates/core/src/flows/model/flow_instance.rs @@ -0,0 +1,151 @@ +//! flow_instance model + queries (dev-docs/workflow db-schema.md). +//! One run (wf_trace root = id); waiting fields denormalized. +use serde::Serialize; +use serde_json::Value; + +use crate::db::{DbDriver, Driver}; +use crate::errors::app_error::{AppError, AppResult}; +use crate::types::snowflake_id::SnowflakeId; +use crate::utils::tz::Timestamp; + +const FLOW_INSTANCE_COLS: &str = "id, tenant_id, flow_id, flow_version_id, status, \ + has_exceptions, trigger_kind, trigger_payload, inputs_summary, outputs, error, started_by, \ + started_at, finished_at, waiting_kind, waiting_needed, waiting_received, resume_until, \ + created_at"; + +pub(crate) fn row_not_found(table: &str) -> AppError { + AppError::not_found(table) +} +#[cfg_attr(feature = "export-types", derive(ts_rs::TS))] +#[derive(Debug, Clone, Serialize, sqlx::FromRow)] +pub struct FlowInstance { + pub id: SnowflakeId, + pub tenant_id: String, + pub flow_id: SnowflakeId, + pub flow_version_id: SnowflakeId, + pub status: String, + pub has_exceptions: bool, + pub trigger_kind: String, + #[cfg_attr(feature = "export-types", ts(type = "unknown"))] + pub trigger_payload: Option, + #[cfg_attr(feature = "export-types", ts(type = "unknown"))] + pub inputs_summary: Option, + #[cfg_attr(feature = "export-types", ts(type = "unknown"))] + pub outputs: Option, + #[cfg_attr(feature = "export-types", ts(type = "unknown"))] + pub error: Option, + pub started_by: Option, + pub started_at: Option, + pub finished_at: Option, + pub waiting_kind: Option, + #[cfg_attr(feature = "export-types", ts(type = "number"))] + pub waiting_needed: Option, + #[cfg_attr(feature = "export-types", ts(type = "number"))] + pub waiting_received: i64, + pub resume_until: Option, + pub created_at: Timestamp, +} + +pub async fn insert_flow_instance(pool: &crate::db::Pool, i: &FlowInstance) -> AppResult<()> { + let ph = (1..=19).map(Driver::ph).collect::>().join(", "); + let sql = format!("INSERT INTO flow_instance ({FLOW_INSTANCE_COLS}) VALUES ({ph})"); + sqlx::query(crate::db::safe_sql(&sql)) + .bind(*i.id) + .bind(&i.tenant_id) + .bind(*i.flow_id) + .bind(*i.flow_version_id) + .bind(&i.status) + .bind(i.has_exceptions) + .bind(&i.trigger_kind) + .bind(&i.trigger_payload) + .bind(&i.inputs_summary) + .bind(&i.outputs) + .bind(&i.error) + .bind(i.started_by) + .bind(i.started_at) + .bind(i.finished_at) + .bind(&i.waiting_kind) + .bind(i.waiting_needed) + .bind(i.waiting_received) + .bind(i.resume_until) + .bind(i.created_at) + .execute(pool) + .await?; + Ok(()) +} + +pub async fn find_instance_by_id( + pool: &crate::db::Pool, + id: SnowflakeId, +) -> AppResult { + let sql = format!( + "SELECT {FLOW_INSTANCE_COLS} FROM flow_instance WHERE id = {}", + Driver::ph(1) + ); + sqlx::query_as::(crate::db::safe_sql(&sql)) + .bind(*id) + .fetch_optional(pool) + .await? + .ok_or_else(|| row_not_found("flow_instance")) +} + +/// Update instance terminal/running status + optional finished_at. +#[allow(clippy::too_many_arguments)] +pub async fn update_instance_status( + pool: &crate::db::Pool, + id: SnowflakeId, + status: &str, + has_exceptions: bool, + error: Option<&Value>, + finished_at: Option, +) -> AppResult<()> { + let sql = format!( + "UPDATE flow_instance SET status = {}, has_exceptions = {}, error = {}, \ + finished_at = {}, waiting_kind = NULL WHERE id = {}", + Driver::ph(1), + Driver::ph(2), + Driver::ph(3), + Driver::ph(4), + Driver::ph(5) + ); + sqlx::query(crate::db::safe_sql(&sql)) + .bind(status) + .bind(has_exceptions) + .bind(error) + .bind(finished_at) + .bind(*id) + .execute(pool) + .await?; + Ok(()) +} + +/// Finalize a finished instance (success/failed): outputs + error + finished_at. +pub async fn finalize_instance( + pool: &crate::db::Pool, + id: SnowflakeId, + status: &str, + has_exceptions: bool, + outputs: Option<&Value>, + error: Option<&Value>, +) -> AppResult<()> { + let sql = format!( + "UPDATE flow_instance SET status = {}, has_exceptions = {}, outputs = {}, error = {}, \ + finished_at = {}, waiting_kind = NULL WHERE id = {}", + Driver::ph(1), + Driver::ph(2), + Driver::ph(3), + Driver::ph(4), + Driver::ph(5), + Driver::ph(6) + ); + sqlx::query(crate::db::safe_sql(&sql)) + .bind(status) + .bind(has_exceptions) + .bind(outputs) + .bind(error) + .bind(crate::utils::tz::now_utc()) + .bind(*id) + .execute(pool) + .await?; + Ok(()) +} diff --git a/crates/core/src/flows/model/flow_instance_snapshot.rs b/crates/core/src/flows/model/flow_instance_snapshot.rs new file mode 100644 index 00000000..ed47dc7e --- /dev/null +++ b/crates/core/src/flows/model/flow_instance_snapshot.rs @@ -0,0 +1,62 @@ +//! flow_instance_snapshot model + queries (dev-docs/workflow db-schema.md). +//! Durable whole-snapshot (1:1 per instance, single writer). +use serde_json::Value; + +use crate::db::{DbDriver, Driver}; +use crate::errors::app_error::AppResult; +use crate::types::snowflake_id::SnowflakeId; + +pub async fn upsert_snapshot( + pool: &crate::db::Pool, + instance_id: SnowflakeId, + snapshot: &Value, +) -> AppResult<()> { + let assignments = format!( + "snapshot = {}, snapshot_version = 1, updated_at = {}", + Driver::excluded_col("snapshot"), + Driver::excluded_col("updated_at"), + ); + let sql = format!( + "INSERT INTO flow_instance_snapshot (instance_id, snapshot, snapshot_version, updated_at) \ + VALUES ({}, {}, 1, {}) {}", + Driver::ph(1), + Driver::ph(2), + Driver::ph(3), + Driver::upsert_clause("instance_id", &assignments), + ); + sqlx::query(crate::db::safe_sql(&sql)) + .bind(*instance_id) + .bind(snapshot) + .bind(crate::utils::tz::now_utc()) + .execute(pool) + .await?; + Ok(()) +} + +pub async fn find_snapshot( + pool: &crate::db::Pool, + instance_id: SnowflakeId, +) -> AppResult> { + let sql = format!( + "SELECT snapshot FROM flow_instance_snapshot WHERE instance_id = {}", + Driver::ph(1) + ); + let row: Option = sqlx::query_scalar(crate::db::safe_sql(&sql)) + .bind(*instance_id) + .fetch_optional(pool) + .await?; + Ok(row) +} + +/// Delete the snapshot row (instance terminal / cleanup). +pub async fn delete_snapshot(pool: &crate::db::Pool, instance_id: SnowflakeId) -> AppResult<()> { + let sql = format!( + "DELETE FROM flow_instance_snapshot WHERE instance_id = {}", + Driver::ph(1) + ); + sqlx::query(crate::db::safe_sql(&sql)) + .bind(*instance_id) + .execute(pool) + .await?; + Ok(()) +} diff --git a/crates/core/src/flows/model/flow_version.rs b/crates/core/src/flows/model/flow_version.rs new file mode 100644 index 00000000..bdf798b7 --- /dev/null +++ b/crates/core/src/flows/model/flow_version.rs @@ -0,0 +1,80 @@ +//! flow_version model + queries (dev-docs/workflow db-schema.md). +//! Immutable definition snapshots; publish = append. +use serde::Serialize; +use serde_json::Value; + +use crate::db::{DbDriver, Driver}; +use crate::errors::app_error::AppResult; +use crate::types::snowflake_id::SnowflakeId; +use crate::utils::tz::Timestamp; + +const FLOW_VERSION_COLS: &str = "id, flow_id, version_number, definition, created_by, created_at"; + +#[cfg_attr(feature = "export-types", derive(ts_rs::TS))] +#[derive(Debug, Clone, Serialize, sqlx::FromRow)] +pub struct FlowVersion { + pub id: SnowflakeId, + pub flow_id: SnowflakeId, + #[cfg_attr(feature = "export-types", ts(type = "number"))] + pub version_number: i64, + #[cfg_attr(feature = "export-types", ts(type = "unknown"))] + pub definition: Value, + pub created_by: Option, + pub created_at: Timestamp, +} + +pub async fn insert_flow_version(pool: &crate::db::Pool, v: &FlowVersion) -> AppResult<()> { + let sql = format!( + "INSERT INTO flow_version ({FLOW_VERSION_COLS}) VALUES ({}, {}, {}, {}, {}, {})", + Driver::ph(1), + Driver::ph(2), + Driver::ph(3), + Driver::ph(4), + Driver::ph(5), + Driver::ph(6) + ); + sqlx::query(crate::db::safe_sql(&sql)) + .bind(*v.id) + .bind(*v.flow_id) + .bind(v.version_number) + .bind(&v.definition) + .bind(v.created_by) + .bind(v.created_at) + .execute(pool) + .await?; + Ok(()) +} + +/// Latest published version of a flow (None = not published yet). +pub async fn latest_version( + pool: &crate::db::Pool, + flow_id: SnowflakeId, +) -> AppResult> { + let sql = format!( + "SELECT {FLOW_VERSION_COLS} FROM flow_version WHERE flow_id = {} \ + ORDER BY version_number DESC LIMIT 1", + Driver::ph(1) + ); + Ok( + sqlx::query_as::(crate::db::safe_sql(&sql)) + .bind(*flow_id) + .fetch_optional(pool) + .await?, + ) +} + +pub async fn find_version_by_id( + pool: &crate::db::Pool, + id: SnowflakeId, +) -> AppResult> { + let sql = format!( + "SELECT {FLOW_VERSION_COLS} FROM flow_version WHERE id = {}", + Driver::ph(1) + ); + Ok( + sqlx::query_as::(crate::db::safe_sql(&sql)) + .bind(*id) + .fetch_optional(pool) + .await?, + ) +} diff --git a/crates/core/src/flows/model/mod.rs b/crates/core/src/flows/model/mod.rs new file mode 100644 index 00000000..1bc087e6 --- /dev/null +++ b/crates/core/src/flows/model/mod.rs @@ -0,0 +1,120 @@ +//! Flow engine models — one file per table (dev-docs/workflow db-schema.md). +//! +//! - [`flow`] — flow metadata +//! - [`flow_version`] — immutable definition snapshots +//! - [`flow_instance`] — one run +//! - [`flow_instance_snapshot`] — durable whole-snapshot (1:1) +//! +//! (flow_node_run added with P1.9 observability.) + +pub mod flow; +pub mod flow_instance; +pub mod flow_instance_snapshot; +pub mod flow_version; + +pub use flow::{ + Flow, delete_flow, find_flow_by_id, find_flows_by_tenant, insert_flow, + set_flow_current_version, update_flow_meta, +}; +pub use flow_instance::{ + FlowInstance, finalize_instance, find_instance_by_id, insert_flow_instance, + update_instance_status, +}; +pub use flow_instance_snapshot::{delete_snapshot, find_snapshot, upsert_snapshot}; +pub use flow_version::{FlowVersion, find_version_by_id, insert_flow_version, latest_version}; + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn now() -> crate::utils::tz::Timestamp { + crate::utils::tz::now_utc() + } + + #[tokio::test] + async fn flow_version_instance_roundtrip() { + let p = crate::test_pool!(); + let tenant = format!("wf-test-{}", crate::utils::id::new_id()); + + let flow = Flow { + id: crate::utils::id::new_snowflake_id(), + tenant_id: tenant.clone(), + name: "test flow".into(), + description: None, + enabled: true, + current_version: None, + extra: Some(json!({"tag": "x"})), + created_at: now(), + updated_at: now(), + }; + insert_flow(&p, &flow).await.unwrap(); + let got = find_flow_by_id(&p, flow.id).await.unwrap(); + assert_eq!(got.name, "test flow"); + assert_eq!(got.extra.unwrap()["tag"], "x"); + + let version = FlowVersion { + id: crate::utils::id::new_snowflake_id(), + flow_id: flow.id, + version_number: 1, + definition: json!({"nodes": []}), + created_by: None, + created_at: now(), + }; + insert_flow_version(&p, &version).await.unwrap(); + let latest = latest_version(&p, flow.id).await.unwrap().unwrap(); + set_flow_current_version(&p, flow.id, latest.id) + .await + .unwrap(); + assert_eq!( + find_flow_by_id(&p, flow.id) + .await + .unwrap() + .current_version + .unwrap(), + latest.id + ); + assert!(find_version_by_id(&p, latest.id).await.unwrap().is_some()); + + let instance = FlowInstance { + id: crate::utils::id::new_snowflake_id(), + tenant_id: tenant.clone(), + flow_id: flow.id, + flow_version_id: latest.id, + status: "running".into(), + has_exceptions: false, + trigger_kind: "api".into(), + trigger_payload: Some(json!({"msg": "hi"})), + inputs_summary: None, + outputs: None, + error: None, + started_by: None, + started_at: Some(now()), + finished_at: None, + waiting_kind: None, + waiting_needed: None, + waiting_received: 0, + resume_until: None, + created_at: now(), + }; + insert_flow_instance(&p, &instance).await.unwrap(); + let got_inst = find_instance_by_id(&p, instance.id).await.unwrap(); + assert_eq!(got_inst.status, "running"); + + update_instance_status(&p, instance.id, "success", false, None, Some(now())) + .await + .unwrap(); + assert_eq!( + find_instance_by_id(&p, instance.id).await.unwrap().status, + "success" + ); + + // snapshot upsert/find/delete + upsert_snapshot(&p, instance.id, &json!({"pool": {}})) + .await + .unwrap(); + assert!(find_snapshot(&p, instance.id).await.unwrap().is_some()); + delete_snapshot(&p, instance.id).await.unwrap(); + assert!(find_snapshot(&p, instance.id).await.unwrap().is_none()); + } +} diff --git a/crates/core/src/flows/nodes.rs b/crates/core/src/flows/nodes.rs new file mode 100644 index 00000000..0aeb1ea4 --- /dev/null +++ b/crates/core/src/flows/nodes.rs @@ -0,0 +1,337 @@ +//! Node registry + config schemas (contracts.md C1). +//! +//! v1 types: start/end/script/egress/branch (+ await config reserved for P2). +//! `validate_node(type, version, config)` deserializes config into a strong Rust +//! struct — shape errors surface as 400 here, not at runtime. Unknown keys are +//! tolerated (extra=allow) so frontend can carry display fields; required keys +//! and value shapes are enforced. + +use serde::Deserialize; +use serde_json::Value; + +use crate::errors::app_error::{AppError, AppResult}; + +/// Reserved handle names (contracts.md C1.3). +pub const H_IN: &str = "in"; +pub const H_OUT: &str = "out"; +pub const H_ERROR_OUT: &str = "error_out"; + +/// Known node types for v1. +pub const T_START: &str = "start"; +pub const T_END: &str = "end"; +pub const T_SCRIPT: &str = "script"; +pub const T_EGRESS: &str = "egress"; +pub const T_BRANCH: &str = "branch"; +pub const T_AWAIT: &str = "await"; + +#[cfg_attr(feature = "export-types", derive(ts_rs::TS))] +#[derive(Debug, Clone, Deserialize)] +pub struct StartParam { + pub name: String, + #[serde(default)] + pub label: String, + #[serde(rename = "type")] + pub kind: String, + #[serde(default)] + pub required: bool, + #[serde(default)] + #[cfg_attr(feature = "export-types", ts(type = "unknown"))] + pub default: Option, + #[serde(default)] + #[cfg_attr(feature = "export-types", ts(type = "number"))] + pub max_length: Option, + #[serde(default)] + #[cfg_attr(feature = "export-types", ts(type = "unknown"))] + pub options: Option>, +} + +#[cfg_attr(feature = "export-types", derive(ts_rs::TS))] +#[derive(Debug, Clone, Deserialize)] +pub struct StartConfig { + #[serde(default)] + pub params: Vec, +} + +#[cfg_attr(feature = "export-types", derive(ts_rs::TS))] +#[derive(Debug, Clone, Deserialize)] +pub struct EndOutput { + pub name: String, + #[serde(default)] + #[cfg_attr(feature = "export-types", ts(type = "unknown"))] + pub value: Value, +} + +#[cfg_attr(feature = "export-types", derive(ts_rs::TS))] +#[derive(Debug, Clone, Deserialize)] +pub struct EndConfig { + #[serde(default)] + pub outputs: Vec, +} + +#[cfg_attr(feature = "export-types", derive(ts_rs::TS))] +#[derive(Debug, Clone, Deserialize)] +pub struct SandboxLimits { + #[serde(default)] + #[cfg_attr(feature = "export-types", ts(type = "number"))] + pub timeout_ms: Option, + #[serde(default)] + #[cfg_attr(feature = "export-types", ts(type = "number"))] + pub memory_mb: Option, +} + +#[cfg_attr(feature = "export-types", derive(ts_rs::TS))] +#[derive(Debug, Clone, Deserialize)] +pub struct HostPermissions { + #[serde(default)] + pub call_api: Option>, + #[serde(default)] + pub data: Option, + #[serde(default)] + pub emit: Option, +} + +#[cfg_attr(feature = "export-types", derive(ts_rs::TS))] +#[derive(Debug, Clone, Deserialize)] +pub struct ScriptConfig { + pub language: String, + #[serde(default)] + pub code: String, + #[serde(default)] + pub plugin_id: Option, + #[serde(default)] + pub fn_name: Option, + #[serde(default)] + #[cfg_attr(feature = "export-types", ts(type = "unknown"))] + pub input: Option>, + #[serde(default)] + #[cfg_attr(feature = "export-types", ts(type = "unknown"))] + pub output_schema: Option>, + #[serde(default)] + pub sandbox: Option, + #[serde(default)] + pub host_permissions: Option, +} + +#[cfg_attr(feature = "export-types", derive(ts_rs::TS))] +#[derive(Debug, Clone, Deserialize)] +pub struct EgressConfig { + pub client_key: String, + #[serde(default)] + pub op: String, + #[serde(default)] + #[cfg_attr(feature = "export-types", ts(type = "unknown"))] + pub input: Option>, + #[serde(default)] + #[cfg_attr(feature = "export-types", ts(type = "unknown"))] + pub output_schema: Option>, + #[serde(default)] + pub response_field: Option, +} + +#[cfg_attr(feature = "export-types", derive(ts_rs::TS))] +#[derive(Debug, Clone, Deserialize)] +pub struct BranchRule { + #[serde(default)] + pub id: String, + #[serde(default)] + pub label: String, + /// Structured condition or expression string. + #[cfg_attr(feature = "export-types", ts(type = "unknown"))] + pub when: Value, + #[serde(default)] + pub handle: Option, +} + +#[cfg_attr(feature = "export-types", derive(ts_rs::TS))] +#[derive(Debug, Clone, Deserialize)] +pub struct BranchConfig { + #[serde(default)] + pub branches: Vec, + #[serde(default)] + pub else_handle: Option, +} + +#[cfg_attr(feature = "export-types", derive(ts_rs::TS))] +#[derive(Debug, Clone, Deserialize)] +pub struct AwaitConfig { + pub kind: String, + #[serde(default)] + #[cfg_attr(feature = "export-types", ts(type = "unknown"))] + pub form: Option, + #[serde(default)] + pub approvers: Option, + #[serde(default)] + #[cfg_attr(feature = "export-types", ts(type = "number"))] + pub timeout_secs: Option, + #[serde(default)] + #[cfg_attr(feature = "export-types", ts(type = "unknown"))] + pub events: Option>, +} + +/// Validate an input value: scalar (literal shorthand) OR exactly one of +/// `{literal|ref|expr}`. `ref` must be an array of strings. +pub fn validate_value_expr(where_: &str, v: &Value) -> AppResult<()> { + if v.is_object() { + let obj = v.as_object().unwrap(); + let keys: Vec<&String> = obj.keys().collect(); + let variant = keys + .iter() + .find(|k| k.as_str() == "literal" || k.as_str() == "ref" || k.as_str() == "expr"); + if let Some(k) = variant { + if keys.len() > 1 { + return Err(AppError::BadRequest(format!( + "{where_}: ValueExpr {v} 只能含一个键 (literal|ref|expr)" + ))); + } + if k.as_str() == "ref" { + let arr = obj.get("ref").and_then(Value::as_array).ok_or_else(|| { + AppError::BadRequest(format!("{where_}: ref 必须是字符串数组")) + })?; + if arr.iter().any(|s| !s.is_string()) { + return Err(AppError::BadRequest(format!( + "{where_}: ref 元素必须是字符串" + ))); + } + } + } + } + Ok(()) +} + +fn validate_input_map( + where_: &str, + input: Option<&serde_json::Map>, +) -> AppResult<()> { + if let Some(map) = input { + for (k, v) in map { + validate_value_expr(&format!("{where_}.input.{k}"), v)?; + } + } + Ok(()) +} + +/// Deserialize + validate a node config against its known schema. Unknown type +/// or version → `BadRequest`. +pub fn validate_node(kind: &str, _version: i64, config: &Value) -> AppResult<()> { + let type_error = + |e: serde_json::Error| AppError::BadRequest(format!("node '{kind}' config invalid: {e}")); + match kind { + T_START => { + let c: StartConfig = serde_json::from_value(config.clone()).map_err(type_error)?; + for p in &c.params { + if p.name.is_empty() { + return Err(AppError::BadRequest("start.params[].name 不能为空".into())); + } + } + } + T_END => { + let _c: EndConfig = serde_json::from_value(config.clone()).map_err(type_error)?; + } + T_SCRIPT => { + let c: ScriptConfig = serde_json::from_value(config.clone()).map_err(type_error)?; + if c.code.is_empty() && c.plugin_id.is_none() { + return Err(AppError::BadRequest( + "script: code 与 plugin_id 至少给一个".into(), + )); + } + validate_input_map("script", c.input.as_ref())?; + } + T_EGRESS => { + let c: EgressConfig = serde_json::from_value(config.clone()).map_err(type_error)?; + if c.client_key.is_empty() || c.op.is_empty() { + return Err(AppError::BadRequest("egress: client_key 与 op 必填".into())); + } + validate_input_map("egress", c.input.as_ref())?; + } + T_BRANCH => { + let c: BranchConfig = serde_json::from_value(config.clone()).map_err(type_error)?; + if c.branches.is_empty() { + return Err(AppError::BadRequest("branch: 至少一个 branches".into())); + } + } + T_AWAIT => { + let _c: AwaitConfig = serde_json::from_value(config.clone()).map_err(type_error)?; + } + other => { + return Err(AppError::BadRequest(format!( + "node type '{other}' not supported (v1: start|end|script|egress|branch)" + ))); + } + } + Ok(()) +} + +/// Node type string union (TS literal union; wire = `data.type`). +#[cfg_attr(feature = "export-types", derive(ts_rs::TS))] +#[cfg_attr(feature = "export-types", ts(rename_all = "lowercase"))] +#[allow(dead_code)] +pub enum NodeKind { + Start, + End, + Script, + Egress, + Branch, + Await, +} + +/// TS-only union of every node's config shape (editor drives panels off it). +/// Discriminator lives at `node.data.type`; config payloads are the inner +/// object (no tag inside). +#[cfg_attr(feature = "export-types", derive(ts_rs::TS))] +#[cfg_attr(feature = "export-types", ts(untagged))] +#[allow(dead_code)] +pub enum NodeConfigVariant { + Start(StartConfig), + End(EndConfig), + Script(ScriptConfig), + Egress(EgressConfig), + Branch(BranchConfig), + Await(AwaitConfig), +} + +/// TS-only union for ValueExpr (literal | ref selector | expr string). +#[cfg_attr(feature = "export-types", derive(ts_rs::TS))] +#[cfg_attr(feature = "export-types", ts(untagged))] +#[allow(dead_code)] +pub enum ValueExpr { + Literal { + #[cfg_attr(feature = "export-types", ts(type = "unknown"))] + literal: Value, + }, + Ref { + #[cfg_attr(feature = "export-types", ts(rename = "ref"))] + ref_: Vec, + }, + Expr { + expr: String, + }, +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn validates_known_types_and_rejects_unknown() { + assert!(validate_node(T_SCRIPT, 1, &json!({"language": "js", "code": "return 1"})).is_ok()); + assert!( + validate_node(T_SCRIPT, 1, &json!({"language": "js"})).is_err(), + "无 code/plugin" + ); + assert!(validate_node(T_EGRESS, 1, &json!({"client_key": "llm", "op": "chat"})).is_ok()); + assert!(validate_node("nope", 1, &json!({})).is_err(), "未知 type"); + } + + #[test] + fn value_expr_shapes() { + assert!(validate_value_expr("x", &json!({"ref": ["start", "msg"]})).is_ok()); + assert!(validate_value_expr("x", &json!({"literal": 5})).is_ok()); + assert!(validate_value_expr("x", &json!({"expr": "{{#start.msg#}}.length > 0"})).is_ok()); + assert!(validate_value_expr("x", &json!({"ref": "start"})).is_err()); + assert!( + validate_value_expr("x", &json!({"ref": ["a"], "literal": 1})).is_err(), + "多键" + ); + } +} diff --git a/crates/core/src/flows/params.rs b/crates/core/src/flows/params.rs new file mode 100644 index 00000000..b48309c2 --- /dev/null +++ b/crates/core/src/flows/params.rs @@ -0,0 +1,132 @@ +//! Start-parameter validation & defaulting (contracts C1.4b, Dify-inspired). +//! +//! Applied at run entry before seeding the pool: `required` missing → error, +//! missing optional → default, then type checks (text/paragraph/select/number/ +//! boolean/…). Callers mutate the start-namespace map in place. + +use std::collections::HashMap; + +use serde_json::Value; + +use crate::errors::app_error::{AppError, AppResult}; + +use super::nodes::StartParam; + +/// Validate + default `inputs` against the declared start params. +pub fn apply(params: &[StartParam], inputs: &mut HashMap) -> AppResult<()> { + for p in params { + let present = inputs.contains_key(&p.name); + if !present { + match (&p.default, p.required) { + (Some(d), _) => { + inputs.insert(p.name.clone(), d.clone()); + } + (None, true) => { + return Err(AppError::BadRequest(format!( + "start 入参 '{}' 必填", + p.name + ))); + } + (None, false) => continue, + } + } + let v = inputs.get_mut(&p.name).unwrap(); + validate_one(p, v)?; + } + Ok(()) +} + +fn validate_one(p: &StartParam, v: &mut Value) -> AppResult<()> { + match p.kind.as_str() { + "text-input" | "paragraph" | "select" => { + let s = v + .as_str() + .ok_or_else(|| AppError::BadRequest(format!("入参 '{}' 需为字符串", p.name)))?; + if let Some(max) = p.max_length + && (s.len() as i64) > max + { + return Err(AppError::BadRequest(format!( + "入参 '{}' 超过 max_length={max}", + p.name + ))); + } + if p.kind == "select" + && let Some(opts) = &p.options + && !opts.iter().any(|o| o.as_str() == Some(s)) + { + return Err(AppError::BadRequest(format!( + "入参 '{}' 不在 options 内", + p.name + ))); + } + } + "number" => { + if let Some(s) = v.as_str() { + let n = s + .parse::() + .map_err(|_| AppError::BadRequest(format!("入参 '{}' 不是数字", p.name)))?; + *v = serde_json::json!(n); + } else if !v.is_number() { + return Err(AppError::BadRequest(format!("入参 '{}' 需为数字", p.name))); + } + } + "boolean" => { + if !v.is_boolean() { + let b = match v.as_str() { + Some(s) => matches!(s, "true" | "1" | "on"), + None => { + return Err(AppError::BadRequest(format!("入参 '{}' 需为布尔", p.name))); + } + }; + *v = Value::Bool(b); + } + } + "json" if !v.is_object() => { + return Err(AppError::BadRequest(format!("入参 '{}' 需为对象", p.name))); + } + "json" => {} + _ => {} + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn sp(name: &str, kind: &str, required: bool, default: Option) -> StartParam { + StartParam { + name: name.into(), + label: String::new(), + kind: kind.into(), + required, + default, + max_length: None, + options: None, + } + } + + #[test] + fn required_missing_errors() { + let p = [sp("msg", "paragraph", true, None)]; + let mut m = HashMap::new(); + assert!(apply(&p, &mut m).is_err()); + } + + #[test] + fn default_filled_and_number_coerced() { + let p = [ + sp("level", "number", false, Some(json!(2))), + sp("msg", "text-input", true, None), + ]; + let mut m = HashMap::new(); + m.insert("msg".into(), json!("hi")); + apply(&p, &mut m).unwrap(); + assert_eq!(m["level"], 2); + // numeric string coerced + m.insert("level".into(), json!("5")); + apply(&p, &mut m).unwrap(); + assert_eq!(m["level"], 5.0); + } +} diff --git a/crates/core/src/flows/run.rs b/crates/core/src/flows/run.rs new file mode 100644 index 00000000..a6dac76b --- /dev/null +++ b/crates/core/src/flows/run.rs @@ -0,0 +1,332 @@ +//! Instance coordinator: DB-backed execution with per-node durable snapshots +//! (dev-docs/workflow db-schema.md §5, reliability A.3). +//! +//! Pipeline: load instance+version → build/seed snapshot → engine `run_persisted` +//! with a DB [`Persist`] (claim + every node completion → upsert snapshot) → +//! finalize instance status/outputs. Completed nodes never re-run on resume +//! (snapshot carries their `success` state). + +use async_trait::async_trait; + +use crate::errors::app_error::{AppError, AppResult}; +use crate::types::snowflake_id::SnowflakeId; + +use super::engine::{self, NodeExecutor, Persist, S_FAILED, S_SUCCESS, S_WAITING, Snapshot}; +use super::graph::{self, Graph}; +use super::model; + +/// Persist the snapshot to `flow_instance_snapshot` (1:1 upsert). +struct DbPersist { + pool: crate::db::Pool, + instance_id: SnowflakeId, +} + +#[async_trait] +impl Persist for DbPersist { + async fn persist(&self, snap: &Snapshot) -> AppResult<()> { + let value = serde_json::to_value(snap) + .map_err(|e| AppError::Internal(anyhow::anyhow!("snapshot serialize: {e}")))?; + model::upsert_snapshot(&self.pool, self.instance_id, &value).await + } +} + +/// Load a graph from the instance's locked flow version. +async fn load_graph_for_instance( + pool: &crate::db::Pool, + inst: &model::FlowInstance, +) -> AppResult { + let version = model::find_version_by_id(pool, inst.flow_version_id) + .await? + .ok_or_else(|| AppError::not_found("flow_version"))?; + graph::load_definition(&version.definition) +} + +/// Seed the start-namespace pool from the instance trigger inputs. +fn seed_pool( + inputs: Option<&serde_json::Value>, +) -> std::collections::HashMap { + let mut ns = std::collections::HashMap::new(); + if let Some(obj) = inputs.and_then(|v| v.as_object()) { + for (k, v) in obj { + ns.insert(k.clone(), v.clone()); + } + } + ns +} + +/// Run an instance to completion (idempotent: terminals return early; completed +/// nodes in a persisted snapshot are skipped). +/// +/// # Errors +/// +/// `AppError` on load/persist failures; node failures are reflected in the +/// instance status (`failed`), not bubbled up. +pub async fn execute_instance( + pool: &crate::db::Pool, + instance_id: SnowflakeId, + exec: &dyn NodeExecutor, +) -> AppResult<()> { + let inst = model::find_instance_by_id(pool, instance_id).await?; + if matches!(inst.status.as_str(), S_SUCCESS | S_FAILED | "canceled") { + return Ok(()); // terminal + } + let graph = load_graph_for_instance(pool, &inst).await?; + + // Load persisted snapshot, or seed a fresh one. + let mut snap: Snapshot = match model::find_snapshot(pool, instance_id).await? { + Some(v) => serde_json::from_value(v) + .map_err(|e| AppError::Internal(anyhow::anyhow!("snapshot parse: {e}")))?, + None => { + let mut ns = seed_pool(inst.trigger_payload.as_ref()); + if let Some(start_node) = graph.nodes.get(&graph.start) { + let cfg: super::nodes::StartConfig = + serde_json::from_value(start_node.data.config.clone()) + .map_err(|e| AppError::BadRequest(format!("start config: {e}")))?; + super::params::apply(&cfg.params, &mut ns)?; + } + let mut s = Snapshot::new(); + s.pool.insert(graph.start.clone(), ns); + s + } + }; + + let persist = DbPersist { + pool: pool.clone(), + instance_id, + }; + engine::run_persisted(&graph, &mut snap, exec, &persist).await?; + + if snap.status == S_WAITING { + // Parked on an await node: keep the snapshot; a resume call continues. + model::update_instance_status(pool, instance_id, "waiting", false, None, None).await?; + return Ok(()); + } + if snap.status == S_SUCCESS { + model::finalize_instance( + pool, + instance_id, + S_SUCCESS, + false, + snap.outputs.as_ref(), + None, + ) + .await?; + // Keep the snapshot? Terminal success can drop it; keep for replay (P2). + } else if snap.status == S_FAILED { + model::finalize_instance( + pool, + instance_id, + S_FAILED, + false, + None, + snap.error.as_ref(), + ) + .await?; + } + Ok(()) +} + +/// Complete a parked `await` node with `payload` and continue the instance. +/// +/// # Errors +/// +/// `BadRequest` when the instance is not waiting on any node. +pub async fn resume_instance( + pool: &crate::db::Pool, + instance_id: SnowflakeId, + payload: Option, +) -> AppResult<()> { + let inst = model::find_instance_by_id(pool, instance_id).await?; + if inst.status != "waiting" { + return Err(AppError::BadRequest(format!( + "实例状态不是 waiting: {}", + inst.status + ))); + } + let Some(value) = model::find_snapshot(pool, instance_id).await? else { + return Err(AppError::BadRequest("实例无快照".into())); + }; + let mut snap: Snapshot = serde_json::from_value(value) + .map_err(|e| AppError::Internal(anyhow::anyhow!("snapshot parse: {e}")))?; + engine::resume_snapshot(&mut snap, payload)?; + let persist = DbPersist { + pool: pool.clone(), + instance_id, + }; + persist.persist(&snap).await?; + execute_instance(pool, instance_id, &NoopExec).await?; + Ok(()) +} + +struct NoopExec; +#[async_trait] +impl NodeExecutor for NoopExec { + async fn exec( + &self, + _node: &super::graph::GraphNode, + _input: serde_json::Value, + ) -> AppResult { + Ok(super::engine::ExecOutcome { + output: serde_json::Value::Null, + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::{Value, json}; + + struct StubExec; + #[async_trait] + impl NodeExecutor for StubExec { + async fn exec( + &self, + _node: &crate::flows::graph::GraphNode, + _input: Value, + ) -> AppResult { + Ok(engine::ExecOutcome { + output: json!({"stub": true}), + }) + } + } + + fn now() -> crate::utils::tz::Timestamp { + crate::utils::tz::now_utc() + } + + async fn seed_linear_flow(pool: &crate::db::Pool) -> SnowflakeId { + let flow_id = crate::utils::id::new_snowflake_id(); + let flow = model::Flow { + id: flow_id, + tenant_id: "default".into(), + name: "durable".into(), + description: None, + enabled: true, + current_version: None, + extra: None, + created_at: now(), + updated_at: now(), + }; + model::insert_flow(pool, &flow).await.unwrap(); + + let def = json!({ + "name": "durable", + "graph": { + "nodes": [ + {"id": "start", "data": {"type": "start", "config": {}}}, + {"id": "e1", "data": {"type": "egress", "config": {"client_key": "k", "op": "o"}}}, + {"id": "end", "data": {"type": "end", "config": {"outputs": [{"name": "v", "value": {"ref": ["e1", "output"]}}]}}} + ], + "edges": [ + {"source": "start", "target": "e1"}, + {"source": "e1", "target": "end"} + ] + } + }); + let version = model::FlowVersion { + id: crate::utils::id::new_snowflake_id(), + flow_id, + version_number: 1, + definition: def, + created_by: None, + created_at: now(), + }; + model::insert_flow_version(pool, &version).await.unwrap(); + model::set_flow_current_version(pool, flow_id, version.id) + .await + .unwrap(); + + let instance_id = crate::utils::id::new_snowflake_id(); + let inst = model::FlowInstance { + id: instance_id, + tenant_id: "default".into(), + flow_id, + flow_version_id: version.id, + status: "running".into(), + has_exceptions: false, + trigger_kind: "api".into(), + trigger_payload: Some(json!({"msg": "hi"})), + inputs_summary: None, + outputs: None, + error: None, + started_by: None, + started_at: Some(now()), + finished_at: None, + waiting_kind: None, + waiting_needed: None, + waiting_received: 0, + resume_until: None, + created_at: now(), + }; + model::insert_flow_instance(pool, &inst).await.unwrap(); + instance_id + } + + #[tokio::test] + async fn durable_run_finishes_and_resume_skips_completed() { + let pool = crate::test_pool!(); + let instance_id = seed_linear_flow(&pool).await; + + // First pass to completion. + execute_instance(&pool, instance_id, &StubExec) + .await + .unwrap(); + let inst = model::find_instance_by_id(&pool, instance_id) + .await + .unwrap(); + assert_eq!(inst.status, "success"); + assert_eq!(inst.outputs.unwrap()["v"]["stub"], true); + let snap_val = model::find_snapshot(&pool, instance_id) + .await + .unwrap() + .unwrap(); + let snap: Snapshot = serde_json::from_value(snap_val).unwrap(); + assert_eq!(snap.node_states["start"].status, engine::N_SUCCESS); + assert_eq!(snap.node_states["e1"].status, engine::N_SUCCESS); + + // Terminal → no-op on second call. + execute_instance(&pool, instance_id, &StubExec) + .await + .unwrap(); + } + + #[tokio::test] + async fn resume_does_not_rerun_completed_start() { + let pool = crate::test_pool!(); + let instance_id = seed_linear_flow(&pool).await; + + // Simulate crash after start completed: snapshot with start success, + // e1 undecided/untouched. + let mut snap = Snapshot::new(); + snap.node_states.insert( + "start".into(), + engine::NodeState { + status: engine::N_SUCCESS.into(), + output: Some(Value::Null), + error: None, + attempt: 1, + }, + ); + let snap_json = serde_json::to_value(&snap).unwrap(); + model::upsert_snapshot(&pool, instance_id, &snap_json) + .await + .unwrap(); + + execute_instance(&pool, instance_id, &StubExec) + .await + .unwrap(); + let inst = model::find_instance_by_id(&pool, instance_id) + .await + .unwrap(); + assert_eq!(inst.status, "success"); + let snap_val = model::find_snapshot(&pool, instance_id) + .await + .unwrap() + .unwrap(); + let snap2: Snapshot = serde_json::from_value(snap_val).unwrap(); + // start not re-run (attempt stays 1), e1 ran once. + assert_eq!(snap2.node_states["start"].attempt, 1); + assert_eq!(snap2.node_states["e1"].status, engine::N_SUCCESS); + } +} diff --git a/crates/core/src/lib.rs b/crates/core/src/lib.rs index 89570478..e06817e1 100644 --- a/crates/core/src/lib.rs +++ b/crates/core/src/lib.rs @@ -30,6 +30,7 @@ pub mod dto; pub mod errors; pub mod event; pub mod eventbus; +pub mod flows; pub mod graphql; pub mod handlers; pub mod integration; diff --git a/crates/core/src/plugins.rs b/crates/core/src/plugins.rs index b6e3861c..a09b8fca 100644 --- a/crates/core/src/plugins.rs +++ b/crates/core/src/plugins.rs @@ -1055,6 +1055,55 @@ impl PluginManager { Ok(()) } + /// Execute an inline script and capture its JSON return value (flow script + /// nodes). Same load/call/unload + zero-state-leakage model as + /// [`Self::run_inline_script`], but uses each engine's value-returning + /// filter path. The code must export `func_name`, which receives the input + /// as a JSON string and returns a JSON string. + /// + /// v1 supports `js`; lua/rhai return a clear not-wired error. + pub async fn run_inline_script_value( + &self, + runtime: &str, + id: &str, + code: &str, + func_name: &str, + data: &serde_json::Value, + permissions: Permissions, + ) -> AppResult { + match runtime { + #[cfg(feature = "plugin-js")] + "js" => { + let sdk_source = sdk_v1::get_sdk_source("js", "v1").ok_or_else(|| { + AppError::Internal(anyhow::anyhow!("SDK source for js/v1 not found")) + })?; + // `host` alias so inline scripts can use host.* without imports. + let wrapped = format!("const host = globalThis.RaisFastHost;\n{code}"); + self.js_engine + .load_plugin( + id, + &wrapped, + permissions, + std::path::Path::new("."), + sdk_source, + ) + .await + .map_err(|e| AppError::Internal(anyhow::anyhow!("load inline js: {e}")))?; + let result = self + .js_engine + .call_filter::(id, func_name, data, None) + .await + .map_err(|e| AppError::Internal(anyhow::anyhow!("exec inline js: {e}"))) + .map(|v| v.unwrap_or(serde_json::Value::Null)); + self.js_engine.unload_plugin(id).await; + result + } + other => Err(AppError::BadRequest(format!( + "script runtime '{other}' 尚未接入 flow（v1 支持 js）" + ))), + } + } + /// Sync plugin Cron schedules to the database async fn sync_crons_for_plugin(&self, plugin_id: &str, entries: &[CronEntry]) { if let Some(ref pool) = self.pool @@ -2283,6 +2332,32 @@ mod tests { assert_eq!(mgr.plugin_count().await, 0); } + #[cfg(feature = "plugin-js")] + #[tokio::test] + async fn inline_script_value_returns_and_denies_host_by_default() { + let config = test_config(); + let mgr = PluginManager::new(config).await; + let code = r#" +export function main(__in) { + const input = JSON.parse(__in || '{}'); + return { sum: input.a + input.b }; +} +"#; + let data = serde_json::json!({"a": 2, "b": 3}); + let out = mgr + .run_inline_script_value( + "js", + "__test_inline", + code, + "main", + &data, + Permissions::default(), + ) + .await + .unwrap(); + assert_eq!(out["sum"], 5, "script return value: {out}"); + } + #[tokio::test] async fn dispatch_filter_passthrough_with_no_plugins() { let config = test_config(); diff --git a/crates/core/src/server.rs b/crates/core/src/server.rs index f85bedd3..a630a1bd 100644 --- a/crates/core/src/server.rs +++ b/crates/core/src/server.rs @@ -256,6 +256,7 @@ async fn build_app( if config.builtins.workflow { api_v1 = api_v1.merge(workflow::handler::routes(&mut registry, config)); + api_v1 = api_v1.merge(crate::flows::handler::routes(&mut registry, config)); } api_v1 = crate::content_type::handler::register_content_routes( diff --git a/crates/core/tests/api.rs b/crates/core/tests/api.rs index 072299e5..d39f8597 100644 --- a/crates/core/tests/api.rs +++ b/crates/core/tests/api.rs @@ -897,6 +897,8 @@ mod category; mod comment; #[path = "api/cron.rs"] mod cron; +#[path = "api/flows/mod.rs"] +mod flows; #[path = "api/health.rs"] mod health; #[path = "api/media.rs"] @@ -7681,3 +7683,150 @@ type = "text" .output() .await; } + +#[tokio::test] +async fn flow_engine_egress_e2e_acceptance() { + // P1.11 acceptance: HTTP-free engine e2e with real plane + egress + plugins. + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(egress_mock_api(listener)); + + let (mut app, state) = test_app().await; + let _ = &mut app; + let plane = state.integration.as_ref().unwrap(); + + // api-client wired to the mock egress server (LLM-ish chat op). + let ops = json!({ + "chat": {"method": "POST", "path": "/v1/chat-messages", "output": {"text": "$.answer"}} + }); + let client = insert_egress_client( + &state, + plane, + &format!("wf-eg-{}", raisfast::utils::id::new_id()), + json!({"kind": "bearer"}), + Some("sk"), + ops, + None, + ) + .await; + let sql = format!( + "UPDATE itg_api_clients SET base_url = {} WHERE id = {}", + raisfast::db::Driver::ph(1), + raisfast::db::Driver::ph(2) + ); + sqlx::query(raisfast::db::safe_sql(&sql)) + .bind(format!("http://{addr}")) + .bind(*client.id) + .execute(&state.pool) + .await + .unwrap(); + + use raisfast::flows::exec::FlowsExec; + use raisfast::flows::model::{self, Flow, FlowInstance, FlowVersion}; + use raisfast::flows::run; + use raisfast::utils::tz::now_utc; + + let now = now_utc(); + let flow_id = raisfast::utils::id::new_snowflake_id(); + let flow = Flow { + id: flow_id, + tenant_id: "default".into(), + name: format!("wf-e2e-{}", raisfast::utils::id::new_id()), + description: None, + enabled: true, + current_version: None, + extra: None, + created_at: now, + updated_at: now, + }; + model::insert_flow(&state.pool, &flow).await.unwrap(); + + let def = json!({ + "name": "egress-e2e", + "graph": { + "nodes": [ + {"id": "start", "data": {"type": "start", "config": {}}}, + {"id": "e1", "data": {"type": "egress", "config": { + "client_key": client.client_key, + "op": "chat", + "input": {"query": {"literal": "hi"}} + }}}, + {"id": "end", "data": {"type": "end", "config": { + "outputs": [{"name": "ans", "value": {"ref": ["e1", "output", "text"]}}] + }}} + ], + "edges": [ + {"source": "start", "target": "e1"}, + {"source": "e1", "target": "end"} + ] + } + }); + let version_id = raisfast::utils::id::new_snowflake_id(); + let version = FlowVersion { + id: version_id, + flow_id, + version_number: 1, + definition: def, + created_by: None, + created_at: now, + }; + model::insert_flow_version(&state.pool, &version) + .await + .unwrap(); + model::set_flow_current_version(&state.pool, flow_id, version_id) + .await + .unwrap(); + + let instance_id = raisfast::utils::id::new_snowflake_id(); + let instance = FlowInstance { + id: instance_id, + tenant_id: "default".into(), + flow_id, + flow_version_id: version_id, + status: "running".into(), + has_exceptions: false, + trigger_kind: "api".into(), + trigger_payload: Some(json!({"msg": "hi"})), + inputs_summary: None, + outputs: None, + error: None, + started_by: None, + started_at: Some(now), + finished_at: None, + waiting_kind: None, + waiting_needed: None, + waiting_received: 0, + resume_until: None, + created_at: now, + }; + model::insert_flow_instance(&state.pool, &instance) + .await + .unwrap(); + + let exec = FlowsExec { + plane: Some(plane.clone()), + plugins: Some(state.plugins.clone()), + }; + run::execute_instance(&state.pool, instance_id, &exec) + .await + .unwrap(); + + let done = model::find_instance_by_id(&state.pool, instance_id) + .await + .unwrap(); + assert_eq!(done.status, "success", "instance failed: {done:?}"); + let outputs = done.outputs.unwrap(); + assert_eq!( + outputs["ans"], "mock-reply", + "real egress round-trip: {outputs}" + ); + + let snap = model::find_snapshot(&state.pool, instance_id) + .await + .unwrap() + .unwrap(); + assert!( + snap.get("node_states").is_some(), + "durable snapshot persisted" + ); +} diff --git a/crates/core/tests/api/flows/helpers.rs b/crates/core/tests/api/flows/helpers.rs new file mode 100644 index 00000000..ad4dab03 --- /dev/null +++ b/crates/core/tests/api/flows/helpers.rs @@ -0,0 +1,78 @@ +//! Shared builders for flow-engine scenario tests (pure/in-memory by default; +//! DB-backed durable scenarios call `super::super::test_pool`). + +use raisfast::flows::engine::{self, ExecOutcome, NodeExecutor, Snapshot}; +use raisfast::flows::graph::{self, Graph}; +use serde_json::{Value, json}; +use std::collections::HashMap; + +/// Build a `Graph` from a raw flow definition object. +pub fn def_graph(def: Value) -> Graph { + graph::load_definition(&def).expect("valid definition") +} + +/// Minimal flow-definition helper: nodes + edges wrapped in `graph`. +pub fn def_of(nodes: Value, edges: Value) -> Value { + json!({"name": "scenario", "graph": {"nodes": nodes, "edges": edges}}) +} + +pub fn node(id: &str, kind: &str, config: Value) -> Value { + json!({"id": id, "data": {"type": kind, "config": config}}) +} + +pub fn node_m(id: &str, kind: &str, config: Value, mods: Value) -> Value { + json!({"id": id, "data": {"type": kind, "config": config, "modifiers": mods}}) +} + +pub fn edge(s: &str, h: &str, t: &str) -> Value { + json!({"source": s, "sourceHandle": h, "target": t}) +} + +/// Deterministic action executor: script/egress nodes return the configured +/// per-node output (fallback: echoes the input). Lets scenario tests drive the +/// engine without network. +pub struct Deterministic { + outputs: HashMap, +} + +impl Deterministic { + pub fn new() -> Self { + Self { + outputs: HashMap::new(), + } + } + pub fn with(mut self, node_id: &str, output: Value) -> Self { + self.outputs.insert(node_id.to_string(), output); + self + } +} + +#[async_trait::async_trait] +impl NodeExecutor for Deterministic { + async fn exec( + &self, + node: &raisfast::flows::graph::GraphNode, + input: Value, + ) -> raisfast::errors::app_error::AppResult { + let out = self.outputs.get(&node.id).cloned().unwrap_or(input); + Ok(ExecOutcome { output: out }) + } +} + +/// Seed a snapshot with the given start-namespace inputs. +pub fn seeded_snapshot(start_inputs: Value) -> Snapshot { + let mut snap = Snapshot::new(); + let mut ns = HashMap::new(); + if let Some(obj) = start_inputs.as_object() { + for (k, v) in obj { + ns.insert(k.clone(), v.clone()); + } + } + snap.pool.insert("start".to_string(), ns); + snap +} + +/// Run an in-memory scenario to completion and return the snapshot. +pub async fn run_pure(graph: &Graph, snap: &mut Snapshot, exec: &dyn NodeExecutor) { + engine::run(graph, snap, exec).await.expect("run ok"); +} diff --git a/crates/core/tests/api/flows/mod.rs b/crates/core/tests/api/flows/mod.rs new file mode 100644 index 00000000..5bfcfa69 --- /dev/null +++ b/crates/core/tests/api/flows/mod.rs @@ -0,0 +1,10 @@ +//! Flow engine scenario test modules. +//! +//! Dedicated corpus for all current + future orchestration scenarios. Add one +//! file per concern (control flow, resilience, dataflow, durable, egress, +//! await/resume, concurrency…) as the engine grows. + +pub mod helpers; +pub mod scenario_await; +pub mod scenario_control_flow; +pub mod scenario_resilience; diff --git a/crates/core/tests/api/flows/scenario_await.rs b/crates/core/tests/api/flows/scenario_await.rs new file mode 100644 index 00000000..5c8c68ef --- /dev/null +++ b/crates/core/tests/api/flows/scenario_await.rs @@ -0,0 +1,112 @@ +//! Await/resume (HITL) scenarios against the DB-backed coordinator. + +use super::helpers::*; +use raisfast::flows::engine::Snapshot; +use raisfast::flows::model::{self, Flow, FlowInstance, FlowVersion}; +use raisfast::flows::run; +use raisfast::utils::tz::now_utc; +use serde_json::json; + +async fn seed_waiting_instance(pool: &raisfast::db::Pool) -> i64 { + let now = now_utc(); + let flow_id = raisfast::utils::id::new_snowflake_id(); + model::insert_flow( + pool, + &Flow { + id: flow_id, + tenant_id: "default".into(), + name: format!("await-{}", raisfast::utils::id::new_id()), + description: None, + enabled: true, + current_version: None, + extra: None, + created_at: now, + updated_at: now, + }, + ) + .await + .unwrap(); + let def = def_of( + json!([ + node("start", "start", json!({})), + node("gate", "await", json!({"kind": "human"})), + node( + "end", + "end", + json!({ + "outputs": [{"name": "ok", "value": {"ref": ["gate", "resume", "approved"]}}] + }) + ) + ]), + json!([edge("start", "out", "gate"), edge("gate", "out", "end")]), + ); + let vid = raisfast::utils::id::new_snowflake_id(); + model::insert_flow_version( + pool, + &FlowVersion { + id: vid, + flow_id, + version_number: 1, + definition: def, + created_by: None, + created_at: now, + }, + ) + .await + .unwrap(); + model::set_flow_current_version(pool, flow_id, vid) + .await + .unwrap(); + + let iid = raisfast::utils::id::new_snowflake_id(); + model::insert_flow_instance( + pool, + &FlowInstance { + id: iid, + tenant_id: "default".into(), + flow_id, + flow_version_id: vid, + status: "running".into(), + has_exceptions: false, + trigger_kind: "api".into(), + trigger_payload: Some(json!({})), + inputs_summary: None, + outputs: None, + error: None, + started_by: None, + started_at: Some(now), + finished_at: None, + waiting_kind: None, + waiting_needed: None, + waiting_received: 0, + resume_until: None, + created_at: now, + }, + ) + .await + .unwrap(); + iid.0 +} + +#[tokio::test] +async fn await_parks_instance_then_resume_approves() { + let pool = super::super::test_pool().await; + let det = Deterministic::new(); + let iid = seed_waiting_instance(&pool).await; + let iid = raisfast::types::snowflake_id::SnowflakeId(iid); + + run::execute_instance(&pool, iid, &det).await.unwrap(); + let inst = model::find_instance_by_id(&pool, iid).await.unwrap(); + assert_eq!(inst.status, "waiting", "parks on await: {inst:?}"); + let snap: Snapshot = + serde_json::from_value(model::find_snapshot(&pool, iid).await.unwrap().unwrap()).unwrap(); + assert!(snap.waiting_nodes.contains(&"gate".to_string())); + + // Resume with approval. + run::resume_instance(&pool, iid, Some(json!({"approved": true}))) + .await + .unwrap(); + let done = model::find_instance_by_id(&pool, iid).await.unwrap(); + assert_eq!(done.status, "success"); + assert_eq!(done.outputs.unwrap()["ok"], true); +} diff --git a/crates/core/tests/api/flows/scenario_control_flow.rs b/crates/core/tests/api/flows/scenario_control_flow.rs new file mode 100644 index 00000000..365c9ec6 --- /dev/null +++ b/crates/core/tests/api/flows/scenario_control_flow.rs @@ -0,0 +1,94 @@ +//! Control-flow scenarios: linear, branch (structured + expr), fan-out join, +//! skip propagation. + +use super::helpers::*; +use raisfast::flows::engine::{N_SKIPPED, N_SUCCESS, S_SUCCESS}; +use serde_json::json; + +#[tokio::test] +async fn linear_chain_with_template_and_refs() { + let g = def_graph(def_of( + json!([ + node("start", "start", json!({})), + node("s1", "egress", json!({"client_key": "k", "op": "o"})), + node( + "end", + "end", + json!({ + "outputs": [ + {"name": "echo", "value": {"ref": ["start", "msg"]}}, + {"name": "n", "value": {"literal": 5}} + ] + }) + ) + ]), + json!([edge("start", "out", "s1"), edge("s1", "out", "end")]), + )); + let mut snap = seeded_snapshot(json!({"msg": "hello"})); + let exec = Deterministic::new().with("s1", json!({"ok": true})); + run_pure(&g, &mut snap, &exec).await; + assert_eq!(snap.status, S_SUCCESS); + let out = snap.outputs.clone().unwrap(); + assert_eq!(out["echo"], "hello"); + assert_eq!(out["n"], 5); +} + +#[tokio::test] +async fn branch_takes_true_path_and_skips_false() { + let g = def_graph(def_of( + json!([ + node("start", "start", json!({})), + node_m( + "br", + "branch", + json!({ + "branches": [{"id": "b1", "when": "{{#start.level#}} >= 3", "handle": "true"}], + "else_handle": "false" + }), + json!({}) + ), + node("na", "egress", json!({"client_key": "k", "op": "o"})), + node("nb", "egress", json!({"client_key": "k", "op": "o"})) + ]), + json!([ + edge("start", "out", "br"), + edge("br", "true", "na"), + edge("br", "false", "nb") + ]), + )); + let mut snap = seeded_snapshot(json!({"level": 5})); + let exec = Deterministic::new(); + run_pure(&g, &mut snap, &exec).await; + assert_eq!(snap.node_states["na"].status, N_SUCCESS); + assert_eq!(snap.node_states["nb"].status, N_SKIPPED); +} + +#[tokio::test] +async fn join_waits_for_both_fanout_branches() { + let g = def_graph(def_of( + json!([ + node("start", "start", json!({})), + node("a", "egress", json!({"client_key": "k", "op": "o"})), + node("b", "egress", json!({"client_key": "k", "op": "o"})), + node("join", "egress", json!({"client_key": "k", "op": "o"})), + node("end", "end", json!({"outputs": []})) + ]), + json!([ + edge("start", "out", "a"), + edge("start", "out", "b"), + edge("a", "out", "join"), + edge("b", "out", "join"), + edge("join", "out", "end") + ]), + )); + let mut snap = seeded_snapshot(json!({})); + let exec = Deterministic::new(); + run_pure(&g, &mut snap, &exec).await; + assert_eq!(snap.status, S_SUCCESS); + assert_eq!(snap.node_states["a"].status, N_SUCCESS); + assert_eq!(snap.node_states["b"].status, N_SUCCESS); + assert_eq!( + snap.node_states["join"].status, N_SUCCESS, + "join waits both" + ); +} diff --git a/crates/core/tests/api/flows/scenario_resilience.rs b/crates/core/tests/api/flows/scenario_resilience.rs new file mode 100644 index 00000000..e2605582 --- /dev/null +++ b/crates/core/tests/api/flows/scenario_resilience.rs @@ -0,0 +1,191 @@ +//! Resilience scenarios: retry, continue_on_error, fail-stop, durable resume. + +use super::helpers::*; +use raisfast::errors::app_error::AppError; +use raisfast::flows::engine::{N_FAILED, N_SUCCESS, S_SUCCESS}; +use serde_json::{Value, json}; + +use std::sync::atomic::{AtomicUsize, Ordering}; + +/// Executor failing the first `fail_until` calls. +struct Flaky { + calls: AtomicUsize, + fail_until: usize, +} +#[async_trait::async_trait] +impl raisfast::flows::engine::NodeExecutor for Flaky { + async fn exec( + &self, + _n: &raisfast::flows::graph::GraphNode, + _i: Value, + ) -> raisfast::errors::app_error::AppResult { + let c = self.calls.fetch_add(1, Ordering::SeqCst); + if c < self.fail_until { + Err(AppError::BadRequest("boom".into())) + } else { + Ok(raisfast::flows::engine::ExecOutcome { + output: json!({"ok": true}), + }) + } + } +} + +fn linear_through(e1_cfg: Value, e1_mods: Value) -> raisfast::flows::graph::Graph { + def_graph(def_of( + json!([ + node("start", "start", json!({})), + node_m("e1", "egress", e1_cfg, e1_mods), + node("end", "end", json!({"outputs": []})) + ]), + json!([edge("start", "out", "e1"), edge("e1", "out", "end")]), + )) +} + +#[tokio::test] +async fn retry_recovers_after_transient_failure() { + let g = linear_through( + json!({"client_key": "k", "op": "o"}), + json!({"retry": {"attempts": 3}}), + ); + let mut snap = seeded_snapshot(json!({})); + let flaky = Flaky { + calls: AtomicUsize::new(0), + fail_until: 1, + }; + run_pure(&g, &mut snap, &flaky).await; + assert_eq!(snap.status, S_SUCCESS); + assert_eq!(snap.node_states["e1"].status, N_SUCCESS); + assert_eq!(snap.node_states["e1"].attempt, 2, "fail then retry"); +} + +#[tokio::test] +async fn continue_on_error_lets_flow_finish() { + let g = linear_through( + json!({"client_key": "k", "op": "o"}), + json!({"continue_on_error": true}), + ); + let mut snap = seeded_snapshot(json!({})); + let always = Flaky { + calls: AtomicUsize::new(0), + fail_until: usize::MAX, + }; + run_pure(&g, &mut snap, &always).await; + assert_eq!(snap.status, S_SUCCESS); + assert_eq!(snap.node_states["e1"].status, N_FAILED); + assert_eq!(snap.node_states["end"].status, N_SUCCESS); +} + +#[tokio::test] +async fn unhandled_failure_stops_flow() { + let g = linear_through(json!({"client_key": "k", "op": "o"}), json!({})); + let mut snap = seeded_snapshot(json!({})); + let always = Flaky { + calls: AtomicUsize::new(0), + fail_until: usize::MAX, + }; + run_pure(&g, &mut snap, &always).await; + assert_eq!(snap.status, raisfast::flows::engine::S_FAILED); +} + +#[tokio::test] +async fn durable_resume_does_not_rerun_completed_nodes() { + let pool = super::super::test_pool().await; + use raisfast::flows::model::{self, Flow, FlowInstance, FlowVersion}; + use raisfast::flows::run; + use raisfast::utils::tz::now_utc; + + let now = now_utc(); + let flow_id = raisfast::utils::id::new_snowflake_id(); + model::insert_flow( + &pool, + &Flow { + id: flow_id, + tenant_id: "default".into(), + name: format!("dup-{}", raisfast::utils::id::new_id()), + description: None, + enabled: true, + current_version: None, + extra: None, + created_at: now, + updated_at: now, + }, + ) + .await + .unwrap(); + + let def = def_of( + json!([ + node("start", "start", json!({})), + node("e1", "egress", json!({"client_key": "k", "op": "o"})), + node("end", "end", json!({"outputs": []})) + ]), + json!([edge("start", "out", "e1"), edge("e1", "out", "end")]), + ); + let vid = raisfast::utils::id::new_snowflake_id(); + model::insert_flow_version( + &pool, + &FlowVersion { + id: vid, + flow_id, + version_number: 1, + definition: def, + created_by: None, + created_at: now, + }, + ) + .await + .unwrap(); + + // Crash simulation: snapshot already has `start` success (completed). + let mut snap = raisfast::flows::engine::Snapshot::new(); + snap.node_states.insert( + "start".into(), + raisfast::flows::engine::NodeState { + status: N_SUCCESS.into(), + output: Some(json!(null)), + error: None, + attempt: 1, + }, + ); + model::upsert_snapshot(&pool, flow_id, &serde_json::to_value(&snap).unwrap()) + .await + .unwrap(); + + let iid = raisfast::utils::id::new_snowflake_id(); + model::insert_flow_instance( + &pool, + &FlowInstance { + id: iid, + tenant_id: "default".into(), + flow_id, + flow_version_id: vid, + status: "running".into(), + has_exceptions: false, + trigger_kind: "api".into(), + trigger_payload: Some(json!({})), + inputs_summary: None, + outputs: None, + error: None, + started_by: None, + started_at: Some(now), + finished_at: None, + waiting_kind: None, + waiting_needed: None, + waiting_received: 0, + resume_until: None, + created_at: now, + }, + ) + .await + .unwrap(); + + let det = Deterministic::new(); + run::execute_instance(&pool, iid, &det).await.unwrap(); + let done = model::find_instance_by_id(&pool, iid).await.unwrap(); + assert_eq!(done.status, "success"); + let s2: raisfast::flows::engine::Snapshot = + serde_json::from_value(model::find_snapshot(&pool, iid).await.unwrap().unwrap()).unwrap(); + // start completed before the "crash" and must not re-run. + assert_eq!(s2.node_states["start"].attempt, 1); + assert_eq!(s2.node_states["e1"].status, N_SUCCESS); +} diff --git a/migrations/mysql/schema.mysql.sql b/migrations/mysql/schema.mysql.sql index 8ff6903a..d37d0654 100644 --- a/migrations/mysql/schema.mysql.sql +++ b/migrations/mysql/schema.mysql.sql @@ -1319,3 +1319,89 @@ INSERT IGNORE INTO options (id, tenant_id, `option_key`, value, `type`, group_na (10014, 'default', 'maintenance_mode', 'false', 'boolean', 'appearance', 'Maintenance mode', 'When enabled, a maintenance page is shown to visitors', NULL, TRUE, TRUE, 31, NOW()), (10015, 'default', 'default_currency', '"USD"', 'select', 'ecommerce', 'Default currency', 'Currency code for products and orders', '{"values":["USD","CNY","EUR","GBP","JPY","KRW","HKD","TWD","SGD","AUD","CAD"]}', TRUE, TRUE, 40, NOW()), (10017, 'default', 'reserved_usernames', '"admin,administrator,root,system,official,support,staff,moderator,mod,help,info,mail,webmaster,security,billing,sales,owner,superuser,operator"', 'text', 'general', 'Reserved usernames', 'Comma-separated usernames that cannot be registered', '{"max_length":10000}', FALSE, TRUE, 5, NOW()); + +-- ============================================================ +-- Flow orchestration engine v2 (dev-docs/workflow) — P0-P1 5 tables +-- Data is engine-own (independent namespace); egress only referenced. +-- ============================================================ + +CREATE TABLE IF NOT EXISTS flow ( + id BIGINT PRIMARY KEY, + tenant_id VARCHAR(36) NOT NULL DEFAULT 'default', + name VARCHAR(255) NOT NULL, + description TEXT, + enabled TINYINT(1) NOT NULL DEFAULT 1, + current_version BIGINT NULL, + extra JSON, + created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + INDEX idx_flow_tenant (tenant_id) +); + +-- Immutable definition snapshots (publish = append; instance locks a version) +CREATE TABLE IF NOT EXISTS flow_version ( + id BIGINT PRIMARY KEY, + flow_id BIGINT NOT NULL, + version_number BIGINT NOT NULL, + definition JSON NOT NULL, + created_by BIGINT NULL, + created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + UNIQUE KEY uq_flow_version (flow_id, version_number), + INDEX idx_flow_version_flow (flow_id) +); + +-- One run of a flow (wf_trace root = instance id) +CREATE TABLE IF NOT EXISTS flow_instance ( + id BIGINT PRIMARY KEY, + tenant_id VARCHAR(36) NOT NULL DEFAULT 'default', + flow_id BIGINT NOT NULL, + flow_version_id BIGINT NOT NULL, + status VARCHAR(20) NOT NULL DEFAULT 'running', + has_exceptions TINYINT(1) NOT NULL DEFAULT 0, + trigger_kind VARCHAR(10) NOT NULL, + trigger_payload JSON, + inputs_summary JSON, + outputs JSON, + error JSON, + started_by BIGINT NULL, + started_at DATETIME NULL, + finished_at DATETIME NULL, + waiting_kind VARCHAR(10) NULL, + waiting_needed BIGINT NULL, + waiting_received BIGINT NOT NULL DEFAULT 0, + resume_until DATETIME NULL, + created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + INDEX idx_flow_instance_flow_status (flow_id, status), + INDEX idx_flow_instance_status_time (status, created_at), + INDEX idx_flow_instance_waiting (status, resume_until) +); + +-- Durable runnable state (whole snapshot, 1:1, rewritten each step) +CREATE TABLE IF NOT EXISTS flow_instance_snapshot ( + instance_id BIGINT PRIMARY KEY, + snapshot JSON NOT NULL, + snapshot_version BIGINT NOT NULL DEFAULT 1, + updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP +); + +-- Per-node run history (engine-own observability) +CREATE TABLE IF NOT EXISTS flow_node_run ( + id BIGINT PRIMARY KEY, + instance_id BIGINT NOT NULL, + node_id VARCHAR(64) NOT NULL, + node_type VARCHAR(32) NOT NULL, + seq BIGINT NOT NULL, + attempt BIGINT NOT NULL DEFAULT 1, + status VARCHAR(20) NOT NULL, + started_at DATETIME NULL, + finished_at DATETIME NULL, + latency_ms BIGINT NULL, + input_summary JSON, + output_summary JSON, + error JSON, + egress_log_id BIGINT NULL, + container_ref JSON, + created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + INDEX idx_flow_node_run_instance_seq (instance_id, seq), + INDEX idx_flow_node_run_egress (egress_log_id) +); diff --git a/migrations/postgres/schema.postgres.sql b/migrations/postgres/schema.postgres.sql index 402b9181..30157c8e 100644 --- a/migrations/postgres/schema.postgres.sql +++ b/migrations/postgres/schema.postgres.sql @@ -1382,3 +1382,89 @@ INSERT INTO options (id, tenant_id, option_key, value, type, group_name, label, (10015, 'default', 'default_currency', '"USD"', 'select', 'ecommerce', 'Default currency', 'Currency code for products and orders', '{"values":["USD","CNY","EUR","GBP","JPY","KRW","HKD","TWD","SGD","AUD","CAD"]}', TRUE, TRUE, 40, NOW()), (10017, 'default', 'reserved_usernames', '"admin,administrator,root,system,official,support,staff,moderator,mod,help,info,mail,webmaster,security,billing,sales,owner,superuser,operator"', 'text', 'general', 'Reserved usernames', 'Comma-separated usernames that cannot be registered', '{"max_length":10000}', FALSE, TRUE, 5, NOW()) ON CONFLICT (tenant_id, option_key) DO NOTHING; + +-- ============================================================ +-- Flow orchestration engine v2 (dev-docs/workflow) — P0-P1 5 tables +-- Data is engine-own (independent namespace); egress only referenced. +-- ============================================================ + +CREATE TABLE IF NOT EXISTS flow ( + id BIGINT PRIMARY KEY, + tenant_id TEXT NOT NULL DEFAULT 'default', + name TEXT NOT NULL, + description TEXT, + enabled BOOLEAN NOT NULL DEFAULT TRUE, + current_version BIGINT, + extra JSONB, + created_at TIMESTAMPTZ(0) NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ(0) NOT NULL DEFAULT NOW() +); +CREATE INDEX IF NOT EXISTS idx_flow_tenant ON flow(tenant_id); + +-- Immutable definition snapshots (publish = append; instance locks a version) +CREATE TABLE IF NOT EXISTS flow_version ( + id BIGINT PRIMARY KEY, + flow_id BIGINT NOT NULL, + version_number BIGINT NOT NULL, + definition JSONB NOT NULL, + created_by BIGINT, + created_at TIMESTAMPTZ(0) NOT NULL DEFAULT NOW(), + UNIQUE (flow_id, version_number) +); +CREATE INDEX IF NOT EXISTS idx_flow_version_flow ON flow_version(flow_id); + +-- One run of a flow (wf_trace root = instance id) +CREATE TABLE IF NOT EXISTS flow_instance ( + id BIGINT PRIMARY KEY, + tenant_id TEXT NOT NULL DEFAULT 'default', + flow_id BIGINT NOT NULL, + flow_version_id BIGINT NOT NULL, + status TEXT NOT NULL DEFAULT 'running', + has_exceptions BOOLEAN NOT NULL DEFAULT FALSE, + trigger_kind TEXT NOT NULL, + trigger_payload JSONB, + inputs_summary JSONB, + outputs JSONB, + error JSONB, + started_by BIGINT, + started_at TIMESTAMPTZ(0), + finished_at TIMESTAMPTZ(0), + waiting_kind TEXT, + waiting_needed BIGINT, + waiting_received BIGINT NOT NULL DEFAULT 0, + resume_until TIMESTAMPTZ(0), + created_at TIMESTAMPTZ(0) NOT NULL DEFAULT NOW() +); +CREATE INDEX IF NOT EXISTS idx_flow_instance_flow_status ON flow_instance(flow_id, status); +CREATE INDEX IF NOT EXISTS idx_flow_instance_status_time ON flow_instance(status, created_at); +CREATE INDEX IF NOT EXISTS idx_flow_instance_waiting ON flow_instance(status, resume_until); + +-- Durable runnable state (whole snapshot, 1:1, rewritten each step) +CREATE TABLE IF NOT EXISTS flow_instance_snapshot ( + instance_id BIGINT PRIMARY KEY, + snapshot JSONB NOT NULL, + snapshot_version BIGINT NOT NULL DEFAULT 1, + updated_at TIMESTAMPTZ(0) NOT NULL DEFAULT NOW() +); + +-- Per-node run history (engine-own observability) +CREATE TABLE IF NOT EXISTS flow_node_run ( + id BIGINT PRIMARY KEY, + instance_id BIGINT NOT NULL, + node_id TEXT NOT NULL, + node_type TEXT NOT NULL, + seq BIGINT NOT NULL, + attempt BIGINT NOT NULL DEFAULT 1, + status TEXT NOT NULL, + started_at TIMESTAMPTZ(0), + finished_at TIMESTAMPTZ(0), + latency_ms BIGINT, + input_summary JSONB, + output_summary JSONB, + error JSONB, + egress_log_id BIGINT, + container_ref JSONB, + created_at TIMESTAMPTZ(0) NOT NULL DEFAULT NOW() +); +CREATE INDEX IF NOT EXISTS idx_flow_node_run_instance_seq ON flow_node_run(instance_id, seq); +CREATE INDEX IF NOT EXISTS idx_flow_node_run_egress ON flow_node_run(egress_log_id); diff --git a/migrations/sqlite/schema.sqlite.sql b/migrations/sqlite/schema.sqlite.sql index 6b369202..4dd46049 100644 --- a/migrations/sqlite/schema.sqlite.sql +++ b/migrations/sqlite/schema.sqlite.sql @@ -1380,3 +1380,89 @@ INSERT OR IGNORE INTO options (id, tenant_id, option_key, value, type, group_nam (10014, 'default', 'maintenance_mode', 'false', 'boolean', 'appearance', 'Maintenance mode', 'When enabled, a maintenance page is shown to visitors', NULL, 1, 1, 31, strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), (10015, 'default', 'default_currency', '"USD"', 'select', 'ecommerce', 'Default currency', 'Currency code for products and orders', '{"values":["USD","CNY","EUR","GBP","JPY","KRW","HKD","TWD","SGD","AUD","CAD"]}', 1, 1, 40, strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), (10017, 'default', 'reserved_usernames', '"admin,administrator,root,system,official,support,staff,moderator,mod,help,info,mail,webmaster,security,billing,sales,owner,superuser,operator"', 'text', 'general', 'Reserved usernames', 'Comma-separated usernames that cannot be registered', '{"max_length":10000}', 0, 1, 5, strftime('%Y-%m-%dT%H:%M:%SZ', 'now')); + +-- ============================================================ +-- Flow orchestration engine v2 (dev-docs/workflow) — P0-P1 5 tables +-- Data is engine-own (independent namespace); egress only referenced. +-- ============================================================ + +CREATE TABLE IF NOT EXISTS flow ( + id INTEGER PRIMARY KEY, + tenant_id TEXT NOT NULL DEFAULT 'default', + name TEXT NOT NULL, + description TEXT, + enabled INTEGER NOT NULL DEFAULT 1, + current_version INTEGER, + extra TEXT, + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ','now')), + updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ','now')) +); +CREATE INDEX IF NOT EXISTS idx_flow_tenant ON flow(tenant_id); + +-- Immutable definition snapshots (publish = append; instance locks a version) +CREATE TABLE IF NOT EXISTS flow_version ( + id INTEGER PRIMARY KEY, + flow_id INTEGER NOT NULL, + version_number INTEGER NOT NULL, + definition TEXT NOT NULL, + created_by INTEGER, + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ','now')), + UNIQUE (flow_id, version_number) +); +CREATE INDEX IF NOT EXISTS idx_flow_version_flow ON flow_version(flow_id); + +-- One run of a flow (wf_trace root = instance id) +CREATE TABLE IF NOT EXISTS flow_instance ( + id INTEGER PRIMARY KEY, + tenant_id TEXT NOT NULL DEFAULT 'default', + flow_id INTEGER NOT NULL, + flow_version_id INTEGER NOT NULL, + status TEXT NOT NULL DEFAULT 'running', + has_exceptions INTEGER NOT NULL DEFAULT 0, + trigger_kind TEXT NOT NULL, + trigger_payload TEXT, + inputs_summary TEXT, + outputs TEXT, + error TEXT, + started_by INTEGER, + started_at TEXT, + finished_at TEXT, + waiting_kind TEXT, + waiting_needed INTEGER, + waiting_received INTEGER NOT NULL DEFAULT 0, + resume_until TEXT, + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ','now')) +); +CREATE INDEX IF NOT EXISTS idx_flow_instance_flow_status ON flow_instance(flow_id, status); +CREATE INDEX IF NOT EXISTS idx_flow_instance_status_time ON flow_instance(status, created_at); +CREATE INDEX IF NOT EXISTS idx_flow_instance_waiting ON flow_instance(status, resume_until); + +-- Durable runnable state (whole snapshot, 1:1, rewritten each step) +CREATE TABLE IF NOT EXISTS flow_instance_snapshot ( + instance_id INTEGER PRIMARY KEY, + snapshot TEXT NOT NULL, + snapshot_version INTEGER NOT NULL DEFAULT 1, + updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ','now')) +); + +-- Per-node run history (engine-own observability) +CREATE TABLE IF NOT EXISTS flow_node_run ( + id INTEGER PRIMARY KEY, + instance_id INTEGER NOT NULL, + node_id TEXT NOT NULL, + node_type TEXT NOT NULL, + seq INTEGER NOT NULL, + attempt INTEGER NOT NULL DEFAULT 1, + status TEXT NOT NULL, + started_at TEXT, + finished_at TEXT, + latency_ms INTEGER, + input_summary TEXT, + output_summary TEXT, + error TEXT, + egress_log_id INTEGER, + container_ref TEXT, + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ','now')) +); +CREATE INDEX IF NOT EXISTS idx_flow_node_run_instance_seq ON flow_node_run(instance_id, seq); +CREATE INDEX IF NOT EXISTS idx_flow_node_run_egress ON flow_node_run(egress_log_id);