import asyncio
import io
import json
import os
import tempfile
from collections.abc import Iterator, Mapping
from typing import TYPE_CHECKING, Any, Callable, Literal, Sequence, cast

import duckdb
import pandas as pd
import pyarrow as pa
import pyarrow.compute as pc
import pyarrow.dataset as ds
import pyarrow.fs as pafs
import pyarrow.parquet as pq
from inspect_ai._util.asyncfiles import AsyncFilesystem
from inspect_ai._util.file import file, filesystem
from inspect_ai._util.json import to_json_str_safe
from typing_extensions import override
from upath import UPath

if TYPE_CHECKING:
    from pyarrow import Scalar

from inspect_scout._recorder.summary import Summary

from .._recorder.buffer import (
    SCAN_ERRORS,
    SCAN_SUMMARY,
    RecorderBuffer,
    cleanup_buffer_dir,
    read_scan_errors,
    read_scan_summary,
    scanner_table,
)
from .._scanner.result import Error, ResultReport
from .._scanspec import ScanSpec, ScanTranscripts
from .._transcript.types import TranscriptInfo
from .recorder import (
    ScanRecorder,
    ScanResultsArrow,
    ScanResultsDB,
    ScanResultsDF,
    Status,
)

SCAN_JSON = "_scan.json"


class LazyScannerMapping(Mapping[str, pd.DataFrame]):
    """A lazy mapping that loads DataFrames on demand (no caching).

    This class implements the Mapping protocol but defers loading of pandas
    DataFrames until they are actually accessed. Each access loads the
    DataFrame fresh - no caching is performed since DataFrames can be large
    and users can cache themselves if needed.
    """

    def __init__(
        self,
        scanner_names: list[str],
        loader: Callable[[str], pd.DataFrame],
        transformer: Callable[[pd.DataFrame], pd.DataFrame] | None = None,
    ) -> None:
        """Initialize the lazy mapping.

        Args:
            scanner_names: List of available scanner names (keys).
            loader: Function that loads a DataFrame given a scanner name.
            transformer: Optional function to transform DataFrame after loading.
        """
        self._scanner_names = scanner_names
        self._loader = loader
        self._transformer = transformer

    def __getitem__(self, key: str) -> pd.DataFrame:
        if key not in self._scanner_names:
            raise KeyError(key)
        df = self._loader(key)
        if self._transformer is not None:
            df = self._transformer(df)
        return df

    def __iter__(self) -> Iterator[str]:
        return iter(self._scanner_names)

    def __len__(self) -> int:
        return len(self._scanner_names)

    def __contains__(self, key: object) -> bool:
        return key in self._scanner_names

    def __repr__(self) -> str:
        return f"LazyScannerMapping(scanners={self._scanner_names})"


