temporalio_workflow/
workflow_interceptors.rs

1//! Intercept inbound and outbound calls made during workflow execution.
2//!
3//! Workflow interceptors allow observing, transforming, or short-circuit workflow
4//! operations without putting that behavior in each workflow implementation.
5//!
6//! [`WorkflowInterceptor`] has two groups of methods:
7//!
8//! - Inbound methods wrap calls into workflow code, such as executing the workflow or handling
9//!   a signal, query, or update.
10//! - Outbound methods wrap commands issued by workflow code, such as scheduling an activity,
11//!   starting a timer, or signaling another workflow.
12//!
13//! Each method receives a [`WorkflowNext`] continuation. An interceptor can change the input
14//! before calling [`WorkflowNext::run`], inspect or change the returned value, or deliberately not
15//! call `next` to short-circuit the operation. Most interceptors should call `next` exactly once.
16//!
17//! Async operation interceptors return [`WorkflowInterceptorFuture`]. Wrap an `async` block with
18//! [`WorkflowInterceptorFuture::new`] when work must happen after the next interceptor completes.
19//! Synchronous methods, including queries and update validators, cannot await workflow operations.
20//!
21//! Workers register interceptors with `register_workflow_interceptors` on their
22//! worker options. Interceptors are entered in insertion order for inbound calls and in reverse
23//! insertion order for outbound calls.
24//!
25//! # Determinism
26//!
27//! Interceptors execute as part of the workflow and are replayed with it. They must follow the same
28//! determinism rules as workflow code: do not read wall-clock time, perform network or filesystem
29//! I/O, use nondeterministic randomness, or await arbitrary futures. Use values from the
30//! interceptor context and SDK-provided workflow futures instead. [`WorkflowInterceptorFuture`]
31//! identifies a future for the workflow scheduler; it does not make an arbitrary future
32//! deterministic.
33//!
34//! [`WorkflowInterceptorContext::is_replaying`] and
35//! [`WorkflowInterceptorContext::is_replaying_history_events`] can be used to suppress duplicate external
36//! observability during replay, but replay state must not change commands or results that affect
37//! workflow behavior.
38//!
39//! # Example
40//!
41//! This interceptor wraps workflow execution and transforms string outputs after the workflow has
42//! completed. The constructor is passed to the worker during worker setup.
43//!
44//! ```
45//! # use temporalio_workflow::{
46//! #     WorkflowContextView,
47//! #     workflow_interceptors::{
48//! #         ExecuteWorkflowInput, ExecuteWorkflowResult, WorkflowInterceptor,
49//! #         WorkflowInterceptorConstructor, WorkflowInterceptorContext, WorkflowInterceptorFuture,
50//! #         WorkflowNext, WorkflowOutputValue,
51//! #     },
52//! # };
53//!
54//! struct UppercaseStringOutput;
55//!
56//! impl WorkflowInterceptor for UppercaseStringOutput {
57//!     fn execute<'a>(
58//!         &'a self,
59//!         _ctx: WorkflowInterceptorContext,
60//!         input: ExecuteWorkflowInput,
61//!         next: WorkflowNext<
62//!             'a,
63//!             ExecuteWorkflowInput,
64//!             WorkflowInterceptorFuture<'a, ExecuteWorkflowResult>,
65//!         >,
66//!     ) -> WorkflowInterceptorFuture<'a, ExecuteWorkflowResult> {
67//!         WorkflowInterceptorFuture::new(async move {
68//!             let output = next.run(input).await?;
69//!             if let Some(value) = output.downcast_ref::<String>() {
70//!                 return Ok(Box::new(value.to_uppercase()) as Box<dyn WorkflowOutputValue>);
71//!             }
72//!             Ok(output)
73//!         })
74//!     }
75//! }
76//!
77//! fn interceptor_constructor() -> WorkflowInterceptorConstructor {
78//!     WorkflowInterceptorConstructor::new(|_ctx: &WorkflowContextView| UppercaseStringOutput)
79//! }
80//!
81//! # let _ = interceptor_constructor();
82//! ```
83
84use crate::{
85    ActivityOptions, BaseWorkflowContext, CancellableFuture, CancellableFutureWithReason,
86    ChildWorkflowOptions, ContinueAsNewOptions, ExternalWorkflowHandle, LocalActivityOptions,
87    NexusOperationOptions, StartChildWorkflowOutput, StartedChildWorkflow, StartedNexusOperation,
88    TimerOptions, WorkflowContextView,
89    runtime::{
90        entry::WorkflowError,
91        model::{
92            CancelExternalWfResult, NexusStartResult, TimerResult, WorkflowResult,
93            WorkflowTermination,
94        },
95    },
96};
97use futures_util::{
98    FutureExt,
99    future::{Fuse, FusedFuture, LocalBoxFuture},
100};
101use std::{
102    any::Any,
103    collections::HashMap,
104    convert::Infallible,
105    future::Future,
106    pin::Pin,
107    rc::Rc,
108    sync::Arc,
109    task::{Context, Poll},
110    time::SystemTime,
111};
112use temporalio_common_wasm::{
113    ActivityDefinition, WorkflowDefinition,
114    data_converters::{
115        GenericPayloadConverter, PayloadConversionError, PayloadConverter, SerializationContext,
116        SerializationContextData, TemporalDeserializable, TemporalSerializable,
117    },
118    error::{
119        ActivityExecutionError, ChildWorkflowExecutionError, ChildWorkflowStartError,
120        WorkflowSignalError,
121    },
122    protos::temporal::api::{common::v1::Payload, failure::v1::Failure},
123    search_attributes::SearchAttributes,
124};
125
126mod workflow_output_value {
127    use super::*;
128
129    pub trait Sealed {
130        fn to_workflow_payload(
131            &self,
132            context: &SerializationContext<'_>,
133        ) -> Result<Payload, PayloadConversionError>;
134    }
135
136    impl<T> Sealed for T
137    where
138        T: Any + TemporalSerializable,
139    {
140        fn to_workflow_payload(
141            &self,
142            context: &SerializationContext<'_>,
143        ) -> Result<Payload, PayloadConversionError> {
144            context.converter.to_payload(context, self)
145        }
146    }
147}
148
149/// Type-erased workflow output carried through the workflow interceptor chain.
150pub trait WorkflowOutputValue: Any + TemporalSerializable + workflow_output_value::Sealed {
151    /// Access this value as [`Any`] for type-specific inspection.
152    fn as_any(&self) -> &dyn Any;
153}
154
155impl<T> WorkflowOutputValue for T
156where
157    T: Any + TemporalSerializable,
158{
159    fn as_any(&self) -> &dyn Any {
160        self
161    }
162}
163
164impl dyn WorkflowOutputValue {
165    /// Attempt to access the workflow output as a concrete type.
166    pub fn downcast_ref<T: Any>(&self) -> Option<&T> {
167        self.as_any().downcast_ref()
168    }
169
170    pub(crate) fn serialize_payload(
171        &self,
172        context: &SerializationContext<'_>,
173    ) -> Result<Payload, PayloadConversionError> {
174        self.to_workflow_payload(context)
175    }
176}
177
178pub(crate) fn serialize_workflow_output(
179    output: &dyn WorkflowOutputValue,
180    converter: &PayloadConverter,
181) -> Result<Payload, PayloadConversionError> {
182    let ctx = SerializationContext {
183        data: &SerializationContextData::Workflow,
184        converter,
185    };
186    output.serialize_payload(&ctx)
187}
188
189/// Result of an intercepted workflow execution.
190pub type ExecuteWorkflowResult = WorkflowResult<Box<dyn WorkflowOutputValue>>;
191
192/// Result of an intercepted signal handler.
193pub type HandleSignalResult = Result<(), WorkflowError>;
194
195/// Result of an intercepted update handler.
196pub type HandleUpdateResult = Result<Box<dyn WorkflowOutputValue>, WorkflowError>;
197
198/// Result of an intercepted query handler.
199pub type HandleQueryResult = Result<Box<dyn WorkflowOutputValue>, WorkflowError>;
200
201/// Result of an intercepted update validator.
202pub type ValidateUpdateResult = Result<(), WorkflowError>;
203
204/// Future produced by workflow interceptors.
205///
206/// The SDK polls a newly created interceptor future once while processing its activation. This
207/// runs synchronous interceptor and synchronous handler work through the first genuine
208/// [`Poll::Pending`]. Async workflow and handler bodies are not entered
209/// until normal routine polling. Awaiting a pending workflow future before calling
210/// [`WorkflowNext::run`] intentionally delays the underlying handler.
211///
212/// This type identifies futures that are polled inside workflow execution. It does not make
213/// arbitrary Rust futures deterministic. Interceptor implementations must only await workflow
214/// scheduler primitives or SDK-provided workflow futures.
215pub struct WorkflowInterceptorFuture<'a, T>(LocalBoxFuture<'a, T>);
216
217impl<'a, T> WorkflowInterceptorFuture<'a, T> {
218    /// Create a workflow interceptor future from a local future.
219    pub fn new(fut: impl Future<Output = T> + 'a) -> Self {
220        Self(fut.boxed_local())
221    }
222}
223
224impl<'a, T> Unpin for WorkflowInterceptorFuture<'a, T> {}
225
226impl<T> Future for WorkflowInterceptorFuture<'_, T> {
227    type Output = T;
228
229    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
230        self.0.as_mut().poll(cx)
231    }
232}
233
234/// Continuation for a workflow interceptor operation.
235pub struct WorkflowNext<'a, I, O> {
236    inner: Box<dyn FnOnce(I) -> O + 'a>,
237}
238
239impl<'a, I, O> WorkflowNext<'a, I, O> {
240    pub(crate) fn new(f: impl FnOnce(I) -> O + 'a) -> Self {
241        Self { inner: Box::new(f) }
242    }
243
244    /// Continue the call chain with the provided input.
245    pub fn run(self, input: I) -> O {
246        (self.inner)(input)
247    }
248}
249
250/// Workflow execution context available to async-capable inbound interceptors.
251#[derive(Clone)]
252pub struct WorkflowInterceptorContext {
253    base: BaseWorkflowContext,
254}
255
256impl WorkflowInterceptorContext {
257    pub(crate) fn new(base: BaseWorkflowContext) -> Self {
258        Self { base }
259    }
260
261    /// Return the workflow's unique identifier.
262    pub fn workflow_id(&self) -> &str {
263        self.base.workflow_id()
264    }
265
266    /// Return the run id of this workflow execution.
267    pub fn run_id(&self) -> &str {
268        self.base.run_id()
269    }
270
271    /// Return the namespace the workflow is executing in.
272    pub fn namespace(&self) -> &str {
273        self.base.namespace()
274    }
275
276    /// Return the task queue the workflow is executing in.
277    pub fn task_queue(&self) -> &str {
278        self.base.task_queue()
279    }
280
281    /// Return the workflow type name.
282    pub fn workflow_type(&self) -> &str {
283        self.base.workflow_type()
284    }
285
286    /// Return the current time according to the workflow.
287    pub fn workflow_time(&self) -> Option<SystemTime> {
288        self.base.workflow_time()
289    }
290
291    /// Return the length of history so far at this point in the workflow.
292    pub fn history_length(&self) -> u32 {
293        self.base.history_length()
294    }
295
296    /// Return current values for workflow search attributes.
297    pub fn search_attributes(&self) -> SearchAttributes {
298        self.base.search_attributes()
299    }
300
301    /// Returns true if the workflow is replaying (including during queries and update validators), false otherwise.
302    pub fn is_replaying(&self) -> bool {
303        self.base.is_replaying()
304    }
305
306    /// Return true if the workflow is replaying history events (excluding queries and update validators), false otherwise.
307    pub fn is_replaying_history_events(&self) -> bool {
308        self.base.is_replaying_history_events()
309    }
310
311    /// Returns the payload converter used by the worker running this workflow.
312    pub fn payload_converter(&self) -> &PayloadConverter {
313        self.base.payload_converter()
314    }
315
316    /// Request to create a timer through the workflow outbound interceptor chain.
317    pub fn timer<T: Into<TimerOptions>>(
318        &self,
319        opts: T,
320    ) -> impl CancellableFuture<TimerResult> + use<T> {
321        self.base.timer(opts)
322    }
323
324    /// Request to run an activity through the workflow outbound interceptor chain.
325    pub fn execute_activity<AD: ActivityDefinition>(
326        &self,
327        activity: AD,
328        input: impl Into<AD::Input>,
329        opts: ActivityOptions,
330    ) -> impl CancellableFuture<Result<AD::Output, ActivityExecutionError>>
331    where
332        AD::Output: TemporalDeserializable,
333    {
334        self.base.execute_activity(activity, input, opts)
335    }
336
337    /// Request to run a local activity through the workflow outbound interceptor chain.
338    pub fn execute_local_activity<AD: ActivityDefinition>(
339        &self,
340        activity: AD,
341        input: impl Into<AD::Input>,
342        opts: LocalActivityOptions,
343    ) -> impl CancellableFuture<Result<AD::Output, ActivityExecutionError>>
344    where
345        AD::Output: TemporalDeserializable,
346    {
347        self.base.execute_local_activity(activity, input, opts)
348    }
349
350    /// Start a child workflow through the workflow outbound interceptor chain.
351    pub fn start_child_workflow<WD: WorkflowDefinition + 'static>(
352        &self,
353        workflow: WD,
354        input: impl Into<WD::Input>,
355        opts: ChildWorkflowOptions,
356    ) -> impl CancellableFutureWithReason<Result<StartedChildWorkflow<WD>, ChildWorkflowStartError>>
357    where
358        WD::Output: TemporalDeserializable,
359    {
360        self.base.start_child_workflow(workflow, input, opts)
361    }
362
363    /// Get a handle to an external workflow for signaling or requesting cancellation.
364    pub fn external_workflow(
365        &self,
366        workflow_id: impl Into<String>,
367        run_id: Option<String>,
368    ) -> ExternalWorkflowHandle {
369        self.base.external_workflow(workflow_id, run_id)
370    }
371
372    /// Start a Nexus operation through the workflow outbound interceptor chain.
373    pub fn start_nexus_operation(
374        &self,
375        opts: NexusOperationOptions,
376    ) -> impl CancellableFuture<NexusStartResult> {
377        self.base.start_nexus_operation(opts)
378    }
379}
380
381/// Workflow execution context available to sync-only inbound interceptors.
382#[derive(Clone)]
383pub struct SyncWorkflowInterceptorContext {
384    base: BaseWorkflowContext,
385}
386
387impl SyncWorkflowInterceptorContext {
388    pub(crate) fn new(base: BaseWorkflowContext) -> Self {
389        Self { base }
390    }
391
392    /// Return the workflow's unique identifier.
393    pub fn workflow_id(&self) -> &str {
394        self.base.workflow_id()
395    }
396
397    /// Return the run id of this workflow execution.
398    pub fn run_id(&self) -> &str {
399        self.base.run_id()
400    }
401
402    /// Return the namespace the workflow is executing in.
403    pub fn namespace(&self) -> &str {
404        self.base.namespace()
405    }
406
407    /// Return the task queue the workflow is executing in.
408    pub fn task_queue(&self) -> &str {
409        self.base.task_queue()
410    }
411
412    /// Return the workflow type name.
413    pub fn workflow_type(&self) -> &str {
414        self.base.workflow_type()
415    }
416
417    /// Return the current time according to the workflow.
418    pub fn workflow_time(&self) -> Option<SystemTime> {
419        self.base.workflow_time()
420    }
421
422    /// Return the length of history so far at this point in the workflow.
423    pub fn history_length(&self) -> u32 {
424        self.base.history_length()
425    }
426
427    /// Return current values for workflow search attributes.
428    pub fn search_attributes(&self) -> SearchAttributes {
429        self.base.search_attributes()
430    }
431
432    /// Returns true if the current workflow task is happening under replay.
433    pub fn is_replaying(&self) -> bool {
434        self.base.is_replaying()
435    }
436
437    /// Returns true if the current work is replaying history events.
438    pub fn is_replaying_history_events(&self) -> bool {
439        self.base.is_replaying_history_events()
440    }
441
442    /// Returns the payload converter used by the worker running this workflow.
443    pub fn payload_converter(&self) -> &PayloadConverter {
444        self.base.payload_converter()
445    }
446}
447
448struct DecodedInput {
449    value: Option<Box<dyn Any>>,
450    headers: HashMap<String, Payload>,
451}
452
453impl DecodedInput {
454    fn new(value: Option<Box<dyn Any>>, headers: HashMap<String, Payload>) -> Self {
455        Self { value, headers }
456    }
457
458    fn input_ref<T: Any>(&self) -> Option<&T> {
459        self.value.as_ref()?.downcast_ref()
460    }
461
462    fn input_mut<T: Any>(&mut self) -> Option<&mut T> {
463        self.value.as_mut()?.downcast_mut()
464    }
465
466    fn headers(&self) -> &HashMap<String, Payload> {
467        &self.headers
468    }
469
470    fn headers_mut(&mut self) -> &mut HashMap<String, Payload> {
471        &mut self.headers
472    }
473
474    fn into_parts(self) -> (Option<Box<dyn Any>>, HashMap<String, Payload>) {
475        (self.value, self.headers)
476    }
477}
478
479/// Input passed to [`WorkflowInterceptor::initialize_workflow`].
480///
481/// The decoded input provided to workflow's `#[init]` method.
482/// If a workflow has no `#[init]`, inputs are instead passed to [`WorkflowInterceptor::execute`].
483#[non_exhaustive]
484pub struct InitializeWorkflowInput {
485    decoded: DecodedInput,
486}
487
488impl InitializeWorkflowInput {
489    pub(crate) fn new(value: Option<Box<dyn Any>>, headers: HashMap<String, Payload>) -> Self {
490        Self {
491            decoded: DecodedInput::new(value, headers),
492        }
493    }
494
495    pub(crate) fn into_parts(self) -> (Option<Box<dyn Any>>, HashMap<String, Payload>) {
496        self.decoded.into_parts()
497    }
498
499    /// Attempt to access the decoded workflow input as a concrete type.
500    pub fn input_ref<T: Any>(&self) -> Option<&T> {
501        self.decoded.input_ref()
502    }
503
504    /// Attempt to mutably access the decoded workflow input as a concrete type.
505    pub fn input_mut<T: Any>(&mut self) -> Option<&mut T> {
506        self.decoded.input_mut()
507    }
508
509    /// Headers attached to the workflow execution.
510    pub fn headers(&self) -> &HashMap<String, Payload> {
511        self.decoded.headers()
512    }
513
514    /// Mutably access headers attached to the workflow execution.
515    pub fn headers_mut(&mut self) -> &mut HashMap<String, Payload> {
516        self.decoded.headers_mut()
517    }
518}
519
520/// Result of workflow initialization.
521pub struct InitializeWorkflowOutput {
522    _private: (),
523}
524
525impl InitializeWorkflowOutput {
526    pub(crate) fn new() -> Self {
527        Self { _private: () }
528    }
529}
530
531/// Input passed to [`WorkflowInterceptor::execute`].
532///
533/// The decoded input provided to workflow's `#[run]` method.
534/// Inputs consumed by `#[init]` are instead passed to [`WorkflowInterceptor::initialize_workflow`].
535#[non_exhaustive]
536pub struct ExecuteWorkflowInput {
537    decoded: DecodedInput,
538}
539
540impl ExecuteWorkflowInput {
541    pub(crate) fn new(value: Option<Box<dyn Any>>, headers: HashMap<String, Payload>) -> Self {
542        Self {
543            decoded: DecodedInput::new(value, headers),
544        }
545    }
546
547    pub(crate) fn into_parts(self) -> (Option<Box<dyn Any>>, HashMap<String, Payload>) {
548        self.decoded.into_parts()
549    }
550
551    /// Attempt to access the decoded workflow input as a concrete type.
552    pub fn input_ref<T: Any>(&self) -> Option<&T> {
553        self.decoded.input_ref()
554    }
555
556    /// Attempt to mutably access the decoded workflow input as a concrete type.
557    pub fn input_mut<T: Any>(&mut self) -> Option<&mut T> {
558        self.decoded.input_mut()
559    }
560
561    /// Headers attached to the workflow execution.
562    pub fn headers(&self) -> &HashMap<String, Payload> {
563        self.decoded.headers()
564    }
565
566    /// Mutably access headers attached to the workflow execution.
567    pub fn headers_mut(&mut self) -> &mut HashMap<String, Payload> {
568        self.decoded.headers_mut()
569    }
570}
571
572macro_rules! handler_input {
573    ($name:ident, $doc:literal, $field:ident, $field_doc:literal $(, $id_field:ident, $id_doc:literal)?) => {
574        #[doc = $doc]
575        #[non_exhaustive]
576        pub struct $name {
577            $($id_field: String,)?
578            $field: String,
579            decoded: DecodedInput,
580        }
581
582        impl $name {
583            pub(crate) fn new(
584                $($id_field: String,)?
585                $field: String,
586                value: Box<dyn Any>,
587                headers: HashMap<String, Payload>,
588            ) -> Self {
589                Self {
590                    $($id_field,)?
591                    $field,
592                    decoded: DecodedInput::new(Some(value), headers),
593                }
594            }
595
596            pub(crate) fn into_parts(self) -> (String, Box<dyn Any>, HashMap<String, Payload>) {
597                let (value, headers) = self.decoded.into_parts();
598                (
599                    self.$field,
600                    value.expect("handler input must exist after typed decode"),
601                    headers,
602                )
603            }
604
605            #[doc = $field_doc]
606            pub fn name(&self) -> &str {
607                &self.$field
608            }
609
610            $(
611                #[doc = $id_doc]
612                pub fn id(&self) -> &str {
613                    &self.$id_field
614                }
615            )?
616
617            /// Attempt to access the decoded input as a concrete type.
618            pub fn input_ref<T: Any>(&self) -> Option<&T> {
619                self.decoded.input_ref()
620            }
621
622            /// Attempt to mutably access the decoded input as a concrete type.
623            pub fn input_mut<T: Any>(&mut self) -> Option<&mut T> {
624                self.decoded.input_mut()
625            }
626
627            /// Headers attached to this handler invocation.
628            pub fn headers(&self) -> &HashMap<String, Payload> {
629                self.decoded.headers()
630            }
631
632            /// Mutably access headers attached to this handler invocation.
633            pub fn headers_mut(&mut self) -> &mut HashMap<String, Payload> {
634                self.decoded.headers_mut()
635            }
636        }
637    };
638}
639
640handler_input!(
641    HandleSignalInput,
642    "Input passed to [`WorkflowInterceptor::handle_signal`].",
643    signal_name,
644    "Return the signal name."
645);
646
647handler_input!(
648    HandleUpdateInput,
649    "Input passed to [`WorkflowInterceptor::handle_update`].",
650    update_name,
651    "Return the update name.",
652    update_id,
653    "Return the update ID."
654);
655
656handler_input!(
657    HandleQueryInput,
658    "Input passed to [`WorkflowInterceptor::handle_query`].",
659    query_name,
660    "Return the query name.",
661    query_id,
662    "Return the query ID."
663);
664
665/// Input passed to [`WorkflowInterceptor::validate_update`].
666#[non_exhaustive]
667pub struct ValidateUpdateInput {
668    update_id: String,
669    update_name: String,
670    decoded: DecodedInput,
671}
672
673impl ValidateUpdateInput {
674    pub(crate) fn new(
675        update_id: String,
676        update_name: String,
677        value: Box<dyn Any>,
678        headers: HashMap<String, Payload>,
679    ) -> Self {
680        Self {
681            update_id,
682            update_name,
683            decoded: DecodedInput::new(Some(value), headers),
684        }
685    }
686
687    pub(crate) fn into_parts(self) -> (String, Box<dyn Any>, HashMap<String, Payload>) {
688        let (value, headers) = self.decoded.into_parts();
689        (
690            self.update_name,
691            value.expect("update validation input must exist after typed decode"),
692            headers,
693        )
694    }
695
696    /// Return the update name.
697    pub fn name(&self) -> &str {
698        &self.update_name
699    }
700
701    /// Return the update ID.
702    pub fn id(&self) -> &str {
703        &self.update_id
704    }
705
706    /// Attempt to access the decoded input as a concrete type.
707    pub fn input_ref<T: Any>(&self) -> Option<&T> {
708        self.decoded.input_ref()
709    }
710
711    /// Attempt to mutably access the decoded input as a concrete type.
712    pub fn input_mut<T: Any>(&mut self) -> Option<&mut T> {
713        self.decoded.input_mut()
714    }
715
716    /// Headers attached to this update invocation.
717    pub fn headers(&self) -> &HashMap<String, Payload> {
718        self.decoded.headers()
719    }
720
721    /// Mutably access headers attached to this update invocation.
722    pub fn headers_mut(&mut self) -> &mut HashMap<String, Payload> {
723        self.decoded.headers_mut()
724    }
725}
726
727/// Type-erased output returned by an intercepted outbound workflow call.
728pub trait WorkflowOutboundValue: Any {
729    /// Access the concrete value through [`Any`].
730    fn as_any(&self) -> &dyn Any;
731
732    /// Convert this value into [`Any`] for a consuming downcast.
733    fn into_any(self: Box<Self>) -> Box<dyn Any>;
734}
735
736impl<T: Any> WorkflowOutboundValue for T {
737    fn as_any(&self) -> &dyn Any {
738        self
739    }
740
741    fn into_any(self: Box<Self>) -> Box<dyn Any> {
742        self
743    }
744}
745
746impl dyn WorkflowOutboundValue {
747    /// Attempt to access the output as a concrete type.
748    pub fn downcast_ref<T: Any>(&self) -> Option<&T> {
749        self.as_any().downcast_ref()
750    }
751
752    /// Attempt to convert the output into a concrete type.
753    pub fn downcast<T: Any>(self: Box<Self>) -> Result<Box<T>, Box<dyn Any>> {
754        self.into_any().downcast()
755    }
756}
757
758/// Future returned by a non-cancellable outbound interceptor operation.
759pub struct WorkflowOutboundFuture<T> {
760    state: WorkflowOutboundFutureState<T>,
761}
762
763enum WorkflowOutboundFutureState<T> {
764    Running(Fuse<LocalBoxFuture<'static, T>>),
765    Prefetched(Option<T>),
766    Terminated,
767}
768
769impl<T> WorkflowOutboundFuture<T> {
770    /// Create an outbound future.
771    pub fn new(future: impl Future<Output = T> + 'static) -> Self {
772        Self {
773            state: WorkflowOutboundFutureState::Running(future.boxed_local().fuse()),
774        }
775    }
776
777    /// Create an immediately ready outbound future.
778    pub fn ready(value: T) -> Self
779    where
780        T: 'static,
781    {
782        Self::new(async move { value })
783    }
784
785    /// Transform the result of this future.
786    pub fn map<U>(self, map: impl FnOnce(T) -> U + 'static) -> WorkflowOutboundFuture<U>
787    where
788        T: 'static,
789        U: 'static,
790    {
791        WorkflowOutboundFuture::new(async move { map(self.await) })
792    }
793
794    pub(crate) fn poll_for_construction(&mut self, cx: &mut Context<'_>) {
795        let WorkflowOutboundFutureState::Running(future) = &mut self.state else {
796            return;
797        };
798        if let Poll::Ready(value) = future.poll_unpin(cx) {
799            self.state = WorkflowOutboundFutureState::Prefetched(Some(value));
800        }
801    }
802}
803
804impl<T> Unpin for WorkflowOutboundFuture<T> {}
805
806impl<T> Future for WorkflowOutboundFuture<T> {
807    type Output = T;
808
809    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
810        match &mut self.state {
811            WorkflowOutboundFutureState::Running(future) => {
812                let result = future.poll_unpin(cx);
813                if result.is_ready() {
814                    self.state = WorkflowOutboundFutureState::Terminated;
815                }
816                result
817            }
818            WorkflowOutboundFutureState::Prefetched(value) => {
819                let value = value
820                    .take()
821                    .expect("outbound future polled after completion");
822                self.state = WorkflowOutboundFutureState::Terminated;
823                Poll::Ready(value)
824            }
825            WorkflowOutboundFutureState::Terminated => {
826                panic!("outbound future polled after completion")
827            }
828        }
829    }
830}
831
832impl<T> FusedFuture for WorkflowOutboundFuture<T> {
833    fn is_terminated(&self) -> bool {
834        matches!(self.state, WorkflowOutboundFutureState::Terminated)
835    }
836}
837
838/// Cancellation callback retained when an interceptor wraps an operation future.
839#[derive(Clone)]
840pub struct WorkflowCancellationHandle {
841    cancel: Rc<dyn Fn(Option<String>)>,
842}
843
844impl WorkflowCancellationHandle {
845    /// Create a cancellation handle.
846    pub fn new(cancel: impl Fn(Option<String>) + 'static) -> Self {
847        Self {
848            cancel: Rc::new(cancel),
849        }
850    }
851
852    pub(crate) fn noop() -> Self {
853        Self::new(|_| {})
854    }
855
856    /// Cancel with an optional reason.
857    pub fn cancel(&self, reason: Option<String>) {
858        (self.cancel)(reason);
859    }
860}
861
862/// Future returned by a cancellable outbound interceptor operation.
863pub struct CancellableWorkflowOutboundFuture<T> {
864    inner: WorkflowOutboundFuture<T>,
865    cancellation: WorkflowCancellationHandle,
866}
867
868impl<T> CancellableWorkflowOutboundFuture<T> {
869    /// Create a cancellable outbound future.
870    pub fn new(
871        future: impl Future<Output = T> + 'static,
872        cancellation: WorkflowCancellationHandle,
873    ) -> Self {
874        Self {
875            inner: WorkflowOutboundFuture::new(future),
876            cancellation,
877        }
878    }
879
880    /// Return the operation's cancellation handle.
881    pub fn cancellation_handle(&self) -> WorkflowCancellationHandle {
882        self.cancellation.clone()
883    }
884
885    /// Transform the result while retaining cancellation behavior.
886    pub fn map<U>(self, map: impl FnOnce(T) -> U + 'static) -> CancellableWorkflowOutboundFuture<U>
887    where
888        T: 'static,
889        U: 'static,
890    {
891        let cancellation = self.cancellation.clone();
892        CancellableWorkflowOutboundFuture::new(async move { map(self.await) }, cancellation)
893    }
894
895    pub(crate) fn poll_for_construction(&mut self, cx: &mut Context<'_>) {
896        self.inner.poll_for_construction(cx);
897    }
898}
899
900impl<T> Unpin for CancellableWorkflowOutboundFuture<T> {}
901
902impl<T> Future for CancellableWorkflowOutboundFuture<T> {
903    type Output = T;
904
905    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
906        Pin::new(&mut self.inner).poll(cx)
907    }
908}
909
910impl<T> FusedFuture for CancellableWorkflowOutboundFuture<T> {
911    fn is_terminated(&self) -> bool {
912        self.inner.is_terminated()
913    }
914}
915
916impl<T> CancellableFuture<T> for CancellableWorkflowOutboundFuture<T> {
917    fn cancel(&self) {
918        if !self.inner.is_terminated() {
919            self.cancellation.cancel(None);
920        }
921    }
922}
923
924impl<T> CancellableFutureWithReason<T> for CancellableWorkflowOutboundFuture<T> {
925    fn cancel_with_reason(&self, reason: String) {
926        if !self.inner.is_terminated() {
927            self.cancellation.cancel(Some(reason));
928        }
929    }
930}
931
932macro_rules! typed_outbound_input {
933    ($name:ident) => {
934        impl $name {
935            /// Attempt to access the decoded input as a concrete type.
936            pub fn input_ref<T: Any>(&self) -> Option<&T> {
937                self.decoded.input_ref()
938            }
939
940            /// Attempt to mutably access the decoded input as a concrete type.
941            pub fn input_mut<T: Any>(&mut self) -> Option<&mut T> {
942                self.decoded.input_mut()
943            }
944
945            /// Headers attached to this outbound call.
946            pub fn headers(&self) -> &HashMap<String, Payload> {
947                self.decoded.headers()
948            }
949
950            /// Mutably access headers attached to this outbound call.
951            pub fn headers_mut(&mut self) -> &mut HashMap<String, Payload> {
952                self.decoded.headers_mut()
953            }
954        }
955    };
956}
957
958/// Input passed to [`WorkflowInterceptor::start_timer`].
959#[non_exhaustive]
960pub struct StartTimerInput {
961    options: TimerOptions,
962}
963
964impl StartTimerInput {
965    pub(crate) fn new(options: TimerOptions) -> Self {
966        Self { options }
967    }
968
969    pub(crate) fn into_options(self) -> TimerOptions {
970        self.options
971    }
972
973    /// Timer options.
974    pub fn options(&self) -> &TimerOptions {
975        &self.options
976    }
977
978    /// Mutably access timer options.
979    pub fn options_mut(&mut self) -> &mut TimerOptions {
980        &mut self.options
981    }
982}
983
984/// Input passed to [`WorkflowInterceptor::schedule_activity`].
985#[non_exhaustive]
986pub struct ScheduleActivityInput {
987    activity_type: String,
988    decoded: DecodedInput,
989    options: ActivityOptions,
990}
991
992impl ScheduleActivityInput {
993    pub(crate) fn new(
994        activity_type: String,
995        input: Box<dyn Any>,
996        options: ActivityOptions,
997    ) -> Self {
998        Self {
999            activity_type,
1000            decoded: DecodedInput::new(Some(input), HashMap::new()),
1001            options,
1002        }
1003    }
1004
1005    pub(crate) fn into_parts(
1006        self,
1007    ) -> (
1008        String,
1009        Box<dyn Any>,
1010        HashMap<String, Payload>,
1011        ActivityOptions,
1012    ) {
1013        let (input, headers) = self.decoded.into_parts();
1014        (
1015            self.activity_type,
1016            input.expect("activity input must exist"),
1017            headers,
1018            self.options,
1019        )
1020    }
1021
1022    /// Activity type.
1023    pub fn activity_type(&self) -> &str {
1024        &self.activity_type
1025    }
1026
1027    /// Mutably access the activity type.
1028    pub fn activity_type_mut(&mut self) -> &mut String {
1029        &mut self.activity_type
1030    }
1031
1032    /// Activity options.
1033    pub fn options(&self) -> &ActivityOptions {
1034        &self.options
1035    }
1036
1037    /// Mutably access activity options.
1038    pub fn options_mut(&mut self) -> &mut ActivityOptions {
1039        &mut self.options
1040    }
1041}
1042
1043typed_outbound_input!(ScheduleActivityInput);
1044
1045/// Input passed to [`WorkflowInterceptor::schedule_local_activity`].
1046#[non_exhaustive]
1047pub struct ScheduleLocalActivityInput {
1048    activity_type: String,
1049    decoded: DecodedInput,
1050    options: LocalActivityOptions,
1051}
1052
1053impl ScheduleLocalActivityInput {
1054    pub(crate) fn new(
1055        activity_type: String,
1056        input: Box<dyn Any>,
1057        options: LocalActivityOptions,
1058    ) -> Self {
1059        Self {
1060            activity_type,
1061            decoded: DecodedInput::new(Some(input), HashMap::new()),
1062            options,
1063        }
1064    }
1065
1066    pub(crate) fn into_parts(
1067        self,
1068    ) -> (
1069        String,
1070        Box<dyn Any>,
1071        HashMap<String, Payload>,
1072        LocalActivityOptions,
1073    ) {
1074        let (input, headers) = self.decoded.into_parts();
1075        (
1076            self.activity_type,
1077            input.expect("local activity input must exist"),
1078            headers,
1079            self.options,
1080        )
1081    }
1082
1083    /// Activity type.
1084    pub fn activity_type(&self) -> &str {
1085        &self.activity_type
1086    }
1087
1088    /// Mutably access the activity type.
1089    pub fn activity_type_mut(&mut self) -> &mut String {
1090        &mut self.activity_type
1091    }
1092
1093    /// Local activity options.
1094    pub fn options(&self) -> &LocalActivityOptions {
1095        &self.options
1096    }
1097
1098    /// Mutably access local activity options.
1099    pub fn options_mut(&mut self) -> &mut LocalActivityOptions {
1100        &mut self.options
1101    }
1102}
1103
1104typed_outbound_input!(ScheduleLocalActivityInput);
1105
1106/// Input passed to [`WorkflowInterceptor::start_child_workflow`].
1107#[non_exhaustive]
1108pub struct StartChildWorkflowInput {
1109    workflow_type: String,
1110    decoded: DecodedInput,
1111    options: ChildWorkflowOptions,
1112}
1113
1114impl StartChildWorkflowInput {
1115    pub(crate) fn new(
1116        workflow_type: String,
1117        input: Box<dyn Any>,
1118        options: ChildWorkflowOptions,
1119    ) -> Self {
1120        Self {
1121            workflow_type,
1122            decoded: DecodedInput::new(Some(input), HashMap::new()),
1123            options,
1124        }
1125    }
1126
1127    pub(crate) fn into_parts(
1128        self,
1129    ) -> (
1130        String,
1131        Box<dyn Any>,
1132        HashMap<String, Payload>,
1133        ChildWorkflowOptions,
1134    ) {
1135        let (input, headers) = self.decoded.into_parts();
1136        (
1137            self.workflow_type,
1138            input.expect("child workflow input must exist"),
1139            headers,
1140            self.options,
1141        )
1142    }
1143
1144    /// Workflow type.
1145    pub fn workflow_type(&self) -> &str {
1146        &self.workflow_type
1147    }
1148
1149    /// Mutably access the workflow type.
1150    pub fn workflow_type_mut(&mut self) -> &mut String {
1151        &mut self.workflow_type
1152    }
1153
1154    /// Child workflow options.
1155    pub fn options(&self) -> &ChildWorkflowOptions {
1156        &self.options
1157    }
1158
1159    /// Mutably access child workflow options.
1160    pub fn options_mut(&mut self) -> &mut ChildWorkflowOptions {
1161        &mut self.options
1162    }
1163}
1164
1165typed_outbound_input!(StartChildWorkflowInput);
1166
1167/// Workflow targeted by an outbound signal.
1168#[derive(Clone, Debug, PartialEq, Eq)]
1169#[non_exhaustive]
1170pub enum SignalWorkflowTarget {
1171    /// A child workflow identified by workflow ID.
1172    Child {
1173        /// Child workflow ID.
1174        workflow_id: String,
1175    },
1176    /// An external workflow execution.
1177    External {
1178        /// Target namespace.
1179        namespace: String,
1180        /// Target workflow ID.
1181        workflow_id: String,
1182        /// Target run ID, or the latest run when absent.
1183        run_id: Option<String>,
1184    },
1185}
1186
1187/// Input passed to [`WorkflowInterceptor::signal_workflow`].
1188#[non_exhaustive]
1189pub struct SignalWorkflowInput {
1190    signal_name: String,
1191    target: SignalWorkflowTarget,
1192    decoded: DecodedInput,
1193}
1194
1195impl SignalWorkflowInput {
1196    pub(crate) fn new(
1197        signal_name: String,
1198        target: SignalWorkflowTarget,
1199        input: Box<dyn Any>,
1200    ) -> Self {
1201        Self {
1202            signal_name,
1203            target,
1204            decoded: DecodedInput::new(Some(input), HashMap::new()),
1205        }
1206    }
1207
1208    pub(crate) fn into_parts(
1209        self,
1210    ) -> (
1211        String,
1212        SignalWorkflowTarget,
1213        Box<dyn Any>,
1214        HashMap<String, Payload>,
1215    ) {
1216        let (input, headers) = self.decoded.into_parts();
1217        (
1218            self.signal_name,
1219            self.target,
1220            input.expect("signal input must exist"),
1221            headers,
1222        )
1223    }
1224
1225    /// Signal name.
1226    pub fn signal_name(&self) -> &str {
1227        &self.signal_name
1228    }
1229
1230    /// Mutably access the signal name.
1231    pub fn signal_name_mut(&mut self) -> &mut String {
1232        &mut self.signal_name
1233    }
1234
1235    /// Signal target.
1236    pub fn target(&self) -> &SignalWorkflowTarget {
1237        &self.target
1238    }
1239
1240    /// Mutably access the signal target.
1241    pub fn target_mut(&mut self) -> &mut SignalWorkflowTarget {
1242        &mut self.target
1243    }
1244}
1245
1246typed_outbound_input!(SignalWorkflowInput);
1247
1248/// Input passed to [`WorkflowInterceptor::cancel_external_workflow`].
1249#[derive(Clone, Debug)]
1250#[non_exhaustive]
1251pub struct CancelExternalWorkflowInput {
1252    /// Target workflow ID.
1253    pub workflow_id: String,
1254    /// Target run ID, or the latest run when absent.
1255    pub run_id: Option<String>,
1256    /// Cancellation reason.
1257    pub reason: Option<String>,
1258}
1259
1260/// Input passed to [`WorkflowInterceptor::continue_as_new`].
1261#[non_exhaustive]
1262pub struct ContinueAsNewInput {
1263    decoded: DecodedInput,
1264    options: ContinueAsNewOptions,
1265}
1266
1267impl ContinueAsNewInput {
1268    pub(crate) fn new(input: Box<dyn Any>, options: ContinueAsNewOptions) -> Self {
1269        Self {
1270            decoded: DecodedInput::new(Some(input), HashMap::new()),
1271            options,
1272        }
1273    }
1274
1275    pub(crate) fn into_parts(
1276        self,
1277    ) -> (Box<dyn Any>, HashMap<String, Payload>, ContinueAsNewOptions) {
1278        let (input, headers) = self.decoded.into_parts();
1279        (
1280            input.expect("continue-as-new input must exist"),
1281            headers,
1282            self.options,
1283        )
1284    }
1285
1286    /// Continue-as-new options.
1287    pub fn options(&self) -> &ContinueAsNewOptions {
1288        &self.options
1289    }
1290
1291    /// Mutably access continue-as-new options.
1292    pub fn options_mut(&mut self) -> &mut ContinueAsNewOptions {
1293        &mut self.options
1294    }
1295}
1296
1297typed_outbound_input!(ContinueAsNewInput);
1298
1299/// Input passed to [`WorkflowInterceptor::start_nexus_operation`].
1300#[non_exhaustive]
1301pub struct StartNexusOperationInput {
1302    options: NexusOperationOptions,
1303}
1304
1305impl StartNexusOperationInput {
1306    pub(crate) fn new(options: NexusOperationOptions) -> Self {
1307        Self { options }
1308    }
1309
1310    pub(crate) fn into_options(self) -> NexusOperationOptions {
1311        self.options
1312    }
1313
1314    /// Nexus operation options.
1315    pub fn options(&self) -> &NexusOperationOptions {
1316        &self.options
1317    }
1318
1319    /// Mutably access Nexus operation options.
1320    pub fn options_mut(&mut self) -> &mut NexusOperationOptions {
1321        &mut self.options
1322    }
1323}
1324
1325/// Result of an intercepted activity call.
1326pub type ScheduleActivityResult = Result<Box<dyn WorkflowOutboundValue>, ActivityExecutionError>;
1327
1328/// Result of an intercepted child workflow completion.
1329pub type ChildWorkflowOutboundResult =
1330    Result<Box<dyn WorkflowOutboundValue>, ChildWorkflowExecutionError>;
1331
1332/// Result of an intercepted signal call.
1333pub type SignalWorkflowResult = Result<(), WorkflowSignalError>;
1334
1335/// Result of an intercepted child workflow start.
1336pub type StartChildWorkflowResult = Result<StartChildWorkflowOutput, ChildWorkflowStartError>;
1337
1338/// Result of an intercepted Nexus operation start.
1339pub type StartNexusOperationResult = Result<StartedNexusOperation, Failure>;
1340
1341/// Result of an intercepted continue-as-new call.
1342pub type ContinueAsNewResult = Result<Infallible, WorkflowTermination>;
1343
1344/// Interceptor for calls into workflow code and commands issued by workflow code.
1345///
1346/// Implement this trait for behavior that should wrap workflow operations. Inbound
1347/// methods intercept operations such as workflow execution and handler dispatch;
1348/// outbound methods intercept timers, activities, child workflows, external workflow calls,
1349/// continue-as-new, and Nexus operations.
1350///
1351/// Implementations normally calls [`WorkflowNext::run`] exactly once. It may transform the input first,
1352/// then inspect or transform the result. Not calling `next` short-circuits the operation, so it
1353/// should only be done if intentionally skipping the operation.
1354///
1355/// The async inbound methods return [`WorkflowInterceptorFuture`]. Use
1356/// [`WorkflowInterceptorFuture::new`] to wrap an `async` block around the downstream future:
1357/// call `next.run(input).await` to call the next interceptor. See the [module-level guide](self)
1358/// for a complete example and the determinism requirements.
1359///
1360/// Interceptors run as workflow code and are recreated when an evicted workflow is rebuilt. Their
1361/// behavior and any instance-local state must remain deterministic under replay. Async methods may
1362/// await only workflow scheduler primitives or SDK-provided workflow futures.
1363///
1364/// A [`WorkflowInterceptorConstructor`] creates one interceptor for each in-memory workflow
1365/// instance. The same interceptor object handles inbound and outbound calls for that instance and
1366/// is recreated if the workflow is evicted and rebuilt.
1367/// Inbound interceptors are called in regsitration order and outbound interceptors are called in reverse order.
1368pub trait WorkflowInterceptor: 'static {
1369    /// Called to invoke the workflow's `#[init]` method.
1370    ///
1371    /// It is only called for workflows that define `#[init]`, before the workflow instance exists.
1372    fn initialize_workflow(
1373        &self,
1374        _ctx: WorkflowContextView,
1375        input: InitializeWorkflowInput,
1376        next: WorkflowNext<'_, InitializeWorkflowInput, InitializeWorkflowOutput>,
1377    ) -> InitializeWorkflowOutput {
1378        next.run(input)
1379    }
1380
1381    /// Called to invoke the workflow run method.
1382    ///
1383    /// Inputs consumed by `#[init]` are instead passed to
1384    /// [`WorkflowInterceptor::initialize_workflow`].
1385    fn execute<'a>(
1386        &'a self,
1387        _ctx: WorkflowInterceptorContext,
1388        input: ExecuteWorkflowInput,
1389        next: WorkflowNext<
1390            'a,
1391            ExecuteWorkflowInput,
1392            WorkflowInterceptorFuture<'a, ExecuteWorkflowResult>,
1393        >,
1394    ) -> WorkflowInterceptorFuture<'a, ExecuteWorkflowResult> {
1395        next.run(input)
1396    }
1397
1398    /// Called to invoke a signal handler.
1399    fn handle_signal<'a>(
1400        &'a self,
1401        _ctx: WorkflowInterceptorContext,
1402        input: HandleSignalInput,
1403        next: WorkflowNext<
1404            'a,
1405            HandleSignalInput,
1406            WorkflowInterceptorFuture<'a, HandleSignalResult>,
1407        >,
1408    ) -> WorkflowInterceptorFuture<'a, HandleSignalResult> {
1409        next.run(input)
1410    }
1411
1412    /// Called to invoke an update handler.
1413    fn handle_update<'a>(
1414        &'a self,
1415        _ctx: WorkflowInterceptorContext,
1416        input: HandleUpdateInput,
1417        next: WorkflowNext<
1418            'a,
1419            HandleUpdateInput,
1420            WorkflowInterceptorFuture<'a, HandleUpdateResult>,
1421        >,
1422    ) -> WorkflowInterceptorFuture<'a, HandleUpdateResult> {
1423        next.run(input)
1424    }
1425
1426    /// Called to invoke a query handler.
1427    fn handle_query(
1428        &self,
1429        _ctx: SyncWorkflowInterceptorContext,
1430        input: HandleQueryInput,
1431        next: WorkflowNext<'_, HandleQueryInput, HandleQueryResult>,
1432    ) -> HandleQueryResult {
1433        next.run(input)
1434    }
1435
1436    /// Called to validate an update.
1437    fn validate_update(
1438        &self,
1439        _ctx: SyncWorkflowInterceptorContext,
1440        input: ValidateUpdateInput,
1441        next: WorkflowNext<'_, ValidateUpdateInput, ValidateUpdateResult>,
1442    ) -> ValidateUpdateResult {
1443        next.run(input)
1444    }
1445
1446    /// Called when the workflow starts a timer.
1447    fn start_timer(
1448        &self,
1449        _ctx: WorkflowInterceptorContext,
1450        input: StartTimerInput,
1451        next: WorkflowNext<
1452            'static,
1453            StartTimerInput,
1454            CancellableWorkflowOutboundFuture<TimerResult>,
1455        >,
1456    ) -> CancellableWorkflowOutboundFuture<TimerResult> {
1457        next.run(input)
1458    }
1459
1460    /// Called when the workflow schedules an activity.
1461    fn schedule_activity(
1462        &self,
1463        _ctx: WorkflowInterceptorContext,
1464        input: ScheduleActivityInput,
1465        next: WorkflowNext<
1466            'static,
1467            ScheduleActivityInput,
1468            CancellableWorkflowOutboundFuture<ScheduleActivityResult>,
1469        >,
1470    ) -> CancellableWorkflowOutboundFuture<ScheduleActivityResult> {
1471        next.run(input)
1472    }
1473
1474    /// Called when the workflow schedules a local activity.
1475    fn schedule_local_activity(
1476        &self,
1477        _ctx: WorkflowInterceptorContext,
1478        input: ScheduleLocalActivityInput,
1479        next: WorkflowNext<
1480            'static,
1481            ScheduleLocalActivityInput,
1482            CancellableWorkflowOutboundFuture<ScheduleActivityResult>,
1483        >,
1484    ) -> CancellableWorkflowOutboundFuture<ScheduleActivityResult> {
1485        next.run(input)
1486    }
1487
1488    /// Called when the workflow starts a child workflow.
1489    fn start_child_workflow(
1490        &self,
1491        _ctx: WorkflowInterceptorContext,
1492        input: StartChildWorkflowInput,
1493        next: WorkflowNext<
1494            'static,
1495            StartChildWorkflowInput,
1496            CancellableWorkflowOutboundFuture<StartChildWorkflowResult>,
1497        >,
1498    ) -> CancellableWorkflowOutboundFuture<StartChildWorkflowResult> {
1499        next.run(input)
1500    }
1501
1502    /// Called when the workflow signals a child or external workflow.
1503    fn signal_workflow(
1504        &self,
1505        _ctx: WorkflowInterceptorContext,
1506        input: SignalWorkflowInput,
1507        next: WorkflowNext<
1508            'static,
1509            SignalWorkflowInput,
1510            CancellableWorkflowOutboundFuture<SignalWorkflowResult>,
1511        >,
1512    ) -> CancellableWorkflowOutboundFuture<SignalWorkflowResult> {
1513        next.run(input)
1514    }
1515
1516    /// Called when the workflow requests cancellation of an external workflow.
1517    fn cancel_external_workflow(
1518        &self,
1519        _ctx: WorkflowInterceptorContext,
1520        input: CancelExternalWorkflowInput,
1521        next: WorkflowNext<
1522            'static,
1523            CancelExternalWorkflowInput,
1524            WorkflowOutboundFuture<CancelExternalWfResult>,
1525        >,
1526    ) -> WorkflowOutboundFuture<CancelExternalWfResult> {
1527        next.run(input)
1528    }
1529
1530    /// Called when the workflow continues as new.
1531    fn continue_as_new(
1532        &self,
1533        _ctx: SyncWorkflowInterceptorContext,
1534        input: ContinueAsNewInput,
1535        next: WorkflowNext<'static, ContinueAsNewInput, ContinueAsNewResult>,
1536    ) -> ContinueAsNewResult {
1537        next.run(input)
1538    }
1539
1540    /// Called when the workflow starts a Nexus operation.
1541    fn start_nexus_operation(
1542        &self,
1543        _ctx: WorkflowInterceptorContext,
1544        input: StartNexusOperationInput,
1545        next: WorkflowNext<
1546            'static,
1547            StartNexusOperationInput,
1548            CancellableWorkflowOutboundFuture<StartNexusOperationResult>,
1549        >,
1550    ) -> CancellableWorkflowOutboundFuture<StartNexusOperationResult> {
1551        next.run(input)
1552    }
1553}
1554
1555macro_rules! outbound_chain {
1556    ($fn_name:ident, $method:ident, $context:ty, $input:ty, $output:ty) => {
1557        pub(crate) fn $fn_name(
1558            interceptors: Rc<[Arc<dyn WorkflowInterceptor>]>,
1559            ctx: $context,
1560            input: $input,
1561            next: WorkflowNext<'static, $input, $output>,
1562        ) -> $output {
1563            fn call(
1564                interceptors: Rc<[Arc<dyn WorkflowInterceptor>]>,
1565                interceptor_count: usize,
1566                ctx: $context,
1567                input: $input,
1568                next: WorkflowNext<'static, $input, $output>,
1569            ) -> $output {
1570                if let Some(interceptor_index) = interceptor_count.checked_sub(1) {
1571                    let interceptor = interceptors[interceptor_index].clone();
1572                    let next_ctx = ctx.clone();
1573                    let downstream = WorkflowNext::new(move |input| {
1574                        call(interceptors, interceptor_index, next_ctx, input, next)
1575                    });
1576                    interceptor.$method(ctx, input, downstream)
1577                } else {
1578                    next.run(input)
1579                }
1580            }
1581
1582            let interceptor_count = interceptors.len();
1583            call(interceptors, interceptor_count, ctx, input, next)
1584        }
1585    };
1586}
1587
1588outbound_chain!(
1589    call_start_timer,
1590    start_timer,
1591    WorkflowInterceptorContext,
1592    StartTimerInput,
1593    CancellableWorkflowOutboundFuture<TimerResult>
1594);
1595outbound_chain!(
1596    call_schedule_activity,
1597    schedule_activity,
1598    WorkflowInterceptorContext,
1599    ScheduleActivityInput,
1600    CancellableWorkflowOutboundFuture<ScheduleActivityResult>
1601);
1602outbound_chain!(
1603    call_schedule_local_activity,
1604    schedule_local_activity,
1605    WorkflowInterceptorContext,
1606    ScheduleLocalActivityInput,
1607    CancellableWorkflowOutboundFuture<ScheduleActivityResult>
1608);
1609outbound_chain!(
1610    call_start_child_workflow,
1611    start_child_workflow,
1612    WorkflowInterceptorContext,
1613    StartChildWorkflowInput,
1614    CancellableWorkflowOutboundFuture<StartChildWorkflowResult>
1615);
1616outbound_chain!(
1617    call_signal_workflow,
1618    signal_workflow,
1619    WorkflowInterceptorContext,
1620    SignalWorkflowInput,
1621    CancellableWorkflowOutboundFuture<SignalWorkflowResult>
1622);
1623outbound_chain!(
1624    call_cancel_external_workflow,
1625    cancel_external_workflow,
1626    WorkflowInterceptorContext,
1627    CancelExternalWorkflowInput,
1628    WorkflowOutboundFuture<CancelExternalWfResult>
1629);
1630outbound_chain!(
1631    call_continue_as_new,
1632    continue_as_new,
1633    SyncWorkflowInterceptorContext,
1634    ContinueAsNewInput,
1635    ContinueAsNewResult
1636);
1637outbound_chain!(
1638    call_start_nexus_operation,
1639    start_nexus_operation,
1640    WorkflowInterceptorContext,
1641    StartNexusOperationInput,
1642    CancellableWorkflowOutboundFuture<StartNexusOperationResult>
1643);
1644
1645type WorkflowInterceptorConstructorFn =
1646    dyn Fn(&WorkflowContextView) -> Arc<dyn WorkflowInterceptor> + Send + Sync + 'static;
1647
1648/// Creates one interceptor for each in-memory workflow instance.
1649///
1650/// The constructor receives a read-only view of the workflow's initialization context. It may use
1651/// that context to initialize interceptor state but must remain deterministic because it runs again
1652/// when an evicted workflow is rebuilt.
1653#[derive(Clone)]
1654pub struct WorkflowInterceptorConstructor {
1655    constructor: Arc<WorkflowInterceptorConstructorFn>,
1656}
1657
1658impl WorkflowInterceptorConstructor {
1659    /// Create a workflow interceptor constructor.
1660    pub fn new<F, I>(constructor: F) -> Self
1661    where
1662        F: Fn(&WorkflowContextView) -> I + Send + Sync + 'static,
1663        I: WorkflowInterceptor,
1664    {
1665        Self {
1666            constructor: Arc::new(move |ctx| Arc::new(constructor(ctx))),
1667        }
1668    }
1669
1670    pub(crate) fn construct(&self, ctx: &WorkflowContextView) -> Arc<dyn WorkflowInterceptor> {
1671        (self.constructor)(ctx)
1672    }
1673}
1674
1675pub(crate) fn wrong_workflow_input_type(type_name: &'static str) -> WorkflowTermination {
1676    WorkflowTermination::failed_application(temporalio_common_wasm::error::ApplicationFailure::new(
1677        anyhow::anyhow!(
1678            "Workflow inbound interceptor returned arguments with wrong concrete type for workflow {type_name}"
1679        ),
1680    ))
1681}
temporalio_workflow/workflow_interceptors.rs

temporalio_workflow/
workflow_interceptors.rs
