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