_workflow_instance.py

"""Workflow worker runner and instance."""

from __future__ import annotations

import asyncio
import collections
import contextvars
import inspect
import json
import logging
import random
import sys
import threading
import traceback
import warnings
from abc import ABC, abstractmethod
from collections import deque
from collections.abc import (
    Awaitable,
    Callable,
    Coroutine,
    Generator,
    Iterable,
    Iterator,
    Mapping,
    MutableMapping,
    Sequence,
)
from contextlib import contextmanager
from dataclasses import dataclass
from datetime import timedelta
from enum import IntEnum
from typing import (
    Any,
    Generic,
    NoReturn,
    TypeAlias,
    TypeVar,
    cast,
)

import nexusrpc
from nexusrpc import InputT, OutputT
from typing_extensions import Self, TypedDict, TypeVarTuple, Unpack

import temporalio.activity
import temporalio.api.common.v1
import temporalio.api.enums.v1
import temporalio.api.sdk.v1
import temporalio.bridge.proto.activity_result
import temporalio.bridge.proto.child_workflow
import temporalio.bridge.proto.common
import temporalio.bridge.proto.nexus
import temporalio.bridge.proto.workflow_activation
import temporalio.bridge.proto.workflow_commands
import temporalio.bridge.proto.workflow_completion
import temporalio.common
import temporalio.converter
import temporalio.exceptions
import temporalio.workflow
from temporalio.converter import StorageDriverStoreContext, StorageDriverWorkflowInfo
from temporalio.service import __version__

from ..api.failure.v1.message_pb2 import Failure
from . import _command_aware_visitor
from ._interceptor import (
    ContinueAsNewInput,
    ExecuteWorkflowInput,
    HandleQueryInput,
    HandleSignalInput,
    HandleUpdateInput,
    SignalChildWorkflowInput,
    SignalExternalWorkflowInput,
    StartActivityInput,
    StartChildWorkflowInput,
    StartLocalActivityInput,
    StartNexusOperationInput,
    WorkflowInboundInterceptor,
    WorkflowOutboundInterceptor,
)

logger = logging.getLogger(__name__)

# Set to true to log all cases where we're ignoring things during delete
LOG_IGNORE_DURING_DELETE = False


class WorkflowRunner(ABC):
    """Abstract runner for workflows that creates workflow instances to run.

    :py:class:`UnsandboxedWorkflowRunner` is an implementation that locally runs
    the workflow.
    """

    @abstractmethod
    def prepare_workflow(self, defn: temporalio.workflow._Definition) -> None:
        """Prepare a workflow for future execution.

        This is run once for each workflow definition when a worker starts. This
        allows the runner to do anything necessarily to prepare for this
        definition to be used multiple times in create_instance.

        Args:
            defn: The workflow definition.
        """
        raise NotImplementedError

    @abstractmethod
    def create_instance(self, det: WorkflowInstanceDetails) -> WorkflowInstance:
        """Create a workflow instance that can handle activations.

        Args:
            det: Details that can be used to create the instance.

        Returns:
            Workflow instance that can handle activations.
        """
        raise NotImplementedError

    def set_worker_level_failure_exception_types(
        self,
        types: Sequence[type[BaseException]],  # type:ignore[reportUnusedParameter]
    ) -> None:
        """Set worker-level failure exception types that will be used to
        validate in the sandbox when calling ``prepare_workflow``.

        Args:
            types: Exception types.
        """
        pass


@dataclass(frozen=True)
class WorkflowInstanceDetails:
    """Immutable details for creating a workflow instance."""

    payload_converter_class: type[temporalio.converter.PayloadConverter]
    failure_converter_class: type[temporalio.converter.FailureConverter]
    interceptor_classes: Sequence[type[WorkflowInboundInterceptor]]
    defn: temporalio.workflow._Definition
    info: temporalio.workflow.Info
    randomness_seed: int
    extern_functions: Mapping[str, Callable]
    disable_eager_activity_execution: bool
    worker_level_failure_exception_types: Sequence[type[BaseException]]
    last_completion_result: temporalio.api.common.v1.Payloads
    last_failure: Failure | None


class WorkflowInstance(ABC):
    """Instance of a workflow that can handle activations."""

    @abstractmethod
    def activate(
        self, act: temporalio.bridge.proto.workflow_activation.WorkflowActivation
    ) -> temporalio.bridge.proto.workflow_completion.WorkflowActivationCompletion:
        """Handle an activation and return completion.

        This should never raise an exception, but instead catch all exceptions
        and set as completion failure.

        Args:
            act: Protobuf activation.

        Returns:
            Completion object with successful commands set or failure info set.
        """
        raise NotImplementedError

    @abstractmethod
    def get_serialization_context(
        self,
        command_info: _command_aware_visitor.CommandInfo | None,
    ) -> temporalio.converter.SerializationContext | None:
        """Return appropriate serialization context.

        Args:
            command_info: Optional information identifying the associated command. If set, the payload
                         codec will have serialization context set appropriately for that command.

        Returns:
            The serialization context, or None if no context should be set.
        """
        raise NotImplementedError

    @abstractmethod
    def get_external_store_context(
        self,
        command_info: _command_aware_visitor.CommandInfo | None,
    ) -> StorageDriverStoreContext:
        """Return appropriate store context for external storage operations.

        Args:
            command_info: Optional information identifying the associated command.

        Returns:
            The store context associated with the command.
        """
        raise NotImplementedError

    @abstractmethod
    def get_info(self) -> temporalio.workflow.Info:
        """Return the workflow info for this instance."""
        raise NotImplementedError

    def get_thread_id(self) -> int | None:
        """Return the thread identifier that this workflow is running on.

        Not an abstractmethod because it is not mandatory to implement. Used primarily for getting the frames of a deadlocked thread.

        Returns:
            Thread ID if the workflow is running, None if not.
        """
        return None


class UnsandboxedWorkflowRunner(WorkflowRunner):
    """Workflow runner that does not do any sandboxing."""

    def prepare_workflow(self, defn: temporalio.workflow._Definition) -> None:
        """Implements :py:meth:`WorkflowRunner.prepare_workflow` as a no-op."""
        pass

    def create_instance(self, det: WorkflowInstanceDetails) -> WorkflowInstance:
        """Implements :py:meth:`WorkflowRunner.create_instance`."""
        # We ignore MyPy failing to instantiate this because it's not _really_
        # abstract at runtime. All of the asyncio.AbstractEventLoop calls that
        # we are not implementing are not abstract, they just throw not-impl'd
        # errors.
        return _WorkflowInstanceImpl(det)  # type: ignore[abstract]


_T = TypeVar("_T")
_Context: TypeAlias = dict[str, Any]
_ExceptionHandler: TypeAlias = Callable[[asyncio.AbstractEventLoop, _Context], Any]


