temporalio-workflow 0.6.0

mod options;
mod view;

pub use options::{
    ActivityCancellationType, ActivityCloseTimeouts, ActivityOptions,
    ChildWorkflowCancellationType, ChildWorkflowOptions, ContinueAsNewOptions,
    ContinueAsNewVersioningBehavior, LocalActivityOptions, NexusOperationCancellationType,
    NexusOperationOptions, ParentClosePolicy, Signal, SignalData, TimerOptions, VersioningIntent,
    WorkflowIdReusePolicy,
};
pub use temporalio_common_wasm::protos::coresdk::child_workflow::StartChildWorkflowExecutionFailedCause;
pub use view::{NamespacedWorkflowInfo, WorkflowContextView};

use crate::{
    MemoValue,
    runtime::{
        SdkGuardedFuture, SdkWakeGuard,
        entry::WorkflowImplementation,
        host::WorkflowHost,
        mark_intercepted_future_activation,
        model::{
            CancelExternalWfResult, CancellableID, NexusStartResult, SignalExternalWfResult,
            TimerResult, UnblockEvent, Unblockable, WorkflowTermination,
        },
        types::WorkflowInit,
    },
    workflow_interceptors::{
        CancelExternalWorkflowInput, CancellableWorkflowOutboundFuture,
        ChildWorkflowOutboundResult, ContinueAsNewInput, ScheduleActivityInput,
        ScheduleLocalActivityInput, SignalWorkflowInput, SignalWorkflowResult,
        SignalWorkflowTarget, StartChildWorkflowInput, StartChildWorkflowResult,
        StartNexusOperationInput, StartTimerInput, WorkflowCancellationHandle, WorkflowInterceptor,
        WorkflowInterceptorConstructor, WorkflowInterceptorContext, WorkflowNext,
        WorkflowOutboundFuture, WorkflowOutboundValue, call_cancel_external_workflow,
        call_continue_as_new, call_schedule_activity, call_schedule_local_activity,
        call_signal_workflow, call_start_child_workflow, call_start_nexus_operation,
        call_start_timer,
    },
};
use futures_channel::oneshot;
use futures_util::{
    FutureExt,
    future::{FusedFuture, Shared},
    task::Context,
};
use rand::SeedableRng;
use rand_pcg::Pcg64Mcg;
use std::{
    cell::{Cell, RefCell},
    collections::{HashMap, HashSet},
    future::{self, Future},
    marker::PhantomData,
    pin::Pin,
    rc::Rc,
    sync::{
        Arc,
        atomic::{AtomicBool, Ordering},
    },
    task::{Poll, Waker},
    time::{Duration, SystemTime},
};
use temporalio_common_wasm::{
    ActivityDefinition, Memo, SignalDefinition, WorkflowDefinition,
    data_converters::{
        ActivityExecutionDecodeHint, ChildWorkflowExecutionDecodeHint,
        ChildWorkflowStartDecodeHint, DataConverter, GenericPayloadConverter,
        PayloadConversionError, PayloadConverter, SerializationContext, SerializationContextData,
        TemporalDeserializable, WorkflowSignalDecodeHint,
    },
    error::{
        ActivityExecutionError, ChildWorkflowExecutionError, ChildWorkflowStartError,
        WorkflowSignalError,
    },
    protos::{
        coresdk::{
            activity_result::{ActivityResolution, Cancellation, activity_resolution},
            child_workflow::{ChildWorkflowResult, child_workflow_result},
            common::NamespacedWorkflowExecution,
            nexus::NexusOperationResult,
            workflow_activation::{
                InitializeWorkflow, WorkflowActivation as CoreWorkflowActivation,
                resolve_child_workflow_execution_start::Status as ChildWorkflowStartStatus,
                workflow_activation_job::Variant as ActivationVariant,
            },
            workflow_commands::{
                CancelChildWorkflowExecution, CancelSignalWorkflow, CancelTimer,
                ModifyWorkflowProperties, RequestCancelActivity,
                RequestCancelExternalWorkflowExecution, RequestCancelLocalActivity,
                RequestCancelNexusOperation, SetPatchMarker, SignalExternalWorkflowExecution,
                UpsertWorkflowSearchAttributes, signal_external_workflow_execution,
                workflow_command,
            },
        },
        temporal::api::{
            common::v1::{Memo as ProtoMemo, Payload, SearchAttributes as ProtoSearchAttributes},
            failure::v1::{CanceledFailureInfo, Failure, failure::FailureInfo},
        },
        utilities::TryIntoOrNone,
    },
    search_attributes::{SearchAttributeUpdate, SearchAttributes},
    worker::WorkerDeploymentVersion,
};
use uuid::Builder;

mod private {
    use rand::distr::{Distribution, StandardUniform};
    use rand_pcg::Pcg64Mcg;

    pub trait Sealed: Sized {
        fn sample(rng: &mut Pcg64Mcg) -> Self;
    }

    pub(super) fn sample<T>(rng: &mut Pcg64Mcg) -> T
    where
        StandardUniform: Distribution<T>,
    {
        StandardUniform.sample(rng)
    }
}

/// A numeric value that can be generated by [`WorkflowContext::random`].
///
/// This trait is sealed and implemented for `u8`, `u16`, `u32`, `u64`, `u128`, `i8`, `i16`,
/// `i32`, `i64`, `i128`, `f32`, and `f64`.
pub trait WorkflowRandomValue: private::Sealed + Sized {}

macro_rules! impl_random_value {
    ($($ty:ty),* $(,)?) => {
        $(
            impl private::Sealed for $ty {
                fn sample(rng: &mut Pcg64Mcg) -> Self {
                    private::sample(rng)
                }
            }

            impl WorkflowRandomValue for $ty {}
        )*
    };
}

impl_random_value!(u8, u16, u32, u64, u128, i8, i16, i32, i64, i128, f32, f64);

/// Non-generic base context containing all workflow execution infrastructure.
///
/// This is used internally by futures and commands that don't need typed workflow state.
#[derive(Clone)]
pub struct BaseWorkflowContext {
    inner: Rc<WorkflowContextInner>,
}

/// Input provided to a worker's patch activation callback.
#[derive(Clone, Debug)]
#[non_exhaustive]
pub struct PatchActivationInput {
    /// Information about the workflow execution calling [`SyncWorkflowContext::patched`].
    pub workflow_info: WorkflowContextView,
    /// Patch ID passed to [`SyncWorkflowContext::patched`].
    pub patch_id: String,
}

/// Callback that decides whether a newly encountered patch should be activated.
pub type PatchActivationCallback =
    Arc<dyn Fn(PatchActivationInput) -> bool + Send + Sync + 'static>;

/// Invokes a patch activation callback with a workflow information snapshot.
#[doc(hidden)]
pub struct PatchActivationCaller {
    callback: PatchActivationCallback,
    workflow_info: WorkflowContextView,
}

impl PatchActivationCaller {
    /// Creates a caller from workflow initialization data.
    pub fn new(
        callback: PatchActivationCallback,
        namespace: String,
        task_queue: String,
        run_id: String,
        init: InitializeWorkflow,
        payload_converter: PayloadConverter,
    ) -> Self {
        Self {
            callback,
            workflow_info: WorkflowContextView::new(
                namespace,
                task_queue,
                run_id,
                init,
                payload_converter,
            ),
        }
    }

    /// Invokes the callback for a patch ID.
    pub fn call(&self, patch_id: String) -> bool {
        (self.callback)(PatchActivationInput {
            workflow_info: self.workflow_info.clone(),
            patch_id,
        })
    }
}

pub(crate) struct WorkflowPollWakerGuard<'a> {
    current_waker: &'a RefCell<Option<Waker>>,
    previous: Option<Waker>,
}

impl Drop for WorkflowPollWakerGuard<'_> {
    fn drop(&mut self) {
        self.current_waker.replace(self.previous.take());
    }
}

fn outbound_type_error(
    value: &str,
) -> temporalio_common_wasm::data_converters::PayloadConversionError {
    temporalio_common_wasm::data_converters::PayloadConversionError::EncodingError(Box::new(
        std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            format!("workflow interceptor returned the wrong concrete {value} type"),
        ),
    ))
}

impl BaseWorkflowContext {
    pub(crate) fn apply_activation_context(
        &self,
        activation: &CoreWorkflowActivation,
        is_replaying_history_events: bool,
    ) {
        let mut shared = self.inner.shared.borrow_mut();
        shared.activation = activation.clone();
        shared.is_replaying_history_events = is_replaying_history_events;
        if let Some(seed) = activation.jobs.iter().find_map(|job| match &job.variant {
            Some(ActivationVariant::UpdateRandomSeed(attrs)) => Some(attrs.randomness_seed),
            _ => None,
        }) {
            shared.random = Pcg64Mcg::seed_from_u64(seed);
        }
    }

    fn random<T>(&self) -> T
    where
        T: WorkflowRandomValue,
    {
        let random = &mut self.inner.shared.borrow_mut().random;
        <T as private::Sealed>::sample(random)
    }

    fn uuid4(&self) -> String {
        Builder::from_random_bytes(self.random::<u128>().to_be_bytes())
            .into_uuid()
            .hyphenated()
            .to_string()
    }

    /// Returns the [`DataConverter`] associated with this workflow's worker.
    pub fn data_converter(&self) -> &DataConverter {
        &self.inner.data_converter
    }

    /// Return the workflow's unique identifier.
    pub fn workflow_id(&self) -> &str {
        &self.inner.initial_information.workflow_id
    }

    /// Return the run id of this workflow execution.
    pub fn run_id(&self) -> &str {
        &self.inner.run_id
    }

    /// Return the namespace the workflow is executing in.
    pub fn namespace(&self) -> &str {
        &self.inner.namespace
    }

    /// Return the task queue the workflow is executing in.
    pub fn task_queue(&self) -> &str {
        &self.inner.task_queue
    }

    /// Return the workflow type name.
    pub fn workflow_type(&self) -> &str {
        &self.inner.initial_information.workflow_type
    }

    pub(crate) fn initial_headers(&self) -> HashMap<String, Payload> {
        self.inner.initial_information.headers.clone()
    }

    /// Return the current time according to the workflow.
    pub fn workflow_time(&self) -> Option<SystemTime> {
        self.inner
            .shared
            .borrow()
            .activation
            .timestamp
            .try_into_or_none()
    }

    /// Return the length of history so far at this point in the workflow.
    pub fn history_length(&self) -> u32 {
        self.inner.shared.borrow().activation.history_length
    }

    /// Return current values for workflow search attributes.
    pub fn search_attributes(&self) -> SearchAttributes {
        SearchAttributes::from_proto(&self.inner.shared.borrow().search_attributes)
    }

    /// Returns true if the workflow is replaying (including during queries and update validators), false otherwise.
    pub fn is_replaying(&self) -> bool {
        self.inner.shared.borrow().activation.is_replaying
    }

    /// Return true if the workflow is replaying history events (excluding queries and update validators), false otherwise.
    pub fn is_replaying_history_events(&self) -> bool {
        self.inner.shared.borrow().is_replaying_history_events
    }

    /// Returns the payload converter used by the worker running this workflow.
    pub fn payload_converter(&self) -> &PayloadConverter {
        self.inner.data_converter.payload_converter()
    }

    pub(crate) fn construction_waker(&self) -> Waker {
        self.inner
            .current_waker
            .borrow()
            .clone()
            .unwrap_or_else(|| Waker::noop().clone())
    }

    pub(crate) fn enter_runtime_poll<'a>(&'a self, waker: &Waker) -> WorkflowPollWakerGuard<'a> {
        WorkflowPollWakerGuard {
            previous: self.inner.current_waker.replace(Some(waker.clone())),
            current_waker: &self.inner.current_waker,
        }
    }

    pub(crate) fn notify_patch(&self, patch_id: String) {
        self.inner
            .shared
            .borrow_mut()
            .notified_patches
            .insert(patch_id);
    }

    fn prepare_outbound_future<T>(
        &self,
        mut future: WorkflowOutboundFuture<T>,
    ) -> WorkflowOutboundFuture<T> {
        let waker = self.construction_waker();
        let mut cx = Context::from_waker(&waker);
        future.poll_for_construction(&mut cx);
        future
    }

    fn prepare_cancellable_outbound_future<T>(
        &self,
        mut future: CancellableWorkflowOutboundFuture<T>,
    ) -> CancellableWorkflowOutboundFuture<T> {
        let waker = self.construction_waker();
        let mut cx = Context::from_waker(&waker);
        future.poll_for_construction(&mut cx);
        future
    }

    /// Create a read-only view of this context.
    pub(crate) fn view(&self) -> WorkflowContextView {
        let shared = self.inner.shared.borrow();
        let mut initial_information = self.inner.initial_information.clone();
        if initial_information.memo.is_some() || !shared.memo.fields.is_empty() {
            initial_information.memo = Some(shared.memo.clone());
        }
        WorkflowContextView::new(
            self.inner.namespace.clone(),
            self.inner.task_queue.clone(),
            self.inner.run_id.clone(),
            initial_information,
            self.inner.data_converter.payload_converter().clone(),
        )
    }
}

#[derive(Debug, PartialEq, Eq, Hash, Clone, Copy)]
enum PendingCommandId {
    Timer(u32),
    Activity(u32),
    ChildWorkflowStart(u32),
    ChildWorkflowComplete(u32),
    SignalExternal(u32),
    CancelExternal(u32),
    NexusOpStart(u32),
    NexusOpComplete(u32),
}

impl PendingCommandId {
    fn from_unblock_event(event: &UnblockEvent) -> Self {
        match event {
            UnblockEvent::Timer(seq, _) => Self::Timer(*seq),
            UnblockEvent::Activity(seq, _) => Self::Activity(*seq),
            UnblockEvent::WorkflowStart(seq, _) => Self::ChildWorkflowStart(*seq),
            UnblockEvent::WorkflowComplete(seq, _) => Self::ChildWorkflowComplete(*seq),
            UnblockEvent::SignalExternal(seq, _) => Self::SignalExternal(*seq),
            UnblockEvent::CancelExternal(seq, _) => Self::CancelExternal(*seq),
            UnblockEvent::NexusOperationStart(seq, _) => Self::NexusOpStart(*seq),
            UnblockEvent::NexusOperationComplete(seq, _) => Self::NexusOpComplete(*seq),
        }
    }
}

struct WorkflowRuntimeState {
    host: Rc<dyn WorkflowHost>,
    pending_unblocks: RefCell<HashMap<PendingCommandId, oneshot::Sender<UnblockEvent>>>,
    forced_wft_failure: RefCell<Option<Box<dyn std::error::Error + Send + Sync>>>,
    progress_made: Cell<bool>,
}

impl WorkflowRuntimeState {
    fn new(host: Rc<dyn WorkflowHost>) -> Self {
        Self {
            host,
            pending_unblocks: RefCell::new(HashMap::new()),
            forced_wft_failure: RefCell::new(None),
            progress_made: Cell::new(false),
        }
    }

    fn register_unblocker(&self, id: PendingCommandId, unblocker: oneshot::Sender<UnblockEvent>) {
        self.pending_unblocks.borrow_mut().insert(id, unblocker);
    }

    fn unblock(&self, event: UnblockEvent) -> Result<(), anyhow::Error> {
        let id = PendingCommandId::from_unblock_event(&event);
        let unblocker = self
            .pending_unblocks
            .borrow_mut()
            .remove(&id)
            .ok_or_else(|| anyhow::anyhow!("Command {id:?} not found to unblock"))?;
        self.progress_made.set(true);
        let _guard = SdkWakeGuard::new();
        let _ = unblocker.send(event);
        Ok(())
    }

    fn maybe_unblock(&self, event: UnblockEvent) -> bool {
        let id = PendingCommandId::from_unblock_event(&event);
        let Some(unblocker) = self.pending_unblocks.borrow_mut().remove(&id) else {
            return false;
        };
        self.progress_made.set(true);
        let _guard = SdkWakeGuard::new();
        let _ = unblocker.send(event);
        true
    }

    fn set_forced_wft_failure(&self, err: Box<dyn std::error::Error + Send + Sync>) {
        *self.forced_wft_failure.borrow_mut() = Some(err);
        self.progress_made.set(true);
    }

    fn take_forced_wft_failure(&self) -> Option<Box<dyn std::error::Error + Send + Sync>> {
        self.forced_wft_failure.borrow_mut().take()
    }

    fn mark_progress(&self) {
        self.progress_made.set(true);
    }

    fn take_progress(&self) -> bool {
        self.progress_made.replace(false)
    }
}

struct WorkflowContextInner {
    namespace: String,
    task_queue: String,
    run_id: String,
    initial_information: InitializeWorkflow,
    runtime: WorkflowRuntimeState,
    cancelled_reason: RefCell<Option<String>>,
    cancel_wakers: RefCell<Vec<Waker>>,
    shared: RefCell<WorkflowContextSharedData>,
    seq_nums: RefCell<WfCtxProtectedDat>,
    data_converter: DataConverter,
    patch_activation_callback: Option<PatchActivationCallback>,
    state_mutated: Cell<bool>,
    current_waker: RefCell<Option<Waker>>,
    workflow_interceptors: Rc<[Arc<dyn WorkflowInterceptor>]>,
}