class FileRecorder(ScanRecorder):
    def __init__(self) -> None:
        self._scan_dir: UPath | None = None
        self._scan_spec: ScanSpec | None = None

    @override
    async def init(
        self,
        spec: ScanSpec,
        scans_location: str,
    ) -> None:
        # create the scan dir
        self._scan_dir = await _ensure_scan_dir(UPath(scans_location), spec.scan_id)
        self._scanners_completed: list[str] = []
        # save the spec
        self._scan_spec = spec
        await self._write_scan_spec()

        # fresh start: clear any stale buffer state from a prior scan that
        # used the same scan_location (the buffer dir is deterministic per
        # scan_location, so unrelated remnants would otherwise carry over)
        cleanup_buffer_dir(RecorderBuffer.buffer_dir(self._scan_dir.as_posix()))

        self._scan_buffer = RecorderBuffer(
            self._scan_dir.as_posix(),
            self._scan_spec,
        )

    async def snapshot_transcripts(self, snapshot: ScanTranscripts) -> None:
        assert self._scan_spec
        self._scan_spec.transcripts = snapshot
        await self._write_scan_spec()

    @override
    async def resume(
        self,
        scan_location: str,
    ) -> ScanSpec:
        """Resume an interrupted scan, retrying transcripts that errored.

        `is_recorded` returns False for errored transcripts so they're
        re-processed; we truncate `_errors.jsonl` to clear stale entries
        that would otherwise outlive a successful retry. To attach to a
        completed scan to add records for *new* transcripts, use
        `attach` instead.
        """
        self._scan_dir = UPath(scan_location)
        self._scan_fs = filesystem(self._scan_dir.as_posix())
        self._scan_spec = await _read_scan_spec(self._scan_dir)

        self._seed_buffer_summary_from_scan_dir()
        # clear errors of transcripts that are about to be retried
        buffer_dir = RecorderBuffer.buffer_dir(self._scan_dir.as_posix())
        buffer_dir.mkdir(parents=True, exist_ok=True)
        (buffer_dir / SCAN_ERRORS).write_text("")

        self._scan_buffer = RecorderBuffer(
            self._scan_dir.as_posix(),
            self.scan_spec,
        )
        return self._scan_spec

    async def attach(
        self,
        scan_location: str,
    ) -> ScanSpec:
        """Attach to an existing scan to add records for *new* transcripts.

        Unlike `resume`, no transcripts are retried — every `record()` is
        for a transcript not previously seen. `_errors.jsonl` is preserved
        across calls so prior runs' errors aren't clobbered.
        """
        self._scan_dir = UPath(scan_location)
        self._scan_fs = filesystem(self._scan_dir.as_posix())
        self._scan_spec = await _read_scan_spec(self._scan_dir)

        self._seed_buffer_summary_from_scan_dir()
        self._seed_buffer_errors_from_scan_dir()

        self._scan_buffer = RecorderBuffer(
            self._scan_dir.as_posix(),
            self.scan_spec,
        )
        return self._scan_spec

    def _seed_buffer_summary_from_scan_dir(self) -> None:
        """Bootstrap the buffer's summary from the scan dir if missing.

        After `sync(complete=True)` cleans the buffer, the accumulated
        summary only lives in the scan dir. When we attach via `resume`
        or `attach`, copy it back so the new `RecorderBuffer` picks up
        the prior counts instead of starting fresh.
        """
        buffer_dir = RecorderBuffer.buffer_dir(self.scan_dir.as_posix())
        buffer_summary = buffer_dir / SCAN_SUMMARY
        scan_summary = self.scan_dir / SCAN_SUMMARY
        if not buffer_summary.exists() and scan_summary.exists():
            buffer_dir.mkdir(parents=True, exist_ok=True)
            buffer_summary.write_text(scan_summary.read_text())

    def _seed_buffer_errors_from_scan_dir(self) -> None:
        """Bootstrap the buffer's errors file from the scan dir if missing.

        Same shape as the summary seed, used by `attach` so prior runs'
        error entries survive across attaches and `record()` calls append
        to the accumulated history.
        """
        buffer_dir = RecorderBuffer.buffer_dir(self.scan_dir.as_posix())
        buffer_errors = buffer_dir / SCAN_ERRORS
        scan_errors = self.scan_dir / SCAN_ERRORS
        if not buffer_errors.exists() and scan_errors.exists():
            buffer_dir.mkdir(parents=True, exist_ok=True)
            buffer_errors.write_text(scan_errors.read_text())

    @override
    async def location(self) -> str:
        return self.scan_dir.as_posix()

    @override
    async def is_recorded(self, transcript_id: str, scanner: str) -> bool:
        return await self._scan_buffer.is_recorded(transcript_id, scanner)

    @override
    async def record(
        self,
        transcript: TranscriptInfo,
        scanner: str,
        results: Sequence[ResultReport],
        metrics: dict[str, dict[str, float]] | None,
    ) -> None:
        await self._scan_buffer.record(transcript, scanner, results, metrics)

    @override
    async def record_metrics(
        self,
        scanner: str,
        metrics: dict[str, dict[str, float]] | None,
    ) -> None:
        await self._scan_buffer.record_metrics(scanner, metrics)

    @override
    async def flush(self) -> None:
        pass

    @override
    async def errors(self) -> list[Error]:
        return self._scan_buffer.errors()

    @override
    async def summary(self) -> Summary:
        return self._scan_buffer.scan_summary().model_copy(deep=True)

    @property
    def scan_dir(self) -> UPath:
        if self._scan_dir is None:
            raise RuntimeError(
                "File recorder must be initialized or resumed before use."
            )
        return self._scan_dir

    @property
    def scan_spec(self) -> ScanSpec:
        if self._scan_spec is None:
            raise RuntimeError(
                "File recorder must be initialized or resumed before use."
            )
        return self._scan_spec

    async def _write_scan_spec(self) -> None:
        async with AsyncFilesystem() as fs:
            await fs.write_file(
                (self.scan_dir / SCAN_JSON).as_posix(),
                to_json_str_safe(self._scan_spec).encode("utf-8"),
            )

    @override
    @staticmethod
    async def sync(scan_location: str, complete: bool) -> Status:
        # get state
        scan_dir = UPath(scan_location)
        scan_spec = await _read_scan_spec(scan_dir)
        scan_buffer_dir = RecorderBuffer.buffer_dir(scan_location)

        # write each scanner's compacted parquet. when a prior sync has
        # already written one (e.g. on a multi-call eval_set lifecycle),
        # merge it in so the new output is the union of prior data + this
        # call's buffer rows. once that's in place, cleaning up the buffer
        # after sync is safe — the data lives on in the compacted output.
        #
        # `scanner_table` uses pyarrow's `ds.dataset(list)` which requires
        # uniform path types: the buffer dir is always local (per scout's
        # existing convention — see `RecorderBuffer.buffer_dir`), but the
        # scan dir may be on a remote filesystem (e.g. s3://). When the
        # prior compacted parquet is remote, download it to a local temp
        # file before passing to `scanner_table`.
        #
        # write to a sibling `.tmp` then atomically rename so the same path
        # is never simultaneously an input to scanner_table and the target
        # of an in-progress write (avoids any read/write overlap and gives
        # crash resilience: a failure mid-write leaves the prior compacted
        # file intact).
        sync_fs = filesystem(scan_dir.as_posix())
        async with AsyncFilesystem() as fs:
            for scanner in sorted(scan_spec.scanners.keys()):
                output_path = _scanner_parquet_file(scan_dir, scanner)
                parquet_bytes = await _compact_with_prior(
                    scan_buffer_dir, scanner, prior=UPath(output_path)
                )
                if parquet_bytes is not None:
                    tmp_path = f"{output_path}.tmp"
                    await fs.write_file(tmp_path, parquet_bytes)
                    sync_fs.mv(tmp_path, output_path)

        # sync summary and errors
        _sync_status_files(scan_dir, scan_buffer_dir, scan_spec, complete)

        # cleanup scan buffer if we are complete
        if complete:
            cleanup_buffer_dir(scan_buffer_dir)

        return Status(
            complete=complete,
            spec=scan_spec,
            location=scan_dir.as_posix(),
            summary=read_scan_summary(scan_dir, scan_spec),
            errors=_read_scan_errors(scan_dir),
        )

    @override
    @staticmethod
    async def status(scan_location: str) -> Status:
        buffer_dir = RecorderBuffer.buffer_dir(scan_location)
        spec = await _read_scan_spec(UPath(scan_location))

        # Buffer dir is always local (see RecorderBuffer.buffer_dir); sync
        # reads here are microseconds — acceptable on the loop.
        if buffer_dir.exists():
            return Status(
                complete=False,
                spec=spec,
                location=scan_location,
                summary=read_scan_summary(buffer_dir, spec),
                errors=_read_scan_errors(buffer_dir),
            )

        # Scan dir may be remote; use async reads via AsyncFilesystem so
        # the loop yields during S3 I/O.
        scan_dir = UPath(scan_location)
        async with AsyncFilesystem() as fs:
            summary = await _async_read_scan_summary(fs, scan_dir, spec)
            errors = await _async_read_scan_errors(fs, scan_dir)
        return Status(
            complete=summary.complete,
            spec=spec,
            location=scan_location,
            summary=summary,
            errors=errors,
        )

    @override
    @staticmethod
    async def results_arrow(
        scan_location: str,
    ) -> ScanResultsArrow:
        scan_path = UPath(scan_location)

        class _ScanResultsArrowFiles(ScanResultsArrow):
            def _resolve_parquet_source(
                self, scanner: str
            ) -> tuple[str | io.BytesIO, pafs.FileSystem | None]:
                """Return (path, filesystem) for opening a parquet file.

                For cloud paths with native PyArrow support (S3, GCS, ABFS),
                returns a pyarrow filesystem that uses HTTP range requests.
                For az:// (no native PyArrow support), downloads via fsspec
                and returns a BytesIO. For local paths, returns (str, None).
                """
                scanner_path = scan_path / f"{scanner}.parquet"
                path_str = scanner_path.as_posix()

                if path_str.startswith(
                    ("s3://", "gs://", "gcs://", "abfs://", "abfss://")
                ):
                    pa_fs, pa_path = pafs.FileSystem.from_uri(path_str)
                    return pa_path, pa_fs

                if path_str.startswith("az://"):
                    with file(path_str, "rb") as f:
                        return io.BytesIO(f.read()), None

                return str(scanner_path), None

            @override
            def reader(
                self,
                scanner: str,
                streaming_batch_size: int = 1024,
                exclude_columns: list[str] | None = None,
            ) -> pa.RecordBatchReader:
                pa_path, pa_fs = self._resolve_parquet_source(scanner)

                if pa_fs is not None:
                    # Use pre_buffer=True to coalesce HTTP range requests,
                    # then read() + to_reader() instead of iter_batches()
                    # which doesn't support pre_buffer.
                    parquet = pq.ParquetFile(pa_path, filesystem=pa_fs, pre_buffer=True)
                else:
                    parquet = pq.ParquetFile(pa_path)

                exclude = set(exclude_columns) if exclude_columns else set()
                columns = [c for c in parquet.schema.names if c not in exclude]

                if pa_fs is not None:
                    table = parquet.read(columns=columns)
                    return table.to_reader(max_chunksize=streaming_batch_size)

                fields_by_name = {f.name: f for f in parquet.schema_arrow}
                arrow_schema = pa.schema([fields_by_name[name] for name in columns])

                return pa.RecordBatchReader.from_batches(
                    arrow_schema,
                    parquet.iter_batches(
                        batch_size=streaming_batch_size, columns=columns
                    ),
                )

            def _point_lookup(
                self,
                scanner: str,
                id_column: str,
                id_value: Any,
                target_columns: list[str],
            ) -> pa.Table:
                """Look up a single row by id, returning a 1-row table.

                For remote files, scans only the id column across row groups
                via range requests, then fetches target columns from the
                matching group only. For local files, uses a dataset filter.
                """
                pa_path, pa_fs = self._resolve_parquet_source(scanner)

                if pa_fs is not None:
                    pf = pq.ParquetFile(pa_path, filesystem=pa_fs)
                    for i in range(pf.metadata.num_row_groups):
                        rg_ids = pf.read_row_group(i, columns=[id_column])
                        mask = pc.equal(rg_ids[id_column], id_value)
                        if pc.any(mask).as_py():
                            rg_data = pf.read_row_group(i, columns=target_columns)
                            return rg_data.filter(mask)
                    return pa.table({c: [] for c in target_columns})

                dataset: ds.Dataset = (
                    ds.dataset(pq.read_table(pa_path))
                    if isinstance(pa_path, io.BytesIO)
                    else ds.dataset(pa_path, format="parquet")
                )
                return dataset.to_table(
                    columns=target_columns,
                    filter=(pc.field(id_column) == id_value),
                )

            def get_field(
                self, scanner: str, id_column: str, id_value: Any, target_column: str
            ) -> "Scalar[Any]":
                table = self._point_lookup(
                    scanner, id_column, id_value, [target_column]
                )

                if len(table) == 0:
                    raise KeyError(f"{id_value!r} not found in {id_column}")

                if len(table) > 1:
                    raise ValueError(
                        f"Multiple rows found for {id_column}={id_value!r}"
                    )

                return cast("Scalar[Any]", table[target_column][0])

            def _schema_names(self, scanner: str) -> set[str]:
                """Return the set of column names in a scanner's parquet file."""
                pa_path, pa_fs = self._resolve_parquet_source(scanner)
                if pa_fs is not None:
                    pf = pq.ParquetFile(pa_path, filesystem=pa_fs)
                else:
                    pf = pq.ParquetFile(pa_path)
                return set(pf.schema.names)

            def get_fields(
                self,
                scanner: str,
                id_column: str,
                id_value: Any,
                target_columns: list[str],
            ) -> dict[str, Any]:
                """Fetch multiple columns for a single row in one parquet scan.

                Missing columns (e.g. in old parquet files) return None.
                """
                schema_names = self._schema_names(scanner)
                present = [c for c in target_columns if c in schema_names]
                missing = [c for c in target_columns if c not in schema_names]

                table = self._point_lookup(scanner, id_column, id_value, present)

                if len(table) == 0:
                    raise KeyError(f"{id_value!r} not found in {id_column}")

                if len(table) > 1:
                    raise ValueError(
                        f"Multiple rows found for {id_column}={id_value!r}"
                    )

                result = {c: table[c][0].as_py() for c in present}
                for c in missing:
                    result[c] = None
                return result

        # get the status
        status = await FileRecorder.status(scan_location)

        # enumerate the scanners
        async with AsyncFilesystem() as fs:
            parquet_uris = sorted(
                [u async for u in fs.iter_files(scan_location, "*.parquet")]
            )
        scanners = [UPath(u).stem for u in parquet_uris]

        return _ScanResultsArrowFiles(
            status=status.complete,
            spec=status.spec,
            location=status.location,
            summary=status.summary,
            errors=status.errors,
            scanners=scanners,
        )

    @override
    @staticmethod
    async def results_df(
        scan_location: str,
        *,
        scanner: str | None = None,
        exclude_columns: list[str] | None = None,
    ) -> ScanResultsDF:
        scan_dir = UPath(scan_location)
        status = await FileRecorder.status(scan_location)

        # Determine available scanner names
        if scanner is not None:
            scanner_names = [scanner]
        else:
            async with AsyncFilesystem() as fs:
                parquet_uris = sorted(
                    [u async for u in fs.iter_files(scan_location, "*.parquet")]
                )
            scanner_names = [UPath(u).stem for u in parquet_uris]

        # Create lazy mapping with a loader that reads DataFrames on demand
        scanners = LazyScannerMapping(
            scanner_names=scanner_names,
            loader=lambda name: _load_scanner_df(
                scan_dir, name, exclude_columns=exclude_columns
            ),
        )

        return ScanResultsDF(
            complete=status.complete,
            spec=status.spec,
            location=status.location,
            summary=status.summary,
            errors=status.errors,
            scanners=scanners,
        )

    @override
    @staticmethod
    async def results_db(
        scan_location: str, *, rows: Literal["results", "transcripts"] = "results"
    ) -> ScanResultsDB:
        from upath import UPath

        status = await FileRecorder.status(scan_location)

        # Create in-memory DuckDB connection
        conn = duckdb.connect(":memory:")

        # Create views for each parquet file
        async with AsyncFilesystem() as fs:
            parquet_uris = sorted(
                [u async for u in fs.iter_files(scan_location, "*.parquet")]
            )
        for parquet_uri in parquet_uris:
            parquet_file = UPath(parquet_uri)
            scanner_name = parquet_file.stem
            # Create a view that references the parquet file
            # Use absolute path to ensure it works regardless of working directory
            # iter_files already yields absolute URIs.
            abs_path = parquet_file.as_posix()

            # Check if we need to expand resultsets
            if rows == "results" and _has_resultsets(conn, abs_path):
                # Create expanded view with UNNEST for resultset rows
                sql = _create_expanded_view_sql(conn, abs_path, scanner_name)
                conn.execute(sql)
            else:
                # Use existing simple view logic (for transcripts mode or non-resultset scanners)
                # Check if value_type is uniform and needs casting
                uniform_type = _get_uniform_value_type(conn, abs_path)
                if uniform_type in ("boolean", "number"):
                    cast_expr = _cast_value_sql(uniform_type)
                    select_clause = f"SELECT * REPLACE ({cast_expr} AS value)"
                else:
                    select_clause = "SELECT *"

                conn.execute(
                    f"CREATE VIEW {scanner_name} AS {select_clause} FROM read_parquet('{abs_path}')"
                )

        return ScanResultsDB(
            status=status.complete,
            spec=status.spec,
            location=status.location,
            summary=status.summary,
            errors=status.errors,
            conn=conn,
        )

    @override
    @staticmethod
    async def list(scans_location: str) -> list[Status]:
        async with AsyncFilesystem() as fs:
            # iter_dirs yields URIs with a trailing slash; normalize via
            # UPath.as_posix() to match the no-slash form used everywhere
            # else (e.g. sync(), location()) so Status.location compares
            # equal across code paths.
            scan_dirs = [
                UPath(uri).as_posix()
                async for uri in fs.iter_dirs(
                    scans_location, "scan_id=*", recursive=True
                )
            ]

        # Fetch in parallel; skip scans that no longer exist and propagate
        # any other error. return_exceptions=True keeps one failure from
        # cancelling its siblings.
        results = await asyncio.gather(
            *(FileRecorder.status(d) for d in scan_dirs), return_exceptions=True
        )
        scans: list[Status] = []
        for r in results:
            if isinstance(r, FileNotFoundError):
                continue
            if isinstance(r, BaseException):
                raise r
            scans.append(r)
        return scans