class _WorkflowInstanceImpl(  # type: ignore[reportImplicitAbstractClass]
    WorkflowInstance, temporalio.workflow._Runtime, asyncio.AbstractEventLoop
):
    def __init__(self, det: WorkflowInstanceDetails) -> None:
        # No init for AbstractEventLoop
        WorkflowInstance.__init__(self)
        temporalio.workflow._Runtime.__init__(self)
        self._defn = det.defn
        self._workflow_input: ExecuteWorkflowInput | None = None
        self._info = det.info
        self._context_free_payload_converter = det.payload_converter_class()
        self._context_free_failure_converter = det.failure_converter_class()
        workflow_context = temporalio.converter.WorkflowSerializationContext(
            namespace=det.info.namespace,
            workflow_id=det.info.workflow_id,
        )
        self._workflow_context_payload_converter = self._payload_converter_with_context(
            workflow_context
        )
        self._workflow_context_failure_converter = self._failure_converter_with_context(
            workflow_context
        )

        self._extern_functions = det.extern_functions
        self._disable_eager_activity_execution = det.disable_eager_activity_execution
        self._worker_level_failure_exception_types = (
            det.worker_level_failure_exception_types
        )
        self._primary_task: asyncio.Task[None] | None = None
        self._time_ns = 0
        self._cancel_requested = False
        self._deployment_version_for_current_task: None | (
            temporalio.bridge.proto.common.WorkerDeploymentVersion
        ) = None
        self._current_history_length = 0
        self._current_history_size = 0
        self._continue_as_new_suggested = False
        self._target_worker_deployment_version_changed = False
        # Lazily loaded
        self._untyped_converted_memo: MutableMapping[str, Any] | None = None
        # Handles which are ready to run on the next event loop iteration
        self._ready: deque[asyncio.Handle] = collections.deque()
        self._conditions: list[tuple[Callable[[], bool], asyncio.Future]] = []
        # Keyed by seq
        self._pending_timers: dict[int, _TimerHandle] = {}
        self._pending_activities: dict[int, _ActivityHandle] = {}
        self._pending_child_workflows: dict[int, _ChildWorkflowHandle] = {}
        self._pending_nexus_operations: dict[int, _NexusOperationHandle] = {}
        self._pending_external_signals: dict[int, tuple[asyncio.Future, str]] = {}
        self._pending_external_cancels: dict[int, tuple[asyncio.Future, str]] = {}
        # Keyed by type
        self._curr_seqs: dict[str, int] = {}
        # TODO(cretz): Any concerns about not sharing this? Maybe the types I
        # need to lookup should be done at definition time?
        self._exception_handler: _ExceptionHandler | None = None
        # The actual instance, instantiated on first _run_once
        self._object: Any = None
        self._is_replaying: bool = False
        self._random = random.Random(det.randomness_seed)
        self._current_seed = det.randomness_seed
        self._seed_callbacks: list[Callable[[int], None]] = []
        self._read_only = False
        self._in_query_or_validator = False

        # Patches we have been notified of and memoized patch responses
        self._patches_notified: set[str] = set()
        self._patches_memoized: dict[str, bool] = {}

        # Tasks stored by asyncio are weak references and therefore can get GC'd
        # which can cause warnings like "Task was destroyed but it is pending!".
        # So we store the tasks ourselves.
        # See https://bugs.python.org/issue21163 and others.
        self._tasks: set[asyncio.Task] = set()

        # We maintain signals, queries, and updates on this class since handlers can be
        # added during workflow execution
        self._signals = dict(self._defn.signals)
        self._queries = dict(self._defn.queries)
        self._updates = dict(self._defn.updates)

        # We record in-progress signals and updates in order to support waiting for handlers to
        # finish, and issuing warnings when the workflow exits with unfinished handlers. Since
        # signals lack a unique per-invocation identifier, we introduce a sequence number for the
        # purpose.
        self._handled_signals_seq = 0
        self._in_progress_signals: dict[int, HandlerExecution] = {}
        self._in_progress_updates: dict[str, HandlerExecution] = {}

        # Add stack trace handler
        # TODO(cretz): Is it ok that this can be forcefully overridden by the
        # workflow author? They could technically override in interceptor
        # anyways. Putting here ensures it ends up in the query list on
        # query-not-found error.
        self._queries["__stack_trace"] = temporalio.workflow._QueryDefinition(
            name="__stack_trace",
            fn=self._stack_trace,
            is_method=False,
            arg_types=[],
            ret_type=str,
        )

        self._queries["__enhanced_stack_trace"] = temporalio.workflow._QueryDefinition(
            name="__enhanced_stack_trace",
            fn=self._enhanced_stack_trace,
            is_method=False,
            arg_types=[],
            ret_type=temporalio.api.sdk.v1.EnhancedStackTrace,
        )

        self._queries["__temporal_workflow_metadata"] = (
            temporalio.workflow._QueryDefinition(
                name="__temporal_workflow_metadata",
                fn=self._temporal_workflow_metadata,
                is_method=False,
                arg_types=[],
                ret_type=temporalio.api.sdk.v1.WorkflowMetadata,
            )
        )

        # Maintain buffered signals for later-added dynamic handlers
        self._buffered_signals: dict[
            str, list[temporalio.bridge.proto.workflow_activation.SignalWorkflow]
        ] = {}

        # When we evict, we have to mark the workflow as deleting so we don't
        # add any commands and we swallow exceptions on tear down
        self._deleting = False

        # We only create the metric meter lazily
        self._metric_meter: _ReplaySafeMetricMeter | None = None

        # For tracking the thread this workflow is running on (primarily for deadlock situations)
        self._current_thread_id: int | None = None

        # The current details (as opposed to static details on workflow start), returned in the
        # metadata query
        self._current_details = ""

        self._last_completion_result = det.last_completion_result
        self._last_failure = det.last_failure

        # The versioning behavior of this workflow, as established by annotation or by the dynamic
        # config function. Is only set once upon initialization.
        self._versioning_behavior: temporalio.common.VersioningBehavior | None = None

        # Dynamic failure exception types as overridden by the dynamic config function
        self._dynamic_failure_exception_types: (
            None | (Sequence[type[BaseException]])
        ) = None

        # Create interceptors. We do this with our runtime on the loop just in
        # case they want to access info() during init(). This should remain at the end of the constructor so that variables are defined during interceptor creation
        temporalio.workflow._Runtime.set_on_loop(asyncio.get_running_loop(), self)
        try:
            root_inbound = _WorkflowInboundImpl(self)
            self._inbound: WorkflowInboundInterceptor = root_inbound
            for interceptor_class in reversed(list(det.interceptor_classes)):
                self._inbound = interceptor_class(self._inbound)
            # During init we set ourselves on the current loop
            self._inbound.init(_WorkflowOutboundImpl(self))
            self._outbound = root_inbound._outbound
        finally:
            # Remove our runtime from the loop
            temporalio.workflow._Runtime.set_on_loop(asyncio.get_running_loop(), None)

        # Set ourselves on our own loop
        temporalio.workflow._Runtime.set_on_loop(self, self)

    def get_thread_id(self) -> int | None:
        return self._current_thread_id

    #### Activation functions ####
    # These are in alphabetical order and besides "activate", and
    # "_make_workflow_input", all other calls are "_apply_" + the job field
    # name.

    def activate(
        self, act: temporalio.bridge.proto.workflow_activation.WorkflowActivation
    ) -> temporalio.bridge.proto.workflow_completion.WorkflowActivationCompletion:
        # Reset current completion, time, and whether replaying
        self._current_completion = (
            temporalio.bridge.proto.workflow_completion.WorkflowActivationCompletion()
        )
        self._current_completion.successful.SetInParent()

        self._current_activation_error: Exception | None = None
        self._deployment_version_for_current_task = (
            act.deployment_version_for_current_task
        )
        self._current_history_length = act.history_length
        self._current_history_size = act.history_size_bytes
        self._continue_as_new_suggested = act.continue_as_new_suggested
        self._target_worker_deployment_version_changed = (
            act.target_worker_deployment_version_changed
        )
        self._time_ns = act.timestamp.ToNanoseconds()
        self._is_replaying = act.is_replaying
        self._current_thread_id = threading.get_ident()
        self._current_internal_flags = act.available_internal_flags
        activation_err: Exception | None = None
        try:
            # Split into job sets with patches, then signals + updates, then
            # non-queries, then queries
            start_job = None
            job_sets: list[
                list[temporalio.bridge.proto.workflow_activation.WorkflowActivationJob]
            ] = [[], [], [], []]
            for job in act.jobs:
                if job.HasField("notify_has_patch"):
                    job_sets[0].append(job)
                elif job.HasField("signal_workflow") or job.HasField("do_update"):
                    job_sets[1].append(job)
                elif not job.HasField("query_workflow"):
                    if job.HasField("initialize_workflow"):
                        start_job = job.initialize_workflow
                    job_sets[2].append(job)
                else:
                    job_sets[3].append(job)

            if start_job:
                self._workflow_input = self._make_workflow_input(start_job)

            # Apply every job set, running after each set
            for index, job_set in enumerate(job_sets):
                if not job_set:
                    continue
                # Separate local-activity resolutions from other jobs.
                # Local activity results may arrive in non-deterministic
                # order (completion order varies between first-execution
                # and replay). Sort them by sequence number so coroutines
                # always resume in scheduling order, ensuring deterministic
                # interleaving regardless of execution mode.
                local_activity_jobs = []
                other_jobs = []
                for job in job_set:
                    if (
                        job.HasField("resolve_activity")
                        and job.resolve_activity.is_local
                    ):
                        local_activity_jobs.append(job)
                    else:
                        other_jobs.append(job)

                # Apply non-local-activity jobs first
                for job in other_jobs:
                    self._apply(job)

                # Apply local activity resolutions sorted by seq number.
                local_activity_jobs.sort(key=lambda j: j.resolve_activity.seq)
                for job in local_activity_jobs:
                    self._apply(job)

                # Run one iteration of the loop. We do not allow conditions to
                # be checked in patch jobs (first index) or query jobs (last
                # index).
                self._run_once(check_conditions=index == 1 or index == 2)
        except Exception as err:
            # We want some errors during activation, like those that can happen
            # during payload conversion, to be able to fail the workflow not the
            # task
            if self.workflow_is_failure_exception(err):
                try:
                    self._set_workflow_failure(err)
                except Exception as inner_err:
                    activation_err = inner_err
            else:
                # Otherwise all exceptions are activation errors
                activation_err = err
            # If we're deleting, swallow any activation error
            if self._deleting:
                if LOG_IGNORE_DURING_DELETE:
                    logger.debug(
                        "Ignoring exception while deleting workflow", exc_info=True
                    )
                activation_err = None

        # Apply versioning behavior if one was established
        if self._versioning_behavior:
            self._current_completion.successful.versioning_behavior = (
                self._versioning_behavior.value
            )

        # If we're deleting, there better be no more tasks. It is important for
        # the integrity of the system that we check this. If there are tasks
        # remaining, they and any associated coroutines will get garbage
        # collected which can trigger a GeneratorExit exception thrown in the
        # coroutine which can cause it to wakeup on a different thread which may
        # have a different workflow/event-loop going.
        if self._deleting and self._tasks:
            raise RuntimeError(
                f"Eviction processed, but {len(self._tasks)} tasks remain. "
                + f"Stack traces below:\n\n{self._stack_trace()}"
            )

        if activation_err:
            logger.warning(
                f"Failed activation on workflow {self._info.workflow_type} with ID {self._info.workflow_id} and run ID {self._info.run_id}",
                exc_info=activation_err,
                extra={
                    "temporal_workflow": self._info._logger_details(),
                    "__temporal_error_identifier": "WorkflowTaskFailure",
                },
            )
            # Set completion failure
            self._current_completion.failed.failure.SetInParent()
            try:
                self._workflow_context_failure_converter.to_failure(
                    activation_err,
                    self._workflow_context_payload_converter,
                    self._current_completion.failed.failure,
                )
            except Exception as inner_err:
                logger.exception(
                    f"Failed converting activation exception on workflow with run ID {act.run_id}"
                )
                self._current_completion.failed.failure.message = (
                    f"Failed converting activation exception: {inner_err}"
                )
                self._current_completion.failed.failure.application_failure_info.SetInParent()

        def is_completion(
            command: temporalio.bridge.proto.workflow_commands.workflow_commands_pb2.WorkflowCommand,
        ):
            return (
                command.HasField("complete_workflow_execution")
                or command.HasField("continue_as_new_workflow_execution")
                or command.HasField("fail_workflow_execution")
                or command.HasField("cancel_workflow_execution")
            )

        if any(map(is_completion, self._current_completion.successful.commands)):
            self._warn_if_unfinished_handlers()

        return self._current_completion

    def _apply(
        self, job: temporalio.bridge.proto.workflow_activation.WorkflowActivationJob
    ) -> None:
        if job.HasField("cancel_workflow"):
            self._apply_cancel_workflow(job.cancel_workflow)
        elif job.HasField("do_update"):
            self._apply_do_update(job.do_update)
        elif job.HasField("fire_timer"):
            self._apply_fire_timer(job.fire_timer)
        elif job.HasField("query_workflow"):
            self._apply_query_workflow(job.query_workflow)
        elif job.HasField("notify_has_patch"):
            self._apply_notify_has_patch(job.notify_has_patch)
        elif job.HasField("remove_from_cache"):
            self._apply_remove_from_cache(job.remove_from_cache)
        elif job.HasField("resolve_activity"):
            self._apply_resolve_activity(job.resolve_activity)
        elif job.HasField("resolve_child_workflow_execution"):
            self._apply_resolve_child_workflow_execution(
                job.resolve_child_workflow_execution
            )
        elif job.HasField("resolve_child_workflow_execution_start"):
            self._apply_resolve_child_workflow_execution_start(
                job.resolve_child_workflow_execution_start
            )
        elif job.HasField("resolve_nexus_operation_start"):
            self._apply_resolve_nexus_operation_start(job.resolve_nexus_operation_start)
        elif job.HasField("resolve_nexus_operation"):
            self._apply_resolve_nexus_operation(job.resolve_nexus_operation)
        elif job.HasField("resolve_request_cancel_external_workflow"):
            self._apply_resolve_request_cancel_external_workflow(
                job.resolve_request_cancel_external_workflow
            )
        elif job.HasField("resolve_signal_external_workflow"):
            self._apply_resolve_signal_external_workflow(
                job.resolve_signal_external_workflow
            )
        elif job.HasField("signal_workflow"):
            self._apply_signal_workflow(job.signal_workflow)
        elif job.HasField("initialize_workflow"):
            self._apply_initialize_workflow(job.initialize_workflow)
        elif job.HasField("update_random_seed"):
            self._apply_update_random_seed(job.update_random_seed)
        else:
            raise RuntimeError(f"Unrecognized job: {job.WhichOneof('variant')}")

    def _apply_cancel_workflow(
        self, _job: temporalio.bridge.proto.workflow_activation.CancelWorkflow
    ) -> None:
        self._cancel_requested = True
        # TODO(cretz): Details or cancel message or whatever?
        if self._primary_task:
            # The primary task may not have started yet and we want to give the
            # workflow the ability to receive the cancellation, so we must defer
            # this cancellation to the next iteration of the event loop.
            self.call_soon(self._primary_task.cancel)

    def _apply_do_update(
        self, job: temporalio.bridge.proto.workflow_activation.DoUpdate
    ):
        # Run the validator & handler in a task. Everything, including looking up the update definition, needs to be
        # inside the task, since the update may not be defined until after we have started the workflow - for example
        # if an update is in the first WFT & is also registered dynamically at the top of workflow code.
        async def run_update() -> None:
            # Set the current update for the life of this task
            temporalio.workflow._set_current_update_info(
                temporalio.workflow.UpdateInfo(id=job.id, name=job.name)
            )

            command = self._add_command()
            command.update_response.protocol_instance_id = job.protocol_instance_id
            past_validation = False
            try:
                defn = self._updates.get(job.name) or self._updates.get(None)
                if not defn:
                    known_updates = sorted([k for k in self._updates.keys() if k])
                    raise RuntimeError(
                        f"Update handler for '{job.name}' expected but not found, and there is no dynamic handler. "
                        f"known updates: [{' '.join(known_updates)}]"
                    )
                self._in_progress_updates[job.id] = HandlerExecution(
                    job.name, defn.unfinished_policy, job.id
                )
                args = self._process_handler_args(
                    job.name,
                    job.input,
                    defn.name,
                    defn.arg_types,
                    defn.dynamic_vararg,
                )
                handler_input = HandleUpdateInput(
                    id=job.id,
                    update=job.name,
                    args=args,
                    headers=job.headers,
                )

                if job.run_validator and defn.validator is not None:
                    with self._as_read_only(in_query_or_validator=True):
                        self._inbound.handle_update_validator(handler_input)
                        # Re-process arguments to avoid any problems caused by user mutation of them during validation
                        args = self._process_handler_args(
                            job.name,
                            job.input,
                            defn.name,
                            defn.arg_types,
                            defn.dynamic_vararg,
                        )
                        handler_input.args = args

                past_validation = True
                # Accept the update
                command.update_response.accepted.SetInParent()
                command = None  # type: ignore

                # Run the handler
                success = await self._inbound.handle_update_handler(handler_input)
                result_payloads = self._workflow_context_payload_converter.to_payloads(
                    [success]
                )
                if len(result_payloads) != 1:
                    raise ValueError(
                        f"Expected 1 result payload, got {len(result_payloads)}"
                    )
                command = self._add_command()
                command.update_response.protocol_instance_id = job.protocol_instance_id
                command.update_response.completed.CopyFrom(result_payloads[0])
            except (Exception, asyncio.CancelledError) as err:
                logger.debug(
                    f"Update raised failure with run ID {self._info.run_id}",
                    exc_info=True,
                )
                # All asyncio cancelled errors become Temporal cancelled errors
                if isinstance(err, asyncio.CancelledError):
                    err = temporalio.exceptions.CancelledError(
                        f"Cancellation raised within update {err}"
                    )
                # Read-only issues during validation should fail the task
                if isinstance(err, temporalio.workflow.ReadOnlyContextError):
                    self._current_activation_error = err
                    return
                # Validation failures are always update failures. We reuse
                # workflow failure logic to decide task failure vs update
                # failure after validation.
                if not past_validation or self.workflow_is_failure_exception(err):
                    if command is None:
                        command = self._add_command()
                        command.update_response.protocol_instance_id = (
                            job.protocol_instance_id
                        )
                    command.update_response.rejected.SetInParent()
                    self._workflow_context_failure_converter.to_failure(
                        err,
                        self._workflow_context_payload_converter,
                        command.update_response.rejected,
                    )
                else:
                    self._current_activation_error = err
                    return
            except BaseException:
                if self._deleting:
                    if LOG_IGNORE_DURING_DELETE:
                        logger.debug(
                            "Ignoring exception while deleting workflow", exc_info=True
                        )
                    return
                raise
            finally:
                self._in_progress_updates.pop(job.id, None)

        self.create_task(
            run_update(),
            name=f"update: {job.name}",
        )

    def _apply_fire_timer(
        self, job: temporalio.bridge.proto.workflow_activation.FireTimer
    ) -> None:
        # We ignore an absent handler because it may have been cancelled and
        # removed earlier this activation by a signal
        handle = self._pending_timers.pop(job.seq, None)
        if handle:
            self._ready.append(handle)

    def _apply_query_workflow(
        self, job: temporalio.bridge.proto.workflow_activation.QueryWorkflow
    ) -> None:
        # Wrap entire bunch of work in a task
        async def run_query() -> None:
            try:
                with self._as_read_only(in_query_or_validator=True):
                    # Named query or dynamic
                    defn = self._queries.get(job.query_type) or self._queries.get(None)
                    if not defn:
                        known_queries = sorted([k for k in self._queries.keys() if k])
                        raise RuntimeError(
                            f"Query handler for '{job.query_type}' expected but not found, "
                            f"known queries: [{' '.join(known_queries)}]"
                        )

                    # Create input
                    args = self._process_handler_args(
                        job.query_type,
                        job.arguments,
                        defn.name,
                        defn.arg_types,
                        defn.dynamic_vararg,
                    )
                    input = HandleQueryInput(
                        id=job.query_id,
                        query=job.query_type,
                        args=args,
                        headers=job.headers,
                    )
                    success = await self._inbound.handle_query(input)
                    result_payloads = (
                        self._workflow_context_payload_converter.to_payloads([success])
                    )
                    if len(result_payloads) != 1:
                        raise ValueError(
                            f"Expected 1 result payload, got {len(result_payloads)}"
                        )
                command = self._add_command()
                command.respond_to_query.query_id = job.query_id
                command.respond_to_query.succeeded.response.CopyFrom(result_payloads[0])
            except Exception as err:
                try:
                    command = self._add_command()
                    command.respond_to_query.query_id = job.query_id
                    self._workflow_context_failure_converter.to_failure(
                        err,
                        self._workflow_context_payload_converter,
                        command.respond_to_query.failed,
                    )
                except Exception as inner_err:
                    raise ValueError(
                        "Failed converting application error"
                    ) from inner_err

        # Schedule it
        self.create_task(run_query(), name=f"query: {job.query_type}")

    def _apply_notify_has_patch(
        self, job: temporalio.bridge.proto.workflow_activation.NotifyHasPatch
    ) -> None:
        self._patches_notified.add(job.patch_id)

    def _apply_remove_from_cache(
        self, _job: temporalio.bridge.proto.workflow_activation.RemoveFromCache
    ) -> None:
        self._deleting = True
        self._cancel_requested = True
        # We consider eviction to be under replay so that certain code like
        # logging that avoids replaying doesn't run during eviction either
        self._is_replaying = True
        # Cancel everything
        for task in self._tasks:
            task.cancel()

    def _apply_resolve_activity(
        self, job: temporalio.bridge.proto.workflow_activation.ResolveActivity
    ) -> None:
        handle = self._pending_activities.pop(job.seq, None)
        if not handle:
            raise RuntimeError(f"Failed finding activity handle for sequence {job.seq}")
        activity_context = temporalio.converter.ActivitySerializationContext(
            namespace=self._info.namespace,
            workflow_id=self._info.workflow_id,
            workflow_type=self._info.workflow_type,
            activity_type=handle._input.activity,
            activity_id=handle._input.activity_id,
            activity_task_queue=(
                handle._input.task_queue or self._info.task_queue
                if isinstance(handle._input, StartActivityInput)
                else self._info.task_queue
            ),
            is_local=isinstance(handle._input, StartLocalActivityInput),
        )
        payload_converter = self._payload_converter_with_context(activity_context)
        failure_converter = self._failure_converter_with_context(activity_context)
        if job.result.HasField("completed"):
            ret: Any | None = None
            if job.result.completed.HasField("result"):
                ret_types = [handle._input.ret_type] if handle._input.ret_type else None
                ret_vals = self._convert_payloads(
                    [job.result.completed.result],
                    ret_types,
                    payload_converter,
                )
                ret = ret_vals[0]
            handle._resolve_success(ret)
        elif job.result.HasField("failed"):
            handle._resolve_failure(
                failure_converter.from_failure(
                    job.result.failed.failure, payload_converter
                )
            )
        elif job.result.HasField("cancelled"):
            handle._resolve_failure(
                failure_converter.from_failure(
                    job.result.cancelled.failure, payload_converter
                )
            )
        elif job.result.HasField("backoff"):
            handle._resolve_backoff(job.result.backoff)
        else:
            raise RuntimeError("Activity did not have result")

    def _apply_resolve_child_workflow_execution(
        self,
        job: temporalio.bridge.proto.workflow_activation.ResolveChildWorkflowExecution,
    ) -> None:
        handle = self._pending_child_workflows.pop(job.seq, None)
        if not handle:
            raise RuntimeError(
                f"Failed finding child workflow handle for sequence {job.seq}"
            )

        if job.result.HasField("completed"):
            ret: Any | None = None
            if job.result.completed.HasField("result"):
                ret_types = [handle._input.ret_type] if handle._input.ret_type else None
                ret_vals = self._convert_payloads(
                    [job.result.completed.result],
                    ret_types,
                    handle._payload_converter,
                )
                ret = ret_vals[0]
            handle._resolve_success(ret)
        elif job.result.HasField("failed"):
            handle._resolve_failure(
                handle._failure_converter.from_failure(
                    job.result.failed.failure, handle._payload_converter
                )
            )
        elif job.result.HasField("cancelled"):
            handle._resolve_failure(
                handle._failure_converter.from_failure(
                    job.result.cancelled.failure, handle._payload_converter
                )
            )
        else:
            raise RuntimeError("Child workflow did not have result")

    def _apply_resolve_child_workflow_execution_start(
        self,
        job: temporalio.bridge.proto.workflow_activation.ResolveChildWorkflowExecutionStart,
    ) -> None:
        handle = self._pending_child_workflows.get(job.seq)
        if not handle:
            raise RuntimeError(
                f"Failed finding child workflow handle for sequence {job.seq}"
            )
        if job.HasField("succeeded"):
            # We intentionally do not pop here because this same handle is used
            # for waiting on the entire workflow to complete
            handle._resolve_start_success(job.succeeded.run_id)
        elif job.HasField("failed"):
            self._pending_child_workflows.pop(job.seq)
            if (
                job.failed.cause
                == temporalio.bridge.proto.child_workflow.StartChildWorkflowExecutionFailedCause.START_CHILD_WORKFLOW_EXECUTION_FAILED_CAUSE_WORKFLOW_ALREADY_EXISTS
            ):
                handle._resolve_failure(
                    temporalio.exceptions.WorkflowAlreadyStartedError(
                        job.failed.workflow_id, job.failed.workflow_type
                    )
                )
            else:
                handle._resolve_failure(
                    RuntimeError(f"Unknown child start fail cause: {job.failed.cause}")
                )
        elif job.HasField("cancelled"):
            self._pending_child_workflows.pop(job.seq)
            handle._resolve_failure(
                handle._failure_converter.from_failure(
                    job.cancelled.failure, handle._payload_converter
                )
            )
        else:
            raise RuntimeError("Child workflow start did not have a known status")

    def _apply_resolve_nexus_operation_start(
        self,
        job: temporalio.bridge.proto.workflow_activation.ResolveNexusOperationStart,
    ) -> None:
        handle = self._pending_nexus_operations.get(job.seq)
        if not handle:
            raise RuntimeError(
                f"Failed to find nexus operation handle for job sequence number {job.seq}"
            )
        if job.HasField("operation_token"):
            # The nexus operation started asynchronously. A `ResolveNexusOperation` job
            # will follow in a future activation.
            handle._resolve_start_success(job.operation_token)
        elif job.HasField("started_sync"):
            # The nexus operation 'started' in the sense that it's already resolved. A
            # `ResolveNexusOperation` job will be in the same activation.
            handle._resolve_start_success(None)
        elif job.HasField("failed"):
            # The nexus operation start failed; no ResolveNexusOperation will follow.
            self._pending_nexus_operations.pop(job.seq, None)
            handle._resolve_failure(
                handle._failure_converter.from_failure(
                    job.failed, handle._payload_converter
                )
            )
        else:
            raise ValueError(f"Unknown Nexus operation start status: {job}")

    def _apply_resolve_nexus_operation(
        self,
        job: temporalio.bridge.proto.workflow_activation.ResolveNexusOperation,
    ) -> None:
        handle = self._pending_nexus_operations.pop(job.seq, None)
        if not handle:
            # One way this can occur is:
            # 1. Cancel request issued with cancellation_type=WaitRequested.
            # 2. Server receives nexus cancel handler task completion and writes
            #    NexusOperationCancelRequestCompleted / NexusOperationCancelRequestFailed. On
            #    consuming this event, core sends an activation resolving the handle future as
            #    completed / failed.
            # 4. Subsequently, the nexus operation completes as completed/failed, causing the server
            #    to write NexusOperationCompleted / NexusOperationFailed. On consuming this event,
            #    core sends an activation which would attempt to resolve the handle future as
            #    completed / failed, but it has already been resolved.
            return

        # Handle the four oneof variants of NexusOperationResult
        result = job.result
        if result.HasField("completed"):
            [output] = self._convert_payloads(
                [result.completed],
                [handle._input.output_type] if handle._input.output_type else None,
                handle._payload_converter,
            )
            handle._resolve_success(output)
        elif result.HasField("failed"):
            handle._resolve_failure(
                handle._failure_converter.from_failure(
                    result.failed, handle._payload_converter
                )
            )
        elif result.HasField("cancelled"):
            handle._resolve_failure(
                handle._failure_converter.from_failure(
                    result.cancelled, handle._payload_converter
                )
            )
        elif result.HasField("timed_out"):
            handle._resolve_failure(
                handle._failure_converter.from_failure(
                    result.timed_out, handle._payload_converter
                )
            )
        else:
            raise RuntimeError("Nexus operation did not have a result")

    def _apply_resolve_request_cancel_external_workflow(
        self,
        job: temporalio.bridge.proto.workflow_activation.ResolveRequestCancelExternalWorkflow,
    ) -> None:
        pending = self._pending_external_cancels.pop(job.seq, None)
        if not pending:
            raise RuntimeError(
                f"Failed finding pending external cancel for sequence {job.seq}"
            )
        fut, external_workflow_id = pending
        # We intentionally let this error if future is already done
        if job.HasField("failure"):
            workflow_context = temporalio.converter.WorkflowSerializationContext(
                namespace=self._info.namespace,
                workflow_id=external_workflow_id,
            )
            payload_converter = self._payload_converter_with_context(workflow_context)
            failure_converter = self._failure_converter_with_context(workflow_context)
            fut.set_exception(
                failure_converter.from_failure(job.failure, payload_converter)
            )
        else:
            fut.set_result(None)

    def _apply_resolve_signal_external_workflow(
        self,
        job: temporalio.bridge.proto.workflow_activation.ResolveSignalExternalWorkflow,
    ) -> None:
        pending = self._pending_external_signals.pop(job.seq, None)
        if not pending:
            raise RuntimeError(
                f"Failed finding pending external signal for sequence {job.seq}"
            )
        fut, external_workflow_id = pending
        # We intentionally let this error if future is already done
        if job.HasField("failure"):
            workflow_context = temporalio.converter.WorkflowSerializationContext(
                namespace=self._info.namespace,
                workflow_id=external_workflow_id,
            )
            payload_converter = self._payload_converter_with_context(workflow_context)
            failure_converter = self._failure_converter_with_context(workflow_context)
            fut.set_exception(
                failure_converter.from_failure(job.failure, payload_converter)
            )
        else:
            fut.set_result(None)

    def _apply_signal_workflow(
        self, job: temporalio.bridge.proto.workflow_activation.SignalWorkflow
    ) -> None:
        # Apply to named or to dynamic or buffer
        signal_defn = self._signals.get(job.signal_name) or self._signals.get(None)
        if not signal_defn:
            self._buffered_signals.setdefault(job.signal_name, []).append(job)
            return
        self._process_signal_job(signal_defn, job)

    def _apply_initialize_workflow(
        self, _job: temporalio.bridge.proto.workflow_activation.InitializeWorkflow
    ) -> None:
        # Async call to run on the scheduler thread. This will be wrapped in
        # another function which applies exception handling.
        async def run_workflow(input: ExecuteWorkflowInput) -> None:
            try:
                result = await self._inbound.execute_workflow(input)
                result_payloads = self._workflow_context_payload_converter.to_payloads(
                    [result]
                )
                if len(result_payloads) != 1:
                    raise ValueError(
                        f"Expected 1 result payload, got {len(result_payloads)}"
                    )
                command = self._add_command()
                command.complete_workflow_execution.result.CopyFrom(result_payloads[0])
            except Exception:
                if self._deleting:
                    if LOG_IGNORE_DURING_DELETE:
                        logger.debug(
                            "Ignoring exception while deleting workflow", exc_info=True
                        )
                    return
                raise

        if not self._workflow_input:
            raise RuntimeError(
                "Expected workflow input to be set. This is an SDK Python bug."
            )
        self._primary_task = self.create_task(
            self._run_top_level_workflow_function(run_workflow(self._workflow_input)),
            name="run",
        )

    def _apply_update_random_seed(
        self, job: temporalio.bridge.proto.workflow_activation.UpdateRandomSeed
    ) -> None:
        self._random.seed(job.randomness_seed)
        self._current_seed = job.randomness_seed
        # Notify all registered callbacks
        for callback in self._seed_callbacks:
            try:
                callback(job.randomness_seed)
            except Exception:
                # Ignore callback errors to avoid disrupting workflow execution
                pass

    def _make_workflow_input(
        self, init_job: temporalio.bridge.proto.workflow_activation.InitializeWorkflow
    ) -> ExecuteWorkflowInput:
        # Set arg types, using raw values for dynamic
        arg_types = self._defn.arg_types
        if not self._defn.name:
            # Dynamic is just the raw value for each input value
            arg_types = [temporalio.common.RawValue] * len(init_job.arguments)

        args = self._convert_payloads(
            init_job.arguments, arg_types, self._workflow_context_payload_converter
        )
        # Put args in a list if dynamic
        if not self._defn.name:
            args = [args]

        return ExecuteWorkflowInput(
            type=self._defn.cls,
            # TODO(cretz): Remove cast when https://github.com/python/mypy/issues/5485 fixed
            run_fn=cast(Callable[..., Awaitable[Any]], self._defn.run_fn),
            args=args,
            headers=init_job.headers,
        )

    #### _Runtime direct workflow call overrides ####
    # These are in alphabetical order and all start with "workflow_".

    def workflow_all_handlers_finished(self) -> bool:
        return not self._in_progress_updates and not self._in_progress_signals

    def workflow_continue_as_new(
        self,
        *args: Any,
        workflow: None | Callable | str,
        task_queue: str | None,
        run_timeout: timedelta | None,
        task_timeout: timedelta | None,
        retry_policy: temporalio.common.RetryPolicy | None,
        memo: Mapping[str, Any] | None,
        search_attributes: None
        | (
            temporalio.common.SearchAttributes | temporalio.common.TypedSearchAttributes
        ),
        versioning_intent: temporalio.workflow.VersioningIntent | None,
        initial_versioning_behavior: temporalio.workflow.ContinueAsNewVersioningBehavior
        | None,
    ) -> NoReturn:
        self._assert_not_read_only("continue as new")
        # Use definition if callable
        name: str | None = None
        arg_types: list[type] | None = None
        if isinstance(workflow, str):
            name = workflow
        elif callable(workflow):
            defn = temporalio.workflow._Definition.must_from_run_fn(workflow)
            name = defn.name
            arg_types = defn.arg_types
        elif workflow is not None:
            raise TypeError("Workflow must be None, a string, or callable")  # type:ignore[reportUnreachable]

        self._outbound.continue_as_new(
            ContinueAsNewInput(
                workflow=name,
                args=args,
                task_queue=task_queue,
                run_timeout=run_timeout,
                task_timeout=task_timeout,
                retry_policy=retry_policy,
                memo=memo,
                search_attributes=search_attributes,
                headers={},
                arg_types=arg_types,
                versioning_intent=versioning_intent,
                initial_versioning_behavior=initial_versioning_behavior,
            )
        )

    def workflow_extern_functions(self) -> Mapping[str, Callable]:
        return self._extern_functions

    def workflow_get_current_build_id(self) -> str:
        if not self._deployment_version_for_current_task:
            return ""
        return self._deployment_version_for_current_task.build_id

    def workflow_get_current_deployment_version(
        self,
    ) -> temporalio.common.WorkerDeploymentVersion | None:
        if not self._deployment_version_for_current_task:
            return None
        return temporalio.common.WorkerDeploymentVersion(
            build_id=self._deployment_version_for_current_task.build_id,
            deployment_name=self._deployment_version_for_current_task.deployment_name,
        )

    def get_info(self) -> temporalio.workflow.Info:
        return self._info

    def workflow_get_current_history_length(self) -> int:
        return self._current_history_length

    def workflow_get_current_history_size(self) -> int:
        return self._current_history_size

    def workflow_get_external_workflow_handle(
        self, id: str, *, run_id: str | None
    ) -> temporalio.workflow.ExternalWorkflowHandle[Any]:
        return _ExternalWorkflowHandle(self, id, run_id)

    def workflow_get_query_handler(self, name: str | None) -> Callable | None:
        defn = self._queries.get(name)
        if not defn:
            return None
        # Bind if a method
        return defn.bind_fn(self._object) if defn.is_method else defn.fn

    def workflow_get_signal_handler(self, name: str | None) -> Callable | None:
        defn = self._signals.get(name)
        if not defn:
            return None
        # Bind if a method
        return defn.bind_fn(self._object) if defn.is_method else defn.fn

    def workflow_get_update_handler(self, name: str | None) -> Callable | None:
        defn = self._updates.get(name)
        if not defn:
            return None
        # Bind if a method
        return defn.bind_fn(self._object) if defn.is_method else defn.fn

    def workflow_get_update_validator(self, name: str | None) -> Callable | None:
        defn = self._updates.get(name) or self._updates.get(None)
        if not defn or not defn.validator:
            return None
        # Bind if a method
        return defn.bind_validator(self._object) if defn.is_method else defn.validator

    def workflow_info(self) -> temporalio.workflow.Info:
        return self._outbound.info()

    def workflow_instance(self) -> Any:
        return self._object

    def workflow_is_continue_as_new_suggested(self) -> bool:
        return self._continue_as_new_suggested

    def workflow_is_target_worker_deployment_version_changed(self) -> bool:
        return self._target_worker_deployment_version_changed

    def workflow_is_replaying(self) -> bool:
        return self._is_replaying

    def workflow_is_replaying_history_events(self) -> bool:
        return self._is_replaying and not self._in_query_or_validator

    def workflow_is_read_only(self) -> bool:
        return self._read_only

    def workflow_memo(self) -> Mapping[str, Any]:
        if self._untyped_converted_memo is None:
            self._untyped_converted_memo = {
                k: self._workflow_context_payload_converter.from_payload(v)
                for k, v in self._info.raw_memo.items()
            }
        return self._untyped_converted_memo

    def workflow_memo_value(
        self, key: str, default: Any, *, type_hint: type | None
    ) -> Any:
        payload = self._info.raw_memo.get(key)
        if not payload:
            if default is temporalio.common._arg_unset:
                raise KeyError(f"Memo does not have a value for key {key}")
            return default
        return self._workflow_context_payload_converter.from_payload(
            payload,
            type_hint,  # type: ignore[arg-type]
        )

    def workflow_upsert_memo(self, updates: Mapping[str, Any]) -> None:
        # Converting before creating a command so that we don't leave a partial command in case of conversion failure.
        update_payloads = {}
        removals = []
        for k, v in updates.items():
            if v is None:
                # Intentionally not checking if memo exists, so that no-op removals show up in history too.
                removals.append(k)
            else:
                update_payloads[k] = (
                    self._workflow_context_payload_converter.to_payload(v)
                )

        if not update_payloads and not removals:
            return

        command = self._add_command()
        fields = command.modify_workflow_properties.upserted_memo.fields

        # Updating memo inside info by downcasting to mutable mapping.
        mut_raw_memo = cast(
            MutableMapping[str, temporalio.api.common.v1.Payload],
            self._info.raw_memo,
        )

        for k, v in update_payloads.items():
            fields[k].CopyFrom(v)
            mut_raw_memo[k] = v

        if removals:
            null_payload = self._workflow_context_payload_converter.to_payload(None)
            for k in removals:
                fields[k].CopyFrom(null_payload)
                mut_raw_memo.pop(k, None)

        # Keeping deserialized memo dict in sync, if exists
        if self._untyped_converted_memo is not None:
            for k, v in update_payloads.items():
                self._untyped_converted_memo[k] = (
                    self._workflow_context_payload_converter.from_payload(v)
                )
            for k in removals:
                self._untyped_converted_memo.pop(k, None)

    def workflow_metric_meter(self) -> temporalio.common.MetricMeter:
        # Create if not present, which means using an extern function
        if not self._metric_meter:
            metric_meter = cast(_WorkflowExternFunctions, self._extern_functions)[
                "__temporal_get_metric_meter"
            ]()
            metric_meter = metric_meter.with_additional_attributes(
                {
                    "namespace": self._info.namespace,
                    "task_queue": self._info.task_queue,
                    "workflow_type": self._info.workflow_type,
                }
            )
            self._metric_meter = _ReplaySafeMetricMeter(metric_meter)
        return self._metric_meter

    def workflow_patch(self, id: str, *, deprecated: bool) -> bool:
        self._assert_not_read_only("patch")
        # We use a previous memoized result of this if present. If this is being
        # deprecated, we can still use memoized result and skip the command.
        use_patch = self._patches_memoized.get(id)
        if use_patch is not None:
            return use_patch

        use_patch = not self._is_replaying or id in self._patches_notified
        self._patches_memoized[id] = use_patch
        if use_patch:
            command = self._add_command()
            command.set_patch_marker.patch_id = id
            command.set_patch_marker.deprecated = deprecated
        return use_patch

    def workflow_payload_converter(self) -> temporalio.converter.PayloadConverter:
        return self._workflow_context_payload_converter

    def workflow_random(self) -> random.Random:
        self._assert_not_read_only("random")
        return self._random

    def workflow_set_query_handler(
        self, name: str | None, handler: Callable | None
    ) -> None:
        self._assert_not_read_only("set query handler")
        if handler:
            if inspect.iscoroutinefunction(handler):
                warnings.warn(
                    "Queries as async def functions are deprecated",
                    DeprecationWarning,
                    stacklevel=3,
                )
            defn = temporalio.workflow._QueryDefinition(
                name=name, fn=handler, is_method=False
            )
            self._queries[name] = defn
            if defn.dynamic_vararg:
                warnings.warn(
                    "Dynamic queries with vararg third param is deprecated, use Sequence[RawValue]",
                    DeprecationWarning,
                    stacklevel=3,
                )
        else:
            self._queries.pop(name, None)

    def workflow_set_signal_handler(
        self, name: str | None, handler: Callable | None
    ) -> None:
        self._assert_not_read_only("set signal handler")
        if handler:
            defn = temporalio.workflow._SignalDefinition(
                name=name, fn=handler, is_method=False
            )
            self._signals[name] = defn
            if defn.dynamic_vararg:
                warnings.warn(
                    "Dynamic signals with vararg third param is deprecated, use Sequence[RawValue]",
                    DeprecationWarning,
                    stacklevel=3,
                )
            # We have to send buffered signals to the handler if they apply
            if name:
                for job in self._buffered_signals.pop(name, []):
                    self._process_signal_job(defn, job)
            else:
                for jobs in self._buffered_signals.values():
                    for job in jobs:
                        self._process_signal_job(defn, job)
                self._buffered_signals.clear()
        else:
            self._signals.pop(name, None)

    def workflow_set_update_handler(
        self,
        name: str | None,
        handler: Callable | None,
        validator: Callable | None,
    ) -> None:
        self._assert_not_read_only("set update handler")
        if handler:
            defn = temporalio.workflow._UpdateDefinition(
                name=name, fn=handler, is_method=False
            )
            if validator is not None:
                defn.set_validator(validator)
            self._updates[name] = defn
            if defn.dynamic_vararg:
                raise RuntimeError(
                    "Dynamic updates do not support a vararg third param, use Sequence[RawValue]",
                )
        else:
            self._updates.pop(name, None)

    def workflow_start_activity(
        self,
        activity: Any,
        *args: Any,
        task_queue: str | None,
        result_type: type | None,
        schedule_to_close_timeout: timedelta | None,
        schedule_to_start_timeout: timedelta | None,
        start_to_close_timeout: timedelta | None,
        heartbeat_timeout: timedelta | None,
        retry_policy: temporalio.common.RetryPolicy | None,
        cancellation_type: temporalio.workflow.ActivityCancellationType,
        activity_id: str | None,
        versioning_intent: temporalio.workflow.VersioningIntent | None,
        summary: str | None = None,
        priority: temporalio.common.Priority = temporalio.common.Priority.default,
    ) -> temporalio.workflow.ActivityHandle[Any]:
        self._assert_not_read_only("start activity")
        # Get activity definition if it's callable
        name: str
        arg_types: list[type] | None = None
        ret_type = result_type
        if isinstance(activity, str):
            name = activity
        elif callable(activity):
            defn = temporalio.activity._Definition.must_from_callable(activity)
            if not defn.name:
                raise ValueError("Cannot invoke dynamic activity explicitly")
            name = defn.name
            arg_types = defn.arg_types
            ret_type = defn.ret_type
        else:
            raise TypeError("Activity must be a string or callable")

        return self._outbound.start_activity(
            StartActivityInput(
                activity=name,
                args=args,
                activity_id=activity_id,
                task_queue=task_queue,
                schedule_to_close_timeout=schedule_to_close_timeout,
                schedule_to_start_timeout=schedule_to_start_timeout,
                start_to_close_timeout=start_to_close_timeout,
                heartbeat_timeout=heartbeat_timeout,
                retry_policy=retry_policy,
                cancellation_type=cancellation_type,
                headers={},
                disable_eager_execution=self._disable_eager_activity_execution,
                arg_types=arg_types,
                ret_type=ret_type,
                versioning_intent=versioning_intent,
                summary=summary,
                priority=priority,
            )
        )

    # workflow_start_child_workflow ret_type
    async def workflow_start_child_workflow(
        self,
        workflow: Any,
        *args: Any,
        id: str,
        task_queue: str | None,
        result_type: type | None,
        cancellation_type: temporalio.workflow.ChildWorkflowCancellationType,
        parent_close_policy: temporalio.workflow.ParentClosePolicy,
        execution_timeout: timedelta | None,
        run_timeout: timedelta | None,
        task_timeout: timedelta | None,
        id_reuse_policy: temporalio.common.WorkflowIDReusePolicy,
        retry_policy: temporalio.common.RetryPolicy | None,
        cron_schedule: str,
        memo: Mapping[str, Any] | None,
        search_attributes: None
        | (
            temporalio.common.SearchAttributes | temporalio.common.TypedSearchAttributes
        ),
        versioning_intent: temporalio.workflow.VersioningIntent | None,
        static_summary: str | None = None,
        static_details: str | None = None,
        priority: temporalio.common.Priority = temporalio.common.Priority.default,
    ) -> temporalio.workflow.ChildWorkflowHandle[Any, Any]:
        # Use definition if callable
        name: str
        arg_types: list[type] | None = None
        ret_type = result_type
        if isinstance(workflow, str):
            name = workflow
        elif callable(workflow):
            defn = temporalio.workflow._Definition.must_from_run_fn(workflow)  # pyright: ignore
            if not defn.name:
                raise TypeError("Cannot invoke dynamic workflow explicitly")
            name = defn.name
            arg_types = defn.arg_types
            ret_type = defn.ret_type
        else:
            raise TypeError("Workflow must be a string or callable")

        return await self._outbound.start_child_workflow(
            StartChildWorkflowInput(
                workflow=name,
                args=args,
                id=id,
                task_queue=task_queue,
                cancellation_type=cancellation_type,
                parent_close_policy=parent_close_policy,
                execution_timeout=execution_timeout,
                run_timeout=run_timeout,
                task_timeout=task_timeout,
                id_reuse_policy=id_reuse_policy,
                retry_policy=retry_policy,
                cron_schedule=cron_schedule,
                memo=memo,
                search_attributes=search_attributes,
                headers={},
                arg_types=arg_types,
                ret_type=ret_type,
                versioning_intent=versioning_intent,
                static_summary=static_summary,
                static_details=static_details,
                priority=priority,
            )
        )

    def workflow_start_local_activity(
        self,
        activity: Any,
        *args: Any,
        result_type: type | None,
        schedule_to_close_timeout: timedelta | None,
        schedule_to_start_timeout: timedelta | None,
        start_to_close_timeout: timedelta | None,
        retry_policy: temporalio.common.RetryPolicy | None,
        local_retry_threshold: timedelta | None,
        cancellation_type: temporalio.workflow.ActivityCancellationType,
        activity_id: str | None,
        summary: str | None,
    ) -> temporalio.workflow.ActivityHandle[Any]:
        # Get activity definition if it's callable
        name: str
        arg_types: list[type] | None = None
        ret_type = result_type
        if isinstance(activity, str):
            name = activity
        elif callable(activity):
            defn = temporalio.activity._Definition.must_from_callable(activity)
            if not defn.name:
                raise ValueError("Cannot invoke dynamic activity explicitly")
            name = defn.name
            arg_types = defn.arg_types
            ret_type = defn.ret_type
        else:
            raise TypeError("Activity must be a string or callable")

        cast(_WorkflowExternFunctions, self._extern_functions)[
            "__temporal_assert_local_activity_valid"
        ](name)

        return self._outbound.start_local_activity(
            StartLocalActivityInput(
                activity=name,
                args=args,
                activity_id=activity_id,
                schedule_to_close_timeout=schedule_to_close_timeout,
                schedule_to_start_timeout=schedule_to_start_timeout,
                start_to_close_timeout=start_to_close_timeout,
                retry_policy=retry_policy,
                local_retry_threshold=local_retry_threshold,
                cancellation_type=cancellation_type,
                summary=summary,
                headers={},
                arg_types=arg_types,
                ret_type=ret_type,
            )
        )

    async def workflow_start_nexus_operation(
        self,
        endpoint: str,
        service: str,
        operation: nexusrpc.Operation[InputT, OutputT] | str | Callable[..., Any],
        input: Any,
        output_type: type[OutputT] | None,
        schedule_to_close_timeout: timedelta | None,
        schedule_to_start_timeout: timedelta | None,
        start_to_close_timeout: timedelta | None,
        cancellation_type: temporalio.workflow.NexusOperationCancellationType,
        headers: Mapping[str, str] | None,
        summary: str | None,
    ) -> temporalio.workflow.NexusOperationHandle[OutputT]:
        # start_nexus_operation
        return await self._outbound.start_nexus_operation(
            StartNexusOperationInput(
                endpoint=endpoint,
                service=service,
                operation=operation,
                input=input,
                output_type=output_type,
                schedule_to_close_timeout=schedule_to_close_timeout,
                schedule_to_start_timeout=schedule_to_start_timeout,
                start_to_close_timeout=start_to_close_timeout,
                cancellation_type=cancellation_type,
                headers=headers,
                summary=summary,
            )
        )

    def workflow_time_ns(self) -> int:
        return self._time_ns

    def workflow_upsert_search_attributes(
        self,
        attributes: (
            temporalio.common.SearchAttributes
            | Sequence[temporalio.common.SearchAttributeUpdate]
        ),
    ) -> None:
        v = self._add_command().upsert_workflow_search_attributes

        # Update the attrs on info, casting to their mutable forms first
        mut_attrs = cast(
            MutableMapping[str, temporalio.common.SearchAttributeValues],
            self._info.search_attributes,
        )
        mut_typed_attrs = cast(
            list[temporalio.common.SearchAttributePair],
            self._info.typed_search_attributes.search_attributes,
        )

        # Do different things for untyped (i.e. Mapping) and typed
        if isinstance(attributes, Mapping):
            # Update the keys in the existing dictionary. We keep exact values
            # sent in instead of any kind of normalization. This means empty
            # lists remain as empty lists which matches what the server does.
            mut_attrs.update(attributes)
            for k, vals in attributes.items():
                # Add to command
                v.search_attributes.indexed_fields[k].CopyFrom(
                    temporalio.converter.encode_search_attribute_values(vals)
                )

                # To apply to typed search attributes we remove, replace, or add. We
                # don't know any of the key types, so we do our best.
                index = next(
                    (i for i, a in enumerate(mut_typed_attrs) if a.key.name == k), None
                )
                if not vals:
                    if index is not None:
                        del mut_typed_attrs[index]
                else:
                    # Attempt to derive type the value and ignore if we can't.
                    # No need to warn because we're already warning on using
                    # this deprecated form.
                    key = (
                        temporalio.common.SearchAttributeKey._guess_from_untyped_values(
                            k, vals
                        )
                    )
                    if key:
                        val: Any = vals
                        # If the key is not a keyword list, we only support
                        # single item lists
                        if (
                            key.indexed_value_type
                            != temporalio.common.SearchAttributeIndexedValueType.KEYWORD_LIST
                        ):
                            if len(vals) != 1:
                                continue
                            val = vals[0]
                        pair = temporalio.common.SearchAttributePair(key, val)
                        if index is None:
                            mut_typed_attrs.append(pair)
                        else:
                            mut_typed_attrs[index] = pair
        else:
            # Update typed and untyped keys, replacing typed as needed
            for update in attributes:
                # Set on command (delete is a proper null)
                v.search_attributes.indexed_fields[update.key.name].CopyFrom(
                    temporalio.converter.encode_typed_search_attribute_value(
                        update.key, update.value
                    )
                )

                # Update typed and untyped in info
                index = next(
                    (
                        i
                        for i, a in enumerate(mut_typed_attrs)
                        if a.key.name == update.key.name
                    ),
                    None,
                )
                if update.value is None:
                    # Delete
                    if index is not None:
                        del mut_typed_attrs[index]
                    # Just empty-list the untyped one
                    mut_attrs[update.key.name] = cast(list[str], [])
                else:
                    # Update
                    pair = temporalio.common.SearchAttributePair(
                        update.key, update.value
                    )
                    if index is None:
                        mut_typed_attrs.append(pair)
                    else:
                        mut_typed_attrs[index] = pair
                    # Single-item list if not already a sequence for untyped
                    mut_attrs[update.key.name] = (
                        list(update.value)
                        if update.key.indexed_value_type
                        == temporalio.common.SearchAttributeIndexedValueType.KEYWORD_LIST
                        else [update.value]
                    )

    async def workflow_sleep(
        self, duration: float, *, summary: str | None = None
    ) -> None:
        user_metadata = (
            temporalio.api.sdk.v1.UserMetadata(
                summary=self._workflow_context_payload_converter.to_payload(summary)
            )
            if summary
            else None
        )
        fut = self.create_future()
        timer_handle = self._timer_impl(
            duration,
            _TimerOptions(user_metadata=user_metadata),
            lambda: fut.set_result(None) if not fut.done() else None,
        )
        fut.add_done_callback(
            lambda f: timer_handle.cancel() if f.cancelled() else None
        )
        await fut

    async def workflow_wait_condition(
        self,
        fn: Callable[[], bool],
        *,
        timeout: float | None = None,
        timeout_summary: str | None = None,
    ) -> None:
        self._assert_not_read_only("wait condition")
        fut = self.create_future()
        self._conditions.append((fn, fut))
        user_metadata = (
            temporalio.api.sdk.v1.UserMetadata(
                summary=self._workflow_context_payload_converter.to_payload(
                    timeout_summary
                )
            )
            if timeout_summary
            else None
        )
        ctxvars = contextvars.copy_context()

        async def in_context():
            _TimerOptionsCtxVar.set(_TimerOptions(user_metadata=user_metadata))
            await asyncio.wait_for(fut, timeout)

        await ctxvars.run(in_context)

    def workflow_get_current_details(self) -> str:
        return self._current_details

    def workflow_set_current_details(self, details: str):
        self._assert_not_read_only("set current details")
        self._current_details = details

    def workflow_is_failure_exception(self, err: BaseException) -> bool:
        # An exception causes the workflow to fail (rather than the task) if it
        # is already a failure error, a timeout error, or an instance of any of the
        # failure exception types configured at the worker or workflow level.
        wf_failure_exception_types = self._defn.failure_exception_types
        if self._dynamic_failure_exception_types is not None:
            wf_failure_exception_types = self._dynamic_failure_exception_types
        return (
            isinstance(err, temporalio.exceptions.FailureError)
            or isinstance(err, asyncio.TimeoutError)
            or any(isinstance(err, typ) for typ in wf_failure_exception_types)
            or any(
                isinstance(err, typ)
                for typ in self._worker_level_failure_exception_types
            )
        )

    def workflow_has_last_completion_result(self) -> bool:
        return len(self._last_completion_result.payloads) > 0

    def workflow_last_completion_result(self, type_hint: type | None) -> Any | None:
        if len(self._last_completion_result.payloads) == 0:
            return None
        elif len(self._last_completion_result.payloads) > 1:
            warnings.warn(
                f"Expected single last completion result, got {len(self._last_completion_result.payloads)}"
            )
            return None

        if type_hint is None:
            return self._workflow_context_payload_converter.from_payload(
                self._last_completion_result.payloads[0]
            )
        else:
            return self._workflow_context_payload_converter.from_payload(
                self._last_completion_result.payloads[0], type_hint
            )

    def workflow_last_failure(self) -> BaseException | None:
        if self._last_failure:
            return self._workflow_context_failure_converter.from_failure(
                self._last_failure, self._workflow_context_payload_converter
            )

        return None

    def workflow_random_seed(self) -> int:
        return self._current_seed

    def workflow_register_random_seed_callback(
        self, callback: Callable[[int], None]
    ) -> None:
        self._seed_callbacks.append(callback)

    #### Calls from outbound impl ####
    # These are in alphabetical order and all start with "_outbound_".

    def _outbound_continue_as_new(self, input: ContinueAsNewInput) -> NoReturn:
        raise _ContinueAsNewError(self, input)

    def _outbound_schedule_activity(
        self,
        input: StartActivityInput | StartLocalActivityInput,
    ) -> _ActivityHandle:
        # Validate
        if not input.start_to_close_timeout and not input.schedule_to_close_timeout:
            raise ValueError(
                "Activity must have start_to_close_timeout or schedule_to_close_timeout"
            )

        handle: _ActivityHandle

        # Function that runs in the handle
        async def run_activity() -> Any:
            while True:
                # Mark it as started each loop because backoff could cause it to
                # be marked as unstarted
                handle._started = True
                try:
                    # We have to shield because we don't want the underlying
                    # result future to be cancelled
                    return await asyncio.shield(handle._result_fut)
                except _ActivityDoBackoffError as err:
                    # We have to sleep then reschedule. Note this sleep can be
                    # cancelled like any other timer.
                    await asyncio.sleep(
                        err.backoff.backoff_duration.ToTimedelta().total_seconds()
                    )
                    handle._apply_schedule_command(err.backoff)
                    # We have to put the handle back on the pending activity
                    # dict with its new seq
                    self._pending_activities[handle._seq] = handle
                except asyncio.CancelledError:
                    # If an activity future completes at the same time as a cancellation is being processed, the cancellation would be swallowed
                    # _WorkflowLogicFlag.RAISE_ON_CANCELLING_COMPLETED_ACTIVITY will correctly reraise the exception
                    if handle._result_fut.done():
                        if (
                            not self._is_replaying
                            or _WorkflowLogicFlag.RAISE_ON_CANCELLING_COMPLETED_ACTIVITY
                            in self._current_internal_flags
                        ):
                            self._current_completion.successful.used_internal_flags.append(
                                _WorkflowLogicFlag.RAISE_ON_CANCELLING_COMPLETED_ACTIVITY
                            )
                            raise
                    # Send a cancel request to the activity
                    handle._apply_cancel_command(self._add_command())
                    # Clear the cancellation counter on Python 3.11+ so the
                    # next await does not immediately re-raise CancelledError
                    if (
                        sys.version_info >= (3, 11)
                        and (t := asyncio.current_task()) is not None
                    ):
                        t.uncancel()  # type: ignore[union-attr]

        # Create the handle and set as pending
        handle = _ActivityHandle(self, input, run_activity())
        handle._apply_schedule_command()
        self._pending_activities[handle._seq] = handle
        return handle

    async def _outbound_signal_child_workflow(
        self, input: SignalChildWorkflowInput
    ) -> None:
        payload_converter = self._payload_converter_with_context(
            temporalio.converter.WorkflowSerializationContext(
                namespace=self._info.namespace,
                workflow_id=input.child_workflow_id,
            )
        )
        payloads = payload_converter.to_payloads(input.args) if input.args else None
        command = self._add_command()
        v = command.signal_external_workflow_execution
        v.child_workflow_id = input.child_workflow_id
        v.signal_name = input.signal
        if payloads:
            v.args.extend(payloads)
        if input.headers:
            temporalio.common._apply_headers(input.headers, v.headers)
        await self._signal_external_workflow(command)

    async def _outbound_signal_external_workflow(
        self, input: SignalExternalWorkflowInput
    ) -> None:
        payload_converter = self._payload_converter_with_context(
            temporalio.converter.WorkflowSerializationContext(
                namespace=input.namespace,
                workflow_id=input.workflow_id,
            )
        )
        payloads = payload_converter.to_payloads(input.args) if input.args else None
        command = self._add_command()
        v = command.signal_external_workflow_execution
        v.workflow_execution.namespace = input.namespace
        v.workflow_execution.workflow_id = input.workflow_id
        if input.workflow_run_id:
            v.workflow_execution.run_id = input.workflow_run_id
        v.signal_name = input.signal
        if payloads:
            v.args.extend(payloads)
        if input.headers:
            temporalio.common._apply_headers(input.headers, v.headers)
        await self._signal_external_workflow(command)

    async def _outbound_start_child_workflow(
        self, input: StartChildWorkflowInput
    ) -> _ChildWorkflowHandle:
        handle: _ChildWorkflowHandle

        # Common code for handling cancel for start and run
        def apply_child_cancel_error() -> None:
            # Send a cancel request to the child
            cancel_command = self._add_command()
            handle._apply_cancel_command(cancel_command)
            # If the cancel command is for external workflow, we
            # have to add a seq and mark it pending
            if cancel_command.HasField("request_cancel_external_workflow_execution"):
                cancel_seq = self._next_seq("external_cancel")
                cancel_command.request_cancel_external_workflow_execution.seq = (
                    cancel_seq
                )
                # TODO(cretz): Nothing waits on this future, so how
                # if at all should we report child-workflow cancel
                # request failure?
                self._pending_external_cancels[cancel_seq] = (
                    self.create_future(),
                    input.id,
                )

        # Function that runs in the handle
        async def run_child() -> Any:
            while True:
                try:
                    # We have to shield because we don't want the future itself
                    # to be cancelled
                    return await asyncio.shield(handle._result_fut)
                except asyncio.CancelledError:
                    apply_child_cancel_error()
                    # Clear the cancellation counter on Python 3.11+ so the
                    # next await does not immediately re-raise CancelledError
                    if (
                        sys.version_info >= (3, 11)
                        and (t := asyncio.current_task()) is not None
                    ):
                        t.uncancel()  # type: ignore[union-attr]

        # Create the handle and set as pending
        handle = _ChildWorkflowHandle(
            self, self._next_seq("child_workflow"), input, run_child()
        )
        handle._apply_start_command()
        self._pending_child_workflows[handle._seq] = handle

        # Wait on start before returning
        while True:
            try:
                # We have to shield because we don't want the future itself
                # to be cancelled
                await asyncio.shield(handle._start_fut)
                return handle
            except asyncio.CancelledError:
                apply_child_cancel_error()
                # Clear the cancellation counter on Python 3.11+ so the
                # next await does not immediately re-raise CancelledError
                if (
                    sys.version_info >= (3, 11)
                    and (t := asyncio.current_task()) is not None
                ):
                    t.uncancel()  # type: ignore[union-attr]
                if self._cancel_requested:
                    raise

    async def _outbound_start_nexus_operation(
        self, input: StartNexusOperationInput[Any, OutputT]
    ) -> _NexusOperationHandle[OutputT]:
        # A Nexus operation handle contains two futures: self._start_fut is resolved as a result of
        # the Nexus operation starting (activation job: resolve_nexus_operation_start), and
        # self._result_fut is resolved as a result of the Nexus operation completing (activation
        # job: resolve_nexus_operation). The handle itself corresponds to an asyncio.Task which
        # waits on self.result_fut, handling CancelledError by emitting a
        # RequestCancelNexusOperation command. We do not return the handle until we receive
        # resolve_nexus_operation_start, like ChildWorkflowHandle and unlike ActivityHandle. Note
        # that a Nexus operation may complete synchronously (in which case both jobs will be sent in
        # the same activation, and start will be resolved without an operation token), or
        # asynchronously (in which case they may be sent in separate activations, and start will be
        # resolved with an operation token). See comments in tests/worker/test_nexus.py for worked
        # examples of the evolution of the resulting handle state machine in the sync and async
        # Nexus response cases.
        handle: _NexusOperationHandle[OutputT]

        async def operation_handle_fn() -> OutputT:
            while True:
                try:
                    return cast(OutputT, await asyncio.shield(handle._result_fut))
                except asyncio.CancelledError:
                    cancel_command = self._add_command()
                    handle._apply_cancel_command(cancel_command)
                    # Clear the cancellation counter on Python 3.11+ so the
                    # next await does not immediately re-raise CancelledError
                    if (
                        sys.version_info >= (3, 11)
                        and (t := asyncio.current_task()) is not None
                    ):
                        t.uncancel()  # type: ignore[union-attr]

        handle = _NexusOperationHandle(
            self, self._next_seq("nexus_operation"), input, operation_handle_fn()
        )
        handle._apply_schedule_command()
        self._pending_nexus_operations[handle._seq] = handle

        while True:
            try:
                await asyncio.shield(handle._start_fut)
                return handle
            except asyncio.CancelledError:
                cancel_command = self._add_command()
                handle._apply_cancel_command(cancel_command)
                # Clear the cancellation counter on Python 3.11+ so the
                # next await does not immediately re-raise CancelledError
                if (
                    sys.version_info >= (3, 11)
                    and (t := asyncio.current_task()) is not None
                ):
                    t.uncancel()  # type: ignore[union-attr]
                if self._cancel_requested:
                    raise

    #### Miscellaneous helpers ####
    # These are in alphabetical order.

    def _add_command(self) -> temporalio.bridge.proto.workflow_commands.WorkflowCommand:
        self._assert_not_read_only("add command")
        return self._current_completion.successful.commands.add()

    @contextmanager
    def _as_read_only(self, *, in_query_or_validator: bool) -> Iterator[None]:
        prev_read_only = self._read_only
        prev_in_query_or_validator = self._in_query_or_validator
        self._read_only = True
        self._in_query_or_validator = in_query_or_validator
        try:
            yield None
        finally:
            self._read_only = prev_read_only
            self._in_query_or_validator = prev_in_query_or_validator

    def _assert_not_read_only(
        self, action_attempted: str, *, allow_during_delete: bool = False
    ) -> None:
        if self._deleting and not allow_during_delete:
            raise _WorkflowBeingEvictedError(
                f"Ignoring {action_attempted} while evicting workflow. This is not an error."
            )
        if self._read_only:
            raise temporalio.workflow.ReadOnlyContextError(
                f"While in read-only function, action attempted: {action_attempted}"
            )

    async def _cancel_external_workflow(
        self,
        # Should not have seq set
        command: temporalio.bridge.proto.workflow_commands.WorkflowCommand,
    ) -> None:
        seq = self._next_seq("external_cancel")
        done_fut = self.create_future()
        command.request_cancel_external_workflow_execution.seq = seq

        # Set as pending with the target workflow ID for later context use
        target_workflow_id = command.request_cancel_external_workflow_execution.workflow_execution.workflow_id
        self._pending_external_cancels[seq] = (done_fut, target_workflow_id)

        # Wait until done (there is no cancelling a cancel request)
        await done_fut

    def _check_condition(self, fn: Callable[[], bool], fut: asyncio.Future) -> bool:
        if fn():
            fut.set_result(True)
            return True
        return False

    def _convert_payloads(
        self,
        payloads: Sequence[temporalio.api.common.v1.Payload],
        types: list[type] | None,
        payload_converter: temporalio.converter.PayloadConverter,
    ) -> list[Any]:
        if not payloads:
            return []
        # Only use type hints if they match count
        if types and len(types) != len(payloads):
            types = None
        try:
            return payload_converter.from_payloads(payloads, type_hints=types)
        except temporalio.exceptions.FailureError:
            # Don't wrap payload conversion errors that would fail the workflow
            raise
        except Exception as err:
            if self.workflow_is_failure_exception(err):
                raise
            raise RuntimeError("Failed decoding arguments") from err

    def _payload_converter_with_context(
        self,
        context: temporalio.converter.SerializationContext,
    ) -> temporalio.converter.PayloadConverter:
        """Construct workflow payload converter with the given context.

        This plays a similar role to DataConverter._with_context, but operates on PayloadConverter
        only (payload encoding/decoding is done by the worker, outside the workflow sandbox).
        """
        payload_converter = self._context_free_payload_converter
        if isinstance(payload_converter, temporalio.converter.WithSerializationContext):
            payload_converter = payload_converter.with_context(context)
        return payload_converter

    def _failure_converter_with_context(
        self,
        context: temporalio.converter.SerializationContext,
    ) -> temporalio.converter.FailureConverter:
        """Construct workflow failure converter with the given context.

        This plays a similar role to DataConverter._with_context, but operates on FailureConverter
        only (payload encoding/decoding is done by the worker, outside the workflow sandbox).
        """
        failure_converter = self._context_free_failure_converter
        if isinstance(failure_converter, temporalio.converter.WithSerializationContext):
            failure_converter = failure_converter.with_context(context)
        return failure_converter

    def get_serialization_context(
        self,
        command_info: _command_aware_visitor.CommandInfo | None,
    ) -> temporalio.converter.SerializationContext | None:
        if command_info is None:
            # Use payload codec with workflow context by default (i.e. for payloads not associated
            # with a pending command)
            return temporalio.converter.WorkflowSerializationContext(
                namespace=self._info.namespace,
                workflow_id=self._info.workflow_id,
            )

        if (
            command_info.command_type
            == temporalio.api.enums.v1.command_type_pb2.CommandType.COMMAND_TYPE_SCHEDULE_ACTIVITY_TASK
            and command_info.command_seq in self._pending_activities
        ):
            # Use the activity's context
            activity_handle = self._pending_activities[command_info.command_seq]
            return temporalio.converter.ActivitySerializationContext(
                namespace=self._info.namespace,
                workflow_id=self._info.workflow_id,
                workflow_type=self._info.workflow_type,
                activity_type=activity_handle._input.activity,
                activity_id=activity_handle._input.activity_id,
                activity_task_queue=(
                    activity_handle._input.task_queue
                    if isinstance(activity_handle._input, StartActivityInput)
                    and activity_handle._input.task_queue
                    else self._info.task_queue
                ),
                is_local=isinstance(activity_handle._input, StartLocalActivityInput),
            )

        elif (
            command_info.command_type
            == temporalio.api.enums.v1.command_type_pb2.CommandType.COMMAND_TYPE_START_CHILD_WORKFLOW_EXECUTION
            and command_info.command_seq in self._pending_child_workflows
        ):
            # Use the child workflow's context
            child_wf_handle = self._pending_child_workflows[command_info.command_seq]
            return temporalio.converter.WorkflowSerializationContext(
                namespace=self._info.namespace,
                workflow_id=child_wf_handle._input.id,
            )

        elif (
            command_info.command_type
            == temporalio.api.enums.v1.command_type_pb2.CommandType.COMMAND_TYPE_SIGNAL_EXTERNAL_WORKFLOW_EXECUTION
            and command_info.command_seq in self._pending_external_signals
        ):
            # Use the target workflow's context
            _, target_workflow_id = self._pending_external_signals[
                command_info.command_seq
            ]
            return temporalio.converter.WorkflowSerializationContext(
                namespace=self._info.namespace,
                workflow_id=target_workflow_id,
            )

        elif (
            command_info.command_type
            == temporalio.api.enums.v1.command_type_pb2.CommandType.COMMAND_TYPE_SCHEDULE_NEXUS_OPERATION
            and command_info.command_seq in self._pending_nexus_operations
        ):
            # Use empty context for nexus operations: users will never want to encrypt using a
            # key derived from caller workflow context because the caller workflow context is
            # not available on the handler side for decryption.
            return None

        else:
            # Use payload codec with workflow context for all other payloads
            return temporalio.converter.WorkflowSerializationContext(
                namespace=self._info.namespace,
                workflow_id=self._info.workflow_id,
            )

    def get_external_store_context(
        self,
        command_info: _command_aware_visitor.CommandInfo | None,
    ) -> StorageDriverStoreContext:
        # The current workflow is the default target for external store
        # operations. For commands that target other workflows, those workflows
        # are the target for that command's external store operation. For
        # workflow activities, the target is the current workflow since the
        # activity is bound to the lifetime of the current workflow, the
        # activity run information is the same as the current workflow, and
        # successfully completed activities are not involved in replay.
        # Otherwise, the storage space for a given workflow would be disparate
        # if stored under activity information.
        current_wf = StorageDriverWorkflowInfo(
            id=self._info.workflow_id,
            run_id=self._info.run_id,
            type=self._info.workflow_type,
            namespace=self._info.namespace,
        )

        if command_info is None:
            return StorageDriverStoreContext(target=current_wf)

        COMMAND_TYPE = temporalio.api.enums.v1.command_type_pb2.CommandType

        if (
            command_info.command_type
            == COMMAND_TYPE.COMMAND_TYPE_START_CHILD_WORKFLOW_EXECUTION
            and command_info.command_seq in self._pending_child_workflows
        ):
            child = self._pending_child_workflows[command_info.command_seq]
            return StorageDriverStoreContext(
                target=StorageDriverWorkflowInfo(
                    id=child._input.id,
                    type=child._input.workflow,
                    namespace=self._info.namespace,
                ),
            )

        elif (
            command_info.command_type
            == COMMAND_TYPE.COMMAND_TYPE_SIGNAL_EXTERNAL_WORKFLOW_EXECUTION
            and command_info.command_seq in self._pending_external_signals
        ):
            _, target_id = self._pending_external_signals[command_info.command_seq]
            return StorageDriverStoreContext(
                target=StorageDriverWorkflowInfo(
                    id=target_id, namespace=self._info.namespace
                ),
            )

        elif (
            command_info.command_type
            == COMMAND_TYPE.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION
            and self._info.parent is not None
            and self._info.continued_run_id is None
        ):
            return StorageDriverStoreContext(
                target=StorageDriverWorkflowInfo(
                    id=self._info.parent.workflow_id,
                    run_id=self._info.parent.run_id,
                    namespace=self._info.parent.namespace,
                ),
            )

        else:
            return StorageDriverStoreContext(target=current_wf)

    def _instantiate_workflow_object(self) -> Any:
        if not self._workflow_input:
            raise RuntimeError("Expected workflow input. This is a Python SDK bug.")

        if hasattr(self._defn.cls.__init__, "__temporal_workflow_init"):  # type:ignore[misc]
            workflow_instance = self._defn.cls(*self._workflow_input.args)
        else:
            workflow_instance = self._defn.cls()

        if self._defn.versioning_behavior:
            self._versioning_behavior = self._defn.versioning_behavior
        # If there's a dynamic config function, call it now after we've instantiated the object
        # but before we start executing the workflow
        if self._defn.name is None and self._defn.dynamic_config_fn is not None:
            dynamic_config = None
            try:
                with self._as_read_only(in_query_or_validator=False):
                    dynamic_config = self._defn.dynamic_config_fn(workflow_instance)
            except Exception as err:
                logger.exception(
                    f"Failed to run dynamic config function in workflow {self._info.workflow_type}"
                )
                # Treat as a task failure
                self._current_activation_error = err
                raise self._current_activation_error

            if dynamic_config:
                if dynamic_config.failure_exception_types is not None:
                    self._dynamic_failure_exception_types = (
                        dynamic_config.failure_exception_types
                    )
                if (
                    dynamic_config.versioning_behavior
                    != temporalio.common.VersioningBehavior.UNSPECIFIED
                ):
                    self._versioning_behavior = dynamic_config.versioning_behavior

        return workflow_instance

    def _warn_if_unfinished_handlers(self) -> None:
        def warnable(handler_executions: Iterable[HandlerExecution]):
            return [
                ex
                for ex in handler_executions
                if ex.unfinished_policy
                == temporalio.workflow.HandlerUnfinishedPolicy.WARN_AND_ABANDON
            ]

        warnable_updates = warnable(self._in_progress_updates.values())
        if warnable_updates:
            warnings.warn(
                temporalio.workflow.UnfinishedUpdateHandlersWarning(
                    _make_unfinished_update_handler_message(warnable_updates)
                )
            )

        warnable_signals = warnable(self._in_progress_signals.values())
        if warnable_signals:
            warnings.warn(
                temporalio.workflow.UnfinishedSignalHandlersWarning(
                    _make_unfinished_signal_handler_message(warnable_signals)
                )
            )

    def _next_seq(self, type: str) -> int:
        seq = self._curr_seqs.get(type, 0) + 1
        self._curr_seqs[type] = seq
        return seq

    def _process_handler_args(
        self,
        job_name: str,
        job_input: Sequence[temporalio.api.common.v1.Payload],
        defn_name: str | None,
        defn_arg_types: list[type] | None,
        defn_dynamic_vararg: bool,
    ) -> list[Any]:
        # If dynamic old-style vararg, args become name + varargs of given arg
        # types. If dynamic new-style raw value sequence, args become name +
        # seq of raw values.
        if not defn_name and defn_dynamic_vararg:
            # Take off the string type hint for conversion
            arg_types = defn_arg_types[1:] if defn_arg_types else None
            return [job_name] + self._convert_payloads(
                job_input, arg_types, self._workflow_context_payload_converter
            )
        if not defn_name:
            return [
                job_name,
                self._convert_payloads(
                    job_input,
                    [temporalio.common.RawValue] * len(job_input),
                    self._workflow_context_payload_converter,
                ),
            ]
        return self._convert_payloads(
            job_input, defn_arg_types, self._workflow_context_payload_converter
        )

    def _process_signal_job(
        self,
        defn: temporalio.workflow._SignalDefinition,
        job: temporalio.bridge.proto.workflow_activation.SignalWorkflow,
    ) -> None:
        try:
            args = self._process_handler_args(
                job.signal_name,
                job.input,
                defn.name,
                defn.arg_types,
                defn.dynamic_vararg,
            )
        except Exception:
            logger.exception(
                f"Failed deserializing signal input for {job.signal_name}"
                f" on workflow {self._info.workflow_type} with ID {self._info.workflow_id}"
                f" and run ID {self._info.run_id}, dropping the signal",
                extra={"temporal_workflow": self._info._logger_details()},
            )
            return
        input = HandleSignalInput(
            signal=job.signal_name, args=args, headers=job.headers
        )

        self._handled_signals_seq += 1
        id = self._handled_signals_seq
        self._in_progress_signals[id] = HandlerExecution(
            job.signal_name, defn.unfinished_policy
        )

        def done_callback(_f: Any):
            self._in_progress_signals.pop(id, None)

        task = self.create_task(
            self._run_top_level_workflow_function(self._inbound.handle_signal(input)),
            name=f"signal: {job.signal_name}",
        )
        task.add_done_callback(done_callback)

    def _register_task(
        self,
        task: asyncio.Task,
        *,
        name: str | None,
    ) -> None:
        self._assert_not_read_only("create task")
        # Name not supported on older Python versions
        # Put the workflow info at the end of the task name
        name = name or task.get_name()
        name += f" (workflow: {self._info.workflow_type}, id: {self._info.workflow_id}, run: {self._info.run_id})"
        task.set_name(name)
        # Add to and remove from our own non-weak set instead of relying on
        # Python's weak set which can collect these too early
        self._tasks.add(task)
        task.add_done_callback(self._tasks.remove)
        # When the workflow is GC'd (for whatever reason, e.g. eviction or
        # worker shutdown), still-open tasks get a message logged to the
        # exception handler about still pending. We disable this if the
        # attribute is available.
        if hasattr(task, "_log_destroy_pending"):
            setattr(task, "_log_destroy_pending", False)

    def _run_once(self, *, check_conditions: bool) -> None:
        try:
            asyncio._set_running_loop(self)

            # We instantiate the workflow class _inside_ here because __init__
            # needs to run with this event loop set. If we're deleting and
            # we've never initialized, we don't need to bother.
            if not self._object and not self._deleting:
                self._object = self._instantiate_workflow_object()

            # Run while there is anything ready
            while self._ready:
                # Run and remove all ready ones
                while self._ready:
                    handle = self._ready.popleft()
                    handle._run()

                    # Must throw here if not deleting. Only really set inside
                    # _run_top_level_workflow_function.
                    if self._current_activation_error and not self._deleting:
                        raise self._current_activation_error

                # Check conditions which may add to the ready list. Also remove
                # conditions whose futures have already cancelled (e.g. when
                # timed out).
                if check_conditions:
                    self._conditions[:] = [
                        t
                        for t in self._conditions
                        if not t[1].done() and not self._check_condition(*t)
                    ]
        finally:
            asyncio._set_running_loop(None)

    # This is used for the primary workflow function and signal handlers in
    # order to apply common exception handling to each
    async def _run_top_level_workflow_function(self, coro: Awaitable[None]) -> None:
        try:
            await coro
        except _ContinueAsNewError as err:
            logger.debug("Workflow requested continue as new")
            err._apply_command()
        except (Exception, asyncio.CancelledError) as err:
            # During tear down we can ignore exceptions. Technically the
            # command-adding done later would throw a not-in-workflow exception
            # we'd ignore later, but it's better to preempt it
            if self._deleting:
                if LOG_IGNORE_DURING_DELETE:
                    logger.debug(
                        "Ignoring exception while deleting workflow", exc_info=True
                    )
                return

            # Handle continue as new
            if isinstance(err, _ContinueAsNewError):
                logger.debug("Workflow requested continue as new")
                err._apply_command()
                return

            logger.debug(
                f"Workflow raised failure with run ID {self._info.run_id}",
                exc_info=True,
            )

            # All asyncio cancelled errors become Temporal cancelled errors
            if isinstance(err, asyncio.CancelledError):
                err = temporalio.exceptions.CancelledError(str(err))

            # If a cancel was ever requested and this is a cancellation, or an
            # activity/child cancellation, we add a cancel command. Technically
            # this means that a swallowed cancel followed by, say, an activity
            # cancel later on will show the workflow as cancelled. But this is
            # a Temporal limitation in that cancellation is a state not an
            # event.
            if self._cancel_requested and temporalio.exceptions.is_cancelled_exception(
                err
            ):
                self._add_command().cancel_workflow_execution.SetInParent()
            elif self.workflow_is_failure_exception(err):
                # All other failure errors fail the workflow
                self._set_workflow_failure(err)
            else:
                # All other exceptions fail the task
                self._current_activation_error = err

    def _set_workflow_failure(self, err: BaseException) -> None:
        # All other failure errors fail the workflow
        failure = self._add_command().fail_workflow_execution.failure
        failure.SetInParent()
        try:
            self._workflow_context_failure_converter.to_failure(
                err, self._workflow_context_payload_converter, failure
            )
        except Exception as inner_err:
            raise ValueError("Failed converting workflow exception") from inner_err

    async def _signal_external_workflow(
        self,
        # Should not have seq set
        command: temporalio.bridge.proto.workflow_commands.WorkflowCommand,
    ) -> None:
        seq = self._next_seq("external_signal")
        done_fut = self.create_future()
        command.signal_external_workflow_execution.seq = seq

        target_workflow_id = (
            command.signal_external_workflow_execution.child_workflow_id
            or command.signal_external_workflow_execution.workflow_execution.workflow_id
        )
        self._pending_external_signals[seq] = (done_fut, target_workflow_id)

        # Wait until completed or cancelled
        while True:
            try:
                # We have to shield because we don't want the future itself
                # to be cancelled
                return await asyncio.shield(done_fut)
            except asyncio.CancelledError:
                cancel_command = self._add_command()
                cancel_command.cancel_signal_workflow.seq = seq
                # Clear the cancellation counter on Python 3.11+ so the
                # next await does not immediately re-raise CancelledError
                if (
                    sys.version_info >= (3, 11)
                    and (t := asyncio.current_task()) is not None
                ):
                    t.uncancel()  # type: ignore[union-attr]

    def _stack_trace(self) -> str:
        stacks = []
        for task in list(self._tasks):
            # TODO(cretz): These stacks are not very clean currently
            frames = []
            for frame in task.get_stack():
                frames.append(
                    traceback.FrameSummary(
                        frame.f_code.co_filename, frame.f_lineno, frame.f_code.co_name
                    )
                )
            stacks.append(
                f"Stack for {task!r} (most recent call last):\n"
                + "\n".join(traceback.format_list(frames))
            )
        return "\n\n".join(stacks)

    def _enhanced_stack_trace(self) -> temporalio.api.sdk.v1.EnhancedStackTrace:
        sdk = temporalio.api.sdk.v1.StackTraceSDKInfo(
            name="sdk-python", version=__version__
        )

        # this is to use `open`
        with temporalio.workflow.unsafe.sandbox_unrestricted():
            sources: dict[str, temporalio.api.sdk.v1.StackTraceFileSlice] = dict()
            stacks: list[temporalio.api.sdk.v1.StackTrace] = []

            # future TODO
            # site package filter list -- we want to filter out traces from Python's internals and our sdk's internals. This is what `internal_code` is for, but right now it's just set to false.

            for task in list(self._tasks):
                locations: list[temporalio.api.sdk.v1.StackTraceFileLocation] = []

                for frame in task.get_stack():
                    filename = frame.f_code.co_filename
                    line_number = frame.f_lineno
                    func_name = frame.f_code.co_name

                    if filename not in sources.keys():
                        try:
                            with open(filename) as f:
                                code = f.read()
                        except OSError as ose:
                            code = f"Cannot access code.\n---\n{ose.strerror}"
                            # TODO possibly include sentinel/property for success of src scrape? work out with ui
                        except Exception:
                            code = f"Generic Error.\n\n{traceback.format_exc()}"

                        file_slice = temporalio.api.sdk.v1.StackTraceFileSlice(
                            line_offset=0, content=code
                        )

                        sources[filename] = file_slice

                    file_location = temporalio.api.sdk.v1.StackTraceFileLocation(
                        file_path=filename,
                        line=line_number,
                        column=-1,
                        function_name=func_name,
                        internal_code=False,
                    )

                    locations.append(file_location)

                stacks.append(temporalio.api.sdk.v1.StackTrace(locations=locations))

            est = temporalio.api.sdk.v1.EnhancedStackTrace(
                sdk=sdk, sources=sources, stacks=stacks
            )
            return est

    def _temporal_workflow_metadata(self) -> temporalio.api.sdk.v1.WorkflowMetadata:
        query_definitions = [
            temporalio.api.sdk.v1.WorkflowInteractionDefinition(
                name=qd.name or "",
                description=qd.description or "",
            )
            for qd in self._queries.values()
        ]
        query_definitions.sort(key=lambda qd: qd.name)
        signal_definitions = [
            temporalio.api.sdk.v1.WorkflowInteractionDefinition(
                name=sd.name or "",
                description=sd.description or "",
            )
            for sd in self._signals.values()
        ]
        signal_definitions.sort(key=lambda sd: sd.name)
        update_definitions = [
            temporalio.api.sdk.v1.WorkflowInteractionDefinition(
                name=ud.name or "",
                description=ud.description or "",
            )
            for ud in self._updates.values()
        ]
        update_definitions.sort(key=lambda ud: ud.name)
        wf_def = temporalio.api.sdk.v1.WorkflowDefinition(
            type=self._info.workflow_type,
            query_definitions=query_definitions,
            signal_definitions=signal_definitions,
            update_definitions=update_definitions,
        )
        cur_details = self.workflow_get_current_details()
        return temporalio.api.sdk.v1.WorkflowMetadata(
            definition=wf_def, current_details=cur_details
        )

    def _timer_impl(
        self,
        delay: float,
        options: _TimerOptions,
        callback: Callable[..., Any],
        *args: Any,
        context: contextvars.Context | None = None,
    ):
        self._assert_not_read_only("schedule timer")
        # Delay must be positive
        if delay < 0:
            raise RuntimeError("Attempting to schedule timer with negative delay")

        # Create, schedule, and return
        seq = self._next_seq("timer")
        handle = _TimerHandle(
            seq, self.time() + delay, options, callback, args, self, context
        )
        handle._apply_start_command(self._add_command(), delay)
        self._pending_timers[seq] = handle
        return handle

    #### asyncio.AbstractEventLoop function impls ####
    # These are in the order defined in CPython's impl of the base class. Many
    # functions are intentionally not implemented/supported.

    def _timer_handle_cancelled(self, handle: asyncio.TimerHandle) -> None:
        if not isinstance(handle, _TimerHandle):
            raise TypeError("Expected Temporal timer handle")
        if not self._pending_timers.pop(handle._seq, None):
            return
        handle._apply_cancel_command(self._add_command())

    _Ts = TypeVarTuple("_Ts")

    def call_soon(
        self,
        callback: Callable[[Unpack[_Ts]], object],
        *args: Unpack[_Ts],
        context: contextvars.Context | None = None,
    ) -> asyncio.Handle:
        # We need to allow this during delete because this is how tasks schedule
        # entire cancellation calls
        self._assert_not_read_only("schedule task", allow_during_delete=True)
        handle = asyncio.Handle(callback, args, self, context)
        self._ready.append(handle)
        return handle

    def call_later(
        self,
        delay: float,
        callback: Callable[[Unpack[_Ts]], object],
        *args: Unpack[_Ts],
        context: contextvars.Context | None = None,
    ) -> asyncio.TimerHandle:
        options = _TimerOptionsCtxVar.get()
        return self._timer_impl(delay, options, callback, *args, context=context)

    def call_at(
        self,
        when: float,
        callback: Callable[[Unpack[_Ts]], object],
        *args: Unpack[_Ts],
        context: contextvars.Context | None = None,
    ) -> asyncio.TimerHandle:
        # We usually would not support fixed-future-time call (and we didn't
        # previously), but 3.11 added asyncio.timeout which uses it and 3.12
        # changed wait_for to use asyncio.timeout. We now simply count on users
        # to only add to loop.time() and not an actual fixed point. Due to the
        # fact that loop.time() is at a nanosecond level which floats can't
        # always express well, we round to the nearest millisecond. We do this
        # after the subtraction not before because this may be the result of a
        # previous addition in Python code.
        return self.call_later(
            round(when - self.time(), 3), callback, *args, context=context
        )

    def time(self) -> float:
        return self._time_ns / 1e9

    def create_future(self) -> asyncio.Future[Any]:
        return asyncio.Future(loop=self)

    def create_task(
        self,
        coro: Awaitable[_T] | Generator[Any, None, _T],
        *,
        name: str | None = None,
        context: contextvars.Context | None = None,
    ) -> asyncio.Task[_T]:
        # Context only supported on newer Python versions
        if sys.version_info >= (3, 11):
            task = asyncio.Task(coro, loop=self, context=context)  # type: ignore
        else:
            task = asyncio.Task(coro, loop=self)  # type: ignore
        self._register_task(task, name=name)
        return task

    def get_exception_handler(self) -> _ExceptionHandler | None:
        return self._exception_handler

    def get_task_factory(self) -> None:
        return None

    def set_exception_handler(self, handler: _ExceptionHandler | None) -> None:
        self._exception_handler = handler

    def default_exception_handler(self, context: _Context) -> None:
        # Copied and slightly modified from
        # asyncio.BaseEventLoop.default_exception_handler
        message = context.get("message")
        if not message:
            message = "Unhandled exception in event loop"

        exception = context.get("exception")
        exc_info: Any
        if exception is not None:
            exc_info = (type(exception), exception, exception.__traceback__)
        else:
            exc_info = False

        log_lines = [message]
        for key in sorted(context):
            if key in {"message", "exception"}:
                continue
            value = context[key]
            if key == "source_traceback":
                tb = "".join(traceback.format_list(value))
                value = "Object created at (most recent call last):\n"
                value += tb.rstrip()
            elif key == "handle_traceback":
                tb = "".join(traceback.format_list(value))
                value = "Handle created at (most recent call last):\n"
                value += tb.rstrip()
            else:
                value = repr(value)
            log_lines.append(f"{key}: {value}")

        logger.error("\n".join(log_lines), exc_info=exc_info)

    def call_exception_handler(self, context: _Context) -> None:
        # Do nothing with any uncaught exceptions while deleting
        if self._deleting:
            return
        # Copied and slightly modified from
        # asyncio.BaseEventLoop.call_exception_handler
        if self._exception_handler is None:
            try:
                self.default_exception_handler(context)
            except (SystemExit, KeyboardInterrupt):
                raise
            except BaseException:
                # Second protection layer for unexpected errors
                # in the default implementation, as well as for subclassed
                # event loops with overloaded "default_exception_handler".
                logger.error("Exception in default exception handler", exc_info=True)
        else:
            try:
                self._exception_handler(self, context)
            except (SystemExit, KeyboardInterrupt):
                raise
            except BaseException as exc:
                # Exception in the user set custom exception handler.
                try:
                    # Let's try default handler.
                    self.default_exception_handler(
                        {
                            "message": "Unhandled error in exception handler",
                            "exception": exc,
                            "context": context,
                        }
                    )
                except (SystemExit, KeyboardInterrupt):
                    raise
                except BaseException:
                    # Guard 'default_exception_handler' in case it is
                    # overloaded.
                    logger.error(
                        "Exception in default exception handler "
                        "while handling an unexpected error "
                        "in custom exception handler",
                        exc_info=True,
                    )

    def get_debug(self) -> bool:
        return False

    def is_running(self) -> bool:
        return True