/// Context provided to synchronous signal and update handlers.
///
/// This type provides all workflow context capabilities except `state()`, `state_mut()`,
/// and `wait_condition()`. Those methods are not applicable in sync handler contexts.
///
/// Sync handlers receive `&mut self` directly, so they can reference and mutate workflow state without
/// needing `state()`/`state_mut()`.
pub struct SyncWorkflowContext<W> {
    base: BaseWorkflowContext,
    /// Headers from the current handler invocation (signal, update, etc.)
    headers: Rc<HashMap<String, Payload>>,
    _phantom: PhantomData<W>,
}

impl<W> Clone for SyncWorkflowContext<W> {
    fn clone(&self) -> Self {
        Self {
            base: self.base.clone(),
            headers: self.headers.clone(),
            _phantom: PhantomData,
        }
    }
}

/// Used within workflows to issue commands, get info, etc.
///
/// The type parameter `W` represents the workflow type. This enables type-safe
/// access to workflow state via `state_mut()` for mutations.
pub struct WorkflowContext<W> {
    sync: SyncWorkflowContext<W>,
    /// The workflow instance
    workflow_state: Rc<RefCell<W>>,
    /// Wakers registered by `wait_condition` futures. Drained and woken on
    /// every `state_mut` call so that waker-based combinators (e.g.
    /// `FuturesOrdered`) re-poll the condition after state changes.
    condition_wakers: Rc<RefCell<Vec<Waker>>>,
}

impl<W> Clone for WorkflowContext<W> {
    fn clone(&self) -> Self {
        Self {
            sync: self.sync.clone(),
            workflow_state: self.workflow_state.clone(),
            condition_wakers: self.condition_wakers.clone(),
        }
    }
}

impl BaseWorkflowContext {
    /// Construct a base context and its interceptors from initial workflow information.
    #[doc(hidden)]
    pub fn from_raw(
        init: WorkflowInit,
        data_converter: DataConverter,
        host: Rc<dyn WorkflowHost>,
        patch_activation_callback: Option<PatchActivationCallback>,
        workflow_interceptor_constructors: Vec<WorkflowInterceptorConstructor>,
    ) -> Self {
        let WorkflowInit {
            namespace,
            task_queue,
            run_id,
            initialize_workflow,
        } = init;
        let view = WorkflowContextView::new(
            namespace,
            task_queue,
            run_id,
            initialize_workflow,
            data_converter.payload_converter().clone(),
        );
        let workflow_interceptors = workflow_interceptor_constructors
            .into_iter()
            .map(|constructor| constructor.construct(&view))
            .collect::<Vec<_>>()
            .into();
        let (namespace, task_queue, run_id, init_workflow_job) = view.into_parts();
        Self {
            inner: Rc::new(WorkflowContextInner {
                namespace,
                task_queue,
                run_id,
                shared: RefCell::new(WorkflowContextSharedData {
                    random: Pcg64Mcg::seed_from_u64(init_workflow_job.randomness_seed),
                    memo: init_workflow_job.memo.clone().unwrap_or_default(),
                    search_attributes: init_workflow_job
                        .search_attributes
                        .clone()
                        .unwrap_or_default(),
                    is_replaying_history_events: false,
                    changes: Default::default(),
                    activation: Default::default(),
                    current_details: Default::default(),
                    notified_patches: Default::default(),
                }),
                initial_information: init_workflow_job,
                runtime: WorkflowRuntimeState::new(host),
                cancelled_reason: RefCell::new(None),
                cancel_wakers: RefCell::new(Vec::new()),
                seq_nums: RefCell::new(WfCtxProtectedDat {
                    next_timer_sequence_number: 1,
                    next_activity_sequence_number: 1,
                    next_child_workflow_sequence_number: 1,
                    next_cancel_external_wf_sequence_number: 1,
                    next_signal_external_wf_sequence_number: 1,
                    next_nexus_op_sequence_number: 1,
                }),
                data_converter,
                patch_activation_callback,
                state_mutated: Cell::new(false),
                current_waker: RefCell::new(None),
                workflow_interceptors,
            }),
        }
    }

    pub(crate) fn workflow_interceptors(&self) -> Rc<[Arc<dyn WorkflowInterceptor>]> {
        self.inner.workflow_interceptors.clone()
    }

    /// Check and clear the state_mutated flag. Returns `true` if `state_mut`
    /// was called since the last time this method was invoked.
    pub(crate) fn take_state_mutated(&self) -> bool {
        self.inner.state_mutated.replace(false)
    }

    /// Mark that workflow state has been mutated.
    pub(crate) fn set_state_mutated(&self) {
        self.inner.state_mutated.set(true);
    }

    pub(crate) fn take_runtime_progress(&self) -> bool {
        self.inner.runtime.take_progress()
    }

    pub(crate) fn take_forced_wft_failure(
        &self,
    ) -> Option<Box<dyn std::error::Error + Send + Sync>> {
        self.inner.runtime.take_forced_wft_failure()
    }

    pub(crate) fn notify_cancel(&self, reason: String) {
        let _guard = SdkWakeGuard::new();
        *self.inner.cancelled_reason.borrow_mut() = Some(reason);
        for waker in self.inner.cancel_wakers.borrow_mut().drain(..) {
            waker.wake();
        }
        self.inner.runtime.mark_progress();
    }

    pub(crate) fn unblock(&self, event: UnblockEvent) -> Result<(), anyhow::Error> {
        self.inner.runtime.unblock(event)
    }

    /// Cancel any cancellable operation by ID
    fn cancel(&self, cancellable_id: CancellableID) {
        match cancellable_id {
            CancellableID::Timer(seq) => {
                if self
                    .inner
                    .runtime
                    .maybe_unblock(UnblockEvent::Timer(seq, TimerResult::Cancelled))
                {
                    self.inner.runtime.host.push_command(
                        workflow_command::Variant::CancelTimer(CancelTimer { seq }).into(),
                    );
                }
            }
            CancellableID::Activity(seq) => {
                self.inner.runtime.host.push_command(
                    workflow_command::Variant::RequestCancelActivity(RequestCancelActivity { seq })
                        .into(),
                );
            }
            CancellableID::LocalActivity(seq) => {
                self.inner.runtime.host.push_command(
                    workflow_command::Variant::RequestCancelLocalActivity(
                        RequestCancelLocalActivity { seq },
                    )
                    .into(),
                );
            }
            CancellableID::ChildWorkflow { seqnum, reason } => {
                self.inner.runtime.host.push_command(
                    workflow_command::Variant::CancelChildWorkflowExecution(
                        CancelChildWorkflowExecution {
                            child_workflow_seq: seqnum,
                            reason,
                        },
                    )
                    .into(),
                );
            }
            CancellableID::SignalExternalWorkflow(seq) => {
                self.inner.runtime.host.push_command(
                    workflow_command::Variant::CancelSignalWorkflow(CancelSignalWorkflow { seq })
                        .into(),
                );
            }
            CancellableID::NexusOp(seq) => {
                self.inner.runtime.host.push_command(
                    workflow_command::Variant::RequestCancelNexusOperation(
                        RequestCancelNexusOperation { seq },
                    )
                    .into(),
                );
            }
        }
    }

    fn cancellation_handle(&self, cancellable_id: CancellableID) -> WorkflowCancellationHandle {
        let base_ctx = self.clone();
        WorkflowCancellationHandle::new(move |reason| {
            let id = reason.map_or_else(
                || cancellable_id.clone(),
                |reason| cancellable_id.clone().with_reason(reason),
            );
            base_ctx.cancel(id);
        })
    }

    /// Return the current value of current_details.
    pub fn current_details(&self) -> String {
        self.inner.shared.borrow().current_details.clone()
    }

    /// Request to create a timer
    pub fn timer<T: Into<TimerOptions>>(
        &self,
        opts: T,
    ) -> impl CancellableFuture<TimerResult> + use<T> {
        let input = StartTimerInput::new(opts.into());
        let base_ctx = self.clone();
        let next = WorkflowNext::new(move |input: StartTimerInput| {
            let opts = input.into_options();
            let seq = base_ctx.inner.seq_nums.borrow_mut().next_timer_seq();
            let (cmd, unblocker) =
                CancellableWFCommandFut::new(CancellableID::Timer(seq), base_ctx.clone());
            base_ctx
                .inner
                .runtime
                .register_unblocker(PendingCommandId::Timer(seq), unblocker);
            base_ctx
                .inner
                .runtime
                .host
                .push_command(opts.into_command(seq));
            CancellableWorkflowOutboundFuture::new(
                cmd,
                base_ctx.cancellation_handle(CancellableID::Timer(seq)),
            )
        });
        let interceptors = self.inner.workflow_interceptors.clone();
        let future = call_start_timer(
            interceptors,
            WorkflowInterceptorContext::new(self.clone()),
            input,
            next,
        );
        self.prepare_cancellable_outbound_future(future)
    }

    /// Request to run an activity
    #[allow(clippy::result_large_err)]
    pub fn execute_activity<AD: ActivityDefinition>(
        &self,
        activity: AD,
        input: impl Into<AD::Input>,
        opts: ActivityOptions,
    ) -> impl CancellableFuture<Result<AD::Output, ActivityExecutionError>>
    where
        AD::Output: TemporalDeserializable,
    {
        let input =
            ScheduleActivityInput::new(activity.name().to_string(), Box::new(input.into()), opts);
        let base_ctx = self.clone();
        let next = WorkflowNext::new(move |input: ScheduleActivityInput| {
            let (activity_type, input, headers, mut opts) = input.into_parts();
            let input = match input.downcast::<AD::Input>() {
                Ok(input) => *input,
                Err(_) => {
                    return CancellableWorkflowOutboundFuture::new(
                        async {
                            Err(ActivityExecutionError::Serialization(outbound_type_error(
                                "activity input",
                            )))
                        },
                        WorkflowCancellationHandle::noop(),
                    );
                }
            };
            let payload_converter = base_ctx.inner.data_converter.payload_converter();
            let ctx = SerializationContext {
                data: &SerializationContextData::Workflow,
                converter: payload_converter,
            };
            match payload_converter.to_payloads(&ctx, &input) {
                Ok(payloads) => {
                    let seq = base_ctx.inner.seq_nums.borrow_mut().next_activity_seq();
                    let (cmd, unblocker) = CancellableWFCommandFut::new(
                        CancellableID::Activity(seq),
                        base_ctx.clone(),
                    );
                    base_ctx
                        .inner
                        .runtime
                        .register_unblocker(PendingCommandId::Activity(seq), unblocker);
                    if opts.task_queue.is_none() {
                        opts.task_queue = Some(base_ctx.inner.task_queue.clone());
                    }
                    base_ctx.inner.runtime.host.push_command(opts.into_command(
                        seq,
                        activity_type,
                        payloads,
                        headers,
                    ));
                    CancellableWorkflowOutboundFuture::new(
                        ActivityFut::running(cmd, base_ctx.inner.data_converter.clone()),
                        base_ctx.cancellation_handle(CancellableID::Activity(seq)),
                    )
                }
                Err(err) => CancellableWorkflowOutboundFuture::new(
                    ActivityFut::<future::Ready<ActivityResolution>, AD::Output>::eager(err.into()),
                    WorkflowCancellationHandle::noop(),
                ),
            }
            .map(|result| result.map(|output| Box::new(output) as Box<dyn WorkflowOutboundValue>))
        });
        let interceptors = self.inner.workflow_interceptors.clone();
        let future = call_schedule_activity(
            interceptors,
            WorkflowInterceptorContext::new(self.clone()),
            input,
            next,
        )
        .map(|result| {
            result.and_then(|output| {
                output
                    .downcast::<AD::Output>()
                    .map(|output| *output)
                    .map_err(|_| {
                        ActivityExecutionError::Serialization(outbound_type_error(
                            "activity output",
                        ))
                    })
            })
        });
        self.prepare_cancellable_outbound_future(future)
    }

    /// Request to run a local activity
    #[allow(clippy::result_large_err)]
    pub fn execute_local_activity<AD: ActivityDefinition>(
        &self,
        activity: AD,
        input: impl Into<AD::Input>,
        opts: LocalActivityOptions,
    ) -> impl CancellableFuture<Result<AD::Output, ActivityExecutionError>>
    where
        AD::Output: TemporalDeserializable,
    {
        let input = ScheduleLocalActivityInput::new(
            activity.name().to_string(),
            Box::new(input.into()),
            opts,
        );
        let base_ctx = self.clone();
        let next = WorkflowNext::new(move |input: ScheduleLocalActivityInput| {
            let (activity_type, input, headers, opts) = input.into_parts();
            let input = match input.downcast::<AD::Input>() {
                Ok(input) => *input,
                Err(_) => {
                    return CancellableWorkflowOutboundFuture::new(
                        async {
                            Err(ActivityExecutionError::Serialization(outbound_type_error(
                                "local activity input",
                            )))
                        },
                        WorkflowCancellationHandle::noop(),
                    );
                }
            };
            let payload_converter = base_ctx.inner.data_converter.payload_converter();
            let ctx = SerializationContext {
                data: &SerializationContextData::Workflow,
                converter: payload_converter,
            };
            match payload_converter.to_payloads(&ctx, &input) {
                Ok(payloads) => {
                    let future = LATimerBackoffFut::new(
                        activity_type,
                        payloads,
                        headers,
                        opts,
                        base_ctx.clone(),
                    );
                    cancellable_outbound(ActivityFut::running(
                        future,
                        base_ctx.inner.data_converter.clone(),
                    ))
                }
                Err(err) => CancellableWorkflowOutboundFuture::new(
                    ActivityFut::<future::Ready<ActivityResolution>, AD::Output>::eager(err.into()),
                    WorkflowCancellationHandle::noop(),
                ),
            }
            .map(|result| result.map(|output| Box::new(output) as Box<dyn WorkflowOutboundValue>))
        });
        let interceptors = self.inner.workflow_interceptors.clone();
        let future = call_schedule_local_activity(
            interceptors,
            WorkflowInterceptorContext::new(self.clone()),
            input,
            next,
        )
        .map(|result| {
            result.and_then(|output| {
                output
                    .downcast::<AD::Output>()
                    .map(|output| *output)
                    .map_err(|_| {
                        ActivityExecutionError::Serialization(outbound_type_error(
                            "local activity output",
                        ))
                    })
            })
        });
        self.prepare_cancellable_outbound_future(future)
    }

