From 2cd39b5171ba7a80213af8d070076309d8198710 Mon Sep 17 00:00:00 2001 From: DABH Date: Fri, 11 Sep 2026 01:07:28 -0500 Subject: [PATCH 1/2] Apply Google ADK deterministic providers inside workflow tasks ADK keeps its time, id, and random providers in contextvars.ContextVars. GoogleAdkPlugin set them in the worker's context, but workflow tasks run on the workflow task executor's threads, which start with an empty context, so ADK code inside a workflow read the defaults: wall-clock time and uuid.uuid4() for session, event, invocation, and function-call ids. Only debug mode, which runs activations inline, saw the deterministic values. Rebind each google.adk.platform ContextVar to one whose default is the Temporal provider so it is visible from every context, on Worker and Replayer alike. Also install the random provider ADK added in 2.8.0 and raise the google-adk floor to 2.8.0. --- CHANGELOG.md | 12 ++ pyproject.toml | 2 +- .../contrib/google_adk_agents/README.md | 6 +- .../contrib/google_adk_agents/_plugin.py | 116 ++++++++--- .../test_adk_platform_providers.py | 194 ++++++++++++++++++ uv.lock | 71 +++---- 6 files changed, 337 insertions(+), 64 deletions(-) create mode 100644 tests/contrib/google_adk_agents/test_adk_platform_providers.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 14a225277..71a34f37a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -37,6 +37,11 @@ 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. - Experimental external storage: `ExternalStorage.driver_selector` is now called with a `StorageDriverSelectContext` instead of a `StorageDriverStoreContext`. Update the annotation; the new type carries the same `target` field. Since selectors are plain callables, a stale @@ -52,6 +57,13 @@ to include examples, links to docs, or any other relevant information. ### Fixed +- `GoogleAdkPlugin` now applies its deterministic time, id, and random providers inside + workflow tasks. ADK reads them from `contextvars` and workflow tasks run on worker threads + whose context is empty, so on a standard `Worker` or `Replayer` ADK-generated session, + event, invocation, and function-call ids came from wall-clock time and `uuid.uuid4()`; only + debug mode, which runs tasks inline, saw the deterministic values. The providers are now + installed as process-wide defaults that fall back to the real clock and RNG outside a + workflow. - **Experimental**: External storage metrics now report the wall-clock time storage was in flight. Previously each batch's duration was summed, over-reporting the time whenever storage operations ran concurrently. diff --git a/pyproject.toml b/pyproject.toml index 27114dc67..3d0e095ee 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,7 +29,7 @@ grpc = ["grpcio>=1.48.2,<2"] opentelemetry = ["opentelemetry-api>=1.26,<2", "opentelemetry-sdk>=1.26,<2"] pydantic = ["pydantic>=2.0.0,<3"] openai-agents = ["openai-agents>=0.19.2,<0.20", "mcp>=1.9.4, <2"] -google-adk = ["google-adk>=2.2.0,<3", "mcp>=1.24,<2"] +google-adk = ["google-adk>=2.8.0,<3", "mcp>=1.24,<2"] langgraph = ["langgraph>=1.1.0"] langsmith = ["langsmith>=0.7.34,<0.9"] deepagents = [ diff --git a/temporalio/contrib/google_adk_agents/README.md b/temporalio/contrib/google_adk_agents/README.md index 294e5dddd..9d644e890 100644 --- a/temporalio/contrib/google_adk_agents/README.md +++ b/temporalio/contrib/google_adk_agents/README.md @@ -38,13 +38,13 @@ 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 the providers return `workflow.now()`, `workflow.uuid4()`, and `workflow.random()`, so ADK-generated session, event, invocation, and function-call ids and retry jitter are reproducible on replay +- Outside a workflow (for example `adk run` or `adk web`) they fall back to `time.time()`, `uuid.uuid4()`, and a process-wide `random.Random` - Automatic setup when using `GoogleAdkPlugin` #### 2. Activity-Based Model Execution diff --git a/temporalio/contrib/google_adk_agents/_plugin.py b/temporalio/contrib/google_adk_agents/_plugin.py index 36515aa1a..ec06b06cf 100644 --- a/temporalio/contrib/google_adk_agents/_plugin.py +++ b/temporalio/contrib/google_adk_agents/_plugin.py @@ -1,7 +1,10 @@ from __future__ import annotations +import contextvars import dataclasses import inspect +import random +import threading import time import uuid import warnings @@ -96,37 +99,97 @@ def _warn_if_global_otel_providers_not_replay_safe() -> None: ) -def setup_deterministic_runtime(): - """Configures ADK runtime for Temporal determinism. +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()) + + +# ADK's own default is one process-wide random.Random, and its +# set_random_provider docstring asks providers to return an existing instance +# so RNG state carries across get_random() calls; keep one for outside +# workflows too. +_random_outside_workflow = random.Random() + + +def _deterministic_random_provider() -> random.Random: + if workflow.in_workflow(): + return workflow.random() + return _random_outside_workflow + + +_install_provider_lock = threading.Lock() + + +def _install_provider(module: Any, var_name: str, provider: Callable[[], Any]) -> None: + """Rebinds an ADK platform ContextVar to one whose default is ``provider``. + + 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``. It keeps its name, and ADK's ``set_*_provider`` and + ``reset_*_provider`` keep working on it. A no-op when ``provider`` is + already the default. + """ + current: contextvars.ContextVar[Callable[[], Any]] = getattr(module, var_name) + try: + installed = contextvars.Context().run(current.get) is provider + except LookupError: + installed = False + if not installed: + setattr( + module, var_name, contextvars.ContextVar(current.name, default=provider) + ) + + +def setup_deterministic_runtime() -> None: + """Installs Temporal's deterministic time, id, and random providers for ADK. .. warning:: This function is experimental and may change in future versions. Use with caution in production environments. - This should be called at the start of a Temporal Workflow before any ADK components - (like SessionService) are used, if they rely on runtime.get_time() or runtime.new_uuid(). + 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 they return + ``workflow.now()``, ``workflow.uuid4()``, and ``workflow.random()``, so + ADK-generated ids and retry jitter are reproducible on replay; outside a + workflow they fall back to ``time.time()``, ``uuid.uuid4()``, and a + process-wide ``random.Random``. + + :class:`GoogleAdkPlugin` calls this when a worker or replayer starts. + Calling it again is a no-op. """ - try: - import google.adk.platform.time - import google.adk.platform.uuid - - # 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()) - - google.adk.platform.time.set_time_provider(_deterministic_time_provider) - google.adk.platform.uuid.set_id_provider(_deterministic_id_provider) - except ImportError: - pass - except Exception as e: - print(f"Warning: Failed to set deterministic runtime providers: {e}") + 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", + _deterministic_time_provider, + ) + _install_provider( + google.adk.platform.uuid, + "_id_provider_context_var", + _deterministic_id_provider, + ) + _install_provider( + google.adk.platform._random, + "_random_provider_context_var", + _deterministic_random_provider, + ) class GoogleAdkPlugin(SimplePlugin): @@ -139,6 +202,9 @@ class GoogleAdkPlugin(SimplePlugin): 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 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 diff --git a/tests/contrib/google_adk_agents/test_adk_platform_providers.py b/tests/contrib/google_adk_agents/test_adk_platform_providers.py new file mode 100644 index 000000000..2d01c8b6e --- /dev/null +++ b/tests/contrib/google_adk_agents/test_adk_platform_providers.py @@ -0,0 +1,194 @@ +"""Tests that GoogleAdkPlugin's deterministic providers reach workflow code. + +ADK reads its time, id, and random providers from contextvars.ContextVars. +Workflow tasks run on the worker's thread pool, whose threads start with an +empty context, so a provider merely set in the worker's context is invisible +there and ADK falls back to wall-clock time and random UUIDs. The plugin must +install the providers so that they are visible from every context. +""" + +import contextvars +import random +import time +import uuid +from concurrent.futures import ThreadPoolExecutor +from dataclasses import dataclass +from datetime import timedelta + +import pytest +from google.adk.platform import _random as adk_random +from google.adk.platform import time as adk_time +from google.adk.platform import uuid as adk_uuid + +from temporalio import workflow +from temporalio.client import Client +from temporalio.contrib.google_adk_agents import GoogleAdkPlugin +from temporalio.contrib.google_adk_agents._plugin import setup_deterministic_runtime +from temporalio.worker import ( + Replayer, + UnsandboxedWorkflowRunner, + Worker, + WorkflowRunner, +) +from temporalio.worker.workflow_sandbox import SandboxedWorkflowRunner + + +@dataclass +class PlatformProviderReadings: + adk_time: float + workflow_time: float + adk_id: str + expected_id: str + random_is_workflow_random: bool + + +# Appended to by PlatformProviderWorkflow when it runs on an unsandboxed +# runner, which shares this module with the test (the sandbox imports its own +# copy). Lets a Replayer run hand its readings back to the test. +unsandboxed_readings: list[PlatformProviderReadings] = [] + + +def reset_adk_providers_to_shipped_state() -> None: + """Undo any earlier plugin install so a test proves its own install. + + Rebuilds each ADK seam as it ships: a fresh ContextVar defaulting to + ADK's own provider. + """ + adk_time._time_provider_context_var = contextvars.ContextVar( + "time_provider", default=adk_time._default_time_provider + ) + adk_uuid._id_provider_context_var = contextvars.ContextVar( + "id_provider", default=adk_uuid._default_id_provider + ) + adk_random._random_provider_context_var = contextvars.ContextVar( + "random_provider", default=adk_random._default_random_provider + ) + + +@workflow.defn +class PlatformProviderWorkflow: + @workflow.run + async def run(self) -> PlatformProviderReadings: + rng = workflow.random() + # new_uuid() and workflow.uuid4() both consume the random stream, so + # rewind it in between: from the same state they must agree. + state = rng.getstate() + adk_id = adk_uuid.new_uuid() + rng.setstate(state) + readings = PlatformProviderReadings( + adk_time=adk_time.get_time(), + workflow_time=workflow.now().timestamp(), + adk_id=adk_id, + expected_id=str(workflow.uuid4()), + random_is_workflow_random=adk_random.get_random() is rng, + ) + unsandboxed_readings.append(readings) + return readings + + +@pytest.mark.parametrize( + "workflow_runner", + [SandboxedWorkflowRunner(), UnsandboxedWorkflowRunner()], + ids=["sandboxed", "unsandboxed"], +) +async def test_providers_apply_inside_workflow_tasks( + client: Client, workflow_runner: WorkflowRunner +) -> None: + reset_adk_providers_to_shipped_state() + new_config = client.config() + new_config["plugins"] = [GoogleAdkPlugin()] + client = Client(**new_config) + + task_queue = f"adk-platform-providers-{uuid.uuid4()}" + # Not debug mode, so activations run on the workflow task executor's + # threads as they do in production. + async with Worker( + client, + task_queue=task_queue, + workflows=[PlatformProviderWorkflow], + workflow_runner=workflow_runner, + ): + handle = await client.start_workflow( + PlatformProviderWorkflow.run, + id=f"adk-platform-providers-{uuid.uuid4()}", + task_queue=task_queue, + execution_timeout=timedelta(seconds=60), + ) + readings = await handle.result() + history = await handle.fetch_history() + + assert readings.adk_time == readings.workflow_time + assert readings.adk_id == readings.expected_id + assert readings.random_is_workflow_random + + # The values derive from history, so a replay reproduces them exactly. + # Replay unsandboxed so the workflow can hand its readings back. + reset_adk_providers_to_shipped_state() + unsandboxed_readings.clear() + await Replayer( + workflows=[PlatformProviderWorkflow], + plugins=[GoogleAdkPlugin()], + workflow_runner=UnsandboxedWorkflowRunner(), + ).replay_workflow(history) + assert unsandboxed_readings == [readings] + + +def test_providers_are_defaults_visible_from_new_threads() -> None: + reset_adk_providers_to_shipped_state() + setup_deterministic_runtime() + + def read_providers() -> tuple[object, object, object]: + # A new thread starts with an empty context; make that explicit so the + # check does not depend on the interpreter's thread-inheritance flag. + return contextvars.Context().run( + lambda: ( + adk_time._time_provider_context_var.get(), + adk_uuid._id_provider_context_var.get(), + adk_random._random_provider_context_var.get(), + ) + ) + + with ThreadPoolExecutor(max_workers=1) as executor: + time_provider, id_provider, random_provider = executor.submit( + read_providers + ).result() + + assert time_provider is not adk_time._default_time_provider + assert id_provider is not adk_uuid._default_id_provider + assert random_provider is not adk_random._default_random_provider + + +def test_providers_fall_back_outside_workflow() -> None: + setup_deterministic_runtime() + + assert adk_time.get_time() == pytest.approx(time.time(), abs=5) + assert uuid.UUID(adk_uuid.new_uuid()).version == 4 + rng = adk_random.get_random() + assert isinstance(rng, random.Random) + # One shared instance, so RNG state carries across calls as ADK expects. + assert adk_random.get_random() is rng + + +def test_setup_deterministic_runtime_is_idempotent() -> None: + setup_deterministic_runtime() + time_var = adk_time._time_provider_context_var + id_var = adk_uuid._id_provider_context_var + random_var = adk_random._random_provider_context_var + + setup_deterministic_runtime() + + assert adk_time._time_provider_context_var is time_var + assert adk_uuid._id_provider_context_var is id_var + assert adk_random._random_provider_context_var is random_var + + +def test_adk_setters_still_override_in_calling_context() -> None: + setup_deterministic_runtime() + + def override_and_read() -> float: + adk_time.set_time_provider(lambda: 1.0) + return adk_time.get_time() + + # Run in a copied context so the override does not leak into other tests. + assert contextvars.copy_context().run(override_and_read) == 1.0 + assert adk_time.get_time() != 1.0 diff --git a/uv.lock b/uv.lock index da13829ce..f49587292 100644 --- a/uv.lock +++ b/uv.lock @@ -9,7 +9,7 @@ resolution-markers = [ ] [options] -exclude-newer = "2026-08-10T18:40:15.391197Z" +exclude-newer = "0001-01-01T00:00:00Z" # This has no effect and is included for backwards compatibility when using relative exclude-newer values. exclude-newer-span = "P2W" [[package]] @@ -257,14 +257,14 @@ name = "anthropic" version = "0.117.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "anyio", marker = "python_full_version >= '3.11'" }, - { name = "distro", marker = "python_full_version >= '3.11'" }, - { name = "docstring-parser", marker = "python_full_version >= '3.11'" }, - { name = "httpx", marker = "python_full_version >= '3.11'" }, - { name = "jiter", marker = "python_full_version >= '3.11'" }, - { name = "pydantic", marker = "python_full_version >= '3.11'" }, - { name = "sniffio", marker = "python_full_version >= '3.11'" }, - { name = "typing-extensions", marker = "python_full_version >= '3.11'" }, + { name = "anyio" }, + { name = "distro" }, + { name = "docstring-parser" }, + { name = "httpx" }, + { name = "jiter" }, + { name = "pydantic" }, + { name = "sniffio" }, + { name = "typing-extensions" }, ] sdist = { url = "https://files.pythonhosted.org/packages/41/0d/8f71d535edb0d438f023bd825fb65f67c14fa88a2bd6b75f292a58a63de4/anthropic-0.117.0.tar.gz", hash = "sha256:98107f2b76439641e0ae2a1754087534b8f178dbab99d6eb1bc4b7bc8c744496", size = 989933, upload-time = "2026-07-16T19:36:13.07Z" } wheels = [ @@ -942,12 +942,12 @@ name = "deepagents" version = "0.6.12" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "langchain", marker = "python_full_version >= '3.11'" }, - { name = "langchain-anthropic", marker = "python_full_version >= '3.11'" }, - { name = "langchain-core", marker = "python_full_version >= '3.11'" }, - { name = "langchain-google-genai", marker = "python_full_version >= '3.11'" }, - { name = "langsmith", marker = "python_full_version >= '3.11'" }, - { name = "wcmatch", marker = "python_full_version >= '3.11'" }, + { name = "langchain" }, + { name = "langchain-anthropic" }, + { name = "langchain-core" }, + { name = "langchain-google-genai" }, + { name = "langsmith" }, + { name = "wcmatch" }, ] sdist = { url = "https://files.pythonhosted.org/packages/e0/db/a6acdc72a9e90c3f07ed10de35c951734a02d4facb693bb59684ad368801/deepagents-0.6.12.tar.gz", hash = "sha256:1f281c0bc5a63132f62e2ee345c1dc593b23188da6e23016401f6879fbe54b5f", size = 211364, upload-time = "2026-06-25T17:26:52.775Z" } wheels = [ @@ -1022,7 +1022,7 @@ name = "exceptiongroup" version = "1.3.1" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "typing-extensions", marker = "python_full_version < '3.11'" }, + { name = "typing-extensions" }, ] sdist = { url = "https://files.pythonhosted.org/packages/50/79/66800aadf48771f6b62f7eb014e352e5d06856655206165d775e675a02c9/exceptiongroup-1.3.1.tar.gz", hash = "sha256:8b412432c6055b0b7d14c310000ae93352ed6754f70fa8f7c34141f91c4e3219", size = 30371, upload-time = "2025-11-21T23:01:54.787Z" } wheels = [ @@ -1298,9 +1298,10 @@ wheels = [ [[package]] name = "google-adk" -version = "2.4.0" +version = "2.8.0" source = { registry = "https://pypi.org/simple" } dependencies = [ + { name = "aiohttp" }, { name = "aiosqlite" }, { name = "authlib" }, { name = "click" }, @@ -1326,9 +1327,9 @@ dependencies = [ { name = "watchdog" }, { name = "websockets" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/2a/a1/6048b1c22817859bafc1101f8ba26f704233d9acb07e715dff5fb41b9b55/google_adk-2.4.0.tar.gz", hash = "sha256:5a2996b288d591deefcb277eeeeb7da838d72056675763bfef52ad3b36975dde", size = 3566788, upload-time = "2026-07-07T19:46:14.802Z" } +sdist = { url = "https://files.pythonhosted.org/packages/c7/29/db8042eb489515ef64fc16d24f1a621523451bfc2cbbae0103527b4066f4/google_adk-2.8.0.tar.gz", hash = "sha256:f51524e18cf0a0cdeb4fdd6f0fa16f31bc5e53021647b3e6c72a90f40f583915", size = 3902380, upload-time = "2026-08-26T23:26:20.501Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/fd/ab/12ec18054990ac69f37dae8327f5d8d5e04557bada0d8d725afd872d3020/google_adk-2.4.0-py3-none-any.whl", hash = "sha256:fba91f1a693e5fc2fd13dc40d625562bd52e44a7baaeecedb01811a68063d847", size = 4123277, upload-time = "2026-07-07T19:46:13.026Z" }, + { url = "https://files.pythonhosted.org/packages/0c/9d/8447a0912dcba1fa5dae83697e1af0a7ca9afa4701f93c82f3a1caf20561/google_adk-2.8.0-py3-none-any.whl", hash = "sha256:616bfa21959ae2726432670cb0b1c549e4908d9bf6908bff6fbc08382b075429", size = 4502706, upload-time = "2026-08-26T23:26:17.53Z" }, ] [[package]] @@ -1354,7 +1355,7 @@ requests = [ [[package]] name = "google-genai" -version = "2.11.0" +version = "2.20.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "anyio" }, @@ -1368,9 +1369,9 @@ dependencies = [ { name = "typing-extensions" }, { name = "websockets" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/a6/01/e7b5f3aac89200c78318ed7643401e7f5ed3131b0cd353c07483606b1e61/google_genai-2.11.0.tar.gz", hash = "sha256:4c5e524d24b145c96be327f9a7f8f04b0fe4efee0533877795e9848afed01749", size = 622366, upload-time = "2026-07-09T17:49:43.862Z" } +sdist = { url = "https://files.pythonhosted.org/packages/87/dd/eacd43097318ea6b3e648862713a964d5de261a2eabcc7826db9b9de9758/google_genai-2.20.0.tar.gz", hash = "sha256:d382186f024e9050a7a4b25af6eacba9aa16c6e09594f5d1b530f22ff7f9d76f", size = 664965, upload-time = "2026-08-25T21:28:27.136Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/93/ef/d296c23390160a8b0b1dafb36dd3cb36a39ed40c81cd27e04e6233334186/google_genai-2.11.0-py3-none-any.whl", hash = "sha256:5bc8186100e1d34d691fbe0cba392b7e04e98d286ca952323a6672d054accf95", size = 984162, upload-time = "2026-07-09T17:49:42.15Z" }, + { url = "https://files.pythonhosted.org/packages/ae/a7/a979230234c9df019e008085c923726dc4d92c14a5701ad698e369c9ab2a/google_genai-2.20.0-py3-none-any.whl", hash = "sha256:49bddeccd29a4e6bf1706c5de67735f7115f537f08b6c36a70b8023c99399095", size = 1064276, upload-time = "2026-08-25T21:28:25.287Z" }, ] [[package]] @@ -1971,9 +1972,9 @@ name = "langchain" version = "1.3.14" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "langchain-core", marker = "python_full_version >= '3.11'" }, - { name = "langgraph", marker = "python_full_version >= '3.11'" }, - { name = "pydantic", marker = "python_full_version >= '3.11'" }, + { name = "langchain-core" }, + { name = "langgraph" }, + { name = "pydantic" }, ] sdist = { url = "https://files.pythonhosted.org/packages/29/68/a6dbad9c22df4087a0f9e79ddd46226c442b30128bfeee538d5889492a73/langchain-1.3.14.tar.gz", hash = "sha256:1b6696c72ba3bbbce54d745e0180742c9f6ece8bbc59ed5a46c3e20b9a435929", size = 645181, upload-time = "2026-07-16T13:28:18.29Z" } wheels = [ @@ -1985,9 +1986,9 @@ name = "langchain-anthropic" version = "1.4.8" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "anthropic", marker = "python_full_version >= '3.11'" }, - { name = "langchain-core", marker = "python_full_version >= '3.11'" }, - { name = "pydantic", marker = "python_full_version >= '3.11'" }, + { name = "anthropic" }, + { name = "langchain-core" }, + { name = "pydantic" }, ] sdist = { url = "https://files.pythonhosted.org/packages/98/22/40ab129b08329ca295b391aa1d48267692b42594757084c6918e22b655ac/langchain_anthropic-1.4.8.tar.gz", hash = "sha256:c76891b2044d56105ff13c106ed12650637b53bd598a4bdf15b4796eefa2a4ec", size = 708524, upload-time = "2026-06-26T21:28:46.916Z" } wheels = [ @@ -2019,10 +2020,10 @@ name = "langchain-google-genai" version = "4.2.7" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "filetype", marker = "python_full_version >= '3.11'" }, - { name = "google-genai", marker = "python_full_version >= '3.11'" }, - { name = "langchain-core", marker = "python_full_version >= '3.11'" }, - { name = "pydantic", marker = "python_full_version >= '3.11'" }, + { name = "filetype" }, + { name = "google-genai" }, + { name = "langchain-core" }, + { name = "pydantic" }, ] sdist = { url = "https://files.pythonhosted.org/packages/1d/0c/bc60dabc362ca7c6ffe8c4bcc2f724c7e566b43eb230cee51419f88f784c/langchain_google_genai-4.2.7.tar.gz", hash = "sha256:03b1463ffe4d42435f43c7870467f2215f684bb46400d2543435d10157c80ac7", size = 281605, upload-time = "2026-07-06T13:51:58.724Z" } wheels = [ @@ -2838,7 +2839,7 @@ wheels = [ [package.optional-dependencies] litellm = [ - { name = "litellm", marker = "python_full_version < '3.14'" }, + { name = "litellm" }, ] [[package]] @@ -4812,7 +4813,7 @@ dev = [ requires-dist = [ { name = "aioboto3", marker = "extra == 'aioboto3'", specifier = ">=10.4.0" }, { name = "deepagents", marker = "python_full_version >= '3.11' and extra == 'deepagents'", specifier = ">=0.6.12,<0.7" }, - { name = "google-adk", marker = "extra == 'google-adk'", specifier = ">=2.2.0,<3" }, + { name = "google-adk", marker = "extra == 'google-adk'", specifier = ">=2.8.0,<3" }, { name = "google-genai", marker = "extra == 'google-genai'", specifier = ">=2.10.0,<3.0.0" }, { name = "grpcio", marker = "extra == 'grpc'", specifier = ">=1.48.2,<2" }, { name = "langchain", marker = "python_full_version >= '3.11' and extra == 'deepagents'", specifier = ">=1.3.11,<2" }, @@ -5377,7 +5378,7 @@ name = "wcmatch" version = "11.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "bracex", marker = "python_full_version >= '3.11'" }, + { name = "bracex" }, ] sdist = { url = "https://files.pythonhosted.org/packages/16/25/1da725838132221e33568973da484ff43813662ccc06ebf7f6e3abddfcd5/wcmatch-11.0.tar.gz", hash = "sha256:55d95c2447789712774b198ceec72939e88b5618f1f8f0a9b605bf7740b63b96", size = 141360, upload-time = "2026-07-10T05:50:24.183Z" } wheels = [ From 593912d440e4d7820530bfc3308c63bd8e5490a1 Mon Sep 17 00:00:00 2001 From: DABH Date: Fri, 11 Sep 2026 01:29:44 -0500 Subject: [PATCH 2/2] Address review on ADK provider install Use workflow.time() for the time provider, warn when installing replaces a provider set earlier in the calling context, and document that overrides must be made after the worker starts and that ADK id and random generation raise ReadOnlyContextError in read-only contexts. Tests assert provider identity. --- CHANGELOG.md | 7 ++- .../contrib/google_adk_agents/README.md | 5 +- .../contrib/google_adk_agents/_plugin.py | 40 +++++++++----- .../test_adk_platform_providers.py | 54 ++++++++++++++----- 4 files changed, 78 insertions(+), 28 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 71a34f37a..118f5f040 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -37,7 +37,8 @@ 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`. +- 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. - `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 @@ -63,7 +64,9 @@ to include examples, links to docs, or any other relevant information. event, invocation, and function-call ids came from wall-clock time and `uuid.uuid4()`; only debug mode, which runs tasks inline, saw the deterministic values. The providers are now installed as process-wide defaults that fall back to the real clock and RNG outside a - workflow. + workflow. As with `workflow.uuid4()` and `workflow.random()`, ADK id generation and + `get_random()` inside a query handler or update validator now raise `ReadOnlyContextError` + rather than returning a random value. - **Experimental**: External storage metrics now report the wall-clock time storage was in flight. Previously each batch's duration was summed, over-reporting the time whenever storage operations ran concurrently. diff --git a/temporalio/contrib/google_adk_agents/README.md b/temporalio/contrib/google_adk_agents/README.md index 9d644e890..7c99952c9 100644 --- a/temporalio/contrib/google_adk_agents/README.md +++ b/temporalio/contrib/google_adk_agents/README.md @@ -43,8 +43,9 @@ ADK provides: (from the [ADK overview](https://google.github.io/adk-docs/#learn- #### 1. Deterministic Runtime - 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 the providers return `workflow.now()`, `workflow.uuid4()`, and `workflow.random()`, so ADK-generated session, event, invocation, and function-call ids and retry jitter are reproducible on replay -- Outside a workflow (for example `adk run` or `adk web`) they fall back to `time.time()`, `uuid.uuid4()`, and a process-wide `random.Random` +- Inside a workflow the providers return `workflow.time()`, `workflow.uuid4()`, and `workflow.random()`, so ADK-generated session, event, invocation, and function-call ids and retry jitter are reproducible on replay. Like those functions, ADK id generation and `get_random()` raise `ReadOnlyContextError` inside query handlers and update validators +- 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 - Automatic setup when using `GoogleAdkPlugin` #### 2. Activity-Based Model Execution diff --git a/temporalio/contrib/google_adk_agents/_plugin.py b/temporalio/contrib/google_adk_agents/_plugin.py index ec06b06cf..0c4d1d117 100644 --- a/temporalio/contrib/google_adk_agents/_plugin.py +++ b/temporalio/contrib/google_adk_agents/_plugin.py @@ -101,7 +101,7 @@ def _warn_if_global_otel_providers_not_replay_safe() -> None: def _deterministic_time_provider() -> float: if workflow.in_workflow(): - return workflow.now().timestamp() + return workflow.time() return time.time() @@ -136,19 +136,28 @@ def _install_provider(module: Any, var_name: str, provider: Callable[[], Any]) - 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``. It keeps its name, and ADK's ``set_*_provider`` and - ``reset_*_provider`` keep working on it. A no-op when ``provider`` is - already the default. + ``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 the + default. """ current: contextvars.ContextVar[Callable[[], Any]] = getattr(module, var_name) try: - installed = contextvars.Context().run(current.get) is provider + default = contextvars.Context().run(current.get) except LookupError: - installed = False - if not installed: - setattr( - module, var_name, contextvars.ContextVar(current.name, default=provider) + default = None + if default 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, var_name, contextvars.ContextVar(current.name, default=provider)) def setup_deterministic_runtime() -> None: @@ -162,11 +171,18 @@ def setup_deterministic_runtime() -> None: ``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 they return - ``workflow.now()``, ``workflow.uuid4()``, and ``workflow.random()``, so - ADK-generated ids and retry jitter are reproducible on replay; outside a - workflow they fall back to ``time.time()``, ``uuid.uuid4()``, and a + ``workflow.time()``, ``workflow.uuid4()``, and ``workflow.random()``, so + ADK-generated ids and retry jitter are reproducible on replay; like those + functions, id and random generation raise + :class:`temporalio.workflow.ReadOnlyContextError` in query handlers and + update validators. Outside a workflow in the same process (activities, + client code) they fall back to ``time.time()``, ``uuid.uuid4()``, and a process-wide ``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. + :class:`GoogleAdkPlugin` calls this when a worker or replayer starts. Calling it again is a no-op. """ diff --git a/tests/contrib/google_adk_agents/test_adk_platform_providers.py b/tests/contrib/google_adk_agents/test_adk_platform_providers.py index 2d01c8b6e..41631c0ec 100644 --- a/tests/contrib/google_adk_agents/test_adk_platform_providers.py +++ b/tests/contrib/google_adk_agents/test_adk_platform_providers.py @@ -11,6 +11,7 @@ import random import time import uuid +import warnings from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass from datetime import timedelta @@ -22,8 +23,7 @@ from temporalio import workflow from temporalio.client import Client -from temporalio.contrib.google_adk_agents import GoogleAdkPlugin -from temporalio.contrib.google_adk_agents._plugin import setup_deterministic_runtime +from temporalio.contrib.google_adk_agents import GoogleAdkPlugin, _plugin from temporalio.worker import ( Replayer, UnsandboxedWorkflowRunner, @@ -77,7 +77,7 @@ async def run(self) -> PlatformProviderReadings: rng.setstate(state) readings = PlatformProviderReadings( adk_time=adk_time.get_time(), - workflow_time=workflow.now().timestamp(), + workflow_time=workflow.time(), adk_id=adk_id, expected_id=str(workflow.uuid4()), random_is_workflow_random=adk_random.get_random() is rng, @@ -135,7 +135,7 @@ async def test_providers_apply_inside_workflow_tasks( def test_providers_are_defaults_visible_from_new_threads() -> None: reset_adk_providers_to_shipped_state() - setup_deterministic_runtime() + _plugin.setup_deterministic_runtime() def read_providers() -> tuple[object, object, object]: # A new thread starts with an empty context; make that explicit so the @@ -153,37 +153,67 @@ def read_providers() -> tuple[object, object, object]: read_providers ).result() - assert time_provider is not adk_time._default_time_provider - assert id_provider is not adk_uuid._default_id_provider - assert random_provider is not adk_random._default_random_provider + assert time_provider is _plugin._deterministic_time_provider + assert id_provider is _plugin._deterministic_id_provider + assert random_provider is _plugin._deterministic_random_provider def test_providers_fall_back_outside_workflow() -> None: - setup_deterministic_runtime() + _plugin.setup_deterministic_runtime() + assert ( + adk_time._time_provider_context_var.get() + is _plugin._deterministic_time_provider + ) assert adk_time.get_time() == pytest.approx(time.time(), abs=5) + assert adk_uuid._id_provider_context_var.get() is _plugin._deterministic_id_provider assert uuid.UUID(adk_uuid.new_uuid()).version == 4 + assert ( + adk_random._random_provider_context_var.get() + is _plugin._deterministic_random_provider + ) + # One shared instance, so RNG state carries across calls as ADK expects. rng = adk_random.get_random() assert isinstance(rng, random.Random) - # One shared instance, so RNG state carries across calls as ADK expects. + assert rng is _plugin._random_outside_workflow assert adk_random.get_random() is rng def test_setup_deterministic_runtime_is_idempotent() -> None: - setup_deterministic_runtime() + _plugin.setup_deterministic_runtime() time_var = adk_time._time_provider_context_var id_var = adk_uuid._id_provider_context_var random_var = adk_random._random_provider_context_var - setup_deterministic_runtime() + _plugin.setup_deterministic_runtime() assert adk_time._time_provider_context_var is time_var assert adk_uuid._id_provider_context_var is id_var assert adk_random._random_provider_context_var is random_var +def test_install_warns_when_replacing_provider_set_before_install() -> None: + reset_adk_providers_to_shipped_state() + + def set_then_install() -> None: + adk_time.set_time_provider(lambda: 1.0) + with pytest.warns(UserWarning, match="set in this context before"): + _plugin.setup_deterministic_runtime() + # The earlier override lives on the replaced variable and is ignored. + assert adk_time.get_time() != 1.0 + + # Run in a copied context so the override does not leak into other tests. + contextvars.copy_context().run(set_then_install) + + # Installing over untouched seams is silent. + reset_adk_providers_to_shipped_state() + with warnings.catch_warnings(): + warnings.simplefilter("error") + _plugin.setup_deterministic_runtime() + + def test_adk_setters_still_override_in_calling_context() -> None: - setup_deterministic_runtime() + _plugin.setup_deterministic_runtime() def override_and_read() -> float: adk_time.set_time_provider(lambda: 1.0)