class _WorkflowInboundImpl(WorkflowInboundInterceptor):
    def __init__(  # type: ignore
        self,
        instance: _WorkflowInstanceImpl,
    ) -> None:
        # We are intentionally not calling the base class's __init__ here
        self._instance = instance

    def init(self, outbound: WorkflowOutboundInterceptor) -> None:
        self._outbound = outbound  # type: ignore

    async def execute_workflow(self, input: ExecuteWorkflowInput) -> Any:
        args = [self._instance._object] + list(input.args)
        return await input.run_fn(*args)

    async def handle_signal(self, input: HandleSignalInput) -> None:
        handler = self._instance.workflow_get_signal_handler(
            input.signal
        ) or self._instance.workflow_get_signal_handler(None)
        # Handler should always be present at this point
        assert handler
        if inspect.iscoroutinefunction(handler):
            await handler(*input.args)
        else:
            handler(*input.args)

    async def handle_query(self, input: HandleQueryInput) -> Any:
        handler = self._instance.workflow_get_query_handler(
            input.query
        ) or self._instance.workflow_get_query_handler(None)
        # Handler should always be present at this point
        assert handler
        if inspect.iscoroutinefunction(handler):
            return await handler(*input.args)
        else:
            return handler(*input.args)

    def handle_update_validator(self, input: HandleUpdateInput) -> None:
        # Do not "or None" the validator, since we only want to use the validator for
        # the specific named update - we shouldn't fall back to the dynamic validator
        # for some defined, named update which doesn't have a defined validator.
        handler = self._instance.workflow_get_update_validator(input.update)
        # Validator may not be defined
        if handler is not None:
            handler(*input.args)

    async def handle_update_handler(self, input: HandleUpdateInput) -> Any:
        handler = self._instance.workflow_get_update_handler(
            input.update
        ) or self._instance.workflow_get_update_handler(None)
        # Handler should always be present at this point
        assert handler
        if inspect.iscoroutinefunction(handler):
            return await handler(*input.args)
        else:
            return handler(*input.args)


