"""Fetch job outputs back from where the job ran.

Two pairs of functions, one per "kind" of payload to retrieve:

* :func:`fetch_local` / :func:`fetch_remote` — copy ``spec.cwd`` (the
  job's *workspace*: the submitted-source root the job ran out of).
* :func:`fetch_workdir_local` / :func:`fetch_workdir_remote` —
  v0.7.7: copy ``spec.workdir`` (the per-job scratch directory the
  daemon materializes at dispatch, exposed to the job as
  ``$VQ_WORKDIR``). Operators dump intermediate / large artefacts
  there per the v0.6.54 agent-protocol convention.

Both pairs place their content under ``output_dir/<dest>/`` so
multiple fetches don't collide; the workdir variants append
``-workdir`` to the dest name so a workspace fetch and a workdir
fetch of the same job can sit side-by-side without a name conflict.

Two internal verbs feed the remote variants:

* ``vq tar-workspace JOBID`` (hidden from --help) feeds
  :func:`fetch_remote`. Knows the remote's state-dir layout via the
  remote vq's own :mod:`paths` module.
* ``vq tar-workdir JOBID`` (hidden from --help) feeds
  :func:`fetch_workdir_remote`. Same delegation pattern.
"""

from __future__ import annotations

import contextlib
import io
import json
import logging
import os
import shutil
import sys
import tarfile
import tempfile
from collections.abc import Callable
from dataclasses import dataclass
from pathlib import Path

from vq import config, paths, transport
from vq.config import HostConfig
from vq.ownership import check_owner
from vq.scheduler_dispatch import (
    SchedulerError,
    SchedulerHandle,
    scheduler_dispatcher_for,
)
from vq.spec import JobSpec, describe_exit_code, utcnow_iso
from vq.status import terminal_diagnosis_for_spec

log = logging.getLogger(__name__)

TERMINAL_DIAGNOSIS_SIDECAR = "_vq/terminal-diagnosis.json"


def _diagnosis_sidecar_payload(spec: JobSpec) -> dict[str, object]:
    """Machine-readable queue outcome copied with fetched artifacts."""
    return {
        "schema": "vq.terminal-diagnosis.v1",
        "jobid": spec.id,
        "job_name": spec.job_name,
        "state": spec.state.value,
        "exit_code": spec.exit_code,
        "exit_code_description": (
            describe_exit_code(spec.exit_code)
            if spec.exit_code is not None
            else None
        ),
        "failure_reason": spec.failure_reason,
        "failure_tail": spec.failure_tail,
        "submitted_at": spec.submitted_at,
        "started_at": spec.started_at,
        "finished_at": spec.finished_at,
        "scheduler_target": spec.scheduler_target,
        "scheduler_job_id": spec.scheduler_job_id,
        "scheduler_state": spec.scheduler_state,
        "scheduler_walltime_used": spec.scheduler_walltime_used,
        "scheduler_walltime_limit": spec.scheduler_walltime_limit,
        "terminal_diagnosis": terminal_diagnosis_for_spec(spec),
        "generated_at": utcnow_iso(),
    }


def _diagnosis_sidecar_bytes(spec: JobSpec) -> bytes:
    return (
        json.dumps(
            _diagnosis_sidecar_payload(spec),
            indent=2,
            sort_keys=True,
            default=str,
        )
        + "\n"
    ).encode("utf-8")


def _write_diagnosis_sidecar(dst: Path, spec: JobSpec) -> None:
    path = dst / TERMINAL_DIAGNOSIS_SIDECAR
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_bytes(_diagnosis_sidecar_bytes(spec))


def _add_diagnosis_sidecar_to_tar(
    tf: tarfile.TarFile,
    *,
    arcname_root: str,
    spec: JobSpec,
) -> None:
    data = _diagnosis_sidecar_bytes(spec)
    info = tarfile.TarInfo(f"{arcname_root}/{TERMINAL_DIAGNOSIS_SIDECAR}")
    info.size = len(data)
    info.mode = 0o644
    tf.addfile(info, io.BytesIO(data))


