Merge branch 'main' into fix/1188-converter-py313-isinstance

Author: BrunodsLilly

.github/workflows/nightly-throughput-stress.yml

@@ -25,6 +25,9 @@ on:
         default: 360
         type: number
 
+permissions:
+    contents: read
+
 env:
   # Workflow configuration
   TEST_DURATION: ${{ inputs.duration || vars.NIGHTLY_TEST_DURATION || '5h' }}
@@ -164,7 +167,7 @@ jobs:
                   "type": "section",
                   "text": {
                     "type": "mrkdwn",
-                    "text": "*Nightly Throughput Stress Failed* :x:\n\n*Duration:* ${{ env.TEST_DURATION }}\n*Run:* <${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}|View Logs>\n*Triggered by:* ${{ github.event_name == 'schedule' && 'Scheduled' || github.actor }}"
+                    "text": "*Nightly Throughput Stress Failed* :x:\n\n*Repository:* ${{ github.repository }}\n*Duration:* ${{ env.TEST_DURATION }}\n*Run:* <${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}|View Logs>\n*Triggered by:* ${{ github.event_name == 'schedule' && 'Scheduled' || github.actor }}"
                   }
                 }
               ]

.gitignore

@@ -10,3 +10,4 @@ temporalio/bridge/temporal_sdk_bridge*
 /sdk-python.iml
 /.zed
 *.DS_Store
+tags

scripts/gen_payload_visitor.py

@@ -172,8 +172,8 @@ def walk(self, desc: Descriptor) -> bool:
         if key in self.generated:
             return self.generated[key]
         if key in self.in_progress:
-            # Break cycles; if another path proves this node needed, we'll revisit
-            return False
+            # Break cycles; Assume the child will be needed (Used by Failure -> Cause)
+            return True
 
         has_payload = False
         self.in_progress.add(key)

temporalio/bridge/_visitor.py

@@ -74,6 +74,8 @@ async def _visit_temporal_api_failure_v1_ResetWorkflowFailureInfo(self, fs, o):
     async def _visit_temporal_api_failure_v1_Failure(self, fs, o):
         if o.HasField("encoded_attributes"):
             await self._visit_temporal_api_common_v1_Payload(fs, o.encoded_attributes)
+        if o.HasField("cause"):
+            await self._visit_temporal_api_failure_v1_Failure(fs, o.cause)
         if o.HasField("application_failure_info"):
             await self._visit_temporal_api_failure_v1_ApplicationFailureInfo(
                 fs, o.application_failure_info

temporalio/contrib/openai_agents/_invoke_model_activity.py

@@ -248,15 +248,18 @@ def make_tool(tool: ToolInput) -> Tool:
                 ) from e
 
             # Specifically retryable status codes
-            if e.response.status_code in [408, 409, 429, 500]:
+            if (
+                e.response.status_code in [408, 409, 429]
+                or e.response.status_code >= 500
+            ):
                 raise ApplicationError(
-                    "Retryable OpenAI status code",
+                    f"Retryable OpenAI status code: {e.response.status_code}",
                     non_retryable=False,
                     next_retry_delay=retry_after,
                 ) from e
 
             raise ApplicationError(
-                "Non retryable OpenAI status code",
+                f"Non retryable OpenAI status code: {e.response.status_code}",
                 non_retryable=True,
                 next_retry_delay=retry_after,
             ) from e

temporalio/contrib/openai_agents/_model_parameters.py

@@ -70,3 +70,6 @@ class ModelActivityParameters:
 
     priority: Priority = Priority.default
     """Priority for the activity execution."""
+
+    use_local_activity: bool = False
+    """Whether to use a local activity. If changed during a workflow execution, that would break determinism."""

temporalio/contrib/openai_agents/_temporal_model_stub.py

@@ -154,20 +154,32 @@ def make_tool_info(tool: Tool) -> ToolInput:
         else:
             summary = None
 
