ironflow_engine/context/accessors.rs
1//! Constructors and plain accessors for [`WorkflowContext`].
2//!
3//! Everything here reads or writes a single field: the run identity, the
4//! wiring the engine attaches before execution, the run-level budget, and the
5//! counters a handler can inspect while it runs.
6
7use std::collections::HashMap;
8use std::sync::Arc;
9
10use chrono::{DateTime, Utc};
11use rust_decimal::Decimal;
12use serde::Serialize;
13use serde_json::{Value, from_value, to_value};
14use tracing::error;
15use uuid::Uuid;
16
17use ironflow_core::decision::DecisionProvider;
18use ironflow_core::provider::AgentProvider;
19use ironflow_core::trace_context::WorkflowTraceContext;
20use ironflow_store::error::StoreError;
21use ironflow_store::models::{Step, TriggerKind};
22use ironflow_store::store::Store;
23#[cfg(feature = "secret-store")]
24use ironflow_store::workflow_secrets::ScopedSecretStore;
25
26use crate::artifact::ArtifactSink;
27use crate::error::EngineError;
28use crate::executor::{StepInterceptor, StepResult};
29use crate::guard::{SharedGuardState, WorkflowGuardConfig};
30use crate::log_sender::LogSender;
31use crate::notify::WorkflowEventBus;
32#[cfg(not(feature = "secret-store"))]
33use crate::operation::NoopSecretResolver;
34use crate::operation::{OperationContext, SecretResolver};
35use crate::plan::{SharedPlanRecorder, lock_plan};
36
37use super::cost::BudgetReservation;
38use super::{HandlerResolver, WorkflowContext};
39
40impl WorkflowContext {
41 /// Create a new context for a run.
42 ///
43 /// Not typically called directly — the [`Engine`](crate::engine::Engine)
44 /// creates this when executing a [`WorkflowHandler`](crate::handler::WorkflowHandler).
45 pub fn new(
46 run_id: Uuid,
47 workflow_name: String,
48 store: Arc<dyn Store>,
49 provider: Arc<dyn AgentProvider>,
50 ) -> Self {
51 let trace_context = WorkflowTraceContext::from_workflow_run_id(&run_id.to_string());
52 Self {
53 run_id,
54 root_run_id: run_id,
55 workflow_name,
56 store,
57 provider,
58 decision_provider: None,
59 handler_resolver: None,
60 position: 0,
61 last_step_ids: Vec::new(),
62 total_cost_usd: Decimal::ZERO,
63 total_duration_ms: 0,
64 max_cost_usd: None,
65 inherited_cost_usd: Decimal::ZERO,
66 shared_cost: None,
67 replay_steps: HashMap::new(),
68 replay_wave_steps: HashMap::new(),
69 granted_approvals: HashMap::new(),
70 answered_inputs: HashMap::new(),
71 interrupted_children: HashMap::new(),
72 attempt: 1,
73 carried_duration_ms: 0,
74 log_sender: None,
75 artifact_sink: None,
76 has_allowed_failure: false,
77 error_handlers: Vec::new(),
78 guard_state: None,
79 guard_config: None,
80 step_results: Vec::new(),
81 event_bus: None,
82 interceptor: None,
83 trace_context,
84 operation_ctx: None,
85 run_created_at: None,
86 plan: None,
87 output: None,
88 worker_tags: None,
89 sibling_suspensions: Vec::new(),
90 }
91 }
92
93 /// Create a new context with a handler resolver for sub-workflow support.
94 ///
95 /// The resolver is called when [`workflow`](Self::workflow) is invoked to
96 /// look up registered handlers by name.
97 pub(crate) fn with_handler_resolver(
98 run_id: Uuid,
99 workflow_name: String,
100 store: Arc<dyn Store>,
101 provider: Arc<dyn AgentProvider>,
102 resolver: HandlerResolver,
103 ) -> Self {
104 let trace_context = WorkflowTraceContext::from_workflow_run_id(&run_id.to_string());
105 Self {
106 run_id,
107 root_run_id: run_id,
108 workflow_name,
109 store,
110 provider,
111 decision_provider: None,
112 handler_resolver: Some(resolver),
113 position: 0,
114 last_step_ids: Vec::new(),
115 total_cost_usd: Decimal::ZERO,
116 total_duration_ms: 0,
117 max_cost_usd: None,
118 inherited_cost_usd: Decimal::ZERO,
119 shared_cost: None,
120 replay_steps: HashMap::new(),
121 replay_wave_steps: HashMap::new(),
122 granted_approvals: HashMap::new(),
123 answered_inputs: HashMap::new(),
124 interrupted_children: HashMap::new(),
125 attempt: 1,
126 carried_duration_ms: 0,
127 log_sender: None,
128 artifact_sink: None,
129 has_allowed_failure: false,
130 error_handlers: Vec::new(),
131 guard_state: None,
132 guard_config: None,
133 step_results: Vec::new(),
134 event_bus: None,
135 interceptor: None,
136 trace_context,
137 operation_ctx: None,
138 run_created_at: None,
139 plan: None,
140 output: None,
141 worker_tags: None,
142 sibling_suspensions: Vec::new(),
143 }
144 }
145
146 /// Attach a log sender for real-time step output streaming.
147 pub fn set_log_sender(&mut self, sender: LogSender) {
148 self.log_sender = Some(sender);
149 }
150
151 /// Attach the backend that stores and serves artifact bytes.
152 ///
153 /// Without one, any step that declares an output or calls
154 /// [`put_artifact`](Self::put_artifact) fails with
155 /// [`EngineError::ArtifactsUnavailable`]. Every other step is unaffected,
156 /// so an existing deployment keeps working until artifacts are configured.
157 ///
158 /// # Examples
159 ///
160 /// ```no_run
161 /// use std::sync::Arc;
162 ///
163 /// use ironflow_engine::artifact::ArtifactSink;
164 /// use ironflow_engine::context::WorkflowContext;
165 ///
166 /// # fn example(ctx: &mut WorkflowContext, sink: Arc<dyn ArtifactSink>) {
167 /// ctx.set_artifact_sink(sink);
168 /// # }
169 /// ```
170 pub fn set_artifact_sink(&mut self, sink: Arc<dyn ArtifactSink>) {
171 self.artifact_sink = Some(sink);
172 }
173
174 /// Return the W3C trace context for this workflow run.
175 ///
176 /// The trace context is derived from the run ID and can be used to
177 /// correlate spans across distributed services. Each step automatically
178 /// receives a [`child`](WorkflowTraceContext::child) context.
179 pub fn trace_context(&self) -> &WorkflowTraceContext {
180 &self.trace_context
181 }
182
183 /// Attach a workflow guard configuration and shared state.
184 ///
185 /// When set, the guard is checked before every sub-workflow invocation.
186 /// The shared state is propagated to child workflows so that limits
187 /// apply globally across the entire run tree.
188 ///
189 /// # Examples
190 ///
191 /// ```no_run
192 /// use ironflow_engine::context::WorkflowContext;
193 /// use ironflow_engine::guard::{WorkflowGuardConfig, new_shared_guard_state};
194 ///
195 /// # fn example(ctx: &mut WorkflowContext) {
196 /// ctx.set_guard(WorkflowGuardConfig::default(), new_shared_guard_state());
197 /// # }
198 /// ```
199 pub fn set_guard(&mut self, config: WorkflowGuardConfig, state: SharedGuardState) {
200 self.guard_config = Some(config);
201 self.guard_state = Some(state);
202 }
203
204 /// The current guard configuration, if any.
205 pub fn guard_config(&self) -> Option<&WorkflowGuardConfig> {
206 self.guard_config.as_ref()
207 }
208
209 /// Attach a [`WorkflowEventBus`] for per-run real-time monitoring.
210 ///
211 /// When set, step transitions automatically publish
212 /// [`WorkflowEvent`](crate::notify::WorkflowEvent)s to the bus.
213 pub fn set_event_bus(&mut self, bus: WorkflowEventBus) {
214 self.event_bus = Some(bus);
215 }
216
217 /// Attach a [`StepInterceptor`] that resolves steps without executing them.
218 ///
219 /// Wired by the [`Engine`](crate::engine::Engine) from
220 /// [`Engine::with_step_interceptor`](crate::engine::Engine::with_step_interceptor).
221 /// Intended for tests: see [`crate::testing::TestEngine`].
222 ///
223 /// # Examples
224 ///
225 /// ```no_run
226 /// use std::sync::Arc;
227 ///
228 /// use ironflow_engine::context::WorkflowContext;
229 /// use ironflow_engine::executor::StepInterceptor;
230 ///
231 /// # fn example(ctx: &mut WorkflowContext, interceptor: Arc<dyn StepInterceptor>) {
232 /// ctx.set_step_interceptor(interceptor);
233 /// # }
234 /// ```
235 pub fn set_step_interceptor(&mut self, interceptor: Arc<dyn StepInterceptor>) {
236 self.interceptor = Some(interceptor);
237 }
238
239 /// Declare the worker tags carried by the process running this context.
240 ///
241 /// Wired by the [`Engine`](crate::engine::Engine) from
242 /// [`Engine::set_worker_tags`](crate::engine::Engine::set_worker_tags).
243 pub(crate) fn set_worker_tags(&mut self, tags: Arc<Vec<String>>) {
244 self.worker_tags = Some(tags);
245 }
246
247 /// The step interceptor attached to this context, if any.
248 pub(crate) fn step_interceptor(&self) -> Option<&Arc<dyn StepInterceptor>> {
249 self.interceptor.as_ref()
250 }
251
252 /// Attach a [`DecisionProvider`] backend for `ctx.decision(...)` steps.
253 ///
254 /// Not typically called directly -- the [`Engine`](crate::engine::Engine)
255 /// wires this from [`Engine::with_decision_provider`](crate::engine::Engine::with_decision_provider).
256 pub fn set_decision_provider(&mut self, provider: Arc<dyn DecisionProvider>) {
257 self.decision_provider = Some(provider);
258 }
259
260 /// Attach a plan recorder, switching this context to plan mode.
261 ///
262 /// Every step method then records its intent instead of executing it.
263 pub(crate) fn set_plan(&mut self, plan: SharedPlanRecorder) {
264 self.plan = Some(plan);
265 }
266
267 /// The shared recorder, when this context is planning.
268 pub(crate) fn plan(&self) -> Option<&SharedPlanRecorder> {
269 self.plan.as_ref()
270 }
271
272 /// Whether this context records a plan instead of executing steps.
273 ///
274 /// # Examples
275 ///
276 /// ```no_run
277 /// use ironflow_engine::context::WorkflowContext;
278 ///
279 /// # fn example(ctx: &WorkflowContext) {
280 /// if ctx.is_planning() {
281 /// // No command runs, no request is sent: only the plan is recorded.
282 /// }
283 /// # }
284 /// ```
285 pub fn is_planning(&self) -> bool {
286 self.plan.is_some()
287 }
288
289 /// Set the typed output of this run.
290 ///
291 /// The value is persisted on the run when the execution ends, whether it
292 /// completes, fails or is cancelled, so a verdict set before a handler
293 /// error is kept. A parent workflow reads it back with
294 /// [`SubWorkflowOutput::output`](crate::executor::SubWorkflowOutput::output).
295 /// The last call wins. While the context is planning, the call is a no-op
296 /// and nothing is serialized.
297 ///
298 /// # Errors
299 ///
300 /// Returns [`EngineError::Serialization`] if `value` cannot be serialized
301 /// to JSON.
302 ///
303 /// # Examples
304 ///
305 /// ```no_run
306 /// use ironflow_engine::context::WorkflowContext;
307 /// use ironflow_engine::error::EngineError;
308 /// use serde::Serialize;
309 ///
310 /// #[derive(Serialize)]
311 /// struct Review {
312 /// approved: bool,
313 /// findings: Vec<String>,
314 /// }
315 ///
316 /// # fn example(ctx: &mut WorkflowContext) -> Result<(), EngineError> {
317 /// ctx.set_output(&Review {
318 /// approved: true,
319 /// findings: Vec::new(),
320 /// })?;
321 /// # Ok(())
322 /// # }
323 /// ```
324 pub fn set_output<T: Serialize>(&mut self, value: &T) -> Result<(), EngineError> {
325 if self.plan.is_some() {
326 return Ok(());
327 }
328 self.output = Some(to_value(value)?);
329 Ok(())
330 }
331
332 /// The output set with [`set_output`](Self::set_output), if any.
333 pub(crate) fn output(&self) -> Option<&Value> {
334 self.output.as_ref()
335 }
336
337 /// Suspensions of wave children that the error the run returned does not
338 /// carry, each one still to publish.
339 pub(crate) fn sibling_suspensions(&self) -> &[EngineError] {
340 &self.sibling_suspensions
341 }
342
343 /// Seed the context with the run's attempt number and the totals already
344 /// accumulated by previous attempts.
345 ///
346 /// Called by the engine before executing a handler. Steps created by this
347 /// context belong to `attempt`, and the cost and duration it reports at the
348 /// end cover the whole run, not just this attempt.
349 pub(crate) fn carry_over_run_totals(
350 &mut self,
351 attempt: u32,
352 cost_usd: Decimal,
353 duration_ms: u64,
354 ) {
355 self.attempt = attempt;
356 self.total_cost_usd = cost_usd;
357 self.carried_duration_ms = duration_ms;
358 }
359
360 /// Record when the run was created, so a wait step can bound the signals
361 /// it accepts without reading the run back.
362 pub(crate) fn set_run_created_at(&mut self, created_at: DateTime<Utc>) {
363 self.run_created_at = Some(created_at);
364 }
365
366 /// When the run was created: the value set by the engine, otherwise read
367 /// from the store.
368 pub(crate) async fn run_created_at(&self) -> Result<DateTime<Utc>, EngineError> {
369 if let Some(created_at) = self.run_created_at {
370 return Ok(created_at);
371 }
372 let run = self
373 .store
374 .get_run(self.run_id)
375 .await?
376 .ok_or(EngineError::Store(StoreError::RunNotFound(self.run_id)))?;
377 Ok(run.created_at)
378 }
379
380 /// Wall-clock duration already recorded on the run by previous attempts.
381 pub(crate) fn carried_duration_ms(&self) -> u64 {
382 self.carried_duration_ms
383 }
384
385 /// The run attempt this context is executing (1-based).
386 pub fn attempt(&self) -> u32 {
387 self.attempt
388 }
389
390 /// Set the cumulative cost cap enforced before every agent step.
391 ///
392 /// Called by the [`Engine`](crate::engine::Engine) with the run's persisted
393 /// `max_cost_usd`. `None` disables the check.
394 ///
395 /// # Examples
396 ///
397 /// ```no_run
398 /// use ironflow_engine::context::WorkflowContext;
399 /// use rust_decimal::Decimal;
400 ///
401 /// # fn example(ctx: &mut WorkflowContext) {
402 /// ctx.set_max_cost_usd(Some(Decimal::new(200, 2))); // $2.00
403 /// # }
404 /// ```
405 pub fn set_max_cost_usd(&mut self, cap: Option<Decimal>) {
406 self.max_cost_usd = cap;
407 }
408
409 /// The cumulative cost cap of this run, if any.
410 pub fn max_cost_usd(&self) -> Option<Decimal> {
411 self.max_cost_usd
412 }
413
414 /// Total cost charged against the cap: this run plus every ancestor run.
415 ///
416 /// For a top-level run this equals [`total_cost_usd`](Self::total_cost_usd).
417 /// For a sub-workflow it also includes what the parent chain already spent.
418 /// Inside a [`workflows_with`](Self::workflows_with) wave it is the live
419 /// total of the whole run tree, siblings included.
420 pub fn charged_cost_usd(&self) -> Decimal {
421 match &self.shared_cost {
422 Some(shared) => shared.spent(),
423 None => self.inherited_cost_usd + self.total_cost_usd,
424 }
425 }
426
427 /// Add the cost a step of this run reported, to the run and to the live
428 /// total it shares with its wave, if any.
429 pub(super) fn charge_cost(&mut self, cost: Decimal) {
430 self.total_cost_usd += cost;
431 if let Some(shared) = &self.shared_cost {
432 shared.charge(cost);
433 }
434 }
435
436 /// Reject the upcoming agent work when it would cross the run's cost cap.
437 ///
438 /// `step_budget` is the declared budget of the step (or the sum of budgets
439 /// for a parallel wave). Called *before* any step record is created so a
440 /// refused run never launches the work it could not afford.
441 ///
442 /// Inside a wave, the admitted budget stays reserved in the live total
443 /// until the returned reservation is dropped: a sibling checking the cap
444 /// meanwhile counts it, so concurrent steps never share the same room.
445 /// Keep the reservation until the step's cost has been charged.
446 ///
447 /// # Errors
448 ///
449 /// Returns [`EngineError::RunBudgetExceeded`] when
450 /// `charged_cost + step_budget` exceeds the cap, the charged cost
451 /// including, inside a wave, the budgets reserved by running siblings.
452 pub(super) fn check_run_budget(
453 &self,
454 step_budget: Decimal,
455 ) -> Result<BudgetReservation, EngineError> {
456 let Some(limit) = self.max_cost_usd else {
457 return Ok(BudgetReservation::none());
458 };
459
460 let spent = match &self.shared_cost {
461 Some(shared) => match shared.reserve(step_budget, limit) {
462 Ok(reservation) => return Ok(reservation),
463 Err(committed) => committed,
464 },
465 None => {
466 let spent = self.charged_cost_usd();
467 if spent + step_budget <= limit {
468 return Ok(BudgetReservation::none());
469 }
470 spent
471 }
472 };
473
474 error!(
475 run_id = %self.run_id,
476 limit_usd = %limit,
477 spent_usd = %spent,
478 step_budget_usd = %step_budget,
479 "run cost cap reached, refusing agent step"
480 );
481
482 Err(EngineError::RunBudgetExceeded {
483 run_id: self.run_id,
484 limit_usd: limit,
485 spent_usd: spent,
486 step_budget_usd: step_budget,
487 })
488 }
489
490 /// The run ID this context is executing for.
491 pub fn run_id(&self) -> Uuid {
492 self.run_id
493 }
494
495 /// The top-level run this context belongs to: [`run_id`](Self::run_id)
496 /// itself, or, inside a sub-workflow, the run that started the chain.
497 ///
498 /// The engine stamps it on agent pods as
499 /// [`LABEL_ROOT_RUN_ID`](ironflow_core::provider::LABEL_ROOT_RUN_ID); set
500 /// the same label on a pod you create yourself (`PodRun`, `JobRun`) so
501 /// that a retry of the top-level run deletes what a dead attempt left.
502 ///
503 /// # Examples
504 ///
505 /// ```no_run
506 /// use ironflow_core::provider::LABEL_ROOT_RUN_ID;
507 /// use ironflow_engine::context::WorkflowContext;
508 ///
509 /// fn root_label(ctx: &WorkflowContext) -> (&'static str, String) {
510 /// (LABEL_ROOT_RUN_ID, ctx.root_run_id().to_string())
511 /// }
512 /// ```
513 pub fn root_run_id(&self) -> Uuid {
514 self.root_run_id
515 }
516
517 /// The workflow name this run belongs to.
518 pub fn workflow_name(&self) -> &str {
519 &self.workflow_name
520 }
521
522 /// Accumulated cost across all executed steps so far.
523 pub fn total_cost_usd(&self) -> Decimal {
524 self.total_cost_usd
525 }
526
527 /// Whether at least one `allow_failure` step failed during this run.
528 pub fn has_allowed_failure(&self) -> bool {
529 self.has_allowed_failure
530 }
531
532 /// Accumulated duration across all executed steps so far.
533 pub fn total_duration_ms(&self) -> u64 {
534 self.total_duration_ms
535 }
536
537 /// Enriched results of all completed steps in execution order.
538 pub fn step_results(&self) -> &[StepResult] {
539 &self.step_results
540 }
541
542 /// Return a [`ScopedSecretStore`] scoped to this workflow.
543 ///
544 /// Secrets are namespaced under `workflows/<uuid>/` where the UUID is
545 /// deterministically derived from the workflow name (UUID v5). This means
546 /// all runs of the same workflow share the same secret namespace.
547 ///
548 /// Requires the `secret-store` feature.
549 ///
550 /// # Errors
551 ///
552 /// Returns [`EngineError::InvalidWorkflow`] in plan mode: planning has no
553 /// side effect, so it never reads or writes a secret.
554 ///
555 /// # Examples
556 ///
557 /// ```no_run
558 /// use ironflow_engine::context::WorkflowContext;
559 /// use ironflow_engine::error::EngineError;
560 ///
561 /// # async fn example(ctx: &WorkflowContext) -> Result<(), EngineError> {
562 /// let secrets = ctx.secrets()?;
563 /// secrets.set("api_token", "sk-ant-12345").await.map_err(EngineError::Store)?;
564 /// let token = secrets.get("api_token").await.map_err(EngineError::Store)?;
565 /// # Ok(())
566 /// # }
567 /// ```
568 #[cfg(feature = "secret-store")]
569 pub fn secrets(&self) -> Result<ScopedSecretStore, EngineError> {
570 if self.plan().is_some() {
571 return Err(EngineError::InvalidWorkflow(
572 "ctx.secrets() is not available in plan mode: planning has no side effect".into(),
573 ));
574 }
575 let workflow_uuid = Uuid::new_v5(&Uuid::NAMESPACE_OID, self.workflow_name.as_bytes());
576 Ok(ScopedSecretStore::for_workflow(
577 workflow_uuid,
578 self.store.clone(),
579 ))
580 }
581
582 pub(super) fn ensure_operation_ctx(&mut self) -> &OperationContext {
583 self.operation_ctx.get_or_insert_with(|| {
584 #[cfg(feature = "secret-store")]
585 let secrets: Arc<dyn SecretResolver> = {
586 let workflow_uuid =
587 Uuid::new_v5(&Uuid::NAMESPACE_OID, self.workflow_name.as_bytes());
588 Arc::new(ScopedSecretStore::for_workflow(
589 workflow_uuid,
590 self.store.clone(),
591 ))
592 };
593 #[cfg(not(feature = "secret-store"))]
594 let secrets: Arc<dyn SecretResolver> = Arc::new(NoopSecretResolver);
595
596 OperationContext::new(secrets)
597 })
598 }
599
600 /// Access the store directly (advanced usage).
601 ///
602 /// # Errors
603 ///
604 /// Returns [`EngineError::InvalidWorkflow`] in plan mode: planning has no
605 /// side effect, so it never reaches the real store.
606 ///
607 /// # Examples
608 ///
609 /// ```no_run
610 /// use ironflow_engine::context::WorkflowContext;
611 /// use ironflow_engine::error::EngineError;
612 ///
613 /// # async fn example(ctx: &WorkflowContext) -> Result<(), EngineError> {
614 /// let steps = ctx.store()?.list_steps(ctx.run_id()).await?;
615 /// println!("{} steps so far", steps.len());
616 /// # Ok(())
617 /// # }
618 /// ```
619 pub fn store(&self) -> Result<&Arc<dyn Store>, EngineError> {
620 if self.plan().is_some() {
621 return Err(EngineError::InvalidWorkflow(
622 "ctx.store() is not available in plan mode: planning has no side effect".into(),
623 ));
624 }
625 Ok(&self.store)
626 }
627
628 /// Get and increment the current position counter.
629 pub(crate) fn next_position(&mut self) -> u32 {
630 let pos = self.position;
631 self.position += 1;
632 pos
633 }
634
635 /// Access the replay steps from a previous execution.
636 pub(crate) fn replay_steps(&self) -> &HashMap<u32, Step> {
637 &self.replay_steps
638 }
639
640 /// Set the last step IDs (for dependency tracking).
641 pub(crate) fn set_last_step_ids(&mut self, ids: Vec<Uuid>) {
642 self.last_step_ids = ids;
643 }
644
645 /// Access the payload that triggered this run.
646 ///
647 /// Fetches the run from the store and returns its payload.
648 ///
649 /// # Errors
650 ///
651 /// Returns [`EngineError::Store`] if the run is not found.
652 pub async fn payload(&self) -> Result<Value, EngineError> {
653 // In plan mode the payload comes from the recorder: planning must not
654 // create, or even read, a run record.
655 if let Some(plan) = &self.plan {
656 return Ok(lock_plan(plan).payload());
657 }
658
659 let run = self
660 .store
661 .get_run(self.run_id)
662 .await?
663 .ok_or(EngineError::Store(StoreError::RunNotFound(self.run_id)))?;
664 Ok(run.payload)
665 }
666
667 /// Access the trigger that started this run.
668 ///
669 /// A run fired by a cron schedule carries
670 /// [`TriggerKind::Cron`] with the schedule id and the occurrence it
671 /// stands for, so a caught-up run knows which slot it covers. In plan
672 /// mode no run record exists and this returns [`TriggerKind::Manual`].
673 ///
674 /// # Errors
675 ///
676 /// Returns [`EngineError::Store`] if the run is not found.
677 ///
678 /// # Examples
679 ///
680 /// ```no_run
681 /// # use ironflow_engine::context::WorkflowContext;
682 /// # use ironflow_engine::error::EngineError;
683 /// use ironflow_store::models::TriggerKind;
684 ///
685 /// # async fn example(ctx: &WorkflowContext) -> Result<(), EngineError> {
686 /// if let TriggerKind::Cron { schedule_id, scheduled_for, .. } = ctx.trigger().await? {
687 /// println!("schedule {schedule_id:?} occurrence {scheduled_for:?}");
688 /// }
689 /// # Ok(())
690 /// # }
691 /// ```
692 pub async fn trigger(&self) -> Result<TriggerKind, EngineError> {
693 if self.plan.is_some() {
694 return Ok(TriggerKind::Manual);
695 }
696
697 let run = self
698 .store
699 .get_run(self.run_id)
700 .await?
701 .ok_or(EngineError::Store(StoreError::RunNotFound(self.run_id)))?;
702 Ok(run.trigger)
703 }
704
705 /// Deserialize the run payload into a typed input struct.
706 ///
707 /// Shorthand for `serde_json::from_value(ctx.payload().await?)`.
708 ///
709 /// # Errors
710 ///
711 /// Returns [`EngineError::Store`] if the run is not found, or
712 /// [`EngineError::Serialization`] if the payload does not match `T`.
713 ///
714 /// # Examples
715 ///
716 /// ```no_run
717 /// # use ironflow_engine::context::WorkflowContext;
718 /// # use ironflow_engine::error::EngineError;
719 /// use serde::Deserialize;
720 ///
721 /// #[derive(Deserialize)]
722 /// struct DeployInput {
723 /// environment: String,
724 /// dry_run: Option<bool>,
725 /// }
726 ///
727 /// # async fn example(ctx: &WorkflowContext) -> Result<(), EngineError> {
728 /// let input: DeployInput = ctx.input().await?;
729 /// # Ok(())
730 /// # }
731 /// ```
732 pub async fn input<T: serde::de::DeserializeOwned>(&self) -> Result<T, EngineError> {
733 let payload = self.payload().await?;
734 from_value(payload).map_err(EngineError::Serialization)
735 }
736}