def _stream_archive_with_diagnosis(archive: Path, spec: JobSpec) -> None:
    """Stream an archived workspace tar while appending fresh vq metadata."""
    arcname_root = spec.dest_dirname
    with (
        tarfile.open(archive, mode="r:*") as src_tf,
        tarfile.open(fileobj=sys.stdout.buffer, mode="w|") as out_tf,
    ):
        for member in src_tf:
            if member.name:
                arcname_root = member.name.split("/", 1)[0]
            fileobj = src_tf.extractfile(member) if member.isfile() else None
            try:
                out_tf.addfile(member, fileobj)
            finally:
                if fileobj is not None:
                    fileobj.close()
        _add_diagnosis_sidecar_to_tar(
            out_tf,
            arcname_root=arcname_root,
            spec=spec,
        )


def _refresh_live_scheduler_workspace(spec: JobSpec) -> None:
    """Stage a live scheduler job's current remote workspace back to ``spec.cwd``.

    Terminal scheduler jobs are already staged back by the daemon's reap path, and
    ordinary local jobs have no scheduler metadata. For live scheduler jobs, this
    gives ``vq fetch`` an explicit mid-run snapshot without changing the normal
    local/remote tar extraction contract.
    """
    if (
        spec.is_archived
        or spec.is_terminal
        or spec.scheduler_target is None
        or spec.scheduler_job_id is None
    ):
        return
    try:
        host_cfg = config.load_config().host(spec.scheduler_target)
        dispatcher = scheduler_dispatcher_for(host_cfg)
        handle = SchedulerHandle(
            job_id=spec.scheduler_job_id,
            remote_workspace=dispatcher.remote_workspace(spec.id),
        )
        dispatcher.fetch_results(handle, Path(spec.cwd))
    except config.ConfigError as exc:
        raise transport.RemoteError(
            f"cannot refresh live scheduler workspace for {spec.id}: {exc}"
        ) from exc
    except SchedulerError as exc:
        raise transport.RemoteError(
            f"failed to refresh live scheduler workspace for {spec.id}: {exc}"
        ) from exc


def fetch_local(jobid: str, output_dir: Path, *, multi_user: bool = False) -> Path:
    """Copy the local workspace for ``jobid`` into ``output_dir/<dest>/``.

    v0.5.34: ``<dest>`` is ``<job_name>-<jobid>`` when the spec has
    ``job_name`` set, else ``<jobid>`` (pre-v0.5.34 behaviour). The
    name-prefixed form makes a directory of fetched workspaces self-
    documenting — ``ls ./results`` shows what each job was.

    Returns the path of the copied workspace. If the job has been
    archived by ``vq cleanup --archive``, un-tars from the archive
    instead of copying the (gone) workspace dir. Raises
    :class:`FileNotFoundError` if the job isn't in the local queue, or
    if the workspace and archive are both missing.
    """
    if multi_user:
        spec_path = paths.resolve_spec_path(jobid, multi_user=True)
    else:
        spec_path = paths.spec_path(jobid)
        if not spec_path.exists():
            raise FileNotFoundError(f"no such job: {jobid}")
    spec = JobSpec.read(spec_path)
    # v0.6.x: multi-user ownership check.
    check_owner(spec)
    output_dir.mkdir(parents=True, exist_ok=True)
    dst = output_dir / spec.dest_dirname
    if dst.exists():
        raise FileExistsError(f"destination already exists: {dst} (remove it or pick another -o)")
    _refresh_live_scheduler_workspace(spec)
    if spec.is_archived and spec.archive_path:
        archive = Path(spec.archive_path)
        if not archive.is_file():
            raise FileNotFoundError(
                f"archive for job {jobid} not found at {archive} "
                "(record exists in spec but file is gone)"
            )
        # Tarball top-level is ``<dest_dirname>/`` (= ``<name>-<jobid>``
        # for v0.5.34+ named jobs, or just ``<jobid>`` for pre-v0.5.34
        # archives and unnamed jobs). Either way extracting under
        # output_dir yields the right path because we computed
        # dst = output_dir / spec.dest_dirname above.
        with tarfile.open(archive, mode="r:bz2") as tf:
            tf.extractall(output_dir, filter="data")
    else:
        src = Path(spec.cwd)
        if not src.is_dir():
            raise FileNotFoundError(
                f"workspace for job {jobid} not found at {src} "
                "(was it cleaned up or never materialized?)"
            )
        shutil.copytree(src, dst)
    _write_diagnosis_sidecar(dst, spec)
    # Stamp last_fetched_at on terminal specs. v0.8.13 *Gray's Transaction*:
    # lock + fresh re-read (STATE-4) — the `spec` read at the top is stale
    # after the copy/extract, and a concurrent `vq status` / `vq cleanup`
    # write must not be clobbered. Best-effort; never fail the fetch for it.
    if spec.is_terminal:
        with contextlib.suppress(OSError), paths.spec_lock(spec_path):
            fresh = JobSpec.read(spec_path)
            if fresh.is_terminal:
                fresh.last_fetched_at = utcnow_iso()
                fresh.write(spec_path)
    return dst


