Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 28 additions & 0 deletions src/agents/extensions/sandbox/daytona/sandbox.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@
WorkspaceArchiveReadError,
WorkspaceArchiveWriteError,
WorkspaceReadNotFoundError,
WorkspaceStartError,
WorkspaceWriteTypeError,
)
from ....sandbox.manifest import Manifest
Expand Down Expand Up @@ -361,6 +362,33 @@ async def _validate_path_access(self, path: Path | str, *, for_write: bool = Fal
def _runtime_helpers(self) -> tuple[RuntimeHelperScript, ...]:
return (RESOLVE_WORKSPACE_PATH_HELPER,)

async def _prepare_workspace_root(self) -> None:
"""Create the workspace root before SDK exec calls use it as cwd."""
root = Path(self.state.manifest.root)
try:
envs = await self._resolved_envs()
result = await self._sandbox.process.exec(
f"mkdir -p -- {shlex.quote(str(root))}",
env=envs or None,
timeout=self.state.timeouts.fast_op_s,
)
except Exception as e:
raise WorkspaceStartError(path=root, cause=e) from e

exit_code = int(getattr(result, "exit_code", 0) or 0)
if exit_code != 0:
raise WorkspaceStartError(
path=root,
context={
"reason": "workspace_root_nonzero_exit",
"exit_code": exit_code,
"output": str(getattr(result, "result", "") or ""),
},
)

async def _prepare_backend_workspace(self) -> None:
await self._prepare_workspace_root()

async def mkdir(
self,
path: Path | str,
Expand Down
72 changes: 72 additions & 0 deletions tests/extensions/test_sandbox_daytona.py
Original file line number Diff line number Diff line change
Expand Up @@ -117,9 +117,14 @@ def __init__(self) -> None:
self._pty_handles: dict[str, _FakePtyHandle] = {}
self.create_pty_session_error: BaseException | None = None
self.symlinks: dict[str, str] = {}
self.workspace_roots: set[str] = set()
self.require_workspace_root_for_cd = False

async def exec(self, cmd: str, **kwargs: object) -> _FakeExecResult:
self.exec_calls.append((cmd, dict(kwargs)))
parts = shlex.split(cmd)
if len(parts) >= 4 and parts[:3] == ["mkdir", "-p", "--"]:
self.workspace_roots.add(parts[3])
if "sleep 0.5" in cmd:
await asyncio.sleep(0.5)
result = self.next_result
Expand Down Expand Up @@ -150,6 +155,21 @@ async def execute_session_command(
) -> object:
self.execute_session_command_calls.append((session_id, request, dict(kwargs)))
command = cast(str, getattr(request, "command", ""))
parts = shlex.split(command)
if (
self.require_workspace_root_for_cd
and len(parts) >= 3
and parts[0] == "cd"
and parts[2] == "&&"
and parts[1] not in self.workspace_roots
):
return types.SimpleNamespace(
cmd_id="cmd-123",
exit_code=1,
stdout="",
stderr=f"cd: no such file or directory: {parts[1]}",
output=f"cd: no such file or directory: {parts[1]}",
)
resolved = resolve_fake_workspace_path(
command,
symlinks=self.symlinks,
Expand Down Expand Up @@ -511,6 +531,58 @@ async def test_create_uses_daytona_safe_default_workspace_root(

assert session.state.manifest.root == daytona_module.DEFAULT_DAYTONA_WORKSPACE_ROOT

@pytest.mark.asyncio
async def test_start_prepares_workspace_root_before_runtime_helpers(
self,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Verify Daytona creates the root before exec uses it as cwd."""

daytona_module = _load_daytona_module(monkeypatch)

async with daytona_module.DaytonaSandboxClient() as client:
session = await client.create(options=daytona_module.DaytonaSandboxClientOptions())
sandbox = _FakeAsyncDaytona.current_sandbox
assert sandbox is not None
sandbox.process.require_workspace_root_for_cd = True

await session.start()

root = daytona_module.DEFAULT_DAYTONA_WORKSPACE_ROOT
assert root in sandbox.process.workspace_roots
assert sandbox.process.exec_calls[0][0] == f"mkdir -p -- {root}"
assert sandbox.process.execute_session_command_calls
_session_id, request, _kwargs = sandbox.process.execute_session_command_calls[0]
assert cast(str, cast(Any, request).command).startswith(f"cd {root} && ")
assert session.state.workspace_root_ready is True

@pytest.mark.asyncio
async def test_start_wraps_workspace_root_prepare_failure(
self,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Verify Daytona surfaces root preparation failures as start errors."""

daytona_module = _load_daytona_module(monkeypatch)

async with daytona_module.DaytonaSandboxClient() as client:
session = await client.create(options=daytona_module.DaytonaSandboxClientOptions())
sandbox = _FakeAsyncDaytona.current_sandbox
assert sandbox is not None
sandbox.process.next_result = _FakeExecResult(exit_code=2, result="mkdir failed")

with pytest.raises(daytona_module.WorkspaceStartError) as exc_info:
await session.start()

assert exc_info.value.context == {
"path": daytona_module.DEFAULT_DAYTONA_WORKSPACE_ROOT,
"reason": "workspace_root_nonzero_exit",
"exit_code": 2,
"output": "mkdir failed",
}
assert sandbox.process.execute_session_command_calls == []
assert session.state.workspace_root_ready is False

@pytest.mark.asyncio
async def test_create_passes_only_option_env_vars_to_daytona(
self,
Expand Down
Loading