from __future__ import annotations

import hashlib
import io
import os
from collections.abc import Awaitable, Callable, Sequence
from pathlib import Path, PureWindowsPath

import pytest

import agents.sandbox.entries.artifacts as artifacts_module
from agents.sandbox import SandboxConcurrencyLimits, SandboxPathGrant
from agents.sandbox.entries import (
    Dir,
    File,
    GitRepo,
    LocalDir,
    LocalFile,
    resolve_workspace_path,
)
from agents.sandbox.errors import (
    ExecNonZeroError,
    GitCloneError,
    GitCopyError,
    GitSubpathError,
    InvalidManifestPathError,
    LocalDirReadError,
    LocalFileReadError,
)
from agents.sandbox.manifest import Manifest
from agents.sandbox.materialization import MaterializedFile
from agents.sandbox.session.base_sandbox_session import BaseSandboxSession
from agents.sandbox.snapshot import NoopSnapshot
from agents.sandbox.types import ExecResult, User
from tests.utils.factories import TestSessionState


class _RecordingSession(BaseSandboxSession):
    def __init__(self, manifest: Manifest | None = None) -> None:
        self.state = TestSessionState(
            manifest=manifest or Manifest(),
            snapshot=NoopSnapshot(id="noop"),
        )
        self.exec_calls: list[tuple[str, ...]] = []
        self.writes: dict[Path, bytes] = {}

    async def _exec_internal(
        self,
        *command: str | Path,
        timeout: float | None = None,
    ) -> ExecResult:
        _ = timeout
        cmd = tuple(str(part) for part in command)
        self.exec_calls.append(cmd)
        return ExecResult(stdout=b"", stderr=b"", exit_code=0)

    async def read(self, path: Path, *, user: object = None) -> io.IOBase:
        _ = user
        return io.BytesIO(self.writes[path])

    async def write(self, path: Path, data: io.IOBase, *, user: object = None) -> None:
        _ = user
        self.writes[path] = data.read()

    async def running(self) -> bool:
        return True

    async def persist_workspace(self) -> io.IOBase:
        return io.BytesIO()

    async def hydrate_workspace(self, data: io.IOBase) -> None:
        _ = data

    async def shutdown(self) -> None:
        return


class _GitRefSession(_RecordingSession):
    async def _exec_internal(
        self,
        *command: str | Path,
        timeout: float | None = None,
    ) -> ExecResult:
        _ = timeout
        cmd = tuple(str(part) for part in command)
        self.exec_calls.append(cmd)
        if cmd == ("command -v git >/dev/null 2>&1",):
            return ExecResult(stdout=b"/usr/bin/git\n", stderr=b"", exit_code=0)
        if cmd[:2] == ("git", "clone"):
            return ExecResult(stdout=b"", stderr=b"unexpected clone path", exit_code=1)
        return ExecResult(stdout=b"", stderr=b"", exit_code=0)


class _GitFailureSession(_RecordingSession):
    def __init__(self, *, fail_on: str) -> None:
        super().__init__()
        self.fail_on = fail_on

    async def _exec_internal(
        self,
        *command: str | Path,
        timeout: float | None = None,
    ) -> ExecResult:
        _ = timeout
        cmd = tuple(str(part) for part in command)
        self.exec_calls.append(cmd)
        if cmd == ("command -v git >/dev/null 2>&1",):
            return ExecResult(stdout=b"/usr/bin/git\n", stderr=b"", exit_code=0)
        if self.fail_on == "clone" and cmd[:2] == ("git", "clone"):
            return ExecResult(stdout=b"", stderr=b"clone failed", exit_code=1)
        if self.fail_on == "copy" and cmd[:1] == ("cp",):
            return ExecResult(stdout=b"", stderr=b"copy failed", exit_code=1)
        return ExecResult(stdout=b"", stderr=b"", exit_code=0)


class _MetadataFailureSession(_RecordingSession):
    def __init__(
        self,
        manifest: Manifest | None = None,
        *,
        fail_commands: set[str],
    ) -> None:
        super().__init__(manifest)
        self.fail_commands = fail_commands

    async def _exec_internal(
        self,
        *command: str | Path,
        timeout: float | None = None,
    ) -> ExecResult:
        _ = timeout
        cmd = tuple(str(part) for part in command)
        self.exec_calls.append(cmd)
        if cmd and cmd[0] in self.fail_commands:
            return ExecResult(stdout=b"", stderr=b"metadata failed", exit_code=1)
        return ExecResult(stdout=b"", stderr=b"", exit_code=0)


