Apply Google ADK deterministic providers inside workflow tasks

CHANGELOG.md

@@ -20,6 +20,9 @@ to include examples, links to docs, or any other relevant information.
 
 ### Added
 
+- `workflow.uuid4()` now accepts an optional keyword-only `rng` argument to derive the
+  UUID from a caller-supplied generator (e.g. a private stream from `workflow.new_random()`)
+  without reading or advancing any workflow state.
 - **Experimental**: `temporalio.contrib.google_adk_agents` now supports ADK v2
   graph workflows, dynamic `@node` workflows, and durable HITL.
 - **Experimental**: Experimental support for _Event Groups_. **Event Groups** is a new form of
@@ -44,14 +47,13 @@ to include examples, links to docs, or any other relevant information.
 
 ### :boom: Breaking Changes
 
-- The `google-adk` extra now requires `google-adk>=2.8.0,<3`, up from `>=2.2.0`.
-- `temporalio.contrib.google_adk_agents`: ADK-generated ids and retry jitter now draw from the
-  workflow's deterministic random stream. A workflow started under an earlier release that calls
-  `workflow.random()` or `workflow.uuid4()` after ADK code may not replay deterministically
-  across the upgrade; drain such workflows or use worker versioning.
+- The `google-adk` extra now requires `google-adk>=2.8.0,<3`, up from `>=2.2.0`; 2.8.0 is the
+  first release with the `google.adk.platform._random` seam the plugin now installs a provider for.
 
 ### Fixed
 
+- `GoogleAdkPlugin` now applies its deterministic time, id, and random providers inside
+  workflow tasks.
 - `GoogleAdkPlugin` now passes the optional `anthropic`, `litellm`, and `openai` SDKs through
   the workflow sandbox.
 - `contrib.deepagents`: prevent duplicate input messages after continue-as-new.

temporalio/contrib/google_adk_agents/README.md

