From 6a92f655897935ce78e9313cccf96503cb2b2b3a Mon Sep 17 00:00:00 2001 From: Brian Strauch Date: Mon, 14 Sep 2026 15:36:37 -0700 Subject: [PATCH 1/5] Add durable Strands sandbox support --- CHANGELOG.md | 6 + pyproject.toml | 4 +- temporalio/contrib/strands/README.md | 200 ++++ temporalio/contrib/strands/__init__.py | 13 + temporalio/contrib/strands/_plugin.py | 44 +- .../contrib/strands/_sandbox_activity.py | 477 +++++++++ temporalio/contrib/strands/_temporal_agent.py | 9 +- .../contrib/strands/_temporal_sandbox.py | 204 ++++ temporalio/contrib/strands/_worker_env_ref.py | 86 ++ tests/contrib/strands/test_sandbox.py | 973 ++++++++++++++++++ uv.lock | 11 +- 11 files changed, 2015 insertions(+), 12 deletions(-) create mode 100644 temporalio/contrib/strands/_sandbox_activity.py create mode 100644 temporalio/contrib/strands/_temporal_sandbox.py create mode 100644 temporalio/contrib/strands/_worker_env_ref.py create mode 100644 tests/contrib/strands/test_sandbox.py diff --git a/CHANGELOG.md b/CHANGELOG.md index f1a526e7e..933a75277 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,6 +24,12 @@ to include examples, links to docs, or any other relevant information. - `ActivityHandle` now supports operator commands for standalone activities: `pause`, `unpause`, `update_options` and `restore_original_options`. +- **Experimental**: `temporalio.contrib.strands` now supports durable, + Workflow-isolated Strands sandboxes through `TemporalSandbox` and + worker-side factories with run and Workflow-chain context registered with + `StrandsPlugin(sandboxes=...)`, using one shared activity set with optional + correlated live Workflow Streams output, worker-side environment references, + and async-context-manager factories for adapter cleanup. - Added GCP Cloud Run serverless-worker OpenTelemetry plugin in `temporalio.contrib.opentelemetry`. - Added new options to ActivityHandle.describe() to retrieve associated payloads, such as activity input and outcome. - New properties and methods in ActivityExecution and ActivityExecutionDescription. diff --git a/pyproject.toml b/pyproject.toml index 27114dc67..4c8615b43 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -56,7 +56,7 @@ cloud-run-worker-otel = [ ] aioboto3 = ["aioboto3>=10.4.0", "types-aioboto3[s3]>=10.4.0"] google-genai = ["google-genai>=2.10.0,<3.0.0"] -strands-agents = ["strands-agents>=1.39.0"] +strands-agents = ["strands-agents>=1.51.0"] [project.urls] Homepage = "https://github.com/temporalio/sdk-python" @@ -107,7 +107,7 @@ dev = [ "opentelemetry-sdk-extension-aws>=2.0.0,<3", "pytest-flakefinder>=1.1.0", "async-timeout>=4.0,<6; python_version < '3.11'", - "strands-agents>=1.39.0", + "strands-agents>=1.51.0", "strands-agents-tools>=0.5.2", "mcp>=1.9.4,<2", ] diff --git a/temporalio/contrib/strands/README.md b/temporalio/contrib/strands/README.md index 5381f29f1..9fe1f91af 100644 --- a/temporalio/contrib/strands/README.md +++ b/temporalio/contrib/strands/README.md @@ -175,6 +175,206 @@ async for item in WorkflowStreamClient.create(client, workflow_id).subscribe( print(item.data) ``` +## Sandboxes + +`TemporalSandbox` implements Strands' sandbox API by scheduling every command, +code, and filesystem operation as a Temporal Activity. Register the real +worker-side sandbox under a name, then select that name in workflow code: + +```python +from strands.sandbox.docker import DockerSandbox +from temporalio.contrib.strands import ( + SandboxWorkflowContext, + StrandsPlugin, + TemporalAgent, + TemporalSandbox, +) + +async def build_sandbox(context: SandboxWorkflowContext) -> DockerSandbox: + # Application-specific and idempotent: return the existing container when + # another activity worker has already provisioned this Workflow's sandbox. + container = await get_or_create_build_container( + context.chain.first_execution_run_id + ) + return DockerSandbox(container.name) + +# workflow +agent = TemporalAgent( + sandbox=TemporalSandbox( + "build", + start_to_close_timeout=timedelta(minutes=5), + ), +) + +# worker +Worker( + ..., + plugins=[StrandsPlugin(sandboxes={ + "build": build_sandbox, + })], +) +``` + +The plugin registers one shared set of sandbox activities regardless of how +many factories are configured. Each operation carries the selected sandbox +name in its activity input so the worker can dispatch it to the matching +factory. + +The factory is called lazily with a `SandboxWorkflowContext` containing the +current `run_id` and a `chain` identity with the Workflow's namespace, Workflow +ID, and first execution Run ID. The worker-local cache uses the sandbox name and +chain identity, so each Workflow chain gets a separate sandbox for each +registered name. Retries, Continue-As-New, Reset, and Cron runs belong to the +same chain and therefore use the same sandbox; unrelated Workflow chains do not +share one. Multiple `TemporalSandbox` objects with the same name in one chain +intentionally share that chain's sandbox. + +The factory receives the current Run ID only when a worker-local cache entry is +created. A later run in the same chain reuses a warm entry without calling the +factory again. After eviction, the next factory call receives the Run ID of the +run that recreates the entry. + +The first execution Run ID is supplied by workflow code because Activity +metadata only identifies the current run. Treat sandbox activities as trusted +worker endpoints: workflow code with access to their task queue can call them +directly and choose this value. Do not share the task queue between mutually +untrusted workflows. If Workflow IDs can be reused, the backing environment +lookup must also prevent a new execution from reconnecting to an old execution's +environment, for example by expiring old environments before ID reuse. + +Factories may be synchronous or asynchronous. Synchronous factories must only +construct a lightweight adapter and must not block the activity event loop; +use an asynchronous factory for remote lookup or provisioning. A factory may +run more than once for the same context after cache eviction or on different +workers, so provisioning must be idempotent. Strands' `DockerSandbox` only +connects to an already-running container; it does not create one. + +Worker-local adapters are reused until they have been idle for five minutes. +Set `sandbox_cache_idle_timeout` on `StrandsPlugin` to change that duration. +Return an async context manager from the factory when an adapter owns clients, +sockets, subprocess handles, tunnels, or leases. Its exit method runs after +cache eviction and Worker shutdown. Returning a plain `Sandbox` only drops the +adapter from the cache: + +```python +from contextlib import asynccontextmanager + +@asynccontextmanager +async def build_managed_sandbox(context: SandboxWorkflowContext): + adapter = await connect_to_sandbox(context) + try: + yield adapter + finally: + await adapter.aclose() +``` + +In either case, provisioning, teardown, and cleanup of orphaned backing +environments remain the application's responsibility; use a backend TTL or +reaper for workflows that are terminated before normal cleanup. + +That cache is per worker *process*, while successive sandbox activities from one +workflow are routed independently across the task queue. With more than one +worker on the queue, a `write-file` can land on one worker and the following +`read-file` on another. The context factory must therefore reconnect every +worker to the same Workflow-scoped backing environment rather than relying on +per-process state. A single worker on the queue also satisfies this. + +Reset does not roll back commands or filesystem mutations already performed in +the external sandbox, just as it does not roll back other Activity side effects. +Account for that when resetting a Workflow that uses a sandbox. + +`SandboxTimeoutError` is serialized as an Activity failure and retried under the +`retry_policy` you pass to `TemporalSandbox`. If retries are exhausted, it is +reconstructed inside the workflow with the sandbox's own message. Any +`FileNotFoundError` raised by a sandbox filesystem operation — including its +`SandboxPathNotFoundError` subclass — is non-retryable because the requested path +is absent, and is reconstructed in the same way. Factory failures and other +sandbox failures, including the `OSError` that Strands documents for a failed +`write_file`, are also retryable. + +Like all Temporal Activities, sandbox operations have at-least-once execution +semantics. A worker can finish a command or filesystem mutation and fail before +recording its result, causing a retry to perform the operation again. Use a +bounded `retry_policy`, and make commands and mutations idempotent when repeated +execution would be unsafe. + +By default, `TemporalSandbox.get_tools()` vends `sandbox_shell` and +`sandbox_file_editor`. A tool passed explicitly through `TemporalAgent(tools=...)` +with either name takes precedence, following Strands' normal sandbox-tool +override behavior. + +Execution output is always buffered into the activity result so workflow replay +observes the same ordered `StreamChunk` and `ExecutionResult` values. For live, +observer-facing output, set `streaming_topic` and host a `WorkflowStream` on the +workflow. The activity publishes each `StreamChunk` as it arrives; the final +`ExecutionResult` is returned only through the buffered activity result: + +```python +from datetime import timedelta + +from temporalio.contrib.strands import SandboxStreamEvent, TemporalSandbox +from temporalio.contrib.workflow_streams import WorkflowStream, WorkflowStreamClient + +# workflow __init__ +self.stream = WorkflowStream() +self.sandbox = TemporalSandbox( + "build", + start_to_close_timeout=timedelta(minutes=5), + streaming_topic="sandbox-events", +) + +# external client +async for item in WorkflowStreamClient.create(client, workflow_id).subscribe( + ["sandbox-events"], result_type=SandboxStreamEvent +): + event = item.data + print(event.execution_id, event.sequence, event.chunk.data) +``` + +Each `SandboxStreamEvent` includes the sandbox name, an execution ID composed of +the Workflow Run ID and Activity ID, the Activity attempt, chunk sequence, and +`StreamChunk`. Events from concurrent executions may interleave on a shared +topic; group them by `execution_id` and `attempt`, then order them by `sequence`. +Workflow code still receives the correctly separated, complete buffered result +for each call. Because publications are observer-facing side effects of an +Activity attempt, a failed attempt may leave chunks in the topic before a retry +publishes its own output. + +Streaming is disabled by default. When `streaming_topic=None`, sandbox +activities do not construct a `WorkflowStreamClient` and the workflow does not +need to host a `WorkflowStream`. + +All sandbox Activity arguments and results are serialized into workflow history. +Keep command output and files within the server's configured payload size limits; +use external storage for large artifacts. + +Do not put secret values directly in `env`, because those values are serialized +into workflow history. Use `temporal_worker_env_ref()` to serialize only the +name of a worker environment variable, then allow that name on every worker that +runs the sandbox activities: + +```python +from temporalio.contrib.strands import temporal_worker_env_ref + +# workflow +await sandbox.execute( + "build", + env={"API_KEY": temporal_worker_env_ref("BUILD_API_KEY")}, +) + +# worker +plugin = StrandsPlugin( + sandboxes={"build": build_sandbox}, + resolvable_worker_env_vars=["BUILD_API_KEY"], +) +``` + +Names are matched exactly. A reference to a name the worker does not allow is +passed to the sandbox unchanged; an allowed but unset variable resolves to an +empty string. `AllowAllWorkerEnvVars()` permits any name, but should only be used +when all workflow code on the task queue is trusted to read every environment +variable available to that worker. + ## Tools Decorate non-deterministic tools with `@activity.defn`, or if you're importing tools from `strands_tools`, wrap them in a thin async function. Then, register the activity on the worker via `Worker(activities=[...])` and pass it to the agent with `workflow.activity_as_tool(activity, **options)` along with any activity options (e.g. `start_to_close_timeout`): diff --git a/temporalio/contrib/strands/__init__.py b/temporalio/contrib/strands/__init__.py index 39a8e7401..70d038079 100644 --- a/temporalio/contrib/strands/__init__.py +++ b/temporalio/contrib/strands/__init__.py @@ -2,12 +2,25 @@ from . import workflow from ._plugin import StrandsPlugin +from ._sandbox_activity import ( + SandboxStreamEvent, + SandboxWorkflowChain, + SandboxWorkflowContext, +) from ._temporal_agent import TemporalAgent from ._temporal_mcp_client import TemporalMCPClient +from ._temporal_sandbox import TemporalSandbox +from ._worker_env_ref import AllowAllWorkerEnvVars, temporal_worker_env_ref __all__ = [ + "AllowAllWorkerEnvVars", + "SandboxStreamEvent", "StrandsPlugin", + "SandboxWorkflowChain", + "SandboxWorkflowContext", "TemporalAgent", "TemporalMCPClient", + "TemporalSandbox", + "temporal_worker_env_ref", "workflow", ] diff --git a/temporalio/contrib/strands/_plugin.py b/temporalio/contrib/strands/_plugin.py index 9d70e10fa..57b0bc954 100644 --- a/temporalio/contrib/strands/_plugin.py +++ b/temporalio/contrib/strands/_plugin.py @@ -1,4 +1,4 @@ -from collections.abc import AsyncGenerator, Callable +from collections.abc import AsyncGenerator, Callable, Collection from contextlib import asynccontextmanager from dataclasses import replace from datetime import timedelta @@ -16,11 +16,13 @@ from ._failure_converter import StrandsFailureConverter from ._model_activity import ModelActivity +from ._sandbox_activity import SandboxActivities, SandboxFactory from ._temporal_mcp_client import ( _evict_connection, build_call_tool_activity, build_list_tools_activity, ) +from ._worker_env_ref import AllowAllWorkerEnvVars def _default_bedrock_model() -> Model: @@ -51,6 +53,16 @@ class StrandsPlugin(SimplePlugin): ``mcp_connection_idle_timeout`` controls how long a worker-process MCP connection is kept open between ``call-tool`` activities before it is disconnected; the timer resets on every reuse. Defaults to 5 minutes. + + When ``sandboxes`` is supplied, registers one stable set of activities that + dispatches each operation by sandbox name. Each factory receives the + requesting Workflow run's context and may return a sandbox directly or + awaitably. Worker-local adapters are cached by sandbox name and Workflow + chain until ``sandbox_cache_idle_timeout`` elapses. Use the same name in + workflow-side ``TemporalSandbox(name)`` instances. + + ``resolvable_worker_env_vars`` controls which worker environment variables + sandbox command ``env`` references may resolve immediately before execution. """ def __init__( @@ -58,12 +70,18 @@ def __init__( *, models: dict[str, Callable[[], Model]] | None = None, mcp_clients: dict[str, Callable[[], MCPClient]] | None = None, + sandboxes: dict[str, SandboxFactory] | None = None, mcp_connection_idle_timeout: timedelta | None = None, + sandbox_cache_idle_timeout: timedelta | None = None, + resolvable_worker_env_vars: Collection[str] | AllowAllWorkerEnvVars = (), ) -> None: - """Build the plugin from optional model and MCP transport factories. + """Build the plugin from optional model, MCP, and sandbox factories. If ``models`` is omitted, registers a single ``BedrockModel`` factory under the name ``"bedrock"`` with Botocore retries disabled. + + A sandbox factory may return an async context manager when its worker-local + adapter needs cleanup after cache eviction or Worker shutdown. """ default_name: str | None = None if models is None: @@ -74,6 +92,18 @@ def __init__( ma = ModelActivity(models, default_name=default_name) activities.extend([ma.invoke_model, ma.invoke_model_streaming]) + sandbox_activities = ( + SandboxActivities( + sandboxes, + sandbox_cache_idle_timeout, + resolvable_worker_env_vars, + ) + if sandboxes + else None + ) + if sandbox_activities is not None: + activities.extend(sandbox_activities.activities()) + mcp_clients = mcp_clients or {} for server, client_factory in mcp_clients.items(): activities.append( @@ -87,11 +117,21 @@ def __init__( ) ) + sandbox_run_contexts = 0 + @asynccontextmanager async def run_context() -> AsyncGenerator[None, None]: + nonlocal sandbox_run_contexts + if sandbox_activities is not None: + sandbox_run_contexts += 1 try: yield finally: + if sandbox_activities is not None: + sandbox_run_contexts -= 1 + # One plugin instance can be shared by multiple Workers. + if sandbox_run_contexts == 0: + await sandbox_activities.aclose() for server in mcp_clients: await _evict_connection(server) diff --git a/temporalio/contrib/strands/_sandbox_activity.py b/temporalio/contrib/strands/_sandbox_activity.py new file mode 100644 index 000000000..1a567e019 --- /dev/null +++ b/temporalio/contrib/strands/_sandbox_activity.py @@ -0,0 +1,477 @@ +from __future__ import annotations + +import asyncio +import base64 +import inspect +import logging +from collections.abc import ( + AsyncGenerator, + Awaitable, + Callable, + Collection, +) +from contextlib import AbstractAsyncContextManager, asynccontextmanager +from dataclasses import dataclass, field +from datetime import timedelta +from typing import Any + +from strands.sandbox import ( + ExecutionResult, + FileInfo, + Sandbox, + StreamChunk, +) +from strands.sandbox.errors import SandboxTimeoutError + +from temporalio import activity +from temporalio.contrib.workflow_streams import WorkflowStreamClient +from temporalio.exceptions import ApplicationError + +from ._heartbeat_decorator import auto_heartbeater +from ._worker_env_ref import AllowAllWorkerEnvVars, _WorkerEnvRefResolver + +SANDBOX_TIMEOUT_ERROR_TYPE = "StrandsSandboxTimeoutError" +SANDBOX_PATH_NOT_FOUND_ERROR_TYPE = "StrandsSandboxPathNotFoundError" +SANDBOX_NOT_FOUND_ERROR_TYPE = "StrandsSandboxNotFoundError" +_SANDBOX_CACHE_IDLE_TIMEOUT = timedelta(minutes=5) +logger = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class SandboxWorkflowChain: + """Identity shared by every run in a Workflow chain.""" + + namespace: str + workflow_id: str + first_execution_run_id: str + + +@dataclass(frozen=True) +class SandboxWorkflowContext: + """Workflow execution requesting a worker-side sandbox.""" + + chain: SandboxWorkflowChain + run_id: str + + @property + def namespace(self) -> str: + """Namespace containing the Workflow.""" + return self.chain.namespace + + @property + def workflow_id(self) -> str: + """Workflow ID shared by the execution chain.""" + return self.chain.workflow_id + + @property + def first_execution_run_id(self) -> str: + """Run ID identifying the execution chain.""" + return self.chain.first_execution_run_id + + +@dataclass(frozen=True) +class SandboxStreamEvent: + """A correlated output chunk from a sandbox execution Activity.""" + + sandbox_name: str + execution_id: str + attempt: int + sequence: int + chunk: StreamChunk + + +_SandboxFactoryResult = Sandbox | AbstractAsyncContextManager[Sandbox] +SandboxFactory = Callable[ + [SandboxWorkflowContext], _SandboxFactoryResult | Awaitable[_SandboxFactoryResult] +] +_SandboxKey = tuple[str, SandboxWorkflowChain] + + +@dataclass +class _WorkflowScopedInput: + sandbox_name: str = field(default="", kw_only=True) + first_execution_run_id: str = field(default="", kw_only=True) + + +@dataclass +class _ExecuteInput(_WorkflowScopedInput): + command: str + timeout: float | None = None + cwd: str | None = None + env: dict[str, str] | None = None + kwargs: dict[str, Any] = field(default_factory=dict) + streaming_topic: str | None = None + streaming_batch_interval_seconds: float = 0.1 + + +@dataclass +class _ExecuteCodeInput(_WorkflowScopedInput): + code: str + language: str + timeout: float | None = None + cwd: str | None = None + env: dict[str, str] | None = None + kwargs: dict[str, Any] = field(default_factory=dict) + streaming_topic: str | None = None + streaming_batch_interval_seconds: float = 0.1 + + +@dataclass +class _PathInput(_WorkflowScopedInput): + path: str + kwargs: dict[str, Any] = field(default_factory=dict) + + +@dataclass +class _WriteFileInput(_PathInput): + content_base64: str = "" + + +@dataclass +class _StreamItem: + value: dict[str, Any] + + +class _SandboxRecord: + def __init__( + self, + owner: SandboxActivities, + key: _SandboxKey, + context: SandboxWorkflowContext, + factory: SandboxFactory, + idle_timeout: timedelta, + ) -> None: + self._owner = owner + self._key = key + self._context = context + self._idle_timeout = idle_timeout + self._inflight = 0 + self._idle_handle: asyncio.TimerHandle | None = None + self._sandbox_context_manager: AbstractAsyncContextManager[Sandbox] | None = ( + None + ) + self._sandbox_task = asyncio.create_task(self._create(factory)) + + async def _create(self, factory: SandboxFactory) -> Sandbox: + result = factory(self._context) + if inspect.isawaitable(result): + result = await result + if isinstance(result, Sandbox): + return result + self._sandbox_context_manager = result + return await result.__aenter__() + + def acquire(self) -> None: + self._inflight += 1 + if self._idle_handle is not None: + self._idle_handle.cancel() + self._idle_handle = None + + def release(self) -> None: + self._inflight -= 1 + if self._inflight == 0 and self._owner._has_record(self._key, self): + self._idle_handle = asyncio.get_running_loop().call_later( + self._idle_timeout.total_seconds(), self._on_idle + ) + + def _on_idle(self) -> None: + self._idle_handle = None + if self._inflight == 0: + self._owner._evict(self._key, self) + + async def sandbox(self) -> Sandbox: + return await asyncio.shield(self._sandbox_task) + + def creation_failed(self) -> bool: + return self._sandbox_task.done() and ( + self._sandbox_task.cancelled() or self._sandbox_task.exception() is not None + ) + + async def aclose(self) -> None: + if self._idle_handle is not None: + self._idle_handle.cancel() + self._idle_handle = None + if not self._sandbox_task.done(): + self._sandbox_task.cancel() + try: + await self._sandbox_task + except BaseException: + return + if self._sandbox_context_manager is not None: + await self._sandbox_context_manager.__aexit__(None, None, None) + + +class SandboxActivities: + """Lazily resolves Workflow-scoped sandboxes and exposes their activities.""" + + def __init__( + self, + factories: dict[str, SandboxFactory], + idle_timeout: timedelta | None = None, + resolvable_worker_env_vars: Collection[str] | AllowAllWorkerEnvVars = (), + ) -> None: + """Store named Workflow-scoped worker-side sandbox factories.""" + self._factories = dict(factories) + self._idle_timeout = ( + idle_timeout if idle_timeout is not None else _SANDBOX_CACHE_IDLE_TIMEOUT + ) + if self._idle_timeout <= timedelta(0): + raise ValueError("Sandbox cache idle timeout must be positive") + self._records: dict[_SandboxKey, _SandboxRecord] = {} + self._closing_records: set[asyncio.Task[None]] = set() + self._worker_env_refs = _WorkerEnvRefResolver(resolvable_worker_env_vars) + + @asynccontextmanager + async def _sandbox( + self, input: _WorkflowScopedInput + ) -> AsyncGenerator[Sandbox, None]: + info = activity.info() + if ( + not info.workflow_id + or not info.workflow_run_id + or not input.first_execution_run_id + ): + raise RuntimeError("Sandbox activities must be started by a Workflow") + context = SandboxWorkflowContext( + chain=SandboxWorkflowChain( + namespace=info.namespace, + workflow_id=info.workflow_id, + first_execution_run_id=input.first_execution_run_id, + ), + run_id=info.workflow_run_id, + ) + factory = self._factories.get(input.sandbox_name) + if factory is None: + raise ApplicationError( + f"Unknown sandbox name {input.sandbox_name!r}. " + f"Known: {sorted(self._factories)}", + type=SANDBOX_NOT_FOUND_ERROR_TYPE, + ) + key = (input.sandbox_name, context.chain) + record = self._records.get(key) + if record is None: + record = _SandboxRecord(self, key, context, factory, self._idle_timeout) + self._records[key] = record + record.acquire() + try: + try: + sandbox = await record.sandbox() + except asyncio.CancelledError: + # A cancelled waiter must not discard a sandbox that another + # activity for the same Workflow may already be using. + if record.creation_failed(): + self._evict(key, record) + raise + except BaseException: + self._evict(key, record) + raise + yield sandbox + finally: + record.release() + + def _has_record(self, key: _SandboxKey, record: _SandboxRecord) -> bool: + return self._records.get(key) is record + + def _evict(self, key: _SandboxKey, record: _SandboxRecord) -> None: + if self._has_record(key, record): + del self._records[key] + task = asyncio.create_task(record.aclose()) + self._closing_records.add(task) + task.add_done_callback(self._record_closed) + + def _record_closed(self, task: asyncio.Task[None]) -> None: + self._closing_records.discard(task) + if not task.cancelled() and (error := task.exception()) is not None: + logger.error("Failed closing a sandbox adapter", exc_info=error) + + async def aclose(self) -> None: + """Cancel cache timers and discard all worker-local sandbox adapters.""" + records = list(self._records.values()) + self._records.clear() + for record in records: + await record.aclose() + if self._closing_records: + await asyncio.gather(*self._closing_records, return_exceptions=True) + + def activities(self) -> list[Callable[..., Any]]: + """Build one stable activity set that dispatches by sandbox name.""" + + @activity.defn(name=_activity_name("execute")) + @auto_heartbeater + async def execute(input: _ExecuteInput) -> list[_StreamItem]: + async with self._sandbox(input) as sandbox: + return await self._run_stream( + sandbox.execute_streaming( + input.command, + timeout=input.timeout, + cwd=input.cwd, + env=self._worker_env_refs.resolve(input.env), + **input.kwargs, + ), + timeout=input.timeout, + sandbox_name=input.sandbox_name, + streaming_topic=input.streaming_topic, + streaming_batch_interval_seconds=input.streaming_batch_interval_seconds, + ) + + @activity.defn(name=_activity_name("execute-code")) + @auto_heartbeater + async def execute_code( + input: _ExecuteCodeInput, + ) -> list[_StreamItem]: + async with self._sandbox(input) as sandbox: + return await self._run_stream( + sandbox.execute_code_streaming( + input.code, + input.language, + timeout=input.timeout, + cwd=input.cwd, + env=self._worker_env_refs.resolve(input.env), + **input.kwargs, + ), + timeout=input.timeout, + sandbox_name=input.sandbox_name, + streaming_topic=input.streaming_topic, + streaming_batch_interval_seconds=input.streaming_batch_interval_seconds, + ) + + @activity.defn(name=_activity_name("read-file")) + @auto_heartbeater + async def read_file(input: _PathInput) -> bytes: + async with self._sandbox(input) as sandbox: + try: + return await sandbox.read_file(input.path, **input.kwargs) + except SandboxTimeoutError as err: + raise _timeout_error(err, None) from err + except FileNotFoundError as err: + raise _path_not_found_error(err, input.path) from err + + @activity.defn(name=_activity_name("write-file")) + @auto_heartbeater + async def write_file(input: _WriteFileInput) -> None: + async with self._sandbox(input) as sandbox: + try: + await sandbox.write_file( + input.path, + base64.b64decode(input.content_base64), + **input.kwargs, + ) + except SandboxTimeoutError as err: + raise _timeout_error(err, None) from err + except FileNotFoundError as err: + raise _path_not_found_error(err, input.path) from err + + @activity.defn(name=_activity_name("remove-file")) + @auto_heartbeater + async def remove_file(input: _PathInput) -> None: + async with self._sandbox(input) as sandbox: + try: + await sandbox.remove_file(input.path, **input.kwargs) + except SandboxTimeoutError as err: + raise _timeout_error(err, None) from err + except FileNotFoundError as err: + raise _path_not_found_error(err, input.path) from err + + @activity.defn(name=_activity_name("list-files")) + @auto_heartbeater + async def list_files(input: _PathInput) -> list[FileInfo]: + async with self._sandbox(input) as sandbox: + try: + return await sandbox.list_files(input.path, **input.kwargs) + except SandboxTimeoutError as err: + raise _timeout_error(err, None) from err + except FileNotFoundError as err: + raise _path_not_found_error(err, input.path) from err + + return [execute, execute_code, read_file, write_file, remove_file, list_files] + + async def _run_stream( + self, + stream: AsyncGenerator[StreamChunk | ExecutionResult, None], + *, + timeout: float | None, + sandbox_name: str, + streaming_topic: str | None, + streaming_batch_interval_seconds: float, + ) -> list[_StreamItem]: + items: list[_StreamItem] = [] + try: + if streaming_topic is None: + async for item in stream: + items.append(_StreamItem(_item_to_json(item))) + return items + + client = WorkflowStreamClient.from_within_activity( + batch_interval=timedelta(seconds=streaming_batch_interval_seconds), + ) + topic = client.topic(streaming_topic, type=SandboxStreamEvent) + info = activity.info() + if not info.workflow_run_id: + raise RuntimeError("Sandbox activities must be started by a Workflow") + sequence = 0 + async with client: + async for item in stream: + items.append(_StreamItem(_item_to_json(item))) + if isinstance(item, StreamChunk): + topic.publish( + SandboxStreamEvent( + sandbox_name=sandbox_name, + execution_id=( + f"{info.workflow_run_id}:{info.activity_id}" + ), + attempt=info.attempt, + sequence=sequence, + chunk=item, + ) + ) + sequence += 1 + return items + except SandboxTimeoutError as err: + raise _timeout_error(err, timeout) from err + + +def _activity_name(operation: str) -> str: + return f"strands-sandbox-{operation}" + + +def _timeout_error(err: SandboxTimeoutError, timeout: float | None) -> ApplicationError: + return ApplicationError( + str(err), + timeout, + type=SANDBOX_TIMEOUT_ERROR_TYPE, + ) + + +def _path_not_found_error(err: FileNotFoundError, path: str) -> ApplicationError: + # Strands documents FileNotFoundError, not SandboxPathNotFoundError, for + # read/remove/list; only list_files raises the sandbox-specific subclass. + # Either way the path is missing on every attempt, so retrying is futile. + return ApplicationError( + str(err), + path, + type=SANDBOX_PATH_NOT_FOUND_ERROR_TYPE, + non_retryable=True, + ) + + +def _item_to_json(item: StreamChunk | ExecutionResult) -> dict[str, Any]: + if isinstance(item, StreamChunk): + return { + "kind": "stream_chunk", + "data": item.data, + "stream_type": item.stream_type, + } + return { + "kind": "execution_result", + "exit_code": item.exit_code, + "stdout": item.stdout, + "stderr": item.stderr, + "output_files": [ + { + "name": output.name, + "content_base64": base64.b64encode(output.content).decode("ascii"), + "mime_type": output.mime_type, + } + for output in item.output_files + ], + } diff --git a/temporalio/contrib/strands/_temporal_agent.py b/temporalio/contrib/strands/_temporal_agent.py index c2f9f14c7..41b13afa4 100644 --- a/temporalio/contrib/strands/_temporal_agent.py +++ b/temporalio/contrib/strands/_temporal_agent.py @@ -9,6 +9,7 @@ from ._temporal_mcp_client import TemporalMCPClient from ._temporal_model import TemporalModel +from ._temporal_sandbox import TemporalSandbox _SNAPSHOT_DISABLED = ( "TemporalAgent disables take_snapshot()/load_snapshot(). Temporal " @@ -23,8 +24,9 @@ class TemporalAgent(Agent): ``model`` is the name of a factory registered in ``StrandsPlugin(models={...})``. The activity options apply to every model - invocation this agent makes. All other keyword arguments are forwarded to - Strands' ``Agent`` (``tools``, ``hooks``, ``system_prompt``, + invocation this agent makes. ``sandbox`` is a workflow-side + ``TemporalSandbox`` whose name selects a worker-side factory. All other + keyword arguments are forwarded to Strands' ``Agent`` (``tools``, ``hooks``, ``system_prompt``, ``structured_output_model``, ``messages``, etc.). Strands' ``retry_strategy`` is disabled; configure retries via @@ -48,6 +50,7 @@ def __init__( priority: Priority = Priority.default, streaming_topic: str | None = None, streaming_batch_interval: timedelta = timedelta(milliseconds=100), + sandbox: TemporalSandbox | None = None, **agent_kwargs: Any, ) -> None: """Build a TemporalAgent from a registered model name and activity options.""" @@ -76,7 +79,7 @@ def __init__( streaming_topic=streaming_topic, streaming_batch_interval=streaming_batch_interval, ) - super().__init__(model=temporal_model, **agent_kwargs) + super().__init__(model=temporal_model, sandbox=sandbox, **agent_kwargs) # Strands invokes ToolProvider.load_tools() once at construction on a # separate run_async thread that has no workflow runtime, so a diff --git a/temporalio/contrib/strands/_temporal_sandbox.py b/temporalio/contrib/strands/_temporal_sandbox.py new file mode 100644 index 000000000..001456b98 --- /dev/null +++ b/temporalio/contrib/strands/_temporal_sandbox.py @@ -0,0 +1,204 @@ +import base64 +from collections.abc import AsyncGenerator +from datetime import timedelta +from typing import Any, TypeVar + +from strands.sandbox import ExecutionResult, FileInfo, OutputFile, Sandbox, StreamChunk +from strands.sandbox.errors import SandboxPathNotFoundError, SandboxTimeoutError +from strands.types.tools import AgentTool +from strands.vended_tools import make_file_editor, make_shell + +from temporalio import workflow +from temporalio.common import Priority, RetryPolicy +from temporalio.exceptions import ActivityError, ApplicationError +from temporalio.workflow import ActivityCancellationType, VersioningIntent + +from ._sandbox_activity import ( + SANDBOX_PATH_NOT_FOUND_ERROR_TYPE, + SANDBOX_TIMEOUT_ERROR_TYPE, + _activity_name, + _ExecuteCodeInput, + _ExecuteInput, + _PathInput, + _StreamItem, + _WriteFileInput, +) + +_ErrorT = TypeVar("_ErrorT", bound=OSError) + + +class TemporalSandbox(Sandbox): + """Workflow-side sandbox that dispatches operations as Temporal activities.""" + + def __init__( + self, + name: str, + *, + task_queue: str | None = None, + schedule_to_close_timeout: timedelta | None = None, + schedule_to_start_timeout: timedelta | None = None, + start_to_close_timeout: timedelta | None = None, + heartbeat_timeout: timedelta | None = None, + retry_policy: RetryPolicy | None = None, + cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL, + versioning_intent: VersioningIntent | None = None, + summary: str | None = None, + priority: Priority = Priority.default, + streaming_topic: str | None = None, + streaming_batch_interval: timedelta = timedelta(milliseconds=100), + ) -> None: + """Configure a registered sandbox name and its activity options.""" + self._name = name + self._streaming_topic = streaming_topic + self._streaming_batch_interval = streaming_batch_interval + self._options: dict[str, Any] = { + "task_queue": task_queue, + "schedule_to_close_timeout": schedule_to_close_timeout, + "schedule_to_start_timeout": schedule_to_start_timeout, + "start_to_close_timeout": start_to_close_timeout, + "heartbeat_timeout": heartbeat_timeout, + "retry_policy": retry_policy, + "cancellation_type": cancellation_type, + "versioning_intent": versioning_intent, + "summary": summary, + "priority": priority, + } + + async def execute_streaming( + self, + command: str, + *, + timeout: float | None = None, + cwd: str | None = None, + env: dict[str, str] | None = None, + **kwargs: Any, + ) -> AsyncGenerator[StreamChunk | ExecutionResult, None]: + """Execute a command in the registered worker-side sandbox.""" + items = await self._execute( + "execute", + _ExecuteInput( + command=command, + timeout=timeout, + cwd=cwd, + env=env, + kwargs=kwargs, + streaming_topic=self._streaming_topic, + streaming_batch_interval_seconds=self._streaming_batch_interval.total_seconds(), + ), + result_type=list[_StreamItem], + ) + for item in items: + yield _item_from_json(item.value) + + async def execute_code_streaming( + self, + code: str, + language: str, + *, + timeout: float | None = None, + cwd: str | None = None, + env: dict[str, str] | None = None, + **kwargs: Any, + ) -> AsyncGenerator[StreamChunk | ExecutionResult, None]: + """Execute code in the registered worker-side sandbox.""" + items = await self._execute( + "execute-code", + _ExecuteCodeInput( + code=code, + language=language, + timeout=timeout, + cwd=cwd, + env=env, + kwargs=kwargs, + streaming_topic=self._streaming_topic, + streaming_batch_interval_seconds=self._streaming_batch_interval.total_seconds(), + ), + result_type=list[_StreamItem], + ) + for item in items: + yield _item_from_json(item.value) + + async def read_file(self, path: str, **kwargs: Any) -> bytes: + """Read bytes from the registered worker-side sandbox.""" + return await self._execute( + "read-file", _PathInput(path, kwargs), result_type=bytes + ) + + async def write_file(self, path: str, content: bytes, **kwargs: Any) -> None: + """Write bytes to the registered worker-side sandbox.""" + await self._execute( + "write-file", + _WriteFileInput(path, kwargs, base64.b64encode(content).decode("ascii")), + ) + + async def remove_file(self, path: str, **kwargs: Any) -> None: + """Remove a file from the registered worker-side sandbox.""" + await self._execute("remove-file", _PathInput(path, kwargs)) + + async def list_files(self, path: str, **kwargs: Any) -> list[FileInfo]: + """List a directory in the registered worker-side sandbox.""" + return await self._execute( + "list-files", _PathInput(path, kwargs), result_type=list[FileInfo] + ) + + def get_tools(self) -> list[AgentTool]: + """Vend Strands' standard shell and file-editor sandbox tools.""" + return [ + make_file_editor(sandbox=self, name="sandbox_file_editor"), + make_shell(sandbox=self, name="sandbox_shell"), + ] + + async def _execute( + self, operation: str, input: Any, *, result_type: type | None = None + ) -> Any: + input.sandbox_name = self._name + input.first_execution_run_id = workflow.info().first_execution_run_id + try: + return await workflow.execute_activity( + _activity_name(operation), + input, + result_type=result_type, + **self._options, + ) + except ActivityError as err: + cause = err.__cause__ + if isinstance(cause, ApplicationError): + if cause.type == SANDBOX_TIMEOUT_ERROR_TYPE: + seconds = cause.details[0] if cause.details else None + raise _with_message(SandboxTimeoutError(seconds), cause) from err + if cause.type == SANDBOX_PATH_NOT_FOUND_ERROR_TYPE: + path = cause.details[0] if cause.details else "" + raise _with_message(SandboxPathNotFoundError(path), cause) from err + raise + + +def _with_message(error: _ErrorT, cause: ApplicationError) -> _ErrorT: + # The details only carry what the workflow needs to rebuild the error type. + # Restore the sandbox's own message so a timeout reports the duration the + # sandbox actually enforced, not the one the caller requested, and a missing + # path keeps whatever the backing environment said about it. + if cause.message: + error.args = (cause.message,) + return error + + +def _item_from_json(value: Any) -> StreamChunk | ExecutionResult: + if not isinstance(value, dict): + raise TypeError("Sandbox stream item must be an object") + if value.get("kind") == "stream_chunk": + return StreamChunk(value["data"], value["stream_type"]) + if value.get("kind") == "execution_result": + return ExecutionResult( + exit_code=value["exit_code"], + stdout=value["stdout"], + stderr=value["stderr"], + output_files=[ + OutputFile( + name=output["name"], + content=base64.b64decode(output["content_base64"]), + mime_type=output["mime_type"], + ) + for output in value["output_files"] + ], + ) + raise ValueError(f"Unknown sandbox stream item kind: {value.get('kind')!r}") diff --git a/temporalio/contrib/strands/_worker_env_ref.py b/temporalio/contrib/strands/_worker_env_ref.py new file mode 100644 index 000000000..474226757 --- /dev/null +++ b/temporalio/contrib/strands/_worker_env_ref.py @@ -0,0 +1,86 @@ +"""References to environment variables resolved by sandbox activity workers.""" + +from __future__ import annotations + +import os +import re +from collections.abc import Collection +from dataclasses import dataclass +from typing import cast + +_REF_PREFIX = "temporal.worker_env_ref:" +_REF_PATTERN = re.compile(re.escape(_REF_PREFIX) + r"\{([^}{]*)\}") + + +@dataclass(frozen=True) +class AllowAllWorkerEnvVars: + """Make every environment variable on the worker resolvable. + + .. warning:: + This class is experimental and may change in future versions. + Use with caution in production environments. + + Pass an instance in place of a list of names:: + + StrandsPlugin(resolvable_worker_env_vars=AllowAllWorkerEnvVars()) + + This lets workflow code name any variable on the worker and make its value + available to commands in the sandbox. + """ + + +def temporal_worker_env_ref(name: str) -> str: + """Refer to an environment variable held by a sandbox activity worker. + + .. warning:: + This function is experimental and may change in future versions. + Use with caution in production environments. + + The returned string contains only the variable name. A worker resolves it + immediately before invoking the sandbox when the name is allowed by + ``StrandsPlugin(resolvable_worker_env_vars=...)``. + """ + return f"{_REF_PREFIX}{{{name}}}" + + +class _WorkerEnvRefResolver: # type:ignore[reportUnusedClass] + def __init__( + self, + resolvable_worker_env_vars: Collection[str] | AllowAllWorkerEnvVars, + ) -> None: + if isinstance(resolvable_worker_env_vars, AllowAllWorkerEnvVars): + self._allowed: frozenset[str] | AllowAllWorkerEnvVars = ( + resolvable_worker_env_vars + ) + elif isinstance(resolvable_worker_env_vars, str): + raise TypeError( + "resolvable_worker_env_vars takes a collection of environment " + 'variable names, such as ["MY_API_KEY"], or ' + "AllowAllWorkerEnvVars(). A single string is read as the collection " + "of its characters, so pass a list even for one name." + ) + elif cast(object, resolvable_worker_env_vars) is AllowAllWorkerEnvVars: + raise TypeError( + "resolvable_worker_env_vars takes an AllowAllWorkerEnvVars instance, " + "not the class itself. Pass AllowAllWorkerEnvVars()." + ) + else: + self._allowed = frozenset(resolvable_worker_env_vars) + + def resolve(self, env: dict[str, str] | None) -> dict[str, str] | None: + """Return a copy with allowed worker environment references resolved.""" + if env is None: + return None + return {name: self._resolve_value(value) for name, value in env.items()} + + def _resolve_value(self, value: str) -> str: + def substitute(match: re.Match[str]) -> str: + name = match.group(1) + if not ( + isinstance(self._allowed, AllowAllWorkerEnvVars) + or name in self._allowed + ): + return match.group(0) + return os.environ.get(name, "") + + return _REF_PATTERN.sub(substitute, value) diff --git a/tests/contrib/strands/test_sandbox.py b/tests/contrib/strands/test_sandbox.py new file mode 100644 index 000000000..bcecb093e --- /dev/null +++ b/tests/contrib/strands/test_sandbox.py @@ -0,0 +1,973 @@ +import asyncio +from collections.abc import AsyncGenerator +from contextlib import asynccontextmanager +from dataclasses import dataclass +from datetime import timedelta +from typing import Any +from uuid import uuid4 + +import pytest +from strands import SandboxPathNotFoundError, SandboxTimeoutError, tool +from strands.sandbox import ExecutionResult, FileInfo, OutputFile, Sandbox, StreamChunk + +import temporalio.contrib.strands._sandbox_activity +import temporalio.contrib.strands._worker_env_ref +import temporalio.exceptions +import temporalio.testing +from temporalio import workflow +from temporalio.client import Client +from temporalio.common import RetryPolicy +from temporalio.contrib.strands import ( + AllowAllWorkerEnvVars, + SandboxStreamEvent, + SandboxWorkflowChain, + SandboxWorkflowContext, + StrandsPlugin, + TemporalAgent, + TemporalSandbox, + temporal_worker_env_ref, +) +from temporalio.contrib.workflow_streams import WorkflowStream, WorkflowStreamClient +from temporalio.worker import Replayer, Worker +from tests.contrib.strands.common import get_activities + + +class RecordingSandbox(Sandbox): + def __init__(self) -> None: + self.calls: list[tuple[Any, ...]] = [] + self.files = {"/binary": b"\x00\xff"} + + async def execute_streaming( + self, + command: str, + *, + timeout: float | None = None, + cwd: str | None = None, + env: dict[str, str] | None = None, + **kwargs: Any, + ) -> AsyncGenerator[StreamChunk | ExecutionResult, None]: + self.calls.append(("execute", command, timeout, cwd, env, kwargs)) + yield StreamChunk("out") + yield StreamChunk("err", "stderr") + yield ExecutionResult( + 0, + "out", + "err", + [OutputFile("artifact.bin", b"\x80\xff")], + ) + + async def execute_code_streaming( + self, + code: str, + language: str, + *, + timeout: float | None = None, + cwd: str | None = None, + env: dict[str, str] | None = None, + **kwargs: Any, + ) -> AsyncGenerator[StreamChunk | ExecutionResult, None]: + self.calls.append(("execute_code", code, language, timeout, cwd, env, kwargs)) + yield ExecutionResult(0, code, "") + + async def read_file(self, path: str, **kwargs: Any) -> bytes: + self.calls.append(("read_file", path, kwargs)) + return self.files[path] + + async def write_file(self, path: str, content: bytes, **kwargs: Any) -> None: + self.calls.append(("write_file", path, content, kwargs)) + self.files[path] = content + + async def remove_file(self, path: str, **kwargs: Any) -> None: + self.calls.append(("remove_file", path, kwargs)) + del self.files[path] + + async def list_files(self, path: str, **kwargs: Any) -> list[FileInfo]: + self.calls.append(("list_files", path, kwargs)) + return [FileInfo("binary", False, len(self.files["/binary"]))] + + +@dataclass +class SandboxWorkflowResult: + command_items_match: bool + code_result: ExecutionResult + binary_values_match: bool + files: list[FileInfo] + + +@workflow.defn +class SandboxWorkflow: + @workflow.run + async def run(self) -> SandboxWorkflowResult: + sandbox = TemporalSandbox( + "recording", start_to_close_timeout=timedelta(seconds=15) + ) + command_items = [ + item + async for item in sandbox.execute_streaming( + "echo hi", + timeout=2, + cwd="/work", + env={"SECRET": temporal_worker_env_ref("STRANDS_TEST_SECRET")}, + future_option=True, + ) + ] + expected_command_items = [ + StreamChunk("out"), + StreamChunk("err", "stderr"), + ExecutionResult( + 0, + "out", + "err", + [OutputFile("artifact.bin", b"\x80\xff")], + ), + ] + code_result = await sandbox.execute_code( + "print('hi')", "python3", future_option=2 + ) + original = await sandbox.read_file("/binary", future_option=3) + await sandbox.write_file("/other", b"\x01\xfe", future_option=4) + written = await sandbox.read_file("/other") + await sandbox.remove_file("/other", future_option=5) + files = await sandbox.list_files("/", future_option=6) + return SandboxWorkflowResult( + command_items == expected_command_items, + code_result, + original == b"\x00\xff" and written == b"\x01\xfe", + files, + ) + + +async def test_sandbox_operations_are_durable_and_cached( + client: Client, monkeypatch: pytest.MonkeyPatch +): + task_queue = f"test_sandbox-{uuid4()}" + secret = f"secret-{uuid4()}" + monkeypatch.setenv("STRANDS_TEST_SECRET", secret) + constructed: list[RecordingSandbox] = [] + contexts: list[SandboxWorkflowContext] = [] + + def factory(context: SandboxWorkflowContext) -> RecordingSandbox: + sandbox = RecordingSandbox() + contexts.append(context) + constructed.append(sandbox) + return sandbox + + plugin = StrandsPlugin( + models={}, + sandboxes={"recording": factory}, + resolvable_worker_env_vars=["STRANDS_TEST_SECRET"], + ) + async with Worker( + client, + task_queue=task_queue, + workflows=[SandboxWorkflow], + plugins=[plugin], + max_cached_workflows=0, + ): + handle = await client.start_workflow( + SandboxWorkflow.run, + id=f"test_sandbox-{uuid4()}", + task_queue=task_queue, + ) + result = await handle.result() + + assert result.command_items_match + assert result.code_result == ExecutionResult(0, "print('hi')", "") + assert result.binary_values_match + assert result.files == [FileInfo("binary", False, 2)] + assert len(constructed) == 1 + assert contexts == [ + SandboxWorkflowContext( + chain=SandboxWorkflowChain( + namespace=client.namespace, + workflow_id=handle.id, + first_execution_run_id=handle.first_execution_run_id or "", + ), + run_id=handle.result_run_id or "", + ) + ] + assert constructed[0].calls == [ + ( + "execute", + "echo hi", + 2, + "/work", + {"SECRET": secret}, + {"future_option": True}, + ), + ( + "execute_code", + "print('hi')", + "python3", + None, + None, + None, + {"future_option": 2}, + ), + ("read_file", "/binary", {"future_option": 3}), + ("write_file", "/other", b"\x01\xfe", {"future_option": 4}), + ("read_file", "/other", {}), + ("remove_file", "/other", {"future_option": 5}), + ("list_files", "/", {"future_option": 6}), + ] + + history = await handle.fetch_history() + assert secret.encode() not in b"".join( + event.SerializeToString() for event in history.events + ) + assert get_activities(history) == [ + "strands-sandbox-execute", + "strands-sandbox-execute-code", + "strands-sandbox-read-file", + "strands-sandbox-write-file", + "strands-sandbox-read-file", + "strands-sandbox-remove-file", + "strands-sandbox-list-files", + ] + await Replayer(workflows=[SandboxWorkflow], plugins=[plugin]).replay_workflow( + history + ) + + +@workflow.defn +class IsolatedSandboxWorkflow: + @workflow.run + async def run(self, value: bytes) -> bytes: + sandbox = TemporalSandbox( + "recording", start_to_close_timeout=timedelta(seconds=15) + ) + await sandbox.write_file("/value", value) + return await sandbox.read_file("/value") + + +async def test_sandbox_isolated_per_workflow_with_async_factory(client: Client): + task_queue = f"test_sandbox_isolation-{uuid4()}" + sandboxes: dict[SandboxWorkflowContext, RecordingSandbox] = {} + + async def factory(context: SandboxWorkflowContext) -> RecordingSandbox: + await asyncio.sleep(0) + sandbox = RecordingSandbox() + sandboxes[context] = sandbox + return sandbox + + plugin = StrandsPlugin(models={}, sandboxes={"recording": factory}) + async with Worker( + client, + task_queue=task_queue, + workflows=[IsolatedSandboxWorkflow], + plugins=[plugin], + max_cached_workflows=0, + ): + handles = [ + await client.start_workflow( + IsolatedSandboxWorkflow.run, + value, + id=f"test_sandbox_isolation-{uuid4()}", + task_queue=task_queue, + ) + for value in (b"first", b"second") + ] + assert await asyncio.gather(*(handle.result() for handle in handles)) == [ + b"first", + b"second", + ] + + assert len(sandboxes) == 2 + assert {context.workflow_id for context in sandboxes} == { + handle.id for handle in handles + } + assert {sandbox.files["/value"] for sandbox in sandboxes.values()} == { + b"first", + b"second", + } + + +@workflow.defn +class ContinueAsNewSandboxWorkflow: + @workflow.run + async def run(self, continued: bool = False) -> bytes: + sandbox = TemporalSandbox( + "recording", start_to_close_timeout=timedelta(seconds=15) + ) + if not continued: + await sandbox.write_file("/continued", b"same sandbox") + workflow.continue_as_new(True) + return await sandbox.read_file("/continued") + + +async def test_sandbox_reused_across_continue_as_new(client: Client): + task_queue = f"test_sandbox_continue_as_new-{uuid4()}" + contexts: list[SandboxWorkflowContext] = [] + + def factory(context: SandboxWorkflowContext) -> RecordingSandbox: + contexts.append(context) + return RecordingSandbox() + + plugin = StrandsPlugin(models={}, sandboxes={"recording": factory}) + async with Worker( + client, + task_queue=task_queue, + workflows=[ContinueAsNewSandboxWorkflow], + plugins=[plugin], + max_cached_workflows=0, + ): + handle = await client.start_workflow( + ContinueAsNewSandboxWorkflow.run, + id=f"test_sandbox_continue_as_new-{uuid4()}", + task_queue=task_queue, + ) + assert await handle.result() == b"same sandbox" + + assert len(contexts) == 1 + assert contexts[0].first_execution_run_id == handle.first_execution_run_id + assert contexts[0].run_id == handle.first_execution_run_id + + +@workflow.defn +class RetriedSandboxWorkflow: + @workflow.run + async def run(self) -> bytes: + sandbox = TemporalSandbox( + "recording", start_to_close_timeout=timedelta(seconds=15) + ) + if workflow.info().attempt == 1: + await sandbox.write_file("/retried", b"same sandbox") + raise RuntimeError("retry workflow") + return await sandbox.read_file("/retried") + + +async def test_sandbox_reused_across_workflow_retry(client: Client): + task_queue = f"test_sandbox_workflow_retry-{uuid4()}" + contexts: list[SandboxWorkflowContext] = [] + + def factory(context: SandboxWorkflowContext) -> RecordingSandbox: + contexts.append(context) + return RecordingSandbox() + + plugin = StrandsPlugin(models={}, sandboxes={"recording": factory}) + async with Worker( + client, + task_queue=task_queue, + workflows=[RetriedSandboxWorkflow], + plugins=[plugin], + max_cached_workflows=0, + workflow_failure_exception_types=[RuntimeError], + ): + handle = await client.start_workflow( + RetriedSandboxWorkflow.run, + id=f"test_sandbox_workflow_retry-{uuid4()}", + task_queue=task_queue, + retry_policy=RetryPolicy( + initial_interval=timedelta(milliseconds=1), maximum_attempts=2 + ), + ) + assert await handle.result() == b"same sandbox" + + assert len(contexts) == 1 + assert contexts[0].first_execution_run_id == handle.first_execution_run_id + assert contexts[0].run_id == handle.first_execution_run_id + + +@workflow.defn +class IdleSandboxWorkflow: + @workflow.run + async def run(self) -> None: + sandbox = TemporalSandbox( + "recording", start_to_close_timeout=timedelta(seconds=15) + ) + await sandbox.read_file("/binary") + await workflow.sleep(0.25) + await sandbox.read_file("/binary") + + +async def test_sandbox_cache_evicts_when_idle(client: Client): + task_queue = f"test_sandbox_idle-{uuid4()}" + contexts: list[SandboxWorkflowContext] = [] + closed: list[SandboxWorkflowContext] = [] + + @asynccontextmanager + async def factory(context: SandboxWorkflowContext): + contexts.append(context) + try: + yield RecordingSandbox() + finally: + closed.append(context) + + plugin = StrandsPlugin( + models={}, + sandboxes={"recording": factory}, + sandbox_cache_idle_timeout=timedelta(milliseconds=50), + ) + async with Worker( + client, + task_queue=task_queue, + workflows=[IdleSandboxWorkflow], + plugins=[plugin], + max_cached_workflows=0, + ): + await client.execute_workflow( + IdleSandboxWorkflow.run, + id=f"test_sandbox_idle-{uuid4()}", + task_queue=task_queue, + ) + + assert len(contexts) == 2 + assert contexts[0] == contexts[1] + assert closed == contexts + + +class SlowSandbox(RecordingSandbox): + async def read_file(self, path: str, **kwargs: Any) -> bytes: + await asyncio.sleep(0.15) + return await super().read_file(path, **kwargs) + + +@workflow.defn +class ConcurrentSandboxWorkflow: + @workflow.run + async def run(self) -> list[bytes]: + sandbox = TemporalSandbox( + "recording", start_to_close_timeout=timedelta(seconds=15) + ) + return list( + await asyncio.gather( + sandbox.read_file("/binary"), sandbox.read_file("/binary") + ) + ) + + +async def test_sandbox_factory_is_single_flight_and_not_evicted_in_use( + client: Client, +): + task_queue = f"test_sandbox_single_flight-{uuid4()}" + contexts: list[SandboxWorkflowContext] = [] + + async def factory(context: SandboxWorkflowContext) -> SlowSandbox: + contexts.append(context) + await asyncio.sleep(0.05) + return SlowSandbox() + + plugin = StrandsPlugin( + models={}, + sandboxes={"recording": factory}, + sandbox_cache_idle_timeout=timedelta(milliseconds=25), + ) + async with Worker( + client, + task_queue=task_queue, + workflows=[ConcurrentSandboxWorkflow], + plugins=[plugin], + max_cached_workflows=0, + ): + result = await client.execute_workflow( + ConcurrentSandboxWorkflow.run, + id=f"test_sandbox_single_flight-{uuid4()}", + task_queue=task_queue, + ) + + assert result == [b"\x00\xff", b"\x00\xff"] + assert len(contexts) == 1 + + +async def test_cancelled_waiter_preserves_completed_sandbox( + monkeypatch: pytest.MonkeyPatch, +) -> None: + factory_calls = 0 + + def factory(_: SandboxWorkflowContext) -> RecordingSandbox: + nonlocal factory_calls + factory_calls += 1 + return RecordingSandbox() + + sandbox_activities = temporalio.contrib.strands._sandbox_activity.SandboxActivities( + {"recording": factory} + ) + read_file = sandbox_activities.activities()[2] + input = temporalio.contrib.strands._sandbox_activity._PathInput( + "/binary", + sandbox_name="recording", + first_execution_run_id="first-run", + ) + activity_environment = temporalio.testing.ActivityEnvironment() + original_sandbox = ( + temporalio.contrib.strands._sandbox_activity._SandboxRecord.sandbox + ) + + async def cancel_after_creation( + record: temporalio.contrib.strands._sandbox_activity._SandboxRecord, + ) -> Sandbox: + await original_sandbox(record) + raise asyncio.CancelledError + + with monkeypatch.context() as patch: + patch.setattr( + temporalio.contrib.strands._sandbox_activity._SandboxRecord, + "sandbox", + cancel_after_creation, + ) + with pytest.raises(asyncio.CancelledError): + await activity_environment.run(read_file, input) + + assert await activity_environment.run(read_file, input) == b"\x00\xff" + assert factory_calls == 1 + await sandbox_activities.aclose() + + +@pytest.mark.parametrize( + ("activity_index", "input"), + [ + ( + 2, + temporalio.contrib.strands._sandbox_activity._PathInput("/missing"), + ), + ( + 3, + temporalio.contrib.strands._sandbox_activity._WriteFileInput( + "/missing", content_base64="" + ), + ), + ( + 4, + temporalio.contrib.strands._sandbox_activity._PathInput("/missing"), + ), + ( + 5, + temporalio.contrib.strands._sandbox_activity._PathInput("/missing"), + ), + ], + ids=["read", "write", "remove", "list"], +) +async def test_file_not_found_from_factory_remains_retryable( + activity_index: int, + input: temporalio.contrib.strands._sandbox_activity._WorkflowScopedInput, +) -> None: + input.sandbox_name = "recording" + input.first_execution_run_id = "first-run" + + def factory(_: SandboxWorkflowContext) -> Sandbox: + raise FileNotFoundError("transient factory failure") + + sandbox_activities = temporalio.contrib.strands._sandbox_activity.SandboxActivities( + {"recording": factory} + ) + activity_environment = temporalio.testing.ActivityEnvironment() + + with pytest.raises(FileNotFoundError, match="transient factory failure"): + await activity_environment.run( + sandbox_activities.activities()[activity_index], input + ) + + +async def test_unknown_sandbox_name_is_retryable() -> None: + sandbox_activities = temporalio.contrib.strands._sandbox_activity.SandboxActivities( + {} + ) + input = temporalio.contrib.strands._sandbox_activity._PathInput( + "/missing", + sandbox_name="missing", + first_execution_run_id="first-run", + ) + + with pytest.raises(temporalio.exceptions.ApplicationError) as err: + await temporalio.testing.ActivityEnvironment().run( + sandbox_activities.activities()[2], input + ) + + assert ( + err.value.type + == temporalio.contrib.strands._sandbox_activity.SANDBOX_NOT_FOUND_ERROR_TYPE + ) + assert not err.value.non_retryable + + +async def test_shared_sandbox_cache_closes_after_last_run_context( + monkeypatch: pytest.MonkeyPatch, +) -> None: + close_calls = 0 + + async def aclose( + _: temporalio.contrib.strands._sandbox_activity.SandboxActivities, + ) -> None: + nonlocal close_calls + close_calls += 1 + + monkeypatch.setattr( + temporalio.contrib.strands._sandbox_activity.SandboxActivities, + "aclose", + aclose, + ) + plugin = StrandsPlugin( + models={}, sandboxes={"recording": lambda _: RecordingSandbox()} + ) + assert plugin.run_context is not None + + async with plugin.run_context(): + async with plugin.run_context(): + pass + assert close_calls == 0 + + assert close_calls == 1 + + +def test_sandbox_cache_idle_timeout_must_be_positive() -> None: + with pytest.raises(ValueError, match="must be positive"): + StrandsPlugin( + models={}, + sandboxes={"recording": lambda _: RecordingSandbox()}, + sandbox_cache_idle_timeout=timedelta(0), + ) + + +@pytest.mark.parametrize( + ("resolvable", "expected"), + [ + (["STRANDS_TEST_SECRET"], "Bearer secret"), + (AllowAllWorkerEnvVars(), "Bearer secret"), + (["OTHER_SECRET"], "Bearer temporal.worker_env_ref:{STRANDS_TEST_SECRET}"), + ], + ids=["allowed", "allow_all", "not_allowed"], +) +def test_sandbox_worker_env_refs_respect_worker_allowlist( + monkeypatch: pytest.MonkeyPatch, + resolvable: Any, + expected: str, +) -> None: + monkeypatch.setenv("STRANDS_TEST_SECRET", "secret") + resolver = temporalio.contrib.strands._worker_env_ref._WorkerEnvRefResolver( + resolvable + ) + + assert resolver.resolve( + {"AUTHORIZATION": f"Bearer {temporal_worker_env_ref('STRANDS_TEST_SECRET')}"} + ) == {"AUTHORIZATION": expected} + + +def test_sandbox_worker_env_vars_rejects_a_single_string() -> None: + with pytest.raises(TypeError, match="collection of environment variable names"): + StrandsPlugin( + models={}, + sandboxes={"recording": lambda _: RecordingSandbox()}, + resolvable_worker_env_vars="STRANDS_TEST_SECRET", # type: ignore[arg-type] + ) + + +def test_sandbox_factories_share_one_activity_set() -> None: + plugin = StrandsPlugin( + models={}, + sandboxes={ + "first": lambda _: RecordingSandbox(), + "second": lambda _: RecordingSandbox(), + }, + ) + + assert plugin.activities is not None + assert not callable(plugin.activities) + assert len(plugin.activities) == 6 + + +def test_sandbox_workflow_context_separates_run_and_chain_identity() -> None: + chain = SandboxWorkflowChain("namespace", "workflow", "first-run") + first = SandboxWorkflowContext(chain, "run-1") + second = SandboxWorkflowContext(chain, "run-2") + + assert first != second + assert first.chain == second.chain + assert first.namespace == "namespace" + assert first.workflow_id == "workflow" + assert first.first_execution_run_id == "first-run" + + +@tool(name="sandbox_shell") +def custom_shell(command: str) -> str: + return command + + +def test_sandbox_default_tools_and_override() -> None: + default_agent = TemporalAgent( + model="mock", + sandbox=TemporalSandbox("recording"), + start_to_close_timeout=timedelta(seconds=15), + ) + assert "sandbox_shell" in default_agent.tool_registry.registry + assert "sandbox_file_editor" in default_agent.tool_registry.registry + + override_agent = TemporalAgent( + model="mock", + sandbox=TemporalSandbox("recording"), + tools=[custom_shell], + start_to_close_timeout=timedelta(seconds=15), + ) + assert override_agent.tool_registry.registry["sandbox_shell"] is custom_shell + assert "sandbox_file_editor" in override_agent.tool_registry.registry + + +@workflow.defn +class StreamingSandboxWorkflow: + def __init__(self) -> None: + self.stream = WorkflowStream() + + @workflow.run + async def run(self) -> bool: + sandbox = TemporalSandbox( + "recording", + start_to_close_timeout=timedelta(seconds=15), + streaming_topic="sandbox-events", + ) + result = [item async for item in sandbox.execute_streaming("echo hi")] + return result[-1] == ExecutionResult( + 0, + "out", + "err", + [OutputFile("artifact.bin", b"\x80\xff")], + ) + + +async def test_sandbox_streaming_publishes_correlated_events(client: Client): + task_queue = f"test_sandbox_streaming-{uuid4()}" + workflow_id = f"test_sandbox_streaming-{uuid4()}" + plugin = StrandsPlugin( + models={}, sandboxes={"recording": lambda _: RecordingSandbox()} + ) + async with Worker( + client, + task_queue=task_queue, + workflows=[StreamingSandboxWorkflow], + plugins=[plugin], + max_cached_workflows=0, + ): + handle = await client.start_workflow( + StreamingSandboxWorkflow.run, + id=workflow_id, + task_queue=task_queue, + ) + stream = WorkflowStreamClient.create(client, workflow_id) + events: list[SandboxStreamEvent] = [] + + async def collect() -> None: + async for stream_item in stream.subscribe( + ["sandbox-events"], + result_type=SandboxStreamEvent, + poll_cooldown=timedelta(milliseconds=50), + ): + events.append(stream_item.data) + if len(events) == 2: + break + + collect_task = asyncio.create_task(collect()) + assert await handle.result() + await asyncio.wait_for(collect_task, timeout=10) + + assert [event.sandbox_name for event in events] == ["recording"] * 2 + assert handle.result_run_id + assert events[0].execution_id.startswith(f"{handle.result_run_id}:") + assert events[1].execution_id == events[0].execution_id + assert [event.attempt for event in events] == [1, 1] + assert [event.sequence for event in events] == [0, 1] + assert [event.chunk for event in events] == [ + StreamChunk("out"), + StreamChunk("err", "stderr"), + ] + await Replayer( + workflows=[StreamingSandboxWorkflow], plugins=[plugin] + ).replay_workflow(await handle.fetch_history()) + + +class ErrorSandbox(RecordingSandbox): + def __init__(self, *, always_timeout: bool = False) -> None: + super().__init__() + self.attempts = 0 + self.always_timeout = always_timeout + + async def execute_streaming( + self, + command: str, + *, + timeout: float | None = None, + cwd: str | None = None, + env: dict[str, str] | None = None, + **kwargs: Any, + ) -> AsyncGenerator[StreamChunk | ExecutionResult, None]: + self.attempts += 1 + if self.always_timeout: + # The backing sandbox enforces its own limit, not the requested one. + raise SandboxTimeoutError(90) + if self.attempts == 1: + raise RuntimeError("transient") + yield ExecutionResult(0, "retried", "") + + async def read_file(self, path: str, **kwargs: Any) -> bytes: + raise FileNotFoundError(f"cat: {path}: No such file or directory") + + async def list_files(self, path: str, **kwargs: Any) -> list[FileInfo]: + raise SandboxPathNotFoundError(path) + + +class FileTimeoutSandbox(RecordingSandbox): + def __init__(self) -> None: + super().__init__() + self.attempts = { + operation: 0 for operation in ("read", "write", "remove", "list") + } + + def _record_attempt(self, operation: str) -> None: + self.attempts[operation] += 1 + + async def read_file(self, path: str, **kwargs: Any) -> bytes: + self._record_attempt("read") + raise SandboxTimeoutError(90) + + async def write_file(self, path: str, content: bytes, **kwargs: Any) -> None: + self._record_attempt("write") + raise SandboxTimeoutError(90) + + async def remove_file(self, path: str, **kwargs: Any) -> None: + self._record_attempt("remove") + raise SandboxTimeoutError(90) + + async def list_files(self, path: str, **kwargs: Any) -> list[FileInfo]: + self._record_attempt("list") + raise SandboxTimeoutError(90) + + +@workflow.defn +class SandboxFileTimeoutWorkflow: + @workflow.run + async def run(self) -> list[str]: + sandbox = TemporalSandbox( + "file-timeouts", + start_to_close_timeout=timedelta(seconds=5), + schedule_to_close_timeout=timedelta(seconds=15), + retry_policy=RetryPolicy( + initial_interval=timedelta(milliseconds=1), maximum_attempts=3 + ), + ) + messages: list[str] = [] + try: + await sandbox.read_file("/file") + except SandboxTimeoutError as err: + messages.append(str(err)) + try: + await sandbox.write_file("/file", b"value") + except SandboxTimeoutError as err: + messages.append(str(err)) + try: + await sandbox.remove_file("/file") + except SandboxTimeoutError as err: + messages.append(str(err)) + try: + await sandbox.list_files("/") + except SandboxTimeoutError as err: + messages.append(str(err)) + return messages + + +async def test_sandbox_file_timeouts_are_retryable_and_reconstructed( + client: Client, +): + task_queue = f"test_sandbox_file_timeouts-{uuid4()}" + sandbox = FileTimeoutSandbox() + plugin = StrandsPlugin(models={}, sandboxes={"file-timeouts": lambda _: sandbox}) + async with Worker( + client, + task_queue=task_queue, + workflows=[SandboxFileTimeoutWorkflow], + plugins=[plugin], + max_cached_workflows=0, + ): + result = await client.execute_workflow( + SandboxFileTimeoutWorkflow.run, + id=f"test_sandbox_file_timeouts-{uuid4()}", + task_queue=task_queue, + ) + + assert result == ["Execution timed out after 90 seconds"] * 4 + assert sandbox.attempts == {"read": 3, "write": 3, "remove": 3, "list": 3} + + +@workflow.defn +class SandboxErrorWorkflow: + @workflow.run + async def run(self) -> tuple[str, bool, bool, str, str]: + retried = TemporalSandbox( + "retried", + start_to_close_timeout=timedelta(seconds=15), + retry_policy=RetryPolicy( + initial_interval=timedelta(milliseconds=1), maximum_attempts=3 + ), + ) + result = await retried.execute("command", timeout=3) + try: + await retried.list_files("/missing") + except SandboxPathNotFoundError: + path_error = True + else: + path_error = False + + # A plain FileNotFoundError from the sandbox arrives as the sandbox + # subclass, keeping the backing environment's own message. + try: + await retried.read_file("/missing") + except SandboxPathNotFoundError as err: + read_message = str(err) + else: + read_message = "" + + failing = TemporalSandbox( + "failing", + start_to_close_timeout=timedelta(seconds=5), + schedule_to_close_timeout=timedelta(seconds=15), + retry_policy=RetryPolicy( + initial_interval=timedelta(milliseconds=1), maximum_attempts=2 + ), + ) + try: + await failing.execute("command", timeout=4) + except SandboxTimeoutError as err: + timeout_error = True + timeout_message = str(err) + else: + timeout_error = False + timeout_message = "" + return result.stdout, path_error, timeout_error, read_message, timeout_message + + +async def test_sandbox_retries_and_reconstructs_errors(client: Client): + task_queue = f"test_sandbox_errors-{uuid4()}" + retried = ErrorSandbox() + failing = ErrorSandbox(always_timeout=True) + factory_attempts = 0 + + def retried_factory(_: SandboxWorkflowContext) -> ErrorSandbox: + nonlocal factory_attempts + factory_attempts += 1 + if factory_attempts == 1: + raise RuntimeError("transient factory failure") + return retried + + plugin = StrandsPlugin( + models={}, + sandboxes={"retried": retried_factory, "failing": lambda _: failing}, + ) + async with Worker( + client, + task_queue=task_queue, + workflows=[SandboxErrorWorkflow], + plugins=[plugin], + max_cached_workflows=0, + ): + result = await client.execute_workflow( + SandboxErrorWorkflow.run, + id=f"test_sandbox_errors-{uuid4()}", + task_queue=task_queue, + ) + + assert result == ( + "retried", + True, + True, + "cat: /missing: No such file or directory", + "Execution timed out after 90 seconds", + ) + assert factory_attempts == 2 + assert retried.attempts == 2 + assert failing.attempts == 2 diff --git a/uv.lock b/uv.lock index 767c4e587..b62ec0466 100644 --- a/uv.lock +++ b/uv.lock @@ -4633,12 +4633,13 @@ wheels = [ [[package]] name = "strands-agents" -version = "1.47.0" +version = "1.51.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "boto3" }, { name = "botocore" }, { name = "docstring-parser" }, + { name = "httpx" }, { name = "jsonschema" }, { name = "mcp" }, { name = "opentelemetry-api" }, @@ -4649,9 +4650,9 @@ dependencies = [ { name = "typing-extensions" }, { name = "watchdog" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/2e/05/5eb8431ff340739b1c21b0da7d19ec9604b9c4953a1c4638df915321573c/strands_agents-1.47.0.tar.gz", hash = "sha256:97770cb6beb6e5fd1a58849f41201eb7edb43fd67ad3de6544ada2f342c2bcf0", size = 1157139, upload-time = "2026-07-10T14:45:05.809Z" } +sdist = { url = "https://files.pythonhosted.org/packages/1a/07/cb77bb1773d885e96dc0412920c620cdc660ef5113e9b61c3cd026240102/strands_agents-1.51.0.tar.gz", hash = "sha256:4cbd0b183aad0bb60ac7c6043d29841c8933937c97228e91669d2d97fc40776d", size = 1285092, upload-time = "2026-08-07T18:07:43.833Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/d0/cb/ceae892c5823bea9254160efc02a4553f2ced9d183676110c03c3a9ff0f3/strands_agents-1.47.0-py3-none-any.whl", hash = "sha256:1f6ce17404ff02079244ad7a4a180a2a7150546275e4b91187d71472ba8fe3f2", size = 600611, upload-time = "2026-07-10T14:45:03.942Z" }, + { url = "https://files.pythonhosted.org/packages/99/4c/937595709a56e966bc471d2fe14071910b2f57d1975a7fd701da18eedabe/strands_agents-1.51.0-py3-none-any.whl", hash = "sha256:1e0a8d125652d7a863cae8fd717b22e4fa5c7983b499438d1f0f50bf328e8832", size = 663778, upload-time = "2026-08-07T18:07:42.253Z" }, ] [[package]] @@ -4837,7 +4838,7 @@ requires-dist = [ { name = "protobuf", marker = "extra == 'cloud-run-worker-otel'", specifier = "<7" }, { name = "pydantic", marker = "extra == 'pydantic'", specifier = ">=2.0.0,<3" }, { name = "python-dateutil", marker = "python_full_version < '3.11'", specifier = ">=2.8.2,<3" }, - { name = "strands-agents", marker = "extra == 'strands-agents'", specifier = ">=1.39.0" }, + { name = "strands-agents", marker = "extra == 'strands-agents'", specifier = ">=1.51.0" }, { name = "types-aioboto3", extras = ["s3"], marker = "extra == 'aioboto3'", specifier = ">=10.4.0" }, { name = "types-protobuf", specifier = ">=3.20,<8.0.0" }, { name = "typing-extensions", specifier = ">=4.2.0,<5" }, @@ -4886,7 +4887,7 @@ dev = [ { name = "pytest-xdist", specifier = ">=3.6,<4" }, { name = "ruff", specifier = ">=0.15.12,<0.16" }, { name = "setuptools", specifier = "<82" }, - { name = "strands-agents", specifier = ">=1.39.0" }, + { name = "strands-agents", specifier = ">=1.51.0" }, { name = "strands-agents-tools", specifier = ">=0.5.2" }, { name = "toml", specifier = ">=0.10.2,<0.11" }, { name = "twine", specifier = ">=4.0.1,<5" }, From fa7ea9d61deb027c09f5bf099d27920977bf8863 Mon Sep 17 00:00:00 2001 From: Brian Strauch Date: Mon, 14 Sep 2026 19:25:41 -0700 Subject: [PATCH 2/5] Move Strands sandbox changelog entry to Unreleased --- CHANGELOG.md | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index afc59f096..dba8a51ba 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,13 @@ to include examples, links to docs, or any other relevant information. ### Added +- **Experimental**: `temporalio.contrib.strands` now supports durable, + Workflow-isolated Strands sandboxes through `TemporalSandbox` and + worker-side factories with run and Workflow-chain context registered with + `StrandsPlugin(sandboxes=...)`, using one shared activity set with optional + correlated live Workflow Streams output, worker-side environment references, + and async-context-manager factories for adapter cleanup. + ### Changed ### Deprecated @@ -38,12 +45,6 @@ to include examples, links to docs, or any other relevant information. - `ActivityHandle` now supports operator commands for standalone activities: `pause`, `unpause`, `update_options` and `restore_original_options`. -- **Experimental**: `temporalio.contrib.strands` now supports durable, - Workflow-isolated Strands sandboxes through `TemporalSandbox` and - worker-side factories with run and Workflow-chain context registered with - `StrandsPlugin(sandboxes=...)`, using one shared activity set with optional - correlated live Workflow Streams output, worker-side environment references, - and async-context-manager factories for adapter cleanup. - Added GCP Cloud Run serverless-worker OpenTelemetry plugin in `temporalio.contrib.opentelemetry`. - Added new options to ActivityHandle.describe() to retrieve associated payloads, such as activity input and outcome. - New properties and methods in ActivityExecution and ActivityExecutionDescription. From 08bb73edde53ddba46a9533ef295c403cbeb164f Mon Sep 17 00:00:00 2001 From: Brian Strauch Date: Thu, 17 Sep 2026 12:34:07 -0700 Subject: [PATCH 3/5] Remove Strands sandbox adapter cache --- CHANGELOG.md | 8 +- temporalio/contrib/strands/README.md | 53 +++--- temporalio/contrib/strands/_plugin.py | 22 +-- .../contrib/strands/_sandbox_activity.py | 137 +-------------- tests/contrib/strands/test_sandbox.py | 158 +++++------------- 5 files changed, 88 insertions(+), 290 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index dba8a51ba..8abf1fae6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -23,9 +23,11 @@ to include examples, links to docs, or any other relevant information. - **Experimental**: `temporalio.contrib.strands` now supports durable, Workflow-isolated Strands sandboxes through `TemporalSandbox` and worker-side factories with run and Workflow-chain context registered with - `StrandsPlugin(sandboxes=...)`, using one shared activity set with optional - correlated live Workflow Streams output, worker-side environment references, - and async-context-manager factories for adapter cleanup. + `StrandsPlugin(sandboxes=...)`. Each Activity calls the selected factory so it + can find or recover the backing environment. The integration uses one shared + activity set with optional correlated live Workflow Streams output, + worker-side environment references, and Activity-scoped async-context-manager + factories. ### Changed diff --git a/temporalio/contrib/strands/README.md b/temporalio/contrib/strands/README.md index 9fe1f91af..dacd5c1b7 100644 --- a/temporalio/contrib/strands/README.md +++ b/temporalio/contrib/strands/README.md @@ -220,19 +220,12 @@ many factories are configured. Each operation carries the selected sandbox name in its activity input so the worker can dispatch it to the matching factory. -The factory is called lazily with a `SandboxWorkflowContext` containing the -current `run_id` and a `chain` identity with the Workflow's namespace, Workflow -ID, and first execution Run ID. The worker-local cache uses the sandbox name and -chain identity, so each Workflow chain gets a separate sandbox for each -registered name. Retries, Continue-As-New, Reset, and Cron runs belong to the -same chain and therefore use the same sandbox; unrelated Workflow chains do not -share one. Multiple `TemporalSandbox` objects with the same name in one chain -intentionally share that chain's sandbox. - -The factory receives the current Run ID only when a worker-local cache entry is -created. A later run in the same chain reuses a warm entry without calling the -factory again. After eviction, the next factory call receives the Run ID of the -run that recreates the entry. +The factory is called for every sandbox Activity with a `SandboxWorkflowContext` +containing the current `run_id` and a `chain` identity with the Workflow's +namespace, Workflow ID, and first execution Run ID. Use the chain identity to +find or recreate the same backing environment across Activity retries, Worker +restarts, Continue-As-New, Reset, and Cron runs. Unrelated Workflow chains +should not share an environment unless the application explicitly intends that. The first execution Run ID is supplied by workflow code because Activity metadata only identifies the current run. Treat sandbox activities as trusted @@ -244,17 +237,21 @@ environment, for example by expiring old environments before ID reuse. Factories may be synchronous or asynchronous. Synchronous factories must only construct a lightweight adapter and must not block the activity event loop; -use an asynchronous factory for remote lookup or provisioning. A factory may -run more than once for the same context after cache eviction or on different -workers, so provisioning must be idempotent. Strands' `DockerSandbox` only -connects to an already-running container; it does not create one. +use an asynchronous factory for remote lookup or provisioning. Calls may run +concurrently or on different workers, so lookup and provisioning must be +idempotent. Strands' `DockerSandbox` only connects to an already-running +container; it does not create one. + +The plugin does not cache adapters. If constructing an adapter is expensive, +cache it in the factory along with any backend-specific liveness and recovery +logic. The cache remains an optimization: every worker process must still be +able to reconnect to the same Workflow-scoped backing environment. Protect +in-process cache misses when concurrent Activities could otherwise provision +the same environment twice. -Worker-local adapters are reused until they have been idle for five minutes. -Set `sandbox_cache_idle_timeout` on `StrandsPlugin` to change that duration. Return an async context manager from the factory when an adapter owns clients, -sockets, subprocess handles, tunnels, or leases. Its exit method runs after -cache eviction and Worker shutdown. Returning a plain `Sandbox` only drops the -adapter from the cache: +sockets, subprocess handles, tunnels, or leases that should be scoped to one +Activity. Its exit method runs after that Activity finishes: ```python from contextlib import asynccontextmanager @@ -272,12 +269,12 @@ In either case, provisioning, teardown, and cleanup of orphaned backing environments remain the application's responsibility; use a backend TTL or reaper for workflows that are terminated before normal cleanup. -That cache is per worker *process*, while successive sandbox activities from one -workflow are routed independently across the task queue. With more than one -worker on the queue, a `write-file` can land on one worker and the following -`read-file` on another. The context factory must therefore reconnect every -worker to the same Workflow-scoped backing environment rather than relying on -per-process state. A single worker on the queue also satisfies this. +Successive sandbox Activities from one Workflow are routed independently across +the task queue. With more than one worker on the queue, a `write-file` can land +on one worker and the following `read-file` on another. The context factory must +therefore reconnect every worker to the same Workflow-scoped backing environment +rather than relying on per-process state. A single worker on the queue also +satisfies this. Reset does not roll back commands or filesystem mutations already performed in the external sandbox, just as it does not roll back other Activity side effects. diff --git a/temporalio/contrib/strands/_plugin.py b/temporalio/contrib/strands/_plugin.py index 57b0bc954..0544306e5 100644 --- a/temporalio/contrib/strands/_plugin.py +++ b/temporalio/contrib/strands/_plugin.py @@ -57,9 +57,9 @@ class StrandsPlugin(SimplePlugin): When ``sandboxes`` is supplied, registers one stable set of activities that dispatches each operation by sandbox name. Each factory receives the requesting Workflow run's context and may return a sandbox directly or - awaitably. Worker-local adapters are cached by sandbox name and Workflow - chain until ``sandbox_cache_idle_timeout`` elapses. Use the same name in - workflow-side ``TemporalSandbox(name)`` instances. + awaitably. The factory is called for every Activity so it can find or recover + the backing environment. Use the same name in workflow-side + ``TemporalSandbox(name)`` instances. ``resolvable_worker_env_vars`` controls which worker environment variables sandbox command ``env`` references may resolve immediately before execution. @@ -72,7 +72,6 @@ def __init__( mcp_clients: dict[str, Callable[[], MCPClient]] | None = None, sandboxes: dict[str, SandboxFactory] | None = None, mcp_connection_idle_timeout: timedelta | None = None, - sandbox_cache_idle_timeout: timedelta | None = None, resolvable_worker_env_vars: Collection[str] | AllowAllWorkerEnvVars = (), ) -> None: """Build the plugin from optional model, MCP, and sandbox factories. @@ -80,8 +79,8 @@ def __init__( If ``models`` is omitted, registers a single ``BedrockModel`` factory under the name ``"bedrock"`` with Botocore retries disabled. - A sandbox factory may return an async context manager when its worker-local - adapter needs cleanup after cache eviction or Worker shutdown. + A sandbox factory may return an async context manager when its adapter + needs cleanup after the Activity finishes. """ default_name: str | None = None if models is None: @@ -95,7 +94,6 @@ def __init__( sandbox_activities = ( SandboxActivities( sandboxes, - sandbox_cache_idle_timeout, resolvable_worker_env_vars, ) if sandboxes @@ -117,21 +115,11 @@ def __init__( ) ) - sandbox_run_contexts = 0 - @asynccontextmanager async def run_context() -> AsyncGenerator[None, None]: - nonlocal sandbox_run_contexts - if sandbox_activities is not None: - sandbox_run_contexts += 1 try: yield finally: - if sandbox_activities is not None: - sandbox_run_contexts -= 1 - # One plugin instance can be shared by multiple Workers. - if sandbox_run_contexts == 0: - await sandbox_activities.aclose() for server in mcp_clients: await _evict_connection(server) diff --git a/temporalio/contrib/strands/_sandbox_activity.py b/temporalio/contrib/strands/_sandbox_activity.py index 1a567e019..0253f1933 100644 --- a/temporalio/contrib/strands/_sandbox_activity.py +++ b/temporalio/contrib/strands/_sandbox_activity.py @@ -1,9 +1,7 @@ from __future__ import annotations -import asyncio import base64 import inspect -import logging from collections.abc import ( AsyncGenerator, Awaitable, @@ -33,8 +31,6 @@ SANDBOX_TIMEOUT_ERROR_TYPE = "StrandsSandboxTimeoutError" SANDBOX_PATH_NOT_FOUND_ERROR_TYPE = "StrandsSandboxPathNotFoundError" SANDBOX_NOT_FOUND_ERROR_TYPE = "StrandsSandboxNotFoundError" -_SANDBOX_CACHE_IDLE_TIMEOUT = timedelta(minutes=5) -logger = logging.getLogger(__name__) @dataclass(frozen=True) @@ -84,7 +80,6 @@ class SandboxStreamEvent: SandboxFactory = Callable[ [SandboxWorkflowContext], _SandboxFactoryResult | Awaitable[_SandboxFactoryResult] ] -_SandboxKey = tuple[str, SandboxWorkflowChain] @dataclass @@ -132,93 +127,16 @@ class _StreamItem: value: dict[str, Any] -class _SandboxRecord: - def __init__( - self, - owner: SandboxActivities, - key: _SandboxKey, - context: SandboxWorkflowContext, - factory: SandboxFactory, - idle_timeout: timedelta, - ) -> None: - self._owner = owner - self._key = key - self._context = context - self._idle_timeout = idle_timeout - self._inflight = 0 - self._idle_handle: asyncio.TimerHandle | None = None - self._sandbox_context_manager: AbstractAsyncContextManager[Sandbox] | None = ( - None - ) - self._sandbox_task = asyncio.create_task(self._create(factory)) - - async def _create(self, factory: SandboxFactory) -> Sandbox: - result = factory(self._context) - if inspect.isawaitable(result): - result = await result - if isinstance(result, Sandbox): - return result - self._sandbox_context_manager = result - return await result.__aenter__() - - def acquire(self) -> None: - self._inflight += 1 - if self._idle_handle is not None: - self._idle_handle.cancel() - self._idle_handle = None - - def release(self) -> None: - self._inflight -= 1 - if self._inflight == 0 and self._owner._has_record(self._key, self): - self._idle_handle = asyncio.get_running_loop().call_later( - self._idle_timeout.total_seconds(), self._on_idle - ) - - def _on_idle(self) -> None: - self._idle_handle = None - if self._inflight == 0: - self._owner._evict(self._key, self) - - async def sandbox(self) -> Sandbox: - return await asyncio.shield(self._sandbox_task) - - def creation_failed(self) -> bool: - return self._sandbox_task.done() and ( - self._sandbox_task.cancelled() or self._sandbox_task.exception() is not None - ) - - async def aclose(self) -> None: - if self._idle_handle is not None: - self._idle_handle.cancel() - self._idle_handle = None - if not self._sandbox_task.done(): - self._sandbox_task.cancel() - try: - await self._sandbox_task - except BaseException: - return - if self._sandbox_context_manager is not None: - await self._sandbox_context_manager.__aexit__(None, None, None) - - class SandboxActivities: - """Lazily resolves Workflow-scoped sandboxes and exposes their activities.""" + """Resolves Workflow-scoped sandboxes and exposes their activities.""" def __init__( self, factories: dict[str, SandboxFactory], - idle_timeout: timedelta | None = None, resolvable_worker_env_vars: Collection[str] | AllowAllWorkerEnvVars = (), ) -> None: """Store named Workflow-scoped worker-side sandbox factories.""" self._factories = dict(factories) - self._idle_timeout = ( - idle_timeout if idle_timeout is not None else _SANDBOX_CACHE_IDLE_TIMEOUT - ) - if self._idle_timeout <= timedelta(0): - raise ValueError("Sandbox cache idle timeout must be positive") - self._records: dict[_SandboxKey, _SandboxRecord] = {} - self._closing_records: set[asyncio.Task[None]] = set() self._worker_env_refs = _WorkerEnvRefResolver(resolvable_worker_env_vars) @asynccontextmanager @@ -247,51 +165,14 @@ async def _sandbox( f"Known: {sorted(self._factories)}", type=SANDBOX_NOT_FOUND_ERROR_TYPE, ) - key = (input.sandbox_name, context.chain) - record = self._records.get(key) - if record is None: - record = _SandboxRecord(self, key, context, factory, self._idle_timeout) - self._records[key] = record - record.acquire() - try: - try: - sandbox = await record.sandbox() - except asyncio.CancelledError: - # A cancelled waiter must not discard a sandbox that another - # activity for the same Workflow may already be using. - if record.creation_failed(): - self._evict(key, record) - raise - except BaseException: - self._evict(key, record) - raise - yield sandbox - finally: - record.release() - - def _has_record(self, key: _SandboxKey, record: _SandboxRecord) -> bool: - return self._records.get(key) is record - - def _evict(self, key: _SandboxKey, record: _SandboxRecord) -> None: - if self._has_record(key, record): - del self._records[key] - task = asyncio.create_task(record.aclose()) - self._closing_records.add(task) - task.add_done_callback(self._record_closed) - - def _record_closed(self, task: asyncio.Task[None]) -> None: - self._closing_records.discard(task) - if not task.cancelled() and (error := task.exception()) is not None: - logger.error("Failed closing a sandbox adapter", exc_info=error) - - async def aclose(self) -> None: - """Cancel cache timers and discard all worker-local sandbox adapters.""" - records = list(self._records.values()) - self._records.clear() - for record in records: - await record.aclose() - if self._closing_records: - await asyncio.gather(*self._closing_records, return_exceptions=True) + result = factory(context) + if inspect.isawaitable(result): + result = await result + if isinstance(result, Sandbox): + yield result + else: + async with result as sandbox: + yield sandbox def activities(self) -> list[Callable[..., Any]]: """Build one stable activity set that dispatches by sandbox name.""" diff --git a/tests/contrib/strands/test_sandbox.py b/tests/contrib/strands/test_sandbox.py index bcecb093e..011214a36 100644 --- a/tests/contrib/strands/test_sandbox.py +++ b/tests/contrib/strands/test_sandbox.py @@ -137,7 +137,7 @@ async def run(self) -> SandboxWorkflowResult: ) -async def test_sandbox_operations_are_durable_and_cached( +async def test_sandbox_operations_are_durable_and_factory_runs_for_each_activity( client: Client, monkeypatch: pytest.MonkeyPatch ): task_queue = f"test_sandbox-{uuid4()}" @@ -145,11 +145,15 @@ async def test_sandbox_operations_are_durable_and_cached( monkeypatch.setenv("STRANDS_TEST_SECRET", secret) constructed: list[RecordingSandbox] = [] contexts: list[SandboxWorkflowContext] = [] + sandboxes: dict[SandboxWorkflowChain, RecordingSandbox] = {} def factory(context: SandboxWorkflowContext) -> RecordingSandbox: - sandbox = RecordingSandbox() contexts.append(context) - constructed.append(sandbox) + sandbox = sandboxes.get(context.chain) + if sandbox is None: + sandbox = RecordingSandbox() + sandboxes[context.chain] = sandbox + constructed.append(sandbox) return sandbox plugin = StrandsPlugin( @@ -176,16 +180,20 @@ def factory(context: SandboxWorkflowContext) -> RecordingSandbox: assert result.binary_values_match assert result.files == [FileInfo("binary", False, 2)] assert len(constructed) == 1 - assert contexts == [ - SandboxWorkflowContext( - chain=SandboxWorkflowChain( - namespace=client.namespace, - workflow_id=handle.id, - first_execution_run_id=handle.first_execution_run_id or "", - ), - run_id=handle.result_run_id or "", - ) - ] + assert ( + contexts + == [ + SandboxWorkflowContext( + chain=SandboxWorkflowChain( + namespace=client.namespace, + workflow_id=handle.id, + first_execution_run_id=handle.first_execution_run_id or "", + ), + run_id=handle.result_run_id or "", + ) + ] + * 7 + ) assert constructed[0].calls == [ ( "execute", @@ -242,13 +250,11 @@ async def run(self, value: bytes) -> bytes: async def test_sandbox_isolated_per_workflow_with_async_factory(client: Client): task_queue = f"test_sandbox_isolation-{uuid4()}" - sandboxes: dict[SandboxWorkflowContext, RecordingSandbox] = {} + sandboxes: dict[SandboxWorkflowChain, RecordingSandbox] = {} async def factory(context: SandboxWorkflowContext) -> RecordingSandbox: await asyncio.sleep(0) - sandbox = RecordingSandbox() - sandboxes[context] = sandbox - return sandbox + return sandboxes.setdefault(context.chain, RecordingSandbox()) plugin = StrandsPlugin(models={}, sandboxes={"recording": factory}) async with Worker( @@ -273,7 +279,7 @@ async def factory(context: SandboxWorkflowContext) -> RecordingSandbox: ] assert len(sandboxes) == 2 - assert {context.workflow_id for context in sandboxes} == { + assert {chain.workflow_id for chain in sandboxes} == { handle.id for handle in handles } assert {sandbox.files["/value"] for sandbox in sandboxes.values()} == { @@ -298,10 +304,11 @@ async def run(self, continued: bool = False) -> bytes: async def test_sandbox_reused_across_continue_as_new(client: Client): task_queue = f"test_sandbox_continue_as_new-{uuid4()}" contexts: list[SandboxWorkflowContext] = [] + sandboxes: dict[SandboxWorkflowChain, RecordingSandbox] = {} def factory(context: SandboxWorkflowContext) -> RecordingSandbox: contexts.append(context) - return RecordingSandbox() + return sandboxes.setdefault(context.chain, RecordingSandbox()) plugin = StrandsPlugin(models={}, sandboxes={"recording": factory}) async with Worker( @@ -318,9 +325,12 @@ def factory(context: SandboxWorkflowContext) -> RecordingSandbox: ) assert await handle.result() == b"same sandbox" - assert len(contexts) == 1 - assert contexts[0].first_execution_run_id == handle.first_execution_run_id + assert len(contexts) == 2 + assert {context.first_execution_run_id for context in contexts} == { + handle.first_execution_run_id + } assert contexts[0].run_id == handle.first_execution_run_id + assert contexts[1].run_id != contexts[0].run_id @workflow.defn @@ -339,10 +349,11 @@ async def run(self) -> bytes: async def test_sandbox_reused_across_workflow_retry(client: Client): task_queue = f"test_sandbox_workflow_retry-{uuid4()}" contexts: list[SandboxWorkflowContext] = [] + sandboxes: dict[SandboxWorkflowChain, RecordingSandbox] = {} def factory(context: SandboxWorkflowContext) -> RecordingSandbox: contexts.append(context) - return RecordingSandbox() + return sandboxes.setdefault(context.chain, RecordingSandbox()) plugin = StrandsPlugin(models={}, sandboxes={"recording": factory}) async with Worker( @@ -363,9 +374,12 @@ def factory(context: SandboxWorkflowContext) -> RecordingSandbox: ) assert await handle.result() == b"same sandbox" - assert len(contexts) == 1 - assert contexts[0].first_execution_run_id == handle.first_execution_run_id + assert len(contexts) == 2 + assert {context.first_execution_run_id for context in contexts} == { + handle.first_execution_run_id + } assert contexts[0].run_id == handle.first_execution_run_id + assert contexts[1].run_id != contexts[0].run_id @workflow.defn @@ -380,7 +394,7 @@ async def run(self) -> None: await sandbox.read_file("/binary") -async def test_sandbox_cache_evicts_when_idle(client: Client): +async def test_sandbox_context_manager_is_scoped_to_activity(client: Client): task_queue = f"test_sandbox_idle-{uuid4()}" contexts: list[SandboxWorkflowContext] = [] closed: list[SandboxWorkflowContext] = [] @@ -396,7 +410,6 @@ async def factory(context: SandboxWorkflowContext): plugin = StrandsPlugin( models={}, sandboxes={"recording": factory}, - sandbox_cache_idle_timeout=timedelta(milliseconds=50), ) async with Worker( client, @@ -436,10 +449,10 @@ async def run(self) -> list[bytes]: ) -async def test_sandbox_factory_is_single_flight_and_not_evicted_in_use( +async def test_sandbox_factory_called_for_concurrent_activities( client: Client, ): - task_queue = f"test_sandbox_single_flight-{uuid4()}" + task_queue = f"test_sandbox_concurrent_factory-{uuid4()}" contexts: list[SandboxWorkflowContext] = [] async def factory(context: SandboxWorkflowContext) -> SlowSandbox: @@ -450,7 +463,6 @@ async def factory(context: SandboxWorkflowContext) -> SlowSandbox: plugin = StrandsPlugin( models={}, sandboxes={"recording": factory}, - sandbox_cache_idle_timeout=timedelta(milliseconds=25), ) async with Worker( client, @@ -461,56 +473,12 @@ async def factory(context: SandboxWorkflowContext) -> SlowSandbox: ): result = await client.execute_workflow( ConcurrentSandboxWorkflow.run, - id=f"test_sandbox_single_flight-{uuid4()}", + id=f"test_sandbox_concurrent_factory-{uuid4()}", task_queue=task_queue, ) assert result == [b"\x00\xff", b"\x00\xff"] - assert len(contexts) == 1 - - -async def test_cancelled_waiter_preserves_completed_sandbox( - monkeypatch: pytest.MonkeyPatch, -) -> None: - factory_calls = 0 - - def factory(_: SandboxWorkflowContext) -> RecordingSandbox: - nonlocal factory_calls - factory_calls += 1 - return RecordingSandbox() - - sandbox_activities = temporalio.contrib.strands._sandbox_activity.SandboxActivities( - {"recording": factory} - ) - read_file = sandbox_activities.activities()[2] - input = temporalio.contrib.strands._sandbox_activity._PathInput( - "/binary", - sandbox_name="recording", - first_execution_run_id="first-run", - ) - activity_environment = temporalio.testing.ActivityEnvironment() - original_sandbox = ( - temporalio.contrib.strands._sandbox_activity._SandboxRecord.sandbox - ) - - async def cancel_after_creation( - record: temporalio.contrib.strands._sandbox_activity._SandboxRecord, - ) -> Sandbox: - await original_sandbox(record) - raise asyncio.CancelledError - - with monkeypatch.context() as patch: - patch.setattr( - temporalio.contrib.strands._sandbox_activity._SandboxRecord, - "sandbox", - cancel_after_creation, - ) - with pytest.raises(asyncio.CancelledError): - await activity_environment.run(read_file, input) - - assert await activity_environment.run(read_file, input) == b"\x00\xff" - assert factory_calls == 1 - await sandbox_activities.aclose() + assert len(contexts) == 2 @pytest.mark.parametrize( @@ -580,44 +548,6 @@ async def test_unknown_sandbox_name_is_retryable() -> None: assert not err.value.non_retryable -async def test_shared_sandbox_cache_closes_after_last_run_context( - monkeypatch: pytest.MonkeyPatch, -) -> None: - close_calls = 0 - - async def aclose( - _: temporalio.contrib.strands._sandbox_activity.SandboxActivities, - ) -> None: - nonlocal close_calls - close_calls += 1 - - monkeypatch.setattr( - temporalio.contrib.strands._sandbox_activity.SandboxActivities, - "aclose", - aclose, - ) - plugin = StrandsPlugin( - models={}, sandboxes={"recording": lambda _: RecordingSandbox()} - ) - assert plugin.run_context is not None - - async with plugin.run_context(): - async with plugin.run_context(): - pass - assert close_calls == 0 - - assert close_calls == 1 - - -def test_sandbox_cache_idle_timeout_must_be_positive() -> None: - with pytest.raises(ValueError, match="must be positive"): - StrandsPlugin( - models={}, - sandboxes={"recording": lambda _: RecordingSandbox()}, - sandbox_cache_idle_timeout=timedelta(0), - ) - - @pytest.mark.parametrize( ("resolvable", "expected"), [ @@ -968,6 +898,6 @@ def retried_factory(_: SandboxWorkflowContext) -> ErrorSandbox: "cat: /missing: No such file or directory", "Execution timed out after 90 seconds", ) - assert factory_attempts == 2 + assert factory_attempts == 5 assert retried.attempts == 2 assert failing.attempts == 2 From bccce001eb15a89093be4372c00cb3037a2868fb Mon Sep 17 00:00:00 2001 From: Brian Strauch Date: Thu, 17 Sep 2026 14:40:11 -0700 Subject: [PATCH 4/5] Apply suggestion from @brianstrauch --- CHANGELOG.md | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0ddb6af1e..c8844420d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -23,11 +23,7 @@ to include examples, links to docs, or any other relevant information. - **Experimental**: `temporalio.contrib.strands` now supports durable, Workflow-isolated Strands sandboxes through `TemporalSandbox` and worker-side factories with run and Workflow-chain context registered with - `StrandsPlugin(sandboxes=...)`. Each Activity calls the selected factory so it - can find or recover the backing environment. The integration uses one shared - activity set with optional correlated live Workflow Streams output, - worker-side environment references, and Activity-scoped async-context-manager - factories. + `StrandsPlugin(sandboxes=...)`. - Added the `temporalio.contrib.gcp.cloud_run.id` module with the `CloudRunIdPlugin` client plugin to set the worker identity on Cloud Run. ### Changed From 0d4fb7becffe15d2bd4ddb518e4f7facb9b124e8 Mon Sep 17 00:00:00 2001 From: Brian Strauch Date: Thu, 17 Sep 2026 14:41:45 -0700 Subject: [PATCH 5/5] Apply suggestion from @brianstrauch --- CHANGELOG.md | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c8844420d..2f3dd1075 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,8 +22,7 @@ to include examples, links to docs, or any other relevant information. - **Experimental**: `temporalio.contrib.strands` now supports durable, Workflow-isolated Strands sandboxes through `TemporalSandbox` and - worker-side factories with run and Workflow-chain context registered with - `StrandsPlugin(sandboxes=...)`. + worker-side factories registered with `StrandsPlugin(sandboxes=...)`. - Added the `temporalio.contrib.gcp.cloud_run.id` module with the `CloudRunIdPlugin` client plugin to set the worker identity on Cloud Run. ### Changed