docs/docs/cli/zot-fetch.md

@@ -0,0 +1,35 @@
+---
+title: stack zot fetch
+sidebar_position: 87
+---
+
+# `stack zot fetch`
+
+```
+Usage: stack zot fetch [OPTIONS]                                               
+                                                                                
+ Attach full text (PDF/EPUB) to items that lack it.                             
+                                                                                
+ DOI → Unpaywall, then the libgen index; ISBN → libgen; delivery by             
+ libgen or Anna's fast download (member key). Items with a PDF/EPUB,            
+ deleted items and ``document`` items are always skipped. Every                 
+ attempt is a line in the ledger; misses wait 30 days unless --retry.           
+                                                                                
+╭─ Options ────────────────────────────────────────────────────────────────────╮
+│ --collection                 TEXT     Collection name.                       │
+│ --tag                        TEXT     Tag name.                              │
+│ --type                       TEXT     Item type, e.g. journalArticle, book.  │
+│ --item                       TEXT     One item key.                          │
+│ --all                                 Every eligible item.                   │
+│ --limit                      INTEGER  Stop after N attempts (0 = all).       │
+│                                       [default: 0]                           │
+│ --dry-run                             List candidates, write nothing.        │
+│ --retry                               Retry items the ledger marked as       │
+│                                       misses.                                │
+│ --db                         PATH                                            │
+│ --ledger                     PATH     [default:                              │
+│                                       data/zotero/fetch_ledger.jsonl]        │
+│ --hold          --no-hold             [default: hold]                        │
+│ --help                                Show this message and exit.            │
+╰──────────────────────────────────────────────────────────────────────────────╯
+```

docs/docs/cli/zot.md

@@ -21,5 +21,6 @@ Usage: stack zot [OPTIONS] COMMAND [ARGS]...
 │ fix-fields     Remap base-field rows to type-specific fieldIDs; drop         │
 │                orphans.                                                      │
 │ verify-parity  Check that ``zot.db`` constants match the live Zotero schema. │
