mlua-swarm-dsl 0.23.1

//! Lua internal DSL for Blueprint authoring (`flow_dsl` + `bp_dsl`).
//!
//! Raw AST (JSON) authoring cost is high at real Blueprint scale (hundreds
//! of lines and deep nesting for a multi-stage flow) — that cost motivated
//! this module. `flow_dsl.lua` (flow.ir vocabulary) and `bp_dsl.lua`
//! (Blueprint vocabulary, depends on `flow_dsl`) are baked into this
//! binary via `include_str!` and preloaded into a fresh `mlua::Lua` VM so
//! `require("flow_dsl")` / `require("bp_dsl")` resolve without touching
//! the filesystem. The `flow-ir` / `mlua-swarm-schema` crates are not
//! touched by this module — canonical JSON stays the wire format; the DSL
//! is purely an authoring-time convenience that emits it.
//!
//! # Crate positioning
//!
//! This crate is the DSL frontend only (`.bp.lua` → `serde_json::Value`).
//! The compile pipeline (linker → shape lint → BPReady) lives in the
//! sibling `mlua-swarm-compile` crate, which consumes the JSON this crate
//! produces. `mlua-swarm-schema` (types) stays free of the `mlua` runtime
//! dep so that consumers who only need type surfaces (e.g. the server's
//! wire codec) do not transitively pull the Lua interpreter.

const FLOW_DSL_SRC: &str = include_str!("flow_dsl.lua");
const BP_DSL_SRC: &str = include_str!("bp_dsl.lua");

/// The wire-level key `F.obj()` (`flow_dsl.lua`) emits — must match the
/// Lua-side `M.EMPTY_OBJECT_MARKER_KEY` literal exactly.
const EMPTY_OBJECT_MARKER_KEY: &str = "__mse_empty_object__";

/// Walk `value` in place and replace every JSON object shaped exactly
/// like `{ "<EMPTY_OBJECT_MARKER_KEY>": true }` (the wire shape `F.obj()`
/// emits) with a genuine empty JSON object (`{}`). Limited to that exact
/// single-key shape so an ordinary data field that happens to carry a key
/// with the same name is left untouched.
fn replace_empty_object_markers(value: &mut serde_json::Value) {
    match value {
        serde_json::Value::Object(map) => {
            let is_marker = map.len() == 1
                && map.get(EMPTY_OBJECT_MARKER_KEY) == Some(&serde_json::Value::Bool(true));
            if is_marker {
                *value = serde_json::Value::Object(serde_json::Map::new());
                return;
            }
            for v in map.values_mut() {
                replace_empty_object_markers(v);
            }
        }
        serde_json::Value::Array(arr) => {
            for v in arr.iter_mut() {
                replace_empty_object_markers(v);
            }
        }
        _ => {}
    }
}

/// Register `flow_dsl` and `bp_dsl` in `lua`'s `package.preload` table so
/// `require("flow_dsl")` / `require("bp_dsl")` resolve to the baked-in Lua
/// source. Idempotent to call more than once on the same `Lua` (each call
/// simply re-sets the same two `preload` entries).
pub fn preload(lua: &mlua::Lua) -> mlua::Result<()> {
    let package: mlua::Table = lua.globals().get("package")?;
    let preload: mlua::Table = package.get("preload")?;

    preload.set(
        "flow_dsl",
        lua.create_function(|lua, ()| {
            lua.load(FLOW_DSL_SRC)
                .set_name("flow_dsl.lua")
                .eval::<mlua::Value>()
        })?,
    )?;
    preload.set(
        "bp_dsl",
        lua.create_function(|lua, ()| {
            lua.load(BP_DSL_SRC)
                .set_name("bp_dsl.lua")
                .eval::<mlua::Value>()
        })?,
    )?;
    Ok(())
}

