Add retrying of workflow validation errors only on python 3.10 (#1223)

* Add retrying of workflow validation errors only on python 3.10

* use session scoped shared state manager to ensure shutdown occurs.

* run formatter

Author: VegetarianOrc

.github/workflows/ci.yml

@@ -32,6 +32,8 @@ jobs:
           - os: ubuntu-latest
             python: "3.10"
             protoCheckTarget: true
+          - python: "3.10"
+            pytestExtraArgs: "--reruns 3 --only-rerun \"RuntimeError: Failed validating workflow\""
           - os: ubuntu-arm
             runsOn: ubuntu-24.04-arm64-2-core
           - os: macos-intel

pyproject.toml

@@ -59,6 +59,7 @@ dev = [
   "openai-agents>=0.3,<0.5; python_version >= '3.14'",
   "openai-agents[litellm]>=0.3,<0.4; python_version < '3.14'",
   "googleapis-common-protos==1.70.0",
+  "pytest-rerunfailures>=16.1",
 ]
 
 [tool.poe.tasks]

tests/conftest.py

@@ -7,6 +7,11 @@
 import pytest
 import pytest_asyncio
 
+from temporalio.client import Client
+from temporalio.testing import WorkflowEnvironment
+from temporalio.worker import SharedStateManager
+from tests.helpers.worker import ExternalPythonWorker, ExternalWorker
+
 from . import DEV_SERVER_DOWNLOAD_VERSION
 
 # If there is an integration test environment variable set, we must remove the
@@ -38,10 +43,6 @@
         or protobuf_version.startswith("6.")
     ), f"Expected protobuf 4.x/5.x/6.x, got {protobuf_version}"
 
-from temporalio.client import Client
-from temporalio.testing import WorkflowEnvironment
-from tests.helpers.worker import ExternalPythonWorker, ExternalWorker
-
 
 def pytest_runtest_setup(item):
     """Print a newline so that custom printed output starts on new line."""
@@ -134,6 +135,17 @@ async def env(env_type: str) -> AsyncGenerator[WorkflowEnvironment, None]:
     await env.shutdown()
 
 
+@pytest.fixture(scope="session")
+def shared_state_manager() -> Iterator[SharedStateManager]:
+    mp_mgr = multiprocessing.Manager()
+    mgr = SharedStateManager.create_from_multiprocessing(mp_mgr)
+
+    try:
+        yield mgr
+    finally:
+        mp_mgr.shutdown()
+
+
 @pytest.fixture(scope="session")
 def mp_fork_ctx() -> Iterator[multiprocessing.context.BaseContext | None]:
     mp_ctx = None