async def _compact_with_prior(
    scan_buffer_dir: UPath, scanner: str, *, prior: UPath
) -> bytes | None:
    """Compact buffer parquets, optionally merging in a prior compacted output.

    `scanner_table` requires uniform local paths. If `prior` exists on a
    remote filesystem, download it to a local temp file before passing it
    in, then clean up.
    """
    if not prior.exists():
        return scanner_table(scan_buffer_dir, scanner)

    local_prior: UPath | None = None
    try:
        if prior.protocol in ("", "file"):
            extra = [prior]
        else:
            tmp_fd, tmp_name = tempfile.mkstemp(suffix=".parquet")
            os.close(tmp_fd)
            local_prior = UPath(tmp_name)
            local_prior.write_bytes(prior.read_bytes())
            extra = [local_prior]
        return scanner_table(scan_buffer_dir, scanner, extra_inputs=extra)
    finally:
        if local_prior is not None:
            local_prior.unlink(missing_ok=True)


def _scanner_parquet_file(scan_dir: UPath, scanner: str) -> str:
    return (scan_dir / f"{scanner}.parquet").as_posix()


async def _read_scan_spec(scan_dir: UPath) -> ScanSpec:
    scan_json = scan_dir / SCAN_JSON
    async with AsyncFilesystem() as fs:
        try:
            data = await fs.read_file(scan_json.as_posix())
        except Exception as e:
            if _is_not_found(e):
                raise FileNotFoundError(
                    f"The specified directory '{scan_dir}' does not contain a scan."
                ) from e
            raise
    return ScanSpec.model_validate_json(data)


