From 7fcd20e86eeba282428d374b0281555310e4c145 Mon Sep 17 00:00:00 2001 From: Brian Strauch Date: Wed, 19 Aug 2026 19:47:53 -0700 Subject: [PATCH 1/7] Add durable Strands sandbox support --- CHANGELOG.md | 3 + pyproject.toml | 4 +- temporalio/contrib/strands/README.md | 76 ++++ temporalio/contrib/strands/__init__.py | 2 + temporalio/contrib/strands/_plugin.py | 13 +- .../contrib/strands/_sandbox_activity.py | 220 +++++++++++ temporalio/contrib/strands/_temporal_agent.py | 9 +- .../contrib/strands/_temporal_sandbox.py | 191 +++++++++ tests/contrib/strands/test_sandbox.py | 361 ++++++++++++++++++ uv.lock | 4 +- 10 files changed, 875 insertions(+), 8 deletions(-) create mode 100644 temporalio/contrib/strands/_sandbox_activity.py create mode 100644 temporalio/contrib/strands/_temporal_sandbox.py create mode 100644 tests/contrib/strands/test_sandbox.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 17d90a43d..875ef2745 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,9 @@ to include examples, links to docs, or any other relevant information. ### Added +- **Experimental**: `temporalio.contrib.strands` now supports durable Strands + sandboxes through `TemporalSandbox` and worker-side factories registered with + `StrandsPlugin(sandboxes=...)`, with optional live Workflow Streams output. - Added experimental `temporalio.contrib.opentelemetry.ReplaySafeMeterProvider` and `ReplaySafeLoggerProvider` (and exported `ReplaySafeTracerProvider`): wrap an OpenTelemetry provider so metrics and log events recorded from workflow code (e.g. by diff --git a/pyproject.toml b/pyproject.toml index d6397a7b6..bce3c86d7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -46,7 +46,7 @@ lambda-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.47.0"] [project.urls] Homepage = "https://github.com/temporalio/sdk-python" @@ -97,7 +97,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.47.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 126f4bd95..0d5a7ef61 100644 --- a/temporalio/contrib/strands/README.md +++ b/temporalio/contrib/strands/README.md @@ -170,6 +170,82 @@ 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 import DockerSandbox +from temporalio.contrib.strands import StrandsPlugin, TemporalAgent, TemporalSandbox + +# workflow +agent = TemporalAgent( + sandbox=TemporalSandbox( + "build", + start_to_close_timeout=timedelta(minutes=5), + ), +) + +# worker +Worker( + ..., + plugins=[StrandsPlugin(sandboxes={ + "build": lambda: DockerSandbox("agent-build-container"), + })], +) +``` + +The factory is called lazily on first use. Its sandbox instance is cached and +shared by all activities for that name for the worker's lifetime, so tools see +the same filesystem and working state. Provisioning and teardown of the backing +environment remain the application's responsibility. + +By default, `TemporalSandbox.get_tools()` vends `sandbox_bash` 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 strands.sandbox import StreamChunk +from temporalio.contrib.strands import TemporalSandbox +from temporalio.contrib.workflow_streams import WorkflowStream, WorkflowStreamClient + +# workflow __init__ +self.stream = WorkflowStream() +self.sandbox = TemporalSandbox("build", streaming_topic="sandbox-events") + +# external client +async for item in WorkflowStreamClient.create(client, workflow_id).subscribe( + ["sandbox-events"], result_type=StreamChunk +): + print(item.data.stream_type, item.data.data) +``` + +The topic is an observer-facing merged log. If sandbox executions overlap, +their chunks may interleave. Use different `streaming_topic` values when the +consumer needs separate logs; 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 that +Temporal retries may leave chunks in the topic before the 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 arguments and results cross Temporal's payload boundary and enter workflow +history. Keep command output and files within the server's configured payload +size limits; use external storage for large artifacts. In particular, `env` +values are recorded in history and must not contain secrets. + ## 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..5b13c7acd 100644 --- a/temporalio/contrib/strands/__init__.py +++ b/temporalio/contrib/strands/__init__.py @@ -4,10 +4,12 @@ from ._plugin import StrandsPlugin from ._temporal_agent import TemporalAgent from ._temporal_mcp_client import TemporalMCPClient +from ._temporal_sandbox import TemporalSandbox __all__ = [ "StrandsPlugin", "TemporalAgent", "TemporalMCPClient", + "TemporalSandbox", "workflow", ] diff --git a/temporalio/contrib/strands/_plugin.py b/temporalio/contrib/strands/_plugin.py index 0f1972666..d1ad51c11 100644 --- a/temporalio/contrib/strands/_plugin.py +++ b/temporalio/contrib/strands/_plugin.py @@ -4,6 +4,7 @@ from datetime import timedelta from strands.models import BedrockModel, Model +from strands.sandbox import Sandbox from strands.tools.mcp import MCPClient from temporalio.contrib.pydantic import pydantic_data_converter @@ -14,6 +15,7 @@ from ._failure_converter import StrandsFailureConverter from ._model_activity import ModelActivity +from ._sandbox_activity import SandboxActivities from ._temporal_mcp_client import ( _evict_connection, build_call_tool_activity, @@ -39,6 +41,11 @@ 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 a stable set of name-prefixed + activities for every sandbox factory. Each factory is called lazily and + its sandbox is shared by those activities for the worker's lifetime. Use + the same name in workflow-side ``TemporalSandbox(name)`` instances. """ def __init__( @@ -46,9 +53,10 @@ def __init__( *, models: dict[str, Callable[[], Model]] | None = None, mcp_clients: dict[str, Callable[[], MCPClient]] | None = None, + sandboxes: dict[str, Callable[[], Sandbox]] | None = None, mcp_connection_idle_timeout: timedelta | None = None, ) -> 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"``, matching Strands' own implicit default. @@ -62,6 +70,9 @@ def __init__( ma = ModelActivity(models, default_name=default_name) activities.extend([ma.invoke_model, ma.invoke_model_streaming]) + for name, sandbox_factory in (sandboxes or {}).items(): + activities.extend(SandboxActivities(name, sandbox_factory).activities()) + mcp_clients = mcp_clients or {} for server, client_factory in mcp_clients.items(): activities.append( diff --git a/temporalio/contrib/strands/_sandbox_activity.py b/temporalio/contrib/strands/_sandbox_activity.py new file mode 100644 index 000000000..6edb9e90e --- /dev/null +++ b/temporalio/contrib/strands/_sandbox_activity.py @@ -0,0 +1,220 @@ +import base64 +from collections.abc import AsyncGenerator, Callable +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 SandboxPathNotFoundError, SandboxTimeoutError + +from temporalio import activity +from temporalio.contrib.workflow_streams import WorkflowStreamClient +from temporalio.exceptions import ApplicationError + +from ._heartbeat_decorator import auto_heartbeater + +SANDBOX_TIMEOUT_ERROR_TYPE = "StrandsSandboxTimeoutError" +SANDBOX_PATH_NOT_FOUND_ERROR_TYPE = "StrandsSandboxPathNotFoundError" + + +@dataclass +class _ExecuteInput: + 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: + 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: + 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 SandboxActivities: + """Lazily resolves one registered sandbox and exposes its activities.""" + + def __init__(self, name: str, factory: Callable[[], Sandbox]) -> None: + """Store a sandbox name and its lazy worker-side factory.""" + self._name = name + self._factory = factory + self._sandbox: Sandbox | None = None + + def _get_sandbox(self) -> Sandbox: + if self._sandbox is None: + self._sandbox = self._factory() + return self._sandbox + + def activities(self) -> list[Callable[..., Any]]: + """Build stable, name-prefixed activities for this sandbox.""" + + @activity.defn(name=_activity_name(self._name, "execute")) + @auto_heartbeater + async def execute(input: _ExecuteInput) -> list[_StreamItem]: + return await self._run_stream( + self._get_sandbox().execute_streaming( + input.command, + timeout=input.timeout, + cwd=input.cwd, + env=input.env, + **input.kwargs, + ), + timeout=input.timeout, + streaming_topic=input.streaming_topic, + streaming_batch_interval_seconds=input.streaming_batch_interval_seconds, + ) + + @activity.defn(name=_activity_name(self._name, "execute-code")) + @auto_heartbeater + async def execute_code( + input: _ExecuteCodeInput, + ) -> list[_StreamItem]: + return await self._run_stream( + self._get_sandbox().execute_code_streaming( + input.code, + input.language, + timeout=input.timeout, + cwd=input.cwd, + env=input.env, + **input.kwargs, + ), + timeout=input.timeout, + streaming_topic=input.streaming_topic, + streaming_batch_interval_seconds=input.streaming_batch_interval_seconds, + ) + + @activity.defn(name=_activity_name(self._name, "read-file")) + @auto_heartbeater + async def read_file(input: _PathInput) -> bytes: + try: + return await self._get_sandbox().read_file(input.path, **input.kwargs) + except SandboxPathNotFoundError as err: + raise _path_not_found_error(err, input.path) from err + + @activity.defn(name=_activity_name(self._name, "write-file")) + @auto_heartbeater + async def write_file(input: _WriteFileInput) -> None: + try: + await self._get_sandbox().write_file( + input.path, + base64.b64decode(input.content_base64), + **input.kwargs, + ) + except SandboxPathNotFoundError as err: + raise _path_not_found_error(err, input.path) from err + + @activity.defn(name=_activity_name(self._name, "remove-file")) + @auto_heartbeater + async def remove_file(input: _PathInput) -> None: + try: + await self._get_sandbox().remove_file(input.path, **input.kwargs) + except SandboxPathNotFoundError as err: + raise _path_not_found_error(err, input.path) from err + + @activity.defn(name=_activity_name(self._name, "list-files")) + @auto_heartbeater + async def list_files(input: _PathInput) -> list[FileInfo]: + try: + return await self._get_sandbox().list_files(input.path, **input.kwargs) + except SandboxPathNotFoundError 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, + 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=StreamChunk) + async with client: + async for item in stream: + items.append(_StreamItem(_item_to_json(item))) + if isinstance(item, StreamChunk): + topic.publish(item) + return items + except SandboxTimeoutError as err: + raise ApplicationError( + str(err), + timeout, + type=SANDBOX_TIMEOUT_ERROR_TYPE, + ) from err + + +def _activity_name(sandbox_name: str, operation: str) -> str: + return f"{sandbox_name}-sandbox-{operation}" + + +def _path_not_found_error(err: SandboxPathNotFoundError, path: str) -> ApplicationError: + 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..d10068563 --- /dev/null +++ b/temporalio/contrib/strands/_temporal_sandbox.py @@ -0,0 +1,191 @@ +import base64 +from collections.abc import AsyncGenerator +from datetime import timedelta +from typing import Any + +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.bash import make_bash +from strands.vended_tools.file_editor import make_file_editor + +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, +) + + +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 bash and file-editor sandbox tools.""" + return [ + make_file_editor(sandbox=self, name="sandbox_file_editor"), + make_bash(sandbox=self, name="sandbox_bash"), + ] + + async def _execute( + self, operation: str, input: Any, *, result_type: type | None = None + ) -> Any: + try: + return await workflow.execute_activity( + _activity_name(self._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 SandboxTimeoutError(seconds) from err + if cause.type == SANDBOX_PATH_NOT_FOUND_ERROR_TYPE: + path = cause.details[0] if cause.details else "" + raise SandboxPathNotFoundError(path) from err + raise + + +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/tests/contrib/strands/test_sandbox.py b/tests/contrib/strands/test_sandbox.py new file mode 100644 index 000000000..d7f9ead79 --- /dev/null +++ b/tests/contrib/strands/test_sandbox.py @@ -0,0 +1,361 @@ +import asyncio +from collections.abc import AsyncGenerator +from dataclasses import dataclass +from datetime import timedelta +from typing import Any +from uuid import uuid4 + +from strands import SandboxPathNotFoundError, SandboxTimeoutError, tool +from strands.sandbox import ExecutionResult, FileInfo, OutputFile, Sandbox, StreamChunk + +from temporalio import workflow +from temporalio.client import Client +from temporalio.common import RetryPolicy +from temporalio.contrib.strands import StrandsPlugin, TemporalAgent, TemporalSandbox +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={"VISIBLE": "history"}, + 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): + task_queue = f"test_sandbox-{uuid4()}" + constructed: list[RecordingSandbox] = [] + + def factory() -> RecordingSandbox: + sandbox = RecordingSandbox() + constructed.append(sandbox) + return sandbox + + plugin = StrandsPlugin(models={}, sandboxes={"recording": factory}) + 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 constructed[0].calls == [ + ( + "execute", + "echo hi", + 2, + "/work", + {"VISIBLE": "history"}, + {"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 get_activities(history) == [ + "recording-sandbox-execute", + "recording-sandbox-execute-code", + "recording-sandbox-read-file", + "recording-sandbox-write-file", + "recording-sandbox-read-file", + "recording-sandbox-remove-file", + "recording-sandbox-list-files", + ] + await Replayer(workflows=[SandboxWorkflow], plugins=[plugin]).replay_workflow( + history + ) + + +@tool(name="sandbox_bash") +def custom_bash(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_bash" 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_bash], + start_to_close_timeout=timedelta(seconds=15), + ) + assert override_agent.tool_registry.registry["sandbox_bash"] is custom_bash + 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_raw_chunks(client: Client): + task_queue = f"test_sandbox_streaming-{uuid4()}" + workflow_id = f"test_sandbox_streaming-{uuid4()}" + plugin = StrandsPlugin(models={}, sandboxes={"recording": 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[StreamChunk] = [] + + async def collect() -> None: + async for stream_item in stream.subscribe( + ["sandbox-events"], + result_type=StreamChunk, + 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 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 or self.attempts == 1: + raise SandboxTimeoutError(timeout) + yield ExecutionResult(0, "retried", "") + + async def list_files(self, path: str, **kwargs: Any) -> list[FileInfo]: + raise SandboxPathNotFoundError(path) + + +@workflow.defn +class SandboxErrorWorkflow: + @workflow.run + async def run(self) -> tuple[str, bool, bool]: + retried = TemporalSandbox( + "retried", + start_to_close_timeout=timedelta(seconds=15), + retry_policy=RetryPolicy( + initial_interval=timedelta(milliseconds=1), maximum_attempts=2 + ), + ) + result = await retried.execute("command", timeout=3) + try: + await retried.list_files("/missing") + except SandboxPathNotFoundError: + path_error = True + else: + path_error = False + + failing = TemporalSandbox( + "failing", + start_to_close_timeout=timedelta(seconds=15), + retry_policy=RetryPolicy(maximum_attempts=1), + ) + try: + await failing.execute("command", timeout=4) + except SandboxTimeoutError: + timeout_error = True + else: + timeout_error = False + return result.stdout, path_error, timeout_error + + +async def test_sandbox_retries_and_reconstructs_errors(client: Client): + task_queue = f"test_sandbox_errors-{uuid4()}" + retried = ErrorSandbox() + failing = ErrorSandbox(always_timeout=True) + plugin = StrandsPlugin( + models={}, + sandboxes={"retried": lambda: retried, "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) + assert retried.attempts == 2 + assert failing.attempts == 1 diff --git a/uv.lock b/uv.lock index 22d2244c6..c4993b583 100644 --- a/uv.lock +++ b/uv.lock @@ -4827,7 +4827,7 @@ requires-dist = [ { name = "protobuf", specifier = ">=3.20,<8.0.0" }, { 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.47.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" }, @@ -4876,7 +4876,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.47.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 59f3dfcb29a529e8a76057346a6c13f59e10261b Mon Sep 17 00:00:00 2001 From: Brian Strauch Date: Thu, 20 Aug 2026 16:22:08 -0700 Subject: [PATCH 2/7] Make sandbox timeouts non-retryable A timeout is the deterministic outcome the caller's own `timeout` argument asked for, so retrying just re-runs the same hanging command. Under Temporal's unlimited-attempt default this meant SandboxTimeoutError never reached workflow code and the agent could never observe the timeout and adapt. Co-Authored-By: Claude Opus 5 (1M context) --- temporalio/contrib/strands/README.md | 6 ++++++ temporalio/contrib/strands/_sandbox_activity.py | 17 ++++++++++++----- tests/contrib/strands/test_sandbox.py | 11 ++++++++--- 3 files changed, 26 insertions(+), 8 deletions(-) diff --git a/temporalio/contrib/strands/README.md b/temporalio/contrib/strands/README.md index 0d5a7ef61..c15e9caa9 100644 --- a/temporalio/contrib/strands/README.md +++ b/temporalio/contrib/strands/README.md @@ -202,6 +202,12 @@ shared by all activities for that name for the worker's lifetime, so tools see the same filesystem and working state. Provisioning and teardown of the backing environment remain the application's responsibility. +`SandboxTimeoutError` and `SandboxPathNotFoundError` cross the activity +boundary as non-retryable failures and are re-raised inside the workflow, so a +command that exceeds its `timeout` or a path that does not exist surfaces to the +agent on the first attempt instead of retrying. Other sandbox failures are +retried under the `retry_policy` you pass to `TemporalSandbox`. + By default, `TemporalSandbox.get_tools()` vends `sandbox_bash` and `sandbox_file_editor`. A tool passed explicitly through `TemporalAgent(tools=...)` with either name takes precedence, following Strands' normal sandbox-tool diff --git a/temporalio/contrib/strands/_sandbox_activity.py b/temporalio/contrib/strands/_sandbox_activity.py index 6edb9e90e..0c9a3b7e1 100644 --- a/temporalio/contrib/strands/_sandbox_activity.py +++ b/temporalio/contrib/strands/_sandbox_activity.py @@ -177,17 +177,24 @@ async def _run_stream( topic.publish(item) return items except SandboxTimeoutError as err: - raise ApplicationError( - str(err), - timeout, - type=SANDBOX_TIMEOUT_ERROR_TYPE, - ) from err + raise _timeout_error(err, timeout) from err def _activity_name(sandbox_name: str, operation: str) -> str: return f"{sandbox_name}-sandbox-{operation}" +def _timeout_error(err: SandboxTimeoutError, timeout: float | None) -> ApplicationError: + # A timeout is the deterministic outcome the caller asked for, so retrying + # just repeats it. Surface it to workflow code on the first attempt instead. + return ApplicationError( + str(err), + timeout, + type=SANDBOX_TIMEOUT_ERROR_TYPE, + non_retryable=True, + ) + + def _path_not_found_error(err: SandboxPathNotFoundError, path: str) -> ApplicationError: return ApplicationError( str(err), diff --git a/tests/contrib/strands/test_sandbox.py b/tests/contrib/strands/test_sandbox.py index d7f9ead79..6b07207d3 100644 --- a/tests/contrib/strands/test_sandbox.py +++ b/tests/contrib/strands/test_sandbox.py @@ -294,8 +294,10 @@ async def execute_streaming( **kwargs: Any, ) -> AsyncGenerator[StreamChunk | ExecutionResult, None]: self.attempts += 1 - if self.always_timeout or self.attempts == 1: + if self.always_timeout: raise SandboxTimeoutError(timeout) + if self.attempts == 1: + raise RuntimeError("transient") yield ExecutionResult(0, "retried", "") async def list_files(self, path: str, **kwargs: Any) -> list[FileInfo]: @@ -321,10 +323,13 @@ async def run(self) -> tuple[str, bool, bool]: else: path_error = False + # No retry policy: a timeout must surface on the first attempt rather + # than retrying under Temporal's unlimited-attempt default. The + # schedule-to-close timeout bounds the failure if that ever regresses. failing = TemporalSandbox( "failing", - start_to_close_timeout=timedelta(seconds=15), - retry_policy=RetryPolicy(maximum_attempts=1), + start_to_close_timeout=timedelta(seconds=5), + schedule_to_close_timeout=timedelta(seconds=15), ) try: await failing.execute("command", timeout=4) From 2b0f98943e3f85bebef9e32105668bab5ecbe4d8 Mon Sep 17 00:00:00 2001 From: Brian Strauch Date: Thu, 20 Aug 2026 16:25:08 -0700 Subject: [PATCH 3/7] Preserve sandbox file and timeout errors across the activity boundary MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every built-in Strands sandbox raises a plain FileNotFoundError from read/write/remove — only list_files raises the SandboxPathNotFoundError subclass — so the old handlers were dead code and a missing path retried forever instead of reaching workflow code. Catch the documented base class instead, and carry the sandbox's own message through so a timeout reports the duration it actually enforced rather than the one the caller requested. Also fix the DockerSandbox import in the README, which is not re-exported from strands.sandbox, and note that the sandbox cache is per worker process. Co-Authored-By: Claude Opus 5 (1M context) --- temporalio/contrib/strands/README.md | 22 +++++++++---- .../contrib/strands/_sandbox_activity.py | 15 +++++---- .../contrib/strands/_temporal_sandbox.py | 18 +++++++++-- tests/contrib/strands/test_sandbox.py | 31 ++++++++++++++++--- 4 files changed, 66 insertions(+), 20 deletions(-) diff --git a/temporalio/contrib/strands/README.md b/temporalio/contrib/strands/README.md index c15e9caa9..cde77e75b 100644 --- a/temporalio/contrib/strands/README.md +++ b/temporalio/contrib/strands/README.md @@ -177,7 +177,7 @@ 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 import DockerSandbox +from strands.sandbox.docker import DockerSandbox from temporalio.contrib.strands import StrandsPlugin, TemporalAgent, TemporalSandbox # workflow @@ -202,11 +202,21 @@ shared by all activities for that name for the worker's lifetime, so tools see the same filesystem and working state. Provisioning and teardown of the backing environment remain the application's responsibility. -`SandboxTimeoutError` and `SandboxPathNotFoundError` cross the activity -boundary as non-retryable failures and are re-raised inside the workflow, so a -command that exceeds its `timeout` or a path that does not exist surfaces to the -agent on the first attempt instead of retrying. Other sandbox failures are -retried under the `retry_policy` you pass to `TemporalSandbox`. +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, so the factory must point at state the whole queue +shares — a named Docker container, an SSH host — rather than a per-process +temporary directory. A single worker on the queue also satisfies this. + +`SandboxTimeoutError` and any `FileNotFoundError` — including its +`SandboxPathNotFoundError` subclass — cross the activity boundary as +non-retryable failures and are re-raised inside the workflow with the sandbox's +own message, so a command that exceeds its `timeout` or a path that does not +exist surfaces to the agent on the first attempt instead of retrying. Other +sandbox failures, including the `OSError` that Strands documents for a failed +`write_file`, are retried under the `retry_policy` you pass to +`TemporalSandbox`. By default, `TemporalSandbox.get_tools()` vends `sandbox_bash` and `sandbox_file_editor`. A tool passed explicitly through `TemporalAgent(tools=...)` diff --git a/temporalio/contrib/strands/_sandbox_activity.py b/temporalio/contrib/strands/_sandbox_activity.py index 0c9a3b7e1..f5877de38 100644 --- a/temporalio/contrib/strands/_sandbox_activity.py +++ b/temporalio/contrib/strands/_sandbox_activity.py @@ -10,7 +10,7 @@ Sandbox, StreamChunk, ) -from strands.sandbox.errors import SandboxPathNotFoundError, SandboxTimeoutError +from strands.sandbox.errors import SandboxTimeoutError from temporalio import activity from temporalio.contrib.workflow_streams import WorkflowStreamClient @@ -118,7 +118,7 @@ async def execute_code( async def read_file(input: _PathInput) -> bytes: try: return await self._get_sandbox().read_file(input.path, **input.kwargs) - except SandboxPathNotFoundError as err: + except FileNotFoundError as err: raise _path_not_found_error(err, input.path) from err @activity.defn(name=_activity_name(self._name, "write-file")) @@ -130,7 +130,7 @@ async def write_file(input: _WriteFileInput) -> None: base64.b64decode(input.content_base64), **input.kwargs, ) - except SandboxPathNotFoundError as err: + except FileNotFoundError as err: raise _path_not_found_error(err, input.path) from err @activity.defn(name=_activity_name(self._name, "remove-file")) @@ -138,7 +138,7 @@ async def write_file(input: _WriteFileInput) -> None: async def remove_file(input: _PathInput) -> None: try: await self._get_sandbox().remove_file(input.path, **input.kwargs) - except SandboxPathNotFoundError as err: + except FileNotFoundError as err: raise _path_not_found_error(err, input.path) from err @activity.defn(name=_activity_name(self._name, "list-files")) @@ -146,7 +146,7 @@ async def remove_file(input: _PathInput) -> None: async def list_files(input: _PathInput) -> list[FileInfo]: try: return await self._get_sandbox().list_files(input.path, **input.kwargs) - except SandboxPathNotFoundError as 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] @@ -195,7 +195,10 @@ def _timeout_error(err: SandboxTimeoutError, timeout: float | None) -> Applicati ) -def _path_not_found_error(err: SandboxPathNotFoundError, path: str) -> ApplicationError: +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, diff --git a/temporalio/contrib/strands/_temporal_sandbox.py b/temporalio/contrib/strands/_temporal_sandbox.py index d10068563..f4491a4da 100644 --- a/temporalio/contrib/strands/_temporal_sandbox.py +++ b/temporalio/contrib/strands/_temporal_sandbox.py @@ -1,7 +1,7 @@ import base64 from collections.abc import AsyncGenerator from datetime import timedelta -from typing import Any +from typing import Any, TypeVar from strands.sandbox import ExecutionResult, FileInfo, OutputFile, Sandbox, StreamChunk from strands.sandbox.errors import SandboxPathNotFoundError, SandboxTimeoutError @@ -25,6 +25,8 @@ _WriteFileInput, ) +_ErrorT = TypeVar("_ErrorT", bound=OSError) + class TemporalSandbox(Sandbox): """Workflow-side sandbox that dispatches operations as Temporal activities.""" @@ -162,13 +164,23 @@ async def _execute( if isinstance(cause, ApplicationError): if cause.type == SANDBOX_TIMEOUT_ERROR_TYPE: seconds = cause.details[0] if cause.details else None - raise SandboxTimeoutError(seconds) from err + 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 SandboxPathNotFoundError(path) from err + 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") diff --git a/tests/contrib/strands/test_sandbox.py b/tests/contrib/strands/test_sandbox.py index 6b07207d3..de1db7b6d 100644 --- a/tests/contrib/strands/test_sandbox.py +++ b/tests/contrib/strands/test_sandbox.py @@ -295,11 +295,15 @@ async def execute_streaming( ) -> AsyncGenerator[StreamChunk | ExecutionResult, None]: self.attempts += 1 if self.always_timeout: - raise SandboxTimeoutError(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) @@ -307,7 +311,7 @@ async def list_files(self, path: str, **kwargs: Any) -> list[FileInfo]: @workflow.defn class SandboxErrorWorkflow: @workflow.run - async def run(self) -> tuple[str, bool, bool]: + async def run(self) -> tuple[str, bool, bool, str, str]: retried = TemporalSandbox( "retried", start_to_close_timeout=timedelta(seconds=15), @@ -323,6 +327,15 @@ async def run(self) -> tuple[str, bool, bool]: 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 = "" + # No retry policy: a timeout must surface on the first attempt rather # than retrying under Temporal's unlimited-attempt default. The # schedule-to-close timeout bounds the failure if that ever regresses. @@ -333,11 +346,13 @@ async def run(self) -> tuple[str, bool, bool]: ) try: await failing.execute("command", timeout=4) - except SandboxTimeoutError: + except SandboxTimeoutError as err: timeout_error = True + timeout_message = str(err) else: timeout_error = False - return result.stdout, path_error, timeout_error + timeout_message = "" + return result.stdout, path_error, timeout_error, read_message, timeout_message async def test_sandbox_retries_and_reconstructs_errors(client: Client): @@ -361,6 +376,12 @@ async def test_sandbox_retries_and_reconstructs_errors(client: Client): task_queue=task_queue, ) - assert result == ("retried", True, True) + assert result == ( + "retried", + True, + True, + "cat: /missing: No such file or directory", + "Execution timed out after 90 seconds", + ) assert retried.attempts == 2 assert failing.attempts == 1 From ef2c383e3c13b519672a4f2f0a8ab529799493fc Mon Sep 17 00:00:00 2001 From: Brian Strauch Date: Thu, 20 Aug 2026 16:32:37 -0700 Subject: [PATCH 4/7] Document sandbox activity retry semantics --- temporalio/contrib/strands/README.md | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/temporalio/contrib/strands/README.md b/temporalio/contrib/strands/README.md index cde77e75b..a55c839ab 100644 --- a/temporalio/contrib/strands/README.md +++ b/temporalio/contrib/strands/README.md @@ -218,6 +218,12 @@ sandbox failures, including the `OSError` that Strands documents for a failed `write_file`, are retried under the `retry_policy` you pass to `TemporalSandbox`. +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_bash` and `sandbox_file_editor`. A tool passed explicitly through `TemporalAgent(tools=...)` with either name takes precedence, following Strands' normal sandbox-tool @@ -230,13 +236,19 @@ 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 strands.sandbox import StreamChunk from temporalio.contrib.strands import TemporalSandbox from temporalio.contrib.workflow_streams import WorkflowStream, WorkflowStreamClient # workflow __init__ self.stream = WorkflowStream() -self.sandbox = TemporalSandbox("build", streaming_topic="sandbox-events") +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( From da17bd315d8a33ba996385b8433ae9b024949443 Mon Sep 17 00:00:00 2001 From: Brian Strauch Date: Thu, 27 Aug 2026 14:17:22 -0700 Subject: [PATCH 5/7] Isolate Strands sandboxes by workflow --- CHANGELOG.md | 5 +- temporalio/contrib/strands/README.md | 51 +++- temporalio/contrib/strands/__init__.py | 2 + temporalio/contrib/strands/_plugin.py | 35 ++- .../contrib/strands/_sandbox_activity.py | 228 +++++++++++--- .../contrib/strands/_temporal_sandbox.py | 1 + tests/contrib/strands/test_sandbox.py | 278 +++++++++++++++++- 7 files changed, 532 insertions(+), 68 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 70039b14e..ae4293851 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,8 +20,9 @@ to include examples, links to docs, or any other relevant information. ### Added -- **Experimental**: `temporalio.contrib.strands` now supports durable Strands - sandboxes through `TemporalSandbox` and worker-side factories registered with +- **Experimental**: `temporalio.contrib.strands` now supports durable, + Workflow-isolated Strands sandboxes through `TemporalSandbox` and + context-aware worker-side factories registered with `StrandsPlugin(sandboxes=...)`, with optional live Workflow Streams output. - Added `temporalio.converter.create_payload_validation_error` to create the non-retryable application error used when a converted payload fails validation. diff --git a/temporalio/contrib/strands/README.md b/temporalio/contrib/strands/README.md index a55c839ab..46d40e74c 100644 --- a/temporalio/contrib/strands/README.md +++ b/temporalio/contrib/strands/README.md @@ -178,7 +178,20 @@ 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 StrandsPlugin, TemporalAgent, TemporalSandbox +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.first_execution_run_id + ) + return DockerSandbox(container.name) # workflow agent = TemporalAgent( @@ -192,22 +205,42 @@ agent = TemporalAgent( Worker( ..., plugins=[StrandsPlugin(sandboxes={ - "build": lambda: DockerSandbox("agent-build-container"), + "build": build_sandbox, })], ) ``` -The factory is called lazily on first use. Its sandbox instance is cached and -shared by all activities for that name for the worker's lifetime, so tools see -the same filesystem and working state. Provisioning and teardown of the backing -environment remain the application's responsibility. +The factory is called lazily with a `SandboxWorkflowContext` that identifies the +Workflow's namespace, Workflow ID, and first execution Run ID. 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. + +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. +Eviction only drops the local adapter. 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, so the factory must point at state the whole queue -shares — a named Docker container, an SSH host — rather than a per-process -temporary directory. A single worker on the queue also satisfies this. +`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` and any `FileNotFoundError` — including its `SandboxPathNotFoundError` subclass — cross the activity boundary as diff --git a/temporalio/contrib/strands/__init__.py b/temporalio/contrib/strands/__init__.py index 5b13c7acd..e9f5e735c 100644 --- a/temporalio/contrib/strands/__init__.py +++ b/temporalio/contrib/strands/__init__.py @@ -2,12 +2,14 @@ from . import workflow from ._plugin import StrandsPlugin +from ._sandbox_activity import SandboxWorkflowContext from ._temporal_agent import TemporalAgent from ._temporal_mcp_client import TemporalMCPClient from ._temporal_sandbox import TemporalSandbox __all__ = [ "StrandsPlugin", + "SandboxWorkflowContext", "TemporalAgent", "TemporalMCPClient", "TemporalSandbox", diff --git a/temporalio/contrib/strands/_plugin.py b/temporalio/contrib/strands/_plugin.py index d1ad51c11..9d49954db 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, Awaitable, Callable from contextlib import asynccontextmanager from dataclasses import replace from datetime import timedelta @@ -15,7 +15,10 @@ from ._failure_converter import StrandsFailureConverter from ._model_activity import ModelActivity -from ._sandbox_activity import SandboxActivities +from ._sandbox_activity import ( + SandboxActivities, + SandboxWorkflowContext, +) from ._temporal_mcp_client import ( _evict_connection, build_call_tool_activity, @@ -43,9 +46,11 @@ class StrandsPlugin(SimplePlugin): disconnected; the timer resets on every reuse. Defaults to 5 minutes. When ``sandboxes`` is supplied, registers a stable set of name-prefixed - activities for every sandbox factory. Each factory is called lazily and - its sandbox is shared by those activities for the worker's lifetime. Use - the same name in workflow-side ``TemporalSandbox(name)`` instances. + activities for every sandbox factory. Each factory receives the owning + Workflow chain's context and may return a sandbox directly or awaitably. + Worker-local adapters are cached until ``sandbox_cache_idle_timeout`` + elapses. Use the same name in workflow-side ``TemporalSandbox(name)`` + instances. """ def __init__( @@ -53,8 +58,16 @@ def __init__( *, models: dict[str, Callable[[], Model]] | None = None, mcp_clients: dict[str, Callable[[], MCPClient]] | None = None, - sandboxes: dict[str, Callable[[], Sandbox]] | None = None, + sandboxes: dict[ + str, + Callable[ + [SandboxWorkflowContext], + Sandbox | Awaitable[Sandbox], + ], + ] + | None = None, mcp_connection_idle_timeout: timedelta | None = None, + sandbox_cache_idle_timeout: timedelta | None = None, ) -> None: """Build the plugin from optional model, MCP, and sandbox factories. @@ -70,8 +83,12 @@ def __init__( ma = ModelActivity(models, default_name=default_name) activities.extend([ma.invoke_model, ma.invoke_model_streaming]) - for name, sandbox_factory in (sandboxes or {}).items(): - activities.extend(SandboxActivities(name, sandbox_factory).activities()) + sandbox_activity_groups = [ + SandboxActivities(name, sandbox_factory, sandbox_cache_idle_timeout) + for name, sandbox_factory in (sandboxes or {}).items() + ] + for sandbox_activities in sandbox_activity_groups: + activities.extend(sandbox_activities.activities()) mcp_clients = mcp_clients or {} for server, client_factory in mcp_clients.items(): @@ -91,6 +108,8 @@ async def run_context() -> AsyncGenerator[None, None]: try: yield finally: + for sandbox_activities in sandbox_activity_groups: + 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 f5877de38..d413f5457 100644 --- a/temporalio/contrib/strands/_sandbox_activity.py +++ b/temporalio/contrib/strands/_sandbox_activity.py @@ -1,5 +1,10 @@ +from __future__ import annotations + +import asyncio import base64 -from collections.abc import AsyncGenerator, Callable +import inspect +from collections.abc import AsyncGenerator, Awaitable, Callable +from contextlib import asynccontextmanager from dataclasses import dataclass, field from datetime import timedelta from typing import Any @@ -20,10 +25,28 @@ SANDBOX_TIMEOUT_ERROR_TYPE = "StrandsSandboxTimeoutError" SANDBOX_PATH_NOT_FOUND_ERROR_TYPE = "StrandsSandboxPathNotFoundError" +_SANDBOX_CACHE_IDLE_TIMEOUT = timedelta(minutes=5) + + +@dataclass(frozen=True) +class SandboxWorkflowContext: + """Identity of the Workflow chain that owns a worker-side sandbox.""" + + namespace: str + workflow_id: str + first_execution_run_id: str + + +SandboxFactory = Callable[[SandboxWorkflowContext], Sandbox | Awaitable[Sandbox]] + + +@dataclass +class _WorkflowScopedInput: + first_execution_run_id: str = field(default="", kw_only=True) @dataclass -class _ExecuteInput: +class _ExecuteInput(_WorkflowScopedInput): command: str timeout: float | None = None cwd: str | None = None @@ -34,7 +57,7 @@ class _ExecuteInput: @dataclass -class _ExecuteCodeInput: +class _ExecuteCodeInput(_WorkflowScopedInput): code: str language: str timeout: float | None = None @@ -46,7 +69,7 @@ class _ExecuteCodeInput: @dataclass -class _PathInput: +class _PathInput(_WorkflowScopedInput): path: str kwargs: dict[str, Any] = field(default_factory=dict) @@ -61,19 +84,130 @@ class _StreamItem: value: dict[str, Any] +class _SandboxRecord: + def __init__( + self, + owner: SandboxActivities, + context: SandboxWorkflowContext, + factory: SandboxFactory, + idle_timeout: timedelta, + ) -> None: + self._owner = owner + self._context = context + self._idle_timeout = idle_timeout + self._inflight = 0 + self._idle_handle: asyncio.TimerHandle | None = None + self._sandbox_task = asyncio.create_task(self._create(factory)) + + async def _create(self, factory: SandboxFactory) -> Sandbox: + sandbox = factory(self._context) + if inspect.isawaitable(sandbox): + return await sandbox + return sandbox + + 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._context, 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._context, self) + + async def sandbox(self) -> Sandbox: + return await asyncio.shield(self._sandbox_task) + + def creation_done(self) -> bool: + return self._sandbox_task.done() + + 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: + pass + + class SandboxActivities: - """Lazily resolves one registered sandbox and exposes its activities.""" + """Lazily resolves Workflow-scoped sandboxes and exposes their activities.""" - def __init__(self, name: str, factory: Callable[[], Sandbox]) -> None: - """Store a sandbox name and its lazy worker-side factory.""" + def __init__( + self, + name: str, + factory: SandboxFactory, + idle_timeout: timedelta | None = None, + ) -> None: + """Store a sandbox name and its Workflow-scoped worker-side factory.""" self._name = name self._factory = factory - self._sandbox: Sandbox | None = None - - def _get_sandbox(self) -> Sandbox: - if self._sandbox is None: - self._sandbox = self._factory() - return self._sandbox + 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[SandboxWorkflowContext, _SandboxRecord] = {} + + @asynccontextmanager + async def _sandbox( + self, input: _WorkflowScopedInput + ) -> AsyncGenerator[Sandbox, None]: + info = activity.info() + if not info.workflow_id or not input.first_execution_run_id: + raise RuntimeError("Sandbox activities must be started by a Workflow") + context = SandboxWorkflowContext( + namespace=info.namespace, + workflow_id=info.workflow_id, + first_execution_run_id=input.first_execution_run_id, + ) + record = self._records.get(context) + if record is None: + record = _SandboxRecord(self, context, self._factory, self._idle_timeout) + self._records[context] = record + record.acquire() + try: + try: + sandbox = await record.sandbox() + except asyncio.CancelledError: + # One cancelled activity must not cancel or evict initialization + # that another activity for the same Workflow is awaiting. + if record.creation_done(): + self._evict(context, record) + raise + except BaseException: + self._evict(context, record) + raise + yield sandbox + finally: + record.release() + + def _has_record( + self, context: SandboxWorkflowContext, record: _SandboxRecord + ) -> bool: + return self._records.get(context) is record + + def _evict(self, context: SandboxWorkflowContext, record: _SandboxRecord) -> None: + if self._has_record(context, record): + del self._records[context] + + 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() def activities(self) -> list[Callable[..., Any]]: """Build stable, name-prefixed activities for this sandbox.""" @@ -81,43 +215,46 @@ def activities(self) -> list[Callable[..., Any]]: @activity.defn(name=_activity_name(self._name, "execute")) @auto_heartbeater async def execute(input: _ExecuteInput) -> list[_StreamItem]: - return await self._run_stream( - self._get_sandbox().execute_streaming( - input.command, + async with self._sandbox(input) as sandbox: + return await self._run_stream( + sandbox.execute_streaming( + input.command, + timeout=input.timeout, + cwd=input.cwd, + env=input.env, + **input.kwargs, + ), timeout=input.timeout, - cwd=input.cwd, - env=input.env, - **input.kwargs, - ), - timeout=input.timeout, - streaming_topic=input.streaming_topic, - streaming_batch_interval_seconds=input.streaming_batch_interval_seconds, - ) + streaming_topic=input.streaming_topic, + streaming_batch_interval_seconds=input.streaming_batch_interval_seconds, + ) @activity.defn(name=_activity_name(self._name, "execute-code")) @auto_heartbeater async def execute_code( input: _ExecuteCodeInput, ) -> list[_StreamItem]: - return await self._run_stream( - self._get_sandbox().execute_code_streaming( - input.code, - input.language, + 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=input.env, + **input.kwargs, + ), timeout=input.timeout, - cwd=input.cwd, - env=input.env, - **input.kwargs, - ), - timeout=input.timeout, - streaming_topic=input.streaming_topic, - streaming_batch_interval_seconds=input.streaming_batch_interval_seconds, - ) + streaming_topic=input.streaming_topic, + streaming_batch_interval_seconds=input.streaming_batch_interval_seconds, + ) @activity.defn(name=_activity_name(self._name, "read-file")) @auto_heartbeater async def read_file(input: _PathInput) -> bytes: try: - return await self._get_sandbox().read_file(input.path, **input.kwargs) + async with self._sandbox(input) as sandbox: + return await sandbox.read_file(input.path, **input.kwargs) except FileNotFoundError as err: raise _path_not_found_error(err, input.path) from err @@ -125,11 +262,12 @@ async def read_file(input: _PathInput) -> bytes: @auto_heartbeater async def write_file(input: _WriteFileInput) -> None: try: - await self._get_sandbox().write_file( - input.path, - base64.b64decode(input.content_base64), - **input.kwargs, - ) + async with self._sandbox(input) as sandbox: + await sandbox.write_file( + input.path, + base64.b64decode(input.content_base64), + **input.kwargs, + ) except FileNotFoundError as err: raise _path_not_found_error(err, input.path) from err @@ -137,7 +275,8 @@ async def write_file(input: _WriteFileInput) -> None: @auto_heartbeater async def remove_file(input: _PathInput) -> None: try: - await self._get_sandbox().remove_file(input.path, **input.kwargs) + async with self._sandbox(input) as sandbox: + await sandbox.remove_file(input.path, **input.kwargs) except FileNotFoundError as err: raise _path_not_found_error(err, input.path) from err @@ -145,7 +284,8 @@ async def remove_file(input: _PathInput) -> None: @auto_heartbeater async def list_files(input: _PathInput) -> list[FileInfo]: try: - return await self._get_sandbox().list_files(input.path, **input.kwargs) + async with self._sandbox(input) as sandbox: + return await sandbox.list_files(input.path, **input.kwargs) except FileNotFoundError as err: raise _path_not_found_error(err, input.path) from err diff --git a/temporalio/contrib/strands/_temporal_sandbox.py b/temporalio/contrib/strands/_temporal_sandbox.py index f4491a4da..ba95e66a4 100644 --- a/temporalio/contrib/strands/_temporal_sandbox.py +++ b/temporalio/contrib/strands/_temporal_sandbox.py @@ -152,6 +152,7 @@ def get_tools(self) -> list[AgentTool]: async def _execute( self, operation: str, input: Any, *, result_type: type | None = None ) -> Any: + input.first_execution_run_id = workflow.info().first_execution_run_id try: return await workflow.execute_activity( _activity_name(self._name, operation), diff --git a/tests/contrib/strands/test_sandbox.py b/tests/contrib/strands/test_sandbox.py index de1db7b6d..af01f4b1f 100644 --- a/tests/contrib/strands/test_sandbox.py +++ b/tests/contrib/strands/test_sandbox.py @@ -5,13 +5,19 @@ 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 from temporalio import workflow from temporalio.client import Client from temporalio.common import RetryPolicy -from temporalio.contrib.strands import StrandsPlugin, TemporalAgent, TemporalSandbox +from temporalio.contrib.strands import ( + SandboxWorkflowContext, + StrandsPlugin, + TemporalAgent, + TemporalSandbox, +) from temporalio.contrib.workflow_streams import WorkflowStream, WorkflowStreamClient from temporalio.worker import Replayer, Worker from tests.contrib.strands.common import get_activities @@ -125,9 +131,11 @@ async def run(self) -> SandboxWorkflowResult: async def test_sandbox_operations_are_durable_and_cached(client: Client): task_queue = f"test_sandbox-{uuid4()}" constructed: list[RecordingSandbox] = [] + contexts: list[SandboxWorkflowContext] = [] - def factory() -> RecordingSandbox: + def factory(context: SandboxWorkflowContext) -> RecordingSandbox: sandbox = RecordingSandbox() + contexts.append(context) constructed.append(sandbox) return sandbox @@ -151,6 +159,13 @@ def factory() -> RecordingSandbox: assert result.binary_values_match assert result.files == [FileInfo("binary", False, 2)] assert len(constructed) == 1 + assert contexts == [ + SandboxWorkflowContext( + namespace=client.namespace, + workflow_id=handle.id, + first_execution_run_id=handle.first_execution_run_id or "", + ) + ] assert constructed[0].calls == [ ( "execute", @@ -191,6 +206,247 @@ def factory() -> RecordingSandbox: ) +@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 + + +@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 + + +@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] = [] + + def factory(context: SandboxWorkflowContext) -> RecordingSandbox: + contexts.append(context) + return RecordingSandbox() + + 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] + + +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 + + +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), + ) + + @tool(name="sandbox_bash") def custom_bash(command: str) -> str: return command @@ -239,7 +495,9 @@ async def run(self) -> bool: async def test_sandbox_streaming_publishes_raw_chunks(client: Client): task_queue = f"test_sandbox_streaming-{uuid4()}" workflow_id = f"test_sandbox_streaming-{uuid4()}" - plugin = StrandsPlugin(models={}, sandboxes={"recording": RecordingSandbox}) + plugin = StrandsPlugin( + models={}, sandboxes={"recording": lambda _: RecordingSandbox()} + ) async with Worker( client, task_queue=task_queue, @@ -316,7 +574,7 @@ async def run(self) -> tuple[str, bool, bool, str, str]: "retried", start_to_close_timeout=timedelta(seconds=15), retry_policy=RetryPolicy( - initial_interval=timedelta(milliseconds=1), maximum_attempts=2 + initial_interval=timedelta(milliseconds=1), maximum_attempts=3 ), ) result = await retried.execute("command", timeout=3) @@ -359,9 +617,18 @@ 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": lambda: retried, "failing": lambda: failing}, + sandboxes={"retried": retried_factory, "failing": lambda _: failing}, ) async with Worker( client, @@ -383,5 +650,6 @@ async def test_sandbox_retries_and_reconstructs_errors(client: Client): "cat: /missing: No such file or directory", "Execution timed out after 90 seconds", ) + assert factory_attempts == 2 assert retried.attempts == 2 assert failing.attempts == 1 From 37031bbeca99fa377f10fd25660c8cc1ad50c008 Mon Sep 17 00:00:00 2001 From: Brian Strauch Date: Sun, 30 Aug 2026 17:22:52 -0700 Subject: [PATCH 6/7] Expose sandbox workflow run context --- CHANGELOG.md | 2 +- temporalio/contrib/strands/README.md | 14 ++-- temporalio/contrib/strands/__init__.py | 3 +- temporalio/contrib/strands/_plugin.py | 10 +-- .../contrib/strands/_sandbox_activity.py | 69 +++++++++++++------ tests/contrib/strands/test_sandbox.py | 24 ++++++- 6 files changed, 88 insertions(+), 34 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ae4293851..e476d2191 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,7 +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 - context-aware worker-side factories registered with + worker-side factories with run and Workflow-chain context registered with `StrandsPlugin(sandboxes=...)`, with optional live Workflow Streams output. - Added `temporalio.converter.create_payload_validation_error` to create the non-retryable application error used when a converted payload fails validation. diff --git a/temporalio/contrib/strands/README.md b/temporalio/contrib/strands/README.md index 46d40e74c..57da73dd4 100644 --- a/temporalio/contrib/strands/README.md +++ b/temporalio/contrib/strands/README.md @@ -189,7 +189,7 @@ 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.first_execution_run_id + context.chain.first_execution_run_id ) return DockerSandbox(container.name) @@ -210,14 +210,20 @@ Worker( ) ``` -The factory is called lazily with a `SandboxWorkflowContext` that identifies the -Workflow's namespace, Workflow ID, and first execution Run ID. Each Workflow -chain gets a separate sandbox for each registered name. Retries, +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 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. + 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 diff --git a/temporalio/contrib/strands/__init__.py b/temporalio/contrib/strands/__init__.py index e9f5e735c..35d839e39 100644 --- a/temporalio/contrib/strands/__init__.py +++ b/temporalio/contrib/strands/__init__.py @@ -2,13 +2,14 @@ from . import workflow from ._plugin import StrandsPlugin -from ._sandbox_activity import SandboxWorkflowContext +from ._sandbox_activity import SandboxWorkflowChain, SandboxWorkflowContext from ._temporal_agent import TemporalAgent from ._temporal_mcp_client import TemporalMCPClient from ._temporal_sandbox import TemporalSandbox __all__ = [ "StrandsPlugin", + "SandboxWorkflowChain", "SandboxWorkflowContext", "TemporalAgent", "TemporalMCPClient", diff --git a/temporalio/contrib/strands/_plugin.py b/temporalio/contrib/strands/_plugin.py index 9d49954db..4951f6052 100644 --- a/temporalio/contrib/strands/_plugin.py +++ b/temporalio/contrib/strands/_plugin.py @@ -46,11 +46,11 @@ class StrandsPlugin(SimplePlugin): disconnected; the timer resets on every reuse. Defaults to 5 minutes. When ``sandboxes`` is supplied, registers a stable set of name-prefixed - activities for every sandbox factory. Each factory receives the owning - Workflow chain's context and may return a sandbox directly or awaitably. - Worker-local adapters are cached until ``sandbox_cache_idle_timeout`` - elapses. Use the same name in workflow-side ``TemporalSandbox(name)`` - instances. + activities for every sandbox factory. Each factory receives the requesting + Workflow run's context and may return a sandbox directly or awaitably. + Worker-local adapters are cached by Workflow chain until + ``sandbox_cache_idle_timeout`` elapses. Use the same name in workflow-side + ``TemporalSandbox(name)`` instances. """ def __init__( diff --git a/temporalio/contrib/strands/_sandbox_activity.py b/temporalio/contrib/strands/_sandbox_activity.py index d413f5457..389e19c15 100644 --- a/temporalio/contrib/strands/_sandbox_activity.py +++ b/temporalio/contrib/strands/_sandbox_activity.py @@ -29,14 +29,37 @@ @dataclass(frozen=True) -class SandboxWorkflowContext: - """Identity of the Workflow chain that owns a worker-side sandbox.""" +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 + + SandboxFactory = Callable[[SandboxWorkflowContext], Sandbox | Awaitable[Sandbox]] @@ -94,6 +117,7 @@ def __init__( ) -> None: self._owner = owner self._context = context + self._chain = context.chain self._idle_timeout = idle_timeout self._inflight = 0 self._idle_handle: asyncio.TimerHandle | None = None @@ -113,7 +137,7 @@ def acquire(self) -> None: def release(self) -> None: self._inflight -= 1 - if self._inflight == 0 and self._owner._has_record(self._context, self): + if self._inflight == 0 and self._owner._has_record(self._chain, self): self._idle_handle = asyncio.get_running_loop().call_later( self._idle_timeout.total_seconds(), self._on_idle ) @@ -121,7 +145,7 @@ def release(self) -> None: def _on_idle(self) -> None: self._idle_handle = None if self._inflight == 0: - self._owner._evict(self._context, self) + self._owner._evict(self._chain, self) async def sandbox(self) -> Sandbox: return await asyncio.shield(self._sandbox_task) @@ -158,24 +182,31 @@ def __init__( ) if self._idle_timeout <= timedelta(0): raise ValueError("Sandbox cache idle timeout must be positive") - self._records: dict[SandboxWorkflowContext, _SandboxRecord] = {} + self._records: dict[SandboxWorkflowChain, _SandboxRecord] = {} @asynccontextmanager async def _sandbox( self, input: _WorkflowScopedInput ) -> AsyncGenerator[Sandbox, None]: info = activity.info() - if not info.workflow_id or not input.first_execution_run_id: + 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( - namespace=info.namespace, - workflow_id=info.workflow_id, - first_execution_run_id=input.first_execution_run_id, + 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, ) - record = self._records.get(context) + record = self._records.get(context.chain) if record is None: record = _SandboxRecord(self, context, self._factory, self._idle_timeout) - self._records[context] = record + self._records[context.chain] = record record.acquire() try: try: @@ -184,23 +215,21 @@ async def _sandbox( # One cancelled activity must not cancel or evict initialization # that another activity for the same Workflow is awaiting. if record.creation_done(): - self._evict(context, record) + self._evict(context.chain, record) raise except BaseException: - self._evict(context, record) + self._evict(context.chain, record) raise yield sandbox finally: record.release() - def _has_record( - self, context: SandboxWorkflowContext, record: _SandboxRecord - ) -> bool: - return self._records.get(context) is record + def _has_record(self, chain: SandboxWorkflowChain, record: _SandboxRecord) -> bool: + return self._records.get(chain) is record - def _evict(self, context: SandboxWorkflowContext, record: _SandboxRecord) -> None: - if self._has_record(context, record): - del self._records[context] + def _evict(self, chain: SandboxWorkflowChain, record: _SandboxRecord) -> None: + if self._has_record(chain, record): + del self._records[chain] async def aclose(self) -> None: """Cancel cache timers and discard all worker-local sandbox adapters.""" diff --git a/tests/contrib/strands/test_sandbox.py b/tests/contrib/strands/test_sandbox.py index af01f4b1f..a193dbdb0 100644 --- a/tests/contrib/strands/test_sandbox.py +++ b/tests/contrib/strands/test_sandbox.py @@ -13,6 +13,7 @@ from temporalio.client import Client from temporalio.common import RetryPolicy from temporalio.contrib.strands import ( + SandboxWorkflowChain, SandboxWorkflowContext, StrandsPlugin, TemporalAgent, @@ -161,9 +162,12 @@ def factory(context: SandboxWorkflowContext) -> RecordingSandbox: assert len(constructed) == 1 assert contexts == [ SandboxWorkflowContext( - namespace=client.namespace, - workflow_id=handle.id, - first_execution_run_id=handle.first_execution_run_id or "", + 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 == [ @@ -297,6 +301,7 @@ def factory(context: SandboxWorkflowContext) -> RecordingSandbox: 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 @@ -341,6 +346,7 @@ def factory(context: SandboxWorkflowContext) -> RecordingSandbox: 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 @@ -447,6 +453,18 @@ def test_sandbox_cache_idle_timeout_must_be_positive() -> None: ) +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_bash") def custom_bash(command: str) -> str: return command From 18d709b210f9854dc88935c07b253462dd49624a Mon Sep 17 00:00:00 2001 From: Brian Strauch Date: Mon, 31 Aug 2026 12:40:31 -0700 Subject: [PATCH 7/7] Share Strands sandbox activities --- CHANGELOG.md | 3 +- temporalio/contrib/strands/README.md | 17 +++-- temporalio/contrib/strands/_plugin.py | 25 +++---- .../contrib/strands/_sandbox_activity.py | 67 +++++++++++-------- .../contrib/strands/_temporal_sandbox.py | 3 +- tests/contrib/strands/test_sandbox.py | 28 ++++++-- 6 files changed, 88 insertions(+), 55 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 672304813..f4cdf8d89 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -23,7 +23,8 @@ 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=...)`, with optional live Workflow Streams output. + `StrandsPlugin(sandboxes=...)`, using one shared activity set with optional + live Workflow Streams output. ### Changed ### Deprecated diff --git a/temporalio/contrib/strands/README.md b/temporalio/contrib/strands/README.md index 57da73dd4..aa7d67db9 100644 --- a/temporalio/contrib/strands/README.md +++ b/temporalio/contrib/strands/README.md @@ -210,14 +210,19 @@ Worker( ) ``` +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 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. +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 diff --git a/temporalio/contrib/strands/_plugin.py b/temporalio/contrib/strands/_plugin.py index 4951f6052..75e035b51 100644 --- a/temporalio/contrib/strands/_plugin.py +++ b/temporalio/contrib/strands/_plugin.py @@ -45,12 +45,12 @@ class StrandsPlugin(SimplePlugin): 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 a stable set of name-prefixed - activities for every sandbox factory. Each factory receives the requesting - Workflow run's context and may return a sandbox directly or awaitably. - Worker-local adapters are cached by Workflow chain until - ``sandbox_cache_idle_timeout`` elapses. Use the same name in workflow-side - ``TemporalSandbox(name)`` instances. + 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. """ def __init__( @@ -83,11 +83,12 @@ def __init__( ma = ModelActivity(models, default_name=default_name) activities.extend([ma.invoke_model, ma.invoke_model_streaming]) - sandbox_activity_groups = [ - SandboxActivities(name, sandbox_factory, sandbox_cache_idle_timeout) - for name, sandbox_factory in (sandboxes or {}).items() - ] - for sandbox_activities in sandbox_activity_groups: + sandbox_activities = ( + SandboxActivities(sandboxes, sandbox_cache_idle_timeout) + if sandboxes + else None + ) + if sandbox_activities is not None: activities.extend(sandbox_activities.activities()) mcp_clients = mcp_clients or {} @@ -108,7 +109,7 @@ async def run_context() -> AsyncGenerator[None, None]: try: yield finally: - for sandbox_activities in sandbox_activity_groups: + if sandbox_activities is not None: 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 389e19c15..1654d3afd 100644 --- a/temporalio/contrib/strands/_sandbox_activity.py +++ b/temporalio/contrib/strands/_sandbox_activity.py @@ -25,6 +25,7 @@ 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) @@ -61,10 +62,12 @@ def first_execution_run_id(self) -> str: SandboxFactory = Callable[[SandboxWorkflowContext], Sandbox | Awaitable[Sandbox]] +_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) @@ -111,13 +114,14 @@ 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._chain = context.chain self._idle_timeout = idle_timeout self._inflight = 0 self._idle_handle: asyncio.TimerHandle | None = None @@ -137,7 +141,7 @@ def acquire(self) -> None: def release(self) -> None: self._inflight -= 1 - if self._inflight == 0 and self._owner._has_record(self._chain, self): + 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 ) @@ -145,7 +149,7 @@ def release(self) -> None: def _on_idle(self) -> None: self._idle_handle = None if self._inflight == 0: - self._owner._evict(self._chain, self) + self._owner._evict(self._key, self) async def sandbox(self) -> Sandbox: return await asyncio.shield(self._sandbox_task) @@ -170,19 +174,17 @@ class SandboxActivities: def __init__( self, - name: str, - factory: SandboxFactory, + factories: dict[str, SandboxFactory], idle_timeout: timedelta | None = None, ) -> None: - """Store a sandbox name and its Workflow-scoped worker-side factory.""" - self._name = name - self._factory = factory + """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[SandboxWorkflowChain, _SandboxRecord] = {} + self._records: dict[_SandboxKey, _SandboxRecord] = {} @asynccontextmanager async def _sandbox( @@ -203,10 +205,19 @@ async def _sandbox( ), run_id=info.workflow_run_id, ) - record = self._records.get(context.chain) + 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, + non_retryable=True, + ) + key = (input.sandbox_name, context.chain) + record = self._records.get(key) if record is None: - record = _SandboxRecord(self, context, self._factory, self._idle_timeout) - self._records[context.chain] = record + record = _SandboxRecord(self, key, context, factory, self._idle_timeout) + self._records[key] = record record.acquire() try: try: @@ -215,21 +226,21 @@ async def _sandbox( # One cancelled activity must not cancel or evict initialization # that another activity for the same Workflow is awaiting. if record.creation_done(): - self._evict(context.chain, record) + self._evict(key, record) raise except BaseException: - self._evict(context.chain, record) + self._evict(key, record) raise yield sandbox finally: record.release() - def _has_record(self, chain: SandboxWorkflowChain, record: _SandboxRecord) -> bool: - return self._records.get(chain) is record + def _has_record(self, key: _SandboxKey, record: _SandboxRecord) -> bool: + return self._records.get(key) is record - def _evict(self, chain: SandboxWorkflowChain, record: _SandboxRecord) -> None: - if self._has_record(chain, record): - del self._records[chain] + def _evict(self, key: _SandboxKey, record: _SandboxRecord) -> None: + if self._has_record(key, record): + del self._records[key] async def aclose(self) -> None: """Cancel cache timers and discard all worker-local sandbox adapters.""" @@ -239,9 +250,9 @@ async def aclose(self) -> None: await record.aclose() def activities(self) -> list[Callable[..., Any]]: - """Build stable, name-prefixed activities for this sandbox.""" + """Build one stable activity set that dispatches by sandbox name.""" - @activity.defn(name=_activity_name(self._name, "execute")) + @activity.defn(name=_activity_name("execute")) @auto_heartbeater async def execute(input: _ExecuteInput) -> list[_StreamItem]: async with self._sandbox(input) as sandbox: @@ -258,7 +269,7 @@ async def execute(input: _ExecuteInput) -> list[_StreamItem]: streaming_batch_interval_seconds=input.streaming_batch_interval_seconds, ) - @activity.defn(name=_activity_name(self._name, "execute-code")) + @activity.defn(name=_activity_name("execute-code")) @auto_heartbeater async def execute_code( input: _ExecuteCodeInput, @@ -278,7 +289,7 @@ async def execute_code( streaming_batch_interval_seconds=input.streaming_batch_interval_seconds, ) - @activity.defn(name=_activity_name(self._name, "read-file")) + @activity.defn(name=_activity_name("read-file")) @auto_heartbeater async def read_file(input: _PathInput) -> bytes: try: @@ -287,7 +298,7 @@ async def read_file(input: _PathInput) -> bytes: except FileNotFoundError as err: raise _path_not_found_error(err, input.path) from err - @activity.defn(name=_activity_name(self._name, "write-file")) + @activity.defn(name=_activity_name("write-file")) @auto_heartbeater async def write_file(input: _WriteFileInput) -> None: try: @@ -300,7 +311,7 @@ async def write_file(input: _WriteFileInput) -> None: except FileNotFoundError as err: raise _path_not_found_error(err, input.path) from err - @activity.defn(name=_activity_name(self._name, "remove-file")) + @activity.defn(name=_activity_name("remove-file")) @auto_heartbeater async def remove_file(input: _PathInput) -> None: try: @@ -309,7 +320,7 @@ async def remove_file(input: _PathInput) -> None: except FileNotFoundError as err: raise _path_not_found_error(err, input.path) from err - @activity.defn(name=_activity_name(self._name, "list-files")) + @activity.defn(name=_activity_name("list-files")) @auto_heartbeater async def list_files(input: _PathInput) -> list[FileInfo]: try: @@ -349,8 +360,8 @@ async def _run_stream( raise _timeout_error(err, timeout) from err -def _activity_name(sandbox_name: str, operation: str) -> str: - return f"{sandbox_name}-sandbox-{operation}" +def _activity_name(operation: str) -> str: + return f"strands-sandbox-{operation}" def _timeout_error(err: SandboxTimeoutError, timeout: float | None) -> ApplicationError: diff --git a/temporalio/contrib/strands/_temporal_sandbox.py b/temporalio/contrib/strands/_temporal_sandbox.py index ba95e66a4..e06e086c2 100644 --- a/temporalio/contrib/strands/_temporal_sandbox.py +++ b/temporalio/contrib/strands/_temporal_sandbox.py @@ -152,10 +152,11 @@ def get_tools(self) -> list[AgentTool]: 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(self._name, operation), + _activity_name(operation), input, result_type=result_type, **self._options, diff --git a/tests/contrib/strands/test_sandbox.py b/tests/contrib/strands/test_sandbox.py index a193dbdb0..132ce489c 100644 --- a/tests/contrib/strands/test_sandbox.py +++ b/tests/contrib/strands/test_sandbox.py @@ -197,13 +197,13 @@ def factory(context: SandboxWorkflowContext) -> RecordingSandbox: history = await handle.fetch_history() assert get_activities(history) == [ - "recording-sandbox-execute", - "recording-sandbox-execute-code", - "recording-sandbox-read-file", - "recording-sandbox-write-file", - "recording-sandbox-read-file", - "recording-sandbox-remove-file", - "recording-sandbox-list-files", + "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 @@ -453,6 +453,20 @@ def test_sandbox_cache_idle_timeout_must_be_positive() -> None: ) +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")