def _stream_extract_remote_tar(
    host_cfg: HostConfig, verb: str, jobid: str, output_dir: Path
) -> Path:
    """Stream ``<remote_vq> <verb> <jobid>`` (a tarball on stdout) and extract
    it under ``output_dir/<dest>/``, where ``<dest>`` is the tar's top-level
    directory. Shared by :func:`fetch_remote` (``verb="tar-workspace"``) and
    :func:`fetch_workdir_remote` (``verb="tar-workdir"``).

    ``<dest>`` (v0.5.34) is whatever the streaming tar's top-level dir says —
    the remote emitter sets ``arcname=<name>-<jobid>`` when ``job_name`` is
    set, else ``arcname=<jobid>``. Streaming tar (``mode="r|"``) can't seek
    back, so we learn the dest name from the FIRST member, pre-check
    existence, then extract.

    REMOTE-2/3/4/5/6 hardening (v0.8.17):

    * Transport via :func:`transport.stream_remote_vq` — ``_ssh_base`` +
      ``shlex.join`` (ConnectTimeout / BatchMode / ServerAlive, and no
      remote-shell word-split of the argv), a concurrently-drained stderr
      (no pipe-buffer deadlock when the remote is chatty on stderr), and
      ssh-exit-255-vs-real-remote-rc discipline.
    * **Temp-then-rename**: members extract into a hidden staging dir on the
      SAME filesystem as ``output_dir``; only on full success is the
      top-level dir ``os.replace``-d into place. A mid-stream failure (the
      remote dies, the tar truncates, the disk fills) therefore leaves NO
      half-written ``<dest>/`` for the next fetch / ``vq cleanup --restore``
      to trip over — just a staging dir we ``rmtree`` in ``finally``.
    """
    what = verb.removeprefix("tar-")  # "workspace" / "workdir", for messages
    output_dir.mkdir(parents=True, exist_ok=True)
    staging = Path(tempfile.mkdtemp(dir=output_dir, prefix=".vq-fetch-"))
    dst: Path | None = None
    stream: transport.RemoteStream | None = None
    try:
        with (
            transport.stream_remote_vq(host_cfg, verb, jobid) as stream,
            tarfile.open(fileobj=stream.stdout, mode="r|") as tf,
        ):
            for member in tf:
                if dst is None:
                    # Top-level is the part before the first '/'. For the
                    # typical ``tf.add(dir, arcname=NAME)`` shape the first
                    # member is just NAME (a dir entry).
                    top_level = member.name.split("/", 1)[0]
                    dst = output_dir / top_level
                    if dst.exists():
                        raise FileExistsError(
                            f"destination already exists: {dst} "
                            "(remove it or pick another -o)"
                        )
                tf.extract(member, staging, filter="data")
        # stream_remote_vq.__exit__ has run: a non-zero remote / ssh exit
        # already raised RemoteError, and stream.stderr_text is fully drained.
        if dst is None:
            raise transport.RemoteError(
                f"remote {verb} produced an empty tarball (job {jobid}); "
                f"remote stderr: {stream.stderr_text or '(empty)'}"
            )
        # Atomic promote: staging/<dest> -> output_dir/<dest>. Same
        # filesystem (staging is under output_dir), so os.replace is atomic.
        os.replace(staging / dst.name, dst)
        return dst
    except tarfile.TarError as e:
        # A truncated / malformed stream — usually the remote vq errored and
        # wrote a partial (or no) tarball. stream_remote_vq drained the
        # remote stderr into stream.stderr_text as the exception propagated
        # out of its ``with`` block.
        remote_stderr = stream.stderr_text if stream is not None else ""
        raise transport.RemoteError(
            f"failed to extract remote {what} tarball for {jobid}: {e}; "
            f"remote stderr: {remote_stderr or '(empty)'}"
        ) from e
    finally:
        shutil.rmtree(staging, ignore_errors=True)


