from __future__ import annotations

import io
import os
import tarfile
import zipfile
from pathlib import Path

import pytest

from agents.sandbox import SandboxArchiveLimits
from agents.sandbox.entries import GCSMount, InContainerMountStrategy, MountpointMountPattern
from agents.sandbox.errors import (
    InvalidCompressionSchemeError,
    InvalidManifestPathError,
    WorkspaceArchiveWriteError,
)
from agents.sandbox.files import EntryKind, FileEntry
from agents.sandbox.manifest import Manifest
from agents.sandbox.sandboxes.unix_local import (
    UnixLocalSandboxSession,
    UnixLocalSandboxSessionState,
)
from agents.sandbox.session.archive_extraction import zipfile_compatible_stream
from agents.sandbox.session.archive_ops import extract_archive
from agents.sandbox.session.base_sandbox_session import BaseSandboxSession
from agents.sandbox.snapshot import NoopSnapshot
from agents.sandbox.types import ExecResult, Permissions


def _build_session(tmp_path: Path) -> UnixLocalSandboxSession:
    state = UnixLocalSandboxSessionState(
        manifest=Manifest(root=str(tmp_path / "workspace")),
        snapshot=NoopSnapshot(id="noop"),
    )
    return UnixLocalSandboxSession.from_state(state)


class _CountingExtractSession(BaseSandboxSession):
    def __init__(self, workspace_root: Path) -> None:
        self.state = UnixLocalSandboxSessionState(
            manifest=Manifest(root=str(workspace_root)),
            snapshot=NoopSnapshot(id="noop"),
        )
        self.ls_calls: list[Path] = []

    async def _exec_internal(
        self,
        *command: str | Path,
        timeout: float | None = None,
    ) -> ExecResult:
        _ = (command, timeout)
        raise AssertionError("exec() should not be called in this test")

    async def read(self, path: Path, *, user: object = None) -> io.IOBase:
        _ = user
        return self.normalize_path(path).open("rb")

    async def write(self, path: Path, data: io.IOBase, *, user: object = None) -> None:
        _ = user
        workspace_path = self.normalize_path(path)
        workspace_path.parent.mkdir(parents=True, exist_ok=True)
        payload = data.read()
        if isinstance(payload, str):
            payload = payload.encode("utf-8")
        workspace_path.write_bytes(payload)

    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

    async def mkdir(
        self,
        path: Path | str,
        *,
        parents: bool = False,
        user: object = None,
    ) -> None:
        _ = user
        self.normalize_path(path).mkdir(parents=parents, exist_ok=True)

    async def ls(
        self,
        path: Path | str,
        *,
        user: object = None,
    ) -> list[FileEntry]:
        _ = user
        directory = self.normalize_path(path)
        self.ls_calls.append(directory)
        if not directory.exists():
            raise AssertionError(f"ls() called for missing directory: {directory}")

        entries: list[FileEntry] = []
        for child in directory.iterdir():
            if child.is_symlink():
                kind = EntryKind.SYMLINK
            elif child.is_dir():
                kind = EntryKind.DIRECTORY
            else:
                kind = EntryKind.FILE
            entries.append(
                FileEntry(
                    path=str(child),
                    permissions=Permissions(),
                    owner="root",
                    group="root",
                    size=0,
                    kind=kind,
                )
            )
        return entries


def _tar_bytes(*, members: dict[str, bytes]) -> io.BytesIO:
    buf = io.BytesIO()
    with tarfile.open(fileobj=buf, mode="w") as archive:
        for name, payload in members.items():
            info = tarfile.TarInfo(name=name)
            info.size = len(payload)
            archive.addfile(info, io.BytesIO(payload))
    buf.seek(0)
    return buf


def _zip_bytes(*, members: dict[str, bytes]) -> io.BytesIO:
    buf = io.BytesIO()
    with zipfile.ZipFile(buf, mode="w") as archive:
        for name, payload in members.items():
            archive.writestr(name, payload)
    buf.seek(0)
    return buf


