fix(sandbox): scope exclusive Add File creation to UnixLocal

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,24 @@ 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:
+        # Backends with a native no-clobber primitive can reject occupied names.
+        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/errors.py

@@ -506,9 +506,10 @@ def __init__(
         context: Mapping[str, object] | None = None,
         cause: BaseException | None = None,
         retryable: bool | None = None,
+        message: str | None = None,
     ) -> None:
         super().__init__(
-            message=f"failed to write archive for path: {path}",
+            message=message if message is not None else f"failed to write archive for path: {path}",
             error_code=ErrorCode.WORKSPACE_ARCHIVE_WRITE_ERROR,
             op="write",
             context={"path": str(path), **_as_context(context)},

src/agents/sandbox/sandboxes/_unix_local_file_ops.py

@@ -26,6 +26,14 @@
 )
 
 
+_EXISTING_TARGET_EXIT_CODE = 13
+_INCOMPLETE_CREATE_EXIT_CODE = 14
+
+
+class _IncompleteCreateError(OSError):
+    """The destination was claimed, but writing its contents did not complete."""
+
+
 class _FileOps:
     """Operate on canonical absolute paths already authorized by the owning session."""
 
@@ -118,6 +126,34 @@ def write(self, path: Path, stream: io.IOBase) -> None:
         with out:
             shutil.copyfileobj(stream, out)
 
+    def write_new(self, path: Path, stream: io.IOBase) -> None:
+        """Create a file that must not already exist.
+
+        O_EXCL fails with EEXIST when the name is taken by anything, including a directory
+        or a dangling symlink, and it claims the name in the same syscall that creates the
+        file, so a concurrent creator either loses the race or keeps its own content.
+        """
+        with self.parent(path, for_write=True, create_parents=True) as (parent_fd, name):
+            fd = os.open(
+                name,
+                os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
+                0o666,
+                dir_fd=parent_fd,
+            )
+        try:
+            try:
+                out = os.fdopen(fd, "wb")
+            except BaseException:
+                os.close(fd)
+                raise
+            with out:
+                shutil.copyfileobj(stream, out)
+        except OSError as exc:
+            # Do not unlink the name: another workspace operation may have replaced it.
+            # Keep filesystem compatibility (no hard-link staging requirement) and let
+            # the caller inspect and recover the partial result explicitly.
+            raise _IncompleteCreateError("File creation did not complete") from exc
+
     def mkdir(self, path: Path, *, parents: bool) -> None:
         with self.parent(path, for_write=True, create_parents=parents) as (parent_fd, name):
             try:
@@ -209,6 +245,14 @@ def _main() -> None:
     path = Path(raw_path)
     if operation == "write":
         files.write(path, cast(io.IOBase, sys.stdin.buffer))
+    elif operation == "write_new":
+        try:
+            files.write_new(path, cast(io.IOBase, sys.stdin.buffer))
+        except FileExistsError:
+            # A distinct status keeps "already exists" separable from a real write failure.
+            sys.exit(_EXISTING_TARGET_EXIT_CODE)
+        except _IncompleteCreateError:
+            sys.exit(_INCOMPLETE_CREATE_EXIT_CODE)
     elif operation == "ls":
         print(json.dumps(files.listing(path), ensure_ascii=True))
     else:

src/agents/sandbox/sandboxes/unix_local.py

@@ -110,6 +110,12 @@
 
 logger = logging.getLogger(__name__)
 
+_INCOMPLETE_CREATE_MESSAGE = (
+    "File creation failed after claiming the destination. The file may contain incomplete "
+    "contents. Inspect the destination before using update_file or removing it to retry; "
+    "retrying Add File without inspection is unsafe."
+)
+
 
 def _mount_path_diagnostic_extra(mount_path: Path) -> dict[str, object]:
     return {"mount_path": str(mount_path)}
@@ -1065,6 +1071,73 @@ 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:
+        payload = coerce_write_payload(path=path, data=data)
+        # Resolve the parent the way the ordinary write path does, so a supported internal
+        # symlink such as "internal -> real" still works, then keep the leaf name
+        # unresolved so the file ops open it with O_NOFOLLOW and a symlink at the target
+        # name is rejected rather than followed.
+        requested = Path(path)
+        target = self.normalize_path(requested.parent, for_write=True) / requested.name
+        if user is not None:
+            await self._write_new_stream_with_exec(target, payload.stream, user=user)
+            return
+
+        try:
+            self._files.write_new(target, payload.stream)
+        except _unix_local_file_ops._IncompleteCreateError as e:
+            raise WorkspaceArchiveWriteError(
+                path=target,
+                cause=e,
+                retryable=False,
+                message=_INCOMPLETE_CREATE_MESSAGE,
+            ) from e
+        except FileExistsError:
+            raise
+        except OSError as e:
+            raise WorkspaceArchiveWriteError(path=target, cause=e) from e
+
+    async def _write_new_stream_with_exec(
+        self,
+        path: Path,
+        stream: io.IOBase,
+        *,
+        user: str | User,
+    ) -> None:
+        payload = stream.read()
+        if isinstance(payload, str):
+            payload = payload.encode("utf-8")
+        elif not isinstance(payload, bytes):
+            payload = bytes(payload)
+        try:
+            result = await self._run_file_operation_as_user(
+                "write_new", path, user=user, payload=payload
+            )
+        except OSError as e:
+            raise WorkspaceArchiveWriteError(path=path, cause=e) from e
+        if result.returncode == _unix_local_file_ops._EXISTING_TARGET_EXIT_CODE:
+            raise FileExistsError(str(path))
+        if result.returncode == _unix_local_file_ops._INCOMPLETE_CREATE_EXIT_CODE:
+            raise WorkspaceArchiveWriteError(
+                path=path,
+                retryable=False,
+                message=_INCOMPLETE_CREATE_MESSAGE,
+            )
+        if result.returncode:
+            raise WorkspaceArchiveWriteError(
+                path=path,
+                context={
+                    "stderr": result.stderr.decode("utf-8", errors="replace"),
+                    "operation": "write_new",
+                },
+            )
+
     async def _write_stream_with_exec(
         self,
         path: Path,
@@ -1094,14 +1167,14 @@ async def _write_stream_with_exec(
 
     async def _run_file_operation_as_user(
         self,
-        operation: Literal["ls", "write"],
+        operation: Literal["ls", "write", "write_new"],
         path: Path,
         *,
         user: str | User,
         payload: bytes = b"",
     ) -> subprocess.CompletedProcess[bytes]:
         # Authorization is synchronous and captured for this operation before dispatch.
-        path = self._files.authorize(path, for_write=operation == "write")
+        path = self._files.authorize(path, for_write=operation != "ls")
         command = self._prepare_exec_command(
             "python3",
             "-I",

src/agents/sandbox/session/base_sandbox_session.py

@@ -995,6 +995,23 @@ 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:
+        """Backend hook for apply_patch creation.
+
+        The default retains the provider's existing mkdir/write semantics. UnixLocal
+        overrides this hook to claim the leaf exclusively; other backends need a native
+        primitive before they can offer the same guarantee.
+        """
+        target = self.normalize_path(path)
+        await self.mkdir(target.parent, parents=True, user=user)
+        await self.write(target, data, 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

@@ -684,6 +684,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,50 @@ 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.
+
+    This stands in for a backend that provides an exclusive create. 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 backend primitive rather than to a probe.
+    """
+
+    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"