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