def fetch_remote(host_cfg: HostConfig, jobid: str, output_dir: Path) -> Path:
    """Stream the remote *workspace* (``spec.cwd``) for ``jobid`` into
    ``output_dir/<dest>/`` via the ``vq tar-workspace`` verb. See
    :func:`_stream_extract_remote_tar` for the transport + extraction
    contract."""
    return _stream_extract_remote_tar(host_cfg, "tar-workspace", jobid, output_dir)


def _tar_spec_path(jobid: str, *, multi_user: bool) -> Path:
    if multi_user:
        return paths.resolve_spec_path(jobid, multi_user=True)
    spec_path = paths.spec_path(jobid)
    if not spec_path.exists():
        raise FileNotFoundError(f"no such job: {jobid}")
    return spec_path


def emit_workspace_tar(jobid: str, *, multi_user: bool = False) -> None:
    """Write the local workspace tarball for ``jobid`` to ``sys.stdout.buffer``.

    Used by the internal ``vq tar-workspace`` verb. Single-file functions
    like this stay separate from :func:`fetch_local` because the latter
    writes to a directory and this one writes a stream.

    Archive-aware (v0.5.10.1+): when the spec has ``archived_at`` set, the
    workspace dir is gone but the tarball lives at ``spec.archive_path``.
    We stream those bytes directly to stdout. The receiving end
    (``fetch_remote``) opens with ``tarfile.open(mode="r|")`` which
    autodetects compression, so the bz2-compressed archive bytes flow
    through unchanged.
    """
    spec_path = _tar_spec_path(jobid, multi_user=multi_user)
    spec = JobSpec.read(spec_path)
    if spec.is_archived and spec.archive_path:
        archive = Path(spec.archive_path)
        if not archive.is_file():
            raise FileNotFoundError(
                f"archive for job {jobid} not found at {archive} "
                "(record exists in spec but file is gone)"
            )
        _stream_archive_with_diagnosis(archive, spec)
        return
    _refresh_live_scheduler_workspace(spec)
    src = Path(spec.cwd)
    if not src.is_dir():
        raise FileNotFoundError(f"workspace for job {jobid} not found at {src}")
    # v0.5.34: arcname is the spec's ``dest_dirname`` (= ``<name>-<jobid>``
    # when job_name is set, else just ``<jobid>``). The streaming tar's
    # top-level directory becomes the receiver's destination directory
    # name — see ``fetch_remote``'s peek-first-member logic.
    with tarfile.open(fileobj=sys.stdout.buffer, mode="w|") as tf:
        tf.add(src, arcname=spec.dest_dirname)
        _add_diagnosis_sidecar_to_tar(
            tf,
            arcname_root=spec.dest_dirname,
            spec=spec,
        )