class _WorkflowOutboundImpl(WorkflowOutboundInterceptor):
    def __init__(self, instance: _WorkflowInstanceImpl) -> None:  # type: ignore
        # We are intentionally not calling the base class's __init__ here
        self._instance = instance

    def continue_as_new(self, input: ContinueAsNewInput) -> NoReturn:
        self._instance._outbound_continue_as_new(input)

    def info(self) -> temporalio.workflow.Info:
        return self._instance._info

    async def signal_child_workflow(self, input: SignalChildWorkflowInput) -> None:
        return await self._instance._outbound_signal_child_workflow(input)

    async def signal_external_workflow(
        self, input: SignalExternalWorkflowInput
    ) -> None:
        await self._instance._outbound_signal_external_workflow(input)

    def start_activity(
        self, input: StartActivityInput
    ) -> temporalio.workflow.ActivityHandle[Any]:
        return self._instance._outbound_schedule_activity(input)

    async def start_child_workflow(
        self, input: StartChildWorkflowInput
    ) -> temporalio.workflow.ChildWorkflowHandle[Any, Any]:
        return await self._instance._outbound_start_child_workflow(input)

    async def start_nexus_operation(
        self, input: StartNexusOperationInput[Any, OutputT]
    ) -> _NexusOperationHandle[OutputT]:
        return await self._instance._outbound_start_nexus_operation(input)

    def start_local_activity(
        self, input: StartLocalActivityInput
    ) -> temporalio.workflow.ActivityHandle[Any]:
        return self._instance._outbound_schedule_activity(input)


