fix: better custom backend support (#1071)

* fix: better custom backend support

* cr

CUSTOMIZATION.md

@@ -37,78 +37,48 @@ This is useful for pre-installing languages, frameworks, or internal tools that
 
 ### Using a different sandbox provider
 
-The `deepagents` ecosystem includes several sandbox providers out of the box. To swap providers, replace the `create_langsmith_sandbox()` call in `agent/server.py` with one of the following:
+Set the `SANDBOX_TYPE` environment variable to switch providers. Each provider has a corresponding integration file in `agent/integrations/` and a factory function registered in `agent/utils/sandbox.py`:
 
-#### Modal
+| `SANDBOX_TYPE` | Integration file | Required env vars |
+|---|---|---|
+| `langsmith` (default) | `agent/integrations/langsmith.py` | `LANGSMITH_API_KEY_PROD`, `SANDBOX_TYPE="langsmith"` |
+| `daytona` | `agent/integrations/daytona.py` | `DAYTONA_API_KEY`, `SANDBOX_TYPE="daytona"` |
+| `runloop` | `agent/integrations/runloop.py` | `RUNLOOP_API_KEY`, `SANDBOX_TYPE="runloop"` |
+| `modal` | `agent/integrations/modal.py` | Modal credentials, `SANDBOX_TYPE="modal"` |
+| `local` | `agent/integrations/local.py` | None (no isolation — development only), `SANDBOX_TYPE="local"` |
 
-```bash
-pip install langchain-modal
-```
+> **Warning**: `local` runs commands directly on your host with no sandboxing. Only use for local development with human-in-the-loop enabled.
+
+### Adding a new sandbox provider
+
+1. **Create an integration file** at `agent/integrations/my_provider.py` with a factory function matching this signature:
 
 ```python
-import modal
-from langchain_modal import ModalSandbox
+def create_my_provider_sandbox(sandbox_id: str | None = None):
+    """Create or reconnect to a sandbox.
 
-app = modal.App.lookup("open-swe")
-sandbox_backend = ModalSandbox(sandbox=modal.Sandbox.create(app=app))
+    Args:
+        sandbox_id: Optional existing sandbox ID to reconnect to.
+            If None, creates a new sandbox.
+
+    Returns:
+        An object implementing SandboxBackendProtocol.
+    """
+    ...
 ```
 
-This is what Ramp uses for their Inspect agent — container-based isolation with fast spin-up.
-
-#### Daytona
-
-```bash
-pip install langchain-daytona
-```
+2. **Register it** in `agent/utils/sandbox.py` by importing your factory and adding it to `SANDBOX_FACTORIES`:
 
 ```python
-from daytona import Daytona
-from langchain_daytona import DaytonaSandbox
+from agent.integrations.my_provider import create_my_provider_sandbox
 
-sandbox = Daytona().create()
-sandbox_backend = DaytonaSandbox(sandbox=sandbox)
+SANDBOX_FACTORIES = {
+    ...
+    "my_provider": create_my_provider_sandbox,
+}
 ```
 
-#### Runloop
-
-```bash
-pip install langchain-runloop
-```
-
-```python
-import os
-from runloop_api_client import RunloopSDK
-from langchain_runloop import RunloopSandbox
-
-client = RunloopSDK(bearer_token=os.environ["RUNLOOP_API_KEY"])
-devbox = client.devbox.create()
-sandbox_backend = RunloopSandbox(devbox=devbox)
-```
-
-#### Local shell (no isolation — development only)
-
-```python
-from deepagents.backends import LocalShellBackend
-
-sandbox_backend = LocalShellBackend(
-    root_dir="/path/to/repo",
-    inherit_env=True,
-)
-```
-
-> **Warning**: `LocalShellBackend` runs commands directly on your host machine with no sandboxing. Only use for local development with human-in-the-loop enabled.
-
-#### Wiring it up
-
-All providers implement `SandboxBackendProtocol` and are interchangeable. Replace the sandbox creation in `agent/server.py`:
-
-```python
-# Before (LangSmith)
-sandbox_backend = await asyncio.to_thread(create_langsmith_sandbox)
-
-# After (any provider)
-sandbox_backend = await asyncio.to_thread(create_my_sandbox)
-```
+The factory must return an object implementing `SandboxBackendProtocol` from `deepagents`. See the existing integration files for reference.
 
 ### Building a custom sandbox provider
 

agent/integrations/daytona.py

@@ -0,0 +1,22 @@
+import os
+
+from daytona import CreateSandboxFromSnapshotParams, Daytona, DaytonaConfig
+from langchain_daytona import DaytonaSandbox
+
+# TODO: Update this to include your specific sandbox configuration
+DAYTONA_SANDBOX_PARAMS = CreateSandboxFromSnapshotParams(snapshot="daytonaio/sandbox:0.6.0")
+
+
+def create_daytona_sandbox(sandbox_id: str | None = None):
+    api_key = os.getenv("DAYTONA_API_KEY")
+    if not api_key:
+        raise ValueError("DAYTONA_API_KEY environment variable is required")
+
+    daytona = Daytona(config=DaytonaConfig(api_key=api_key))
+
+    if sandbox_id:
+        sandbox = daytona.get(sandbox_id)
+    else:
+        sandbox = daytona.create(params=DAYTONA_SANDBOX_PARAMS)
+
+    return DaytonaSandbox(sandbox=sandbox)

agent/integrations/local.py

@@ -0,0 +1,26 @@
+import os
+
+from deepagents.backends import LocalShellBackend
+
+
+def create_local_sandbox(sandbox_id: str | None = None):
+    """Create a local shell sandbox with no isolation.
+
+    WARNING: This runs commands directly on the host machine with no sandboxing.
+    Only use for local development with human-in-the-loop enabled.
+
+    The root directory defaults to the current working directory and can be
+    overridden via the LOCAL_SANDBOX_ROOT_DIR environment variable.
+
+    Args:
+        sandbox_id: Ignored for local sandboxes; accepted for interface compatibility.
+
+    Returns:
+        LocalShellBackend instance implementing SandboxBackendProtocol.
+    """
+    root_dir = os.getenv("LOCAL_SANDBOX_ROOT_DIR", os.getcwd())
+
+    return LocalShellBackend(
+        root_dir=root_dir,
+        inherit_env=True,
+    )

agent/integrations/modal.py

@@ -0,0 +1,26 @@
+import os
+
+import modal
+from langchain_modal import ModalSandbox
+
+MODAL_APP_NAME = os.getenv("MODAL_APP_NAME", "open-swe")
+
+
+def create_modal_sandbox(sandbox_id: str | None = None):
+    """Create or reconnect to a Modal sandbox.
+
+    Args:
+        sandbox_id: Optional existing sandbox ID to reconnect to.
+            If None, creates a new sandbox.
+
+    Returns:
+        ModalSandbox instance implementing SandboxBackendProtocol.
+    """
+    app = modal.App.lookup(MODAL_APP_NAME)
+
+    if sandbox_id:
+        sandbox = modal.Sandbox.from_id(sandbox_id, app=app)
+    else:
+        sandbox = modal.Sandbox.create(app=app)
+
+    return ModalSandbox(sandbox=sandbox)

agent/integrations/runloop.py

@@ -0,0 +1,30 @@
+import os
+
+from langchain_runloop import RunloopSandbox
+from runloop_api_client import Client
+
+
+def create_runloop_sandbox(sandbox_id: str | None = None):
+    """Create or reconnect to a Runloop devbox sandbox.
+
+    Requires the RUNLOOP_API_KEY environment variable to be set.
+
+    Args:
+        sandbox_id: Optional existing devbox ID to reconnect to.
+            If None, creates a new devbox.
+
+    Returns:
+        RunloopSandbox instance implementing SandboxBackendProtocol.
+    """
+    api_key = os.getenv("RUNLOOP_API_KEY")
+    if not api_key:
+        raise ValueError("RUNLOOP_API_KEY environment variable is required")
+
+    client = Client(bearer_token=api_key)
+
+    if sandbox_id:
+        devbox = client.devboxes.retrieve(sandbox_id)
+    else:
+        devbox = client.devboxes.create()
+
+    return RunloopSandbox(devbox=devbox)

agent/middleware/open_pr.py

@@ -30,6 +30,7 @@ from ..utils.github import (
     git_push,
 )
 from ..utils.github_token import get_github_token
+from ..utils.sandbox_paths import aresolve_repo_dir
 from ..utils.sandbox_state import get_sandbox_backend
 
 logger = logging.getLogger(__name__)
@@ -92,10 +93,9 @@ async def open_pr_if_needed(
         repo_name = repo_config.get("name")
 
         sandbox_backend = await get_sandbox_backend(thread_id)
-        repo_dir = f"/workspace/{repo_name}"
-
-        if not sandbox_backend or not repo_dir:
+        if not sandbox_backend or not repo_name:
             return None
+        repo_dir = await aresolve_repo_dir(sandbox_backend, repo_name)
 
         has_uncommitted_changes = await asyncio.to_thread(
             git_has_uncommitted_changes, sandbox_backend, repo_dir

agent/server.py

@@ -4,6 +4,7 @@
 # Suppress deprecation warnings from langchain_core (e.g., Pydantic V1 on Python 3.14+)
 # ruff: noqa: E402
 import logging
+import shlex
 import warnings
 
 logger = logging.getLogger(__name__)
@@ -25,7 +26,6 @@ from deepagents import create_deep_agent
 from deepagents.backends.protocol import SandboxBackendProtocol
 from langsmith.sandbox import SandboxClientError
 
-from .integrations.langsmith import create_langsmith_sandbox
 from .middleware import (
     ToolErrorMiddleware,
     check_message_queue_before_model,
@@ -43,6 +43,7 @@ from .tools import (
 )
 from .utils.auth import resolve_github_token
 from .utils.model import make_model
+from .utils.sandbox import create_sandbox
 
 client = get_client()
 
@@ -59,6 +60,7 @@ from .utils.github import (
     remove_directory,
     setup_git_credentials,
 )
+from .utils.sandbox_paths import aresolve_repo_dir, aresolve_sandbox_work_dir
 from .utils.sandbox_state import SANDBOX_BACKENDS, get_sandbox_id_from_metadata
 
 
@@ -88,9 +90,14 @@ async def _clone_or_pull_repo_in_sandbox(  # noqa: PLR0915
         logger.error(msg)
         raise ValueError(msg)
 
-    repo_dir = f"/workspace/{repo}"
+    work_dir = await aresolve_sandbox_work_dir(sandbox_backend)
+    repo_dir = await aresolve_repo_dir(sandbox_backend, repo)
     clean_url = f"https://github.com/{owner}/{repo}.git"
     cred_helper_arg = f"-c credential.helper='store --file={_CRED_FILE_PATH}'"
+    safe_repo_dir = shlex.quote(repo_dir)
+    safe_clean_url = shlex.quote(clean_url)
+
+    logger.info("Resolved sandbox work dir to %s", work_dir)
 
     is_git_repo = await loop.run_in_executor(None, is_valid_git_repo, sandbox_backend, repo_dir)
 
@@ -147,7 +154,7 @@ async def _clone_or_pull_repo_in_sandbox(  # noqa: PLR0915
         result = await loop.run_in_executor(
             None,
             sandbox_backend.execute,
-            f"git {cred_helper_arg} clone {clean_url} {repo_dir}",
+            f"git {cred_helper_arg} clone {safe_clean_url} {safe_repo_dir}",
         )
         logger.debug("Git clone result: exit_code=%s", result.exit_code)
     except Exception:
@@ -183,7 +190,7 @@ async def _recreate_sandbox(
         metadata={"sandbox_id": SANDBOX_CREATING},
     )
     try:
-        sandbox_backend = await asyncio.to_thread(create_langsmith_sandbox)
+        sandbox_backend = await asyncio.to_thread(create_sandbox)
         repo_dir = await _clone_or_pull_repo_in_sandbox(
             sandbox_backend, repo_owner, repo_name, github_token
         )
@@ -283,7 +290,7 @@ async def get_agent(config: RunnableConfig) -> Pregel:  # noqa: PLR0915
 
         try:
             # Create sandbox without context manager cleanup (sandbox persists)
-            sandbox_backend = await asyncio.to_thread(create_langsmith_sandbox)
+            sandbox_backend = await asyncio.to_thread(create_sandbox)
             logger.info("Sandbox created: %s", sandbox_backend.id)
 
             repo_dir = None
@@ -310,7 +317,7 @@ async def get_agent(config: RunnableConfig) -> Pregel:  # noqa: PLR0915
         logger.info("Connecting to existing sandbox %s", sandbox_id)
         try:
             # Connect to existing sandbox without context manager cleanup
-            sandbox_backend = await asyncio.to_thread(create_langsmith_sandbox, sandbox_id)
+            sandbox_backend = await asyncio.to_thread(create_sandbox, sandbox_id)
             logger.info("Connected to existing sandbox %s", sandbox_id)
         except Exception:
             logger.warning("Failed to connect to existing sandbox %s, creating new one", sandbox_id)
@@ -321,7 +328,7 @@ async def get_agent(config: RunnableConfig) -> Pregel:  # noqa: PLR0915
             )
 
             try:
-                sandbox_backend = await asyncio.to_thread(create_langsmith_sandbox)
+                sandbox_backend = await asyncio.to_thread(create_sandbox)
                 logger.info("New sandbox created: %s", sandbox_backend.id)
             except Exception:
                 logger.exception("Failed to create replacement sandbox")

agent/tools/commit_and_open_pr.py

@@ -18,6 +18,7 @@ from ..utils.github import (
     git_push,
 )
 from ..utils.github_token import get_github_token
+from ..utils.sandbox_paths import resolve_repo_dir
 from ..utils.sandbox_state import get_sandbox_backend_sync
 
 logger = logging.getLogger(__name__)
@@ -129,7 +130,7 @@ def commit_and_open_pr(
         if not sandbox_backend:
             return {"success": False, "error": "No sandbox found for thread", "pr_url": None}
 
-        repo_dir = f"/workspace/{repo_name}"
+        repo_dir = resolve_repo_dir(sandbox_backend, repo_name)
 
         has_uncommitted_changes = git_has_uncommitted_changes(sandbox_backend, repo_dir)
         git_fetch_origin(sandbox_backend, repo_dir)

agent/utils/sandbox.py

@@ -0,0 +1,35 @@
+import os
+
+from agent.integrations.daytona import create_daytona_sandbox
+from agent.integrations.langsmith import create_langsmith_sandbox
+from agent.integrations.local import create_local_sandbox
+from agent.integrations.modal import create_modal_sandbox
+from agent.integrations.runloop import create_runloop_sandbox
+
+SANDBOX_FACTORIES = {
+    "langsmith": create_langsmith_sandbox,
+    "daytona": create_daytona_sandbox,
+    "modal": create_modal_sandbox,
+    "runloop": create_runloop_sandbox,
+    "local": create_local_sandbox,
+}
+
+
+def create_sandbox(sandbox_id: str | None = None):
+    """Create or reconnect to a sandbox using the configured provider.
+
+    The provider is selected via the SANDBOX_TYPE environment variable.
+    Supported values: langsmith (default), daytona, modal, runloop, local.
+
+    Args:
+        sandbox_id: Optional existing sandbox ID to reconnect to.
+
+    Returns:
+        A sandbox backend implementing SandboxBackendProtocol.
+    """
+    sandbox_type = os.getenv("SANDBOX_TYPE", "langsmith")
+    factory = SANDBOX_FACTORIES.get(sandbox_type)
+    if not factory:
+        supported = ", ".join(sorted(SANDBOX_FACTORIES))
+        raise ValueError(f"Invalid sandbox type: {sandbox_type}. Supported types: {supported}")
+    return factory(sandbox_id)

agent/utils/sandbox_paths.py

@@ -0,0 +1,153 @@
+"""Helpers for resolving portable writable paths inside sandboxes."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import posixpath
+import shlex
+from collections.abc import Iterable
+from typing import Any
+
+from deepagents.backends.protocol import SandboxBackendProtocol
+
+logger = logging.getLogger(__name__)
+
+_WORK_DIR_CACHE_ATTR = "_open_swe_resolved_work_dir"
+_PROVIDER_ATTR_NAMES = ("sandbox", "_sandbox")
+
+
+def resolve_repo_dir(sandbox_backend: SandboxBackendProtocol, repo_name: str) -> str:
+    """Resolve the repository directory for a sandbox backend."""
+    if not repo_name:
+        raise ValueError("repo_name must be a non-empty string")
+
+    work_dir = resolve_sandbox_work_dir(sandbox_backend)
+    return posixpath.join(work_dir, repo_name)
+
+
+async def aresolve_repo_dir(sandbox_backend: SandboxBackendProtocol, repo_name: str) -> str:
+    """Async wrapper around resolve_repo_dir for use in event-loop code."""
+    return await asyncio.to_thread(resolve_repo_dir, sandbox_backend, repo_name)
+
+
+def resolve_sandbox_work_dir(sandbox_backend: SandboxBackendProtocol) -> str:
+    """Resolve a writable base directory for repository operations."""
+    cached_work_dir = getattr(sandbox_backend, _WORK_DIR_CACHE_ATTR, None)
+    if isinstance(cached_work_dir, str) and cached_work_dir:
+        return cached_work_dir
+
+    checked_candidates: list[str] = []
+    for candidate in _iter_work_dir_candidates(sandbox_backend):
+        checked_candidates.append(candidate)
+        if _is_writable_directory(sandbox_backend, candidate):
+            _cache_work_dir(sandbox_backend, candidate)
+            return candidate
+
+    msg = "Failed to resolve a writable sandbox work directory"
+    if checked_candidates:
+        msg = f"{msg}. Candidates checked: {', '.join(checked_candidates)}"
+    raise RuntimeError(msg)
+
+
+async def aresolve_sandbox_work_dir(sandbox_backend: SandboxBackendProtocol) -> str:
+    """Async wrapper around resolve_sandbox_work_dir for use in event-loop code."""
+    return await asyncio.to_thread(resolve_sandbox_work_dir, sandbox_backend)
+
+
+def _iter_work_dir_candidates(
+    sandbox_backend: SandboxBackendProtocol,
+) -> Iterable[str]:
+    seen: set[str] = set()
+
+    for candidate in _iter_provider_paths(sandbox_backend, "get_work_dir"):
+        if candidate not in seen:
+            seen.add(candidate)
+            yield candidate
+
+    shell_work_dir = _resolve_shell_path(sandbox_backend, "pwd")
+    if shell_work_dir and shell_work_dir not in seen:
+        seen.add(shell_work_dir)
+        yield shell_work_dir
+
+    for candidate in _iter_provider_paths(
+        sandbox_backend,
+        "get_user_home_dir",
+        "get_user_root_dir",
+    ):
+        if candidate not in seen:
+            seen.add(candidate)
+            yield candidate
+
+    shell_home_dir = _resolve_shell_path(sandbox_backend, "printf '%s' \"$HOME\"")
+    if shell_home_dir and shell_home_dir not in seen:
+        seen.add(shell_home_dir)
+        yield shell_home_dir
+
+
+def _iter_provider_paths(
+    sandbox_backend: SandboxBackendProtocol,
+    *method_names: str,
+) -> Iterable[str]:
+    for provider in _iter_path_providers(sandbox_backend):
+        for method_name in method_names:
+            path = _call_path_method(provider, method_name)
+            if path:
+                yield path
+
+
+def _iter_path_providers(sandbox_backend: SandboxBackendProtocol) -> Iterable[Any]:
+    yield sandbox_backend
+    for attr_name in _PROVIDER_ATTR_NAMES:
+        provider = getattr(sandbox_backend, attr_name, None)
+        if provider is not None:
+            yield provider
+
+
+def _call_path_method(provider: Any, method_name: str) -> str | None:
+    method = getattr(provider, method_name, None)
+    if not callable(method):
+        return None
+
+    try:
+        return _normalize_path(method())
+    except Exception:
+        logger.debug("Failed to call %s on %s", method_name, type(provider).__name__, exc_info=True)
+        return None
+
+
+def _resolve_shell_path(
+    sandbox_backend: SandboxBackendProtocol,
+    command: str,
+) -> str | None:
+    result = sandbox_backend.execute(command)
+    if result.exit_code != 0:
+        return None
+    return _normalize_path(result.output)
+
+
+def _normalize_path(raw_path: str | None) -> str | None:
+    if raw_path is None:
+        return None
+
+    path = raw_path.strip()
+    if not path or not path.startswith("/"):
+        return None
+
+    return posixpath.normpath(path)
+
+
+def _is_writable_directory(
+    sandbox_backend: SandboxBackendProtocol,
+    directory: str,
+) -> bool:
+    safe_directory = shlex.quote(directory)
+    result = sandbox_backend.execute(f"test -d {safe_directory} && test -w {safe_directory}")
+    return result.exit_code == 0
+
+
+def _cache_work_dir(sandbox_backend: SandboxBackendProtocol, work_dir: str) -> None:
+    try:
+        setattr(sandbox_backend, _WORK_DIR_CACHE_ATTR, work_dir)
+    except Exception:
+        logger.debug("Failed to cache sandbox work dir on %s", type(sandbox_backend).__name__)

agent/utils/sandbox_state.py

@@ -8,7 +8,7 @@ from typing import Any
 
 from langgraph.config import get_config
 
-from ..integrations.langsmith import create_langsmith_sandbox
+from .sandbox import create_sandbox
 
 logger = logging.getLogger(__name__)
 
@@ -36,7 +36,7 @@ async def get_sandbox_backend(thread_id: str) -> Any | None:
     if not sandbox_id:
         raise ValueError(f"Missing sandbox_id in thread metadata for {thread_id}")
 
-    sandbox_backend = await asyncio.to_thread(create_langsmith_sandbox, sandbox_id)
+    sandbox_backend = await asyncio.to_thread(create_sandbox, sandbox_id)
     SANDBOX_BACKENDS[thread_id] = sandbox_backend
     return sandbox_backend
 

pyproject.toml

@@ -20,6 +20,9 @@ dependencies = [
     "langgraph-cli[inmem]>=0.4.12",
     "langsmith>=0.7.1",
     "langchain-openai==1.1.10",
+    "langchain-daytona>=0.0.3",
+    "langchain-modal>=0.0.2",
+    "langchain-runloop>=0.0.3",
 ]
 
 [project.optional-dependencies]

tests/test_sandbox_paths.py

@@ -0,0 +1,121 @@
+from __future__ import annotations
+
+import shlex
+
+from deepagents.backends.protocol import ExecuteResponse
+
+from agent.utils.sandbox_paths import (
+    aresolve_repo_dir,
+    resolve_repo_dir,
+    resolve_sandbox_work_dir,
+)
+
+
+class _FakeProvider:
+    def __init__(self, work_dir: str | None = None, home_dir: str | None = None) -> None:
+        self._work_dir = work_dir
+        self._home_dir = home_dir
+
+    def get_work_dir(self) -> str:
+        if self._work_dir is None:
+            raise RuntimeError("work dir unavailable")
+        return self._work_dir
+
+    def get_user_home_dir(self) -> str:
+        if self._home_dir is None:
+            raise RuntimeError("home dir unavailable")
+        return self._home_dir
+
+
+class _FakeSandboxBackend:
+    def __init__(
+        self,
+        *,
+        provider: _FakeProvider | None = None,
+        shell_paths: dict[str, str] | None = None,
+        writable_dirs: set[str] | None = None,
+    ) -> None:
+        self.sandbox = provider
+        self.shell_paths = shell_paths or {}
+        self.writable_dirs = writable_dirs or set()
+        self.commands: list[str] = []
+
+    @property
+    def id(self) -> str:
+        return "fake-sandbox"
+
+    def execute(self, command: str, *, timeout: int | None = None) -> ExecuteResponse:
+        del timeout
+        self.commands.append(command)
+
+        if command in self.shell_paths:
+            return ExecuteResponse(
+                output=self.shell_paths[command],
+                exit_code=0,
+                truncated=False,
+            )
+
+        if command.startswith("test -d "):
+            path = shlex.split(command)[2]
+            exit_code = 0 if path in self.writable_dirs else 1
+            return ExecuteResponse(output="", exit_code=exit_code, truncated=False)
+
+        return ExecuteResponse(output="", exit_code=1, truncated=False)
+
+
+def test_resolve_repo_dir_uses_provider_work_dir() -> None:
+    backend = _FakeSandboxBackend(
+        provider=_FakeProvider(work_dir="/workspace"),
+        writable_dirs={"/workspace"},
+    )
+
+    repo_dir = resolve_repo_dir(backend, "open-swe")
+
+    assert repo_dir == "/workspace/open-swe"
+    assert backend.commands == ["test -d /workspace && test -w /workspace"]
+
+
+def test_resolve_sandbox_work_dir_falls_back_to_home_when_work_dir_is_not_writable() -> None:
+    backend = _FakeSandboxBackend(
+        provider=_FakeProvider(work_dir="/workspace", home_dir="/home/daytona"),
+        shell_paths={
+            "pwd": "/workspace",
+            "printf '%s' \"$HOME\"": "/home/daytona",
+        },
+        writable_dirs={"/home/daytona"},
+    )
+
+    work_dir = resolve_sandbox_work_dir(backend)
+
+    assert work_dir == "/home/daytona"
+    assert backend.commands == [
+        "test -d /workspace && test -w /workspace",
+        "pwd",
+        "test -d /home/daytona && test -w /home/daytona",
+    ]
+
+
+def test_resolve_sandbox_work_dir_caches_the_result() -> None:
+    backend = _FakeSandboxBackend(
+        provider=_FakeProvider(work_dir="/workspace"),
+        writable_dirs={"/workspace"},
+    )
+
+    first = resolve_sandbox_work_dir(backend)
+    second = resolve_sandbox_work_dir(backend)
+
+    assert first == "/workspace"
+    assert second == "/workspace"
+    assert backend.commands == ["test -d /workspace && test -w /workspace"]
+
+
+async def test_aresolve_repo_dir_offloads_sync_resolution() -> None:
+    backend = _FakeSandboxBackend(
+        provider=_FakeProvider(work_dir="/home/daytona"),
+        writable_dirs={"/home/daytona"},
+    )
+
+    repo_dir = await aresolve_repo_dir(backend, "open-swe")
+
+    assert repo_dir == "/home/daytona/open-swe"
+    assert backend.commands == ["test -d /home/daytona && test -w /home/daytona"]