#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
cryptostruct-dl — the official download tool for CryptoStruct market-data orders.

Single file, Python >= 3.9, standard library only.

    python3 cryptostruct_dl.py --order ORDER_ID --token T   [options]
    python3 cryptostruct_dl.py --key cs_dl_...               [options]
    python3 cryptostruct_dl.py --manifest FILE_OR_URL [--token T] [--key K] [options]

Sections (in file order):
    Helpers · Http · Manifest · Ledger · Planner · Poller · Bundles · Downloader · Verify · Report · CLI

Exit codes: 0 everything complete · 1 incomplete (failed / cold / left pending) ·
2 usage error, manifest 404, expired link (410) · 3 interrupted (Ctrl-C).

Large orders (1.1): a manifest with a `page` block (or an account manifest with
`large_orders`) is walked page by page — bounded memory, a SQLite ledger, files
that are not ready within --page-wait-min are left for the next run.

Direct downloads (1.2): every request carries `X-CryptoStruct-Direct: 1`, so a
single file may answer 302 to a short-lived storage URL and the bytes come
straight from storage. `--key` travels as an unredirected header — it never
leaves the CryptoStruct host (storage rejects a second credential anyway).

Partial delivery per day (1.4): a range chunk whose day makes no progress for
--stuck-day-min (or that turns out cold without --restore) is split — every day
already complete in the partial tar is unpacked into <exchange>/<bundle>/<date>/,
the rest comes one day at a time, the stuck day last.
"""

import argparse
import base64
import concurrent.futures
import datetime as _dt
import email.utils
import hashlib
import http.client
import json
import os
import random
import re
import shutil
import socket
import subprocess
import sys
import tarfile
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
import zipfile

try:  # stdlib, but missing from some minimal Python builds — only large orders need it
    import sqlite3
except ImportError:  # pragma: no cover
    sqlite3 = None

__version__ = "1.4.0"
USER_AGENT = "cryptostruct-dl/" + __version__
# Opt-in for single-file 302s to storage (see Http): we follow them, never with the Bearer.
DIRECT_HEADER = "X-CryptoStruct-Direct"

DEFAULT_BASE = "https://cryptostruct.com"
SCHEMA_ID = "cryptostruct.download-manifest/1"
LEDGER_SCHEMA = "cryptostruct-dl.ledger/1"
LEDGER_DIR = ".cryptostruct-dl"
LEDGER_FILE = "ledger.json"
LEDGER_SQLITE = "ledger.sqlite"          # large orders (paged manifests) — one row per file
SIDECAR_NAME = "cryptostruct-dl.sha256"
PART_SUFFIX = ".part"
META_SUFFIX = ".part.json"

FORMAT_TOKENS = ("trades.csv.gz", "bbo.csv.gz", "liquidations.csv.gz", "trades.parquet", "liquidations.parquet")

CHUNK_SIZE = 1024 * 1024          # bytes per read while streaming
JSON_BODY_LIMIT = 8 * 1024 * 1024  # never read more than this for a JSON/NDJSON body
HTTP_TIMEOUT_S = 60.0              # socket connect/read timeout
MAX_TRANSIENT_TRIES = 8            # 5xx / network errors per item
BACKOFF_BASE_S = 5.0
BACKOFF_CAP_S = 120.0
DEFAULT_RETRY_AFTER_S = 30.0       # when a pending answer carries no Retry-After
MIN_POLL_S = 1.0
MAX_POLL_S = 900.0
ZIP_PARALLEL = 2                   # bundle days in flight at once (zips, or direct member lists)
DIRECT_PARALLEL = 8                # member files of one direct bundle day in flight at once
DIRECT_REFRESH_MAX = 3             # fresh URL lists per member before giving up (storage URLs expire)
SHUTDOWN_GRACE_S = 10.0
DEFAULT_PAGE_WAIT_MIN = 2.0        # paged mode: per-file wait before it is left for the next run
DEFAULT_RESTORE_MAX = 500          # paged mode: --restore requests per run
DEFAULT_STUCK_DAY_MIN = 15.0       # range chunk: minutes without progress on one day before it is split into days
SALVAGE_DISK_MARGIN = 256 * 1024 * 1024  # free bytes kept on top of the partial tar when unpacking it
SALVAGE_DIR_PREFIX = ".salvage-"   # staging dir of a day unpacked from a partial range tar
REPORT_MAX_ROWS = 1000             # paged mode: problem rows kept for the report
# Tests scale every sleep down to zero; production keeps 1.0.
SLEEP_SCALE = float(os.environ.get("CRYPTOSTRUCT_DL_SLEEP_SCALE", "1") or 1)
ZSTD_BIN = shutil.which("zstd")

DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$")
DATE_IN_NAME_RE = re.compile(r"(\d{4}-\d{2}-\d{2})")
MANIFEST_LINE_RE = re.compile(r"^(\d+)\t(.+)$")   # "<size>\t<member>" lines of a day's MANIFEST.txt


# ───────────────────────────── Helpers ─────────────────────────────


class DlError(Exception):
    """Base class for everything the tool raises on purpose."""


class UsageError(DlError):
    """Wrong invocation, bad manifest, manifest 404/403 → exit 2."""


class ManifestUnreachable(DlError):
    """Manifest could not be fetched (network/5xx after retries) → exit 1."""


class ExpiredLink(DlError):
    """HTTP 410 anywhere → stop everything → exit 2."""


class ItemFailed(DlError):
    """Permanent per-item failure (404/403/verify/…) → item failed, run continues."""


class ItemCold(DlError):
    """Item sits in cold storage and --restore was not given."""


class ItemPending(DlError):
    """Waited longer than --max-wait-min on one item."""


class Interrupted(DlError):
    """Ctrl-C (or a fatal error elsewhere) asked this worker to stop."""


class TransientError(DlError):
    """Network error / 5xx — retried with backoff by the caller."""


class VerifyError(DlError):
    """A completed file did not pass verification."""


class SplitChunk(DlError):
    """413 RANGE_TOO_LARGE: split the chunk at `suggested_to`."""

    def __init__(self, suggested_to):
        DlError.__init__(self, "range too large, split at %s" % suggested_to)
        self.suggested_to = suggested_to


class RangeNotFound(DlError):
    """404 on the range route → fall back to per-day zips."""


class StuckRange(DlError):
    """A range chunk waits on one day that made no progress for --stuck-day-min → split into days."""

    def __init__(self, date, waited_s):
        DlError.__init__(self, "no progress on %s for %s" % (date or "the range", human_duration(waited_s)))
        self.date = date
        self.waited_s = waited_s


class UseTar(DlError):
    """413 TOO_LARGE on a day zip → use the day's tar_url."""


def utc_now():
    return _dt.datetime.now(_dt.timezone.utc)


def utc_now_iso():
    return utc_now().strftime("%Y-%m-%dT%H:%M:%SZ")


def human_bytes(n):
    if n is None:
        return "?"
    try:
        f = float(n)
    except (TypeError, ValueError):
        return "?"
    units = ("B", "KB", "MB", "GB", "TB", "PB")
    i = 0
    while f >= 1000.0 and i < len(units) - 1:
        f /= 1000.0
        i += 1
    if i == 0:
        return "%d B" % int(f)
    return "%.1f %s" % (f, units[i])