def test_resolve_workspace_path_rejects_windows_drive_absolute_path() -> None:
    with pytest.raises(InvalidManifestPathError) as exc_info:
        resolve_workspace_path(
            Path("/workspace"),
            PureWindowsPath("C:/tmp/secret.txt"),
            allow_absolute_within_root=True,
        )

    assert str(exc_info.value) == "manifest path must be relative: C:/tmp/secret.txt"
    assert exc_info.value.context == {"rel": "C:/tmp/secret.txt", "reason": "absolute"}


def test_resolve_workspace_path_rejects_absolute_escape_after_normalization() -> None:
    with pytest.raises(InvalidManifestPathError) as exc_info:
        resolve_workspace_path(
            Path("/workspace"),
            "/workspace/../etc/passwd",
            allow_absolute_within_root=True,
        )

    assert str(exc_info.value) == "manifest path must be relative: /etc/passwd"
    assert exc_info.value.context == {"rel": "/etc/passwd", "reason": "absolute"}


def test_resolve_workspace_path_rejects_absolute_symlink_escape_for_host_root(
    tmp_path: Path,
) -> None:
    root = tmp_path / "workspace"
    outside = tmp_path / "outside"
    root.mkdir()
    outside.mkdir()
    link = root / "link"
    try:
        os.symlink(outside, link, target_is_directory=True)
    except (NotImplementedError, OSError) as exc:
        pytest.skip(f"symlink unavailable: {exc}")

    escaped = link / "secret.txt"

    with pytest.raises(InvalidManifestPathError) as exc_info:
        resolve_workspace_path(
            root,
            escaped,
            allow_absolute_within_root=True,
        )

    assert str(exc_info.value) == f"manifest path must be relative: {escaped.as_posix()}"
    assert exc_info.value.context == {"rel": escaped.as_posix(), "reason": "absolute"}


def _symlink_or_skip(path: Path, target: Path, *, target_is_directory: bool = False) -> None:
    try:
        path.symlink_to(target, target_is_directory=target_is_directory)
    except OSError as e:
        if os.name == "nt" and getattr(e, "winerror", None) == 1314:
            pytest.skip("symlink creation requires elevated privileges on Windows")
        raise


