1#[cfg(feature = "ir")]
34use crate::expr::evaluate as evaluate_expression;
35use crate::expr::ExprFailure;
36#[cfg(feature = "ir")]
37use crate::plan::{run_plan, unproduced_value, ExecutionPlanSpec, OpSpec, RelationKind};
38use crate::plan::{ExecOutcome, PlanFailure};
39use crate::value::Value;
40#[cfg(feature = "ir")]
41use serde_json::Value as J;
42
43#[derive(Debug, Clone, Copy, PartialEq, Eq)]
45pub enum BehaviorFailureCode {
46 UnknownComponent,
47 UnknownNodeKind,
48 MapOverNotArray,
49 MapIntoElementNotObject,
50 MapBatchResultMismatch,
51 FanoutOverNotArray,
52 FanoutBatchResultMismatch,
53 UnknownEntry,
54}
55
56impl BehaviorFailureCode {
57 pub fn as_str(self) -> &'static str {
58 match self {
59 BehaviorFailureCode::UnknownComponent => "UNKNOWN_COMPONENT",
60 BehaviorFailureCode::UnknownNodeKind => "UNKNOWN_NODE_KIND",
61 BehaviorFailureCode::MapOverNotArray => "MAP_OVER_NOT_ARRAY",
62 BehaviorFailureCode::MapIntoElementNotObject => "MAP_INTO_ELEMENT_NOT_OBJECT",
63 BehaviorFailureCode::MapBatchResultMismatch => "MAP_BATCH_RESULT_MISMATCH",
64 BehaviorFailureCode::FanoutOverNotArray => "FANOUT_OVER_NOT_ARRAY",
65 BehaviorFailureCode::FanoutBatchResultMismatch => "FANOUT_BATCH_RESULT_MISMATCH",
66 BehaviorFailureCode::UnknownEntry => "UNKNOWN_ENTRY",
67 }
68 }
69}
70
71#[derive(Debug, Clone)]
74pub struct BehaviorError {
75 code: String,
76 pub message: String,
77 pub detail: Option<Box<crate::plan::ErrorDetail>>,
81}
82
83impl BehaviorError {
84 pub fn new(code: impl Into<String>, message: impl Into<String>) -> Self {
89 BehaviorError {
90 code: code.into(),
91 message: message.into(),
92 detail: None,
93 }
94 }
95
96 pub fn code(&self) -> &str {
98 &self.code
99 }
100
101 pub fn detail(&self) -> Option<&crate::plan::ErrorDetail> {
103 self.detail.as_deref()
104 }
105}
106
107impl std::fmt::Display for BehaviorError {
108 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
109 write!(f, "{}: {}", self.code, self.message)
110 }
111}
112impl std::error::Error for BehaviorError {}
113
114impl From<ExprFailure> for BehaviorError {
115 fn from(e: ExprFailure) -> Self {
116 BehaviorError {
117 code: e.code.as_str().to_string(),
118 message: e.message,
119 detail: e.detail,
120 }
121 }
122}
123impl From<PlanFailure> for BehaviorError {
124 fn from(e: PlanFailure) -> Self {
125 BehaviorError {
126 code: e.code.as_str().to_string(),
127 message: e.message,
128 detail: e.detail,
129 }
130 }
131}
132
133#[cfg(feature = "ir")]
134fn bfail<T>(code: BehaviorFailureCode, message: impl Into<String>) -> Result<T, BehaviorError> {
135 Err(BehaviorError {
136 code: code.as_str().to_string(),
137 message: message.into(),
138 detail: None,
139 })
140}
141
142#[cfg(feature = "ir")]
145fn wire_type_name(v: &Value) -> &'static str {
146 v.type_name()
147}
148
149#[cfg(feature = "ir")]
151fn raw_value_of(v: &Value) -> String {
152 match v {
153 Value::Str(s) => s.clone(),
154 Value::Int(i) => i.to_string(),
155 Value::Float(f) => crate::canonical::py_float_repr(*f).unwrap_or_else(|_| f.to_string()),
156 Value::Bool(b) => b.to_string(),
157 Value::Null => "null".to_string(),
158 _ => crate::canonical::canonical_json(v).unwrap_or_else(|_| v.type_name().to_string()),
159 }
160}
161
162#[cfg(feature = "ir")]
164fn portable_type_notation(t: &J) -> String {
165 if let Some(s) = t.as_str() {
166 return s.to_string();
167 }
168 for key in ["opt", "arr", "map"] {
169 if let Some(inner) = t.get(key) {
170 return format!("{key}({})", portable_type_notation(inner));
171 }
172 }
173 if let Some(J::Object(o)) = t.get("obj") {
174 let parts: Vec<String> = o
175 .iter()
176 .map(|(k, v)| format!("{k}:{}", portable_type_notation(v)))
177 .collect();
178 return format!("obj{{{}}}", parts.join(","));
179 }
180 "?".to_string()
181}
182
183#[cfg(feature = "ir")]
186fn result_type_of(node: &J, out_type: &J) -> J {
187 if node.get("map").is_some() {
188 serde_json::json!({ "arr": out_type })
189 } else {
190 out_type.clone()
191 }
192}
193
194#[cfg(feature = "ir")]
200fn portable_type_of_input_port(schema: &J) -> Option<J> {
201 let base: J = match schema.get("type").and_then(|v| v.as_str()) {
202 Some(t @ ("string" | "int" | "float" | "bool" | "null")) => J::String(t.to_string()),
203 Some("array") => serde_json::json!({ "arr": schema.get("elemType")? }),
204 Some("map") => serde_json::json!({ "map": schema.get("elemType")? }),
205 _ => return None,
206 };
207 if schema.get("required") == Some(&J::Bool(false)) {
208 Some(serde_json::json!({ "opt": base }))
209 } else {
210 Some(base)
211 }
212}
213
214#[cfg(feature = "ir")]
216fn descend(t: &J, field: &str) -> Option<J> {
217 if t.is_string() {
218 return None;
219 }
220 if let Some(obj) = t.get("obj") {
221 return obj.get(field).cloned();
222 }
223 if let Some(inner) = t.get("opt") {
224 let d = descend(inner, field)?;
225 return Some(serde_json::json!({ "opt": d }));
226 }
227 None
228}
229
230#[cfg(feature = "ir")]
232fn bare_ref_path(over: &J) -> Option<Vec<String>> {
233 let obj = over.as_object()?;
234 if obj.len() != 1 {
235 return None;
236 }
237 let (k, v) = obj.iter().next()?;
238 if k != "ref" && k != "refOpt" {
239 return None;
240 }
241 let arr = v.as_array()?;
242 if arr.is_empty() {
243 return None;
244 }
245 arr.iter()
246 .map(|p| p.as_str().map(|s| s.to_string()))
247 .collect()
248}
249
250#[cfg(feature = "ir")]
252fn is_opt_arr_type(t: &J) -> bool {
253 t.get("opt").and_then(|inner| inner.get("arr")).is_some()
254}
255
256#[cfg(feature = "ir")]
260fn over_is_declared_opt_arr(
261 over: &J,
262 body: &[J],
263 input_ports: Option<&serde_json::Map<String, J>>,
264) -> bool {
265 let path = match bare_ref_path(over) {
266 Some(p) => p,
267 None => return false,
268 };
269 let (head, rest) = (&path[0], &path[1..]);
270 let node = body
271 .iter()
272 .find(|n| n.get("id").and_then(|v| v.as_str()) == Some(head.as_str()));
273 let mut cur: J = if let Some(node) = node {
274 match node.get("outType") {
275 Some(ot) => result_type_of(node, ot),
276 None => return false,
277 }
278 } else {
279 let schema = match input_ports.and_then(|ip| ip.get(head)) {
280 Some(s) => s,
281 None => return false,
282 };
283 if rest.is_empty() {
284 return schema.get("required") == Some(&J::Bool(false))
285 && schema.get("type").and_then(|v| v.as_str()) == Some("array");
286 }
287 match portable_type_of_input_port(schema) {
288 Some(t) => t,
289 None => return false,
290 }
291 };
292 for f in rest {
293 cur = match descend(&cur, f) {
294 Some(t) => t,
295 None => return false,
296 };
297 }
298 is_opt_arr_type(&cur)
299}
300
301#[cfg(feature = "ir")]
305fn resolve_over_array(
306 over: &Value,
307 over_expr: &J,
308 body: &[J],
309 input_ports: Option<&serde_json::Map<String, J>>,
310 code: BehaviorFailureCode,
311 op_id: &str,
312 label: &str,
313) -> Result<Vec<Value>, BehaviorError> {
314 if let Value::Arr(a) = over {
315 return Ok(a.clone());
316 }
317 if matches!(over, Value::Null) && over_is_declared_opt_arr(over_expr, body, input_ports) {
318 return Ok(Vec::new());
319 }
320 Err(BehaviorError {
321 code: code.as_str().to_string(),
322 message: format!("{label} '{op_id}': 'over' is not an array"),
323 detail: None,
324 })
325}
326
327#[cfg(feature = "ir")]
328fn conform_fail(
329 code: crate::expr::ExprFailureCode,
330 message: String,
331 kind: crate::plan::ErrorKind,
332 node_id: &str,
333 field: &str,
334 expected: &J,
335 actual: Option<&Value>,
336) -> ExprFailure {
337 let mut detail = crate::plan::ErrorDetail {
338 kind: Some(kind),
339 model: Some(node_id.to_string()),
340 field: Some(field.to_string()),
341 expected_type: Some(portable_type_notation(expected)),
342 ..Default::default()
343 };
344 if let Some(a) = actual {
345 detail.actual_wire_type = Some(wire_type_name(a).to_string());
346 detail.raw_value = Some(raw_value_of(a));
347 }
348 ExprFailure {
349 code,
350 message,
351 detail: Some(Box::new(detail)),
352 }
353}
354
355#[cfg(feature = "ir")]
357fn obj_set(pairs: &mut Vec<(String, Value)>, key: &str, val: Value) {
358 if let Some(slot) = pairs.iter_mut().find(|(k, _)| k == key) {
359 slot.1 = val;
360 } else {
361 pairs.push((key.to_string(), val));
362 }
363}
364
365#[cfg(feature = "ir")]
372fn assert_conforms_to_out_type(
373 node_id: &str,
374 v: &Value,
375 t: &J,
376 field: &str,
377) -> Result<Value, ExprFailure> {
378 use crate::expr::ExprFailureCode::{MissingProp, TypeMismatch};
379 use crate::plan::ErrorKind;
380 if let Some(scalar) = t.as_str() {
381 if scalar == "float" {
382 if let Value::Int(i) = v {
383 return Ok(Value::Float(*i as f64)); }
385 }
386 let ok = match scalar {
387 "string" => matches!(v, Value::Str(_)),
388 "int" => matches!(v, Value::Int(_)),
389 "float" => matches!(v, Value::Float(_)),
390 "bool" => matches!(v, Value::Bool(_)),
391 "null" => matches!(v, Value::Null),
392 _ => true,
393 };
394 if !ok {
395 return Err(conform_fail(
396 TypeMismatch,
397 format!(
398 "node '{node_id}': {field}: expected {scalar}, got {}",
399 wire_type_name(v)
400 ),
401 ErrorKind::TypeMismatch,
402 node_id,
403 field,
404 t,
405 Some(v),
406 ));
407 }
408 return Ok(v.clone());
409 }
410 if let Some(inner) = t.get("opt") {
411 if matches!(v, Value::Null) {
412 return Ok(Value::Null);
413 }
414 return assert_conforms_to_out_type(node_id, v, inner, field);
415 }
416 if let Some(inner) = t.get("arr") {
417 match v {
418 Value::Arr(items) => {
419 let mut out = Vec::with_capacity(items.len());
420 for (i, el) in items.iter().enumerate() {
421 out.push(assert_conforms_to_out_type(
422 node_id,
423 el,
424 inner,
425 &format!("{field}[{i}]"),
426 )?);
427 }
428 return Ok(Value::Arr(out));
429 }
430 _ => {
431 return Err(conform_fail(
432 TypeMismatch,
433 format!(
434 "node '{node_id}': {field}: expected arr, got {}",
435 wire_type_name(v)
436 ),
437 ErrorKind::TypeMismatch,
438 node_id,
439 field,
440 t,
441 Some(v),
442 ))
443 }
444 }
445 }
446 if let Some(inner) = t.get("map") {
447 match v {
448 Value::Obj(pairs) => {
449 let mut out = Vec::with_capacity(pairs.len());
450 for (k, mv) in pairs {
451 out.push((
452 k.clone(),
453 assert_conforms_to_out_type(node_id, mv, inner, &format!("{field}.{k}"))?,
454 ));
455 }
456 return Ok(Value::Obj(out));
457 }
458 _ => {
459 return Err(conform_fail(
460 TypeMismatch,
461 format!(
462 "node '{node_id}': {field}: expected map, got {}",
463 wire_type_name(v)
464 ),
465 ErrorKind::TypeMismatch,
466 node_id,
467 field,
468 t,
469 Some(v),
470 ))
471 }
472 }
473 }
474 if let Some(J::Object(fields)) = t.get("obj") {
479 let mut out: Vec<(String, Value)> = match v {
480 Value::Obj(_) => Vec::with_capacity(fields.len()),
481 _ => {
482 return Err(conform_fail(
483 TypeMismatch,
484 format!(
485 "node '{node_id}': {field}: expected obj, got {}",
486 wire_type_name(v)
487 ),
488 ErrorKind::TypeMismatch,
489 node_id,
490 field,
491 t,
492 Some(v),
493 ));
494 }
495 };
496 for (k, ft) in fields {
497 match v.obj_get(k) {
498 None => {
499 if ft.get("opt").is_some() {
500 obj_set(&mut out, k, Value::Null); } else {
502 return Err(conform_fail(
503 MissingProp,
504 format!("node '{node_id}': {field}: missing property .{k}"),
505 ErrorKind::MissingField,
506 node_id,
507 &format!("{field}.{k}"),
508 ft,
509 None,
510 ));
511 }
512 }
513 Some(fv) => {
514 let nv = assert_conforms_to_out_type(node_id, fv, ft, &format!("{field}.{k}"))?;
515 obj_set(&mut out, k, nv);
516 }
517 }
518 }
519 return Ok(Value::Obj(out));
520 }
521 Ok(v.clone())
522}
523
524pub trait ComponentExec {
529 fn exec(
530 &mut self,
531 component: &str,
532 ports: &[(String, Value)],
533 bound: Option<&Value>,
534 ) -> Option<ExecOutcome>;
535
536 fn exec_ctx(
542 &mut self,
543 node_id: &str,
544 component: &str,
545 ports: &[(String, Value)],
546 bound: Option<&Value>,
547 ) -> Option<ExecOutcome> {
548 let _ = node_id;
549 self.exec(component, ports, bound)
550 }
551}
552
553impl ComponentExec for &mut dyn ComponentExec {
570 fn exec(
571 &mut self,
572 component: &str,
573 ports: &[(String, Value)],
574 bound: Option<&Value>,
575 ) -> Option<ExecOutcome> {
576 (**self).exec(component, ports, bound)
577 }
578
579 fn exec_ctx(
580 &mut self,
581 node_id: &str,
582 component: &str,
583 ports: &[(String, Value)],
584 bound: Option<&Value>,
585 ) -> Option<ExecOutcome> {
586 (**self).exec_ctx(node_id, component, ports, bound)
587 }
588}
589
590#[cfg(feature = "ir")]
591fn scope_get<'a>(scope: &'a [(String, Value)], key: &str) -> Option<&'a Value> {
592 scope.iter().find(|(k, _)| k == key).map(|(_, v)| v)
593}
594
595#[cfg(feature = "ir")]
601fn fanout_dedup_drop(
602 aligned_bodies: Vec<Value>,
603 dedupe_key: &str,
604 drop: &str,
605 implicit_source: Option<&str>,
606) -> Vec<Value> {
607 use std::collections::HashSet;
608 let mut items: Vec<Value> = Vec::with_capacity(aligned_bodies.len());
609 let mut seen: HashSet<String> = HashSet::new();
610 for body in aligned_bodies.into_iter() {
611 let fields = match &body {
612 Value::Obj(f) => Some(f),
613 _ => None,
614 };
615 let key_val = fields.and_then(|f| f.iter().find(|(k, _)| k == dedupe_key).map(|(_, v)| v));
616 let has_key = matches!(key_val, Some(v) if !matches!(v, Value::Null));
617 if !has_key {
618 if drop == "dangling" {
619 continue;
620 }
621 items.push(body); continue;
623 }
624 let seen_key = match key_val.unwrap() {
625 Value::Str(s) => format!("s:{s}"),
626 other => format!("j:{other:?}"),
627 };
628 if seen.contains(&seen_key) {
629 continue;
630 }
631 seen.insert(seen_key);
632 match (implicit_source, fields) {
633 (Some(src), Some(f)) if f.iter().any(|(k, _)| k == src) => {
634 let stripped: Vec<(String, Value)> =
635 f.iter().filter(|(k, _)| k != src).cloned().collect();
636 items.push(Value::Obj(stripped));
637 }
638 _ => items.push(body),
639 }
640 }
641 items
642}
643
644#[cfg(feature = "ir")]
645fn node_kind(n: &J) -> Result<&'static str, BehaviorError> {
646 if n.get("fanout").is_some() {
647 Ok("fanout")
648 } else if n.get("map").is_some() {
649 Ok("map")
650 } else if n.get("cond").is_some() {
651 Ok("cond")
652 } else if n.get("component").is_some() {
653 Ok("componentRef")
654 } else {
655 bfail(
656 BehaviorFailureCode::UnknownNodeKind,
657 format!(
658 "body node '{}' is not componentRef/map/cond/fanout",
659 n.get("id").and_then(|v| v.as_str()).unwrap_or("?")
660 ),
661 )
662 }
663}
664
665#[cfg(feature = "ir")]
666fn node_sub(n: &J) -> &J {
667 if let Some(f) = n.get("fanout") {
668 f
669 } else if let Some(m) = n.get("map") {
670 m
671 } else if let Some(c) = n.get("cond") {
672 c
673 } else {
674 n
675 }
676}
677
678#[cfg(feature = "ir")]
679fn node_parent(n: &J) -> Option<&str> {
680 node_sub(n).get("parent").and_then(|v| v.as_str())
681}
682
683#[cfg(feature = "ir")]
684fn node_bind_field(n: &J) -> Option<String> {
685 if n.get("map").is_some() || n.get("cond").is_some() || n.get("fanout").is_some() {
686 return None;
687 }
688 n.get("bindField")
689 .and_then(|v| v.as_str())
690 .map(str::to_string)
691}
692
693#[cfg(feature = "ir")]
694fn node_relation_kind(n: &J) -> Option<RelationKind> {
695 if n.get("cond").is_some() {
696 return None;
697 }
698 let sub = if n.get("map").is_some() || n.get("fanout").is_some() {
699 node_sub(n)
700 } else {
701 n
702 };
703 match sub.get("relationKind").and_then(|v| v.as_str()) {
705 Some("connection") => Some(RelationKind::Connection),
706 _ => None,
707 }
708}
709
710#[cfg(feature = "ir")]
711fn node_policy(n: &J) -> Option<String> {
712 if n.get("cond").is_some() {
713 return None;
714 }
715 let sub = if n.get("map").is_some() || n.get("fanout").is_some() {
716 node_sub(n)
717 } else {
718 n
719 };
720 sub.get("policy")
721 .and_then(|v| v.as_str())
722 .map(str::to_string)
723}
724
725#[cfg(feature = "ir")]
726fn eval_ports(ports: &J, scope: &[(String, Value)]) -> Result<Vec<(String, Value)>, BehaviorError> {
727 let obj = ports.as_object().ok_or_else(|| BehaviorError {
728 code: BehaviorFailureCode::UnknownNodeKind.as_str().to_string(),
729 message: "ports must be an object".into(),
730 detail: None,
731 })?;
732 let mut out = Vec::with_capacity(obj.len());
733 for (k, v) in obj {
734 out.push((k.clone(), evaluate_expression(v, scope)?));
735 }
736 Ok(out)
737}
738
739#[cfg(feature = "ir")]
741fn parse_plan(plan: Option<&J>) -> Option<ExecutionPlanSpec> {
742 let p = plan?;
743 if p.is_null() {
744 return None;
745 }
746 let groups = p
747 .get("groups")?
748 .as_array()?
749 .iter()
750 .map(|g| {
751 g.as_array()
752 .map(|row| {
753 row.iter()
754 .filter_map(|i| i.as_u64().map(|x| x as usize))
755 .collect()
756 })
757 .unwrap_or_default()
758 })
759 .collect();
760 let concurrency = p.get("concurrency").and_then(|c| c.as_i64()).unwrap_or(1);
761 Some(ExecutionPlanSpec {
762 groups,
763 concurrency,
764 })
765}
766
767#[cfg(feature = "ir")]
776pub fn run_behavior(
777 ir: &J,
778 handlers: &mut dyn ComponentExec,
779 input: &[(String, Value)],
780 entry: Option<&str>,
781) -> Result<Value, BehaviorError> {
782 let components = ir
783 .get("components")
784 .and_then(|c| c.as_array())
785 .ok_or_else(|| BehaviorError {
786 code: BehaviorFailureCode::UnknownEntry.as_str().to_string(),
787 message: "IR.components must be an array".into(),
788 detail: None,
789 })?;
790 let comp = match entry {
791 Some(name) => components
792 .iter()
793 .find(|c| c.get("name").and_then(|n| n.as_str()) == Some(name)),
794 None => components.first(),
795 };
796 let comp = match comp {
797 Some(c) => c,
798 None => {
799 return bfail(
800 BehaviorFailureCode::UnknownEntry,
801 format!("component '{}' not found in IR", entry.unwrap_or("<first>")),
802 )
803 }
804 };
805
806 let body: Vec<J> = comp
807 .get("body")
808 .and_then(|b| b.as_array())
809 .cloned()
810 .unwrap_or_default();
811
812 let index_of = |id: &str| {
813 body.iter()
814 .position(|n| n.get("id").and_then(|v| v.as_str()) == Some(id))
815 };
816
817 let mut ops: Vec<OpSpec> = Vec::with_capacity(body.len());
819 for n in &body {
820 let parent = node_parent(n).and_then(index_of);
821 ops.push(OpSpec {
822 id: n
823 .get("id")
824 .and_then(|v| v.as_str())
825 .unwrap_or_default()
826 .to_string(),
827 parent,
828 bind_field: node_bind_field(n),
829 relation_kind: node_relation_kind(n),
830 policy: node_policy(n),
831 });
832 }
833
834 run_behavior_inner(
835 &body,
836 &ops,
837 parse_plan(comp.get("plan")),
838 comp.get("output").cloned().unwrap_or(J::Null),
839 &bind_input_scope(comp, input),
840 comp.get("inputPorts").and_then(|p| p.as_object()),
841 handlers,
842 )
843}
844
845#[cfg(feature = "ir")]
857fn bind_input_scope(comp: &J, input: &[(String, Value)]) -> Vec<(String, Value)> {
858 let mut scope: Vec<(String, Value)> = input.to_vec();
859 if let Some(ports) = comp.get("inputPorts").and_then(|p| p.as_object()) {
860 for (name, schema) in ports {
861 let optional = schema.get("required").and_then(|r| r.as_bool()) == Some(false);
862 if optional && !scope.iter().any(|(k, _)| k == name) {
863 scope.push((name.clone(), Value::Null));
864 }
865 }
866 }
867 scope
868}
869
870#[cfg(feature = "ir")]
874fn run_behavior_inner(
875 body: &[J],
876 ops: &[OpSpec],
877 plan: Option<ExecutionPlanSpec>,
878 output: J,
879 input: &[(String, Value)],
880 input_ports: Option<&serde_json::Map<String, J>>,
881 handlers: &mut dyn ComponentExec,
882) -> Result<Value, BehaviorError> {
883 let index_of = |id: &str| {
884 body.iter()
885 .position(|n| n.get("id").and_then(|v| v.as_str()) == Some(id))
886 };
887
888 let mut results: Vec<(String, Value)> = Vec::new();
889 let mut pending_err: Option<BehaviorError> = None;
890
891 let run = {
892 let results_cell = &mut results;
893 let err_cell = &mut pending_err;
894 let exec = |op: &OpSpec, _bound: Option<&Value>| -> ExecOutcome {
895 if err_cell.is_some() {
896 return ExecOutcome::error("aborted");
897 }
898 let idx = match index_of(&op.id) {
899 Some(i) => i,
900 None => {
901 *err_cell = Some(BehaviorError {
902 code: BehaviorFailureCode::UnknownNodeKind.as_str().to_string(),
903 message: format!("no body node for op '{}'", op.id),
904 detail: None,
905 });
906 return ExecOutcome::error("aborted");
907 }
908 };
909 let node = &body[idx];
910
911 let base_scope = |extra: Option<(&str, &Value)>| -> Vec<(String, Value)> {
912 let mut s: Vec<(String, Value)> = input.to_vec();
913 for (k, v) in results_cell.iter() {
914 s.push((k.clone(), v.clone()));
915 }
916 if let Some((k, v)) = extra {
917 s.push((k.to_string(), v.clone()));
918 }
919 s
920 };
921
922 let outcome = match node_kind(node) {
923 Err(e) => {
924 *err_cell = Some(e);
925 return ExecOutcome::error("aborted");
926 }
927 Ok("cond") => {
928 let c = node.get("cond").unwrap();
929 let cond_expr = serde_json::json!({
930 "cond": [c.get("if"), c.get("then"), c.get("else")]
931 });
932 match evaluate_expression(&cond_expr, &base_scope(None)) {
933 Ok(v) => ExecOutcome::Ok(v),
934 Err(e) => {
935 *err_cell = Some(e.into());
936 return ExecOutcome::error("aborted");
937 }
938 }
939 }
940 Ok("map") => 'map_arm: {
944 let m = node.get("map").unwrap();
945 let over = match evaluate_expression(
946 m.get("over").unwrap_or(&J::Null),
947 &base_scope(None),
948 ) {
949 Ok(v) => v,
950 Err(e) => {
951 *err_cell = Some(e.into());
952 return ExecOutcome::error("aborted");
953 }
954 };
955 let arr = match resolve_over_array(
956 &over,
957 m.get("over").unwrap_or(&J::Null),
958 body,
959 input_ports,
960 BehaviorFailureCode::MapOverNotArray,
961 &op.id,
962 "map",
963 ) {
964 Ok(a) => a,
965 Err(e) => {
966 *err_cell = Some(e);
967 return ExecOutcome::error("aborted");
968 }
969 };
970 let component = m.get("component").and_then(|v| v.as_str()).unwrap_or("");
971 let as_name = m.get("as").and_then(|v| v.as_str()).unwrap_or("$");
972 if m.get("component").is_none() {
976 if let Some(transform) = m.get("transform") {
977 let when_t = m.get("when");
978 let mut out: Vec<Value> = Vec::with_capacity(arr.len());
979 for el in arr.iter() {
980 let scope = base_scope(Some((as_name, el)));
981 if let Some(w) = when_t {
982 let keep_expr = serde_json::json!({ "cond": [w, true, false] });
983 match evaluate_expression(&keep_expr, &scope) {
984 Ok(Value::Bool(true)) => {}
985 Ok(_) => continue,
986 Err(e) => {
987 *err_cell = Some(e.into());
988 return ExecOutcome::error("aborted");
989 }
990 }
991 }
992 match evaluate_expression(transform, &scope) {
993 Ok(v) => out.push(v),
994 Err(e) => {
995 *err_cell = Some(e.into());
996 return ExecOutcome::error("aborted");
997 }
998 }
999 }
1000 break 'map_arm ExecOutcome::Ok(Value::Arr(out));
1001 }
1002 }
1003 let ports_j = m
1004 .get("ports")
1005 .cloned()
1006 .unwrap_or(J::Object(Default::default()));
1007 let when = m.get("when");
1008 let batched = m.get("batched").and_then(|v| v.as_bool()).unwrap_or(false);
1009 let element_policy = {
1013 let raw = m
1014 .get("elementPolicy")
1015 .and_then(|v| v.as_str())
1016 .unwrap_or("error");
1017 match crate::plan::ElementPolicyKind::parse(raw) {
1018 None => {
1019 *err_cell = Some(BehaviorError {
1020 code: "UNKNOWN_ELEMENT_POLICY".to_string(),
1021 message: format!(
1022 "map '{}': unknown element policy '{raw}' (fail-closed)",
1023 op.id
1024 ),
1025 detail: None,
1026 });
1027 return ExecOutcome::error("aborted");
1028 }
1029 Some(crate::plan::ElementPolicyKind::Skip) if batched => {
1030 *err_cell = Some(BehaviorError {
1031 code: "ELEMENT_POLICY_NOT_APPLICABLE".to_string(),
1032 message: format!("map '{}': elementPolicy 'skip' needs a per-element Failure, but a batched map takes ONE outcome for the whole batch (fail-closed)", op.id),
1033 detail: None,
1034 });
1035 return ExecOutcome::error("aborted");
1036 }
1037 Some(p) => p,
1038 }
1039 };
1040 let into = m.get("into").and_then(|v| v.as_str());
1041
1042 let keep = |scope: &[(String, Value)]| -> Result<bool, BehaviorError> {
1045 match when {
1046 None => Ok(true),
1047 Some(w) => {
1048 let cond_expr = serde_json::json!({ "cond": [w, true, false] });
1049 Ok(matches!(
1050 evaluate_expression(&cond_expr, scope)?,
1051 Value::Bool(true)
1052 ))
1053 }
1054 }
1055 };
1056
1057 let mut kept_idx: Vec<usize> = Vec::new(); let collected: Vec<Value>;
1059 if batched {
1060 let mut items: Vec<Value> = Vec::new();
1062 for (i, el) in arr.iter().enumerate() {
1063 let scope = base_scope(Some((as_name, el)));
1064 match keep(&scope) {
1065 Ok(false) => continue,
1066 Ok(true) => {}
1067 Err(e) => {
1068 *err_cell = Some(e);
1069 return ExecOutcome::error("aborted");
1070 }
1071 }
1072 match eval_ports(&ports_j, &scope) {
1073 Ok(p) => items.push(Value::Obj(p)),
1074 Err(e) => {
1075 *err_cell = Some(e);
1076 return ExecOutcome::error("aborted");
1077 }
1078 }
1079 kept_idx.push(i);
1080 }
1081 if items.is_empty() {
1082 collected = Vec::new(); } else {
1084 let want = items.len();
1085 let batch_ports = vec![("items".to_string(), Value::Arr(items))];
1086 match handlers.exec_ctx(&op.id, component, &batch_ports, None) {
1087 None => {
1088 *err_cell = Some(BehaviorError {
1089 code: BehaviorFailureCode::UnknownComponent
1090 .as_str()
1091 .to_string(),
1092 message: format!(
1093 "component '{component}' has no handler (fail-closed)"
1094 ),
1095 detail: None,
1096 });
1097 return ExecOutcome::error("aborted");
1098 }
1099 Some(ExecOutcome::Error(e, d)) => return ExecOutcome::Error(e, d),
1100 Some(ExecOutcome::Ok(Value::Arr(r))) if r.len() == want => {
1101 collected = r;
1102 }
1103 Some(ExecOutcome::Ok(_)) => {
1104 *err_cell = Some(BehaviorError {
1105 code: BehaviorFailureCode::MapBatchResultMismatch
1106 .as_str()
1107 .to_string(),
1108 message: format!(
1109 "map '{}': batched handler must return a list aligned to items (want {want})",
1110 op.id
1111 ),
1112 detail: None,
1113 });
1114 return ExecOutcome::error("aborted");
1115 }
1116 }
1117 }
1118 } else {
1119 let mut out: Vec<Value> = Vec::with_capacity(arr.len());
1120 for (i, el) in arr.iter().enumerate() {
1121 let scope = base_scope(Some((as_name, el)));
1122 match keep(&scope) {
1123 Ok(false) => continue,
1124 Ok(true) => {}
1125 Err(e) => {
1126 *err_cell = Some(e);
1127 return ExecOutcome::error("aborted");
1128 }
1129 }
1130 let ports = match eval_ports(&ports_j, &scope) {
1131 Ok(p) => p,
1132 Err(e) => {
1133 *err_cell = Some(e);
1134 return ExecOutcome::error("aborted");
1135 }
1136 };
1137 match handlers.exec_ctx(&op.id, component, &ports, Some(el)) {
1138 None => {
1139 *err_cell = Some(BehaviorError {
1140 code: BehaviorFailureCode::UnknownComponent
1141 .as_str()
1142 .to_string(),
1143 message: format!(
1144 "component '{component}' has no handler (fail-closed)"
1145 ),
1146 detail: None,
1147 });
1148 return ExecOutcome::error("aborted");
1149 }
1150 Some(ExecOutcome::Error(e, d)) => {
1154 if element_policy == crate::plan::ElementPolicyKind::Skip {
1155 continue;
1156 }
1157 return ExecOutcome::Error(e, d);
1158 }
1159 Some(ExecOutcome::Ok(v)) => out.push(v),
1160 }
1161 kept_idx.push(i);
1162 }
1163 collected = out;
1164 }
1165
1166 match into {
1167 None => ExecOutcome::Ok(Value::Arr(collected)),
1168 Some(key) => {
1169 let node_rk = node_relation_kind(node);
1174 let mut augmented: Vec<Value> = Vec::with_capacity(arr.len());
1175 let mut k = 0usize;
1176 for (i, el) in arr.iter().enumerate() {
1177 let into_val: Option<Value> =
1178 if k < kept_idx.len() && kept_idx[k] == i {
1179 let v = collected[k].clone();
1180 k += 1;
1181 Some(v)
1182 } else if node_rk == Some(RelationKind::Connection) {
1183 Some(unproduced_value(node_rk))
1184 } else {
1185 None
1186 };
1187 let val = match into_val {
1188 None => {
1189 augmented.push(el.clone());
1190 continue;
1191 }
1192 Some(v) => v,
1193 };
1194 let mut fields = match el {
1195 Value::Obj(f) => f.clone(),
1196 _ => {
1197 *err_cell = Some(BehaviorError {
1198 code: BehaviorFailureCode::MapIntoElementNotObject
1199 .as_str()
1200 .to_string(),
1201 message: format!(
1202 "map '{}': 'into' requires object elements (element {i} is not an object)",
1203 op.id
1204 ),
1205 detail: None,
1206 });
1207 return ExecOutcome::error("aborted");
1208 }
1209 };
1210 match fields.iter_mut().find(|(fk, _)| fk == key) {
1211 Some(slot) => slot.1 = val,
1212 None => fields.push((key.to_string(), val)),
1213 }
1214 augmented.push(Value::Obj(fields));
1215 }
1216 ExecOutcome::Ok(Value::Arr(augmented))
1217 }
1218 }
1219 }
1220 Ok("fanout") => {
1221 let f = node.get("fanout").unwrap();
1225 let over = match evaluate_expression(
1226 f.get("over").unwrap_or(&J::Null),
1227 &base_scope(None),
1228 ) {
1229 Ok(v) => v,
1230 Err(e) => {
1231 *err_cell = Some(e.into());
1232 return ExecOutcome::error("aborted");
1233 }
1234 };
1235 let arr = match resolve_over_array(
1236 &over,
1237 f.get("over").unwrap_or(&J::Null),
1238 body,
1239 input_ports,
1240 BehaviorFailureCode::FanoutOverNotArray,
1241 &op.id,
1242 "fanout",
1243 ) {
1244 Ok(a) => a,
1245 Err(e) => {
1246 *err_cell = Some(e);
1247 return ExecOutcome::error("aborted");
1248 }
1249 };
1250 let component = f.get("component").and_then(|v| v.as_str()).unwrap_or("");
1251 let as_name = f.get("as").and_then(|v| v.as_str()).unwrap_or("$");
1252 let ports_j = f
1253 .get("ports")
1254 .cloned()
1255 .unwrap_or(J::Object(Default::default()));
1256 let dedupe_key = f.get("dedupeKey").and_then(|v| v.as_str()).unwrap_or("");
1257 let drop = f.get("drop").and_then(|v| v.as_str()).unwrap_or("dangling");
1258 let implicit_source = f.get("implicitSource").and_then(|v| v.as_str());
1259 let mut items: Vec<Value> = Vec::with_capacity(arr.len());
1260 for el in arr.iter() {
1261 let scope = base_scope(Some((as_name, el)));
1262 match eval_ports(&ports_j, &scope) {
1263 Ok(p) => items.push(Value::Obj(p)),
1264 Err(e) => {
1265 *err_cell = Some(e);
1266 return ExecOutcome::error("aborted");
1267 }
1268 }
1269 }
1270 if items.is_empty() {
1271 ExecOutcome::Ok(Value::Obj(vec![
1272 ("items".into(), Value::Arr(vec![])),
1273 ("cursor".into(), Value::Null),
1274 ]))
1275 } else {
1276 let want = items.len();
1277 let batch_ports = vec![("items".to_string(), Value::Arr(items))];
1278 match handlers.exec_ctx(&op.id, component, &batch_ports, None) {
1279 None => {
1280 *err_cell = Some(BehaviorError {
1281 code: BehaviorFailureCode::UnknownComponent
1282 .as_str()
1283 .to_string(),
1284 message: format!(
1285 "component '{component}' has no handler (fail-closed)"
1286 ),
1287 detail: None,
1288 });
1289 ExecOutcome::error("aborted")
1290 }
1291 Some(ExecOutcome::Error(e, d)) => ExecOutcome::Error(e, d),
1292 Some(ExecOutcome::Ok(Value::Arr(r))) if r.len() == want => {
1293 let deduped =
1294 fanout_dedup_drop(r, dedupe_key, drop, implicit_source);
1295 ExecOutcome::Ok(Value::Obj(vec![
1296 ("items".into(), Value::Arr(deduped)),
1297 ("cursor".into(), Value::Null),
1298 ]))
1299 }
1300 Some(ExecOutcome::Ok(_)) => {
1301 *err_cell = Some(BehaviorError {
1302 code: BehaviorFailureCode::FanoutBatchResultMismatch
1303 .as_str()
1304 .to_string(),
1305 message: format!(
1306 "fanout '{}': batched handler must return a list aligned to the deduped id list (want {want})",
1307 op.id
1308 ),
1309 detail: None,
1310 });
1311 ExecOutcome::error("aborted")
1312 }
1313 }
1314 }
1315 }
1316 Ok(_) => {
1317 let component = node.get("component").and_then(|v| v.as_str()).unwrap_or("");
1318 let ports_j = node
1319 .get("ports")
1320 .cloned()
1321 .unwrap_or(J::Object(Default::default()));
1322 let ports = match eval_ports(&ports_j, &base_scope(None)) {
1323 Ok(p) => p,
1324 Err(e) => {
1325 *err_cell = Some(e);
1326 return ExecOutcome::error("aborted");
1327 }
1328 };
1329 match handlers.exec_ctx(&op.id, component, &ports, None) {
1330 None => {
1331 *err_cell = Some(BehaviorError {
1332 code: BehaviorFailureCode::UnknownComponent.as_str().to_string(),
1333 message: format!(
1334 "component '{component}' has no handler (fail-closed)"
1335 ),
1336 detail: None,
1337 });
1338 return ExecOutcome::error("aborted");
1339 }
1340 Some(o) => o,
1341 }
1342 }
1343 };
1344
1345 if let ExecOutcome::Ok(v) = &outcome {
1346 let mut value = v.clone();
1351 if let Some(ot) = node.get("outType") {
1352 let want = result_type_of(node, ot);
1353 match assert_conforms_to_out_type(&op.id, &value, &want, "result") {
1354 Ok(normalized) => value = normalized,
1355 Err(e) => {
1356 *err_cell = Some(e.into());
1357 return ExecOutcome::error("aborted");
1358 }
1359 }
1360 }
1361 results_cell.push((op.id.clone(), value));
1362 }
1363 outcome
1364 };
1365
1366 run_plan(plan.as_ref(), ops, exec)
1367 };
1368
1369 if let Some(e) = pending_err {
1372 return Err(e);
1373 }
1374 let run = run?;
1375
1376 for (i, id_val) in run.skipped.iter().enumerate() {
1379 let _ = i;
1380 if let Some(idx) = index_of(id_val) {
1381 let unproduced = unproduced_value(node_relation_kind(&body[idx]));
1382 if scope_get(&results, id_val).is_none() {
1383 results.push((id_val.clone(), unproduced));
1384 }
1385 }
1386 }
1387
1388 let scope: Vec<(String, Value)> = {
1389 let mut s: Vec<(String, Value)> = input.to_vec();
1390 s.extend(results.iter().cloned());
1391 s
1392 };
1393 Ok(evaluate_expression(&output, &scope)?)
1394}