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 if scalar == "int" {
389 if let Value::Float(f) = v {
390 if *f >= -9223372036854775808.0 && *f < 9223372036854775808.0 && *f == f.trunc() {
391 return Ok(Value::Int(*f as i64));
392 }
393 }
394 }
395 let ok = match scalar {
396 "string" => matches!(v, Value::Str(_)),
397 "int" => matches!(v, Value::Int(_)),
398 "float" => matches!(v, Value::Float(_)),
399 "bool" => matches!(v, Value::Bool(_)),
400 "null" => matches!(v, Value::Null),
401 _ => true,
402 };
403 if !ok {
404 return Err(conform_fail(
405 TypeMismatch,
406 format!(
407 "node '{node_id}': {field}: expected {scalar}, got {}",
408 wire_type_name(v)
409 ),
410 ErrorKind::TypeMismatch,
411 node_id,
412 field,
413 t,
414 Some(v),
415 ));
416 }
417 return Ok(v.clone());
418 }
419 if let Some(inner) = t.get("opt") {
420 if matches!(v, Value::Null) {
421 return Ok(Value::Null);
422 }
423 return assert_conforms_to_out_type(node_id, v, inner, field);
424 }
425 if let Some(inner) = t.get("arr") {
426 match v {
427 Value::Arr(items) => {
428 let mut out = Vec::with_capacity(items.len());
429 for (i, el) in items.iter().enumerate() {
430 out.push(assert_conforms_to_out_type(
431 node_id,
432 el,
433 inner,
434 &format!("{field}[{i}]"),
435 )?);
436 }
437 return Ok(Value::Arr(out));
438 }
439 _ => {
440 return Err(conform_fail(
441 TypeMismatch,
442 format!(
443 "node '{node_id}': {field}: expected arr, got {}",
444 wire_type_name(v)
445 ),
446 ErrorKind::TypeMismatch,
447 node_id,
448 field,
449 t,
450 Some(v),
451 ))
452 }
453 }
454 }
455 if let Some(inner) = t.get("map") {
456 match v {
457 Value::Obj(pairs) => {
458 let mut out = Vec::with_capacity(pairs.len());
459 for (k, mv) in pairs {
460 out.push((
461 k.clone(),
462 assert_conforms_to_out_type(node_id, mv, inner, &format!("{field}.{k}"))?,
463 ));
464 }
465 return Ok(Value::Obj(out));
466 }
467 _ => {
468 return Err(conform_fail(
469 TypeMismatch,
470 format!(
471 "node '{node_id}': {field}: expected map, got {}",
472 wire_type_name(v)
473 ),
474 ErrorKind::TypeMismatch,
475 node_id,
476 field,
477 t,
478 Some(v),
479 ))
480 }
481 }
482 }
483 if let Some(J::Object(fields)) = t.get("obj") {
488 let mut out: Vec<(String, Value)> = match v {
489 Value::Obj(_) => Vec::with_capacity(fields.len()),
490 _ => {
491 return Err(conform_fail(
492 TypeMismatch,
493 format!(
494 "node '{node_id}': {field}: expected obj, got {}",
495 wire_type_name(v)
496 ),
497 ErrorKind::TypeMismatch,
498 node_id,
499 field,
500 t,
501 Some(v),
502 ));
503 }
504 };
505 for (k, ft) in fields {
506 match v.obj_get(k) {
507 None => {
508 if ft.get("opt").is_some() {
509 obj_set(&mut out, k, Value::Null); } else {
511 return Err(conform_fail(
512 MissingProp,
513 format!("node '{node_id}': {field}: missing property .{k}"),
514 ErrorKind::MissingField,
515 node_id,
516 &format!("{field}.{k}"),
517 ft,
518 None,
519 ));
520 }
521 }
522 Some(fv) => {
523 let nv = assert_conforms_to_out_type(node_id, fv, ft, &format!("{field}.{k}"))?;
524 obj_set(&mut out, k, nv);
525 }
526 }
527 }
528 return Ok(Value::Obj(out));
529 }
530 Ok(v.clone())
531}
532
533pub trait ComponentExec {
538 fn exec(
539 &mut self,
540 component: &str,
541 ports: &[(String, Value)],
542 bound: Option<&Value>,
543 ) -> Option<ExecOutcome>;
544
545 fn exec_ctx(
551 &mut self,
552 node_id: &str,
553 component: &str,
554 ports: &[(String, Value)],
555 bound: Option<&Value>,
556 ) -> Option<ExecOutcome> {
557 let _ = node_id;
558 self.exec(component, ports, bound)
559 }
560}
561
562impl ComponentExec for &mut dyn ComponentExec {
579 fn exec(
580 &mut self,
581 component: &str,
582 ports: &[(String, Value)],
583 bound: Option<&Value>,
584 ) -> Option<ExecOutcome> {
585 (**self).exec(component, ports, bound)
586 }
587
588 fn exec_ctx(
589 &mut self,
590 node_id: &str,
591 component: &str,
592 ports: &[(String, Value)],
593 bound: Option<&Value>,
594 ) -> Option<ExecOutcome> {
595 (**self).exec_ctx(node_id, component, ports, bound)
596 }
597}
598
599#[cfg(feature = "ir")]
600fn scope_get<'a>(scope: &'a [(String, Value)], key: &str) -> Option<&'a Value> {
601 scope.iter().find(|(k, _)| k == key).map(|(_, v)| v)
602}
603
604#[cfg(feature = "ir")]
610fn fanout_dedup_drop(
611 aligned_bodies: Vec<Value>,
612 dedupe_key: &str,
613 drop: &str,
614 implicit_source: Option<&str>,
615) -> Vec<Value> {
616 use std::collections::HashSet;
617 let mut items: Vec<Value> = Vec::with_capacity(aligned_bodies.len());
618 let mut seen: HashSet<String> = HashSet::new();
619 for body in aligned_bodies.into_iter() {
620 let fields = match &body {
621 Value::Obj(f) => Some(f),
622 _ => None,
623 };
624 let key_val = fields.and_then(|f| f.iter().find(|(k, _)| k == dedupe_key).map(|(_, v)| v));
625 let has_key = matches!(key_val, Some(v) if !matches!(v, Value::Null));
626 if !has_key {
627 if drop == "dangling" {
628 continue;
629 }
630 items.push(body); continue;
632 }
633 let seen_key = match key_val.unwrap() {
634 Value::Str(s) => format!("s:{s}"),
635 other => format!("j:{other:?}"),
636 };
637 if seen.contains(&seen_key) {
638 continue;
639 }
640 seen.insert(seen_key);
641 match (implicit_source, fields) {
642 (Some(src), Some(f)) if f.iter().any(|(k, _)| k == src) => {
643 let stripped: Vec<(String, Value)> =
644 f.iter().filter(|(k, _)| k != src).cloned().collect();
645 items.push(Value::Obj(stripped));
646 }
647 _ => items.push(body),
648 }
649 }
650 items
651}
652
653#[cfg(feature = "ir")]
654fn node_kind(n: &J) -> Result<&'static str, BehaviorError> {
655 if n.get("fanout").is_some() {
656 Ok("fanout")
657 } else if n.get("map").is_some() {
658 Ok("map")
659 } else if n.get("cond").is_some() {
660 Ok("cond")
661 } else if n.get("component").is_some() {
662 Ok("componentRef")
663 } else {
664 bfail(
665 BehaviorFailureCode::UnknownNodeKind,
666 format!(
667 "body node '{}' is not componentRef/map/cond/fanout",
668 n.get("id").and_then(|v| v.as_str()).unwrap_or("?")
669 ),
670 )
671 }
672}
673
674#[cfg(feature = "ir")]
675fn node_sub(n: &J) -> &J {
676 if let Some(f) = n.get("fanout") {
677 f
678 } else if let Some(m) = n.get("map") {
679 m
680 } else if let Some(c) = n.get("cond") {
681 c
682 } else {
683 n
684 }
685}
686
687#[cfg(feature = "ir")]
688fn node_parent(n: &J) -> Option<&str> {
689 node_sub(n).get("parent").and_then(|v| v.as_str())
690}
691
692#[cfg(feature = "ir")]
693fn node_bind_field(n: &J) -> Option<String> {
694 if n.get("map").is_some() || n.get("cond").is_some() || n.get("fanout").is_some() {
695 return None;
696 }
697 n.get("bindField")
698 .and_then(|v| v.as_str())
699 .map(str::to_string)
700}
701
702#[cfg(feature = "ir")]
703fn node_relation_kind(n: &J) -> Option<RelationKind> {
704 if n.get("cond").is_some() {
705 return None;
706 }
707 let sub = if n.get("map").is_some() || n.get("fanout").is_some() {
708 node_sub(n)
709 } else {
710 n
711 };
712 match sub.get("relationKind").and_then(|v| v.as_str()) {
714 Some("connection") => Some(RelationKind::Connection),
715 _ => None,
716 }
717}
718
719#[cfg(feature = "ir")]
720fn node_policy(n: &J) -> Option<String> {
721 if n.get("cond").is_some() {
722 return None;
723 }
724 let sub = if n.get("map").is_some() || n.get("fanout").is_some() {
725 node_sub(n)
726 } else {
727 n
728 };
729 sub.get("policy")
730 .and_then(|v| v.as_str())
731 .map(str::to_string)
732}
733
734#[cfg(feature = "ir")]
735fn eval_ports(ports: &J, scope: &[(String, Value)]) -> Result<Vec<(String, Value)>, BehaviorError> {
736 let obj = ports.as_object().ok_or_else(|| BehaviorError {
737 code: BehaviorFailureCode::UnknownNodeKind.as_str().to_string(),
738 message: "ports must be an object".into(),
739 detail: None,
740 })?;
741 let mut out = Vec::with_capacity(obj.len());
742 for (k, v) in obj {
743 out.push((k.clone(), evaluate_expression(v, scope)?));
744 }
745 Ok(out)
746}
747
748#[cfg(feature = "ir")]
750fn parse_plan(plan: Option<&J>) -> Option<ExecutionPlanSpec> {
751 let p = plan?;
752 if p.is_null() {
753 return None;
754 }
755 let groups = p
756 .get("groups")?
757 .as_array()?
758 .iter()
759 .map(|g| {
760 g.as_array()
761 .map(|row| {
762 row.iter()
763 .filter_map(|i| i.as_u64().map(|x| x as usize))
764 .collect()
765 })
766 .unwrap_or_default()
767 })
768 .collect();
769 let concurrency = p.get("concurrency").and_then(|c| c.as_i64()).unwrap_or(1);
770 Some(ExecutionPlanSpec {
771 groups,
772 concurrency,
773 })
774}
775
776#[cfg(feature = "ir")]
785pub fn run_behavior(
786 ir: &J,
787 handlers: &mut dyn ComponentExec,
788 input: &[(String, Value)],
789 entry: Option<&str>,
790) -> Result<Value, BehaviorError> {
791 let components = ir
792 .get("components")
793 .and_then(|c| c.as_array())
794 .ok_or_else(|| BehaviorError {
795 code: BehaviorFailureCode::UnknownEntry.as_str().to_string(),
796 message: "IR.components must be an array".into(),
797 detail: None,
798 })?;
799 let comp = match entry {
800 Some(name) => components
801 .iter()
802 .find(|c| c.get("name").and_then(|n| n.as_str()) == Some(name)),
803 None => components.first(),
804 };
805 let comp = match comp {
806 Some(c) => c,
807 None => {
808 return bfail(
809 BehaviorFailureCode::UnknownEntry,
810 format!("component '{}' not found in IR", entry.unwrap_or("<first>")),
811 )
812 }
813 };
814
815 let body: Vec<J> = comp
816 .get("body")
817 .and_then(|b| b.as_array())
818 .cloned()
819 .unwrap_or_default();
820
821 let index_of = |id: &str| {
822 body.iter()
823 .position(|n| n.get("id").and_then(|v| v.as_str()) == Some(id))
824 };
825
826 let mut ops: Vec<OpSpec> = Vec::with_capacity(body.len());
828 for n in &body {
829 let parent = node_parent(n).and_then(index_of);
830 ops.push(OpSpec {
831 id: n
832 .get("id")
833 .and_then(|v| v.as_str())
834 .unwrap_or_default()
835 .to_string(),
836 parent,
837 bind_field: node_bind_field(n),
838 relation_kind: node_relation_kind(n),
839 policy: node_policy(n),
840 });
841 }
842
843 run_behavior_inner(
844 &body,
845 &ops,
846 parse_plan(comp.get("plan")),
847 comp.get("output").cloned().unwrap_or(J::Null),
848 &bind_input_scope(comp, input),
849 comp.get("inputPorts").and_then(|p| p.as_object()),
850 handlers,
851 )
852}
853
854#[cfg(feature = "ir")]
866fn bind_input_scope(comp: &J, input: &[(String, Value)]) -> Vec<(String, Value)> {
867 let mut scope: Vec<(String, Value)> = input.to_vec();
868 if let Some(ports) = comp.get("inputPorts").and_then(|p| p.as_object()) {
869 for (name, schema) in ports {
870 let optional = schema.get("required").and_then(|r| r.as_bool()) == Some(false);
871 if optional && !scope.iter().any(|(k, _)| k == name) {
872 scope.push((name.clone(), Value::Null));
873 }
874 }
875 }
876 scope
877}
878
879#[cfg(feature = "ir")]
883fn run_behavior_inner(
884 body: &[J],
885 ops: &[OpSpec],
886 plan: Option<ExecutionPlanSpec>,
887 output: J,
888 input: &[(String, Value)],
889 input_ports: Option<&serde_json::Map<String, J>>,
890 handlers: &mut dyn ComponentExec,
891) -> Result<Value, BehaviorError> {
892 let index_of = |id: &str| {
893 body.iter()
894 .position(|n| n.get("id").and_then(|v| v.as_str()) == Some(id))
895 };
896
897 let mut results: Vec<(String, Value)> = Vec::new();
898 let mut pending_err: Option<BehaviorError> = None;
899
900 let run = {
901 let results_cell = &mut results;
902 let err_cell = &mut pending_err;
903 let exec = |op: &OpSpec, _bound: Option<&Value>| -> ExecOutcome {
904 if err_cell.is_some() {
905 return ExecOutcome::error("aborted");
906 }
907 let idx = match index_of(&op.id) {
908 Some(i) => i,
909 None => {
910 *err_cell = Some(BehaviorError {
911 code: BehaviorFailureCode::UnknownNodeKind.as_str().to_string(),
912 message: format!("no body node for op '{}'", op.id),
913 detail: None,
914 });
915 return ExecOutcome::error("aborted");
916 }
917 };
918 let node = &body[idx];
919
920 let base_scope = |extra: Option<(&str, &Value)>| -> Vec<(String, Value)> {
921 let mut s: Vec<(String, Value)> = input.to_vec();
922 for (k, v) in results_cell.iter() {
923 s.push((k.clone(), v.clone()));
924 }
925 if let Some((k, v)) = extra {
926 s.push((k.to_string(), v.clone()));
927 }
928 s
929 };
930
931 let outcome = match node_kind(node) {
932 Err(e) => {
933 *err_cell = Some(e);
934 return ExecOutcome::error("aborted");
935 }
936 Ok("cond") => {
937 let c = node.get("cond").unwrap();
938 let cond_expr = serde_json::json!({
939 "cond": [c.get("if"), c.get("then"), c.get("else")]
940 });
941 match evaluate_expression(&cond_expr, &base_scope(None)) {
942 Ok(v) => ExecOutcome::Ok(v),
943 Err(e) => {
944 *err_cell = Some(e.into());
945 return ExecOutcome::error("aborted");
946 }
947 }
948 }
949 Ok("map") => 'map_arm: {
953 let m = node.get("map").unwrap();
954 let over = match evaluate_expression(
955 m.get("over").unwrap_or(&J::Null),
956 &base_scope(None),
957 ) {
958 Ok(v) => v,
959 Err(e) => {
960 *err_cell = Some(e.into());
961 return ExecOutcome::error("aborted");
962 }
963 };
964 let arr = match resolve_over_array(
965 &over,
966 m.get("over").unwrap_or(&J::Null),
967 body,
968 input_ports,
969 BehaviorFailureCode::MapOverNotArray,
970 &op.id,
971 "map",
972 ) {
973 Ok(a) => a,
974 Err(e) => {
975 *err_cell = Some(e);
976 return ExecOutcome::error("aborted");
977 }
978 };
979 let component = m.get("component").and_then(|v| v.as_str()).unwrap_or("");
980 let as_name = m.get("as").and_then(|v| v.as_str()).unwrap_or("$");
981 if m.get("component").is_none() {
985 if let Some(transform) = m.get("transform") {
986 let when_t = m.get("when");
987 let mut out: Vec<Value> = Vec::with_capacity(arr.len());
988 for el in arr.iter() {
989 let scope = base_scope(Some((as_name, el)));
990 if let Some(w) = when_t {
991 let keep_expr = serde_json::json!({ "cond": [w, true, false] });
992 match evaluate_expression(&keep_expr, &scope) {
993 Ok(Value::Bool(true)) => {}
994 Ok(_) => continue,
995 Err(e) => {
996 *err_cell = Some(e.into());
997 return ExecOutcome::error("aborted");
998 }
999 }
1000 }
1001 match evaluate_expression(transform, &scope) {
1002 Ok(v) => out.push(v),
1003 Err(e) => {
1004 *err_cell = Some(e.into());
1005 return ExecOutcome::error("aborted");
1006 }
1007 }
1008 }
1009 break 'map_arm ExecOutcome::Ok(Value::Arr(out));
1010 }
1011 }
1012 let ports_j = m
1013 .get("ports")
1014 .cloned()
1015 .unwrap_or(J::Object(Default::default()));
1016 let when = m.get("when");
1017 let batched = m.get("batched").and_then(|v| v.as_bool()).unwrap_or(false);
1018 let element_policy = {
1022 let raw = m
1023 .get("elementPolicy")
1024 .and_then(|v| v.as_str())
1025 .unwrap_or("error");
1026 match crate::plan::ElementPolicyKind::parse(raw) {
1027 None => {
1028 *err_cell = Some(BehaviorError {
1029 code: "UNKNOWN_ELEMENT_POLICY".to_string(),
1030 message: format!(
1031 "map '{}': unknown element policy '{raw}' (fail-closed)",
1032 op.id
1033 ),
1034 detail: None,
1035 });
1036 return ExecOutcome::error("aborted");
1037 }
1038 Some(crate::plan::ElementPolicyKind::Skip) if batched => {
1039 *err_cell = Some(BehaviorError {
1040 code: "ELEMENT_POLICY_NOT_APPLICABLE".to_string(),
1041 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),
1042 detail: None,
1043 });
1044 return ExecOutcome::error("aborted");
1045 }
1046 Some(p) => p,
1047 }
1048 };
1049 let into = m.get("into").and_then(|v| v.as_str());
1050
1051 let keep = |scope: &[(String, Value)]| -> Result<bool, BehaviorError> {
1054 match when {
1055 None => Ok(true),
1056 Some(w) => {
1057 let cond_expr = serde_json::json!({ "cond": [w, true, false] });
1058 Ok(matches!(
1059 evaluate_expression(&cond_expr, scope)?,
1060 Value::Bool(true)
1061 ))
1062 }
1063 }
1064 };
1065
1066 let mut kept_idx: Vec<usize> = Vec::new(); let collected: Vec<Value>;
1068 if batched {
1069 let mut items: Vec<Value> = Vec::new();
1071 for (i, el) in arr.iter().enumerate() {
1072 let scope = base_scope(Some((as_name, el)));
1073 match keep(&scope) {
1074 Ok(false) => continue,
1075 Ok(true) => {}
1076 Err(e) => {
1077 *err_cell = Some(e);
1078 return ExecOutcome::error("aborted");
1079 }
1080 }
1081 match eval_ports(&ports_j, &scope) {
1082 Ok(p) => items.push(Value::Obj(p)),
1083 Err(e) => {
1084 *err_cell = Some(e);
1085 return ExecOutcome::error("aborted");
1086 }
1087 }
1088 kept_idx.push(i);
1089 }
1090 if items.is_empty() {
1091 collected = Vec::new(); } else {
1093 let want = items.len();
1094 let batch_ports = vec![("items".to_string(), Value::Arr(items))];
1095 match handlers.exec_ctx(&op.id, component, &batch_ports, None) {
1096 None => {
1097 *err_cell = Some(BehaviorError {
1098 code: BehaviorFailureCode::UnknownComponent
1099 .as_str()
1100 .to_string(),
1101 message: format!(
1102 "component '{component}' has no handler (fail-closed)"
1103 ),
1104 detail: None,
1105 });
1106 return ExecOutcome::error("aborted");
1107 }
1108 Some(ExecOutcome::Error(e, d)) => return ExecOutcome::Error(e, d),
1109 Some(ExecOutcome::Ok(Value::Arr(r))) if r.len() == want => {
1110 collected = r;
1111 }
1112 Some(ExecOutcome::Ok(_)) => {
1113 *err_cell = Some(BehaviorError {
1114 code: BehaviorFailureCode::MapBatchResultMismatch
1115 .as_str()
1116 .to_string(),
1117 message: format!(
1118 "map '{}': batched handler must return a list aligned to items (want {want})",
1119 op.id
1120 ),
1121 detail: None,
1122 });
1123 return ExecOutcome::error("aborted");
1124 }
1125 }
1126 }
1127 } else {
1128 let mut out: Vec<Value> = Vec::with_capacity(arr.len());
1129 for (i, el) in arr.iter().enumerate() {
1130 let scope = base_scope(Some((as_name, el)));
1131 match keep(&scope) {
1132 Ok(false) => continue,
1133 Ok(true) => {}
1134 Err(e) => {
1135 *err_cell = Some(e);
1136 return ExecOutcome::error("aborted");
1137 }
1138 }
1139 let ports = match eval_ports(&ports_j, &scope) {
1140 Ok(p) => p,
1141 Err(e) => {
1142 *err_cell = Some(e);
1143 return ExecOutcome::error("aborted");
1144 }
1145 };
1146 match handlers.exec_ctx(&op.id, component, &ports, Some(el)) {
1147 None => {
1148 *err_cell = Some(BehaviorError {
1149 code: BehaviorFailureCode::UnknownComponent
1150 .as_str()
1151 .to_string(),
1152 message: format!(
1153 "component '{component}' has no handler (fail-closed)"
1154 ),
1155 detail: None,
1156 });
1157 return ExecOutcome::error("aborted");
1158 }
1159 Some(ExecOutcome::Error(e, d)) => {
1163 if element_policy == crate::plan::ElementPolicyKind::Skip {
1164 continue;
1165 }
1166 return ExecOutcome::Error(e, d);
1167 }
1168 Some(ExecOutcome::Ok(v)) => out.push(v),
1169 }
1170 kept_idx.push(i);
1171 }
1172 collected = out;
1173 }
1174
1175 match into {
1176 None => ExecOutcome::Ok(Value::Arr(collected)),
1177 Some(key) => {
1178 let node_rk = node_relation_kind(node);
1183 let mut augmented: Vec<Value> = Vec::with_capacity(arr.len());
1184 let mut k = 0usize;
1185 for (i, el) in arr.iter().enumerate() {
1186 let into_val: Option<Value> =
1187 if k < kept_idx.len() && kept_idx[k] == i {
1188 let v = collected[k].clone();
1189 k += 1;
1190 Some(v)
1191 } else if node_rk == Some(RelationKind::Connection) {
1192 Some(unproduced_value(node_rk))
1193 } else {
1194 None
1195 };
1196 let val = match into_val {
1197 None => {
1198 augmented.push(el.clone());
1199 continue;
1200 }
1201 Some(v) => v,
1202 };
1203 let mut fields = match el {
1204 Value::Obj(f) => f.clone(),
1205 _ => {
1206 *err_cell = Some(BehaviorError {
1207 code: BehaviorFailureCode::MapIntoElementNotObject
1208 .as_str()
1209 .to_string(),
1210 message: format!(
1211 "map '{}': 'into' requires object elements (element {i} is not an object)",
1212 op.id
1213 ),
1214 detail: None,
1215 });
1216 return ExecOutcome::error("aborted");
1217 }
1218 };
1219 match fields.iter_mut().find(|(fk, _)| fk == key) {
1220 Some(slot) => slot.1 = val,
1221 None => fields.push((key.to_string(), val)),
1222 }
1223 augmented.push(Value::Obj(fields));
1224 }
1225 ExecOutcome::Ok(Value::Arr(augmented))
1226 }
1227 }
1228 }
1229 Ok("fanout") => {
1230 let f = node.get("fanout").unwrap();
1234 let over = match evaluate_expression(
1235 f.get("over").unwrap_or(&J::Null),
1236 &base_scope(None),
1237 ) {
1238 Ok(v) => v,
1239 Err(e) => {
1240 *err_cell = Some(e.into());
1241 return ExecOutcome::error("aborted");
1242 }
1243 };
1244 let arr = match resolve_over_array(
1245 &over,
1246 f.get("over").unwrap_or(&J::Null),
1247 body,
1248 input_ports,
1249 BehaviorFailureCode::FanoutOverNotArray,
1250 &op.id,
1251 "fanout",
1252 ) {
1253 Ok(a) => a,
1254 Err(e) => {
1255 *err_cell = Some(e);
1256 return ExecOutcome::error("aborted");
1257 }
1258 };
1259 let component = f.get("component").and_then(|v| v.as_str()).unwrap_or("");
1260 let as_name = f.get("as").and_then(|v| v.as_str()).unwrap_or("$");
1261 let ports_j = f
1262 .get("ports")
1263 .cloned()
1264 .unwrap_or(J::Object(Default::default()));
1265 let dedupe_key = f.get("dedupeKey").and_then(|v| v.as_str()).unwrap_or("");
1266 let drop = f.get("drop").and_then(|v| v.as_str()).unwrap_or("dangling");
1267 let implicit_source = f.get("implicitSource").and_then(|v| v.as_str());
1268 let mut items: Vec<Value> = Vec::with_capacity(arr.len());
1269 for el in arr.iter() {
1270 let scope = base_scope(Some((as_name, el)));
1271 match eval_ports(&ports_j, &scope) {
1272 Ok(p) => items.push(Value::Obj(p)),
1273 Err(e) => {
1274 *err_cell = Some(e);
1275 return ExecOutcome::error("aborted");
1276 }
1277 }
1278 }
1279 if items.is_empty() {
1280 ExecOutcome::Ok(Value::Obj(vec![
1281 ("items".into(), Value::Arr(vec![])),
1282 ("cursor".into(), Value::Null),
1283 ]))
1284 } else {
1285 let want = items.len();
1286 let batch_ports = vec![("items".to_string(), Value::Arr(items))];
1287 match handlers.exec_ctx(&op.id, component, &batch_ports, None) {
1288 None => {
1289 *err_cell = Some(BehaviorError {
1290 code: BehaviorFailureCode::UnknownComponent
1291 .as_str()
1292 .to_string(),
1293 message: format!(
1294 "component '{component}' has no handler (fail-closed)"
1295 ),
1296 detail: None,
1297 });
1298 ExecOutcome::error("aborted")
1299 }
1300 Some(ExecOutcome::Error(e, d)) => ExecOutcome::Error(e, d),
1301 Some(ExecOutcome::Ok(Value::Arr(r))) if r.len() == want => {
1302 let deduped =
1303 fanout_dedup_drop(r, dedupe_key, drop, implicit_source);
1304 ExecOutcome::Ok(Value::Obj(vec![
1305 ("items".into(), Value::Arr(deduped)),
1306 ("cursor".into(), Value::Null),
1307 ]))
1308 }
1309 Some(ExecOutcome::Ok(_)) => {
1310 *err_cell = Some(BehaviorError {
1311 code: BehaviorFailureCode::FanoutBatchResultMismatch
1312 .as_str()
1313 .to_string(),
1314 message: format!(
1315 "fanout '{}': batched handler must return a list aligned to the deduped id list (want {want})",
1316 op.id
1317 ),
1318 detail: None,
1319 });
1320 ExecOutcome::error("aborted")
1321 }
1322 }
1323 }
1324 }
1325 Ok(_) => {
1326 let component = node.get("component").and_then(|v| v.as_str()).unwrap_or("");
1327 let ports_j = node
1328 .get("ports")
1329 .cloned()
1330 .unwrap_or(J::Object(Default::default()));
1331 let ports = match eval_ports(&ports_j, &base_scope(None)) {
1332 Ok(p) => p,
1333 Err(e) => {
1334 *err_cell = Some(e);
1335 return ExecOutcome::error("aborted");
1336 }
1337 };
1338 match handlers.exec_ctx(&op.id, component, &ports, None) {
1339 None => {
1340 *err_cell = Some(BehaviorError {
1341 code: BehaviorFailureCode::UnknownComponent.as_str().to_string(),
1342 message: format!(
1343 "component '{component}' has no handler (fail-closed)"
1344 ),
1345 detail: None,
1346 });
1347 return ExecOutcome::error("aborted");
1348 }
1349 Some(o) => o,
1350 }
1351 }
1352 };
1353
1354 if let ExecOutcome::Ok(v) = &outcome {
1355 let mut value = v.clone();
1360 if let Some(ot) = node.get("outType") {
1361 let want = result_type_of(node, ot);
1362 match assert_conforms_to_out_type(&op.id, &value, &want, "result") {
1363 Ok(normalized) => value = normalized,
1364 Err(e) => {
1365 *err_cell = Some(e.into());
1366 return ExecOutcome::error("aborted");
1367 }
1368 }
1369 }
1370 results_cell.push((op.id.clone(), value));
1371 }
1372 outcome
1373 };
1374
1375 run_plan(plan.as_ref(), ops, exec)
1376 };
1377
1378 if let Some(e) = pending_err {
1381 return Err(e);
1382 }
1383 let run = run?;
1384
1385 for (i, id_val) in run.skipped.iter().enumerate() {
1388 let _ = i;
1389 if let Some(idx) = index_of(id_val) {
1390 let unproduced = unproduced_value(node_relation_kind(&body[idx]));
1391 if scope_get(&results, id_val).is_none() {
1392 results.push((id_val.clone(), unproduced));
1393 }
1394 }
1395 }
1396
1397 let scope: Vec<(String, Value)> = {
1398 let mut s: Vec<(String, Value)> = input.to_vec();
1399 s.extend(results.iter().cloned());
1400 s
1401 };
1402 Ok(evaluate_expression(&output, &scope)?)
1403}