def _is_not_found(e: BaseException) -> bool:
    """True if the exception indicates a missing file (local or S3 404).

    Duck-types ``botocore.exceptions.ClientError`` by inspecting the
    ``.response`` attribute shape; avoids a hard dependency on the type.
    """
    if isinstance(e, FileNotFoundError):
        return True
    response = getattr(e, "response", None)
    if isinstance(response, dict):
        code = response.get("Error", {}).get("Code")
        if code in ("404", "NoSuchKey", "NotFound"):
            return True
    return False


def _read_scan_errors(scan_dir: UPath) -> list[Error]:
    scan_errors = scan_dir / SCAN_ERRORS
    return read_scan_errors(str(scan_errors))


async def _async_read_scan_summary(
    fs: AsyncFilesystem, scan_dir: UPath, spec: ScanSpec
) -> Summary:
    """Async counterpart to `read_scan_summary` for remote scan dirs.

    The logic intentionally duplicates `read_scan_summary` rather than
    delegating. We can't convert `read_scan_summary` itself to async
    (it has sync callers — `RecorderBuffer.__init__`, `_sync_status_files`,
    and the test suite), and we don't want to wrap it in
    `anyio.to_thread.run_sync` because that puts sync fsspec calls inside
    a thread, which interacts poorly with fsspec's internal loop-in-a-thread
    + instance cache + adaptive retry behavior. So async callers that may be
    reading from remote storage call this directly, and `AsyncFilesystem.read_file`
    handles the async I/O natively.
    """
    try:
        data = await fs.read_file((scan_dir / SCAN_SUMMARY).as_posix())
    except Exception as e:
        if _is_not_found(e):
            return Summary(complete=False, scanners=list(spec.scanners.keys()))
        raise
    text = data.decode().strip()
    if text:
        return Summary.model_validate_json(text)
    return Summary(complete=False, scanners=list(spec.scanners.keys()))


