fix(sandbox): reject apply_patch create_file on an existing file

src/agents/sandbox/apply_patch.py

@@ -127,7 +127,11 @@ async def apply_operation(
                     path=operation.path,
                     cause=exc,
                 ) from exc
-            await self._write_text(destination, created_text)
+            # Hand over the unresolved path. destination has already been through
+            # normalize_path(), which resolves leaf symlinks on some backends, so passing
+            # it would ask the backend to create the link target instead of the requested
+            # name and a dangling link would be reported as a successful create.
+            await self._write_new_text(relative_path, created_text, display_path=display_path)
             return ApplyPatchResult(output=f"Created {display_path}")
 
         raise ApplyPatchDiffError(
@@ -186,6 +190,25 @@ async def _ensure_exists(self, destination: Path, *, display_path: str) -> None:
         else:
             handle.close()
 
+    async def _write_new_text(self, destination: Path, text: str, *, display_path: str) -> None:
+        # Add File is documented as creating a new file, so the name is claimed
+        # exclusively by the backend rather than checked and then overwritten.
+        try:
+            await self._session.write_new_file(
+                destination,
+                io.BytesIO(text.encode("utf-8")),
+                user=self._user,
+            )
+        except FileExistsError as exc:
+            raise ApplyPatchDiffError(
+                message=(
+                    f"apply_patch cannot create {display_path} because it already exists. "
+                    "Use an update_file operation to change an existing file."
+                ),
+                path=display_path,
+                cause=exc,
+            ) from exc
+
     async def _read_text(self, destination: Path, *, op_path: str, decode_path: Path) -> str:
         try:
             handle = await self._session.read(destination, user=self._user)

src/agents/sandbox/sandboxes/unix_local.py

@@ -1015,6 +1015,56 @@ async def write(
         except OSError as e:
             raise WorkspaceArchiveWriteError(path=workspace_path, cause=e) from e
 
+    async def write_new_file(
+        self,
+        path: Path,
+        data: io.IOBase,
+        *,
+        user: str | User | None = None,
+    ) -> None:
+        if user is not None:
+            await super().write_new_file(path, data, user=user)
+            return
+
+        payload = coerce_write_payload(path=path, data=data)
+        # Validate the parent with the normal policy so grants and symlinked parents are
+        # still enforced, then keep the final component unresolved. normalize_path()
+        # resolves symlinks, which would turn a dangling link at the target name into its
+        # absent target and let the write land there instead of being rejected.
+        requested = Path(path)
+        parent_path = self.normalize_path(requested.parent, for_write=True)
+        workspace_path = parent_path / requested.name
+        staging_path = parent_path / f".apply-patch-create-{uuid.uuid4().hex}"
+        # Classify a visible collision before staging, so a target inside a non-writable
+        # parent reports the collision rather than a permission failure from the staging
+        # write. os.path.lexists does not follow a symlink at the target name.
+        if os.path.lexists(workspace_path):
+            raise FileExistsError(str(workspace_path))
+
+        try:
+            # Only the link may report a collision. A parent that is a regular file also
+            # raises FileExistsError from mkdir, and reporting that as "the target already
+            # exists" would send the model to update_file for a target that is absent.
+            try:
+                parent_path.mkdir(parents=True, exist_ok=True)
+                with staging_path.open("wb") as staged:
+                    shutil.copyfileobj(payload.stream, staged)
+            except OSError as e:
+                raise WorkspaceArchiveWriteError(path=workspace_path, cause=e) from e
+
+            # os.link claims the name in one step and fails with EEXIST when it is taken
+            # by anything, including a directory or a dangling symlink. Linking a complete
+            # payload means a failed write never leaves a file holding the name.
+            try:
+                os.link(staging_path, workspace_path)
+            except FileExistsError:
+                raise
+            except OSError as e:
+                raise WorkspaceArchiveWriteError(path=workspace_path, cause=e) from e
+        finally:
+            with suppress(OSError):
+                staging_path.unlink()
+
     async def _write_stream_with_exec(
         self,
         path: Path,

src/agents/sandbox/session/base_sandbox_session.py

@@ -2,7 +2,9 @@
 import asyncio
 import io
 import shlex
+import uuid
 from collections.abc import Awaitable, Callable, Mapping, Sequence
+from contextlib import suppress
 from pathlib import Path, PurePath
 from typing import Literal, NoReturn, TypeVar
 
@@ -149,6 +151,31 @@
     fi
 done
 """.strip()
+_EXCLUSIVE_CREATE_EXISTS_CODE = 13
+# Classify an already-visible target before any payload is staged. Without this the
+# staging write runs first, so a target inside an executable but non-writable parent
+# fails on permissions and the caller sees a write error instead of the collision error
+# that tells it to use update_file. The atomic claim below still decides real races.
+_TARGET_EXISTS_SCRIPT = 'target="$1"\nif [ -e "$target" ] || [ -L "$target" ]; then exit 13; fi\n'
+
+# ``ln`` claims the target name and fails when that name is already taken. Linking a
+# fully written staging file means the content is complete before the name exists, so a
+# failed or cancelled upload cannot leave a file behind that holds the name. The leading
+# test rejects a name held by a directory, which ``ln`` would otherwise treat as a target
+# directory and populate; the trailing test only classifies a failure, so "already
+# exists" stays separable from any other error without parsing shell-specific stderr.
+# ``ln`` is a regular command, unlike ``:``, so its failure still reaches the explicit
+# exit mapping on shells where ``:`` is a special builtin. The caller creates the parent,
+# so this script never has to create one as a different identity.
+_EXCLUSIVE_CREATE_SCRIPT = (
+    'target="$1"\n'
+    'source="$2"\n'
+    'if [ -e "$target" ] || [ -L "$target" ]; then exit 13; fi\n'
+    'ln "$source" "$target" 2>/dev/null && exit 0\n'
+    'if [ -e "$target" ] || [ -L "$target" ]; then exit 13; fi\n'
+    "exit 14\n"
+)
+
 _WRITE_ACCESS_CHECK_SCRIPT = (
     'target="$1"\n'
     'if [ -e "$target" ]; then\n'
@@ -945,6 +972,79 @@ async def write(
         :param user: Optional sandbox user to perform the write as.
         """
 
+    async def write_new_file(
+        self,
+        path: Path,
+        data: io.IOBase,
+        *,
+        user: str | User | None = None,
+    ) -> None:
+        """Write a file that must not already exist.
+
+        The target name is claimed in a single atomic step once the payload is complete,
+        so a concurrent creator either loses the race or keeps its own content, and a
+        failed write does not leave a partial file holding the name.
+
+        :param path: Absolute path in the container or path relative to the
+                workspace root.
+        :param data: A file-like object positioned at the start of the payload.
+        :param user: Optional sandbox user to perform the write as.
+        :raises FileExistsError: If the path already exists, including a dangling symlink.
+        """
+        # Validate the parent so grants and symlinked parents are still enforced, then
+        # keep the final component unresolved. A path policy that resolves symlinks would
+        # otherwise turn a dangling link at the target name into its absent target and let
+        # the create land there instead of being rejected.
+        requested = Path(path)
+        parent_path = await self._validate_path_access(requested.parent, for_write=True)
+        workspace_path = parent_path / requested.name
+        path_arg = sandbox_path_str(workspace_path)
+        # A fixed-length staging basename. Deriving it from the destination made the
+        # staging name longer than the destination, so a name that fits the filesystem's
+        # component limit could still fail to stage.
+        staging_path = parent_path / f".apply-patch-create-{uuid.uuid4().hex}"
+        staging_arg = sandbox_path_str(staging_path)
+
+        preflight = await self.exec(
+            "sh", "-c", _TARGET_EXISTS_SCRIPT, "sh", path_arg, shell=False, user=user
+        )
+        if preflight.exit_code == _EXCLUSIVE_CREATE_EXISTS_CODE:
+            raise FileExistsError(path_arg)
+
+        try:
+            # Create the parent as the bound user so a fresh nested path is owned the same
+            # way the ordinary write path owned it.
+            await self.mkdir(parent_path, parents=True, user=user)
+            await self.write(staging_path, data, user=user)
+            # -c rather than -lc: this runs on a filesystem-only capability set, so it
+            # must not source workspace-writable shell startup files.
+            result = await self.exec(
+                "sh",
+                "-c",
+                _EXCLUSIVE_CREATE_SCRIPT,
+                "sh",
+                path_arg,
+                staging_arg,
+                shell=False,
+                user=user,
+            )
+            if result.exit_code == _EXCLUSIVE_CREATE_EXISTS_CODE:
+                raise FileExistsError(path_arg)
+            if not result.ok():
+                raise WorkspaceArchiveWriteError(
+                    path=workspace_path,
+                    context={
+                        "command": ["sh", "-c", "<exclusive_create>", path_arg, staging_arg],
+                        "stdout": result.stdout.decode("utf-8", errors="replace"),
+                        "stderr": result.stderr.decode("utf-8", errors="replace"),
+                    },
+                )
+        finally:
+            # The staging entry is an implementation detail, and removing it must not
+            # replace the outcome of the create.
+            with suppress(Exception):
+                await self.rm(staging_path, user=user)
+
     async def _check_read_with_exec(
         self, path: Path | str, *, user: str | User | None = None
     ) -> Path:

src/agents/sandbox/session/sandbox_session.py

@@ -677,6 +677,19 @@ async def write(
     ) -> None:
         await self._inner.write(path, data, user=user)
 
+    @instrumented_op("write", data=_write_start_data)
+    async def write_new_file(
+        self,
+        path: Path,
+        data: io.IOBase,
+        *,
+        user: str | User | None = None,
+    ) -> None:
+        # Forwarded so a backend with a native exclusive-create primitive is actually
+        # used. Without this the wrapper would fall back to the shared implementation and
+        # bypass the inner session's override.
+        await self._inner.write_new_file(path, data, user=user)
+
     @instrumented_op(
         "running",
         finish_data=_running_finish_data,

tests/sandbox/_apply_patch_test_session.py

@@ -56,6 +56,20 @@ async def write(
         else:
             self.files[normalized] = bytes(payload)
 
+    async def write_new_file(
+        self,
+        path: Path,
+        data: io.IOBase,
+        *,
+        user: str | User | None = None,
+    ) -> None:
+        normalized = self.normalize_path(path)
+        if normalized in self.files:
+            raise FileExistsError(str(normalized))
+        # Real backends create the parents inside the primitive, so record that here too.
+        await self.mkdir(normalized.parent, parents=True, user=user)
+        await self.write(path, data, user=user)
+
     async def _exec_internal(
         self,
         *command: str | Path,

tests/sandbox/test_apply_patch.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import io
 from pathlib import Path
 
 import pytest
@@ -11,7 +12,9 @@
     ApplyPatchDiffError,
     ApplyPatchFileNotFoundError,
     ApplyPatchPathError,
+    WorkspaceReadNotFoundError,
 )
+from agents.sandbox.types import User
 from tests.sandbox._apply_patch_test_session import (
     ApplyPatchSession,
     ProviderNotFoundApplyPatchSession,
@@ -411,3 +414,49 @@ async def test_apply_patch_mapping_operation_rejects_non_string_move_to() -> Non
         )
 
     assert session.files[Path("/workspace/old.txt")] == b"alpha\n"
+
+
+@pytest.mark.asyncio
+async def test_apply_patch_create_rejects_an_existing_file() -> None:
+    session = ApplyPatchSession()
+    session.files[Path("/workspace/notes.txt")] = b"alpha\n"
+
+    with pytest.raises(ApplyPatchDiffError):
+        await session.apply_patch(
+            ApplyPatchOperation(
+                type="create_file",
+                path="notes.txt",
+                diff="+beta\n",
+            )
+        )
+
+    assert session.files[Path("/workspace/notes.txt")] == b"alpha\n"
+
+
+class _AlwaysMissingReadApplyPatchSession(ApplyPatchSession):
+    """Reports every path as missing while still holding the file.
+
+    A create that only probed with read() would be told the path is free and would
+    overwrite the stored content, so this pins the rejection to the write boundary.
+    """
+
+    async def read(self, path: Path, *, user: str | User | None = None) -> io.BytesIO:
+        _ = (path, user)
+        raise WorkspaceReadNotFoundError(path=path)
+
+
+@pytest.mark.asyncio
+async def test_apply_patch_create_rejects_an_existing_file_without_reading_it() -> None:
+    session = _AlwaysMissingReadApplyPatchSession()
+    session.files[Path("/workspace/notes.txt")] = b"alpha\n"
+
+    with pytest.raises(ApplyPatchDiffError):
+        await session.apply_patch(
+            ApplyPatchOperation(
+                type="create_file",
+                path="notes.txt",
+                diff="+beta\n",
+            )
+        )
+
+    assert session.files[Path("/workspace/notes.txt")] == b"alpha\n"