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