1use async_trait::async_trait;
2use serde::{Deserialize, Serialize};
3
4use crate::CamelError;
5use crate::declarative::LanguageExpressionDef;
6use crate::splitter::StreamSplitConfig;
7
8pub const CANONICAL_CONTRACT_NAME: &str = "canonical-v1";
9pub const CANONICAL_CONTRACT_VERSION: u32 = 2;
10pub const CANONICAL_CONTRACT_SUPPORTED_STEPS: &[&str] = &[
11 "to",
12 "log",
13 "wire_tap",
14 "script",
15 "filter",
16 "choice",
17 "split",
18 "aggregate",
19 "stop",
20 "delay",
21];
22pub const CANONICAL_CONTRACT_DECLARATIVE_ONLY_STEPS: &[&str] =
23 &["script", "filter", "choice", "split"];
24pub const CANONICAL_CONTRACT_EXCLUDED_DECLARATIVE_STEPS: &[&str] = &[
25 "set_header",
26 "set_property",
27 "set_body",
28 "multicast",
29 "convert_body_to",
30 "bean",
31 "marshal",
32 "unmarshal",
33];
34pub const CANONICAL_CONTRACT_RUST_ONLY_STEPS: &[&str] = &[
35 "processor",
36 "process",
37 "process_fn",
38 "map_body",
39 "set_body_fn",
40 "set_header_fn",
41];
42
43pub fn canonical_contract_supports_step(step: &str) -> bool {
44 CANONICAL_CONTRACT_SUPPORTED_STEPS.contains(&step)
45}
46
47pub fn canonical_contract_rejection_reason(step: &str) -> Option<&'static str> {
48 if CANONICAL_CONTRACT_EXCLUDED_DECLARATIVE_STEPS.contains(&step) {
49 return Some(
50 "declared out-of-scope for canonical v1; use declarative route compilation path outside CQRS canonical commands",
51 );
52 }
53
54 if CANONICAL_CONTRACT_RUST_ONLY_STEPS.contains(&step) {
55 return Some("rust-only programmable step; not representable in canonical v1 contract");
56 }
57
58 if canonical_contract_supports_step(step)
59 && CANONICAL_CONTRACT_DECLARATIVE_ONLY_STEPS.contains(&step)
60 {
61 return Some(
62 "supported only as declarative/serializable expression form; closure/processor variants are outside canonical v1",
63 );
64 }
65
66 None
67}
68
69#[derive(
70 Debug,
71 Clone,
72 PartialEq,
73 Eq,
74 serde::Serialize,
75 serde::Deserialize,
76 schemars::JsonSchema,
77 ts_rs::TS,
78)]
79#[serde(rename_all = "snake_case")]
80#[ts(rename_all = "snake_case")]
81pub struct CanonicalRouteSpec {
82 pub route_id: String,
92 pub from: String,
93 pub steps: Vec<CanonicalStepSpec>,
94 pub circuit_breaker: Option<CanonicalCircuitBreakerSpec>,
95 pub auto_startup: Option<bool>,
96 pub startup_order: Option<i32>,
97 pub concurrency: Option<CanonicalConcurrencySpec>,
98 pub version: u32,
99}
100
101#[derive(
102 Debug,
103 Clone,
104 PartialEq,
105 Eq,
106 serde::Serialize,
107 serde::Deserialize,
108 schemars::JsonSchema,
109 ts_rs::TS,
110)]
111#[serde(tag = "step", content = "config", rename_all = "snake_case")]
112#[ts(rename_all = "snake_case")]
113pub enum CanonicalStepSpec {
114 To {
115 uri: String,
116 },
117 Log {
118 message: String,
119 },
120 WireTap {
121 uri: String,
122 },
123 Script {
124 expression: LanguageExpressionDef,
125 },
126 Filter {
127 predicate: LanguageExpressionDef,
128 steps: Vec<CanonicalStepSpec>,
129 },
130 Choice {
131 whens: Vec<CanonicalWhenSpec>,
132 otherwise: Option<Vec<CanonicalStepSpec>>,
133 },
134 Split {
135 expression: CanonicalSplitExpressionSpec,
136 aggregation: CanonicalSplitAggregationSpec,
137 parallel: bool,
138 parallel_limit: Option<usize>,
139 stop_on_exception: bool,
140 steps: Vec<CanonicalStepSpec>,
141 },
142 Aggregate(CanonicalAggregateSpec),
143 Stop,
144 Delay {
145 #[ts(type = "number")]
146 delay_ms: u64,
147 dynamic_header: Option<String>,
148 },
149}
150
151#[derive(
152 Debug,
153 Clone,
154 PartialEq,
155 Eq,
156 serde::Serialize,
157 serde::Deserialize,
158 schemars::JsonSchema,
159 ts_rs::TS,
160)]
161#[serde(rename_all = "snake_case")]
162#[ts(rename_all = "snake_case")]
163pub struct CanonicalWhenSpec {
164 pub predicate: LanguageExpressionDef,
165 pub steps: Vec<CanonicalStepSpec>,
166}
167
168#[derive(
169 Debug,
170 Clone,
171 PartialEq,
172 Eq,
173 serde::Serialize,
174 serde::Deserialize,
175 schemars::JsonSchema,
176 ts_rs::TS,
177)]
178#[serde(rename_all = "snake_case")]
179#[ts(rename_all = "snake_case")]
180pub enum CanonicalSplitExpressionSpec {
181 BodyLines,
182 BodyJsonArray,
183 Language(LanguageExpressionDef),
184 Stream(StreamSplitConfig),
185}
186
187#[derive(
188 Debug,
189 Clone,
190 PartialEq,
191 Eq,
192 serde::Serialize,
193 serde::Deserialize,
194 schemars::JsonSchema,
195 ts_rs::TS,
196)]
197#[serde(rename_all = "snake_case")]
198#[ts(rename_all = "snake_case")]
199pub enum CanonicalSplitAggregationSpec {
200 LastWins,
201 CollectAll,
202 Original,
203}
204
205#[derive(
206 Debug,
207 Clone,
208 PartialEq,
209 Eq,
210 serde::Serialize,
211 serde::Deserialize,
212 schemars::JsonSchema,
213 ts_rs::TS,
214)]
215#[serde(rename_all = "snake_case")]
216#[ts(rename_all = "snake_case")]
217pub enum CanonicalAggregateStrategySpec {
218 CollectAll,
219}
220
221#[derive(
222 Debug,
223 Clone,
224 PartialEq,
225 Eq,
226 serde::Serialize,
227 serde::Deserialize,
228 schemars::JsonSchema,
229 ts_rs::TS,
230)]
231#[serde(rename_all = "snake_case")]
232#[ts(rename_all = "snake_case")]
233pub struct CanonicalAggregateSpec {
234 pub header: String,
235 pub completion_size: Option<usize>,
236 #[ts(type = "number")]
237 pub completion_timeout_ms: Option<u64>,
238 pub correlation_key: Option<String>,
239 pub force_completion_on_stop: Option<bool>,
240 pub discard_on_timeout: Option<bool>,
241 pub strategy: CanonicalAggregateStrategySpec,
242 pub max_buckets: Option<usize>,
243 #[ts(type = "number")]
244 pub bucket_ttl_ms: Option<u64>,
245}
246
247#[derive(
248 Debug,
249 Clone,
250 PartialEq,
251 Eq,
252 serde::Serialize,
253 serde::Deserialize,
254 schemars::JsonSchema,
255 ts_rs::TS,
256)]
257#[serde(rename_all = "snake_case")]
258#[ts(rename_all = "snake_case")]
259pub struct CanonicalCircuitBreakerSpec {
260 pub failure_threshold: u32,
261 #[ts(type = "number")]
262 pub open_duration_ms: u64,
263}
264
265#[derive(
266 Debug,
267 Clone,
268 PartialEq,
269 Eq,
270 serde::Serialize,
271 serde::Deserialize,
272 schemars::JsonSchema,
273 ts_rs::TS,
274)]
275#[serde(tag = "mode", rename_all = "snake_case")]
276pub enum CanonicalConcurrencySpec {
277 Sequential,
278 Concurrent { max: usize },
279}
280
281impl CanonicalRouteSpec {
282 pub fn new(route_id: impl Into<String>, from: impl Into<String>) -> Self {
283 Self {
284 route_id: route_id.into(),
285 from: from.into(),
286 steps: Vec::new(),
287 circuit_breaker: None,
288 auto_startup: None,
289 startup_order: None,
290 concurrency: None,
291 version: CANONICAL_CONTRACT_VERSION,
292 }
293 }
294
295 pub fn with_auto_startup(mut self, auto: bool) -> Self {
296 self.auto_startup = Some(auto);
297 self
298 }
299
300 pub fn with_startup_order(mut self, order: i32) -> Self {
301 self.startup_order = Some(order);
302 self
303 }
304
305 pub fn with_concurrency(mut self, concurrency: CanonicalConcurrencySpec) -> Self {
306 self.concurrency = Some(concurrency);
307 self
308 }
309
310 pub fn validate_contract(&self) -> Result<(), CamelError> {
311 if self.route_id.trim().is_empty() {
312 return Err(CamelError::RouteError(
313 "canonical contract violation: route_id cannot be empty".to_string(),
314 ));
315 }
316 if self.from.trim().is_empty() {
317 return Err(CamelError::RouteError(
318 "canonical contract violation: from cannot be empty".to_string(),
319 ));
320 }
321 if self.version == 0 || self.version > CANONICAL_CONTRACT_VERSION {
322 return Err(CamelError::RouteError(format!(
323 "canonical contract violation: expected version {}, got {}",
324 CANONICAL_CONTRACT_VERSION, self.version
325 )));
326 }
327 validate_steps(&self.steps)?;
328 if let Some(cb) = &self.circuit_breaker {
329 if cb.failure_threshold == 0 {
330 return Err(CamelError::RouteError(
331 "canonical contract violation: circuit_breaker.failure_threshold must be > 0"
332 .to_string(),
333 ));
334 }
335 if cb.open_duration_ms == 0 {
336 return Err(CamelError::RouteError(
337 "canonical contract violation: circuit_breaker.open_duration_ms must be > 0"
338 .to_string(),
339 ));
340 }
341 }
342 if let Some(CanonicalConcurrencySpec::Concurrent { max: 0 }) = &self.concurrency {
343 return Err(CamelError::RouteError(
344 "canonical contract violation: concurrency max must be > 0".to_string(),
345 ));
346 }
347 Ok(())
348 }
349}
350
351#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
352pub struct CanonicalFieldLoss {
353 pub field: &'static str,
354 pub reason: String,
355 pub target_version: u32,
356}
357
358#[derive(Debug, Clone, PartialEq, Eq, Default, serde::Serialize)]
359pub struct CanonicalLossReport {
360 pub dropped_fields: Vec<CanonicalFieldLoss>,
361}
362
363impl CanonicalLossReport {
364 pub fn from_field(field: &'static str, reason: &str, target_version: u32) -> Self {
365 Self {
366 dropped_fields: vec![CanonicalFieldLoss {
367 field,
368 reason: reason.to_string(),
369 target_version,
370 }],
371 }
372 }
373
374 pub fn is_empty(&self) -> bool {
375 self.dropped_fields.is_empty()
376 }
377}
378
379fn validate_steps(steps: &[CanonicalStepSpec]) -> Result<(), CamelError> {
380 for step in steps {
381 match step {
382 CanonicalStepSpec::To { uri } | CanonicalStepSpec::WireTap { uri } => {
383 if uri.trim().is_empty() {
384 return Err(CamelError::RouteError(
385 "canonical contract violation: endpoint uri cannot be empty".to_string(),
386 ));
387 }
388 }
389 CanonicalStepSpec::Filter { steps, .. } => validate_steps(steps)?,
390 CanonicalStepSpec::Choice { whens, otherwise } => {
391 for when in whens {
392 validate_steps(&when.steps)?;
393 }
394 if let Some(otherwise) = otherwise {
395 validate_steps(otherwise)?;
396 }
397 }
398 CanonicalStepSpec::Split {
399 parallel_limit,
400 steps,
401 ..
402 } => {
403 if let Some(limit) = parallel_limit
404 && *limit == 0
405 {
406 return Err(CamelError::RouteError(
407 "canonical contract violation: split.parallel_limit must be > 0"
408 .to_string(),
409 ));
410 }
411 validate_steps(steps)?;
412 }
413 CanonicalStepSpec::Aggregate(config) => {
414 if config.header.trim().is_empty() {
415 return Err(CamelError::RouteError(
416 "canonical contract violation: aggregate.header cannot be empty"
417 .to_string(),
418 ));
419 }
420 if let Some(size) = config.completion_size
421 && size == 0
422 {
423 return Err(CamelError::RouteError(
424 "canonical contract violation: aggregate.completion_size must be > 0"
425 .to_string(),
426 ));
427 }
428 }
429 CanonicalStepSpec::Log { .. }
430 | CanonicalStepSpec::Script { .. }
431 | CanonicalStepSpec::Stop
432 | CanonicalStepSpec::Delay { .. } => {}
433 }
434 }
435 Ok(())
436}
437
438#[derive(Debug, Clone, PartialEq, Eq)]
439pub enum RuntimeCommand {
440 RegisterRoute {
441 spec: CanonicalRouteSpec,
442 command_id: String,
443 causation_id: Option<String>,
444 },
445 StartRoute {
446 route_id: String,
447 command_id: String,
448 causation_id: Option<String>,
449 },
450 StopRoute {
451 route_id: String,
452 command_id: String,
453 causation_id: Option<String>,
454 },
455 SuspendRoute {
456 route_id: String,
457 command_id: String,
458 causation_id: Option<String>,
459 },
460 ResumeRoute {
461 route_id: String,
462 command_id: String,
463 causation_id: Option<String>,
464 },
465 ReloadRoute {
466 route_id: String,
467 command_id: String,
468 causation_id: Option<String>,
469 },
470 FailRoute {
474 route_id: String,
475 error: String,
476 command_id: String,
477 causation_id: Option<String>,
478 },
479 RemoveRoute {
480 route_id: String,
481 command_id: String,
482 causation_id: Option<String>,
483 },
484 ReloadTlsCerts {
485 scheme: String,
486 host: String,
487 port: u16,
488 command_id: String,
489 causation_id: Option<String>,
490 },
491}
492
493impl RuntimeCommand {
494 pub fn command_id(&self) -> &str {
495 match self {
496 RuntimeCommand::RegisterRoute { command_id, .. }
497 | RuntimeCommand::StartRoute { command_id, .. }
498 | RuntimeCommand::StopRoute { command_id, .. }
499 | RuntimeCommand::SuspendRoute { command_id, .. }
500 | RuntimeCommand::ResumeRoute { command_id, .. }
501 | RuntimeCommand::ReloadRoute { command_id, .. }
502 | RuntimeCommand::FailRoute { command_id, .. }
503 | RuntimeCommand::RemoveRoute { command_id, .. }
504 | RuntimeCommand::ReloadTlsCerts { command_id, .. } => command_id,
505 }
506 }
507
508 pub fn causation_id(&self) -> Option<&str> {
509 match self {
510 RuntimeCommand::RegisterRoute { causation_id, .. }
511 | RuntimeCommand::StartRoute { causation_id, .. }
512 | RuntimeCommand::StopRoute { causation_id, .. }
513 | RuntimeCommand::SuspendRoute { causation_id, .. }
514 | RuntimeCommand::ResumeRoute { causation_id, .. }
515 | RuntimeCommand::ReloadRoute { causation_id, .. }
516 | RuntimeCommand::FailRoute { causation_id, .. }
517 | RuntimeCommand::RemoveRoute { causation_id, .. }
518 | RuntimeCommand::ReloadTlsCerts { causation_id, .. } => causation_id.as_deref(),
519 }
520 }
521}
522
523#[derive(Debug, Clone, PartialEq, Eq)]
524pub enum RuntimeCommandResult {
525 Accepted,
526 Duplicate {
527 command_id: String,
528 },
529 RouteRegistered {
530 route_id: String,
531 },
532 RouteStateChanged {
533 route_id: String,
534 status: String,
535 },
536 TlsCertsReloaded {
537 scheme: String,
538 host: String,
539 port: u16,
540 },
541}
542
543#[derive(Debug, Clone, PartialEq, Eq)]
544pub enum RuntimeQuery {
545 GetRouteStatus {
546 route_id: String,
547 },
548 InFlightCount {
552 route_id: String,
553 },
554 ListRoutes,
555}
556
557#[derive(Debug, Clone, PartialEq, Eq)]
558pub enum RuntimeQueryResult {
559 InFlightCount { route_id: String, count: u64 },
560 RouteNotFound { route_id: String },
561 RouteStatus { route_id: String, status: String },
562 Routes { route_ids: Vec<String> },
563}
564
565#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
566pub enum RuntimeEvent {
567 RouteRegistered { route_id: String },
568 RouteStartRequested { route_id: String },
569 RouteStarted { route_id: String },
570 RouteFailed { route_id: String, error: String },
571 RouteStopped { route_id: String },
572 RouteSuspended { route_id: String },
573 RouteResumed { route_id: String },
574 RouteReloaded { route_id: String },
575 RouteRemoved { route_id: String },
576}
577
578#[async_trait]
579pub trait RuntimeCommandBus: Send + Sync {
580 async fn execute(&self, cmd: RuntimeCommand) -> Result<RuntimeCommandResult, CamelError>;
581}
582
583#[async_trait]
584pub trait RuntimeQueryBus: Send + Sync {
585 async fn ask(&self, query: RuntimeQuery) -> Result<RuntimeQueryResult, CamelError>;
586}
587
588pub trait RuntimeHandle: RuntimeCommandBus + RuntimeQueryBus {}
589
590impl<T> RuntimeHandle for T where T: RuntimeCommandBus + RuntimeQueryBus {}
591
592#[cfg(test)]
593mod tests {
594 use super::*;
595 use async_trait::async_trait;
596 use futures::executor::block_on;
597
598 struct NoopRuntime;
599
600 #[async_trait]
601 impl RuntimeCommandBus for NoopRuntime {
602 async fn execute(&self, cmd: RuntimeCommand) -> Result<RuntimeCommandResult, CamelError> {
603 Ok(match cmd {
604 RuntimeCommand::RegisterRoute { spec, .. } => {
605 RuntimeCommandResult::RouteRegistered {
606 route_id: spec.route_id,
607 }
608 }
609 RuntimeCommand::StartRoute { route_id, .. }
610 | RuntimeCommand::StopRoute { route_id, .. }
611 | RuntimeCommand::SuspendRoute { route_id, .. }
612 | RuntimeCommand::ResumeRoute { route_id, .. }
613 | RuntimeCommand::ReloadRoute { route_id, .. }
614 | RuntimeCommand::FailRoute { route_id, .. }
615 | RuntimeCommand::RemoveRoute { route_id, .. } => {
616 RuntimeCommandResult::RouteStateChanged {
617 route_id,
618 status: "ok".to_string(),
619 }
620 }
621 RuntimeCommand::ReloadTlsCerts {
622 scheme, host, port, ..
623 } => RuntimeCommandResult::TlsCertsReloaded { scheme, host, port },
624 })
625 }
626 }
627
628 #[async_trait]
629 impl RuntimeQueryBus for NoopRuntime {
630 async fn ask(&self, query: RuntimeQuery) -> Result<RuntimeQueryResult, CamelError> {
631 Ok(match query {
632 RuntimeQuery::GetRouteStatus { route_id } => RuntimeQueryResult::RouteStatus {
633 route_id,
634 status: "Started".to_string(),
635 },
636 RuntimeQuery::InFlightCount { route_id } => {
637 RuntimeQueryResult::InFlightCount { route_id, count: 0 }
638 }
639 RuntimeQuery::ListRoutes => RuntimeQueryResult::Routes {
640 route_ids: vec!["r1".to_string()],
641 },
642 })
643 }
644 }
645
646 #[test]
647 fn command_and_query_ids_are_exposed() {
648 let cmd = RuntimeCommand::StartRoute {
649 route_id: "r1".into(),
650 command_id: "c1".into(),
651 causation_id: None,
652 };
653 assert_eq!(cmd.command_id(), "c1");
654 }
655
656 #[test]
657 fn canonical_spec_requires_route_id_and_from() {
658 let spec = CanonicalRouteSpec::new("r1", "timer:tick");
659 assert_eq!(spec.route_id, "r1");
660 assert_eq!(spec.from, "timer:tick");
661 assert_eq!(spec.version, CANONICAL_CONTRACT_VERSION);
662 assert!(spec.steps.is_empty());
663 assert!(spec.circuit_breaker.is_none());
664 }
665
666 #[test]
667 fn canonical_contract_rejects_invalid_version() {
668 let mut spec = CanonicalRouteSpec::new("r1", "timer:tick");
669 spec.version = 3;
670 let err = spec.validate_contract().unwrap_err().to_string();
671 assert!(err.contains("expected version"));
672 }
673
674 #[test]
675 fn canonical_contract_declares_subset_scope() {
676 assert!(canonical_contract_supports_step("to"));
677 assert!(canonical_contract_supports_step("split"));
678 assert!(!canonical_contract_supports_step("set_header"));
679 assert!(!canonical_contract_supports_step("set_property"));
680
681 assert!(CANONICAL_CONTRACT_DECLARATIVE_ONLY_STEPS.contains(&"split"));
682 assert!(CANONICAL_CONTRACT_EXCLUDED_DECLARATIVE_STEPS.contains(&"set_header"));
683 assert!(CANONICAL_CONTRACT_EXCLUDED_DECLARATIVE_STEPS.contains(&"set_property"));
684 assert!(CANONICAL_CONTRACT_RUST_ONLY_STEPS.contains(&"processor"));
685 }
686
687 #[test]
688 fn canonical_contract_rejection_reason_is_explicit() {
689 let set_header_reason = canonical_contract_rejection_reason("set_header")
690 .expect("set_header should have explicit reason");
691 assert!(set_header_reason.contains("out-of-scope"));
692
693 let set_property_reason = canonical_contract_rejection_reason("set_property")
694 .expect("set_property should have explicit reason");
695 assert!(set_property_reason.contains("out-of-scope"));
696
697 let processor_reason = canonical_contract_rejection_reason("processor")
698 .expect("processor should be rust-only");
699 assert!(processor_reason.contains("rust-only"));
700
701 let split_reason = canonical_contract_rejection_reason("split")
702 .expect("split should require declarative form");
703 assert!(split_reason.contains("declarative"));
704 }
705
706 #[test]
707 fn command_causation_id_is_exposed() {
708 let cmd = RuntimeCommand::StopRoute {
709 route_id: "r1".into(),
710 command_id: "c2".into(),
711 causation_id: Some("c1".into()),
712 };
713 assert_eq!(cmd.command_id(), "c2");
714 assert_eq!(cmd.causation_id(), Some("c1"));
715 }
716
717 #[test]
718 fn canonical_contract_rejects_empty_route_id_and_from() {
719 let spec = CanonicalRouteSpec::new(" ", "timer:tick");
720 let err = spec.validate_contract().unwrap_err().to_string();
721 assert!(err.contains("route_id cannot be empty"));
722
723 let spec = CanonicalRouteSpec::new("r1", " ");
724 let err = spec.validate_contract().unwrap_err().to_string();
725 assert!(err.contains("from cannot be empty"));
726 }
727
728 #[test]
729 fn canonical_contract_rejects_invalid_nested_steps() {
730 let mut spec = CanonicalRouteSpec::new("r1", "timer:tick");
731 spec.steps = vec![CanonicalStepSpec::Split {
732 expression: CanonicalSplitExpressionSpec::BodyLines,
733 aggregation: CanonicalSplitAggregationSpec::CollectAll,
734 parallel: true,
735 parallel_limit: Some(0),
736 stop_on_exception: false,
737 steps: vec![CanonicalStepSpec::To {
738 uri: "log:ok".to_string(),
739 }],
740 }];
741 let err = spec.validate_contract().unwrap_err().to_string();
742 assert!(err.contains("split.parallel_limit must be > 0"));
743
744 spec.steps = vec![CanonicalStepSpec::To {
745 uri: " ".to_string(),
746 }];
747 let err = spec.validate_contract().unwrap_err().to_string();
748 assert!(err.contains("endpoint uri cannot be empty"));
749 }
750
751 #[test]
752 fn canonical_contract_rejects_invalid_aggregate_and_circuit_breaker() {
753 let mut spec = CanonicalRouteSpec::new("r1", "timer:tick");
754 spec.steps = vec![CanonicalStepSpec::Aggregate(CanonicalAggregateSpec {
755 header: " ".to_string(),
756 completion_size: Some(1),
757 completion_timeout_ms: None,
758 correlation_key: None,
759 force_completion_on_stop: None,
760 discard_on_timeout: None,
761 strategy: CanonicalAggregateStrategySpec::CollectAll,
762 max_buckets: None,
763 bucket_ttl_ms: None,
764 })];
765 let err = spec.validate_contract().unwrap_err().to_string();
766 assert!(err.contains("aggregate.header cannot be empty"));
767
768 spec.steps = vec![CanonicalStepSpec::Aggregate(CanonicalAggregateSpec {
769 header: "k".to_string(),
770 completion_size: Some(0),
771 completion_timeout_ms: None,
772 correlation_key: None,
773 force_completion_on_stop: None,
774 discard_on_timeout: None,
775 strategy: CanonicalAggregateStrategySpec::CollectAll,
776 max_buckets: None,
777 bucket_ttl_ms: None,
778 })];
779 let err = spec.validate_contract().unwrap_err().to_string();
780 assert!(err.contains("aggregate.completion_size must be > 0"));
781
782 spec.steps = vec![];
783 spec.circuit_breaker = Some(CanonicalCircuitBreakerSpec {
784 failure_threshold: 0,
785 open_duration_ms: 10,
786 });
787 let err = spec.validate_contract().unwrap_err().to_string();
788 assert!(err.contains("failure_threshold must be > 0"));
789
790 spec.circuit_breaker = Some(CanonicalCircuitBreakerSpec {
791 failure_threshold: 1,
792 open_duration_ms: 0,
793 });
794 let err = spec.validate_contract().unwrap_err().to_string();
795 assert!(err.contains("open_duration_ms must be > 0"));
796 }
797
798 #[test]
799 fn canonical_contract_rejection_reason_none_for_regular_steps() {
800 assert!(canonical_contract_rejection_reason("to").is_none());
801 assert!(canonical_contract_rejection_reason("unknown-step").is_none());
802 }
803
804 #[test]
805 fn command_helpers_cover_all_variants() {
806 let spec = CanonicalRouteSpec::new("r1", "timer:tick");
807 let cmds = [
808 RuntimeCommand::RegisterRoute {
809 spec,
810 command_id: "c1".into(),
811 causation_id: Some("root".into()),
812 },
813 RuntimeCommand::StartRoute {
814 route_id: "r1".into(),
815 command_id: "c2".into(),
816 causation_id: None,
817 },
818 RuntimeCommand::StopRoute {
819 route_id: "r1".into(),
820 command_id: "c3".into(),
821 causation_id: None,
822 },
823 RuntimeCommand::SuspendRoute {
824 route_id: "r1".into(),
825 command_id: "c4".into(),
826 causation_id: None,
827 },
828 RuntimeCommand::ResumeRoute {
829 route_id: "r1".into(),
830 command_id: "c5".into(),
831 causation_id: None,
832 },
833 RuntimeCommand::ReloadRoute {
834 route_id: "r1".into(),
835 command_id: "c6".into(),
836 causation_id: None,
837 },
838 RuntimeCommand::FailRoute {
839 route_id: "r1".into(),
840 error: "boom".into(),
841 command_id: "c7".into(),
842 causation_id: None,
843 },
844 RuntimeCommand::RemoveRoute {
845 route_id: "r1".into(),
846 command_id: "c8".into(),
847 causation_id: None,
848 },
849 RuntimeCommand::ReloadTlsCerts {
850 scheme: "https".into(),
851 host: "example.com".into(),
852 port: 8443,
853 command_id: "c9".into(),
854 causation_id: None,
855 },
856 ];
857
858 let ids: Vec<&str> = cmds.iter().map(RuntimeCommand::command_id).collect();
859 assert_eq!(
860 ids,
861 vec!["c1", "c2", "c3", "c4", "c5", "c6", "c7", "c8", "c9"]
862 );
863 assert_eq!(cmds[0].causation_id(), Some("root"));
864 assert_eq!(cmds[1].causation_id(), None);
865 }
866
867 #[test]
868 fn canonical_route_spec_serde_roundtrip() {
869 let mut spec = CanonicalRouteSpec::new("test-route", "timer:tick?period=1000");
870 spec.steps.push(CanonicalStepSpec::Log {
871 message: "Hello".into(),
872 });
873 spec.steps.push(CanonicalStepSpec::To {
874 uri: "log:info".into(),
875 });
876 spec.steps.push(CanonicalStepSpec::Stop);
877
878 let json = serde_json::to_string(&spec).unwrap();
879 let deserialized: CanonicalRouteSpec = serde_json::from_str(&json).unwrap();
880 assert_eq!(spec, deserialized);
881 }
882
883 #[test]
884 fn canonical_step_spec_serde_variants() {
885 let steps = vec![
886 CanonicalStepSpec::To {
887 uri: "direct:a".into(),
888 },
889 CanonicalStepSpec::Log {
890 message: "msg".into(),
891 },
892 CanonicalStepSpec::WireTap {
893 uri: "direct:audit".into(),
894 },
895 CanonicalStepSpec::Stop,
896 CanonicalStepSpec::Delay {
897 delay_ms: 100,
898 dynamic_header: None,
899 },
900 ];
901 let json = serde_json::to_string_pretty(&steps).unwrap();
902 let back: Vec<CanonicalStepSpec> = serde_json::from_str(&json).unwrap();
903 assert_eq!(steps, back);
904 }
905
906 #[test]
907 fn canonical_route_spec_json_schema_generates() {
908 let schema = schemars::schema_for!(CanonicalRouteSpec);
909 let json = serde_json::to_string(&schema).unwrap();
910 assert!(json.contains("CanonicalRouteSpec"));
911 assert!(json.contains("route_id"));
912 }
913
914 #[test]
915 fn canonical_json_schema_has_no_function_step() {
916 let schema = schemars::schema_for!(CanonicalRouteSpec);
917 let json = serde_json::to_string(&schema).unwrap();
918 assert!(
919 !json.contains("\"function\""),
920 "canonical JSON schema must not contain 'function' step"
921 );
922 }
923
924 #[test]
925 fn canonical_contract_does_not_support_function() {
926 assert!(
927 !canonical_contract_supports_step("function"),
928 "function must not be in CANONICAL_CONTRACT_SUPPORTED_STEPS"
929 );
930 }
931
932 #[test]
933 fn runtime_command_result_all_variants_are_distinct() {
934 let accepted = RuntimeCommandResult::Accepted;
935 let dup = RuntimeCommandResult::Duplicate {
936 command_id: "c1".into(),
937 };
938 let registered = RuntimeCommandResult::RouteRegistered {
939 route_id: "r1".into(),
940 };
941 let changed = RuntimeCommandResult::RouteStateChanged {
942 route_id: "r1".into(),
943 status: "Started".into(),
944 };
945
946 assert_ne!(accepted, dup);
947 assert_ne!(dup, registered);
948 assert_ne!(registered, changed);
949
950 let dup2 = RuntimeCommandResult::Duplicate {
951 command_id: "c1".into(),
952 };
953 assert_eq!(dup, dup2);
954 }
955
956 #[test]
957 fn runtime_event_serialization_round_trip() {
958 let event = RuntimeEvent::RouteFailed {
959 route_id: "route-a".to_string(),
960 error: "boom".to_string(),
961 };
962 let json = serde_json::to_string(&event).unwrap();
963 let back: RuntimeEvent = serde_json::from_str(&json).unwrap();
964 assert_eq!(event, back);
965 }
966
967 #[test]
968 fn noop_runtime_execute_and_ask_return_expected_shapes() {
969 let rt = NoopRuntime;
970 let cmd = RuntimeCommand::RegisterRoute {
971 spec: CanonicalRouteSpec::new("r2", "timer:tick"),
972 command_id: "c1".into(),
973 causation_id: None,
974 };
975 let cmd_result = block_on(rt.execute(cmd)).unwrap();
976 assert_eq!(
977 cmd_result,
978 RuntimeCommandResult::RouteRegistered {
979 route_id: "r2".into()
980 }
981 );
982
983 let query_result = block_on(rt.ask(RuntimeQuery::GetRouteStatus {
984 route_id: "r2".into(),
985 }))
986 .unwrap();
987 assert_eq!(
988 query_result,
989 RuntimeQueryResult::RouteStatus {
990 route_id: "r2".into(),
991 status: "Started".into()
992 }
993 );
994 }
995
996 #[test]
997 fn canonical_contract_name_and_version_constants_match() {
998 assert_eq!(CANONICAL_CONTRACT_NAME, "canonical-v1");
999 assert_eq!(CANONICAL_CONTRACT_VERSION, 2);
1000 }
1001
1002 #[test]
1003 fn canonical_concurrency_spec_rejects_zero_max() {
1004 let spec = CanonicalRouteSpec::new("r1", "timer:tick")
1005 .with_concurrency(CanonicalConcurrencySpec::Concurrent { max: 0 });
1006 let err = spec.validate_contract().unwrap_err().to_string();
1007 assert!(err.contains("concurrency max must be > 0"), "{err}");
1008 }
1009
1010 #[test]
1011 fn canonical_v2_round_trip() {
1012 let spec = CanonicalRouteSpec::new("r1", "timer:tick")
1013 .with_auto_startup(false)
1014 .with_startup_order(42)
1015 .with_concurrency(CanonicalConcurrencySpec::Concurrent { max: 8 });
1016 spec.validate_contract().unwrap();
1017 }
1018
1019 #[test]
1020 fn canonical_v2_version_is_2() {
1021 assert_eq!(CANONICAL_CONTRACT_VERSION, 2);
1022 }
1023
1024 #[test]
1025 fn canonical_loss_report_builder() {
1026 let report =
1027 CanonicalLossReport::from_field("error_handler", "not supported by canonical path", 2);
1028 assert_eq!(report.dropped_fields.len(), 1);
1029 assert_eq!(report.dropped_fields[0].field, "error_handler");
1030 }
1031
1032 #[test]
1033 fn canonical_v1_json_deserializes_in_v2() {
1034 let json = r#"{"route_id":"r1","from":"timer:tick","steps":[],"version":1}"#;
1035 let spec: CanonicalRouteSpec = serde_json::from_str(json).unwrap();
1036 assert_eq!(spec.route_id, "r1");
1037 assert!(spec.auto_startup.is_none());
1038 assert!(spec.startup_order.is_none());
1039 assert!(spec.concurrency.is_none());
1040 spec.validate_contract().unwrap();
1042 }
1043}