    /// Start a child workflow with typed input/output.
    pub(crate) fn start_child_workflow<WD: WorkflowDefinition + 'static>(
        &self,
        workflow: WD,
        input: impl Into<WD::Input>,
        opts: ChildWorkflowOptions,
    ) -> impl CancellableFutureWithReason<Result<StartedChildWorkflow<WD>, ChildWorkflowStartError>>
    where
        WD::Output: TemporalDeserializable,
    {
        let input =
            StartChildWorkflowInput::new(workflow.name().to_string(), Box::new(input.into()), opts);
        let base_ctx = self.clone();
        let next = WorkflowNext::new(move |input: StartChildWorkflowInput| {
            let (workflow_type, input, headers, mut opts) = input.into_parts();
            let input = match input.downcast::<WD::Input>() {
                Ok(input) => *input,
                Err(_) => {
                    return CancellableWorkflowOutboundFuture::new(
                        async {
                            Err(ChildWorkflowStartError::Serialization(outbound_type_error(
                                "child workflow input",
                            )))
                        },
                        WorkflowCancellationHandle::noop(),
                    );
                }
            };
            let payload_converter = base_ctx.inner.data_converter.payload_converter();
            let ctx = SerializationContext {
                data: &SerializationContextData::Workflow,
                converter: payload_converter,
            };
            let payloads = match payload_converter.to_payloads(&ctx, &input) {
                Ok(payloads) => payloads,
                Err(err) => {
                    return CancellableWorkflowOutboundFuture::new(
                        ChildWorkflowStartFut::<future::Ready<PendingChildWorkflow<WD>>, WD>::eager(
                            err.into(),
                        ),
                        WorkflowCancellationHandle::noop(),
                    );
                }
            };
            let workflow_id = opts
                .workflow_id
                .take()
                .filter(|id| !id.is_empty())
                .unwrap_or_else(|| base_ctx.uuid4());

            let child_seq = base_ctx
                .inner
                .seq_nums
                .borrow_mut()
                .next_child_workflow_seq();
            // Immediately create the command/future for the result, otherwise if the user does
            // not await the result until *after* we receive an activation for it, there will be nothing
            // to match when unblocking.
            let (result_cmd, unblocker) = CancellableWFCommandFut::new(
                CancellableID::ChildWorkflow {
                    seqnum: child_seq,
                    reason: String::new(),
                },
                base_ctx.clone(),
            );
            base_ctx.inner.runtime.register_unblocker(
                PendingCommandId::ChildWorkflowComplete(child_seq),
                unblocker,
            );

            let common = ChildWfCommon {
                workflow_id: workflow_id.clone(),
                child_seq,
                result_future: result_cmd,
                base_ctx: base_ctx.clone(),
                data_converter: base_ctx.inner.data_converter.clone(),
            };

            let (cmd, unblocker) =
                CancellableWFCommandFut::<PendingChildWorkflow<WD>, ChildWfCommon>::new_with_dat(
                    CancellableID::ChildWorkflow {
                        seqnum: child_seq,
                        reason: String::new(),
                    },
                    common,
                    base_ctx.clone(),
                );
            base_ctx
                .inner
                .runtime
                .register_unblocker(PendingCommandId::ChildWorkflowStart(child_seq), unblocker);
            base_ctx.inner.runtime.host.push_command(opts.into_command(
                child_seq,
                workflow_type,
                payloads,
                headers,
                workflow_id,
            ));

            cancellable_outbound_with_reason(ChildWorkflowStartFut::Running(cmd))
        });
        let interceptors = self.inner.workflow_interceptors.clone();
        let future = call_start_child_workflow(
            interceptors,
            WorkflowInterceptorContext::new(self.clone()),
            input,
            next,
        )
        .map(|result| result.map(StartChildWorkflowOutput::into_started));
        self.prepare_cancellable_outbound_future(future)
    }

    /// Request to run a local activity with no implementation of timer-backoff based retrying.
    fn local_activity_no_timer_retry(
        self,
        activity_type: String,
        arguments: Vec<Payload>,
        headers: HashMap<String, Payload>,
        opts: LocalActivityOptions,
    ) -> impl CancellableFuture<ActivityResolution> {
        let seq = self.inner.seq_nums.borrow_mut().next_activity_seq();
        let (cmd, unblocker) =
            CancellableWFCommandFut::new(CancellableID::LocalActivity(seq), self.clone());
        self.inner
            .runtime
            .register_unblocker(PendingCommandId::Activity(seq), unblocker);
        self.inner.runtime.host.push_command(opts.into_command(
            seq,
            activity_type,
            arguments,
            headers,
        ));
        cmd
    }

    fn signal_workflow<S: SignalDefinition + 'static>(
        &self,
        target: SignalWorkflowTarget,
        signal: S,
        input: S::Input,
    ) -> CancellableWorkflowOutboundFuture<SignalWorkflowResult> {
        let input = SignalWorkflowInput::new(S::name(&signal).to_string(), target, Box::new(input));
        let base_ctx = self.clone();
        let next = WorkflowNext::new(move |input: SignalWorkflowInput| {
            let (signal_name, target, input, headers) = input.into_parts();
            let input = match input.downcast::<S::Input>() {
                Ok(input) => *input,
                Err(_) => {
                    return CancellableWorkflowOutboundFuture::new(
                        async {
                            Err(WorkflowSignalError::Serialization(outbound_type_error(
                                "signal input",
                            )))
                        },
                        WorkflowCancellationHandle::noop(),
                    );
                }
            };
            let payload_converter = base_ctx.data_converter().payload_converter();
            let ctx = SerializationContext {
                data: &SerializationContextData::Workflow,
                converter: payload_converter,
            };
            let payloads = match payload_converter.to_payloads(&ctx, &input) {
                Ok(payloads) => payloads,
                Err(err) => {
                    return CancellableWorkflowOutboundFuture::new(
                        async move { Err(err.into()) },
                        WorkflowCancellationHandle::noop(),
                    );
                }
            };
            let target = match target {
                SignalWorkflowTarget::Child { workflow_id } => {
                    signal_external_workflow_execution::Target::ChildWorkflowId(workflow_id)
                }
                SignalWorkflowTarget::External {
                    namespace,
                    workflow_id,
                    run_id,
                } => signal_external_workflow_execution::Target::WorkflowExecution(
                    NamespacedWorkflowExecution {
                        namespace,
                        workflow_id,
                        run_id: run_id.unwrap_or_default(),
                    },
                ),
            };
            let mut signal = Signal::new(signal_name, payloads);
            signal.data.headers = headers;
            let seq = base_ctx
                .inner
                .seq_nums
                .borrow_mut()
                .next_signal_external_wf_seq();
            let (cmd, unblocker) = CancellableWFCommandFut::new(
                CancellableID::SignalExternalWorkflow(seq),
                base_ctx.clone(),
            );
            base_ctx
                .inner
                .runtime
                .register_unblocker(PendingCommandId::SignalExternal(seq), unblocker);
            let signal = signal.into_invocation();
            base_ctx.inner.runtime.host.push_command(
                workflow_command::Variant::SignalExternalWorkflowExecution(
                    SignalExternalWorkflowExecution {
                        seq,
                        signal_name: signal.signal_name,
                        args: signal.input,
                        target: Some(target),
                        headers: signal.headers,
                    },
                )
                .into(),
            );
            cancellable_outbound(SignalChildFut::Running {
                inner: cmd,
                data_converter: base_ctx.data_converter().clone(),
            })
        });
        let interceptors = self.inner.workflow_interceptors.clone();
        let future = call_signal_workflow(
            interceptors,
            WorkflowInterceptorContext::new(self.clone()),
            input,
            next,
        );
        self.prepare_cancellable_outbound_future(future)
    }

    pub(crate) fn external_workflow(
        &self,
        workflow_id: impl Into<String>,
        run_id: Option<String>,
    ) -> ExternalWorkflowHandle {
        ExternalWorkflowHandle {
            workflow_id: workflow_id.into(),
            run_id,
            namespace: self.inner.namespace.clone(),
            base_ctx: self.clone(),
        }
    }

    fn cancel_external_workflow(
        &self,
        input: CancelExternalWorkflowInput,
    ) -> WorkflowOutboundFuture<CancelExternalWfResult> {
        let base_ctx = self.clone();
        let next = WorkflowNext::new(move |input: CancelExternalWorkflowInput| {
            let seq = base_ctx
                .inner
                .seq_nums
                .borrow_mut()
                .next_cancel_external_wf_seq();
            let (cmd, unblocker) = WFCommandFut::new();
            base_ctx
                .inner
                .runtime
                .register_unblocker(PendingCommandId::CancelExternal(seq), unblocker);
            base_ctx.inner.runtime.host.push_command(
                workflow_command::Variant::RequestCancelExternalWorkflowExecution(
                    RequestCancelExternalWorkflowExecution {
                        seq,
                        workflow_execution: Some(NamespacedWorkflowExecution {
                            namespace: base_ctx.inner.namespace.clone(),
                            workflow_id: input.workflow_id,
                            run_id: input.run_id.unwrap_or_default(),
                        }),
                        reason: input.reason.unwrap_or_default(),
                    },
                )
                .into(),
            );
            WorkflowOutboundFuture::new(cmd)
        });
        let interceptors = self.inner.workflow_interceptors.clone();
        let future = call_cancel_external_workflow(
            interceptors,
            WorkflowInterceptorContext::new(self.clone()),
            input,
            next,
        );
        self.prepare_outbound_future(future)
    }

    pub(crate) fn start_nexus_operation(
        &self,
        opts: NexusOperationOptions,
    ) -> impl CancellableFuture<NexusStartResult> {
        let input = StartNexusOperationInput::new(opts);
        let base_ctx = self.clone();
        let next = WorkflowNext::new(move |input: StartNexusOperationInput| {
            let opts = input.into_options();
            let seq = base_ctx.inner.seq_nums.borrow_mut().next_nexus_op_seq();
            let (result_future, unblocker) = WFCommandFut::new();
            base_ctx
                .inner
                .runtime
                .register_unblocker(PendingCommandId::NexusOpComplete(seq), unblocker);
            let (cmd, unblocker) = CancellableWFCommandFut::new_with_dat(
                CancellableID::NexusOp(seq),
                NexusUnblockData {
                    result_future: result_future.shared(),
                    schedule_seq: seq,
                    base_ctx: base_ctx.clone(),
                },
                base_ctx.clone(),
            );
            base_ctx
                .inner
                .runtime
                .register_unblocker(PendingCommandId::NexusOpStart(seq), unblocker);
            base_ctx
                .inner
                .runtime
                .host
                .push_command(opts.into_command(seq));
            cancellable_outbound(cmd)
        });
        let interceptors = self.inner.workflow_interceptors.clone();
        let future = call_start_nexus_operation(
            interceptors,
            WorkflowInterceptorContext::new(self.clone()),
            input,
            next,
        );
        self.prepare_cancellable_outbound_future(future)
    }
}

impl<W> SyncWorkflowContext<W> {
    /// Return the workflow's unique identifier
    pub fn workflow_id(&self) -> &str {
        &self.base.inner.initial_information.workflow_id
    }

    /// Return the run id of this workflow execution
    pub fn run_id(&self) -> &str {
        &self.base.inner.run_id
    }

    /// Return the namespace the workflow is executing in
    pub fn namespace(&self) -> &str {
        &self.base.inner.namespace
    }

    /// Return the task queue the workflow is executing in
    pub fn task_queue(&self) -> &str {
        &self.base.inner.task_queue
    }

    /// Return the current time according to the workflow (which is not wall-clock time).
    pub fn workflow_time(&self) -> Option<SystemTime> {
        self.base
            .inner
            .shared
            .borrow()
            .activation
            .timestamp
            .try_into_or_none()
    }

    /// Return the length of history so far at this point in the workflow
    pub fn history_length(&self) -> u32 {
        self.base.inner.shared.borrow().activation.history_length
    }

    /// Return the deployment version, if any,  as it was when this point in the workflow was first
    /// reached. If this code is being executed for the first time, return this Worker's deployment
    /// version if it has one.
    pub fn current_deployment_version(&self) -> Option<WorkerDeploymentVersion> {
        self.base
            .inner
            .shared
            .borrow()
            .activation
            .clone()
            .deployment_version_for_current_task
            .map(Into::into)
    }

    /// Return current values for workflow search attributes.
    pub fn search_attributes(&self) -> SearchAttributes {
        SearchAttributes::from_proto(&self.base.inner.shared.borrow().search_attributes)
    }

    /// Return the current workflow memo values.
    pub fn memo(&self) -> Memo {
        Memo::from_raw(
            Some(self.base.inner.shared.borrow().memo.clone()),
            self.payload_converter().clone(),
            SerializationContextData::Workflow,
        )
    }

    /// Generates a deterministic pseudo-random value of type `T`.
    ///
    /// The value is derived from Temporal's workflow randomness seed, making it safe during
    /// replay. This generator is not cryptographically secure.
    pub fn random<T>(&self) -> T
    where
        T: WorkflowRandomValue,
    {
        self.base.random()
    }

    /// Generates a deterministic lowercase, hyphenated version 4 UUID string.
    ///
    /// This uses [`Self::random`] and is not cryptographically secure.
    pub fn uuid4(&self) -> String {
        self.base.uuid4()
    }

    /// Returns true if the current workflow task is happening under replay
    pub fn is_replaying(&self) -> bool {
        self.base.inner.shared.borrow().activation.is_replaying
    }

    /// Returns true if the current work is replaying history events
    pub fn is_replaying_history_events(&self) -> bool {
        self.base.inner.shared.borrow().is_replaying_history_events
    }

    /// Returns true if the server suggests this workflow should continue-as-new
    pub fn continue_as_new_suggested(&self) -> bool {
        self.base
            .inner
            .shared
            .borrow()
            .activation
            .continue_as_new_suggested
    }

    /// Returns true if the workflow's target worker deployment version changed.
    ///
    /// This experimental signal is intended for workers using worker deployment versioning.
    pub fn target_worker_deployment_version_changed(&self) -> bool {
        self.base
            .inner
            .shared
            .borrow()
            .activation
            .target_worker_deployment_version_changed
    }

    /// Returns the headers for the current handler invocation (signal, update, query, etc.).
    ///
    /// When called from within a signal handler, returns the headers that were sent with that
    /// signal. When called from the main workflow run method, returns an empty map.
    pub fn headers(&self) -> &HashMap<String, Payload> {
        &self.headers
    }

    /// Returns the [PayloadConverter] currently used by the worker running this workflow.
    pub fn payload_converter(&self) -> &PayloadConverter {
        self.base.inner.data_converter.payload_converter()
    }

    /// Return Rust-native information about this workflow execution.
    pub fn info(&self) -> WorkflowContextView {
        self.view()
    }

