1use std::collections::BTreeMap;
2use std::fs;
3use std::path::Path;
4
5use anyhow::{Context, Result};
6use serde::{Deserialize, Serialize};
7use serde_json::Value;
8use sha2::{Digest, Sha256};
9
10use crate::{
11 BeamPaths, EventDraft, EventLog, ParamDef, WorkflowActor, WorkflowDefinition,
12 parse_workflow_definition,
13};
14
15#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
16#[serde(rename_all = "camelCase")]
17pub struct RunChatBinding {
18 pub chat_id: String,
19 pub lark_app_id: String,
20}
21
22#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
23#[serde(rename_all = "camelCase")]
24pub struct WorkflowOutputRef {
25 pub output_hash: String,
26 pub output_path: String,
27 pub output_bytes: usize,
28 pub output_schema_version: u32,
29 #[serde(default, skip_serializing_if = "Option::is_none")]
30 pub content_type: Option<String>,
31}
32
33#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
34pub struct WorkflowRunBootstrap {
35 pub run_id: String,
36 pub workflow_id: String,
37 pub revision_id: String,
38 pub input_ref: WorkflowOutputRef,
39}
40
41#[derive(Debug, Clone)]
42pub struct BootstrapWorkflowRunInput<'a> {
43 pub run_id: &'a str,
44 pub workflow_json: &'a str,
45 pub expected_workflow_id: Option<&'a str>,
46 pub params: &'a BTreeMap<String, Value>,
47 pub initiator: &'a str,
48 pub chat_binding: Option<RunChatBinding>,
49}
50
51pub fn mint_workflow_run_id(workflow_id: &str, now_ms: u64) -> String {
52 let safe = workflow_id
53 .chars()
54 .map(|ch| match ch {
55 'a'..='z' | 'A'..='Z' | '0'..='9' | '.' | '_' | '-' => ch,
56 _ => '_',
57 })
58 .collect::<String>();
59 format!("{}-{}", safe, now_ms)
60}
61
62pub fn bootstrap_workflow_run(
63 paths: &BeamPaths,
64 input: BootstrapWorkflowRunInput<'_>,
65) -> Result<WorkflowRunBootstrap> {
66 let workflow = parse_workflow_definition(input.workflow_json)?;
67 let workflow_id = workflow.workflow_id.clone();
68 if let Some(expected) = input.expected_workflow_id {
69 if expected != workflow_id {
70 anyhow::bail!(
71 "workflowId mismatch: requested={} file={}",
72 expected,
73 workflow_id
74 );
75 }
76 }
77
78 let normalized_params = normalize_workflow_params(&workflow, input.params)?;
80
81 let run_dir = paths.workflow_run_dir(input.run_id);
82 fs::create_dir_all(run_dir.join("blobs"))?;
83 fs::write(run_dir.join("workflow.json"), input.workflow_json)?;
84 if let Some(binding) = input.chat_binding {
85 fs::write(
86 run_dir.join("chat-binding.json"),
87 serde_json::to_vec_pretty(&binding)?,
88 )?;
89 }
90
91 let params_json = serde_json::to_vec(&normalized_params)?;
92 let params_hash = sha256_hex(¶ms_json);
93 let input_path = run_dir.join("blobs").join(¶ms_hash);
94 fs::write(&input_path, ¶ms_json)?;
95 let input_ref = WorkflowOutputRef {
96 output_hash: format!("sha256:{}", params_hash),
97 output_path: input_path.display().to_string(),
98 output_bytes: params_json.len(),
99 output_schema_version: 1,
100 content_type: Some("application/json".to_string()),
101 };
102
103 let mut log = EventLog::new(input.run_id.to_string(), paths.workflow_runs_dir())?;
104 let revision_id = sha256_hex(&serde_json::to_vec(&workflow)?);
105 let _run_created = log.append(EventDraft {
106 event_type: "runCreated".to_string(),
107 actor: WorkflowActor::System,
108 payload: serde_json::json!({
109 "workflowId": workflow_id,
110 "revisionId": revision_id,
111 "inputRef": input_ref,
112 "initiator": input.initiator,
113 }),
114 timestamp: None,
115 payload_hash: None,
116 })?;
117 let _run_started = log.append(EventDraft {
118 event_type: "runStarted".to_string(),
119 actor: WorkflowActor::Scheduler,
120 payload: serde_json::json!({}),
121 timestamp: None,
122 payload_hash: None,
123 })?;
124
125 Ok(WorkflowRunBootstrap {
126 run_id: input.run_id.to_string(),
127 workflow_id,
128 revision_id,
129 input_ref,
130 })
131}
132
133pub fn normalize_workflow_params(
140 def: &WorkflowDefinition,
141 input: &BTreeMap<String, Value>,
142) -> Result<BTreeMap<String, Value>> {
143 let Some(params_def) = &def.params else {
144 if input.is_empty() {
146 return Ok(BTreeMap::new());
147 }
148 let unknown: Vec<String> = input.keys().cloned().collect();
149 anyhow::bail!(
150 "unknown workflow parameter(s): {}. No parameters are declared for this workflow.",
151 unknown.join(", ")
152 );
153 };
154
155 if params_def.is_empty() {
157 if input.is_empty() {
158 return Ok(BTreeMap::new());
159 }
160 let unknown: Vec<String> = input.keys().cloned().collect();
161 anyhow::bail!(
162 "unknown workflow parameter(s): {}. No parameters are declared for this workflow.",
163 unknown.join(", ")
164 );
165 }
166
167 let defined_keys: std::collections::HashSet<&String> = params_def.keys().collect();
169 let unknown: Vec<String> = input
170 .keys()
171 .filter(|k| !defined_keys.contains(*k))
172 .cloned()
173 .collect();
174
175 if !unknown.is_empty() {
176 let available: Vec<&str> = params_def.keys().map(String::as_str).collect();
177 anyhow::bail!(
178 "unknown workflow parameter(s): {}. Available parameters: [{}]",
179 unknown.join(", "),
180 available.join(", ")
181 );
182 }
183
184 let mut normalized = BTreeMap::new();
186 let mut missing: Vec<String> = Vec::new();
187
188 for (name, param_def) in params_def {
189 match input.get(name) {
190 Some(value) => {
191 let coerced = coerce_param_value(name, param_def, value)?;
195
196 validate_param_type(name, param_def, &coerced)?;
197 validate_param_format(name, param_def, &coerced)?;
198
199 if param_def.required == Some(true) && param_def.param_type == "string" {
202 if let Value::String(s) = &coerced {
203 if s.trim().is_empty() {
204 missing.push(name.clone());
205 continue;
206 }
207 }
208 }
209 normalized.insert(name.clone(), coerced);
212 }
213 None => {
214 if param_def.required == Some(true) {
215 missing.push(name.clone());
216 } else if let Some(default) = ¶m_def.default {
217 validate_param_type(name, param_def, default)?;
220 validate_param_format(name, param_def, default)?;
221 normalized.insert(name.clone(), default.clone());
222 }
223 }
225 }
226 }
227
228 if !missing.is_empty() {
229 anyhow::bail!(
230 "missing required workflow parameter(s): {}",
231 missing.join(", ")
232 );
233 }
234
235 Ok(normalized)
236}
237
238fn validate_param_type(name: &str, def: &ParamDef, value: &Value) -> Result<()> {
240 match def.param_type.as_str() {
241 "string" => {
242 if !value.is_string() {
243 anyhow::bail!(
244 "workflow parameter '{}' expects type 'string', got {}",
245 name,
246 describe_value_kind(value)
247 );
248 }
249 }
250 "number" => {
251 if !value.is_number() {
252 anyhow::bail!(
253 "workflow parameter '{}' expects type 'number', got {}",
254 name,
255 describe_value_kind(value)
256 );
257 }
258 }
259 "integer" => match value {
260 Value::Number(n) => {
261 let is_int = n.as_i64().is_some()
262 || n.as_u64().is_some()
263 || n.as_f64().map(|f| f.fract() == 0.0).unwrap_or(false);
264 if !is_int {
265 anyhow::bail!(
266 "workflow parameter '{}' expects type 'integer', got non-integer number",
267 name
268 );
269 }
270 }
271 _ => anyhow::bail!(
272 "workflow parameter '{}' expects type 'integer', got {}",
273 name,
274 describe_value_kind(value)
275 ),
276 },
277 "boolean" => {
278 if !value.is_boolean() {
279 anyhow::bail!(
280 "workflow parameter '{}' expects type 'boolean', got {}",
281 name,
282 describe_value_kind(value)
283 );
284 }
285 }
286 "object" => {
287 if !value.is_object() {
288 anyhow::bail!(
289 "workflow parameter '{}' expects type 'object', got {}",
290 name,
291 describe_value_kind(value)
292 );
293 }
294 }
295 "array" => {
296 if !value.is_array() {
297 anyhow::bail!(
298 "workflow parameter '{}' expects type 'array', got {}",
299 name,
300 describe_value_kind(value)
301 );
302 }
303 }
304 unknown => {
305 anyhow::bail!(
306 "workflow parameter '{}' has unknown type '{}'",
307 name,
308 unknown
309 );
310 }
311 }
312 Ok(())
313}
314
315fn validate_param_format(name: &str, def: &ParamDef, value: &Value) -> Result<()> {
320 let Some(format) = &def.format else {
321 return Ok(());
322 };
323
324 if def.param_type != "string" {
326 anyhow::bail!(
327 "workflow parameter '{}' has format '{}' but type is '{}'; format is only valid for string type",
328 name,
329 format,
330 def.param_type,
331 );
332 }
333
334 let Value::String(s) = value else {
335 return Ok(());
337 };
338
339 match format.as_str() {
340 "date" => validate_date(name, s),
341 "date-time" => validate_date_time(name, s),
342 "email" => validate_email(name, s),
343 unknown => {
344 anyhow::bail!(
345 "workflow parameter '{}' has unknown format '{}'",
346 name,
347 unknown,
348 );
349 }
350 }
351}
352
353fn validate_date(name: &str, value: &str) -> Result<()> {
355 chrono::NaiveDate::parse_from_str(value, "%Y-%m-%d").map_err(|_| {
356 anyhow::anyhow!(
357 "workflow parameter '{}' with format 'date' must be a valid date (YYYY-MM-DD), got: {}",
358 name,
359 value,
360 )
361 })?;
362 Ok(())
363}
364
365fn validate_date_time(name: &str, value: &str) -> Result<()> {
367 chrono::DateTime::parse_from_rfc3339(value).map_err(|_| {
369 anyhow::anyhow!(
370 "workflow parameter '{}' with format 'date-time' must be valid RFC 3339, got: {}",
371 name,
372 value,
373 )
374 })?;
375 Ok(())
376}
377
378fn validate_email(name: &str, value: &str) -> Result<()> {
382 let at_count = value.chars().filter(|&c| c == '@').count();
384 if at_count != 1 {
385 anyhow::bail!(
386 "workflow parameter '{}' with format 'email' must contain exactly one '@', got: {}",
387 name,
388 value,
389 );
390 }
391
392 let at_pos = value.find('@').unwrap();
393 let local = &value[..at_pos];
394 let domain = &value[at_pos + 1..];
395
396 if local.is_empty() {
397 anyhow::bail!(
398 "workflow parameter '{}' with format 'email' has empty local part, got: {}",
399 name,
400 value,
401 );
402 }
403 if domain.is_empty() {
404 anyhow::bail!(
405 "workflow parameter '{}' with format 'email' has empty domain part, got: {}",
406 name,
407 value,
408 );
409 }
410
411 if !domain.contains('.') {
413 anyhow::bail!(
414 "workflow parameter '{}' with format 'email' domain must contain at least one dot, got: {}",
415 name,
416 value,
417 );
418 }
419
420 if value.chars().any(|c| c.is_whitespace()) {
422 anyhow::bail!(
423 "workflow parameter '{}' with format 'email' must not contain whitespace, got: {}",
424 name,
425 value,
426 );
427 }
428
429 for label in domain.split('.') {
431 if label.is_empty() {
432 anyhow::bail!(
433 "workflow parameter '{}' with format 'email' domain has empty label, got: {}",
434 name,
435 value,
436 );
437 }
438 if label.starts_with('-') || label.ends_with('-') {
439 anyhow::bail!(
440 "workflow parameter '{}' with format 'email' domain label must not start or end with '-', got: {}",
441 name,
442 value,
443 );
444 }
445 }
446
447 Ok(())
448}
449
450fn describe_value_kind(value: &Value) -> &'static str {
451 match value {
452 Value::Null => "null",
453 Value::Bool(_) => "bool",
454 Value::Number(_) => "number",
455 Value::String(_) => "string",
456 Value::Array(_) => "array",
457 Value::Object(_) => "object",
458 }
459}
460
461fn coerce_param_value(name: &str, def: &ParamDef, value: &Value) -> Result<Value> {
482 let target_type = def.param_type.as_str();
483
484 let already_typed = match target_type {
486 "string" => value.is_string(),
487 "number" | "integer" => value.is_number(),
488 "boolean" => value.is_boolean(),
489 "object" => value.is_object(),
490 "array" => value.is_array(),
491 _ => false,
492 };
493 if already_typed {
494 return Ok(value.clone());
495 }
496
497 if !value.is_string() {
499 return Ok(value.clone());
501 }
502
503 let s = value.as_str().unwrap();
504
505 match target_type {
506 "string" => {
507 Ok(value.clone())
509 }
510 "boolean" => {
511 let trimmed = s.trim();
512 if trimmed.is_empty() {
513 anyhow::bail!(
514 "workflow parameter '{}' expects type 'boolean', got empty string",
515 name
516 );
517 }
518 match trimmed {
519 "true" => Ok(Value::Bool(true)),
520 "false" => Ok(Value::Bool(false)),
521 _ => anyhow::bail!(
522 "workflow parameter '{}' expects type 'boolean', got string '{}' (only 'true' or 'false' are accepted)",
523 name,
524 trimmed
525 ),
526 }
527 }
528 "number" => {
529 let trimmed = s.trim();
530 if trimmed.is_empty() {
531 anyhow::bail!(
532 "workflow parameter '{}' expects type 'number', got empty string",
533 name
534 );
535 }
536 let v: Value = serde_json::from_str(trimmed).map_err(|e| {
537 anyhow::anyhow!(
538 "workflow parameter '{}' expects type 'number', failed to parse '{}' as a number: {}",
539 name, trimmed, e
540 )
541 })?;
542 if !v.is_number() {
543 anyhow::bail!(
544 "workflow parameter '{}' expects type 'number', but string '{}' parsed to {}",
545 name,
546 trimmed,
547 describe_value_kind(&v)
548 );
549 }
550 Ok(v)
551 }
552 "integer" => {
553 let trimmed = s.trim();
554 if trimmed.is_empty() {
555 anyhow::bail!(
556 "workflow parameter '{}' expects type 'integer', got empty string",
557 name
558 );
559 }
560
561 let is_decimal_integer = trimmed.chars().enumerate().all(|(i, c)| {
564 if i == 0 && c == '-' {
565 trimmed.len() > 1 } else {
567 c.is_ascii_digit()
568 }
569 });
570
571 if !is_decimal_integer {
572 anyhow::bail!(
573 "workflow parameter '{}' expects type 'integer', got string '{}' (only decimal integer strings like '42' or '-1' are accepted)",
574 name,
575 trimmed
576 );
577 }
578
579 let n: i128 = trimmed.parse().map_err(|_| {
581 anyhow::anyhow!(
582 "workflow parameter '{}' expects type 'integer', failed to parse '{}'",
583 name,
584 trimmed
585 )
586 })?;
587 let num = serde_json::Number::from_i128(n).ok_or_else(|| {
588 anyhow::anyhow!(
589 "workflow parameter '{}' expects type 'integer', number out of range: '{}'",
590 name,
591 trimmed
592 )
593 })?;
594 Ok(Value::Number(num))
595 }
596 "object" => {
597 let trimmed = s.trim();
598 if trimmed.is_empty() {
599 anyhow::bail!(
600 "workflow parameter '{}' expects type 'object', got empty string",
601 name
602 );
603 }
604 let v: Value = serde_json::from_str(trimmed).map_err(|e| {
605 anyhow::anyhow!(
606 "workflow parameter '{}' expects type 'object', but string value must be valid JSON (failed to parse: {})",
607 name, e
608 )
609 })?;
610 if !v.is_object() {
611 anyhow::bail!(
612 "workflow parameter '{}' expects type 'object', but string value parsed to {} (must be a JSON object like '{{\"a\":1}}')",
613 name,
614 describe_value_kind(&v)
615 );
616 }
617 Ok(v)
618 }
619 "array" => {
620 let trimmed = s.trim();
621 if trimmed.is_empty() {
622 anyhow::bail!(
623 "workflow parameter '{}' expects type 'array', got empty string",
624 name
625 );
626 }
627 let v: Value = serde_json::from_str(trimmed).map_err(|e| {
628 anyhow::anyhow!(
629 "workflow parameter '{}' expects type 'array', but string value must be valid JSON (failed to parse: {})",
630 name, e
631 )
632 })?;
633 if !v.is_array() {
634 anyhow::bail!(
635 "workflow parameter '{}' expects type 'array', but string value parsed to {} (must be a JSON array like '[\"a\",\"b\"]')",
636 name,
637 describe_value_kind(&v)
638 );
639 }
640 Ok(v)
641 }
642 unknown => {
643 anyhow::bail!(
644 "workflow parameter '{}' has unknown type '{}'",
645 name,
646 unknown
647 );
648 }
649 }
650}
651
652fn sha256_hex(bytes: &[u8]) -> String {
653 let mut hasher = Sha256::new();
654 hasher.update(bytes);
655 format!("{:x}", hasher.finalize())
656}
657
658pub fn read_workflow_definition_from_path(path: &Path) -> Result<String> {
659 Ok(fs::read_to_string(path).with_context(|| format!("读取 {} 失败", path.display()))?)
660}
661
662#[cfg(test)]
663mod tests {
664 use super::*;
665 use std::collections::BTreeMap;
666 use std::time::{SystemTime, UNIX_EPOCH};
667
668 fn temp_paths(label: &str) -> BeamPaths {
669 let nanos = SystemTime::now()
670 .duration_since(UNIX_EPOCH)
671 .unwrap_or_default()
672 .as_nanos();
673 BeamPaths::from_root(std::env::temp_dir().join(format!(
674 "beam-workflow-run-{label}-{nanos}-{}",
675 std::process::id()
676 )))
677 }
678
679 #[test]
680 fn mint_workflow_run_id_sanitizes_input() {
681 let run_id = mint_workflow_run_id("flow/a:b", 123);
682 assert!(run_id.starts_with("flow_a_b-123"));
683 assert!(!run_id.contains('/'));
684 assert!(!run_id.contains(':'));
685 }
686
687 #[test]
688 fn bootstrap_workflow_run_writes_snapshot_and_events() {
689 let paths = temp_paths("bootstrap");
690 let params: BTreeMap<String, Value> =
691 BTreeMap::from([(String::from("foo"), Value::String("bar".to_string()))]);
692 let result = bootstrap_workflow_run(
693 &paths,
694 BootstrapWorkflowRunInput {
695 run_id: "run-1",
696 workflow_json: r#"{"workflowId":"flow-a","version":1,"params":{"foo":{"type":"string"}},"nodes":{"node-a":{"type":"subagent","bot":"bot-a","prompt":"hi"}}}"#,
697 expected_workflow_id: Some("flow-a"),
698 params: ¶ms,
699 initiator: "cli",
700 chat_binding: Some(RunChatBinding {
701 chat_id: "chat-1".to_string(),
702 lark_app_id: "app-1".to_string(),
703 }),
704 },
705 )
706 .expect("bootstrap");
707 assert_eq!(result.run_id, "run-1");
708 assert_eq!(result.workflow_id, "flow-a");
709 assert!(
710 paths
711 .workflow_run_dir("run-1")
712 .join("workflow.json")
713 .exists()
714 );
715 assert!(
716 paths
717 .workflow_run_dir("run-1")
718 .join("chat-binding.json")
719 .exists()
720 );
721 assert!(paths.workflow_run_dir("run-1").join("blobs").exists());
722 let log = EventLog::new("run-1", paths.workflow_runs_dir()).expect("log");
723 let events = log.read_all().expect("events");
724 assert_eq!(events.len(), 2);
725 assert_eq!(events[0].event_type, "runCreated");
726 assert_eq!(events[1].event_type, "runStarted");
727 let _ = std::fs::remove_dir_all(paths.root());
728 }
729
730 #[test]
731 fn bootstrap_workflow_run_hashes_canonical_definition_bytes() {
732 let params: BTreeMap<String, Value> = BTreeMap::new();
733 let raw_a = r#"{"workflowId":"flow-a","version":1,"nodes":{"node-a":{"type":"subagent","bot":"bot-a","prompt":"hi","workingDir":"/tmp/demo"}}}"#;
734 let raw_b = r#"
735 {
736 "nodes": {
737 "node-a": {
738 "prompt": "hi",
739 "type": "subagent",
740 "bot": "bot-a",
741 "workingDir": "/tmp/demo"
742 }
743 },
744 "version": 1,
745 "workflowId": "flow-a"
746 }
747 "#;
748
749 let paths_a = temp_paths("bootstrap-canonical-a");
750 let paths_b = temp_paths("bootstrap-canonical-b");
751 let rev_a = bootstrap_workflow_run(
752 &paths_a,
753 BootstrapWorkflowRunInput {
754 run_id: "run-a",
755 workflow_json: raw_a,
756 expected_workflow_id: Some("flow-a"),
757 params: ¶ms,
758 initiator: "cli",
759 chat_binding: None,
760 },
761 )
762 .expect("bootstrap a")
763 .revision_id;
764 let rev_b = bootstrap_workflow_run(
765 &paths_b,
766 BootstrapWorkflowRunInput {
767 run_id: "run-b",
768 workflow_json: raw_b,
769 expected_workflow_id: Some("flow-a"),
770 params: ¶ms,
771 initiator: "cli",
772 chat_binding: None,
773 },
774 )
775 .expect("bootstrap b")
776 .revision_id;
777 assert_eq!(rev_a, rev_b);
778 let _ = std::fs::remove_dir_all(paths_a.root());
779 let _ = std::fs::remove_dir_all(paths_b.root());
780 }
781
782 #[test]
785 fn bootstrap_rejects_missing_required_param() {
786 let paths = temp_paths("missing-param");
787 let params: BTreeMap<String, Value> = BTreeMap::new(); let workflow_json = r#"{
789 "workflowId": "flow-req",
790 "version": 1,
791 "params": {
792 "task": {
793 "type": "string",
794 "required": true,
795 "description": "what to do"
796 }
797 },
798 "nodes": {
799 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
800 }
801 }"#;
802 let err = bootstrap_workflow_run(
803 &paths,
804 BootstrapWorkflowRunInput {
805 run_id: "run-missing",
806 workflow_json,
807 expected_workflow_id: Some("flow-req"),
808 params: ¶ms,
809 initiator: "test",
810 chat_binding: None,
811 },
812 )
813 .unwrap_err();
814 let msg = err.to_string();
815 assert!(
816 msg.contains("missing required workflow parameter"),
817 "got: {msg}"
818 );
819 assert!(msg.contains("task"), "expected 'task' in error, got: {msg}");
820 let run_dir = paths.workflow_run_dir("run-missing");
822 assert!(
823 !run_dir.exists(),
824 "run directory should NOT exist after param validation failure, but found: {}",
825 run_dir.display()
826 );
827 let _ = std::fs::remove_dir_all(paths.root());
828 }
829
830 #[test]
831 fn bootstrap_rejects_empty_required_param_value() {
832 let paths = temp_paths("empty-param");
833 let params: BTreeMap<String, Value> =
834 BTreeMap::from([(String::from("task"), Value::String(" ".to_string()))]); let workflow_json = r#"{
836 "workflowId": "flow-req",
837 "version": 1,
838 "params": {
839 "task": {
840 "type": "string",
841 "required": true,
842 "description": "what to do"
843 }
844 },
845 "nodes": {
846 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
847 }
848 }"#;
849 let err = bootstrap_workflow_run(
850 &paths,
851 BootstrapWorkflowRunInput {
852 run_id: "run-empty",
853 workflow_json,
854 expected_workflow_id: Some("flow-req"),
855 params: ¶ms,
856 initiator: "test",
857 chat_binding: None,
858 },
859 )
860 .unwrap_err();
861 let msg = err.to_string();
862 assert!(
863 msg.contains("missing required workflow parameter"),
864 "got: {msg}"
865 );
866 assert!(msg.contains("task"), "expected 'task' in error, got: {msg}");
867 let run_dir = paths.workflow_run_dir("run-empty");
869 assert!(
870 !run_dir.exists(),
871 "run directory should NOT exist after param validation failure, but found: {}",
872 run_dir.display()
873 );
874 let _ = std::fs::remove_dir_all(paths.root());
875 }
876
877 #[test]
878 fn bootstrap_succeeds_with_required_params_provided() {
879 let paths = temp_paths("provided-param");
880 let params: BTreeMap<String, Value> =
881 BTreeMap::from([(String::from("task"), Value::String("build XYZ".to_string()))]);
882 let workflow_json = r#"{
883 "workflowId": "flow-req",
884 "version": 1,
885 "params": {
886 "task": {
887 "type": "string",
888 "required": true,
889 "description": "what to do"
890 }
891 },
892 "nodes": {
893 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
894 }
895 }"#;
896 let result = bootstrap_workflow_run(
897 &paths,
898 BootstrapWorkflowRunInput {
899 run_id: "run-provided",
900 workflow_json,
901 expected_workflow_id: Some("flow-req"),
902 params: ¶ms,
903 initiator: "test",
904 chat_binding: None,
905 },
906 )
907 .expect("bootstrap with required params provided should succeed");
908 assert_eq!(result.workflow_id, "flow-req");
909 assert!(
910 paths
911 .workflow_run_dir("run-provided")
912 .join("workflow.json")
913 .exists()
914 );
915 let log = EventLog::new("run-provided", paths.workflow_runs_dir()).expect("log");
916 let events = log.read_all().expect("events");
917 assert_eq!(events.len(), 2);
918 assert_eq!(events[0].event_type, "runCreated");
919 assert_eq!(events[1].event_type, "runStarted");
920 let _ = std::fs::remove_dir_all(paths.root());
921 }
922
923 #[test]
924 fn bootstrap_ignores_optional_param_when_missing() {
925 let paths = temp_paths("optional-param");
926 let params: BTreeMap<String, Value> = BTreeMap::new(); let workflow_json = r#"{
928 "workflowId": "flow-opt",
929 "version": 1,
930 "params": {
931 "verbose": {
932 "type": "boolean",
933 "required": false,
934 "default": false
935 }
936 },
937 "nodes": {
938 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
939 }
940 }"#;
941 let result = bootstrap_workflow_run(
942 &paths,
943 BootstrapWorkflowRunInput {
944 run_id: "run-opt",
945 workflow_json,
946 expected_workflow_id: Some("flow-opt"),
947 params: ¶ms,
948 initiator: "test",
949 chat_binding: None,
950 },
951 )
952 .expect("bootstrap should ignore missing optional param");
953 assert_eq!(result.workflow_id, "flow-opt");
954 let _ = std::fs::remove_dir_all(paths.root());
955 }
956
957 #[test]
958 fn bootstrap_rejects_multiple_missing_required_params() {
959 let paths = temp_paths("multi-missing");
960 let params: BTreeMap<String, Value> = BTreeMap::new(); let workflow_json = r#"{
962 "workflowId": "flow-multi",
963 "version": 1,
964 "params": {
965 "task": {
966 "type": "string",
967 "required": true,
968 "description": "what to do"
969 },
970 "target": {
971 "type": "string",
972 "required": true,
973 "description": "where to deploy"
974 }
975 },
976 "nodes": {
977 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
978 }
979 }"#;
980 let err = bootstrap_workflow_run(
981 &paths,
982 BootstrapWorkflowRunInput {
983 run_id: "run-multi",
984 workflow_json,
985 expected_workflow_id: Some("flow-multi"),
986 params: ¶ms,
987 initiator: "test",
988 chat_binding: None,
989 },
990 )
991 .unwrap_err();
992 let msg = err.to_string();
993 assert!(
994 msg.contains("missing required workflow parameter"),
995 "got: {msg}"
996 );
997 assert!(msg.contains("task"), "expected 'task' in error, got: {msg}");
998 assert!(
999 msg.contains("target"),
1000 "expected 'target' in error, got: {msg}"
1001 );
1002 let run_dir = paths.workflow_run_dir("run-multi");
1003 assert!(
1004 !run_dir.exists(),
1005 "run directory should NOT exist on failure"
1006 );
1007 let _ = std::fs::remove_dir_all(paths.root());
1008 }
1009
1010 #[test]
1013 fn normalize_writes_default_value_for_optional_param() {
1014 let def = parse_workflow_definition(
1015 r#"{
1016 "workflowId": "flow-default",
1017 "version": 1,
1018 "params": {
1019 "verbose": {
1020 "type": "boolean",
1021 "required": false,
1022 "default": false
1023 },
1024 "level": {
1025 "type": "integer",
1026 "required": false,
1027 "default": 3
1028 }
1029 },
1030 "nodes": {
1031 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1032 }
1033 }"#,
1034 )
1035 .unwrap();
1036 let input: BTreeMap<String, Value> = BTreeMap::new();
1037 let normalized = normalize_workflow_params(&def, &input).expect("should succeed");
1038 assert_eq!(normalized.get("verbose"), Some(&Value::Bool(false)));
1039 assert_eq!(normalized.get("level"), Some(&serde_json::json!(3)));
1040 assert!(!normalized.contains_key("unknown_key"));
1042 }
1043
1044 #[test]
1045 fn normalize_type_validation_success() {
1046 let def = parse_workflow_definition(
1047 r#"{
1048 "workflowId": "flow-types",
1049 "version": 1,
1050 "params": {
1051 "name": { "type": "string" },
1052 "count": { "type": "integer" },
1053 "ratio": { "type": "number" },
1054 "enabled": { "type": "boolean" },
1055 "tags": { "type": "array" },
1056 "meta": { "type": "object" }
1057 },
1058 "nodes": {
1059 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1060 }
1061 }"#,
1062 )
1063 .unwrap();
1064 let input: BTreeMap<String, Value> = BTreeMap::from([
1065 (String::from("name"), Value::String("test".to_string())),
1066 (String::from("count"), serde_json::json!(42)),
1067 (String::from("ratio"), serde_json::json!(3.14)),
1068 (String::from("enabled"), Value::Bool(true)),
1069 (String::from("tags"), serde_json::json!(["a", "b"])),
1070 (String::from("meta"), serde_json::json!({"key": "val"})),
1071 ]);
1072 let _ = normalize_workflow_params(&def, &input).expect("all types valid");
1073 }
1074
1075 #[test]
1076 fn normalize_type_mismatch_fails() {
1077 let def = parse_workflow_definition(
1078 r#"{
1079 "workflowId": "flow-type-err",
1080 "version": 1,
1081 "params": {
1082 "enabled": { "type": "boolean" }
1083 },
1084 "nodes": {
1085 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1086 }
1087 }"#,
1088 )
1089 .unwrap();
1090 let input: BTreeMap<String, Value> =
1091 BTreeMap::from([(String::from("enabled"), Value::String("yes".to_string()))]);
1092 let err = normalize_workflow_params(&def, &input).unwrap_err();
1093 let msg = err.to_string();
1094 assert!(msg.contains("enabled"), "got: {msg}");
1095 assert!(msg.contains("boolean"), "got: {msg}");
1096 }
1097
1098 #[test]
1099 fn normalize_unknown_key_fails_when_schema_defined() {
1100 let def = parse_workflow_definition(
1101 r#"{
1102 "workflowId": "flow-unknown",
1103 "version": 1,
1104 "params": {
1105 "task": { "type": "string" }
1106 },
1107 "nodes": {
1108 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1109 }
1110 }"#,
1111 )
1112 .unwrap();
1113 let input: BTreeMap<String, Value> = BTreeMap::from([
1114 (String::from("task"), Value::String("hello".to_string())),
1115 (String::from("extra"), Value::String("bad".to_string())),
1116 ]);
1117 let err = normalize_workflow_params(&def, &input).unwrap_err();
1118 let msg = err.to_string();
1119 assert!(msg.contains("unknown workflow parameter"), "got: {msg}");
1120 assert!(msg.contains("extra"), "got: {msg}");
1121 assert!(
1122 msg.contains("task"),
1123 "expected available params list, got: {msg}"
1124 );
1125 }
1126
1127 #[test]
1128 fn normalize_no_schema_rejects_extra_params() {
1129 let def = parse_workflow_definition(
1130 r#"{
1131 "workflowId": "flow-no-schema",
1132 "version": 1,
1133 "nodes": {
1134 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1135 }
1136 }"#,
1137 )
1138 .unwrap();
1139 let input: BTreeMap<String, Value> = BTreeMap::from([
1140 (String::from("anything"), Value::String("goes".to_string())),
1141 (String::from("extra"), serde_json::json!({"deep": true})),
1142 ]);
1143 let err = normalize_workflow_params(&def, &input).unwrap_err();
1144 let msg = err.to_string();
1145 assert!(msg.contains("unknown workflow parameter"), "got: {msg}");
1146 assert!(msg.contains("anything"), "got: {msg}");
1147 assert!(msg.contains("extra"), "got: {msg}");
1148 assert!(msg.contains("No parameters are declared"), "got: {msg}");
1149 }
1150
1151 #[test]
1152 fn normalize_no_schema_with_empty_input_succeeds() {
1153 let def = parse_workflow_definition(
1154 r#"{
1155 "workflowId": "flow-no-schema",
1156 "version": 1,
1157 "nodes": {
1158 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1159 }
1160 }"#,
1161 )
1162 .unwrap();
1163 let input: BTreeMap<String, Value> = BTreeMap::new();
1164 let normalized = normalize_workflow_params(&def, &input)
1165 .expect("empty params with no schema should succeed");
1166 assert!(normalized.is_empty());
1167 }
1168
1169 #[test]
1170 fn normalize_empty_params_schema_rejects_extra_params() {
1171 let def = parse_workflow_definition(
1172 r#"{
1173 "workflowId": "flow-empty-params",
1174 "version": 1,
1175 "params": {},
1176 "nodes": {
1177 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1178 }
1179 }"#,
1180 )
1181 .unwrap();
1182 let input: BTreeMap<String, Value> =
1183 BTreeMap::from([(String::from("x"), Value::String("y".to_string()))]);
1184 let err = normalize_workflow_params(&def, &input).unwrap_err();
1185 let msg = err.to_string();
1186 assert!(msg.contains("unknown workflow parameter"), "got: {msg}");
1187 assert!(msg.contains("No parameters are declared"), "got: {msg}");
1188 }
1189
1190 #[test]
1191 fn normalize_empty_params_schema_with_empty_input_succeeds() {
1192 let def = parse_workflow_definition(
1193 r#"{
1194 "workflowId": "flow-empty-params",
1195 "version": 1,
1196 "params": {},
1197 "nodes": {
1198 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1199 }
1200 }"#,
1201 )
1202 .unwrap();
1203 let input: BTreeMap<String, Value> = BTreeMap::new();
1204 let normalized = normalize_workflow_params(&def, &input)
1205 .expect("empty params with empty schema should succeed");
1206 assert!(normalized.is_empty());
1207 }
1208
1209 #[test]
1210 fn normalize_default_value_type_mismatch_fails() {
1211 let def = parse_workflow_definition(
1213 r#"{
1214 "workflowId": "flow-bad-default",
1215 "version": 1,
1216 "params": {
1217 "enabled": {
1218 "type": "boolean",
1219 "required": false,
1220 "default": "not-a-bool"
1221 }
1222 },
1223 "nodes": {
1224 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1225 }
1226 }"#,
1227 )
1228 .unwrap();
1229 let input: BTreeMap<String, Value> = BTreeMap::new();
1230 let err = normalize_workflow_params(&def, &input).unwrap_err();
1231 let msg = err.to_string();
1232 assert!(msg.contains("enabled"), "got: {msg}");
1233 assert!(msg.contains("boolean"), "got: {msg}");
1234 }
1235
1236 #[test]
1237 fn normalize_unknown_param_type_fails() {
1238 let def = parse_workflow_definition(
1239 r#"{
1240 "workflowId": "flow-unknown-type",
1241 "version": 1,
1242 "params": {
1243 "x": {
1244 "type": "unknown-type-xyz",
1245 "required": false
1246 }
1247 },
1248 "nodes": {
1249 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1250 }
1251 }"#,
1252 )
1253 .unwrap();
1254 let input: BTreeMap<String, Value> =
1255 BTreeMap::from([(String::from("x"), Value::String("v".to_string()))]);
1256 let err = normalize_workflow_params(&def, &input).unwrap_err();
1257 let msg = err.to_string();
1258 assert!(msg.contains("unknown type"), "got: {msg}");
1259 assert!(msg.contains("x"), "got: {msg}");
1260 assert!(msg.contains("unknown-type-xyz"), "got: {msg}");
1261 }
1262
1263 #[test]
1264 fn normalize_integer_rejects_non_integer_number() {
1265 let def = parse_workflow_definition(
1266 r#"{
1267 "workflowId": "flow-int",
1268 "version": 1,
1269 "params": {
1270 "count": { "type": "integer" }
1271 },
1272 "nodes": {
1273 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1274 }
1275 }"#,
1276 )
1277 .unwrap();
1278 let input: BTreeMap<String, Value> =
1279 BTreeMap::from([(String::from("count"), serde_json::json!(3.5))]);
1280 let err = normalize_workflow_params(&def, &input).unwrap_err();
1281 let msg = err.to_string();
1282 assert!(msg.contains("integer"), "got: {msg}");
1283 assert!(msg.contains("count"), "got: {msg}");
1284 }
1285
1286 #[test]
1287 fn normalize_integer_accepts_integer_number() {
1288 let def = parse_workflow_definition(
1289 r#"{
1290 "workflowId": "flow-int-ok",
1291 "version": 1,
1292 "params": {
1293 "count": { "type": "integer" }
1294 },
1295 "nodes": {
1296 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1297 }
1298 }"#,
1299 )
1300 .unwrap();
1301 let input: BTreeMap<String, Value> =
1302 BTreeMap::from([(String::from("count"), serde_json::json!(42))]);
1303 let _ = normalize_workflow_params(&def, &input).expect("integer 42 OK");
1304 let input2: BTreeMap<String, Value> =
1306 BTreeMap::from([(String::from("count"), serde_json::json!(42.0))]);
1307 let _ = normalize_workflow_params(&def, &input2).expect("integer 42.0 OK");
1308 }
1309
1310 #[test]
1311 fn bootstrap_integration_rejects_unknown_key_with_params_schema() {
1312 let paths = temp_paths("bootstrap-unknown");
1313 let params: BTreeMap<String, Value> = BTreeMap::from([
1314 (String::from("task"), Value::String("hello".to_string())),
1315 (String::from("bad_key"), Value::String("x".to_string())),
1316 ]);
1317 let workflow_json = r#"{
1318 "workflowId": "flow-req",
1319 "version": 1,
1320 "params": {
1321 "task": {
1322 "type": "string",
1323 "required": true,
1324 "description": "what to do"
1325 }
1326 },
1327 "nodes": {
1328 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1329 }
1330 }"#;
1331 let err = bootstrap_workflow_run(
1332 &paths,
1333 BootstrapWorkflowRunInput {
1334 run_id: "run-unk",
1335 workflow_json,
1336 expected_workflow_id: Some("flow-req"),
1337 params: ¶ms,
1338 initiator: "test",
1339 chat_binding: None,
1340 },
1341 )
1342 .unwrap_err();
1343 let msg = err.to_string();
1344 assert!(msg.contains("unknown workflow parameter"), "got: {msg}");
1345 assert!(msg.contains("bad_key"), "got: {msg}");
1346 let run_dir = paths.workflow_run_dir("run-unk");
1347 assert!(!run_dir.exists(), "run dir should NOT exist");
1348 let _ = std::fs::remove_dir_all(paths.root());
1349 }
1350
1351 #[test]
1352 fn bootstrap_integration_default_written_to_params_blob() {
1353 let paths = temp_paths("bootstrap-default");
1354 let params: BTreeMap<String, Value> =
1355 BTreeMap::from([(String::from("task"), Value::String("build".to_string()))]);
1356 let workflow_json = r#"{
1357 "workflowId": "flow-def",
1358 "version": 1,
1359 "params": {
1360 "task": { "type": "string", "required": true },
1361 "verbose": { "type": "boolean", "required": false, "default": false }
1362 },
1363 "nodes": {
1364 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1365 }
1366 }"#;
1367 let result = bootstrap_workflow_run(
1368 &paths,
1369 BootstrapWorkflowRunInput {
1370 run_id: "run-def",
1371 workflow_json,
1372 expected_workflow_id: Some("flow-def"),
1373 params: ¶ms,
1374 initiator: "test",
1375 chat_binding: None,
1376 },
1377 )
1378 .expect("bootstrap with default param");
1379 assert_eq!(result.workflow_id, "flow-def");
1380 let blob_path = result.input_ref.output_path;
1382 let blob_bytes = std::fs::read(&blob_path).expect("read params blob");
1383 let blob: BTreeMap<String, Value> =
1384 serde_json::from_slice(&blob_bytes).expect("parse params blob");
1385 assert_eq!(blob.get("task"), Some(&Value::String("build".to_string())));
1386 assert_eq!(blob.get("verbose"), Some(&Value::Bool(false)));
1387 let _ = std::fs::remove_dir_all(paths.root());
1388 }
1389
1390 #[test]
1393 fn format_date_success() {
1394 let def = parse_workflow_definition(
1395 r#"{
1396 "workflowId": "flow-fmt",
1397 "version": 1,
1398 "params": {
1399 "d": { "type": "string", "format": "date" }
1400 },
1401 "nodes": {
1402 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1403 }
1404 }"#,
1405 )
1406 .unwrap();
1407 let input: BTreeMap<String, Value> =
1408 BTreeMap::from([(String::from("d"), Value::String("2024-02-29".to_string()))]);
1409 let _ = normalize_workflow_params(&def, &input).expect("leap date should pass");
1410 }
1411
1412 #[test]
1413 fn format_date_failure_invalid_date() {
1414 let def = parse_workflow_definition(
1415 r#"{
1416 "workflowId": "flow-fmt",
1417 "version": 1,
1418 "params": {
1419 "d": { "type": "string", "format": "date" }
1420 },
1421 "nodes": {
1422 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1423 }
1424 }"#,
1425 )
1426 .unwrap();
1427 let input: BTreeMap<String, Value> =
1428 BTreeMap::from([(String::from("d"), Value::String("2023-02-29".to_string()))]);
1429 let err = normalize_workflow_params(&def, &input).unwrap_err();
1430 let msg = err.to_string();
1431 assert!(msg.contains("date"), "got: {msg}");
1432 assert!(msg.contains("2023-02-29"), "got: {msg}");
1433 }
1434
1435 #[test]
1436 fn format_date_failure_invalid_month() {
1437 let def = parse_workflow_definition(
1438 r#"{
1439 "workflowId": "flow-fmt",
1440 "version": 1,
1441 "params": {
1442 "d": { "type": "string", "format": "date" }
1443 },
1444 "nodes": {
1445 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1446 }
1447 }"#,
1448 )
1449 .unwrap();
1450 let input: BTreeMap<String, Value> =
1451 BTreeMap::from([(String::from("d"), Value::String("2024-13-01".to_string()))]);
1452 let err = normalize_workflow_params(&def, &input).unwrap_err();
1453 let msg = err.to_string();
1454 assert!(msg.contains("date"), "got: {msg}");
1455 assert!(msg.contains("2024-13-01"), "got: {msg}");
1456 }
1457
1458 #[test]
1459 fn format_date_failure_wrong_format() {
1460 let def = parse_workflow_definition(
1461 r#"{
1462 "workflowId": "flow-fmt",
1463 "version": 1,
1464 "params": {
1465 "d": { "type": "string", "format": "date" }
1466 },
1467 "nodes": {
1468 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1469 }
1470 }"#,
1471 )
1472 .unwrap();
1473 let input: BTreeMap<String, Value> =
1474 BTreeMap::from([(String::from("d"), Value::String("01-01-2024".to_string()))]);
1475 let err = normalize_workflow_params(&def, &input).unwrap_err();
1476 let msg = err.to_string();
1477 assert!(msg.contains("date"), "got: {msg}");
1478 }
1479
1480 #[test]
1481 fn format_date_time_success() {
1482 let def = parse_workflow_definition(
1483 r#"{
1484 "workflowId": "flow-fmt",
1485 "version": 1,
1486 "params": {
1487 "ts": { "type": "string", "format": "date-time" }
1488 },
1489 "nodes": {
1490 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1491 }
1492 }"#,
1493 )
1494 .unwrap();
1495 let input: BTreeMap<String, Value> = BTreeMap::from([(
1496 String::from("ts"),
1497 Value::String("2026-06-17T12:34:56Z".to_string()),
1498 )]);
1499 let _ = normalize_workflow_params(&def, &input).expect("RFC3339 should pass");
1500 let input2: BTreeMap<String, Value> = BTreeMap::from([(
1502 String::from("ts"),
1503 Value::String("2026-06-17T12:34:56+08:00".to_string()),
1504 )]);
1505 let _ = normalize_workflow_params(&def, &input2).expect("RFC3339 with offset should pass");
1506 }
1507
1508 #[test]
1509 fn format_date_time_failure() {
1510 let def = parse_workflow_definition(
1511 r#"{
1512 "workflowId": "flow-fmt",
1513 "version": 1,
1514 "params": {
1515 "ts": { "type": "string", "format": "date-time" }
1516 },
1517 "nodes": {
1518 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1519 }
1520 }"#,
1521 )
1522 .unwrap();
1523 let input: BTreeMap<String, Value> = BTreeMap::from([(
1524 String::from("ts"),
1525 Value::String("2026-06-17 12:34:56".to_string()),
1526 )]);
1527 let err = normalize_workflow_params(&def, &input).unwrap_err();
1528 let msg = err.to_string();
1529 assert!(msg.contains("date-time"), "got: {msg}");
1530 }
1531
1532 #[test]
1533 fn format_email_success() {
1534 let def = parse_workflow_definition(
1535 r#"{
1536 "workflowId": "flow-fmt",
1537 "version": 1,
1538 "params": {
1539 "email": { "type": "string", "format": "email" }
1540 },
1541 "nodes": {
1542 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1543 }
1544 }"#,
1545 )
1546 .unwrap();
1547 let input: BTreeMap<String, Value> = BTreeMap::from([(
1548 String::from("email"),
1549 Value::String("user@example.com".to_string()),
1550 )]);
1551 let _ = normalize_workflow_params(&def, &input).expect("valid email should pass");
1552 }
1553
1554 #[test]
1555 fn format_email_failure_no_at() {
1556 let def = parse_workflow_definition(
1557 r#"{
1558 "workflowId": "flow-fmt",
1559 "version": 1,
1560 "params": {
1561 "email": { "type": "string", "format": "email" }
1562 },
1563 "nodes": {
1564 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1565 }
1566 }"#,
1567 )
1568 .unwrap();
1569 let input: BTreeMap<String, Value> = BTreeMap::from([(
1570 String::from("email"),
1571 Value::String("notanemail".to_string()),
1572 )]);
1573 let err = normalize_workflow_params(&def, &input).unwrap_err();
1574 let msg = err.to_string();
1575 assert!(msg.contains("email"), "got: {msg}");
1576 }
1577
1578 #[test]
1579 fn format_email_failure_double_at() {
1580 let def = parse_workflow_definition(
1581 r#"{
1582 "workflowId": "flow-fmt",
1583 "version": 1,
1584 "params": {
1585 "email": { "type": "string", "format": "email" }
1586 },
1587 "nodes": {
1588 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1589 }
1590 }"#,
1591 )
1592 .unwrap();
1593 let input: BTreeMap<String, Value> = BTreeMap::from([(
1594 String::from("email"),
1595 Value::String("a@b@c.com".to_string()),
1596 )]);
1597 let err = normalize_workflow_params(&def, &input).unwrap_err();
1598 let msg = err.to_string();
1599 assert!(msg.contains("email"), "got: {msg}");
1600 }
1601
1602 #[test]
1603 fn format_email_failure_empty_local() {
1604 let def = parse_workflow_definition(
1605 r#"{
1606 "workflowId": "flow-fmt",
1607 "version": 1,
1608 "params": {
1609 "email": { "type": "string", "format": "email" }
1610 },
1611 "nodes": {
1612 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1613 }
1614 }"#,
1615 )
1616 .unwrap();
1617 let input: BTreeMap<String, Value> = BTreeMap::from([(
1618 String::from("email"),
1619 Value::String("@example.com".to_string()),
1620 )]);
1621 let err = normalize_workflow_params(&def, &input).unwrap_err();
1622 let msg = err.to_string();
1623 assert!(msg.contains("email"), "got: {msg}");
1624 }
1625
1626 #[test]
1627 fn format_email_failure_no_dot_in_domain() {
1628 let def = parse_workflow_definition(
1629 r#"{
1630 "workflowId": "flow-fmt",
1631 "version": 1,
1632 "params": {
1633 "email": { "type": "string", "format": "email" }
1634 },
1635 "nodes": {
1636 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1637 }
1638 }"#,
1639 )
1640 .unwrap();
1641 let input: BTreeMap<String, Value> = BTreeMap::from([(
1642 String::from("email"),
1643 Value::String("user@localhost".to_string()),
1644 )]);
1645 let err = normalize_workflow_params(&def, &input).unwrap_err();
1646 let msg = err.to_string();
1647 assert!(msg.contains("dot"), "got: {msg}");
1648 }
1649
1650 #[test]
1651 fn format_email_failure_whitespace() {
1652 let def = parse_workflow_definition(
1653 r#"{
1654 "workflowId": "flow-fmt",
1655 "version": 1,
1656 "params": {
1657 "email": { "type": "string", "format": "email" }
1658 },
1659 "nodes": {
1660 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1661 }
1662 }"#,
1663 )
1664 .unwrap();
1665 let input: BTreeMap<String, Value> = BTreeMap::from([(
1666 String::from("email"),
1667 Value::String("user @example.com".to_string()),
1668 )]);
1669 let err = normalize_workflow_params(&def, &input).unwrap_err();
1670 let msg = err.to_string();
1671 assert!(msg.contains("whitespace"), "got: {msg}");
1672 }
1673
1674 #[test]
1675 fn format_on_non_string_type_fails() {
1676 let def = parse_workflow_definition(
1677 r#"{
1678 "workflowId": "flow-fmt-nonstring",
1679 "version": 1,
1680 "params": {
1681 "count": { "type": "integer", "format": "date" }
1682 },
1683 "nodes": {
1684 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1685 }
1686 }"#,
1687 )
1688 .unwrap();
1689 let input: BTreeMap<String, Value> =
1690 BTreeMap::from([(String::from("count"), serde_json::json!(42))]);
1691 let err = normalize_workflow_params(&def, &input).unwrap_err();
1692 let msg = err.to_string();
1693 assert!(msg.contains("count"), "got: {msg}");
1694 assert!(msg.contains("format"), "got: {msg}");
1695 assert!(msg.contains("integer"), "got: {msg}");
1696 assert!(msg.contains("string"), "got: {msg}");
1697 }
1698
1699 #[test]
1700 fn format_unknown_format_fails() {
1701 let def = parse_workflow_definition(
1702 r#"{
1703 "workflowId": "flow-fmt-unknown",
1704 "version": 1,
1705 "params": {
1706 "x": { "type": "string", "format": "unknown-format-xyz" }
1707 },
1708 "nodes": {
1709 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1710 }
1711 }"#,
1712 )
1713 .unwrap();
1714 let input: BTreeMap<String, Value> =
1715 BTreeMap::from([(String::from("x"), Value::String("v".to_string()))]);
1716 let err = normalize_workflow_params(&def, &input).unwrap_err();
1717 let msg = err.to_string();
1718 assert!(msg.contains("unknown format"), "got: {msg}");
1719 assert!(msg.contains("x"), "got: {msg}");
1720 assert!(msg.contains("unknown-format-xyz"), "got: {msg}");
1721 }
1722
1723 #[test]
1726 fn coerce_string_to_boolean_true_and_false() {
1727 let def = parse_workflow_definition(
1728 r#"{
1729 "workflowId": "flow-coerce",
1730 "version": 1,
1731 "params": {
1732 "enabled": { "type": "boolean" }
1733 },
1734 "nodes": {
1735 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1736 }
1737 }"#,
1738 )
1739 .unwrap();
1740
1741 let input: BTreeMap<String, Value> =
1743 BTreeMap::from([(String::from("enabled"), Value::String("true".to_string()))]);
1744 let normalized = normalize_workflow_params(&def, &input).expect("true coerces to bool");
1745 assert_eq!(normalized.get("enabled"), Some(&Value::Bool(true)));
1746
1747 let input2: BTreeMap<String, Value> =
1749 BTreeMap::from([(String::from("enabled"), Value::String("false".to_string()))]);
1750 let normalized2 = normalize_workflow_params(&def, &input2).expect("false coerces to bool");
1751 assert_eq!(normalized2.get("enabled"), Some(&Value::Bool(false)));
1752 }
1753
1754 #[test]
1755 fn coerce_boolean_rejects_case_insensitive_and_junk() {
1756 let def = parse_workflow_definition(
1757 r#"{
1758 "workflowId": "flow-coerce",
1759 "version": 1,
1760 "params": {
1761 "enabled": { "type": "boolean" }
1762 },
1763 "nodes": {
1764 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1765 }
1766 }"#,
1767 )
1768 .unwrap();
1769
1770 for bad in &["True", "TRUE", "yes", "1", "0", "FALSE", "on", "off"] {
1771 let input: BTreeMap<String, Value> =
1772 BTreeMap::from([(String::from("enabled"), Value::String(bad.to_string()))]);
1773 let err = normalize_workflow_params(&def, &input).unwrap_err();
1774 let msg = err.to_string();
1775 assert!(
1776 msg.contains("boolean"),
1777 "expected 'boolean' in error for input '{}', got: {}",
1778 bad,
1779 msg
1780 );
1781 assert!(
1782 msg.contains("enabled"),
1783 "expected 'enabled' in error for input '{}', got: {}",
1784 bad,
1785 msg
1786 );
1787 }
1788 }
1789
1790 #[test]
1791 fn coerce_boolean_accepts_already_typed_value() {
1792 let def = parse_workflow_definition(
1793 r#"{
1794 "workflowId": "flow-coerce",
1795 "version": 1,
1796 "params": {
1797 "enabled": { "type": "boolean" }
1798 },
1799 "nodes": {
1800 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1801 }
1802 }"#,
1803 )
1804 .unwrap();
1805
1806 let input: BTreeMap<String, Value> =
1808 BTreeMap::from([(String::from("enabled"), Value::Bool(true))]);
1809 let normalized = normalize_workflow_params(&def, &input).expect("bool passthrough");
1810 assert_eq!(normalized.get("enabled"), Some(&Value::Bool(true)));
1811 }
1812
1813 #[test]
1814 fn coerce_number_from_strings() {
1815 let def = parse_workflow_definition(
1816 r#"{
1817 "workflowId": "flow-coerce",
1818 "version": 1,
1819 "params": {
1820 "ratio": { "type": "number" }
1821 },
1822 "nodes": {
1823 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1824 }
1825 }"#,
1826 )
1827 .unwrap();
1828
1829 let cases: Vec<(&str, Value)> = vec![
1830 ("1", serde_json::json!(1)),
1831 ("1.5", serde_json::json!(1.5)),
1832 ("-2", serde_json::json!(-2)),
1833 ("1e3", serde_json::json!(1000.0)),
1834 ];
1835 for (input_str, expected) in &cases {
1836 let input: BTreeMap<String, Value> =
1837 BTreeMap::from([(String::from("ratio"), Value::String(input_str.to_string()))]);
1838 let normalized =
1839 normalize_workflow_params(&def, &input).expect("number string should coerce");
1840 assert_eq!(
1841 normalized.get("ratio"),
1842 Some(expected),
1843 "failed for input '{}'",
1844 input_str
1845 );
1846 }
1847 }
1848
1849 #[test]
1850 fn coerce_number_rejects_junk() {
1851 let def = parse_workflow_definition(
1852 r#"{
1853 "workflowId": "flow-coerce",
1854 "version": 1,
1855 "params": {
1856 "ratio": { "type": "number" }
1857 },
1858 "nodes": {
1859 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1860 }
1861 }"#,
1862 )
1863 .unwrap();
1864
1865 for bad in &[
1866 "NaN",
1867 "Infinity",
1868 "-Infinity",
1869 "not-a-number",
1870 "true",
1871 "\"hi\"",
1872 "[]",
1873 ] {
1874 let input: BTreeMap<String, Value> =
1875 BTreeMap::from([(String::from("ratio"), Value::String(bad.to_string()))]);
1876 let err = normalize_workflow_params(&def, &input).unwrap_err();
1877 let msg = err.to_string();
1878 assert!(
1879 msg.contains("number") || msg.contains("number"),
1880 "expected 'number' in error for input '{}', got: {}",
1881 bad,
1882 msg
1883 );
1884 }
1885 }
1886
1887 #[test]
1888 fn coerce_integer_from_decimal_strings() {
1889 let def = parse_workflow_definition(
1890 r#"{
1891 "workflowId": "flow-coerce",
1892 "version": 1,
1893 "params": {
1894 "count": { "type": "integer" }
1895 },
1896 "nodes": {
1897 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1898 }
1899 }"#,
1900 )
1901 .unwrap();
1902
1903 let cases: Vec<(&str, Value)> = vec![
1904 ("42", serde_json::json!(42)),
1905 ("-1", serde_json::json!(-1)),
1906 ("0", serde_json::json!(0)),
1907 ];
1908 for (input_str, expected) in &cases {
1909 let input: BTreeMap<String, Value> =
1910 BTreeMap::from([(String::from("count"), Value::String(input_str.to_string()))]);
1911 let normalized =
1912 normalize_workflow_params(&def, &input).expect("integer string should coerce");
1913 assert_eq!(
1914 normalized.get("count"),
1915 Some(expected),
1916 "failed for input '{}'",
1917 input_str
1918 );
1919 }
1920 }
1921
1922 #[test]
1923 fn coerce_integer_with_whitespace_padding() {
1924 let def = parse_workflow_definition(
1925 r#"{
1926 "workflowId": "flow-coerce",
1927 "version": 1,
1928 "params": {
1929 "count": { "type": "integer" }
1930 },
1931 "nodes": {
1932 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1933 }
1934 }"#,
1935 )
1936 .unwrap();
1937
1938 let input: BTreeMap<String, Value> =
1940 BTreeMap::from([(String::from("count"), Value::String(" 42 ".to_string()))]);
1941 let normalized =
1942 normalize_workflow_params(&def, &input).expect("padded integer should coerce");
1943 assert_eq!(normalized.get("count"), Some(&serde_json::json!(42)));
1944 }
1945
1946 #[test]
1947 fn coerce_integer_rejects_non_decimal_formats() {
1948 let def = parse_workflow_definition(
1949 r#"{
1950 "workflowId": "flow-coerce",
1951 "version": 1,
1952 "params": {
1953 "count": { "type": "integer" }
1954 },
1955 "nodes": {
1956 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1957 }
1958 }"#,
1959 )
1960 .unwrap();
1961
1962 for bad in &["1.0", "1.5", "1e3", "0x10", "3.0"] {
1963 let input: BTreeMap<String, Value> =
1964 BTreeMap::from([(String::from("count"), Value::String(bad.to_string()))]);
1965 let err = normalize_workflow_params(&def, &input).unwrap_err();
1966 let msg = err.to_string();
1967 assert!(
1968 msg.contains("integer"),
1969 "expected 'integer' in error for input '{}', got: {}",
1970 bad,
1971 msg
1972 );
1973 }
1974 }
1975
1976 #[test]
1977 fn coerce_integer_accepts_already_typed_value() {
1978 let def = parse_workflow_definition(
1979 r#"{
1980 "workflowId": "flow-coerce",
1981 "version": 1,
1982 "params": {
1983 "count": { "type": "integer" }
1984 },
1985 "nodes": {
1986 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
1987 }
1988 }"#,
1989 )
1990 .unwrap();
1991
1992 let input: BTreeMap<String, Value> =
1994 BTreeMap::from([(String::from("count"), serde_json::json!(42))]);
1995 let normalized = normalize_workflow_params(&def, &input).expect("integer passthrough");
1996 assert_eq!(normalized.get("count"), Some(&serde_json::json!(42)));
1997 }
1998
1999 #[test]
2000 fn coerce_object_from_json_string() {
2001 let def = parse_workflow_definition(
2002 r#"{
2003 "workflowId": "flow-coerce",
2004 "version": 1,
2005 "params": {
2006 "payload": { "type": "object" }
2007 },
2008 "nodes": {
2009 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
2010 }
2011 }"#,
2012 )
2013 .unwrap();
2014
2015 let input: BTreeMap<String, Value> = BTreeMap::from([(
2016 String::from("payload"),
2017 Value::String(r#"{"a":1,"b":"x"}"#.to_string()),
2018 )]);
2019 let normalized =
2020 normalize_workflow_params(&def, &input).expect("object string should coerce");
2021 assert_eq!(
2022 normalized.get("payload"),
2023 Some(&serde_json::json!({"a":1,"b":"x"}))
2024 );
2025 }
2026
2027 #[test]
2028 fn coerce_array_from_json_string() {
2029 let def = parse_workflow_definition(
2030 r#"{
2031 "workflowId": "flow-coerce",
2032 "version": 1,
2033 "params": {
2034 "tags": { "type": "array" }
2035 },
2036 "nodes": {
2037 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
2038 }
2039 }"#,
2040 )
2041 .unwrap();
2042
2043 let input: BTreeMap<String, Value> = BTreeMap::from([(
2044 String::from("tags"),
2045 Value::String(r#"["a","b"]"#.to_string()),
2046 )]);
2047 let normalized =
2048 normalize_workflow_params(&def, &input).expect("array string should coerce");
2049 assert_eq!(normalized.get("tags"), Some(&serde_json::json!(["a", "b"])));
2050 }
2051
2052 #[test]
2053 fn coerce_object_rejects_non_object_json() {
2054 let def = parse_workflow_definition(
2055 r#"{
2056 "workflowId": "flow-coerce",
2057 "version": 1,
2058 "params": {
2059 "payload": { "type": "object" }
2060 },
2061 "nodes": {
2062 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
2063 }
2064 }"#,
2065 )
2066 .unwrap();
2067
2068 let input: BTreeMap<String, Value> = BTreeMap::from([(
2070 String::from("payload"),
2071 Value::String(r#"["a"]"#.to_string()),
2072 )]);
2073 let err = normalize_workflow_params(&def, &input).unwrap_err();
2074 let msg = err.to_string();
2075 assert!(msg.contains("object"), "got: {msg}");
2076 assert!(msg.contains("payload"), "got: {msg}");
2077 assert!(msg.contains("array"), "got: {msg}");
2078 }
2079
2080 #[test]
2081 fn coerce_array_rejects_non_array_json() {
2082 let def = parse_workflow_definition(
2083 r#"{
2084 "workflowId": "flow-coerce",
2085 "version": 1,
2086 "params": {
2087 "tags": { "type": "array" }
2088 },
2089 "nodes": {
2090 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
2091 }
2092 }"#,
2093 )
2094 .unwrap();
2095
2096 let input: BTreeMap<String, Value> =
2098 BTreeMap::from([(String::from("tags"), Value::String("1".to_string()))]);
2099 let err = normalize_workflow_params(&def, &input).unwrap_err();
2100 let msg = err.to_string();
2101 assert!(msg.contains("array"), "got: {msg}");
2102 assert!(msg.contains("tags"), "got: {msg}");
2103 }
2104
2105 #[test]
2106 fn coerce_object_invalid_json_fails() {
2107 let def = parse_workflow_definition(
2108 r#"{
2109 "workflowId": "flow-coerce",
2110 "version": 1,
2111 "params": {
2112 "payload": { "type": "object" }
2113 },
2114 "nodes": {
2115 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
2116 }
2117 }"#,
2118 )
2119 .unwrap();
2120
2121 let input: BTreeMap<String, Value> = BTreeMap::from([(
2122 String::from("payload"),
2123 Value::String("not-json".to_string()),
2124 )]);
2125 let err = normalize_workflow_params(&def, &input).unwrap_err();
2126 let msg = err.to_string();
2127 assert!(msg.contains("object"), "got: {msg}");
2128 assert!(msg.contains("valid JSON"), "got: {msg}");
2129 }
2130
2131 #[test]
2132 fn coerce_default_not_coerced_for_boolean() {
2133 let def = parse_workflow_definition(
2136 r#"{
2137 "workflowId": "flow-coerce",
2138 "version": 1,
2139 "params": {
2140 "enabled": {
2141 "type": "boolean",
2142 "required": false,
2143 "default": "true"
2144 }
2145 },
2146 "nodes": {
2147 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
2148 }
2149 }"#,
2150 )
2151 .unwrap();
2152
2153 let input: BTreeMap<String, Value> = BTreeMap::new(); let err = normalize_workflow_params(&def, &input).unwrap_err();
2155 let msg = err.to_string();
2156 assert!(
2157 msg.contains("boolean"),
2158 "default string 'true' should fail type check for boolean, got: {}",
2159 msg
2160 );
2161 }
2162
2163 #[test]
2164 fn coerce_string_param_unchanged() {
2165 let def = parse_workflow_definition(
2167 r#"{
2168 "workflowId": "flow-coerce",
2169 "version": 1,
2170 "params": {
2171 "name": { "type": "string" }
2172 },
2173 "nodes": {
2174 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
2175 }
2176 }"#,
2177 )
2178 .unwrap();
2179
2180 let input: BTreeMap<String, Value> =
2181 BTreeMap::from([(String::from("name"), Value::String("true".to_string()))]);
2182 let normalized =
2183 normalize_workflow_params(&def, &input).expect("string param should not be coerced");
2184 assert_eq!(
2186 normalized.get("name"),
2187 Some(&Value::String("true".to_string()))
2188 );
2189 }
2190
2191 #[test]
2192 fn coerce_boolean_with_whitespace_padding() {
2193 let def = parse_workflow_definition(
2194 r#"{
2195 "workflowId": "flow-coerce",
2196 "version": 1,
2197 "params": {
2198 "enabled": { "type": "boolean" }
2199 },
2200 "nodes": {
2201 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
2202 }
2203 }"#,
2204 )
2205 .unwrap();
2206
2207 let input: BTreeMap<String, Value> = BTreeMap::from([(
2209 String::from("enabled"),
2210 Value::String(" true ".to_string()),
2211 )]);
2212 let normalized =
2213 normalize_workflow_params(&def, &input).expect("padded true coerces to bool");
2214 assert_eq!(normalized.get("enabled"), Some(&Value::Bool(true)));
2215 }
2216
2217 #[test]
2218 fn coerce_empty_string_to_boolean_fails() {
2219 let def = parse_workflow_definition(
2220 r#"{
2221 "workflowId": "flow-coerce",
2222 "version": 1,
2223 "params": {
2224 "enabled": { "type": "boolean" }
2225 },
2226 "nodes": {
2227 "a": { "type": "subagent", "bot": "bot-a", "prompt": "hi" }
2228 }
2229 }"#,
2230 )
2231 .unwrap();
2232
2233 let input: BTreeMap<String, Value> =
2234 BTreeMap::from([(String::from("enabled"), Value::String("".to_string()))]);
2235 let err = normalize_workflow_params(&def, &input).unwrap_err();
2236 let msg = err.to_string();
2237 assert!(msg.contains("boolean"), "got: {msg}");
2238 assert!(msg.contains("empty"), "got: {msg}");
2239 }
2240}