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 UnknownEntry,
52}
53
54impl BehaviorFailureCode {
55 pub fn as_str(self) -> &'static str {
56 match self {
57 BehaviorFailureCode::UnknownComponent => "UNKNOWN_COMPONENT",
58 BehaviorFailureCode::UnknownNodeKind => "UNKNOWN_NODE_KIND",
59 BehaviorFailureCode::MapOverNotArray => "MAP_OVER_NOT_ARRAY",
60 BehaviorFailureCode::MapIntoElementNotObject => "MAP_INTO_ELEMENT_NOT_OBJECT",
61 BehaviorFailureCode::MapBatchResultMismatch => "MAP_BATCH_RESULT_MISMATCH",
62 BehaviorFailureCode::UnknownEntry => "UNKNOWN_ENTRY",
63 }
64 }
65}
66
67#[derive(Debug, Clone)]
70pub struct BehaviorError {
71 code: String,
72 pub message: String,
73}
74
75impl BehaviorError {
76 pub fn new(code: impl Into<String>, message: impl Into<String>) -> Self {
81 BehaviorError {
82 code: code.into(),
83 message: message.into(),
84 }
85 }
86
87 pub fn code(&self) -> &str {
89 &self.code
90 }
91}
92
93impl std::fmt::Display for BehaviorError {
94 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
95 write!(f, "{}: {}", self.code, self.message)
96 }
97}
98impl std::error::Error for BehaviorError {}
99
100impl From<ExprFailure> for BehaviorError {
101 fn from(e: ExprFailure) -> Self {
102 BehaviorError {
103 code: e.code.as_str().to_string(),
104 message: e.message,
105 }
106 }
107}
108impl From<PlanFailure> for BehaviorError {
109 fn from(e: PlanFailure) -> Self {
110 BehaviorError {
111 code: e.code.as_str().to_string(),
112 message: e.message,
113 }
114 }
115}
116
117#[cfg(feature = "ir")]
118fn bfail<T>(code: BehaviorFailureCode, message: impl Into<String>) -> Result<T, BehaviorError> {
119 Err(BehaviorError {
120 code: code.as_str().to_string(),
121 message: message.into(),
122 })
123}
124
125pub trait ComponentExec {
130 fn exec(
131 &mut self,
132 component: &str,
133 ports: &[(String, Value)],
134 bound: Option<&Value>,
135 ) -> Option<ExecOutcome>;
136
137 fn exec_ctx(
143 &mut self,
144 node_id: &str,
145 component: &str,
146 ports: &[(String, Value)],
147 bound: Option<&Value>,
148 ) -> Option<ExecOutcome> {
149 let _ = node_id;
150 self.exec(component, ports, bound)
151 }
152}
153
154impl ComponentExec for &mut dyn ComponentExec {
171 fn exec(
172 &mut self,
173 component: &str,
174 ports: &[(String, Value)],
175 bound: Option<&Value>,
176 ) -> Option<ExecOutcome> {
177 (**self).exec(component, ports, bound)
178 }
179
180 fn exec_ctx(
181 &mut self,
182 node_id: &str,
183 component: &str,
184 ports: &[(String, Value)],
185 bound: Option<&Value>,
186 ) -> Option<ExecOutcome> {
187 (**self).exec_ctx(node_id, component, ports, bound)
188 }
189}
190
191#[cfg(feature = "ir")]
192fn scope_get<'a>(scope: &'a [(String, Value)], key: &str) -> Option<&'a Value> {
193 scope.iter().find(|(k, _)| k == key).map(|(_, v)| v)
194}
195
196#[cfg(feature = "ir")]
197fn node_kind(n: &J) -> Result<&'static str, BehaviorError> {
198 if n.get("map").is_some() {
199 Ok("map")
200 } else if n.get("cond").is_some() {
201 Ok("cond")
202 } else if n.get("component").is_some() {
203 Ok("componentRef")
204 } else {
205 bfail(
206 BehaviorFailureCode::UnknownNodeKind,
207 format!(
208 "body node '{}' is not componentRef/map/cond",
209 n.get("id").and_then(|v| v.as_str()).unwrap_or("?")
210 ),
211 )
212 }
213}
214
215#[cfg(feature = "ir")]
216fn node_parent(n: &J) -> Option<&str> {
217 let sub = if let Some(m) = n.get("map") {
218 m
219 } else if let Some(c) = n.get("cond") {
220 c
221 } else {
222 n
223 };
224 sub.get("parent").and_then(|v| v.as_str())
225}
226
227#[cfg(feature = "ir")]
228fn node_bind_field(n: &J) -> Option<String> {
229 if n.get("map").is_some() || n.get("cond").is_some() {
230 return None;
231 }
232 n.get("bindField")
233 .and_then(|v| v.as_str())
234 .map(str::to_string)
235}
236
237#[cfg(feature = "ir")]
238fn node_relation_kind(n: &J) -> Option<RelationKind> {
239 let sub = if let Some(m) = n.get("map") { m } else { n };
240 if n.get("cond").is_some() {
241 return None;
242 }
243 match sub.get("relationKind").and_then(|v| v.as_str()) {
244 Some("connection") => Some(RelationKind::Connection),
245 Some(_) => Some(RelationKind::Single),
246 None => None,
247 }
248}
249
250#[cfg(feature = "ir")]
251fn node_relation_kind_str(n: &J) -> Option<&str> {
252 let sub = if let Some(m) = n.get("map") { m } else { n };
253 if n.get("cond").is_some() {
254 return None;
255 }
256 sub.get("relationKind").and_then(|v| v.as_str())
257}
258
259#[cfg(feature = "ir")]
260fn node_policy(n: &J) -> Option<String> {
261 if n.get("cond").is_some() {
262 return None;
263 }
264 let sub = if let Some(m) = n.get("map") { m } else { n };
265 sub.get("policy")
266 .and_then(|v| v.as_str())
267 .map(str::to_string)
268}
269
270#[cfg(feature = "ir")]
271fn eval_ports(ports: &J, scope: &[(String, Value)]) -> Result<Vec<(String, Value)>, BehaviorError> {
272 let obj = ports.as_object().ok_or_else(|| BehaviorError {
273 code: BehaviorFailureCode::UnknownNodeKind.as_str().to_string(),
274 message: "ports must be an object".into(),
275 })?;
276 let mut out = Vec::with_capacity(obj.len());
277 for (k, v) in obj {
278 out.push((k.clone(), evaluate_expression(v, scope)?));
279 }
280 Ok(out)
281}
282
283#[cfg(feature = "ir")]
285fn parse_plan(plan: Option<&J>) -> Option<ExecutionPlanSpec> {
286 let p = plan?;
287 if p.is_null() {
288 return None;
289 }
290 let groups = p
291 .get("groups")?
292 .as_array()?
293 .iter()
294 .map(|g| {
295 g.as_array()
296 .map(|row| {
297 row.iter()
298 .filter_map(|i| i.as_u64().map(|x| x as usize))
299 .collect()
300 })
301 .unwrap_or_default()
302 })
303 .collect();
304 let concurrency = p.get("concurrency").and_then(|c| c.as_i64()).unwrap_or(1);
305 Some(ExecutionPlanSpec {
306 groups,
307 concurrency,
308 })
309}
310
311#[cfg(feature = "ir")]
320pub fn run_behavior(
321 ir: &J,
322 handlers: &mut dyn ComponentExec,
323 input: &[(String, Value)],
324 entry: Option<&str>,
325) -> Result<Value, BehaviorError> {
326 let components = ir
327 .get("components")
328 .and_then(|c| c.as_array())
329 .ok_or_else(|| BehaviorError {
330 code: BehaviorFailureCode::UnknownEntry.as_str().to_string(),
331 message: "IR.components must be an array".into(),
332 })?;
333 let comp = match entry {
334 Some(name) => components
335 .iter()
336 .find(|c| c.get("name").and_then(|n| n.as_str()) == Some(name)),
337 None => components.first(),
338 };
339 let comp = match comp {
340 Some(c) => c,
341 None => {
342 return bfail(
343 BehaviorFailureCode::UnknownEntry,
344 format!("component '{}' not found in IR", entry.unwrap_or("<first>")),
345 )
346 }
347 };
348
349 let body: Vec<J> = comp
350 .get("body")
351 .and_then(|b| b.as_array())
352 .cloned()
353 .unwrap_or_default();
354
355 let index_of = |id: &str| {
356 body.iter()
357 .position(|n| n.get("id").and_then(|v| v.as_str()) == Some(id))
358 };
359
360 let mut ops: Vec<OpSpec> = Vec::with_capacity(body.len());
362 for n in &body {
363 let parent = node_parent(n).and_then(index_of);
364 ops.push(OpSpec {
365 id: n
366 .get("id")
367 .and_then(|v| v.as_str())
368 .unwrap_or_default()
369 .to_string(),
370 parent,
371 bind_field: node_bind_field(n),
372 relation_kind: node_relation_kind(n),
373 policy: node_policy(n),
374 });
375 }
376
377 run_behavior_inner(
378 &body,
379 &ops,
380 parse_plan(comp.get("plan")),
381 comp.get("output").cloned().unwrap_or(J::Null),
382 input,
383 handlers,
384 )
385}
386
387#[cfg(feature = "ir")]
391fn run_behavior_inner(
392 body: &[J],
393 ops: &[OpSpec],
394 plan: Option<ExecutionPlanSpec>,
395 output: J,
396 input: &[(String, Value)],
397 handlers: &mut dyn ComponentExec,
398) -> Result<Value, BehaviorError> {
399 let index_of = |id: &str| {
400 body.iter()
401 .position(|n| n.get("id").and_then(|v| v.as_str()) == Some(id))
402 };
403
404 let mut results: Vec<(String, Value)> = Vec::new();
405 let mut pending_err: Option<BehaviorError> = None;
406
407 let run = {
408 let results_cell = &mut results;
409 let err_cell = &mut pending_err;
410 let exec = |op: &OpSpec, _bound: Option<&Value>| -> ExecOutcome {
411 if err_cell.is_some() {
412 return ExecOutcome::Error("aborted".into());
413 }
414 let idx = match index_of(&op.id) {
415 Some(i) => i,
416 None => {
417 *err_cell = Some(BehaviorError {
418 code: BehaviorFailureCode::UnknownNodeKind.as_str().to_string(),
419 message: format!("no body node for op '{}'", op.id),
420 });
421 return ExecOutcome::Error("aborted".into());
422 }
423 };
424 let node = &body[idx];
425
426 let base_scope = |extra: Option<(&str, &Value)>| -> Vec<(String, Value)> {
427 let mut s: Vec<(String, Value)> = input.to_vec();
428 for (k, v) in results_cell.iter() {
429 s.push((k.clone(), v.clone()));
430 }
431 if let Some((k, v)) = extra {
432 s.push((k.to_string(), v.clone()));
433 }
434 s
435 };
436
437 let outcome = match node_kind(node) {
438 Err(e) => {
439 *err_cell = Some(e);
440 return ExecOutcome::Error("aborted".into());
441 }
442 Ok("cond") => {
443 let c = node.get("cond").unwrap();
444 let cond_expr = serde_json::json!({
445 "cond": [c.get("if"), c.get("then"), c.get("else")]
446 });
447 match evaluate_expression(&cond_expr, &base_scope(None)) {
448 Ok(v) => ExecOutcome::Ok(v),
449 Err(e) => {
450 *err_cell = Some(e.into());
451 return ExecOutcome::Error("aborted".into());
452 }
453 }
454 }
455 Ok("map") => {
456 let m = node.get("map").unwrap();
457 let over = match evaluate_expression(
458 m.get("over").unwrap_or(&J::Null),
459 &base_scope(None),
460 ) {
461 Ok(v) => v,
462 Err(e) => {
463 *err_cell = Some(e.into());
464 return ExecOutcome::Error("aborted".into());
465 }
466 };
467 let arr = match &over {
468 Value::Arr(a) => a.clone(),
469 _ => {
470 *err_cell = Some(BehaviorError {
471 code: BehaviorFailureCode::MapOverNotArray.as_str().to_string(),
472 message: format!("map '{}': 'over' is not an array", op.id),
473 });
474 return ExecOutcome::Error("aborted".into());
475 }
476 };
477 let component = m.get("component").and_then(|v| v.as_str()).unwrap_or("");
478 let as_name = m.get("as").and_then(|v| v.as_str()).unwrap_or("$");
479 let ports_j = m
480 .get("ports")
481 .cloned()
482 .unwrap_or(J::Object(Default::default()));
483 let when = m.get("when");
484 let batched = m.get("batched").and_then(|v| v.as_bool()).unwrap_or(false);
485 let into = m.get("into").and_then(|v| v.as_str());
486
487 let keep = |scope: &[(String, Value)]| -> Result<bool, BehaviorError> {
490 match when {
491 None => Ok(true),
492 Some(w) => {
493 let cond_expr = serde_json::json!({ "cond": [w, true, false] });
494 Ok(matches!(
495 evaluate_expression(&cond_expr, scope)?,
496 Value::Bool(true)
497 ))
498 }
499 }
500 };
501
502 let mut kept_idx: Vec<usize> = Vec::new(); let collected: Vec<Value>;
504 if batched {
505 let mut items: Vec<Value> = Vec::new();
507 for (i, el) in arr.iter().enumerate() {
508 let scope = base_scope(Some((as_name, el)));
509 match keep(&scope) {
510 Ok(false) => continue,
511 Ok(true) => {}
512 Err(e) => {
513 *err_cell = Some(e);
514 return ExecOutcome::Error("aborted".into());
515 }
516 }
517 match eval_ports(&ports_j, &scope) {
518 Ok(p) => items.push(Value::Obj(p)),
519 Err(e) => {
520 *err_cell = Some(e);
521 return ExecOutcome::Error("aborted".into());
522 }
523 }
524 kept_idx.push(i);
525 }
526 if items.is_empty() {
527 collected = Vec::new(); } else {
529 let want = items.len();
530 let batch_ports = vec![("items".to_string(), Value::Arr(items))];
531 match handlers.exec_ctx(&op.id, component, &batch_ports, None) {
532 None => {
533 *err_cell = Some(BehaviorError {
534 code: BehaviorFailureCode::UnknownComponent
535 .as_str()
536 .to_string(),
537 message: format!(
538 "component '{component}' has no handler (fail-closed)"
539 ),
540 });
541 return ExecOutcome::Error("aborted".into());
542 }
543 Some(ExecOutcome::Error(e)) => return ExecOutcome::Error(e),
544 Some(ExecOutcome::Ok(Value::Arr(r))) if r.len() == want => {
545 collected = r;
546 }
547 Some(ExecOutcome::Ok(_)) => {
548 *err_cell = Some(BehaviorError {
549 code: BehaviorFailureCode::MapBatchResultMismatch
550 .as_str()
551 .to_string(),
552 message: format!(
553 "map '{}': batched handler must return a list aligned to items (want {want})",
554 op.id
555 ),
556 });
557 return ExecOutcome::Error("aborted".into());
558 }
559 }
560 }
561 } else {
562 let mut out: Vec<Value> = Vec::with_capacity(arr.len());
563 for (i, el) in arr.iter().enumerate() {
564 let scope = base_scope(Some((as_name, el)));
565 match keep(&scope) {
566 Ok(false) => continue,
567 Ok(true) => {}
568 Err(e) => {
569 *err_cell = Some(e);
570 return ExecOutcome::Error("aborted".into());
571 }
572 }
573 let ports = match eval_ports(&ports_j, &scope) {
574 Ok(p) => p,
575 Err(e) => {
576 *err_cell = Some(e);
577 return ExecOutcome::Error("aborted".into());
578 }
579 };
580 match handlers.exec_ctx(&op.id, component, &ports, Some(el)) {
581 None => {
582 *err_cell = Some(BehaviorError {
583 code: BehaviorFailureCode::UnknownComponent
584 .as_str()
585 .to_string(),
586 message: format!(
587 "component '{component}' has no handler (fail-closed)"
588 ),
589 });
590 return ExecOutcome::Error("aborted".into());
591 }
592 Some(ExecOutcome::Error(e)) => return ExecOutcome::Error(e),
593 Some(ExecOutcome::Ok(v)) => out.push(v),
594 }
595 kept_idx.push(i);
596 }
597 collected = out;
598 }
599
600 match into {
601 None => ExecOutcome::Ok(Value::Arr(collected)),
602 Some(key) => {
603 let mut augmented: Vec<Value> = Vec::with_capacity(arr.len());
606 let mut k = 0usize;
607 for (i, el) in arr.iter().enumerate() {
608 if k < kept_idx.len() && kept_idx[k] == i {
609 let fields = match el {
610 Value::Obj(f) => f.clone(),
611 _ => {
612 *err_cell = Some(BehaviorError {
613 code: BehaviorFailureCode::MapIntoElementNotObject
614 .as_str()
615 .to_string(),
616 message: format!(
617 "map '{}': 'into' requires object elements (element {i} is not an object)",
618 op.id
619 ),
620 });
621 return ExecOutcome::Error("aborted".into());
622 }
623 };
624 let mut fields = fields;
625 let val = collected[k].clone();
626 match fields.iter_mut().find(|(fk, _)| fk == key) {
627 Some(slot) => slot.1 = val,
628 None => fields.push((key.to_string(), val)),
629 }
630 augmented.push(Value::Obj(fields));
631 k += 1;
632 } else {
633 augmented.push(el.clone());
634 }
635 }
636 ExecOutcome::Ok(Value::Arr(augmented))
637 }
638 }
639 }
640 Ok(_) => {
641 let component = node.get("component").and_then(|v| v.as_str()).unwrap_or("");
642 let ports_j = node
643 .get("ports")
644 .cloned()
645 .unwrap_or(J::Object(Default::default()));
646 let ports = match eval_ports(&ports_j, &base_scope(None)) {
647 Ok(p) => p,
648 Err(e) => {
649 *err_cell = Some(e);
650 return ExecOutcome::Error("aborted".into());
651 }
652 };
653 match handlers.exec_ctx(&op.id, component, &ports, None) {
654 None => {
655 *err_cell = Some(BehaviorError {
656 code: BehaviorFailureCode::UnknownComponent.as_str().to_string(),
657 message: format!(
658 "component '{component}' has no handler (fail-closed)"
659 ),
660 });
661 return ExecOutcome::Error("aborted".into());
662 }
663 Some(o) => o,
664 }
665 }
666 };
667
668 if let ExecOutcome::Ok(v) = &outcome {
669 results_cell.push((op.id.clone(), v.clone()));
670 }
671 outcome
672 };
673
674 run_plan(plan.as_ref(), ops, exec)
675 };
676
677 if let Some(e) = pending_err {
680 return Err(e);
681 }
682 let run = run?;
683
684 for (i, id_val) in run.skipped.iter().enumerate() {
687 let _ = i;
688 if let Some(idx) = index_of(id_val) {
689 let rk = node_relation_kind_str(&body[idx]);
690 let unproduced = if rk == Some("connection") {
691 Value::Obj(vec![
692 ("items".into(), Value::Arr(vec![])),
693 ("cursor".into(), Value::Null),
694 ])
695 } else {
696 Value::Null
697 };
698 if scope_get(&results, id_val).is_none() {
699 results.push((id_val.clone(), unproduced));
700 }
701 }
702 }
703
704 let scope: Vec<(String, Value)> = {
705 let mut s: Vec<(String, Value)> = input.to_vec();
706 s.extend(results.iter().cloned());
707 s
708 };
709 Ok(evaluate_expression(&output, &scope)?)
710}