# ----------------------------------------------------------------------
# v0.7.7 *Cerf's Datagram* — workdir variants
# ----------------------------------------------------------------------
#
# `spec.workdir` is the per-job scratch dir the daemon materializes at
# dispatch (v0.6.54). The job sees it as ``$VQ_WORKDIR`` and is
# expected to dump intermediate / large artefacts there. Operators
# need a way to pull that content back to the laptop without ssh'ing
# in and reading the workdir path out of `vq status`.
#
# The implementation mirrors workspace fetch closely — same streaming-
# tar pipeline, same per-host helper split, same destination naming
# rule with ``-workdir`` appended. The few differences:
#
# * No archive path. Workdirs are not archived by `vq cleanup
#   --archive` (which only touches the workspace + spec). If a workdir
#   has been swept by `clean_workdir_on_terminal=True` or by the
#   daemon's auto-cleanup pass, the fetch surfaces a precise "workdir
#   for job X was cleaned up at <time>" error rather than letting
#   `FileNotFoundError` bubble up cryptically.
# * `spec.workdir` is `Optional[str]` — pre-v0.6.54 specs and jobs
#   submitted with ``--no-workdir`` (a future opt-out) won't have one
#   to fetch. The error message names the gap explicitly.
# * Destination dir is ``<dest_dirname>-workdir`` so a workspace fetch
#   and a workdir fetch of the same job can coexist under one
#   ``-o DIR``.


def _workdir_dest_name(spec: JobSpec) -> str:
    """v0.7.7: pick the destination directory name for a workdir fetch.

    Same shape as :py:attr:`JobSpec.dest_dirname` but with
    ``-workdir`` appended so the two payloads (workspace + workdir) of
    the same job land in distinct sibling directories under the
    operator's ``-o DIR``.
    """
    return f"{spec.dest_dirname}-workdir"


def _workdir_missing_hint(spec: JobSpec) -> str:
    """Explain WHY a job's workdir directory is gone, for the 'not found'
    errors below.

    CLEAN-3: the daemon's auto-cleanup *age-sweep* and the opt-in
    ``--clean-tmp`` immediate cleanup are different things, and the previous
    hint blamed ``--clean-tmp`` for both — misdirecting an operator whose
    workdir was simply swept for age (they never passed ``--clean-tmp``). The
    sweep now stamps ``workdir_swept_at``, so we can name the real cause.
    """
    if spec.workdir_swept_at:
        return (
            " (the daemon's auto-cleanup age-sweep removed this workdir at "
            f"{spec.workdir_swept_at} — workdirs are swept once their job has "
            "been terminal longer than the configured workdir_max_age; fetch "
            "sooner, or raise / disable that retention)"
        )
    if spec.is_terminal and spec.clean_workdir_on_terminal:
        return (
            " (workdir was swept on terminal — the job was submitted with "
            "--clean-tmp; re-submit without it to keep the workdir fetchable)"
        )
    return ""


def _no_workdir_message(spec: JobSpec, jobid: str) -> str:
    """Explain the workspace-only contract for jobs without a workdir field."""
    if spec.scheduler_target:
        return (
            f"job {jobid} is scheduler-backed for {spec.scheduler_target}; "
            "scheduler jobs are workspace-only in vq. Use "
            f"`vq fetch {spec.scheduler_target} {jobid} -o DIR` without "
            "--workdir to copy the preserved scheduler workspace, including "
            "stdout, stderr, _vq markers, and generated output files."
        )
    return (
        f"job {jobid} has no workdir (pre-v0.6.54 spec or --no-workdir "
        "submit); only the workspace is fetchable with "
        f"`vq fetch {jobid} -o DIR`"
    )


