Instantiate interceptors later in workflow instance construction

temporalio/worker/_workflow_instance.py

@@ -301,24 +301,6 @@ def __init__(self, det: WorkflowInstanceDetails) -> None:
             str, List[temporalio.bridge.proto.workflow_activation.SignalWorkflow]
         ] = {}
 
-        # Create interceptors. We do this with our runtime on the loop just in
-        # case they want to access info() during init().
-        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)
-
         # 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
@@ -342,6 +324,24 @@ def __init__(self, det: WorkflowInstanceDetails) -> 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) -> Optional[int]:
         return self._current_thread_id
 

tests/worker/test_workflow.py

@@ -7973,3 +7973,33 @@ async def test_quick_activity_swallows_cancellation(client: Client):
             assert cause.message == "Workflow cancelled"
 
         temporalio.worker._workflow_instance._raise_on_cancelling_completed_activity_override = False
+
+
+class SignalInterceptor(temporalio.worker.Interceptor):
+    def workflow_interceptor_class(
+        self, input: temporalio.worker.WorkflowInterceptorClassInput
+    ) -> Type[SignalInboundInterceptor]:
+        return SignalInboundInterceptor
+
+
+class SignalInboundInterceptor(temporalio.worker.WorkflowInboundInterceptor):
+    def init(self, outbound: temporalio.worker.WorkflowOutboundInterceptor) -> None:
+        def unblock() -> None:
+            return None
+
+        workflow.set_signal_handler("my_random_signal", unblock)
+        super().init(outbound)
+
+
+async def test_signal_handler_in_interceptor(client: Client):
+    async with new_worker(
+        client,
+        HelloWorkflow,
+        interceptors=[SignalInterceptor()],
+    ) as worker:
+        await client.execute_workflow(
+            HelloWorkflow.run,
+            "Temporal",
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )