ClawLibrary v1: open-access research vault harvester

Topic-driven harvester that builds a deduplicated, metadata-rich offline PDF
vault from legitimate open-access APIs (OpenAlex, Unpaywall, arXiv, Semantic
Scholar, CORE) plus an optional, config-gated sanctioned-TDM path for
subscription content. Replaces the original PRD's authenticated discovery-layer
scraping with channels designed for programmatic research access.

- config/topics loaders (Pydantic), PaperCandidate/VaultEntry models
- OpenAlex search + multi-source OA PDF resolver chain (tries alternatives)
- atomic dedup state, polite rate limiting + backoff, streaming PDF downloader
  with content-type/magic-byte validation, vault writer + index.json/csv
- CLI: run/stats/export; structured JSONL run logs
- 47 tests, 87% coverage; verified live against real OA APIs

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

.env.example

@@ -0,0 +1,13 @@
+# Identity for polite API access (required by OpenAlex polite pool & Unpaywall).
+OPENALEX_MAILTO=[email protected]
+UNPAYWALL_EMAIL=[email protected]
+
+# Optional API keys for additional open-access sources.
+# CORE_API_KEY=...
+# SEMANTIC_SCHOLAR_KEY=...
+
+# Sanctioned publisher TDM tokens (request from your subject librarian).
+# Any var ending in _TDM_TOKEN is auto-detected; the part before it is the publisher.
+# Set sources.tdm: true in config.yaml to activate once a token is present.
+# ELSEVIER_TDM_TOKEN=...
+# WILEY_TDM_TOKEN=...

.gitignore

@@ -0,0 +1,25 @@
+# Secrets & local state
+.env
+state.json
+
+# Vault & logs (local research data)
+vault/
+logs/
+
+# Python
+__pycache__/
+*.py[cod]
+*.egg-info/
+.eggs/
+build/
+dist/
+.venv/
+venv/
+.pytest_cache/
+.ruff_cache/
+.coverage
+htmlcov/
+
+# Generated exports
+export.csv
+export.json

README.md