@dataclass(frozen=True)
class _TimerOptions:
    user_metadata: temporalio.api.sdk.v1.UserMetadata | None = None


_TimerOptionsCtxVar: contextvars.ContextVar[_TimerOptions] = contextvars.ContextVar(
    "__temporal_timer_options", default=_TimerOptions()
)


class _TimerHandle(asyncio.TimerHandle):
    def __init__(
        self,
        seq: int,
        when: float,
        options: _TimerOptions | None,
        callback: Callable[..., Any],
        args: Sequence[Any],
        loop: asyncio.AbstractEventLoop,
        context: contextvars.Context | None,
    ) -> None:
        super().__init__(when, callback, args, loop, context)
        self._seq = seq
        self._options = options

    def _apply_start_command(
        self,
        command: temporalio.bridge.proto.workflow_commands.WorkflowCommand,
        delay: float,
    ) -> None:
        command.start_timer.seq = self._seq
        if self._options and self._options.user_metadata:
            command.user_metadata.CopyFrom(self._options.user_metadata)
        command.start_timer.start_to_fire_timeout.FromNanoseconds(int(delay * 1e9))

    def _apply_cancel_command(
        self,
        command: temporalio.bridge.proto.workflow_commands.WorkflowCommand,
    ) -> None:
        command.cancel_timer.seq = self._seq


