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