def fetch_workdir_local(
    jobid: str, output_dir: Path, *, multi_user: bool = False
) -> Path:
    """v0.7.7: copy the local workdir for ``jobid`` into
    ``output_dir/<dest_dirname>-workdir/``.

    Returns the path of the copied workdir. Raises
    :class:`FileNotFoundError` if the job isn't in the local queue,
    if the spec has no workdir field, or if the workdir directory is
    gone (cleaned up by ``clean_workdir_on_terminal`` or the daemon's
    auto-cleanup sweep). Raises :class:`FileExistsError` if the
    destination directory already exists.
    """
    if multi_user:
        spec_path = paths.resolve_spec_path(jobid, multi_user=True)
    else:
        spec_path = paths.spec_path(jobid)
        if not spec_path.exists():
            raise FileNotFoundError(f"no such job: {jobid}")
    spec = JobSpec.read(spec_path)
    # v0.6.x: multi-user ownership check — same as workspace fetch.
    check_owner(spec)
    if not spec.workdir:
        raise FileNotFoundError(_no_workdir_message(spec, jobid))
    src = Path(spec.workdir)
    if not src.is_dir():
        # Be explicit about WHY the directory is gone (CLEAN-3): the age-sweep
        # and --clean-tmp are different causes; the hint helper names the real
        # one from workdir_swept_at / clean_workdir_on_terminal.
        raise FileNotFoundError(
            f"workdir for job {jobid} not found at {src}"
            f"{_workdir_missing_hint(spec)}"
        )
    output_dir.mkdir(parents=True, exist_ok=True)
    dst = output_dir / _workdir_dest_name(spec)
    if dst.exists():
        raise FileExistsError(
            f"destination already exists: {dst} (remove it or pick another -o)"
        )
    shutil.copytree(src, dst)
    _write_diagnosis_sidecar(dst, spec)
    return dst


def fetch_workdir_remote(
    host_cfg: HostConfig, jobid: str, output_dir: Path
) -> Path:
    """v0.7.7: stream the remote *workdir* (``spec.workdir``, the per-job
    scratch dir) for ``jobid`` into ``output_dir/<dest>/`` via the
    ``vq tar-workdir`` verb. See :func:`_stream_extract_remote_tar` for the
    transport + extraction contract."""
    return _stream_extract_remote_tar(host_cfg, "tar-workdir", jobid, output_dir)


def emit_workdir_tar(jobid: str, *, multi_user: bool = False) -> None:
    """v0.7.7: write the local workdir tarball for ``jobid`` to
    ``sys.stdout.buffer``.

    Backs the internal ``vq tar-workdir`` verb. Unlike
    :func:`emit_workspace_tar`, there is no archive-aware path —
    workdirs aren't archived by ``vq cleanup --archive`` (which only
    touches the workspace + spec). If the workdir has been swept
    (via ``clean_workdir_on_terminal`` or the daemon's auto-cleanup
    pass), we surface a precise error so the operator knows why.
    """
    spec_path = _tar_spec_path(jobid, multi_user=multi_user)
    spec = JobSpec.read(spec_path)
    if not spec.workdir:
        raise FileNotFoundError(_no_workdir_message(spec, jobid))
    src = Path(spec.workdir)
    if not src.is_dir():
        raise FileNotFoundError(
            f"workdir for job {jobid} not found at {src}"
            f"{_workdir_missing_hint(spec)}"  # CLEAN-3: name the real cause
        )
    # arcname mirrors `_workdir_dest_name(spec)` so the receiver's
    # peek-first-member logic lands on a name with the ``-workdir``
    # suffix.
    with tarfile.open(fileobj=sys.stdout.buffer, mode="w|") as tf:
        tf.add(src, arcname=_workdir_dest_name(spec))
        _add_diagnosis_sidecar_to_tar(
            tf,
            arcname_root=_workdir_dest_name(spec),
            spec=spec,
        )