    /// A future that resolves if/when the workflow is cancelled, with the user provided cause
    pub fn cancelled(&self) -> impl FusedFuture<Output = String> + '_ {
        let inner = self.base.inner.clone();
        future::poll_fn(move |cx| {
            if let Some(reason) = inner.cancelled_reason.borrow().as_ref() {
                Poll::Ready(reason.clone())
            } else {
                inner.cancel_wakers.borrow_mut().push(cx.waker().clone());
                Poll::Pending
            }
        })
        .fuse()
    }

    /// Signal that this workflow should continue as a new workflow execution with the given input and
    /// options.
    ///
    /// This always returns an `Err` which should be propigated.
    pub fn continue_as_new(
        &self,
        input: <W::Run as WorkflowDefinition>::Input,
        opts: ContinueAsNewOptions,
    ) -> Result<std::convert::Infallible, WorkflowTermination>
    where
        W: WorkflowImplementation,
    {
        let input = ContinueAsNewInput::new(Box::new(input), opts);
        let base_ctx = self.base.clone();
        let workflow_type = base_ctx.workflow_type().to_string();
        let next = WorkflowNext::new(move |input: ContinueAsNewInput| {
            let (input, headers, opts) = input.into_parts();
            let input = match input.downcast::<<W::Run as WorkflowDefinition>::Input>() {
                Ok(input) => input,
                Err(_) => return Err(outbound_type_error("continue-as-new input").into()),
            };
            let pc = base_ctx.data_converter().payload_converter();
            let ctx = SerializationContext {
                data: &SerializationContextData::Workflow,
                converter: pc,
            };
            let arguments = pc
                .to_payloads(&ctx, &*input)
                .map_err(WorkflowTermination::from)?;
            let request = opts.into_request(workflow_type, arguments, headers, pc)?;
            Err(WorkflowTermination::continue_as_new(request))
        });
        let interceptors = self.base.inner.workflow_interceptors.clone();
        call_continue_as_new(
            interceptors,
            crate::workflow_interceptors::SyncWorkflowInterceptorContext::new(self.base.clone()),
            input,
            next,
        )
    }

    /// Request to create a timer
    pub fn timer<T: Into<TimerOptions>>(&self, opts: T) -> impl CancellableFuture<TimerResult> {
        self.base.timer(opts)
    }

    /// Request to run an activity
    pub fn execute_activity<AD: ActivityDefinition>(
        &self,
        activity: AD,
        input: impl Into<AD::Input>,
        opts: ActivityOptions,
    ) -> impl CancellableFuture<Result<AD::Output, ActivityExecutionError>>
    where
        AD::Output: TemporalDeserializable,
    {
        self.base.execute_activity(activity, input, opts)
    }

    /// Request to run an activity
    ///
    /// Deprecated alias for [`SyncWorkflowContext::execute_activity`].
    #[deprecated(note = "use `execute_activity` instead")]
    pub fn start_activity<AD: ActivityDefinition>(
        &self,
        activity: AD,
        input: impl Into<AD::Input>,
        opts: ActivityOptions,
    ) -> impl CancellableFuture<Result<AD::Output, ActivityExecutionError>>
    where
        AD::Output: TemporalDeserializable,
    {
        self.execute_activity(activity, input, opts)
    }

    /// Request to run a local activity
    pub fn execute_local_activity<AD: ActivityDefinition>(
        &self,
        activity: AD,
        input: impl Into<AD::Input>,
        opts: LocalActivityOptions,
    ) -> impl CancellableFuture<Result<AD::Output, ActivityExecutionError>>
    where
        AD::Output: TemporalDeserializable,
    {
        self.base.execute_local_activity(activity, input, opts)
    }

    /// Request to run a local activity
    ///
    /// Deprecated alias for [`SyncWorkflowContext::execute_local_activity`].
    #[deprecated(note = "use `execute_local_activity` instead")]
    pub fn start_local_activity<AD: ActivityDefinition>(
        &self,
        activity: AD,
        input: impl Into<AD::Input>,
        opts: LocalActivityOptions,
    ) -> impl CancellableFuture<Result<AD::Output, ActivityExecutionError>>
    where
        AD::Output: TemporalDeserializable,
    {
        self.execute_local_activity(activity, input, opts)
    }

    /// Start a child workflow. Returns a future that resolves to a [StartedChildWorkflow]
    /// which can be used to await the result, send signals, or cancel the child.
    pub fn start_child_workflow<WD: WorkflowDefinition + 'static>(
        &self,
        workflow: WD,
        input: impl Into<WD::Input>,
        opts: ChildWorkflowOptions,
    ) -> impl CancellableFutureWithReason<Result<StartedChildWorkflow<WD>, ChildWorkflowStartError>>
    where
        WD::Output: TemporalDeserializable,
    {
        self.base.start_child_workflow(workflow, input, opts)
    }

    /// Deprecated alias for [`SyncWorkflowContext::start_child_workflow`].
    #[deprecated(note = "use `start_child_workflow` instead")]
    pub fn child_workflow<WD: WorkflowDefinition + 'static>(
        &self,
        workflow: WD,
        input: impl Into<WD::Input>,
        opts: ChildWorkflowOptions,
    ) -> impl CancellableFutureWithReason<Result<StartedChildWorkflow<WD>, ChildWorkflowStartError>>
    where
        WD::Output: TemporalDeserializable,
    {
        self.start_child_workflow(workflow, input, opts)
    }

    /// Check (or record) that this workflow history was created with the provided patch.
    ///
    /// Workers can use their experimental patch activation callback to delay activating a newly
    /// introduced patch during a rolling deployment. The callback is only consulted when the
    /// marker would otherwise be created for the first time.
    pub fn patched(&self, patch_id: &str) -> bool {
        self.patch_impl(patch_id, false)
    }

    /// Record that this workflow history was created with the provided patch, and it is being
    /// phased out.
    pub fn deprecate_patch(&self, patch_id: &str) -> bool {
        self.patch_impl(patch_id, true)
    }

    fn patch_impl(&self, patch_id: &str, deprecated: bool) -> bool {
        if let Some(present) = self.base.inner.shared.borrow().changes.get(patch_id) {
            return *present;
        }

        let shared = self.base.inner.shared.borrow();
        let replaying = shared.activation.is_replaying;
        let notified = shared.notified_patches.contains(patch_id);
        drop(shared);

        // Replay and deprecation must follow history; only a fresh patch consults rollout policy.
        let res = if deprecated || replaying || notified {
            !replaying || notified
        } else if let Some(callback) = &self.base.inner.patch_activation_callback {
            callback(PatchActivationInput {
                workflow_info: self.base.view(),
                patch_id: patch_id.to_string(),
            })
        } else {
            true
        };

        if res {
            self.base.inner.runtime.host.push_command(
                workflow_command::Variant::SetPatchMarker(SetPatchMarker {
                    patch_id: patch_id.to_string(),
                    deprecated,
                })
                .into(),
            );
        }

        self.base
            .inner
            .shared
            .borrow_mut()
            .changes
            .insert(patch_id.to_string(), res);

        res
    }

    /// Get a handle to an external workflow for sending signals or requesting cancellation.
    pub fn external_workflow(
        &self,
        workflow_id: impl Into<String>,
        run_id: Option<String>,
    ) -> ExternalWorkflowHandle {
        self.base.external_workflow(workflow_id, run_id)
    }

    /// Add, update, or remove search attributes using typed keys.
    ///
    /// Updates are applied to the local in-memory view immediately so that
    /// subsequent calls to [`search_attributes()`](Self::search_attributes)
    /// reflect the changes. The command is also sent to the server.
    pub fn upsert_search_attributes(
        &self,
        updates: impl IntoIterator<Item = SearchAttributeUpdate>,
    ) {
        // Collect so we can iterate twice: once for local state, once for the
        // wire proto (which uses a different encoding for "unset").
        let updates: Vec<SearchAttributeUpdate> = updates.into_iter().collect();

        // Update local state using the typed API, which correctly removes keys
        // on unset (rather than inserting empty payloads like the wire format).
        {
            let mut shared = self.base.inner.shared.borrow_mut();
            let mut attrs = SearchAttributes::from_proto(&shared.search_attributes);
            for update in updates.iter().cloned() {
                attrs.apply(update);
            }
            shared.search_attributes = attrs.into_proto();
        }

        let proto = SearchAttributes::updates_to_proto(updates);
        self.base.inner.runtime.host.push_command(
            workflow_command::Variant::UpsertWorkflowSearchAttributes(
                UpsertWorkflowSearchAttributes {
                    search_attributes: Some(proto),
                },
            )
            .into(),
        );
    }

    /// Add or replace memo values with `Some`; remove memo keys with `None`.
    pub fn upsert_memo<K>(
        &self,
        updates: impl IntoIterator<Item = (K, Option<MemoValue>)>,
    ) -> Result<(), PayloadConversionError>
    where
        K: Into<String>,
    {
        let mut fields = HashMap::new();
        let mut local_updates = Vec::new();
        for (key, value) in updates {
            let key = key.into();
            let (command_payload, local_payload) = match value {
                Some(value) => {
                    let payload = value.to_payload(self.payload_converter())?;
                    (payload.clone(), Some(payload))
                }
                None => (
                    MemoValue::new(()).to_payload(self.payload_converter())?,
                    None,
                ),
            };
            fields.insert(key.clone(), command_payload);
            local_updates.push((key, local_payload));
        }
        {
            let mut shared = self.base.inner.shared.borrow_mut();
            for (key, payload) in local_updates {
                match payload {
                    Some(payload) => {
                        shared.memo.fields.insert(key, payload);
                    }
                    None => {
                        shared.memo.fields.remove(&key);
                    }
                }
            }
        }
        self.base.inner.runtime.host.push_command(
            workflow_command::Variant::ModifyWorkflowProperties(ModifyWorkflowProperties {
                upserted_memo: Some(ProtoMemo { fields }),
            })
            .into(),
        );
        Ok(())
    }

    /// Set the current details string for this workflow execution.
    ///
    /// The value is surfaced to the Temporal server UI in real time via the
    /// the workflow metadata query.
    pub fn set_current_details(&self, details: impl Into<String>) {
        let details = details.into();
        self.base.inner.shared.borrow_mut().current_details = details.clone();
        self.base.inner.runtime.host.set_current_details(details);
    }

    /// Force a workflow task failure (EX: in order to retry on non-sticky queue)
    pub fn force_task_fail(&self, with: impl Into<Box<dyn std::error::Error + Send + Sync>>) {
        self.base.inner.runtime.set_forced_wft_failure(with.into());
    }

    /// Start a nexus operation
    pub fn start_nexus_operation(
        &self,
        opts: NexusOperationOptions,
    ) -> impl CancellableFuture<NexusStartResult> {
        self.base.start_nexus_operation(opts)
    }

    /// Create a read-only view of this context.
    pub(crate) fn view(&self) -> WorkflowContextView {
        self.base.view()
    }
}

impl<W> WorkflowContext<W> {
    /// Create a new wf context from a base context and workflow state.
    pub(crate) fn from_base(base: BaseWorkflowContext, workflow_state: Rc<RefCell<W>>) -> Self {
        Self {
            sync: SyncWorkflowContext {
                base,
                headers: Rc::new(HashMap::new()),
                _phantom: PhantomData,
            },
            workflow_state,
            condition_wakers: Rc::new(RefCell::new(Vec::new())),
        }
    }

    /// Returns a new context with the specified headers set.
    pub(crate) fn with_headers(&self, headers: HashMap<String, Payload>) -> Self {
        Self {
            sync: SyncWorkflowContext {
                base: self.sync.base.clone(),
                headers: Rc::new(headers),
                _phantom: PhantomData,
            },
            workflow_state: self.workflow_state.clone(),
            condition_wakers: self.condition_wakers.clone(),
        }
    }

    /// Returns a [`SyncWorkflowContext`] extracted from this context.
    pub(crate) fn sync_context(&self) -> SyncWorkflowContext<W> {
        self.sync.clone()
    }

    /// Create a read-only view of this context.
    pub(crate) fn view(&self) -> WorkflowContextView {
        self.sync.view()
    }

    // --- Delegated methods from SyncWorkflowContext ---

    /// Return the workflow's unique identifier
    pub fn workflow_id(&self) -> &str {
        self.sync.workflow_id()
    }

    /// Return the run id of this workflow execution
    pub fn run_id(&self) -> &str {
        self.sync.run_id()
    }

    /// Return the namespace the workflow is executing in
    pub fn namespace(&self) -> &str {
        self.sync.namespace()
    }

    /// Return the task queue the workflow is executing in
    pub fn task_queue(&self) -> &str {
        self.sync.task_queue()
    }

    /// Return the current time according to the workflow (which is not wall-clock time).
    pub fn workflow_time(&self) -> Option<SystemTime> {
        self.sync.workflow_time()
    }

    /// Return the length of history so far at this point in the workflow
    pub fn history_length(&self) -> u32 {
        self.sync.history_length()
    }

    /// Return the deployment version, if any, as it was when this point in the workflow was first
    /// reached. If this code is being executed for the first time, return this Worker's deployment
    /// version if it has one.
    pub fn current_deployment_version(&self) -> Option<WorkerDeploymentVersion> {
        self.sync.current_deployment_version()
    }

    /// Return current values for workflow search attributes.
    pub fn search_attributes(&self) -> SearchAttributes {
        self.sync.search_attributes()
    }

    /// Return the current workflow memo values.
    pub fn memo(&self) -> Memo {
        self.sync.memo()
    }

    /// Generates a deterministic pseudo-random value of type `T`.
    ///
    /// See [`SyncWorkflowContext::random`].
    pub fn random<T>(&self) -> T
    where
        T: WorkflowRandomValue,
    {
        self.sync.random()
    }

    /// Generates a deterministic lowercase, hyphenated version 4 UUID string.
    ///
    /// See [`SyncWorkflowContext::uuid4`].
    pub fn uuid4(&self) -> String {
        self.sync.uuid4()
    }

    /// Returns true if the current workflow task is happening under replay
    pub fn is_replaying(&self) -> bool {
        self.sync.is_replaying()
    }

    /// Returns true if the current work is replaying history events
    pub fn is_replaying_history_events(&self) -> bool {
        self.sync.is_replaying_history_events()
    }

    /// Returns true if the server suggests this workflow should continue-as-new
    pub fn continue_as_new_suggested(&self) -> bool {
        self.sync.continue_as_new_suggested()
    }

    /// Returns true if the workflow's target worker deployment version changed.
    ///
    /// This experimental signal is intended for workers using worker deployment versioning.
    pub fn target_worker_deployment_version_changed(&self) -> bool {
        self.sync.target_worker_deployment_version_changed()
    }

    /// Returns the headers for the current handler invocation (signal, update, query, etc.).
    pub fn headers(&self) -> &HashMap<String, Payload> {
        self.sync.headers()
    }

    /// Returns the [PayloadConverter] currently used by the worker running this workflow.
    pub fn payload_converter(&self) -> &PayloadConverter {
        self.sync.payload_converter()
    }

    /// Return Rust-native information about this workflow execution.
    pub fn info(&self) -> WorkflowContextView {
        self.sync.info()
    }

    /// A future that resolves if/when the workflow is cancelled, with the user provided cause
    pub fn cancelled(&self) -> impl FusedFuture<Output = String> + '_ {
        self.sync.cancelled()
    }

    /// Request to create a timer
    pub fn timer<T: Into<TimerOptions>>(&self, opts: T) -> impl CancellableFuture<TimerResult> {
        self.sync.timer(opts)
    }

    /// Request to run an activity
    pub fn execute_activity<AD: ActivityDefinition>(
        &self,
        activity: AD,
        input: impl Into<AD::Input>,
        opts: ActivityOptions,
    ) -> impl CancellableFuture<Result<AD::Output, ActivityExecutionError>>
    where
        AD::Output: TemporalDeserializable,
    {
        self.sync.execute_activity(activity, input, opts)
    }

    /// Request to run an activity
    ///
    /// Deprecated alias for [`WorkflowContext::execute_activity`].
    #[deprecated(note = "use `execute_activity` instead")]
    pub fn start_activity<AD: ActivityDefinition>(
        &self,
        activity: AD,
        input: impl Into<AD::Input>,
        opts: ActivityOptions,
    ) -> impl CancellableFuture<Result<AD::Output, ActivityExecutionError>>
    where
        AD::Output: TemporalDeserializable,
    {
        self.execute_activity(activity, input, opts)
    }

    /// Request to run a local activity
    pub fn execute_local_activity<AD: ActivityDefinition>(
        &self,
        activity: AD,
        input: impl Into<AD::Input>,
        opts: LocalActivityOptions,
    ) -> impl CancellableFuture<Result<AD::Output, ActivityExecutionError>>
    where
        AD::Output: TemporalDeserializable,
    {
        self.sync.execute_local_activity(activity, input, opts)
    }

    /// Request to run a local activity
    ///
    /// Deprecated alias for [`WorkflowContext::execute_local_activity`].
    #[deprecated(note = "use `execute_local_activity` instead")]
    pub fn start_local_activity<AD: ActivityDefinition>(
        &self,
        activity: AD,
        input: impl Into<AD::Input>,
        opts: LocalActivityOptions,
    ) -> impl CancellableFuture<Result<AD::Output, ActivityExecutionError>>
    where
        AD::Output: TemporalDeserializable,
    {
        self.execute_local_activity(activity, input, opts)
    }

    /// Start a child workflow. See [SyncWorkflowContext::start_child_workflow] for details.
    pub fn start_child_workflow<WD: WorkflowDefinition + 'static>(
        &self,
        workflow: WD,
        input: impl Into<WD::Input>,
        opts: ChildWorkflowOptions,
    ) -> impl CancellableFutureWithReason<Result<StartedChildWorkflow<WD>, ChildWorkflowStartError>>
    where
        WD::Output: TemporalDeserializable,
    {
        self.sync.start_child_workflow(workflow, input, opts)
    }

    /// Deprecated alias for [`WorkflowContext::start_child_workflow`].
    #[deprecated(note = "use `start_child_workflow` instead")]
    pub fn child_workflow<WD: WorkflowDefinition + 'static>(
        &self,
        workflow: WD,
        input: impl Into<WD::Input>,
        opts: ChildWorkflowOptions,
    ) -> impl CancellableFutureWithReason<Result<StartedChildWorkflow<WD>, ChildWorkflowStartError>>
    where
        WD::Output: TemporalDeserializable,
    {
        self.start_child_workflow(workflow, input, opts)
    }

    /// Check (or record) that this workflow history was created with the provided patch
    pub fn patched(&self, patch_id: &str) -> bool {
        self.sync.patched(patch_id)
    }

    /// Record that this workflow history was created with the provided patch, and it is being
    /// phased out.
    pub fn deprecate_patch(&self, patch_id: &str) -> bool {
        self.sync.deprecate_patch(patch_id)
    }

    /// Get a handle to an external workflow. See [SyncWorkflowContext::external_workflow].
    pub fn external_workflow(
        &self,
        workflow_id: impl Into<String>,
        run_id: Option<String>,
    ) -> ExternalWorkflowHandle {
        self.sync.external_workflow(workflow_id, run_id)
    }

    /// Add, update, or remove search attributes using typed keys.
    pub fn upsert_search_attributes(
        &self,
        updates: impl IntoIterator<Item = SearchAttributeUpdate>,
    ) {
        self.sync.upsert_search_attributes(updates)
    }

    /// Add or replace memo values with `Some`; remove memo keys with `None`.
    pub fn upsert_memo<K>(
        &self,
        updates: impl IntoIterator<Item = (K, Option<MemoValue>)>,
    ) -> Result<(), PayloadConversionError>
    where
        K: Into<String>,
    {
        self.sync.upsert_memo(updates)
    }

    /// Set the current details string for this workflow execution.
    ///
    /// See [`SyncWorkflowContext::set_current_details`].
    pub fn set_current_details(&self, details: impl Into<String>) {
        self.sync.set_current_details(details)
    }

    /// Force a workflow task failure (EX: in order to retry on non-sticky queue)
    pub fn force_task_fail(&self, with: impl Into<Box<dyn std::error::Error + Send + Sync>>) {
        self.sync.force_task_fail(with)
    }

    /// Start a nexus operation
    pub fn start_nexus_operation(
        &self,
        opts: NexusOperationOptions,
    ) -> impl CancellableFuture<NexusStartResult> {
        self.sync.start_nexus_operation(opts)
    }

    /// Access workflow state immutably via closure.
    ///
    /// The borrow is scoped to the closure and cannot escape, preventing
    /// borrows from being held across await points.
    pub fn state<R>(&self, f: impl FnOnce(&W) -> R) -> R {
        f(&*self.workflow_state.borrow())
    }

    /// Access workflow state mutably via closure.
    ///
    /// The borrow is scoped to the closure and cannot escape, preventing
    /// borrows from being held across await points.
    ///
    /// After the mutation, all wakers registered by pending `wait_condition`
    /// futures are woken so that waker-based combinators (e.g.
    /// `FuturesOrdered`) re-poll them on the next pass.
    pub fn state_mut<R>(&self, f: impl FnOnce(&mut W) -> R) -> R {
        let result = f(&mut *self.workflow_state.borrow_mut());
        let _guard = SdkWakeGuard::new();
        for waker in self.condition_wakers.borrow_mut().drain(..) {
            waker.wake();
        }
        self.sync.base.set_state_mutated();
        result
    }

    /// Signal that this workflow should continue as a new workflow execution with the given input and
    /// options.
    ///
    /// This always returns an `Err` which should be propigated
    pub fn continue_as_new(
        &self,
        input: <W::Run as WorkflowDefinition>::Input,
        opts: ContinueAsNewOptions,
    ) -> Result<std::convert::Infallible, WorkflowTermination>
    where
        W: WorkflowImplementation,
    {
        self.sync.continue_as_new(input, opts)
    }

    /// Wait for some condition on workflow state to become true, yielding the workflow if not.
    ///
    /// The condition closure receives an immutable reference to the workflow state,
    /// which is borrowed only for the duration of each poll (not across await points).
    pub fn wait_condition<'a>(
        &'a self,
        mut condition: impl FnMut(&W) -> bool + 'a,
    ) -> impl FusedFuture<Output = ()> + 'a {
        future::poll_fn(move |cx: &mut Context<'_>| {
            if condition(&*self.workflow_state.borrow()) {
                Poll::Ready(())
            } else {
                self.condition_wakers.borrow_mut().push(cx.waker().clone());
                Poll::Pending
            }
        })
        .fuse()
    }
}

