diff --git a/src/agents/sandbox/apply_patch.py b/src/agents/sandbox/apply_patch.py index 30623fdf82..5e6f8d98fc 100644 --- a/src/agents/sandbox/apply_patch.py +++ b/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) diff --git a/src/agents/sandbox/sandboxes/unix_local.py b/src/agents/sandbox/sandboxes/unix_local.py index 28eeb265ef..890299a30d 100644 --- a/src/agents/sandbox/sandboxes/unix_local.py +++ b/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, diff --git a/src/agents/sandbox/session/base_sandbox_session.py b/src/agents/sandbox/session/base_sandbox_session.py index d377bea9ef..784f30c7a3 100644 --- a/src/agents/sandbox/session/base_sandbox_session.py +++ b/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", "", 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: diff --git a/src/agents/sandbox/session/sandbox_session.py b/src/agents/sandbox/session/sandbox_session.py index 923f025857..593ff1054a 100644 --- a/src/agents/sandbox/session/sandbox_session.py +++ b/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, diff --git a/tests/sandbox/_apply_patch_test_session.py b/tests/sandbox/_apply_patch_test_session.py index 24ce567011..911581b707 100644 --- a/tests/sandbox/_apply_patch_test_session.py +++ b/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, diff --git a/tests/sandbox/test_apply_patch.py b/tests/sandbox/test_apply_patch.py index c4cd676fec..0fda2bf2c7 100644 --- a/tests/sandbox/test_apply_patch.py +++ b/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" diff --git a/tests/sandbox/test_unix_local.py b/tests/sandbox/test_unix_local.py index 9188b1fc33..2413bf9452 100644 --- a/tests/sandbox/test_unix_local.py +++ b/tests/sandbox/test_unix_local.py @@ -2,7 +2,10 @@ import asyncio import io +import os +import shutil import signal +import subprocess import tarfile import threading import time @@ -12,8 +15,13 @@ import pytest +from agents.editor import ApplyPatchOperation from agents.sandbox import SandboxPathGrant -from agents.sandbox.errors import PtySessionNotFoundError +from agents.sandbox.errors import ( + ApplyPatchDiffError, + PtySessionNotFoundError, + WorkspaceArchiveWriteError, +) from agents.sandbox.manifest import Environment, Manifest from agents.sandbox.sandboxes import unix_local as unix_local_module from agents.sandbox.sandboxes.unix_local import ( @@ -22,6 +30,10 @@ UnixLocalSandboxSessionState, _UnixPtyProcessEntry, ) +from agents.sandbox.session.base_sandbox_session import ( + _EXCLUSIVE_CREATE_EXISTS_CODE, + _EXCLUSIVE_CREATE_SCRIPT, +) from agents.sandbox.snapshot import NoopSnapshot from agents.sandbox.types import ExecResult, User @@ -610,3 +622,358 @@ def _slow_extract(tar: object, **kwargs: object) -> None: # the workspace root are only released once nothing is still writing to them. assert events == ["extract-start", "extract-end"] assert not buf.closed + + +def _exclusive_write_session(root: Path) -> UnixLocalSandboxSession: + return UnixLocalSandboxSession( + state=UnixLocalSandboxSessionState( + manifest=Manifest(root=str(root)), + snapshot=NoopSnapshot(id="noop"), + ) + ) + + +@pytest.mark.asyncio +async def test_write_new_file_keeps_an_intervening_creator_content(tmp_path: Path) -> None: + """The name is claimed by the write itself, so a creator that got there first wins.""" + session = _exclusive_write_session(tmp_path) + target = tmp_path / "notes.txt" + target.write_bytes(b"written by someone else\n") + + with pytest.raises(FileExistsError): + await session.write_new_file(Path("notes.txt"), io.BytesIO(b"clobbered")) + + assert target.read_bytes() == b"written by someone else\n" + + +@pytest.mark.asyncio +async def test_write_new_file_rejects_a_dangling_symlink(tmp_path: Path) -> None: + """A symlink entry is not absent, and the write must not follow it to its target.""" + session = _exclusive_write_session(tmp_path) + link = tmp_path / "link.txt" + link.symlink_to(tmp_path / "missing.txt") + + with pytest.raises(FileExistsError): + await session.write_new_file(Path("link.txt"), io.BytesIO(b"clobbered")) + + assert link.is_symlink() + assert not (tmp_path / "missing.txt").exists() + + +@pytest.mark.asyncio +async def test_write_new_file_creates_a_file_and_its_parents(tmp_path: Path) -> None: + session = _exclusive_write_session(tmp_path) + + await session.write_new_file(Path("nested/dir/new.txt"), io.BytesIO(b"payload")) + + assert (tmp_path / "nested" / "dir" / "new.txt").read_bytes() == b"payload" + + +class _ExitCodeUnixLocalSession(UnixLocalSandboxSession): + """Drives the shared exec-based exclusive create with a chosen exit code.""" + + def __init__(self, root: Path, exit_code: int, *, preflight_exit_code: int = 0) -> None: + super().__init__( + state=UnixLocalSandboxSessionState( + manifest=Manifest(root=str(root)), + snapshot=NoopSnapshot(id="noop"), + ) + ) + self._exit_code = exit_code + self._preflight_exit_code = preflight_exit_code + self.exec_commands: list[tuple[str, ...]] = [] + self.writes: list[Path] = [] + self.removed: list[Path] = [] + self.made_dirs: list[Path] = [] + + async def _exec_internal( + self, + *command: str | Path, + timeout: float | None = None, + ) -> ExecResult: + _ = timeout + parts = tuple(str(part) for part in command) + self.exec_commands.append(parts) + # The collision preflight is the invocation that receives only the target. + is_preflight = not any("ln " in part for part in parts) + code = self._preflight_exit_code if is_preflight else self._exit_code + return ExecResult(stdout=b"", stderr=b"", exit_code=code) + + async def write(self, path: Path, data: io.IOBase, *, user: object = None) -> None: + _ = (data, user) + self.writes.append(path) + + async def rm( + self, + path: Path | str, + *, + recursive: bool = False, + user: object = None, + ) -> None: + _ = (recursive, user) + self.removed.append(Path(path)) + + async def mkdir( + self, + path: Path | str, + *, + parents: bool = False, + user: object = None, + ) -> None: + _ = (parents, user) + self.made_dirs.append(Path(path)) + + +@pytest.mark.asyncio +async def test_write_new_file_with_a_bound_user_reports_an_existing_name(tmp_path: Path) -> None: + """A target that is already visible is rejected before any payload is staged.""" + session = _ExitCodeUnixLocalSession(tmp_path, exit_code=0, preflight_exit_code=13) + + with pytest.raises(FileExistsError): + await session.write_new_file( + Path("notes.txt"), io.BytesIO(b"payload"), user=User(name="sandbox-user") + ) + + # No payload bytes were uploaded, so a create onto an occupied name costs one probe + # rather than a full staged write that is then discarded. + assert session.writes == [] + assert session.removed == [] + + +@pytest.mark.asyncio +async def test_write_new_file_with_a_bound_user_reports_a_racing_creator(tmp_path: Path) -> None: + """A creator that wins between the preflight and the link still loses the name.""" + session = _ExitCodeUnixLocalSession(tmp_path, exit_code=13, preflight_exit_code=0) + + with pytest.raises(FileExistsError): + await session.write_new_file( + Path("notes.txt"), io.BytesIO(b"payload"), user=User(name="sandbox-user") + ) + + # Here the payload was staged before the race was detected, and the staging entry is + # still cleaned up rather than left in the workspace. + assert all(path.name.startswith(".apply-patch-create-") for path in session.writes) + assert session.removed == session.writes + assert session.made_dirs != [] + + +@pytest.mark.asyncio +async def test_write_new_file_with_a_bound_user_links_the_completed_payload( + tmp_path: Path, +) -> None: + """The payload is written first, then the target name is claimed by linking it.""" + session = _ExitCodeUnixLocalSession(tmp_path, exit_code=0) + + await session.write_new_file( + Path("notes.txt"), io.BytesIO(b"payload"), user=User(name="sandbox-user") + ) + + staged = session.writes[0] + assert staged.name.startswith(".apply-patch-create-") + dispatched = [part for cmd in session.exec_commands for part in cmd] + assert any("ln " in part for part in dispatched) + assert any(part.endswith("notes.txt") for part in dispatched) + assert str(staged) in dispatched + assert session.removed == [staged] + + +@pytest.mark.asyncio +async def test_write_new_file_with_a_bound_user_keeps_a_symlink_name_unresolved( + tmp_path: Path, +) -> None: + """The exclusive create must act on the link name, not on the target it points at.""" + session = _ExitCodeUnixLocalSession(tmp_path, exit_code=13) + (tmp_path / "link.txt").symlink_to(tmp_path / "missing.txt") + + with pytest.raises(FileExistsError): + await session.write_new_file( + Path("link.txt"), io.BytesIO(b"payload"), user=User(name="sandbox-user") + ) + + dispatched = [part for cmd in session.exec_commands for part in cmd] + assert any(part.endswith("link.txt") for part in dispatched) + assert not any(part.endswith("missing.txt") for part in dispatched) + + +@pytest.mark.parametrize("shell", ["sh", "dash", "bash"]) +def test_exclusive_create_script_reports_a_taken_name_on_each_shell( + shell: str, tmp_path: Path +) -> None: + """Run the shipped script through real shells. + + The script is dispatched as ``sh -lc``, so whichever shell provides ``/bin/sh`` + decides how a failing command is handled. An earlier version used ``:``, which is a + POSIX special builtin, so a redirection failure terminated dash before the explicit + exit mapping ran and the collision surfaced as a generic write error. This lives with + the Unix-local tests because tests/conftest.py already skips them on Windows. + """ + executable = shutil.which(shell) + if executable is None: + pytest.skip(f"{shell} is not available") + + staging = tmp_path / "staging" + staging.write_bytes(b"payload") + taken = tmp_path / "taken.txt" + taken.write_bytes(b"existing\n") + dangling = tmp_path / "dangling.txt" + dangling.symlink_to(tmp_path / "missing.txt") + + def run(target: Path) -> int: + return subprocess.run( + [executable, "-c", _EXCLUSIVE_CREATE_SCRIPT, shell, str(target), str(staging)], + capture_output=True, + ).returncode + + assert run(taken) == _EXCLUSIVE_CREATE_EXISTS_CODE + assert taken.read_bytes() == b"existing\n" + + assert run(dangling) == _EXCLUSIVE_CREATE_EXISTS_CODE + assert not (tmp_path / "missing.txt").exists() + + # The caller creates the parent, so the script only has to claim the name. + fresh = tmp_path / "nested" / "fresh.txt" + fresh.parent.mkdir() + assert run(fresh) == 0 + assert fresh.read_bytes() == b"payload" + + existing_directory = tmp_path / "adir" + existing_directory.mkdir() + assert run(existing_directory) == _EXCLUSIVE_CREATE_EXISTS_CODE + assert list(existing_directory.iterdir()) == [] + + +@pytest.mark.asyncio +async def test_apply_patch_create_through_the_session_rejects_a_dangling_symlink( + tmp_path: Path, +) -> None: + """Drive the real caller path. + + WorkspaceEditor normalizes the destination before dispatching, and this backend + resolves leaf symlinks, so a create aimed at a dangling link used to land on the + link's absent target and report success. + """ + session = _exclusive_write_session(tmp_path) + (tmp_path / "link.txt").symlink_to(tmp_path / "missing.txt") + + with pytest.raises(ApplyPatchDiffError): + await session.apply_patch( + ApplyPatchOperation(type="create_file", path="link.txt", diff="+clobbered\n") + ) + + assert not (tmp_path / "missing.txt").exists() + assert (tmp_path / "link.txt").is_symlink() + + +@pytest.mark.asyncio +async def test_apply_patch_create_through_the_session_rejects_a_directory( + tmp_path: Path, +) -> None: + session = _exclusive_write_session(tmp_path) + (tmp_path / "adir").mkdir() + + with pytest.raises(ApplyPatchDiffError): + await session.apply_patch( + ApplyPatchOperation(type="create_file", path="adir", diff="+clobbered\n") + ) + + assert list((tmp_path / "adir").iterdir()) == [] + + +@pytest.mark.asyncio +async def test_apply_patch_create_through_the_session_keeps_existing_content( + tmp_path: Path, +) -> None: + session = _exclusive_write_session(tmp_path) + (tmp_path / "notes.txt").write_bytes(b"important\n") + + with pytest.raises(ApplyPatchDiffError): + await session.apply_patch( + ApplyPatchOperation(type="create_file", path="notes.txt", diff="+clobbered\n") + ) + + assert (tmp_path / "notes.txt").read_bytes() == b"important\n" + + +@pytest.mark.asyncio +async def test_apply_patch_create_through_the_session_writes_a_new_nested_file( + tmp_path: Path, +) -> None: + session = _exclusive_write_session(tmp_path) + + await session.apply_patch( + ApplyPatchOperation(type="create_file", path="nested/dir/new.txt", diff="+hello\n") + ) + + assert (tmp_path / "nested" / "dir" / "new.txt").read_text() == "hello" + assert not any(p.name.startswith(".") for p in (tmp_path / "nested" / "dir").iterdir()) + + +@pytest.mark.asyncio +async def test_apply_patch_create_through_the_session_reports_a_file_parent_as_a_write_error( + tmp_path: Path, +) -> None: + """A parent that is a regular file is not a collision on the requested name. + + Reporting it as one would tell the model to use update_file for a target that does + not exist and cannot be updated. + """ + session = _exclusive_write_session(tmp_path) + (tmp_path / "parent").write_bytes(b"i am a file\n") + + with pytest.raises(WorkspaceArchiveWriteError): + await session.apply_patch( + ApplyPatchOperation(type="create_file", path="parent/child.txt", diff="+hi\n") + ) + + +@pytest.mark.asyncio +async def test_apply_patch_create_accepts_a_destination_at_the_component_limit( + tmp_path: Path, +) -> None: + """Staging must not push a valid destination name past the filesystem's limit. + + Deriving the staging basename from the destination made it longer than the + destination itself, so a name the ordinary write path accepts failed to create. + """ + session = _exclusive_write_session(tmp_path) + long_name = "a" * 250 + ".txt" + # Confirm the platform really does accept this name, so the test fails for the + # right reason rather than because the limit is lower here. + probe = tmp_path / long_name + probe.write_text("probe") + probe.unlink() + + await session.apply_patch( + ApplyPatchOperation(type="create_file", path=long_name, diff="+hello\n") + ) + + assert (tmp_path / long_name).read_text() == "hello" + + +@pytest.mark.skipif(os.geteuid() == 0, reason="root bypasses directory write permissions") +@pytest.mark.asyncio +async def test_apply_patch_create_reports_collision_inside_a_read_only_parent( + tmp_path: Path, +) -> None: + """A visible collision must classify as a collision, not as a permission failure. + + Staging before classifying meant a target inside an executable but non-writable + parent failed on the staging write, so the caller was told the write failed instead + of being told to use update_file. + """ + session = _exclusive_write_session(tmp_path) + parent = tmp_path / "locked" + parent.mkdir() + target = parent / "notes.txt" + target.write_bytes(b"important\n") + parent.chmod(0o555) + try: + with pytest.raises(ApplyPatchDiffError): + await session.apply_patch( + ApplyPatchOperation( + type="create_file", path="locked/notes.txt", diff="+clobbered\n" + ) + ) + assert target.read_bytes() == b"important\n" + finally: + parent.chmod(0o755)