1#[cfg(feature = "ir")]
34use crate::expr::evaluate as evaluate_expression;
35use crate::expr::ExprFailure;
36#[cfg(feature = "ir")]
37use crate::plan::{run_plan, 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")]
195fn conform_fail(
196 code: crate::expr::ExprFailureCode,
197 message: String,
198 kind: crate::plan::ErrorKind,
199 node_id: &str,
200 field: &str,
201 expected: &J,
202 actual: Option<&Value>,
203) -> ExprFailure {
204 let mut detail = crate::plan::ErrorDetail {
205 kind: Some(kind),
206 model: Some(node_id.to_string()),
207 field: Some(field.to_string()),
208 expected_type: Some(portable_type_notation(expected)),
209 ..Default::default()
210 };
211 if let Some(a) = actual {
212 detail.actual_wire_type = Some(wire_type_name(a).to_string());
213 detail.raw_value = Some(raw_value_of(a));
214 }
215 ExprFailure {
216 code,
217 message,
218 detail: Some(Box::new(detail)),
219 }
220}
221
222#[cfg(feature = "ir")]
227fn assert_conforms_to_out_type(
228 node_id: &str,
229 v: &Value,
230 t: &J,
231 field: &str,
232) -> Result<(), ExprFailure> {
233 use crate::expr::ExprFailureCode::{MissingProp, TypeMismatch};
234 use crate::plan::ErrorKind;
235 if let Some(scalar) = t.as_str() {
236 let ok = match scalar {
237 "string" => matches!(v, Value::Str(_)),
238 "int" => matches!(v, Value::Int(_)),
239 "float" => matches!(v, Value::Float(_) | Value::Int(_)), "bool" => matches!(v, Value::Bool(_)),
241 "null" => matches!(v, Value::Null),
242 _ => true,
243 };
244 if !ok {
245 return Err(conform_fail(
246 TypeMismatch,
247 format!(
248 "node '{node_id}': {field}: expected {scalar}, got {}",
249 wire_type_name(v)
250 ),
251 ErrorKind::TypeMismatch,
252 node_id,
253 field,
254 t,
255 Some(v),
256 ));
257 }
258 return Ok(());
259 }
260 if let Some(inner) = t.get("opt") {
261 if matches!(v, Value::Null) {
262 return Ok(());
263 }
264 return assert_conforms_to_out_type(node_id, v, inner, field);
265 }
266 if let Some(inner) = t.get("arr") {
267 match v {
268 Value::Arr(items) => {
269 for (i, el) in items.iter().enumerate() {
270 assert_conforms_to_out_type(node_id, el, inner, &format!("{field}[{i}]"))?;
271 }
272 return Ok(());
273 }
274 _ => {
275 return Err(conform_fail(
276 TypeMismatch,
277 format!(
278 "node '{node_id}': {field}: expected arr, got {}",
279 wire_type_name(v)
280 ),
281 ErrorKind::TypeMismatch,
282 node_id,
283 field,
284 t,
285 Some(v),
286 ))
287 }
288 }
289 }
290 if let Some(inner) = t.get("map") {
291 match v {
292 Value::Obj(pairs) => {
293 for (k, mv) in pairs {
294 assert_conforms_to_out_type(node_id, mv, inner, &format!("{field}.{k}"))?;
295 }
296 return Ok(());
297 }
298 _ => {
299 return Err(conform_fail(
300 TypeMismatch,
301 format!(
302 "node '{node_id}': {field}: expected map, got {}",
303 wire_type_name(v)
304 ),
305 ErrorKind::TypeMismatch,
306 node_id,
307 field,
308 t,
309 Some(v),
310 ))
311 }
312 }
313 }
314 if let Some(J::Object(fields)) = t.get("obj") {
316 let Value::Obj(_) = v else {
317 return Err(conform_fail(
318 TypeMismatch,
319 format!(
320 "node '{node_id}': {field}: expected obj, got {}",
321 wire_type_name(v)
322 ),
323 ErrorKind::TypeMismatch,
324 node_id,
325 field,
326 t,
327 Some(v),
328 ));
329 };
330 for (k, ft) in fields {
331 match v.obj_get(k) {
332 None => {
333 return Err(conform_fail(
334 MissingProp,
335 format!("node '{node_id}': {field}: missing property .{k}"),
336 ErrorKind::MissingField,
337 node_id,
338 &format!("{field}.{k}"),
339 ft,
340 None,
341 ))
342 }
343 Some(fv) => assert_conforms_to_out_type(node_id, fv, ft, &format!("{field}.{k}"))?,
344 }
345 }
346 }
347 Ok(())
348}
349
350pub trait ComponentExec {
355 fn exec(
356 &mut self,
357 component: &str,
358 ports: &[(String, Value)],
359 bound: Option<&Value>,
360 ) -> Option<ExecOutcome>;
361
362 fn exec_ctx(
368 &mut self,
369 node_id: &str,
370 component: &str,
371 ports: &[(String, Value)],
372 bound: Option<&Value>,
373 ) -> Option<ExecOutcome> {
374 let _ = node_id;
375 self.exec(component, ports, bound)
376 }
377}
378
379impl ComponentExec for &mut dyn ComponentExec {
396 fn exec(
397 &mut self,
398 component: &str,
399 ports: &[(String, Value)],
400 bound: Option<&Value>,
401 ) -> Option<ExecOutcome> {
402 (**self).exec(component, ports, bound)
403 }
404
405 fn exec_ctx(
406 &mut self,
407 node_id: &str,
408 component: &str,
409 ports: &[(String, Value)],
410 bound: Option<&Value>,
411 ) -> Option<ExecOutcome> {
412 (**self).exec_ctx(node_id, component, ports, bound)
413 }
414}
415
416#[cfg(feature = "ir")]
417fn scope_get<'a>(scope: &'a [(String, Value)], key: &str) -> Option<&'a Value> {
418 scope.iter().find(|(k, _)| k == key).map(|(_, v)| v)
419}
420
421#[cfg(feature = "ir")]
427fn fanout_dedup_drop(
428 aligned_bodies: Vec<Value>,
429 dedupe_key: &str,
430 drop: &str,
431 implicit_source: Option<&str>,
432) -> Vec<Value> {
433 use std::collections::HashSet;
434 let mut items: Vec<Value> = Vec::with_capacity(aligned_bodies.len());
435 let mut seen: HashSet<String> = HashSet::new();
436 for body in aligned_bodies.into_iter() {
437 let fields = match &body {
438 Value::Obj(f) => Some(f),
439 _ => None,
440 };
441 let key_val = fields.and_then(|f| f.iter().find(|(k, _)| k == dedupe_key).map(|(_, v)| v));
442 let has_key = matches!(key_val, Some(v) if !matches!(v, Value::Null));
443 if !has_key {
444 if drop == "dangling" {
445 continue;
446 }
447 items.push(body); continue;
449 }
450 let seen_key = match key_val.unwrap() {
451 Value::Str(s) => format!("s:{s}"),
452 other => format!("j:{other:?}"),
453 };
454 if seen.contains(&seen_key) {
455 continue;
456 }
457 seen.insert(seen_key);
458 match (implicit_source, fields) {
459 (Some(src), Some(f)) if f.iter().any(|(k, _)| k == src) => {
460 let stripped: Vec<(String, Value)> =
461 f.iter().filter(|(k, _)| k != src).cloned().collect();
462 items.push(Value::Obj(stripped));
463 }
464 _ => items.push(body),
465 }
466 }
467 items
468}
469
470#[cfg(feature = "ir")]
471fn node_kind(n: &J) -> Result<&'static str, BehaviorError> {
472 if n.get("fanout").is_some() {
473 Ok("fanout")
474 } else if n.get("map").is_some() {
475 Ok("map")
476 } else if n.get("cond").is_some() {
477 Ok("cond")
478 } else if n.get("component").is_some() {
479 Ok("componentRef")
480 } else {
481 bfail(
482 BehaviorFailureCode::UnknownNodeKind,
483 format!(
484 "body node '{}' is not componentRef/map/cond/fanout",
485 n.get("id").and_then(|v| v.as_str()).unwrap_or("?")
486 ),
487 )
488 }
489}
490
491#[cfg(feature = "ir")]
492fn node_sub(n: &J) -> &J {
493 if let Some(f) = n.get("fanout") {
494 f
495 } else if let Some(m) = n.get("map") {
496 m
497 } else if let Some(c) = n.get("cond") {
498 c
499 } else {
500 n
501 }
502}
503
504#[cfg(feature = "ir")]
505fn node_parent(n: &J) -> Option<&str> {
506 node_sub(n).get("parent").and_then(|v| v.as_str())
507}
508
509#[cfg(feature = "ir")]
510fn node_bind_field(n: &J) -> Option<String> {
511 if n.get("map").is_some() || n.get("cond").is_some() || n.get("fanout").is_some() {
512 return None;
513 }
514 n.get("bindField")
515 .and_then(|v| v.as_str())
516 .map(str::to_string)
517}
518
519#[cfg(feature = "ir")]
520fn node_relation_kind(n: &J) -> Option<RelationKind> {
521 if n.get("cond").is_some() {
522 return None;
523 }
524 let sub = if n.get("map").is_some() || n.get("fanout").is_some() {
525 node_sub(n)
526 } else {
527 n
528 };
529 match sub.get("relationKind").and_then(|v| v.as_str()) {
530 Some("connection") => Some(RelationKind::Connection),
531 Some(_) => Some(RelationKind::Single),
532 None => None,
533 }
534}
535
536#[cfg(feature = "ir")]
537fn node_relation_kind_str(n: &J) -> Option<&str> {
538 if n.get("cond").is_some() {
539 return None;
540 }
541 let sub = if n.get("map").is_some() || n.get("fanout").is_some() {
542 node_sub(n)
543 } else {
544 n
545 };
546 sub.get("relationKind").and_then(|v| v.as_str())
547}
548
549#[cfg(feature = "ir")]
550fn node_policy(n: &J) -> Option<String> {
551 if n.get("cond").is_some() {
552 return None;
553 }
554 let sub = if n.get("map").is_some() || n.get("fanout").is_some() {
555 node_sub(n)
556 } else {
557 n
558 };
559 sub.get("policy")
560 .and_then(|v| v.as_str())
561 .map(str::to_string)
562}
563
564#[cfg(feature = "ir")]
565fn eval_ports(ports: &J, scope: &[(String, Value)]) -> Result<Vec<(String, Value)>, BehaviorError> {
566 let obj = ports.as_object().ok_or_else(|| BehaviorError {
567 code: BehaviorFailureCode::UnknownNodeKind.as_str().to_string(),
568 message: "ports must be an object".into(),
569 detail: None,
570 })?;
571 let mut out = Vec::with_capacity(obj.len());
572 for (k, v) in obj {
573 out.push((k.clone(), evaluate_expression(v, scope)?));
574 }
575 Ok(out)
576}
577
578#[cfg(feature = "ir")]
580fn parse_plan(plan: Option<&J>) -> Option<ExecutionPlanSpec> {
581 let p = plan?;
582 if p.is_null() {
583 return None;
584 }
585 let groups = p
586 .get("groups")?
587 .as_array()?
588 .iter()
589 .map(|g| {
590 g.as_array()
591 .map(|row| {
592 row.iter()
593 .filter_map(|i| i.as_u64().map(|x| x as usize))
594 .collect()
595 })
596 .unwrap_or_default()
597 })
598 .collect();
599 let concurrency = p.get("concurrency").and_then(|c| c.as_i64()).unwrap_or(1);
600 Some(ExecutionPlanSpec {
601 groups,
602 concurrency,
603 })
604}
605
606#[cfg(feature = "ir")]
615pub fn run_behavior(
616 ir: &J,
617 handlers: &mut dyn ComponentExec,
618 input: &[(String, Value)],
619 entry: Option<&str>,
620) -> Result<Value, BehaviorError> {
621 let components = ir
622 .get("components")
623 .and_then(|c| c.as_array())
624 .ok_or_else(|| BehaviorError {
625 code: BehaviorFailureCode::UnknownEntry.as_str().to_string(),
626 message: "IR.components must be an array".into(),
627 detail: None,
628 })?;
629 let comp = match entry {
630 Some(name) => components
631 .iter()
632 .find(|c| c.get("name").and_then(|n| n.as_str()) == Some(name)),
633 None => components.first(),
634 };
635 let comp = match comp {
636 Some(c) => c,
637 None => {
638 return bfail(
639 BehaviorFailureCode::UnknownEntry,
640 format!("component '{}' not found in IR", entry.unwrap_or("<first>")),
641 )
642 }
643 };
644
645 let body: Vec<J> = comp
646 .get("body")
647 .and_then(|b| b.as_array())
648 .cloned()
649 .unwrap_or_default();
650
651 let index_of = |id: &str| {
652 body.iter()
653 .position(|n| n.get("id").and_then(|v| v.as_str()) == Some(id))
654 };
655
656 let mut ops: Vec<OpSpec> = Vec::with_capacity(body.len());
658 for n in &body {
659 let parent = node_parent(n).and_then(index_of);
660 ops.push(OpSpec {
661 id: n
662 .get("id")
663 .and_then(|v| v.as_str())
664 .unwrap_or_default()
665 .to_string(),
666 parent,
667 bind_field: node_bind_field(n),
668 relation_kind: node_relation_kind(n),
669 policy: node_policy(n),
670 });
671 }
672
673 run_behavior_inner(
674 &body,
675 &ops,
676 parse_plan(comp.get("plan")),
677 comp.get("output").cloned().unwrap_or(J::Null),
678 &bind_input_scope(comp, input),
679 handlers,
680 )
681}
682
683#[cfg(feature = "ir")]
695fn bind_input_scope(comp: &J, input: &[(String, Value)]) -> Vec<(String, Value)> {
696 let mut scope: Vec<(String, Value)> = input.to_vec();
697 if let Some(ports) = comp.get("inputPorts").and_then(|p| p.as_object()) {
698 for (name, schema) in ports {
699 let optional = schema.get("required").and_then(|r| r.as_bool()) == Some(false);
700 if optional && !scope.iter().any(|(k, _)| k == name) {
701 scope.push((name.clone(), Value::Null));
702 }
703 }
704 }
705 scope
706}
707
708#[cfg(feature = "ir")]
712fn run_behavior_inner(
713 body: &[J],
714 ops: &[OpSpec],
715 plan: Option<ExecutionPlanSpec>,
716 output: J,
717 input: &[(String, Value)],
718 handlers: &mut dyn ComponentExec,
719) -> Result<Value, BehaviorError> {
720 let index_of = |id: &str| {
721 body.iter()
722 .position(|n| n.get("id").and_then(|v| v.as_str()) == Some(id))
723 };
724
725 let mut results: Vec<(String, Value)> = Vec::new();
726 let mut pending_err: Option<BehaviorError> = None;
727
728 let run = {
729 let results_cell = &mut results;
730 let err_cell = &mut pending_err;
731 let exec = |op: &OpSpec, _bound: Option<&Value>| -> ExecOutcome {
732 if err_cell.is_some() {
733 return ExecOutcome::error("aborted");
734 }
735 let idx = match index_of(&op.id) {
736 Some(i) => i,
737 None => {
738 *err_cell = Some(BehaviorError {
739 code: BehaviorFailureCode::UnknownNodeKind.as_str().to_string(),
740 message: format!("no body node for op '{}'", op.id),
741 detail: None,
742 });
743 return ExecOutcome::error("aborted");
744 }
745 };
746 let node = &body[idx];
747
748 let base_scope = |extra: Option<(&str, &Value)>| -> Vec<(String, Value)> {
749 let mut s: Vec<(String, Value)> = input.to_vec();
750 for (k, v) in results_cell.iter() {
751 s.push((k.clone(), v.clone()));
752 }
753 if let Some((k, v)) = extra {
754 s.push((k.to_string(), v.clone()));
755 }
756 s
757 };
758
759 let outcome = match node_kind(node) {
760 Err(e) => {
761 *err_cell = Some(e);
762 return ExecOutcome::error("aborted");
763 }
764 Ok("cond") => {
765 let c = node.get("cond").unwrap();
766 let cond_expr = serde_json::json!({
767 "cond": [c.get("if"), c.get("then"), c.get("else")]
768 });
769 match evaluate_expression(&cond_expr, &base_scope(None)) {
770 Ok(v) => ExecOutcome::Ok(v),
771 Err(e) => {
772 *err_cell = Some(e.into());
773 return ExecOutcome::error("aborted");
774 }
775 }
776 }
777 Ok("map") => {
778 let m = node.get("map").unwrap();
779 let over = match evaluate_expression(
780 m.get("over").unwrap_or(&J::Null),
781 &base_scope(None),
782 ) {
783 Ok(v) => v,
784 Err(e) => {
785 *err_cell = Some(e.into());
786 return ExecOutcome::error("aborted");
787 }
788 };
789 let arr = match &over {
790 Value::Arr(a) => a.clone(),
791 _ => {
792 *err_cell = Some(BehaviorError {
793 code: BehaviorFailureCode::MapOverNotArray.as_str().to_string(),
794 message: format!("map '{}': 'over' is not an array", op.id),
795 detail: None,
796 });
797 return ExecOutcome::error("aborted");
798 }
799 };
800 let component = m.get("component").and_then(|v| v.as_str()).unwrap_or("");
801 let as_name = m.get("as").and_then(|v| v.as_str()).unwrap_or("$");
802 let ports_j = m
803 .get("ports")
804 .cloned()
805 .unwrap_or(J::Object(Default::default()));
806 let when = m.get("when");
807 let batched = m.get("batched").and_then(|v| v.as_bool()).unwrap_or(false);
808 let element_policy = {
812 let raw = m
813 .get("elementPolicy")
814 .and_then(|v| v.as_str())
815 .unwrap_or("error");
816 match crate::plan::ElementPolicyKind::parse(raw) {
817 None => {
818 *err_cell = Some(BehaviorError {
819 code: "UNKNOWN_ELEMENT_POLICY".to_string(),
820 message: format!(
821 "map '{}': unknown element policy '{raw}' (fail-closed)",
822 op.id
823 ),
824 detail: None,
825 });
826 return ExecOutcome::error("aborted");
827 }
828 Some(crate::plan::ElementPolicyKind::Skip) if batched => {
829 *err_cell = Some(BehaviorError {
830 code: "ELEMENT_POLICY_NOT_APPLICABLE".to_string(),
831 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),
832 detail: None,
833 });
834 return ExecOutcome::error("aborted");
835 }
836 Some(p) => p,
837 }
838 };
839 let into = m.get("into").and_then(|v| v.as_str());
840
841 let keep = |scope: &[(String, Value)]| -> Result<bool, BehaviorError> {
844 match when {
845 None => Ok(true),
846 Some(w) => {
847 let cond_expr = serde_json::json!({ "cond": [w, true, false] });
848 Ok(matches!(
849 evaluate_expression(&cond_expr, scope)?,
850 Value::Bool(true)
851 ))
852 }
853 }
854 };
855
856 let mut kept_idx: Vec<usize> = Vec::new(); let collected: Vec<Value>;
858 if batched {
859 let mut items: Vec<Value> = Vec::new();
861 for (i, el) in arr.iter().enumerate() {
862 let scope = base_scope(Some((as_name, el)));
863 match keep(&scope) {
864 Ok(false) => continue,
865 Ok(true) => {}
866 Err(e) => {
867 *err_cell = Some(e);
868 return ExecOutcome::error("aborted");
869 }
870 }
871 match eval_ports(&ports_j, &scope) {
872 Ok(p) => items.push(Value::Obj(p)),
873 Err(e) => {
874 *err_cell = Some(e);
875 return ExecOutcome::error("aborted");
876 }
877 }
878 kept_idx.push(i);
879 }
880 if items.is_empty() {
881 collected = Vec::new(); } else {
883 let want = items.len();
884 let batch_ports = vec![("items".to_string(), Value::Arr(items))];
885 match handlers.exec_ctx(&op.id, component, &batch_ports, None) {
886 None => {
887 *err_cell = Some(BehaviorError {
888 code: BehaviorFailureCode::UnknownComponent
889 .as_str()
890 .to_string(),
891 message: format!(
892 "component '{component}' has no handler (fail-closed)"
893 ),
894 detail: None,
895 });
896 return ExecOutcome::error("aborted");
897 }
898 Some(ExecOutcome::Error(e, d)) => return ExecOutcome::Error(e, d),
899 Some(ExecOutcome::Ok(Value::Arr(r))) if r.len() == want => {
900 collected = r;
901 }
902 Some(ExecOutcome::Ok(_)) => {
903 *err_cell = Some(BehaviorError {
904 code: BehaviorFailureCode::MapBatchResultMismatch
905 .as_str()
906 .to_string(),
907 message: format!(
908 "map '{}': batched handler must return a list aligned to items (want {want})",
909 op.id
910 ),
911 detail: None,
912 });
913 return ExecOutcome::error("aborted");
914 }
915 }
916 }
917 } else {
918 let mut out: Vec<Value> = Vec::with_capacity(arr.len());
919 for (i, el) in arr.iter().enumerate() {
920 let scope = base_scope(Some((as_name, el)));
921 match keep(&scope) {
922 Ok(false) => continue,
923 Ok(true) => {}
924 Err(e) => {
925 *err_cell = Some(e);
926 return ExecOutcome::error("aborted");
927 }
928 }
929 let ports = match eval_ports(&ports_j, &scope) {
930 Ok(p) => p,
931 Err(e) => {
932 *err_cell = Some(e);
933 return ExecOutcome::error("aborted");
934 }
935 };
936 match handlers.exec_ctx(&op.id, component, &ports, Some(el)) {
937 None => {
938 *err_cell = Some(BehaviorError {
939 code: BehaviorFailureCode::UnknownComponent
940 .as_str()
941 .to_string(),
942 message: format!(
943 "component '{component}' has no handler (fail-closed)"
944 ),
945 detail: None,
946 });
947 return ExecOutcome::error("aborted");
948 }
949 Some(ExecOutcome::Error(e, d)) => {
953 if element_policy == crate::plan::ElementPolicyKind::Skip {
954 continue;
955 }
956 return ExecOutcome::Error(e, d);
957 }
958 Some(ExecOutcome::Ok(v)) => out.push(v),
959 }
960 kept_idx.push(i);
961 }
962 collected = out;
963 }
964
965 match into {
966 None => ExecOutcome::Ok(Value::Arr(collected)),
967 Some(key) => {
968 let mut augmented: Vec<Value> = Vec::with_capacity(arr.len());
971 let mut k = 0usize;
972 for (i, el) in arr.iter().enumerate() {
973 if k < kept_idx.len() && kept_idx[k] == i {
974 let fields = match el {
975 Value::Obj(f) => f.clone(),
976 _ => {
977 *err_cell = Some(BehaviorError {
978 code: BehaviorFailureCode::MapIntoElementNotObject
979 .as_str()
980 .to_string(),
981 message: format!(
982 "map '{}': 'into' requires object elements (element {i} is not an object)",
983 op.id
984 ),
985 detail: None,
986 });
987 return ExecOutcome::error("aborted");
988 }
989 };
990 let mut fields = fields;
991 let val = collected[k].clone();
992 match fields.iter_mut().find(|(fk, _)| fk == key) {
993 Some(slot) => slot.1 = val,
994 None => fields.push((key.to_string(), val)),
995 }
996 augmented.push(Value::Obj(fields));
997 k += 1;
998 } else {
999 augmented.push(el.clone());
1000 }
1001 }
1002 ExecOutcome::Ok(Value::Arr(augmented))
1003 }
1004 }
1005 }
1006 Ok("fanout") => {
1007 let f = node.get("fanout").unwrap();
1011 let over = match evaluate_expression(
1012 f.get("over").unwrap_or(&J::Null),
1013 &base_scope(None),
1014 ) {
1015 Ok(v) => v,
1016 Err(e) => {
1017 *err_cell = Some(e.into());
1018 return ExecOutcome::error("aborted");
1019 }
1020 };
1021 let arr = match &over {
1022 Value::Arr(a) => a.clone(),
1023 _ => {
1024 *err_cell = Some(BehaviorError {
1025 code: BehaviorFailureCode::FanoutOverNotArray.as_str().to_string(),
1026 message: format!("fanout '{}': 'over' is not an array", op.id),
1027 detail: None,
1028 });
1029 return ExecOutcome::error("aborted");
1030 }
1031 };
1032 let component = f.get("component").and_then(|v| v.as_str()).unwrap_or("");
1033 let as_name = f.get("as").and_then(|v| v.as_str()).unwrap_or("$");
1034 let ports_j = f
1035 .get("ports")
1036 .cloned()
1037 .unwrap_or(J::Object(Default::default()));
1038 let dedupe_key = f.get("dedupeKey").and_then(|v| v.as_str()).unwrap_or("");
1039 let drop = f.get("drop").and_then(|v| v.as_str()).unwrap_or("dangling");
1040 let implicit_source = f.get("implicitSource").and_then(|v| v.as_str());
1041 let mut items: Vec<Value> = Vec::with_capacity(arr.len());
1042 for el in arr.iter() {
1043 let scope = base_scope(Some((as_name, el)));
1044 match eval_ports(&ports_j, &scope) {
1045 Ok(p) => items.push(Value::Obj(p)),
1046 Err(e) => {
1047 *err_cell = Some(e);
1048 return ExecOutcome::error("aborted");
1049 }
1050 }
1051 }
1052 if items.is_empty() {
1053 ExecOutcome::Ok(Value::Obj(vec![
1054 ("items".into(), Value::Arr(vec![])),
1055 ("cursor".into(), Value::Null),
1056 ]))
1057 } else {
1058 let want = items.len();
1059 let batch_ports = vec![("items".to_string(), Value::Arr(items))];
1060 match handlers.exec_ctx(&op.id, component, &batch_ports, None) {
1061 None => {
1062 *err_cell = Some(BehaviorError {
1063 code: BehaviorFailureCode::UnknownComponent
1064 .as_str()
1065 .to_string(),
1066 message: format!(
1067 "component '{component}' has no handler (fail-closed)"
1068 ),
1069 detail: None,
1070 });
1071 ExecOutcome::error("aborted")
1072 }
1073 Some(ExecOutcome::Error(e, d)) => ExecOutcome::Error(e, d),
1074 Some(ExecOutcome::Ok(Value::Arr(r))) if r.len() == want => {
1075 let deduped =
1076 fanout_dedup_drop(r, dedupe_key, drop, implicit_source);
1077 ExecOutcome::Ok(Value::Obj(vec![
1078 ("items".into(), Value::Arr(deduped)),
1079 ("cursor".into(), Value::Null),
1080 ]))
1081 }
1082 Some(ExecOutcome::Ok(_)) => {
1083 *err_cell = Some(BehaviorError {
1084 code: BehaviorFailureCode::FanoutBatchResultMismatch
1085 .as_str()
1086 .to_string(),
1087 message: format!(
1088 "fanout '{}': batched handler must return a list aligned to the deduped id list (want {want})",
1089 op.id
1090 ),
1091 detail: None,
1092 });
1093 ExecOutcome::error("aborted")
1094 }
1095 }
1096 }
1097 }
1098 Ok(_) => {
1099 let component = node.get("component").and_then(|v| v.as_str()).unwrap_or("");
1100 let ports_j = node
1101 .get("ports")
1102 .cloned()
1103 .unwrap_or(J::Object(Default::default()));
1104 let ports = match eval_ports(&ports_j, &base_scope(None)) {
1105 Ok(p) => p,
1106 Err(e) => {
1107 *err_cell = Some(e);
1108 return ExecOutcome::error("aborted");
1109 }
1110 };
1111 match handlers.exec_ctx(&op.id, component, &ports, None) {
1112 None => {
1113 *err_cell = Some(BehaviorError {
1114 code: BehaviorFailureCode::UnknownComponent.as_str().to_string(),
1115 message: format!(
1116 "component '{component}' has no handler (fail-closed)"
1117 ),
1118 detail: None,
1119 });
1120 return ExecOutcome::error("aborted");
1121 }
1122 Some(o) => o,
1123 }
1124 }
1125 };
1126
1127 if let ExecOutcome::Ok(v) = &outcome {
1128 if let Some(ot) = node.get("outType") {
1132 let want = result_type_of(node, ot);
1133 if let Err(e) = assert_conforms_to_out_type(&op.id, v, &want, "result") {
1134 *err_cell = Some(e.into());
1135 return ExecOutcome::error("aborted");
1136 }
1137 }
1138 results_cell.push((op.id.clone(), v.clone()));
1139 }
1140 outcome
1141 };
1142
1143 run_plan(plan.as_ref(), ops, exec)
1144 };
1145
1146 if let Some(e) = pending_err {
1149 return Err(e);
1150 }
1151 let run = run?;
1152
1153 for (i, id_val) in run.skipped.iter().enumerate() {
1156 let _ = i;
1157 if let Some(idx) = index_of(id_val) {
1158 let rk = node_relation_kind_str(&body[idx]);
1159 let unproduced = if rk == Some("connection") {
1160 Value::Obj(vec![
1161 ("items".into(), Value::Arr(vec![])),
1162 ("cursor".into(), Value::Null),
1163 ])
1164 } else {
1165 Value::Null
1166 };
1167 if scope_get(&results, id_val).is_none() {
1168 results.push((id_val.clone(), unproduced));
1169 }
1170 }
1171 }
1172
1173 let scope: Vec<(String, Value)> = {
1174 let mut s: Vec<(String, Value)> = input.to_vec();
1175 s.extend(results.iter().cloned());
1176 s
1177 };
1178 Ok(evaluate_expression(&output, &scope)?)
1179}