async def _assert_extract_rejects_member(
    tmp_path: Path,
    archive_name: str,
    data: io.IOBase,
    *,
    expected_member: str,
    expected_reason: str,
) -> Path:
    session = _build_session(tmp_path)
    await session.start()
    try:
        workspace = Path(session.state.manifest.root)
        with pytest.raises(WorkspaceArchiveWriteError) as exc_info:
            await session.extract(archive_name, data)

        assert exc_info.value.context["member"] == expected_member
        assert exc_info.value.context["reason"] == expected_reason
        return workspace
    finally:
        await session.shutdown()


@pytest.mark.asyncio
async def test_extract_tar_writes_archive_and_unpacks_contents(tmp_path: Path) -> None:
    session = _build_session(tmp_path)
    await session.start()
    try:
        await session.extract(
            "bundle.tar",
            _tar_bytes(members={"nested/hello.txt": b"hello from tar"}),
        )
    finally:
        await session.shutdown()

    workspace = Path(session.state.manifest.root)
    assert (workspace / "bundle.tar").is_file()
    assert (workspace / "nested" / "hello.txt").read_text(encoding="utf-8") == "hello from tar"


@pytest.mark.asyncio
async def test_extract_zip_writes_archive_and_unpacks_contents(tmp_path: Path) -> None:
    session = _build_session(tmp_path)
    await session.start()
    try:
        await session.extract(
            "bundle.zip",
            _zip_bytes(members={"nested/hello.txt": b"hello from zip"}),
        )
    finally:
        await session.shutdown()

    workspace = Path(session.state.manifest.root)
    assert (workspace / "bundle.zip").is_file()
    assert (workspace / "nested" / "hello.txt").read_text(encoding="utf-8") == "hello from zip"


@pytest.mark.asyncio
async def test_extract_default_archive_limits_none_preserves_no_resource_limit_behavior(
    tmp_path: Path,
) -> None:
    session = _build_session(tmp_path)
    await session.start()
    try:
        await session.extract(
            "bundle.tar",
            _tar_bytes(members={"one.txt": b"1", "two.txt": b"2"}),
        )
    finally:
        await session.shutdown()

    workspace = Path(session.state.manifest.root)
    assert (workspace / "one.txt").read_text(encoding="utf-8") == "1"
    assert (workspace / "two.txt").read_text(encoding="utf-8") == "2"


def test_sandbox_archive_limits_defaults_enable_sdk_thresholds() -> None:
    limits = SandboxArchiveLimits()

    assert limits.max_input_bytes == 1024 * 1024 * 1024
    assert limits.max_extracted_bytes == 4 * 1024 * 1024 * 1024
    assert limits.max_members == 100_000


@pytest.mark.asyncio
async def test_extract_archive_rejects_missing_compression_scheme(tmp_path: Path) -> None:
    session = _CountingExtractSession(tmp_path / "workspace")

    with pytest.raises(InvalidCompressionSchemeError) as exc_info:
        await extract_archive(session, "bundle", io.BytesIO(b"not an archive"))

    assert exc_info.value.context["path"] == "bundle"
    assert exc_info.value.context["scheme"] is None


@pytest.mark.parametrize(
    ("kwargs", "message"),
    [
        ({"max_input_bytes": 0}, "archive_limits.max_input_bytes must be at least 1"),
        ({"max_extracted_bytes": 0}, "archive_limits.max_extracted_bytes must be at least 1"),
        ({"max_members": 0}, "archive_limits.max_members must be at least 1"),
    ],
)
def test_sandbox_archive_limits_rejects_non_positive_values(
    kwargs: dict[str, int],
    message: str,
) -> None:
    with pytest.raises(ValueError) as exc_info:
        SandboxArchiveLimits(**kwargs)

    assert str(exc_info.value) == message


