fix: async interceptor should_intercept must be sync + temporal example fixes

burr/lifecycle/base.py

@@ -709,14 +709,18 @@ class ActionExecutionInterceptorHookAsync(abc.ABC):
     """Async version of ActionExecutionInterceptorHook for intercepting async actions."""
 
     @abc.abstractmethod
-    async def should_intercept(
+    def should_intercept(
         self,
         *,
         action: "Action",
         **future_kwargs: Any,
     ) -> bool:
         """Determine if this action should be intercepted.
 
+        Note: This method is intentionally synchronous even on async interceptors.
+        It is a pure predicate used inside lambdas that cannot be awaited. If you
+        need I/O to decide, cache the result during __init__ or pre_run_step.
+
         :param action: Action to potentially intercept
         :param future_kwargs: Future keyword arguments
         :return: True if this hook should intercept execution
@@ -799,22 +803,26 @@ class StreamingActionInterceptorHookAsync(abc.ABC):
     """Async version for intercepting async streaming actions."""
 
     @abc.abstractmethod
-    async def should_intercept(
+    def should_intercept(
         self,
         *,
         action: "Action",
         **future_kwargs: Any,
     ) -> bool:
         """Determine if this streaming action should be intercepted.
 
+        Note: This method is intentionally synchronous even on async interceptors.
+        It is a pure predicate used inside lambdas that cannot be awaited. If you
+        need I/O to decide, cache the result during __init__ or pre_run_step.
+
         :param action: Streaming action to potentially intercept
         :param future_kwargs: Future keyword arguments
         :return: True if this hook should intercept execution
         """
         pass
 
     @abc.abstractmethod
-    def intercept_stream_run_and_update(
+    async def intercept_stream_run_and_update(
         self,
         *,
         action: "Action",

examples/remote-execution-temporal/temporal_interceptor.py

@@ -113,7 +113,7 @@ async def _get_client(self):
             self._client = await Client.connect("localhost:7233")
         return self._client
 
-    async def should_intercept(self, *, action: Action, **future_kwargs: Any) -> bool:
+    def should_intercept(self, *, action: Action, **future_kwargs: Any) -> bool:
         """Intercept actions tagged with 'durable'."""
         return hasattr(action, "tags") and "durable" in getattr(action, "tags", {})
 
@@ -142,12 +142,12 @@ async def intercept_run(
         # Call worker hooks if provided
         worker_adapter_set = future_kwargs.get("worker_adapter_set")
         if worker_adapter_set:
-            for hook in worker_adapter_set.get_hooks(PreRunStepHookWorkerAsync):
-                await hook.pre_run_step_worker(
-                    action=action.name,
-                    state=state,
-                    inputs=inputs,
-                )
+            await worker_adapter_set.call_all_lifecycle_hooks_async(
+                "pre_run_step_worker",
+                action=action,
+                state=state,
+                inputs=inputs,
+            )
 
         # Execute as Temporal activity
         result_data = await client.execute_workflow(
@@ -167,12 +167,13 @@ async def intercept_run(
 
         # Call post-run worker hooks
         if worker_adapter_set:
-            for hook in worker_adapter_set.get_hooks(PostRunStepHookWorkerAsync):
-                await hook.post_run_step_worker(
-                    action=action.name,
-                    state=state,
-                    result=result,
-                )
+            await worker_adapter_set.call_all_lifecycle_hooks_async(
+                "post_run_step_worker",
+                action=action,
+                state=state,
+                result=result,
+                exception=None,
+            )
 
         # For single-step actions, wrap state update
         if hasattr(action, "single_step") and action.single_step:
@@ -189,14 +190,22 @@ class TemporalWorkerLogger(PreRunStepHookWorkerAsync, PostRunStepHookWorkerAsync
     """Example worker hook that logs action execution on the Temporal worker side."""
 
     async def pre_run_step_worker(
-        self, *, action: str, state: "State", inputs: Dict[str, Any], **kwargs
+        self, *, action: Action, state: "State", inputs: Dict[str, Any], **future_kwargs
     ):
         """Log before action runs on worker."""
-        logger.info(f"[Worker Hook] PRE: {action}")
+        logger.info(f"[Worker Hook] PRE: {action.name}")
 
-    async def post_run_step_worker(self, *, action: str, state: "State", result: dict, **kwargs):
+    async def post_run_step_worker(
+        self,
+        *,
+        action: Action,
+        state: "State",
+        result: dict,
+        exception: Exception,
+        **future_kwargs,
+    ):
         """Log after action runs on worker."""
-        logger.info(f"[Worker Hook] POST: {action} -> {list(result.keys())}")
+        logger.info(f"[Worker Hook] POST: {action.name} -> {list(result.keys())}")
 
 
 # --- Example Application ---