class _ActivityDoBackoffError(BaseException):
    def __init__(
        self, backoff: temporalio.bridge.proto.activity_result.DoBackoff
    ) -> None:
        super().__init__("Backoff")
        self.backoff = backoff


class _ActivityHandle(temporalio.workflow.ActivityHandle[Any]):
    def __init__(
        self,
        instance: _WorkflowInstanceImpl,
        input: StartActivityInput | StartLocalActivityInput,
        fn: Coroutine[Any, Any, Any],
    ) -> None:
        super().__init__(fn)
        self._instance = instance
        self._seq = instance._next_seq("activity")
        self._input = input
        self._result_fut = instance.create_future()
        self._started = False
        instance._register_task(self, name=f"activity: {input.activity}")
        self._payload_converter = self._instance._payload_converter_with_context(
            temporalio.converter.ActivitySerializationContext(
                namespace=self._instance._info.namespace,
                workflow_id=self._instance._info.workflow_id,
                workflow_type=self._instance._info.workflow_type,
                activity_type=self._input.activity,
                activity_id=self._input.activity_id,
                activity_task_queue=(
                    self._input.task_queue or self._instance._info.task_queue
                    if isinstance(self._input, StartActivityInput)
                    else self._instance._info.task_queue
                ),
                is_local=isinstance(self._input, StartLocalActivityInput),
            )
        )

    def cancel(self, msg: Any | None = None) -> bool:
        # Allow the cancel to go through for the task even if we're deleting,
        # just don't do any commands
        if not self._instance._deleting:
            self._instance._assert_not_read_only("cancel activity handle")
            # We override this because if it's not yet started and not done, we need
            # to send a cancel command because the async function won't run to trap
            # the cancel (i.e. cancelled before started)
            if not self._started and not self.done():
                self._apply_cancel_command(self._instance._add_command())
        return super().cancel(msg)

    def _resolve_success(self, result: Any) -> None:
        # We intentionally let this error if already done
        self._result_fut.set_result(result)

    def _resolve_failure(self, err: BaseException) -> None:
        # If it was never started, we don't need to set this failure. In cases
        # where this is cancelled before started, setting this exception causes
        # a Python warning to be emitted because this future is never awaited
        # on.
        if self._started:
            self._result_fut.set_exception(err)

    def _resolve_backoff(
        self, backoff: temporalio.bridge.proto.activity_result.DoBackoff
    ) -> None:
        # Change the sequence and set backoff exception on the current future
        self._seq = self._instance._next_seq("activity")
        self._result_fut.set_exception(_ActivityDoBackoffError(backoff))
        # Replace the result future since that one is resolved now
        self._result_fut = self._instance.create_future()
        # Mark this as not started since it's a new future
        self._started = False

    def _apply_schedule_command(
        self,
        local_backoff: None
        | (temporalio.bridge.proto.activity_result.DoBackoff) = None,
    ) -> None:
        # Convert arguments before creating command in case it raises error
        payloads = (
            self._payload_converter.to_payloads(self._input.args)
            if self._input.args
            else None
        )

        command = self._instance._add_command()
        # TODO(cretz): Why can't MyPy infer this?
        v: (
            temporalio.bridge.proto.workflow_commands.ScheduleActivity
            | temporalio.bridge.proto.workflow_commands.ScheduleLocalActivity
        ) = (
            command.schedule_local_activity
            if isinstance(self._input, StartLocalActivityInput)
            else command.schedule_activity
        )
        v.seq = self._seq
        v.activity_id = self._input.activity_id or str(self._seq)
        v.activity_type = self._input.activity
        if self._input.headers:
            temporalio.common._apply_headers(self._input.headers, v.headers)
        if payloads:
            v.arguments.extend(payloads)
        if self._input.schedule_to_close_timeout:
            v.schedule_to_close_timeout.FromTimedelta(
                self._input.schedule_to_close_timeout
            )
        if self._input.schedule_to_start_timeout:
            v.schedule_to_start_timeout.FromTimedelta(
                self._input.schedule_to_start_timeout
            )
        if self._input.start_to_close_timeout:
            v.start_to_close_timeout.FromTimedelta(self._input.start_to_close_timeout)
        if self._input.retry_policy:
            self._input.retry_policy.apply_to_proto(v.retry_policy)
        if self._input.summary:
            command.user_metadata.summary.CopyFrom(
                self._payload_converter.to_payload(self._input.summary)
            )
        v.cancellation_type = cast(
            temporalio.bridge.proto.workflow_commands.ActivityCancellationType.ValueType,
            int(self._input.cancellation_type),
        )

        # Things specific to local or remote
        if isinstance(self._input, StartActivityInput):
            command.schedule_activity.task_queue = (
                self._input.task_queue or self._instance._info.task_queue
            )
            if self._input.heartbeat_timeout:
                command.schedule_activity.heartbeat_timeout.FromTimedelta(
                    self._input.heartbeat_timeout
                )
            command.schedule_activity.do_not_eagerly_execute = (
                self._input.disable_eager_execution
            )
            if self._input.versioning_intent:
                command.schedule_activity.versioning_intent = (
                    self._input.versioning_intent._to_proto()
                )
            if self._input.priority:
                command.schedule_activity.priority.CopyFrom(
                    self._input.priority._to_proto()
                )
        if isinstance(self._input, StartLocalActivityInput):
            if self._input.local_retry_threshold:
                command.schedule_local_activity.local_retry_threshold.FromTimedelta(
                    self._input.local_retry_threshold
                )
            if local_backoff:
                command.schedule_local_activity.attempt = local_backoff.attempt
                command.schedule_local_activity.original_schedule_time.CopyFrom(
                    local_backoff.original_schedule_time
                )

    def _apply_cancel_command(
        self,
        command: temporalio.bridge.proto.workflow_commands.WorkflowCommand,
    ) -> None:
        if isinstance(self._input, StartActivityInput):
            command.request_cancel_activity.seq = self._seq
        else:
            command.request_cancel_local_activity.seq = self._seq