@pytest.mark.asyncio
async def test_extract_rejects_archive_input_over_limit(tmp_path: Path) -> None:
    session = _build_session(tmp_path)
    await session.start()
    try:
        with pytest.raises(WorkspaceArchiveWriteError) as exc_info:
            await session.extract(
                "bundle.zip",
                _ChunkedBinaryStream([b"123", b"45"]),
                archive_limits=SandboxArchiveLimits(
                    max_input_bytes=4,
                    max_extracted_bytes=None,
                    max_members=None,
                ),
            )

        assert exc_info.value.context["reason"] == "archive input size exceeds limit"
        assert exc_info.value.context["limit"] == 4
        assert exc_info.value.context["actual"] == 5
    finally:
        await session.shutdown()

    workspace = Path(session.state.manifest.root)
    assert not (workspace / "bundle.zip").exists()


@pytest.mark.asyncio
async def test_extract_tar_rejects_extracted_bytes_over_limit(
    tmp_path: Path,
) -> None:
    session = _build_session(tmp_path)
    await session.start()
    try:
        with pytest.raises(WorkspaceArchiveWriteError) as exc_info:
            await session.extract(
                "bundle.tar",
                _tar_bytes(members={"large.txt": b"12345"}),
                archive_limits=SandboxArchiveLimits(
                    max_input_bytes=None,
                    max_extracted_bytes=4,
                    max_members=None,
                ),
            )

        assert exc_info.value.context["member"] == "large.txt"
        assert exc_info.value.context["reason"] == "archive extracted size exceeds limit"
        assert exc_info.value.context["limit"] == 4
        assert exc_info.value.context["actual"] == 5
    finally:
        await session.shutdown()

    workspace = Path(session.state.manifest.root)
    assert not (workspace / "large.txt").exists()


@pytest.mark.asyncio
async def test_extract_zip_rejects_extracted_bytes_over_limit(
    tmp_path: Path,
) -> None:
    session = _build_session(tmp_path)
    await session.start()
    try:
        with pytest.raises(WorkspaceArchiveWriteError) as exc_info:
            await session.extract(
                "bundle.zip",
                _zip_bytes(members={"large.txt": b"12345"}),
                archive_limits=SandboxArchiveLimits(
                    max_input_bytes=None,
                    max_extracted_bytes=4,
                    max_members=None,
                ),
            )

        assert exc_info.value.context["member"] == "large.txt"
        assert exc_info.value.context["reason"] == "archive extracted size exceeds limit"
        assert exc_info.value.context["limit"] == 4
        assert exc_info.value.context["actual"] == 5
    finally:
        await session.shutdown()

    workspace = Path(session.state.manifest.root)
    assert not (workspace / "large.txt").exists()