struct WfCtxProtectedDat {
    next_timer_sequence_number: u32,
    next_activity_sequence_number: u32,
    next_child_workflow_sequence_number: u32,
    next_cancel_external_wf_sequence_number: u32,
    next_signal_external_wf_sequence_number: u32,
    next_nexus_op_sequence_number: u32,
}

impl WfCtxProtectedDat {
    fn next_timer_seq(&mut self) -> u32 {
        let seq = self.next_timer_sequence_number;
        self.next_timer_sequence_number += 1;
        seq
    }
    fn next_activity_seq(&mut self) -> u32 {
        let seq = self.next_activity_sequence_number;
        self.next_activity_sequence_number += 1;
        seq
    }
    fn next_child_workflow_seq(&mut self) -> u32 {
        let seq = self.next_child_workflow_sequence_number;
        self.next_child_workflow_sequence_number += 1;
        seq
    }
    fn next_cancel_external_wf_seq(&mut self) -> u32 {
        let seq = self.next_cancel_external_wf_sequence_number;
        self.next_cancel_external_wf_sequence_number += 1;
        seq
    }
    fn next_signal_external_wf_seq(&mut self) -> u32 {
        let seq = self.next_signal_external_wf_sequence_number;
        self.next_signal_external_wf_sequence_number += 1;
        seq
    }
    fn next_nexus_op_seq(&mut self) -> u32 {
        let seq = self.next_nexus_op_sequence_number;
        self.next_nexus_op_sequence_number += 1;
        seq
    }
}

#[derive(Clone, Debug)]
struct WorkflowContextSharedData {
    /// Maps change ids -> resolved status
    changes: HashMap<String, bool>,
    /// Kept separate from memoized decisions so replay still emits the matching patch command.
    notified_patches: HashSet<String>,
    activation: CoreWorkflowActivation,
    memo: ProtoMemo,
    is_replaying_history_events: bool,
    search_attributes: ProtoSearchAttributes,
    random: Pcg64Mcg,
    /// Current details string, surfaced via the workflow metadata query.
    current_details: String,
}

/// A Future that can be cancelled.
/// Used in the prototype SDK for cancelling operations like timers and activities.
pub trait CancellableFuture<T>: Future<Output = T> + FusedFuture {
    /// Cancel this Future
    fn cancel(&self);
}

/// A Future that can be cancelled with a reason
pub trait CancellableFutureWithReason<T>: CancellableFuture<T> {
    /// Cancel this Future with a reason
    fn cancel_with_reason(&self, reason: String);
}

fn cancellable_outbound<T: 'static>(
    future: impl CancellableFuture<T> + 'static,
) -> CancellableWorkflowOutboundFuture<T> {
    let future = Rc::new(RefCell::new(Box::pin(future)));
    let polled = future.clone();
    let cancellation = WorkflowCancellationHandle::new(move |_| {
        future.borrow().as_ref().get_ref().cancel();
    });
    CancellableWorkflowOutboundFuture::new(
        future::poll_fn(move |cx| polled.borrow_mut().as_mut().poll(cx)),
        cancellation,
    )
}

fn cancellable_outbound_with_reason<T: 'static>(
    future: impl CancellableFutureWithReason<T> + 'static,
) -> CancellableWorkflowOutboundFuture<T> {
    let future = Rc::new(RefCell::new(Box::pin(future)));
    let polled = future.clone();
    let cancellation = WorkflowCancellationHandle::new(move |reason| {
        let future = future.borrow();
        let future = future.as_ref().get_ref();
        if let Some(reason) = reason {
            future.cancel_with_reason(reason);
        } else {
            future.cancel();
        }
    });
    CancellableWorkflowOutboundFuture::new(
        future::poll_fn(move |cx| polled.borrow_mut().as_mut().poll(cx)),
        cancellation,
    )
}

pub(crate) struct WFCommandFut<T, D> {
    _unused: PhantomData<T>,
    result_rx: oneshot::Receiver<UnblockEvent>,
    other_dat: Option<D>,
}
impl<T> WFCommandFut<T, ()> {
    fn new() -> (Self, oneshot::Sender<UnblockEvent>) {
        Self::new_with_dat(())
    }
}

impl<T, D> WFCommandFut<T, D> {
    fn new_with_dat(other_dat: D) -> (Self, oneshot::Sender<UnblockEvent>) {
        let (tx, rx) = oneshot::channel();
        (
            Self {
                _unused: PhantomData,
                result_rx: rx,
                other_dat: Some(other_dat),
            },
            tx,
        )
    }
}

impl<T, D> Unpin for WFCommandFut<T, D> where T: Unblockable<OtherDat = D> {}
impl<T, D> Future for WFCommandFut<T, D>
where
    T: Unblockable<OtherDat = D>,
{
    type Output = T;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let poll = self.result_rx.poll_unpin(cx).map(|x| {
            let od = self
                .other_dat
                .take()
                .expect("Other data must exist when resolving command future");
            Unblockable::unblock(x.unwrap(), od)
        });
        if poll.is_pending() {
            mark_intercepted_future_activation();
        }
        poll
    }
}
impl<T, D> FusedFuture for WFCommandFut<T, D>
where
    T: Unblockable<OtherDat = D>,
{
    fn is_terminated(&self) -> bool {
        self.other_dat.is_none()
    }
}

struct CancellableWFCommandFut<T, D> {
    cmd_fut: WFCommandFut<T, D>,
    cancellable_id: CancellableID,
    base_ctx: BaseWorkflowContext,
}
impl<T> CancellableWFCommandFut<T, ()> {
    fn new(
        cancellable_id: CancellableID,
        base_ctx: BaseWorkflowContext,
    ) -> (Self, oneshot::Sender<UnblockEvent>) {
        Self::new_with_dat(cancellable_id, (), base_ctx)
    }
}
impl<T, D> CancellableWFCommandFut<T, D> {
    fn new_with_dat(
        cancellable_id: CancellableID,
        other_dat: D,
        base_ctx: BaseWorkflowContext,
    ) -> (Self, oneshot::Sender<UnblockEvent>) {
        let (cmd_fut, sender) = WFCommandFut::new_with_dat(other_dat);
        (
            Self {
                cmd_fut,
                cancellable_id,
                base_ctx,
            },
            sender,
        )
    }
}
impl<T, D> Unpin for CancellableWFCommandFut<T, D> where T: Unblockable<OtherDat = D> {}
impl<T, D> Future for CancellableWFCommandFut<T, D>
where
    T: Unblockable<OtherDat = D>,
{
    type Output = T;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        self.cmd_fut.poll_unpin(cx)
    }
}
impl<T, D> FusedFuture for CancellableWFCommandFut<T, D>
where
    T: Unblockable<OtherDat = D>,
{
    fn is_terminated(&self) -> bool {
        self.cmd_fut.is_terminated()
    }
}

impl<T, D> CancellableFuture<T> for CancellableWFCommandFut<T, D>
where
    T: Unblockable<OtherDat = D>,
{
    fn cancel(&self) {
        self.base_ctx.cancel(self.cancellable_id.clone());
    }
}
impl<T, D> CancellableFutureWithReason<T> for CancellableWFCommandFut<T, D>
where
    T: Unblockable<OtherDat = D>,
{
    fn cancel_with_reason(&self, reason: String) {
        self.base_ctx
            .cancel(self.cancellable_id.clone().with_reason(reason));
    }
}

struct LATimerBackoffFut {
    la_opts: LocalActivityOptions,
    activity_type: String,
    arguments: Vec<Payload>,
    headers: HashMap<String, Payload>,
    current_fut: Pin<Box<dyn CancellableFuture<ActivityResolution> + Unpin>>,
    timer_fut: Option<Pin<Box<dyn CancellableFuture<TimerResult> + Unpin>>>,
    base_ctx: BaseWorkflowContext,
    next_attempt: u32,
    next_sched_time: Option<prost_types::Timestamp>,
    did_cancel: AtomicBool,
    terminated: bool,
}
impl LATimerBackoffFut {
    fn new(
        activity_type: String,
        arguments: Vec<Payload>,
        headers: HashMap<String, Payload>,
        opts: LocalActivityOptions,
        base_ctx: BaseWorkflowContext,
    ) -> Self {
        let current_fut = Box::pin(base_ctx.clone().local_activity_no_timer_retry(
            activity_type.clone(),
            arguments.clone(),
            headers.clone(),
            opts.clone(),
        ));
        Self {
            la_opts: opts,
            activity_type,
            arguments,
            headers,
            current_fut,
            timer_fut: None,
            base_ctx,
            next_attempt: 1,
            next_sched_time: None,
            did_cancel: AtomicBool::new(false),
            terminated: false,
        }
    }
}
impl Unpin for LATimerBackoffFut {}
impl Future for LATimerBackoffFut {
    type Output = ActivityResolution;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        // If the timer exists, wait for it first
        if let Some(tf) = self.timer_fut.as_mut() {
            return match tf.poll_unpin(cx) {
                Poll::Ready(tr) => {
                    self.timer_fut = None;
                    // Schedule next LA if this timer wasn't cancelled
                    if let TimerResult::Fired = tr {
                        let mut opts = self.la_opts.clone();
                        opts.attempt = Some(self.next_attempt);
                        opts.original_schedule_time
                            .clone_from(&self.next_sched_time);
                        self.current_fut =
                            Box::pin(self.base_ctx.clone().local_activity_no_timer_retry(
                                self.activity_type.clone(),
                                self.arguments.clone(),
                                self.headers.clone(),
                                opts,
                            ));
                        Poll::Pending
                    } else {
                        self.terminated = true;
                        Poll::Ready(ActivityResolution {
                            status: Some(activity_resolution::Status::Cancelled(Cancellation {
                                failure: Some(Failure {
                                    message: "Activity cancelled".to_owned(),
                                    failure_info: Some(FailureInfo::CanceledFailureInfo(
                                        CanceledFailureInfo::default(),
                                    )),
                                    ..Default::default()
                                }),
                            })),
                        })
                    }
                }
                Poll::Pending => Poll::Pending,
            };
        }
        let poll_res = self.current_fut.poll_unpin(cx);
        if let Poll::Ready(ref r) = poll_res
            && let Some(activity_resolution::Status::Backoff(b)) = r.status.as_ref()
        {
            // If we've already said we want to cancel, don't schedule the backoff timer. Just
            // return cancel status. This can happen if cancel comes after the LA says it wants
            // to back off but before we have scheduled the timer.
            if self.did_cancel.load(Ordering::Acquire) {
                self.terminated = true;
                return Poll::Ready(ActivityResolution {
                    status: Some(activity_resolution::Status::Cancelled(Cancellation {
                        failure: Some(Failure {
                            message: "Activity cancelled".to_owned(),
                            failure_info: Some(FailureInfo::CanceledFailureInfo(
                                CanceledFailureInfo::default(),
                            )),
                            ..Default::default()
                        }),
                    })),
                });
            }

            let timer_f = self.base_ctx.timer::<Duration>(
                b.backoff_duration
                    .expect("Duration is set")
                    .try_into()
                    .expect("duration converts ok"),
            );
            self.timer_fut = Some(Box::pin(timer_f));
            self.next_attempt = b.attempt;
            self.next_sched_time.clone_from(&b.original_schedule_time);
            return Poll::Pending;
        }
        if poll_res.is_ready() {
            self.terminated = true;
        }
        poll_res
    }
}
impl FusedFuture for LATimerBackoffFut {
    fn is_terminated(&self) -> bool {
        self.terminated
    }
}
impl CancellableFuture<ActivityResolution> for LATimerBackoffFut {
    fn cancel(&self) {
        self.did_cancel.store(true, Ordering::Release);
        if let Some(tf) = self.timer_fut.as_ref() {
            tf.cancel();
        }
        self.current_fut.cancel();
    }
}

/// Future for activity results. Either an immediate error or a running activity.
enum ActivityFut<F, Output> {
    /// Immediate error (e.g., input serialization failure). Resolves on first poll.
    Errored {
        error: Option<Box<ActivityExecutionError>>,
        _phantom: PhantomData<Output>,
    },
    /// Running activity that will deserialize output on completion.
    Running {
        inner: F,
        data_converter: DataConverter,
        _phantom: PhantomData<Output>,
    },
    Terminated,
}

impl<F, Output> ActivityFut<F, Output> {
    fn eager(err: ActivityExecutionError) -> Self {
        Self::Errored {
            error: Some(Box::new(err)),
            _phantom: PhantomData,
        }
    }

    fn running(inner: F, data_converter: DataConverter) -> Self {
        Self::Running {
            inner,
            data_converter,
            _phantom: PhantomData,
        }
    }
}

impl<F, Output> Unpin for ActivityFut<F, Output> where F: Unpin {}

impl<F, Output> Future for ActivityFut<F, Output>
where
    F: Future<Output = ActivityResolution> + Unpin,
    Output: TemporalDeserializable + 'static,
{
    type Output = Result<Output, ActivityExecutionError>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.get_mut();
        let poll = match this {
            ActivityFut::Errored { error, .. } => {
                Poll::Ready(Err(*error.take().expect("polled after completion")))
            }
            ActivityFut::Running {
                inner,
                data_converter,
                ..
            } => match Pin::new(inner).poll(cx) {
                Poll::Pending => Poll::Pending,
                Poll::Ready(resolution) => Poll::Ready({
                    let status = resolution.status.ok_or_else(|| {
                        data_converter
                            .to_error(
                                &SerializationContextData::Workflow,
                                Failure {
                                    message: "Activity completed without a status".to_string(),
                                    ..Default::default()
                                },
                                ActivityExecutionDecodeHint { cancelled: false },
                            )
                            .expect("synthetic activity failure should decode")
                    })?;

                    match status {
                        activity_resolution::Status::Completed(success) => {
                            let payload = success.result.unwrap_or_default();
                            let ctx = SerializationContext {
                                data: &SerializationContextData::Workflow,
                                converter: data_converter.payload_converter(),
                            };
                            data_converter
                                .payload_converter()
                                .from_payload::<Output>(&ctx, payload)
                                .map_err(ActivityExecutionError::Serialization)
                        }
                        activity_resolution::Status::Failed(f) => Err(data_converter.to_error(
                            &SerializationContextData::Workflow,
                            f.failure.unwrap_or_default(),
                            ActivityExecutionDecodeHint { cancelled: false },
                        )?),
                        activity_resolution::Status::Cancelled(c) => Err(data_converter.to_error(
                            &SerializationContextData::Workflow,
                            c.failure.unwrap_or_default(),
                            ActivityExecutionDecodeHint { cancelled: true },
                        )?),
                        activity_resolution::Status::Backoff(_) => {
                            panic!("DoBackoff should be handled by LATimerBackoffFut")
                        }
                    }
                }),
            },
            ActivityFut::Terminated => panic!("polled after termination"),
        };
        if poll.is_ready() {
            *this = ActivityFut::Terminated;
        }
        poll
    }
}

impl<F, Output> FusedFuture for ActivityFut<F, Output>
where
    F: Future<Output = ActivityResolution> + Unpin,
    Output: TemporalDeserializable + 'static,
{
    fn is_terminated(&self) -> bool {
        matches!(self, ActivityFut::Terminated)
    }
}

impl<F, Output> CancellableFuture<Result<Output, ActivityExecutionError>> for ActivityFut<F, Output>
where
    F: CancellableFuture<ActivityResolution> + Unpin,
    Output: TemporalDeserializable + 'static,
{
    fn cancel(&self) {
        if let ActivityFut::Running { inner, .. } = self {
            inner.cancel()
        }
    }
}

pub(crate) struct ChildWfCommon {
    workflow_id: String,
    child_seq: u32,
    result_future: CancellableWFCommandFut<ChildWorkflowResult, ()>,
    base_ctx: BaseWorkflowContext,
    data_converter: DataConverter,
}

/// Child workflow in pending state. Internal type used during the start handshake;
/// `ChildWorkflowStartFut` converts this into `Result<StartedChildWorkflow, _>` before
/// the caller sees it.
#[derive(derive_more::Debug)]
pub(crate) struct PendingChildWorkflow<WD: WorkflowDefinition> {
    pub(crate) status: ChildWorkflowStartStatus,
    #[debug(skip)]
    pub(crate) common: ChildWfCommon,
    pub(crate) _phantom: PhantomData<WD>,
}