/// Run a `.bp.lua` DSL script (source text, not a file path) in a fresh
/// `mlua::Lua` VM and return its result as `serde_json::Value`.
///
/// The script is expected to `require("flow_dsl")` and/or
/// `require("bp_dsl")` and `return` a Blueprint-shaped (or Expr/Node
/// -shaped, for narrower scripts) Lua table as its last expression.
///
/// Empty Lua tables are treated as empty JSON arrays rather than empty
/// objects (`encode_empty_tables_as_array`) — every plain empty table this
/// DSL can emit is a `Node`/`Expr` list field (`seq.children`, `and.args`,
/// `or.args`), never a legitimately-empty JSON object. A field that must
/// serialize as an empty JSON object uses the `F.obj()` marker
/// (`flow_dsl.lua`) instead of a bare `{}` table literal; this function
/// replaces every occurrence of that marker with a genuine empty JSON
/// object as a post-pass (`replace_empty_object_markers`) over the
/// converted value.
pub fn build_bp_from_script(script: &str) -> anyhow::Result<serde_json::Value> {
    Ok(build_bp_from_script_with_warnings(script)?.0)
}

/// Like [`build_bp_from_script`], but also drains the authoring-time
/// warnings `bp_dsl.lua` accumulated during the run (currently the
/// B.pipeline dead-halt lint: pipeline-level `halt_on` with zero
/// gate-emitting stages). Best-effort: a script that never
/// `require`s `bp_dsl` yields an empty list.
pub fn build_bp_from_script_with_warnings(
    script: &str,
) -> anyhow::Result<(serde_json::Value, Vec<String>)> {
    use mlua::LuaSerdeExt;

    // `mlua::Error` wraps a boxed `dyn std::error::Error` without a
    // `Send + Sync` bound, so it does not satisfy anyhow's blanket `From`
    // impl (`?` cannot convert it directly) — stringify explicitly instead.
    let lua = mlua::Lua::new();
    preload(&lua).map_err(|e| anyhow::anyhow!("dsl preload failed: {e}"))?;
    let result: mlua::Value = lua
        .load(script)
        .set_name("<bp-script>")
        .eval()
        .map_err(|e| anyhow::anyhow!("bp-script eval failed: {e}"))?;
    let options = mlua::serde::de::Options::new().encode_empty_tables_as_array(true);
    let mut value: serde_json::Value = lua
        .from_value_with(result, options)
        .map_err(|e| anyhow::anyhow!("lua value -> json conversion failed: {e}"))?;
    replace_empty_object_markers(&mut value);
    let warnings = drain_authoring_warnings(&lua);
    Ok((value, warnings))
}

