Fix Pydantic constraints in workflow sandboxes

CHANGELOG.md

@@ -41,6 +41,9 @@ to include examples, links to docs, or any other relevant information.
 
 ### Fixed
 
+- Fixed Pydantic constraints being silently ignored inside or outside workflow
+  sandboxes by passing through `annotated_types` by default.
+
 - `contrib.google_adk_agents`: agents with an `output_schema` no longer fail every workflow task
   when calling the model. The schema type is now sent to the model activity as its JSON schema.
   Custom Pydantic schema generation is preserved.

temporalio/worker/workflow_sandbox/_restrictions.py

@@ -515,6 +515,8 @@ def with_child_unrestricted(self, *child_path: str) -> SandboxMatcher:
         "pydantic",
         # Same for its compiled core, which Pydantic-based libraries import lazily
         "pydantic_core",
+        # Pydantic caches constraint classes by identity across sandboxes.
+        "annotated_types",
     }
 )
 

tests/worker/workflow_sandbox/test_importer.py

@@ -1,7 +1,9 @@
 import dataclasses
 import importlib
 import importlib.machinery
+import subprocess
 import sys
+import textwrap
 import types
 from typing import Any
 
@@ -23,6 +25,71 @@
 from .testmodules import restrictions
 
 
+@pytest.mark.parametrize("first", ["host", "sandbox"])
+def test_workflow_sandbox_importer_pydantic_constraints(first: str):
+    pytest.importorskip("pydantic", minversion="2")
+    # Pydantic caches constraint classes process-wide, so each creation order
+    # needs a fresh process to catch constraints being ignored on either side.
+    result = subprocess.run(
+        [
+            sys.executable,
+            "-c",
+            textwrap.dedent(
+                """
+                import sys
+                from pydantic import BaseModel, Field, ValidationError, conint
+                from temporalio.worker.workflow_sandbox._importer import Importer
+                from temporalio.worker.workflow_sandbox._restrictions import (
+                    RestrictionContext, SandboxRestrictions,
+                )
+
+                def on_host():
+                    class Host(BaseModel):
+                        field: int = Field(ge=0)
+                        constrained: conint(ge=0)
+                    return Host
+
+                def in_sandbox():
+                    with Importer(
+                        SandboxRestrictions.default, RestrictionContext()
+                    ).applied():
+                        from typing import Annotated
+                        import annotated_types
+
+                        class Sandboxed(BaseModel):
+                            value: Annotated[int, annotated_types.Ge(1)]
+                    return Sandboxed
+
+                if sys.argv[1] == "host":
+                    host, sandboxed = on_host(), in_sandbox()
+                else:
+                    sandboxed, host = in_sandbox(), on_host()
+
+                assert sandboxed(value=1).value == 1
+                assert host(field=0, constrained=0).field == 0
+                for model, values, field in (
+                    (sandboxed, {"value": 0}, "value"),
+                    (host, {"field": -1, "constrained": 0}, "field"),
+                    (host, {"field": 0, "constrained": -1}, "constrained"),
+                ):
+                    try:
+                        model(**values)
+                    except ValidationError as err:
+                        assert err.errors()[0]["loc"] == (field,)
+                        assert err.errors()[0]["type"] == "greater_than_equal"
+                    else:
+                        raise AssertionError(f"Constraint ignored for {field}")
+                """
+            ),
+            first,
+        ],
+        capture_output=True,
+        text=True,
+        timeout=30,
+    )
+    assert result.returncode == 0, result.stdout + result.stderr
+
+
 def test_workflow_sandbox_importer_invalid_module():
     with pytest.raises(RestrictedWorkflowAccessError) as err:
         with Importer(restrictions, RestrictionContext()).applied():