diff --git a/CHANGELOG.md b/CHANGELOG.md index a84835436..63215c787 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,9 @@ to include examples, links to docs, or any other relevant information. ### Added +- `workflow.new_random()` accepts an optional `name` that is mixed into the seed, so differently + named generators, and `workflow.random()`, produce different sequences. + ### Changed - Payload converters exposed by data converters and workflow/activity accessors @@ -30,6 +33,11 @@ to include examples, links to docs, or any other relevant information. ### :boom: Breaking Changes +- `temporalio.contrib.google_adk_agents`: ADK-generated ids and retry jitter now draw from a + workflow-private deterministic stream (a `workflow.new_random()` per run) instead of + `workflow.random()`. A workflow started under 1.34.0 that generated + ADK ids or jitter (for example one waiting on a HITL response) may not replay + deterministically across this upgrade; drain such workflows or use worker versioning. - The OpenAI Agents integration has moved to the independently versioned [`temporalio-openai-agents`](https://pypi.org/project/temporalio-openai-agents/) package. The existing `temporalio[openai-agents]` extra now installs that @@ -41,6 +49,9 @@ to include examples, links to docs, or any other relevant information. ### Fixed +- `GoogleAdkPlugin`'s deterministic providers now work in read-only contexts (query handlers, + update validators), returning the workflow's deterministic time and fresh entropy without + touching the workflow's random state. - `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. diff --git a/temporalio/contrib/google_adk_agents/README.md b/temporalio/contrib/google_adk_agents/README.md index c2691e318..4d3392b42 100644 --- a/temporalio/contrib/google_adk_agents/README.md +++ b/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 is still `workflow.time()` and ids and randoms come from fresh 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 @@ -385,13 +386,13 @@ instead (for example inside an activity or an MCP toolset factory). > generated interrupt/function-call ids, so those ids must regenerate > identically on replay. The plugin installs ADK's platform time/uuid/random > providers as process-wide defaults, so the ids ADK generates (including -> default `RequestInput` interrupt ids) derive from `workflow.uuid4()` and -> replay identically. +> default `RequestInput` interrupt ids) derive from the workflow-private +> deterministic stream and replay identically. ## Determinism Notes - The plugin patches ADK's `google.adk.platform` time, uuid, and random - providers to `workflow.now()`, `workflow.uuid4()`, and `workflow.random()` + providers to `workflow.time()` and a workflow-private deterministic stream inside workflows. - ADK node `timeout=`/`RetryConfig` map onto durable timers (`asyncio.wait_for`/`asyncio.sleep`). For activity-backed nodes, prefer diff --git a/temporalio/contrib/google_adk_agents/_plugin.py b/temporalio/contrib/google_adk_agents/_plugin.py index ad33b34e1..83e161610 100644 --- a/temporalio/contrib/google_adk_agents/_plugin.py +++ b/temporalio/contrib/google_adk_agents/_plugin.py @@ -1,11 +1,14 @@ from __future__ import annotations +import contextvars import dataclasses import inspect import random +import threading import time import uuid import warnings +import weakref from collections.abc import AsyncIterator, Callable from contextlib import asynccontextmanager from types import FrameType @@ -37,14 +40,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 +100,165 @@ 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: + # workflow.time() in every in-workflow context, read-only ones included: a + # dynamic workflow's ``dynamic_config`` runs read-only and is replayed, so + # a wall-clock value there would not be replay-safe. In a query handler + # the value is the current activation's timestamp rather than the wall + # clock, which is harmless because nothing a query computes is persisted. + if workflow.in_workflow(): + return workflow.time() + return time.time() - .. 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. +# Each run's private stream, keyed by the SDK's per-run runtime object: that +# exists during the workflow's __init__ (workflow.instance() does not yet) and +# leaves the user's class alone (it may use __slots__). Entries go away with +# the run. +_adk_randoms: weakref.WeakKeyDictionary[workflow._Runtime, random.Random] = ( + weakref.WeakKeyDictionary() +) +_adk_randoms_lock = threading.Lock() + + +def _workflow_adk_random() -> random.Random: + # ADK draws from a private, named workflow.new_random() stream rather than + # sharing workflow.random(): ADK's draw count never shifts the sequence user + # code sees, and the name keeps the two from coinciding (an unnamed stream + # starts out identical to workflow.random(), so the Nth ADK id would equal + # the Nth workflow.uuid4()). Read-only code (query handlers, update + # validators) must not touch that stream, since a draw there would advance + # it and diverge later activations from replay; as in the opentelemetry id + # generator, it gets a fresh unseeded generator instead. + if workflow.unsafe.is_read_only(): + return random.Random() + runtime = workflow._Runtime.current() + with _adk_randoms_lock: + rng = _adk_randoms.get(runtime) + if rng is None: + rng = workflow.new_random("temporalio.contrib.google_adk_agents") + _adk_randoms[runtime] = rng + return rng + + +def _uuid4_from(rng: random.Random) -> uuid.UUID: + # Same construction as workflow.uuid4(), drawn from the given stream. + return uuid.UUID(bytes=rng.getrandbits(16 * 8).to_bytes(16, "big"), version=4) + + +def _deterministic_id_provider() -> str: + if workflow.in_workflow(): + return str(_uuid4_from(_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()`` per run; ids are v4 + UUIDs built from that stream), 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 is still ``workflow.time()``, + while ids and randoms come from fresh entropy 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 +269,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, google.genai, and OpenTelemetry 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 diff --git a/temporalio/workflow/_context.py b/temporalio/workflow/_context.py index 37928afb9..41eebb2c1 100644 --- a/temporalio/workflow/_context.py +++ b/temporalio/workflow/_context.py @@ -879,21 +879,32 @@ def register_random_seed_callback(callback: Callable[[int], None]) -> None: return _Runtime.current().workflow_register_random_seed_callback(callback) -def new_random() -> Random: +def new_random(name: str | None = None) -> Random: """Create a Random instance that automatically reseeds when the workflow seed changes. This creates a new Random instance that is initially seeded with the current workflow seed, and automatically registers a callback to reseed itself whenever the workflow receives a new seed from core. + Args: + name: Mixed into the seed when given, so differently named instances, + and :py:func:`random`, produce different sequences. Without it the + instance starts out identical to :py:func:`random`. + Returns: A Random instance that stays synchronized with the workflow's randomness. """ - current_seed = random_seed() - auto_random = Random(current_seed) + + def seed_for(workflow_seed: int) -> int | str: + if name is None: + # Unchanged: the same integer seed as :py:func:`random`. + return workflow_seed + return f"{workflow_seed}:{name}" + + auto_random = Random(seed_for(random_seed())) def reseed_callback(new_seed: int) -> None: - auto_random.seed(new_seed) + auto_random.seed(seed_for(new_seed)) register_random_seed_callback(reseed_callback) return auto_random diff --git a/tests/contrib/google_adk_agents/test_adk_graph_workflows.py b/tests/contrib/google_adk_agents/test_adk_graph_workflows.py index 084cc8398..1ea804287 100644 --- a/tests/contrib/google_adk_agents/test_adk_graph_workflows.py +++ b/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) diff --git a/tests/contrib/google_adk_agents/test_adk_hitl.py b/tests/contrib/google_adk_agents/test_adk_hitl.py index d4fbf054a..59ea626e4 100644 --- a/tests/contrib/google_adk_agents/test_adk_hitl.py +++ b/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( 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..7f2330d33 --- /dev/null +++ b/tests/contrib/google_adk_agents/test_adk_platform_providers.py @@ -0,0 +1,540 @@ +"""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 +import warnings +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, _plugin +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_private_cached_stream: bool + workflow_stream_unperturbed: bool + distinct_from_user_stream: 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: the standard-library ``_default_*`` + provider and a fresh ContextVar defaulting to it. The plugin rebinds + both, so both must be restored. + """ + adk_time._default_time_provider = time.time + adk_time._time_provider_context_var = contextvars.ContextVar( + "time_provider", default=adk_time._default_time_provider + ) + adk_uuid._default_id_provider = lambda: str(uuid.uuid4()) + adk_uuid._id_provider_context_var = contextvars.ContextVar( + "id_provider", default=adk_uuid._default_id_provider + ) + adk_random._default_random_provider = lambda: adk_random._default_random + 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: + # ADK ids and randoms come from a private named stream created via + # workflow.new_random() on first use, so a mirror made the same way + # reproduces the id from the same 128 bits. + adk_id = adk_uuid.new_uuid() + mirror = workflow.new_random("temporalio.contrib.google_adk_agents") + expected_id = str(uuid.UUID(int=mirror.getrandbits(128), version=4)) + adk_rng = adk_random.get_random() + # The private stream and workflow.random() start from the same seed, + # so if ADK's id draw had gone through workflow.random(), the user + # stream's next value would no longer match a fresh same-seed stream. + probe = workflow.new_random() + readings = PlatformProviderReadings( + adk_time=adk_time.get_time(), + workflow_time=workflow.time(), + adk_id=adk_id, + expected_id=expected_id, + random_is_private_cached_stream=( + adk_rng is not workflow.random() and adk_random.get_random() is adk_rng + ), + workflow_stream_unperturbed=workflow.random().random() == probe.random(), + # An unnamed stream starts out as workflow.random() does, so its first + # 128 bits are what workflow.uuid4() would mint first for user code. + distinct_from_user_stream=adk_id + != str(uuid.UUID(int=workflow.new_random().getrandbits(128), version=4)), + ) + unsandboxed_readings.append(readings) + return readings + + @workflow.query + def query_adk_id(self) -> str: + # Read-only contexts get nondeterministic entropy; the cached private + # stream must stay untouched (QueryDuringRunWorkflow proves that). + return adk_uuid.new_uuid() + + +@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() + # Read-only fallback: a query still gets a valid (nondeterministic) + # uuid rather than an error. + assert uuid.UUID(await handle.query(PlatformProviderWorkflow.query_adk_id)) + history = await handle.fetch_history() + + assert readings.adk_time == readings.workflow_time + assert readings.adk_id == readings.expected_id + assert readings.random_is_private_cached_stream + assert readings.workflow_stream_unperturbed + assert readings.distinct_from_user_stream + + # 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] + + +@dataclass +class SetResetReadings: + overridden_time: float + time_after_reset: float + workflow_time: float + overridden_id: str + id_after_reset: str + overridden_random_was_adk_default: bool + random_after_reset_is_private_stream: bool + + +# Same hand-back mechanism as unsandboxed_readings above. +unsandboxed_set_reset_readings: list[SetResetReadings] = [] + + +@workflow.defn +class SetResetProviderWorkflow: + """Exercises ADK's public set-then-reset cycle inside a workflow. + + reset_*_provider() restores the module's _default_* binding, so the + plugin must have rebound that too: otherwise a reset lands on the + standard-library provider and the rest of the run is nondeterministic. + """ + + @workflow.run + async def run(self) -> SetResetReadings: + private_rng = adk_random.get_random() + + adk_time.set_time_provider(lambda: -1.0) + overridden_time = adk_time.get_time() + adk_time.reset_time_provider() + + adk_uuid.set_id_provider(lambda: "fixed-id") + overridden_id = adk_uuid.new_uuid() + adk_uuid.reset_id_provider() + + # ADK's shipped default instance still exists on the module; use it + # as the override to avoid constructing randomness in workflow code. + adk_random.set_random_provider(lambda: adk_random._default_random) + overridden_random_was_adk_default = ( + adk_random.get_random() is adk_random._default_random + ) + adk_random.reset_random_provider() + after_reset_rng = adk_random.get_random() + + readings = SetResetReadings( + overridden_time=overridden_time, + time_after_reset=adk_time.get_time(), + workflow_time=workflow.time(), + overridden_id=overridden_id, + id_after_reset=adk_uuid.new_uuid(), + overridden_random_was_adk_default=overridden_random_was_adk_default, + random_after_reset_is_private_stream=( + after_reset_rng is private_rng + and after_reset_rng is not adk_random._default_random + ), + ) + unsandboxed_set_reset_readings.append(readings) + return readings + + +@pytest.mark.parametrize( + "workflow_runner", + [SandboxedWorkflowRunner(), UnsandboxedWorkflowRunner()], + ids=["sandboxed", "unsandboxed"], +) +async def test_reset_in_workflow_restores_deterministic_providers( + 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-set-reset-{uuid.uuid4()}" + async with Worker( + client, + task_queue=task_queue, + workflows=[SetResetProviderWorkflow], + workflow_runner=workflow_runner, + ): + handle = await client.start_workflow( + SetResetProviderWorkflow.run, + id=f"adk-set-reset-{uuid.uuid4()}", + task_queue=task_queue, + execution_timeout=timedelta(seconds=60), + ) + readings = await handle.result() + history = await handle.fetch_history() + + assert readings.overridden_time == -1.0 + assert readings.time_after_reset == readings.workflow_time + assert readings.overridden_id == "fixed-id" + assert uuid.UUID(readings.id_after_reset).version == 4 + assert readings.overridden_random_was_adk_default + assert readings.random_after_reset_is_private_stream + + # If reset had restored wall-clock/stdlib providers, the post-reset + # readings could not reproduce from history. + reset_adk_providers_to_shipped_state() + unsandboxed_set_reset_readings.clear() + await Replayer( + workflows=[SetResetProviderWorkflow], + plugins=[GoogleAdkPlugin()], + workflow_runner=UnsandboxedWorkflowRunner(), + ).replay_workflow(history) + assert unsandboxed_set_reset_readings == [readings] + + +# Same hand-back mechanism as unsandboxed_readings above. +unsandboxed_query_run_ids: list[list[str]] = [] + + +@workflow.defn +class QueryDuringRunWorkflow: + """Proves query-handler draws never advance the private ADK stream. + + The run draws one id, waits for a signal (queries happen here), then + draws another. Queries do not run during replay, so if a query had + advanced the cached stream, the second id could not reproduce on replay. + """ + + def __init__(self) -> None: + self.proceed = False + + @workflow.run + async def run(self) -> list[str]: + ids = [adk_uuid.new_uuid()] + await workflow.wait_condition(lambda: self.proceed) + ids.append(adk_uuid.new_uuid()) + unsandboxed_query_run_ids.append(ids) + return ids + + @workflow.signal + def go(self) -> None: + self.proceed = True + + @workflow.query + def query_adk_id(self) -> str: + return adk_uuid.new_uuid() + + @workflow.query + def query_adk_time(self) -> list[float]: + # Read-only contexts keep the deterministic workflow clock: a dynamic + # workflow's ``dynamic_config`` is read-only too and is replayed. + return [adk_time.get_time(), workflow.time()] + + +@pytest.mark.parametrize( + "workflow_runner", + [SandboxedWorkflowRunner(), UnsandboxedWorkflowRunner()], + ids=["sandboxed", "unsandboxed"], +) +async def test_query_draws_do_not_advance_private_stream( + 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-query-stream-{uuid.uuid4()}" + async with Worker( + client, + task_queue=task_queue, + workflows=[QueryDuringRunWorkflow], + workflow_runner=workflow_runner, + ): + handle = await client.start_workflow( + QueryDuringRunWorkflow.run, + id=f"adk-query-stream-{uuid.uuid4()}", + task_queue=task_queue, + execution_timeout=timedelta(seconds=60), + ) + # Draw through the read-only fallback between the run's two draws. + for _ in range(3): + assert uuid.UUID(await handle.query(QueryDuringRunWorkflow.query_adk_id)) + # get_time() in a read-only context is still workflow.time(). + adk_queried_time, workflow_queried_time = await handle.query( + QueryDuringRunWorkflow.query_adk_time + ) + assert adk_queried_time == workflow_queried_time + await handle.signal(QueryDuringRunWorkflow.go) + ids = await handle.result() + history = await handle.fetch_history() + + assert len(ids) == 2 and ids[0] != ids[1] + + # Replay never runs the queries; the ids only reproduce if the query + # draws left the private stream untouched. + reset_adk_providers_to_shipped_state() + unsandboxed_query_run_ids.clear() + await Replayer( + workflows=[QueryDuringRunWorkflow], + plugins=[GoogleAdkPlugin()], + workflow_runner=UnsandboxedWorkflowRunner(), + ).replay_workflow(history) + assert unsandboxed_query_run_ids == [ids] + + +def test_reset_outside_workflow_restores_installed_provider() -> None: + reset_adk_providers_to_shipped_state() + _plugin.setup_deterministic_runtime() + + def set_reset_read() -> None: + adk_time.set_time_provider(lambda: 1.0) + assert adk_time.get_time() == 1.0 + adk_time.reset_time_provider() + assert ( + adk_time._time_provider_context_var.get() + is _plugin._deterministic_time_provider + ) + adk_uuid.set_id_provider(lambda: "fixed-id") + adk_uuid.reset_id_provider() + assert ( + adk_uuid._id_provider_context_var.get() + is _plugin._deterministic_id_provider + ) + adk_random.set_random_provider(lambda: adk_random._default_random) + adk_random.reset_random_provider() + assert ( + adk_random._random_provider_context_var.get() + is _plugin._deterministic_random_provider + ) + + # Run in a copied context so the overrides do not leak into other tests. + contextvars.copy_context().run(set_reset_read) + + +def test_providers_are_defaults_visible_from_new_threads() -> None: + reset_adk_providers_to_shipped_state() + _plugin.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 _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: + _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 + ) + # A fresh unseeded generator, not a workflow stream. + assert isinstance(adk_random.get_random(), random.Random) + + +def test_setup_deterministic_runtime_is_idempotent() -> None: + _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 + + _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: + _plugin.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 + + +@workflow.defn +class IdInInitWorkflow: + def __init__(self) -> None: + # workflow.instance() is still None here, so the private stream cannot + # live on the workflow object. + self._init_id = adk_uuid.new_uuid() + + @workflow.run + async def run(self) -> list[str]: + return [self._init_id, adk_uuid.new_uuid()] + + +@workflow.defn +class SlottedIdWorkflow: + # No instance attributes can be added to this object. + __slots__ = () + + @workflow.run + async def run(self) -> str: + return adk_uuid.new_uuid() + + +@pytest.mark.parametrize( + "workflow_runner", + [SandboxedWorkflowRunner(), UnsandboxedWorkflowRunner()], + ids=["sandboxed", "unsandboxed"], +) +async def test_ids_in_init_and_in_slotted_workflow( + 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()}" + async with Worker( + client, + task_queue=task_queue, + workflows=[IdInInitWorkflow, SlottedIdWorkflow], + workflow_runner=workflow_runner, + ): + init_id, run_id = await client.execute_workflow( + IdInInitWorkflow.run, + id=f"adk-platform-providers-{uuid.uuid4()}", + task_queue=task_queue, + execution_timeout=timedelta(seconds=30), + ) + slotted_id = await client.execute_workflow( + SlottedIdWorkflow.run, + id=f"adk-platform-providers-{uuid.uuid4()}", + task_queue=task_queue, + execution_timeout=timedelta(seconds=30), + ) + + assert uuid.UUID(init_id).version == 4 + assert uuid.UUID(run_id).version == 4 + assert init_id != run_id + assert uuid.UUID(slotted_id).version == 4 diff --git a/tests/contrib/google_adk_agents/test_google_adk_agents.py b/tests/contrib/google_adk_agents/test_google_adk_agents.py index c0aef0ba4..5d3cabaa8 100644 --- a/tests/contrib/google_adk_agents/test_google_adk_agents.py +++ b/tests/contrib/google_adk_agents/test_google_adk_agents.py @@ -102,7 +102,7 @@ async def run(self, prompt: str, model_name: str) -> Event | None: app_name="test_app", ) - # 3. Create Session (uses runtime.new_uuid() -> workflow.uuid4()) + # 3. Create Session (uses runtime.new_uuid() -> the plugin's private stream) logger.info("Create session.") session = await runner.session_service.create_session( app_name="test_app", user_id="test" @@ -421,7 +421,7 @@ async def run(self, prompt: str, model_name: str) -> str: # 1. Define Agent using Temporal Helpers agent = mcp_agent(model_name) - # 2. Create Session (uses runtime.new_uuid() -> workflow.uuid4()) + # 2. Create Session (uses runtime.new_uuid() -> the plugin's private stream) session_service = InMemorySessionService() logger.info("Create session.") session = await session_service.create_session(