diff --git a/CHANGELOG.md b/CHANGELOG.md index 03a2243ed..d3851740b 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/temporalio/contrib/google_adk_agents/README.md b/temporalio/contrib/google_adk_agents/README.md index 64ae28fd2..a36e0e222 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 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 diff --git a/temporalio/contrib/google_adk_agents/_plugin.py b/temporalio/contrib/google_adk_agents/_plugin.py index e770d6d11..1c7986a28 100644 --- a/temporalio/contrib/google_adk_agents/_plugin.py +++ b/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 diff --git a/temporalio/contrib/langsmith/_interceptor.py b/temporalio/contrib/langsmith/_interceptor.py index a7eea0714..dbc8c8364 100644 --- a/temporalio/contrib/langsmith/_interceptor.py +++ b/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) # --------------------------------------------------------------------------- diff --git a/temporalio/workflow/_context.py b/temporalio/workflow/_context.py index 37928afb9..caf808c7a 100644 --- a/temporalio/workflow/_context.py +++ b/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: 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..05fdb5ce5 --- /dev/null +++ b/tests/contrib/google_adk_agents/test_adk_platform_providers.py @@ -0,0 +1,471 @@ +"""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 + + +# 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 stream created via + # workflow.new_random() on first use, so a mirror stream made the + # same way reproduces the id from the same 128 bits. + adk_id = adk_uuid.new_uuid() + mirror = workflow.new_random() + 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(), + ) + 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 + + # 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) -> float: + # Read-only contexts get wall-clock time, not the stale timestamp of + # the activation the workflow last parked on. + return adk_time.get_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 returns wall-clock time. + queried_time = await handle.query(QueryDuringRunWorkflow.query_adk_time) + assert queried_time == pytest.approx(time.time(), abs=60) + 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 diff --git a/tests/test_workflow.py b/tests/test_workflow.py index 5618c34f5..4900a8555 100644 --- a/tests/test_workflow.py +++ b/tests/test_workflow.py @@ -588,3 +588,16 @@ async def test_child_workflow_config_parity_with_start_child_workflow(): await workflow.start_child_workflow( "workflow", **workflow.ChildWorkflowConfig() ) + + +def test_uuid4_accepts_explicit_random() -> None: + # workflow.uuid4(rng=...) touches no workflow state, so it works + # outside a workflow and derives the same uuid from the same stream state. + import random + import uuid + + seeded = random.Random(42) + expected = uuid.UUID(int=random.Random(42).getrandbits(128), version=4) + got = workflow.uuid4(rng=seeded) + assert got == expected + assert got.version == 4 diff --git a/uv.lock b/uv.lock index cee51de68..d850bcf98 100644 --- a/uv.lock +++ b/uv.lock @@ -9,7 +9,7 @@ resolution-markers = [ ] [options] -exclude-newer = "2026-08-31T19:12:49.465398Z" +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 = [ @@ -1972,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 = [ @@ -1986,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 = [ @@ -2020,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 = [ @@ -2839,7 +2839,7 @@ wheels = [ [package.optional-dependencies] litellm = [ - { name = "litellm", marker = "python_full_version < '3.14'" }, + { name = "litellm" }, ] [[package]] @@ -5379,7 +5379,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 = [