class _ChildWorkflowHandle(temporalio.workflow.ChildWorkflowHandle[Any, Any]):
    def __init__(
        self,
        instance: _WorkflowInstanceImpl,
        seq: int,
        input: StartChildWorkflowInput,
        fn: Coroutine[Any, Any, Any],
    ) -> None:
        super().__init__(fn)
        self._instance = instance
        self._seq = seq
        self._input = input
        self._start_fut: asyncio.Future[None] = instance.create_future()
        self._result_fut: asyncio.Future[Any] = instance.create_future()
        self._first_execution_run_id = "<unknown>"
        instance._register_task(self, name=f"child: {input.workflow}")
        workflow_context = temporalio.converter.WorkflowSerializationContext(
            namespace=self._instance._info.namespace,
            workflow_id=self._input.id,
        )
        self._payload_converter = self._instance._payload_converter_with_context(
            workflow_context
        )
        self._failure_converter = self._instance._failure_converter_with_context(
            workflow_context
        )

    @property
    def id(self) -> str:
        return self._input.id

    @property
    def first_execution_run_id(self) -> str | None:
        return self._first_execution_run_id

    async def signal(
        self,
        signal: str | Callable,
        arg: Any = temporalio.common._arg_unset,
        *,
        args: Sequence[Any] = [],
    ) -> None:
        self._instance._assert_not_read_only("signal child handle")
        await self._instance._outbound.signal_child_workflow(
            SignalChildWorkflowInput(
                signal=temporalio.workflow._SignalDefinition.must_name_from_fn_or_str(
                    signal
                ),
                args=temporalio.common._arg_or_args(arg, args),
                child_workflow_id=self._input.id,
                headers={},
            )
        )

    def _resolve_start_success(self, run_id: str) -> None:
        self._first_execution_run_id = run_id
        # We intentionally let this error if already done
        self._start_fut.set_result(None)

    def _resolve_success(self, result: Any) -> None:
        # We intentionally let this error if already done
        self._result_fut.set_result(result)

    def _resolve_failure(self, err: BaseException) -> None:
        if self._start_fut.done():
            # We intentionally let this error if already done
            self._result_fut.set_exception(err)
        else:
            self._start_fut.set_exception(err)
            # Set the result as none to avoid Python warning about unhandled
            # future
            self._result_fut.set_result(None)

    def _apply_start_command(self) -> None:
        # Convert arguments before creating command in case it raises error
        payloads = (
            self._payload_converter.to_payloads(self._input.args)
            if self._input.args
            else None
        )

        command = self._instance._add_command()
        v = command.start_child_workflow_execution
        v.seq = self._seq
        v.namespace = self._instance._info.namespace
        v.workflow_id = self._input.id
        v.workflow_type = self._input.workflow
        v.task_queue = self._input.task_queue or self._instance._info.task_queue
        if payloads:
            v.input.extend(payloads)
        if self._input.execution_timeout:
            v.workflow_execution_timeout.FromTimedelta(self._input.execution_timeout)
        if self._input.run_timeout:
            v.workflow_run_timeout.FromTimedelta(self._input.run_timeout)
        if self._input.task_timeout:
            v.workflow_task_timeout.FromTimedelta(self._input.task_timeout)
        v.parent_close_policy = cast(
            temporalio.bridge.proto.child_workflow.ParentClosePolicy.ValueType,
            int(self._input.parent_close_policy),
        )
        v.workflow_id_reuse_policy = cast(
            "temporalio.api.enums.v1.WorkflowIdReusePolicy.ValueType",
            int(self._input.id_reuse_policy),
        )
        if self._input.retry_policy:
            self._input.retry_policy.apply_to_proto(v.retry_policy)
        v.cron_schedule = self._input.cron_schedule
        if self._input.headers:
            temporalio.common._apply_headers(self._input.headers, v.headers)
        if self._input.memo:
            for k, val in self._input.memo.items():
                v.memo[k].CopyFrom(self._payload_converter.to_payloads([val])[0])
        if self._input.search_attributes:
            _encode_search_attributes(
                self._input.search_attributes, v.search_attributes.indexed_fields
            )
        v.cancellation_type = cast(
            temporalio.bridge.proto.child_workflow.ChildWorkflowCancellationType.ValueType,
            int(self._input.cancellation_type),
        )
        if self._input.versioning_intent:
            v.versioning_intent = self._input.versioning_intent._to_proto()
        if self._input.static_summary:
            command.user_metadata.summary.CopyFrom(
                self._payload_converter.to_payload(self._input.static_summary)
            )
        if self._input.static_details:
            command.user_metadata.details.CopyFrom(
                self._payload_converter.to_payload(self._input.static_details)
            )
        if self._input.priority:
            v.priority.CopyFrom(self._input.priority._to_proto())

    # If request cancel external, result does _not_ have seq
    def _apply_cancel_command(
        self,
        command: temporalio.bridge.proto.workflow_commands.WorkflowCommand,
    ) -> None:
        command.cancel_child_workflow_execution.child_workflow_seq = self._seq


