mlua_flow_ir/
lib.rs

1#![deny(unsafe_code)]
2//! flow.ir async runtime + mlua binding.
3//!
4//! Layer 3 of the 4-layer flow.ir stack:
5//!
6//! 1. `flow-ir-lua` — Pure Lua DSL (separate repo, ecosystem-neutral)
7//! 2. `flow-ir-core` — Pure Rust schema + sync interpreter (no mlua, no async)
8//! 3. `mlua-flow-ir` — **this crate**: re-export of `flow-ir-core` +
9//!    `AsyncDispatcher` + `eval_async` + `fanout_eval` + Lua `module()` binding
10//! 4. `mlua-swarm-engine` — host concerns (Spawner / Worker / Loop /
11//!    AuthzPolicy / cp_state persist)
12//!
13//! All schema types (`Node` / `Expr` / `JoinMode` / `EvalError` / `Dispatcher`)
14//! are re-exported verbatim from `flow-ir-core` so callers can keep a single
15//! import path:
16//!
17//! ```
18//! use mlua_flow_ir::{eval, eval_async, AsyncDispatcher, Dispatcher, EvalError, Expr, Node};
19//! ```
20
21// ──────────────────────────────────────────────────────────────────────────
22// Re-export Pure Rust core (flow-ir-core)
23// ──────────────────────────────────────────────────────────────────────────
24
25pub use flow_ir_core::{
26    eval, eval_expr, is_truthy, read_path, write_path, Dispatcher, EvalError, Expr, JoinMode, Node,
27};
28
29use serde_json::Value;
30
31// ══════════════════════════════════════════════════════════════════════════
32// v0.0.2 — Async core (eval_async + AsyncDispatcher trait)
33// ══════════════════════════════════════════════════════════════════════════
34
35use async_recursion::async_recursion;
36use async_trait::async_trait;
37
38/// Async dispatcher trait — async 版 `Dispatcher`。
39///
40/// `async_trait` macro 経由 (= Rust 2021 互換 + dyn safe)。 Host crate
41/// (e.g. mlua-swarm-engine `AsyncSpawner`) が impl する。 substrate には
42/// tokio dep 入れない (= Pure 維持)、 executor は caller (host) 責務。
43#[async_trait]
44pub trait AsyncDispatcher: Send + Sync {
45    async fn dispatch(&self, ref_: &str, input: Value) -> Result<Value, EvalError>;
46}
47
48/// Evaluate a `Node` against a context value asynchronously,
49/// using the given async dispatcher for `Step` resolution.
50///
51/// `eval` (sync) と同型 logic、 dispatch を `.await` に置き換え。 Seq / Branch
52/// は recursive async fn (= `async_recursion` macro で `Pin<Box>` wrap)。
53///
54/// # Quick start
55///
56/// ```
57/// use async_trait::async_trait;
58/// use mlua_flow_ir::{eval_async, AsyncDispatcher, EvalError, Expr, Node};
59/// use serde_json::{json, Value};
60///
61/// struct Fixture;
62///
63/// #[async_trait]
64/// impl AsyncDispatcher for Fixture {
65///     async fn dispatch(&self, _r: &str, input: Value) -> Result<Value, EvalError> {
66///         if let Value::String(s) = input {
67///             Ok(Value::String(s.to_uppercase()))
68///         } else {
69///             Ok(input)
70///         }
71///     }
72/// }
73///
74/// let rt = tokio::runtime::Runtime::new().unwrap();
75/// rt.block_on(async {
76///     let node = Node::Step {
77///         ref_: "up".into(),
78///         in_: Expr::Path { at: "$.input".into() },
79///         out: Expr::Path { at: "$.output".into() },
80///     };
81///     let out = eval_async(&node, json!({ "input": "hello" }), &Fixture).await.unwrap();
82///     assert_eq!(out, json!({ "input": "hello", "output": "HELLO" }));
83/// });
84/// ```
85#[async_recursion]
86pub async fn eval_async<D>(node: &Node, ctx: Value, dispatcher: &D) -> Result<Value, EvalError>
87where
88    D: AsyncDispatcher + ?Sized,
89{
90    match node {
91        Node::Step { ref_, in_, out } => {
92            let input = eval_expr(in_, &ctx)?;
93            let output =
94                dispatcher
95                    .dispatch(ref_, input)
96                    .await
97                    .map_err(|e| EvalError::DispatcherError {
98                        ref_: ref_.clone(),
99                        msg: e.to_string(),
100                    })?;
101            write_path(out, ctx, output)
102        }
103        Node::Seq { children } => {
104            let mut cur = ctx;
105            for child in children {
106                cur = eval_async(child, cur, dispatcher).await?;
107            }
108            Ok(cur)
109        }
110        Node::Branch { cond, then_, else_ } => match eval_expr(cond, &ctx)? {
111            Value::Bool(true) => eval_async(then_, ctx, dispatcher).await,
112            Value::Bool(false) => eval_async(else_, ctx, dispatcher).await,
113            other => Err(EvalError::NonBoolCond(other)),
114        },
115        Node::Fanout {
116            items,
117            bind,
118            body,
119            join,
120            out,
121        } => fanout_eval(items, bind, body, *join, out, ctx, dispatcher).await,
122        Node::Loop {
123            counter,
124            cond,
125            body,
126            max,
127        } => {
128            let mut cur = write_path(counter, ctx, Value::Number(serde_json::Number::from(0u32)))?;
129            let mut n: u32 = 0;
130            while n < *max && is_truthy(&eval_expr(cond, &cur)?) {
131                cur = eval_async(body, cur, dispatcher).await?;
132                n += 1;
133                cur = write_path(counter, cur, Value::Number(serde_json::Number::from(n)))?;
134            }
135            Ok(cur)
136        }
137        Node::Try {
138            body,
139            catch,
140            err_at,
141        } => match eval_async(body, ctx.clone(), dispatcher).await {
142            Ok(v) => Ok(v),
143            Err(e) => {
144                let cur = match err_at {
145                    Some(at) => write_path(at, ctx, Value::String(e.to_string()))?,
146                    None => ctx,
147                };
148                eval_async(catch, cur, dispatcher).await
149            }
150        },
151    }
152}
153
154/// Fanout 並列 evaluator。 executor 不問 (futures crate のみ)、 caller の async
155/// runtime (tokio / async-std / 自前) がそのまま並列性を出す。
156#[async_recursion]
157async fn fanout_eval<D>(
158    items: &Expr,
159    bind: &Expr,
160    body: &Node,
161    join: JoinMode,
162    out: &Expr,
163    ctx: Value,
164    dispatcher: &D,
165) -> Result<Value, EvalError>
166where
167    D: AsyncDispatcher + ?Sized,
168{
169    use futures::future::{join_all, select_ok, FutureExt};
170
171    let items_val = eval_expr(items, &ctx)?;
172    let items_arr = match items_val {
173        Value::Array(a) => a,
174        other => {
175            return Err(EvalError::DispatcherError {
176                ref_: "fanout.items".into(),
177                msg: format!("expected array, got {other:?}"),
178            })
179        }
180    };
181
182    // 各 branch を Pin<Box<dyn Future>> として並列化、 ctx は caller の snapshot を
183    // clone して各 branch に渡す (= disjoint state)。
184    let branch_futs: Vec<_> = items_arr
185        .into_iter()
186        .map(|item| {
187            let branch_ctx = write_path(bind, ctx.clone(), item)?;
188            Ok::<_, EvalError>(eval_async(body, branch_ctx, dispatcher))
189        })
190        .collect::<Result<_, _>>()?;
191
192    let joined: Value = match join {
193        JoinMode::All => {
194            // try_join_all = 全成功で Vec、 1 つでも fail で即 abort + error
195            let results = futures::future::try_join_all(branch_futs).await?;
196            Value::Array(results)
197        }
198        JoinMode::Any => {
199            if branch_futs.is_empty() {
200                Value::Array(vec![])
201            } else {
202                // select_ok = 最初に成功した branch の ctx を winner、 全 fail で last error
203                let mapped = branch_futs
204                    .into_iter()
205                    .map(|f| f.boxed())
206                    .collect::<Vec<_>>();
207                let (winner, _rest) = select_ok(mapped).await?;
208                winner
209            }
210        }
211        JoinMode::Race => {
212            if branch_futs.is_empty() {
213                Value::Array(vec![])
214            } else {
215                // select = 最初に settle した branch (Ok / Err 問わず) の結果
216                let mapped = branch_futs
217                    .into_iter()
218                    .map(|f| f.boxed())
219                    .collect::<Vec<_>>();
220                let (first, _idx, _rest) = futures::future::select_all(mapped).await;
221                first?
222            }
223        }
224        JoinMode::AllSettled => {
225            // 全 branch 完走、 fail も rejected record として残す
226            let results = join_all(branch_futs).await;
227            let records: Vec<Value> = results
228                .into_iter()
229                .map(|r| match r {
230                    Ok(v) => serde_json::json!({"status": "fulfilled", "value": v}),
231                    Err(e) => serde_json::json!({"status": "rejected", "reason": e.to_string()}),
232                })
233                .collect();
234            Value::Array(records)
235        }
236    };
237
238    write_path(out, ctx, joined)
239}
240
241// ══════════════════════════════════════════════════════════════════════════
242// v0.0.3 — mlua bridge full
243// ══════════════════════════════════════════════════════════════════════════
244
245use mlua::LuaSerdeExt;
246
247/// Lua function を Rust `Dispatcher` trait に wrap した adapter。
248///
249/// Lua 側 dispatcher function `function(ref, input) return ... end` を受けて、
250/// Rust `eval(node, ctx, &lua_dispatcher)` から呼び出せるようにする。
251/// 内部で serde Value ↔ Lua value 変換 (= mlua serde feature) を経由。
252struct LuaDispatcher<'a> {
253    lua: &'a mlua::Lua,
254    func: mlua::Function,
255}
256
257impl<'a> Dispatcher for LuaDispatcher<'a> {
258    fn dispatch(&self, ref_: &str, input: Value) -> Result<Value, EvalError> {
259        let lua_input = self
260            .lua
261            .to_value(&input)
262            .map_err(|e| EvalError::DispatcherError {
263                ref_: ref_.into(),
264                msg: format!("to_value: {}", e),
265            })?;
266        let result: mlua::Value = self.func.call((ref_.to_string(), lua_input)).map_err(|e| {
267            EvalError::DispatcherError {
268                ref_: ref_.into(),
269                msg: format!("lua call: {}", e),
270            }
271        })?;
272        let value: Value = self
273            .lua
274            .from_value(result)
275            .map_err(|e| EvalError::DispatcherError {
276                ref_: ref_.into(),
277                msg: format!("from_value: {}", e),
278            })?;
279        Ok(value)
280    }
281}
282
283/// Register the flow module table with Lua.
284///
285/// v0.0.3 full impl — exposes:
286///
287/// - `flow.version` (= string): crate version
288/// - `flow.eval(node_table, ctx_table, dispatcher_fn) -> result_table`:
289///   Lua-side entry to evaluate a flow.ir BluePrint with a Lua dispatcher fn
290///
291/// # Lua usage
292///
293/// ```lua
294/// local flow = require("flow")  -- or set via lua.globals():set("flow", module(lua))
295///
296/// local node = {
297///   kind = "step",
298///   ref = "uppercase",
299///   ["in"] = { op = "path", at = "$.input" },
300///   out = { op = "path", at = "$.output" },
301/// }
302///
303/// local function dispatcher(ref, input)
304///   if ref == "uppercase" then
305///     return string.upper(input)
306///   end
307/// end
308///
309/// local result = flow.eval(node, { input = "hello" }, dispatcher)
310/// assert(result.output == "HELLO")
311/// ```
312pub fn module(lua: &mlua::Lua) -> mlua::Result<mlua::Table> {
313    let t = lua.create_table()?;
314    t.set("version", env!("CARGO_PKG_VERSION"))?;
315
316    let eval_fn = lua.create_function(
317        |lua_inner: &mlua::Lua,
318         (node_val, ctx_val, dispatcher_fn): (mlua::Value, mlua::Value, mlua::Function)| {
319            let node: Node = lua_inner
320                .from_value(node_val)
321                .map_err(|e| mlua::Error::external(format!("node parse: {}", e)))?;
322            let ctx: Value = lua_inner
323                .from_value(ctx_val)
324                .map_err(|e| mlua::Error::external(format!("ctx parse: {}", e)))?;
325
326            let dispatcher = LuaDispatcher {
327                lua: lua_inner,
328                func: dispatcher_fn,
329            };
330            let result = eval(&node, ctx, &dispatcher)
331                .map_err(|e| mlua::Error::external(format!("eval: {}", e)))?;
332            lua_inner.to_value(&result)
333        },
334    )?;
335    t.set("eval", eval_fn)?;
336
337    Ok(t)
338}
mlua_flow_ir/lib.rs

mlua_flow_ir/
lib.rs