-        return await workflow.execute_activity_method(
-            ModelActivity.invoke_model_activity,
-            activity_input,
-            summary=summary,
-            task_queue=self.model_params.task_queue,
-            schedule_to_close_timeout=self.model_params.schedule_to_close_timeout,
-            schedule_to_start_timeout=self.model_params.schedule_to_start_timeout,
-            start_to_close_timeout=self.model_params.start_to_close_timeout,
-            heartbeat_timeout=self.model_params.heartbeat_timeout,
-            retry_policy=self.model_params.retry_policy,
-            cancellation_type=self.model_params.cancellation_type,
-            versioning_intent=self.model_params.versioning_intent,
-            priority=self.model_params.priority,
-        )
+        if self.model_params.use_local_activity:
+            return await workflow.execute_local_activity_method(
+                ModelActivity.invoke_model_activity,
+                activity_input,
+                summary=summary,
+                schedule_to_close_timeout=self.model_params.schedule_to_close_timeout,
+                schedule_to_start_timeout=self.model_params.schedule_to_start_timeout,
+                start_to_close_timeout=self.model_params.start_to_close_timeout,
+                retry_policy=self.model_params.retry_policy,
+                cancellation_type=self.model_params.cancellation_type,
+            )
+        else:
+            return await workflow.execute_activity_method(
+                ModelActivity.invoke_model_activity,
+                activity_input,
+                summary=summary,
+                task_queue=self.model_params.task_queue,
+                schedule_to_close_timeout=self.model_params.schedule_to_close_timeout,
+                schedule_to_start_timeout=self.model_params.schedule_to_start_timeout,
+                start_to_close_timeout=self.model_params.start_to_close_timeout,
+                heartbeat_timeout=self.model_params.heartbeat_timeout,
+                retry_policy=self.model_params.retry_policy,
+                cancellation_type=self.model_params.cancellation_type,
+                versioning_intent=self.model_params.versioning_intent,
+                priority=self.model_params.priority,
+            )
 
     def stream_response(
         self,

temporalio/contrib/opentelemetry.py

@@ -172,29 +172,34 @@ def _start_as_current_span(
         kind: opentelemetry.trace.SpanKind,
         context: Optional[Context] = None,
     ) -> Iterator[None]:
-        with self.tracer.start_as_current_span(
-            name,
-            attributes=attributes,
-            kind=kind,
-            context=context,
-            set_status_on_exception=False,
-        ) as span:
-            if input:
-                input.headers = self._context_to_headers(input.headers)
-            try:
-                yield None
-            except Exception as exc:
-                if (
-                    not isinstance(exc, ApplicationError)
-                    or exc.category != ApplicationErrorCategory.BENIGN
-                ):
-                    span.set_status(
-                        Status(
-                            status_code=StatusCode.ERROR,
-                            description=f"{type(exc).__name__}: {exc}",
+        token = opentelemetry.context.attach(context) if context else None
+        try:
+            with self.tracer.start_as_current_span(
+                name,
+                attributes=attributes,
+                kind=kind,
+                context=context,
+                set_status_on_exception=False,
+            ) as span:
+                if input:
+                    input.headers = self._context_to_headers(input.headers)
+                try:
+                    yield None
+                except Exception as exc:
+                    if (
+                        not isinstance(exc, ApplicationError)
+                        or exc.category != ApplicationErrorCategory.BENIGN
+                    ):
+                        span.set_status(
+                            Status(
+                                status_code=StatusCode.ERROR,
+                                description=f"{type(exc).__name__}: {exc}",
+                            )
                         )
-                    )
-                raise
+                    raise
+        finally:
+            if token and context is opentelemetry.context.get_current():
+                opentelemetry.context.detach(token)
 
     def _completed_workflow_span(
         self, params: _CompletedWorkflowSpanParams

temporalio/worker/workflow_sandbox/__init__.py

@@ -58,11 +58,13 @@
     RestrictedWorkflowAccessError,
     SandboxMatcher,
     SandboxRestrictions,
+    UnintentionalPassthroughError,
 )
 from ._runner import SandboxedWorkflowRunner
 
 __all__ = [
     "RestrictedWorkflowAccessError",
+    "UnintentionalPassthroughError",
     "SandboxedWorkflowRunner",
     "SandboxMatcher",
     "SandboxRestrictions",

temporalio/worker/workflow_sandbox/_importer.py

@@ -42,6 +42,7 @@
     RestrictedWorkflowAccessError,
     RestrictionContext,
     SandboxRestrictions,
+    UnintentionalPassthroughError,
 )
 
 logger = logging.getLogger(__name__)
@@ -200,6 +201,17 @@ def _import(
 
         # Check module restrictions and passthrough modules
         if full_name not in sys.modules:
+            # Issue a warning if appropriate
+            if (
+                self.restriction_context.in_activation
+                and self._is_import_notification_policy_applied(
+                    temporalio.workflow.SandboxImportNotificationPolicy.WARN_ON_DYNAMIC_IMPORT
+                )
+            ):
+                warnings.warn(
+                    f"Module {full_name} was imported after initial workflow load."
+                )
+
             # Make sure not an entirely invalid module
             self._assert_valid_module(full_name)
 
@@ -282,13 +294,36 @@ def module_configured_passthrough(self, name: str) -> bool:
                 break
         return True
 
+    def _is_import_notification_policy_applied(
+        self, policy: temporalio.workflow.SandboxImportNotificationPolicy
+    ) -> bool:
+        override_policy = (
+            temporalio.workflow.unsafe.current_import_notification_policy_override()
+        )
+        if override_policy:
+            return policy in override_policy
+
+        return policy in self.restrictions.import_notification_policy
+
     def _maybe_passthrough_module(self, name: str) -> Optional[types.ModuleType]:
         # If imports not passed through and all modules are not passed through
         # and name not in passthrough modules, check parents
         if (
             not temporalio.workflow.unsafe.is_imports_passed_through()
             and not self.module_configured_passthrough(name)
         ):
+            if self._is_import_notification_policy_applied(
+                temporalio.workflow.SandboxImportNotificationPolicy.RAISE_ON_UNINTENTIONAL_PASSTHROUGH
+            ):
+                raise UnintentionalPassthroughError(name)
+
+            if self._is_import_notification_policy_applied(
+                temporalio.workflow.SandboxImportNotificationPolicy.WARN_ON_UNINTENTIONAL_PASSTHROUGH
+            ):
+                warnings.warn(
+                    f"Module {name} was not intentionally passed through to the sandbox."
+                )
+
             return None
         # Do the pass through
         with self._unapplied():