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") {
476 let mut out: Vec<(String, Value)> = match v {
477 Value::Obj(pairs) => pairs.clone(),
478 _ => {
479 return Err(conform_fail(
480 TypeMismatch,
481 format!(
482 "node '{node_id}': {field}: expected obj, got {}",
483 wire_type_name(v)
484 ),
485 ErrorKind::TypeMismatch,
486 node_id,
487 field,
488 t,
489 Some(v),
490 ));
491 }
492 };
493 for (k, ft) in fields {
494 match v.obj_get(k) {
495 None => {
496 if ft.get("opt").is_some() {
497 obj_set(&mut out, k, Value::Null); } else {
499 return Err(conform_fail(
500 MissingProp,
501 format!("node '{node_id}': {field}: missing property .{k}"),
502 ErrorKind::MissingField,
503 node_id,
504 &format!("{field}.{k}"),
505 ft,
506 None,
507 ));
508 }
509 }
510 Some(fv) => {
511 let nv = assert_conforms_to_out_type(node_id, fv, ft, &format!("{field}.{k}"))?;
512 obj_set(&mut out, k, nv);
513 }
514 }
515 }
516 return Ok(Value::Obj(out));
517 }
518 Ok(v.clone())
519}
520
521pub trait ComponentExec {
526 fn exec(
527 &mut self,
528 component: &str,
529 ports: &[(String, Value)],
530 bound: Option<&Value>,
531 ) -> Option<ExecOutcome>;
532
533 fn exec_ctx(
539 &mut self,
540 node_id: &str,
541 component: &str,
542 ports: &[(String, Value)],
543 bound: Option<&Value>,
544 ) -> Option<ExecOutcome> {
545 let _ = node_id;
546 self.exec(component, ports, bound)
547 }
548}
549
550impl ComponentExec for &mut dyn ComponentExec {
567 fn exec(
568 &mut self,
569 component: &str,
570 ports: &[(String, Value)],
571 bound: Option<&Value>,
572 ) -> Option<ExecOutcome> {
573 (**self).exec(component, ports, bound)
574 }
575
576 fn exec_ctx(
577 &mut self,
578 node_id: &str,
579 component: &str,
580 ports: &[(String, Value)],
581 bound: Option<&Value>,
582 ) -> Option<ExecOutcome> {
583 (**self).exec_ctx(node_id, component, ports, bound)
584 }
585}
586
587#[cfg(feature = "ir")]
588fn scope_get<'a>(scope: &'a [(String, Value)], key: &str) -> Option<&'a Value> {
589 scope.iter().find(|(k, _)| k == key).map(|(_, v)| v)
590}
591
592#[cfg(feature = "ir")]
598fn fanout_dedup_drop(
599 aligned_bodies: Vec<Value>,
600 dedupe_key: &str,
601 drop: &str,
602 implicit_source: Option<&str>,
603) -> Vec<Value> {
604 use std::collections::HashSet;
605 let mut items: Vec<Value> = Vec::with_capacity(aligned_bodies.len());
606 let mut seen: HashSet<String> = HashSet::new();
607 for body in aligned_bodies.into_iter() {
608 let fields = match &body {
609 Value::Obj(f) => Some(f),
610 _ => None,
611 };
612 let key_val = fields.and_then(|f| f.iter().find(|(k, _)| k == dedupe_key).map(|(_, v)| v));
613 let has_key = matches!(key_val, Some(v) if !matches!(v, Value::Null));
614 if !has_key {
615 if drop == "dangling" {
616 continue;
617 }
618 items.push(body); continue;
620 }
621 let seen_key = match key_val.unwrap() {
622 Value::Str(s) => format!("s:{s}"),
623 other => format!("j:{other:?}"),
624 };
625 if seen.contains(&seen_key) {
626 continue;
627 }
628 seen.insert(seen_key);
629 match (implicit_source, fields) {
630 (Some(src), Some(f)) if f.iter().any(|(k, _)| k == src) => {
631 let stripped: Vec<(String, Value)> =
632 f.iter().filter(|(k, _)| k != src).cloned().collect();
633 items.push(Value::Obj(stripped));
634 }
635 _ => items.push(body),
636 }
637 }
638 items
639}
640
641#[cfg(feature = "ir")]
642fn node_kind(n: &J) -> Result<&'static str, BehaviorError> {
643 if n.get("fanout").is_some() {
644 Ok("fanout")
645 } else if n.get("map").is_some() {
646 Ok("map")
647 } else if n.get("cond").is_some() {
648 Ok("cond")
649 } else if n.get("component").is_some() {
650 Ok("componentRef")
651 } else {
652 bfail(
653 BehaviorFailureCode::UnknownNodeKind,
654 format!(
655 "body node '{}' is not componentRef/map/cond/fanout",
656 n.get("id").and_then(|v| v.as_str()).unwrap_or("?")
657 ),
658 )
659 }
660}
661
662#[cfg(feature = "ir")]
663fn node_sub(n: &J) -> &J {
664 if let Some(f) = n.get("fanout") {
665 f
666 } else if let Some(m) = n.get("map") {
667 m
668 } else if let Some(c) = n.get("cond") {
669 c
670 } else {
671 n
672 }
673}
674
675#[cfg(feature = "ir")]
676fn node_parent(n: &J) -> Option<&str> {
677 node_sub(n).get("parent").and_then(|v| v.as_str())
678}
679
680#[cfg(feature = "ir")]
681fn node_bind_field(n: &J) -> Option<String> {
682 if n.get("map").is_some() || n.get("cond").is_some() || n.get("fanout").is_some() {
683 return None;
684 }
685 n.get("bindField")
686 .and_then(|v| v.as_str())
687 .map(str::to_string)
688}
689
690#[cfg(feature = "ir")]
691fn node_relation_kind(n: &J) -> Option<RelationKind> {
692 if n.get("cond").is_some() {
693 return None;
694 }
695 let sub = if n.get("map").is_some() || n.get("fanout").is_some() {
696 node_sub(n)
697 } else {
698 n
699 };
700 match sub.get("relationKind").and_then(|v| v.as_str()) {
701 Some("connection") => Some(RelationKind::Connection),
702 Some(_) => Some(RelationKind::Single),
703 None => None,
704 }
705}
706
707#[cfg(feature = "ir")]
708fn node_policy(n: &J) -> Option<String> {
709 if n.get("cond").is_some() {
710 return None;
711 }
712 let sub = if n.get("map").is_some() || n.get("fanout").is_some() {
713 node_sub(n)
714 } else {
715 n
716 };
717 sub.get("policy")
718 .and_then(|v| v.as_str())
719 .map(str::to_string)
720}
721
722#[cfg(feature = "ir")]
723fn eval_ports(ports: &J, scope: &[(String, Value)]) -> Result<Vec<(String, Value)>, BehaviorError> {
724 let obj = ports.as_object().ok_or_else(|| BehaviorError {
725 code: BehaviorFailureCode::UnknownNodeKind.as_str().to_string(),
726 message: "ports must be an object".into(),
727 detail: None,
728 })?;
729 let mut out = Vec::with_capacity(obj.len());
730 for (k, v) in obj {
731 out.push((k.clone(), evaluate_expression(v, scope)?));
732 }
733 Ok(out)
734}
735
736#[cfg(feature = "ir")]
738fn parse_plan(plan: Option<&J>) -> Option<ExecutionPlanSpec> {
739 let p = plan?;
740 if p.is_null() {
741 return None;
742 }
743 let groups = p
744 .get("groups")?
745 .as_array()?
746 .iter()
747 .map(|g| {
748 g.as_array()
749 .map(|row| {
750 row.iter()
751 .filter_map(|i| i.as_u64().map(|x| x as usize))
752 .collect()
753 })
754 .unwrap_or_default()
755 })
756 .collect();
757 let concurrency = p.get("concurrency").and_then(|c| c.as_i64()).unwrap_or(1);
758 Some(ExecutionPlanSpec {
759 groups,
760 concurrency,
761 })
762}
763
764#[cfg(feature = "ir")]
773pub fn run_behavior(
774 ir: &J,
775 handlers: &mut dyn ComponentExec,
776 input: &[(String, Value)],
777 entry: Option<&str>,
778) -> Result<Value, BehaviorError> {
779 let components = ir
780 .get("components")
781 .and_then(|c| c.as_array())
782 .ok_or_else(|| BehaviorError {
783 code: BehaviorFailureCode::UnknownEntry.as_str().to_string(),
784 message: "IR.components must be an array".into(),
785 detail: None,
786 })?;
787 let comp = match entry {
788 Some(name) => components
789 .iter()
790 .find(|c| c.get("name").and_then(|n| n.as_str()) == Some(name)),
791 None => components.first(),
792 };
793 let comp = match comp {
794 Some(c) => c,
795 None => {
796 return bfail(
797 BehaviorFailureCode::UnknownEntry,
798 format!("component '{}' not found in IR", entry.unwrap_or("<first>")),
799 )
800 }
801 };
802
803 let body: Vec<J> = comp
804 .get("body")
805 .and_then(|b| b.as_array())
806 .cloned()
807 .unwrap_or_default();
808
809 let index_of = |id: &str| {
810 body.iter()
811 .position(|n| n.get("id").and_then(|v| v.as_str()) == Some(id))
812 };
813
814 let mut ops: Vec<OpSpec> = Vec::with_capacity(body.len());
816 for n in &body {
817 let parent = node_parent(n).and_then(index_of);
818 ops.push(OpSpec {
819 id: n
820 .get("id")
821 .and_then(|v| v.as_str())
822 .unwrap_or_default()
823 .to_string(),
824 parent,
825 bind_field: node_bind_field(n),
826 relation_kind: node_relation_kind(n),
827 policy: node_policy(n),
828 });
829 }
830
831 run_behavior_inner(
832 &body,
833 &ops,
834 parse_plan(comp.get("plan")),
835 comp.get("output").cloned().unwrap_or(J::Null),
836 &bind_input_scope(comp, input),
837 comp.get("inputPorts").and_then(|p| p.as_object()),
838 handlers,
839 )
840}
841
842#[cfg(feature = "ir")]
854fn bind_input_scope(comp: &J, input: &[(String, Value)]) -> Vec<(String, Value)> {
855 let mut scope: Vec<(String, Value)> = input.to_vec();
856 if let Some(ports) = comp.get("inputPorts").and_then(|p| p.as_object()) {
857 for (name, schema) in ports {
858 let optional = schema.get("required").and_then(|r| r.as_bool()) == Some(false);
859 if optional && !scope.iter().any(|(k, _)| k == name) {
860 scope.push((name.clone(), Value::Null));
861 }
862 }
863 }
864 scope
865}
866
867#[cfg(feature = "ir")]
871fn run_behavior_inner(
872 body: &[J],
873 ops: &[OpSpec],
874 plan: Option<ExecutionPlanSpec>,
875 output: J,
876 input: &[(String, Value)],
877 input_ports: Option<&serde_json::Map<String, J>>,
878 handlers: &mut dyn ComponentExec,
879) -> Result<Value, BehaviorError> {
880 let index_of = |id: &str| {
881 body.iter()
882 .position(|n| n.get("id").and_then(|v| v.as_str()) == Some(id))
883 };
884
885 let mut results: Vec<(String, Value)> = Vec::new();
886 let mut pending_err: Option<BehaviorError> = None;
887
888 let run = {
889 let results_cell = &mut results;
890 let err_cell = &mut pending_err;
891 let exec = |op: &OpSpec, _bound: Option<&Value>| -> ExecOutcome {
892 if err_cell.is_some() {
893 return ExecOutcome::error("aborted");
894 }
895 let idx = match index_of(&op.id) {
896 Some(i) => i,
897 None => {
898 *err_cell = Some(BehaviorError {
899 code: BehaviorFailureCode::UnknownNodeKind.as_str().to_string(),
900 message: format!("no body node for op '{}'", op.id),
901 detail: None,
902 });
903 return ExecOutcome::error("aborted");
904 }
905 };
906 let node = &body[idx];
907
908 let base_scope = |extra: Option<(&str, &Value)>| -> Vec<(String, Value)> {
909 let mut s: Vec<(String, Value)> = input.to_vec();
910 for (k, v) in results_cell.iter() {
911 s.push((k.clone(), v.clone()));
912 }
913 if let Some((k, v)) = extra {
914 s.push((k.to_string(), v.clone()));
915 }
916 s
917 };
918
919 let outcome = match node_kind(node) {
920 Err(e) => {
921 *err_cell = Some(e);
922 return ExecOutcome::error("aborted");
923 }
924 Ok("cond") => {
925 let c = node.get("cond").unwrap();
926 let cond_expr = serde_json::json!({
927 "cond": [c.get("if"), c.get("then"), c.get("else")]
928 });
929 match evaluate_expression(&cond_expr, &base_scope(None)) {
930 Ok(v) => ExecOutcome::Ok(v),
931 Err(e) => {
932 *err_cell = Some(e.into());
933 return ExecOutcome::error("aborted");
934 }
935 }
936 }
937 Ok("map") => {
938 let m = node.get("map").unwrap();
939 let over = match evaluate_expression(
940 m.get("over").unwrap_or(&J::Null),
941 &base_scope(None),
942 ) {
943 Ok(v) => v,
944 Err(e) => {
945 *err_cell = Some(e.into());
946 return ExecOutcome::error("aborted");
947 }
948 };
949 let arr = match resolve_over_array(
950 &over,
951 m.get("over").unwrap_or(&J::Null),
952 body,
953 input_ports,
954 BehaviorFailureCode::MapOverNotArray,
955 &op.id,
956 "map",
957 ) {
958 Ok(a) => a,
959 Err(e) => {
960 *err_cell = Some(e);
961 return ExecOutcome::error("aborted");
962 }
963 };
964 let component = m.get("component").and_then(|v| v.as_str()).unwrap_or("");
965 let as_name = m.get("as").and_then(|v| v.as_str()).unwrap_or("$");
966 let ports_j = m
967 .get("ports")
968 .cloned()
969 .unwrap_or(J::Object(Default::default()));
970 let when = m.get("when");
971 let batched = m.get("batched").and_then(|v| v.as_bool()).unwrap_or(false);
972 let element_policy = {
976 let raw = m
977 .get("elementPolicy")
978 .and_then(|v| v.as_str())
979 .unwrap_or("error");
980 match crate::plan::ElementPolicyKind::parse(raw) {
981 None => {
982 *err_cell = Some(BehaviorError {
983 code: "UNKNOWN_ELEMENT_POLICY".to_string(),
984 message: format!(
985 "map '{}': unknown element policy '{raw}' (fail-closed)",
986 op.id
987 ),
988 detail: None,
989 });
990 return ExecOutcome::error("aborted");
991 }
992 Some(crate::plan::ElementPolicyKind::Skip) if batched => {
993 *err_cell = Some(BehaviorError {
994 code: "ELEMENT_POLICY_NOT_APPLICABLE".to_string(),
995 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),
996 detail: None,
997 });
998 return ExecOutcome::error("aborted");
999 }
1000 Some(p) => p,
1001 }
1002 };
1003 let into = m.get("into").and_then(|v| v.as_str());
1004
1005 let keep = |scope: &[(String, Value)]| -> Result<bool, BehaviorError> {
1008 match when {
1009 None => Ok(true),
1010 Some(w) => {
1011 let cond_expr = serde_json::json!({ "cond": [w, true, false] });
1012 Ok(matches!(
1013 evaluate_expression(&cond_expr, scope)?,
1014 Value::Bool(true)
1015 ))
1016 }
1017 }
1018 };
1019
1020 let mut kept_idx: Vec<usize> = Vec::new(); let collected: Vec<Value>;
1022 if batched {
1023 let mut items: Vec<Value> = Vec::new();
1025 for (i, el) in arr.iter().enumerate() {
1026 let scope = base_scope(Some((as_name, el)));
1027 match keep(&scope) {
1028 Ok(false) => continue,
1029 Ok(true) => {}
1030 Err(e) => {
1031 *err_cell = Some(e);
1032 return ExecOutcome::error("aborted");
1033 }
1034 }
1035 match eval_ports(&ports_j, &scope) {
1036 Ok(p) => items.push(Value::Obj(p)),
1037 Err(e) => {
1038 *err_cell = Some(e);
1039 return ExecOutcome::error("aborted");
1040 }
1041 }
1042 kept_idx.push(i);
1043 }
1044 if items.is_empty() {
1045 collected = Vec::new(); } else {
1047 let want = items.len();
1048 let batch_ports = vec![("items".to_string(), Value::Arr(items))];
1049 match handlers.exec_ctx(&op.id, component, &batch_ports, None) {
1050 None => {
1051 *err_cell = Some(BehaviorError {
1052 code: BehaviorFailureCode::UnknownComponent
1053 .as_str()
1054 .to_string(),
1055 message: format!(
1056 "component '{component}' has no handler (fail-closed)"
1057 ),
1058 detail: None,
1059 });
1060 return ExecOutcome::error("aborted");
1061 }
1062 Some(ExecOutcome::Error(e, d)) => return ExecOutcome::Error(e, d),
1063 Some(ExecOutcome::Ok(Value::Arr(r))) if r.len() == want => {
1064 collected = r;
1065 }
1066 Some(ExecOutcome::Ok(_)) => {
1067 *err_cell = Some(BehaviorError {
1068 code: BehaviorFailureCode::MapBatchResultMismatch
1069 .as_str()
1070 .to_string(),
1071 message: format!(
1072 "map '{}': batched handler must return a list aligned to items (want {want})",
1073 op.id
1074 ),
1075 detail: None,
1076 });
1077 return ExecOutcome::error("aborted");
1078 }
1079 }
1080 }
1081 } else {
1082 let mut out: Vec<Value> = Vec::with_capacity(arr.len());
1083 for (i, el) in arr.iter().enumerate() {
1084 let scope = base_scope(Some((as_name, el)));
1085 match keep(&scope) {
1086 Ok(false) => continue,
1087 Ok(true) => {}
1088 Err(e) => {
1089 *err_cell = Some(e);
1090 return ExecOutcome::error("aborted");
1091 }
1092 }
1093 let ports = match eval_ports(&ports_j, &scope) {
1094 Ok(p) => p,
1095 Err(e) => {
1096 *err_cell = Some(e);
1097 return ExecOutcome::error("aborted");
1098 }
1099 };
1100 match handlers.exec_ctx(&op.id, component, &ports, Some(el)) {
1101 None => {
1102 *err_cell = Some(BehaviorError {
1103 code: BehaviorFailureCode::UnknownComponent
1104 .as_str()
1105 .to_string(),
1106 message: format!(
1107 "component '{component}' has no handler (fail-closed)"
1108 ),
1109 detail: None,
1110 });
1111 return ExecOutcome::error("aborted");
1112 }
1113 Some(ExecOutcome::Error(e, d)) => {
1117 if element_policy == crate::plan::ElementPolicyKind::Skip {
1118 continue;
1119 }
1120 return ExecOutcome::Error(e, d);
1121 }
1122 Some(ExecOutcome::Ok(v)) => out.push(v),
1123 }
1124 kept_idx.push(i);
1125 }
1126 collected = out;
1127 }
1128
1129 match into {
1130 None => ExecOutcome::Ok(Value::Arr(collected)),
1131 Some(key) => {
1132 let node_rk = node_relation_kind(node);
1137 let mut augmented: Vec<Value> = Vec::with_capacity(arr.len());
1138 let mut k = 0usize;
1139 for (i, el) in arr.iter().enumerate() {
1140 let into_val: Option<Value> =
1141 if k < kept_idx.len() && kept_idx[k] == i {
1142 let v = collected[k].clone();
1143 k += 1;
1144 Some(v)
1145 } else if node_rk == Some(RelationKind::Connection) {
1146 Some(unproduced_value(node_rk))
1147 } else {
1148 None
1149 };
1150 let val = match into_val {
1151 None => {
1152 augmented.push(el.clone());
1153 continue;
1154 }
1155 Some(v) => v,
1156 };
1157 let mut fields = match el {
1158 Value::Obj(f) => f.clone(),
1159 _ => {
1160 *err_cell = Some(BehaviorError {
1161 code: BehaviorFailureCode::MapIntoElementNotObject
1162 .as_str()
1163 .to_string(),
1164 message: format!(
1165 "map '{}': 'into' requires object elements (element {i} is not an object)",
1166 op.id
1167 ),
1168 detail: None,
1169 });
1170 return ExecOutcome::error("aborted");
1171 }
1172 };
1173 match fields.iter_mut().find(|(fk, _)| fk == key) {
1174 Some(slot) => slot.1 = val,
1175 None => fields.push((key.to_string(), val)),
1176 }
1177 augmented.push(Value::Obj(fields));
1178 }
1179 ExecOutcome::Ok(Value::Arr(augmented))
1180 }
1181 }
1182 }
1183 Ok("fanout") => {
1184 let f = node.get("fanout").unwrap();
1188 let over = match evaluate_expression(
1189 f.get("over").unwrap_or(&J::Null),
1190 &base_scope(None),
1191 ) {
1192 Ok(v) => v,
1193 Err(e) => {
1194 *err_cell = Some(e.into());
1195 return ExecOutcome::error("aborted");
1196 }
1197 };
1198 let arr = match resolve_over_array(
1199 &over,
1200 f.get("over").unwrap_or(&J::Null),
1201 body,
1202 input_ports,
1203 BehaviorFailureCode::FanoutOverNotArray,
1204 &op.id,
1205 "fanout",
1206 ) {
1207 Ok(a) => a,
1208 Err(e) => {
1209 *err_cell = Some(e);
1210 return ExecOutcome::error("aborted");
1211 }
1212 };
1213 let component = f.get("component").and_then(|v| v.as_str()).unwrap_or("");
1214 let as_name = f.get("as").and_then(|v| v.as_str()).unwrap_or("$");
1215 let ports_j = f
1216 .get("ports")
1217 .cloned()
1218 .unwrap_or(J::Object(Default::default()));
1219 let dedupe_key = f.get("dedupeKey").and_then(|v| v.as_str()).unwrap_or("");
1220 let drop = f.get("drop").and_then(|v| v.as_str()).unwrap_or("dangling");
1221 let implicit_source = f.get("implicitSource").and_then(|v| v.as_str());
1222 let mut items: Vec<Value> = Vec::with_capacity(arr.len());
1223 for el in arr.iter() {
1224 let scope = base_scope(Some((as_name, el)));
1225 match eval_ports(&ports_j, &scope) {
1226 Ok(p) => items.push(Value::Obj(p)),
1227 Err(e) => {
1228 *err_cell = Some(e);
1229 return ExecOutcome::error("aborted");
1230 }
1231 }
1232 }
1233 if items.is_empty() {
1234 ExecOutcome::Ok(Value::Obj(vec![
1235 ("items".into(), Value::Arr(vec![])),
1236 ("cursor".into(), Value::Null),
1237 ]))
1238 } else {
1239 let want = items.len();
1240 let batch_ports = vec![("items".to_string(), Value::Arr(items))];
1241 match handlers.exec_ctx(&op.id, component, &batch_ports, None) {
1242 None => {
1243 *err_cell = Some(BehaviorError {
1244 code: BehaviorFailureCode::UnknownComponent
1245 .as_str()
1246 .to_string(),
1247 message: format!(
1248 "component '{component}' has no handler (fail-closed)"
1249 ),
1250 detail: None,
1251 });
1252 ExecOutcome::error("aborted")
1253 }
1254 Some(ExecOutcome::Error(e, d)) => ExecOutcome::Error(e, d),
1255 Some(ExecOutcome::Ok(Value::Arr(r))) if r.len() == want => {
1256 let deduped =
1257 fanout_dedup_drop(r, dedupe_key, drop, implicit_source);
1258 ExecOutcome::Ok(Value::Obj(vec![
1259 ("items".into(), Value::Arr(deduped)),
1260 ("cursor".into(), Value::Null),
1261 ]))
1262 }
1263 Some(ExecOutcome::Ok(_)) => {
1264 *err_cell = Some(BehaviorError {
1265 code: BehaviorFailureCode::FanoutBatchResultMismatch
1266 .as_str()
1267 .to_string(),
1268 message: format!(
1269 "fanout '{}': batched handler must return a list aligned to the deduped id list (want {want})",
1270 op.id
1271 ),
1272 detail: None,
1273 });
1274 ExecOutcome::error("aborted")
1275 }
1276 }
1277 }
1278 }
1279 Ok(_) => {
1280 let component = node.get("component").and_then(|v| v.as_str()).unwrap_or("");
1281 let ports_j = node
1282 .get("ports")
1283 .cloned()
1284 .unwrap_or(J::Object(Default::default()));
1285 let ports = match eval_ports(&ports_j, &base_scope(None)) {
1286 Ok(p) => p,
1287 Err(e) => {
1288 *err_cell = Some(e);
1289 return ExecOutcome::error("aborted");
1290 }
1291 };
1292 match handlers.exec_ctx(&op.id, component, &ports, None) {
1293 None => {
1294 *err_cell = Some(BehaviorError {
1295 code: BehaviorFailureCode::UnknownComponent.as_str().to_string(),
1296 message: format!(
1297 "component '{component}' has no handler (fail-closed)"
1298 ),
1299 detail: None,
1300 });
1301 return ExecOutcome::error("aborted");
1302 }
1303 Some(o) => o,
1304 }
1305 }
1306 };
1307
1308 if let ExecOutcome::Ok(v) = &outcome {
1309 let mut value = v.clone();
1314 if let Some(ot) = node.get("outType") {
1315 let want = result_type_of(node, ot);
1316 match assert_conforms_to_out_type(&op.id, &value, &want, "result") {
1317 Ok(normalized) => value = normalized,
1318 Err(e) => {
1319 *err_cell = Some(e.into());
1320 return ExecOutcome::error("aborted");
1321 }
1322 }
1323 }
1324 results_cell.push((op.id.clone(), value));
1325 }
1326 outcome
1327 };
1328
1329 run_plan(plan.as_ref(), ops, exec)
1330 };
1331
1332 if let Some(e) = pending_err {
1335 return Err(e);
1336 }
1337 let run = run?;
1338
1339 for (i, id_val) in run.skipped.iter().enumerate() {
1342 let _ = i;
1343 if let Some(idx) = index_of(id_val) {
1344 let unproduced = unproduced_value(node_relation_kind(&body[idx]));
1345 if scope_get(&results, id_val).is_none() {
1346 results.push((id_val.clone(), unproduced));
1347 }
1348 }
1349 }
1350
1351 let scope: Vec<(String, Value)> = {
1352 let mut s: Vec<(String, Value)> = input.to_vec();
1353 s.extend(results.iter().cloned());
1354 s
1355 };
1356 Ok(evaluate_expression(&output, &scope)?)
1357}