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

temporalio_workflow/
workflow_interceptors.rs