async def _async_read_scan_errors(fs: AsyncFilesystem, scan_dir: UPath) -> list[Error]:
    """Async counterpart to `_read_scan_errors` for remote scan dirs.

    Duplicates the parsing logic intentionally — see
    `_async_read_scan_summary` for the reasoning. In short: the sync
    helpers must stay sync for their sync callers, and we don't put
    sync fsspec inside `to_thread`.
    """
    try:
        data = await fs.read_file((scan_dir / SCAN_ERRORS).as_posix())
    except Exception as e:
        if _is_not_found(e):
            return []
        raise
    errors: list[Error] = []
    for line in data.decode().splitlines():
        line = line.strip()
        if line:
            errors.append(Error(**json.loads(line)))
    return errors


async def _find_scan_dir(scans_path: UPath, scan_id: str) -> UPath | None:
    _ensure_scans_dir(scans_path)
    async with AsyncFilesystem() as fs:
        async for uri in fs.iter_dirs(scans_path.as_posix(), f"scan_id={scan_id}"):
            return UPath(uri)
    return None


async def _ensure_scan_dir(scans_path: UPath, scan_id: str) -> UPath:
    # look for an existing scan dir
    scan_dir = await _find_scan_dir(scans_path, scan_id)

    # if there is no scan_dir then create one
    if scan_dir is None:
        scan_dir = scans_path / f"scan_id={scan_id}"
        scan_dir.mkdir(parents=True, exist_ok=False)

    # return scan dir
    return scan_dir