/// Output produced when an intercepted child workflow successfully starts.
#[derive(derive_more::Debug)]
pub struct StartChildWorkflowOutput {
    /// Run ID of the child workflow
    pub run_id: String,
    #[debug(skip)]
    result_future: CancellableWorkflowOutboundFuture<ChildWorkflowOutboundResult>,
    workflow_id: String,
    child_seq: u32,
    #[debug(skip)]
    base_ctx: BaseWorkflowContext,
}

impl StartChildWorkflowOutput {
    /// Replace the intercepted child completion future.
    pub fn map_result(
        mut self,
        map: impl FnOnce(
            CancellableWorkflowOutboundFuture<ChildWorkflowOutboundResult>,
        ) -> CancellableWorkflowOutboundFuture<ChildWorkflowOutboundResult>,
    ) -> Self {
        self.result_future = map(self.result_future);
        self
    }

    fn into_started<WD: WorkflowDefinition>(self) -> StartedChildWorkflow<WD> {
        StartedChildWorkflow {
            run_id: self.run_id,
            result_future: self.result_future,
            workflow_id: self.workflow_id,
            child_seq: self.child_seq,
            base_ctx: self.base_ctx,
            _phantom: PhantomData,
        }
    }
}

/// Child workflow in started state.
#[derive(derive_more::Debug)]
pub struct StartedChildWorkflow<WD: WorkflowDefinition> {
    /// Run ID of the child workflow
    pub run_id: String,
    #[debug(skip)]
    result_future: CancellableWorkflowOutboundFuture<ChildWorkflowOutboundResult>,
    workflow_id: String,
    child_seq: u32,
    #[debug(skip)]
    base_ctx: BaseWorkflowContext,
    _phantom: PhantomData<WD>,
}

/// Future for child workflow results. Wraps the raw result future and deserializes
/// the output on completion.
enum ChildWorkflowFut<F, Output> {
    Running {
        inner: F,
        data_converter: DataConverter,
        _phantom: PhantomData<Output>,
    },
    Terminated,
}

impl<F, Output> Unpin for ChildWorkflowFut<F, Output> where F: Unpin {}

impl<F, Output> Future for ChildWorkflowFut<F, Output>
where
    F: Future<Output = ChildWorkflowResult> + Unpin,
    Output: TemporalDeserializable + 'static,
{
    type Output = Result<Output, ChildWorkflowExecutionError>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.get_mut();
        let poll = match this {
            ChildWorkflowFut::Running {
                inner,
                data_converter,
                ..
            } => match Pin::new(inner).poll(cx) {
                Poll::Pending => Poll::Pending,
                Poll::Ready(result) => Poll::Ready({
                    let status = result.status.ok_or_else(|| {
                        data_converter
                            .to_error(
                                &SerializationContextData::Workflow,
                                Failure {
                                    message: "Child workflow completed without a status"
                                        .to_string(),
                                    ..Default::default()
                                },
                                ChildWorkflowExecutionDecodeHint,
                            )
                            .expect("synthetic child workflow failure should decode")
                    })?;
                    match status {
                        child_workflow_result::Status::Completed(success) => {
                            let payloads = success.result.into_iter().collect();
                            let ctx = SerializationContext {
                                data: &SerializationContextData::Workflow,
                                converter: data_converter.payload_converter(),
                            };
                            data_converter
                                .payload_converter()
                                .from_payloads::<Output>(&ctx, payloads)
                                .map_err(ChildWorkflowExecutionError::Serialization)
                        }
                        child_workflow_result::Status::Failed(f) => Err(data_converter.to_error(
                            &SerializationContextData::Workflow,
                            f.failure.unwrap_or_default(),
                            ChildWorkflowExecutionDecodeHint,
                        )?),
                        child_workflow_result::Status::Cancelled(c) => Err(data_converter
                            .to_error(
                                &SerializationContextData::Workflow,
                                c.failure.unwrap_or_default(),
                                ChildWorkflowExecutionDecodeHint,
                            )?),
                    }
                }),
            },
            ChildWorkflowFut::Terminated => panic!("polled after termination"),
        };
        if poll.is_ready() {
            *this = ChildWorkflowFut::Terminated;
        }
        poll
    }
}

impl<F, Output> FusedFuture for ChildWorkflowFut<F, Output>
where
    F: Future<Output = ChildWorkflowResult> + Unpin,
    Output: TemporalDeserializable + 'static,
{
    fn is_terminated(&self) -> bool {
        matches!(self, ChildWorkflowFut::Terminated)
    }
}

impl<F, Output> CancellableFutureWithReason<Result<Output, ChildWorkflowExecutionError>>
    for ChildWorkflowFut<F, Output>
where
    F: CancellableFutureWithReason<ChildWorkflowResult> + Unpin,
    Output: TemporalDeserializable + 'static,
{
    fn cancel_with_reason(&self, reason: String) {
        if let ChildWorkflowFut::Running { inner, .. } = self {
            inner.cancel_with_reason(reason)
        }
    }
}

impl<F, Output> CancellableFuture<Result<Output, ChildWorkflowExecutionError>>
    for ChildWorkflowFut<F, Output>
where
    F: CancellableFutureWithReason<ChildWorkflowResult> + Unpin,
    Output: TemporalDeserializable + 'static,
{
    fn cancel(&self) {
        if let ChildWorkflowFut::Running { inner, .. } = self {
            inner.cancel()
        }
    }
}

/// Wrapper future for starting a child workflow. Mirrors `ActivityFut` to allow returning
/// serialization errors eagerly.
enum ChildWorkflowStartFut<F, WD: WorkflowDefinition> {
    /// Immediate error (e.g., input serialization failure). Resolves on first poll.
    Errored {
        error: Option<Box<ChildWorkflowStartError>>,
        _phantom: PhantomData<WD>,
    },
    Running(F),
    Terminated,
}

impl<F, WD: WorkflowDefinition> ChildWorkflowStartFut<F, WD> {
    fn eager(err: ChildWorkflowStartError) -> Self {
        Self::Errored {
            error: Some(Box::new(err)),
            _phantom: PhantomData,
        }
    }
}

impl<F, WD: WorkflowDefinition> Unpin for ChildWorkflowStartFut<F, WD> where F: Unpin {}

impl<F, WD> Future for ChildWorkflowStartFut<F, WD>
where
    F: Future<Output = PendingChildWorkflow<WD>> + Unpin,
    WD: WorkflowDefinition,
{
    type Output = StartChildWorkflowResult;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.get_mut();
        let poll = match this {
            ChildWorkflowStartFut::Errored { error, .. } => {
                Poll::Ready(Err(*error.take().expect("polled after completion")))
            }
            ChildWorkflowStartFut::Running(inner) => match Pin::new(inner).poll(cx) {
                Poll::Pending => Poll::Pending,
                Poll::Ready(pending) => Poll::Ready(match pending.status {
                    ChildWorkflowStartStatus::Succeeded(s) => {
                        let ChildWfCommon {
                            workflow_id,
                            child_seq,
                            result_future,
                            base_ctx,
                            data_converter,
                        } = pending.common;
                        let result_future = cancellable_outbound_with_reason(ChildWorkflowFut::<
                            _,
                            WD::Output,
                        >::Running {
                            inner: result_future,
                            data_converter,
                            _phantom: PhantomData,
                        })
                        .map(|result| {
                            result.map(|output| Box::new(output) as Box<dyn WorkflowOutboundValue>)
                        });
                        Ok(StartChildWorkflowOutput {
                            run_id: s.run_id,
                            result_future,
                            workflow_id,
                            child_seq,
                            base_ctx,
                        })
                    }
                    ChildWorkflowStartStatus::Failed(f) => {
                        Err(ChildWorkflowStartError::StartFailed {
                            workflow_id: f.workflow_id,
                            workflow_type: f.workflow_type,
                            cause: StartChildWorkflowExecutionFailedCause::try_from(f.cause)
                                .unwrap_or(StartChildWorkflowExecutionFailedCause::Unspecified),
                        })
                    }
                    ChildWorkflowStartStatus::Cancelled(c) => {
                        Err(pending.common.data_converter.to_error(
                            &SerializationContextData::Workflow,
                            c.failure.unwrap_or_default(),
                            ChildWorkflowStartDecodeHint,
                        )?)
                    }
                }),
            },
            ChildWorkflowStartFut::Terminated => panic!("polled after termination"),
        };
        if poll.is_ready() {
            *this = ChildWorkflowStartFut::Terminated;
        }
        poll
    }
}

impl<F, WD> FusedFuture for ChildWorkflowStartFut<F, WD>
where
    F: Future<Output = PendingChildWorkflow<WD>> + Unpin,
    WD: WorkflowDefinition,
{
    fn is_terminated(&self) -> bool {
        matches!(self, ChildWorkflowStartFut::Terminated)
    }
}

impl<F, WD> CancellableFuture<StartChildWorkflowResult> for ChildWorkflowStartFut<F, WD>
where
    F: CancellableFutureWithReason<PendingChildWorkflow<WD>> + Unpin,
    WD: WorkflowDefinition,
{
    fn cancel(&self) {
        if let ChildWorkflowStartFut::Running(inner) = self {
            inner.cancel()
        }
    }
}

impl<F, WD> CancellableFutureWithReason<StartChildWorkflowResult> for ChildWorkflowStartFut<F, WD>
where
    F: CancellableFutureWithReason<PendingChildWorkflow<WD>> + Unpin,
    WD: WorkflowDefinition,
{
    fn cancel_with_reason(&self, reason: String) {
        if let ChildWorkflowStartFut::Running(inner) = self {
            inner.cancel_with_reason(reason)
        }
    }
}

/// Wrapper future for signaling a child workflow.
enum SignalChildFut<F> {
    Running {
        inner: F,
        data_converter: DataConverter,
    },
    Terminated,
}

impl<F> Unpin for SignalChildFut<F> where F: Unpin {}

impl<F> Future for SignalChildFut<F>
where
    F: Future<Output = SignalExternalWfResult> + Unpin,
{
    type Output = Result<(), WorkflowSignalError>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.get_mut();
        let poll = match this {
            SignalChildFut::Running {
                inner,
                data_converter,
            } => match Pin::new(inner).poll(cx) {
                Poll::Pending => Poll::Pending,
                Poll::Ready(Ok(_)) => Poll::Ready(Ok(())),
                Poll::Ready(Err(failure)) => Poll::Ready(Err(data_converter.to_error(
                    &SerializationContextData::Workflow,
                    failure,
                    WorkflowSignalDecodeHint,
                )?)),
            },
            SignalChildFut::Terminated => panic!("polled after termination"),
        };
        if poll.is_ready() {
            *this = SignalChildFut::Terminated;
        }
        poll
    }
}

impl<F> FusedFuture for SignalChildFut<F>
where
    F: Future<Output = SignalExternalWfResult> + Unpin,
{
    fn is_terminated(&self) -> bool {
        matches!(self, SignalChildFut::Terminated)
    }
}

impl<F> CancellableFuture<Result<(), WorkflowSignalError>> for SignalChildFut<F>
where
    F: CancellableFuture<SignalExternalWfResult> + Unpin,
{
    fn cancel(&self) {
        if let SignalChildFut::Running { inner, .. } = self {
            inner.cancel()
        }
    }
}

impl<WD: WorkflowDefinition> StartedChildWorkflow<WD>
where
    WD::Output: TemporalDeserializable + 'static,
{
    /// Consumes self and returns a future that deserializes the child workflow result
    /// into `WD::Output`.
    pub fn result(
        self,
    ) -> impl CancellableFutureWithReason<Result<WD::Output, ChildWorkflowExecutionError>> {
        self.result_future.map(|result| {
            result.and_then(|output| {
                output
                    .downcast::<WD::Output>()
                    .map(|output| *output)
                    .map_err(|_| {
                        ChildWorkflowExecutionError::Serialization(outbound_type_error(
                            "child workflow output",
                        ))
                    })
            })
        })
    }

    /// Cancel the child workflow
    pub fn cancel(&self, reason: String) {
        self.base_ctx.inner.runtime.host.push_command(
            workflow_command::Variant::CancelChildWorkflowExecution(CancelChildWorkflowExecution {
                child_workflow_seq: self.child_seq,
                reason,
            })
            .into(),
        );
    }

    /// Send a typed signal to the child workflow.
    pub fn signal<S: SignalDefinition<Workflow = WD> + 'static>(
        &self,
        signal: S,
        input: S::Input,
    ) -> impl CancellableFuture<Result<(), WorkflowSignalError>> + 'static {
        self.base_ctx.signal_workflow(
            SignalWorkflowTarget::Child {
                workflow_id: self.workflow_id.clone(),
            },
            signal,
            input,
        )
    }
}

/// Handle to an external workflow for sending signals or requesting cancellation.
///
/// Obtained via [`SyncWorkflowContext::external_workflow`],
/// [`WorkflowContext::external_workflow`], or
/// [`WorkflowInterceptorContext::external_workflow`].
#[derive(derive_more::Debug)]
pub struct ExternalWorkflowHandle {
    workflow_id: String,
    run_id: Option<String>,
    namespace: String,
    #[debug(skip)]
    base_ctx: BaseWorkflowContext,
}

impl ExternalWorkflowHandle {
    /// The workflow ID of the external workflow.
    pub fn workflow_id(&self) -> &str {
        &self.workflow_id
    }

    /// The run ID of the external workflow, or `None` if targeting the latest run.
    pub fn run_id(&self) -> Option<&str> {
        self.run_id.as_deref()
    }

    /// Send a signal to the external workflow.
    pub fn signal<S: SignalDefinition + 'static>(
        &self,
        signal: S,
        input: S::Input,
    ) -> impl CancellableFuture<Result<(), WorkflowSignalError>> + 'static {
        self.base_ctx.signal_workflow(
            SignalWorkflowTarget::External {
                namespace: self.namespace.clone(),
                workflow_id: self.workflow_id.clone(),
                run_id: self.run_id.clone(),
            },
            signal,
            input,
        )
    }

    /// Request cancellation of the external workflow.
    pub fn cancel(
        &self,
        reason: Option<String>,
    ) -> impl FusedFuture<Output = CancelExternalWfResult> {
        self.base_ctx
            .cancel_external_workflow(CancelExternalWorkflowInput {
                workflow_id: self.workflow_id.clone(),
                run_id: self.run_id.clone(),
                reason,
            })
    }
}

#[derive(derive_more::Debug)]
#[debug("StartedNexusOperation{{ operation_token: {operation_token:?} }}")]
/// Handle to a started Nexus operation.
pub struct StartedNexusOperation {
    /// The operation token, if the operation started asynchronously
    pub operation_token: Option<String>,
    #[debug(skip)]
    pub(crate) result_future: Shared<WFCommandFut<NexusOperationResult, ()>>,
    pub(crate) schedule_seq: u32,
    #[debug(skip)]
    pub(crate) base_ctx: BaseWorkflowContext,
}

pub(crate) struct NexusUnblockData {
    pub(crate) result_future: Shared<WFCommandFut<NexusOperationResult, ()>>,
    pub(crate) schedule_seq: u32,
    pub(crate) base_ctx: BaseWorkflowContext,
}

impl StartedNexusOperation {
    /// Wait for the operation result.
    pub async fn result(&self) -> NexusOperationResult {
        // The result future is a `Shared`; poll it inside an `SdkWakeGuard` (via
        // `SdkGuardedFuture`) so its internal waker machinery isn't mistaken for a non-SDK wake on
        // replay (which would fail the workflow task with TMPRL1100).
        SdkGuardedFuture(self.result_future.clone()).await
    }