@pytest.mark.asyncio
async def test_extract_tar_rejects_member_count_over_limit(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    data = _tar_bytes(members={"one.txt": b"1", "two.txt": b"2"})

    def fail_getmembers(_self: tarfile.TarFile) -> list[tarfile.TarInfo]:
        raise AssertionError("tar extraction should not materialize all members")

    monkeypatch.setattr(tarfile.TarFile, "getmembers", fail_getmembers)

    session = _build_session(tmp_path)
    await session.start()
    try:
        with pytest.raises(WorkspaceArchiveWriteError) as exc_info:
            await session.extract(
                "bundle.tar",
                data,
                archive_limits=SandboxArchiveLimits(
                    max_input_bytes=None,
                    max_extracted_bytes=None,
                    max_members=1,
                ),
            )

        assert exc_info.value.context["member"] == "two.txt"
        assert exc_info.value.context["reason"] == "archive member count exceeds limit"
        assert exc_info.value.context["limit"] == 1
        assert exc_info.value.context["actual"] == 2
    finally:
        await session.shutdown()

    workspace = Path(session.state.manifest.root)
    assert not (workspace / "one.txt").exists()
    assert not (workspace / "two.txt").exists()


@pytest.mark.asyncio
async def test_extract_zip_rejects_member_count_over_limit(
    tmp_path: Path,
) -> None:
    session = _build_session(tmp_path)
    await session.start()
    try:
        with pytest.raises(WorkspaceArchiveWriteError) as exc_info:
            await session.extract(
                "bundle.zip",
                _zip_bytes(members={"one.txt": b"1", "two.txt": b"2"}),
                archive_limits=SandboxArchiveLimits(
                    max_input_bytes=None,
                    max_extracted_bytes=None,
                    max_members=1,
                ),
            )

        assert exc_info.value.context["member"] == "two.txt"
        assert exc_info.value.context["reason"] == "archive member count exceeds limit"
        assert exc_info.value.context["limit"] == 1
        assert exc_info.value.context["actual"] == 2
    finally:
        await session.shutdown()

    workspace = Path(session.state.manifest.root)
    assert not (workspace / "one.txt").exists()
    assert not (workspace / "two.txt").exists()


@pytest.mark.asyncio
async def test_extract_archive_limits_none_disables_only_selected_limits(
    tmp_path: Path,
) -> None:
    session = _build_session(tmp_path)
    await session.start()
    try:
        await session.extract(
            "bundle.tar",
            _tar_bytes(members={"large.txt": b"12345"}),
            archive_limits=SandboxArchiveLimits(
                max_input_bytes=None,
                max_extracted_bytes=None,
                max_members=1,
            ),
        )
    finally:
        await session.shutdown()

    workspace = Path(session.state.manifest.root)
    assert (workspace / "large.txt").read_text(encoding="utf-8") == "12345"


@pytest.mark.asyncio
async def test_extract_archive_limits_per_call_override_session_default(
    tmp_path: Path,
) -> None:
    session = _build_session(tmp_path)
    session._set_archive_limits(
        SandboxArchiveLimits(max_input_bytes=None, max_extracted_bytes=None, max_members=1)
    )
    await session.start()
    try:
        await session.extract(
            "bundle.tar",
            _tar_bytes(members={"one.txt": b"1", "two.txt": b"2"}),
            archive_limits=SandboxArchiveLimits(
                max_input_bytes=None,
                max_extracted_bytes=None,
                max_members=2,
            ),
        )
    finally:
        await session.shutdown()

    workspace = Path(session.state.manifest.root)
    assert (workspace / "one.txt").read_text(encoding="utf-8") == "1"
    assert (workspace / "two.txt").read_text(encoding="utf-8") == "2"


@pytest.mark.asyncio
async def test_extract_uses_session_default_archive_limits(
    tmp_path: Path,
) -> None:
    session = _build_session(tmp_path)
    session._set_archive_limits(
        SandboxArchiveLimits(max_input_bytes=None, max_extracted_bytes=None, max_members=1)
    )
    await session.start()
    try:
        with pytest.raises(WorkspaceArchiveWriteError) as exc_info:
            await session.extract(
                "bundle.tar",
                _tar_bytes(members={"one.txt": b"1", "two.txt": b"2"}),
            )

        assert exc_info.value.context["member"] == "two.txt"
        assert exc_info.value.context["reason"] == "archive member count exceeds limit"
        assert exc_info.value.context["limit"] == 1
        assert exc_info.value.context["actual"] == 2
    finally:
        await session.shutdown()


@pytest.mark.asyncio
async def test_extract_archive_limits_object_with_all_none_overrides_session_default(
    tmp_path: Path,
) -> None:
    session = _build_session(tmp_path)
    session._set_archive_limits(
        SandboxArchiveLimits(max_input_bytes=None, max_extracted_bytes=None, max_members=1)
    )
    await session.start()
    try:
        await session.extract(
            "bundle.tar",
            _tar_bytes(members={"one.txt": b"1", "two.txt": b"2"}),
            archive_limits=SandboxArchiveLimits(
                max_input_bytes=None,
                max_extracted_bytes=None,
                max_members=None,
            ),
        )
    finally:
        await session.shutdown()

    workspace = Path(session.state.manifest.root)
    assert (workspace / "one.txt").read_text(encoding="utf-8") == "1"
    assert (workspace / "two.txt").read_text(encoding="utf-8") == "2"


@pytest.mark.asyncio
async def test_extract_rejects_invalid_per_call_archive_limits(
    tmp_path: Path,
) -> None:
    limits = SandboxArchiveLimits(max_input_bytes=1)
    limits.max_input_bytes = 0
    session = _build_session(tmp_path)
    await session.start()
    try:
        with pytest.raises(ValueError) as exc_info:
            await session.extract(
                "bundle.tar",
                _tar_bytes(members={"one.txt": b"1"}),
                archive_limits=limits,
            )

        assert str(exc_info.value) == "archive_limits.max_input_bytes must be at least 1"
    finally:
        await session.shutdown()


class _NoSeekableZipStream(io.IOBase):
    def __init__(self, payload: bytes) -> None:
        self._buffer = io.BytesIO(payload)

    def tell(self) -> int:
        return self._buffer.tell()

    def seek(self, offset: int, whence: int = io.SEEK_SET) -> int:
        return self._buffer.seek(offset, whence)

    def read(self, size: int = -1) -> bytes:
        return self._buffer.read(size)


class _ChunkedBinaryStream(io.IOBase):
    def __init__(self, chunks: list[bytes]) -> None:
        self._chunks = list(chunks)
        self.headers = {"Content-Length": str(sum(len(chunk) for chunk in chunks))}

    def read(self, size: int = -1) -> bytes:
        if not self._chunks:
            return b""
        if size < 0:
            data = b"".join(self._chunks)
            self._chunks.clear()
            return data

        remaining = size
        out = bytearray()
        while remaining > 0 and self._chunks:
            chunk = self._chunks[0]
            if len(chunk) <= remaining:
                out.extend(self._chunks.pop(0))
                remaining -= len(chunk)
                continue
            out.extend(chunk[:remaining])
            self._chunks[0] = chunk[remaining:]
            remaining = 0
        return bytes(out)


class _SeekableFalseZipStream(io.IOBase):
    def __init__(self, payload: bytes) -> None:
        self._buffer = io.BytesIO(payload)

    def seekable(self) -> bool:
        return False

    def read(self, size: int = -1) -> bytes:
        return self._buffer.read(size)


def test_zipfile_compatible_stream_supports_streams_without_seekable() -> None:
    raw_stream = _NoSeekableZipStream(_zip_bytes(members={"file.txt": b"hello"}).getvalue())

    with zipfile_compatible_stream(raw_stream) as compatible:
        assert compatible.seekable() is True
        with zipfile.ZipFile(compatible) as archive:
            assert archive.read("file.txt") == b"hello"


def test_zipfile_compatible_stream_buffers_streams_with_seekable_false() -> None:
    raw_stream = _SeekableFalseZipStream(_zip_bytes(members={"file.txt": b"hello"}).getvalue())

    with zipfile_compatible_stream(raw_stream) as compatible:
        assert compatible.seekable() is True
        with zipfile.ZipFile(compatible) as archive:
            assert archive.read("file.txt") == b"hello"


@pytest.mark.asyncio
async def test_unix_local_write_accepts_chunked_non_seekable_binary_stream(tmp_path: Path) -> None:
    session = _build_session(tmp_path)
    await session.start()
    try:
        await session.write(
            Path("streamed.bin"),
            _ChunkedBinaryStream([b"hello ", b"from ", b"stream"]),
        )
    finally:
        await session.shutdown()

    workspace = Path(session.state.manifest.root)
    assert (workspace / "streamed.bin").read_bytes() == b"hello from stream"


@pytest.mark.asyncio
async def test_extract_tar_rejects_symlinked_parent_paths(tmp_path: Path) -> None:
    session = _build_session(tmp_path)
    await session.start()
    try:
        workspace = Path(session.state.manifest.root)
        outside = tmp_path / "outside"
        outside.mkdir()
        os.symlink(outside, workspace / "link", target_is_directory=True)

        with pytest.raises(WorkspaceArchiveWriteError) as exc_info:
            await session.extract(
                "bundle.tar",
                _tar_bytes(members={"link/hello.txt": b"hello from tar"}),
            )

        assert exc_info.value.context["member"] == "link/hello.txt"
        assert exc_info.value.context["reason"] == "symlink in parent path: link"
        assert not (outside / "hello.txt").exists()
    finally:
        await session.shutdown()


@pytest.mark.asyncio
async def test_extract_zip_rejects_symlinked_parent_paths(tmp_path: Path) -> None:
    session = _build_session(tmp_path)
    await session.start()
    try:
        workspace = Path(session.state.manifest.root)
        outside = tmp_path / "outside"
        outside.mkdir()
        os.symlink(outside, workspace / "link", target_is_directory=True)

        with pytest.raises(WorkspaceArchiveWriteError) as exc_info:
            await session.extract(
                "bundle.zip",
                _zip_bytes(members={"link/hello.txt": b"hello from zip"}),
            )

        assert exc_info.value.context["member"] == "link/hello.txt"
        assert exc_info.value.context["reason"] == "symlink in parent path: link"
        assert not (outside / "hello.txt").exists()
    finally:
        await session.shutdown()


@pytest.mark.asyncio
async def test_unix_local_hydrate_workspace_rejects_external_symlink_targets(
    tmp_path: Path,
) -> None:
    session = _build_session(tmp_path)
    await session.start()
    try:
        archive = io.BytesIO()
        with tarfile.open(fileobj=archive, mode="w") as tar:
            info = tarfile.TarInfo(name="leak")
            info.type = tarfile.SYMTYPE
            info.linkname = "/etc/passwd"
            tar.addfile(info)
        archive.seek(0)

        with pytest.raises(WorkspaceArchiveWriteError) as exc_info:
            await session.hydrate_workspace(archive)

        assert exc_info.value.context["member"] == "leak"
        assert (
            exc_info.value.context["reason"] == "absolute symlink target not allowed: /etc/passwd"
        )
        assert not (Path(session.state.manifest.root) / "leak").exists()
    finally:
        await session.shutdown()


@pytest.mark.asyncio
async def test_extract_tar_rejects_windows_drive_member_paths(tmp_path: Path) -> None:
    await _assert_extract_rejects_member(
        tmp_path,
        "bundle.tar",
        _tar_bytes(members={"C:/tmp/evil.txt": b"evil"}),
        expected_member="C:/tmp/evil.txt",
        expected_reason="windows drive path",
    )


@pytest.mark.asyncio
async def test_extract_zip_rejects_windows_drive_member_paths(tmp_path: Path) -> None:
    await _assert_extract_rejects_member(
        tmp_path,
        "bundle.zip",
        _zip_bytes(members={r"C:\tmp\evil.txt": b"evil"}),
        expected_member=r"C:\tmp\evil.txt",
        expected_reason="windows drive path",
    )


@pytest.mark.asyncio
async def test_extract_tar_rejects_windows_separator_member_paths(tmp_path: Path) -> None:
    await _assert_extract_rejects_member(
        tmp_path,
        "bundle.tar",
        _tar_bytes(members={r"..\evil.txt": b"evil"}),
        expected_member=r"..\evil.txt",
        expected_reason="windows path separator",
    )


@pytest.mark.asyncio
async def test_extract_zip_rejects_windows_separator_member_paths(tmp_path: Path) -> None:
    await _assert_extract_rejects_member(
        tmp_path,
        "bundle.zip",
        _zip_bytes(members={r"\evil.txt": b"evil"}),
        expected_member=r"\evil.txt",
        expected_reason="windows path separator",
    )


@pytest.mark.asyncio
async def test_extract_tar_rejects_member_under_non_directory_member(tmp_path: Path) -> None:
    workspace = await _assert_extract_rejects_member(
        tmp_path,
        "bundle.tar",
        _tar_bytes(
            members={
                "nested/hello.txt": b"hello from tar",
                "nested": b"not a directory",
            }
        ),
        expected_member="nested/hello.txt",
        expected_reason="archive path descends through non-directory: nested",
    )

    assert not (workspace / "nested").exists()


@pytest.mark.asyncio
async def test_extract_zip_rejects_member_under_non_directory_member(tmp_path: Path) -> None:
    workspace = await _assert_extract_rejects_member(
        tmp_path,
        "bundle.zip",
        _zip_bytes(
            members={
                "nested/hello.txt": b"hello from zip",
                "nested": b"not a directory",
            }
        ),
        expected_member="nested/hello.txt",
        expected_reason="archive path descends through non-directory: nested",
    )

    assert not (workspace / "nested").exists()


@pytest.mark.asyncio
async def test_unix_local_persist_workspace_excludes_resolved_mount_path(tmp_path: Path) -> None:
    workspace_root = tmp_path / "workspace"
    actual_mount_path = workspace_root / "actual"
    actual_mount_path.mkdir(parents=True)
    (actual_mount_path / "remote.txt").write_text("remote", encoding="utf-8")
    (workspace_root / "keep.txt").write_text("keep", encoding="utf-8")

    state = UnixLocalSandboxSessionState(
        manifest=Manifest(
            root=str(workspace_root),
            entries={
                "logical": GCSMount(
                    bucket="bucket",
                    mount_path=Path("actual"),
                    mount_strategy=InContainerMountStrategy(pattern=MountpointMountPattern()),
                )
            },
        ),
        snapshot=NoopSnapshot(id="noop"),
    )
    session = UnixLocalSandboxSession.from_state(state)

    archive = await session.persist_workspace()

    with tarfile.open(fileobj=archive, mode="r:*") as tar:
        names = set(tar.getnames())

    assert "./keep.txt" in names
    assert "./actual" not in names
    assert "./actual/remote.txt" not in names


@pytest.mark.asyncio
async def test_extract_tar_reuses_directory_listings_during_symlink_checks(tmp_path: Path) -> None:
    workspace = tmp_path / "workspace"
    workspace.mkdir()
    session = _CountingExtractSession(workspace)

    await session.extract(
        "bundle.tar",
        _tar_bytes(
            members={
                "nested/one.txt": b"one",
                "nested/two.txt": b"two",
            }
        ),
    )

    assert (workspace / "nested" / "one.txt").read_text(encoding="utf-8") == "one"
    assert (workspace / "nested" / "two.txt").read_text(encoding="utf-8") == "two"
    assert session.ls_calls == [
        workspace,
        workspace / "nested",
    ]


@pytest.mark.asyncio
async def test_unix_local_helpers_reject_paths_outside_workspace_root(tmp_path: Path) -> None:
    session = _build_session(tmp_path)
    await session.start()
    try:
        with pytest.raises(InvalidManifestPathError, match="must not escape root"):
            await session.ls("../outside")
        with pytest.raises(InvalidManifestPathError, match="must not escape root"):
            await session.mkdir("../outside", parents=True)
        with pytest.raises(InvalidManifestPathError, match="must not escape root"):
            await session.rm("../outside")
        with pytest.raises(InvalidManifestPathError, match="must be relative"):
            await session.extract("/tmp/bundle.tar", _tar_bytes(members={"a.txt": b"a"}))
    finally:
        await session.shutdown()


@pytest.mark.asyncio
async def test_unix_local_helpers_reject_symlink_escape_paths(tmp_path: Path) -> None:
    session = _build_session(tmp_path)
    await session.start()
    try:
        workspace = Path(session.state.manifest.root)
        outside = tmp_path / "outside"
        outside.mkdir()
        os.symlink(outside, workspace / "link", target_is_directory=True)

        with pytest.raises(InvalidManifestPathError, match="must not escape root"):
            await session.mkdir("link/nested", parents=True)
        with pytest.raises(InvalidManifestPathError, match="must not escape root"):
            await session.ls("link")
    finally:
        await session.shutdown()