@@ -38,13 +38,14 @@ ADK provides: (from the [ADK overview](https://google.github.io/adk-docs/#learn-
 ### OpenTelemetry Integration
 - Automatic instrumentation for ADK components when exporters are provided
 - Tracing integration that works within Temporal's execution context
-- Support for custom span exporters
 
 ### Key Features
 
 #### 1. Deterministic Runtime
-- Replaces `time.time()` with `workflow.now()` when in workflow context
-- Replaces `uuid.uuid4()` with `workflow.uuid4()` for deterministic IDs
+- Installs ADK's `google.adk.platform` time, uuid, and random providers as process-wide defaults, so they apply inside workflow tasks (which run on worker threads with an empty `contextvars` context)
+- Inside a workflow, time comes from `workflow.time()` and ids and randoms come from a workflow-private deterministic stream (a `workflow.new_random()` cached per run), so ADK-generated session, event, invocation, and function-call ids and retry jitter are reproducible on replay without shifting the sequences user code sees from `workflow.random()` and `workflow.uuid4()`. In read-only contexts (query handlers, update validators) time comes from the wall clock and ids and randoms come from nondeterministic entropy that leaves the private stream untouched
+- Outside a workflow in the same process (activities, client code) they fall back to the standard library
+- Overrides through ADK's `set_*_provider` functions must be made after the Worker starts or from workflow code; one made earlier is replaced (with a warning) when the plugin installs its providers, and `reset_*_provider` restores the deterministic providers rather than the standard-library ones
 - Automatic setup when using `GoogleAdkPlugin`
 
 #### 2. Activity-Based Model Execution

temporalio/contrib/google_adk_agents/_plugin.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
+import contextvars
 import dataclasses
 import inspect
 import random
+import threading
 import time
 import uuid
 import warnings
@@ -37,14 +39,6 @@
 from temporalio.worker.workflow_sandbox import SandboxedWorkflowRunner
 
 
-def _install_provider(module: Any, var_name: str, provider: Callable[[], Any]) -> None:
-    """Rebinds an ADK platform ContextVar so ``provider`` is its default in every context."""
-    from contextvars import ContextVar
-
-    context_var = getattr(module, var_name)
-    setattr(module, var_name, ContextVar(context_var.name, default=provider))
-
-
 def _stacklevel_outside_temporalio() -> int:
     # Attribute provider warnings to the nearest frame outside temporalio,
     # e.g. the user's Worker(...)/Replayer(...) call or a user plugin that
@@ -105,61 +99,152 @@ def _warn_if_global_otel_providers_not_replay_safe() -> None:
         )
 
 
-def setup_deterministic_runtime():
-    """Configures ADK runtime for Temporal determinism.
-
-    .. warning::
-        This function is experimental and may change in future versions.
-        Use with caution in production environments.
-
-    Installs Temporal-aware time, uuid, and random providers as the
-    process-wide defaults for ADK's ``google.adk.platform`` seams. Inside a
-    workflow they derive from ``workflow.now()`` / ``workflow.uuid4()`` /
-    ``workflow.random()`` so replays are deterministic; outside a workflow
-    they fall back to the real primitives.
+def _deterministic_time_provider() -> float:
+    # Read-only contexts (query handlers, update validators) get wall-clock
+    # time: their results are never replayed, and workflow.time() would hand
+    # them the last activation's timestamp, which is stale by however long the
+    # workflow has been parked.
+    if workflow.in_workflow() and not workflow.unsafe.is_read_only():
+        return workflow.time()
+    return time.time()
+
+
+_ADK_RANDOM_ATTR = "__temporal_adk_random"
+
+
+def _workflow_adk_random() -> random.Random:
+    # ADK draws from a private stream (a workflow.new_random() cached on the
+    # workflow instance, as the opentelemetry and langsmith integrations do)
+    # rather than sharing workflow.random(), so how many values ADK consumes
+    # never shifts the sequence user code sees. The read-only check must come
+    # first, so a query handler can never touch the cached stream: read-only
+    # contexts (query handlers, update validators) get a fresh unseeded
+    # generator instead, since their results are never replayed while a draw
+    # from the cached stream would advance it and diverge later activations
+    # from replay.
+    if workflow.unsafe.is_read_only():
+        return random.Random()
+    inst = workflow.instance()
+    rng: random.Random | None = getattr(inst, _ADK_RANDOM_ATTR, None)
+    if rng is None:
+        rng = workflow.new_random()
+        setattr(inst, _ADK_RANDOM_ATTR, rng)
+    return rng
+
+
+def _deterministic_id_provider() -> str:
+    if workflow.in_workflow():
+        return str(workflow.uuid4(rng=_workflow_adk_random()))
+    return str(uuid.uuid4())
+
+
+def _deterministic_random_provider() -> random.Random:
+    # Outside a workflow, a fresh unseeded generator per call. ADK's
+    # set_random_provider docstring asks providers to return an existing
+    # instance so a seeded generator keeps its sequence across get_random()
+    # calls; an unseeded one draws fresh OS entropy either way, and ADK's only
+    # caller uses the result immediately (retry jitter).
+    if workflow.in_workflow():
+        return _workflow_adk_random()
+    return random.Random()
+
+
+_install_provider_lock = threading.Lock()
+
+
+def _install_provider(
+    module: Any, var_name: str, default_name: str, provider: Callable[[], Any]
+) -> None:
+    """Makes ``provider`` an ADK platform seam's default, everywhere.
+
+    ADK's ``set_*_provider`` functions set a value in the calling context only.
+    Workflow tasks run on worker threads, which start with an empty
+    contextvars context, so a value set from the worker's event loop never
+    reaches them and ADK falls back to its wall-clock and random defaults
+    there. A ContextVar's default, unlike a set value, is visible from every
+    context, so the module's variable is replaced with one that defaults to
+    ``provider``. The module's ``_default_*`` binding is rebound too, because
+    ``reset_*_provider`` restores that binding: without this, an override
+    followed by a reset would land on the standard-library provider rather
+    than back on ``provider``. ADK's ``set_*_provider`` and
+    ``reset_*_provider`` operate on the new variable from then on; a value set
+    on the old one beforehand is orphaned, so it is warned about. A no-op when
+    ``provider`` is already installed.
     """
+    current: contextvars.ContextVar[Callable[[], Any]] = getattr(module, var_name)
     try:
-        import google.adk.platform._random
-        import google.adk.platform.time
-        import google.adk.platform.uuid
+        default = contextvars.Context().run(current.get)
+    except LookupError:
+        default = None
+    if default is provider and getattr(module, default_name) is provider:
+        return
+    if current.get(default) is not default:
+        warnings.warn(
+            f"Replacing the {module.__name__} provider set in this context before "
+            "GoogleAdkPlugin installed its deterministic providers; it will not "
+            "take effect. Set ADK provider overrides after the worker starts or "
+            "from workflow code.",
+            UserWarning,
+            stacklevel=_stacklevel_outside_temporalio(),
+        )
+    setattr(module, default_name, provider)
+    setattr(module, var_name, contextvars.ContextVar(current.name, default=provider))
 
-        # Define safer, context-aware providers
-        def _deterministic_time_provider() -> float:
-            if workflow.in_workflow():
-                return workflow.now().timestamp()
-            return time.time()
 
-        def _deterministic_id_provider() -> str:
-            if workflow.in_workflow():
-                return str(workflow.uuid4())
-            return str(uuid.uuid4())
+def setup_deterministic_runtime() -> None:
+    """Installs Temporal's deterministic time, id, and random providers for ADK.
 
-        _local_random = random.Random()
+    .. warning::
+        This function is experimental and may change in future versions.
+        Use with caution in production environments.
 
-        def _deterministic_random_provider() -> random.Random:
-            if workflow.in_workflow():
-                return workflow.random()
-            return _local_random
+    The providers become the process-wide defaults of ADK's
+    ``google.adk.platform`` time, uuid, and random seams, so they apply inside
+    workflow tasks (which run on worker threads with an empty contextvars
+    context) as well as in the calling context. Inside a workflow, time comes
+    from ``workflow.time()``, and ids and randoms come from a workflow-private
+    deterministic stream (a ``workflow.new_random()`` cached on the workflow
+    instance), so ADK-generated ids and retry jitter are reproducible on
+    replay without shifting the sequence user code sees from
+    ``workflow.random()`` and ``workflow.uuid4()``. In read-only contexts
+    (query handlers, update validators) time comes from the wall clock and ids
+    and randoms come from a nondeterministic fallback stream that leaves the
+    private stream untouched, since read-only results are never replayed.
+    Outside a workflow in the same process (activities, client code) they fall
+    back to ``time.time()``, ``uuid.uuid4()``, and an unseeded
+    ``random.Random()``.
+
+    Overrides through ADK's ``set_*_provider`` functions must be made after
+    this runs (after the worker starts, or from workflow code); one made
+    earlier is replaced, with a warning. ADK's ``reset_*_provider`` functions
+    restore these deterministic providers, not the standard-library ones.
+
+    :class:`GoogleAdkPlugin` calls this when a worker or replayer starts.
+    Calling it again is a no-op.
+    """
+    import google.adk.platform._random
+    import google.adk.platform.time
+    import google.adk.platform.uuid
 
+    with _install_provider_lock:
         _install_provider(
             google.adk.platform.time,
             "_time_provider_context_var",
+            "_default_time_provider",
             _deterministic_time_provider,
         )
         _install_provider(
             google.adk.platform.uuid,
             "_id_provider_context_var",
+            "_default_id_provider",
             _deterministic_id_provider,
         )
         _install_provider(
             google.adk.platform._random,
             "_random_provider_context_var",
+            "_default_random_provider",
             _deterministic_random_provider,
         )
-    except ImportError:
-        pass
-    except Exception as e:
-        print(f"Warning: Failed to set deterministic runtime providers: {e}")
 
 
 class GoogleAdkPlugin(SimplePlugin):
@@ -170,8 +255,13 @@ class GoogleAdkPlugin(SimplePlugin):
         Use with caution in production environments.
 
     This plugin configures:
+
     - Pydantic Payload Converter (required for ADK objects).
     - Sandbox Passthrough for google.adk and google.genai modules.
+    - ADK's time, id, and random providers, so ADK-generated ids and retry
+      jitter come from the workflow's deterministic clock and a
+      workflow-private deterministic random stream
+      (see :func:`setup_deterministic_runtime`).
 
     At worker and replayer configuration time it also warns when the global
     OpenTelemetry meter or tracer provider is not replay-safe, since ADK

temporalio/contrib/langsmith/_interceptor.py

@@ -238,7 +238,7 @@ def _get_workflow_random() -> random.Random | None:
 
 def _uuid_from_random(rng: random.Random) -> uuid.UUID:
     """Generate a deterministic UUID4 from a workflow-bound random generator."""
-    return uuid.UUID(int=rng.getrandbits(128), version=4)
+    return temporalio.workflow.uuid4(rng=rng)
 
 
 # ---------------------------------------------------------------------------

temporalio/workflow/_context.py

@@ -949,16 +949,24 @@ def upsert_search_attributes(
     )
 
 
-def uuid4() -> uuid.UUID:
+def uuid4(*, rng: Random | None = None) -> uuid.UUID:
     """Get a new, determinism-safe v4 UUID based on :py:func:`random`.
 
     Note, this UUID is not cryptographically safe and should not be used for
     security purposes.
 
+    Args:
+        rng: Generator to draw from instead of the workflow's shared one,
+            e.g. a private stream from :py:func:`new_random`. When provided,
+            no workflow state is read or advanced, so this form also works in
+            read-only contexts and outside a workflow.
+
     Returns:
-        A deterministically-seeded v4 UUID.
+        A v4 UUID deterministically derived from the generator.
     """
-    return uuid.UUID(bytes=random().getrandbits(16 * 8).to_bytes(16, "big"), version=4)
+    if rng is None:
+        rng = random()
+    return uuid.UUID(bytes=rng.getrandbits(16 * 8).to_bytes(16, "big"), version=4)
 
 
 def uuid7() -> uuid.UUID:

tests/contrib/google_adk_agents/test_adk_graph_workflows.py

@@ -279,8 +279,9 @@ class JitteredRetryGraphWorkflow:
     """A retried node with default-style jitter must replay deterministically.
 
     Retry jitter feeds asyncio.sleep, i.e. a durable timer; unless the delay is
-    drawn from workflow.random() (via ADK's platform random seam), replays
-    compute a different timer duration and diverge.
+    drawn from the workflow's deterministic random stream (the plugin's
+    provider behind ADK's platform random seam), replays compute a different
+    timer duration and diverge.
     """
 
     @workflow.run
@@ -504,7 +505,7 @@ async def test_graph_node_retry_jitter_replay_safe(client: Client):
         assert result == "ok-after-2"
         history = await handle.fetch_history()
     # The jittered retry delay is a durable timer; replay must recompute the
-    # exact same duration from workflow.random().
+    # exact same duration from the plugin's deterministic random provider.
     await Replayer(
         workflows=[JitteredRetryGraphWorkflow], plugins=[GoogleAdkPlugin()]
     ).replay_workflow(history)

tests/contrib/google_adk_agents/test_adk_hitl.py

@@ -406,8 +406,9 @@ async def test_tool_confirmation_activity_as_tool(client: Client, confirmed: boo
     # max_cached_workflows=0 forces a full history replay on every workflow
     # task, proving the confirmation resume is replay-safe: the recorded human
     # response references the confirmation function-call id, which must
-    # regenerate identically on replay (it derives from workflow.uuid4() via
-    # the platform uuid seam the plugin installs).
+    # regenerate identically on replay (it derives from the workflow's
+    # deterministic random stream via the platform uuid seam the plugin
+    # installs).
     async with _worker(client):
         LLMRegistry.register(ConfirmationModel)
         handle = await client.start_workflow(