def _ensure_scans_dir(scans_dir: UPath) -> None:
    scans_dir.mkdir(parents=True, exist_ok=True)


def _has_resultsets(conn: duckdb.DuckDBPyConnection, parquet_path: str) -> bool:
    """
    Check if a parquet file contains any resultset rows.

    Args:
        conn: DuckDB connection
        parquet_path: Path to the parquet file

    Returns:
        True if any rows have value_type == 'resultset', False otherwise
    """
    result = conn.execute(
        f"SELECT COUNT(*) FROM read_parquet('{parquet_path}') WHERE value_type = 'resultset' LIMIT 1"
    ).fetchone()
    return result is not None and result[0] > 0


def _create_expanded_view_sql(
    conn: duckdb.DuckDBPyConnection, parquet_path: str, scanner_name: str
) -> str:
    """
    Generate SQL to create an expanded view that unnests resultset rows.

    For rows where value_type == 'resultset', the value field contains a JSON array
    of Result objects. This function generates SQL that:
    1. Passes through non-resultset rows as-is
    2. Unnests resultset rows using json_extract and UNNEST
    3. Extracts Result fields into columns
    4. Extracts value as string (typed conversion handled downstream)

    Args:
        conn: DuckDB connection to query schema
        parquet_path: Path to the parquet file
        scanner_name: Name for the view

    Returns:
        SQL CREATE VIEW statement with UNION of non-resultset and expanded resultset rows
    """
    # Query the actual column names from the parquet file to avoid hardcoding
    # We use LIMIT 0 to get just the schema without reading data
    result = conn.execute(
        f"SELECT * FROM read_parquet('{parquet_path}') LIMIT 0"
    ).description
    all_columns = [col[0] for col in result]

    # These are the Result fields that need special handling during expansion
    result_fields = {
        "uuid",
        "label",
        "value",
        "value_type",
        "answer",
        "explanation",
        "metadata",
        "message_references",
        "event_references",
    }

    # These columns should be NULL in expanded rows to avoid incorrect aggregation
    # (they represent the scan execution, not individual results)
    scan_execution_fields = {"scan_total_tokens", "scan_model_usage"}

    # Validation columns should also be NULL in expanded rows because:
    # 1. Label-based validation can't create synthetic rows in SQL (too complex)
    # 2. Without synthetic rows, validation results would be incomplete/misleading
    # 3. Users should use Pandas (rows="results") for full validation support
    validation_fields = {
        "validation_target",
        "validation_result",
        "validation_predicate",
        "validation_split",
    }

    # Build the non-resultset rows query with explicit column selection
    non_resultset_cols = ", ".join(all_columns)
    non_resultset_query = f"""
    SELECT {non_resultset_cols} FROM read_parquet('{parquet_path}')
    WHERE value_type != 'resultset' OR value_type IS NULL
    """

    # Build the expanded resultset rows query
    # Base columns are everything except the result fields and scan execution fields
    base_columns = [
        col
        for col in all_columns
        if col not in result_fields and col not in scan_execution_fields
    ]

    # Extract value as string — downstream _cast_value_column() (Python) and
    # castValue() (TypeScript) handle typed conversion. Kept as string to avoid
    # DuckDB 1.5 mixed-type CASE expression errors.
    cast_value_expr = "json_extract_string(CAST(elem AS JSON), '$.value')"

    # Build column selects for expanded query in the same order as all_columns
    expanded_col_selects = []
    for col in all_columns:
        if col in base_columns:
            # Base columns from the parquet table
            expanded_col_selects.append(f"base_row.{col}")
        elif col in scan_execution_fields:
            # NULL out scan execution fields to avoid incorrect aggregation
            expanded_col_selects.append(f"NULL AS {col}")
        elif col in validation_fields or col.startswith("validation_result_"):
            # NULL out validation fields in expanded rows because:
            # 1. Label-based validation can't create synthetic rows in SQL
            # 2. Without synthetic rows, validation results are incomplete/misleading
            expanded_col_selects.append(f"NULL AS {col}")
        elif col == "uuid":
            expanded_col_selects.append(
                "json_extract_string(CAST(elem AS JSON), '$.uuid') AS uuid"
            )
        elif col == "label":
            expanded_col_selects.append(
                "json_extract_string(CAST(elem AS JSON), '$.label') AS label"
            )
        elif col == "value":
            expanded_col_selects.append(f"({cast_value_expr}) AS value")
        elif col == "value_type":
            expanded_col_selects.append(
                "COALESCE(json_extract_string(CAST(elem AS JSON), '$.type'), 'null') AS value_type"
            )
        elif col == "answer":
            expanded_col_selects.append(
                "json_extract_string(CAST(elem AS JSON), '$.answer') AS answer"
            )
        elif col == "explanation":
            expanded_col_selects.append(
                "json_extract_string(CAST(elem AS JSON), '$.explanation') AS explanation"
            )
        elif col == "metadata":
            expanded_col_selects.append(
                "COALESCE(json_extract_string(CAST(elem AS JSON), '$.metadata'), '{{}}') AS metadata"
            )
        elif col == "message_references":
            # Extract references array and filter by type="message"
            # Use CASE to handle NULL references, scalar subquery to filter
            expanded_col_selects.append(
                "CAST(CASE WHEN json_extract(CAST(elem AS JSON), '$.references') IS NULL THEN '[]'::JSON "
                "ELSE (SELECT COALESCE(list(ref), []) FROM UNNEST(CAST(json_extract(CAST(elem AS JSON), '$.references') AS JSON[])) AS t(ref) "
                "WHERE json_extract_string(CAST(ref AS JSON), '$.type') = 'message') END AS JSON) AS message_references"
            )
        elif col == "event_references":
            # Extract references array and filter by type="event"
            # Use CASE to handle NULL references, scalar subquery to filter
            expanded_col_selects.append(
                "CAST(CASE WHEN json_extract(CAST(elem AS JSON), '$.references') IS NULL THEN '[]'::JSON "
                "ELSE (SELECT COALESCE(list(ref), []) FROM UNNEST(CAST(json_extract(CAST(elem AS JSON), '$.references') AS JSON[])) AS t(ref) "
                "WHERE json_extract_string(CAST(ref AS JSON), '$.type') = 'event') END AS JSON) AS event_references"
            )

    expanded_resultset_query = f"""
    SELECT
        {", ".join(expanded_col_selects)}
    FROM read_parquet('{parquet_path}') base_row,
    UNNEST(CAST(json_extract(base_row.value, '$') AS JSON[])) AS t(elem)
    WHERE base_row.value_type = 'resultset'
    """

    # Combine both queries with UNION ALL
    return f"""CREATE VIEW {scanner_name} AS
    SELECT * FROM (
        {non_resultset_query}
        UNION ALL
        {expanded_resultset_query}
    )"""


