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_file_ops.py

@@ -25,6 +25,9 @@
 )
 
 
+_EXISTING_TARGET_EXIT_CODE = 13
+
+
 class _FileOps:
     """Operate on canonical absolute paths already authorized by the owning session."""
 
@@ -76,6 +79,28 @@ 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:
+            out = os.fdopen(fd, "wb")
+        except BaseException:
+            os.close(fd)
+            raise
+        with out:
+            shutil.copyfileobj(stream, out)
+
     def mkdir(self, path: Path, *, parents: bool) -> None:
         with self.parent(path, for_write=True, create_parents=parents) as (parent_fd, name):
             try:
@@ -167,6 +192,12 @@ 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)
     elif operation == "ls":
         print(json.dumps(files.listing(path), ensure_ascii=True))
     else:

src/agents/sandbox/sandboxes/unix_local.py

@@ -1054,6 +1054,60 @@ 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 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:
+            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,
@@ -1083,14 +1137,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

@@ -149,6 +149,12 @@
     fi
 done
 """.strip()
+_EXISTING_TARGET_EXIT_CODE = 13
+# A bare existence test. It needs only execute permission on the parent, never reads the
+# target, and reports absent when the parent itself is missing, so a nested create still
+# reaches write() and lets the backend create the parents.
+_TARGET_EXISTS_SCRIPT = 'if [ -e "$1" ] || [ -L "$1" ]; then exit 13; fi\n'
+
 _WRITE_ACCESS_CHECK_SCRIPT = (
     'target="$1"\n'
     'if [ -e "$target" ]; then\n'
@@ -945,6 +951,55 @@ 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.
+
+        This default checks the target and then writes, so it rejects the ordinary case of
+        a create aimed at a path that is already occupied and leaves the existing content
+        alone. It is not atomic: a creator that arrives between the check and the write is
+        overwritten. A backend that can express an exclusive create should override this
+        and claim the name in one step; ``UnixLocalSandboxSession`` does.
+
+        :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.
+        """
+        workspace_path = await self._validate_path_access(path, for_write=True)
+        path_arg = sandbox_path_str(workspace_path)
+        # The probe reports absent as 0, including when the parent does not exist, so 0 is
+        # the only status that may proceed. Any other status means the probe itself did not
+        # run, and write() can still succeed through a separate upload API on provider
+        # sessions, which would overwrite an existing target exactly when the precondition
+        # could not be checked.
+        probe = await self.exec(
+            "sh", "-c", _TARGET_EXISTS_SCRIPT, "sh", path_arg, shell=False, user=user
+        )
+        if probe.exit_code == _EXISTING_TARGET_EXIT_CODE:
+            raise FileExistsError(path_arg)
+        if probe.exit_code != 0:
+            raise WorkspaceArchiveWriteError(
+                path=workspace_path,
+                context={
+                    "command": ["sh", "-c", "<target_exists>", path_arg],
+                    "exit_code": probe.exit_code,
+                    "stdout": probe.stdout.decode("utf-8", errors="replace"),
+                    "stderr": probe.stderr.decode("utf-8", errors="replace"),
+                },
+            )
+        # Create the parents explicitly, the way the previous create path did, so the call
+        # sequence a caller can observe is unchanged and backends that do not create them
+        # during write() still work.
+        await self.mkdir(workspace_path.parent, parents=True, user=user)
+        await self.write(workspace_path, 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

@@ -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,

src/agents/testing/sandbox.py

@@ -331,6 +331,23 @@ def __init__(self, steps: Sequence[_SandboxStep], *, manifest: Manifest | None)
         self._calls: list[SandboxCall] = []
         self._running = False
 
+    async def write_new_file(
+        self,
+        path: Path,
+        data: io.IOBase,
+        *,
+        user: str | User | None = None,
+    ) -> None:
+        # A scripted session has no filesystem to hold a colliding entry, so an exclusive
+        # create is scripted as the same mkdir and write pair the ordinary create path uses.
+        # Delegating here also keeps the inherited default from reaching for `exec`, which a
+        # script that only configures file steps hides. Normalize first: the create path
+        # hands over an unresolved path so a symlinked leaf is not followed, but a script
+        # matching on arguments expects the workspace paths these calls have always carried.
+        target = self.normalize_path(path)
+        await self.mkdir(target.parent, parents=True, user=user)
+        await self.write(target, data, user=user)
+
     def __getattribute__(self, name: str) -> Any:
         if name in _SCRIPTABLE_METHODS:
             configured = object.__getattribute__(self, "_configured_methods")

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/_filesystem_test_session.py

@@ -21,7 +21,11 @@
     UnixLocalSandboxSessionState,
 )
 from agents.sandbox.session import SandboxSession
-from agents.sandbox.session.base_sandbox_session import BaseSandboxSession
+from agents.sandbox.session.base_sandbox_session import (
+    _EXISTING_TARGET_EXIT_CODE,
+    _TARGET_EXISTS_SCRIPT,
+    BaseSandboxSession,
+)
 from agents.sandbox.session.sandbox_session_state import SandboxSessionState
 from agents.sandbox.snapshot import NoopSnapshot, SnapshotBase, SnapshotSpec
 from agents.sandbox.types import ExecResult, Permissions, User
@@ -56,6 +60,19 @@ async def _exec_internal(
             path = Path(command_parts[2])
             exists = path.is_dir() if command_parts[1] == "-d" else path.is_file()
             return ExecResult(stdout=b"", stderr=b"", exit_code=0 if exists else 1)
+        # The exclusive-create probe is a bare existence test dispatched as `sh -c`. It has
+        # to see a dangling symlink as present, so it uses lexists rather than exists.
+        if (
+            len(command_parts) == 5
+            and command_parts[:2] == ("sh", "-c")
+            and command_parts[2] == _TARGET_EXISTS_SCRIPT
+        ):
+            present = os.path.lexists(command_parts[4])
+            return ExecResult(
+                stdout=b"",
+                stderr=b"",
+                exit_code=_EXISTING_TARGET_EXIT_CODE if present else 0,
+            )
         raise AssertionError(f"Unexpected filesystem test command: {command_parts!r}")
 
     @staticmethod