@pytest.mark.asyncio
async def test_base_sandbox_session_uses_current_working_directory_for_local_file_sources(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    source = tmp_path / "source.txt"
    source.write_text("hello", encoding="utf-8")
    monkeypatch.chdir(tmp_path)
    session = _RecordingSession(
        Manifest(
            entries={"copied.txt": LocalFile(src=Path("source.txt"))},
        ),
    )

    result = await session.apply_manifest()

    assert result.files[0].path == Path("/workspace/copied.txt")
    assert result.files[0].sha256 == hashlib.sha256(b"hello").hexdigest()
    assert session.writes[Path("/workspace/copied.txt")] == b"hello"


@pytest.mark.asyncio
async def test_local_file_rejects_absolute_source_outside_base_dir(tmp_path: Path) -> None:
    base = tmp_path / "base"
    outside = tmp_path / "outside"
    base.mkdir()
    outside.mkdir()
    (outside / "secret.txt").write_text("secret", encoding="utf-8")
    session = _RecordingSession()

    with pytest.raises(LocalFileReadError) as excinfo:
        await LocalFile(src=outside / "secret.txt").apply(
            session,
            Path("/workspace/copied.txt"),
            base,
        )

    assert excinfo.value.context["reason"] == "outside_base_dir"
    assert excinfo.value.context["base_dir"] == str(base)
    assert session.writes == {}


@pytest.mark.asyncio
async def test_local_file_rejects_relative_source_outside_base_dir(tmp_path: Path) -> None:
    base = tmp_path / "base"
    outside = tmp_path / "outside"
    base.mkdir()
    outside.mkdir()
    (outside / "secret.txt").write_text("secret", encoding="utf-8")
    session = _RecordingSession()

    with pytest.raises(LocalFileReadError) as excinfo:
        await LocalFile(src=Path("../outside/secret.txt")).apply(
            session,
            Path("/workspace/copied.txt"),
            base,
        )

    assert excinfo.value.context["reason"] == "outside_base_dir"
    assert excinfo.value.context["base_dir"] == str(base)
    assert session.writes == {}


@pytest.mark.asyncio
async def test_local_file_allows_extra_path_granted_source_outside_base_dir(
    tmp_path: Path,
) -> None:
    base = tmp_path / "base"
    outside = tmp_path / "outside"
    base.mkdir()
    outside.mkdir()
    (outside / "secret.txt").write_text("secret", encoding="utf-8")
    session = _RecordingSession(
        Manifest(extra_path_grants=(SandboxPathGrant(path=str(outside)),)),
    )

    result = await LocalFile(src=outside / "secret.txt").apply(
        session,
        Path("/workspace/copied.txt"),
        base,
    )

    assert result[0].path == Path("/workspace/copied.txt")
    assert session.writes[Path("/workspace/copied.txt")] == b"secret"


@pytest.mark.asyncio
async def test_local_file_rejects_source_outside_extra_path_grants(tmp_path: Path) -> None:
    base = tmp_path / "base"
    outside = tmp_path / "outside"
    other = tmp_path / "other"
    base.mkdir()
    outside.mkdir()
    other.mkdir()
    (outside / "secret.txt").write_text("secret", encoding="utf-8")
    session = _RecordingSession(
        Manifest(extra_path_grants=(SandboxPathGrant(path=str(other)),)),
    )

    with pytest.raises(LocalFileReadError) as excinfo:
        await LocalFile(src=outside / "secret.txt").apply(
            session,
            Path("/workspace/copied.txt"),
            base,
        )

    assert excinfo.value.context["reason"] == "outside_base_dir"
    assert excinfo.value.context["extra_path_grants"] == [str(other)]
    assert session.writes == {}


@pytest.mark.asyncio
async def test_serialized_manifest_extra_path_grant_allows_local_file_source(
    tmp_path: Path,
) -> None:
    base = tmp_path / "base"
    outside = tmp_path / "outside"
    base.mkdir()
    outside.mkdir()
    (outside / "secret.txt").write_text("secret", encoding="utf-8")
    manifest = Manifest.model_validate(
        {
            "extra_path_grants": [{"path": str(outside)}],
            "entries": {
                "copied.txt": {
                    "type": "local_file",
                    "src": str(outside / "secret.txt"),
                }
            },
        }
    )
    session = _RecordingSession(manifest)

    result = await session._apply_entry_batch(
        [(Path("/workspace/copied.txt"), manifest.entries["copied.txt"])],
        base_dir=base,
    )

    assert result[0].path == Path("/workspace/copied.txt")
    assert session.writes[Path("/workspace/copied.txt")] == b"secret"


@pytest.mark.asyncio
async def test_local_file_allows_absolute_source_inside_base_dir(tmp_path: Path) -> None:
    base = tmp_path / "base"
    source_dir = base / "source"
    source_dir.mkdir(parents=True)
    (source_dir / "safe.txt").write_text("safe", encoding="utf-8")
    session = _RecordingSession()

    result = await LocalFile(src=source_dir / "safe.txt").apply(
        session,
        Path("/workspace/copied.txt"),
        base,
    )

    assert result[0].path == Path("/workspace/copied.txt")
    assert session.writes[Path("/workspace/copied.txt")] == b"safe"


@pytest.mark.asyncio
async def test_local_file_rejects_symlinked_source_ancestors(tmp_path: Path) -> None:
    target_dir = tmp_path / "secret-dir"
    target_dir.mkdir()
    nested_dir = target_dir / "sub"
    nested_dir.mkdir()
    (nested_dir / "secret.txt").write_text("secret", encoding="utf-8")
    _symlink_or_skip(tmp_path / "link", target_dir, target_is_directory=True)
    session = _RecordingSession()

    with pytest.raises(LocalFileReadError) as excinfo:
        await LocalFile(src=Path("link/sub/secret.txt")).apply(
            session,
            Path("/workspace/copied.txt"),
            tmp_path,
        )

    assert excinfo.value.context["reason"] == "symlink_not_supported"
    assert excinfo.value.context["child"] == "link"
    assert session.writes == {}


@pytest.mark.asyncio
async def test_local_file_rejects_symlinked_source_leaf(tmp_path: Path) -> None:
    secret = tmp_path / "secret.txt"
    secret.write_text("secret", encoding="utf-8")
    _symlink_or_skip(tmp_path / "link.txt", secret)
    session = _RecordingSession()

    with pytest.raises(LocalFileReadError) as excinfo:
        await LocalFile(src=Path("link.txt")).apply(
            session,
            Path("/workspace/copied.txt"),
            tmp_path,
        )

    assert excinfo.value.context["reason"] == "symlink_not_supported"
    assert excinfo.value.context["child"] == "link.txt"
    assert session.writes == {}


@pytest.mark.asyncio
async def test_local_file_rejects_symlinked_source_before_checksum(tmp_path: Path) -> None:
    target_dir = tmp_path / "secret-dir"
    target_dir.mkdir()
    _symlink_or_skip(tmp_path / "link.txt", target_dir, target_is_directory=True)
    session = _RecordingSession()

    with pytest.raises(LocalFileReadError) as excinfo:
        await LocalFile(src=Path("link.txt")).apply(
            session,
            Path("/workspace/copied.txt"),
            tmp_path,
        )

    assert excinfo.value.context["reason"] == "symlink_not_supported"
    assert excinfo.value.context["child"] == "link.txt"
    assert session.writes == {}


@pytest.mark.asyncio
async def test_local_dir_copy_falls_back_when_safe_dir_fd_open_unavailable(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    src_root = tmp_path / "src"
    src_root.mkdir()
    src_file = src_root / "safe.txt"
    src_file.write_text("safe", encoding="utf-8")
    session = _RecordingSession()
    local_dir = LocalDir(src=Path("src"))

    monkeypatch.setattr("agents.sandbox.entries.artifacts._OPEN_SUPPORTS_DIR_FD", False)
    monkeypatch.setattr("agents.sandbox.entries.artifacts._HAS_O_DIRECTORY", False)

    result = await local_dir._copy_local_dir_file(
        base_dir=tmp_path,
        session=session,
        src_root=src_root,
        src=src_file,
        dest_root=Path("/workspace/copied"),
    )

    assert result.path == Path("/workspace/copied/safe.txt")
    assert session.writes[Path("/workspace/copied/safe.txt")] == b"safe"


@pytest.mark.asyncio
async def test_local_dir_copy_revalidates_swapped_paths_during_open(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    if not artifacts_module._OPEN_SUPPORTS_DIR_FD or not artifacts_module._HAS_O_DIRECTORY:
        pytest.skip("safe dir_fd open pinning is unavailable on this platform")

    src_root = tmp_path / "src"
    src_root.mkdir()
    src_file = src_root / "safe.txt"
    src_file.write_text("safe", encoding="utf-8")
    secret = tmp_path / "secret.txt"
    secret.write_text("secret", encoding="utf-8")
    session = _RecordingSession()
    local_dir = LocalDir(src=Path("src"))
    original_open = os.open
    swapped = False

    def swap_then_open(
        path: str | Path,
        flags: int,
        mode: int = 0o777,
        *,
        dir_fd: int | None = None,
    ) -> int:
        nonlocal swapped
        if (path == "safe.txt" or Path(path) == src_file) and not swapped:
            src_file.unlink()
            _symlink_or_skip(src_file, secret)
            swapped = True
        if dir_fd is None:
            return original_open(path, flags, mode)
        return original_open(path, flags, mode, dir_fd=dir_fd)

    monkeypatch.setattr("agents.sandbox.entries.artifacts.os.open", swap_then_open)

    with pytest.raises(LocalDirReadError) as excinfo:
        await local_dir._copy_local_dir_file(
            base_dir=tmp_path,
            session=session,
            src_root=src_root,
            src=src_file,
            dest_root=Path("/workspace/copied"),
        )

    assert excinfo.value.context["reason"] in {
        "symlink_not_supported",
        "path_changed_during_copy",
    }
    assert excinfo.value.context["child"] == "safe.txt"
    assert session.writes == {}


@pytest.mark.asyncio
async def test_local_dir_copy_pins_parent_directories_during_open(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    if not artifacts_module._OPEN_SUPPORTS_DIR_FD or not artifacts_module._HAS_O_DIRECTORY:
        pytest.skip("safe dir_fd open pinning is unavailable on this platform")

    src_root = tmp_path / "src"
    src_root.mkdir()
    nested_dir = src_root / "nested"
    nested_dir.mkdir()
    src_file = nested_dir / "safe.txt"
    src_file.write_text("safe", encoding="utf-8")
    secret_dir = tmp_path / "secret-dir"
    secret_dir.mkdir()
    (secret_dir / "safe.txt").write_text("secret", encoding="utf-8")
    session = _RecordingSession()
    local_dir = LocalDir(src=Path("src"))
    original_open = os.open
    swapped = False

    def swap_parent_then_open(
        path: str | Path,
        flags: int,
        mode: int = 0o777,
        *,
        dir_fd: int | None = None,
    ) -> int:
        nonlocal swapped
        if path == "safe.txt" and not swapped:
            (src_root / "nested").rename(src_root / "nested-original")
            _symlink_or_skip(src_root / "nested", secret_dir, target_is_directory=True)
            swapped = True
        if dir_fd is None:
            return original_open(path, flags, mode)
        return original_open(path, flags, mode, dir_fd=dir_fd)

    monkeypatch.setattr("agents.sandbox.entries.artifacts.os.open", swap_parent_then_open)

    result = await local_dir._copy_local_dir_file(
        base_dir=tmp_path,
        session=session,
        src_root=src_root,
        src=src_file,
        dest_root=Path("/workspace/copied"),
    )

    assert result.path == Path("/workspace/copied/nested/safe.txt")
    assert session.writes[Path("/workspace/copied/nested/safe.txt")] == b"safe"


@pytest.mark.asyncio
async def test_local_dir_copy_fallback_rejects_swapped_parent_directory(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    src_root = tmp_path / "src"
    src_root.mkdir()
    nested_dir = src_root / "nested"
    nested_dir.mkdir()
    src_file = nested_dir / "safe.txt"
    src_file.write_text("safe", encoding="utf-8")
    secret_dir = tmp_path / "secret-dir"
    secret_dir.mkdir()
    (secret_dir / "safe.txt").write_text("secret", encoding="utf-8")
    session = _RecordingSession()
    local_dir = LocalDir(src=Path("src"))
    original_open = os.open
    swapped = False

    monkeypatch.setattr("agents.sandbox.entries.artifacts._OPEN_SUPPORTS_DIR_FD", False)
    monkeypatch.setattr("agents.sandbox.entries.artifacts._HAS_O_DIRECTORY", False)

    def swap_parent_then_open(
        path: str | Path,
        flags: int,
        mode: int = 0o777,
        *,
        dir_fd: int | None = None,
    ) -> int:
        nonlocal swapped
        if Path(path) == src_file and not swapped:
            nested_dir.rename(src_root / "nested-original")
            _symlink_or_skip(src_root / "nested", secret_dir, target_is_directory=True)
            swapped = True
        if dir_fd is None:
            return original_open(path, flags, mode)
        return original_open(path, flags, mode, dir_fd=dir_fd)

    monkeypatch.setattr("agents.sandbox.entries.artifacts.os.open", swap_parent_then_open)

    with pytest.raises(LocalDirReadError) as excinfo:
        await local_dir._copy_local_dir_file(
            base_dir=tmp_path,
            session=session,
            src_root=src_root,
            src=src_file,
            dest_root=Path("/workspace/copied"),
        )

    assert excinfo.value.context["reason"] == "symlink_not_supported"
    assert excinfo.value.context["child"] == "src/nested"
    assert session.writes == {}


@pytest.mark.asyncio
async def test_local_dir_apply_rejects_source_root_swapped_to_symlink_after_validation(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    if not artifacts_module._OPEN_SUPPORTS_DIR_FD or not artifacts_module._HAS_O_DIRECTORY:
        pytest.skip("safe dir_fd open pinning is unavailable on this platform")

    src_root = tmp_path / "src"
    src_root.mkdir()
    (src_root / "safe.txt").write_text("safe", encoding="utf-8")
    secret_dir = tmp_path / "secret-dir"
    secret_dir.mkdir()
    (secret_dir / "secret.txt").write_text("secret", encoding="utf-8")
    session = _RecordingSession()
    local_dir = LocalDir(src=Path("src"))
    original_open = os.open
    swapped = False

    def swap_root_then_open(
        path: str | Path,
        flags: int,
        mode: int = 0o777,
        *,
        dir_fd: int | None = None,
    ) -> int:
        nonlocal swapped
        if (path == "src" or Path(path) in {src_root, src_root / "safe.txt"}) and not swapped:
            src_root.rename(tmp_path / "src-original")
            (tmp_path / "src").symlink_to(secret_dir, target_is_directory=True)
            swapped = True
        if dir_fd is None:
            return original_open(path, flags, mode)
        return original_open(path, flags, mode, dir_fd=dir_fd)

    monkeypatch.setattr("agents.sandbox.entries.artifacts.os.open", swap_root_then_open)

    with pytest.raises(LocalDirReadError) as excinfo:
        await local_dir.apply(session, Path("/workspace/copied"), tmp_path)

    assert excinfo.value.context["reason"] == "symlink_not_supported"
    assert excinfo.value.context["child"] == "src"
    assert session.writes == {}


@pytest.mark.asyncio
async def test_local_dir_apply_fallback_rejects_source_root_swapped_to_symlink_after_validation(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    src_root = tmp_path / "src"
    src_root.mkdir()
    (src_root / "safe.txt").write_text("safe", encoding="utf-8")
    secret_dir = tmp_path / "secret-dir"
    secret_dir.mkdir()
    session = _RecordingSession()
    local_dir = LocalDir(src=Path("src"))
    original_open = os.open
    swapped = False

    monkeypatch.setattr("agents.sandbox.entries.artifacts._OPEN_SUPPORTS_DIR_FD", False)
    monkeypatch.setattr("agents.sandbox.entries.artifacts._HAS_O_DIRECTORY", False)

    def swap_root_then_open(
        path: str | Path,
        flags: int,
        mode: int = 0o777,
        *,
        dir_fd: int | None = None,
    ) -> int:
        nonlocal swapped
        if Path(path) == src_root / "safe.txt" and not swapped:
            src_root.rename(tmp_path / "src-original")
            _symlink_or_skip(tmp_path / "src", secret_dir, target_is_directory=True)
            swapped = True
        if dir_fd is None:
            return original_open(path, flags, mode)
        return original_open(path, flags, mode, dir_fd=dir_fd)

    monkeypatch.setattr("agents.sandbox.entries.artifacts.os.open", swap_root_then_open)

    with pytest.raises(LocalDirReadError) as excinfo:
        await local_dir.apply(session, Path("/workspace/copied"), tmp_path)

    assert excinfo.value.context["reason"] == "symlink_not_supported"
    assert excinfo.value.context["child"] == "src"
    assert session.writes == {}


@pytest.mark.asyncio
async def test_local_dir_apply_uses_configured_file_copy_fanout(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    src_root = tmp_path / "src"
    src_root.mkdir()
    (src_root / "a.txt").write_text("a", encoding="utf-8")
    (src_root / "b.txt").write_text("b", encoding="utf-8")
    session = _RecordingSession()
    session._set_concurrency_limits(
        SandboxConcurrencyLimits(
            manifest_entries=4,
            local_dir_files=2,
        )
    )
    observed_limits: list[int | None] = []

    async def gather_with_limit_recording(
        task_factories: Sequence[Callable[[], Awaitable[MaterializedFile]]],
        *,
        max_concurrency: int | None = None,
    ) -> list[MaterializedFile]:
        observed_limits.append(max_concurrency)
        return [await factory() for factory in task_factories]

    monkeypatch.setattr(
        artifacts_module,
        "gather_in_order",
        gather_with_limit_recording,
    )

    result = await LocalDir(src=Path("src")).apply(
        session,
        Path("/workspace/copied"),
        tmp_path,
    )

    assert observed_limits == [2]
    assert sorted(file.path.as_posix() for file in result) == [
        "/workspace/copied/a.txt",
        "/workspace/copied/b.txt",
    ]
    assert session.writes == {
        Path("/workspace/copied/a.txt"): b"a",
        Path("/workspace/copied/b.txt"): b"b",
    }


@pytest.mark.asyncio
async def test_local_dir_rejects_absolute_source_outside_base_dir(tmp_path: Path) -> None:
    base = tmp_path / "base"
    outside = tmp_path / "outside"
    base.mkdir()
    outside.mkdir()
    (outside / "secret.txt").write_text("secret", encoding="utf-8")
    session = _RecordingSession()

    with pytest.raises(LocalDirReadError) as excinfo:
        await LocalDir(src=outside).apply(session, Path("/workspace/copied"), base)

    assert excinfo.value.context["reason"] == "outside_base_dir"
    assert excinfo.value.context["base_dir"] == str(base)
    assert session.writes == {}


@pytest.mark.asyncio
async def test_local_dir_rejects_relative_source_outside_base_dir(tmp_path: Path) -> None:
    base = tmp_path / "base"
    outside = tmp_path / "outside"
    base.mkdir()
    outside.mkdir()
    (outside / "secret.txt").write_text("secret", encoding="utf-8")
    session = _RecordingSession()

    with pytest.raises(LocalDirReadError) as excinfo:
        await LocalDir(src=Path("../outside")).apply(session, Path("/workspace/copied"), base)

    assert excinfo.value.context["reason"] == "outside_base_dir"
    assert excinfo.value.context["base_dir"] == str(base)
    assert session.writes == {}


@pytest.mark.asyncio
async def test_local_dir_allows_extra_path_granted_source_outside_base_dir(
    tmp_path: Path,
) -> None:
    base = tmp_path / "base"
    outside = tmp_path / "outside"
    base.mkdir()
    outside.mkdir()
    (outside / "secret.txt").write_text("secret", encoding="utf-8")
    session = _RecordingSession(
        Manifest(extra_path_grants=(SandboxPathGrant(path=str(outside), read_only=True),)),
    )

    result = await LocalDir(src=outside).apply(
        session,
        Path("/workspace/copied"),
        base,
    )

    assert result[0].path == Path("/workspace/copied/secret.txt")
    assert session.writes[Path("/workspace/copied/secret.txt")] == b"secret"


@pytest.mark.asyncio
async def test_local_dir_allows_absolute_source_inside_base_dir(tmp_path: Path) -> None:
    base = tmp_path / "base"
    source = base / "source"
    source.mkdir(parents=True)
    (source / "safe.txt").write_text("safe", encoding="utf-8")
    session = _RecordingSession()

    result = await LocalDir(src=source).apply(session, Path("/workspace/copied"), base)

    assert result[0].path == Path("/workspace/copied/safe.txt")
    assert session.writes[Path("/workspace/copied/safe.txt")] == b"safe"


@pytest.mark.asyncio
async def test_local_dir_rejects_symlinked_source_ancestors(tmp_path: Path) -> None:
    target_dir = tmp_path / "secret-dir"
    target_dir.mkdir()
    nested_dir = target_dir / "sub"
    nested_dir.mkdir()
    (nested_dir / "secret.txt").write_text("secret", encoding="utf-8")
    _symlink_or_skip(tmp_path / "link", target_dir, target_is_directory=True)
    session = _RecordingSession()

    with pytest.raises(LocalDirReadError) as excinfo:
        await LocalDir(src=Path("link/sub")).apply(session, Path("/workspace/copied"), tmp_path)

    assert excinfo.value.context["reason"] == "symlink_not_supported"
    assert excinfo.value.context["child"] == "link"
    assert session.writes == {}


@pytest.mark.asyncio
async def test_local_dir_rejects_symlinked_source_root(tmp_path: Path) -> None:
    target_dir = tmp_path / "secret-dir"
    target_dir.mkdir()
    (target_dir / "secret.txt").write_text("secret", encoding="utf-8")
    _symlink_or_skip(tmp_path / "src", target_dir, target_is_directory=True)
    session = _RecordingSession()

    with pytest.raises(LocalDirReadError) as excinfo:
        await LocalDir(src=Path("src")).apply(session, Path("/workspace/copied"), tmp_path)

    assert excinfo.value.context["reason"] == "symlink_not_supported"
    assert excinfo.value.context["child"] == "src"
    assert session.writes == {}


@pytest.mark.asyncio
async def test_local_dir_rejects_symlinked_files(tmp_path: Path) -> None:
    src_root = tmp_path / "src"
    src_root.mkdir()
    (src_root / "safe.txt").write_text("safe", encoding="utf-8")
    secret = tmp_path / "secret.txt"
    secret.write_text("secret", encoding="utf-8")
    _symlink_or_skip(src_root / "link.txt", secret)
    session = _RecordingSession()

    with pytest.raises(LocalDirReadError) as excinfo:
        await LocalDir(src=Path("src")).apply(session, Path("/workspace/copied"), tmp_path)

    assert excinfo.value.context["reason"] == "symlink_not_supported"
    assert excinfo.value.context["child"] == "link.txt"
    assert session.writes == {}


@pytest.mark.asyncio
async def test_local_dir_rejects_symlinked_directories(tmp_path: Path) -> None:
    src_root = tmp_path / "src"
    src_root.mkdir()
    (src_root / "safe.txt").write_text("safe", encoding="utf-8")
    target_dir = tmp_path / "secret-dir"
    target_dir.mkdir()
    (target_dir / "secret.txt").write_text("secret", encoding="utf-8")
    _symlink_or_skip(src_root / "linked-dir", target_dir, target_is_directory=True)
    session = _RecordingSession()

    with pytest.raises(LocalDirReadError) as excinfo:
        await LocalDir(src=Path("src")).apply(session, Path("/workspace/copied"), tmp_path)

    assert excinfo.value.context["reason"] == "symlink_not_supported"
    assert excinfo.value.context["child"] == "linked-dir"
    assert session.writes == {}


@pytest.mark.asyncio
async def test_git_repo_uses_fetch_checkout_path_for_commit_refs() -> None:
    session = _GitRefSession()
    repo = GitRepo(repo="openai/example", ref="deadbeef")

    await repo.apply(session, Path("/workspace/repo"), Path("/ignored"))

    assert not any(call[:2] == ("git", "clone") for call in session.exec_calls)
    assert any(call[:2] == ("git", "init") for call in session.exec_calls)
    assert any(
        len(call) >= 7
        and call[:2] == ("git", "-C")
        and call[3:6] == ("remote", "add", "origin")
        and call[6] == "https://github.com/openai/example.git"
        for call in session.exec_calls
    )
    assert any(
        len(call) >= 9
        and call[:2] == ("git", "-C")
        and call[3:7] == ("fetch", "--depth", "1", "--no-tags")
        and call[-2:] == ("origin", "deadbeef")
        for call in session.exec_calls
    )
    assert any(
        len(call) >= 6
        and call[:2] == ("git", "-C")
        and call[3:5] == ("checkout", "--detach")
        and call[-1] == "FETCH_HEAD"
        for call in session.exec_calls
    )


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("subpath", "reason"),
    [
        ("   ", "empty"),
        ("/docs", "absolute"),
        ("../outside", "parent_traversal"),
        ("docs/../../outside", "parent_traversal"),
        ("C:/repo", "windows_path"),
        ("docs\\outside", "windows_path"),
    ],
)
async def test_git_repo_rejects_invalid_subpath_before_copy(
    subpath: str,
    reason: str,
) -> None:
    session = _GitFailureSession(fail_on="clone")
    repo = GitRepo(repo="openai/example", ref="main", subpath=subpath)

    with pytest.raises(GitSubpathError) as excinfo:
        await repo.apply(session, Path("/workspace/repo"), Path("/ignored"))

    assert excinfo.value.context["reason"] == reason
    assert excinfo.value.context["subpath"] == subpath
    assert session.exec_calls == []


@pytest.mark.asyncio
@pytest.mark.parametrize("subpath", ["", ".", "./", "./.", " ./ "])
async def test_git_repo_root_subpath_alias_copies_repo_root(subpath: str) -> None:
    session = _RecordingSession()
    repo = GitRepo(repo="openai/example", ref="main", subpath=subpath)

    await repo.apply(session, Path("/workspace/repo"), Path("/ignored"))

    copy_call = next(call for call in session.exec_calls if call[:1] == ("cp",))
    assert copy_call[3].startswith("/tmp/sandbox-git-")
    assert copy_call[3].endswith("/.")
    assert not copy_call[3].endswith("//.")
    assert copy_call[4].replace("\\", "/") == "/workspace/repo/"


@pytest.mark.asyncio
async def test_git_repo_allows_relative_subpath_copy() -> None:
    session = _RecordingSession()
    repo = GitRepo(repo="openai/example", ref="main", subpath="docs/reference")

    await repo.apply(session, Path("/workspace/repo"), Path("/ignored"))

    copy_call = next(call for call in session.exec_calls if call[:1] == ("cp",))
    assert copy_call[3].endswith("/docs/reference/.")
    assert copy_call[4].replace("\\", "/") == "/workspace/repo/"


def _git_temp_cleanup_calls(session: _RecordingSession) -> list[tuple[str, ...]]:
    return [call for call in session.exec_calls if call[:3] == ("rm", "-rf", "--")]


def _git_temp_cleanup_call_indices(session: _RecordingSession) -> list[int]:
    return [i for i, call in enumerate(session.exec_calls) if call[:3] == ("rm", "-rf", "--")]


@pytest.mark.asyncio
async def test_git_repo_cleans_temp_clone_after_copy_failure() -> None:
    session = _GitFailureSession(fail_on="copy")
    repo = GitRepo(repo="openai/example", ref="main")

    with pytest.raises(GitCopyError):
        await repo.apply(session, Path("/workspace/repo"), Path("/ignored"))

    cleanup_calls = _git_temp_cleanup_calls(session)
    cleanup_indices = _git_temp_cleanup_call_indices(session)
    assert len(cleanup_calls) == 2
    assert cleanup_calls[0] == cleanup_calls[1]
    assert cleanup_indices[1] > next(
        i for i, call in enumerate(session.exec_calls) if call[:1] == ("cp",)
    )


@pytest.mark.asyncio
async def test_git_repo_cleans_temp_clone_after_clone_failure() -> None:
    session = _GitFailureSession(fail_on="clone")
    repo = GitRepo(repo="openai/example", ref="main")

    with pytest.raises(GitCloneError):
        await repo.apply(session, Path("/workspace/repo"), Path("/ignored"))

    cleanup_calls = _git_temp_cleanup_calls(session)
    cleanup_indices = _git_temp_cleanup_call_indices(session)
    assert len(cleanup_calls) == 2
    assert cleanup_calls[0] == cleanup_calls[1]
    assert cleanup_indices[1] > next(
        i for i, call in enumerate(session.exec_calls) if call[:2] == ("git", "clone")
    )


@pytest.mark.asyncio
async def test_dir_metadata_strips_file_type_bits_before_chmod() -> None:
    session = _RecordingSession()
    dest = Path("/workspace/dir")

    await Dir()._apply_metadata(session, dest)

    assert ("chmod", "0755", "/workspace/dir") in session.exec_calls


@pytest.mark.asyncio
async def test_apply_manifest_raises_on_chmod_failure() -> None:
    session = _MetadataFailureSession(
        Manifest(entries={"copied.txt": File(content=b"hello")}),
        fail_commands={"chmod"},
    )

    with pytest.raises(ExecNonZeroError):
        await session.apply_manifest()


@pytest.mark.asyncio
async def test_apply_manifest_raises_on_chgrp_failure() -> None:
    session = _MetadataFailureSession(
        Manifest(
            entries={
                "copied.txt": File(
                    content=b"hello",
                    group=User(name="sandbox-user"),
                )
            }
        ),
        fail_commands={"chgrp"},
    )

    with pytest.raises(ExecNonZeroError):
        await session.apply_manifest()

    assert ("chgrp", "sandbox-user", "/workspace/copied.txt") in session.exec_calls
    assert not any(call[0] == "chmod" for call in session.exec_calls)