1const FLOW_DSL_SRC: &str = include_str!("flow_dsl.lua");
23const BP_DSL_SRC: &str = include_str!("bp_dsl.lua");
24
25const EMPTY_OBJECT_MARKER_KEY: &str = "__mse_empty_object__";
28
29fn replace_empty_object_markers(value: &mut serde_json::Value) {
35 match value {
36 serde_json::Value::Object(map) => {
37 let is_marker = map.len() == 1
38 && map.get(EMPTY_OBJECT_MARKER_KEY) == Some(&serde_json::Value::Bool(true));
39 if is_marker {
40 *value = serde_json::Value::Object(serde_json::Map::new());
41 return;
42 }
43 for v in map.values_mut() {
44 replace_empty_object_markers(v);
45 }
46 }
47 serde_json::Value::Array(arr) => {
48 for v in arr.iter_mut() {
49 replace_empty_object_markers(v);
50 }
51 }
52 _ => {}
53 }
54}
55
56pub fn preload(lua: &mlua::Lua) -> mlua::Result<()> {
61 let package: mlua::Table = lua.globals().get("package")?;
62 let preload: mlua::Table = package.get("preload")?;
63
64 preload.set(
65 "flow_dsl",
66 lua.create_function(|lua, ()| {
67 lua.load(FLOW_DSL_SRC)
68 .set_name("flow_dsl.lua")
69 .eval::<mlua::Value>()
70 })?,
71 )?;
72 preload.set(
73 "bp_dsl",
74 lua.create_function(|lua, ()| {
75 lua.load(BP_DSL_SRC)
76 .set_name("bp_dsl.lua")
77 .eval::<mlua::Value>()
78 })?,
79 )?;
80 Ok(())
81}
82
83pub fn build_bp_from_script(script: &str) -> anyhow::Result<serde_json::Value> {
100 Ok(build_bp_from_script_with_warnings(script)?.0)
101}
102
103pub fn build_bp_from_script_with_warnings(
109 script: &str,
110) -> anyhow::Result<(serde_json::Value, Vec<String>)> {
111 use mlua::LuaSerdeExt;
112
113 let lua = mlua::Lua::new();
117 preload(&lua).map_err(|e| anyhow::anyhow!("dsl preload failed: {e}"))?;
118 let result: mlua::Value = lua
119 .load(script)
120 .set_name("<bp-script>")
121 .eval()
122 .map_err(|e| anyhow::anyhow!("bp-script eval failed: {e}"))?;
123 let options = mlua::serde::de::Options::new().encode_empty_tables_as_array(true);
124 let mut value: serde_json::Value = lua
125 .from_value_with(result, options)
126 .map_err(|e| anyhow::anyhow!("lua value -> json conversion failed: {e}"))?;
127 replace_empty_object_markers(&mut value);
128 let warnings = drain_authoring_warnings(&lua);
129 Ok((value, warnings))
130}
131
132fn drain_authoring_warnings(lua: &mlua::Lua) -> Vec<String> {
137 let drained: mlua::Result<Vec<String>> = (|| {
138 let package: mlua::Table = lua.globals().get("package")?;
139 let loaded: mlua::Table = package.get("loaded")?;
140 let module: mlua::Value = loaded.get("bp_dsl")?;
141 let mlua::Value::Table(module) = module else {
142 return Ok(Vec::new());
143 };
144 let take: mlua::Function = module.get("take_authoring_warnings")?;
145 let list: mlua::Table = take.call(())?;
146 let mut out = Vec::new();
147 for entry in list.sequence_values::<String>() {
148 out.push(entry?);
149 }
150 Ok(out)
151 })();
152 drained.unwrap_or_default()
153}
154
155#[cfg(test)]
156mod tests {
157 use super::*;
158
159 #[test]
160 fn preload_exposes_flow_dsl_and_bp_dsl() {
161 let lua = mlua::Lua::new();
162 preload(&lua).expect("preload must succeed");
163 let ok: bool = lua
164 .load(
165 r#"
166 local F = require("flow_dsl")
167 local B = require("bp_dsl")
168 return F ~= nil and B ~= nil
169 "#,
170 )
171 .eval()
172 .expect("require must succeed for both modules");
173 assert!(ok, "flow_dsl / bp_dsl must both resolve via require()");
174 }
175
176 #[test]
177 fn build_bp_from_script_returns_json_value() {
178 let out = build_bp_from_script(
179 r#"
180 local F = require("flow_dsl")
181 return { id = "t", flow = F.assign{ at = F.p("$.x"), value = F.lit(1) } }
182 "#,
183 )
184 .expect("script must build");
185 assert_eq!(out["id"], serde_json::json!("t"));
186 assert_eq!(out["flow"]["kind"], serde_json::json!("assign"));
187 assert_eq!(
188 out["flow"]["at"],
189 serde_json::json!({"op": "path", "at": "$.x"})
190 );
191 }
192
193 #[test]
194 fn build_bp_from_script_surfaces_lua_errors() {
195 let err = build_bp_from_script("error(\"boom\")").expect_err("must propagate the error");
196 assert!(err.to_string().contains("boom"));
197 }
198
199 #[test]
200 fn f_obj_marker_becomes_a_genuine_empty_json_object() {
201 let out = build_bp_from_script(
202 r#"
203 local F = require("flow_dsl")
204 return { spec = F.obj(), other = {} }
205 "#,
206 )
207 .expect("script must build");
208 assert_eq!(out["spec"], serde_json::json!({}));
209 assert!(
210 out["spec"].is_object(),
211 "F.obj() must become an object, not an array"
212 );
213 assert_eq!(out["other"], serde_json::json!([]));
218 }
219
220 #[test]
226 fn bp_dsl_skip_on_compiles_to_branch_in_verdict_skip_on_list() {
227 let out = build_bp_from_script(
228 r#"
229 local F = require("flow_dsl")
230 local B = require("bp_dsl")
231 return B.pipeline({
232 B.stage "gate" { agent = "mock-gate" },
233 B.stage "worker" {
234 agent = "mock-worker",
235 input = B.from "gate",
236 skip_on = { "SKIP", "NOT_APPLICABLE" },
237 },
238 halted_at = "$.halted_at",
239 })
240 "#,
241 )
242 .expect("skip_on pipeline must build");
243
244 assert_eq!(out["kind"], serde_json::json!("seq"));
246 let top_children = out["children"].as_array().expect("top seq children");
247 assert_eq!(top_children.len(), 2);
248 assert_eq!(top_children[0]["kind"], serde_json::json!("step"));
249 assert_eq!(top_children[0]["ref"], serde_json::json!("mock-gate"));
250
251 let rest = &top_children[1];
254 assert_eq!(rest["kind"], serde_json::json!("seq"));
255 let rest_children = rest["children"].as_array().expect("rest seq children");
256 let worker_guarded = &rest_children[0];
258 assert_eq!(worker_guarded["kind"], serde_json::json!("branch"));
259
260 let cond = &worker_guarded["cond"];
263 assert_eq!(cond["op"], serde_json::json!("in"));
264 assert_eq!(
265 cond["needle"],
266 serde_json::json!({"op": "path", "at": "$.gate.parts[\"verdict\"]"})
267 );
268 assert_eq!(cond["haystack"]["op"], serde_json::json!("lit"));
269 assert_eq!(
270 cond["haystack"]["value"],
271 serde_json::json!(["SKIP", "NOT_APPLICABLE"])
272 );
273
274 assert_eq!(
276 worker_guarded["then"],
277 serde_json::json!({"kind": "seq", "children": []})
278 );
279
280 let body = &worker_guarded["else"];
283 assert_eq!(body["kind"], serde_json::json!("seq"));
284 assert_eq!(body["children"][0]["ref"], serde_json::json!("mock-worker"));
285 }
286
287 #[test]
294 fn bp_dsl_skip_on_coexists_with_halt_on() {
295 let out = build_bp_from_script(
296 r#"
297 local F = require("flow_dsl")
298 local B = require("bp_dsl")
299 return B.pipeline({
300 B.stage "planner" { agent = "mock-planner" },
301 B.stage "worker" {
302 agent = "mock-worker",
303 input = B.from "planner",
304 skip_on = { "SKIP" },
305 halt_on = { "BLOCKED" },
306 },
307 B.stage "publisher" { agent = "mock-publisher" },
308 halted_at = "$.halted_at",
309 })
310 "#,
311 )
312 .expect("skip_on + halt_on pipeline must build");
313
314 let rest = &out["children"][1];
318 let worker_seq = rest;
319 assert_eq!(worker_seq["kind"], serde_json::json!("seq"));
320 let worker_children = worker_seq["children"]
321 .as_array()
322 .expect("worker seq children");
323 assert_eq!(
324 worker_children.len(),
325 2,
326 "skip guard + halt_on gate (with publisher threaded into gate else)"
327 );
328
329 let skip_branch = &worker_children[0];
331 assert_eq!(skip_branch["kind"], serde_json::json!("branch"));
332 assert_eq!(skip_branch["cond"]["op"], serde_json::json!("in"));
333
334 let halt_gate = &worker_children[1];
337 assert_eq!(halt_gate["kind"], serde_json::json!("branch"));
338 assert_eq!(halt_gate["cond"]["op"], serde_json::json!("eq"));
339 assert_eq!(
340 halt_gate["cond"]["lhs"],
341 serde_json::json!({"op": "path", "at": "$.worker.parts[\"verdict\"]"})
342 );
343 let gate_else = &halt_gate["else"];
345 assert_eq!(gate_else["kind"], serde_json::json!("seq"));
346 let contains_publisher = gate_else["children"]
348 .as_array()
349 .map(|arr| {
350 arr.iter()
351 .any(|c| c["ref"] == serde_json::json!("mock-publisher"))
352 })
353 .unwrap_or(false);
354 assert!(
355 contains_publisher,
356 "halt_on gate else must thread the publisher stage through: {gate_else}"
357 );
358 }
359
360 #[test]
364 fn bp_dsl_skip_on_empty_list_is_noop() {
365 let with_empty = build_bp_from_script(
366 r#"
367 local B = require("bp_dsl")
368 return B.pipeline({
369 B.stage "worker" { agent = "mock-worker", skip_on = {} },
370 halted_at = "$.halted_at",
371 })
372 "#,
373 )
374 .expect("skip_on={} pipeline must build");
375
376 let children = with_empty["children"].as_array().expect("seq children");
379 assert_eq!(
380 children.len(),
381 2,
382 "no skip guard emitted for empty skip_on: {with_empty}"
383 );
384 assert_eq!(children[0]["kind"], serde_json::json!("step"));
385 assert_eq!(children[0]["ref"], serde_json::json!("mock-worker"));
386
387 let baseline = build_bp_from_script(
389 r#"
390 local B = require("bp_dsl")
391 return B.pipeline({
392 B.stage "worker" { agent = "mock-worker" },
393 halted_at = "$.halted_at",
394 })
395 "#,
396 )
397 .expect("baseline pipeline must build");
398 assert_eq!(with_empty, baseline, "skip_on = {{}} must be a no-op");
399 }
400
401 fn warnings_for(script: &str) -> Vec<String> {
403 build_bp_from_script_with_warnings(script)
404 .expect("script must build")
405 .1
406 }
407
408 #[test]
412 fn dead_halt_lint_warns_when_pipeline_halt_on_has_no_gating_stage() {
413 let warnings = warnings_for(
414 r#"
415 local B = require("bp_dsl")
416 return B.pipeline({
417 B.stage "review" { agent = "mock-review" },
418 halt_on = { "BLOCKED" },
419 halted_at = "$.halted_at",
420 })
421 "#,
422 );
423 assert_eq!(
424 warnings.len(),
425 1,
426 "exactly one dead-halt WARN: {warnings:?}"
427 );
428 let w = &warnings[0];
429 assert!(w.contains("can never halt"), "{w}");
430 assert!(w.contains("review"), "must name the stage id: {w}");
431 assert!(w.contains("BLOCKED"), "must name the halt values: {w}");
432 }
433
434 #[test]
437 fn dead_halt_lint_silent_when_a_stage_opts_in_with_gate_true() {
438 let warnings = warnings_for(
439 r#"
440 local B = require("bp_dsl")
441 return B.pipeline({
442 B.stage "review" { agent = "mock-review", gate = true },
443 halt_on = { "BLOCKED" },
444 halted_at = "$.halted_at",
445 })
446 "#,
447 );
448 assert!(warnings.is_empty(), "gate = true opts in: {warnings:?}");
449 }
450
451 #[test]
454 fn dead_halt_lint_silent_under_gate_default_auto() {
455 let warnings = warnings_for(
456 r#"
457 local B = require("bp_dsl")
458 return B.pipeline({
459 B.stage "review" { agent = "mock-review" },
460 halt_on = { "BLOCKED" },
461 halted_at = "$.halted_at",
462 gate_default = "auto",
463 })
464 "#,
465 );
466 assert!(
467 warnings.is_empty(),
468 "auto cascade gates every stage: {warnings:?}"
469 );
470 }
471
472 #[test]
475 fn dead_halt_lint_silent_when_a_stage_declares_retry() {
476 let warnings = warnings_for(
477 r#"
478 local B = require("bp_dsl")
479 return B.pipeline({
480 B.stage "review" {
481 agent = "mock-review",
482 retry = { max = 1, fix = B.stage "fix" { agent = "f" } },
483 },
484 halt_on = { "BLOCKED" },
485 halted_at = "$.halted_at",
486 })
487 "#,
488 );
489 assert!(warnings.is_empty(), "retry implies a gate: {warnings:?}");
490 }
491
492 #[test]
495 fn dead_halt_lint_silent_for_halted_at_without_halt_on() {
496 let warnings = warnings_for(
497 r#"
498 local B = require("bp_dsl")
499 return B.pipeline({
500 B.stage "review" { agent = "mock-review" },
501 halted_at = "$.halted_at",
502 })
503 "#,
504 );
505 assert!(
506 warnings.is_empty(),
507 "halted_at alone is not halt intent: {warnings:?}"
508 );
509 }
510
511 #[test]
514 fn dead_halt_lint_silent_for_done_without_halt_on() {
515 let warnings = warnings_for(
516 r#"
517 local B = require("bp_dsl")
518 return B.pipeline({
519 B.stage "review" { agent = "mock-review" },
520 done = "$.done",
521 })
522 "#,
523 );
524 assert!(
525 warnings.is_empty(),
526 "done without halt_on is not a dead halt: {warnings:?}"
527 );
528 }
529
530 #[test]
533 fn build_bp_from_script_still_builds_a_dead_halt_pipeline() {
534 let out = build_bp_from_script(
535 r#"
536 local B = require("bp_dsl")
537 return B.pipeline({
538 B.stage "review" { agent = "mock-review" },
539 halt_on = { "BLOCKED" },
540 halted_at = "$.halted_at",
541 })
542 "#,
543 )
544 .expect("dead-halt pipeline must still build");
545 assert_eq!(out["kind"], serde_json::json!("seq"));
546 }
547
548 #[test]
553 fn fanout_stage_with_a_verdict_gate_warns_and_still_emits_the_gate() {
554 let (value, warnings) = build_bp_from_script_with_warnings(
555 r#"
556 local B = require("bp_dsl")
557 return B.pipeline({
558 B.stage "gates" {
559 fanout = { lanes = { "danger", "leak" } },
560 gate = true,
561 },
562 halt_on = { "BLOCKED" },
563 halted_at = "$.halted_at",
564 })
565 "#,
566 )
567 .expect("gate-on-fanout must still build");
568
569 assert_eq!(
570 warnings.len(),
571 1,
572 "exactly one fanout-gate WARN: {warnings:?}"
573 );
574 let w = &warnings[0];
575 assert!(w.contains("gates"), "must name the stage id: {w}");
576 assert!(w.contains("fanout stage"), "{w}");
577 assert!(
578 w.contains("aggregate"),
579 "must point at the aggregate-stage fix: {w}"
580 );
581
582 let children = value["children"].as_array().expect("seq children");
584 assert_eq!(children[0]["kind"], serde_json::json!("fanout"));
585 assert_eq!(children[1]["kind"], serde_json::json!("branch"));
586 }
587
588 #[test]
590 fn fanout_stage_with_a_stage_level_halt_on_warns_too() {
591 let warnings = warnings_for(
592 r#"
593 local B = require("bp_dsl")
594 return B.pipeline({
595 B.stage "gates" {
596 fanout = { agent = "check" },
597 halt_on = { "BLOCKED" },
598 },
599 halted_at = "$.halted_at",
600 })
601 "#,
602 );
603 assert_eq!(warnings.len(), 1, "stage halt_on opts in: {warnings:?}");
604 assert!(warnings[0].contains("gates"), "{:?}", warnings);
605 }
606
607 #[test]
612 fn fanout_stage_is_outside_the_auto_cascade_and_reports_a_dead_halt() {
613 let (value, warnings) = build_bp_from_script_with_warnings(
614 r#"
615 local B = require("bp_dsl")
616 return B.pipeline({
617 B.stage "gates" { fanout = { agent = "check" } },
618 halt_on = { "BLOCKED" },
619 halted_at = "$.halted_at",
620 gate_default = "auto",
621 })
622 "#,
623 )
624 .expect("auto cascade + fanout must build");
625
626 assert_eq!(warnings.len(), 1, "only the dead-halt WARN: {warnings:?}");
627 assert!(
628 warnings[0].contains("can never halt"),
629 "the dead-halt lint is the correct report here: {}",
630 warnings[0]
631 );
632
633 let children = value["children"].as_array().expect("seq children");
634 assert_eq!(children[0]["kind"], serde_json::json!("fanout"));
635 assert_ne!(
636 children[1]["kind"],
637 serde_json::json!("branch"),
638 "the auto cascade must not gate a fanout stage: {value}"
639 );
640 }
641
642 fn error_for(script: &str) -> String {
644 build_bp_from_script(script)
645 .expect_err("script must fail to build")
646 .to_string()
647 }
648
649 #[test]
653 fn retry_on_a_fanout_stage_errors() {
654 let message = error_for(
655 r#"
656 local B = require("bp_dsl")
657 return B.pipeline({
658 B.stage "gates" {
659 fanout = { agent = "check" },
660 retry = { max = 1, fix = B.stage "fix" { agent = "f" } },
661 },
662 halted_at = "$.halted_at",
663 })
664 "#,
665 );
666 assert!(message.contains("retry"), "{message}");
667 assert!(message.contains("gates"), "must name the stage: {message}");
668 }
669
670 #[test]
672 fn agent_alongside_fanout_errors() {
673 let message = error_for(
674 r#"
675 local B = require("bp_dsl")
676 return B.pipeline({
677 B.stage "gates" { agent = "check", fanout = { agent = "check" } },
678 halted_at = "$.halted_at",
679 })
680 "#,
681 );
682 assert!(message.contains("mutually exclusive"), "{message}");
683 }
684
685 #[test]
687 fn fanout_without_agent_or_lanes_errors() {
688 let message = error_for(
689 r#"
690 local B = require("bp_dsl")
691 return B.pipeline({
692 B.stage "gates" { fanout = { join = "all" } },
693 halted_at = "$.halted_at",
694 })
695 "#,
696 );
697 assert!(
698 message.contains("fanout.agent") || message.contains("agent ="),
699 "{message}"
700 );
701 assert!(message.contains("lanes"), "{message}");
702 }
703
704 #[test]
707 fn unknown_fanout_join_mode_errors() {
708 let message = error_for(
709 r#"
710 local B = require("bp_dsl")
711 return B.pipeline({
712 B.stage "gates" { fanout = { agent = "check", join = "first" } },
713 halted_at = "$.halted_at",
714 })
715 "#,
716 );
717 assert!(message.contains("join"), "{message}");
718 assert!(
719 message.contains("first"),
720 "must echo the bad value: {message}"
721 );
722 assert!(
723 message.contains("all_settled"),
724 "must list the modes: {message}"
725 );
726 }
727
728 #[test]
731 fn keyed_lanes_table_errors() {
732 let message = error_for(
733 r#"
734 local B = require("bp_dsl")
735 return B.pipeline({
736 B.stage "gates" {
737 fanout = { lanes = { danger = "gate-danger", leak = "gate-leak" } },
738 },
739 halted_at = "$.halted_at",
740 })
741 "#,
742 );
743 assert!(message.contains("ordered array"), "{message}");
744 }
745
746 #[test]
748 fn empty_lanes_list_errors() {
749 let message = error_for(
750 r#"
751 local B = require("bp_dsl")
752 return B.pipeline({
753 B.stage "gates" { fanout = { lanes = {} } },
754 halted_at = "$.halted_at",
755 })
756 "#,
757 );
758 assert!(message.contains("empty"), "{message}");
759 }
760
761 #[test]
762 fn empty_object_marker_replacement_does_not_misfire_on_ordinary_data() {
763 let out = build_bp_from_script(
767 r#"
768 return {
769 a = { __mse_empty_object__ = false },
770 b = { __mse_empty_object__ = true, extra = 1 },
771 }
772 "#,
773 )
774 .expect("script must build");
775 assert_eq!(out["a"], serde_json::json!({"__mse_empty_object__": false}));
776 assert_eq!(
777 out["b"],
778 serde_json::json!({"__mse_empty_object__": true, "extra": 1})
779 );
780 }
781}