+│ fetch          Attach full text (PDF/EPUB) to items that lack it.            │
 ╰──────────────────────────────────────────────────────────────────────────────╯
 ```

docs/superpowers/specs/2026-09-09-zotero-annas-connector-design.md

@@ -1,6 +1,6 @@
 # Zotero full-text connector (Anna's Archive, libgen, Unpaywall) — design
 
-**Status:** draft for operator approval, 2026-09-09.
+**Status:** Z1 approved and built 2026-09-22/23 (plan `docs/superpowers/plans/2026-09-22-zotero-fetch-z1.md`); Z2–Z4 pending.
 **Related:** `dev/scripts/add_ama_coding_to_zotero.py` (PR #697), corwins.media issue #707 (Anna's key), corwins spec `2026-09-09-direct-source-tier-design.md` (egress proxy).
 
 ## Problem

src/cli/zot.py

@@ -7,11 +7,18 @@ they run (SQLite file is WAL-locked by the desktop app). ``--hold``
 
 from __future__ import annotations
 
+import os
 import subprocess
 from pathlib import Path
 
 import typer
 
+from zot.db import Db
+from zot.fetch.deliver import AnnasAuthError, Budget
+from zot.fetch.ledger import Ledger
+from zot.fetch.run import make_clients, run
+from zot.fetch.select import select_items
+
 app = typer.Typer(no_args_is_help=True)
 
 _DEFAULT_DB = Path("data/zotero/data/zotero.sqlite")
@@ -126,3 +133,94 @@ def verify_parity(
 
     with Db(str(_resolve(db_path))):
         typer.echo("parity OK")
+
+
+_DEFAULT_LEDGER = Path("data/zotero/fetch_ledger.jsonl")
+_PROXY_HELP = (
+    "VPN_HTTP_PROXY is not set. libgen and Anna's must go through the gluetun "
+    "HTTP proxy published by corwins.media (docker compose up gluetun → "
+    "127.0.0.1:19035): add VPN_HTTP_PROXY=http://127.0.0.1:19035 to .env and "
+    "source it (set -a; . ./.env; set +a) before running."
+)
+
+
+@app.command(name="fetch")
+def fetch(
+    collection: str = typer.Option("", "--collection", help="Collection name."),
+    tag: str = typer.Option("", "--tag", help="Tag name."),
+    type_: str = typer.Option(
+        "", "--type", help="Item type, e.g. journalArticle, book."
+    ),
+    item: str = typer.Option("", "--item", help="One item key."),
+    all_: bool = typer.Option(False, "--all", help="Every eligible item."),
+    limit: int = typer.Option(0, "--limit", help="Stop after N attempts (0 = all)."),
+    dry_run: bool = typer.Option(
+        False, "--dry-run", help="List candidates, write nothing."
+    ),
+    retry: bool = typer.Option(
+        False, "--retry", help="Retry items the ledger marked as misses."
+    ),
+    db_path: Path | None = typer.Option(None, "--db"),
+    ledger_path: Path = typer.Option(_DEFAULT_LEDGER, "--ledger"),
+    hold: bool = typer.Option(True, "--hold/--no-hold"),
+) -> None:
+    """Attach full text (PDF/EPUB) to items that lack it.
+
+    DOI → Unpaywall, then the libgen index; ISBN → libgen; delivery by
+    libgen or Anna's fast download (member key). Items with a PDF/EPUB,
+    deleted items and ``document`` items are always skipped. Every
+    attempt is a line in the ledger; misses wait 30 days unless --retry.
+    """
+    from conf import path as conf_path
+
+    proxy = os.environ.get("VPN_HTTP_PROXY", "")
+    if not proxy:
+        raise typer.BadParameter(_PROXY_HELP)
+    annas_key = os.environ.get("ANNAS_ARCHIVE_KEY", "")
+    if not annas_key:
+        typer.echo(
+            "ANNAS_ARCHIVE_KEY not set — Anna's fast download disabled for this run."
+        )
+    email = os.environ.get("UNPAYWALL_EMAIL", "dev@fhirworx.io")
+    reserve = int(os.environ.get("ANNAS_DAILY_RESERVE", "5"))
+    path = _resolve(db_path)
+
+    def _go() -> dict[str, int]:
+        with Db(str(path)) as db:
+            try:
+                refs = select_items(
+                    db,
+                    collection=collection,
+                    tag=tag,
+                    type_=type_,
+                    item=item,
+                    all_=all_,
+                )
+            except ValueError as exc:
+                raise typer.BadParameter(str(exc)) from exc
+            typer.echo(f"{len(refs)} eligible item(s)")
+            clients = make_clients(proxy, email=email)
+            try:
+                return run(
+                    db,
+                    refs,
+                    ledger=Ledger(ledger_path),
+                    budget=Budget(annas_key=annas_key, reserve=reserve),
+                    clients=clients,
+                    email=email,
+                    storage_dir=conf_path("storage.zotero"),
+                    scratch_dir=Path(".state/zot/fetch"),
+                    limit=limit,
+                    dry_run=dry_run,
+                    retry=retry,
+                    out=typer.echo,
+                )
+            except AnnasAuthError as exc:
+                typer.echo(f"error: {exc}", err=True)
+                raise typer.Exit(2) from exc
+            finally:
+                clients.direct.close()
+                clients.proxied.close()
+
+    stats = _hold_zotero(_go, hold)
+    typer.echo("fetch: " + " ".join(f"{k}={v}" for k, v in stats.items()))

src/zot/fetch/__init__.py

@@ -0,0 +1,22 @@
+"""Full-text connector for Zotero items (spec 2026-09-09, slice Z1).
+
+select → resolve → deliver → attach, with a JSONL ledger of attempts.
+"""
+
+from zot.fetch.deliver import AnnasAuthError, Budget
+from zot.fetch.ledger import Ledger
+from zot.fetch.resolve import Candidate
+from zot.fetch.run import Clients, make_clients, run
+from zot.fetch.select import ItemRef, select_items
+
+__all__ = [
+    "AnnasAuthError",
+    "Budget",
+    "Candidate",
+    "Clients",
+    "ItemRef",
+    "Ledger",
+    "make_clients",
+    "run",
+    "select_items",
+]

src/zot/fetch/attach.py

@@ -0,0 +1,77 @@
+"""Put a delivered file into Zotero storage and register it as a child.
+
+Generalises ``dev/scripts/add_ama_coding_to_zotero.py::_attach``: the
+storage tree is owned by the zotero container's uid (100999), so the
+default copier shells out to ``sudo``; tests inject a plain copier.
+"""
+
+from __future__ import annotations
+
+import subprocess
+from collections.abc import Callable
+from pathlib import Path
+
+from zot.db import Db, generate_key
+
+FULL_TEXT_TYPES = {"application/pdf", "application/epub+zip"}
+CONTENT_TYPES = {".pdf": "application/pdf", ".epub": "application/epub+zip"}
+ZOTERO_UID_GID = "100999:100999"
+
+Copier = Callable[[Path, Path], None]
+
+
+def has_full_text(db: Db, item_id: int) -> bool:
+    """True when *item_id* already has a PDF or EPUB child attachment."""
+    rows = db.con.execute(
+        "SELECT contentType FROM itemAttachments WHERE parentItemID = ?", (item_id,)
+    ).fetchall()
+    return any((r[0] or "") in FULL_TEXT_TYPES for r in rows)
+
+
+def sudo_copier(src: Path, dest_dir: Path) -> None:
+    """Copy *src* into *dest_dir* as the zotero container's owner."""
+    subprocess.run(["sudo", "mkdir", "-p", str(dest_dir)], check=True)
+    subprocess.run(["sudo", "cp", str(src), str(dest_dir / src.name)], check=True)
+    subprocess.run(["sudo", "chown", "-R", ZOTERO_UID_GID, str(dest_dir)], check=True)
+
+
+def append_extra(db: Db, item_id: int, line: str) -> None:
+    """Append *line* to the item's ``extra`` field unless already present."""
+    existing = db.get_field(item_id, "extra") or ""
+    if line in existing.splitlines():
+        return
+    db.set_field(item_id, "extra", f"{existing.rstrip()}\n{line}".strip())
+
+
+def attach_file(
+    db: Db,
+    parent_id: int,
+    path: Path,
+    *,
+    source: str,
+    storage_dir: Path,
+    md5: str = "",
+    copier: Copier = sudo_copier,
+) -> int:
+    """Attach *path* to *parent_id*; returns the attachment ``itemID``.
+
+    Provenance: tag ``source:<source>`` on the parent and, when an md5
+    is known, an ``md5: <md5>`` line in ``extra``. Nothing else on the
+    parent changes. The caller commits; on any exception it must roll
+    the transaction back.
+    """
+    ctype = CONTENT_TYPES.get(path.suffix.lower())
+    if ctype is None:
+        raise ValueError(f"unsupported attachment type: {path.suffix!r} ({path.name})")
+    att_key = generate_key()
+    att_id = db.add_attachment(
+        parent_id, key=att_key, content_type=ctype, path=f"storage:{path.name}"
+    )
+    db.set_field(att_id, "title", path.name)
+    db.tag_item(parent_id, f"source:{source}")
+    if md5:
+        append_extra(db, parent_id, f"md5: {md5}")
+    # Rows first, file second: a failed copy leaves rows the caller rolls
+    # back; the only orphan case is a failed commit after a good copy.
+    copier(path, Path(storage_dir) / att_key)
+    return att_id

src/zot/fetch/deliver.py

@@ -0,0 +1,229 @@
+"""Candidate → file on disk.
+
+libgen: ``ads.php?md5=`` (browser UA + Referer; a 200 with an error div
+and no ``get.php`` link is a throttle → one retry after 3 s) then the
+keyed ``get.php``. Anna's: the member fast-download JSON API by md5,
+guarded by a daily reserve. Unpaywall URLs download directly.
+"""
+
+from __future__ import annotations
+
+import re
+import time
+from collections.abc import Callable
+from dataclasses import dataclass
+from pathlib import Path
+
+import httpx
+
+from zot.fetch.resolve import LIBGEN, Candidate, Notes, Pacer
+
+ANNAS = "https://annas-archive.gl"
+MIN_BYTES = 10_240
+MAGIC = {"pdf": b"%PDF", "epub": b"PK"}
+_GET_LINK = re.compile(r"get\.php\?md5=([0-9a-f]{32})&(?:amp;)?key=([A-Za-z0-9]+)")
+_ADS_RETRY_SECONDS = 3.0
+_KEY_ERRORS = ("secret key", "not a member", "invalid key", "membership")
+
+
+class AnnasAuthError(RuntimeError):
+    """The Anna's key is wrong or the account is not a member — stop the run."""
+
+
+@dataclass
+class Budget:
+    annas_key: str = ""
+    reserve: int = 5
+    annas_left: int | None = None
+    annas_disabled: bool = False
+    quota_hit: bool = False
+
+    def can_use_annas(self) -> bool:
+        if not self.annas_key or self.annas_disabled:
+            return False
+        return self.annas_left is None or self.annas_left > self.reserve
+
+
+def download(
+    client: httpx.Client,
+    url: str,
+    dest: Path,
+    *,
+    ext: str,
+    headers: dict[str, str] | None = None,
+    notes: Notes | None = None,
+    what: str = "download",
+) -> Path | None:
+    """Stream *url* to *dest*; None unless it is a real file of type *ext*."""
+    notes = notes if notes is not None else Notes()
+    try:
+        with client.stream(
+            "GET", url, headers=headers, timeout=120, follow_redirects=True
+        ) as r:
+            if r.status_code != 200:
+                if r.status_code >= 500:
+                    notes.error(f"{what}: http {r.status_code}")
+                else:
+                    notes.miss(f"{what}: http {r.status_code}")
+                return None
+            if "text/html" in r.headers.get("content-type", "").lower():
+                notes.miss(f"{what}: answered HTML, not a file")
+                return None
+            dest.parent.mkdir(parents=True, exist_ok=True)
+            with dest.open("wb") as fh:
+                for chunk in r.iter_bytes(1 << 16):
+                    fh.write(chunk)
+    except (httpx.HTTPError, OSError) as exc:
+        dest.unlink(missing_ok=True)
+        notes.error(f"{what}: {type(exc).__name__}")
+        return None
+    size = dest.stat().st_size
+    with dest.open("rb") as fh:
+        head = fh.read(4)
+    if size < MIN_BYTES or not head.startswith(MAGIC[ext]):
+        dest.unlink(missing_ok=True)
+        notes.miss(f"{what}: not a {ext} ({size} bytes)")
+        return None
+    return dest
+
+
+def _ads_key(client: httpx.Client, md5: str, pacer: Pacer, notes: Notes) -> str | None:
+    pacer.wait()
+    try:
+        r = client.get(
+            f"{LIBGEN}/ads.php",
+            params={"md5": md5},
+            headers={"Referer": f"{LIBGEN}/"},
+            timeout=30,
+        )
+    except httpx.HTTPError as exc:
+        notes.error(f"libgen ads: {type(exc).__name__}")
+        return None
+    if r.status_code != 200:
+        notes.error(f"libgen ads: http {r.status_code}")
+        return None
+    m = _GET_LINK.search(r.text)
+    if m is None:
+        notes.error("libgen ads: no get link (throttle page?)")
+        return None
+    if m.group(1) != md5:
+        notes.miss("libgen ads: page links a different md5")
+        return None
+    return m.group(2)
+
+
+def fetch_libgen(
+    client: httpx.Client,
+    cand: Candidate,
+    dest: Path,
+    *,
+    pacer: Pacer,
+    sleep: Callable[[float], None],
+    notes: Notes | None = None,
+) -> Path | None:
+    notes = notes if notes is not None else Notes()
+    md5 = cand.md5 or ""
+    key = _ads_key(client, md5, pacer, notes)
+    if key is None:
+        sleep(_ADS_RETRY_SECONDS)
+        key = _ads_key(client, md5, pacer, notes)
+    if key is None:
+        return None
+    return download(
+        client,
+        f"{LIBGEN}/get.php?md5={md5}&key={key}",
+        dest,
+        ext=cand.ext,
+        headers={"Referer": f"{LIBGEN}/ads.php?md5={md5}"},
+        notes=notes,
+        what="libgen get",
+    )
+
+
+def _error_text(r: httpx.Response) -> str:
+    try:
+        data = r.json()
+    except ValueError:
+        return r.text[:200]
+    return str(data.get("error") or "")[:200] if isinstance(data, dict) else ""
+
+
+def fetch_annas(
+    client: httpx.Client,
+    cand: Candidate,
+    dest: Path,
+    *,
+    budget: Budget,
+    notes: Notes | None = None,
+) -> Path | None:
+    notes = notes if notes is not None else Notes()
+    try:
+        r = client.get(
+            f"{ANNAS}/dyn/api/fast_download.json",
+            params={"md5": cand.md5 or "", "key": budget.annas_key},
+            timeout=30,
+        )
+    except httpx.HTTPError as exc:
+        notes.error(f"annas: {type(exc).__name__}")
+        return None
+    if r.status_code in (401, 403):
+        err = _error_text(r)
+        if r.status_code == 401 or any(k in err.lower() for k in _KEY_ERRORS):
+            raise AnnasAuthError(
+                f"Anna's Archive refused the key ({r.status_code}): {err}"
+            )
+        # 403 for anything else (e.g. "No downloads left") → quota exhausted
+        budget.annas_disabled = True
+        budget.annas_left = 0
+        notes.error(f"annas: {err or 'http 403'} — disabled for this run")
+        return None
+    if r.status_code not in (200, 204):
+        budget.annas_disabled = True
+        notes.error(f"annas: http {r.status_code} — disabled for this run")
+        return None
+    try:
+        data = r.json()
+    except ValueError:
+        notes.error("annas: non-JSON answer")
+        return None
+    info = data.get("account_fast_download_info") or {}
+    if isinstance(info.get("downloads_left"), int):
+        budget.annas_left = info["downloads_left"]
+    url = data.get("download_url")
+    if not url:
+        notes.miss(f"annas: {data.get('error') or 'no download_url'}")
+        return None
+    return download(client, url, dest, ext=cand.ext, notes=notes, what="annas get")
+
+
+def fetch(
+    cand: Candidate,
+    dest: Path,
+    *,
+    direct: httpx.Client,
+    proxied: httpx.Client,
+    budget: Budget,
+    pacer: Pacer,
+    sleep: Callable[[float], None] = time.sleep,
+    notes: Notes | None = None,
+) -> str | None:
+    """Deliver *cand* to *dest*; returns the source that worked, else None.
+
+    Sets ``budget.quota_hit`` when Anna's would have been tried but the
+    reserve (or an earlier API failure) blocked it.
+    """
+    notes = notes if notes is not None else Notes()
+    if cand.source == "unpaywall":
+        got = download(
+            direct, cand.url or "", dest, ext=cand.ext, notes=notes, what="unpaywall"
+        )
+        return "unpaywall" if got else None
+    if fetch_libgen(proxied, cand, dest, pacer=pacer, sleep=sleep, notes=notes):
+        return "libgen"
+    if budget.can_use_annas():
+        got = fetch_annas(proxied, cand, dest, budget=budget, notes=notes)
+        return "annas" if got else None
+    if budget.annas_key:
+        budget.quota_hit = True
+        notes.miss("annas: skipped (daily reserve or disabled)")
+    return None

src/zot/fetch/ledger.py

@@ -0,0 +1,95 @@
+"""JSONL ledger of fetch attempts — the report the operator reads.
+
+One line per attempt. ``should_try`` keeps a terminal miss
+(``no_candidate`` / ``all_failed``) from being retried for 30 days,
+never retries an ``attached`` item, and always retries ``quota``.
+"""
+
+from __future__ import annotations
+
+import json
+from collections import Counter
+from dataclasses import asdict, dataclass, field
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+TERMINAL = {"no_candidate", "all_failed"}
+RETRY_AFTER = timedelta(days=30)
+
+
+@dataclass(frozen=True)
+class Attempt:
+    item_key: str
+    outcome: str
+    source: str = ""
+    md5: str = ""
+    ids: dict[str, str] = field(default_factory=dict)
+    note: str = ""
+    ts: str = ""
+
+    @property
+    def when(self) -> datetime:
+        return datetime.fromisoformat(self.ts)
+
+
+def _now_iso(now: datetime | None) -> str:
+    return (now or datetime.now(timezone.utc)).isoformat()
+
+
+class Ledger:
+    def __init__(self, path: Path) -> None:
+        self.path = Path(path)
+        self._last: dict[str, Attempt] = {}
+        self._session: list[Attempt] = []
+        if self.path.is_file():
+            for line in self.path.read_text().splitlines():
+                try:
+                    att = Attempt(**json.loads(line))
+                except (ValueError, TypeError):
+                    continue
+                self._last[att.item_key] = att
+
+    def record(
+        self,
+        item_key: str,
+        *,
+        outcome: str,
+        source: str = "",
+        md5: str = "",
+        ids: dict[str, str] | None = None,
+        note: str = "",
+        now: datetime | None = None,
+    ) -> Attempt:
+        att = Attempt(
+            item_key=item_key,
+            outcome=outcome,
+            source=source,
+            md5=md5,
+            ids=dict(ids or {}),
+            note=note,
+            ts=_now_iso(now),
+        )
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        with self.path.open("a", encoding="utf-8") as fh:
+            fh.write(json.dumps(asdict(att), ensure_ascii=False) + "\n")
+        self._last[item_key] = att
+        self._session.append(att)
+        return att
+
+    def should_try(
+        self, item_key: str, *, retry: bool = False, now: datetime | None = None
+    ) -> bool:
+        if retry:
+            return True
+        last = self._last.get(item_key)
+        if last is None:
+            return True
+        if last.outcome == "attached":
+            return False
+        if last.outcome in TERMINAL:
+            current = now or datetime.now(timezone.utc)
+            return current - last.when >= RETRY_AFTER
+        return True
+
+    def summary(self) -> dict[str, int]:
+        return dict(Counter(a.outcome for a in self._session))

src/zot/fetch/resolve.py

@@ -0,0 +1,323 @@
+"""Identifiers → download candidates.
+
+DOI: Unpaywall (direct, legal) first, then the libgen JSON index (through
+the VPN proxy). ISBN: libgen JSON. A book with no ISBN: libgen title
+search → edition ids → JSON. libgen serves a bare nginx page to
+non-browser user agents, so the proxied client must send ``BROWSER_UA``.
+"""
+
+from __future__ import annotations
+
+import re
+import time
+from collections.abc import Callable, Iterable
+from dataclasses import dataclass
+
+import httpx
+
+from prisma.fetch import unpaywall_urls
+from zot.fetch.select import ItemRef
+
+LIBGEN = "https://libgen.li"
+BROWSER_UA = (
+    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
+    "(KHTML, like Gecko) Chrome/128.0 Safari/537.36"
+)
+_EDITION_RE = re.compile(r'edition\.php\?id=(\d+)"[^>]*>([^<]*)<')
+_YEAR_RE = re.compile(r"\b(19|20)\d\d\b")
+_EXT_ORDER = {"book": ("epub", "pdf"), "paper": ("pdf",)}
+
+
+class Notes:
+    """Why an item got no file: ``miss`` = a definitive answer (no edition,
+    404, bad file); ``error`` = the service never answered (transport,
+    5xx, throttle page). An item whose notes hold any error is retried
+    next run instead of waiting out the 30-day miss rule."""
+
+    def __init__(self) -> None:
+        self.items: list[tuple[str, str]] = []
+
+    def miss(self, text: str) -> None:
+        self.items.append(("miss", text))
+
+    def error(self, text: str) -> None:
+        self.items.append(("error", text))
+
+    @property
+    def had_error(self) -> bool:
+        return any(kind == "error" for kind, _ in self.items)
+
+    def __str__(self) -> str:
+        return "; ".join(f"{kind}: {text}" for kind, text in self.items)
+
+
+@dataclass(frozen=True)
+class Candidate:
+    source: str  # unpaywall | libgen
+    md5: str | None
+    url: str | None
+    ext: str
+    size_bytes: int | None
+    title: str
+
+
+class Pacer:
+    """Keep calls at least *min_interval* seconds apart."""
+
+    def __init__(
+        self,
+        min_interval: float,
+        *,
+        clock: Callable[[], float] = time.monotonic,
+        sleep: Callable[[float], None] = time.sleep,
+    ) -> None:
+        self.min_interval = min_interval
+        self._clock = clock
+        self._sleep = sleep
+        self._last: float | None = None
+
+    def wait(self) -> None:
+        now = self._clock()
+        if self._last is not None:
+            remaining = self.min_interval - (now - self._last)
+            if remaining > 0:
+                self._sleep(remaining)
+        self._last = self._clock()
+
+
+def _get_json(
+    client: httpx.Client,
+    url: str,
+    params: dict[str, str],
+    pacer: Pacer,
+    notes: Notes | None = None,
+) -> dict:
+    notes = notes if notes is not None else Notes()
+    what = f"libgen {params.get('object', '?')}"
+    pacer.wait()
+    try:
+        r = client.get(url, params=params, timeout=30)
+        if r.status_code != 200:
+            notes.error(f"{what}: http {r.status_code}")
+            return {}
+        data = r.json()
+    except httpx.HTTPError as exc:
+        notes.error(f"{what}: {type(exc).__name__}")
+        return {}
+    except ValueError:
+        notes.error(f"{what}: non-JSON answer (missing browser UA?)")
+        return {}
+    return data if isinstance(data, dict) else {}
+
+
+def libgen_editions(
+    client: httpx.Client,
+    *,
+    doi: str = "",
+    isbn: str = "",
+    ids: Iterable[str] = (),
+    pacer: Pacer,
+    notes: Notes | None = None,
+) -> dict:
+    params = {"object": "e", "addkeys": "*"}
+    if doi:
+        params["doi"] = doi
+    elif isbn:
+        params["isbn"] = isbn
+    else:
+        params["ids"] = ",".join(ids)
+    return _get_json(client, f"{LIBGEN}/json.php", params, pacer, notes)
+
+
+def libgen_files(
+    client: httpx.Client,
+    f_ids: list[str],
+    *,
+    pacer: Pacer,
+    notes: Notes | None = None,
+) -> dict:
+    if not f_ids:
+        return {}
+    params = {"object": "f", "addkeys": "*", "ids": ",".join(f_ids)}
+    return _get_json(client, f"{LIBGEN}/json.php", params, pacer, notes)
+
+
+def _int_or_none(value: object) -> int | None:
+    try:
+        return int(value) if value else None
+    except (TypeError, ValueError):
+        return None
+
+
+def _candidates_from_editions(
+    client: httpx.Client,
+    editions: dict,
+    *,
+    kind: str,
+    pacer: Pacer,
+    notes: Notes | None = None,
+) -> list[Candidate]:
+    files: list[tuple[str, str, str]] = []  # (f_id, md5, title)
+    for ed in editions.values():
+        if not isinstance(ed, dict):
+            continue
+        title = ed.get("title") or ""
+        raw = ed.get("files")
+        entries = raw.values() if isinstance(raw, dict) else (raw or [])
+        for f in entries:
+            if not isinstance(f, dict):
+                continue
+            md5 = (f.get("md5") or "").lower()
+            if md5:
+                files.append((str(f.get("f_id") or ""), md5, title))
+    info = libgen_files(
+        client, [fid for fid, _, _ in files if fid], pacer=pacer, notes=notes
+    )
+    out: list[Candidate] = []
+    for fid, md5, title in files:
+        meta = info.get(fid) or {}
+        ext = (meta.get("extension") or "").lower()
+        if not ext and kind == "paper":
+            ext = "pdf"
+        if ext not in _EXT_ORDER[kind]:
+            continue
+        size = _int_or_none(meta.get("filesize"))
+        out.append(Candidate("libgen", md5, None, ext, size, title))
+    order = _EXT_ORDER[kind]
+    out.sort(key=lambda c: order.index(c.ext))
+    return out
+
+
+def libgen_candidates(
+    client: httpx.Client,
+    *,
+    kind: str,
+    doi: str = "",
+    isbn: str = "",
+    pacer: Pacer,
+    notes: Notes | None = None,
+) -> list[Candidate]:
+    notes = notes if notes is not None else Notes()
+    editions = libgen_editions(client, doi=doi, isbn=isbn, pacer=pacer, notes=notes)
+    if not editions:
+        if not notes.had_error:
+            notes.miss(
+                f"libgen: no edition for {'doi ' + doi if doi else 'isbn ' + isbn}"
+            )
+        return []
+    out = _candidates_from_editions(
+        client, editions, kind=kind, pacer=pacer, notes=notes
+    )
+    if not out and not notes.had_error:
+        notes.miss("libgen: edition has no usable file")
+    return out
+
+
+def _norm(s: str) -> str:
+    return re.sub(r"[^a-z0-9]+", " ", s.lower()).strip()
+
+
+def _year(s: str) -> str:
+    m = _YEAR_RE.search(s or "")
+    return m.group(0) if m else ""
+
+
+def title_matches(target: str, year: str, entries: list[dict]) -> list[dict]:
+    """Exact normalised title, else containment with agreeing years
+    (the AMA script's folder-matching rule)."""
+    t = _norm(target)
+    exact = [e for e in entries if _norm(e.get("title", "")) == t]
+    if exact:
+        return exact
+    tight = t.replace(" ", "")
+    out: list[dict] = []
+    for e in entries:
+        ey = _year(e.get("title", "")) or str(e.get("year") or "")
+        if year and ey and year != ey:
+            continue
+        cand = _norm(e.get("title", "")).replace(" ", "")
+        if cand and (cand in tight or tight in cand):
+            out.append(e)
+    return out
+
+
+def title_candidates(
+    client: httpx.Client,
+    *,
+    title: str,
+    year: str,
+    pacer: Pacer,
+    notes: Notes | None = None,
+) -> list[Candidate]:
+    notes = notes if notes is not None else Notes()
+    if not title:
+        return []
+    pacer.wait()
+    params = {
+        "req": title,
+        "columns[]": "t",
+        "objects[]": "e",
+        "topics[]": "l",
+        "res": "25",
+    }
+    try:
+        r = client.get(f"{LIBGEN}/index.php", params=params, timeout=30)
+    except httpx.HTTPError as exc:
+        notes.error(f"libgen search: {type(exc).__name__}")
+        return []
+    if r.status_code != 200:
+        notes.error(f"libgen search: http {r.status_code}")
+        return []
+    entries = [{"id": i, "title": t.strip()} for i, t in _EDITION_RE.findall(r.text)]
+    hits = title_matches(title, year, entries)
+    if not hits:
+        notes.miss("libgen search: no title match")
+        return []
+    editions = libgen_editions(
+        client, ids=[h["id"] for h in hits], pacer=pacer, notes=notes
+    )
+    return _candidates_from_editions(
+        client, editions, kind="book", pacer=pacer, notes=notes
+    )
+
+
+def candidates(
+    ref: ItemRef,
+    *,
+    direct: httpx.Client,
+    proxied: httpx.Client,
+    email: str,
+    up_pacer: Pacer,
+    lg_pacer: Pacer,
+    notes: Notes | None = None,
+) -> list[Candidate]:
+    """Ordered candidates for one item: Unpaywall URLs, then libgen md5s."""
+    notes = notes if notes is not None else Notes()
+    out: list[Candidate] = []
+    if ref.doi:
+        up_pacer.wait()
+        urls = unpaywall_urls(direct, ref.doi, email)
+        if not urls:
+            notes.miss("unpaywall: no OA location (or unreachable)")
+        for url in urls:
+            out.append(Candidate("unpaywall", None, url, "pdf", None, ref.title))
+        out.extend(
+            libgen_candidates(
+                proxied, kind=ref.kind, doi=ref.doi, pacer=lg_pacer, notes=notes
+            )
+        )
+    elif ref.isbn:
+        out.extend(
+            libgen_candidates(
+                proxied, kind=ref.kind, isbn=ref.isbn, pacer=lg_pacer, notes=notes
+            )
+        )
+    elif ref.kind == "book" and ref.title:
+        out.extend(
+            title_candidates(
+                proxied, title=ref.title, year=ref.year, pacer=lg_pacer, notes=notes
+            )
+        )
+    else:
+        notes.miss("no identifiers (no DOI/ISBN; not a book with a title)")
+    return out

src/zot/fetch/run.py

@@ -0,0 +1,223 @@
+"""The per-item loop: ledger gate → candidates → deliver → attach → ledger."""
+
+from __future__ import annotations
+
+import hashlib
+import re
+import time
+from collections.abc import Callable
+from dataclasses import dataclass
+from pathlib import Path
+
+import httpx
+
+from zot.db import Db
+from zot.fetch.attach import Copier, attach_file, sudo_copier
+from zot.fetch.deliver import AnnasAuthError, Budget, fetch
+from zot.fetch.ledger import Ledger
+from zot.fetch.resolve import BROWSER_UA, Notes, Pacer, candidates
+from zot.fetch.select import ItemRef
+
+STATS = (
+    "seen",
+    "skipped",
+    "attempted",
+    "attached",
+    "no_candidate",
+    "all_failed",
+    "quota",
+    "error",
+)
+_UNSAFE = re.compile(r"[^A-Za-z0-9 ._-]+")
+
+
+@dataclass
+class Clients:
+    direct: httpx.Client
+    proxied: httpx.Client
+
+
+def make_clients(proxy: str, *, email: str) -> Clients:
+    direct = httpx.Client(
+        headers={"User-Agent": f"stack-zot-fetch/1 (mailto:{email})"},
+        follow_redirects=True,
+    )
+    proxied = httpx.Client(
+        proxy=proxy, headers={"User-Agent": BROWSER_UA}, follow_redirects=True
+    )
+    return Clients(direct=direct, proxied=proxied)
+
+
+def safe_name(title: str, fallback: str) -> str:
+    cleaned = " ".join(_UNSAFE.sub(" ", title or "").split())[:80].strip()
+    return cleaned or fallback
+
+
+def _scratch_path(scratch_dir: Path, ref: ItemRef, cand_key: str, ext: str) -> Path:
+    tag = hashlib.sha1(cand_key.encode()).hexdigest()[:10]  # noqa: S324 — not security
+    return scratch_dir / f"{ref.key}-{tag}.{ext}"
+
+
+def _cleanup(paths: list[Path]) -> None:
+    for p in paths:
+        p.unlink(missing_ok=True)
+
+
+def _one(
+    db: Db,
+    ref: ItemRef,
+    *,
+    notes: Notes,
+    scratch: list[Path],
+    budget: Budget,
+    clients: Clients,
+    email: str,
+    storage_dir: Path,
+    scratch_dir: Path,
+    dry_run: bool,
+    copier: Copier,
+    sleep: Callable[[float], None],
+    out: Callable[..., None],
+    up_pacer: Pacer,
+    lg_pacer: Pacer,
+    ledger: Ledger,
+) -> str:
+    """One item end to end; returns the stats key for its outcome
+    ("" for a dry run). Raises on unexpected failure — ``run`` isolates it."""
+    cands = candidates(
+        ref,
+        direct=clients.direct,
+        proxied=clients.proxied,
+        email=email,
+        up_pacer=up_pacer,
+        lg_pacer=lg_pacer,
+        notes=notes,
+    )
+    if dry_run:
+        out(f"{ref.key}  {ref.title[:60]}  ids={ref.ids}")
+        for c in cands:
+            out(f"    {c.source:<9} {c.ext:<4} {c.md5 or c.url}")
+        if not cands:
+            out(f"    (no candidates: {notes})")
+        return ""
+    if not cands:
+        outcome = "error" if notes.had_error else "no_candidate"
+        ledger.record(ref.key, outcome=outcome, ids=ref.ids, note=str(notes))
+        out(f"{ref.key}  {outcome}  {ref.title[:60]}")
+        return outcome
+
+    budget.quota_hit = False
+    delivered: tuple[str, Path, str] | None = None
+    for c in cands:
+        dest = _scratch_path(scratch_dir, ref, c.md5 or c.url or "", c.ext)
+        scratch.append(dest)
+        source = fetch(
+            c,
+            dest,
+            direct=clients.direct,
+            proxied=clients.proxied,
+            budget=budget,
+            pacer=lg_pacer,
+            sleep=sleep,
+            notes=notes,
+        )
+        if source:
+            delivered = (source, dest, c.md5 or "")
+            break
+    if delivered is None:
+        if budget.quota_hit:
+            outcome = "quota"
+        elif notes.had_error:
+            outcome = "error"
+        else:
+            outcome = "all_failed"
+        ledger.record(ref.key, outcome=outcome, ids=ref.ids, note=str(notes))
+        out(f"{ref.key}  {outcome}  {ref.title[:60]}")
+        return outcome
+
+    source, path, md5 = delivered
+    final = path.with_name(f"{safe_name(ref.title, ref.key)}{path.suffix}")
+    path.replace(final)
+    scratch.append(final)
+    attach_file(
+        db,
+        ref.item_id,
+        final,
+        source=source,
+        md5=md5,
+        storage_dir=storage_dir,
+        copier=copier,
+    )
+    db.commit()
+    _cleanup(scratch)
+    ledger.record(ref.key, outcome="attached", source=source, md5=md5, ids=ref.ids)
+    out(f"{ref.key}  attached  {source}  {final.name}")
+    return "attached"
+
+
+def run(
+    db: Db,
+    refs: list[ItemRef],
+    *,
+    ledger: Ledger,
+    budget: Budget,
+    clients: Clients,
+    email: str,
+    storage_dir: Path,
+    scratch_dir: Path,
+    limit: int = 0,
+    dry_run: bool = False,
+    retry: bool = False,
+    copier: Copier = sudo_copier,
+    sleep: Callable[[float], None] = time.sleep,
+    out: Callable[..., None] = print,
+) -> dict[str, int]:
+    stats = dict.fromkeys(STATS, 0)
+    up_pacer = Pacer(1.0, sleep=sleep)
+    lg_pacer = Pacer(3.0, sleep=sleep)
+    scratch_dir.mkdir(parents=True, exist_ok=True)
+
+    for ref in refs:
+        if limit and stats["attempted"] >= limit:
+            break
+        stats["seen"] += 1
+        if not ledger.should_try(ref.key, retry=retry):
+            stats["skipped"] += 1
+            continue
+        stats["attempted"] += 1
+        notes = Notes()
+        scratch: list[Path] = []
+        try:
+            outcome = _one(
+                db,
+                ref,
+                notes=notes,
+                scratch=scratch,
+                budget=budget,
+                clients=clients,
+                email=email,
+                storage_dir=storage_dir,
+                scratch_dir=scratch_dir,
+                dry_run=dry_run,
+                copier=copier,
+                sleep=sleep,
+                out=out,
+                up_pacer=up_pacer,
+                lg_pacer=lg_pacer,
+                ledger=ledger,
+            )
+        except AnnasAuthError:
+            db.con.rollback()
+            _cleanup(scratch)
+            raise
+        except Exception as exc:  # per-item failures never abort the run
+            db.con.rollback()
+            _cleanup(scratch)
+            note = f"{type(exc).__name__}: {exc}"
+            ledger.record(ref.key, outcome="error", ids=ref.ids, note=note)
+            stats["error"] += 1
+            out(f"{ref.key}  error  {note[:120]}")
+            continue
+        if outcome:
+            stats[outcome] += 1
+    return stats

src/zot/fetch/select.py

@@ -0,0 +1,151 @@
+"""Which Zotero items a fetch run looks at, and their identifiers."""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+
+from zot.db import TYPE_MAP, Db
+from zot.fetch.attach import has_full_text
+
+BOOK_TYPES = {"book", "bookSection"}
+NEVER = {
+    TYPE_MAP["attachment"],
+    TYPE_MAP["note"],
+    TYPE_MAP["annotation"],
+    TYPE_MAP["document"],
+}
+
+_DOI_PREFIX = re.compile(r"^(?:https?://(?:dx\.)?doi\.org/|doi:)\s*", re.I)
+_ISBN_TOKEN = re.compile(r"(?<![0-9X])(?:[0-9]{13}|[0-9]{9}[0-9X])(?![0-9X])")
+_YEAR = re.compile(r"\b(19|20)\d\d\b")
+
+
+@dataclass(frozen=True)
+class ItemRef:
+    item_id: int
+    key: str
+    item_type: str
+    title: str
+    doi: str
+    isbn: str
+    year: str
+
+    @property
+    def kind(self) -> str:
+        return "book" if self.item_type in BOOK_TYPES else "paper"
+
+    @property
+    def ids(self) -> dict[str, str]:
+        return {k: v for k, v in (("doi", self.doi), ("isbn", self.isbn)) if v}
+
+
+def normalise_doi(raw: str) -> str:
+    return _DOI_PREFIX.sub("", (raw or "").strip()).strip()
+
+
+def first_isbn(raw: str) -> str:
+    """First ISBN in *raw*, digits only; a 13-digit one wins over a 10."""
+    flat = (raw or "").replace("-", "").upper()
+    tokens = _ISBN_TOKEN.findall(flat)
+    for t in tokens:
+        if len(t) == 13:
+            return t
+    return tokens[0] if tokens else ""
+
+
+def year_of(raw: str) -> str:
+    m = _YEAR.search(raw or "")
+    return m.group(0) if m else ""
+
+
+def item_ref(db: Db, item_id: int) -> ItemRef:
+    item = db.get_item(item_id)
+    f = item["fields"]
+    return ItemRef(
+        item_id=item_id,
+        key=item["key"],
+        item_type=item["itemType"],
+        title=f.get("title", "") or "",
+        doi=normalise_doi(f.get("DOI", "")),
+        isbn=first_isbn(f.get("ISBN", "")),
+        year=year_of(f.get("date", "")),
+    )
+
+
+def _collection_keys(db: Db, name: str) -> list[str]:
+    rows = db.con.execute(
+        "SELECT key FROM collections WHERE collectionName = ? ORDER BY collectionID",
+        (name,),
+    ).fetchall()
+    return [r[0] for r in rows]
+
+
+def _all_ids(db: Db) -> list[int]:
+    marks = ",".join("?" * len(NEVER))
+    rows = db.con.execute(
+        f"SELECT itemID FROM items WHERE itemTypeID NOT IN ({marks}) "  # noqa: S608
+        "ORDER BY itemID",
+        tuple(NEVER),
+    ).fetchall()
+    return [r[0] for r in rows]
+
+
+def _deleted(db: Db) -> set[int]:
+    rows = db.con.execute("SELECT itemID FROM deletedItems").fetchall()
+    return {r[0] for r in rows}
+
+
+def select_items(
+    db: Db,
+    *,
+    collection: str = "",
+    tag: str = "",
+    type_: str = "",
+    item: str = "",
+    all_: bool = False,
+) -> list[ItemRef]:
+    """Items to try, in itemID order.
+
+    Exactly one of *collection* (by name), *tag*, *item* (key) or *all_*
+    picks the population and *type_* narrows it; *type_* alone is a
+    population of its own (the spec's ``--type journalArticle``). Items with a PDF/EPUB
+    child, deleted items, and ``document`` items are always excluded.
+    """
+    if type_ and type_ not in TYPE_MAP:
+        raise ValueError(f"unknown item type {type_!r}")
+    primaries = sum([bool(collection), bool(tag), bool(item), all_])
+    if primaries > 1 or (primaries == 0 and not type_):
+        raise ValueError(
+            "exactly one of --collection/--tag/--item/--all (or --type alone) "
+            "is required"
+        )
+
+    if collection:
+        keys = _collection_keys(db, collection)
+        if not keys:
+            raise ValueError(f"no collection named {collection!r}")
+        ids: list[int] = sorted({i for k in keys for i in db.search_by_collection(k)})
+    elif tag:
+        ids = db.search_by_tag(tag)
+    elif item:
+        found = db.find_item_by_key(item)
+        if found is None:
+            raise ValueError(f"no item with key {item!r}")
+        ids = [found]
+    elif all_:
+        ids = _all_ids(db)
+    else:
+        ids = db.search_by_type(type_)
+
+    if type_ and primaries:
+        wanted = TYPE_MAP[type_]
+        ids = [i for i in ids if db.get_item_type(i) == wanted]
+
+    deleted = _deleted(db)
+    out: list[ItemRef] = []
+    for i in ids:
+        if i in deleted or db.get_item_type(i) in NEVER or has_full_text(db, i):
+            continue
+        out.append(item_ref(db, i))
+    return out

tests/cli/conftest.py

@@ -0,0 +1,3 @@
+"""Shared fixtures for CLI tests."""
+
+from tests.zot.fetch.conftest import zdb  # noqa: F401 — scratch Zotero DB

tests/cli/test_zot_fetch.py

@@ -0,0 +1,97 @@
+"""stack zot fetch — CLI wiring (no network, no docker)."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+from typer.testing import CliRunner
+
+from cli.zot import app
+from tests.zot.fetch.conftest import mk_item
+from zot.db import Db
+from zot.fetch.deliver import AnnasAuthError
+from zot.fetch.run import STATS
+
+runner = CliRunner()
+
+
+@pytest.fixture()
+def env(monkeypatch, tmp_path: Path):
+    monkeypatch.setenv("VPN_HTTP_PROXY", "http://127.0.0.1:19035")
+    monkeypatch.setenv("ANNAS_ARCHIVE_KEY", "k")
+    monkeypatch.setenv("ANNAS_DAILY_RESERVE", "7")
+    monkeypatch.setenv("UNPAYWALL_EMAIL", "dev@fhirworx.io")
+    monkeypatch.setattr("cli.zot._hold_zotero", lambda fn, hold: fn())
+    return tmp_path
+
+
+def _db_with_item(zdb: Db) -> str:
+    mk_item(zdb, title="T", doi="10.1/x")
+    zdb.commit()
+    return zdb.path
+
+
+def test_fetch_requires_proxy(monkeypatch, zdb: Db):
+    monkeypatch.delenv("VPN_HTTP_PROXY", raising=False)
+    monkeypatch.setattr("cli.zot._hold_zotero", lambda fn, hold: fn())
+    r = runner.invoke(app, ["fetch", "--all", "--db", _db_with_item(zdb), "--no-hold"])
+    assert r.exit_code != 0
+    assert "VPN_HTTP_PROXY" in r.output and "gluetun" in r.output
+
+
+def test_fetch_dry_run_calls_run_with_selection(env, zdb: Db, monkeypatch):
+    captured: dict = {}
+
+    def fake_run(db, refs, **kw):
+        captured["refs"] = refs
+        captured["kw"] = kw
+        return dict.fromkeys(STATS, 0) | {"seen": 1, "attempted": 1}
+
+    monkeypatch.setattr("cli.zot.run", fake_run)
+    r = runner.invoke(
+        app,
+        [
+            "fetch",
+            "--all",
+            "--dry-run",
+            "--limit",
+            "3",
+            "--db",
+            _db_with_item(zdb),
+            "--ledger",
+            str(env / "ledger.jsonl"),
+            "--no-hold",
+        ],
+    )
+    assert r.exit_code == 0, r.output
+    assert [x.doi for x in captured["refs"]] == ["10.1/x"]
+    kw = captured["kw"]
+    assert kw["dry_run"] is True and kw["limit"] == 3 and kw["retry"] is False
+    assert kw["budget"].annas_key == "k" and kw["budget"].reserve == 7
+    assert kw["email"] == "dev@fhirworx.io"
+    assert kw["ledger"].path == env / "ledger.jsonl"
+    assert kw["storage_dir"].name == "storage"
+    assert "1 eligible" in r.output
+    assert "seen=1" in r.output and "attempted=1" in r.output
+
+
+def test_fetch_selection_error_is_a_usage_error(env, zdb: Db):
+    r = runner.invoke(app, ["fetch", "--db", _db_with_item(zdb), "--no-hold"])
+    assert r.exit_code != 0 and "exactly one" in r.output
+
+
+def test_fetch_annas_auth_error_exits_nonzero_with_message(env, zdb: Db, monkeypatch):
+    def fake_run(db, refs, **kw):
+        raise AnnasAuthError("Anna's Archive refused the key (403)")
+
+    monkeypatch.setattr("cli.zot.run", fake_run)
+    r = runner.invoke(app, ["fetch", "--all", "--db", _db_with_item(zdb), "--no-hold"])
+    assert r.exit_code != 0 and "refused the key" in r.output
+
+
+def test_fetch_without_annas_key_prints_note(env, zdb: Db, monkeypatch):
+    monkeypatch.delenv("ANNAS_ARCHIVE_KEY")
+    monkeypatch.setattr("cli.zot.run", lambda db, refs, **kw: dict.fromkeys(STATS, 0))
+    r = runner.invoke(app, ["fetch", "--all", "--db", _db_with_item(zdb), "--no-hold"])
+    assert r.exit_code == 0 and "ANNAS_ARCHIVE_KEY not set" in r.output

tests/zot/fetch/conftest.py

@@ -0,0 +1,50 @@
+"""Scratch Zotero DB for the fetch tests (no live DB, no network)."""
+
+from __future__ import annotations
+
+import sqlite3
+from pathlib import Path
+
+import pytest
+
+from tests.zot.test_db import ZOTERO_SCHEMA, _seed_schema_maps
+from zot.db import TYPE_MAP, Db
+
+EXTRA_SCHEMA = """
+CREATE TABLE IF NOT EXISTS deletedItems (
+    itemID INTEGER PRIMARY KEY,
+    dateDeleted TEXT DEFAULT CURRENT_TIMESTAMP NOT NULL
+);
+"""
+
+
+@pytest.fixture()
+def zdb(tmp_path: Path) -> Db:
+    path = str(tmp_path / "zotero.sqlite")
+    con = sqlite3.connect(path)
+    con.executescript(ZOTERO_SCHEMA)
+    con.executescript(EXTRA_SCHEMA)
+    _seed_schema_maps(con)
+    con.commit()
+    con.close()
+    d = Db(path)
+    yield d
+    d.close()
+
+
+def mk_item(
+    db: Db,
+    *,
+    type_name: str = "journalArticle",
+    title: str = "",
+    doi: str = "",
+    isbn: str = "",
+    date: str = "",
+    extra: str = "",
+) -> int:
+    item_id = db.create_item(TYPE_MAP[type_name])
+    db.set_fields(
+        item_id,
+        {"title": title, "DOI": doi, "ISBN": isbn, "date": date, "extra": extra},
+    )
+    return item_id

tests/zot/fetch/fixtures/annas_fast_download.json

@@ -0,0 +1 @@
+{"download_url": "https://fast.example.org/dl/a78add1bd62047ead5945134488f470f.pdf", "account_fast_download_info": {"downloads_left": 42, "downloads_per_day": 50, "recently_downloaded_md5s": []}}

tests/zot/fetch/fixtures/libgen_ads.html

@@ -0,0 +1 @@
+<html><body><table><tr><td><a href="get.php?md5=a78add1bd62047ead5945134488f470f&key=AJPW7IO3P1LGRGWA">GET</a></td></tr></table></body></html>

tests/zot/fetch/fixtures/libgen_index.html

@@ -0,0 +1,4 @@
+<html><body><table id="tablelibgen"><tbody>
+<tr><td><a href="edition.php?id=146962484">HCPCS Level II Professional 2023</a></td><td>2023</td></tr>
+<tr><td><a href="edition.php?id=146962485">HCPCS Level II Professional 2019</a></td><td>2019</td></tr>
+</tbody></table></body></html>

tests/zot/fetch/fixtures/libgen_json_doi.json

@@ -0,0 +1 @@
+{"27703795": {"title": "Early Palliative Care for Patients with Metastatic Non–Small-Cell Lung Cancer", "libgen_topic": "a", "type": "a", "author": "Temel, Jennifer S.; Greer, Joseph A.", "year": "2010", "doi": "10.1056/NEJMoa1000678", "files": {"26575841": {"f_id": "27858481", "md5": "a78add1bd62047ead5945134488f470f", "time_added": "2021-07-17 13:59:28"}, "26575842": {"f_id": "27858482", "md5": "b78add1bd62047ead5945134488f470f", "time_added": "2021-07-17 13:59:28"}}, "add": {"68149362": {"key": "904", "value": "20818875", "name_en": "PMID"}}}}

tests/zot/fetch/fixtures/libgen_json_file.json

@@ -0,0 +1 @@
+{"27858481": {"md5": "a78add1bd62047ead5945134488f470f", "filesize": "401662", "extension": "pdf"}, "104009164": {"md5": "120ddd118103ea2ea927aaa489310dc8", "filesize": "9000000", "extension": "pdf"}, "104009165": {"md5": "220ddd118103ea2ea927aaa489310dc8", "filesize": "4000000", "extension": "epub"}}

tests/zot/fetch/fixtures/libgen_json_isbn.json

@@ -0,0 +1 @@
+{"146962484": {"title": "HCPCS Level II Professional 2023", "libgen_topic": "l", "type": "b", "year": "2023", "edition": "1", "files": {"103379671": {"f_id": "104009164", "md5": "120ddd118103ea2ea927aaa489310dc8"}, "103379672": {"f_id": "104009165", "md5": "220ddd118103ea2ea927aaa489310dc8"}}}}

tests/zot/fetch/fixtures/unpaywall_doi.json

@@ -0,0 +1 @@
+{"doi": "10.1056/nejmoa1000678", "is_oa": true, "best_oa_location": {"url_for_pdf": "https://example.org/best.pdf", "url": "https://example.org/best"}, "oa_locations": [{"url_for_pdf": "https://example.org/best.pdf", "url": "https://example.org/best"}, {"url_for_pdf": null, "url": "https://example.org/landing"}]}

tests/zot/fetch/test_attach.py

@@ -0,0 +1,152 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from tests.zot.fetch.conftest import mk_item
+from zot.db import Db
+from zot.fetch.attach import (
+    CONTENT_TYPES,
+    append_extra,
+    attach_file,
+    has_full_text,
+    sudo_copier,
+)
+
+
+def _py_copier(src: Path, dest_dir: Path) -> None:
+    dest_dir.mkdir(parents=True, exist_ok=True)
+    (dest_dir / src.name).write_bytes(src.read_bytes())
+
+
+def _att(db: Db, parent: int, ctype: str) -> None:
+    db.add_attachment(parent, content_type=ctype, path="storage:x")
+
+
+@pytest.mark.parametrize(
+    "ctypes, expected",
+    [
+        ([], False),
+        (["text/html"], False),
+        (["application/pdf"], True),
+        (["application/epub+zip"], True),
+        (["text/html", "application/pdf"], True),
+    ],
+)
+def test_has_full_text_truth_table(zdb: Db, ctypes, expected):
+    parent = mk_item(zdb, title="P")
+    for c in ctypes:
+        _att(zdb, parent, c)
+    assert has_full_text(zdb, parent) is expected
+
+
+def test_has_full_text_ignores_other_parents(zdb: Db):
+    a = mk_item(zdb, title="A")
+    b = mk_item(zdb, title="B")
+    _att(zdb, a, "application/pdf")
+    assert has_full_text(zdb, b) is False
+
+
+def test_attach_file_writes_row_tags_and_extra(zdb: Db, tmp_path: Path):
+    parent = mk_item(zdb, title="Paper", extra="PMID: 1")
+    src = tmp_path / "Paper.pdf"
+    src.write_bytes(b"%PDF-1.4 x")
+    storage = tmp_path / "storage"
+    att_id = attach_file(
+        zdb,
+        parent,
+        src,
+        source="libgen",
+        md5="a" * 32,
+        storage_dir=storage,
+        copier=_py_copier,
+    )
+    row = zdb.con.execute(
+        "SELECT parentItemID, linkMode, contentType, path FROM itemAttachments "
+        "WHERE itemID = ?",
+        (att_id,),
+    ).fetchone()
+    assert tuple(row) == (parent, 0, "application/pdf", "storage:Paper.pdf")
+    assert zdb.get_field(att_id, "title") == "Paper.pdf"
+    att_key = zdb.get_item(att_id)["key"]
+    assert (storage / att_key / "Paper.pdf").read_bytes() == b"%PDF-1.4 x"
+    assert "source:libgen" in zdb.get_tags(parent)
+    assert zdb.get_field(parent, "extra") == "PMID: 1\nmd5: " + "a" * 32
+    assert has_full_text(zdb, parent) is True
+
+
+def test_attach_file_epub_content_type(zdb: Db, tmp_path: Path):
+    parent = mk_item(zdb, type_name="book", title="Book")
+    src = tmp_path / "Book.epub"
+    src.write_bytes(b"PK\x03\x04")
+    att_id = attach_file(
+        zdb, parent, src, source="annas", storage_dir=tmp_path / "s", copier=_py_copier
+    )
+    ctype = zdb.con.execute(
+        "SELECT contentType FROM itemAttachments WHERE itemID = ?", (att_id,)
+    ).fetchone()[0]
+    assert ctype == CONTENT_TYPES[".epub"]
+    assert zdb.get_field(parent, "extra") is None  # no md5 → extra untouched
+
+
+def test_attach_file_rejects_unknown_extension(zdb: Db, tmp_path: Path):
+    parent = mk_item(zdb, title="P")
+    src = tmp_path / "x.mobi"
+    src.write_bytes(b"x")
+    with pytest.raises(ValueError, match="mobi"):
+        attach_file(
+            zdb, parent, src, source="libgen", storage_dir=tmp_path, copier=_py_copier
+        )
+
+
+def test_append_extra_is_idempotent_and_keeps_lines(zdb: Db):
+    item = mk_item(zdb, title="P", extra="PMID: 1")
+    append_extra(zdb, item, "md5: abc")
+    append_extra(zdb, item, "md5: abc")
+    assert zdb.get_field(item, "extra") == "PMID: 1\nmd5: abc"
+
+
+def test_append_extra_on_empty_extra(zdb: Db):
+    item = mk_item(zdb, title="P")
+    append_extra(zdb, item, "md5: abc")
+    assert zdb.get_field(item, "extra") == "md5: abc"
+
+
+def test_sudo_copier_runs_mkdir_cp_chown(monkeypatch, tmp_path: Path):
+    calls: list[list[str]] = []
+
+    def fake_run(cmd, check):
+        calls.append(cmd)
+
+    monkeypatch.setattr("zot.fetch.attach.subprocess.run", fake_run)
+    sudo_copier(tmp_path / "a.pdf", tmp_path / "storage" / "KEY")
+    dest = tmp_path / "storage" / "KEY"
+    assert calls == [
+        ["sudo", "mkdir", "-p", str(dest)],
+        ["sudo", "cp", str(tmp_path / "a.pdf"), str(dest / "a.pdf")],
+        ["sudo", "chown", "-R", "100999:100999", str(dest)],
+    ]
+
+
+def test_attach_file_writes_rows_before_copying(zdb: Db, tmp_path: Path):
+    """A failed copy must leave rows the caller can roll back, never an
+    orphan storage dir with no row."""
+    parent = mk_item(zdb, title="P")
+    src = tmp_path / "P.pdf"
+    src.write_bytes(b"%PDF-1.4 x")
+    seen = {}
+
+    def spy_copier(s: Path, d: Path) -> None:
+        seen["rows"] = zdb.con.execute(
+            "SELECT COUNT(*) FROM itemAttachments WHERE parentItemID = ?", (parent,)
+        ).fetchone()[0]
+        raise OSError("disk full")
+
+    with pytest.raises(OSError):
+        attach_file(
+            zdb, parent, src, source="libgen", storage_dir=tmp_path, copier=spy_copier
+        )
+    assert seen["rows"] == 1
+    zdb.con.rollback()
+    assert has_full_text(zdb, parent) is False

tests/zot/fetch/test_deliver.py

@@ -0,0 +1,429 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import httpx
+import pytest
+
+from zot.fetch.deliver import (
+    MIN_BYTES,
+    AnnasAuthError,
+    Budget,
+    download,
+    fetch,
+    fetch_annas,
+    fetch_libgen,
+)
+from zot.fetch.resolve import LIBGEN, Candidate, Notes, Pacer
+
+FX = Path(__file__).parent / "fixtures"
+PDF = b"%PDF-1.4\n" + b"0" * MIN_BYTES
+EPUB = b"PK\x03\x04" + b"0" * MIN_BYTES
+MD5 = "a78add1bd62047ead5945134488f470f"
+ADS = (FX / "libgen_ads.html").read_text()
+ANNAS_OK = (FX / "annas_fast_download.json").read_text()
+PDF_HDR = {"content-type": "application/pdf"}
+
+
+def _pacer() -> Pacer:
+    return Pacer(3.0, clock=lambda: 0.0, sleep=lambda s: None)
+
+
+def _client(handler) -> httpx.Client:
+    return httpx.Client(transport=httpx.MockTransport(handler))
+
+
+def _lg(md5=MD5) -> Candidate:
+    return Candidate("libgen", md5, None, "pdf", None, "T")
+
+
+def _up() -> Candidate:
+    return Candidate("unpaywall", None, "https://example.org/x.pdf", "pdf", None, "T")
+
+
+def _nosleep(s: float) -> None:
+    return None
+
+
+# ── download ──────────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize(
+    "status, ctype, body, ext, ok",
+    [
+        (200, "application/pdf", PDF, "pdf", True),
+        (200, "application/octet-stream", PDF, "pdf", True),
+        (200, "application/epub+zip", EPUB, "epub", True),
+        (200, "text/html; charset=utf-8", PDF, "pdf", False),
+        (200, "application/pdf", b"%PDF tiny", "pdf", False),
+        (200, "application/pdf", b"<html>" + b"0" * MIN_BYTES, "pdf", False),
+        (503, "application/pdf", PDF, "pdf", False),
+    ],
+)
+def test_download_validates_status_type_size_and_magic(
+    tmp_path, status, ctype, body, ext, ok
+):
+    c = _client(
+        lambda r: httpx.Response(status, content=body, headers={"content-type": ctype})
+    )
+    dest = tmp_path / f"f.{ext}"
+    got = download(c, "https://x/y", dest, ext=ext)
+    assert (got == dest) is ok
+    assert dest.exists() is ok
+
+
+def test_download_transport_error_is_none(tmp_path):
+    def boom(r):
+        raise httpx.ReadTimeout("t")
+
+    assert download(_client(boom), "https://x", tmp_path / "f.pdf", ext="pdf") is None
+
+
+def test_download_passes_headers(tmp_path):
+    seen = {}
+
+    def h(r):
+        seen.update(r.headers)
+        return httpx.Response(200, content=PDF, headers=PDF_HDR)
+
+    download(
+        _client(h),
+        "https://x",
+        tmp_path / "f.pdf",
+        ext="pdf",
+        headers={"Referer": "https://r/"},
+    )
+    assert seen["referer"] == "https://r/"
+
+
+# ── libgen ────────────────────────────────────────────────────────
+
+
+def _libgen_handler(ads_pages: list[httpx.Response], get_status=200):
+    calls: list[str] = []
+
+    def h(r: httpx.Request) -> httpx.Response:
+        calls.append(r.url.path)
+        if r.url.path == "/ads.php":
+            assert r.url.params["md5"] == MD5
+            assert r.headers["referer"] == f"{LIBGEN}/"
+            return ads_pages.pop(0)
+        if r.url.path == "/get.php":
+            assert r.url.params["key"] == "AJPW7IO3P1LGRGWA"
+            return httpx.Response(get_status, content=PDF, headers=PDF_HDR)
+        raise AssertionError(r.url)
+
+    return h, calls
+
+
+def test_fetch_libgen_ads_then_get(tmp_path):
+    h, calls = _libgen_handler([httpx.Response(200, text=ADS)])
+    got = fetch_libgen(
+        _client(h), _lg(), tmp_path / "f.pdf", pacer=_pacer(), sleep=_nosleep
+    )
+    assert got == tmp_path / "f.pdf" and calls == ["/ads.php", "/get.php"]
+
+
+def test_libgen_ads_error_div_retries_once_then_gives_up(tmp_path):
+    err = httpx.Response(
+        200,
+        text='<div class="alert alert-danger">Could not connect to the database 3306</div>',
+    )
+    h, calls = _libgen_handler([err, err])
+    slept: list[float] = []
+    got = fetch_libgen(
+        _client(h), _lg(), tmp_path / "f.pdf", pacer=_pacer(), sleep=slept.append
+    )
+    assert got is None and calls == ["/ads.php", "/ads.php"] and slept == [3.0]
+
+
+def test_libgen_ads_error_then_success_on_retry(tmp_path):
+    h, calls = _libgen_handler([httpx.Response(500), httpx.Response(200, text=ADS)])
+    got = fetch_libgen(
+        _client(h), _lg(), tmp_path / "f.pdf", pacer=_pacer(), sleep=_nosleep
+    )
+    assert got is not None and calls == ["/ads.php", "/ads.php", "/get.php"]
+
+
+def test_libgen_ads_link_for_other_md5_is_ignored(tmp_path):
+    other = httpx.Response(200, text=ADS.replace(MD5, "f" * 32))
+    h, calls = _libgen_handler([other, other])
+    got = fetch_libgen(
+        _client(h), _lg(), tmp_path / "f.pdf", pacer=_pacer(), sleep=_nosleep
+    )
+    assert got is None and calls == ["/ads.php", "/ads.php"]
+
+
+def test_libgen_get_503_is_none(tmp_path):
+    h, _ = _libgen_handler([httpx.Response(200, text=ADS)], get_status=503)
+    got = fetch_libgen(
+        _client(h), _lg(), tmp_path / "f.pdf", pacer=_pacer(), sleep=_nosleep
+    )
+    assert got is None
+
+
+def test_libgen_transport_error_is_none(tmp_path):
+    def boom(r):
+        raise httpx.ConnectError("x")
+
+    got = fetch_libgen(
+        _client(boom), _lg(), tmp_path / "f.pdf", pacer=_pacer(), sleep=_nosleep
+    )
+    assert got is None
+
+
+# ── Anna's ────────────────────────────────────────────────────────
+
+
+def _annas_handler(api: httpx.Response, file_status=200):
+    def h(r: httpx.Request) -> httpx.Response:
+        if r.url.host == "annas-archive.gl":
+            assert r.url.path == "/dyn/api/fast_download.json"
+            assert r.url.params["md5"] == MD5 and r.url.params["key"] == "SECRET"
+            return api
+        return httpx.Response(file_status, content=PDF, headers=PDF_HDR)
+
+    return h
+
+
+def test_fetch_annas_downloads_and_updates_budget(tmp_path):
+    b = Budget(annas_key="SECRET")
+    c = _client(_annas_handler(httpx.Response(200, text=ANNAS_OK)))
+    got = fetch_annas(c, _lg(), tmp_path / "f.pdf", budget=b)
+    assert got == tmp_path / "f.pdf" and b.annas_left == 42
+
+
+@pytest.mark.parametrize("status", [401, 403])
+def test_fetch_annas_auth_errors_stop_the_run(tmp_path, status):
+    b = Budget(annas_key="SECRET")
+    c = _client(_annas_handler(httpx.Response(status, json={"error": "Not a member"})))
+    with pytest.raises(AnnasAuthError):
+        fetch_annas(c, _lg(), tmp_path / "f.pdf", budget=b)
+
+
+def test_fetch_annas_other_error_disables_annas(tmp_path):
+    b = Budget(annas_key="SECRET")
+    c = _client(_annas_handler(httpx.Response(500)))
+    assert fetch_annas(c, _lg(), tmp_path / "f.pdf", budget=b) is None
+    assert b.annas_disabled is True and b.can_use_annas() is False
+
+
+def test_fetch_annas_null_download_url_is_none(tmp_path):
+    b = Budget(annas_key="SECRET")
+    api = httpx.Response(200, json={"download_url": None, "error": "Invalid md5"})
+    assert (
+        fetch_annas(_client(_annas_handler(api)), _lg(), tmp_path / "f.pdf", budget=b)
+        is None
+    )
+
+
+def test_fetch_annas_bad_json_or_transport_is_none(tmp_path):
+    b = Budget(annas_key="SECRET")
+    c = _client(_annas_handler(httpx.Response(200, text="nope")))
+    assert fetch_annas(c, _lg(), tmp_path / "f.pdf", budget=b) is None
+
+    def boom(r):
+        raise httpx.ConnectError("x")
+
+    assert fetch_annas(_client(boom), _lg(), tmp_path / "f.pdf", budget=b) is None
+
+
+def test_fetch_annas_file_failure_still_reports_budget(tmp_path):
+    b = Budget(annas_key="SECRET")
+    c = _client(_annas_handler(httpx.Response(200, text=ANNAS_OK), file_status=503))
+    got = fetch_annas(c, _lg(), tmp_path / "f.pdf", budget=b)
+    assert got is None and b.annas_left == 42  # the API call itself consumed one
+
+
+def test_budget_reserve_blocks_annas():
+    assert Budget(annas_key="").can_use_annas() is False
+    assert Budget(annas_key="k").can_use_annas() is True  # unknown budget → allowed
+    assert Budget(annas_key="k", annas_left=6, reserve=5).can_use_annas() is True
+    assert Budget(annas_key="k", annas_left=5, reserve=5).can_use_annas() is False
+
+
+# ── fetch (the cascade) ───────────────────────────────────────────
+
+
+def test_fetch_unpaywall_uses_direct_client(tmp_path):
+    direct = _client(lambda r: httpx.Response(200, content=PDF, headers=PDF_HDR))
+
+    def never(r):
+        raise AssertionError("proxied client used for Unpaywall")
+
+    src = fetch(
+        _up(),
+        tmp_path / "f.pdf",
+        direct=direct,
+        proxied=_client(never),
+        budget=Budget(),
+        pacer=_pacer(),
+    )
+    assert src == "unpaywall"
+
+
+def test_fetch_unpaywall_failure_is_none(tmp_path):
+    direct = _client(lambda r: httpx.Response(404))
+    src = fetch(
+        _up(),
+        tmp_path / "f.pdf",
+        direct=direct,
+        proxied=direct,
+        budget=Budget(),
+        pacer=_pacer(),
+    )
+    assert src is None
+
+
+def test_fetch_libgen_then_annas_on_failure(tmp_path):
+    def h(r: httpx.Request) -> httpx.Response:
+        if r.url.path == "/ads.php":
+            return httpx.Response(500)
+        if r.url.host == "annas-archive.gl":
+            return httpx.Response(200, text=ANNAS_OK)
+        return httpx.Response(200, content=PDF, headers=PDF_HDR)
+
+    b = Budget(annas_key="SECRET")
+    src = fetch(
+        _lg(),
+        tmp_path / "f.pdf",
+        direct=_client(h),
+        proxied=_client(h),
+        budget=b,
+        pacer=_pacer(),
+        sleep=_nosleep,
+    )
+    assert src == "annas" and b.annas_left == 42
+
+
+def test_fetch_libgen_success_skips_annas(tmp_path):
+    h, calls = _libgen_handler([httpx.Response(200, text=ADS)])
+    b = Budget(annas_key="SECRET")
+    src = fetch(
+        _lg(),
+        tmp_path / "f.pdf",
+        direct=_client(h),
+        proxied=_client(h),
+        budget=b,
+        pacer=_pacer(),
+        sleep=_nosleep,
+    )
+    assert src == "libgen" and calls == ["/ads.php", "/get.php"]
+
+
+def test_fetch_marks_quota_when_annas_blocked_by_reserve(tmp_path):
+    proxied = _client(lambda r: httpx.Response(500))
+    b = Budget(annas_key="SECRET", annas_left=2, reserve=5)
+    src = fetch(
+        _lg(),
+        tmp_path / "f.pdf",
+        direct=proxied,
+        proxied=proxied,
+        budget=b,
+        pacer=_pacer(),
+        sleep=_nosleep,
+    )
+    assert src is None and b.quota_hit is True
+
+
+def test_fetch_all_fail_without_annas_is_not_quota(tmp_path):
+    proxied = _client(lambda r: httpx.Response(500))
+    b = Budget()
+    src = fetch(
+        _lg(),
+        tmp_path / "f.pdf",
+        direct=proxied,
+        proxied=proxied,
+        budget=b,
+        pacer=_pacer(),
+        sleep=_nosleep,
+    )
+    assert src is None and b.quota_hit is False
+
+
+# ── fix pass (review) ─────────────────────────────────────────────
+
+
+def test_fetch_annas_quota_403_disables_instead_of_raising(tmp_path):
+    b = Budget(annas_key="SECRET")
+    api = httpx.Response(403, json={"download_url": None, "error": "No downloads left"})
+    notes = Notes()
+    got = fetch_annas(
+        _client(_annas_handler(api)), _lg(), tmp_path / "f.pdf", budget=b, notes=notes
+    )
+    assert got is None
+    assert b.annas_disabled is True and b.annas_left == 0
+    assert notes.had_error and "No downloads left" in str(notes)
+
+
+@pytest.mark.parametrize("err", ["Invalid secret key", "Not a member"])
+def test_fetch_annas_key_errors_still_raise(tmp_path, err):
+    b = Budget(annas_key="SECRET")
+    api = httpx.Response(403, json={"download_url": None, "error": err})
+    with pytest.raises(AnnasAuthError, match=err):
+        fetch_annas(_client(_annas_handler(api)), _lg(), tmp_path / "f.pdf", budget=b)
+
+
+def test_notes_distinguish_errors_from_misses(tmp_path):
+    # transport error → error note; definitive 404 → miss note
+    def boom(r):
+        raise httpx.ConnectError("x")
+
+    n = Notes()
+    fetch(
+        _lg(),
+        tmp_path / "f.pdf",
+        direct=_client(boom),
+        proxied=_client(boom),
+        budget=Budget(),
+        pacer=_pacer(),
+        sleep=_nosleep,
+        notes=n,
+    )
+    assert n.had_error and "ConnectError" in str(n)
+
+    n2 = Notes()
+    fetch(
+        _up(),
+        tmp_path / "f.pdf",
+        direct=_client(lambda r: httpx.Response(404)),
+        proxied=_client(boom),
+        budget=Budget(),
+        pacer=_pacer(),
+        notes=n2,
+    )
+    assert not n2.had_error and "404" in str(n2)
+
+
+def test_libgen_throttle_page_is_an_error_note(tmp_path):
+    err = httpx.Response(200, text='<div class="alert alert-danger">db 3306</div>')
+    h, _ = _libgen_handler([err, err])
+    n = Notes()
+    fetch_libgen(
+        _client(h), _lg(), tmp_path / "f.pdf", pacer=_pacer(), sleep=_nosleep, notes=n
+    )
+    assert n.had_error and "ads" in str(n)
+
+
+def test_libgen_and_annas_never_use_the_direct_client(tmp_path):
+    def never(r):
+        raise AssertionError("direct client used for libgen/Anna's")
+
+    def h(r: httpx.Request) -> httpx.Response:
+        if r.url.path == "/ads.php":
+            return httpx.Response(500)
+        if r.url.host == "annas-archive.gl":
+            return httpx.Response(200, text=ANNAS_OK)
+        return httpx.Response(200, content=PDF, headers=PDF_HDR)
+
+    b = Budget(annas_key="SECRET")
+    src = fetch(
+        _lg(),
+        tmp_path / "f.pdf",
+        direct=_client(never),
+        proxied=_client(h),
+        budget=b,
+        pacer=_pacer(),
+        sleep=_nosleep,
+    )
+    assert src == "annas"

tests/zot/fetch/test_ledger.py

@@ -0,0 +1,107 @@
+from __future__ import annotations
+
+import json
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+from zot.fetch.ledger import RETRY_AFTER, TERMINAL, Attempt, Ledger
+
+T0 = datetime(2026, 9, 22, 12, 0, tzinfo=timezone.utc)
+
+
+def test_record_appends_one_json_line(tmp_path: Path):
+    led = Ledger(tmp_path / "ledger.jsonl")
+    att = led.record(
+        "ABCD2345",
+        outcome="attached",
+        source="libgen",
+        md5="a" * 32,
+        ids={"doi": "10.1/x"},
+        now=T0,
+    )
+    assert isinstance(att, Attempt)
+    lines = (tmp_path / "ledger.jsonl").read_text().splitlines()
+    assert len(lines) == 1
+    row = json.loads(lines[0])
+    assert row == {
+        "item_key": "ABCD2345",
+        "outcome": "attached",
+        "source": "libgen",
+        "md5": "a" * 32,
+        "ids": {"doi": "10.1/x"},
+        "note": "",
+        "ts": "2026-09-22T12:00:00+00:00",
+    }
+
+
+def test_unknown_item_should_try(tmp_path: Path):
+    led = Ledger(tmp_path / "ledger.jsonl")
+    assert led.should_try("NEW00000") is True
+
+
+def test_attached_item_never_retried(tmp_path: Path):
+    led = Ledger(tmp_path / "ledger.jsonl")
+    led.record("K1", outcome="attached", source="unpaywall", now=T0)
+    assert led.should_try("K1", now=T0 + timedelta(days=400)) is False
+
+
+def test_terminal_miss_waits_thirty_days(tmp_path: Path):
+    led = Ledger(tmp_path / "ledger.jsonl")
+    for outcome in TERMINAL:
+        led.record("K1", outcome=outcome, now=T0)
+        early = T0 + RETRY_AFTER - timedelta(seconds=1)
+        assert led.should_try("K1", now=early) is False
+        assert led.should_try("K1", now=T0 + RETRY_AFTER) is True
+
+
+def test_retry_flag_overrides_terminal_miss(tmp_path: Path):
+    led = Ledger(tmp_path / "ledger.jsonl")
+    led.record("K1", outcome="no_candidate", now=T0)
+    assert led.should_try("K1", retry=True, now=T0) is True
+
+
+def test_quota_miss_retries_immediately(tmp_path: Path):
+    led = Ledger(tmp_path / "ledger.jsonl")
+    led.record("K1", outcome="quota", now=T0)
+    assert led.should_try("K1", now=T0) is True
+
+
+def test_ledger_reloads_last_attempt_per_key(tmp_path: Path):
+    p = tmp_path / "ledger.jsonl"
+    Ledger(p).record("K1", outcome="no_candidate", now=T0)
+    Ledger(p).record("K1", outcome="attached", now=T0)
+    led = Ledger(p)
+    assert led.should_try("K1", now=T0 + timedelta(days=60)) is False
+
+
+def test_ledger_skips_corrupt_lines(tmp_path: Path):
+    p = tmp_path / "ledger.jsonl"
+    good = (
+        '{"item_key": "K1", "outcome": "attached", "ts": "2026-09-22T12:00:00+00:00"}'
+    )
+    p.write_text(good + "\nnot json\n")
+    led = Ledger(p)
+    assert led.should_try("K1") is False
+    assert led.should_try("K2") is True
+
+
+def test_summary_counts_this_session_only(tmp_path: Path):
+    p = tmp_path / "ledger.jsonl"
+    Ledger(p).record("K0", outcome="attached", now=T0)
+    led = Ledger(p)
+    led.record("K1", outcome="attached", now=T0)
+    led.record("K2", outcome="no_candidate", now=T0)
+    led.record("K3", outcome="no_candidate", now=T0)
+    assert led.summary() == {"attached": 1, "no_candidate": 2}
+
+
+def test_default_now_is_utc_iso(tmp_path: Path):
+    led = Ledger(tmp_path / "ledger.jsonl")
+    att = led.record("K1", outcome="attached")
+    assert att.ts.endswith("+00:00")
+
+
+def test_error_outcome_retries_immediately(tmp_path: Path):
+    led = Ledger(tmp_path / "ledger.jsonl")
+    led.record("K1", outcome="error", note="ConnectError: proxy down", now=T0)
+    assert led.should_try("K1", now=T0) is True

tests/zot/fetch/test_resolve.py

@@ -0,0 +1,333 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import httpx
+import pytest
+
+from zot.fetch.resolve import (
+    BROWSER_UA,
+    LIBGEN,
+    Candidate,
+    Notes,
+    Pacer,
+    candidates,
+    libgen_candidates,
+    libgen_editions,
+    libgen_files,
+    title_candidates,
+    title_matches,
+)
+from zot.fetch.select import ItemRef
+
+FX = Path(__file__).parent / "fixtures"
+TEMEL = "Early Palliative Care for Patients with Metastatic Non–Small-Cell Lung Cancer"
+
+
+def _fx(name: str) -> str:
+    return (FX / name).read_text()
+
+
+def _pacer() -> Pacer:
+    return Pacer(3.0, clock=lambda: 0.0, sleep=lambda s: None)
+
+
+def _client(handler) -> httpx.Client:
+    return httpx.Client(
+        transport=httpx.MockTransport(handler), headers={"User-Agent": BROWSER_UA}
+    )
+
+
+def _libgen_handler(seen: list[str]):
+    def handler(req: httpx.Request) -> httpx.Response:
+        seen.append(str(req.url))
+        assert req.headers["user-agent"] == BROWSER_UA
+        q = req.url.params
+        if req.url.path == "/json.php" and q.get("object") == "e":
+            if q.get("doi") == "10.1056/NEJMoa1000678":
+                return httpx.Response(200, text=_fx("libgen_json_doi.json"))
+            if q.get("isbn") == "9781640162280" or q.get("ids") == "146962484":
+                return httpx.Response(200, text=_fx("libgen_json_isbn.json"))
+            return httpx.Response(200, text="[]")
+        if req.url.path == "/json.php" and q.get("object") == "f":
+            want = set(q.get("ids", "").split(","))
+            allf = json.loads(_fx("libgen_json_file.json"))
+            return httpx.Response(
+                200, json={k: v for k, v in allf.items() if k in want}
+            )
+        if req.url.path == "/index.php":
+            return httpx.Response(200, text=_fx("libgen_index.html"))
+        return httpx.Response(404)
+
+    return handler
+
+
+def test_pacer_sleeps_only_the_remaining_interval():
+    now = [100.0]
+    slept: list[float] = []
+    p = Pacer(3.0, clock=lambda: now[0], sleep=slept.append)
+    p.wait()  # first call never sleeps
+    now[0] += 1.0
+    p.wait()
+    assert slept == [2.0]
+    now[0] += 10.0
+    p.wait()
+    assert slept == [2.0]
+
+
+def test_libgen_editions_doi_and_isbn_and_ids():
+    seen: list[str] = []
+    c = _client(_libgen_handler(seen))
+    e = libgen_editions(c, doi="10.1056/NEJMoa1000678", pacer=_pacer())
+    assert list(e) == ["27703795"]
+    assert list(libgen_editions(c, isbn="9781640162280", pacer=_pacer())) == [
+        "146962484"
+    ]
+    assert list(libgen_editions(c, ids=("146962484",), pacer=_pacer())) == ["146962484"]
+    assert seen[0].startswith(f"{LIBGEN}/json.php?object=e&addkeys=%2A&doi=")
+
+
+@pytest.mark.parametrize(
+    "status, body",
+    [(500, "x"), (200, "<html>nginx</html>"), (200, "[]"), (200, "null")],
+)
+def test_libgen_editions_bad_answers_are_empty(status, body):
+    c = _client(lambda req: httpx.Response(status, text=body))
+    assert libgen_editions(c, doi="10.1/x", pacer=_pacer()) == {}
+
+
+def test_libgen_editions_transport_error_is_empty():
+    def boom(req):
+        raise httpx.ConnectError("down")
+
+    assert libgen_editions(_client(boom), doi="10.1/x", pacer=_pacer()) == {}
+
+
+def test_libgen_files_batches_ids_in_one_call():
+    seen: list[str] = []
+    c = _client(_libgen_handler(seen))
+    f = libgen_files(c, ["27858481", "104009164"], pacer=_pacer())
+    assert set(f) == {"27858481", "104009164"}
+    assert len(seen) == 1 and "ids=27858481%2C104009164" in seen[0]
+    assert libgen_files(c, [], pacer=_pacer()) == {}
+
+
+def test_libgen_candidates_paper_pdf_only_with_size():
+    c = _client(_libgen_handler([]))
+    got = libgen_candidates(
+        c, kind="paper", doi="10.1056/NEJMoa1000678", pacer=_pacer()
+    )
+    assert got == [
+        Candidate(
+            "libgen", "a78add1bd62047ead5945134488f470f", None, "pdf", 401662, TEMEL
+        ),
+        Candidate(
+            "libgen", "b78add1bd62047ead5945134488f470f", None, "pdf", None, TEMEL
+        ),
+    ]
+
+
+def test_libgen_candidates_book_epub_before_pdf():
+    c = _client(_libgen_handler([]))
+    got = libgen_candidates(c, kind="book", isbn="9781640162280", pacer=_pacer())
+    assert [(g.ext, g.md5[:3]) for g in got] == [("epub", "220"), ("pdf", "120")]
+
+
+def test_libgen_file_lookup_missing_defaults_paper_to_pdf_drops_book():
+    def handler(req: httpx.Request) -> httpx.Response:
+        q = req.url.params
+        if q.get("object") == "f":
+            return httpx.Response(503)
+        if q.get("doi"):
+            return httpx.Response(200, text=_fx("libgen_json_doi.json"))
+        return httpx.Response(200, text=_fx("libgen_json_isbn.json"))
+
+    c = _client(handler)
+    paper = libgen_candidates(
+        c, kind="paper", doi="10.1056/NEJMoa1000678", pacer=_pacer()
+    )
+    assert [p.ext for p in paper] == ["pdf", "pdf"] and paper[0].size_bytes is None
+    book = libgen_candidates(c, kind="book", isbn="9781640162280", pacer=_pacer())
+    assert book == []
+
+
+def test_libgen_candidates_skips_files_without_md5():
+    body = json.dumps(
+        {"1": {"title": "T", "files": {"9": {"f_id": "9"}}, "year": "2000"}}
+    )
+    c = _client(lambda req: httpx.Response(200, text=body))
+    assert libgen_candidates(c, kind="paper", doi="10.1/x", pacer=_pacer()) == []
+
+
+def test_title_matches_exact_then_containment_with_year():
+    entries = [
+        {"id": "1", "title": "HCPCS Level II Professional 2023", "year": "2023"},
+        {"id": "2", "title": "HCPCS Level II Professional 2019", "year": "2019"},
+        {"id": "3", "title": "CPT 2023 Professional", "year": "2023"},
+    ]
+    exact = title_matches("HCPCS Level II Professional 2023", "", entries)
+    assert [e["id"] for e in exact] == ["1"]
+    by_year = title_matches("HCPCS Level II Professional", "2019", entries)
+    assert [e["id"] for e in by_year] == ["2"]
+    both = title_matches("HCPCS Level II Professional", "", entries)
+    assert [e["id"] for e in both] == ["1", "2"]
+    assert title_matches("Netter's Atlas", "", entries) == []
+
+
+def test_title_candidates_parses_index_then_fetches_editions():
+    seen: list[str] = []
+    c = _client(_libgen_handler(seen))
+    got = title_candidates(
+        c, title="HCPCS Level II Professional", year="2023", pacer=_pacer()
+    )
+    assert [g.ext for g in got] == ["epub", "pdf"]
+    assert any(
+        "/index.php?" in s and "req=HCPCS+Level+II+Professional" in s for s in seen
+    )
+    assert any("object=e" in s and "ids=146962484" in s for s in seen)
+
+
+def test_title_candidates_no_rows_or_error_is_empty():
+    empty = _client(lambda r: httpx.Response(200, text="<html></html>"))
+    assert title_candidates(empty, title="X", year="", pacer=_pacer()) == []
+    err = _client(lambda r: httpx.Response(500))
+    assert title_candidates(err, title="X", year="", pacer=_pacer()) == []
+    ok = _client(lambda r: httpx.Response(200))
+    assert title_candidates(ok, title="", year="", pacer=_pacer()) == []
+
+
+def test_title_candidates_transport_error_is_empty():
+    def boom(req):
+        raise httpx.ConnectError("down")
+
+    assert title_candidates(_client(boom), title="X", year="", pacer=_pacer()) == []
+
+
+def _direct(oa: bool):
+    def handler(req: httpx.Request) -> httpx.Response:
+        assert req.url.host == "api.unpaywall.org"
+        assert req.url.params["email"] == "dev@fhirworx.io"
+        if not oa:
+            return httpx.Response(404)
+        return httpx.Response(200, text=_fx("unpaywall_doi.json"))
+
+    return httpx.Client(transport=httpx.MockTransport(handler))
+
+
+def _ref(**kw) -> ItemRef:
+    base = dict(
+        item_id=1,
+        key="ABCD2345",
+        item_type="journalArticle",
+        title="T",
+        doi="",
+        isbn="",
+        year="",
+    )
+    base.update(kw)
+    return ItemRef(**base)
+
+
+def test_candidates_doi_unpaywall_first_then_libgen():
+    got = candidates(
+        _ref(doi="10.1056/NEJMoa1000678"),
+        direct=_direct(True),
+        proxied=_client(_libgen_handler([])),
+        email="dev@fhirworx.io",
+        up_pacer=_pacer(),
+        lg_pacer=_pacer(),
+    )
+    assert [g.source for g in got] == ["unpaywall"] * 3 + ["libgen"] * 2
+    assert got[0] == Candidate(
+        "unpaywall", None, "https://example.org/best.pdf", "pdf", None, "T"
+    )
+    assert got[3].md5 == "a78add1bd62047ead5945134488f470f"
+
+
+def test_candidates_isbn_book_and_title_fallback_and_nothing():
+    proxied = _client(_libgen_handler([]))
+    kw = dict(
+        direct=_direct(False),
+        proxied=proxied,
+        email="dev@fhirworx.io",
+        up_pacer=_pacer(),
+        lg_pacer=_pacer(),
+    )
+    book = candidates(_ref(item_type="book", isbn="9781640162280"), **kw)
+    assert [b.source for b in book] == ["libgen", "libgen"]
+    by_title = candidates(
+        _ref(item_type="book", title="HCPCS Level II Professional", year="2023"), **kw
+    )
+    assert [b.ext for b in by_title] == ["epub", "pdf"]
+    assert candidates(_ref(title="paper without doi"), **kw) == []
+    assert candidates(_ref(item_type="book", title=""), **kw) == []
+
+
+def test_libgen_candidates_skips_non_dict_editions():
+    body = json.dumps({"1": None, "2": [1, 2], "3": {"title": "T", "files": []}})
+    c = _client(lambda req: httpx.Response(200, text=body))
+    assert libgen_candidates(c, kind="paper", doi="10.1/x", pacer=_pacer()) == []
+
+
+# ── fix pass (review) ─────────────────────────────────────────────
+
+
+def test_notes_error_for_http_and_transport_miss_for_empty():
+    n = Notes()
+    libgen_editions(
+        _client(lambda r: httpx.Response(503)), doi="10.1/x", pacer=_pacer(), notes=n
+    )
+    assert n.had_error and "503" in str(n)
+
+    def boom(req):
+        raise httpx.ConnectError("down")
+
+    n2 = Notes()
+    libgen_editions(_client(boom), doi="10.1/x", pacer=_pacer(), notes=n2)
+    assert n2.had_error and "ConnectError" in str(n2)
+
+    n3 = Notes()
+    libgen_candidates(
+        _client(lambda r: httpx.Response(200, text="[]")),
+        kind="paper",
+        doi="10.1/x",
+        pacer=_pacer(),
+        notes=n3,
+    )
+    assert not n3.had_error and "no edition" in str(n3)
+
+
+def test_candidates_note_unpaywall_miss():
+    n = Notes()
+    got = candidates(
+        _ref(doi="10.1/x"),
+        direct=_direct(False),
+        proxied=_client(lambda r: httpx.Response(200, text="[]")),
+        email="dev@fhirworx.io",
+        up_pacer=_pacer(),
+        lg_pacer=_pacer(),
+        notes=n,
+    )
+    assert got == [] and "unpaywall" in str(n) and not n.had_error
+
+
+def test_libgen_candidates_tolerates_bad_file_entries():
+    body = json.dumps(
+        {
+            "1": {
+                "title": "T",
+                "files": {"a": None, "b": 3, "c": {"f_id": "9", "md5": "c" * 32}},
+            }
+        }
+    )
+
+    def h(r):
+        if r.url.params.get("object") == "f":
+            return httpx.Response(
+                200, json={"9": {"extension": "pdf", "filesize": "n/a"}}
+            )
+        return httpx.Response(200, text=body)
+
+    got = libgen_candidates(_client(h), kind="paper", doi="10.1/x", pacer=_pacer())
+    assert [(g.md5, g.ext, g.size_bytes) for g in got] == [("c" * 32, "pdf", None)]

tests/zot/fetch/test_run.py

@@ -0,0 +1,302 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import httpx
+import pytest
+
+from tests.zot.fetch.conftest import mk_item
+from zot.db import Db
+from zot.fetch.attach import has_full_text
+from zot.fetch.deliver import MIN_BYTES, AnnasAuthError, Budget
+from zot.fetch.ledger import Ledger
+from zot.fetch.resolve import BROWSER_UA
+from zot.fetch.run import STATS, Clients, make_clients, run, safe_name
+from zot.fetch.select import select_items
+
+FX = Path(__file__).parent / "fixtures"
+PDF = b"%PDF-1.4\n" + b"0" * MIN_BYTES
+PDF_HDR = {"content-type": "application/pdf"}
+ADS = (FX / "libgen_ads.html").read_text()
+DOI = "10.1056/NEJMoa1000678"
+
+
+def _py_copier(src: Path, dest_dir: Path) -> None:
+    dest_dir.mkdir(parents=True, exist_ok=True)
+    (dest_dir / src.name).write_bytes(src.read_bytes())
+
+
+def _clients(direct_handler, proxied_handler) -> Clients:
+    return Clients(
+        direct=httpx.Client(transport=httpx.MockTransport(direct_handler)),
+        proxied=httpx.Client(
+            transport=httpx.MockTransport(proxied_handler),
+            headers={"User-Agent": BROWSER_UA},
+        ),
+    )
+
+
+def _unpaywall_ok(r: httpx.Request) -> httpx.Response:
+    if r.url.host == "api.unpaywall.org":
+        return httpx.Response(200, text=(FX / "unpaywall_doi.json").read_text())
+    return httpx.Response(200, content=PDF, headers=PDF_HDR)
+
+
+def _unpaywall_miss(r: httpx.Request) -> httpx.Response:
+    return httpx.Response(404)
+
+
+def _libgen_ok(r: httpx.Request) -> httpx.Response:
+    q = r.url.params
+    if r.url.path == "/json.php" and q.get("object") == "e":
+        return httpx.Response(200, text=(FX / "libgen_json_doi.json").read_text())
+    if r.url.path == "/json.php":
+        return httpx.Response(200, text=(FX / "libgen_json_file.json").read_text())
+    if r.url.path == "/ads.php":
+        return httpx.Response(200, text=ADS)
+    if r.url.path == "/get.php":
+        return httpx.Response(200, content=PDF, headers=PDF_HDR)
+    return httpx.Response(404)
+
+
+def _libgen_none(r: httpx.Request) -> httpx.Response:
+    return httpx.Response(200, text="[]")
+
+
+def _libgen_dead(r: httpx.Request) -> httpx.Response:
+    """Every route answers definitively (a miss), nothing errors."""
+    if r.url.path == "/json.php" and r.url.params.get("object") == "e":
+        return httpx.Response(200, text=(FX / "libgen_json_doi.json").read_text())
+    if r.url.path == "/json.php":
+        return httpx.Response(200, text=(FX / "libgen_json_file.json").read_text())
+    if r.url.path == "/ads.php":
+        return httpx.Response(200, text=ADS)
+    if r.url.host == "annas-archive.gl":
+        return httpx.Response(200, json={"download_url": None, "error": "nope"})
+    return httpx.Response(404)
+
+
+def _setup(tmp_path: Path, **kw):
+    return dict(
+        ledger=Ledger(tmp_path / "ledger.jsonl"),
+        budget=kw.pop("budget", Budget()),
+        email="dev@fhirworx.io",
+        storage_dir=tmp_path / "storage",
+        scratch_dir=tmp_path / "scratch",
+        copier=kw.pop("copier", _py_copier),
+        sleep=lambda s: None,
+        out=kw.pop("out", lambda *a: None),
+        **kw,
+    )
+
+
+def _zero(**over) -> dict[str, int]:
+    d = dict.fromkeys(STATS, 0)
+    d.update(over)
+    return d
+
+
+def test_safe_name():
+    assert safe_name("Early Palliative Care: NSCLC / 2010", "K") == (
+        "Early Palliative Care NSCLC 2010"
+    )
+    assert safe_name("", "ABCD2345") == "ABCD2345"
+    assert len(safe_name("x" * 200, "K")) == 80
+
+
+def test_make_clients_sets_proxy_and_agents():
+    c = make_clients("http://127.0.0.1:19035", email="dev@fhirworx.io")
+    try:
+        assert c.proxied.headers["user-agent"] == BROWSER_UA
+        assert "dev@fhirworx.io" in c.direct.headers["user-agent"]
+        assert c.proxied._transport is not c.direct._transport
+    finally:
+        c.direct.close()
+        c.proxied.close()
+
+
+def test_run_attaches_from_unpaywall_and_ledgers(zdb: Db, tmp_path: Path):
+    item = mk_item(zdb, title="Early Palliative Care", doi=DOI)
+    refs = select_items(zdb, all_=True)
+    lines: list[str] = []
+    kw = _setup(tmp_path, out=lambda *a: lines.append(" ".join(map(str, a))))
+    stats = run(zdb, refs, clients=_clients(_unpaywall_ok, _libgen_none), **kw)
+    assert stats == _zero(seen=1, attempted=1, attached=1)
+    assert has_full_text(zdb, item)
+    assert "source:unpaywall" in zdb.get_tags(item)
+    att = zdb.con.execute(
+        "SELECT path FROM itemAttachments WHERE parentItemID = ?", (item,)
+    ).fetchone()[0]
+    assert att == "storage:Early Palliative Care.pdf"
+    assert kw["ledger"].summary() == {"attached": 1}
+    assert any("attached" in ln for ln in lines)
+    assert not list((tmp_path / "scratch").iterdir())  # scratch file cleaned up
+
+
+def test_run_falls_through_to_libgen_with_md5_provenance(zdb: Db, tmp_path: Path):
+    item = mk_item(zdb, title="T", doi=DOI)
+    kw = _setup(tmp_path)
+    refs = select_items(zdb, all_=True)
+    stats = run(zdb, refs, clients=_clients(_unpaywall_miss, _libgen_ok), **kw)
+    assert stats["attached"] == 1
+    assert "source:libgen" in zdb.get_tags(item)
+    assert zdb.get_field(item, "extra") == "md5: a78add1bd62047ead5945134488f470f"
+
+
+def test_run_no_candidate_and_all_failed_outcomes(zdb: Db, tmp_path: Path):
+    mk_item(zdb, title="no ids")
+    mk_item(zdb, title="dead", doi=DOI)
+    kw = _setup(tmp_path)
+    refs = select_items(zdb, all_=True)
+    stats = run(zdb, refs, clients=_clients(_unpaywall_miss, _libgen_dead), **kw)
+    assert stats == _zero(seen=2, attempted=2, no_candidate=1, all_failed=1)
+    assert kw["ledger"].summary() == {"no_candidate": 1, "all_failed": 1}
+
+
+def test_run_ledgers_quota_when_only_annas_was_left(zdb: Db, tmp_path: Path):
+    mk_item(zdb, title="dead", doi=DOI)
+    kw = _setup(tmp_path, budget=Budget(annas_key="k", annas_left=1, reserve=5))
+    refs = select_items(zdb, all_=True)
+    stats = run(zdb, refs, clients=_clients(_unpaywall_miss, _libgen_dead), **kw)
+    assert stats["quota"] == 1 and stats["all_failed"] == 0
+    assert kw["ledger"].summary() == {"quota": 1}
+
+
+def test_run_skips_ledgered_items_and_honours_limit(zdb: Db, tmp_path: Path):
+    a = mk_item(zdb, title="A", doi=DOI)
+    mk_item(zdb, title="B", doi=DOI)
+    mk_item(zdb, title="C", doi=DOI)
+    refs = select_items(zdb, all_=True)
+    kw = _setup(tmp_path)
+    kw["ledger"].record(refs[0].key, outcome="no_candidate")
+    clients = _clients(_unpaywall_ok, _libgen_none)
+    stats = run(zdb, refs, clients=clients, limit=1, **kw)
+    assert stats == _zero(seen=2, skipped=1, attempted=1, attached=1)
+    assert not has_full_text(zdb, a)
+
+
+def test_run_retry_flag_reprocesses_ledgered_items(zdb: Db, tmp_path: Path):
+    mk_item(zdb, title="A", doi=DOI)
+    refs = select_items(zdb, all_=True)
+    kw = _setup(tmp_path)
+    kw["ledger"].record(refs[0].key, outcome="all_failed")
+    clients = _clients(_unpaywall_ok, _libgen_none)
+    stats = run(zdb, refs, clients=clients, retry=True, **kw)
+    assert stats["skipped"] == 0 and stats["attached"] == 1
+
+
+def test_run_dry_run_prints_candidates_and_writes_nothing(zdb: Db, tmp_path: Path):
+    item = mk_item(zdb, title="T", doi=DOI)
+    mk_item(zdb, title="no ids")
+    lines: list[str] = []
+    kw = _setup(tmp_path, out=lambda *a: lines.append(" ".join(map(str, a))))
+    refs = select_items(zdb, all_=True)
+    clients = _clients(_unpaywall_ok, _libgen_ok)
+    stats = run(zdb, refs, clients=clients, dry_run=True, **kw)
+    assert stats == _zero(seen=2, attempted=2)
+    assert not has_full_text(zdb, item)
+    assert not (tmp_path / "ledger.jsonl").exists()
+    assert any("unpaywall" in ln for ln in lines)
+    assert any("libgen" in ln for ln in lines)
+    assert any("no candidates" in ln for ln in lines)
+
+
+def test_run_annas_auth_error_aborts(zdb: Db, tmp_path: Path):
+    mk_item(zdb, title="T", doi=DOI)
+
+    def proxied(r: httpx.Request) -> httpx.Response:
+        if r.url.host == "annas-archive.gl":
+            return httpx.Response(401)
+        return _libgen_dead(r)
+
+    kw = _setup(tmp_path, budget=Budget(annas_key="bad"))
+    refs = select_items(zdb, all_=True)
+    with pytest.raises(AnnasAuthError):
+        run(zdb, refs, clients=_clients(_unpaywall_miss, proxied), **kw)
+
+
+# ── fix pass (review) ─────────────────────────────────────────────
+
+
+def _proxy_down(r: httpx.Request) -> httpx.Response:
+    raise httpx.ConnectError("proxy down")
+
+
+def test_run_transient_failure_is_error_not_terminal(zdb: Db, tmp_path: Path):
+    mk_item(zdb, title="T", doi=DOI)
+    kw = _setup(tmp_path)
+    refs = select_items(zdb, all_=True)
+    stats = run(zdb, refs, clients=_clients(_unpaywall_miss, _proxy_down), **kw)
+    assert stats == _zero(seen=1, attempted=1, error=1)
+    led = kw["ledger"]
+    assert led.summary() == {"error": 1}
+    assert led.should_try(refs[0].key) is True  # retried next run, not 30 days out
+    line = (tmp_path / "ledger.jsonl").read_text()
+    assert "ConnectError" in line
+
+
+def test_run_all_failed_and_no_candidate_carry_notes(zdb: Db, tmp_path: Path):
+    mk_item(zdb, title="no ids")
+    mk_item(zdb, title="dead", doi=DOI)
+    kw = _setup(tmp_path)
+    refs = select_items(zdb, all_=True)
+    run(zdb, refs, clients=_clients(_unpaywall_miss, _libgen_dead), **kw)
+    lines = (tmp_path / "ledger.jsonl").read_text().splitlines()
+    assert "no identifiers" in lines[0]
+    assert "unpaywall" in lines[1] and "libgen get: http 404" in lines[1]
+
+
+def test_run_isolates_a_failing_item_and_rolls_back(zdb: Db, tmp_path: Path):
+    bad = mk_item(zdb, title="bad", doi=DOI)
+    good = mk_item(zdb, title="good", doi=DOI)
+    calls = {"n": 0}
+
+    def flaky_copier(src: Path, dest_dir: Path) -> None:
+        calls["n"] += 1
+        if calls["n"] == 1:
+            raise RuntimeError("sudo: a password is required")
+        _py_copier(src, dest_dir)
+
+    kw = _setup(tmp_path, copier=flaky_copier)
+    refs = select_items(zdb, all_=True)
+    stats = run(zdb, refs, clients=_clients(_unpaywall_ok, _libgen_none), **kw)
+    assert stats == _zero(seen=2, attempted=2, attached=1, error=1)
+    # the failed item left no rows behind (rolled back) and no scratch file
+    rows = zdb.con.execute(
+        "SELECT COUNT(*) FROM itemAttachments WHERE parentItemID = ?", (bad,)
+    ).fetchone()[0]
+    assert rows == 0 and "source:unpaywall" not in zdb.get_tags(bad)
+    assert has_full_text(zdb, good)
+    assert not list((tmp_path / "scratch").iterdir())
+    assert "RuntimeError" in (tmp_path / "ledger.jsonl").read_text()
+    assert kw["ledger"].should_try(refs[0].key) is True
+
+
+def test_run_annas_auth_error_still_aborts_after_fix(zdb: Db, tmp_path: Path):
+    mk_item(zdb, title="T", doi=DOI)
+
+    def proxied(r: httpx.Request) -> httpx.Response:
+        if r.url.host == "annas-archive.gl":
+            return httpx.Response(403, json={"error": "Invalid secret key"})
+        return _libgen_dead(r)
+
+    kw = _setup(tmp_path, budget=Budget(annas_key="bad"))
+    refs = select_items(zdb, all_=True)
+    with pytest.raises(AnnasAuthError):
+        run(zdb, refs, clients=_clients(_unpaywall_miss, proxied), **kw)
+
+
+def test_run_delivery_error_is_error_outcome(zdb: Db, tmp_path: Path):
+    """Candidates exist but every delivery route errored (5xx) → retry next run."""
+
+    def flaky(r: httpx.Request) -> httpx.Response:
+        if r.url.path == "/ads.php":
+            return httpx.Response(503)
+        return _libgen_ok(r)
+
+    mk_item(zdb, title="T", doi=DOI)
+    kw = _setup(tmp_path)
+    refs = select_items(zdb, all_=True)
+    stats = run(zdb, refs, clients=_clients(_unpaywall_miss, flaky), **kw)
+    assert stats == _zero(seen=1, attempted=1, error=1)
+    assert "503" in (tmp_path / "ledger.jsonl").read_text()

tests/zot/fetch/test_select.py

@@ -0,0 +1,118 @@
+from __future__ import annotations
+
+import pytest
+
+from tests.zot.fetch.conftest import mk_item
+from zot.db import Db
+from zot.fetch.select import (
+    ItemRef,
+    first_isbn,
+    item_ref,
+    normalise_doi,
+    select_items,
+    year_of,
+)
+
+
+@pytest.mark.parametrize(
+    "raw, want",
+    [
+        ("10.1056/NEJMoa1000678", "10.1056/NEJMoa1000678"),
+        ("  https://doi.org/10.1056/NEJMoa1000678 ", "10.1056/NEJMoa1000678"),
+        ("http://dx.doi.org/10.1/ab", "10.1/ab"),
+        ("doi:10.1/ab", "10.1/ab"),
+        ("", ""),
+    ],
+)
+def test_doi_normalised_from_url_form(raw, want):
+    assert normalise_doi(raw) == want
+
+
+@pytest.mark.parametrize(
+    "raw, want",
+    [
+        ("978-1-64016-228-0 1640162283", "9781640162280"),
+        ("1640162283", "1640162283"),
+        ("ISBN 0-19-852663-6", "0198526636"),
+        ("no isbn here", ""),
+        ("", ""),
+    ],
+)
+def test_isbn_prefers_13_digit_and_strips_hyphens(raw, want):
+    assert first_isbn(raw) == want
+
+
+@pytest.mark.parametrize(
+    "raw, want", [("2010-08-19", "2010"), ("August 2010", "2010"), ("", "")]
+)
+def test_year_of(raw, want):
+    assert year_of(raw) == want
+
+
+def test_item_ref_reads_fields_and_kind(zdb: Db):
+    paper = mk_item(zdb, title="P", doi="https://doi.org/10.1/x", date="2020-01-01")
+    book = mk_item(zdb, type_name="book", title="B", isbn="978-1-64016-228-0")
+    p = item_ref(zdb, paper)
+    b = item_ref(zdb, book)
+    key = zdb.get_item(paper)["key"]
+    assert p == ItemRef(paper, key, "journalArticle", "P", "10.1/x", "", "2020")
+    assert p.kind == "paper"
+    assert p.ids == {"doi": "10.1/x"}
+    assert b.isbn == "9781640162280" and b.kind == "book"
+
+
+def test_select_requires_exactly_one_primary_selector(zdb: Db):
+    with pytest.raises(ValueError, match="exactly one"):
+        select_items(zdb)
+    with pytest.raises(ValueError, match="exactly one"):
+        select_items(zdb, tag="t", all_=True)
+
+
+def test_select_by_tag_type_item_collection_all(zdb: Db):
+    a = mk_item(zdb, title="A", doi="10.1/a")
+    b = mk_item(zdb, type_name="book", title="B", isbn="9781640162280")
+    zdb.tag_item(a, "rfi:codeset")
+    ckey = zdb.ensure_collection("AMA Coding Publications")
+    zdb.add_to_collection(b, collection_key=ckey)
+    assert [r.item_id for r in select_items(zdb, tag="rfi:codeset")] == [a]
+    assert [r.item_id for r in select_items(zdb, type_="book")] == [b]
+    akey = zdb.get_item(a)["key"]
+    assert [r.item_id for r in select_items(zdb, item=akey)] == [a]
+    got = select_items(zdb, collection="AMA Coding Publications")
+    assert [r.item_id for r in got] == [b]
+    assert [r.item_id for r in select_items(zdb, all_=True)] == [a, b]
+
+
+def test_type_filter_combines_with_primary_selector(zdb: Db):
+    a = mk_item(zdb, title="A")
+    b = mk_item(zdb, type_name="book", title="B")
+    ckey = zdb.ensure_collection("C")
+    zdb.add_to_collection(a, collection_key=ckey)
+    zdb.add_to_collection(b, collection_key=ckey)
+    got = select_items(zdb, collection="C", type_="book")
+    assert [r.item_id for r in got] == [b]
+
+
+def test_unknown_collection_or_item_or_type_raises(zdb: Db):
+    with pytest.raises(ValueError, match="no collection named"):
+        select_items(zdb, collection="Nope")
+    with pytest.raises(ValueError, match="no item with key"):
+        select_items(zdb, item="ZZZZZZZZ")
+    with pytest.raises(ValueError, match="unknown item type"):
+        select_items(zdb, type_="novel")
+
+
+def test_select_excludes_full_text_deleted_document_and_children(zdb: Db):
+    has_pdf = mk_item(zdb, title="has pdf")
+    zdb.add_attachment(has_pdf, content_type="application/pdf", path="storage:x.pdf")
+    deleted = mk_item(zdb, title="deleted")
+    zdb.con.execute("INSERT INTO deletedItems (itemID) VALUES (?)", (deleted,))
+    mk_item(zdb, type_name="document", title="reg pdf")
+    keep = mk_item(zdb, title="keep")
+    got = [r.item_id for r in select_items(zdb, all_=True)]
+    assert got == [keep]
+
+
+def test_select_all_excludes_document_even_when_typed(zdb: Db):
+    mk_item(zdb, type_name="document", title="reg pdf")
+    assert select_items(zdb, type_="document") == []