# ----------------------------------------------------------------------
# v0.12.0 *Hollerith's Return* — bulk fetch-back (`vq fetch-all`)
# ----------------------------------------------------------------------
#
# `vq fetch-all [HOST]` pulls EVERY terminal job's workspace back in one
# shot, so an operator who fired a fleet of jobs (an `--array` sweep, a
# batch of AICCM runs on mars / maru) gets all the outputs back with a
# single command from the submitting folder, instead of one
# `vq fetch HOST JOBID` per job. In a PBS-style "everything runs in a
# copied scratch dir" world this is how the generated files (vibe-qc
# `.out` / `.system`, CRYSTAL, ORCA) find their way back to the user's
# directory.
#
# Idempotence is by destination existence, not a spec flag: `fetch_local`
# and `fetch_remote` already refuse to overwrite an existing
# `<output_dir>/<dest>/` (FileExistsError), so re-running `vq fetch-all`
# against the same `-o` only pulls jobs whose output isn't there yet. That
# key is uniform across local and remote, where `last_fetched_at` is not:
# a remote fetch streams a tarball and never writes the remote spec, so
# only the local copy path stamps it. Directory existence is the honest
# cross-host signal.


@dataclass(frozen=True)
class BulkFetchResult:
    """One job's outcome in a `vq fetch-all` run.

    ``outcome`` is one of ``"fetched"`` (``detail`` is the destination
    path), ``"skipped"`` (``detail`` is why, e.g. already present), or
    ``"error"`` (``detail`` is the message). Non-terminal and
    state-filtered-out jobs are dropped before a result is recorded, so
    they never show up as a row.

    ``state`` is the job's terminal state and ``failure_hint`` is the
    first line of a failed job's stderr tail (None for a COMPLETED job),
    so a bulk-sweep summary can flag inline WHICH fetched jobs died and
    why, closing the loop with the crash-feedback field.
    """

    jobid: str
    job_name: str | None
    outcome: str
    detail: str
    state: str | None = None
    failure_hint: str | None = None


def bulk_fetch(
    specs: list[JobSpec],
    *,
    states: set[str] | None,
    fetch_one: Callable[[str], Path],
) -> list[BulkFetchResult]:
    """Fetch every terminal spec via ``fetch_one(jobid) -> Path``.

    ``fetch_one`` is bound to the local or remote primitive by the caller
    (``fetch_local`` / ``fetch_remote`` with ``output_dir`` already closed
    over), which keeps this loop transport-agnostic and unit-testable with
    a fake.

    Filtering: a non-terminal spec is dropped silently (an active job has
    nothing to fetch yet, which is not an error). When ``states`` is given,
    only those terminal states pass, and ``None`` means every terminal
    state.

    Each job's fetch is isolated so one bad job never aborts the sweep: a
    FileExistsError becomes a ``"skipped"`` (already present, the
    idempotence path), and a missing workspace or a remote-transport
    failure becomes an ``"error"`` row the caller can report and move past.
    """
    results: list[BulkFetchResult] = []
    for spec in specs:
        if not spec.is_terminal:
            continue
        if states is not None and spec.state.value not in states:
            continue
        st = spec.state.value
        # First line of the crash tail for a non-COMPLETED terminal, capped
        # so the summary stays one line per job. COMPLETED jobs and pre-
        # v0.12.0 specs (no failure_tail) carry no hint.
        hint = None
        if st != "completed" and spec.failure_tail:
            hint = spec.failure_tail.splitlines()[0][:120]
        elif st != "completed" and spec.exit_code is not None:
            # No stderr to tail (a hard SIGKILL/OOM or a segfault often leaves
            # none): fall back to decoding the signal from the exit code, so
            # the sweep still names the crash instead of a bare [FAILED].
            decoded = describe_exit_code(spec.exit_code)
            if decoded != str(spec.exit_code):
                hint = decoded
        try:
            dst = fetch_one(spec.id)
            results.append(
                BulkFetchResult(spec.id, spec.job_name, "fetched", str(dst), st, hint)
            )
        except FileExistsError:
            results.append(
                BulkFetchResult(
                    spec.id, spec.job_name, "skipped", "already present", st, hint
                )
            )
        except (FileNotFoundError, transport.RemoteError) as e:
            results.append(
                BulkFetchResult(spec.id, spec.job_name, "error", str(e), st, hint)
            )
    return results