Skip to main content

flow_ir_core/
lib.rs

1#![deny(unsafe_code)]
2#![warn(missing_docs)]
3//! flow.ir Pure Rust schema + sync interpreter.
4//!
5//! Node kinds (Step / Seq / Branch / Fanout / Loop / Try / Let) + Expr ops
6//! (canonical wire format — comparison / boolean / existence / arithmetic /
7//! aggregate / `call_extern`) + sync `eval` + `Dispatcher` trait + `Externs`
8//! registry + typed [`Path`] read/write (see [`Path`] for the full path
9//! syntax + uniform malformed-path rejection rules — the single authority,
10//! rather than restating them here).
11//!
12//! mlua / futures / async 依存ゼロ。 async runtime + mlua binding は上流
13//! `mlua-flow-ir` crate が担当する 4 層 stack の core 層。
14//!
15//! # Quick start
16//!
17//! ```
18//! use flow_ir_core::{eval, Dispatcher, EvalError, Expr, Node};
19//! use serde_json::{json, Value};
20//!
21//! let node: Node = serde_json::from_value(json!({
22//!     "kind": "step",
23//!     "ref": "uppercase",
24//!     "in": { "op": "path", "at": "$.input" },
25//!     "out": { "op": "path", "at": "$.output" },
26//! })).unwrap();
27//!
28//! struct Fixture;
29//! impl Dispatcher for Fixture {
30//!     fn dispatch(&self, _r: &str, input: Value) -> Result<Value, EvalError> {
31//!         if let Value::String(s) = input {
32//!             Ok(Value::String(s.to_uppercase()))
33//!         } else {
34//!             Ok(input)
35//!         }
36//!     }
37//! }
38//!
39//! let out = eval(&node, json!({ "input": "hello" }), &Fixture).unwrap();
40//! assert_eq!(out, json!({ "input": "hello", "output": "HELLO" }));
41//! ```
42
43use serde::{Deserialize, Serialize};
44use serde_json::Value;
45use thiserror::Error;
46
47mod path;
48pub use path::{Path, PathParseError};
49
50// ──────────────────────────────────────────────────────────────────────────
51// IR: 7 Node kinds + 20 Expr ops
52// ──────────────────────────────────────────────────────────────────────────
53
54/// flow.ir Node kind.
55///
56/// Discriminated with `kind` tag, `deny_unknown_fields` (open=false),
57/// `rename_all = "snake_case"`. Covers the 7 supported kinds: `Step` / `Seq`
58/// / `Branch` / `Fanout` (canonical schema の `fanout` Node、 4 join mode) /
59/// `Loop` / `Try` / `Let` (canonical `let` — see the [`Node::Let`] variant
60/// for the v0.3.0 rename from `Assign` and the accompanying `at` field
61/// shape change). Additional kinds may be added in future versions.
62#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
63#[serde(tag = "kind", deny_unknown_fields, rename_all = "snake_case")]
64pub enum Node {
65    /// `Step` — dispatch a referenced operation with `in` input, write result to `out`.
66    Step {
67        /// Dispatcher key (wire field `ref`), resolved via [`Dispatcher::dispatch`].
68        #[serde(rename = "ref")]
69        ref_: String,
70        /// Input `Expr`, evaluated against the ctx snapshot before dispatch (wire field `in`).
71        #[serde(rename = "in")]
72        in_: Expr,
73        /// `Path` `Expr` the dispatcher's output is written to.
74        out: Expr,
75    },
76    /// `Seq` — evaluate children in order, threading the context value through.
77    Seq {
78        /// Child nodes, evaluated in order.
79        children: Vec<Node>,
80    },
81    /// `Branch` — eval `cond`; if `true` run `then`, else run `else`.
82    Branch {
83        /// Condition `Expr`; must evaluate to a JSON boolean.
84        cond: Expr,
85        /// Branch taken when `cond` is `true` (wire field `then`).
86        #[serde(rename = "then")]
87        then_: Box<Node>,
88        /// Branch taken when `cond` is `false` (wire field `else`).
89        #[serde(rename = "else")]
90        else_: Box<Node>,
91    },
92    /// `Fanout` — eval `items` to an array, run `body` per item against a
93    /// branch-local ctx (caller ctx + item written to `bind`), join results
94    /// per `join` mode into `out`. Async parallel runner uses
95    /// `futures::future::{try_join_all|select_ok|join_all}` (executor-agnostic).
96    Fanout {
97        /// `Expr` evaluated to a JSON array; one branch runs per element.
98        items: Expr,
99        /// `Path` `Expr` each branch's item is written to before running `body`.
100        bind: Expr,
101        /// Node run once per `items` element, against a disjoint branch ctx.
102        body: Box<Node>,
103        /// How per-branch results are combined into `out`.
104        join: JoinMode,
105        /// `Path` `Expr` the joined result is written to.
106        out: Expr,
107    },
108    /// `Loop` — counter を 0 から、 `cond` が truthy かつ `counter < max` の間
109    /// `body` を eval。 各 iter 後 counter を increment して `counter` path に書く。
110    /// VerdictLoop 等の retry/poll パターン primitive (canonical schema 整合)。
111    Loop {
112        /// `Path` `Expr` the iteration counter is written to (starts at `0`).
113        counter: Expr,
114        /// Condition re-evaluated before each iteration; loop stops once falsy.
115        cond: Expr,
116        /// Node evaluated once per iteration.
117        body: Box<Node>,
118        /// Hard iteration cap (loop stops once `counter >= max`, regardless of `cond`).
119        max: u32,
120    },
121    /// `Try` — `body` を eval、 raise した場合 `catch` を eval。
122    /// `err_at` が Some なら catch 開始前に error message を ctx に書く。
123    Try {
124        /// Node evaluated first; failures trigger a ctx rollback + `catch`.
125        body: Box<Node>,
126        /// Node evaluated when `body` raises.
127        catch: Box<Node>,
128        /// Optional `Path` `Expr` the error message is written to before `catch` runs.
129        #[serde(default)]
130        err_at: Option<Expr>,
131    },
132    /// `Let` — pure transform Node (canonical `let`). `value` Expr を ctx
133    /// snapshot 上で評価し、 結果を `at` ([`Path`]) に write する。 dispatcher
134    /// 不要、 副作用は `CtxStorage.write` 1 回のみ。 `Seq` の中で Step 間の
135    /// Adhoc update 表現に使う (= IR primitive、 Command 履歴は CtxStorage の
136    /// write hook 経由で取得)。
137    ///
138    /// **v0.3.0 rename (breaking):** this variant used to be `Node::Assign`
139    /// with `at: Expr` (a `Path`-wrapping `Expr`). It now matches the
140    /// canonical `flow.ir` schema's `let` node: the wire tag is `"let"` and
141    /// `at` is a bare [`Path`], serialized as a plain path string
142    /// (`"ctx.foo"`) rather than a `Path` `Expr` object. The canonical
143    /// write-side prefix is `ctx.`; the parser accepts both `$` and `ctx`
144    /// (root-token distinction is delegated to the caller — see [`Path`]).
145    Let {
146        /// [`Path`] the evaluated `value` is written to.
147        at: Path,
148        /// `Expr` evaluated against the ctx snapshot.
149        value: Expr,
150    },
151}
152
153/// Fanout join semantics (Promise / futures combinators).
154#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
155#[serde(rename_all = "snake_case")]
156pub enum JoinMode {
157    /// every branch runs; out is an array of per-branch final ctx
158    /// (Promise.all / `futures::try_join_all`).
159    All,
160    /// first non-raising branch's ctx wins; all-fail raises
161    /// (Promise.any / `futures::future::select_ok`).
162    Any,
163    /// first branch to settle wins, success OR raise
164    /// (Promise.race / `futures::future::select`).
165    Race,
166    /// every branch runs, never raises; per-item record
167    /// `{status: fulfilled|rejected, value|reason}` (Promise.allSettled).
168    AllSettled,
169}
170
171/// flow.ir Expr op.
172///
173/// Discriminated with `op` tag, `deny_unknown_fields`, `rename_all = "snake_case"`.
174/// Wire format (op tag / field names) follows the canonical `flow-ir-lua`
175/// schema (`flow/ir/schema.lua`) verbatim: `gte`/`lte` (not `ge`/`le`),
176/// `args` on `and`/`or`, `arg` on `not`/`len`/`exists`.
177///
178/// Ops:
179/// - read / literal: `Path` / `Lit`
180/// - comparison: `Eq` / `Ne` / `Lt` / `Lte` / `Gt` / `Gte` (numbers or strings)
181/// - boolean: `Not` / `And` / `Or`
182/// - existence: `Exists` (truthy iff `arg` evaluates to a non-null value)
183/// - arithmetic: `Add` / `Sub` / `Mul` / `Div` / `Mod`
184/// - aggregate: `Len` (length of array / string / object) / `In` (membership in array)
185/// - hatch: `CallExtern` (host-registered pure function, resolved via `Externs`)
186#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
187#[serde(tag = "op", deny_unknown_fields, rename_all = "snake_case")]
188pub enum Expr {
189    /// `Path` — read a value from ctx by simple `$.a.b.c` form.
190    Path {
191        /// The parsed context path — see [`Path`] for syntax + rejection
192        /// rules. Deserialized (and syntax-validated) once, at parse time.
193        at: Path,
194    },
195    /// `Lit` — literal JSON value.
196    Lit {
197        /// The literal value, returned as-is on evaluation.
198        value: Value,
199    },
200    /// `Eq` — boolean equality of two sub-expressions. Numbers compare by
201    /// f64 value (`5 == 5.0` is `true`), matching the ordering ops' numeric
202    /// coercion; integers above 2^53 may lose precision (same caveat as
203    /// `Lt`/`Lte`/`Gt`/`Gte`).
204    Eq {
205        /// Left-hand operand.
206        lhs: Box<Expr>,
207        /// Right-hand operand.
208        rhs: Box<Expr>,
209    },
210    /// `Ne` — boolean inequality. Same numeric coercion as `Eq` (`5 != 5.0`
211    /// is `false`); precision caveat above 2^53 shared with ordering ops.
212    Ne {
213        /// Left-hand operand.
214        lhs: Box<Expr>,
215        /// Right-hand operand.
216        rhs: Box<Expr>,
217    },
218    /// `Lt` — `lhs < rhs`. Both numbers (f64) or both strings (lexicographic),
219    /// mirroring canonical Lua `<` semantics. Mixed / other types raise.
220    Lt {
221        /// Left-hand operand.
222        lhs: Box<Expr>,
223        /// Right-hand operand.
224        rhs: Box<Expr>,
225    },
226    /// `Lte` — `lhs <= rhs` (canonical wire tag `lte`).
227    Lte {
228        /// Left-hand operand.
229        lhs: Box<Expr>,
230        /// Right-hand operand.
231        rhs: Box<Expr>,
232    },
233    /// `Gt` — `lhs > rhs`.
234    Gt {
235        /// Left-hand operand.
236        lhs: Box<Expr>,
237        /// Right-hand operand.
238        rhs: Box<Expr>,
239    },
240    /// `Gte` — `lhs >= rhs` (canonical wire tag `gte`).
241    Gte {
242        /// Left-hand operand.
243        lhs: Box<Expr>,
244        /// Right-hand operand.
245        rhs: Box<Expr>,
246    },
247    /// `Not` — boolean negation of `arg` (truthy-based; null/false → true).
248    Not {
249        /// Operand negated by truthiness.
250        arg: Box<Expr>,
251    },
252    /// `And` — variadic boolean conjunction (short-circuit). Empty list → true.
253    And {
254        /// Operands evaluated left-to-right until one is falsy.
255        args: Vec<Expr>,
256    },
257    /// `Or` — variadic boolean disjunction (short-circuit). Empty list → false.
258    Or {
259        /// Operands evaluated left-to-right until one is truthy.
260        args: Vec<Expr>,
261    },
262    /// `Exists` — evaluate `arg`; `true` iff it resolves to a non-null value.
263    /// A `Path` arg that raises `PathNotFound` yields `false` (canonical
264    /// `arg ~= nil` semantics — JSON null maps to Lua nil).
265    Exists {
266        /// Operand whose presence (non-null, resolvable) is tested.
267        arg: Box<Expr>,
268    },
269    /// `Add` — numeric `lhs + rhs` (f64).
270    Add {
271        /// Left-hand operand.
272        lhs: Box<Expr>,
273        /// Right-hand operand.
274        rhs: Box<Expr>,
275    },
276    /// `Sub` — numeric `lhs - rhs`.
277    Sub {
278        /// Left-hand operand.
279        lhs: Box<Expr>,
280        /// Right-hand operand.
281        rhs: Box<Expr>,
282    },
283    /// `Mul` — numeric `lhs * rhs`.
284    Mul {
285        /// Left-hand operand.
286        lhs: Box<Expr>,
287        /// Right-hand operand.
288        rhs: Box<Expr>,
289    },
290    /// `Div` — numeric `lhs / rhs`. Division by zero raises `ArithError`.
291    Div {
292        /// Left-hand operand (dividend).
293        lhs: Box<Expr>,
294        /// Right-hand operand (divisor).
295        rhs: Box<Expr>,
296    },
297    /// `Mod` — numeric `lhs % rhs` (Lua `%` semantics: result takes the sign
298    /// of `rhs`). Modulo by zero raises `ArithError`.
299    Mod {
300        /// Left-hand operand (dividend).
301        lhs: Box<Expr>,
302        /// Right-hand operand (divisor).
303        rhs: Box<Expr>,
304    },
305    /// `Len` — length of `arg`: array → element count, string → char count,
306    /// object → key count. Other types raise `TypeError`.
307    Len {
308        /// Operand whose length is computed.
309        arg: Box<Expr>,
310    },
311    /// `In` — `true` if `needle` equals any element of `haystack` (which must
312    /// evaluate to an array). Rust-side extension (not in canonical schema).
313    /// Element equality uses the same numeric coercion as `Eq` (`5 == 5.0`);
314    /// precision caveat above 2^53 shared with ordering ops.
315    In {
316        /// Value tested for membership.
317        needle: Box<Expr>,
318        /// Operand that must evaluate to a JSON array.
319        haystack: Box<Expr>,
320    },
321    /// `CallExtern` — value-shape Hatch: resolve a host-injected pure function
322    /// by opaque key via the `Externs` registry, apply it to evaluated args,
323    /// return the value. The registered function MUST be pure (no side
324    /// effects, no flow control) — see canonical `doc/ir.md §call_extern`.
325    CallExtern {
326        /// Extern registry key (wire field `ref`), resolved via [`Externs::call`].
327        #[serde(rename = "ref")]
328        ref_: String,
329        /// Argument expressions, evaluated before the extern call.
330        args: Vec<Expr>,
331    },
332}
333
334// ──────────────────────────────────────────────────────────────────────────
335// Dispatcher trait + EvalError
336// ──────────────────────────────────────────────────────────────────────────
337
338/// Dispatcher callback: resolves a `Step.ref` against the provided input,
339/// returns the step's raw output value.
340///
341/// Host crates (e.g. `mlua-swarm-engine`) provide concrete implementations:
342/// agent-block process spawn, mlua callback, MCP call, direct LLM, etc.
343/// `Fn(&str, Value) -> Result<Value, EvalError>` closures also implement this
344/// trait via the blanket impl below.
345pub trait Dispatcher {
346    /// Resolve `ref_` against `input`, returning the step's raw output value.
347    fn dispatch(&self, ref_: &str, input: Value) -> Result<Value, EvalError>;
348}
349
350impl<F> Dispatcher for F
351where
352    F: Fn(&str, Value) -> Result<Value, EvalError>,
353{
354    fn dispatch(&self, ref_: &str, input: Value) -> Result<Value, EvalError> {
355        self(ref_, input)
356    }
357}
358
359/// Evaluation error.
360///
361/// `#[non_exhaustive]`: new variants may be added in a minor release: match
362/// with a wildcard arm.
363#[derive(Debug, Error)]
364#[non_exhaustive]
365pub enum EvalError {
366    /// A `Path` read did not resolve — the requested key is missing from ctx.
367    #[error("path not found: {0}")]
368    PathNotFound(String),
369    /// A `Path` string is malformed — see [`Path`] for the full syntax +
370    /// rejection rules. Raised by the `read_path` / `write_path` compat
371    /// wrappers when their `&str` argument fails to parse; never raised by
372    /// [`Path::read`] / [`Path::write`] themselves (an already-parsed
373    /// `Path` cannot represent malformed syntax).
374    #[error("invalid path syntax: {0}")]
375    InvalidPath(String),
376    /// `Branch.cond` evaluated to a non-boolean value.
377    #[error("branch cond must be boolean, got: {0}")]
378    NonBoolCond(Value),
379    /// An expression or node received a value of the wrong type (e.g. a
380    /// `Len`/`In`/comparison/arithmetic operand of the wrong JSON type, or a
381    /// `Fanout.items` result that did not evaluate to an array).
382    #[error("type error in '{op}': {msg}")]
383    TypeError {
384        /// The op (or synthetic label, e.g. `"fanout.any"`) that raised.
385        op: String,
386        /// Human-readable description of the type mismatch.
387        msg: String,
388    },
389    /// An arithmetic operation failed (division/modulo by zero, or a
390    /// numeric result/operand that cannot be represented as `f64`).
391    #[error("arithmetic error in '{op}': {msg}")]
392    ArithError {
393        /// The op (e.g. `"div"`, `"mod"`, `"cmp"`) that raised.
394        op: String,
395        /// Human-readable description of the arithmetic failure.
396        msg: String,
397    },
398    /// The [`Dispatcher`] returned an error for the given `Step.ref`.
399    #[error("dispatcher error for ref '{ref_}': {msg}")]
400    DispatcherError {
401        /// The `Step.ref` that raised.
402        ref_: String,
403        /// The dispatcher's error message.
404        msg: String,
405    },
406    /// The [`Externs`] registry raised for the given `call_extern.ref` (e.g.
407    /// unregistered ref, or the extern fn itself returned an error).
408    #[error("extern error for ref '{ref_}': {msg}")]
409    ExternError {
410        /// The `call_extern.ref` that raised.
411        ref_: String,
412        /// The extern fn's error message.
413        msg: String,
414    },
415}
416
417// ──────────────────────────────────────────────────────────────────────────
418// Externs — whitelist registry for `call_extern` Expr (canonical opts.externs)
419// ──────────────────────────────────────────────────────────────────────────
420
421/// Extern registry: resolves a `call_extern.ref` against evaluated args and
422/// returns the value. Mirror of canonical `opts.externs` (flow-ir-lua
423/// `interpreter.lua`): each entry MUST be a pure function — no side effects,
424/// no flow control, value-shape manipulation only.
425///
426/// Same DI pattern as [`Dispatcher`]: host crates provide concrete
427/// implementations ([`ExternMap`] for plain Rust closures, mlua bridge for
428/// Lua functions upstream).
429pub trait Externs {
430    /// Invoke the extern registered under `ref_` with already-evaluated args.
431    /// Unregistered refs raise [`EvalError::ExternError`].
432    fn call(&self, ref_: &str, args: &[Value]) -> Result<Value, EvalError>;
433}
434
435/// Empty registry — every `call_extern` raises `ExternError` (parity with
436/// canonical "requires opts.externs" error). Used by the externs-less
437/// compat wrappers (`eval` / `eval_expr` / `eval_with_storage`).
438pub struct NoExterns;
439
440impl Externs for NoExterns {
441    fn call(&self, ref_: &str, _args: &[Value]) -> Result<Value, EvalError> {
442        Err(EvalError::ExternError {
443            ref_: ref_.into(),
444            msg: "no externs registry configured".into(),
445        })
446    }
447}
448
449/// Boxed pure extern function stored in [`ExternMap`].
450pub type ExternFn = Box<dyn Fn(&[Value]) -> Result<Value, EvalError> + Send + Sync>;
451
452/// `HashMap`-backed [`Externs`] impl for host-side Rust closures.
453///
454/// ```
455/// use flow_ir_core::{eval_expr_with_externs, EvalError, Expr, ExternMap};
456/// use serde_json::{json, Value};
457///
458/// let mut externs = ExternMap::new();
459/// externs.register("math.sqrt", |args: &[Value]| {
460///     let x = args
461///         .first()
462///         .and_then(|v| v.as_f64())
463///         .ok_or_else(|| EvalError::ExternError {
464///             ref_: "math.sqrt".into(),
465///             msg: "expected number".into(),
466///         })?;
467///     Ok(json!(x.sqrt()))
468/// });
469///
470/// let expr: Expr = serde_json::from_value(json!({
471///     "op": "call_extern", "ref": "math.sqrt",
472///     "args": [{ "op": "lit", "value": 9.0 }],
473/// })).unwrap();
474/// let out = eval_expr_with_externs(&expr, &json!({}), &externs).unwrap();
475/// assert_eq!(out, json!(3.0));
476/// ```
477#[derive(Default)]
478pub struct ExternMap {
479    fns: std::collections::HashMap<String, ExternFn>,
480}
481
482impl ExternMap {
483    /// Create an empty registry.
484    pub fn new() -> Self {
485        Self::default()
486    }
487
488    /// Register a pure function under `name` (overwrites an existing entry).
489    pub fn register<F>(&mut self, name: impl Into<String>, f: F)
490    where
491        F: Fn(&[Value]) -> Result<Value, EvalError> + Send + Sync + 'static,
492    {
493        self.fns.insert(name.into(), Box::new(f));
494    }
495
496    /// Whether `name` is registered (compile-time whitelist check parity).
497    pub fn contains(&self, name: &str) -> bool {
498        self.fns.contains_key(name)
499    }
500}
501
502impl Externs for ExternMap {
503    fn call(&self, ref_: &str, args: &[Value]) -> Result<Value, EvalError> {
504        let f = self.fns.get(ref_).ok_or_else(|| EvalError::ExternError {
505            ref_: ref_.into(),
506            msg: "not registered in externs".into(),
507        })?;
508        f(args)
509    }
510}
511
512// ──────────────────────────────────────────────────────────────────────────
513// CtxStorage — Backend DI for ctx state
514// ──────────────────────────────────────────────────────────────────────────
515
516/// Ctx backend trait — `eval(_with_storage)` 系が ctx state を touch する
517/// 唯一の経路。 `&self` write (interior mutability) で **走行中の Flow と
518/// 外部 task が同じ ctx を共有** できる (= dispatch().await suspend 中に外部
519/// task が `ctx.write` で State 注入 → resume 後 Step が read で観測、 という
520/// dynamic injection 経路を成立させる)。
521///
522/// Default impl は `MemoryCtx` (`Arc<Mutex<Value>>` wrapper、 既存
523/// `serde_json::Value` 直保持と挙動互換)。 consumer は typed struct / KV /
524/// 外部 store / observer wrap / event log 等を custom impl で持ち込める。
525pub trait CtxStorage: Send + Sync {
526    /// Read a single path (`$.a.b.c` 形式) from ctx.
527    fn read(&self, path: &str) -> Result<Value, EvalError>;
528    /// Write `value` to `path` (`$.a.b.c` 形式).
529    fn write(&self, path: &str, value: Value) -> Result<(), EvalError>;
530    /// Take a snapshot of the entire ctx (= Expr eval / Fanout fork で使う pure read view).
531    fn snapshot(&self) -> Value;
532    /// Replace the entire ctx with the given value (= Fanout branch restore 等).
533    fn replace(&self, value: Value);
534}
535
536/// Default `CtxStorage` impl — `Arc<Mutex<Value>>` wrapper。
537///
538/// Send + Sync かつ `&self` write OK = `Arc<MemoryCtx>` で外部 task と共有可能。
539pub struct MemoryCtx {
540    inner: std::sync::Mutex<Value>,
541}
542
543impl MemoryCtx {
544    /// Create a new MemoryCtx initialised with `ctx`.
545    pub fn new(ctx: Value) -> Self {
546        Self {
547            inner: std::sync::Mutex::new(ctx),
548        }
549    }
550
551    /// Convenience: wrap in `Arc<dyn CtxStorage>`.
552    pub fn shared(ctx: Value) -> std::sync::Arc<dyn CtxStorage> {
553        std::sync::Arc::new(Self::new(ctx))
554    }
555}
556
557impl CtxStorage for MemoryCtx {
558    fn read(&self, path: &str) -> Result<Value, EvalError> {
559        let guard = self.inner.lock().expect("ctx mutex poisoned");
560        read_path(path, &guard)
561    }
562
563    fn write(&self, path: &str, value: Value) -> Result<(), EvalError> {
564        let parsed: Path = path
565            .parse()
566            .map_err(|e: PathParseError| EvalError::InvalidPath(e.to_string()))?;
567        let mut guard = self.inner.lock().expect("ctx mutex poisoned");
568        parsed.write(&mut guard, value)
569    }
570
571    fn snapshot(&self) -> Value {
572        let guard = self.inner.lock().expect("ctx mutex poisoned");
573        guard.clone()
574    }
575
576    fn replace(&self, value: Value) {
577        let mut guard = self.inner.lock().expect("ctx mutex poisoned");
578        *guard = value;
579    }
580}
581
582/// Extract the already-parsed [`Path`] out of a `Path` `Expr`, or
583/// `InvalidPath` if `expr` is some other `Expr` variant. No parsing happens
584/// here — the `Path` was parsed once, at deserialize (or `Path::from_str`)
585/// time.
586fn path_of(expr: &Expr) -> Result<&Path, EvalError> {
587    match expr {
588        Expr::Path { at } => Ok(at),
589        _ => Err(EvalError::InvalidPath(
590            "expected Path expr for write target".into(),
591        )),
592    }
593}
594
595// ──────────────────────────────────────────────────────────────────────────
596// Evaluator — storage-backed (canonical) + legacy Value-passing wrapper
597// ──────────────────────────────────────────────────────────────────────────
598
599/// Storage-backed sync evaluator — `CtxStorage` 経由で ctx を touch する正本。
600///
601/// 各 Node 評価開始時に `ctx.snapshot()` で Expr eval 用の pure view を取り、
602/// write は `ctx.write(path, value)` 経由。 これにより同じ `Arc<dyn CtxStorage>`
603/// を共有する外部 task が、 Step 間 (sync の場合は 1 Step 評価内では touch
604/// しないが) や eval 間で ctx state を変更できる。
605pub fn eval_with_storage<D: Dispatcher>(
606    node: &Node,
607    ctx: &dyn CtxStorage,
608    dispatcher: &D,
609) -> Result<(), EvalError> {
610    eval_with_storage_externs(node, ctx, dispatcher, &NoExterns)
611}
612
613/// `eval_with_storage` + externs registry for `call_extern` Expr resolution.
614pub fn eval_with_storage_externs<D: Dispatcher>(
615    node: &Node,
616    ctx: &dyn CtxStorage,
617    dispatcher: &D,
618    externs: &dyn Externs,
619) -> Result<(), EvalError> {
620    match node {
621        Node::Step { ref_, in_, out } => {
622            let snap = ctx.snapshot();
623            let input = eval_expr_with_externs(in_, &snap, externs)?;
624            let output =
625                dispatcher
626                    .dispatch(ref_, input)
627                    .map_err(|e| EvalError::DispatcherError {
628                        ref_: ref_.clone(),
629                        msg: e.to_string(),
630                    })?;
631            ctx.write(&path_of(out)?.to_string(), output)
632        }
633        Node::Seq { children } => {
634            for child in children {
635                eval_with_storage_externs(child, ctx, dispatcher, externs)?;
636            }
637            Ok(())
638        }
639        Node::Branch { cond, then_, else_ } => {
640            let snap = ctx.snapshot();
641            match eval_expr_with_externs(cond, &snap, externs)? {
642                Value::Bool(true) => eval_with_storage_externs(then_, ctx, dispatcher, externs),
643                Value::Bool(false) => eval_with_storage_externs(else_, ctx, dispatcher, externs),
644                other => Err(EvalError::NonBoolCond(other)),
645            }
646        }
647        Node::Fanout {
648            items,
649            bind,
650            body,
651            join,
652            out,
653        } => {
654            // Fanout fork = 各 branch を disjoint MemoryCtx に切り出して逐次
655            // (sync) evaluate、 集約結果を共有 ctx の `out` path に書く。
656            let snap = ctx.snapshot();
657            let items_val = eval_expr_with_externs(items, &snap, externs)?;
658            let items_arr = match items_val {
659                Value::Array(a) => a,
660                other => {
661                    return Err(EvalError::TypeError {
662                        op: "fanout.items".into(),
663                        msg: format!("expected array, got {other:?}"),
664                    })
665                }
666            };
667            let joined =
668                fanout_eval_sync(bind, body, *join, &snap, items_arr, dispatcher, externs)?;
669            ctx.write(&path_of(out)?.to_string(), joined)
670        }
671        Node::Loop {
672            counter,
673            cond,
674            body,
675            max,
676        } => {
677            let counter_path = path_of(counter)?.to_string();
678            ctx.write(&counter_path, Value::Number(serde_json::Number::from(0u32)))?;
679            let mut n: u32 = 0;
680            loop {
681                if n >= *max {
682                    break;
683                }
684                let snap = ctx.snapshot();
685                if !is_truthy(&eval_expr_with_externs(cond, &snap, externs)?) {
686                    break;
687                }
688                eval_with_storage_externs(body, ctx, dispatcher, externs)?;
689                n += 1;
690                ctx.write(&counter_path, Value::Number(serde_json::Number::from(n)))?;
691            }
692            Ok(())
693        }
694        Node::Try {
695            body,
696            catch,
697            err_at,
698        } => {
699            // body 失敗時の rollback 用 snapshot
700            let snap_before = ctx.snapshot();
701            match eval_with_storage_externs(body, ctx, dispatcher, externs) {
702                Ok(()) => Ok(()),
703                Err(e) => {
704                    // body の途中 write を破棄 (Try semantic: rollback)
705                    ctx.replace(snap_before);
706                    if let Some(at) = err_at {
707                        ctx.write(&path_of(at)?.to_string(), Value::String(e.to_string()))?;
708                    }
709                    eval_with_storage_externs(catch, ctx, dispatcher, externs)
710                }
711            }
712        }
713        Node::Let { at, value } => {
714            let snap = ctx.snapshot();
715            let v = eval_expr_with_externs(value, &snap, externs)?;
716            ctx.write(&at.to_string(), v)
717        }
718    }
719}
720
721/// Internal: fanout per-item sync evaluator (disjoint branch ctx).
722fn fanout_eval_sync<D: Dispatcher>(
723    bind: &Expr,
724    body: &Node,
725    join: JoinMode,
726    base_snap: &Value,
727    items_arr: Vec<Value>,
728    dispatcher: &D,
729    externs: &dyn Externs,
730) -> Result<Value, EvalError> {
731    match join {
732        JoinMode::All => {
733            let mut results = Vec::with_capacity(items_arr.len());
734            for item in items_arr {
735                let branch_ctx = write_path(bind, base_snap.clone(), item)?;
736                let storage = MemoryCtx::new(branch_ctx);
737                eval_with_storage_externs(body, &storage, dispatcher, externs)?;
738                results.push(storage.snapshot());
739            }
740            Ok(Value::Array(results))
741        }
742        JoinMode::Any => {
743            // Promise.any parity: zero items can never produce a winner, so
744            // (unlike All/AllSettled, whose empty-array result shape is
745            // still meaningful) this raises rather than returning `[]`.
746            if items_arr.is_empty() {
747                return Err(EvalError::TypeError {
748                    op: "fanout.any".into(),
749                    msg: "requires at least one item".into(),
750                });
751            }
752            let mut winner: Option<Value> = None;
753            let mut last_err: Option<EvalError> = None;
754            for item in items_arr {
755                let branch_ctx = write_path(bind, base_snap.clone(), item)?;
756                let storage = MemoryCtx::new(branch_ctx);
757                match eval_with_storage_externs(body, &storage, dispatcher, externs) {
758                    Ok(()) => {
759                        winner = Some(storage.snapshot());
760                        last_err = None;
761                        break;
762                    }
763                    Err(e) => last_err = Some(e),
764                }
765            }
766            if let Some(e) = last_err {
767                return Err(e);
768            }
769            Ok(winner.expect("non-empty items_arr always assigns a winner or returns an error"))
770        }
771        JoinMode::Race => {
772            // Same rationale as Any: zero branches means there is nothing to
773            // race, so this raises rather than returning `[]`.
774            let Some(first) = items_arr.into_iter().next() else {
775                return Err(EvalError::TypeError {
776                    op: "fanout.race".into(),
777                    msg: "requires at least one item".into(),
778                });
779            };
780            let branch_ctx = write_path(bind, base_snap.clone(), first)?;
781            let storage = MemoryCtx::new(branch_ctx);
782            eval_with_storage_externs(body, &storage, dispatcher, externs)?;
783            Ok(storage.snapshot())
784        }
785        JoinMode::AllSettled => {
786            let mut records = Vec::with_capacity(items_arr.len());
787            for item in items_arr {
788                let branch_ctx = write_path(bind, base_snap.clone(), item)?;
789                let storage = MemoryCtx::new(branch_ctx);
790                match eval_with_storage_externs(body, &storage, dispatcher, externs) {
791                    Ok(()) => records.push(
792                        serde_json::json!({"status": "fulfilled", "value": storage.snapshot()}),
793                    ),
794                    Err(e) => records
795                        .push(serde_json::json!({"status": "rejected", "reason": e.to_string()})),
796                }
797            }
798            Ok(Value::Array(records))
799        }
800    }
801}
802
803/// Legacy Value-passing sync evaluator — backward compat wrapper around
804/// `eval_with_storage` + `MemoryCtx`. `Value` を所有権で受け取り、 内部で
805/// `MemoryCtx::new(ctx)` を使って storage 版に委譲、 終了後の snapshot を返す。
806///
807/// 既存 caller (= dynamic injection を要求しない、 1-shot pure eval 用途) は
808/// 引き続きこの API で OK。 動的注入が要る場合は `eval_with_storage` を直接
809/// 呼ぶ。
810///
811/// Returns the updated context (= ctx with `Step.out` path written for each step traversed).
812pub fn eval<D: Dispatcher>(node: &Node, ctx: Value, dispatcher: &D) -> Result<Value, EvalError> {
813    eval_externs(node, ctx, dispatcher, &NoExterns)
814}
815
816/// `eval` + externs registry for `call_extern` Expr resolution.
817pub fn eval_externs<D: Dispatcher>(
818    node: &Node,
819    ctx: Value,
820    dispatcher: &D,
821    externs: &dyn Externs,
822) -> Result<Value, EvalError> {
823    let storage = MemoryCtx::new(ctx);
824    eval_with_storage_externs(node, &storage, dispatcher, externs)?;
825    Ok(storage.snapshot())
826}
827
828/// JSON value の truthy 判定 (= flow.ir Branch cond / Loop cond で使う)。
829/// Bool は値そのまま、 null/false 以外は truthy (Lua / JS と整合)。
830pub fn is_truthy(v: &Value) -> bool {
831    match v {
832        Value::Null => false,
833        Value::Bool(b) => *b,
834        _ => true,
835    }
836}
837
838/// Evaluate an `Expr` against a context value, returning the resolved JSON
839/// value. Externs-less compat wrapper — `call_extern` raises `ExternError`.
840pub fn eval_expr(expr: &Expr, ctx: &Value) -> Result<Value, EvalError> {
841    eval_expr_with_externs(expr, ctx, &NoExterns)
842}
843
844/// `eval_expr` + externs registry for `call_extern` Expr resolution.
845pub fn eval_expr_with_externs(
846    expr: &Expr,
847    ctx: &Value,
848    externs: &dyn Externs,
849) -> Result<Value, EvalError> {
850    let ev = |e: &Expr| eval_expr_with_externs(e, ctx, externs);
851    match expr {
852        Expr::Lit { value } => Ok(value.clone()),
853        // `at` is an already-parsed `Path` — no re-parse in this hot path.
854        Expr::Path { at } => at.read(ctx).cloned(),
855        Expr::Eq { lhs, rhs } => Ok(Value::Bool(json_eq(&ev(lhs)?, &ev(rhs)?))),
856        Expr::Ne { lhs, rhs } => Ok(Value::Bool(!json_eq(&ev(lhs)?, &ev(rhs)?))),
857        Expr::Lt { lhs, rhs } => ord_cmp(&ev(lhs)?, &ev(rhs)?, |o| o.is_lt()),
858        Expr::Lte { lhs, rhs } => ord_cmp(&ev(lhs)?, &ev(rhs)?, |o| o.is_le()),
859        Expr::Gt { lhs, rhs } => ord_cmp(&ev(lhs)?, &ev(rhs)?, |o| o.is_gt()),
860        Expr::Gte { lhs, rhs } => ord_cmp(&ev(lhs)?, &ev(rhs)?, |o| o.is_ge()),
861        Expr::Not { arg } => Ok(Value::Bool(!is_truthy(&ev(arg)?))),
862        Expr::And { args } => {
863            for a in args {
864                if !is_truthy(&ev(a)?) {
865                    return Ok(Value::Bool(false));
866                }
867            }
868            Ok(Value::Bool(true))
869        }
870        Expr::Or { args } => {
871            for a in args {
872                if is_truthy(&ev(a)?) {
873                    return Ok(Value::Bool(true));
874                }
875            }
876            Ok(Value::Bool(false))
877        }
878        Expr::Exists { arg } => match ev(arg) {
879            Ok(Value::Null) => Ok(Value::Bool(false)),
880            Ok(_) => Ok(Value::Bool(true)),
881            // canonical: a path to a missing key reads as nil → exists=false
882            Err(EvalError::PathNotFound(_)) => Ok(Value::Bool(false)),
883            Err(e) => Err(e),
884        },
885        Expr::Add { lhs, rhs } => num_arith(&ev(lhs)?, &ev(rhs)?, "add", |a, b| Some(a + b)),
886        Expr::Sub { lhs, rhs } => num_arith(&ev(lhs)?, &ev(rhs)?, "sub", |a, b| Some(a - b)),
887        Expr::Mul { lhs, rhs } => num_arith(&ev(lhs)?, &ev(rhs)?, "mul", |a, b| Some(a * b)),
888        Expr::Div { lhs, rhs } => num_arith(&ev(lhs)?, &ev(rhs)?, "div", |a, b| {
889            if b == 0.0 {
890                None
891            } else {
892                Some(a / b)
893            }
894        }),
895        // Lua `%` semantics (canonical): a - floor(a/b)*b, sign follows rhs.
896        Expr::Mod { lhs, rhs } => num_arith(&ev(lhs)?, &ev(rhs)?, "mod", |a, b| {
897            if b == 0.0 {
898                None
899            } else {
900                Some(a - (a / b).floor() * b)
901            }
902        }),
903        Expr::Len { arg } => {
904            let v = ev(arg)?;
905            let n = match &v {
906                Value::Array(a) => a.len(),
907                Value::String(s) => s.chars().count(),
908                Value::Object(o) => o.len(),
909                other => {
910                    return Err(EvalError::TypeError {
911                        op: "expr.len".into(),
912                        msg: format!("len: unsupported type {other:?}"),
913                    })
914                }
915            };
916            Ok(Value::Number(serde_json::Number::from(n as u64)))
917        }
918        Expr::In { needle, haystack } => {
919            let n = ev(needle)?;
920            let h = ev(haystack)?;
921            match h {
922                Value::Array(a) => Ok(Value::Bool(a.iter().any(|e| json_eq(e, &n)))),
923                other => Err(EvalError::TypeError {
924                    op: "expr.in".into(),
925                    msg: format!("in: haystack must be array, got {other:?}"),
926                }),
927            }
928        }
929        Expr::CallExtern { ref_, args } => {
930            let mut vals = Vec::with_capacity(args.len());
931            for a in args {
932                vals.push(ev(a)?);
933            }
934            externs.call(ref_, &vals)
935        }
936    }
937}
938
939/// Deep equality with Lua-parity numeric coercion: numbers compare by
940/// f64 value (so `5 == 5.0`), matching the coercion Lt/Lte/Gt/Gte use.
941/// Integers above 2^53 may lose precision — same caveat as ordering ops.
942fn json_eq(a: &Value, b: &Value) -> bool {
943    match (a, b) {
944        (Value::Number(na), Value::Number(nb)) => match (na.as_f64(), nb.as_f64()) {
945            (Some(fa), Some(fb)) => fa == fb,
946            // non-f64-representable (e.g. integers beyond 2^53): fall back
947            // to exact comparison rather than a lossy f64 round-trip.
948            _ => na == nb,
949        },
950        (Value::Array(aa), Value::Array(ab)) => {
951            aa.len() == ab.len() && aa.iter().zip(ab.iter()).all(|(x, y)| json_eq(x, y))
952        }
953        (Value::Object(oa), Value::Object(ob)) => {
954            oa.len() == ob.len()
955                && oa
956                    .iter()
957                    .all(|(k, v)| ob.get(k).is_some_and(|ov| json_eq(v, ov)))
958        }
959        _ => a == b,
960    }
961}
962
963/// Coerce a JSON value to f64 for numeric ops. A non-`Number` value is a
964/// `TypeError` (wrong JSON type); a `Number` that itself cannot be
965/// represented as `f64` (e.g. an integer beyond `f64`'s exact range) is an
966/// `ArithError` (right type, unrepresentable value).
967fn to_f64(v: &Value, op: &str) -> Result<f64, EvalError> {
968    match v {
969        Value::Number(n) => n.as_f64().ok_or_else(|| EvalError::ArithError {
970            op: op.into(),
971            msg: format!("non-f64-representable number: {n}"),
972        }),
973        other => Err(EvalError::TypeError {
974            op: op.into(),
975            msg: format!("expected number, got {other:?}"),
976        }),
977    }
978}
979
980/// Ordering comparison over two evaluated values. Mirrors canonical Lua
981/// `< / <= / > / >=`: both numbers (f64) or both strings (lexicographic
982/// byte order, same as Lua's string comparison for UTF-8); anything else
983/// raises.
984fn ord_cmp<F>(lv: &Value, rv: &Value, f: F) -> Result<Value, EvalError>
985where
986    F: Fn(std::cmp::Ordering) -> bool,
987{
988    let ord = match (lv, rv) {
989        (Value::Number(_), Value::Number(_)) => {
990            let l = to_f64(lv, "cmp")?;
991            let r = to_f64(rv, "cmp")?;
992            l.partial_cmp(&r).ok_or_else(|| EvalError::ArithError {
993                op: "cmp".into(),
994                msg: "non-comparable numbers (NaN)".into(),
995            })?
996        }
997        (Value::String(l), Value::String(r)) => l.cmp(r),
998        (l, r) => {
999            return Err(EvalError::TypeError {
1000                op: "cmp".into(),
1001                msg: format!("cmp: both sides must be numbers or strings, got {l:?} vs {r:?}"),
1002            })
1003        }
1004    };
1005    Ok(Value::Bool(f(ord)))
1006}
1007
1008fn num_arith<F>(lv: &Value, rv: &Value, op: &str, f: F) -> Result<Value, EvalError>
1009where
1010    F: Fn(f64, f64) -> Option<f64>,
1011{
1012    let l = to_f64(lv, op)?;
1013    let r = to_f64(rv, op)?;
1014    let result = f(l, r).ok_or_else(|| EvalError::ArithError {
1015        op: op.into(),
1016        msg: "arithmetic failure (e.g. division by zero)".into(),
1017    })?;
1018    let n = serde_json::Number::from_f64(result).ok_or_else(|| EvalError::ArithError {
1019        op: op.into(),
1020        msg: format!("result not f64-representable: {result}"),
1021    })?;
1022    Ok(Value::Number(n))
1023}
1024
1025// ──────────────────────────────────────────────────────────────────────────
1026// Path compat wrappers — thin `&str` entry points over the typed `Path` in
1027// `path.rs`, which is the single authority for path syntax + rejection
1028// rules. See [`Path`] docs for the full syntax (dot form, RFC 9535-style
1029// bracket notation, uniform malformed-path rejections).
1030// ──────────────────────────────────────────────────────────────────────────
1031
1032/// Parse `path` and read the value it resolves to inside `ctx`.
1033///
1034/// Thin wrapper: parses `path` via [`str::parse`] (surfacing a parse
1035/// failure as [`EvalError::InvalidPath`]) then delegates to [`Path::read`].
1036/// Callers holding an already-parsed `Path` (e.g. from an `Expr::Path`)
1037/// should call [`Path::read`] directly to avoid re-parsing.
1038pub fn read_path(path: &str, ctx: &Value) -> Result<Value, EvalError> {
1039    let parsed: Path = path
1040        .parse()
1041        .map_err(|e: PathParseError| EvalError::InvalidPath(e.to_string()))?;
1042    parsed.read(ctx).cloned()
1043}
1044
1045/// Write a value at the path location inside ctx, returning the updated ctx.
1046/// `out` must be a `Path` Expr (its `at` field is an already-parsed `Path`,
1047/// so no re-parsing happens here — this is a thin wrapper around
1048/// [`Path::write`], adapted to the `Value`-in/`Value`-out shape the rest of
1049/// this crate's legacy (non-`CtxStorage`) API uses).
1050pub fn write_path(out: &Expr, ctx: Value, value: Value) -> Result<Value, EvalError> {
1051    let path = path_of(out)?;
1052    let mut root = ctx;
1053    path.write(&mut root, value)?;
1054    Ok(root)
1055}