Reraise workflow failure errors from OpenAI's UserError

Author: tconley1428

temporalio/contrib/openai_agents/__init__.py

@@ -9,6 +9,7 @@
 """
 
 from temporalio.contrib.openai_agents._model_parameters import ModelActivityParameters
+from temporalio.contrib.openai_agents._openai_runner import AgentsWorkflowFailure
 from temporalio.contrib.openai_agents._temporal_openai_agents import (
     OpenAIAgentsPlugin,
     TestModel,
@@ -21,6 +22,7 @@
 from . import workflow
 
 __all__ = [
+    "AgentsWorkflowFailure",
     "OpenAIAgentsPlugin",
     "ModelActivityParameters",
     "workflow",

temporalio/contrib/openai_agents/_openai_runner.py

@@ -5,6 +5,7 @@
 
 from agents import (
     Agent,
+    AgentsException,
     Handoff,
     RunConfig,
     RunContextWrapper,
@@ -21,6 +22,16 @@
 from temporalio import workflow
 from temporalio.contrib.openai_agents._model_parameters import ModelActivityParameters
 from temporalio.contrib.openai_agents._temporal_model_stub import _TemporalModelStub
+from temporalio.exceptions import ApplicationError, TemporalError
+
+
+class AgentsWorkflowFailure(TemporalError):
+    """Error that occurs when the agents SDK raises an error which should terminate the calling workflow.
+
+    .. warning::
+        This exception is experimental and may change in future versions.
+        Use with caution in production environments.
+    """
 
 
 class TemporalOpenAIRunner(AgentRunner):
@@ -136,16 +147,28 @@ async def on_invoke(
                 handoffs=new_handoffs,
             )
 
-        return await self._runner.run(
-            starting_agent=convert_agent(starting_agent, None),
-            input=input,
-            context=context,
-            max_turns=max_turns,
-            hooks=hooks,
-            run_config=run_config,
-            previous_response_id=previous_response_id,
-            session=session,
-        )
+        try:
+            return await self._runner.run(
+                starting_agent=convert_agent(starting_agent, None),
+                input=input,
+                context=context,
+                max_turns=max_turns,
+                hooks=hooks,
+                run_config=run_config,
+                previous_response_id=previous_response_id,
+                session=session,
+            )
+        except AgentsException as e:
+            # In order for workflow failures to properly fail the workflow, we need to rewrap them in
+            # a Temporal error
+            if e.__cause__ and workflow.is_workflow_failure_exception(e.__cause__):
+                reraise = AgentsWorkflowFailure(
+                    f"Workflow failure exception in Agents Framework: {e}"
+                )
+                reraise.__traceback__ = e.__traceback__
+                raise reraise from e.__cause__
+            else:
+                raise e
 
     def run_sync(
         self,

temporalio/contrib/openai_agents/_temporal_openai_agents.py

@@ -27,7 +27,10 @@
 from temporalio.client import ClientConfig, Plugin
 from temporalio.contrib.openai_agents._invoke_model_activity import ModelActivity
 from temporalio.contrib.openai_agents._model_parameters import ModelActivityParameters
-from temporalio.contrib.openai_agents._openai_runner import TemporalOpenAIRunner
+from temporalio.contrib.openai_agents._openai_runner import (
+    AgentsWorkflowFailure,
+    TemporalOpenAIRunner,
+)
 from temporalio.contrib.openai_agents._temporal_trace_provider import (
     TemporalTraceProvider,
 )
@@ -284,6 +287,9 @@ def configure_worker(self, config: WorkerConfig) -> WorkerConfig:
         config["activities"] = list(config.get("activities") or []) + [
             ModelActivity(self._model_provider).invoke_model_activity
         ]
+        config["workflow_failure_exception_types"] = list(
+            config.get("workflow_failure_exception_types") or []
+        ) + [AgentsWorkflowFailure]
         return self.next_worker_plugin.configure_worker(config)
 
     async def run_worker(self, worker: Worker) -> None:

temporalio/worker/_workflow_instance.py

@@ -414,7 +414,7 @@ def activate(
             # 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._is_workflow_failure_exception(err):
+            if self.is_workflow_failure_exception(err):
                 try:
                     self._set_workflow_failure(err)
                 except Exception as inner_err:
@@ -629,7 +629,7 @@ async def run_update() -> None:
                 # 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._is_workflow_failure_exception(err):
+                if not past_validation or self.is_workflow_failure_exception(err):
                     if command is None:
                         command = self._add_command()
                         command.update_response.protocol_instance_id = (
@@ -1939,7 +1939,7 @@ def _convert_payloads(
             # Don't wrap payload conversion errors that would fail the workflow
             raise
         except Exception as err:
-            if self._is_workflow_failure_exception(err):
+            if self.is_workflow_failure_exception(err):
                 raise
             raise RuntimeError("Failed decoding arguments") from err
 
@@ -1982,7 +1982,7 @@ def _instantiate_workflow_object(self) -> Any:
 
         return workflow_instance
 
-    def _is_workflow_failure_exception(self, err: BaseException) -> bool:
+    def is_workflow_failure_exception(self, err: BaseException) -> bool:
         # An exception is a failure instead of a task fail if it's already a
         # failure error or if it is a timeout error or if it is an instance of
         # any of the failure types in the worker or workflow-level setting
@@ -2192,7 +2192,7 @@ async def _run_top_level_workflow_function(self, coro: Awaitable[None]) -> None:
                 err
             ):
                 self._add_command().cancel_workflow_execution.SetInParent()
-            elif self._is_workflow_failure_exception(err):
+            elif self.is_workflow_failure_exception(err):
                 # All other failure errors fail the workflow
                 self._set_workflow_failure(err)
             else:

temporalio/workflow.py

@@ -897,6 +897,9 @@ def workflow_get_current_details(self) -> str: ...
     @abstractmethod
     def workflow_set_current_details(self, details: str): ...
 
+    @abstractmethod
+    def is_workflow_failure_exception(self, err: BaseException) -> bool: ...
+
 
 _current_update_info: contextvars.ContextVar[UpdateInfo] = contextvars.ContextVar(
     "__temporal_current_update_info"
@@ -981,6 +984,10 @@ def memo() -> Mapping[str, Any]:
     return _Runtime.current().workflow_memo()
 
 
+def is_workflow_failure_exception(err: BaseException) -> bool:
+    return _Runtime.current().is_workflow_failure_exception(err)
+
+
 @overload
 def memo_value(key: str, default: Any = temporalio.common._arg_unset) -> Any: ...
 

tests/contrib/openai_agents/test_openai.py

@@ -318,6 +318,10 @@ async def run(self, question: str) -> str:
                     ActivityWeatherService.get_weather_method,
                     start_to_close_timeout=timedelta(seconds=10),
                 ),
+                openai_agents.workflow.activity_as_tool(
+                    get_weather_failure,
+                    start_to_close_timeout=timedelta(seconds=10),
+                ),
             ],
         )
         result = await Runner.run(
@@ -462,6 +466,53 @@ async def test_tool_workflow(client: Client, use_local_model: bool):
             )
 
 
+@activity.defn
+async def get_weather_failure(city: str) -> Weather:
+    """
+    Get the weather for a given city.
+    """
+    raise ApplicationError("No weather", non_retryable=True)
+
+
+class TestWeatherFailureModel(StaticTestModel):
+    responses = [
+        ResponseBuilders.tool_call('{"city":"Tokyo"}', "get_weather_failure"),
+    ]
+
+
+async def test_tool_failure_workflow(client: Client):
+    new_config = client.config()
+    new_config["plugins"] = [
+        openai_agents.OpenAIAgentsPlugin(
+            model_params=ModelActivityParameters(
+                start_to_close_timeout=timedelta(seconds=30)
+            ),
+            model_provider=TestModelProvider(TestWeatherFailureModel()),
+        )
+    ]
+    client = Client(**new_config)
+
+    async with new_worker(
+        client,
+        ToolsWorkflow,
+        activities=[
+            get_weather_failure,
+        ],
+    ) as worker:
+        workflow_handle = await client.start_workflow(
+            ToolsWorkflow.run,
+            "What is the weather in Tokio?",
+            id=f"tools-failure-workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+            execution_timeout=timedelta(seconds=2),
+        )
+        with pytest.raises(WorkflowFailureError) as e:
+            result = await workflow_handle.result()
+        cause = e.value.cause
+        assert isinstance(cause, ApplicationError)
+        assert "Workflow failure exception in Agents Framework" in cause.message
+
+
 @pytest.mark.parametrize("use_local_model", [True, False])
 async def test_nexus_tool_workflow(
     client: Client, env: WorkflowEnvironment, use_local_model: bool