@@ -0,0 +1,99 @@
+# ClawLibrary
+
+A topic-driven harvester that builds a **deduplicated, metadata-rich, offline
+research vault** of **legally available full-text PDFs** — organized by topic,
+with a JSON sidecar per paper and a vault-level `index.json` / `index.csv` for
+downstream ingestion.
+
+## What it does (and what it deliberately does not)
+
+ClawLibrary acquires full text only through channels designed for programmatic
+research access:
+
+- **OpenAlex** — topic search + rich metadata (DOI, authors, abstract, venue, OA status)
+- **Unpaywall** — DOI → lawfully posted open-access PDF
+- **arXiv / PubMed Central / Semantic Scholar / CORE** — open-access full text
+- **Sanctioned publisher TDM APIs** — *optional, off by default* — the legitimate
+  faculty route for subscription content (see [TDM access](#subscription-content-tdm))
+
+It does **not** automate logins to a library discovery layer, scrape
+authenticated Primo/SFX/publisher sessions, or use timing tricks to stay under
+anti-automation detection. Publisher license agreements prohibit systematic
+downloading by *all* authorized users, and detection cuts off access for the
+*entire institution* — so that path is out of scope by design.
+
+## Install
+
+```bash
+python -m venv .venv && source .venv/bin/activate
+pip install -e ".[dev]"
+playwright_not_required=true   # no browser needed
+cp .env.example .env           # add your contact email(s)
+```
+
+Set at least `OPENALEX_MAILTO` and `UNPAYWALL_EMAIL` in `.env` (these identify
+you to the polite API pools — no passwords involved).
+
+## Usage
+
+```bash
+# Resolve only — see what WOULD be downloaded, fetch nothing
+python -m clawlibrary run --dry-run
+
+# Harvest the topic queue (respects max_downloads_per_session)
+python -m clawlibrary run
+
+# Single topic, capped at 3 PDFs
+python -m clawlibrary run --topic "Rust systems programming memory safety" --max 3
+
+# Vault statistics
+python -m clawlibrary stats
+
+# Export the index
+python -m clawlibrary export --format csv --out export.csv
+```
+
+## Configuration
+
+- **`topics.yaml`** — the topic queue. Each topic: `query`, `max_results`,
+  `priority`, `enabled`, and `filters` (`date_from`, `date_to`, `language`,
+  `open_access_only`).
+- **`config.yaml`** — sources, rate limits, vault path, logging.
+- **`.env`** — contact emails, optional API keys, optional TDM tokens. Never
+  commit this file (it is gitignored).
+
+## Vault layout
+
+```
+vault/
+├── index.json                  # array of all sidecars
+├── index.csv                   # flat table
+└── {topic_slug}/
+    ├── {doi_slug}.pdf
+    └── {doi_slug}.json         # VaultEntry sidecar
+```
+
+Deduplication is by DOI (primary) then PDF-URL hash (fallback), tracked in
+`state.json`. Re-running never re-downloads; an interrupted run resumes cleanly.
+
+## Subscription content (TDM)
+
+For papers with no open-access copy, the legitimate path is your institution's
+**Text & Data Mining** service:
+
+1. Email your subject librarian; request TDM/API access for a research corpus,
+   naming the publishers you need.
+2. They issue an institutional token (or register your IP).
+3. Add it to `.env` (e.g. `ELSEVIER_TDM_TOKEN=...`) and set `sources.tdm: true`
+   in `config.yaml`.
+
+Until a token is present, non-OA papers are simply skipped and logged.
+
+## Logs
+
+Each run writes structured JSONL to `logs/{run_id}.jsonl` and prints a summary
+table on completion.
+
+---
+
+**RedClaw Systems LLC** — internal research infrastructure.

clawlibrary/__init__.py

@@ -0,0 +1,9 @@
+"""ClawLibrary — open-access research vault builder.
+
+Harvests legally available full-text PDFs (open access + sanctioned TDM channels)
+into a deduplicated, metadata-rich local vault. No authenticated scraping of
+subscription discovery layers; acquisition is via APIs designed for programmatic
+research access.
+"""
+
+__version__ = "1.0.0"

clawlibrary/__main__.py

@@ -0,0 +1,171 @@
+"""ClawLibrary CLI.
+
+Commands:
+  run [--topic Q] [--dry-run] [--max N]   harvest the topic queue
+  stats                                    show vault statistics
+  export [--format csv|json] [--out P]     export the vault index
+"""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import json
+import sys
+from pathlib import Path
+
+from loguru import logger
+from rich.console import Console
+from rich.table import Table
+
+from .config import Config, Topic, load_config, load_topics
+from .pipeline import RunStats, run
+from .vault import VaultManager
+
+console = Console()
+
+
+def _load(config_path: str, topics_path: str) -> tuple[Config, list[Topic]]:
+    config = load_config(config_path)
+    queue = load_topics(topics_path)
+    return config, queue.active()
+
+
+def _print_summary(stats: RunStats, *, dry_run: bool) -> None:
+    table = Table(title=f"Run {stats.run_id}", show_header=True, header_style="bold")
+    table.add_column("Metric")
+    table.add_column("Count", justify="right")
+    table.add_row("Topics processed", str(stats.topics_processed))
+    if dry_run:
+        table.add_row("Would download", str(stats.would_download))
+    else:
+        table.add_row("Downloaded", str(stats.downloaded))
+    table.add_row("Skipped (dedup)", str(stats.skipped_dedup))
+    table.add_row("Skipped (no OA PDF)", str(stats.skipped_no_pdf))
+    table.add_row("Failed", str(stats.failed))
+    if stats.suspended:
+        table.add_row("Status", "[red]SUSPENDED (server errors)[/red]")
+    console.print(table)
+
+
+def cmd_run(args: argparse.Namespace) -> int:
+    config, topics = _load(args.config, args.topics)
+    if args.topic:
+        topics = [t for t in topics if t.query == args.topic]
+        if not topics:
+            console.print(f"[red]No enabled topic matching:[/red] {args.topic}")
+            return 1
+
+    def on_event(kind: str, **data):
+        if kind == "topic_start":
+            console.print(f"[bold cyan]→ {data['topic']}[/bold cyan]")
+        elif kind == "downloaded":
+            t = data["title"][:80]
+            console.print(f"  [green]✓[/green] {t}  [dim]({data['source']})[/dim]")
+        elif kind == "would_download":
+            t = data["title"][:80]
+            console.print(f"  [yellow]·[/yellow] {t}  [dim]({data['source']})[/dim]")
+        elif kind == "failed":
+            title = data.get("title", "")[:70]
+            console.print(f"  [red]✗[/red] {title}  [dim]{data['reason']}[/dim]")
+        elif kind == "run_suspended":
+            console.print("[red]Run suspended: too many consecutive server errors.[/red]")
+
+    stats = run(config, topics, dry_run=args.dry_run, max_override=args.max, on_event=on_event)
+    _print_summary(stats, dry_run=args.dry_run)
+    return 0
+
+
+def cmd_stats(args: argparse.Namespace) -> int:
+    config, _ = _load(args.config, args.topics)
+    vault = VaultManager(config.vault.root)
+    entries = vault.load_index()
+    if not entries:
+        console.print("[yellow]Vault is empty (no index.json).[/yellow]")
+        return 0
+
+    by_topic: dict[str, int] = {}
+    by_source: dict[str, int] = {}
+    total_bytes = 0
+    for e in entries:
+        by_topic[e.get("topic", "?")] = by_topic.get(e.get("topic", "?"), 0) + 1
+        by_source[e.get("source") or "?"] = by_source.get(e.get("source") or "?", 0) + 1
+        total_bytes += int(e.get("file_size_bytes", 0) or 0)
+
+    table = Table(title="Vault statistics", header_style="bold")
+    table.add_column("Metric")
+    table.add_column("Value", justify="right")
+    table.add_row("Total papers", str(len(entries)))
+    table.add_row("Total size", f"{total_bytes / 1_048_576:.1f} MiB")
+    table.add_row("Topics", str(len(by_topic)))
+    console.print(table)
+
+    src_table = Table(title="By source", header_style="bold")
+    src_table.add_column("Source")
+    src_table.add_column("Papers", justify="right")
+    for src, n in sorted(by_source.items(), key=lambda kv: -kv[1]):
+        src_table.add_row(src, str(n))
+    console.print(src_table)
+    return 0
+
+
+def cmd_export(args: argparse.Namespace) -> int:
+    config, _ = _load(args.config, args.topics)
+    vault = VaultManager(config.vault.root)
+    entries = vault.load_index()
+    out = Path(args.out) if args.out else Path(f"export.{args.format}")
+
+    if args.format == "json":
+        out.write_text(json.dumps(entries, indent=2))
+    else:
+        fields = [
+            "id", "doi", "title", "authors", "journal", "year", "topic",
+            "source", "vault_path", "file_size_bytes", "downloaded_at",
+        ]
+        with open(out, "w", newline="") as f:
+            w = csv.DictWriter(f, fieldnames=fields, extrasaction="ignore")
+            w.writeheader()
+            for e in entries:
+                row = dict(e)
+                if isinstance(row.get("authors"), list):
+                    row["authors"] = "; ".join(row["authors"])
+                w.writerow(row)
+    console.print(f"[green]Exported {len(entries)} entries to {out}[/green]")
+    return 0
+
+
+def build_parser() -> argparse.ArgumentParser:
+    p = argparse.ArgumentParser(prog="clawlibrary", description=__doc__)
+    p.add_argument("--config", default="config.yaml", help="path to config.yaml")
+    p.add_argument("--topics", default="topics.yaml", help="path to topics.yaml")
+    sub = p.add_subparsers(dest="command", required=True)
+
+    pr = sub.add_parser("run", help="harvest the topic queue")
+    pr.add_argument("--topic", help="run a single topic by exact query string")
+    pr.add_argument("--dry-run", action="store_true", help="resolve only, no downloads")
+    pr.add_argument("--max", type=int, default=None, help="override max downloads this run")
+    pr.set_defaults(func=cmd_run)
+
+    ps = sub.add_parser("stats", help="show vault statistics")
+    ps.set_defaults(func=cmd_stats)
+
+    pe = sub.add_parser("export", help="export the vault index")
+    pe.add_argument("--format", choices=["csv", "json"], default="csv")
+    pe.add_argument("--out", help="output path")
+    pe.set_defaults(func=cmd_export)
+    return p
+
+
+def main(argv: list[str] | None = None) -> int:
+    logger.remove()  # quiet console; pipeline adds a JSONL file sink per run
+    parser = build_parser()
+    args = parser.parse_args(argv)
+    try:
+        return args.func(args)
+    except FileNotFoundError as e:
+        console.print(f"[red]Error:[/red] {e}")
+        return 2
+
+
+if __name__ == "__main__":
+    sys.exit(main())

clawlibrary/config.py

@@ -0,0 +1,166 @@
+"""Configuration loading and validation for ClawLibrary.
+
+Two files drive the tool:
+  * config.yaml  — system config (sources, rate limits, vault path, logging)
+  * topics.yaml  — the topic queue (what to search for)
+
+Secrets/keys come from the environment (.env), never from YAML.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+import yaml
+from dotenv import load_dotenv
+from pydantic import BaseModel, Field, field_validator
+
+
+class TopicFilters(BaseModel):
+    date_from: int | None = None
+    date_to: int | None = None
+    language: str | None = None
+    # If True (default), only download papers with a legally available OA full text.
+    open_access_only: bool = True
+
+
+class Topic(BaseModel):
+    query: str
+    max_results: int = 40
+    priority: int = 100
+    enabled: bool = True
+    filters: TopicFilters = Field(default_factory=TopicFilters)
+
+    @field_validator("query")
+    @classmethod
+    def _nonempty_query(cls, v: str) -> str:
+        if not v or not v.strip():
+            raise ValueError("topic query must be a non-empty string")
+        return v.strip()
+
+
+class TopicQueue(BaseModel):
+    defaults: dict = Field(default_factory=dict)
+    topics: list[Topic]
+
+    def active(self) -> list[Topic]:
+        """Enabled topics, ordered by priority (lower first), stable on input order."""
+        enabled = [t for t in self.topics if t.enabled]
+        return sorted(enabled, key=lambda t: t.priority)
+
+
+class VaultConfig(BaseModel):
+    root: Path = Path("./vault")
+
+
+class SourcesConfig(BaseModel):
+    """Which acquisition sources are active and how to identify ourselves.
+
+    Emails are required by OpenAlex (polite pool) and Unpaywall; they are read
+    from the environment so they can vary per deployment.
+    """
+
+    openalex: bool = True
+    unpaywall: bool = True
+    arxiv: bool = True
+    semantic_scholar: bool = True
+    core: bool = False  # requires a free API key
+    # Sanctioned publisher TDM access — OFF unless a librarian-issued token exists.
+    tdm: bool = False
+
+    mailto: str | None = None
+    unpaywall_email: str | None = None
+    core_api_key: str | None = None
+    semantic_scholar_key: str | None = None
+    # TDM tokens, keyed by publisher (e.g. {"elsevier": "..."}), loaded from env.
+    tdm_tokens: dict[str, str] = Field(default_factory=dict)
+
+
+class RateLimitConfig(BaseModel):
+    base_delay_seconds: float = 1.0
+    jitter_seconds: float = 0.5
+    backoff_base_seconds: float = 5.0
+    backoff_ceiling_seconds: float = 120.0
+    max_downloads_per_session: int = 50
+    max_consecutive_errors: int = 3
+
+
+class LoggingConfig(BaseModel):
+    log_dir: Path = Path("./logs")
+    level: str = "INFO"
+
+
+class StateConfig(BaseModel):
+    state_file: Path = Path("./state.json")
+
+
+class Config(BaseModel):
+    vault: VaultConfig = Field(default_factory=VaultConfig)
+    sources: SourcesConfig = Field(default_factory=SourcesConfig)
+    rate_limit: RateLimitConfig = Field(default_factory=RateLimitConfig)
+    logging: LoggingConfig = Field(default_factory=LoggingConfig)
+    state: StateConfig = Field(default_factory=StateConfig)
+    user_agent: str = (
+        "ClawLibrary/1.0 (open-access research vault; mailto:[email protected])"
+    )
+
+
+def _read_yaml(path: Path) -> dict:
+    if not path.exists():
+        raise FileNotFoundError(f"config file not found: {path}")
+    with open(path) as f:
+        data = yaml.safe_load(f) or {}
+    if not isinstance(data, dict):
+        raise ValueError(f"{path} must contain a YAML mapping at the top level")
+    return data
+
+
+def load_config(path: str | Path = "config.yaml", *, load_env: bool = True) -> Config:
+    """Load config.yaml and overlay secrets/identity from the environment."""
+    if load_env:
+        load_dotenv()
+    data = _read_yaml(Path(path))
+    config = Config.model_validate(data)
+
+    # Environment overlays for identity + keys (never stored in YAML).
+    config.sources.mailto = os.getenv("OPENALEX_MAILTO") or config.sources.mailto
+    config.sources.unpaywall_email = (
+        os.getenv("UNPAYWALL_EMAIL") or config.sources.unpaywall_email or config.sources.mailto
+    )
+    config.sources.core_api_key = os.getenv("CORE_API_KEY") or config.sources.core_api_key
+    config.sources.semantic_scholar_key = (
+        os.getenv("SEMANTIC_SCHOLAR_KEY") or config.sources.semantic_scholar_key
+    )
+    # TDM tokens: collect any ENV var of the form *_TDM_TOKEN.
+    tdm_tokens = dict(config.sources.tdm_tokens)
+    for key, val in os.environ.items():
+        if key.endswith("_TDM_TOKEN") and val:
+            publisher = key[: -len("_TDM_TOKEN")].lower()
+            tdm_tokens[publisher] = val
+    config.sources.tdm_tokens = tdm_tokens
+    # TDM is only truly enabled if both flagged on AND a token is present.
+    config.sources.tdm = config.sources.tdm and bool(tdm_tokens)
+    return config
+
+
+def load_topics(path: str | Path = "topics.yaml") -> TopicQueue:
+    """Load and validate topics.yaml, applying `defaults` to each topic."""
+    data = _read_yaml(Path(path))
+    defaults = data.get("defaults", {}) or {}
+    raw_topics = data.get("topics", []) or []
+    if not raw_topics:
+        raise ValueError("topics.yaml must define at least one topic under 'topics'")
+
+    merged: list[dict] = []
+    for raw in raw_topics:
+        if not isinstance(raw, dict):
+            raise ValueError(f"each topic must be a mapping, got: {raw!r}")
+        topic = {**{k: v for k, v in defaults.items() if k != "filters"}, **raw}
+        # Merge nested filters (topic-level overrides default-level keys).
+        default_filters = defaults.get("filters", {}) or {}
+        topic_filters = raw.get("filters", {}) or {}
+        topic["filters"] = {**default_filters, **topic_filters}
+        merged.append(topic)
+
+    return TopicQueue(defaults=defaults, topics=[Topic.model_validate(t) for t in merged])

clawlibrary/dedup.py

@@ -0,0 +1,69 @@
+"""Persistent, atomic deduplication store (state.json).
+
+Dedup identity is the DOI when present, else a SHA-256 hash of the resolved PDF
+URL. The store is written atomically (temp file + os.replace) so an interrupted
+run never corrupts it, and re-runs skip everything already downloaded.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+from pathlib import Path
+
+from .models import PaperCandidate
+
+
+class StateStore:
+    def __init__(self, path: str | Path):
+        self.path = Path(path)
+        self.seen_dois: dict[str, str] = {}
+        self.seen_url_hashes: dict[str, str] = {}
+        self.total_downloaded: int = 0
+        self.last_updated: str | None = None
+
+    @classmethod
+    def load(cls, path: str | Path) -> StateStore:
+        store = cls(path)
+        if store.path.exists():
+            try:
+                data = json.loads(store.path.read_text())
+            except (json.JSONDecodeError, OSError):
+                data = {}
+            store.seen_dois = data.get("seen_dois", {}) or {}
+            store.seen_url_hashes = data.get("seen_url_hashes", {}) or {}
+            store.total_downloaded = data.get("total_downloaded", 0) or 0
+            store.last_updated = data.get("last_updated")
+        return store
+
+    def is_seen(self, candidate: PaperCandidate) -> bool:
+        if candidate.doi and candidate.doi.lower() in self.seen_dois:
+            return True
+        h = candidate.url_hash
+        return bool(h and h in self.seen_url_hashes)
+
+    def mark_seen(self, candidate: PaperCandidate, run_id: str) -> None:
+        if candidate.doi:
+            self.seen_dois[candidate.doi.lower()] = run_id
+        h = candidate.url_hash
+        if h:
+            self.seen_url_hashes[h] = run_id
+        self.total_downloaded += 1
+
+    def save(self, *, now: str) -> None:
+        self.last_updated = now
+        data = {
+            "version": 1,
+            "seen_dois": self.seen_dois,
+            "seen_url_hashes": self.seen_url_hashes,
+            "total_downloaded": self.total_downloaded,
+            "last_updated": self.last_updated,
+        }
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        with tempfile.NamedTemporaryFile(
+            "w", dir=self.path.parent, delete=False, suffix=".tmp"
+        ) as tf:
+            json.dump(data, tf, indent=2)
+            tmp = tf.name
+        os.replace(tmp, self.path)

clawlibrary/downloader.py

@@ -0,0 +1,108 @@
+"""Streaming PDF downloader with content validation and atomic writes.
+
+Streams the response in 8KB chunks to a `.tmp` file and only renames into place
+on success. Validates that the payload is actually a PDF (by content-type OR the
+``%PDF`` magic bytes, since OA servers often mislabel content-type). Retries
+429/503 with exponential backoff via the shared RateLimiter.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+
+import httpx
+
+from .ratelimit import RateLimiter
+
+CHUNK = 8192
+_PDF_MAGIC = b"%PDF"
+
+
+@dataclass
+class DownloadResult:
+    status: str  # "ok" | "not_pdf" | "incomplete" | "http_error" | "server_error"
+    dest: Path | None = None
+    size_bytes: int = 0
+    detail: str | None = None
+
+    @property
+    def ok(self) -> bool:
+        return self.status == "ok"
+
+    @property
+    def is_server_error(self) -> bool:
+        return self.status == "server_error"
+
+
+def _is_pdf(content_type: str, first_chunk: bytes) -> bool:
+    if "application/pdf" in content_type.lower():
+        return True
+    return first_chunk[:4] == _PDF_MAGIC
+
+
+def download(
+    url: str,
+    dest: Path,
+    client: httpx.Client,
+    rate_limiter: RateLimiter,
+    *,
+    max_retries: int = 3,
+) -> DownloadResult:
+    """Download `url` to `dest`, returning a DownloadResult.
+
+    Rate-limit spacing is applied before each attempt. 429/503 trigger backoff
+    and retry; other 5xx are reported as server errors (for run suspension).
+    """
+    dest.parent.mkdir(parents=True, exist_ok=True)
+    tmp = dest.with_suffix(dest.suffix + ".tmp")
+
+    for attempt in range(max_retries + 1):
+        rate_limiter.wait()
+        try:
+            with client.stream("GET", url, follow_redirects=True) as r:
+                if r.status_code in (429, 503):
+                    if attempt < max_retries:
+                        rate_limiter.backoff(attempt)
+                        continue
+                    return DownloadResult("server_error", detail=f"HTTP {r.status_code}")
+                if r.status_code >= 500:
+                    return DownloadResult("server_error", detail=f"HTTP {r.status_code}")
+                if r.status_code >= 400:
+                    return DownloadResult("http_error", detail=f"HTTP {r.status_code}")
+
+                content_type = r.headers.get("content-type", "")
+                size = 0
+                first = True
+                try:
+                    with open(tmp, "wb") as f:
+                        for chunk in r.iter_bytes(CHUNK):
+                            if first:
+                                if not _is_pdf(content_type, chunk):
+                                    f.close()
+                                    tmp.unlink(missing_ok=True)
+                                    return DownloadResult(
+                                        "not_pdf", detail=f"content-type={content_type!r}"
+                                    )
+                                first = False
+                            f.write(chunk)
+                            size += len(chunk)
+                except httpx.HTTPError as e:
+                    tmp.unlink(missing_ok=True)
+                    return DownloadResult("incomplete", detail=str(e))
+
+                if size == 0:
+                    tmp.unlink(missing_ok=True)
+                    return DownloadResult("incomplete", detail="empty response")
+
+                tmp.replace(dest)
+                return DownloadResult("ok", dest=dest, size_bytes=size)
+
+        except httpx.HTTPError as e:
+            tmp.unlink(missing_ok=True)
+            if attempt < max_retries:
+                rate_limiter.backoff(attempt)
+                continue
+            return DownloadResult("incomplete", detail=str(e))
+
+    return DownloadResult("http_error", detail="retries exhausted")

clawlibrary/models.py

@@ -0,0 +1,103 @@
+"""Core data models: PaperCandidate (pre-download) and VaultEntry (sidecar)."""
+
+from __future__ import annotations
+
+import hashlib
+
+from pydantic import BaseModel, Field
+
+
+def url_hash(url: str) -> str:
+    """SHA-256 of a resolved URL, used as the dedup fallback key."""
+    return hashlib.sha256(url.encode("utf-8")).hexdigest()
+
+
+class PaperCandidate(BaseModel):
+    """A discovered paper, before any download is attempted."""
+
+    title: str
+    authors: list[str] = Field(default_factory=list)
+    doi: str | None = None
+    journal: str | None = None
+    year: int | None = None
+    abstract: str | None = None
+    is_open_access: bool = False
+    # Record-level URL from the discovery source (OpenAlex work id, landing page).
+    record_url: str | None = None
+    # Final, directly downloadable PDF URL (None until a resolver finds one).
+    resolved_pdf_url: str | None = None
+    # Which source supplied the PDF URL (e.g. "openalex", "unpaywall", "arxiv").
+    pdf_source: str | None = None
+    # Originating topic query.
+    topic: str = ""
+
+    @property
+    def dedup_key(self) -> str:
+        """Canonical dedup identity: DOI when present, else hash of the PDF URL."""
+        if self.doi:
+            return self.doi.lower()
+        if self.resolved_pdf_url:
+            return f"urlhash:{url_hash(self.resolved_pdf_url)}"
+        if self.record_url:
+            return f"urlhash:{url_hash(self.record_url)}"
+        return f"title:{self.title.lower()}"
+
+    @property
+    def url_hash(self) -> str | None:
+        if self.resolved_pdf_url:
+            return url_hash(self.resolved_pdf_url)
+        return None
+
+
+class VaultEntry(BaseModel):
+    """A downloaded paper's metadata sidecar (written next to the PDF)."""
+
+    id: str
+    doi: str | None = None
+    doi_slug: str
+    title: str
+    authors: list[str] = Field(default_factory=list)
+    journal: str | None = None
+    year: int | None = None
+    abstract: str | None = None
+    topic: str = ""
+    record_url: str | None = None
+    resolved_pdf_url: str
+    url_hash: str
+    source: str | None = None
+    vault_path: str
+    file_size_bytes: int
+    downloaded_at: str
+    run_id: str
+
+    @classmethod
+    def from_candidate(
+        cls,
+        candidate: PaperCandidate,
+        *,
+        doi_slug: str,
+        vault_path: str,
+        file_size_bytes: int,
+        downloaded_at: str,
+        run_id: str,
+    ) -> VaultEntry:
+        identity = candidate.doi or candidate.url_hash or doi_slug
+        return cls(
+            id=identity,
+            doi=candidate.doi,
+            doi_slug=doi_slug,
+            title=candidate.title,
+            authors=candidate.authors,
+            journal=candidate.journal,
+            year=candidate.year,
+            abstract=candidate.abstract,
+            topic=candidate.topic,
+            record_url=candidate.record_url,
+            resolved_pdf_url=candidate.resolved_pdf_url or "",
+            url_hash=candidate.url_hash or "",
+            source=candidate.pdf_source,
+            vault_path=vault_path,
+            file_size_bytes=file_size_bytes,
+            downloaded_at=downloaded_at,
+            run_id=run_id,
+        )

clawlibrary/pipeline.py

@@ -0,0 +1,201 @@
+"""End-to-end run orchestration: search → resolve → dedup → download → vault."""
+
+from __future__ import annotations
+
+import secrets
+from dataclasses import dataclass, field
+from datetime import UTC, datetime
+from pathlib import Path
+
+import httpx
+from loguru import logger
+
+from .config import Config, Topic
+from .dedup import StateStore
+from .downloader import download
+from .models import VaultEntry
+from .ratelimit import ConsecutiveErrorTracker, RateLimiter
+from .sources import openalex
+from .sources.resolver import resolve_pdf_candidates
+from .vault import VaultManager, doi_slug
+
+
+def generate_run_id() -> str:
+    """ISO-ish timestamp + 6-char hex suffix, e.g. 20260603T142200-4a7f3b."""
+    ts = datetime.now(UTC).strftime("%Y%m%dT%H%M%S")
+    return f"{ts}-{secrets.token_hex(3)}"
+
+
+def _utcnow_iso() -> str:
+    return datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+@dataclass
+class RunStats:
+    run_id: str
+    topics_processed: int = 0
+    downloaded: int = 0
+    skipped_dedup: int = 0
+    skipped_no_pdf: int = 0
+    failed: int = 0
+    would_download: int = 0  # dry-run only
+    suspended: bool = False
+    per_topic: dict[str, int] = field(default_factory=dict)
+
+
+def _setup_logging(config: Config, run_id: str):
+    log_dir = Path(config.logging.log_dir)
+    log_dir.mkdir(parents=True, exist_ok=True)
+    sink_id = logger.add(
+        log_dir / f"{run_id}.jsonl",
+        level=config.logging.level,
+        serialize=True,
+        enqueue=False,
+    )
+    return sink_id
+
+
+def run(
+    config: Config,
+    topics: list[Topic],
+    *,
+    dry_run: bool = False,
+    max_override: int | None = None,
+    on_event=None,
+    transport: httpx.BaseTransport | None = None,
+) -> RunStats:
+    """Process the topic queue. `on_event(kind, **data)` is an optional UI hook.
+
+    `transport` is a test hook to inject an httpx.MockTransport.
+    """
+    run_id = generate_run_id()
+    stats = RunStats(run_id=run_id)
+    sink_id = _setup_logging(config, run_id)
+
+    def emit(kind: str, **data):
+        logger.bind(run_id=run_id, **data).info(kind)
+        if on_event:
+            on_event(kind, **data)
+
+    state = StateStore.load(config.state.state_file)
+    vault = VaultManager(config.vault.root)
+    limiter = RateLimiter(config.rate_limit)
+    errors = ConsecutiveErrorTracker(config.rate_limit.max_consecutive_errors)
+    max_downloads = max_override or config.rate_limit.max_downloads_per_session
+
+    headers = {"User-Agent": config.user_agent}
+    emit("run_start", dry_run=dry_run, max_downloads=max_downloads, topics=len(topics))
+
+    try:
+        with httpx.Client(headers=headers, timeout=30.0, transport=transport) as client:
+            for topic in topics:
+                if stats.suspended:
+                    break
+                stats.topics_processed += 1
+                stats.per_topic.setdefault(topic.query, 0)
+                emit("topic_start", topic=topic.query, max_results=topic.max_results)
+
+                for candidate in openalex.search(client, topic, config.sources.mailto):
+                    if stats.downloaded >= max_downloads and not dry_run:
+                        emit("session_cap_reached", cap=max_downloads)
+                        break
+
+                    # Dedup on metadata identity (DOI) before resolving.
+                    if state.is_seen(candidate):
+                        stats.skipped_dedup += 1
+                        emit("skip", reason="dedup", title=candidate.title, doi=candidate.doi)
+                        continue
+
+                    url_candidates = resolve_pdf_candidates(client, candidate, config.sources)
+                    if not url_candidates:
+                        stats.skipped_no_pdf += 1
+                        emit("skip", reason="no_oa_pdf", title=candidate.title, doi=candidate.doi)
+                        continue
+
+                    # Use the first URL for dedup/identity; we'll try each in turn.
+                    candidate.resolved_pdf_url = url_candidates[0][0]
+                    candidate.pdf_source = url_candidates[0][1]
+
+                    # Re-check dedup now that we have a URL hash.
+                    if state.is_seen(candidate):
+                        stats.skipped_dedup += 1
+                        emit("skip", reason="dedup", title=candidate.title, doi=candidate.doi)
+                        continue
+
+                    if dry_run:
+                        stats.would_download += 1
+                        emit(
+                            "would_download",
+                            title=candidate.title,
+                            doi=candidate.doi,
+                            source=candidate.pdf_source,
+                            url=candidate.resolved_pdf_url,
+                            alternatives=len(url_candidates),
+                        )
+                        continue
+
+                    dest = vault.pdf_path(candidate)
+                    result = None
+                    for pdf_url, source in url_candidates:
+                        result = download(pdf_url, dest, client, limiter)
+                        if result.ok or result.is_server_error:
+                            candidate.resolved_pdf_url = pdf_url
+                            candidate.pdf_source = source
+                            break
+
+                    if result is not None and result.ok:
+                        entry = VaultEntry.from_candidate(
+                            candidate,
+                            doi_slug=doi_slug(candidate),
+                            vault_path=str(dest),
+                            file_size_bytes=result.size_bytes,
+                            downloaded_at=_utcnow_iso(),
+                            run_id=run_id,
+                        )
+                        vault.write_sidecar(entry, dest)
+                        vault.append_to_index(entry)
+                        state.mark_seen(candidate, run_id)
+                        state.save(now=_utcnow_iso())
+                        errors.record_success()
+                        stats.downloaded += 1
+                        stats.per_topic[topic.query] += 1
+                        emit(
+                            "downloaded",
+                            title=candidate.title,
+                            doi=candidate.doi,
+                            source=source,
+                            size=result.size_bytes,
+                            path=str(dest),
+                        )
+                    else:
+                        stats.failed += 1
+                        emit(
+                            "failed",
+                            reason=result.status,
+                            detail=result.detail,
+                            title=candidate.title,
+                            url=pdf_url,
+                        )
+                        if result.is_server_error:
+                            errors.record_error()
+                            if errors.tripped:
+                                stats.suspended = True
+                                emit("run_suspended", reason="server_errors", count=errors.count)
+                                break
+    finally:
+        emit("run_end", **_summary_dict(stats))
+        logger.remove(sink_id)
+
+    return stats
+
+
+def _summary_dict(stats: RunStats) -> dict:
+    return {
+        "topics_processed": stats.topics_processed,
+        "downloaded": stats.downloaded,
+        "skipped_dedup": stats.skipped_dedup,
+        "skipped_no_pdf": stats.skipped_no_pdf,
+        "failed": stats.failed,
+        "would_download": stats.would_download,
+        "suspended": stats.suspended,
+    }

clawlibrary/ratelimit.py

@@ -0,0 +1,53 @@
+"""Polite rate limiting and server-distress backoff.
+
+This is ordinary good-citizen API behavior: a small spacing between calls plus
+exponential backoff when a server signals 429/503. It is NOT anti-detection
+timing — the sources used here welcome programmatic access; we simply avoid
+hammering them and respect their `Retry-After` signals.
+"""
+
+from __future__ import annotations
+
+import random
+import time
+
+from .config import RateLimitConfig
+
+
+class RateLimiter:
+    def __init__(self, config: RateLimitConfig, *, sleeper=time.sleep, rng=None):
+        self.config = config
+        self._sleep = sleeper
+        self._rng = rng or random.Random()
+
+    def wait(self) -> float:
+        """Sleep base_delay ± jitter (floored at 0) before the next request."""
+        jitter = self._rng.uniform(-self.config.jitter_seconds, self.config.jitter_seconds)
+        delay = max(0.0, self.config.base_delay_seconds + jitter)
+        self._sleep(delay)
+        return delay
+
+    def backoff(self, attempt: int) -> float:
+        """Exponential backoff for retry `attempt` (0-based), capped at ceiling."""
+        delay = self.config.backoff_base_seconds * (2 ** attempt)
+        delay = min(delay, self.config.backoff_ceiling_seconds)
+        self._sleep(delay)
+        return delay
+
+
+class ConsecutiveErrorTracker:
+    """Suspends a run after N consecutive server errors (5xx)."""
+
+    def __init__(self, limit: int):
+        self.limit = limit
+        self.count = 0
+
+    def record_error(self) -> None:
+        self.count += 1
+
+    def record_success(self) -> None:
+        self.count = 0
+
+    @property
+    def tripped(self) -> bool:
+        return self.count >= self.limit

clawlibrary/sources/__init__.py

@@ -0,0 +1,4 @@
+"""Acquisition sources: legitimate, programmatic-access APIs for metadata and
+open-access full text. These replace the PRD's authenticated discovery-layer
+scraping with channels designed for research automation.
+"""

clawlibrary/sources/arxiv.py

@@ -0,0 +1,63 @@
+"""arXiv — open-access preprint full text.
+
+Two paths: (1) a DOI minted by arXiv (10.48550/arXiv.XXXX) maps directly to a
+PDF; (2) otherwise query the arXiv API by title and take the top match. arXiv
+PDFs are openly licensed for download.
+
+API docs: https://info.arxiv.org/help/api/
+"""
+
+from __future__ import annotations
+
+import re
+from xml.etree import ElementTree as ET
+
+import httpx
+
+API_URL = "http://export.arxiv.org/api/query"
+_ATOM = "{http://www.w3.org/2005/Atom}"
+
+_ARXIV_DOI_RE = re.compile(r"arxiv\.(?P<id>\d{4}\.\d{4,5}(v\d+)?)", re.IGNORECASE)
+
+
+def _pdf_from_id(arxiv_id: str) -> str:
+    return f"https://arxiv.org/pdf/{arxiv_id}.pdf"
+
+
+def resolve(client: httpx.Client, *, doi: str | None, title: str | None) -> str | None:
+    """Return an arXiv PDF URL for a paper, or None."""
+    if doi:
+        m = _ARXIV_DOI_RE.search(doi)
+        if m:
+            return _pdf_from_id(m.group("id"))
+
+    if not title:
+        return None
+
+    # Title search; arXiv API returns Atom XML.
+    try:
+        resp = client.get(
+            API_URL,
+            params={"search_query": f'ti:"{title}"', "max_results": 1},
+        )
+        if resp.status_code != 200:
+            return None
+        root = ET.fromstring(resp.text)
+    except (httpx.HTTPError, ET.ParseError):
+        return None
+
+    entry = root.find(f"{_ATOM}entry")
+    if entry is None:
+        return None
+    for link in entry.findall(f"{_ATOM}link"):
+        if link.get("title") == "pdf" or link.get("type") == "application/pdf":
+            href = link.get("href")
+            if href:
+                return href if href.endswith(".pdf") else href + ".pdf"
+    # Fall back to the entry id (e.g. http://arxiv.org/abs/XXXX) -> pdf.
+    id_el = entry.find(f"{_ATOM}id")
+    if id_el is not None and id_el.text:
+        m = re.search(r"abs/(?P<id>[^/\s]+)", id_el.text)
+        if m:
+            return _pdf_from_id(m.group("id"))
+    return None

clawlibrary/sources/base.py

@@ -0,0 +1,54 @@
+"""Shared helpers for acquisition sources."""
+
+from __future__ import annotations
+
+import re
+
+import httpx
+
+
+def get_json(client: httpx.Client, url: str, params: dict | None = None) -> dict | None:
+    """GET a JSON document, returning None on any HTTP/transport/parse failure.
+
+    Sources treat absence as "no result" rather than a hard error, so the
+    pipeline can fall through to the next resolver.
+    """
+    try:
+        resp = client.get(url, params=params)
+        if resp.status_code != 200:
+            return None
+        return resp.json()
+    except (httpx.HTTPError, ValueError):
+        return None
+
+
+def normalize_doi(doi: str | None) -> str | None:
+    """Strip URL/prefix decoration from a DOI; return lowercase bare DOI."""
+    if not doi:
+        return None
+    doi = doi.strip()
+    doi = re.sub(r"^https?://(dx\.)?doi\.org/", "", doi, flags=re.IGNORECASE)
+    doi = re.sub(r"^doi:", "", doi, flags=re.IGNORECASE)
+    return doi.lower() or None
+
+
+def reconstruct_abstract(inverted_index: dict | None) -> str | None:
+    """Rebuild plain-text abstract from OpenAlex's inverted-index representation."""
+    if not inverted_index:
+        return None
+    positions: list[tuple[int, str]] = []
+    for word, idxs in inverted_index.items():
+        for i in idxs:
+            positions.append((i, word))
+    if not positions:
+        return None
+    positions.sort(key=lambda p: p[0])
+    return " ".join(word for _, word in positions)
+
+
+def looks_like_pdf_url(url: str | None) -> bool:
+    """Heuristic: does this URL plausibly point at a PDF?"""
+    if not url:
+        return False
+    low = url.lower()
+    return low.endswith(".pdf") or "/pdf" in low or "format=pdf" in low

clawlibrary/sources/core.py

@@ -0,0 +1,42 @@
+"""CORE — aggregated open-access full text (fallback).
+
+CORE harvests OA repositories worldwide. Requires a free API key. Used only as
+a last-resort resolver when other OA sources miss.
+
+API docs: https://api.core.ac.uk/docs/v3
+"""
+
+from __future__ import annotations
+
+import httpx
+
+API_URL = "https://api.core.ac.uk/v3/search/works"
+
+
+def resolve(
+    client: httpx.Client, *, doi: str | None, title: str | None, api_key: str | None
+) -> str | None:
+    if not api_key:
+        return None
+    if doi:
+        query = f'doi:"{doi}"'
+    elif title:
+        query = f'title:"{title}"'
+    else:
+        return None
+    try:
+        resp = client.get(
+            API_URL,
+            params={"q": query, "limit": 1},
+            headers={"Authorization": f"Bearer {api_key}"},
+        )
+        if resp.status_code != 200:
+            return None
+        data = resp.json()
+    except (httpx.HTTPError, ValueError):
+        return None
+    for work in data.get("results") or []:
+        url = work.get("downloadUrl")
+        if url:
+            return url
+    return None

clawlibrary/sources/openalex.py

@@ -0,0 +1,107 @@
+"""OpenAlex — primary topic search and metadata source.
+
+OpenAlex is a free, open catalog of scholarly works with a documented REST API
+and a "polite pool" (faster, more reliable) accessed by passing a contact email.
+We use it for discovery + metadata, and harvest its `best_oa_location` PDF link
+when one is available.
+
+API docs: https://docs.openalex.org/
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterator
+
+import httpx
+
+from ..config import Topic
+from ..models import PaperCandidate
+from .base import get_json, normalize_doi, reconstruct_abstract
+
+API_URL = "https://api.openalex.org/works"
+
+
+def _build_filters(topic: Topic) -> str:
+    parts = ["type:article"]
+    f = topic.filters
+    if f.open_access_only:
+        parts.append("is_oa:true")
+    if f.date_from:
+        parts.append(f"from_publication_date:{f.date_from}-01-01")
+    if f.date_to:
+        parts.append(f"to_publication_date:{f.date_to}-12-31")
+    if f.language:
+        parts.append(f"language:{f.language}")
+    return ",".join(parts)
+
+
+def _pdf_url_from_work(work: dict) -> tuple[str | None, bool]:
+    """Return (best OA pdf url, is_oa) from an OpenAlex work record."""
+    oa = work.get("open_access") or {}
+    is_oa = bool(oa.get("is_oa"))
+    for loc_key in ("best_oa_location", "primary_location"):
+        loc = work.get(loc_key) or {}
+        pdf = loc.get("pdf_url")
+        if pdf:
+            return pdf, is_oa
+    # Fall back to the generic OA URL (may be a landing page, resolved later).
+    return oa.get("oa_url"), is_oa
+
+
+def work_to_candidate(work: dict, topic_query: str) -> PaperCandidate:
+    authors = [
+        a.get("author", {}).get("display_name")
+        for a in (work.get("authorships") or [])
+        if a.get("author", {}).get("display_name")
+    ]
+    journal = None
+    primary = work.get("primary_location") or {}
+    src = primary.get("source") or {}
+    if src:
+        journal = src.get("display_name")
+
+    pdf_url, is_oa = _pdf_url_from_work(work)
+
+    return PaperCandidate(
+        title=work.get("title") or work.get("display_name") or "(untitled)",
+        authors=authors,
+        doi=normalize_doi(work.get("doi")),
+        journal=journal,
+        year=work.get("publication_year"),
+        abstract=reconstruct_abstract(work.get("abstract_inverted_index")),
+        is_open_access=is_oa,
+        record_url=work.get("id"),
+        resolved_pdf_url=pdf_url,
+        pdf_source="openalex" if pdf_url else None,
+        topic=topic_query,
+    )
+
+
+def search(client: httpx.Client, topic: Topic, mailto: str | None) -> Iterator[PaperCandidate]:
+    """Yield up to topic.max_results candidates for a topic via cursor pagination."""
+    per_page = min(200, max(1, topic.max_results))
+    cursor = "*"
+    yielded = 0
+
+    while yielded < topic.max_results and cursor:
+        params = {
+            "search": topic.query,
+            "filter": _build_filters(topic),
+            "per-page": per_page,
+            "cursor": cursor,
+        }
+        if mailto:
+            params["mailto"] = mailto
+
+        data = get_json(client, API_URL, params=params)
+        if not data:
+            return
+        results = data.get("results") or []
+        if not results:
+            return
+        for work in results:
+            yield work_to_candidate(work, topic.query)
+            yielded += 1
+            if yielded >= topic.max_results:
+                return
+        cursor = (data.get("meta") or {}).get("next_cursor")

clawlibrary/sources/resolver.py

@@ -0,0 +1,68 @@
+"""Resolve a PaperCandidate to one or more downloadable PDF URLs across OA sources.
+
+Many OA links are flaky: a publisher-hosted "OA" link may 403 a bare GET while
+Unpaywall or arXiv hold a working copy of the same paper. So we collect an
+ordered list of candidate (url, source) pairs and let the downloader try them in
+turn. Order reflects trust/directness: OpenAlex's own link, then Unpaywall,
+arXiv, Semantic Scholar, CORE, and finally sanctioned TDM (only if configured).
+"""
+
+from __future__ import annotations
+
+import httpx
+
+from ..config import SourcesConfig
+from ..models import PaperCandidate
+from . import arxiv, core, semanticscholar, tdm, unpaywall
+
+
+def resolve_pdf_candidates(
+    client: httpx.Client, candidate: PaperCandidate, sources: SourcesConfig
+) -> list[tuple[str, str]]:
+    """Return an ordered, de-duplicated list of (pdf_url, source_name)."""
+    out: list[tuple[str, str]] = []
+    seen: set[str] = set()
+
+    def add(url: str | None, source: str) -> None:
+        if url and url not in seen:
+            seen.add(url)
+            out.append((url, source))
+
+    doi = candidate.doi
+
+    # 1. OpenAlex's own OA link (already on the candidate).
+    if candidate.resolved_pdf_url and candidate.pdf_source == "openalex":
+        add(candidate.resolved_pdf_url, "openalex")
+
+    # 2. Unpaywall.
+    if sources.unpaywall and doi:
+        add(unpaywall.resolve(client, doi, sources.unpaywall_email), "unpaywall")
+
+    # 3. arXiv.
+    if sources.arxiv:
+        add(arxiv.resolve(client, doi=doi, title=candidate.title), "arxiv")
+
+    # 4. Semantic Scholar.
+    if sources.semantic_scholar and doi:
+        add(semanticscholar.resolve(client, doi, sources.semantic_scholar_key), "semanticscholar")
+
+    # 5. CORE.
+    if sources.core:
+        core_url = core.resolve(
+            client, doi=doi, title=candidate.title, api_key=sources.core_api_key
+        )
+        add(core_url, "core")
+
+    # 6. Sanctioned publisher TDM — only when explicitly enabled with a token.
+    if sources.tdm and sources.tdm_tokens:
+        add(tdm.resolve(client, doi=doi, tokens=sources.tdm_tokens), "tdm")
+
+    return out
+
+
+def resolve_pdf_url(
+    client: httpx.Client, candidate: PaperCandidate, sources: SourcesConfig
+) -> tuple[str | None, str | None]:
+    """Convenience: first resolved (url, source), or (None, None)."""
+    candidates = resolve_pdf_candidates(client, candidate, sources)
+    return candidates[0] if candidates else (None, None)

clawlibrary/sources/semanticscholar.py

@@ -0,0 +1,36 @@
+"""Semantic Scholar — open-access PDF fallback via the Academic Graph API.
+
+Returns only the `openAccessPdf` link the API exposes (lawfully available
+copies). An API key is optional but raises rate limits.
+
+API docs: https://api.semanticscholar.org/api-docs/graph
+"""
+
+from __future__ import annotations
+
+import httpx
+
+from .base import get_json
+
+API_URL = "https://api.semanticscholar.org/graph/v1/paper"
+
+
+def resolve(client: httpx.Client, doi: str, api_key: str | None) -> str | None:
+    if not doi:
+        return None
+    headers = {"x-api-key": api_key} if api_key else None
+    try:
+        # get_json doesn't take headers; do a direct call when a key is present.
+        if headers:
+            resp = client.get(
+                f"{API_URL}/DOI:{doi}", params={"fields": "openAccessPdf"}, headers=headers
+            )
+            data = resp.json() if resp.status_code == 200 else None
+        else:
+            data = get_json(client, f"{API_URL}/DOI:{doi}", params={"fields": "openAccessPdf"})
+    except (httpx.HTTPError, ValueError):
+        return None
+    if not data:
+        return None
+    oa = data.get("openAccessPdf") or {}
+    return oa.get("url") or None

clawlibrary/sources/tdm.py

@@ -0,0 +1,54 @@
+"""Sanctioned publisher TDM (Text & Data Mining) access — config-gated.
+
+This is the *legitimate* channel for full text of subscription content: many
+publishers (Elsevier, Wiley, Springer Nature, ...) expose a TDM/API endpoint
+that an institution authorizes for a named researcher via an issued token.
+
+How to enable (faculty path):
+  1. Email your subject librarian and request TDM/API access for a research
+     corpus, naming the publishers you need.
+  2. They issue an institutional token (or register your IP range).
+  3. Put the token in .env as e.g. ELSEVIER_TDM_TOKEN=...
+  4. Set sources.tdm: true in config.yaml.
+
+Until a token is present this module is inert and returns None for everything,
+so the pipeline simply skips non-OA papers (logged SKIPPED:no_oa_pdf).
+
+NOTE: This intentionally does NOT scrape authenticated discovery layers or
+bypass any access control. It only calls publisher-sanctioned TDM endpoints
+with credentials the institution issued for exactly this purpose.
+"""
+
+from __future__ import annotations
+
+import httpx
+
+# Minimal registry of known publisher TDM endpoints. Resolving a DOI to its
+# publisher and constructing the correct request is publisher-specific; these
+# templates are wired conservatively and only used when a matching token exists.
+_ENDPOINTS = {
+    # Elsevier ScienceDirect article retrieval (TDM): full text by DOI.
+    "elsevier": "https://api.elsevier.com/content/article/doi/{doi}",
+}
+
+
+def resolve(client: httpx.Client, *, doi: str | None, tokens: dict[str, str]) -> str | None:
+    """Return a downloadable PDF URL via a sanctioned TDM endpoint, or None.
+
+    Conservative by design: only acts when both a DOI and a matching publisher
+    token are available. Publisher-to-DOI mapping and request specifics should
+    be extended per the agreement issued by the library.
+    """
+    if not doi or not tokens:
+        return None
+    # Without a DOI->publisher map we cannot safely choose an endpoint; callers
+    # extend this once the library confirms which publishers/tokens apply.
+    for publisher, token in tokens.items():
+        endpoint = _ENDPOINTS.get(publisher)
+        if not endpoint or not token:
+            continue
+        # Return the TDM URL with token as a query param; the downloader sends it
+        # with the configured user-agent. Publisher-specific auth headers can be
+        # added here per the issued agreement.
+        return endpoint.format(doi=doi) + f"?apiKey={token}&httpAccept=application/pdf"
+    return None

clawlibrary/sources/unpaywall.py

@@ -0,0 +1,33 @@
+"""Unpaywall — given a DOI, find a legal open-access PDF URL.
+
+Unpaywall indexes only lawfully posted OA copies (publisher OA, PMC, author
+manuscripts in repositories). It never circumvents access controls.
+
+API docs: https://unpaywall.org/products/api
+"""
+
+from __future__ import annotations
+
+import httpx
+
+from .base import get_json
+
+API_URL = "https://api.unpaywall.org/v2"
+
+
+def resolve(client: httpx.Client, doi: str, email: str | None) -> str | None:
+    """Return a legal OA PDF URL for a DOI, or None."""
+    if not doi or not email:
+        return None
+    data = get_json(client, f"{API_URL}/{doi}", params={"email": email})
+    if not data:
+        return None
+    best = data.get("best_oa_location") or {}
+    pdf = best.get("url_for_pdf") or best.get("url")
+    if pdf:
+        return pdf
+    for loc in data.get("oa_locations") or []:
+        url = loc.get("url_for_pdf") or loc.get("url")
+        if url:
+            return url
+    return None

clawlibrary/vault.py

@@ -0,0 +1,117 @@
+"""Vault manager: directory layout, sidecars, and the vault-level index.
+
+Layout:
+  {root}/{topic_slug}/{doi_slug}.pdf
+  {root}/{topic_slug}/{doi_slug}.json   (VaultEntry sidecar)
+  {root}/index.json                     (array of all sidecars)
+  {root}/index.csv                      (flat table for downstream ingestion)
+"""
+
+from __future__ import annotations
+
+import csv
+import json
+import re
+from pathlib import Path
+
+from .models import PaperCandidate, VaultEntry
+
+_CSV_FIELDS = [
+    "id",
+    "doi",
+    "title",
+    "authors",
+    "journal",
+    "year",
+    "topic",
+    "source",
+    "resolved_pdf_url",
+    "vault_path",
+    "file_size_bytes",
+    "downloaded_at",
+    "run_id",
+]
+
+
+def slugify(text: str, *, maxlen: int = 80) -> str:
+    text = text.strip().lower()
+    text = re.sub(r"[^a-z0-9]+", "-", text)
+    text = re.sub(r"-+", "-", text).strip("-")
+    return text[:maxlen] or "untitled"
+
+
+def topic_slug(topic_query: str) -> str:
+    return slugify(topic_query)
+
+
+def doi_slug(candidate: PaperCandidate) -> str:
+    """Filesystem-safe identifier: from DOI when present, else URL hash, else title."""
+    if candidate.doi:
+        return slugify(candidate.doi)
+    if candidate.url_hash:
+        return f"urlhash-{candidate.url_hash[:16]}"
+    return slugify(candidate.title)
+
+
+class VaultManager:
+    def __init__(self, root: str | Path):
+        self.root = Path(root)
+
+    def pdf_path(self, candidate: PaperCandidate) -> Path:
+        return self.root / topic_slug(candidate.topic) / f"{doi_slug(candidate)}.pdf"
+
+    def sidecar_path(self, pdf_path: Path) -> Path:
+        return pdf_path.with_suffix(".json")
+
+    def write_sidecar(self, entry: VaultEntry, pdf_path: Path) -> Path:
+        sidecar = self.sidecar_path(pdf_path)
+        sidecar.parent.mkdir(parents=True, exist_ok=True)
+        sidecar.write_text(json.dumps(entry.model_dump(), indent=2))
+        return sidecar
+
+    # ---- index ---------------------------------------------------------------
+
+    def _index_json_path(self) -> Path:
+        return self.root / "index.json"
+
+    def _index_csv_path(self) -> Path:
+        return self.root / "index.csv"
+
+    def load_index(self) -> list[dict]:
+        p = self._index_json_path()
+        if not p.exists():
+            return []
+        try:
+            return json.loads(p.read_text())
+        except (json.JSONDecodeError, OSError):
+            return []
+
+    def append_to_index(self, entry: VaultEntry) -> None:
+        entries = self.load_index()
+        # Replace any existing entry with the same id (idempotent re-index).
+        entries = [e for e in entries if e.get("id") != entry.id]
+        entries.append(entry.model_dump())
+        self._write_index(entries)
+
+    def _write_index(self, entries: list[dict]) -> None:
+        self.root.mkdir(parents=True, exist_ok=True)
+        self._index_json_path().write_text(json.dumps(entries, indent=2))
+        with open(self._index_csv_path(), "w", newline="") as f:
+            writer = csv.DictWriter(f, fieldnames=_CSV_FIELDS, extrasaction="ignore")
+            writer.writeheader()
+            for e in entries:
+                row = dict(e)
+                if isinstance(row.get("authors"), list):
+                    row["authors"] = "; ".join(row["authors"])
+                writer.writerow(row)
+
+    def rebuild_index(self) -> list[dict]:
+        """Regenerate index.json/csv by scanning all sidecar files."""
+        entries: list[dict] = []
+        for sidecar in sorted(self.root.glob("*/*.json")):
+            try:
+                entries.append(json.loads(sidecar.read_text()))
+            except (json.JSONDecodeError, OSError):
+                continue
+        self._write_index(entries)
+        return entries

config.yaml

@@ -0,0 +1,30 @@
+# ClawLibrary system configuration.
+# Secrets and identity (emails, API keys, TDM tokens) live in .env, not here.
+
+vault:
+  root: ./vault
+
+state:
+  state_file: ./state.json
+
+sources:
+  openalex: true          # primary topic search + metadata
+  unpaywall: true         # DOI -> legal open-access PDF
+  arxiv: true             # preprint full text
+  semantic_scholar: true  # OA fallback
+  core: false             # requires CORE_API_KEY in .env
+  tdm: false              # sanctioned publisher TDM; needs a librarian-issued token
+
+rate_limit:
+  base_delay_seconds: 1.0       # polite spacing between API/download calls
+  jitter_seconds: 0.5
+  backoff_base_seconds: 5.0     # initial backoff on 429/503
+  backoff_ceiling_seconds: 120.0
+  max_downloads_per_session: 50
+  max_consecutive_errors: 3
+
+logging:
+  log_dir: ./logs
+  level: INFO
+
+user_agent: "ClawLibrary/1.0 (open-access research vault; mailto:[email protected])"

pyproject.toml

@@ -0,0 +1,45 @@
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[project]
+name = "clawlibrary"
+version = "1.0.0"
+description = "Open-access research vault builder: topic-driven harvesting of legally available full-text PDFs into a deduplicated, metadata-rich local library."
+readme = "README.md"
+requires-python = ">=3.11"
+authors = [{ name = "Omar Sobh", email = "[email protected]" }]
+license = { text = "MIT" }
+dependencies = [
+    "httpx>=0.27",
+    "pyyaml>=6.0",
+    "python-dotenv>=1.0",
+    "rich>=13.0",
+    "loguru>=0.7",
+    "tenacity>=8.3",
+    "pydantic>=2.6",
+]
+
+[project.optional-dependencies]
+dev = [
+    "pytest>=8.0",
+    "pytest-cov>=5.0",
+    "ruff>=0.5",
+]
+
+[project.scripts]
+clawlibrary = "clawlibrary.__main__:main"
+
+[tool.hatch.build.targets.wheel]
+packages = ["clawlibrary"]
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+addopts = "-q"
+
+[tool.ruff]
+line-length = 100
+target-version = "py311"
+
+[tool.ruff.lint]
+select = ["E", "F", "I", "W", "UP", "B"]

tests/conftest.py

@@ -0,0 +1,66 @@
+import json
+
+import pytest
+
+from clawlibrary.models import PaperCandidate
+
+# Keep tests hermetic: clear any ambient ClawLibrary env vars (e.g. a developer's
+# real .env in the repo root) so config/overlay tests are deterministic.
+_ENV_VARS = [
+    "OPENALEX_MAILTO",
+    "UNPAYWALL_EMAIL",
+    "CORE_API_KEY",
+    "SEMANTIC_SCHOLAR_KEY",
+]
+
+
+@pytest.fixture(autouse=True)
+def _clean_env(monkeypatch):
+    for var in _ENV_VARS:
+        monkeypatch.delenv(var, raising=False)
+    for key in list(__import__("os").environ):
+        if key.endswith("_TDM_TOKEN"):
+            monkeypatch.delenv(key, raising=False)
+
+
+@pytest.fixture
+def candidate():
+    return PaperCandidate(
+        title="Oxidizing Interfaces: Rust in the Linux Kernel",
+        authors=["Wedson Almeida Filho", "Alex Gaynor"],
+        doi="10.1145/3620666.3651327",
+        journal="SOSP",
+        year=2023,
+        topic="Rust Linux kernel",
+        resolved_pdf_url="https://example.org/paper.pdf",
+        pdf_source="openalex",
+    )
+
+
+@pytest.fixture
+def openalex_work():
+    return json.loads(OPENALEX_WORK_JSON)
+
+
+OPENALEX_WORK_JSON = r"""
+{
+  "id": "https://openalex.org/W123",
+  "doi": "https://doi.org/10.1145/3620666.3651327",
+  "title": "Oxidizing Interfaces: Rust in the Linux Kernel",
+  "display_name": "Oxidizing Interfaces: Rust in the Linux Kernel",
+  "publication_year": 2023,
+  "type": "article",
+  "open_access": {"is_oa": true, "oa_url": "https://example.org/landing"},
+  "authorships": [
+    {"author": {"display_name": "Wedson Almeida Filho"}},
+    {"author": {"display_name": "Alex Gaynor"}}
+  ],
+  "primary_location": {
+    "source": {"display_name": "SOSP"},
+    "pdf_url": null,
+    "landing_page_url": "https://dl.acm.org/doi/10.1145/3620666.3651327"
+  },
+  "best_oa_location": {"pdf_url": "https://example.org/best.pdf"},
+  "abstract_inverted_index": {"Rust": [0], "is": [1], "safe": [2]}
+}
+"""

tests/test_cli.py

@@ -0,0 +1,84 @@
+import json
+
+import pytest
+
+from clawlibrary import __main__ as cli
+from clawlibrary.pipeline import RunStats
+
+
+@pytest.fixture
+def project(tmp_path, monkeypatch):
+    (tmp_path / "config.yaml").write_text(
+        f"vault:\n  root: {tmp_path / 'vault'}\n"
+        f"state:\n  state_file: {tmp_path / 'state.json'}\n"
+        f"logging:\n  log_dir: {tmp_path / 'logs'}\n"
+    )
+    (tmp_path / "topics.yaml").write_text("topics:\n  - query: rust\n")
+    monkeypatch.chdir(tmp_path)
+    return tmp_path
+
+
+def _seed_vault(tmp_path):
+    from clawlibrary.models import VaultEntry
+    from clawlibrary.vault import VaultManager
+
+    vault = VaultManager(tmp_path / "vault")
+    entry = VaultEntry(
+        id="10.1/x", doi="10.1/x", doi_slug="10-1-x", title="A Paper",
+        authors=["Jane Doe"], journal="J", year=2024, topic="rust",
+        resolved_pdf_url="https://e/x.pdf", url_hash="abc", source="arxiv",
+        vault_path="vault/rust/10-1-x.pdf", file_size_bytes=2048,
+        downloaded_at="2026-06-03T00:00:00Z", run_id="r1",
+    )
+    vault.append_to_index(entry)
+
+
+def test_cli_run_dispatches_and_summarizes(project, monkeypatch):
+    captured = {}
+
+    def fake_run(config, topics, **kwargs):
+        # exercise the on_event UI callback branches
+        cb = kwargs["on_event"]
+        cb("topic_start", topic="rust", max_results=5)
+        cb("downloaded", title="A Paper", source="arxiv", doi="10.1/x", size=10, path="p")
+        cb("failed", title="Bad", reason="not_pdf")
+        captured["called"] = True
+        return RunStats(run_id="r1", topics_processed=1, downloaded=1)
+
+    monkeypatch.setattr(cli, "run", fake_run)
+    rc = cli.main(["run"])
+    assert rc == 0
+    assert captured["called"]
+
+
+def test_cli_run_unknown_topic(project, monkeypatch):
+    monkeypatch.setattr(cli, "run", lambda *a, **k: RunStats(run_id="r"))
+    rc = cli.main(["run", "--topic", "does-not-exist"])
+    assert rc == 1
+
+
+def test_cli_stats(project, capsys):
+    _seed_vault(project)
+    rc = cli.main(["stats"])
+    assert rc == 0
+    out = capsys.readouterr().out
+    assert "Total papers" in out
+
+
+def test_cli_stats_empty(project):
+    assert cli.main(["stats"]) == 0
+
+
+def test_cli_export_csv_and_json(project, tmp_path):
+    _seed_vault(project)
+    assert cli.main(["export", "--format", "json", "--out", str(project / "e.json")]) == 0
+    data = json.loads((project / "e.json").read_text())
+    assert data[0]["doi"] == "10.1/x"
+
+    assert cli.main(["export", "--format", "csv", "--out", str(project / "e.csv")]) == 0
+    assert (project / "e.csv").read_text().splitlines()[0].startswith("id,doi,title")
+
+
+def test_cli_missing_config_returns_2(tmp_path, monkeypatch):
+    monkeypatch.chdir(tmp_path)
+    assert cli.main(["stats"]) == 2

tests/test_config.py

@@ -0,0 +1,71 @@
+import pytest
+
+from clawlibrary.config import load_config, load_topics
+
+
+def _write(tmp_path, name, text):
+    p = tmp_path / name
+    p.write_text(text)
+    return p
+
+
+def test_load_topics_applies_defaults_and_priority(tmp_path):
+    path = _write(
+        tmp_path,
+        "topics.yaml",
+        """
+defaults:
+  max_results: 40
+  filters:
+    open_access_only: true
+topics:
+  - query: "alpha"
+    priority: 2
+  - query: "beta"
+    priority: 1
+    max_results: 10
+    filters:
+      date_from: 2020
+  - query: "gamma"
+    enabled: false
+""",
+    )
+    queue = load_topics(path)
+    active = queue.active()
+    # gamma disabled; beta(prio1) before alpha(prio2)
+    assert [t.query for t in active] == ["beta", "alpha"]
+    assert active[0].max_results == 10
+    assert active[0].filters.date_from == 2020
+    # default applied to alpha
+    assert active[1].max_results == 40
+    assert active[1].filters.open_access_only is True
+
+
+def test_load_topics_rejects_empty(tmp_path):
+    path = _write(tmp_path, "topics.yaml", "topics: []\n")
+    with pytest.raises(ValueError):
+        load_topics(path)
+
+
+def test_load_topics_rejects_blank_query(tmp_path):
+    path = _write(tmp_path, "topics.yaml", "topics:\n  - query: '   '\n")
+    with pytest.raises(ValueError):
+        load_topics(path)
+
+
+def test_load_config_env_overlay(tmp_path, monkeypatch):
+    path = _write(tmp_path, "config.yaml", "vault:\n  root: ./myvault\n")
+    monkeypatch.setenv("OPENALEX_MAILTO", "[email protected]")
+    monkeypatch.setenv("ELSEVIER_TDM_TOKEN", "secret")
+    cfg = load_config(path, load_env=False)
+    assert str(cfg.vault.root) == "myvault"
+    assert cfg.sources.mailto == "[email protected]"
+    assert cfg.sources.unpaywall_email == "[email protected]"  # falls back to mailto
+    assert cfg.sources.tdm_tokens == {"elsevier": "secret"}
+    # tdm flag off in yaml -> stays disabled even with a token present
+    assert cfg.sources.tdm is False
+
+
+def test_load_config_missing_file(tmp_path):
+    with pytest.raises(FileNotFoundError):
+        load_config(tmp_path / "nope.yaml", load_env=False)

tests/test_dedup.py

@@ -0,0 +1,35 @@
+from clawlibrary.dedup import StateStore
+from clawlibrary.models import PaperCandidate
+
+
+def test_fresh_state_nothing_seen(tmp_path, candidate):
+    store = StateStore.load(tmp_path / "state.json")
+    assert store.is_seen(candidate) is False
+
+
+def test_mark_and_detect_by_doi(tmp_path, candidate):
+    store = StateStore.load(tmp_path / "state.json")
+    store.mark_seen(candidate, "run-1")
+    store.save(now="2026-06-03T00:00:00Z")
+
+    reloaded = StateStore.load(tmp_path / "state.json")
+    assert reloaded.is_seen(candidate) is True
+    assert reloaded.total_downloaded == 1
+
+
+def test_detect_by_url_hash_when_no_doi(tmp_path):
+    c = PaperCandidate(title="x", resolved_pdf_url="https://e/x.pdf")
+    store = StateStore.load(tmp_path / "state.json")
+    store.mark_seen(c, "run-1")
+    # Same URL, no DOI -> caught by url hash
+    c2 = PaperCandidate(title="x copy", resolved_pdf_url="https://e/x.pdf")
+    assert store.is_seen(c2) is True
+
+
+def test_atomic_save_leaves_no_tmp(tmp_path, candidate):
+    store = StateStore.load(tmp_path / "state.json")
+    store.mark_seen(candidate, "run-1")
+    store.save(now="2026-06-03T00:00:00Z")
+    leftovers = list(tmp_path.glob("*.tmp"))
+    assert leftovers == []
+    assert (tmp_path / "state.json").exists()

tests/test_downloader.py

@@ -0,0 +1,81 @@
+import httpx
+
+from clawlibrary.config import RateLimitConfig
+from clawlibrary.downloader import download
+from clawlibrary.ratelimit import RateLimiter
+
+
+def _limiter():
+    # No real sleeping in tests.
+    return RateLimiter(RateLimitConfig(base_delay_seconds=0, jitter_seconds=0,
+                                       backoff_base_seconds=0, backoff_ceiling_seconds=0),
+                       sleeper=lambda s: None)
+
+
+def _client(handler):
+    return httpx.Client(transport=httpx.MockTransport(handler))
+
+
+def test_download_ok_pdf(tmp_path):
+    def handler(request):
+        return httpx.Response(200, headers={"content-type": "application/pdf"},
+                              content=b"%PDF-1.7\n" + b"x" * 100)
+
+    with _client(handler) as client:
+        dest = tmp_path / "topic" / "p.pdf"
+        result = download("https://e/p.pdf", dest, client, _limiter())
+    assert result.ok
+    assert dest.exists()
+    assert result.size_bytes > 100
+    assert list(tmp_path.glob("**/*.tmp")) == []
+
+
+def test_download_accepts_pdf_by_magic_despite_bad_content_type(tmp_path):
+    def handler(request):
+        return httpx.Response(200, headers={"content-type": "application/octet-stream"},
+                              content=b"%PDF-1.7 body")
+
+    with _client(handler) as client:
+        dest = tmp_path / "t" / "p.pdf"
+        result = download("https://e/p.pdf", dest, client, _limiter())
+    assert result.ok
+
+
+def test_download_rejects_non_pdf(tmp_path):
+    def handler(request):
+        return httpx.Response(200, headers={"content-type": "text/html"},
+                              content=b"<html>not a pdf</html>")
+
+    with _client(handler) as client:
+        dest = tmp_path / "t" / "p.pdf"
+        result = download("https://e/p.pdf", dest, client, _limiter())
+    assert result.status == "not_pdf"
+    assert not dest.exists()
+    assert list(tmp_path.glob("**/*.tmp")) == []
+
+
+def test_download_retries_on_429_then_succeeds(tmp_path):
+    calls = {"n": 0}
+
+    def handler(request):
+        calls["n"] += 1
+        if calls["n"] == 1:
+            return httpx.Response(429)
+        return httpx.Response(200, headers={"content-type": "application/pdf"},
+                              content=b"%PDF ok")
+
+    with _client(handler) as client:
+        dest = tmp_path / "t" / "p.pdf"
+        result = download("https://e/p.pdf", dest, client, _limiter())
+    assert result.ok
+    assert calls["n"] == 2
+
+
+def test_download_server_error(tmp_path):
+    def handler(request):
+        return httpx.Response(500)
+
+    with _client(handler) as client:
+        dest = tmp_path / "t" / "p.pdf"
+        result = download("https://e/p.pdf", dest, client, _limiter())
+    assert result.is_server_error

tests/test_pipeline.py

@@ -0,0 +1,87 @@
+import json
+
+import httpx
+
+from clawlibrary.config import (
+    Config,
+    LoggingConfig,
+    RateLimitConfig,
+    SourcesConfig,
+    StateConfig,
+    Topic,
+    VaultConfig,
+)
+from clawlibrary.pipeline import run
+
+from .conftest import OPENALEX_WORK_JSON
+
+
+def _config(tmp_path):
+    return Config(
+        vault=VaultConfig(root=tmp_path / "vault"),
+        state=StateConfig(state_file=tmp_path / "state.json"),
+        logging=LoggingConfig(log_dir=tmp_path / "logs"),
+        rate_limit=RateLimitConfig(
+            base_delay_seconds=0, jitter_seconds=0,
+            backoff_base_seconds=0, backoff_ceiling_seconds=0,
+            max_downloads_per_session=10,
+        ),
+        sources=SourcesConfig(
+            arxiv=False, semantic_scholar=False, unpaywall=False, mailto="[email protected]"
+        ),
+    )
+
+
+def _transport():
+    work = json.loads(OPENALEX_WORK_JSON)
+
+    def handler(request):
+        url = str(request.url)
+        if "api.openalex.org" in url:
+            return httpx.Response(200, json={"results": [work], "meta": {"next_cursor": None}})
+        if url == "https://example.org/best.pdf":
+            return httpx.Response(200, headers={"content-type": "application/pdf"},
+                                  content=b"%PDF-1.7 hello world")
+        return httpx.Response(404)
+
+    return httpx.MockTransport(handler)
+
+
+def test_end_to_end_download_then_dedup(tmp_path):
+    cfg = _config(tmp_path)
+    topic = Topic(query="rust", max_results=1)
+
+    stats = run(cfg, [topic], transport=_transport())
+    assert stats.downloaded == 1
+    assert stats.skipped_dedup == 0
+
+    # PDF + sidecar + index written
+    pdfs = list((tmp_path / "vault").glob("*/*.pdf"))
+    assert len(pdfs) == 1
+    index = json.loads((tmp_path / "vault" / "index.json").read_text())
+    assert len(index) == 1
+    assert index[0]["source"] == "openalex"
+
+    # Second run: everything deduped, nothing re-downloaded
+    stats2 = run(cfg, [topic], transport=_transport())
+    assert stats2.downloaded == 0
+    assert stats2.skipped_dedup == 1
+
+
+def test_dry_run_downloads_nothing(tmp_path):
+    cfg = _config(tmp_path)
+    topic = Topic(query="rust", max_results=1)
+    stats = run(cfg, [topic], dry_run=True, transport=_transport())
+    assert stats.would_download == 1
+    assert stats.downloaded == 0
+    assert not (tmp_path / "vault").exists() or not list((tmp_path / "vault").glob("*/*.pdf"))
+
+
+def test_logs_jsonl_written(tmp_path):
+    cfg = _config(tmp_path)
+    topic = Topic(query="rust", max_results=1)
+    stats = run(cfg, [topic], transport=_transport())
+    logs = list((tmp_path / "logs").glob(f"{stats.run_id}.jsonl"))
+    assert len(logs) == 1
+    lines = logs[0].read_text().strip().splitlines()
+    assert any('"downloaded"' in line for line in lines)

tests/test_ratelimit.py

@@ -0,0 +1,55 @@
+from clawlibrary.config import RateLimitConfig
+from clawlibrary.ratelimit import ConsecutiveErrorTracker, RateLimiter
+
+
+class FakeRng:
+    def __init__(self, val):
+        self.val = val
+
+    def uniform(self, a, b):
+        return self.val
+
+
+def test_wait_applies_base_plus_jitter():
+    slept = []
+    rl = RateLimiter(
+        RateLimitConfig(base_delay_seconds=10, jitter_seconds=3),
+        sleeper=slept.append,
+        rng=FakeRng(2.0),
+    )
+    delay = rl.wait()
+    assert delay == 12.0
+    assert slept == [12.0]
+
+
+def test_wait_never_negative():
+    slept = []
+    rl = RateLimiter(
+        RateLimitConfig(base_delay_seconds=1, jitter_seconds=5),
+        sleeper=slept.append,
+        rng=FakeRng(-5.0),
+    )
+    assert rl.wait() == 0.0
+
+
+def test_backoff_exponential_capped():
+    slept = []
+    rl = RateLimiter(
+        RateLimitConfig(backoff_base_seconds=10, backoff_ceiling_seconds=35),
+        sleeper=slept.append,
+    )
+    assert rl.backoff(0) == 10
+    assert rl.backoff(1) == 20
+    assert rl.backoff(2) == 35  # 40 capped to 35
+
+
+def test_consecutive_error_tracker():
+    t = ConsecutiveErrorTracker(limit=3)
+    t.record_error()
+    t.record_error()
+    assert not t.tripped
+    t.record_success()
+    t.record_error()
+    t.record_error()
+    t.record_error()
+    assert t.tripped

tests/test_resolvers.py

@@ -0,0 +1,92 @@
+import httpx
+
+from clawlibrary.sources import arxiv, core, semanticscholar, tdm
+
+
+def _client(handler):
+    return httpx.Client(transport=httpx.MockTransport(handler))
+
+
+def test_arxiv_doi_mapped():
+    # arXiv-minted DOI maps directly to a PDF without any HTTP call.
+    def handler(request):  # pragma: no cover
+        raise AssertionError("no HTTP expected for DOI-mapped arXiv")
+
+    with _client(handler) as c:
+        url = arxiv.resolve(c, doi="10.48550/arXiv.2301.01234", title=None)
+    assert url == "https://arxiv.org/pdf/2301.01234.pdf"
+
+
+def test_arxiv_title_search():
+    atom = """<feed xmlns="http://www.w3.org/2005/Atom">
+      <entry>
+        <id>http://arxiv.org/abs/2301.09999v1</id>
+        <link title="pdf" href="http://arxiv.org/pdf/2301.09999v1" type="application/pdf"/>
+      </entry>
+    </feed>"""
+
+    def handler(request):
+        return httpx.Response(200, text=atom)
+
+    with _client(handler) as c:
+        url = arxiv.resolve(c, doi=None, title="Some Paper")
+    assert url == "http://arxiv.org/pdf/2301.09999v1.pdf"
+
+
+def test_arxiv_no_match():
+    def handler(request):
+        return httpx.Response(200, text='<feed xmlns="http://www.w3.org/2005/Atom"></feed>')
+
+    with _client(handler) as c:
+        assert arxiv.resolve(c, doi=None, title="x") is None
+
+
+def test_semanticscholar_resolve():
+    def handler(request):
+        return httpx.Response(200, json={"openAccessPdf": {"url": "https://s2/x.pdf"}})
+
+    with _client(handler) as c:
+        assert semanticscholar.resolve(c, "10.1/x", api_key=None) == "https://s2/x.pdf"
+
+
+def test_semanticscholar_with_key():
+    def handler(request):
+        assert request.headers.get("x-api-key") == "k"
+        return httpx.Response(200, json={"openAccessPdf": {"url": "https://s2/y.pdf"}})
+
+    with _client(handler) as c:
+        assert semanticscholar.resolve(c, "10.1/x", api_key="k") == "https://s2/y.pdf"
+
+
+def test_core_requires_key():
+    def handler(request):  # pragma: no cover
+        raise AssertionError("no HTTP without key")
+
+    with _client(handler) as c:
+        assert core.resolve(c, doi="10.1/x", title=None, api_key=None) is None
+
+
+def test_core_resolve():
+    def handler(request):
+        return httpx.Response(200, json={"results": [{"downloadUrl": "https://core/x.pdf"}]})
+
+    with _client(handler) as c:
+        assert core.resolve(c, doi="10.1/x", title=None, api_key="k") == "https://core/x.pdf"
+
+
+def test_tdm_inert_without_token():
+    def handler(request):  # pragma: no cover
+        raise AssertionError("no HTTP expected")
+
+    with _client(handler) as c:
+        assert tdm.resolve(c, doi="10.1/x", tokens={}) is None
+
+
+def test_tdm_builds_url_for_known_publisher():
+    def handler(request):  # pragma: no cover
+        raise AssertionError("resolve only builds a URL, no HTTP")
+
+    with _client(handler) as c:
+        url = tdm.resolve(c, doi="10.1/x", tokens={"elsevier": "tok"})
+    assert url is not None
+    assert "10.1/x" in url and "apiKey=tok" in url

tests/test_sources.py

@@ -0,0 +1,89 @@
+import httpx
+
+from clawlibrary.config import SourcesConfig, Topic
+from clawlibrary.sources import openalex, unpaywall
+from clawlibrary.sources.base import normalize_doi, reconstruct_abstract
+from clawlibrary.sources.resolver import resolve_pdf_url
+
+
+def test_normalize_doi():
+    assert normalize_doi("https://doi.org/10.1/AbC") == "10.1/abc"
+    assert normalize_doi("doi:10.2/x") == "10.2/x"
+    assert normalize_doi(None) is None
+
+
+def test_reconstruct_abstract():
+    inv = {"Rust": [0], "is": [1], "safe": [2]}
+    assert reconstruct_abstract(inv) == "Rust is safe"
+    assert reconstruct_abstract(None) is None
+
+
+def test_work_to_candidate(openalex_work):
+    c = openalex.work_to_candidate(openalex_work, "rust")
+    assert c.doi == "10.1145/3620666.3651327"
+    assert c.authors == ["Wedson Almeida Filho", "Alex Gaynor"]
+    assert c.year == 2023
+    assert c.journal == "SOSP"
+    assert c.resolved_pdf_url == "https://example.org/best.pdf"
+    assert c.pdf_source == "openalex"
+    assert c.abstract == "Rust is safe"
+    assert c.is_open_access is True
+
+
+def test_openalex_search_paginates(openalex_work):
+    pages = [
+        {"results": [openalex_work], "meta": {"next_cursor": "c2"}},
+        {"results": [openalex_work], "meta": {"next_cursor": None}},
+    ]
+    idx = {"n": 0}
+
+    def handler(request):
+        page = pages[idx["n"]]
+        idx["n"] += 1
+        return httpx.Response(200, json=page)
+
+    with httpx.Client(transport=httpx.MockTransport(handler)) as client:
+        topic = Topic(query="rust", max_results=2)
+        results = list(openalex.search(client, topic, mailto="[email protected]"))
+    assert len(results) == 2
+
+
+def test_unpaywall_resolve():
+    def handler(request):
+        return httpx.Response(200, json={
+            "best_oa_location": {"url_for_pdf": "https://oa/x.pdf"}
+        })
+
+    with httpx.Client(transport=httpx.MockTransport(handler)) as client:
+        url = unpaywall.resolve(client, "10.1/x", "[email protected]")
+    assert url == "https://oa/x.pdf"
+
+
+def test_resolver_trusts_openalex_first(candidate):
+    # The list-based resolver collects alternatives too, but OpenAlex's own link
+    # must come first. Other sources are disabled here so no HTTP is needed.
+    def handler(request):  # pragma: no cover - should not be called
+        raise AssertionError("no HTTP expected")
+
+    src = SourcesConfig(unpaywall=False, arxiv=False, semantic_scholar=False, core=False)
+    with httpx.Client(transport=httpx.MockTransport(handler)) as client:
+        url, source = resolve_pdf_url(client, candidate, src)
+    assert source == "openalex"
+    assert url == candidate.resolved_pdf_url
+
+
+def test_resolver_falls_back_to_unpaywall():
+    from clawlibrary.models import PaperCandidate
+
+    cand = PaperCandidate(title="x", doi="10.1/x", topic="t")  # no pdf yet
+
+    def handler(request):
+        if "unpaywall" in str(request.url):
+            return httpx.Response(200, json={"best_oa_location": {"url_for_pdf": "https://oa/x.pdf"}})
+        return httpx.Response(404)
+
+    src = SourcesConfig(arxiv=False, semantic_scholar=False, unpaywall_email="[email protected]")
+    with httpx.Client(transport=httpx.MockTransport(handler)) as client:
+        url, source = resolve_pdf_url(client, cand, src)
+    assert source == "unpaywall"
+    assert url == "https://oa/x.pdf"

tests/test_vault.py

@@ -0,0 +1,63 @@
+import csv
+import json
+
+from clawlibrary.models import VaultEntry
+from clawlibrary.vault import VaultManager, doi_slug, slugify, topic_slug
+
+
+def test_slugify_and_slugs(candidate):
+    assert slugify("Rust & Linux: Kernel!") == "rust-linux-kernel"
+    assert topic_slug("Rust Linux kernel") == "rust-linux-kernel"
+    assert doi_slug(candidate) == "10-1145-3620666-3651327"
+
+
+def test_doi_slug_url_hash_fallback():
+    from clawlibrary.models import PaperCandidate
+
+    c = PaperCandidate(title="x", resolved_pdf_url="https://e/x.pdf")
+    assert doi_slug(c).startswith("urlhash-")
+
+
+def _entry(candidate, dest):
+    return VaultEntry.from_candidate(
+        candidate,
+        doi_slug=doi_slug(candidate),
+        vault_path=str(dest),
+        file_size_bytes=1234,
+        downloaded_at="2026-06-03T00:00:00Z",
+        run_id="run-1",
+    )
+
+
+def test_write_sidecar_and_index(tmp_path, candidate):
+    vault = VaultManager(tmp_path)
+    dest = vault.pdf_path(candidate)
+    dest.parent.mkdir(parents=True, exist_ok=True)
+    dest.write_bytes(b"%PDF-1.4 fake")
+
+    entry = _entry(candidate, dest)
+    sidecar = vault.write_sidecar(entry, dest)
+    vault.append_to_index(entry)
+
+    assert sidecar.exists()
+    side = json.loads(sidecar.read_text())
+    assert side["doi"] == "10.1145/3620666.3651327"
+    assert side["file_size_bytes"] == 1234
+
+    index = json.loads((tmp_path / "index.json").read_text())
+    assert len(index) == 1
+    assert index[0]["title"].startswith("Oxidizing")
+
+    with open(tmp_path / "index.csv") as f:
+        rows = list(csv.DictReader(f))
+    assert rows[0]["authors"] == "Wedson Almeida Filho; Alex Gaynor"
+
+
+def test_index_is_idempotent(tmp_path, candidate):
+    vault = VaultManager(tmp_path)
+    dest = vault.pdf_path(candidate)
+    entry = _entry(candidate, dest)
+    vault.append_to_index(entry)
+    vault.append_to_index(entry)  # same id again
+    index = json.loads((tmp_path / "index.json").read_text())
+    assert len(index) == 1

topics.yaml

@@ -0,0 +1,30 @@
+# Topic queue. Each topic becomes an OpenAlex search; full text is fetched only
+# from legally available open-access locations (or sanctioned TDM, if configured).
+
+defaults:
+  max_results: 40
+  filters:
+    open_access_only: true
+
+topics:
+  - query: "Rust systems programming memory safety"
+    max_results: 25
+    priority: 1
+    filters:
+      date_from: 2020
+      language: en
+
+  - query: "HDF5 scientific data management high performance"
+    max_results: 20
+    priority: 2
+
+  - query: "protein structure prediction machine learning"
+    max_results: 25
+    priority: 3
+    filters:
+      date_from: 2021
+
+  - query: "distributed agent orchestration LLM inference"
+    max_results: 20
+    priority: 4
+    enabled: false  # paused