def _get_uniform_value_type(
    conn: duckdb.DuckDBPyConnection, parquet_path: str
) -> str | None:
    """
    Check if value_type is uniform across all rows in a parquet file.

    Args:
        conn: DuckDB connection
        parquet_path: Path to the parquet file

    Returns:
        The uniform value_type if all rows have the same type, None otherwise
    """
    result = conn.execute(
        f"SELECT DISTINCT value_type FROM read_parquet('{parquet_path}') WHERE value_type IS NOT NULL"
    ).fetchall()

    if len(result) == 1:
        return str(result[0][0])
    else:
        return None


def _cast_value_sql(value_type: str) -> str:
    """
    Generate SQL CASE expression to cast the value column based on value_type.

    Args:
        value_type: The uniform value type ('boolean', 'number', etc.)

    Returns:
        SQL CASE expression for casting the value column
    """
    if value_type == "boolean":
        return """CASE
            WHEN value IN ('true', 'True') THEN TRUE
            WHEN value IN ('false', 'False') THEN FALSE
            ELSE NULL
        END"""
    elif value_type == "number":
        return "TRY_CAST(value AS DOUBLE)"
    else:
        # For string, null, array, object - keep as-is
        return "value"


def _load_scanner_df(
    scan_dir: UPath,
    scanner_name: str,
    *,
    exclude_columns: list[str] | None = None,
) -> pd.DataFrame:
    """Load a scanner's DataFrame from a parquet file.

    Args:
        scan_dir: Directory containing the scan parquet files.
        scanner_name: Name of the scanner (without .parquet extension).
        exclude_columns: List of column names to exclude when reading.
            Non-existent columns are silently ignored.

    Returns:
        DataFrame with the scanner results, value column cast appropriately.
    """
    parquet_file = scan_dir / f"{scanner_name}.parquet"
    # Use file() from inspect_ai to match original AsyncFilesystem behavior
    with file(parquet_file.as_posix(), "rb") as f:
        file_bytes = f.read()

    # Determine columns to read (exclude specified columns if they exist)
    columns: list[str] | None = None
    if exclude_columns:
        parquet_schema = pq.read_schema(io.BytesIO(file_bytes))
        all_columns = set(parquet_schema.names)
        columns = [col for col in all_columns if col not in exclude_columns]

    table = pq.read_table(io.BytesIO(file_bytes), columns=columns)
    df = table.to_pandas(types_mapper=pd.ArrowDtype)
    return _cast_value_column(df)


