ironflow_engine/handler.rs
1//! [`WorkflowHandler`] trait — dynamic workflows with context chaining.
2//!
3//! Implement this trait to define workflows where steps can reference
4//! outputs from previous steps. The handler receives a [`WorkflowContext`]
5//! that provides step execution methods with automatic persistence.
6//!
7//! # Examples
8//!
9//! ```no_run
10//! use ironflow_engine::handler::WorkflowHandler;
11//! use ironflow_engine::context::WorkflowContext;
12//! use ironflow_engine::config::{ShellConfig, AgentStepConfig};
13//! use ironflow_engine::error::EngineError;
14//! use std::future::Future;
15//! use std::pin::Pin;
16//!
17//! struct DeployWorkflow;
18//!
19//! impl WorkflowHandler for DeployWorkflow {
20//! fn name(&self) -> &str {
21//! "deploy"
22//! }
23//!
24//! fn execute<'a>(
25//! &'a self,
26//! ctx: &'a mut WorkflowContext,
27//! ) -> Pin<Box<dyn Future<Output = Result<(), EngineError>> + Send + 'a>> {
28//! Box::pin(async move {
29//! let build = ctx.shell("build", ShellConfig::new("cargo build --release")).await?;
30//! let tests = ctx.shell("test", ShellConfig::new("cargo test")).await?;
31//!
32//! let review = ctx.agent("review", AgentStepConfig::new(
33//! &format!("Build:\n{}\nTests:\n{}\nReview.",
34//! build.output["stdout"], tests.output["stdout"])
35//! )).await?;
36//!
37//! if review.output.as_str().unwrap_or("").contains("LGTM") {
38//! ctx.shell("deploy", ShellConfig::new("./deploy.sh")).await?;
39//! }
40//!
41//! Ok(())
42//! })
43//! }
44//! }
45//! ```
46
47use std::collections::HashMap;
48use std::future::Future;
49use std::pin::Pin;
50
51use rust_decimal::Decimal;
52use schemars::JsonSchema;
53use serde::Serialize;
54use serde_json::Value;
55
56use crate::context::WorkflowContext;
57use crate::error::EngineError;
58use crate::guard::WorkflowGuardConfig;
59use crate::run_creator::{CreateRunOpts, RunCreator, RunCreatorFuture};
60use crate::schedule::CronSchedule;
61
62/// Generate a JSON Schema [`Value`] from a type that derives [`JsonSchema`].
63///
64/// Use this in [`WorkflowHandler::input_schema`] to automatically derive the
65/// schema from your input struct instead of writing JSON by hand.
66///
67/// # Examples
68///
69/// ```
70/// use schemars::JsonSchema;
71/// use serde::Deserialize;
72/// use ironflow_engine::handler::input_schema_for;
73///
74/// #[derive(Deserialize, JsonSchema)]
75/// struct DeployInput {
76/// environment: String,
77/// dry_run: Option<bool>,
78/// }
79///
80/// let schema = input_schema_for::<DeployInput>();
81/// assert_eq!(schema["type"], "object");
82/// assert!(schema["properties"]["environment"].is_object());
83/// ```
84pub fn input_schema_for<T: JsonSchema>() -> Value {
85 let schema = schemars::schema_for!(T);
86 serde_json::to_value(schema).expect("schema serialization cannot fail")
87}
88
89/// Boxed future returned by [`WorkflowHandler::execute`].
90pub type HandlerFuture<'a> = Pin<Box<dyn Future<Output = Result<(), EngineError>> + Send + 'a>>;
91
92/// Metadata about a workflow, returned by [`WorkflowHandler::describe`].
93///
94/// Contains a human-readable description and optional Rust source code
95/// for display in the dashboard.
96#[derive(Debug, Clone, Serialize)]
97pub struct WorkflowInfo {
98 /// Human-readable description of what the workflow does.
99 pub description: String,
100 /// Optional Rust source code of the handler (for UI display).
101 pub source_code: Option<String>,
102 /// Names of sub-workflows invoked by this handler.
103 #[serde(default, skip_serializing_if = "Vec::is_empty")]
104 pub sub_workflows: Vec<String>,
105 /// Optional `/`-separated category path used to group workflows in the UI tree.
106 ///
107 /// A value like `"data/etl"` places the workflow under `data` → `etl`.
108 /// `None` means the workflow is uncategorized.
109 #[serde(default, skip_serializing_if = "Option::is_none")]
110 pub category: Option<String>,
111 /// Handler version string, used to trace which code produced a given run.
112 #[serde(default, skip_serializing_if = "Option::is_none")]
113 pub version: Option<String>,
114 /// Versions accepted for replay without `force`.
115 #[serde(default, skip_serializing_if = "Vec::is_empty")]
116 pub compatible_versions: Vec<String>,
117 /// JSON Schema describing the expected input payload.
118 ///
119 /// When present, the dashboard renders a dynamic form from this schema
120 /// and the engine validates the payload before creating a run.
121 #[serde(default, skip_serializing_if = "Option::is_none")]
122 pub input_schema: Option<Value>,
123 /// Labels automatically applied to every run of this workflow.
124 #[serde(default, skip_serializing_if = "HashMap::is_empty")]
125 pub default_labels: HashMap<String, String>,
126 /// Optional cron schedule for automatic execution.
127 #[serde(default, skip_serializing_if = "Option::is_none")]
128 pub schedule: Option<CronSchedule>,
129 /// Default cumulative cost cap applied to runs of this workflow, in USD.
130 ///
131 /// Overridden by a cap supplied at run creation, and takes precedence over
132 /// the server-wide default. `None` means the handler declares no default.
133 #[serde(default, skip_serializing_if = "Option::is_none")]
134 pub default_max_cost_usd: Option<Decimal>,
135}
136
137/// A dynamic workflow handler with context-aware step chaining.
138///
139/// Implement this trait to define workflows where each step can use
140/// the output of previous steps. Register handlers with
141/// [`Engine::register`](crate::engine::Engine::register) and execute
142/// them by name.
143///
144/// # Why `Pin<Box<dyn Future>>` instead of `async fn`?
145///
146/// The handler must be object-safe (`dyn WorkflowHandler`) to allow
147/// registering different handler types in the engine's registry.
148pub trait WorkflowHandler: Send + Sync {
149 /// The workflow name used for registration and lookup.
150 fn name(&self) -> &str;
151
152 /// Handler version string, used to trace which code version produced a run.
153 ///
154 /// Override this to return a meaningful version (semver, git SHA, build
155 /// hash, etc.). The default is `"1"`.
156 ///
157 /// The engine records this value on every run it creates so that retries
158 /// can detect when the handler has changed since the original execution.
159 fn version(&self) -> Option<&str> {
160 Some("1")
161 }
162
163 /// Versions of this handler that can replay payloads produced by an
164 /// older run without requiring `force`.
165 ///
166 /// When a retry targets a run whose `handler_version` differs from
167 /// [`version`](Self::version), the engine checks this list. If the
168 /// run's version appears here, the retry proceeds normally; otherwise
169 /// it is refused with `409 HANDLER_VERSION_MISMATCH` unless the caller
170 /// passes `force=true`.
171 ///
172 /// The default is an empty slice (only the current version is accepted).
173 ///
174 /// # Examples
175 ///
176 /// ```
177 /// # use ironflow_engine::handler::{WorkflowHandler, HandlerFuture};
178 /// # use ironflow_engine::context::WorkflowContext;
179 /// struct MigratedHandler;
180 ///
181 /// impl WorkflowHandler for MigratedHandler {
182 /// fn name(&self) -> &str { "migrated" }
183 /// fn version(&self) -> Option<&str> { Some("2.0.0") }
184 /// fn compatible_versions(&self) -> &[&str] { &["1.0.0", "1.5.0"] }
185 /// fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
186 /// Box::pin(async move { Ok(()) })
187 /// }
188 /// }
189 ///
190 /// assert_eq!(MigratedHandler.compatible_versions(), &["1.0.0", "1.5.0"]);
191 /// ```
192 fn compatible_versions(&self) -> &[&str] {
193 &[]
194 }
195
196 /// Optional `/`-separated category path used to group workflows in the UI tree.
197 ///
198 /// Return a value like `"data/etl"` to place the workflow under `data` → `etl`.
199 /// The default is `None` (uncategorized).
200 ///
201 /// Validation (empty segments, leading or trailing `/`, `//`, whitespace
202 /// segments) is enforced at registration time by
203 /// [`Engine::register`](crate::engine::Engine::register).
204 fn category(&self) -> Option<&str> {
205 None
206 }
207
208 /// Return a JSON Schema describing the expected input payload.
209 ///
210 /// When present, the dashboard renders a dynamic form from this schema
211 /// and the engine validates the payload before creating a run.
212 /// The default is `None` (no schema, free-form payload).
213 fn input_schema(&self) -> Option<Value> {
214 None
215 }
216
217 /// Labels automatically applied to every run of this workflow.
218 ///
219 /// These are merged with any labels provided at run creation time.
220 /// User-provided labels take precedence over defaults.
221 fn default_labels(&self) -> HashMap<String, String> {
222 HashMap::new()
223 }
224
225 /// Optional cron schedule for automatic execution.
226 ///
227 /// Return a [`CronSchedule`] built from a cron expression
228 /// (5 or 6 fields, as supported by [`croner`]).
229 ///
230 /// When set, the engine exposes this handler via
231 /// [`Engine::scheduled_handlers`](crate::engine::Engine::scheduled_handlers)
232 /// so the runtime can wire it into a cron scheduler automatically.
233 ///
234 /// The default is `None` (no automatic scheduling).
235 ///
236 /// # Examples
237 ///
238 /// ```
239 /// # use ironflow_engine::handler::{WorkflowHandler, HandlerFuture};
240 /// # use ironflow_engine::context::WorkflowContext;
241 /// # use ironflow_engine::schedule::CronSchedule;
242 /// struct HourlySync;
243 ///
244 /// impl WorkflowHandler for HourlySync {
245 /// fn name(&self) -> &str { "hourly-sync" }
246 /// fn schedule(&self) -> Option<&CronSchedule> {
247 /// // In practice, store as a field or use `std::sync::LazyLock`.
248 /// None
249 /// }
250 /// fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
251 /// Box::pin(async move { Ok(()) })
252 /// }
253 /// }
254 /// ```
255 fn schedule(&self) -> Option<&CronSchedule> {
256 None
257 }
258
259 /// Default cumulative cost cap for runs of this workflow, in USD.
260 ///
261 /// Applied when the run creation request does not supply one. Takes
262 /// precedence over the server-wide
263 /// [`IRONFLOW_DEFAULT_RUN_MAX_COST_USD`](crate::budget::DEFAULT_RUN_MAX_COST_ENV).
264 /// The default is `None` (fall back to the server default, or no cap).
265 ///
266 /// # Examples
267 ///
268 /// ```
269 /// # use ironflow_engine::handler::{WorkflowHandler, HandlerFuture};
270 /// # use ironflow_engine::context::WorkflowContext;
271 /// use rust_decimal::Decimal;
272 ///
273 /// struct ExpensiveAnalysis;
274 ///
275 /// impl WorkflowHandler for ExpensiveAnalysis {
276 /// fn name(&self) -> &str { "expensive-analysis" }
277 /// fn default_max_cost_usd(&self) -> Option<Decimal> {
278 /// Some(Decimal::new(500, 2)) // $5.00
279 /// }
280 /// fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
281 /// Box::pin(async move { Ok(()) })
282 /// }
283 /// }
284 ///
285 /// assert_eq!(ExpensiveAnalysis.default_max_cost_usd(), Some(Decimal::new(500, 2)));
286 /// ```
287 fn default_max_cost_usd(&self) -> Option<Decimal> {
288 None
289 }
290
291 /// Optional guard configuration for this workflow.
292 ///
293 /// When present, overrides the engine's global guard configuration
294 /// for runs of this handler. The default is `None` (use the engine's
295 /// global configuration).
296 ///
297 /// # Examples
298 ///
299 /// ```
300 /// # use ironflow_engine::handler::{WorkflowHandler, HandlerFuture};
301 /// # use ironflow_engine::context::WorkflowContext;
302 /// use ironflow_engine::guard::WorkflowGuardConfig;
303 ///
304 /// struct StrictWorkflow;
305 ///
306 /// impl WorkflowHandler for StrictWorkflow {
307 /// fn name(&self) -> &str { "strict" }
308 /// fn guard_config(&self) -> Option<WorkflowGuardConfig> {
309 /// Some(WorkflowGuardConfig::new().with_max_depth(2))
310 /// }
311 /// fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
312 /// Box::pin(async move { Ok(()) })
313 /// }
314 /// }
315 ///
316 /// assert_eq!(StrictWorkflow.guard_config().unwrap().max_depth, 2);
317 /// ```
318 fn guard_config(&self) -> Option<WorkflowGuardConfig> {
319 None
320 }
321
322 /// Check whether a run carrying `run_version` can be replayed by this
323 /// handler without `force`.
324 ///
325 /// Compatibility rules:
326 /// - `run_version` is `None` (old run predating version tracking): always
327 /// compatible.
328 /// - `run_version` equals [`version`](Self::version): compatible.
329 /// - `run_version` appears in [`compatible_versions`](Self::compatible_versions):
330 /// compatible.
331 /// - Otherwise: incompatible.
332 ///
333 /// # Examples
334 ///
335 /// ```
336 /// # use ironflow_engine::handler::{WorkflowHandler, HandlerFuture};
337 /// # use ironflow_engine::context::WorkflowContext;
338 /// struct MyHandler;
339 ///
340 /// impl WorkflowHandler for MyHandler {
341 /// fn name(&self) -> &str { "my-handler" }
342 /// fn version(&self) -> Option<&str> { Some("2.0.0") }
343 /// fn compatible_versions(&self) -> &[&str] { &["1.0.0"] }
344 /// fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
345 /// Box::pin(async move { Ok(()) })
346 /// }
347 /// }
348 ///
349 /// assert!(MyHandler.is_version_compatible(None));
350 /// assert!(MyHandler.is_version_compatible(Some("2.0.0")));
351 /// assert!(MyHandler.is_version_compatible(Some("1.0.0")));
352 /// assert!(!MyHandler.is_version_compatible(Some("0.5.0")));
353 /// ```
354 fn is_version_compatible(&self, run_version: Option<&str>) -> bool {
355 let Some(rv) = run_version else {
356 return true;
357 };
358 if self.version() == Some(rv) {
359 return true;
360 }
361 self.compatible_versions().contains(&rv)
362 }
363
364 /// Return metadata about this workflow (description, source code).
365 ///
366 /// Override this to provide a description and source code for the
367 /// dashboard UI. The default returns an empty description with no source
368 /// but propagates [`WorkflowHandler::category`],
369 /// [`WorkflowHandler::version`], [`WorkflowHandler::input_schema`],
370 /// [`WorkflowHandler::default_labels`],
371 /// [`WorkflowHandler::compatible_versions`],
372 /// and [`WorkflowHandler::schedule`].
373 fn describe(&self) -> WorkflowInfo {
374 WorkflowInfo {
375 description: String::new(),
376 source_code: None,
377 sub_workflows: Vec::new(),
378 category: self.category().map(str::to_string),
379 version: self.version().map(str::to_string),
380 compatible_versions: self
381 .compatible_versions()
382 .iter()
383 .map(|s| s.to_string())
384 .collect(),
385 input_schema: self.input_schema(),
386 default_labels: self.default_labels(),
387 schedule: self.schedule().cloned(),
388 default_max_cost_usd: self.default_max_cost_usd(),
389 }
390 }
391
392 /// Create a run for this workflow, using handler metadata automatically.
393 ///
394 /// Assembles a [`NewRun`](ironflow_store::entities::NewRun) from [`name`](Self::name),
395 /// [`version`](Self::version), and [`default_max_cost_usd`](Self::default_max_cost_usd),
396 /// then delegates to the given [`RunCreator`].
397 ///
398 /// # Errors
399 ///
400 /// Returns [`EngineError`] if the underlying store rejects the run.
401 ///
402 /// # Examples
403 ///
404 /// ```no_run
405 /// # use ironflow_engine::handler::{WorkflowHandler, HandlerFuture};
406 /// # use ironflow_engine::context::WorkflowContext;
407 /// # use ironflow_engine::run_creator::{CreateRunOpts, RunCreator};
408 /// # use ironflow_store::entities::TriggerKind;
409 /// struct DeployWorkflow;
410 ///
411 /// impl WorkflowHandler for DeployWorkflow {
412 /// fn name(&self) -> &str { "deploy" }
413 /// fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
414 /// Box::pin(async move { Ok(()) })
415 /// }
416 /// }
417 ///
418 /// # async fn example(store: &dyn RunCreator) -> Result<(), ironflow_engine::error::EngineError> {
419 /// let opts = CreateRunOpts::new().trigger(TriggerKind::Api);
420 /// let run = DeployWorkflow.create_run(store, opts).await?.into_run();
421 /// assert_eq!(run.workflow_name, "deploy");
422 /// # Ok(())
423 /// # }
424 /// ```
425 fn create_run<'a>(
426 &self,
427 creator: &'a dyn RunCreator,
428 opts: CreateRunOpts,
429 ) -> RunCreatorFuture<'a> {
430 use tracing::{Instrument, info_span};
431
432 let new_run = opts.build(self.name(), self.version(), self.default_max_cost_usd());
433 let span = info_span!("handler.create_run", workflow = %self.name());
434 Box::pin(creator.create_run(new_run).instrument(span))
435 }
436
437 /// Execute the workflow with the given context.
438 ///
439 /// The context provides [`shell`](WorkflowContext::shell),
440 /// [`http`](WorkflowContext::http), and [`agent`](WorkflowContext::agent)
441 /// methods that automatically persist each step.
442 ///
443 /// # Errors
444 ///
445 /// Return [`EngineError`] if any step fails. The engine will mark
446 /// the run as `Failed` and record the error.
447 fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a>;
448}
449
450#[cfg(test)]
451mod tests {
452 use super::*;
453 use serde::{Deserialize, Serialize};
454
455 #[derive(Debug, Serialize, Deserialize, JsonSchema)]
456 struct TestInput {
457 environment: String,
458 #[serde(default)]
459 dry_run: bool,
460 }
461
462 struct MinimalHandler;
463
464 impl WorkflowHandler for MinimalHandler {
465 fn name(&self) -> &str {
466 "minimal"
467 }
468
469 fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
470 Box::pin(async { Ok(()) })
471 }
472 }
473
474 struct FullFeaturedHandler;
475
476 impl WorkflowHandler for FullFeaturedHandler {
477 fn name(&self) -> &str {
478 "full"
479 }
480
481 fn version(&self) -> Option<&str> {
482 Some("1.2.0")
483 }
484
485 fn category(&self) -> Option<&str> {
486 Some("data/etl")
487 }
488
489 fn input_schema(&self) -> Option<Value> {
490 Some(input_schema_for::<TestInput>())
491 }
492
493 fn default_labels(&self) -> HashMap<String, String> {
494 HashMap::from([
495 ("team".to_string(), "platform".to_string()),
496 ("env".to_string(), "prod".to_string()),
497 ])
498 }
499
500 fn default_max_cost_usd(&self) -> Option<Decimal> {
501 Some(Decimal::new(750, 2))
502 }
503
504 fn describe(&self) -> WorkflowInfo {
505 WorkflowInfo {
506 description: "Full-featured test handler".to_string(),
507 source_code: Some("fn test() {}".to_string()),
508 sub_workflows: vec!["helper".to_string()],
509 category: self.category().map(str::to_string),
510 version: self.version().map(str::to_string),
511 compatible_versions: self
512 .compatible_versions()
513 .iter()
514 .map(|s| s.to_string())
515 .collect(),
516 input_schema: self.input_schema(),
517 default_labels: self.default_labels(),
518 schedule: self.schedule().cloned(),
519 default_max_cost_usd: self.default_max_cost_usd(),
520 }
521 }
522
523 fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
524 Box::pin(async { Ok(()) })
525 }
526 }
527
528 #[test]
529 fn minimal_handler_has_required_name() {
530 let handler = MinimalHandler;
531 assert_eq!(handler.name(), "minimal");
532 }
533
534 #[test]
535 fn minimal_handler_defaults_to_version_1() {
536 let handler = MinimalHandler;
537 assert_eq!(handler.version(), Some("1"));
538 }
539
540 #[test]
541 fn minimal_handler_defaults_to_no_compatible_versions() {
542 let handler = MinimalHandler;
543 assert!(handler.compatible_versions().is_empty());
544 }
545
546 #[test]
547 fn minimal_handler_defaults_to_no_category() {
548 let handler = MinimalHandler;
549 assert_eq!(handler.category(), None);
550 }
551
552 #[test]
553 fn minimal_handler_defaults_to_no_schema() {
554 let handler = MinimalHandler;
555 assert_eq!(handler.input_schema(), None);
556 }
557
558 #[test]
559 fn minimal_handler_defaults_to_empty_labels() {
560 let handler = MinimalHandler;
561 let labels = handler.default_labels();
562 assert!(labels.is_empty());
563 }
564
565 #[test]
566 fn minimal_handler_defaults_to_no_schedule() {
567 let handler = MinimalHandler;
568 assert_eq!(handler.schedule(), None);
569 }
570
571 #[test]
572 fn minimal_handler_describe_reflects_defaults() {
573 let handler = MinimalHandler;
574 let info = handler.describe();
575 assert_eq!(info.description, "");
576 assert_eq!(info.source_code, None);
577 assert_eq!(info.sub_workflows, Vec::<String>::new());
578 assert_eq!(info.category, None);
579 assert_eq!(info.version, Some("1".to_string()));
580 assert!(info.compatible_versions.is_empty());
581 assert_eq!(info.input_schema, None);
582 assert!(info.default_labels.is_empty());
583 assert_eq!(info.schedule, None);
584 }
585
586 #[test]
587 fn full_handler_returns_all_metadata() {
588 let handler = FullFeaturedHandler;
589 assert_eq!(handler.name(), "full");
590 assert_eq!(handler.version(), Some("1.2.0"));
591 assert_eq!(handler.category(), Some("data/etl"));
592 assert!(handler.input_schema().is_some());
593 }
594
595 #[test]
596 fn full_handler_default_labels_are_set() {
597 let handler = FullFeaturedHandler;
598 let labels = handler.default_labels();
599 assert_eq!(labels.get("team"), Some(&"platform".to_string()));
600 assert_eq!(labels.get("env"), Some(&"prod".to_string()));
601 }
602
603 #[test]
604 fn full_handler_describe_includes_all_fields() {
605 let handler = FullFeaturedHandler;
606 let info = handler.describe();
607 assert_eq!(info.description, "Full-featured test handler");
608 assert_eq!(info.source_code, Some("fn test() {}".to_string()));
609 assert_eq!(info.sub_workflows, vec!["helper".to_string()]);
610 assert_eq!(info.category, Some("data/etl".to_string()));
611 assert_eq!(info.version, Some("1.2.0".to_string()));
612 assert!(info.input_schema.is_some());
613 assert_eq!(info.default_labels.len(), 2);
614 }
615
616 #[test]
617 fn input_schema_for_generates_json_schema() {
618 let schema = input_schema_for::<TestInput>();
619 assert_eq!(schema["type"], "object");
620 assert!(schema["properties"]["environment"].is_object());
621 assert!(schema["properties"]["dry_run"].is_object());
622 }
623
624 #[test]
625 fn input_schema_for_preserves_serde_attributes() {
626 let schema = input_schema_for::<TestInput>();
627 let properties = &schema["properties"];
628 assert!(properties.is_object());
629 assert!(properties.get("environment").is_some());
630 assert!(properties.get("dry_run").is_some());
631 }
632
633 #[test]
634 fn minimal_handler_defaults_to_no_max_cost() {
635 assert!(MinimalHandler.default_max_cost_usd().is_none());
636 assert!(MinimalHandler.describe().default_max_cost_usd.is_none());
637 }
638
639 #[test]
640 fn describe_propagates_handler_max_cost() {
641 assert_eq!(
642 FullFeaturedHandler.describe().default_max_cost_usd,
643 Some(Decimal::new(750, 2))
644 );
645 }
646
647 #[test]
648 fn workflow_info_omits_absent_max_cost_from_json() {
649 let json = serde_json::to_value(MinimalHandler.describe()).expect("serialize");
650 assert!(json.get("default_max_cost_usd").is_none());
651 }
652
653 #[test]
654 fn workflow_info_serializes_with_skip_empty() {
655 let info = WorkflowInfo {
656 description: "test".to_string(),
657 source_code: None,
658 sub_workflows: Vec::new(),
659 category: None,
660 version: None,
661 compatible_versions: Vec::new(),
662 input_schema: None,
663 default_labels: HashMap::new(),
664 schedule: None,
665 default_max_cost_usd: None,
666 };
667
668 let json = serde_json::to_value(&info).expect("serialize");
669 assert_eq!(json["description"], "test");
670 // Optional fields with skip_serializing_if may still be present or absent
671 // depending on the serde configuration. Just verify the description is there.
672 assert!(json.is_object());
673 }
674
675 #[test]
676 fn workflow_info_serializes_with_values() {
677 let info = WorkflowInfo {
678 description: "test".to_string(),
679 source_code: Some("code".to_string()),
680 sub_workflows: vec!["sub".to_string()],
681 category: Some("cat".to_string()),
682 version: Some("1.0.0".to_string()),
683 compatible_versions: vec!["0.9.0".to_string()],
684 input_schema: Some(serde_json::json!({"type": "object"})),
685 default_labels: HashMap::from([("key".to_string(), "value".to_string())]),
686 schedule: Some(CronSchedule::new("0 0 * * * *").unwrap()),
687 default_max_cost_usd: Some(Decimal::new(750, 2)),
688 };
689
690 let json = serde_json::to_value(&info).expect("serialize");
691 assert_eq!(json["description"], "test");
692 assert_eq!(json["source_code"], "code");
693 assert_eq!(json["sub_workflows"][0], "sub");
694 assert_eq!(json["category"], "cat");
695 assert_eq!(json["version"], "1.0.0");
696 assert_eq!(json["default_labels"]["key"], "value");
697 assert_eq!(json["schedule"], "0 0 * * * *");
698 assert_eq!(json["compatible_versions"][0], "0.9.0");
699 }
700
701 // ---- is_version_compatible ----
702
703 struct VersionedHandler;
704
705 impl WorkflowHandler for VersionedHandler {
706 fn name(&self) -> &str {
707 "versioned"
708 }
709 fn version(&self) -> Option<&str> {
710 Some("2.0.0")
711 }
712 fn compatible_versions(&self) -> &[&str] {
713 &["1.5.0", "1.9.0"]
714 }
715 fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
716 Box::pin(async { Ok(()) })
717 }
718 }
719
720 #[test]
721 fn version_compatible_with_same_version() {
722 assert!(VersionedHandler.is_version_compatible(Some("2.0.0")));
723 }
724
725 #[test]
726 fn version_compatible_with_none_run_version() {
727 assert!(VersionedHandler.is_version_compatible(None));
728 }
729
730 #[test]
731 fn version_compatible_with_listed_version() {
732 assert!(VersionedHandler.is_version_compatible(Some("1.5.0")));
733 assert!(VersionedHandler.is_version_compatible(Some("1.9.0")));
734 }
735
736 #[test]
737 fn version_incompatible_with_unlisted_version() {
738 assert!(!VersionedHandler.is_version_compatible(Some("1.0.0")));
739 assert!(!VersionedHandler.is_version_compatible(Some("3.0.0")));
740 }
741
742 #[test]
743 fn minimal_handler_compatible_with_same_default() {
744 assert!(MinimalHandler.is_version_compatible(Some("1")));
745 }
746
747 #[test]
748 fn minimal_handler_incompatible_with_different_version() {
749 assert!(!MinimalHandler.is_version_compatible(Some("2")));
750 }
751
752 // ---- WorkflowHandler::create_run ----
753
754 #[tokio::test]
755 async fn handler_create_run_uses_handler_metadata() {
756 use ironflow_store::entities::TriggerKind;
757 use ironflow_store::memory::InMemoryStore;
758
759 let store = InMemoryStore::new();
760
761 let opts = CreateRunOpts::new().trigger(TriggerKind::Api);
762 let creation = FullFeaturedHandler
763 .create_run(&store, opts)
764 .await
765 .expect("create_run");
766 let run = creation.into_run();
767
768 assert_eq!(run.workflow_name, "full");
769 assert_eq!(run.handler_version, Some("1.2.0".to_string()));
770 assert_eq!(run.max_cost_usd, Some(Decimal::new(750, 2)));
771 }
772
773 #[tokio::test]
774 async fn handler_create_run_opts_override_handler_defaults() {
775 use ironflow_store::memory::InMemoryStore;
776
777 let store = InMemoryStore::new();
778
779 let opts = CreateRunOpts::new().max_cost_usd(Decimal::new(100, 2));
780 let creation = FullFeaturedHandler
781 .create_run(&store, opts)
782 .await
783 .expect("create_run");
784 let run = creation.into_run();
785
786 assert_eq!(run.max_cost_usd, Some(Decimal::new(100, 2)));
787 }
788
789 #[tokio::test]
790 async fn handler_create_run_minimal_handler_defaults() {
791 use ironflow_store::memory::InMemoryStore;
792
793 let store = InMemoryStore::new();
794
795 let opts = CreateRunOpts::new();
796 let creation = MinimalHandler
797 .create_run(&store, opts)
798 .await
799 .expect("create_run");
800 let run = creation.into_run();
801
802 assert_eq!(run.workflow_name, "minimal");
803 assert_eq!(run.handler_version, Some("1".to_string()));
804 assert_eq!(run.max_cost_usd, None);
805 }
806}