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

temporalio_workflow/
workflow_interceptors.rs