def human_duration(seconds):
    s = int(round(max(0.0, float(seconds))))
    if s < 60:
        return "%d s" % s
    if s < 3600:
        return "%dm %02ds" % (s // 60, s % 60)
    return "%dh %02dm" % (s // 3600, (s % 3600) // 60)


def parse_date(value, what):
    if value is None:
        return None
    if not DATE_RE.match(str(value)):
        raise UsageError("%s must be YYYY-MM-DD, got %r" % (what, value))
    try:
        return _dt.date.fromisoformat(str(value))
    except ValueError:
        raise UsageError("%s is not a valid date: %r" % (what, value))


def next_day(iso_date):
    return (_dt.date.fromisoformat(iso_date) + _dt.timedelta(days=1)).isoformat()


def days_between(iso_from, iso_to):
    a = _dt.date.fromisoformat(iso_from)
    b = _dt.date.fromisoformat(iso_to)
    return (b - a).days + 1


def safe_component(value):
    """One path component: no separators, no traversal, no control characters."""
    s = str(value if value is not None else "")
    s = s.replace("\\", "_").replace("/", "_")
    s = re.sub(r"[\x00-\x1f\x7f]", "", s).strip()
    if s in ("", ".", ".."):
        return "_"
    return s


def add_query(url, **params):
    """Append query parameters that are not present yet."""
    parts = urllib.parse.urlsplit(url)
    q = urllib.parse.parse_qsl(parts.query, keep_blank_values=True)
    present = set(k for k, _ in q)
    for k, v in params.items():
        if k not in present:
            q.append((k, str(v)))
    return urllib.parse.urlunsplit((parts.scheme, parts.netloc, parts.path, urllib.parse.urlencode(q), parts.fragment))


def set_query(url, **params):
    """Replace (or add) query parameters."""
    parts = urllib.parse.urlsplit(url)
    q = [(k, v) for k, v in urllib.parse.parse_qsl(parts.query, keep_blank_values=True) if k not in params]
    for k, v in params.items():
        q.append((k, str(v)))
    return urllib.parse.urlunsplit((parts.scheme, parts.netloc, parts.path, urllib.parse.urlencode(q), parts.fragment))


def query_values(url, name):
    parts = urllib.parse.urlsplit(url)
    return [v for k, v in urllib.parse.parse_qsl(parts.query, keep_blank_values=True) if k == name]


def looks_like_url(s):
    return bool(re.match(r"^https?://", str(s), re.I))


def parse_retry_after(value):
    """Retry-After as seconds — accepts the delta-seconds and the HTTP-date form."""
    if value is None:
        return None
    v = str(value).strip()
    if not v:
        return None
    if re.fullmatch(r"\d+(\.\d+)?", v):
        return max(0.0, float(v))
    try:
        dt = email.utils.parsedate_to_datetime(v)
    except (TypeError, ValueError, IndexError):
        return None
    if dt is None:
        return None
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=_dt.timezone.utc)
    return max(0.0, (dt - utc_now()).total_seconds())


def parse_repr_digest(value):
    """`Repr-Digest: sha-256=:<base64>:` → lowercase hex, or None."""
    if not value:
        return None
    m = re.search(r"sha-256=:([A-Za-z0-9+/=]+):", str(value))
    if not m:
        return None
    try:
        raw = base64.b64decode(m.group(1), validate=True)
    except (ValueError, TypeError):
        return None
    if len(raw) != 32:
        return None
    return raw.hex()


def parse_content_range(value):
    """`bytes START-END/TOTAL` or `bytes */TOTAL` → (start, end, total); None where unknown."""
    if not value:
        return None
    m = re.match(r"\s*bytes\s+(\*|\d+-\d+)/(\*|\d+)\s*$", str(value))
    if not m:
        return None
    start = end = total = None
    if m.group(1) != "*":
        a, b = m.group(1).split("-")
        start, end = int(a), int(b)
    if m.group(2) != "*":
        total = int(m.group(2))
    return (start, end, total)


def parse_content_disposition_filename(value):
    if not value:
        return None
    v = str(value)
    m = re.search(r"filename\*\s*=\s*(?:UTF-8|utf-8)''([^;]+)", v)
    if m:
        name = urllib.parse.unquote(m.group(1).strip())
    else:
        m = re.search(r'filename\s*=\s*"((?:[^"\\]|\\.)*)"', v)
        if m:
            name = m.group(1).replace('\\"', '"')
        else:
            m = re.search(r"filename\s*=\s*([^;]+)", v)
            if not m:
                return None
            name = m.group(1).strip()
    name = safe_component(name)
    return name if name != "_" else None


def parse_ndjson_summary(text):
    """Last `{summary: …}` line of an NDJSON body, or None."""
    lines = [ln for ln in str(text).splitlines() if ln.strip()]
    for ln in reversed(lines):
        try:
            obj = json.loads(ln)
        except ValueError:
            continue
        if isinstance(obj, dict) and isinstance(obj.get("summary"), dict):
            return obj["summary"]
    return None


def first_unready_date(text):
    """First day line of a range status (NDJSON) that is not ready: pending/cold members or state=error."""
    for ln in str(text).splitlines():
        try:
            obj = json.loads(ln)
        except ValueError:
            continue
        if not isinstance(obj, dict) or not obj.get("date"):
            continue
        counts = obj.get("counts") if isinstance(obj.get("counts"), dict) else {}
        if obj.get("state") == "error" or as_int(counts.get("pending")) > 0 or as_int(counts.get("cold")) > 0:
            return str(obj["date"])
    return None


def backoff_seconds(attempt):
    """Exponential 5 s → 120 s with ±25 % jitter; attempt starts at 1."""
    base = min(BACKOFF_CAP_S, BACKOFF_BASE_S * (2 ** max(0, attempt - 1)))
    return base * random.uniform(0.75, 1.25)


def as_int(value, default=0):
    try:
        return int(value)
    except (TypeError, ValueError):
        return default


def sha256_file(path):
    h = hashlib.sha256()
    with open(path, "rb") as fh:
        while True:
            chunk = fh.read(CHUNK_SIZE)
            if not chunk:
                break
            h.update(chunk)
    return h.hexdigest()


def read_json_file(path):
    with open(path, "r", encoding="utf-8") as fh:
        return json.load(fh)


def write_json_atomic(path, obj):
    d = os.path.dirname(path)
    if d:
        os.makedirs(d, exist_ok=True)
    tmp = "%s.tmp.%d" % (path, os.getpid())
    with open(tmp, "w", encoding="utf-8") as fh:
        json.dump(obj, fh, indent=1, sort_keys=True)
        fh.write("\n")
    os.replace(tmp, path)


def read_part_meta(part_path):
    meta_path = part_path[: -len(PART_SUFFIX)] + META_SUFFIX if part_path.endswith(PART_SUFFIX) else part_path + ".json"
    try:
        obj = read_json_file(meta_path)
        return obj if isinstance(obj, dict) else {}
    except (OSError, ValueError):
        return {}


def write_part_meta(part_path, meta):
    meta_path = part_path[: -len(PART_SUFFIX)] + META_SUFFIX if part_path.endswith(PART_SUFFIX) else part_path + ".json"
    try:
        write_json_atomic(meta_path, meta)
    except OSError:
        pass


def remove_quiet(path):
    try:
        os.remove(path)
    except OSError:
        pass


def meta_path_for(part_path):
    return part_path[: -len(PART_SUFFIX)] + META_SUFFIX


class Masker:
    """Never let a token or key reach a log line."""

    def __init__(self):
        self._secrets = []

    def add(self, secret):
        if not secret:
            return
        s = str(secret)
        if len(s) < 8:
            return
        for variant in (s, urllib.parse.quote(s, safe=""), urllib.parse.quote_plus(s)):
            if variant not in self._secrets:
                self._secrets.append(variant)
        self._secrets.sort(key=len, reverse=True)

    @staticmethod
    def short(secret):
        s = str(secret)
        if s.startswith("cs_dl_"):
            body = s[6:]
            return "cs_dl_%s…%s" % (body[:4], body[-4:])
        return "%s…%s" % (s[:4], s[-4:])

    def mask(self, text):
        out = str(text)
        for s in self._secrets:
            if s in out:
                out = out.replace(s, self.short(s))
        return out


class Log:
    """Progress → stderr. Human report / plan → stdout (stderr in --json mode)."""

    def __init__(self, masker, json_mode=False):
        self.m = masker
        self.json_mode = json_mode
        self._lock = threading.Lock()

    def _emit(self, stream, msg):
        text = self.m.mask(msg)
        with self._lock:
            try:
                stream.write(text + "\n")
                stream.flush()
            except (OSError, ValueError):
                pass

    def progress(self, msg):
        self._emit(sys.stderr, msg)

    def warn(self, msg):
        self._emit(sys.stderr, "warning: " + str(msg))

    def error(self, msg):
        self._emit(sys.stderr, "error: " + str(msg))

    def out(self, msg):
        self._emit(sys.stderr if self.json_mode else sys.stdout, msg)


class Clock:
    """All waiting goes through here so Ctrl-C wakes every worker and tests can run at zero delay."""

    def __init__(self, stop_event):
        self.stop = stop_event

    def sleep(self, seconds):
        if self.stop.is_set():
            raise Interrupted()
        if seconds <= 0:
            return
        if self.stop.wait(seconds * SLEEP_SCALE):
            raise Interrupted()


# ───────────────────────────── Http ─────────────────────────────


class HttpResponse:
    def __init__(self, status, headers, fp, url):
        self.status = int(status)
        self.headers = headers
        self._fp = fp
        self.url = url
        self._body = None

    def header(self, name, default=None):
        if self.headers is None:
            return default
        v = self.headers.get(name)
        return default if v is None else v

    def content_length(self):
        v = self.header("Content-Length")
        if v is None:
            return None
        try:
            return int(v)
        except ValueError:
            return None

    def retry_after(self):
        return parse_retry_after(self.header("Retry-After"))

    def read(self, limit=JSON_BODY_LIMIT):
        if self._body is None:
            try:
                self._body = self._fp.read(limit) if self._fp is not None else b""
            except (OSError, http.client.HTTPException):
                self._body = b""
        return self._body

    def text(self):
        return self.read().decode("utf-8", "replace")

    def json(self):
        try:
            return json.loads(self.text())
        except ValueError:
            return None

    def error_code(self):
        j = self.json()
        if isinstance(j, dict):
            err = j.get("error")
            if isinstance(err, dict):
                return err.get("code")
            if isinstance(err, str):
                return err
            return j.get("code")
        return None

    def error_obj(self):
        j = self.json()
        if isinstance(j, dict):
            err = j.get("error")
            if isinstance(err, dict):
                return err
            return j
        return {}

    def iter_chunks(self, size=CHUNK_SIZE):
        while True:
            chunk = self._fp.read(size)
            if not chunk:
                break
            yield chunk

    def close(self):
        try:
            if self._fp is not None:
                self._fp.close()
        except (OSError, ValueError):
            pass


class Http:
    """urllib opener with the tool's User-Agent on every request, optional Bearer.

    The Bearer is an UNREDIRECTED header: urllib copies ordinary headers onto a
    redirected request, and a single file may 302 to a presigned storage URL —
    the key must never reach that host (it answers 400 to a second credential).
    `X-CryptoStruct-Direct: 1` tells the shop we follow redirects that way."""

    def __init__(self, masker, bearer=None, timeout=None):
        self.masker = masker
        self.bearer = bearer
        self.timeout = timeout
        self.opener = urllib.request.build_opener()
        self.opener.addheaders = []

    def request(self, method, url, headers=None, auth=True):
        """`auth=False` for a storage URL out of a member list: no Bearer (storage answers
        400 to a second credential) and no opt-in header — those belong to the shop only."""
        req = urllib.request.Request(url, method=method)
        req.add_header("User-Agent", USER_AGENT)
        req.add_header("Accept", "*/*")
        if auth:
            req.add_header(DIRECT_HEADER, "1")
            if self.bearer:
                req.add_unredirected_header("Authorization", "Bearer " + self.bearer)
        for k, v in (headers or {}).items():
            req.add_header(k, v)
        timeout = self.timeout if self.timeout is not None else HTTP_TIMEOUT_S
        try:
            raw = self.opener.open(req, timeout=timeout)
            return HttpResponse(raw.status, raw.headers, raw, url)
        except urllib.error.HTTPError as e:
            return HttpResponse(e.code, e.headers, e, url)
        except (urllib.error.URLError, socket.timeout, http.client.HTTPException, ConnectionError, OSError) as e:
            raise TransientError(self.masker.mask("%s: %s" % (type(e).__name__, e)))
        except ValueError as e:
            raise UsageError(self.masker.mask("bad URL: %s" % e))


# ───────────────────────────── Manifest ─────────────────────────────


class Manifest:
    def __init__(self, data, source):
        if not isinstance(data, dict):
            raise UsageError("manifest is not a JSON object")
        if data.get("schema") != SCHEMA_ID:
            raise UsageError("unsupported manifest schema %r (expected %s)" % (data.get("schema"), SCHEMA_ID))
        self.data = data
        self.source = source
        self.scope = data.get("scope") or "order"
        self.site = data.get("site")
        self.orders = [o for o in (data.get("orders") or []) if isinstance(o, dict)]
        self.items = [i for i in (data.get("items") or []) if isinstance(i, dict)]
        self.bundles = [b for b in (data.get("bundles") or []) if isinstance(b, dict)]
        self.totals = data.get("totals") if isinstance(data.get("totals"), dict) else {}
        self.contract = data.get("contract") if isinstance(data.get("contract"), dict) else {}
        self.truncated = bool(data.get("truncated"))
        self.new_only = bool(data.get("new_only"))
        # new_only manifests leave downloaded-before items out and only count them here
        self.omitted = data.get("omitted_downloaded") if isinstance(data.get("omitted_downloaded"), dict) else {}
        # Large orders: one page of the order (follow page["next"]) / account manifests list them by reference
        self.page = data.get("page") if isinstance(data.get("page"), dict) else None
        self.large_orders = [o for o in (data.get("large_orders") or []) if isinstance(o, dict) and o.get("manifest_url")]

    @property
    def paged(self):
        return self.page is not None or bool(self.large_orders)

    # -- discovery -------------------------------------------------------

    def all_urls(self):
        for it in self.items:
            for k in ("url", "status_url", "restore_url"):
                if it.get(k):
                    yield it[k]
            fm = it.get("formats")
            if isinstance(fm, dict):
                for v in fm.values():
                    if v:
                        yield v
        for b in self.bundles:
            for d in b.get("days") or []:
                for k in ("zip_url", "status_url", "restore_url", "manifest_url", "tar_url"):
                    if isinstance(d, dict) and d.get(k):
                        yield d[k]
            for c in b.get("range_chunks") or []:
                for k in ("url", "status_url", "restore_url", "manifest_url"):
                    if isinstance(c, dict) and c.get(k):
                        yield c[k]

    def secrets(self):
        seen = set()
        for u in self.all_urls():
            for v in query_values(u, "t"):
                if v and v not in seen:
                    seen.add(v)
        return sorted(seen)

    def absolutize(self, base):
        """Make relative URLs absolute against the manifest's site (or --base)."""
        root = self.site or base

        def fix(u):
            if u and not looks_like_url(u):
                return urllib.parse.urljoin(root.rstrip("/") + "/", u.lstrip("/"))
            return u

        for it in self.items:
            for k in ("url", "status_url", "restore_url"):
                if it.get(k):
                    it[k] = fix(it[k])
            fm = it.get("formats")
            if isinstance(fm, dict):
                for tok in list(fm.keys()):
                    fm[tok] = fix(fm[tok])
        for b in self.bundles:
            for d in b.get("days") or []:
                for k in ("zip_url", "status_url", "restore_url", "manifest_url", "tar_url"):
                    if d.get(k):
                        d[k] = fix(d[k])
            for c in b.get("range_chunks") or []:
                for k in ("url", "status_url", "restore_url", "manifest_url"):
                    if c.get(k):
                        c[k] = fix(c[k])

    def with_token(self, token):
        """Attach `t=` to every URL that lacks one (local manifest + --token)."""
        if not token:
            return

        def fix(u):
            return add_query(u, t=token) if (u and not query_values(u, "t")) else u

        for it in self.items:
            for k in ("url", "status_url", "restore_url"):
                if it.get(k):
                    it[k] = fix(it[k])
            fm = it.get("formats")
            if isinstance(fm, dict):
                for tok in list(fm.keys()):
                    fm[tok] = fix(fm[tok])
        for b in self.bundles:
            for d in b.get("days") or []:
                for k in ("zip_url", "status_url", "restore_url", "manifest_url", "tar_url"):
                    if d.get(k):
                        d[k] = fix(d[k])
            for c in b.get("range_chunks") or []:
                for k in ("url", "status_url", "restore_url", "manifest_url"):
                    if c.get(k):
                        c[k] = fix(c[k])

    # -- filters ---------------------------------------------------------

    def apply_filters(self, only=None, date_from=None, date_to=None):
        """In-place restriction to --only / --from / --to.

        A range chunk is kept only when it lies fully inside the window; the days of a
        partially covered chunk are fetched as per-day zips instead (never more than asked)."""

        def in_window(d):
            if not d or not DATE_RE.match(str(d)):
                return True
            dd = _dt.date.fromisoformat(str(d))
            if date_from and dd < date_from:
                return False
            if date_to and dd > date_to:
                return False
            return True

        if only == "bundles":
            self.items = []
        if only == "tick":
            self.bundles = []
        if date_from or date_to:
            self.items = [it for it in self.items if in_window(it.get("date"))]
            kept = []
            for b in self.bundles:
                days = [d for d in (b.get("days") or []) if isinstance(d, dict) and in_window(d.get("date"))]
                chunks = []
                for c in b.get("range_chunks") or []:
                    if not isinstance(c, dict):
                        continue
                    if in_window(c.get("from")) and in_window(c.get("to")):
                        chunks.append(c)
                b = dict(b)
                b["days"] = days
                b["range_chunks"] = chunks
                if days or chunks:
                    kept.append(b)
            self.bundles = kept

    # -- loading ---------------------------------------------------------

    @staticmethod
    def load_file(path):
        try:
            data = read_json_file(path)
        except OSError as e:
            raise UsageError("cannot read manifest file %s: %s" % (path, e))
        except ValueError as e:
            raise UsageError("manifest file %s is not valid JSON: %s" % (path, e))
        return Manifest(data, path)

    @staticmethod
    def fetch(http, url, clock, log, masker):
        """One manifest request; transient trouble is retried with backoff."""
        tries = 0
        while True:
            try:
                resp = http.request("GET", url)
            except TransientError as e:
                tries += 1
                if tries >= MAX_TRANSIENT_TRIES:
                    raise ManifestUnreachable("manifest unreachable after %d tries: %s" % (tries, e))
                delay = backoff_seconds(tries)
                log.warn("manifest request failed (%s) — retrying in %s" % (e, human_duration(delay)))
                clock.sleep(delay)
                continue
            try:
                s = resp.status
                if s == 200:
                    try:
                        data = json.loads(resp.text())
                    except ValueError:
                        raise ManifestUnreachable("manifest response is not JSON")
                    return Manifest(data, masker.mask(url))
                if s == 410:
                    raise ExpiredLink(expired_hint(resp, None))
                if s == 404:
                    raise UsageError("manifest not found (404) — check the order id and the token/key")
                if s in (401, 403):
                    raise UsageError("not allowed to read this manifest (HTTP %d) — check the token/key" % s)
                if s in (429, 503):
                    tries += 1
                    if tries >= MAX_TRANSIENT_TRIES:
                        raise ManifestUnreachable("manifest unavailable (HTTP %d) after %d tries" % (s, tries))
                    wait = resp.retry_after()
                    if wait is None:
                        wait = backoff_seconds(tries)
                    log.warn("manifest: HTTP %d — retrying in %s" % (s, human_duration(wait)))
                    clock.sleep(wait)
                    continue
                if s >= 500:
                    tries += 1
                    if tries >= MAX_TRANSIENT_TRIES:
                        raise ManifestUnreachable("manifest unavailable (HTTP %d) after %d tries" % (s, tries))
                    delay = backoff_seconds(tries)
                    log.warn("manifest: HTTP %d — retrying in %s" % (s, human_duration(delay)))
                    clock.sleep(delay)
                    continue
                raise UsageError("unexpected HTTP %d while fetching the manifest" % s)
            finally:
                resp.close()


def expired_hint(resp, contract):
    """Build the 410 message: 'this download link has expired — open <claim hint> or sign in'."""
    hint = None
    try:
        err = resp.error_obj() if resp is not None else {}
    except Exception:
        err = {}
    top = {}
    try:
        j = resp.json() if resp is not None else None
        top = j if isinstance(j, dict) else {}
    except Exception:
        top = {}
    for src in (err, top):
        for k in ("claim_url", "hint", "claim"):
            v = src.get(k) if isinstance(src, dict) else None
            if isinstance(v, str) and v.strip():
                hint = v.strip()
                break
        if hint:
            break
    if hint is None and isinstance(contract, dict):
        v = contract.get("expired")
        if isinstance(v, str) and looks_like_url(v.strip()):
            hint = v.strip()
    if hint and looks_like_url(hint):
        # never print a token: the claim link carries the guest token the caller already has
        hint = strip_query(hint, "t")
    if hint is None:
        hint = "the order link from your confirmation e-mail"
    return "this download link has expired — open %s or sign in" % hint


def strip_query(url, name):
    parts = urllib.parse.urlsplit(url)
    q = [(k, v) for k, v in urllib.parse.parse_qsl(parts.query, keep_blank_values=True) if k != name]
    return urllib.parse.urlunsplit((parts.scheme, parts.netloc, parts.path, urllib.parse.urlencode(q), parts.fragment))


# ───────────────────────────── Ledger ─────────────────────────────


class Ledger:
    """<out>/.cryptostruct-dl/ledger.json — every completed download, keyed by item."""

    def __init__(self, out_dir, log):
        self.out_dir = out_dir
        self.dir = os.path.join(out_dir, LEDGER_DIR)
        self.path = os.path.join(self.dir, LEDGER_FILE)
        self.log = log
        self.entries = {}
        self._lock = threading.Lock()
        self._dirty = False

    def load(self):
        if not os.path.exists(self.path):
            return
        try:
            data = read_json_file(self.path)
        except (OSError, ValueError) as e:
            self.log.warn("ledger %s unreadable (%s) — starting a fresh one" % (self.path, e))
            return
        entries = data.get("entries") if isinstance(data, dict) else None
        if isinstance(entries, dict):
            self.entries = dict((k, v) for k, v in entries.items() if isinstance(v, dict))

    def get(self, key):
        with self._lock:
            return self.entries.get(key)

    def record(self, key, rel_path, size, sha256, order_id):
        with self._lock:
            self.entries[key] = {
                "key": key,
                "path": rel_path,
                "size": int(size) if size is not None else None,
                "sha256": sha256,
                "at": utc_now_iso(),
                "order_id": order_id,
            }
            self._dirty = True

    def save(self):
        with self._lock:
            if not self._dirty:
                return
            try:
                write_json_atomic(self.path, {"schema": LEDGER_SCHEMA, "updated_at": utc_now_iso(), "entries": self.entries})
                self._dirty = False
            except OSError as e:
                self.log.warn("could not write ledger %s: %s" % (self.path, e))


class Sidecar:
    """<out>/cryptostruct-dl.sha256 in `sha256sum` format (one line per verified file)."""

    def __init__(self, out_dir):
        self.path = os.path.join(out_dir, SIDECAR_NAME)
        self.lines = {}
        self._lock = threading.Lock()
        self._dirty = False
        self._loaded = False

    def _load(self):
        if self._loaded:
            return
        self._loaded = True
        try:
            with open(self.path, "r", encoding="utf-8") as fh:
                for ln in fh:
                    m = re.match(r"^([0-9a-fA-F]{64})\s[ *](.+)$", ln.rstrip("\n"))
                    if m:
                        self.lines[m.group(2)] = m.group(1).lower()
        except OSError:
            pass

    def add(self, rel_path, sha_hex):
        if not sha_hex:
            return
        with self._lock:
            self._load()
            self.lines[rel_path] = sha_hex.lower()
            self._dirty = True

    def save(self):
        with self._lock:
            if not self._dirty:
                return
            tmp = self.path + ".tmp"
            try:
                with open(tmp, "w", encoding="utf-8") as fh:
                    for rel in sorted(self.lines):
                        fh.write("%s  %s\n" % (self.lines[rel], rel))
                os.replace(tmp, self.path)
                self._dirty = False
            except OSError:
                remove_quiet(tmp)


class SqliteLedger(Ledger):
    """<out>/.cryptostruct-dl/ledger.sqlite — the ledger for large orders: one row per file,
    a commit per finished file instead of rewriting a JSON file of a million entries.
    A ledger.json of earlier runs is imported once; once the SQLite ledger exists it is used
    for every run in this directory."""

    def __init__(self, out_dir, log):
        Ledger.__init__(self, out_dir, log)
        self.db_path = os.path.join(self.dir, LEDGER_SQLITE)
        self.conn = None

    @staticmethod
    def exists(out_dir):
        return os.path.exists(os.path.join(out_dir, LEDGER_DIR, LEDGER_SQLITE))

    def load(self):
        if sqlite3 is None:
            raise UsageError("this Python has no sqlite3 module — large orders need it (install a full Python 3 build)")
        os.makedirs(self.dir, exist_ok=True)
        fresh = not os.path.exists(self.db_path)
        self.conn = sqlite3.connect(self.db_path, timeout=30, check_same_thread=False)
        self.conn.execute(
            "CREATE TABLE IF NOT EXISTS entries (key TEXT PRIMARY KEY, path TEXT, size INTEGER, sha256 TEXT, at TEXT, order_id TEXT)")
        self.conn.commit()
        if fresh and os.path.exists(self.path):
            Ledger.load(self)
            rows = [(k, e.get("path"), e.get("size"), e.get("sha256"), e.get("at"), e.get("order_id"))
                    for k, e in self.entries.items()]
            self.conn.executemany("INSERT OR REPLACE INTO entries VALUES (?, ?, ?, ?, ?, ?)", rows)
            self.conn.commit()
            self.entries = {}
            if rows:
                self.log.progress("ledger: imported %d entries from %s into %s" % (len(rows), LEDGER_FILE, LEDGER_SQLITE))

    def get(self, key):
        with self._lock:
            row = self.conn.execute("SELECT key, path, size, sha256, at, order_id FROM entries WHERE key = ?", (key,)).fetchone()
        if row is None:
            return None
        return {"key": row[0], "path": row[1], "size": row[2], "sha256": row[3], "at": row[4], "order_id": row[5]}

    def record(self, key, rel_path, size, sha256, order_id):
        with self._lock:
            self.conn.execute("INSERT OR REPLACE INTO entries VALUES (?, ?, ?, ?, ?, ?)",
                              (key, rel_path, int(size) if size is not None else None, sha256, utc_now_iso(), order_id))
            self._dirty = True

    def save(self):
        with self._lock:
            if not self._dirty or self.conn is None:
                return
            try:
                self.conn.commit()
                self._dirty = False
            except Exception as e:  # sqlite3.Error
                self.log.warn("could not write ledger %s: %s" % (self.db_path, e))

    def count(self):
        with self._lock:
            return self.conn.execute("SELECT COUNT(*) FROM entries").fetchone()[0]


class AppendSidecar(Sidecar):
    """Paged mode: append one `sha256sum` line per verified file — never rewrite the whole file."""

    def add(self, rel_path, sha_hex):
        if not sha_hex:
            return
        with self._lock:
            try:
                with open(self.path, "a", encoding="utf-8") as fh:
                    fh.write("%s  %s\n" % (sha_hex.lower(), rel_path))
            except OSError:
                pass

    def save(self):
        return


# ───────────────────────────── Planner ─────────────────────────────


class Task:
    """One transfer unit: a tick/minute file, a derived format, a bundle day zip (or tar), or a range chunk."""

    def __init__(self, kind, key, order_id, filename, rel_dir, date, size, downloaded_at, ref, bundle=None, fmt=None):
        self.kind = kind              # tick | minute_csv | format | bundle_day | day_tar | range
        self.direct = False           # bundle_day: member files straight from storage (zip=1&urls=1)
        self.key = key
        self.order_id = order_id
        self.filename = filename
        self.rel_dir = rel_dir
        self.date = date
        self.size = size              # manifest bytes (exact for tick/minute, approximate for bundles)
        self.downloaded_at = downloaded_at
        self.ref = ref                # the manifest dict (item / day / chunk)
        self.bundle = bundle          # the bundle dict for bundle kinds
        self.fmt = fmt                # format token for kind == format
        self.plan = None              # new | local | downloaded-before
        self.note = None
        self.index = 0
        self.out_dir = None
        self.split_dates = None       # range: some days already here — fetch these one day at a time
        self.split_before = ()        # range (split): days the server marks as downloaded before
        self.split_why = None

    @property
    def rel_path(self):
        return os.path.join(self.rel_dir, self.filename)

    @property
    def abs_path(self):
        return os.path.join(self.out_dir, self.rel_dir, self.filename)

    def label(self):
        return self.filename

    def derive_range(self, chunk):
        t = Task("range", "range:%s:%s:%s:%s" % (self.order_id, self.bundle.get("bundle_id"), chunk.get("from"), chunk.get("to")),
                 self.order_id, chunk.get("filename"), self.rel_dir, chunk.get("from"), chunk.get("bytes"), None, chunk, self.bundle)
        t.out_dir = self.out_dir
        t.index = self.index
        t.plan = "new"
        return t

    def derive_day(self, day):
        t = Task("bundle_day", "bundle-day:%s:%s:%s" % (self.order_id, self.bundle.get("bundle_id"), day.get("date")),
                 self.order_id, day.get("filename") or ("%s-%s.zip" % (self.bundle.get("code"), day.get("date"))),
                 self.rel_dir, day.get("date"), day.get("bytes"), day.get("downloaded_at"), day, self.bundle)
        t.out_dir = self.out_dir
        t.index = self.index
        t.plan = "new"
        t.direct = self.direct and day_is_direct(day)
        return t

    def day_dir(self):
        """Where a direct bundle day's member files land: <rel_dir>/<date>/ (the --extract layout)."""
        return os.path.join(self.out_dir, self.rel_dir, safe_component(str(self.date or "undated")))

    def derive_day_tar(self):
        base = self.filename
        if base.lower().endswith(".zip"):
            base = base[:-4]
        t = Task("day_tar", self.key, self.order_id, base + ".tar", self.rel_dir, self.date, self.size, self.downloaded_at, self.ref, self.bundle)
        t.out_dir = self.out_dir
        t.index = self.index
        t.plan = "new"
        return t


class Plan:
    def __init__(self, tasks):
        self.tasks = tasks
        self.new_tasks = [t for t in tasks if t.plan == "new"]
        self.local_tasks = [t for t in tasks if t.plan == "local"]
        self.before_tasks = [t for t in tasks if t.plan == "downloaded-before"]
        self.new_bytes = sum(int(t.size or 0) for t in self.new_tasks)
        self.before_bytes = sum(int(t.size or 0) for t in self.before_tasks)
        dates = [str(t.downloaded_at)[:10] for t in self.before_tasks if t.downloaded_at]
        self.before_last = max(dates) if dates else None
        for i, t in enumerate(self.new_tasks, 1):
            t.index = i

    def plan_line(self):
        line = "Will download %d new files (%s) · %d already here · %d downloaded before" % (
            len(self.new_tasks), human_bytes(self.new_bytes), len(self.local_tasks), len(self.before_tasks))
        if self.before_last:
            line += " (last %s)" % self.before_last
        return line


class Planner:
    """Classify every item/day/chunk before anything is transferred."""

    def __init__(self, out_dir, ledger, http, log, opts):
        self.out_dir = out_dir
        self.ledger = ledger
        self.http = http
        self.log = log
        self.opts = opts

    # -- task construction ------------------------------------------------

    def build_tasks(self, manifest):
        tasks = []
        wanted_formats = list(self.opts.formats or [])
        for it in manifest.items:
            kind = it.get("kind") or "tick"
            if kind not in ("tick", "minute_csv"):
                self.log.warn("skipping item of unknown kind %r" % kind)
                continue
            order_id = str(it.get("order_id") or "")
            iid = it.get("instrument_id")
            date = it.get("date")
            rel_dir = os.path.join(safe_component(it.get("exchange_code")), safe_component(it.get("code")))
            filename = safe_component(it.get("filename") or "%s-%s-%s.txt.zst" % (it.get("exchange_code"), it.get("code"), date))
            key = "%s:%s:%s:%s" % (kind, order_id, iid, date)
            t = Task(kind, key, order_id, filename, rel_dir, date, it.get("bytes"), it.get("downloaded_at"), it)
            t.out_dir = self.out_dir
            tasks.append(t)
            if kind == "tick" and wanted_formats:
                fm = it.get("formats") if isinstance(it.get("formats"), dict) else {}
                for tok in wanted_formats:
                    url = fm.get(tok)
                    if not url:
                        continue
                    stem = filename[:-8] if filename.endswith(".txt.zst") else os.path.splitext(filename)[0]
                    ft = Task("format", "format:%s:%s:%s:%s" % (order_id, iid, date, tok), order_id, "%s-%s" % (stem, tok),
                              rel_dir, date, None, None, it, fmt=tok)
                    ft.out_dir = self.out_dir
                    tasks.append(ft)
        for b in manifest.bundles:
            order_id = str(b.get("order_id") or "")
            bid = b.get("bundle_id")
            rel_dir = os.path.join(safe_component(b.get("exchange_code")), safe_component(b.get("code")))
            days = [d for d in (b.get("days") or []) if isinstance(d, dict) and d.get("date")]
            chunks = [c for c in (b.get("range_chunks") or []) if isinstance(c, dict) and c.get("from") and c.get("to") and c.get("url")]
            covered = set()
            direct_ok = not self.opts.no_direct
            for c in chunks:
                in_chunk = [d for d in days if c["from"] <= d["date"] <= c["to"]]
                if direct_ok and in_chunk and all(day_is_direct(d) for d in in_chunk):
                    # Every day of this chunk comes straight from storage, day by
                    # day — no tar through the server (see _run_day_direct).
                    continue
                fname = safe_component(c.get("filename") or "%s_%s_%s.tar" % (b.get("code"), c.get("from"), c.get("to")))
                key = "range:%s:%s:%s:%s" % (order_id, bid, c.get("from"), c.get("to"))
                t = Task("range", key, order_id, fname, rel_dir, c.get("from"), c.get("bytes"), None, c, b)
                t.out_dir = self.out_dir
                t.direct = direct_ok
                tasks.append(t)
                for d in in_chunk:
                    covered.add(d["date"])
            for d in days:
                if d["date"] in covered:
                    continue
                fname = safe_component(d.get("filename") or "%s-%s.zip" % (b.get("code"), d.get("date")))
                key = "bundle-day:%s:%s:%s" % (order_id, bid, d.get("date"))
                t = Task("bundle_day", key, order_id, fname, rel_dir, d.get("date"), d.get("bytes"), d.get("downloaded_at"), d, b)
                t.out_dir = self.out_dir
                t.direct = direct_ok and day_is_direct(d)
                tasks.append(t)
        return tasks

    # -- classification ---------------------------------------------------

    def classify(self, tasks, dry_run=False, include_downloaded=False):
        for t in tasks:
            if t.kind == "range":
                self._classify_range(t, dry_run)
            elif t.kind == "bundle_day":
                self._classify_day(t)
            else:
                self._classify_file(t)
            if t.plan == "downloaded-before" and include_downloaded:
                t.plan = "new"
                t.note = (t.note + "; " if t.note else "") + "fetching again (--include-downloaded)"
        return Plan(tasks)

    def _ledger_missing_warning(self, t, led):
        self.log.warn("%s: previously downloaded here on %s, file no longer at %s" % (
            t.label(), str(led.get("at") or "?")[:10], led.get("path") or t.rel_path))

    def _verify_local_hash(self, t, path, led):
        if not (self.opts.verify and led and led.get("sha256")):
            return True
        try:
            actual = sha256_file(path)
        except OSError:
            return False
        if actual.lower() != str(led["sha256"]).lower():
            self.log.warn("%s: local file hash differs from the ledger — downloading again" % t.label())
            return False
        return True

    def _classify_file(self, t):
        path = t.abs_path
        led = self.ledger.get(t.key)
        if led and led.get("path") and led.get("path") != t.rel_path:
            alt = os.path.join(self.out_dir, led["path"])
            if os.path.exists(alt) and not os.path.exists(path):
                path = alt
        if os.path.exists(path):
            size = os.path.getsize(path)
            if t.size is not None and size == int(t.size):
                if self._verify_local_hash(t, path, led):
                    t.plan = "local"
                    return
                t.plan = "new"
                t.note = "hash mismatch"
                return
            if t.size is None and led and led.get("size") == size:
                t.plan = "local"
                return
            if t.size is None and not led:
                t.plan = "local"
                t.note = "size unknown, file present"
                return
            t.plan = "new"
            t.note = "size differs (%s here, %s expected)" % (human_bytes(size), human_bytes(t.size if t.size is not None else (led or {}).get("size")))
            return
        if os.path.exists(path + PART_SUFFIX):
            t.plan = "new"
            t.note = "resumable (%s)" % human_bytes(os.path.getsize(path + PART_SUFFIX))
            return
        if led:
            self._ledger_missing_warning(t, led)
            t.plan = "downloaded-before"
            t.downloaded_at = t.downloaded_at or led.get("at")
            t.note = "in ledger, file missing"
            return
        if t.downloaded_at:
            t.plan = "downloaded-before"
            t.note = "server: downloaded %s" % str(t.downloaded_at)[:10]
            return
        t.plan = "new"

    def _day_is_local(self, rel_dir, day, key):
        """A bundle day counts as here when its zip (or tar) exists and is complete — or when
        its member files are here (direct download / --extract) with the sizes its MANIFEST.txt lists."""
        led = self.ledger.get(key)
        if day.get("date") and day_dir_complete(os.path.join(self.out_dir, rel_dir, safe_component(str(day["date"])))):
            return True
        candidates = []
        fname = safe_component(day.get("filename") or "")
        if fname and fname != "_":
            candidates.append(os.path.join(self.out_dir, rel_dir, fname))
            if fname.lower().endswith(".zip"):
                candidates.append(os.path.join(self.out_dir, rel_dir, fname[:-4] + ".tar"))
        if led and led.get("path"):
            candidates.append(os.path.join(self.out_dir, led["path"]))
        for p in candidates:
            if not os.path.exists(p):
                continue
            size = os.path.getsize(p)
            if led and led.get("size") == size:
                return True
            if p.lower().endswith(".zip") and zipfile.is_zipfile(p):
                return True
            if p.lower().endswith(".tar") and day.get("bytes") is not None and size == int(day["bytes"]):
                return True
        return False

    def _classify_day(self, t):
        led = self.ledger.get(t.key)
        if self._day_is_local(t.rel_dir, t.ref, t.key):
            t.plan = "local"
            return
        path = t.abs_path
        if os.path.exists(path):
            t.plan = "new"
            t.note = "incomplete archive here, fetching again"
            return
        if led:
            self._ledger_missing_warning(t, led)
            t.plan = "downloaded-before"
            t.downloaded_at = t.downloaded_at or led.get("at")
            t.note = "in ledger, file missing"
            return
        if t.downloaded_at:
            t.plan = "downloaded-before"
            t.note = "server: downloaded %s" % str(t.downloaded_at)[:10]
            return
        t.plan = "new"

    def _classify_range(self, t, dry_run):
        chunk = t.ref
        path = t.abs_path
        led = self.ledger.get(t.key)
        if os.path.exists(path):
            size = os.path.getsize(path)
            if led and led.get("size") == size:
                t.plan = "local"
                return
            if dry_run:
                if t.size is not None and size == int(t.size):
                    t.plan = "local"
                    t.note = "size matches the manifest (HEAD check skipped in --dry-run)"
                else:
                    t.plan = "new"
                    t.note = "file here but not in the ledger — will compare with the server"
                return
            total = self._head_length(chunk.get("url"))
            if total is not None and total == size:
                t.plan = "local"
                self.ledger.record(t.key, t.rel_path, size, (led or {}).get("sha256"), t.order_id)
                return
            if total is None and t.size is not None and size == int(t.size):
                t.plan = "local"
                t.note = "size matches the manifest"
                return
            t.plan = "new"
            t.note = "size differs from the server — fetching again"
            return
        if os.path.exists(path + PART_SUFFIX):
            t.plan = "new"
            t.note = "resumable (%s)" % human_bytes(os.path.getsize(path + PART_SUFFIX))
            return
        # Not here as a tar: look at the chunk's days (they may be here as day zips).
        days = [d for d in (t.bundle.get("days") or []) if isinstance(d, dict) and d.get("date") and chunk["from"] <= d["date"] <= chunk["to"]]
        statuses = []
        latest = None
        for d in days:
            key = "bundle-day:%s:%s:%s" % (t.order_id, t.bundle.get("bundle_id"), d["date"])
            if self._day_is_local(t.rel_dir, d, key):
                statuses.append("local")
            elif d.get("downloaded_at") or self.ledger.get(key):
                statuses.append("before")
                stamp = d.get("downloaded_at") or (self.ledger.get(key) or {}).get("at")
                if stamp and (latest is None or str(stamp) > str(latest)):
                    latest = stamp
            else:
                statuses.append("new")
        if days and all(s == "local" for s in statuses):
            t.plan = "local"
            t.note = "all %d days already here as day files" % len(days)
            return
        if "local" in statuses and "new" in statuses:
            # Some days are here as day files (a split range, 1.4): fetch only the
            # missing ones one by one instead of the whole range tar again.
            t.plan = "new"
            t.split_dates = [d["date"] for d, s in zip(days, statuses) if s == "new"]
            t.split_before = [d["date"] for d, s in zip(days, statuses) if s == "before"]
            t.split_why = "%d of %d days already here" % (statuses.count("local"), len(days))
            t.note = t.split_why + " — fetching the rest one day at a time"
            t.size = sum(as_int(d.get("bytes")) for d, s in zip(days, statuses) if s == "new")
            return
        if led:
            self._ledger_missing_warning(t, led)
            t.plan = "downloaded-before"
            t.downloaded_at = led.get("at")
            t.note = "in ledger, file missing"
            return
        if days and "new" not in statuses:
            t.plan = "downloaded-before"
            t.downloaded_at = latest
            t.note = "every day of this chunk was downloaded before"
            return
        t.plan = "new"

    def _head_length(self, url):
        if not url:
            return None
        try:
            resp = self.http.request("HEAD", url)
        except TransientError:
            return None
        try:
            if resp.status == 410:
                raise ExpiredLink(expired_hint(resp, None))
            if resp.status in (200, 206):
                return resp.content_length()
            return None
        finally:
            resp.close()


# ───────────────────────────── Poller ─────────────────────────────


class WaitBudget:
    """--max-wait-min, charged per item for every pending/cold/busy wait."""

    def __init__(self, max_s):
        self.max_s = max_s
        self.used = 0.0

    def charge(self, seconds, what):
        if self.used + seconds > self.max_s:
            raise ItemPending("still %s after waiting %s (limit %s) — re-run later to pick it up" % (
                what, human_duration(self.used), human_duration(self.max_s)))
        self.used += seconds


class AttemptState:
    def __init__(self, max_wait_s):
        self.tries = 0                   # transient tries (network / 5xx)
        self.budget = WaitBudget(max_wait_s)
        self.resumed_from = None
        self.started = time.monotonic()
        self.restore_fired = False
        # Range chunks (1.4): the day the transfer waits on and its best pending count.
        # Progress is measured in waited (budget) seconds, never wall time.
        self.stuck_key = None
        self.stuck_best = None
        self.stuck_since = 0.0
        self.stuck_off = False           # a split was not possible — wait as before


class Session:
    """Everything the workers share."""

    def __init__(self, opts, http, log, masker, ledger, sidecar, verify, out_dir, contract):
        self.opts = opts
        self.http = http
        self.log = log
        self.masker = masker
        self.ledger = ledger
        self.sidecar = sidecar
        self.verify = verify
        self.out_dir = out_dir
        self.contract = contract or {}
        self.stop = threading.Event()
        self.clock = Clock(self.stop)
        self.fatal = None
        self.restored = set()
        self._restored_lock = threading.Lock()
        self.results = []
        self._results_lock = threading.Lock()
        self.total_new = 0

    def check_stop(self):
        if self.stop.is_set():
            raise Interrupted()

    def set_fatal(self, exc):
        if self.fatal is None:
            self.fatal = exc
        self.stop.set()

    def record(self, task, status, message=None, size=None, sha256=None, seconds=None, resumed_from=None, path=None):
        entry = {
            "key": task.key,
            "kind": task.kind,
            "order_id": task.order_id,
            "date": task.date,
            "filename": task.filename,
            "path": path if path is not None else task.rel_path,
            "status": status,
            "bytes": size,
            "sha256": sha256,
            "seconds": round(seconds, 1) if seconds is not None else None,
            "resumed_from": resumed_from,
            "message": self.masker.mask(message) if message else None,
        }
        with self._results_lock:
            self.results.append(entry)
        return entry

    def restore_cap(self):
        return getattr(self.opts, "restore_max", None)

    def restore_once(self, key):
        with self._restored_lock:
            if key in self.restored:
                return False
            cap = self.restore_cap()
            if cap is not None and len(self.restored) >= cap:
                return False
            self.restored.add(key)
            return True

    def restore_cap_reached(self, key):
        with self._restored_lock:
            cap = self.restore_cap()
            return cap is not None and key not in self.restored and len(self.restored) >= cap


class Poller:
    """Waiting primitives shared by the file, zip and range flows."""

    def __init__(self, session):
        self.s = session
        self.log = session.log
        self.http = session.http
        self.clock = session.clock
        self.opts = session.opts

    def tag(self, task):
        return "[%d/%d] %s" % (task.index, max(1, self.s.total_new), task.label())

    def wait(self, task, st, seconds, why):
        if seconds is None:
            seconds = DEFAULT_RETRY_AFTER_S
        seconds = min(MAX_POLL_S, max(MIN_POLL_S, float(seconds)))
        st.budget.charge(seconds, why)
        self.log.progress("%s … %s — waiting %s" % (self.tag(task), why, human_duration(seconds)))
        self.clock.sleep(seconds)

    def transient(self, task, st, msg):
        st.tries += 1
        if st.tries > MAX_TRANSIENT_TRIES:
            raise ItemFailed("gave up after %d tries: %s" % (MAX_TRANSIENT_TRIES, msg))
        delay = backoff_seconds(st.tries)
        self.log.progress("%s … %s — retry %d/%d in %s" % (self.tag(task), msg, st.tries, MAX_TRANSIENT_TRIES, human_duration(delay)))
        self.clock.sleep(delay)

    def eta_text(self, obj):
        eta = obj.get("eta_seconds") if isinstance(obj, dict) else None
        try:
            if eta is not None and float(eta) > 0:
                return ", eta ~%s" % human_duration(float(eta))
        except (TypeError, ValueError):
            pass
        return ""

    def pending(self, task, st, resp, restore_url, what="pending"):
        """A 202 answer: wait (or go cold on reason=archived)."""
        obj = resp.json()
        obj = obj if isinstance(obj, dict) else {}
        reason = obj.get("reason") or obj.get("state") or what
        if reason == "archived":
            self.cold(task, st, obj, restore_url, resp.retry_after())
            return
        wait = resp.retry_after()
        if wait is None:
            wait = DEFAULT_RETRY_AFTER_S
        self.wait(task, st, wait, "%s (%s%s)" % (what, reason, self.eta_text(obj)))

    def busy(self, task, st, resp, label=None):
        """429 / 503: honour Retry-After; without one it is a transient error."""
        wait = resp.retry_after()
        code = None
        try:
            code = resp.error_code()
        except Exception:
            code = None
        why = label or ("HTTP %d%s" % (resp.status, (" " + str(code)) if code else ""))
        if wait is None:
            if resp.status == 429:
                wait = backoff_seconds(max(1, st.tries + 1))
            else:
                self.transient(task, st, why)
                return
        self.wait(task, st, wait, why)

    def cold(self, task, st, obj, restore_url, retry_after=None):
        """Cold storage: without --restore the item counts as cold; with it, restore once and wait."""
        obj = obj if isinstance(obj, dict) else {}
        if not self.opts.restore:
            raise ItemCold("in cold storage — re-run with --restore (a restore takes up to ~12 h)")
        if not st.restore_fired:
            st.restore_fired = True
            if restore_url and self.s.restore_once(task.key):
                eta = self.fire_restore(task, st, restore_url)
                self.log.progress("%s … restore requested%s" % (self.tag(task), eta))
            elif not restore_url:
                raise ItemFailed("in cold storage and the manifest has no restore link")
            elif self.s.restore_cap_reached(task.key):
                raise ItemCold("in cold storage — this run already requested %d restores (--restore-max); re-run later to request more"
                               % self.s.restore_cap())
        wait = retry_after
        if wait is None:
            wait = DEFAULT_RETRY_AFTER_S
        self.wait(task, st, wait, "restoring%s" % self.eta_text(obj))

    def fire_restore(self, task, st, restore_url):
        while True:
            self.s.check_stop()
            try:
                resp = self.http.request("GET", restore_url)
            except TransientError as e:
                self.transient(task, st, "restore request failed: %s" % e)
                continue
            try:
                s = resp.status
                if 200 <= s < 300:
                    return self.eta_text(resp.json() if resp.json() else {})
                if s == 410:
                    raise ExpiredLink(expired_hint(resp, self.s.contract))
                if s in (401, 403, 404):
                    raise ItemFailed("restore request answered HTTP %d" % s)
                if s in (429, 503):
                    self.busy(task, st, resp, "restore busy")
                    continue
                if s >= 500:
                    self.transient(task, st, "restore request: HTTP %d" % s)
                    continue
                raise ItemFailed("restore request answered HTTP %d" % s)
            finally:
                resp.close()

    def get_json(self, task, st, url):
        """GET a JSON (or NDJSON) endpoint; waits/retries transient trouble; returns (status, text, resp-ish)."""
        while True:
            self.s.check_stop()
            try:
                resp = self.http.request("GET", url)
            except TransientError as e:
                self.transient(task, st, str(e))
                continue
            try:
                s = resp.status
                text = resp.text()
                if s == 410:
                    raise ExpiredLink(expired_hint(resp, self.s.contract))
                if s in (429, 503):
                    self.busy(task, st, resp)
                    continue
                if s >= 500:
                    self.transient(task, st, "HTTP %d" % s)
                    continue
                return s, text, resp
            finally:
                resp.close()


# ───────────────────────────── Bundles ─────────────────────────────


def split_chunk(chunk, suggested_to, bundle_code):
    """Two chunk dicts: [from, suggested_to] and [suggested_to + 1, to]."""
    frm = str(chunk.get("from"))
    to = str(chunk.get("to"))
    cut = str(suggested_to or "")
    if not (DATE_RE.match(cut) and frm <= cut < to):
        raise ItemFailed("server suggested an unusable split point %r for %s..%s" % (suggested_to, frm, to))
    name = str(chunk.get("filename") or "")
    m = re.match(r"^(.*)_(\d{4}-\d{2}-\d{2})_(\d{4}-\d{2}-\d{2})(\.tar)$", name)
    prefix = m.group(1) if m else str(bundle_code or "bundle")

    def mk(a, b):
        c = dict(chunk)
        c["from"] = a
        c["to"] = b
        c["days"] = days_between(a, b)
        c["bytes"] = None
        c["filename"] = "%s_%s_%s.tar" % (prefix, a, b)
        for k in ("url", "status_url", "restore_url", "manifest_url"):
            if chunk.get(k):
                c[k] = set_query(chunk[k], **{"from": a, "to": b})
        return c

    return mk(frm, cut), mk(next_day(cut), to)


def date_from_member_name(name, fallback=None):
    m = DATE_IN_NAME_RE.search(os.path.basename(str(name)))
    if m:
        return m.group(1)
    for part in str(name).replace("\\", "/").split("/"):
        if DATE_RE.match(part):
            return part
    return fallback


def day_is_direct(day):
    """Manifest hint: this bundle day is expected to answer `zip=1&urls=1` (B2-era)."""
    return bool(isinstance(day, dict) and day.get("direct") and day.get("urls_url"))


def day_dir_complete(dir_path):
    """A bundle day's member files are here when its MANIFEST.txt lists only files present with the right size."""
    try:
        with open(os.path.join(dir_path, "MANIFEST.txt"), "r", encoding="utf-8", errors="replace") as fh:
            lines = fh.read().splitlines()
    except OSError:
        return False
    n = 0
    for line in lines:
        m = MANIFEST_LINE_RE.match(line)
        if not m:
            continue
        p = os.path.join(dir_path, safe_component(m.group(2)))
        if not os.path.isfile(p) or os.path.getsize(p) != int(m.group(1)):
            return False
        n += 1
    return n > 0


def manifest_text_for(listing, entries):
    """The day's MANIFEST.txt — the server's text (byte-identical to the zip's) or, failing that, size\tname lines."""
    txt = listing.get("manifest_txt") if isinstance(listing, dict) else None
    if isinstance(txt, str) and txt.strip():
        return txt if txt.endswith("\n") else txt + "\n"
    return "".join("%d\t%s\n" % (int(e.get("size") or 0), e["name"]) for e in entries)


def extract_archive(archive_path, dest_root, date_hint, log):
    """Extract a bundle zip/tar into <dest_root>/<date>/<basename>. Never trusts member paths."""
    count = 0
    if archive_path.lower().endswith(".zip"):
        with zipfile.ZipFile(archive_path) as zf:
            for info in zf.infolist():
                if info.is_dir():
                    continue
                name = safe_component(os.path.basename(info.filename))
                if name == "_":
                    continue
                date = date_from_member_name(info.filename, date_hint) or "undated"
                dest_dir = os.path.join(dest_root, safe_component(date))
                os.makedirs(dest_dir, exist_ok=True)
                dest = os.path.join(dest_dir, name)
                with zf.open(info) as src, open(dest + ".tmp", "wb") as dst:
                    shutil.copyfileobj(src, dst, CHUNK_SIZE)
                os.replace(dest + ".tmp", dest)
                count += 1
    else:
        with tarfile.open(archive_path, "r:*") as tf:
            for member in tf:
                if not member.isfile():
                    continue
                name = safe_component(os.path.basename(member.name))
                if name == "_":
                    continue
                date = date_from_member_name(member.name, date_hint) or "undated"
                dest_dir = os.path.join(dest_root, safe_component(date))
                os.makedirs(dest_dir, exist_ok=True)
                dest = os.path.join(dest_dir, name)
                src = tf.extractfile(member)
                if src is None:
                    continue
                with src, open(dest + ".tmp", "wb") as dst:
                    shutil.copyfileobj(src, dst, CHUNK_SIZE)
                os.replace(dest + ".tmp", dest)
                count += 1
    return count


def dir_bytes(path):
    total = 0
    try:
        for name in os.listdir(path):
            p = os.path.join(path, name)
            if os.path.isfile(p):
                total += os.path.getsize(p)
    except OSError:
        pass
    return total


def salvage_range_part(part_path, dest_root, dates=None):
    """Unpack every COMPLETE day of a partial range tar into <dest_root>/<date>/ (the direct / --extract layout).

    A range tar holds RANGE_MANIFEST.txt, then per day <date>/MANIFEST.txt followed by the day's
    member files. Days are staged in <dest_root>/.salvage-<date>/ and moved into place only when
    their MANIFEST.txt lists nothing but files that came out whole — the day cut off by the end of
    the partial file (and anything after it) is dropped. Returns the dates moved into place."""
    wanted = set(dates) if dates else None
    staged = {}
    try:
        with tarfile.open(part_path, "r:") as tf:
            for member in tf:
                if not member.isfile():
                    continue
                pieces = member.name.split("/")
                if len(pieces) != 2 or not DATE_RE.match(pieces[0]):
                    continue  # RANGE_MANIFEST.txt
                date, name = pieces[0], safe_component(pieces[1])
                if name == "_" or (wanted is not None and date not in wanted):
                    continue
                stage = staged.get(date)
                if stage is None:
                    stage = os.path.join(dest_root, SALVAGE_DIR_PREFIX + date)
                    shutil.rmtree(stage, ignore_errors=True)
                    os.makedirs(stage)
                    staged[date] = stage
                src = tf.extractfile(member)
                if src is None:
                    continue
                dest = os.path.join(stage, name)
                with src, open(dest + ".tmp", "wb") as dst:
                    shutil.copyfileobj(src, dst, CHUNK_SIZE)
                if os.path.getsize(dest + ".tmp") != member.size:
                    break
                os.replace(dest + ".tmp", dest)
    except (tarfile.TarError, EOFError, OSError):
        pass  # the end of the partial archive
    moved = []
    for date, stage in sorted(staged.items()):
        final = os.path.join(dest_root, safe_component(date))
        if day_dir_complete(stage) and not day_dir_complete(final):
            shutil.rmtree(final, ignore_errors=True)
            os.replace(stage, final)
            moved.append(date)
        else:
            shutil.rmtree(stage, ignore_errors=True)
    return moved


# ───────────────────────────── Downloader ─────────────────────────────


class StreamResult:
    def __init__(self, size, sha256, digest, server_filename, etag):
        self.size = size
        self.sha256 = sha256
        self.digest = digest
        self.server_filename = server_filename
        self.etag = etag


class UrlListRefresher:
    """A day's member URLs expire (storage TTL, 1 h). When a member meets 401/403 the first worker
    fetches ONE fresh list from the shop and every other worker reuses it until it is half-way to
    expiry itself. Thread-safe; the server logs the handout once per list, so lists stay rare."""

    def __init__(self, downloader, task, url, listing):
        self.d = downloader
        self.task = task
        self.url = url
        self.lock = threading.Lock()
        self.fetched_at = time.monotonic()
        self.urls = self._index(listing)
        self.ttl = self._ttl(listing)

    @staticmethod
    def _index(listing):
        return dict((e.get("name"), e.get("url")) for e in (listing.get("entries") or []) if isinstance(e, dict))

    @staticmethod
    def _ttl(listing):
        try:
            return max(60.0, float(listing.get("expires_in") or 3600))
        except (TypeError, ValueError):
            return 3600.0

    def url_for(self, name):
        with self.lock:
            if time.monotonic() - self.fetched_at > self.ttl * 0.5:
                st = AttemptState(self.d.opts.max_wait_s)
                listing = self.d._fetch_url_list(self.task, st, self.url)
                if listing is None:
                    return None
                self.urls = self._index(listing)
                self.ttl = self._ttl(listing)
                self.fetched_at = time.monotonic()
                self.d.log.progress("%s … storage URLs expired — fetched a fresh list" % self.d.poll.tag(self.task))
            return self.urls.get(name)


class Downloader:
    """Runs the plan: files in parallel, bundle days ≤ 2 at a time (zip, or member files straight
    from storage with DIRECT_PARALLEL in flight), range chunks one at a time per order."""

    def __init__(self, session, plan):
        self.s = session
        self.plan = plan
        self.log = session.log
        self.http = session.http
        self.clock = session.clock
        self.opts = session.opts
        self.poll = Poller(session)
        self.zip_pool = None
        # Day tasks a split range hands to the zip pool (1.4); run() waits for them too,
        # so the range worker can move on to the next chunk at once.
        self._spawned = {}
        self._spawn_lock = threading.Lock()
        session.total_new = len(plan.new_tasks)

    # -- orchestration ----------------------------------------------------

    def run(self):
        opts = self.opts
        file_pool = concurrent.futures.ThreadPoolExecutor(max_workers=max(1, int(opts.parallel)), thread_name_prefix="file")
        zip_pool = concurrent.futures.ThreadPoolExecutor(max_workers=ZIP_PARALLEL, thread_name_prefix="zip")
        self.zip_pool = zip_pool
        range_pools = {}
        futures = {}
        for task in self.plan.new_tasks:
            if task.kind in ("tick", "minute_csv", "format"):
                fut = file_pool.submit(self._guard, self._run_file, task)
            elif task.kind == "bundle_day":
                fut = zip_pool.submit(self._guard, self._run_day, task)
            elif task.kind == "range":
                pool = range_pools.get(task.order_id)
                if pool is None:
                    pool = concurrent.futures.ThreadPoolExecutor(max_workers=1, thread_name_prefix="range")
                    range_pools[task.order_id] = pool
                fut = pool.submit(self._guard, self._run_range, task)
            else:
                continue
            futures[fut] = task
        interrupted = False
        pending = set(futures)
        try:
            while True:
                pending |= self._take_spawned(futures)
                if not pending:
                    break
                _done, pending = concurrent.futures.wait(pending, timeout=0.25)
                if self.s.fatal is not None:
                    break
        except KeyboardInterrupt:
            interrupted = True
            self.s.stop.set()
            self.log.progress("interrupted — finishing the ledger; re-run the same command to resume")
        pools = [file_pool, zip_pool] + list(range_pools.values())
        for p in pools:
            p.shutdown(wait=False, cancel_futures=True)
        self._take_spawned(futures)
        if interrupted or self.s.fatal is not None:
            self.s.stop.set()
        for fut, task in futures.items():
            if fut.cancelled():
                self.s.record(task, "interrupted" if interrupted else "aborted", "not started")
        remaining = [f for f in futures if not f.done()]
        if remaining:
            try:
                concurrent.futures.wait(remaining, timeout=SHUTDOWN_GRACE_S)
            except KeyboardInterrupt:
                pass
        return interrupted

    def _spawn(self, fn, task):
        """Run a derived task on the zip pool without waiting for it (run() collects it)."""
        try:
            fut = self.zip_pool.submit(self._guard, fn, task)
        except RuntimeError:  # pool already shut down (Ctrl-C / fatal error)
            self.s.record(task, "interrupted" if self.s.fatal is None else "aborted", "not started")
            return
        with self._spawn_lock:
            self._spawned[fut] = task

    def _take_spawned(self, futures):
        with self._spawn_lock:
            fresh = dict(self._spawned)
            self._spawned.clear()
        futures.update(fresh)
        return set(fresh)

    def _guard(self, fn, *args):
        task = args[0]
        try:
            fn(*args)
        except ItemFailed as e:
            self.s.record(task, "failed", str(e))
            self.log.progress("%s … failed: %s" % (self.poll.tag(task), e))
        except ItemCold as e:
            self.s.record(task, "cold", str(e))
            self.log.progress("%s … cold: %s" % (self.poll.tag(task), e))
        except ItemPending as e:
            self.s.record(task, "pending", str(e))
            self.log.progress("%s … left pending: %s" % (self.poll.tag(task), e))
        except Interrupted:
            self.s.record(task, "interrupted" if self.s.fatal is None else "aborted", "stopped before completion")
        except ExpiredLink as e:
            self.s.record(task, "aborted", str(e))
            self.s.set_fatal(e)
        except UsageError as e:
            self.s.record(task, "failed", str(e))
        except TransientError as e:
            self.s.record(task, "failed", str(e))
        except Exception as e:  # never let one item kill the run
            self.s.record(task, "failed", "internal error: %s: %s" % (type(e).__name__, e))
            self.log.warn("%s: internal error: %s: %s" % (task.label(), type(e).__name__, e))

    # -- single files (tick / minute_csv / derived formats) ----------------

    def _run_file(self, task):
        item = task.ref
        if task.kind == "format":
            url = (item.get("formats") or {}).get(task.fmt)
            if not url:
                raise ItemFailed("manifest lists no %s link for this file" % task.fmt)
            st = AttemptState(self.opts.max_wait_s)
            if not self._format_probe(task, st, url):
                self.s.record(task, "empty", "no rows for %s — nothing to download" % task.fmt)
                self.log.progress("%s … empty (no %s rows) — skipped" % (self.poll.tag(task), task.fmt))
                return
            restore_url = item.get("restore_url")
        else:
            url = item.get("url")
            if not url:
                raise ItemFailed("manifest item has no url")
            restore_url = item.get("restore_url")
        final = task.abs_path
        part = final + PART_SUFFIX
        for attempt in (1, 2):
            st = AttemptState(self.opts.max_wait_s)
            res = self._download_resumable(task, st, url, part, mode="file", restore_url=restore_url)
            try:
                self._finalize(task, st, part, res)
                return
            except VerifyError as e:
                remove_quiet(part)
                remove_quiet(meta_path_for(part))
                if attempt == 1:
                    self.log.progress("%s … verification failed (%s) — downloading once more" % (self.poll.tag(task), e))
                    continue
                raise ItemFailed("verification failed: %s (file discarded)" % e)

    def _format_probe(self, task, st, url):
        """Poll `url&probe=1` until ready (True) or empty (False); failed/unavailable raise."""
        probe_url = add_query(url, probe="1")
        while True:
            s, text, resp = self.poll.get_json(task, st, probe_url)
            try:
                obj = json.loads(text) if text else {}
            except ValueError:
                obj = {}
            obj = obj if isinstance(obj, dict) else {}
            if s in (200, 202):
                state = obj.get("state") or ("pending" if s == 202 else "ready")
                if state == "ready":
                    return True
                if state == "empty":
                    return False
                if state in ("failed", "unavailable", "error"):
                    raise ItemFailed("%s export %s%s" % (task.fmt, state, (": " + str(obj.get("reason"))) if obj.get("reason") else ""))
                wait = resp.retry_after()
                self.poll.wait(task, st, wait if wait is not None else DEFAULT_RETRY_AFTER_S,
                               "preparing %s (%s%s)" % (task.fmt, obj.get("reason") or state, self.poll.eta_text(obj)))
                continue
            if s == 409:
                self.poll.cold(task, st, obj, task.ref.get("restore_url"), resp.retry_after())
                continue
            if s in (401, 403, 404):
                raise ItemFailed("HTTP %d on the %s export" % (s, task.fmt))
            raise ItemFailed("unexpected HTTP %d on the %s probe" % (s, task.fmt))

    # -- resumable stream (files, formats, range chunks, day tars) --------

    def _download_resumable(self, task, st, url, part, mode, restore_url=None):
        """Loop until `part` holds the complete body. mode: file | range | daytar."""
        is_range = mode == "range"
        while True:
            self.s.check_stop()
            resume_from = os.path.getsize(part) if os.path.exists(part) else 0
            meta = read_part_meta(part) if resume_from > 0 else {}
            headers = {}
            if resume_from > 0:
                headers["Range"] = "bytes=%d-" % resume_from
                etag = meta.get("etag")
                if etag and not str(etag).startswith("W/"):
                    headers["If-Range"] = str(etag)
            try:
                resp = self.http.request("GET", url, headers)
            except TransientError as e:
                self.poll.transient(task, st, str(e))
                continue
            try:
                s = resp.status
                if s in (200, 206):
                    res = self._pump(task, st, resp, part, resume_from, meta)
                    if res is not None:
                        return res
                    continue
                if s == 416:
                    res = self._handle_416(task, st, resp, part, resume_from, meta)
                    if res is not None:
                        return res
                    continue
                if s == 202:
                    self.poll.pending(task, st, resp, restore_url)
                    continue
                if s in (429, 503):
                    if is_range and s == 503 and resp.error_code() == "WARMING":
                        body = resp.json() if isinstance(resp.json(), dict) else {}
                        date = body.get("date")
                        pend = body.get("pending")
                        self._stuck_check(st, "day:%s" % date, as_int(pend) if pend is not None else None, date)
                    self.poll.busy(task, st, resp)
                    continue
                if s == 410:
                    raise ExpiredLink(expired_hint(resp, self.s.contract))
                if s == 409:
                    code = resp.error_code()
                    if code == "SIZE_DRIFT":
                        raise ItemFailed("the server reports a size drift for this range — contact support")
                    self.poll.cold(task, st, resp.error_obj(), restore_url, resp.retry_after())
                    continue
                if s == 413:
                    if is_range:
                        err = resp.error_obj()
                        raise SplitChunk(err.get("suggested_to"))
                    raise ItemFailed("HTTP 413 — too large for this route")
                if s == 404 and is_range:
                    raise RangeNotFound()
                if s in (401, 403, 404):
                    raise ItemFailed("HTTP %d" % s)
                if s >= 500:
                    self.poll.transient(task, st, "HTTP %d" % s)
                    continue
                raise ItemFailed("unexpected HTTP %d" % s)
            finally:
                resp.close()

    def _handle_416(self, task, st, resp, part, resume_from, meta):
        if resume_from <= 0:
            raise ItemFailed("HTTP 416 on a fresh download")
        cr = parse_content_range(resp.header("Content-Range"))
        total = cr[2] if cr else None
        expected = total if total is not None else (int(task.size) if (task.size is not None and task.kind in ("tick", "minute_csv")) else None)
        if expected is None or expected == resume_from:
            sha = sha256_file(part) if self.opts.verify else None
            st.resumed_from = resume_from
            return StreamResult(resume_from, sha, meta.get("digest"), meta.get("server_filename"), meta.get("etag"))
        self.log.progress("%s … local part (%s) does not match the server (%s) — starting over" % (
            self.poll.tag(task), human_bytes(resume_from), human_bytes(expected)))
        remove_quiet(part)
        remove_quiet(meta_path_for(part))
        st.tries += 1
        if st.tries > MAX_TRANSIENT_TRIES:
            raise ItemFailed("could not resume after %d tries" % MAX_TRANSIENT_TRIES)
        return None

    def _pump(self, task, st, resp, part, resume_from, meta):
        """Write a 200/206 body to `part`. Returns a StreamResult, or None to retry (part kept for resume)."""
        status = resp.status
        total = None
        if status == 206:
            cr = parse_content_range(resp.header("Content-Range"))
            if cr is None or cr[0] != resume_from:
                self.log.progress("%s … server resumed at the wrong offset — starting over" % self.poll.tag(task))
                remove_quiet(part)
                remove_quiet(meta_path_for(part))
                st.tries += 1
                if st.tries > MAX_TRANSIENT_TRIES:
                    raise ItemFailed("could not resume after %d tries" % MAX_TRANSIENT_TRIES)
                return None
            total = cr[2]
            file_mode = "ab"
            st.resumed_from = resume_from
        else:
            if resume_from > 0:
                self.log.progress("%s … server restarted the transfer from zero (no resume)" % self.poll.tag(task))
            resume_from = 0
            file_mode = "wb"
            total = resp.content_length()
        etag = resp.header("ETag") or meta.get("etag")
        digest = parse_repr_digest(resp.header("Repr-Digest")) or meta.get("digest")
        server_filename = parse_content_disposition_filename(resp.header("Content-Disposition")) or meta.get("server_filename")
        os.makedirs(os.path.dirname(part), exist_ok=True)
        write_part_meta(part, {"etag": etag, "digest": digest, "server_filename": server_filename, "total": total, "url_kind": task.kind})
        hasher = hashlib.sha256() if self.opts.verify else None
        if hasher is not None and resume_from > 0:
            with open(part, "rb") as fh:
                while True:
                    chunk = fh.read(CHUNK_SIZE)
                    if not chunk:
                        break
                    hasher.update(chunk)
        written = resume_from
        try:
            with open(part, file_mode) as fh:
                for chunk in resp.iter_chunks():
                    if self.s.stop.is_set():
                        raise Interrupted()
                    fh.write(chunk)
                    written += len(chunk)
                    if hasher is not None:
                        hasher.update(chunk)
                    if total is not None and written >= total:
                        break
        except (socket.timeout, http.client.HTTPException, ConnectionError, OSError) as e:
            self.poll.transient(task, st, "connection dropped at %s (%s)" % (human_bytes(written), type(e).__name__))
            return None
        if total is not None and written != total:
            if written > total:
                remove_quiet(part)
                remove_quiet(meta_path_for(part))
                self.poll.transient(task, st, "received more bytes than announced — starting over")
                return None
            self.poll.transient(task, st, "short body: %s of %s" % (human_bytes(written), human_bytes(total)))
            return None
        return StreamResult(written, hasher.hexdigest() if hasher else None, digest, server_filename, etag)

    # -- bundle day zips -----------------------------------------------------

    def _run_day(self, task):
        day = task.ref
        if task.direct and not self.opts.no_direct and day.get("urls_url"):
            if self._run_day_direct(task):
                return
            # 409 NOT_DIRECT: the server serves this day as an archive — carry on below.
        st = AttemptState(self.opts.max_wait_s)
        zip_url = day.get("zip_url")
        if not zip_url:
            raise ItemFailed("manifest day has no zip link")
        outcome = self._zip_probe(task, st, day)
        if outcome == "tar":
            return self._run_day_tar(task)
        final = task.abs_path
        part = final + PART_SUFFIX
        try:
            res = self._download_zip(task, st, zip_url, part, day.get("restore_url"))
        except UseTar:
            remove_quiet(part)
            return self._run_day_tar(task)
        self._finalize(task, st, part, res)

    # -- bundle days straight from storage (zip=1&urls=1, 1.3) ---------------

    def _fetch_url_list(self, task, st, url):
        """GET the day's member-URL list. A dict on 200; None on 409 (NOT_DIRECT or cold — the archive path knows)."""
        while True:
            s, text, resp = self.poll.get_json(task, st, url)
            if s == 200:
                try:
                    obj = json.loads(text) if text else None
                except ValueError:
                    obj = None
                if not isinstance(obj, dict) or not isinstance(obj.get("entries"), list):
                    raise ItemFailed("the URL list is not a JSON object with entries")
                return obj
            if s == 409:
                return None
            if s == 202:
                self.poll.pending(task, st, resp, task.ref.get("restore_url"), "preparing day")
                continue
            if s in (401, 403, 404):
                raise ItemFailed("HTTP %d on the URL list" % s)
            raise ItemFailed("unexpected HTTP %d on the URL list" % s)

    def _run_day_direct(self, task):
        """Fetch every member file of the day from the URL list (B2 → here, nothing through the
        server). True when the day is complete; False when the server answered 409 (use the archive)."""
        day = task.ref
        st = AttemptState(self.opts.max_wait_s)
        listing = self._fetch_url_list(task, st, day["urls_url"])
        if listing is None:
            self.log.progress("%s … not served as a URL list — fetching the archive" % self.poll.tag(task))
            return False
        entries = [e for e in listing.get("entries") if isinstance(e, dict) and e.get("name") and e.get("url")]
        if not entries:
            raise ItemFailed("the URL list for this day is empty")
        dest_dir = task.day_dir()
        os.makedirs(dest_dir, exist_ok=True)
        refresher = UrlListRefresher(self, task, day["urls_url"], listing)
        self.log.progress("%s … %d member files straight from storage" % (self.poll.tag(task), len(entries)))
        errors = []
        total = 0
        total_lock = threading.Lock()

        def one(entry):
            try:
                n = self._fetch_member(task, entry, dest_dir, refresher)
                with total_lock:
                    nonlocal total
                    total += n
                return None
            except Interrupted:
                raise
            except DlError as e:
                return "%s: %s" % (entry["name"], e)
            except Exception as e:  # one broken member never kills the day silently
                return "%s: %s: %s" % (entry["name"], type(e).__name__, e)

        pool = concurrent.futures.ThreadPoolExecutor(max_workers=DIRECT_PARALLEL, thread_name_prefix="member")
        futures = [pool.submit(one, e) for e in entries]
        try:
            for fut in futures:
                err = fut.result()
                if err:
                    errors.append(err)
        except Interrupted:
            self.s.stop.set()
            pool.shutdown(wait=False, cancel_futures=True)
            raise
        pool.shutdown(wait=True)
        if errors:
            raise ItemFailed("%d of %d member files failed (first: %s) — re-run to fetch the rest" % (
                len(errors), len(entries), errors[0]))
        text = manifest_text_for(listing, entries)
        tmp = os.path.join(dest_dir, "MANIFEST.txt.tmp")
        with open(tmp, "w", encoding="utf-8") as fh:
            fh.write(text)
        os.replace(tmp, os.path.join(dest_dir, "MANIFEST.txt"))
        rel = os.path.relpath(dest_dir, self.s.out_dir)
        self.s.ledger.record(task.key, rel, total, None, task.order_id)
        self.s.ledger.save()
        elapsed = time.monotonic() - st.started
        self.s.record(task, "ok", None, size=total, seconds=elapsed, path=rel)
        self.log.progress("%s %s … ok %s (%d files straight from storage)" % (
            self.poll.tag(task), human_bytes(total), human_duration(elapsed), len(entries)))
        return True

    def _fetch_member(self, task, entry, dest_dir, refresher):
        """One member file from storage into <dest_dir>/<name>; resumes a .part with Range. Returns its size."""
        name = safe_component(entry["name"])
        try:
            size = int(entry.get("size"))
        except (TypeError, ValueError):
            size = None
        final = os.path.join(dest_dir, name)
        part = final + PART_SUFFIX
        if os.path.isfile(final) and (size is None or os.path.getsize(final) == size):
            return os.path.getsize(final)
        url = entry["url"]
        tries = 0
        refreshes = 0
        while True:
            self.s.check_stop()
            resume_from = os.path.getsize(part) if os.path.exists(part) else 0
            headers = {"Range": "bytes=%d-" % resume_from} if resume_from > 0 else {}
            try:
                resp = self.http.request("GET", url, headers=headers, auth=False)
            except TransientError as e:
                tries += 1
                if tries > MAX_TRANSIENT_TRIES:
                    raise ItemFailed("gave up after %d tries: %s" % (MAX_TRANSIENT_TRIES, e))
                self.clock.sleep(backoff_seconds(tries))
                continue
            try:
                s = resp.status
                if s in (200, 206):
                    if s == 206:
                        cr = parse_content_range(resp.header("Content-Range"))
                        if cr is None or cr[0] != resume_from:
                            remove_quiet(part)
                            tries += 1
                            if tries > MAX_TRANSIENT_TRIES:
                                raise ItemFailed("storage resumed at the wrong offset %d times" % MAX_TRANSIENT_TRIES)
                            continue
                        mode = "ab"
                    else:
                        resume_from = 0
                        mode = "wb"
                    written = resume_from
                    try:
                        with open(part, mode) as fh:
                            for chunk in resp.iter_chunks():
                                if self.s.stop.is_set():
                                    raise Interrupted()
                                fh.write(chunk)
                                written += len(chunk)
                    except (socket.timeout, http.client.HTTPException, ConnectionError, OSError):
                        tries += 1
                        if tries > MAX_TRANSIENT_TRIES:
                            raise ItemFailed("connection dropped %d times" % MAX_TRANSIENT_TRIES)
                        self.clock.sleep(backoff_seconds(tries))
                        continue
                    if size is not None and written != size:
                        # short body: keep the .part and resume; too long: start over
                        if written > size:
                            remove_quiet(part)
                        tries += 1
                        if tries > MAX_TRANSIENT_TRIES:
                            raise ItemFailed("size %d differs from the list (%d) after %d tries" % (written, size, MAX_TRANSIENT_TRIES))
                        self.clock.sleep(backoff_seconds(tries))
                        continue
                    if self.opts.verify and name.lower().endswith(".zst") and ZSTD_BIN:
                        proc = subprocess.run([ZSTD_BIN, "-t", "-q", part], stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, timeout=3600)
                        if proc.returncode != 0:
                            remove_quiet(part)
                            tries += 1
                            if tries > MAX_TRANSIENT_TRIES:
                                raise ItemFailed("zstd -t failed: %s" % proc.stderr.decode("utf-8", "replace").strip()[:200])
                            continue
                    os.replace(part, final)
                    return written
                if s in (401, 403):
                    # The presigned URL expired (1 h) — ask the shop for a fresh list, once per member and only so often.
                    refreshes += 1
                    if refreshes > DIRECT_REFRESH_MAX:
                        raise ItemFailed("storage keeps answering HTTP %d" % s)
                    fresh = refresher.url_for(name)
                    if not fresh:
                        raise ItemFailed("storage answered HTTP %d and the fresh list has no URL for this file" % s)
                    url = fresh
                    continue
                if s == 416:
                    remove_quiet(part)
                    continue
                if s == 404:
                    raise ItemFailed("storage answered 404")
                if s == 429 or s >= 500:
                    tries += 1
                    if tries > MAX_TRANSIENT_TRIES:
                        raise ItemFailed("storage answered HTTP %d %d times" % (s, MAX_TRANSIENT_TRIES))
                    wait = resp.retry_after()
                    self.clock.sleep(wait if wait is not None else backoff_seconds(tries))
                    continue
                raise ItemFailed("unexpected HTTP %d from storage" % s)
            finally:
                resp.close()

    def _run_day_tar(self, task):
        day = task.ref
        tar_url = day.get("tar_url")
        if not tar_url:
            raise ItemFailed("day is over the zip ceiling and the manifest has no tar link — contact support")
        t = task.derive_day_tar()
        st = AttemptState(self.opts.max_wait_s)
        part = t.abs_path + PART_SUFFIX
        self.log.progress("%s … over the zip ceiling — fetching the day as a tar (%s)" % (self.poll.tag(task), t.filename))
        res = self._download_resumable(t, st, tar_url, part, mode="daytar", restore_url=day.get("restore_url"))
        self._finalize(t, st, part, res)

    def _zip_probe(self, task, st, day):
        probe_url = add_query(day["zip_url"], probe="1")
        while True:
            s, text, resp = self.poll.get_json(task, st, probe_url)
            try:
                obj = json.loads(text) if text else {}
            except ValueError:
                obj = {}
            obj = obj if isinstance(obj, dict) else {}
            if s in (200, 202):
                state = obj.get("state") or ("pending" if s == 202 else "ready")
                if state == "ready":
                    return "ready"
                if state == "too_large":
                    return "tar"
                if state in ("unavailable", "error"):
                    raise ItemFailed("bundle day %s%s" % (state, (": " + str(obj.get("reason"))) if obj.get("reason") else ""))
                cold = as_int(obj.get("cold"))
                if cold > 0:
                    self.poll.cold(task, st, obj, day.get("restore_url"), resp.retry_after())
                    continue
                ready = as_int(obj.get("ready"))
                pend = as_int(obj.get("pending"))
                why = "preparing day (%s)" % (obj.get("reason") or "%d ready, %d pending" % (ready, pend))
                wait = resp.retry_after()
                self.poll.wait(task, st, wait if wait is not None else DEFAULT_RETRY_AFTER_S, why)
                continue
            if s == 409:
                self.poll.cold(task, st, obj, day.get("restore_url"), resp.retry_after())
                continue
            if s == 413:
                return "tar"
            if s in (401, 403, 404):
                raise ItemFailed("HTTP %d" % s)
            raise ItemFailed("unexpected HTTP %d on the day probe" % s)

    def _download_zip(self, task, st, url, part, restore_url):
        """Day zips have no Content-Length and no Range: every attempt starts from zero."""
        while True:
            self.s.check_stop()
            remove_quiet(part)
            try:
                resp = self.http.request("GET", url)
            except TransientError as e:
                self.poll.transient(task, st, str(e))
                continue
            try:
                s = resp.status
                if s == 200:
                    res = self._pump(task, st, resp, part, 0, {})
                    if res is None:
                        continue
                    if not zipfile.is_zipfile(part):
                        remove_quiet(part)
                        self.poll.transient(task, st, "zip stream ended without a central directory (stall) — starting over")
                        continue
                    return res
                if s == 202:
                    self.poll.pending(task, st, resp, restore_url)
                    continue
                if s in (429, 503):
                    self.poll.busy(task, st, resp)
                    continue
                if s == 409:
                    self.poll.cold(task, st, resp.error_obj(), restore_url, resp.retry_after())
                    continue
                if s == 413:
                    raise UseTar()
                if s == 410:
                    raise ExpiredLink(expired_hint(resp, self.s.contract))
                if s in (401, 403, 404):
                    raise ItemFailed("HTTP %d" % s)
                if s >= 500:
                    self.poll.transient(task, st, "HTTP %d" % s)
                    continue
                raise ItemFailed("unexpected HTTP %d" % s)
            finally:
                resp.close()

    # -- bundle range chunks -------------------------------------------------

    def _run_range(self, task):
        if task.split_dates is not None:
            # Some days are here already (an earlier split): only the missing ones, one day at a time.
            dates = list(task.split_dates)
            if self.opts.include_downloaded:
                dates += list(task.split_before)
            self._split_to_days(task, task.ref, task.abs_path + PART_SUFFIX, None, task.split_why or "some days already here",
                                only=set(dates))
            return
        self._run_range_chunk(task, task.ref)

    def _run_range_chunk(self, task, chunk):
        try:
            self._range_chunk(task, chunk)
        except RangeNotFound:
            self._range_fallback_days(task, chunk)

    def _range_chunk(self, task, chunk):
        st = AttemptState(self.opts.max_wait_s)
        final = task.abs_path
        part = final + PART_SUFFIX
        while True:
            try:
                self._range_precheck(task, st, chunk)
                res = self._download_resumable(task, st, chunk["url"], part, mode="range", restore_url=chunk.get("restore_url"))
                break
            except SplitChunk as e:
                remove_quiet(part)
                remove_quiet(meta_path_for(part))
                left, right = split_chunk(chunk, e.suggested_to, (task.bundle or {}).get("code"))
                self.log.progress("%s … range too large — splitting at %s (%s + %s)" % (
                    self.poll.tag(task), e.suggested_to, left["filename"], right["filename"]))
                self.s.record(task, "split", "split at %s" % e.suggested_to)
                for sub in (left, right):
                    self.s.check_stop()
                    self._guard(self._run_range_chunk, task.derive_range(sub), sub)
                return
            except StuckRange as e:
                # Partial delivery per day (1.4): one stuck day must not hold back the others.
                if self._split_to_days(task, chunk, part, e.date, str(e)):
                    return
                st.stuck_off = True  # cannot split here — keep waiting on the range as before
            except ItemCold as e:
                # Cold without --restore: the warm days still come, the cold ones report cold one by one.
                if self._split_to_days(task, chunk, part, None, "cold days in this range"):
                    return
                raise
        self._finalize(task, st, part, res)

    def _stuck_check(self, st, key, pending, date):
        """Called before every wait on a range chunk. Raises StuckRange once the day the transfer waits on
        (`key`) made no progress — its pending count did not drop — for --stuck-day-min waited minutes."""
        limit = float(getattr(self.opts, "stuck_day_s", 0) or 0)
        if limit <= 0 or st.stuck_off:
            return
        progressed = pending is not None and (st.stuck_best is None or pending < st.stuck_best)
        if key != st.stuck_key or progressed:
            st.stuck_key = key
            st.stuck_best = pending
            st.stuck_since = st.budget.used
            return
        waited = st.budget.used - st.stuck_since
        if waited >= limit:
            raise StuckRange(date, waited)

    def _split_to_days(self, task, chunk, part, stuck_date, why, only=None):
        """Partial delivery per day (1.4). Unpacks every day already complete in the partial range tar into
        <exchange>/<bundle>/<date>/, then hands each remaining day of the chunk to the zip pool as its own
        task (member files straight from storage, the day zip, or the day tar) — the stuck day last.
        False when the chunk cannot be split (the manifest lists no days for it, or the disk is too full
        to unpack the partial tar); the caller then keeps its old behaviour."""
        days = [d for d in (task.bundle.get("days") or []) if isinstance(d, dict) and d.get("date")
                and chunk["from"] <= d["date"] <= chunk["to"]]
        if only is not None:
            days = [d for d in days if d["date"] in only]
        if not days and only is None:
            return False
        dest_root = os.path.join(self.s.out_dir, task.rel_dir)
        unpacked = []
        if os.path.exists(part) and os.path.getsize(part) > 0:
            need = os.path.getsize(part) + SALVAGE_DISK_MARGIN
            try:
                free = shutil.disk_usage(dest_root).free
            except OSError:
                free = None
            if free is not None and free < need:
                self.log.warn("%s: not enough free disk to unpack the partial archive (%s needed) — waiting on the range instead"
                              % (task.label(), human_bytes(need)))
                return False
            unpacked = salvage_range_part(part, dest_root, [d["date"] for d in days])
            remove_quiet(part)
            remove_quiet(meta_path_for(part))
        by_date = dict((d["date"], d) for d in days)
        for date in unpacked:
            sub = task.derive_day(by_date[date])
            rel = os.path.join(task.rel_dir, safe_component(date))
            size = dir_bytes(os.path.join(self.s.out_dir, rel))
            self.s.ledger.record(sub.key, rel, size, None, task.order_id)
            self.s.record(sub, "ok", "unpacked from the partial range archive", size=size, path=rel)
        if unpacked:
            self.s.ledger.save()
        planner = Planner(self.s.out_dir, self.s.ledger, self.http, self.log, self.opts)
        rest = [d for d in days if not planner._day_is_local(
            task.rel_dir, d, "bundle-day:%s:%s:%s" % (task.order_id, task.bundle.get("bundle_id"), d["date"]))]
        rest.sort(key=lambda d: (d["date"] == stuck_date, d["date"]))
        parts = []
        if unpacked:
            parts.append("%d day(s) unpacked from the partial archive" % len(unpacked))
        parts.append("%d fetched one day at a time%s" % (
            len(rest), (" (%s last)" % stuck_date) if stuck_date and any(d["date"] == stuck_date for d in rest) else ""))
        msg = "%s — %s" % (why, ", ".join(parts))
        self.log.progress("%s … %s" % (self.poll.tag(task), msg))
        self.s.record(task, "split", msg)
        for d in rest:
            self._spawn(self._run_day, task.derive_day(d))
        return True

    def _range_precheck(self, task, st, chunk):
        """Poll the range status (NDJSON) until pending == 0 and cold == 0 (cold handled by --restore)."""
        url = chunk.get("status_url")
        if not url:
            return
        while True:
            s, text, resp = self.poll.get_json(task, st, url)
            if s == 200:
                summary = parse_ndjson_summary(text)
                if summary is None:
                    return
                if as_int(summary.get("size_drift")) > 0:
                    raise ItemFailed("the server reports a size drift for this range — contact support")
                cold = as_int(summary.get("cold"))
                pend = as_int(summary.get("pending"))
                if cold > 0:
                    self.poll.cold(task, st, summary, chunk.get("restore_url"), resp.retry_after())
                    continue
                if pend > 0:
                    self._stuck_check(st, "status", pend, first_unready_date(text))
                    wait = resp.retry_after()
                    self.poll.wait(task, st, wait if wait is not None else DEFAULT_RETRY_AFTER_S,
                                   "warming %d of %d days" % (pend, as_int(summary.get("days"), pend)))
                    continue
                if as_int(summary.get("errored_days")) > 0:
                    self.log.warn("%s: the server reports %d errored day(s) in this range" % (task.label(), as_int(summary.get("errored_days"))))
                return
            if s == 404:
                return  # no status route — the GET decides (404 there → per-day fallback)
            if s == 409:
                try:
                    obj = json.loads(text)
                except ValueError:
                    obj = {}
                err = obj.get("error") if isinstance(obj, dict) and isinstance(obj.get("error"), dict) else obj
                if isinstance(err, dict) and err.get("code") == "SIZE_DRIFT":
                    raise ItemFailed("the server reports a size drift for this range — contact support")
                self.poll.cold(task, st, err if isinstance(err, dict) else {}, chunk.get("restore_url"), resp.retry_after())
                continue
            if s in (401, 403):
                raise ItemFailed("HTTP %d" % s)
            return

    def _range_fallback_days(self, task, chunk):
        days = [d for d in (task.bundle.get("days") or []) if isinstance(d, dict) and d.get("date") and chunk["from"] <= d["date"] <= chunk["to"]]
        if not days:
            raise ItemFailed("range route not available and the manifest lists no day files for %s..%s" % (chunk["from"], chunk["to"]))
        self.log.progress("%s … range route not available — falling back to %d day file(s)" % (self.poll.tag(task), len(days)))
        self.s.record(task, "fallback", "range route 404 — fetched %d days individually" % len(days))
        planner = Planner(self.s.out_dir, self.s.ledger, self.http, self.log, self.opts)
        futs = []
        for d in days:
            sub = task.derive_day(d)
            planner._classify_day(sub)
            if sub.plan == "local":
                self.s.record(sub, "local", "already here")
                continue
            if sub.plan == "downloaded-before" and not self.opts.include_downloaded:
                self.s.record(sub, "downloaded-before", sub.note)
                continue
            futs.append(self.zip_pool.submit(self._guard, self._run_day, sub))
        if futs:
            concurrent.futures.wait(futs)

    # -- completion -------------------------------------------------------------

    def _finalize(self, task, st, part, res):
        final = task.abs_path
        if task.kind == "format" and res.server_filename:
            final = os.path.join(os.path.dirname(final), res.server_filename)
            task.filename = res.server_filename
        self.s.verify.check(task, part, res)
        os.makedirs(os.path.dirname(final), exist_ok=True)
        os.replace(part, final)
        remove_quiet(meta_path_for(part))
        rel = os.path.relpath(final, self.s.out_dir)
        self.s.ledger.record(task.key, rel, res.size, res.sha256, task.order_id)
        self.s.ledger.save()
        if self.opts.verify and res.sha256:
            self.s.sidecar.add(rel, res.sha256)
            self.s.sidecar.save()
        extracted = None
        if self.opts.extract and task.kind in ("bundle_day", "day_tar", "range"):
            try:
                dest_root = os.path.join(self.s.out_dir, task.rel_dir)
                extracted = extract_archive(final, dest_root, task.date if task.kind != "range" else None, self.log)
            except (OSError, zipfile.BadZipFile, tarfile.TarError) as e:
                self.log.warn("%s: downloaded but could not be extracted: %s" % (task.label(), e))
        elapsed = time.monotonic() - st.started
        self.s.record(task, "ok", None, size=res.size, sha256=res.sha256, seconds=elapsed, resumed_from=st.resumed_from, path=rel)
        line = "%s %s" % (self.poll.tag(task), human_bytes(res.size))
        if st.resumed_from:
            line += " resumed from %s" % human_bytes(st.resumed_from)
        line += " … ok %s" % human_duration(elapsed)
        if extracted is not None:
            line += " (extracted %d files)" % extracted
        self.log.progress(line)


# ───────────────────────────── Verify ─────────────────────────────


class Verify:
    def __init__(self, enabled, log):
        self.enabled = enabled
        self.log = log

    def check(self, task, path, res):
        """Raise VerifyError when the finished `path` (still the .part file) is not what the server/manifest promised."""
        lower = str(task.filename or "").lower()  # the final name decides the suffix checks, not the .part path
        if task.kind == "bundle_day" and not zipfile.is_zipfile(path):
            raise VerifyError("not a complete zip archive")
        if not self.enabled:
            return
        if task.kind in ("tick", "minute_csv") and task.size is not None and res.size != int(task.size):
            raise VerifyError("size %d differs from the manifest (%d)" % (res.size, int(task.size)))
        expected = res.digest
        source = "Repr-Digest"
        if not expected and task.kind in ("tick", "minute_csv"):
            expected = task.ref.get("sha256")
            source = "manifest sha256"
        if expected and res.sha256 and res.sha256.lower() != str(expected).lower():
            raise VerifyError("sha-256 mismatch against the %s" % source)
        if lower.endswith(".zst") and ZSTD_BIN:
            try:
                proc = subprocess.run([ZSTD_BIN, "-t", "-q", path], stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, timeout=3600)
            except (OSError, subprocess.SubprocessError) as e:
                self.log.warn("%s: could not run zstd -t: %s" % (task.label(), e))
                return
            if proc.returncode != 0:
                raise VerifyError("zstd -t failed: %s" % proc.stderr.decode("utf-8", "replace").strip()[:200])
        if lower.endswith(".tar") and not tarfile.is_tarfile(path):
            raise VerifyError("not a tar archive")


# ───────────────────────────── Report ─────────────────────────────


class Report:
    COUNTED = ("ok", "failed", "cold", "pending", "empty", "interrupted", "aborted")

    def __init__(self, plan, results, elapsed, interrupted, opts, manifest, out_dir):
        self.plan = plan
        self.results = results
        self.elapsed = elapsed
        self.interrupted = interrupted
        self.opts = opts
        self.manifest = manifest
        self.out_dir = out_dir

    def counts(self):
        c = dict((k, 0) for k in self.COUNTED)
        c["ok_bytes"] = 0
        for r in self.results:
            s = r.get("status")
            if s in c:
                c[s] += 1
            if s == "ok":
                c["ok_bytes"] += int(r.get("bytes") or 0)
        c["already_here"] = len(self.plan.local_tasks) + sum(1 for r in self.results if r.get("status") == "local")
        c["downloaded_before"] = len(self.plan.before_tasks) + sum(1 for r in self.results if r.get("status") == "downloaded-before")
        return c

    def exit_code(self):
        if self.interrupted:
            return 3
        c = self.counts()
        if c["failed"] or c["cold"] or c["pending"] or c["aborted"] or c["interrupted"]:
            return 1
        return 0

    def human_lines(self):
        c = self.counts()
        head = "Interrupted after" if self.interrupted else "Done in"
        line = "%s %s — %d ok (%s) · %d already here · %d downloaded before" % (
            head, human_duration(self.elapsed), c["ok"], human_bytes(c["ok_bytes"]), c["already_here"], c["downloaded_before"])
        if self.plan.before_last:
            line += " (last %s)" % self.plan.before_last
        extras = []
        for key, label in (("failed", "failed"), ("cold", "cold"), ("pending", "left pending"), ("empty", "empty"), ("interrupted", "not finished")):
            if c[key]:
                extras.append("%d %s" % (c[key], label))
        if extras:
            line += " · " + " · ".join(extras)
        lines = [line]
        if self.plan.before_tasks and not self.opts.include_downloaded:
            lines.append("  %d downloaded before%s — add --include-downloaded to fetch them again" % (
                len(self.plan.before_tasks), (" (last %s)" % self.plan.before_last) if self.plan.before_last else ""))
        groups = (
            ("failed", "Failed:"),
            ("cold", "Cold (re-run with --restore; a restore takes up to ~12 h):"),
            ("pending", "Left pending (re-run later — the server is still preparing these):"),
        )
        for status, title in groups:
            rows = [r for r in self.results if r.get("status") == status]
            if not rows:
                continue
            lines.append(title)
            for r in rows[:50]:
                lines.append("  %s — %s" % (r.get("filename"), r.get("message") or status))
            if len(rows) > 50:
                lines.append("  … and %d more" % (len(rows) - 50))
        if self.interrupted:
            lines.append("Re-run the same command to resume; finished files are in the ledger and will not be fetched again.")
        return lines

    def as_json(self, exit_code, masker):
        c = self.counts()
        return {
            "tool": "cryptostruct-dl",
            "version": __version__,
            "generated_at": utc_now_iso(),
            "exit_code": exit_code,
            "interrupted": self.interrupted,
            "scope": self.manifest.scope if self.manifest else None,
            "manifest_source": masker.mask(self.manifest.source) if self.manifest else None,
            "out": self.out_dir,
            "orders": [o.get("order_id") for o in (self.manifest.orders if self.manifest else [])],
            "plan": {
                "new": len(self.plan.new_tasks),
                "new_bytes": self.plan.new_bytes,
                "already_here": len(self.plan.local_tasks),
                "downloaded_before": len(self.plan.before_tasks),
                "downloaded_before_last": self.plan.before_last,
                "total_tasks": len(self.plan.tasks),
            },
            "results": {
                "ok": c["ok"],
                "ok_bytes": c["ok_bytes"],
                "failed": c["failed"],
                "cold": c["cold"],
                "left_pending": c["pending"],
                "empty": c["empty"],
                "interrupted": c["interrupted"] + c["aborted"],
                "already_here": c["already_here"],
                "downloaded_before": c["downloaded_before"],
            },
            "items": list(self.results),
            "elapsed_seconds": round(self.elapsed, 1),
        }


def dry_run_table(plan):
    rows = []
    width_status = max([len("STATUS")] + [len(t.plan or "") for t in plan.tasks])
    width_kind = max([len("KIND")] + [len(t.kind) for t in plan.tasks])
    header = "%-*s  %-*s  %-10s  %9s  %s" % (width_status, "STATUS", width_kind, "KIND", "DATE", "SIZE", "FILE")
    rows.append(header)
    for t in plan.tasks:
        status = {"local": "already here", "downloaded-before": "downloaded before"}.get(t.plan, t.plan)
        line = "%-*s  %-*s  %-10s  %9s  %s" % (width_status, status, width_kind, t.kind, t.date or "", human_bytes(t.size), t.rel_path)
        if t.note:
            line += "   (%s)" % t.note
        rows.append(line)
    return rows


# ───────────────────────────── CLI ─────────────────────────────


class Options:
    def __init__(self, args):
        self.out = args.out
        self.parallel = args.parallel
        self.dry_run = args.dry_run
        self.restore = args.restore
        self.include_downloaded = args.include_downloaded
        self.verify = args.verify
        self.extract = args.extract
        self.no_direct = args.no_direct
        self.formats = args.formats or []
        self.only = args.only
        self.date_from = args.date_from
        self.date_to = args.date_to
        self.max_wait_min = args.max_wait_min
        self.max_wait_s = float(args.max_wait_min) * 60.0
        self.stuck_day_s = float(args.stuck_day_min) * 60.0
        self.json = args.json
        self.base = args.base.rstrip("/")
        self.yes = args.yes
        self.page_wait_min = args.page_wait_min
        self.restore_max = args.restore_max  # None = unlimited (small manifests); paged mode defaults it


def build_parser():
    p = argparse.ArgumentParser(
        prog="cryptostruct_dl.py",
        description="Download the files of a CryptoStruct order — resumable, verified, never twice by accident.",
        epilog="exit codes: 0 complete · 1 incomplete (failed/cold/left pending) · 2 usage error, 404, expired link · 3 interrupted",
    )
    p.add_argument("--order", metavar="ORDER_ID", help="order id (with --token or --key)")
    p.add_argument("--token", metavar="T", help="guest download token (env CRYPTOSTRUCT_DOWNLOAD_TOKEN)")
    p.add_argument("--key", metavar="K", help="account download key cs_dl_… (env CRYPTOSTRUCT_DOWNLOAD_KEY)")
    p.add_argument("--manifest", metavar="FILE_OR_URL", help="load the manifest from a local JSON file or a URL")
    p.add_argument("--out", default=".", metavar="DIR", help='output directory (default ".")')
    p.add_argument("--parallel", type=int, default=3, metavar="N", help="single files in flight (default 3)")
    p.add_argument("--dry-run", action="store_true", help="plan only: one manifest request, no downloads, no restore")
    p.add_argument("--restore", action="store_true", help="trigger a restore for cold files/days and wait with an ETA")
    p.add_argument("--include-downloaded", action="store_true", help="also fetch items the server marks as downloaded before")
    p.add_argument("--verify", action="store_true", help="size + sha-256 (Repr-Digest / manifest) + zstd -t; writes cryptostruct-dl.sha256")
    p.add_argument("--extract", action="store_true", help="extract bundle tar/zip into <exchange>/<bundle>/<date>/")
    p.add_argument("--no-direct", action="store_true",
                   help="bundle days: always fetch the zip/tar through CryptoStruct instead of the member files straight from storage")
    p.add_argument("--format", dest="formats", action="append", choices=FORMAT_TOKENS, metavar="TOKEN",
                   help="also download a derived format for tick items (%s); repeatable" % "|".join(FORMAT_TOKENS))
    p.add_argument("--only", choices=("tick", "bundles"), help="restrict to single files or to bundles")
    p.add_argument("--from", dest="date_from", metavar="YYYY-MM-DD", help="first UTC day to include")
    p.add_argument("--to", dest="date_to", metavar="YYYY-MM-DD", help="last UTC day to include")
    p.add_argument("--max-wait-min", type=float, default=240, metavar="N", help="max minutes to wait on one pending/cold item (default 240)")
    p.add_argument("--stuck-day-min", type=float, default=DEFAULT_STUCK_DAY_MIN, metavar="N",
                   help="range chunks: when the day they wait on makes no progress for N minutes, deliver the other "
                        "days one by one instead (default %g; 0 = keep waiting on the range)" % DEFAULT_STUCK_DAY_MIN)
    p.add_argument("--page-wait-min", type=float, default=DEFAULT_PAGE_WAIT_MIN, metavar="N",
                   help="large orders: max minutes to wait on one file before leaving it for the next run (default %g)" % DEFAULT_PAGE_WAIT_MIN)
    p.add_argument("--restore-max", type=int, default=None, metavar="N",
                   help="with --restore: max restore requests per run (large orders default %d)" % DEFAULT_RESTORE_MAX)
    p.add_argument("--json", action="store_true", help="machine-readable final report on stdout (progress on stderr)")
    p.add_argument("--base", default=DEFAULT_BASE, metavar="URL", help="site base URL (default %s)" % DEFAULT_BASE)
    p.add_argument("--yes", action="store_true", help="assume yes on interactive prompts")
    p.add_argument("--version", action="version", version="cryptostruct-dl %s" % __version__)
    return p


def stdin_is_tty():
    try:
        return bool(sys.stdin and sys.stdin.isatty())
    except (AttributeError, ValueError):
        return False


# ───────────────────────────── Large orders (paged manifests) ─────────────────────────────


class PagedTotals:
    """Counters over every page — the per-file results are dropped after each page (bounded memory)."""

    COUNTED = ("ok", "failed", "cold", "pending", "empty", "interrupted", "aborted")
    PROBLEMS = ("failed", "cold", "pending", "aborted")

    def __init__(self):
        self.c = dict((k, 0) for k in self.COUNTED)
        self.ok_bytes = 0
        self.already_here = 0
        self.downloaded_before = 0
        self.before_last = None
        self.listed = 0
        self.pages = 0
        self.problems = []
        self.problems_dropped = 0

    def add_plan(self, plan):
        self.pages += 1
        self.listed += len(plan.tasks)
        self.already_here += len(plan.local_tasks)
        self.downloaded_before += len(plan.before_tasks)
        if plan.before_last and (self.before_last is None or plan.before_last > self.before_last):
            self.before_last = plan.before_last

    def add_results(self, results):
        for r in results:
            s = r.get("status")
            if s in self.c:
                self.c[s] += 1
            if s == "ok":
                self.ok_bytes += int(r.get("bytes") or 0)
            if s in self.PROBLEMS:
                if len(self.problems) < REPORT_MAX_ROWS:
                    self.problems.append(r)
                else:
                    self.problems_dropped += 1

    def exit_code(self, interrupted):
        if interrupted:
            return 3
        c = self.c
        if c["failed"] or c["cold"] or c["pending"] or c["aborted"] or c["interrupted"]:
            return 1
        return 0

    def human_lines(self, elapsed, interrupted, opts):
        c = self.c
        head = "Interrupted after" if interrupted else "Done in"
        line = "%s %s — %d pages · %d ok (%s) · %d already here · %d downloaded before" % (
            head, human_duration(elapsed), self.pages, c["ok"], human_bytes(self.ok_bytes), self.already_here, self.downloaded_before)
        extras = []
        for key, label in (("failed", "failed"), ("cold", "cold"), ("pending", "not ready yet"), ("empty", "empty"),
                           ("interrupted", "not finished")):
            if c[key]:
                extras.append("%d %s" % (c[key], label))
        if extras:
            line += " · " + " · ".join(extras)
        lines = [line]
        if c["pending"] or c["cold"]:
            lines.append("  %d files are not ready yet (being prepared, or restored from cold storage — that takes up to ~48 h). "
                         "Re-run the same command later: finished files are skipped, the rest is picked up." % (c["pending"] + c["cold"]))
        groups = (("failed", "Failed:"), ("cold", "Cold:"), ("pending", "Not ready yet (first 20):"))
        for status, title in groups:
            rows = [r for r in self.problems if r.get("status") == status]
            if not rows:
                continue
            lines.append(title)
            for r in rows[:20]:
                lines.append("  %s — %s" % (r.get("filename"), r.get("message") or status))
            n = self.c[status]
            if n > 20:
                lines.append("  … and %d more" % (n - 20))
        if interrupted:
            lines.append("Re-run the same command to resume; finished files are in the ledger and will not be fetched again.")
        return lines

    def as_json(self, exit_code, interrupted, elapsed, opts, first, masker, out_dir):
        c = self.c
        return {
            "tool": "cryptostruct-dl",
            "version": __version__,
            "generated_at": utc_now_iso(),
            "exit_code": exit_code,
            "interrupted": interrupted,
            "scope": first.scope,
            "paged": True,
            "pages": self.pages,
            "manifest_source": masker.mask(first.source),
            "out": out_dir,
            "results": {
                "ok": c["ok"],
                "ok_bytes": self.ok_bytes,
                "failed": c["failed"],
                "cold": c["cold"],
                "left_pending": c["pending"],
                "empty": c["empty"],
                "interrupted": c["interrupted"] + c["aborted"],
                "already_here": self.already_here,
                "downloaded_before": self.downloaded_before,
            },
            "problems": list(self.problems),
            "problems_truncated": self.problems_dropped,
            "elapsed_seconds": round(elapsed, 1),
        }


def run_paged(first, args, opts, http, clock, stop, log, masker, out_dir, token, date_from, date_to):
    """Walk a paged manifest (a large order) — or an account manifest's large orders — page by page.

    Each page is planned and downloaded on its own; a file that is not ready within
    --page-wait-min is left for the next run (the server lists it again until it was
    delivered). Memory stays bounded: no page is kept after it ran."""
    ledger = SqliteLedger(out_dir, log)
    ledger.load()
    sidecar = AppendSidecar(out_dir)
    verify = Verify(args.verify, log)
    opts.max_wait_s = float(opts.page_wait_min) * 60.0
    if opts.restore_max is None:
        opts.restore_max = DEFAULT_RESTORE_MAX
    session = Session(opts, http, log, masker, ledger, sidecar, verify, out_dir, first.contract)
    session.stop = stop
    session.clock = clock
    totals = PagedTotals()
    started = time.monotonic()
    state = {"interrupted": False}

    def prepare(m):
        for sec in m.secrets():
            masker.add(sec)
        m.absolutize(opts.base)
        m.with_token(token)
        m.apply_filters(only=args.only, date_from=date_from, date_to=date_to)

    def run_page(m, label):
        prepare(m)
        planner = Planner(out_dir, ledger, http, log, opts)
        tasks = planner.build_tasks(m)
        plan = planner.classify(tasks, dry_run=args.dry_run, include_downloaded=args.include_downloaded)
        totals.add_plan(plan)
        log.out("%s: %s" % (label, plan.plan_line()))
        if args.dry_run:
            for ln in dry_run_table(plan)[:20]:
                log.out(ln)
            return
        if not plan.new_tasks:
            return
        if state["interrupted"]:
            return
        downloader = Downloader(session, plan)
        if downloader.run():
            state["interrupted"] = True
        totals.add_results(session.results)
        session.results = []
        ledger.save()
        sidecar.save()
        if session.fatal is not None:
            raise session.fatal

    def walk(m, what):
        n = 1
        if m.page is not None:
            p = m.page
            od = p.get("order_downloads") if isinstance(p.get("order_downloads"), dict) else {}
            log.out("%s: large order %s — %s files, %s; listed page by page%s" % (
                what, p.get("order_id"), p.get("order_files"), human_bytes(p.get("order_bytes")),
                (" (%d downloaded before are left out — --include-downloaded fetches them again)" % as_int(od.get("files_downloaded")))
                if m.new_only and as_int(od.get("files_downloaded")) > 0 else ""))
        while True:
            run_page(m, "%s page %d" % (what, n))
            if args.dry_run:
                log.out("(dry run: page 1 only — the order is listed page by page)")
                return
            if state["interrupted"] or session.fatal is not None:
                return
            nxt = m.page.get("next") if m.page else None
            if not nxt:
                return
            m = Manifest.fetch(http, nxt, clock, log, masker)
            n += 1

    if first.page is not None:
        walk(first, "order")
    else:
        if first.items or first.bundles:
            run_page(first, "account files")
        for lo in first.large_orders:
            if state["interrupted"] or session.fatal is not None:
                break
            url = lo.get("manifest_url")
            if not looks_like_url(url):
                url = urllib.parse.urljoin((first.site or opts.base).rstrip("/") + "/", str(url).lstrip("/"))
            if args.include_downloaded:
                url = set_query(url, new="0")
            for sec in query_values(url, "t"):
                masker.add(sec)
            walk(Manifest.fetch(http, url, clock, log, masker), "order %s" % lo.get("order_id"))

    ledger.save()
    elapsed = time.monotonic() - started
    if args.dry_run:
        return 0
    code = totals.exit_code(state["interrupted"])
    for ln in totals.human_lines(elapsed, state["interrupted"], opts):
        log.out(ln)
    if args.json:
        print(json.dumps(totals.as_json(code, state["interrupted"], elapsed, opts, first, masker, out_dir), indent=1, sort_keys=True))
    return code


def main(argv=None):
    try:
        return _main(sys.argv[1:] if argv is None else list(argv))
    except UsageError as e:
        sys.stderr.write("error: %s\n" % e)
        return 2
    except ExpiredLink as e:
        sys.stderr.write("%s\n" % e)
        return 2
    except ManifestUnreachable as e:
        sys.stderr.write("error: %s\n" % e)
        return 1
    except KeyboardInterrupt:
        sys.stderr.write("interrupted\n")
        return 3


def _main(argv):
    parser = build_parser()
    try:
        args = parser.parse_args(argv)
    except SystemExit as e:
        code = e.code
        return int(code) if isinstance(code, int) else (0 if code is None else 2)

    token = args.token or os.environ.get("CRYPTOSTRUCT_DOWNLOAD_TOKEN") or None
    key = args.key or os.environ.get("CRYPTOSTRUCT_DOWNLOAD_KEY") or None
    masker = Masker()
    masker.add(token)
    masker.add(key)
    log = Log(masker, json_mode=args.json)

    if args.parallel < 1:
        raise UsageError("--parallel must be >= 1")
    if args.max_wait_min < 0:
        raise UsageError("--max-wait-min must be >= 0")
    if args.stuck_day_min < 0:
        raise UsageError("--stuck-day-min must be >= 0")
    if args.page_wait_min < 0:
        raise UsageError("--page-wait-min must be >= 0")
    if args.restore_max is not None and args.restore_max < 0:
        raise UsageError("--restore-max must be >= 0")
    opts = Options(args)
    date_from = parse_date(args.date_from, "--from")
    date_to = parse_date(args.date_to, "--to")
    if date_from and date_to and date_from > date_to:
        raise UsageError("--from is after --to")
    if not (args.manifest or args.order or key):
        raise UsageError("give --order ORDER_ID with --token/--key, or --key alone, or --manifest FILE_OR_URL")
    if args.order and not (token or key):
        raise UsageError("--order needs --token (or CRYPTOSTRUCT_DOWNLOAD_TOKEN) or --key (or CRYPTOSTRUCT_DOWNLOAD_KEY)")

    http = Http(masker, bearer=key)
    stop = threading.Event()
    clock = Clock(stop)
    out_dir = os.path.abspath(args.out)
    os.makedirs(out_dir, exist_ok=True)

    # -- manifest -----------------------------------------------------------
    if args.manifest:
        if looks_like_url(args.manifest):
            url = args.manifest
            if token and not query_values(url, "t"):
                url = add_query(url, t=token)
            log.progress("manifest: %s" % url)
            manifest = Manifest.fetch(http, url, clock, log, masker)
        else:
            log.progress("manifest: %s" % args.manifest)
            manifest = Manifest.load_file(args.manifest)
    elif args.order:
        if key:
            url = "%s/api/download/order/%s/manifest" % (opts.base, urllib.parse.quote(str(args.order), safe=""))
        else:
            url = "%s/api/download/order/%s/manifest?t=%s" % (opts.base, urllib.parse.quote(str(args.order), safe=""), urllib.parse.quote(token, safe=""))
        # A large order's manifest is new-only unless asked otherwise (a small
        # order's JSON lists everything either way — same answer).
        if args.include_downloaded:
            url = set_query(url, new="0")
        log.progress("manifest: %s" % url)
        manifest = Manifest.fetch(http, url, clock, log, masker)
    else:
        url = "%s/api/download/manifest" % opts.base
        log.progress("manifest: %s" % url)
        manifest = Manifest.fetch(http, url, clock, log, masker)

    if manifest.paged:
        return run_paged(manifest, args, opts, http, clock, stop, log, masker, out_dir, token, date_from, date_to)

    for s in manifest.secrets():
        masker.add(s)
    manifest.absolutize(opts.base)
    manifest.with_token(token)
    if manifest.truncated:
        log.warn("the server truncated this manifest — it lists only part of your files; re-run after this pass to pick up the rest")
    omitted_files = as_int(manifest.omitted.get("files"))
    omitted_last = str(manifest.omitted.get("last_at") or "")[:10] or None
    if manifest.new_only and omitted_files > 0:
        log.out("  %d files (%s) downloaded before%s are not listed in this new-only manifest — request it with ?new=0 to include them" % (
            omitted_files, human_bytes(manifest.omitted.get("bytes")), (" (last %s)" % omitted_last) if omitted_last else ""))
    n_days = sum(len(b.get("days") or []) for b in manifest.bundles)
    n_chunks = sum(len(b.get("range_chunks") or []) for b in manifest.bundles)
    log.progress("manifest ok: scope %s · %d files · %d bundle days · %d range chunks" % (manifest.scope, len(manifest.items), n_days, n_chunks))
    manifest.apply_filters(only=args.only, date_from=date_from, date_to=date_to)

    # -- plan -----------------------------------------------------------------
    # Once a large order created the SQLite ledger here, every run uses it.
    ledger = SqliteLedger(out_dir, log) if SqliteLedger.exists(out_dir) else Ledger(out_dir, log)
    ledger.load()
    planner = Planner(out_dir, ledger, http, log, opts)
    tasks = planner.build_tasks(manifest)
    plan = planner.classify(tasks, dry_run=args.dry_run, include_downloaded=args.include_downloaded)
    log.out(plan.plan_line())
    if plan.before_tasks and not args.include_downloaded:
        log.out("  %d downloaded before%s — add --include-downloaded to fetch them again" % (
            len(plan.before_tasks), (" (last %s)" % plan.before_last) if plan.before_last else ""))

    if args.dry_run:
        if tasks:
            for ln in dry_run_table(plan):
                log.out(ln)
        else:
            log.out("(the manifest lists nothing to download)")
        if args.json:
            report = Report(plan, [], 0.0, False, opts, manifest, out_dir)
            print(json.dumps(report.as_json(0, masker), indent=1, sort_keys=True))
        ledger.save()
        return 0

    order_scope = bool(args.order) or manifest.scope == "order"
    if not tasks and omitted_files > 0 and order_scope:
        # a new-only manifest of a fully downloaded order lists nothing at all
        log.out("You downloaded this order on %s (%d files, %s). Downloading again transfers everything once more." % (
            omitted_last or "an earlier date", omitted_files, human_bytes(manifest.omitted.get("bytes"))))
        log.out("Nothing new to download — fetch the manifest with ?new=0 and add --include-downloaded to fetch the order again.")
        ledger.save()
        return 0

    if not plan.new_tasks and plan.before_tasks and order_scope:
        last = plan.before_last
        for o in manifest.orders:
            la = ((o.get("downloads") or {}).get("last_at") if isinstance(o.get("downloads"), dict) else None)
            if la and (last is None or str(la)[:10] > str(last)[:10]):
                last = str(la)[:10]
        n_files = len(plan.before_tasks)
        total_bytes = plan.before_bytes or sum(int(o.get("bytes") or 0) for o in manifest.orders)
        log.out("You downloaded this order on %s (%d files, %s). Downloading again transfers everything once more." % (
            last or "an earlier date", n_files, human_bytes(total_bytes)))
        proceed = False
        if stdin_is_tty():
            if args.yes:
                proceed = True
            else:
                try:
                    answer = input("Re-download? [y/N] ")
                except (EOFError, KeyboardInterrupt):
                    answer = ""
                proceed = answer.strip().lower() in ("y", "yes")
        if not proceed:
            log.out("Nothing new to download — add --include-downloaded to fetch the order again.")
            ledger.save()
            return 0
        plan = planner.classify(tasks, dry_run=False, include_downloaded=True)
        opts.include_downloaded = True
        log.out(plan.plan_line())

    if not plan.new_tasks:
        log.out("Nothing to download — everything is already here." if plan.local_tasks and not plan.before_tasks else "Nothing to download.")
        ledger.save()
        return 0

    # -- transfer -------------------------------------------------------------
    sidecar = Sidecar(out_dir)
    verify = Verify(args.verify, log)
    session = Session(opts, http, log, masker, ledger, sidecar, verify, out_dir, manifest.contract)
    session.stop = stop
    session.clock = clock
    started = time.monotonic()
    downloader = Downloader(session, plan)
    interrupted = downloader.run()
    ledger.save()
    sidecar.save()
    if session.fatal is not None:
        raise session.fatal
    elapsed = time.monotonic() - started
    report = Report(plan, session.results, elapsed, interrupted, opts, manifest, out_dir)
    code = report.exit_code()
    if args.json:
        for ln in report.human_lines():
            log.out(ln)
        print(json.dumps(report.as_json(code, masker), indent=1, sort_keys=True))
    else:
        for ln in report.human_lines():
            log.out(ln)
    return code


if __name__ == "__main__":
    sys.exit(main())