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