/// Best-effort drain of `bp_dsl`'s authoring-warning buffer from the VM the
/// script just ran in: `package.loaded["bp_dsl"].take_authoring_warnings()`.
/// A script that never required the module (or any unexpected shape) yields
/// an empty list — a reporting-only lint must never fail a build.
fn drain_authoring_warnings(lua: &mlua::Lua) -> Vec<String> {
    let drained: mlua::Result<Vec<String>> = (|| {
        let package: mlua::Table = lua.globals().get("package")?;
        let loaded: mlua::Table = package.get("loaded")?;
        let module: mlua::Value = loaded.get("bp_dsl")?;
        let mlua::Value::Table(module) = module else {
            return Ok(Vec::new());
        };
        let take: mlua::Function = module.get("take_authoring_warnings")?;
        let list: mlua::Table = take.call(())?;
        let mut out = Vec::new();
        for entry in list.sequence_values::<String>() {
            out.push(entry?);
        }
        Ok(out)
    })();
    drained.unwrap_or_default()
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn preload_exposes_flow_dsl_and_bp_dsl() {
        let lua = mlua::Lua::new();
        preload(&lua).expect("preload must succeed");
        let ok: bool = lua
            .load(
                r#"
                local F = require("flow_dsl")
                local B = require("bp_dsl")
                return F ~= nil and B ~= nil
                "#,
            )
            .eval()
            .expect("require must succeed for both modules");
        assert!(ok, "flow_dsl / bp_dsl must both resolve via require()");
    }

    #[test]
    fn build_bp_from_script_returns_json_value() {
        let out = build_bp_from_script(
            r#"
            local F = require("flow_dsl")
            return { id = "t", flow = F.assign{ at = F.p("$.x"), value = F.lit(1) } }
            "#,
        )
        .expect("script must build");
        assert_eq!(out["id"], serde_json::json!("t"));
        assert_eq!(out["flow"]["kind"], serde_json::json!("assign"));
        assert_eq!(
            out["flow"]["at"],
            serde_json::json!({"op": "path", "at": "$.x"})
        );
    }

    #[test]
    fn build_bp_from_script_surfaces_lua_errors() {
        let err = build_bp_from_script("error(\"boom\")").expect_err("must propagate the error");
        assert!(err.to_string().contains("boom"));
    }

    #[test]
    fn f_obj_marker_becomes_a_genuine_empty_json_object() {
        let out = build_bp_from_script(
            r#"
            local F = require("flow_dsl")
            return { spec = F.obj(), other = {} }
            "#,
        )
        .expect("script must build");
        assert_eq!(out["spec"], serde_json::json!({}));
        assert!(
            out["spec"].is_object(),
            "F.obj() must become an object, not an array"
        );
        // A plain empty Lua table is still converted to an empty JSON
        // array (the pre-existing `encode_empty_tables_as_array` rule),
        // proving the marker replacement is scoped to `F.obj()`'s exact
        // one-key shape and does not affect ordinary empty tables.
        assert_eq!(out["other"], serde_json::json!([]));
    }

    /// GH #76 DSL sugar: `skip_on = { "SKIP", ... }` on a stage record wraps
    /// the stage's own body (step + optional retry loop) in a Branch
    /// whose `cond` is `in(<input>.parts["verdict"], <skip_on_list>)`.
    /// When the skip check hits, the stage body is elided (`then` =
    /// empty `Seq`); the gate/rest chain continues unchanged.
    #[test]
    fn bp_dsl_skip_on_compiles_to_branch_in_verdict_skip_on_list() {
        let out = build_bp_from_script(
            r#"
            local F = require("flow_dsl")
            local B = require("bp_dsl")
            return B.pipeline({
              B.stage "gate" { agent = "mock-gate" },
              B.stage "worker" {
                agent = "mock-worker",
                input = B.from "gate",
                skip_on = { "SKIP", "NOT_APPLICABLE" },
              },
              halted_at = "$.halted_at",
            })
            "#,
        )
        .expect("skip_on pipeline must build");

        // Top-level seq: [gate_step, rest].
        assert_eq!(out["kind"], serde_json::json!("seq"));
        let top_children = out["children"].as_array().expect("top seq children");
        assert_eq!(top_children.len(), 2);
        assert_eq!(top_children[0]["kind"], serde_json::json!("step"));
        assert_eq!(top_children[0]["ref"], serde_json::json!("mock-gate"));

        // `rest` = worker stage's compiled form. With skip_on, the
        // stage's body is wrapped in a branch whose cond is `in(...)`.
        let rest = &top_children[1];
        assert_eq!(rest["kind"], serde_json::json!("seq"));
        let rest_children = rest["children"].as_array().expect("rest seq children");
        // No gate/rest chain past worker (last stage, no halt_on / retry).
        let worker_guarded = &rest_children[0];
        assert_eq!(worker_guarded["kind"], serde_json::json!("branch"));

        // cond: in(needle=path("$.gate.parts[\"verdict\"]"),
        //         haystack=lit(["SKIP", "NOT_APPLICABLE"])).
        let cond = &worker_guarded["cond"];
        assert_eq!(cond["op"], serde_json::json!("in"));
        assert_eq!(
            cond["needle"],
            serde_json::json!({"op": "path", "at": "$.gate.parts[\"verdict\"]"})
        );
        assert_eq!(cond["haystack"]["op"], serde_json::json!("lit"));
        assert_eq!(
            cond["haystack"]["value"],
            serde_json::json!(["SKIP", "NOT_APPLICABLE"])
        );

        // then = empty seq (skip elides body).
        assert_eq!(
            worker_guarded["then"],
            serde_json::json!({"kind": "seq", "children": []})
        );

        // else = the original stage body (just the worker step here —
        // no retry, no gate).
        let body = &worker_guarded["else"];
        assert_eq!(body["kind"], serde_json::json!("seq"));
        assert_eq!(body["children"][0]["ref"], serde_json::json!("mock-worker"));
    }

    /// GH #76 DSL sugar: `skip_on` may coexist with `halt_on` on the same
    /// stage — the skip guard wraps the stage's OWN body (step +
    /// optional retry loop) and sits INSIDE the enclosing gate/rest
    /// chain, so a skipped stage still lets `halt_on`'s gate cond be
    /// evaluated against the (absent) `<out>` and thread through to
    /// `rest`.
    #[test]
    fn bp_dsl_skip_on_coexists_with_halt_on() {
        let out = build_bp_from_script(
            r#"
            local F = require("flow_dsl")
            local B = require("bp_dsl")
            return B.pipeline({
              B.stage "planner" { agent = "mock-planner" },
              B.stage "worker" {
                agent = "mock-worker",
                input = B.from "planner",
                skip_on = { "SKIP" },
                halt_on = { "BLOCKED" },
              },
              B.stage "publisher" { agent = "mock-publisher" },
              halted_at = "$.halted_at",
            })
            "#,
        )
        .expect("skip_on + halt_on pipeline must build");

        // Walk to the worker stage. Structure: top seq -> [planner,
        // rest]; rest = seq -> [worker_body, gate]; worker_body =
        // branch (skip guard).
        let rest = &out["children"][1];
        let worker_seq = rest;
        assert_eq!(worker_seq["kind"], serde_json::json!("seq"));
        let worker_children = worker_seq["children"]
            .as_array()
            .expect("worker seq children");
        assert_eq!(
            worker_children.len(),
            2,
            "skip guard + halt_on gate (with publisher threaded into gate else)"
        );

        // Child 0 = skip guard branch (skip_on).
        let skip_branch = &worker_children[0];
        assert_eq!(skip_branch["kind"], serde_json::json!("branch"));
        assert_eq!(skip_branch["cond"]["op"], serde_json::json!("in"));

        // Child 1 = halt_on gate (`branch`) whose cond is `eq` against
        // the current stage's own out.parts["verdict"].
        let halt_gate = &worker_children[1];
        assert_eq!(halt_gate["kind"], serde_json::json!("branch"));
        assert_eq!(halt_gate["cond"]["op"], serde_json::json!("eq"));
        assert_eq!(
            halt_gate["cond"]["lhs"],
            serde_json::json!({"op": "path", "at": "$.worker.parts[\"verdict\"]"})
        );
        // gate's else is publisher's compiled form (the pipeline tail).
        let gate_else = &halt_gate["else"];
        assert_eq!(gate_else["kind"], serde_json::json!("seq"));
        // publisher's step should be somewhere in that seq's children.
        let contains_publisher = gate_else["children"]
            .as_array()
            .map(|arr| {
                arr.iter()
                    .any(|c| c["ref"] == serde_json::json!("mock-publisher"))
            })
            .unwrap_or(false);
        assert!(
            contains_publisher,
            "halt_on gate else must thread the publisher stage through: {gate_else}"
        );
    }

    /// GH #76 DSL sugar: `skip_on = {}` is a no-op (equivalent to omitting
    /// the option). No branch is emitted, the stage compiles exactly
    /// as if `skip_on` were absent.
    #[test]
    fn bp_dsl_skip_on_empty_list_is_noop() {
        let with_empty = build_bp_from_script(
            r#"
            local B = require("bp_dsl")
            return B.pipeline({
              B.stage "worker" { agent = "mock-worker", skip_on = {} },
              halted_at = "$.halted_at",
            })
            "#,
        )
        .expect("skip_on={} pipeline must build");

        // Without any gate, retry, or a firing skip_on, the pipeline
        // compiles to [step, <final_else>] (no branch wrapping).
        let children = with_empty["children"].as_array().expect("seq children");
        assert_eq!(
            children.len(),
            2,
            "no skip guard emitted for empty skip_on: {with_empty}"
        );
        assert_eq!(children[0]["kind"], serde_json::json!("step"));
        assert_eq!(children[0]["ref"], serde_json::json!("mock-worker"));

        // Byte-identical to the same script without skip_on.
        let baseline = build_bp_from_script(
            r#"
            local B = require("bp_dsl")
            return B.pipeline({
              B.stage "worker" { agent = "mock-worker" },
              halted_at = "$.halted_at",
            })
            "#,
        )
        .expect("baseline pipeline must build");
        assert_eq!(with_empty, baseline, "skip_on = {{}} must be a no-op");
    }

    /// Build `script` and return only the authoring warnings it produced.
    fn warnings_for(script: &str) -> Vec<String> {
        build_bp_from_script_with_warnings(script)
            .expect("script must build")
            .1
    }

    /// The dead-halt lint: pipeline-level `halt_on` with zero
    /// gate-emitting stages compiles to a flow that can never halt, so
    /// one WARN line is emitted naming the stages and the halt values.
    #[test]
    fn dead_halt_lint_warns_when_pipeline_halt_on_has_no_gating_stage() {
        let warnings = warnings_for(
            r#"
            local B = require("bp_dsl")
            return B.pipeline({
              B.stage "review" { agent = "mock-review" },
              halt_on = { "BLOCKED" },
              halted_at = "$.halted_at",
            })
            "#,
        );
        assert_eq!(
            warnings.len(),
            1,
            "exactly one dead-halt WARN: {warnings:?}"
        );
        let w = &warnings[0];
        assert!(w.contains("can never halt"), "{w}");
        assert!(w.contains("review"), "must name the stage id: {w}");
        assert!(w.contains("BLOCKED"), "must name the halt values: {w}");
    }

    /// `gate = true` on any stage is an explicit opt-in — the pipeline can
    /// halt, so the lint stays silent.
    #[test]
    fn dead_halt_lint_silent_when_a_stage_opts_in_with_gate_true() {
        let warnings = warnings_for(
            r#"
            local B = require("bp_dsl")
            return B.pipeline({
              B.stage "review" { agent = "mock-review", gate = true },
              halt_on = { "BLOCKED" },
              halted_at = "$.halted_at",
            })
            "#,
        );
        assert!(warnings.is_empty(), "gate = true opts in: {warnings:?}");
    }

    /// `gate_default = "auto"` restores the pre-flip cascade, so every
    /// stage gates and the lint stays silent.
    #[test]
    fn dead_halt_lint_silent_under_gate_default_auto() {
        let warnings = warnings_for(
            r#"
            local B = require("bp_dsl")
            return B.pipeline({
              B.stage "review" { agent = "mock-review" },
              halt_on = { "BLOCKED" },
              halted_at = "$.halted_at",
              gate_default = "auto",
            })
            "#,
        );
        assert!(
            warnings.is_empty(),
            "auto cascade gates every stage: {warnings:?}"
        );
    }

    /// `retry` implies a gate (the retry loop reads verdict and the
    /// post-retry gate is emitted), so the lint stays silent.
    #[test]
    fn dead_halt_lint_silent_when_a_stage_declares_retry() {
        let warnings = warnings_for(
            r#"
            local B = require("bp_dsl")
            return B.pipeline({
              B.stage "review" {
                agent = "mock-review",
                retry = { max = 1, fix = B.stage "fix" { agent = "f" } },
              },
              halt_on = { "BLOCKED" },
              halted_at = "$.halted_at",
            })
            "#,
        );
        assert!(warnings.is_empty(), "retry implies a gate: {warnings:?}");
    }

    /// An explicit `halted_at` alone is a target path, not halt intent —
    /// deliberately outside the trigger.
    #[test]
    fn dead_halt_lint_silent_for_halted_at_without_halt_on() {
        let warnings = warnings_for(
            r#"
            local B = require("bp_dsl")
            return B.pipeline({
              B.stage "review" { agent = "mock-review" },
              halted_at = "$.halted_at",
            })
            "#,
        );
        assert!(
            warnings.is_empty(),
            "halted_at alone is not halt intent: {warnings:?}"
        );
    }

    /// `done` is deliberately outside the trigger too: without gates the
    /// final assign still runs unconditionally, so nothing is dead.
    #[test]
    fn dead_halt_lint_silent_for_done_without_halt_on() {
        let warnings = warnings_for(
            r#"
            local B = require("bp_dsl")
            return B.pipeline({
              B.stage "review" { agent = "mock-review" },
              done = "$.done",
            })
            "#,
        );
        assert!(
            warnings.is_empty(),
            "done without halt_on is not a dead halt: {warnings:?}"
        );
    }

    /// The legacy entry point keeps building a warning-triggering script:
    /// the lint is report-only, warnings are simply dropped.
    #[test]
    fn build_bp_from_script_still_builds_a_dead_halt_pipeline() {
        let out = build_bp_from_script(
            r#"
            local B = require("bp_dsl")
            return B.pipeline({
              B.stage "review" { agent = "mock-review" },
              halt_on = { "BLOCKED" },
              halted_at = "$.halted_at",
            })
            "#,
        )
        .expect("dead-halt pipeline must still build");
        assert_eq!(out["kind"], serde_json::json!("seq"));
    }

    /// A verdict gate on a fanout stage compares the join result rather
    /// than one agent's verdict, so it can never fire — one WARN line names
    /// the stage and points at the aggregate stage. Report-only: the gate
    /// is still emitted exactly as written.
    #[test]
    fn fanout_stage_with_a_verdict_gate_warns_and_still_emits_the_gate() {
        let (value, warnings) = build_bp_from_script_with_warnings(
            r#"
            local B = require("bp_dsl")
            return B.pipeline({
              B.stage "gates" {
                fanout = { lanes = { "danger", "leak" } },
                gate = true,
              },
              halt_on = { "BLOCKED" },
              halted_at = "$.halted_at",
            })
            "#,
        )
        .expect("gate-on-fanout must still build");

        assert_eq!(
            warnings.len(),
            1,
            "exactly one fanout-gate WARN: {warnings:?}"
        );
        let w = &warnings[0];
        assert!(w.contains("gates"), "must name the stage id: {w}");
        assert!(w.contains("fanout stage"), "{w}");
        assert!(
            w.contains("aggregate"),
            "must point at the aggregate-stage fix: {w}"
        );

        // The gate itself is untouched: seq{fanout, branch}.
        let children = value["children"].as_array().expect("seq children");
        assert_eq!(children[0]["kind"], serde_json::json!("fanout"));
        assert_eq!(children[1]["kind"], serde_json::json!("branch"));
    }

    /// A stage-level `halt_on` is the same opt-in, so it warns the same way.
    #[test]
    fn fanout_stage_with_a_stage_level_halt_on_warns_too() {
        let warnings = warnings_for(
            r#"
            local B = require("bp_dsl")
            return B.pipeline({
              B.stage "gates" {
                fanout = { agent = "check" },
                halt_on = { "BLOCKED" },
              },
              halted_at = "$.halted_at",
            })
            "#,
        );
        assert_eq!(warnings.len(), 1, "stage halt_on opts in: {warnings:?}");
        assert!(warnings[0].contains("gates"), "{:?}", warnings);
    }

    /// A fanout stage is outside the `gate_default = "auto"` cascade, so it
    /// gets no gate and no fanout-gate WARN of its own — and with no other
    /// stage opting in, the pipeline-level `halt_on` is correctly reported
    /// as a dead halt instead.
    #[test]
    fn fanout_stage_is_outside_the_auto_cascade_and_reports_a_dead_halt() {
        let (value, warnings) = build_bp_from_script_with_warnings(
            r#"
            local B = require("bp_dsl")
            return B.pipeline({
              B.stage "gates" { fanout = { agent = "check" } },
              halt_on = { "BLOCKED" },
              halted_at = "$.halted_at",
              gate_default = "auto",
            })
            "#,
        )
        .expect("auto cascade + fanout must build");

        assert_eq!(warnings.len(), 1, "only the dead-halt WARN: {warnings:?}");
        assert!(
            warnings[0].contains("can never halt"),
            "the dead-halt lint is the correct report here: {}",
            warnings[0]
        );

        let children = value["children"].as_array().expect("seq children");
        assert_eq!(children[0]["kind"], serde_json::json!("fanout"));
        assert_ne!(
            children[1]["kind"],
            serde_json::json!("branch"),
            "the auto cascade must not gate a fanout stage: {value}"
        );
    }

    /// Build `script` and return the error message it raised.
    fn error_for(script: &str) -> String {
        build_bp_from_script(script)
            .expect_err("script must fail to build")
            .to_string()
    }

    /// `retry` on a fanout stage is rejected outright: the loop cond would
    /// compare the join result against a verdict, and `retry.fix` has no
    /// lane ctx to write back into.
    #[test]
    fn retry_on_a_fanout_stage_errors() {
        let message = error_for(
            r#"
            local B = require("bp_dsl")
            return B.pipeline({
              B.stage "gates" {
                fanout = { agent = "check" },
                retry = { max = 1, fix = B.stage "fix" { agent = "f" } },
              },
              halted_at = "$.halted_at",
            })
            "#,
        );
        assert!(message.contains("retry"), "{message}");
        assert!(message.contains("gates"), "must name the stage: {message}");
    }

    /// `agent` and `fanout` on the same stage record are mutually exclusive.
    #[test]
    fn agent_alongside_fanout_errors() {
        let message = error_for(
            r#"
            local B = require("bp_dsl")
            return B.pipeline({
              B.stage "gates" { agent = "check", fanout = { agent = "check" } },
              halted_at = "$.halted_at",
            })
            "#,
        );
        assert!(message.contains("mutually exclusive"), "{message}");
    }

    /// A fanout record needs exactly one of `agent` / `lanes`.
    #[test]
    fn fanout_without_agent_or_lanes_errors() {
        let message = error_for(
            r#"
            local B = require("bp_dsl")
            return B.pipeline({
              B.stage "gates" { fanout = { join = "all" } },
              halted_at = "$.halted_at",
            })
            "#,
        );
        assert!(
            message.contains("fanout.agent") || message.contains("agent ="),
            "{message}"
        );
        assert!(message.contains("lanes"), "{message}");
    }

    /// An unknown `join` mode fails loud — typo protection, same posture as
    /// `gate_default`.
    #[test]
    fn unknown_fanout_join_mode_errors() {
        let message = error_for(
            r#"
            local B = require("bp_dsl")
            return B.pipeline({
              B.stage "gates" { fanout = { agent = "check", join = "first" } },
              halted_at = "$.halted_at",
            })
            "#,
        );
        assert!(message.contains("join"), "{message}");
        assert!(
            message.contains("first"),
            "must echo the bad value: {message}"
        );
        assert!(
            message.contains("all_settled"),
            "must list the modes: {message}"
        );
    }

    /// `lanes` must be an ordered array: a keyed table has undefined `pairs`
    /// order, which would emit a non-deterministic lane order.
    #[test]
    fn keyed_lanes_table_errors() {
        let message = error_for(
            r#"
            local B = require("bp_dsl")
            return B.pipeline({
              B.stage "gates" {
                fanout = { lanes = { danger = "gate-danger", leak = "gate-leak" } },
              },
              halted_at = "$.halted_at",
            })
            "#,
        );
        assert!(message.contains("ordered array"), "{message}");
    }

    /// An empty `lanes` list is an error too (the fanout would have no body).
    #[test]
    fn empty_lanes_list_errors() {
        let message = error_for(
            r#"
            local B = require("bp_dsl")
            return B.pipeline({
              B.stage "gates" { fanout = { lanes = {} } },
              halted_at = "$.halted_at",
            })
            "#,
        );
        assert!(message.contains("empty"), "{message}");
    }

    #[test]
    fn empty_object_marker_replacement_does_not_misfire_on_ordinary_data() {
        // A field that legitimately reuses the marker key name for
        // something other than `true` (or carries sibling keys) must not
        // be collapsed to `{}`.
        let out = build_bp_from_script(
            r#"
            return {
              a = { __mse_empty_object__ = false },
              b = { __mse_empty_object__ = true, extra = 1 },
            }
            "#,
        )
        .expect("script must build");
        assert_eq!(out["a"], serde_json::json!({"__mse_empty_object__": false}));
        assert_eq!(
            out["b"],
            serde_json::json!({"__mse_empty_object__": true, "extra": 1})
        );
    }
}