    /// Request cancellation of the operation.
    pub fn cancel(&self) {
        self.base_ctx
            .cancel(CancellableID::NexusOp(self.schedule_seq));
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::MemoValues;
    use std::{
        collections::HashMap,
        sync::{
            Mutex,
            atomic::{AtomicUsize, Ordering as AtomicOrdering},
        },
        task::Wake,
    };
    use temporalio_common_wasm::{
        RetryPolicy,
        data_converters::{TemporalDeserializable, TemporalSerializable},
        protos::{
            coresdk::{
                AsJsonPayloadExt, FromJsonPayloadExt,
                common::VersioningIntent as ProtoVersioningIntent,
                workflow_activation::{UpdateRandomSeed, WorkflowActivationJob},
                workflow_commands::WorkflowCommand,
            },
            temporal::api::{
                common::v1::{Payload, RetryPolicy as ProtoRetryPolicy},
                enums::v1::ContinueAsNewVersioningBehavior as ProtoContinueAsNewVersioningBehavior,
            },
        },
    };
    use temporalio_macros::{workflow, workflow_methods};

    #[derive(Default)]
    struct NoopHost;

    struct CountingWake(Arc<AtomicUsize>);

    impl Wake for CountingWake {
        fn wake(self: Arc<Self>) {
            self.wake_by_ref();
        }

        fn wake_by_ref(self: &Arc<Self>) {
            self.0.fetch_add(1, AtomicOrdering::Relaxed);
        }
    }

    impl WorkflowHost for NoopHost {
        fn set_current_details(&self, _details: String) {}
        fn push_command(&self, _command: WorkflowCommand) {}
    }

    #[derive(Default)]
    struct RecordingHost {
        commands: Rc<RefCell<Vec<WorkflowCommand>>>,
    }

    impl WorkflowHost for RecordingHost {
        fn set_current_details(&self, _details: String) {}

        fn push_command(&self, command: WorkflowCommand) {
            self.commands.borrow_mut().push(command);
        }
    }

    #[derive(Debug)]
    struct FailingMemoValue;

    impl TemporalSerializable for FailingMemoValue {
        fn to_payload(
            &self,
            _ctx: &temporalio_common_wasm::data_converters::SerializationContext<'_>,
        ) -> Result<Payload, temporalio_common_wasm::data_converters::PayloadConversionError>
        {
            Err(
                temporalio_common_wasm::data_converters::PayloadConversionError::EncodingError(
                    std::io::Error::other("memo serialization failure").into(),
                ),
            )
        }
    }

    #[workflow]
    #[derive(Default)]
    struct TestWorkflow;

    #[workflow_methods]
    impl TestWorkflow {
        #[run]
        async fn run(_ctx: &mut WorkflowContext<Self>, _input: u8) -> crate::WorkflowResult<()> {
            unreachable!("test workflow run should not be polled")
        }

        #[signal]
        fn test_signal(&mut self, _ctx: &mut SyncWorkflowContext<Self>, _input: String) {
            unreachable!("test workflow signal should not be dispatched")
        }
    }

    fn test_context() -> WorkflowContext<TestWorkflow> {
        test_context_with_seed(0)
    }

    fn test_context_with_seed(randomness_seed: u64) -> WorkflowContext<TestWorkflow> {
        let init = InitializeWorkflow {
            workflow_type: TestWorkflow.name().to_string(),
            randomness_seed,
            ..Default::default()
        };
        let init = WorkflowInit {
            namespace: "default".to_string(),
            task_queue: "orig-task-queue".to_string(),
            run_id: "run-id".to_string(),
            initialize_workflow: init,
        };
        let base = BaseWorkflowContext::from_raw(
            init,
            DataConverter::default(),
            Rc::new(NoopHost),
            None,
            Vec::new(),
        );
        WorkflowContext::from_base(base, Rc::new(RefCell::new(TestWorkflow)))
    }

    fn patch_test_context(
        callback: Option<PatchActivationCallback>,
    ) -> (
        BaseWorkflowContext,
        WorkflowContext<TestWorkflow>,
        Rc<RefCell<Vec<WorkflowCommand>>>,
    ) {
        let init = InitializeWorkflow {
            workflow_id: "workflow-id".to_string(),
            workflow_type: TestWorkflow.name().to_string(),
            ..Default::default()
        };
        let host = Rc::new(RecordingHost::default());
        let commands = host.commands.clone();
        let init = WorkflowInit {
            namespace: "default".to_string(),
            task_queue: "task-queue".to_string(),
            run_id: "run-id".to_string(),
            initialize_workflow: init,
        };
        let base = BaseWorkflowContext::from_raw(
            init,
            DataConverter::default(),
            host,
            callback,
            Vec::new(),
        );

        let ctx = WorkflowContext::from_base(base.clone(), Rc::new(RefCell::new(TestWorkflow)));
        (base, ctx, commands)
    }

    struct ShortCircuitFirstTimer {
        calls: AtomicUsize,
    }

    impl WorkflowInterceptor for ShortCircuitFirstTimer {
        fn start_timer(
            &self,
            _ctx: WorkflowInterceptorContext,
            input: StartTimerInput,
            next: WorkflowNext<
                'static,
                StartTimerInput,
                CancellableWorkflowOutboundFuture<TimerResult>,
            >,
        ) -> CancellableWorkflowOutboundFuture<TimerResult> {
            if self.calls.fetch_add(1, Ordering::Relaxed) == 0 {
                CancellableWorkflowOutboundFuture::new(
                    async { TimerResult::Cancelled },
                    WorkflowCancellationHandle::new(|_| {}),
                )
            } else {
                next.run(input)
            }
        }
    }

    #[test]
    fn short_circuited_outbound_call_does_not_consume_sequence_number() {
        let host = Rc::new(RecordingHost::default());
        let init = InitializeWorkflow {
            workflow_type: TestWorkflow.name().to_string(),
            ..Default::default()
        };
        let init = WorkflowInit {
            namespace: "default".to_string(),
            task_queue: "task-queue".to_string(),
            run_id: "run-id".to_string(),
            initialize_workflow: init,
        };
        let base = BaseWorkflowContext::from_raw(
            init,
            DataConverter::default(),
            host.clone(),
            None,
            vec![WorkflowInterceptorConstructor::new(|_| {
                ShortCircuitFirstTimer {
                    calls: AtomicUsize::new(0),
                }
            })],
        );

        let first = base.timer(Duration::from_secs(1));
        assert_eq!(first.now_or_never(), Some(TimerResult::Cancelled));
        let _second = base.timer(Duration::from_secs(1));

        let commands = host.commands.borrow();
        assert_eq!(commands.len(), 1);
        let Some(workflow_command::Variant::StartTimer(timer)) = &commands[0].variant else {
            panic!("expected start timer command");
        };
        assert_eq!(timer.seq, 1);
    }

    #[test]
    fn patch_activation_callback_activates_and_memoizes() {
        let calls = Arc::new(AtomicUsize::new(0));
        let input = Arc::new(Mutex::new(None));
        let callback_calls = calls.clone();
        let callback_input = input.clone();
        let callback: PatchActivationCallback = Arc::new(move |value| {
            callback_calls.fetch_add(1, AtomicOrdering::Relaxed);
            *callback_input.lock().unwrap() = Some(value);
            true
        });
        let (_, ctx, commands) = patch_test_context(Some(callback));

        assert!(ctx.patched("my-patch"));
        assert!(ctx.patched("my-patch"));
        assert_eq!(calls.load(AtomicOrdering::Relaxed), 1);
        assert_eq!(commands.borrow().len(), 1);
        let input = input.lock().unwrap();
        let input = input.as_ref().unwrap();
        assert_eq!(input.workflow_info.workflow_id(), "workflow-id");
        assert_eq!(input.workflow_info.run_id(), "run-id");
        assert_eq!(input.patch_id, "my-patch");
    }

    #[test]
    fn patch_activation_callback_can_decline_and_memoizes() {
        let calls = Arc::new(AtomicUsize::new(0));
        let callback_calls = calls.clone();
        let callback: PatchActivationCallback = Arc::new(move |_| {
            callback_calls.fetch_add(1, AtomicOrdering::Relaxed);
            false
        });
        let (_, ctx, commands) = patch_test_context(Some(callback));

        assert!(!ctx.patched("my-patch"));
        assert!(!ctx.patched("my-patch"));
        assert_eq!(calls.load(AtomicOrdering::Relaxed), 1);
        assert!(commands.borrow().is_empty());
    }

    #[test]
    fn patch_activation_callback_bypasses_history_and_deprecation() {
        let callback: PatchActivationCallback = Arc::new(|_| panic!("callback must not run"));

        let (base, ctx, commands) = patch_test_context(Some(callback.clone()));
        base.apply_activation_context(
            &CoreWorkflowActivation {
                is_replaying: true,
                ..Default::default()
            },
            true,
        );
        assert!(!ctx.patched("replay-patch"));
        assert!(commands.borrow().is_empty());

        let (base, ctx, commands) = patch_test_context(Some(callback.clone()));
        base.apply_activation_context(
            &CoreWorkflowActivation {
                is_replaying: true,
                ..Default::default()
            },
            true,
        );
        base.notify_patch("existing-patch".to_string());
        assert!(ctx.patched("existing-patch"));
        assert_eq!(commands.borrow().len(), 1);

        let (_, ctx, commands) = patch_test_context(Some(callback));
        assert!(ctx.deprecate_patch("deprecated-patch"));
        assert_eq!(commands.borrow().len(), 1);
    }

    #[test]
    fn patch_activation_defaults_to_active() {
        let (_, ctx, commands) = patch_test_context(None);

        assert!(ctx.patched("my-patch"));
        assert_eq!(commands.borrow().len(), 1);
    }

    #[test]
    fn random_is_deterministic_for_supported_numeric_types() {
        let first = test_context_with_seed(42);
        let second = test_context_with_seed(42);

        assert_eq!(first.random::<u8>(), second.random::<u8>());
        assert_eq!(first.random::<i64>(), second.random::<i64>());
        assert_eq!(first.random::<u128>(), second.random::<u128>());
        assert_eq!(first.random::<f32>(), second.random::<f32>());
        assert_eq!(first.random::<f64>(), second.random::<f64>());
        assert_eq!(first.uuid4(), second.uuid4());
    }

    #[test]
    fn random_is_reseeded_by_activation() {
        let ctx = test_context_with_seed(123);
        let expected = ctx.random::<u64>();
        let activation = CoreWorkflowActivation {
            jobs: vec![WorkflowActivationJob {
                variant: Some(ActivationVariant::UpdateRandomSeed(UpdateRandomSeed {
                    randomness_seed: 123,
                })),
            }],
            ..Default::default()
        };

        ctx.sync.base.apply_activation_context(&activation, false);

        assert_eq!(ctx.random::<u64>(), expected);
    }

    struct MutatingRemainingOutboundInterceptor;

    impl WorkflowInterceptor for MutatingRemainingOutboundInterceptor {
        fn signal_workflow(
            &self,
            _ctx: WorkflowInterceptorContext,
            mut input: SignalWorkflowInput,
            next: WorkflowNext<
                'static,
                SignalWorkflowInput,
                CancellableWorkflowOutboundFuture<SignalWorkflowResult>,
            >,
        ) -> CancellableWorkflowOutboundFuture<SignalWorkflowResult> {
            *input.signal_name_mut() = "mutated-signal".to_string();
            *input.input_mut::<String>().unwrap() = "mutated-input".to_string();
            *input.target_mut() = SignalWorkflowTarget::External {
                namespace: "mutated-namespace".to_string(),
                workflow_id: "mutated-workflow".to_string(),
                run_id: Some("mutated-run".to_string()),
            };
            input
                .headers_mut()
                .insert("signal-header".to_string(), Payload::default());
            next.run(input)
        }

        fn cancel_external_workflow(
            &self,
            _ctx: WorkflowInterceptorContext,
            mut input: CancelExternalWorkflowInput,
            next: WorkflowNext<
                'static,
                CancelExternalWorkflowInput,
                WorkflowOutboundFuture<CancelExternalWfResult>,
            >,
        ) -> WorkflowOutboundFuture<CancelExternalWfResult> {
            input.workflow_id = "mutated-cancel-workflow".to_string();
            input.run_id = Some("mutated-cancel-run".to_string());
            input.reason = Some("mutated-reason".to_string());
            next.run(input)
        }

        fn continue_as_new(
            &self,
            _ctx: crate::workflow_interceptors::SyncWorkflowInterceptorContext,
            mut input: ContinueAsNewInput,
            next: WorkflowNext<
                'static,
                ContinueAsNewInput,
                crate::workflow_interceptors::ContinueAsNewResult,
            >,
        ) -> crate::workflow_interceptors::ContinueAsNewResult {
            *input.input_mut::<u8>().unwrap() = 42;
            input.options_mut().workflow_type = Some("mutated-workflow-type".to_string());
            input.headers_mut().insert(
                "continue-header".to_string(),
                Payload::from(b"continue-header-value".as_slice()),
            );
            next.run(input)
        }

        fn start_nexus_operation(
            &self,
            _ctx: WorkflowInterceptorContext,
            mut input: StartNexusOperationInput,
            next: WorkflowNext<
                'static,
                StartNexusOperationInput,
                CancellableWorkflowOutboundFuture<
                    crate::workflow_interceptors::StartNexusOperationResult,
                >,
            >,
        ) -> CancellableWorkflowOutboundFuture<
            crate::workflow_interceptors::StartNexusOperationResult,
        > {
            input.options_mut().endpoint = "mutated-endpoint".to_string();
            input.options_mut().service = "mutated-service".to_string();
            input.options_mut().operation = "mutated-operation".to_string();
            next.run(input)
        }
    }

    #[test]
    fn outbound_interceptors_mutate_signal_cancel_continue_as_new_and_nexus() {
        let host = Rc::new(RecordingHost::default());
        let init = InitializeWorkflow {
            workflow_type: TestWorkflow.name().to_string(),
            ..Default::default()
        };
        let init = WorkflowInit {
            namespace: "default".to_string(),
            task_queue: "task-queue".to_string(),
            run_id: "run-id".to_string(),
            initialize_workflow: init,
        };
        let base = BaseWorkflowContext::from_raw(
            init,
            DataConverter::default(),
            host.clone(),
            None,
            vec![WorkflowInterceptorConstructor::new(|_| {
                MutatingRemainingOutboundInterceptor
            })],
        );
        let ctx = WorkflowContext::from_base(base, Rc::new(RefCell::new(TestWorkflow)));

        let signal = ctx
            .external_workflow("original-workflow", Some("original-run".to_string()))
            .signal(TestWorkflow::test_signal, "original-input".to_string());
        let cancel_target =
            ctx.external_workflow("cancel-workflow", Some("cancel-run".to_string()));
        let cancel = cancel_target.cancel(Some("original-reason".to_string()));
        let termination = ctx
            .continue_as_new(7, ContinueAsNewOptions::default())
            .expect_err("continue_as_new should terminate the workflow");
        let sync_ctx = ctx.sync_context();
        let nexus = sync_ctx.start_nexus_operation(NexusOperationOptions {
            endpoint: "original-endpoint".to_string(),
            service: "original-service".to_string(),
            operation: "original-operation".to_string(),
            ..Default::default()
        });
        drop((signal, cancel, nexus));

        let WorkflowTermination::ContinueAsNew(continue_as_new) = termination else {
            panic!("expected continue-as-new termination")
        };
        assert_eq!(continue_as_new.workflow_type, "mutated-workflow-type");
        assert_eq!(
            continue_as_new.arguments,
            vec![42u8.as_json_payload().unwrap()]
        );
        assert!(continue_as_new.headers.contains_key("continue-header"));

        let commands = host.commands.borrow();
        assert_eq!(commands.len(), 3);
        let Some(workflow_command::Variant::SignalExternalWorkflowExecution(signal)) =
            &commands[0].variant
        else {
            panic!("expected signal command")
        };
        assert_eq!(signal.signal_name, "mutated-signal");
        assert_eq!(
            signal.args,
            vec!["mutated-input".to_string().as_json_payload().unwrap()]
        );
        assert!(signal.headers.contains_key("signal-header"));
        let Some(signal_external_workflow_execution::Target::WorkflowExecution(target)) =
            &signal.target
        else {
            panic!("expected external workflow signal target")
        };
        assert_eq!(target.namespace, "mutated-namespace");
        assert_eq!(target.workflow_id, "mutated-workflow");
        assert_eq!(target.run_id, "mutated-run");

        let Some(workflow_command::Variant::RequestCancelExternalWorkflowExecution(cancel)) =
            &commands[1].variant
        else {
            panic!("expected external cancellation command")
        };
        let target = cancel.workflow_execution.as_ref().unwrap();
        assert_eq!(target.workflow_id, "mutated-cancel-workflow");
        assert_eq!(target.run_id, "mutated-cancel-run");
        assert_eq!(cancel.reason, "mutated-reason");

        let Some(workflow_command::Variant::ScheduleNexusOperation(nexus)) = &commands[2].variant
        else {
            panic!("expected Nexus operation command")
        };
        assert_eq!(nexus.endpoint, "mutated-endpoint");
        assert_eq!(nexus.service, "mutated-service");
        assert_eq!(nexus.operation, "mutated-operation");
    }

    #[test]
    fn continue_as_new_interceptor_header_reaches_proto_command() {
        let init = InitializeWorkflow {
            workflow_type: TestWorkflow.name().to_string(),
            ..Default::default()
        };
        let init = WorkflowInit {
            namespace: "default".to_string(),
            task_queue: "task-queue".to_string(),
            run_id: "run-id".to_string(),
            initialize_workflow: init,
        };
        let base = BaseWorkflowContext::from_raw(
            init,
            DataConverter::default(),
            Rc::new(NoopHost),
            None,
            vec![WorkflowInterceptorConstructor::new(|_| {
                MutatingRemainingOutboundInterceptor
            })],
        );
        let ctx = WorkflowContext::from_base(base, Rc::new(RefCell::new(TestWorkflow)));

        let termination = ctx
            .continue_as_new(7, ContinueAsNewOptions::default())
            .expect_err("continue_as_new should terminate the workflow");
        let WorkflowTermination::ContinueAsNew(proto_command) = termination else {
            panic!("expected continue-as-new termination")
        };

        assert_eq!(
            proto_command.headers,
            HashMap::from([(
                "continue-header".to_string(),
                Payload::from(b"continue-header-value".as_slice()),
            )])
        );
    }

    #[test]
    fn construction_waker_uses_runtime_poll_waker() {
        let base = test_context().sync.base;
        let wakes = Arc::new(AtomicUsize::new(0));
        let waker = Waker::from(Arc::new(CountingWake(wakes.clone())));
        let _guard = base.enter_runtime_poll(&waker);
        base.construction_waker().wake_by_ref();
        assert_eq!(wakes.load(AtomicOrdering::Relaxed), 1);
    }

    #[test]
    fn workflow_context_continue_as_new_serializes_input_and_defaults() {
        let ctx = test_context();

        let termination = ctx
            .continue_as_new(7, ContinueAsNewOptions::default())
            .expect_err("continue_as_new should terminate the workflow");
        assert!(
            matches!(termination, WorkflowTermination::ContinueAsNew(_)),
            "expected continue-as-new termination, got {termination:?}"
        );
        let WorkflowTermination::ContinueAsNew(cmd) = termination else {
            unreachable!()
        };

        assert_eq!(
            *cmd,
            crate::runtime::types::ContinueAsNewRequest {
                workflow_type: TestWorkflow.name().to_string(),
                task_queue: String::new(),
                arguments: vec![7u8.as_json_payload().unwrap()],
                workflow_run_timeout: None,
                workflow_task_timeout: None,
                backoff_start_interval: None,
                memo: HashMap::new(),
                headers: HashMap::new(),
                search_attributes: None,
                retry_policy: None,
                versioning_intent: ProtoVersioningIntent::Unspecified.into(),
                initial_versioning_behavior: ProtoContinueAsNewVersioningBehavior::Unspecified
                    .into(),
            }
        );
    }

    #[test]
    fn sync_workflow_context_continue_as_new_applies_options() {
        let ctx = test_context();
        let sync = ctx.sync_context();
        let mut memo = MemoValues::new();
        memo.insert("memo-key", "memo-value".to_string());
        let mut proto_search_attributes = ProtoSearchAttributes::default();
        proto_search_attributes.indexed_fields.insert(
            "CustomKeywordField".to_string(),
            Payload::from(b"value".as_slice()),
        );
        let search_attributes = SearchAttributes::from_proto(&proto_search_attributes);

        let termination = sync
            .continue_as_new(
                11,
                ContinueAsNewOptions {
                    workflow_type: Some("next-workflow".to_string()),
                    task_queue: Some("next-task-queue".to_string()),
                    run_timeout: Some(Duration::from_secs(10)),
                    task_timeout: Some(Duration::from_secs(3)),
                    backoff_start_interval: Some(Duration::from_secs(4)),
                    memo: Some(memo.clone()),
                    search_attributes: Some(search_attributes.clone()),
                    retry_policy: Some(RetryPolicy::builder().maximum_attempts(5).build()),
                    versioning_intent: Some(ProtoVersioningIntent::Compatible.into()),
                    initial_versioning_behavior: Some(
                        ContinueAsNewVersioningBehavior::UseRampingVersion,
                    ),
                },
            )
            .expect_err("continue_as_new should terminate the workflow");
        assert!(
            matches!(termination, WorkflowTermination::ContinueAsNew(_)),
            "expected continue-as-new termination, got {termination:?}"
        );
        let WorkflowTermination::ContinueAsNew(cmd) = termination else {
            unreachable!()
        };

        assert_eq!(
            *cmd,
            crate::runtime::types::ContinueAsNewRequest {
                workflow_type: "next-workflow".to_string(),
                task_queue: "next-task-queue".to_string(),
                arguments: vec![11u8.as_json_payload().unwrap()],
                workflow_run_timeout: Some(Duration::from_secs(10).try_into().unwrap()),
                workflow_task_timeout: Some(Duration::from_secs(3).try_into().unwrap()),
                backoff_start_interval: Some(Duration::from_secs(4).try_into().unwrap()),
                memo: HashMap::from([(
                    "memo-key".to_string(),
                    "memo-value".as_json_payload().unwrap(),
                )]),
                headers: HashMap::new(),
                search_attributes: Some(proto_search_attributes),
                retry_policy: Some(ProtoRetryPolicy {
                    initial_interval: Some(Duration::from_secs(1).try_into().unwrap()),
                    backoff_coefficient: 2.0,
                    maximum_attempts: 5,
                    ..Default::default()
                }),
                versioning_intent: ProtoVersioningIntent::Compatible.into(),
                initial_versioning_behavior: ProtoContinueAsNewVersioningBehavior::UseRampingVersion
                    as i32,
            }
        );
    }

    #[test]
    fn continue_as_new_preserves_explicit_empty_search_attributes() {
        let ctx = test_context();
        let sync = ctx.sync_context();

        let termination = sync
            .continue_as_new(
                11,
                ContinueAsNewOptions {
                    search_attributes: Some(SearchAttributes::default()),
                    ..Default::default()
                },
            )
            .expect_err("continue_as_new should terminate the workflow");
        let WorkflowTermination::ContinueAsNew(cmd) = termination else {
            unreachable!()
        };

        assert_eq!(
            cmd.search_attributes,
            Some(ProtoSearchAttributes::default())
        );
    }

    #[test]
    fn workflow_context_continue_as_new_applies_auto_upgrade_versioning_behavior() {
        let ctx = test_context();

        let termination = ctx
            .continue_as_new(
                13,
                ContinueAsNewOptions {
                    initial_versioning_behavior: Some(ContinueAsNewVersioningBehavior::AutoUpgrade),
                    ..Default::default()
                },
            )
            .expect_err("continue_as_new should terminate the workflow");
        let WorkflowTermination::ContinueAsNew(cmd) = termination else {
            unreachable!()
        };

        assert_eq!(
            cmd.initial_versioning_behavior,
            ProtoContinueAsNewVersioningBehavior::AutoUpgrade as i32
        );
    }

    #[test]
    fn continue_as_new_reports_serialization_errors() {
        #[derive(Debug)]
        struct FailingInput;

        impl TemporalSerializable for FailingInput {
            fn to_payload(
                &self,
                _ctx: &temporalio_common_wasm::data_converters::SerializationContext<'_>,
            ) -> Result<Payload, temporalio_common_wasm::data_converters::PayloadConversionError>
            {
                Err(
                    temporalio_common_wasm::data_converters::PayloadConversionError::EncodingError(
                        std::io::Error::other("serialization failure").into(),
                    ),
                )
            }
        }

        impl TemporalDeserializable for FailingInput {
            fn from_payload(
                _ctx: &temporalio_common_wasm::data_converters::SerializationContext<'_>,
                _payload: Payload,
            ) -> Result<Self, temporalio_common_wasm::data_converters::PayloadConversionError>
            {
                unreachable!("test input is only serialized")
            }
        }

        #[workflow]
        #[derive(Default)]
        struct FailingWorkflow;

        #[workflow_methods]
        impl FailingWorkflow {
            #[run]
            async fn run(
                _ctx: &mut WorkflowContext<Self>,
                _input: FailingInput,
            ) -> crate::WorkflowResult<()> {
                unreachable!("test workflow run should not be polled")
            }
        }

        let init = InitializeWorkflow {
            workflow_type: "failing-workflow".to_string(),
            ..Default::default()
        };
        let init = WorkflowInit {
            namespace: "default".to_string(),
            task_queue: "orig-task-queue".to_string(),
            run_id: "run-id".to_string(),
            initialize_workflow: init,
        };
        let base = BaseWorkflowContext::from_raw(
            init,
            DataConverter::default(),
            Rc::new(NoopHost),
            None,
            Vec::new(),
        );
        let ctx = WorkflowContext::from_base(base, Rc::new(RefCell::new(FailingWorkflow)));

        let err = ctx
            .continue_as_new(FailingInput, ContinueAsNewOptions::default())
            .expect_err("serialization errors should be surfaced");

        let WorkflowTermination::Failed(err) = err else {
            panic!("expected failed termination, got {err:?}");
        };
        assert_eq!(err.to_string(), "Encoding error: serialization failure");
    }

    #[test]
    fn continue_as_new_reports_memo_serialization_errors() {
        let ctx = test_context();
        let mut memo = MemoValues::new();
        memo.insert("invalid", FailingMemoValue);

        let err = ctx
            .continue_as_new(
                7,
                ContinueAsNewOptions {
                    memo: Some(memo),
                    ..Default::default()
                },
            )
            .expect_err("memo serialization errors should be surfaced");

        let WorkflowTermination::Failed(err) = err else {
            panic!("expected failed termination, got {err:?}");
        };
        assert_eq!(
            err.to_string(),
            "Encoding error: memo serialization failure"
        );
    }

    #[test]
    fn upsert_search_attributes_updates_local_state() {
        use temporalio_common_wasm::search_attributes::SearchAttributeKey;

        const K: SearchAttributeKey<i64> = SearchAttributeKey::int("my_int");

        let ctx = test_context();
        assert!(ctx.search_attributes().is_empty());

        ctx.upsert_search_attributes([K.value_set(42)]);
        let attrs = ctx.search_attributes();
        assert_eq!(attrs.get(&K), Some(42));
    }

    #[test]
    fn upsert_memo_updates_local_state_and_encodes_removals() {
        let init = InitializeWorkflow {
            workflow_type: TestWorkflow.name().to_string(),
            memo: Some(ProtoMemo {
                fields: HashMap::from([("old".to_string(), "before".as_json_payload().unwrap())]),
            }),
            ..Default::default()
        };
        let host = Rc::new(RecordingHost::default());
        let init = WorkflowInit {
            namespace: "default".to_string(),
            task_queue: "orig-task-queue".to_string(),
            run_id: "run-id".to_string(),
            initialize_workflow: init,
        };
        let base = BaseWorkflowContext::from_raw(
            init,
            DataConverter::default(),
            host.clone(),
            None,
            Vec::new(),
        );
        let ctx = WorkflowContext::from_base(base, Rc::new(RefCell::new(TestWorkflow)));

        assert_eq!(
            ctx.memo().get::<String>("old").unwrap(),
            Some("before".to_string())
        );
        ctx.upsert_memo([("new", Some(MemoValue::new(42_u32))), ("old", None)])
            .unwrap();

        let current = ctx.memo();
        assert_eq!(current.get::<u32>("new").unwrap(), Some(42));
        assert_eq!(current.get::<String>("old").unwrap(), None);
        let view = ctx.view();
        assert_eq!(view.memo().get::<u32>("new").unwrap(), Some(42));
        assert_eq!(
            view.memo().raw(),
            view.raw()
                .memo
                .as_ref()
                .expect("view memo should be present")
        );

        let commands = host.commands.borrow();
        let [command] = commands.as_slice() else {
            panic!("expected one modify-properties command");
        };
        let Some(workflow_command::Variant::ModifyWorkflowProperties(command)) = &command.variant
        else {
            panic!("expected a modify-properties command");
        };
        let fields = &command.upserted_memo.as_ref().unwrap().fields;
        let payload_converter = PayloadConverter::default();
        let removal_payload = MemoValue::new(()).to_payload(&payload_converter).unwrap();
        assert_eq!(fields.get("old"), Some(&removal_payload));
        assert_eq!(
            u32::from_json_payload(fields.get("new").unwrap()).unwrap(),
            42
        );
    }

    #[test]
    fn upsert_memo_conversion_failure_does_not_mutate_or_emit_command() {
        let host = Rc::new(RecordingHost::default());
        let init = InitializeWorkflow {
            workflow_type: TestWorkflow.name().to_string(),
            ..Default::default()
        };
        let init = WorkflowInit {
            namespace: "default".to_string(),
            task_queue: "orig-task-queue".to_string(),
            run_id: "run-id".to_string(),
            initialize_workflow: init,
        };
        let base = BaseWorkflowContext::from_raw(
            init,
            DataConverter::default(),
            host.clone(),
            None,
            Vec::new(),
        );
        let ctx = WorkflowContext::from_base(base, Rc::new(RefCell::new(TestWorkflow)));
        let err = ctx
            .upsert_memo([
                ("valid", Some(MemoValue::new("value".to_string()))),
                ("invalid", Some(MemoValue::new(FailingMemoValue))),
            ])
            .unwrap_err();

        assert_eq!(
            err.to_string(),
            "Encoding error: memo serialization failure"
        );
        assert_eq!(ctx.memo().get::<String>("valid").unwrap(), None);
        assert!(host.commands.borrow().is_empty());
    }

    #[test]
    fn upsert_search_attributes_unset_removes_from_local_state() {
        use temporalio_common_wasm::search_attributes::SearchAttributeKey;

        const K: SearchAttributeKey<String> = SearchAttributeKey::keyword("my_kw");

        let ctx = test_context();
        // Set, then unset.
        ctx.upsert_search_attributes([K.value_set("hello".into())]);
        assert_eq!(ctx.search_attributes().get(&K), Some("hello".into()));

        ctx.upsert_search_attributes([K.value_unset()]);
        assert!(!ctx.search_attributes().contains_key(&K));
        assert!(ctx.search_attributes().is_empty());
    }

    #[test]
    fn upsert_search_attributes_multiple_updates_last_wins() {
        use temporalio_common_wasm::search_attributes::SearchAttributeKey;

        const K: SearchAttributeKey<i64> = SearchAttributeKey::int("counter");

        let ctx = test_context();
        ctx.upsert_search_attributes([K.value_set(1), K.value_set(2)]);
        assert_eq!(ctx.search_attributes().get(&K), Some(2));
    }

    #[test]
    fn upsert_search_attributes_merges_with_initial() {
        use temporalio_common_wasm::search_attributes::SearchAttributeKey;

        const A: SearchAttributeKey<i64> = SearchAttributeKey::int("attr_a");
        const B: SearchAttributeKey<String> = SearchAttributeKey::keyword("attr_b");

        // Start with initial search attribute A.
        let init_sa = SearchAttributes::new([A.value_set(1)]).into_proto();
        let init = InitializeWorkflow {
            workflow_type: TestWorkflow.name().to_string(),
            search_attributes: Some(init_sa),
            ..Default::default()
        };
        let init = WorkflowInit {
            namespace: "default".to_string(),
            task_queue: "tq".to_string(),
            run_id: "run-id".to_string(),
            initialize_workflow: init,
        };
        let base = BaseWorkflowContext::from_raw(
            init,
            DataConverter::default(),
            Rc::new(NoopHost),
            None,
            Vec::new(),
        );
        let ctx = WorkflowContext::from_base(base, Rc::new(RefCell::new(TestWorkflow)));

        assert_eq!(ctx.search_attributes().get(&A), Some(1));

        // Upsert B — A should still be present.
        ctx.upsert_search_attributes([B.value_set("hello".into())]);
        assert_eq!(ctx.search_attributes().get(&A), Some(1));
        assert_eq!(ctx.search_attributes().get(&B), Some("hello".into()));
    }

    #[test]
    fn view_search_attributes_returns_typed() {
        use temporalio_common_wasm::search_attributes::SearchAttributeKey;

        const K: SearchAttributeKey<bool> = SearchAttributeKey::bool("active");

        let init_sa = SearchAttributes::new([K.value_set(true)]).into_proto();
        let init = InitializeWorkflow {
            workflow_type: TestWorkflow.name().to_string(),
            search_attributes: Some(init_sa),
            ..Default::default()
        };
        let init = WorkflowInit {
            namespace: "default".to_string(),
            task_queue: "tq".to_string(),
            run_id: "run-id".to_string(),
            initialize_workflow: init,
        };
        let base = BaseWorkflowContext::from_raw(
            init,
            DataConverter::default(),
            Rc::new(NoopHost),
            None,
            Vec::new(),
        );
        let ctx = WorkflowContext::from_base(base, Rc::new(RefCell::new(TestWorkflow)));

        let view = ctx.view();
        let sa = view
            .search_attributes()
            .expect("should have search attributes");
        assert_eq!(sa.get(&K), Some(true));
    }

    #[test]
    fn workflow_info_retains_raw_initialization() {
        let init = InitializeWorkflow {
            workflow_type: TestWorkflow.name().to_string(),
            identity: "raw-only-identity".to_owned(),
            ..Default::default()
        };
        let expected = init.clone();
        let init = WorkflowInit {
            namespace: "default".to_string(),
            task_queue: "tq".to_string(),
            run_id: "run-id".to_string(),
            initialize_workflow: init,
        };
        let base = BaseWorkflowContext::from_raw(
            init,
            DataConverter::default(),
            Rc::new(NoopHost),
            None,
            Vec::new(),
        );
        let ctx = WorkflowContext::from_base(base, Rc::new(RefCell::new(TestWorkflow)));
        let info = ctx.info();

        assert_eq!(info.raw().identity, "raw-only-identity");
        assert_eq!(info.raw(), &expected);
        assert_eq!(info.into_raw(), expected);
    }
}