def _cast_value_column(df: pd.DataFrame) -> pd.DataFrame:
    """
    Cast the 'value' column to its appropriate type based on 'value_type'.

    The value column is stored as string in parquet files to handle mixed types
    during compaction. This function restores the original types for analysis.

    Args:
        df: DataFrame with 'value' and 'value_type' columns

    Returns:
        DataFrame with value column cast to appropriate type
    """
    if "value" not in df.columns or "value_type" not in df.columns:
        return df

    # Check if value_type is uniform across all rows
    value_types = df["value_type"].dropna().unique()

    if len(value_types) == 0:
        # No value_type information, keep as-is
        return df
    elif len(value_types) == 1:
        # Uniform type - safe to cast entire column
        vtype = value_types[0]

        try:
            if vtype == "boolean":
                # Handle various string representations of booleans
                df["value"] = df["value"].map(
                    {
                        "true": True,
                        "false": False,
                        "True": True,
                        "False": False,
                        None: None,
                    }
                )
            elif vtype == "number":
                # Use nullable Int64/Float64 to preserve NaN
                # Try int first, fall back to float
                df["value"] = pd.to_numeric(df["value"], errors="coerce")
            elif vtype in ("string", "null"):
                # Already strings or nulls, keep as-is
                pass
            elif vtype in ("array", "object"):
                # Complex types are JSON strings, keep as-is
                # Could optionally parse JSON here if needed
                pass
        except Exception:
            # If casting fails for any reason, keep original string representation
            pass
    else:
        # Mixed types - keep as strings for safety
        pass

    return df


def _sync_status_files(
    scan_dir: UPath, scan_buffer_dir: UPath, scan_spec: ScanSpec, complete: bool
) -> None:
    """Copy summary and errors from buffer to scan directory."""
    # copy scan summary (update with complete status)
    with file((scan_dir / SCAN_SUMMARY).as_posix(), "w") as f:
        summary = read_scan_summary(scan_buffer_dir, scan_spec)
        summary.complete = complete
        f.write(summary.model_dump_json())

    # copy errors
    with file((scan_dir / SCAN_ERRORS).as_posix(), "w") as f:
        for error in _read_scan_errors(scan_buffer_dir):
            f.write(error.model_dump_json(warnings=False) + "\n")
