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::schedule::CronSchedule;
59
60/// Generate a JSON Schema [`Value`] from a type that derives [`JsonSchema`].
61///
62/// Use this in [`WorkflowHandler::input_schema`] to automatically derive the
63/// schema from your input struct instead of writing JSON by hand.
64///
65/// # Examples
66///
67/// ```
68/// use schemars::JsonSchema;
69/// use serde::Deserialize;
70/// use ironflow_engine::handler::input_schema_for;
71///
72/// #[derive(Deserialize, JsonSchema)]
73/// struct DeployInput {
74/// environment: String,
75/// dry_run: Option<bool>,
76/// }
77///
78/// let schema = input_schema_for::<DeployInput>();
79/// assert_eq!(schema["type"], "object");
80/// assert!(schema["properties"]["environment"].is_object());
81/// ```
82pub fn input_schema_for<T: JsonSchema>() -> Value {
83 let schema = schemars::schema_for!(T);
84 serde_json::to_value(schema).expect("schema serialization cannot fail")
85}
86
87/// Boxed future returned by [`WorkflowHandler::execute`].
88pub type HandlerFuture<'a> = Pin<Box<dyn Future<Output = Result<(), EngineError>> + Send + 'a>>;
89
90/// Metadata about a workflow, returned by [`WorkflowHandler::describe`].
91///
92/// Contains a human-readable description and optional Rust source code
93/// for display in the dashboard.
94#[derive(Debug, Clone, Serialize)]
95pub struct WorkflowInfo {
96 /// Human-readable description of what the workflow does.
97 pub description: String,
98 /// Optional Rust source code of the handler (for UI display).
99 pub source_code: Option<String>,
100 /// Names of sub-workflows invoked by this handler.
101 #[serde(default, skip_serializing_if = "Vec::is_empty")]
102 pub sub_workflows: Vec<String>,
103 /// Optional `/`-separated category path used to group workflows in the UI tree.
104 ///
105 /// A value like `"data/etl"` places the workflow under `data` → `etl`.
106 /// `None` means the workflow is uncategorized.
107 #[serde(default, skip_serializing_if = "Option::is_none")]
108 pub category: Option<String>,
109 /// Handler version string, used to trace which code produced a given run.
110 #[serde(default, skip_serializing_if = "Option::is_none")]
111 pub version: Option<String>,
112 /// JSON Schema describing the expected input payload.
113 ///
114 /// When present, the dashboard renders a dynamic form from this schema
115 /// and the engine validates the payload before creating a run.
116 #[serde(default, skip_serializing_if = "Option::is_none")]
117 pub input_schema: Option<Value>,
118 /// Labels automatically applied to every run of this workflow.
119 #[serde(default, skip_serializing_if = "HashMap::is_empty")]
120 pub default_labels: HashMap<String, String>,
121 /// Optional cron schedule for automatic execution.
122 #[serde(default, skip_serializing_if = "Option::is_none")]
123 pub schedule: Option<CronSchedule>,
124 /// Default cumulative cost cap applied to runs of this workflow, in USD.
125 ///
126 /// Overridden by a cap supplied at run creation, and takes precedence over
127 /// the server-wide default. `None` means the handler declares no default.
128 #[serde(default, skip_serializing_if = "Option::is_none")]
129 pub default_max_cost_usd: Option<Decimal>,
130}
131
132/// A dynamic workflow handler with context-aware step chaining.
133///
134/// Implement this trait to define workflows where each step can use
135/// the output of previous steps. Register handlers with
136/// [`Engine::register`](crate::engine::Engine::register) and execute
137/// them by name.
138///
139/// # Why `Pin<Box<dyn Future>>` instead of `async fn`?
140///
141/// The handler must be object-safe (`dyn WorkflowHandler`) to allow
142/// registering different handler types in the engine's registry.
143pub trait WorkflowHandler: Send + Sync {
144 /// The workflow name used for registration and lookup.
145 fn name(&self) -> &str;
146
147 /// Handler version string, used to trace which code version produced a run.
148 ///
149 /// Override this to return a meaningful version (semver, git SHA, build
150 /// hash, etc.). The default is `None`.
151 fn version(&self) -> Option<&str> {
152 None
153 }
154
155 /// Optional `/`-separated category path used to group workflows in the UI tree.
156 ///
157 /// Return a value like `"data/etl"` to place the workflow under `data` → `etl`.
158 /// The default is `None` (uncategorized).
159 ///
160 /// Validation (empty segments, leading or trailing `/`, `//`, whitespace
161 /// segments) is enforced at registration time by
162 /// [`Engine::register`](crate::engine::Engine::register).
163 fn category(&self) -> Option<&str> {
164 None
165 }
166
167 /// Return a JSON Schema describing the expected input payload.
168 ///
169 /// When present, the dashboard renders a dynamic form from this schema
170 /// and the engine validates the payload before creating a run.
171 /// The default is `None` (no schema, free-form payload).
172 fn input_schema(&self) -> Option<Value> {
173 None
174 }
175
176 /// Labels automatically applied to every run of this workflow.
177 ///
178 /// These are merged with any labels provided at run creation time.
179 /// User-provided labels take precedence over defaults.
180 fn default_labels(&self) -> HashMap<String, String> {
181 HashMap::new()
182 }
183
184 /// Optional cron schedule for automatic execution.
185 ///
186 /// Return a [`CronSchedule`] built from a cron expression
187 /// (5 or 6 fields, as supported by [`croner`]).
188 ///
189 /// When set, the engine exposes this handler via
190 /// [`Engine::scheduled_handlers`](crate::engine::Engine::scheduled_handlers)
191 /// so the runtime can wire it into a cron scheduler automatically.
192 ///
193 /// The default is `None` (no automatic scheduling).
194 ///
195 /// # Examples
196 ///
197 /// ```
198 /// # use ironflow_engine::handler::{WorkflowHandler, HandlerFuture};
199 /// # use ironflow_engine::context::WorkflowContext;
200 /// # use ironflow_engine::schedule::CronSchedule;
201 /// struct HourlySync;
202 ///
203 /// impl WorkflowHandler for HourlySync {
204 /// fn name(&self) -> &str { "hourly-sync" }
205 /// fn schedule(&self) -> Option<&CronSchedule> {
206 /// // In practice, store as a field or use `std::sync::LazyLock`.
207 /// None
208 /// }
209 /// fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
210 /// Box::pin(async move { Ok(()) })
211 /// }
212 /// }
213 /// ```
214 fn schedule(&self) -> Option<&CronSchedule> {
215 None
216 }
217
218 /// Default cumulative cost cap for runs of this workflow, in USD.
219 ///
220 /// Applied when the run creation request does not supply one. Takes
221 /// precedence over the server-wide
222 /// [`IRONFLOW_DEFAULT_RUN_MAX_COST_USD`](crate::budget::DEFAULT_RUN_MAX_COST_ENV).
223 /// The default is `None` (fall back to the server default, or no cap).
224 ///
225 /// # Examples
226 ///
227 /// ```
228 /// # use ironflow_engine::handler::{WorkflowHandler, HandlerFuture};
229 /// # use ironflow_engine::context::WorkflowContext;
230 /// use rust_decimal::Decimal;
231 ///
232 /// struct ExpensiveAnalysis;
233 ///
234 /// impl WorkflowHandler for ExpensiveAnalysis {
235 /// fn name(&self) -> &str { "expensive-analysis" }
236 /// fn default_max_cost_usd(&self) -> Option<Decimal> {
237 /// Some(Decimal::new(500, 2)) // $5.00
238 /// }
239 /// fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
240 /// Box::pin(async move { Ok(()) })
241 /// }
242 /// }
243 ///
244 /// assert_eq!(ExpensiveAnalysis.default_max_cost_usd(), Some(Decimal::new(500, 2)));
245 /// ```
246 fn default_max_cost_usd(&self) -> Option<Decimal> {
247 None
248 }
249
250 /// Return metadata about this workflow (description, source code).
251 ///
252 /// Override this to provide a description and source code for the
253 /// dashboard UI. The default returns an empty description with no source
254 /// but propagates [`WorkflowHandler::category`],
255 /// [`WorkflowHandler::version`], [`WorkflowHandler::input_schema`],
256 /// [`WorkflowHandler::default_labels`],
257 /// and [`WorkflowHandler::schedule`].
258 fn describe(&self) -> WorkflowInfo {
259 WorkflowInfo {
260 description: String::new(),
261 source_code: None,
262 sub_workflows: Vec::new(),
263 category: self.category().map(str::to_string),
264 version: self.version().map(str::to_string),
265 input_schema: self.input_schema(),
266 default_labels: self.default_labels(),
267 schedule: self.schedule().cloned(),
268 default_max_cost_usd: self.default_max_cost_usd(),
269 }
270 }
271
272 /// Execute the workflow with the given context.
273 ///
274 /// The context provides [`shell`](WorkflowContext::shell),
275 /// [`http`](WorkflowContext::http), and [`agent`](WorkflowContext::agent)
276 /// methods that automatically persist each step.
277 ///
278 /// # Errors
279 ///
280 /// Return [`EngineError`] if any step fails. The engine will mark
281 /// the run as `Failed` and record the error.
282 fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a>;
283}
284
285#[cfg(test)]
286mod tests {
287 use super::*;
288 use serde::{Deserialize, Serialize};
289
290 #[derive(Debug, Serialize, Deserialize, JsonSchema)]
291 struct TestInput {
292 environment: String,
293 #[serde(default)]
294 dry_run: bool,
295 }
296
297 struct MinimalHandler;
298
299 impl WorkflowHandler for MinimalHandler {
300 fn name(&self) -> &str {
301 "minimal"
302 }
303
304 fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
305 Box::pin(async { Ok(()) })
306 }
307 }
308
309 struct FullFeaturedHandler;
310
311 impl WorkflowHandler for FullFeaturedHandler {
312 fn name(&self) -> &str {
313 "full"
314 }
315
316 fn version(&self) -> Option<&str> {
317 Some("1.2.0")
318 }
319
320 fn category(&self) -> Option<&str> {
321 Some("data/etl")
322 }
323
324 fn input_schema(&self) -> Option<Value> {
325 Some(input_schema_for::<TestInput>())
326 }
327
328 fn default_labels(&self) -> HashMap<String, String> {
329 HashMap::from([
330 ("team".to_string(), "platform".to_string()),
331 ("env".to_string(), "prod".to_string()),
332 ])
333 }
334
335 fn default_max_cost_usd(&self) -> Option<Decimal> {
336 Some(Decimal::new(750, 2))
337 }
338
339 fn describe(&self) -> WorkflowInfo {
340 WorkflowInfo {
341 description: "Full-featured test handler".to_string(),
342 source_code: Some("fn test() {}".to_string()),
343 sub_workflows: vec!["helper".to_string()],
344 category: self.category().map(str::to_string),
345 version: self.version().map(str::to_string),
346 input_schema: self.input_schema(),
347 default_labels: self.default_labels(),
348 schedule: self.schedule().cloned(),
349 default_max_cost_usd: self.default_max_cost_usd(),
350 }
351 }
352
353 fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
354 Box::pin(async { Ok(()) })
355 }
356 }
357
358 #[test]
359 fn minimal_handler_has_required_name() {
360 let handler = MinimalHandler;
361 assert_eq!(handler.name(), "minimal");
362 }
363
364 #[test]
365 fn minimal_handler_defaults_to_no_version() {
366 let handler = MinimalHandler;
367 assert_eq!(handler.version(), None);
368 }
369
370 #[test]
371 fn minimal_handler_defaults_to_no_category() {
372 let handler = MinimalHandler;
373 assert_eq!(handler.category(), None);
374 }
375
376 #[test]
377 fn minimal_handler_defaults_to_no_schema() {
378 let handler = MinimalHandler;
379 assert_eq!(handler.input_schema(), None);
380 }
381
382 #[test]
383 fn minimal_handler_defaults_to_empty_labels() {
384 let handler = MinimalHandler;
385 let labels = handler.default_labels();
386 assert!(labels.is_empty());
387 }
388
389 #[test]
390 fn minimal_handler_defaults_to_no_schedule() {
391 let handler = MinimalHandler;
392 assert_eq!(handler.schedule(), None);
393 }
394
395 #[test]
396 fn minimal_handler_describe_reflects_defaults() {
397 let handler = MinimalHandler;
398 let info = handler.describe();
399 assert_eq!(info.description, "");
400 assert_eq!(info.source_code, None);
401 assert_eq!(info.sub_workflows, Vec::<String>::new());
402 assert_eq!(info.category, None);
403 assert_eq!(info.version, None);
404 assert_eq!(info.input_schema, None);
405 assert!(info.default_labels.is_empty());
406 assert_eq!(info.schedule, None);
407 }
408
409 #[test]
410 fn full_handler_returns_all_metadata() {
411 let handler = FullFeaturedHandler;
412 assert_eq!(handler.name(), "full");
413 assert_eq!(handler.version(), Some("1.2.0"));
414 assert_eq!(handler.category(), Some("data/etl"));
415 assert!(handler.input_schema().is_some());
416 }
417
418 #[test]
419 fn full_handler_default_labels_are_set() {
420 let handler = FullFeaturedHandler;
421 let labels = handler.default_labels();
422 assert_eq!(labels.get("team"), Some(&"platform".to_string()));
423 assert_eq!(labels.get("env"), Some(&"prod".to_string()));
424 }
425
426 #[test]
427 fn full_handler_describe_includes_all_fields() {
428 let handler = FullFeaturedHandler;
429 let info = handler.describe();
430 assert_eq!(info.description, "Full-featured test handler");
431 assert_eq!(info.source_code, Some("fn test() {}".to_string()));
432 assert_eq!(info.sub_workflows, vec!["helper".to_string()]);
433 assert_eq!(info.category, Some("data/etl".to_string()));
434 assert_eq!(info.version, Some("1.2.0".to_string()));
435 assert!(info.input_schema.is_some());
436 assert_eq!(info.default_labels.len(), 2);
437 }
438
439 #[test]
440 fn input_schema_for_generates_json_schema() {
441 let schema = input_schema_for::<TestInput>();
442 assert_eq!(schema["type"], "object");
443 assert!(schema["properties"]["environment"].is_object());
444 assert!(schema["properties"]["dry_run"].is_object());
445 }
446
447 #[test]
448 fn input_schema_for_preserves_serde_attributes() {
449 let schema = input_schema_for::<TestInput>();
450 let properties = &schema["properties"];
451 assert!(properties.is_object());
452 assert!(properties.get("environment").is_some());
453 assert!(properties.get("dry_run").is_some());
454 }
455
456 #[test]
457 fn minimal_handler_defaults_to_no_max_cost() {
458 assert!(MinimalHandler.default_max_cost_usd().is_none());
459 assert!(MinimalHandler.describe().default_max_cost_usd.is_none());
460 }
461
462 #[test]
463 fn describe_propagates_handler_max_cost() {
464 assert_eq!(
465 FullFeaturedHandler.describe().default_max_cost_usd,
466 Some(Decimal::new(750, 2))
467 );
468 }
469
470 #[test]
471 fn workflow_info_omits_absent_max_cost_from_json() {
472 let json = serde_json::to_value(MinimalHandler.describe()).expect("serialize");
473 assert!(json.get("default_max_cost_usd").is_none());
474 }
475
476 #[test]
477 fn workflow_info_serializes_with_skip_empty() {
478 let info = WorkflowInfo {
479 description: "test".to_string(),
480 source_code: None,
481 sub_workflows: Vec::new(),
482 category: None,
483 version: None,
484 input_schema: None,
485 default_labels: HashMap::new(),
486 schedule: None,
487 default_max_cost_usd: None,
488 };
489
490 let json = serde_json::to_value(&info).expect("serialize");
491 assert_eq!(json["description"], "test");
492 // Optional fields with skip_serializing_if may still be present or absent
493 // depending on the serde configuration. Just verify the description is there.
494 assert!(json.is_object());
495 }
496
497 #[test]
498 fn workflow_info_serializes_with_values() {
499 let info = WorkflowInfo {
500 description: "test".to_string(),
501 source_code: Some("code".to_string()),
502 sub_workflows: vec!["sub".to_string()],
503 category: Some("cat".to_string()),
504 version: Some("1.0.0".to_string()),
505 input_schema: Some(serde_json::json!({"type": "object"})),
506 default_labels: HashMap::from([("key".to_string(), "value".to_string())]),
507 schedule: Some(CronSchedule::new("0 0 * * * *").unwrap()),
508 default_max_cost_usd: Some(Decimal::new(750, 2)),
509 };
510
511 let json = serde_json::to_value(&info).expect("serialize");
512 assert_eq!(json["description"], "test");
513 assert_eq!(json["source_code"], "code");
514 assert_eq!(json["sub_workflows"][0], "sub");
515 assert_eq!(json["category"], "cat");
516 assert_eq!(json["version"], "1.0.0");
517 assert_eq!(json["default_labels"]["key"], "value");
518 assert_eq!(json["schedule"], "0 0 * * * *");
519 }
520}