class _ExternalWorkflowHandle(temporalio.workflow.ExternalWorkflowHandle[Any]):
    def __init__(
        self,
        instance: _WorkflowInstanceImpl,
        id: str,
        run_id: str | None,
    ) -> None:
        super().__init__()
        self._instance = instance
        self._id = id
        self._run_id = run_id

    @property
    def id(self) -> str:
        return self._id

    @property
    def run_id(self) -> str | None:
        return self._run_id

    async def signal(
        self,
        signal: str | Callable,
        arg: Any = temporalio.common._arg_unset,
        *,
        args: Sequence[Any] = [],
    ) -> None:
        self._instance._assert_not_read_only("signal external handle")
        await self._instance._outbound.signal_external_workflow(
            SignalExternalWorkflowInput(
                signal=temporalio.workflow._SignalDefinition.must_name_from_fn_or_str(
                    signal
                ),
                args=temporalio.common._arg_or_args(arg, args),
                namespace=self._instance._info.namespace,
                workflow_id=self._id,
                workflow_run_id=self._run_id,
                headers={},
            )
        )

    async def cancel(self) -> None:
        self._instance._assert_not_read_only("cancel external handle")
        command = self._instance._add_command()
        v = command.request_cancel_external_workflow_execution
        v.workflow_execution.namespace = self._instance._info.namespace
        v.workflow_execution.workflow_id = self._id
        if self._run_id:
            v.workflow_execution.run_id = self._run_id
        await self._instance._cancel_external_workflow(command)


class _NexusOperationHandle(temporalio.workflow.NexusOperationHandle[OutputT]):
    def __init__(
        self,
        instance: _WorkflowInstanceImpl,
        seq: int,
        input: StartNexusOperationInput[Any, OutputT],
        fn: Coroutine[Any, Any, OutputT],
    ):
        self._instance = instance
        self._seq = seq
        self._input = input
        self._task = asyncio.Task(fn)
        self._start_fut: asyncio.Future[str | None] = instance.create_future()
        self._result_fut: asyncio.Future[OutputT | None] = instance.create_future()
        self._payload_converter = self._instance._context_free_payload_converter
        self._failure_converter = self._instance._context_free_failure_converter

    @property
    def operation_token(self) -> str | None:
        try:
            return self._start_fut.result()
        except BaseException:
            return None

    def __await__(self) -> Generator[Any, Any, OutputT]:
        return self._task.__await__()

    def cancel(self) -> bool:
        return self._task.cancel()

    def _resolve_start_success(self, operation_token: str | None) -> None:
        # We intentionally let this error if already done
        self._start_fut.set_result(operation_token)

    def _resolve_success(self, result: Any) -> None:
        # We intentionally let this error if already done
        self._result_fut.set_result(result)

    def _resolve_failure(self, err: BaseException) -> None:
        if self._start_fut.done():
            # We intentionally let this error if already done
            self._result_fut.set_exception(err)
        else:
            self._start_fut.set_exception(err)
            # Set null result to avoid warning about unhandled future
            self._result_fut.set_result(None)

    def _apply_schedule_command(self) -> None:
        payload = self._payload_converter.to_payload(self._input.input)
        command = self._instance._add_command()
        v = command.schedule_nexus_operation
        v.seq = self._seq
        v.endpoint = self._input.endpoint
        v.service = self._input.service
        v.operation = self._input.operation_name
        v.input.CopyFrom(payload)
        if self._input.schedule_to_close_timeout is not None:
            v.schedule_to_close_timeout.FromTimedelta(
                self._input.schedule_to_close_timeout
            )
        if self._input.schedule_to_start_timeout is not None:
            v.schedule_to_start_timeout.FromTimedelta(
                self._input.schedule_to_start_timeout
            )
        if self._input.start_to_close_timeout is not None:
            v.start_to_close_timeout.FromTimedelta(self._input.start_to_close_timeout)
        v.cancellation_type = cast(
            temporalio.bridge.proto.nexus.NexusOperationCancellationType.ValueType,
            int(self._input.cancellation_type),
        )

        if self._input.headers:
            for key, val in self._input.headers.items():
                v.nexus_header[key] = val

        if self._input.summary:
            command.user_metadata.summary.CopyFrom(
                self._payload_converter.to_payload(self._input.summary)
            )

    def _apply_cancel_command(
        self,
        command: temporalio.bridge.proto.workflow_commands.WorkflowCommand,
    ) -> None:
        command.request_cancel_nexus_operation.seq = self._seq


class _ContinueAsNewError(temporalio.workflow.ContinueAsNewError):
    def __init__(
        self, instance: _WorkflowInstanceImpl, input: ContinueAsNewInput
    ) -> None:
        super().__init__("Continue as new")
        self._instance = instance
        self._input = input

    def _apply_command(self) -> None:
        # Convert arguments before creating command in case it raises error
        payloads = (
            self._instance._workflow_context_payload_converter.to_payloads(
                self._input.args
            )
            if self._input.args
            else None
        )
        memo_payloads = (
            {
                k: self._instance._workflow_context_payload_converter.to_payloads(
                    [val]
                )[0]
                for k, val in self._input.memo.items()
            }
            if self._input.memo
            else None
        )

        command = self._instance._add_command()
        v = command.continue_as_new_workflow_execution
        v.SetInParent()
        if self._input.workflow:
            v.workflow_type = self._input.workflow
        if self._input.task_queue:
            v.task_queue = self._input.task_queue
        if payloads:
            v.arguments.extend(payloads)
        if self._input.run_timeout:
            v.workflow_run_timeout.FromTimedelta(self._input.run_timeout)
        if self._input.task_timeout:
            v.workflow_task_timeout.FromTimedelta(self._input.task_timeout)
        if self._input.headers:
            temporalio.common._apply_headers(self._input.headers, v.headers)
        if self._input.retry_policy:
            self._input.retry_policy.apply_to_proto(v.retry_policy)
        if memo_payloads:
            for k, val in memo_payloads.items():
                v.memo[k].CopyFrom(val)
        if self._input.search_attributes:
            _encode_search_attributes(
                self._input.search_attributes, v.search_attributes.indexed_fields
            )
        if self._input.versioning_intent:
            v.versioning_intent = self._input.versioning_intent._to_proto()
        if self._input.initial_versioning_behavior:
            v.initial_versioning_behavior = cast(
                "temporalio.api.enums.v1.ContinueAsNewVersioningBehavior.ValueType",
                int(self._input.initial_versioning_behavior),
            )


def _encode_search_attributes(
    attributes: (
        temporalio.common.SearchAttributes | temporalio.common.TypedSearchAttributes
    ),
    payloads: Mapping[str, temporalio.api.common.v1.Payload],
) -> None:
    if isinstance(attributes, temporalio.common.TypedSearchAttributes):
        for pair in attributes:
            payloads[pair.key.name].CopyFrom(
                temporalio.converter.encode_typed_search_attribute_value(
                    pair.key, pair.value
                )
            )
    else:
        for k, vals in attributes.items():
            payloads[k].CopyFrom(
                temporalio.converter.encode_search_attribute_values(vals)
            )


class _WorkflowExternFunctions(TypedDict):
    __temporal_get_metric_meter: Callable[[], temporalio.common.MetricMeter]
    __temporal_assert_local_activity_valid: Callable[[str], None]


class _ReplaySafeMetricMeter(temporalio.common.MetricMeter):
    def __init__(self, underlying: temporalio.common.MetricMeter) -> None:
        self._underlying = underlying

    def create_counter(
        self, name: str, description: str | None = None, unit: str | None = None
    ) -> temporalio.common.MetricCounter:
        return _ReplaySafeMetricCounter(
            self._underlying.create_counter(name, description, unit)
        )

    def create_histogram(
        self, name: str, description: str | None = None, unit: str | None = None
    ) -> temporalio.common.MetricHistogram:
        return _ReplaySafeMetricHistogram(
            self._underlying.create_histogram(name, description, unit)
        )

    def create_histogram_float(
        self, name: str, description: str | None = None, unit: str | None = None
    ) -> temporalio.common.MetricHistogramFloat:
        return _ReplaySafeMetricHistogramFloat(
            self._underlying.create_histogram_float(name, description, unit)
        )

    def create_histogram_timedelta(
        self, name: str, description: str | None = None, unit: str | None = None
    ) -> temporalio.common.MetricHistogramTimedelta:
        return _ReplaySafeMetricHistogramTimedelta(
            self._underlying.create_histogram_timedelta(name, description, unit)
        )

    def create_gauge(
        self, name: str, description: str | None = None, unit: str | None = None
    ) -> temporalio.common.MetricGauge:
        return _ReplaySafeMetricGauge(
            self._underlying.create_gauge(name, description, unit)
        )

    def create_gauge_float(
        self, name: str, description: str | None = None, unit: str | None = None
    ) -> temporalio.common.MetricGaugeFloat:
        return _ReplaySafeMetricGaugeFloat(
            self._underlying.create_gauge_float(name, description, unit)
        )

    def with_additional_attributes(
        self, additional_attributes: temporalio.common.MetricAttributes
    ) -> temporalio.common.MetricMeter:
        return _ReplaySafeMetricMeter(
            self._underlying.with_additional_attributes(additional_attributes)
        )


_MetricType = TypeVar("_MetricType", bound=temporalio.common.MetricCommon)


class _ReplaySafeMetricCommon(temporalio.common.MetricCommon, Generic[_MetricType]):
    def __init__(self, underlying: _MetricType) -> None:
        self._underlying = underlying

    @property
    def name(self) -> str:
        return self._underlying.name

    @property
    def description(self) -> str | None:
        return self._underlying.description

    @property
    def unit(self) -> str | None:
        return self._underlying.unit

    def with_additional_attributes(
        self, additional_attributes: temporalio.common.MetricAttributes
    ) -> Self:
        return self.__class__(
            self._underlying.with_additional_attributes(additional_attributes)
        )


class _ReplaySafeMetricCounter(
    temporalio.common.MetricCounter,
    _ReplaySafeMetricCommon[temporalio.common.MetricCounter],
):
    def add(
        self,
        value: int,
        additional_attributes: temporalio.common.MetricAttributes | None = None,
    ) -> None:
        if not temporalio.workflow.unsafe.is_replaying():
            self._underlying.add(value, additional_attributes)


class _ReplaySafeMetricHistogram(
    temporalio.common.MetricHistogram,
    _ReplaySafeMetricCommon[temporalio.common.MetricHistogram],
):
    def record(
        self,
        value: int,
        additional_attributes: temporalio.common.MetricAttributes | None = None,
    ) -> None:
        if not temporalio.workflow.unsafe.is_replaying():
            self._underlying.record(value, additional_attributes)


class _ReplaySafeMetricHistogramFloat(
    temporalio.common.MetricHistogramFloat,
    _ReplaySafeMetricCommon[temporalio.common.MetricHistogramFloat],
):
    def record(
        self,
        value: float,
        additional_attributes: temporalio.common.MetricAttributes | None = None,
    ) -> None:
        if not temporalio.workflow.unsafe.is_replaying():
            self._underlying.record(value, additional_attributes)


class _ReplaySafeMetricHistogramTimedelta(
    temporalio.common.MetricHistogramTimedelta,
    _ReplaySafeMetricCommon[temporalio.common.MetricHistogramTimedelta],
):
    def record(
        self,
        value: timedelta,
        additional_attributes: temporalio.common.MetricAttributes | None = None,
    ) -> None:
        if not temporalio.workflow.unsafe.is_replaying():
            self._underlying.record(value, additional_attributes)


class _ReplaySafeMetricGauge(
    temporalio.common.MetricGauge,
    _ReplaySafeMetricCommon[temporalio.common.MetricGauge],
):
    def set(
        self,
        value: int,
        additional_attributes: temporalio.common.MetricAttributes | None = None,
    ) -> None:
        if not temporalio.workflow.unsafe.is_replaying():
            self._underlying.set(value, additional_attributes)


class _ReplaySafeMetricGaugeFloat(
    temporalio.common.MetricGaugeFloat,
    _ReplaySafeMetricCommon[temporalio.common.MetricGaugeFloat],
):
    def set(
        self,
        value: float,
        additional_attributes: temporalio.common.MetricAttributes | None = None,
    ) -> None:
        if not temporalio.workflow.unsafe.is_replaying():
            self._underlying.set(value, additional_attributes)


class _WorkflowBeingEvictedError(BaseException):
    pass


@dataclass
class HandlerExecution:
    """Information about an execution of a signal or update handler."""

    name: str
    unfinished_policy: temporalio.workflow.HandlerUnfinishedPolicy
    id: str | None = None


def _make_unfinished_update_handler_message(
    handler_executions: list[HandlerExecution],
) -> str:
    message = """
[TMPRL1102] Workflow finished while update handlers are still running. This may have interrupted work that the
update handler was doing, and the client that sent the update will receive a 'workflow execution
already completed' RPCError instead of the update result. You can wait for all update and signal
handlers to complete by using `await workflow.wait_condition(lambda:
workflow.all_handlers_finished())`. Alternatively, if both you and the clients sending the update
are okay with interrupting running handlers when the workflow finishes, and causing clients to
receive errors, then you can disable this warning via the update handler decorator:
`@workflow.update(unfinished_policy=workflow.HandlerUnfinishedPolicy.ABANDON)`.
""".replace("\n", " ").strip()
    return (
        f"{message} The following updates were unfinished (and warnings were not disabled for their handler): "
        + json.dumps([{"name": ex.name, "id": ex.id} for ex in handler_executions])
    )


def _make_unfinished_signal_handler_message(
    handler_executions: list[HandlerExecution],
) -> str:
    message = """
[TMPRL1102] Workflow finished while signal handlers are still running. This may have interrupted work that the
signal handler was doing. You can wait for all update and signal handlers to complete by using
`await workflow.wait_condition(lambda: workflow.all_handlers_finished())`. Alternatively, if both
you and the clients sending the signal are okay with interrupting running handlers when the workflow
finishes, then you can disable this warning via the signal handler decorator:
`@workflow.signal(unfinished_policy=workflow.HandlerUnfinishedPolicy.ABANDON)`.
""".replace("\n", " ").strip()
    names = collections.Counter(ex.name for ex in handler_executions)
    return (
        f"{message} The following signals were unfinished (and warnings were not disabled for their handler): "
        + json.dumps(
            [{"name": name, "count": count} for name, count in names.most_common()]
        )
    )


class _WorkflowLogicFlag(IntEnum):
    """Flags that may be set on task/activation completion to differentiate new from old workflow behavior."""

    RAISE_ON_CANCELLING_COMPLETED_ACTIVITY = 1