Merge origin/main into docs/session-2026-10-21

Brings the harvest (#24, #41, #42) and the libsql fix (#45) under the redesign
stack so the session's docs can be written against both.

.github/workflows/daily-refresh.yml

@@ -18,6 +18,14 @@
 # collide with it.
 #
 # Budget: ~900 papers/day encoded on CPU, roughly 10-20 minutes a run.
+#
+# Catching up after a gap: run it by hand with `today` set to a past day, a
+# lookback covering the days since the previous run, and a longer
+# `ingest_timeout_min`, then step `today` forward. A GitHub runner encodes about
+# 2,400 papers an hour (measured 2026-10-20), and arXiv adds roughly 770 a day in
+# the configured categories, so a week is about two hours: today=2026-10-07,
+# lookback_days=9, ingest_timeout_min=240. A run that hits its limit keeps every
+# page it finished; the next run skips them.
 
 name: Daily refresh
 
@@ -29,6 +37,12 @@ on:
       lookback_days:
         description: "Days to look back (re-runs skip stored papers)"
         default: "3"
+      today:
+        description: "Run as if on this day, YYYY-MM-DD, to catch up in short windows (empty = today)"
+        default: ""
+      ingest_timeout_min:
+        description: "Fail the run if one ingest attempt takes longer (catch-up runs need more)"
+        default: "60"
       skip_citations:
         description: "Skip the citation refresh"
         type: boolean
@@ -46,7 +60,9 @@ jobs:
   refresh:
     if: vars.DAILY_REFRESH == 'on'
     runs-on: ubuntu-latest
-    timeout-minutes: 120
+    # A ceiling, not the stall guard: the ingest has its own limit (above) and
+    # every network call its own deadline. GitHub allows 360.
+    timeout-minutes: 300
 
     steps:
       - uses: actions/checkout@v7
@@ -75,7 +91,11 @@ jobs:
           TURSO_DB_TOKEN: ${{ secrets.TURSO_DB_TOKEN }}
           S2_API_KEY: ${{ secrets.S2_API_KEY }}
           INGEST_DEVICE: cpu
+          TODAY: ${{ inputs.today }}     # through env, never pasted into the script
+          INGEST_TIMEOUT_MIN: ${{ inputs.ingest_timeout_min || '60' }}
         run: |
           python scripts/daily_refresh.py \
             --lookback-days "${{ inputs.lookback_days || '3' }}" \
+            --ingest-timeout-min "$INGEST_TIMEOUT_MIN" \
+            ${TODAY:+--today "$TODAY"} \
             ${{ inputs.skip_citations && '--skip-citations' || '' }}

docs/CURRENT-STATE.md

@@ -49,7 +49,7 @@ and deployed. Never copy old corpus counts or benchmark results into current cla
 | Figure warm-up | Behind `FIGURE_PREWARM` (off, on). Reads papers and prepares their first figure before anyone opens them, from two sources: papers people were shown (every feed page and the top 5 search results call `figure_prewarm.enqueue()`; queue of 300, oldest dropped) and, every 6 hours, papers Turso dates in the last 2 days (batches of 300). Every request takes the background lane, one per 15 s at most; 800 papers a day; stops while the breaker is open; one runner per container (file lock). Queue, reads today and the cap in `/healthz/sources` | `app/figure_prewarm.py` |
 | Search thumbnails | Behind `SEARCH_FIGURES` (off, preview, on). A small figure (7.5rem; 4.75rem on a phone) beside a search result whose paper has one saved, via `card_data.enrich()` (local reads only). The classic card's markup is unchanged without a figure. The thumbnail repeats the title's §3.11 link with `#sec-fig`, out of the tab order. Browser tests: `tests/test_e2e_search.py` | `partials/paper_card.html`, `app/routers/search.py` |
 | Feature switches | Redesign features read `off`, `shadow`, `preview` or `on` from Space variables; `preview` shows only to a browser that visited `?preview=<PREVIEW_TOKEN>` | `app/features.py` |
-| Corpus freshness | 1,853,777 papers through 2026-09-29 (2026-10-01). New papers go to `arxiv_recent` + Turso via `scripts/ingest_arxiv.py`; the sidecar catches up at the next publish + deploy. A daily job exists (`scripts/daily_refresh.py`, `.github/workflows/daily-refresh.yml`) but is off until `DAILY_REFRESH=on` | PHASE7 §8 |
+| Corpus freshness | 1,853,777 papers through 2026-09-29 (2026-10-01). New papers go to `arxiv_recent` + Turso via `scripts/ingest_arxiv.py` (`--source oai`, arXiv's OAI-PMH feed, for the daily job; `--source search` for backfills), and the OAI path writes each paper's licence to Turso `paper_licences`. The sidecar catches up at the next publish + deploy. A daily job exists (`scripts/daily_refresh.py`, `.github/workflows/daily-refresh.yml`) but is off until `DAILY_REFRESH=on` | PHASE7 §8 |
 | Citations | Semantic Scholar counts for the whole corpus, refreshed 2026-10-01 (99.47% found); the daily job refreshes the last three months | `scripts/refresh_citations.py` |
 
 ## Constants worth protecting

scripts/daily_refresh.py

@@ -24,7 +24,9 @@ Stages, each with a check that fails the run
   4. verify     new Qdrant points == new Turso rows == papers the ingest
                 reported; sampled new vectors are 1024-dim unit vectors and
                 their Turso rows are complete
-  5. summary    printed as JSON, and to $GITHUB_STEP_SUMMARY when set
+  5. licences   how many new papers have a `paper_licences` row, and how many
+                of those may show figures (reported, never fails the run)
+  6. summary    printed as JSON, and to $GITHUB_STEP_SUMMARY when set
 
 Re-running is always safe: the ingest skips papers already stored and point
 ids are derived from the arXiv id, so overlapping runs cannot collide.
@@ -53,6 +55,9 @@ from pathlib import Path
 import httpx
 
 HERE = Path(__file__).resolve().parent
+sys.path.insert(0, str(HERE.parent))
+from app import licences  # noqa: E402  (standard library only)
+
 REQUIRED_ENV = ("QDRANT_RECENT_URL", "QDRANT_RECENT_API_KEY", "TURSO_URL", "TURSO_DB_TOKEN")
 COLLECTION = os.getenv("QDRANT_RECENT_COLLECTION", "arxiv_recent")
 
@@ -147,13 +152,19 @@ def verify_counts(before: dict, after: dict, ingest_new: int | None) -> list[str
 
 # ── stages ───────────────────────────────────────────────────────────────────
 
-def stage_ingest(since: str, until: str) -> tuple[int | None, str]:
+def stage_ingest(since: str, until: str, timeout_s: int = 3600) -> tuple[int | None, str]:
     state = Path(tempfile.mkdtemp()) / "state.json"
-    cmd = [sys.executable, str(HERE / "ingest_arxiv.py"), "--since", since, "--until", until,
-           "--state", str(state)]
+    cmd = [sys.executable, str(HERE / "ingest_arxiv.py"), "--source", "oai",
+           "--since", since, "--until", until, "--state", str(state)]
     log = ""
     for attempt in (1, 2):
-        p = subprocess.run(cmd, capture_output=True, text=True)
+        try:
+            p = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout_s)
+        except subprocess.TimeoutExpired as e:
+            out = (e.stdout or b"").decode(errors="replace") if isinstance(e.stdout, bytes) else (e.stdout or "")
+            err = (e.stderr or b"").decode(errors="replace") if isinstance(e.stderr, bytes) else (e.stderr or "")
+            print(out[-4000:], err[-6000:], sep="\n", flush=True)    # includes faulthandler stacks
+            raise SystemExit(f"ingest exceeded {timeout_s // 60} min; stacks above") from None
         log += p.stdout + p.stderr
         print(p.stdout[-4000:], p.stderr[-2000:], sep="\n", flush=True)
         if p.returncode == 0:
@@ -200,6 +211,27 @@ def stage_citations(st: Stores, prefix: str) -> dict:
     return {"papers": len(ids), "found": found, "changed": len(changes)}
 
 
+def licence_coverage(codes: list[str | None]) -> dict:
+    """One code per new paper; None or "" when it has no licence row."""
+    known = [c for c in codes if c]
+    figures = sum(1 for c in known if licences.allows_figures(c))
+    n = len(codes)
+    return {"new_papers": n, "with_licence": len(known),
+            "with_licence_share": round(len(known) / n, 3) if n else None,
+            "open": sum(1 for c in known if licences.is_open(c)),
+            "allow_figures": figures, "allow_figures_share": round(figures / n, 3) if n else None,
+            "figure_policy": licences.figure_policy()}
+
+
+def stage_licences(st: Stores, after_rowid: int) -> dict:
+    try:
+        rows = st.t("SELECT l.licence FROM papers p LEFT JOIN paper_licences l ON l.arxiv_id = p.arxiv_id "
+                    "WHERE p.rowid > ?", (after_rowid,))
+    except RuntimeError as e:           # no table yet: no licence has been written
+        return {"error": str(e)[:160]}
+    return licence_coverage([r[0] for r in rows])
+
+
 def stage_sample_new(st: Stores, after_rowid: int, n: int = 5) -> list[str]:
     """Spot-check up to n papers this run added."""
     rows = st.t("SELECT arxiv_id, title, authors, abstract_preview, categories, primary_topic "
@@ -234,6 +266,8 @@ def main() -> int:
     ap.add_argument("--max-recent-points", type=int, default=600_000,
                     help="refuse to write past this many points (free-tier headroom)")
     ap.add_argument("--today", default="", help="YYYY-MM-DD, for re-running a past day")
+    ap.add_argument("--ingest-timeout-min", type=int, default=60,
+                    help="fail the run if one ingest attempt takes longer")
     ap.add_argument("--dry-run", action="store_true")
     ap.add_argument("--skip-citations", action="store_true")
     args = ap.parse_args()
@@ -257,7 +291,7 @@ def main() -> int:
         return _finish(summary, 0)
 
     t0 = time.time()
-    ingest_new, _log = stage_ingest(since, until)
+    ingest_new, _log = stage_ingest(since, until, args.ingest_timeout_min * 60)
     summary["ingest"] = {"new_papers": ingest_new, "minutes": round((time.time() - t0) / 60, 1)}
 
     if not args.skip_citations:
@@ -267,6 +301,7 @@ def main() -> int:
 
     after = st.counts()
     summary["after"] = after
+    summary["licences"] = stage_licences(st, before["turso_max_rowid"])
     problems = verify_counts(before, after, ingest_new) + stage_sample_new(st, before["turso_max_rowid"])
     summary["problems"] = problems
     return _finish(summary, 1 if problems else 0)

scripts/ingest_arxiv.py

@@ -46,19 +46,50 @@ from __future__ import annotations
 
 import argparse
 import json
+from collections import deque
 import os
 import re
 import sys
 import time
+import unicodedata
 import urllib.parse
 import urllib.request
 import xml.etree.ElementTree as ET
 
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(1, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+from net_deadline import deadline  # noqa: E402
+from app.licences import code_from_url  # noqa: E402  (standard library only)
+
 ARXIV_API = "https://export.arxiv.org/api/query"
 NS = {"atom": "http://www.w3.org/2005/Atom",
       "arxiv": "http://arxiv.org/schemas/atom",
       "opensearch": "http://a9.com/-/spec/opensearch/1.1/"}
 
+# arXiv asks clients to identify themselves. On 2026-10-01, after a heavy night,
+# oaipmh.arxiv.org answered ListRecords with HTTP 406 for a while (curl got 200
+# at the same moment), then served the identical request normally: a temporary
+# block, treated like a rate limit below.
+USER_AGENT = "ResearchIT-ingest/1.0 (+https://github.com/siddhm11/ResearchIT-priv)"
+
+
+def _arxiv_get(url: str, timeout: int):
+    return urllib.request.urlopen(
+        urllib.request.Request(url, headers={"User-Agent": USER_AGENT, "Accept": "*/*"}),
+        timeout=timeout)
+
+
+# Status codes that mean "slow down / blocked for now": wait and retry.
+SOFT_BLOCK = (403, 406, 429, 503)
+
+
+def _permanent(error: Exception) -> bool:
+    """A malformed or missing-resource request (400, 404, 422...) fails at once;
+    a soft block (SOFT_BLOCK) is waited out."""
+    code = getattr(error, "code", None)
+    return isinstance(code, int) and 400 <= code < 500 and code not in SOFT_BLOCK
+
+
 # arXiv asks for roughly one request every three seconds.
 ARXIV_DELAY = 3.5
 PAGE = 200  # results per API call; arXiv caps this well below its documented max
@@ -91,8 +122,8 @@ def parse_entry(entry) -> dict | None:
         return None
     arxiv_id = m.group(1)
 
-    title = re.sub(r"\s+", " ", _text(entry, "atom:title"))
-    abstract = re.sub(r"\s+", " ", _text(entry, "atom:summary"))
+    title = clean_text(_text(entry, "atom:title"))
+    abstract = clean_text(_text(entry, "atom:summary"))
     if len(title) < 5 or len(abstract) < 20:
         return None
 
@@ -120,6 +151,26 @@ def parse_entry(entry) -> dict | None:
     }
 
 
+FETCH_ATTEMPTS = 6
+
+
+def retry_wait(error: Exception, attempt: int) -> float:
+    """Seconds to wait before retrying a failed arXiv request.
+
+    A rate limit (429/503) gets arXiv's Retry-After when it sends one, else an
+    exponential back-off from 30 s, capped at 5 min: on 2026-10-01 retries 5,
+    10 and 15 s apart failed every time once arXiv had started throttling.
+    Other errors (timeouts, resets) retry sooner.
+    """
+    code = getattr(error, "code", None)
+    if code in SOFT_BLOCK:
+        after = (getattr(error, "headers", None) or {}).get("Retry-After", "")
+        if str(after).strip().isdigit():
+            return min(300.0, float(after))
+        return min(300.0, 30.0 * 2 ** attempt)
+    return 5.0 * (attempt + 1)
+
+
 def fetch_page(category: str, since: str, until: str, start: int) -> tuple[list[dict], int | None]:
     """One page of results. Returns (papers, total_available); total is None
     when the request failed, so a failure is never mistaken for an empty
@@ -133,16 +184,16 @@ def fetch_page(category: str, since: str, until: str, start: int) -> tuple[list[
         "sortBy": "submittedDate",
         "sortOrder": "ascending",
     })
-    for attempt in range(4):
+    for attempt in range(FETCH_ATTEMPTS):
         try:
-            with urllib.request.urlopen(url, timeout=120) as r:
+            with deadline(150, f"arXiv {category} @{start}"), _arxiv_get(url, 120) as r:
                 xml = r.read()
             break
         except Exception as e:
-            if attempt == 3:
+            if _permanent(e) or attempt == FETCH_ATTEMPTS - 1:
                 print(f"    [{category}] fetch failed: {str(e)[:90]}")
                 return [], None
-            time.sleep(5 * (attempt + 1))
+            time.sleep(retry_wait(e, attempt))
     root = ET.fromstring(xml)
     total_el = root.find("opensearch:totalResults", NS)
     total = int(total_el.text) if total_el is not None and total_el.text else 0
@@ -162,37 +213,238 @@ def _midpoint(since: str, until: str) -> str:
     return (a + (b - a) / 2).isoformat()
 
 
-def plan_windows(category: str, since: str, until: str, probe=None) -> list[list[str]]:
-    """Date windows for one category, each small enough for one arXiv query.
+def walk(cats, since, until, state, handle_page, save_state, *, fetch=None,
+         limit: int = 0, delay: float = ARXIV_DELAY, log=print) -> tuple[int, int]:
+    """Fetch every category's window page by page; returns (examined, new).
 
-    Costs one request per window probed. A window that cannot be split further
-    (a single day) is returned as is and will surface in the coverage report.
+    A window is split by date only when its own first page reports more than
+    MAX_RESULTS_PER_QUERY results, so small windows (the daily case) cost no
+    extra requests. Probing every category up front doubled the requests and
+    drew 19 minutes of HTTP 429 back-off from arXiv on 2026-10-01. Splits are
+    recorded in `state` so a resumed run keeps the same windows.
     """
-    probe = probe or fetch_page
-    _papers, total = probe(category, since, until, 0)
-    time.sleep(ARXIV_DELAY if probe is fetch_page else 0)
-    mid = _midpoint(since, until)
-    if total is None or total <= MAX_RESULTS_PER_QUERY or mid in (since, until):
-        return [[since, until]]
-    return (plan_windows(category, since, mid, probe)
-            + plan_windows(category, mid, until, probe))
+    fetch = fetch or fetch_page
+    queue: deque = deque()
+    for cat in cats:
+        windows = state.setdefault(f"{cat}#windows", [[since, until]])
+        queue.extend((cat, w) for w in windows)
+    seen = new = 0
+    t0 = time.time()
+    while queue:
+        cat, w = queue.popleft()
+        windows = state[f"{cat}#windows"]
+        key = window_key(cat, windows, w)
+        start = int(state.get(key, 0))
+        while True:
+            papers, total = fetch(cat, w[0], w[1], start)
+            if delay:
+                time.sleep(delay)
+            if total is not None and start == 0 and total > MAX_RESULTS_PER_QUERY:
+                mid = _midpoint(w[0], w[1])
+                if mid not in (w[0], w[1]):
+                    halves = [[w[0], mid], [mid, w[1]]]
+                    i = windows.index(w)
+                    windows[i:i + 1] = halves
+                    save_state()
+                    queue.extendleft(reversed([(cat, h) for h in halves]))
+                    log(f"  {cat}: {total:,} results in {w[0]}..{w[1]} -> split at {mid}")
+                    break
+            if total is not None:
+                state[f"{key}#total"] = total
+            if not papers:
+                break
+            seen += len(papers)
+            new += handle_page(papers)
+            start += len(papers)
+            state[key] = start
+            save_state()
+            log(f"  {key:<16} {start:>6}/{total:<7} seen={seen:,} new={new:,}  "
+                f"{(time.time() - t0) / 60:.1f} min")
+            if start >= (total or 0) or (limit and seen >= limit):
+                break
+        if limit and seen >= limit:
+            break
+    return seen, new
 
 
 def window_key(category: str, windows: list[list[str]], w: list[str]) -> str:
     return category if len(windows) == 1 else f"{category}@{w[0]}..{w[1]}"
 
 
+# ── OAI-PMH harvest (the daily path) ─────────────────────────────────────────
+#
+# arXiv's OAI-PMH interface is built for incremental harvesting: everything
+# changed between two dates, a whole archive per request stream, ~1,300
+# records a page. A day across our seven archives is ~10-15 requests, where
+# the search API needs one query per category (~100). Measured parity against
+# rows written from the search API (901 overlapping papers, 2026-09-29):
+# identical embedding text once NFKC-normalised (the search API sends U+03C6
+# where OAI sends U+03D5) and identical categories once arXiv's alias pairs
+# are collapsed as the search API does.
+
+OAI_API = "https://oaipmh.arxiv.org/oai"
+OAI_NS = {"oai": "http://www.openarchives.org/OAI/2.0/", "a": "http://arxiv.org/OAI/arXiv/"}
+# OAI lists both halves of an alias pair; the search API keeps the canonical one.
+CATEGORY_ALIASES = {"math.IT": "cs.IT", "cs.SY": "eess.SY", "cs.NA": "math.NA",
+                    "stat.TH": "math.ST", "math.MP": "math-ph", "q-fin.EC": "econ.GN"}
+
+
+def clean_text(text: str) -> str:
+    return unicodedata.normalize("NFKC", re.sub(r"\s+", " ", text or "").strip())
+
+
+def canonical_categories(categories: str) -> str:
+    cats = categories.split()
+    present, out = set(cats), []
+    for c in cats:
+        if c in CATEGORY_ALIASES and CATEGORY_ALIASES[c] in present:
+            continue                  # canonical is listed too: keep it at its own position
+        c = CATEGORY_ALIASES.get(c, c)
+        if c not in out:
+            out.append(c)
+    return " ".join(out)
+
+
+def oai_set(category: str) -> str:
+    """The archive-level OAI set holding `category`."""
+    archive = category.split(".")[0]
+    if archive in ("cs", "math", "stat", "q-bio", "econ", "eess", "q-fin"):
+        return archive
+    return "physics"                  # hep-*, astro-ph, quant-ph, cond-mat, physics.*
+
+
+def parse_oai_record(rec) -> dict | None:
+    header = rec.find("oai:header", OAI_NS)
+    if header is not None and header.get("status") == "deleted":
+        return None
+    m = rec.find(".//a:arXiv", OAI_NS)
+    if m is None:
+        return None
+    title = clean_text(m.findtext("a:title", "", OAI_NS))
+    abstract = clean_text(m.findtext("a:abstract", "", OAI_NS))
+    if len(title) < 5 or len(abstract) < 20:
+        return None
+    authors = []
+    for au in m.findall("a:authors/a:author", OAI_NS):
+        parts = (au.findtext("a:forenames", "", OAI_NS), au.findtext("a:keyname", "", OAI_NS),
+                 au.findtext("a:suffix", "", OAI_NS))
+        authors.append(" ".join(p.strip() for p in parts if p and p.strip()))
+    cats = canonical_categories(" ".join((m.findtext("a:categories", "", OAI_NS) or "").split()))
+    licence_url = (m.findtext("a:license", "", OAI_NS) or "").strip()
+    return {
+        "arxiv_id": m.findtext("a:id", "", OAI_NS).strip(),
+        "title": title,
+        "authors": ", ".join(authors[:20]),
+        "abstract": abstract,
+        "categories": cats,
+        "primary_topic": cats.split()[0] if cats else "",
+        "update_date": (m.findtext("a:updated", "", OAI_NS) or m.findtext("a:created", "", OAI_NS) or "")[:10],
+        "licence_url": licence_url,           # "" when the record names none
+        "licence": code_from_url(licence_url),
+    }
+
+
+def fetch_oai_page(set_spec: str, since: str, until: str, token: str | None):
+    """One ListRecords page: (papers, next_token, complete_list_size, records).
+
+    `records` counts every record on the page (deleted ones too, as arXiv's
+    completeListSize does); it is None when the request failed."""
+    params = ({"verb": "ListRecords", "resumptionToken": token} if token else
+              {"verb": "ListRecords", "metadataPrefix": "arXiv", "set": set_spec,
+               "from": since, "until": until})
+    url = f"{OAI_API}?" + urllib.parse.urlencode(params)
+    for attempt in range(FETCH_ATTEMPTS):
+        try:
+            with deadline(240, f"OAI {set_spec}"), _arxiv_get(url, 180) as r:
+                xml = r.read()
+            break
+        except Exception as e:
+            if _permanent(e) or attempt == FETCH_ATTEMPTS - 1:
+                print(f"    [oai {set_spec}] fetch failed: {str(e)[:90]}")
+                return [], None, None, None
+            time.sleep(retry_wait(e, attempt))
+    root = ET.fromstring(xml)
+    err = root.find("oai:error", OAI_NS)
+    if err is not None:
+        if err.get("code") == "noRecordsMatch":
+            return [], None, 0, 0
+        print(f"    [oai {set_spec}] error {err.get('code')}: {(err.text or '')[:90]}")
+        return [], None, None, None
+    papers = [p for p in (parse_oai_record(r) for r in root.iter(f"{{{OAI_NS['oai']}}}record")) if p]
+    tok = root.find(".//oai:resumptionToken", OAI_NS)
+    size = tok.get("completeListSize") if tok is not None else None
+    records = sum(1 for _ in root.iter(f"{{{OAI_NS['oai']}}}record"))
+    next_token = ((tok.text or "").strip() or None) if tok is not None else None
+    return papers, next_token, (int(size) if size and size.isdigit() else None), records
+
+
+def harvest(cats, since: str, until: str, min_month: str, state: dict, handle_page, save_state,
+            *, fetch=None, delay: float = ARXIV_DELAY, log=print) -> tuple[int, int]:
+    """Harvest every archive set holding `cats`; returns (examined, new).
+
+    Keeps records with at least one of `cats` and a new-style id from
+    `min_month` (YYMM) on. OAI's <created> is not the first-version date for
+    revised papers (1810.04449 showed 2026-09-10), so the id decides "new".
+    A set is complete only when its last page carries no resumption token;
+    `state["oai:<set>"]` records that for the coverage report.
+    """
+    fetch = fetch or fetch_oai_page
+    wanted = set(cats)
+    seen = new = 0
+    for set_spec in sorted({oai_set(c) for c in cats}):
+        state[f"oai:{set_spec}"] = "started"
+        token, records, expected = None, 0, None
+        while True:
+            papers, token, size, got = fetch(set_spec, since, until, token)
+            if delay:
+                time.sleep(delay)
+            if got is None:
+                state[f"oai:{set_spec}"] = "failed"
+                break
+            records += got
+            expected = size if size is not None else expected
+            keep = [p for p in papers
+                    if wanted & set(p["categories"].split())
+                    and re.match(r"\d{4}\.\d{4,5}$", p["arxiv_id"]) and p["arxiv_id"][:4] >= min_month]
+            seen += len(keep)
+            if keep:
+                new += handle_page(keep)
+            log(f"  oai {set_spec:<8} records={records:,}{'/' + format(expected, ',') if expected else ''} "
+                f"kept={seen:,} new={new:,}")
+            if not token:
+                ok = expected is None or records == expected
+                state[f"oai:{set_spec}"] = "done" if ok else f"short {records}/{expected}"
+                break
+        save_state()
+    return seen, new
+
+
+def report_oai_coverage(cats, state: dict) -> int:
+    sets = sorted({oai_set(c) for c in cats})
+    bad = [f"{s}={state.get(f'oai:{s}')}" for s in sets if state.get(f"oai:{s}") != "done"]
+    if bad:
+        print(f"coverage: {len(bad)} OAI sets incomplete -> re-run: " + ", ".join(bad))
+        return 3
+    print(f"coverage: all {len(sets)} OAI sets harvested to the end")
+    return 0
+
+
 # ── Stores ───────────��───────────────────────────────────────────────────────
 
+def _pipeline_url(url: str) -> str:
+    """Turso's HTTP endpoint. Turso hands out libsql:// URLs, which urllib can't open."""
+    return url.replace("libsql://", "https://", 1).rstrip("/") + "/v2/pipeline"
+
+
 def turso_execute(url: str, token: str, stmts: list[dict]) -> None:
     payload = json.dumps({
         "requests": [{"type": "execute", "stmt": s} for s in stmts]
                     + [{"type": "close"}]}).encode()
     req = urllib.request.Request(
-        f"{url.rstrip('/')}/v2/pipeline", data=payload,
+        _pipeline_url(url), data=payload,
         headers={"Authorization": f"Bearer {token}",
                  "Content-Type": "application/json"})
-    with urllib.request.urlopen(req, timeout=180) as r:
+    with deadline(210, "Turso request"), urllib.request.urlopen(req, timeout=180) as r:
         data = json.loads(r.read())
     for res in data.get("results", []):
         if res.get("type") == "error":
@@ -206,10 +458,10 @@ def turso_existing(url: str, token: str, ids: list[str]) -> set[str]:
     payload = json.dumps({"requests": [{"type": "execute", "stmt": stmt},
                                        {"type": "close"}]}).encode()
     req = urllib.request.Request(
-        f"{url.rstrip('/')}/v2/pipeline", data=payload,
+        _pipeline_url(url), data=payload,
         headers={"Authorization": f"Bearer {token}",
                  "Content-Type": "application/json"})
-    with urllib.request.urlopen(req, timeout=180) as r:
+    with deadline(210, "Turso request"), urllib.request.urlopen(req, timeout=180) as r:
         data = json.loads(r.read())
     res = data["results"][0]
     if res.get("type") == "error":
@@ -231,8 +483,15 @@ def main() -> int:
                     help="fetch, parse and encode, but write nothing. Validates "
                          "the GPU path without needing database credentials.")
     ap.add_argument("--state", default="data/ingest_state.json")
+    ap.add_argument("--source", choices=("search", "oai"), default="search",
+                    help="search: per-category query API (backfills); oai: OAI-PMH (daily)")
     args = ap.parse_args()
 
+    # If a run stalls, its log names the line: dump every thread's stack every
+    # INGEST_STACK_DUMP_S seconds (default 10 min) for as long as it runs.
+    import faulthandler
+    faulthandler.dump_traceback_later(int(os.getenv("INGEST_STACK_DUMP_S", "600")), repeat=True)
+
     cats = ([c.strip() for c in args.categories.split(",") if c.strip()]
             or DEFAULT_CATEGORIES)
 
@@ -258,68 +517,58 @@ def main() -> int:
 
     seen_total = new_total = 0
     t0 = time.time()
-
-    units = []
-    for cat in cats:
-        windows = state.get(f"{cat}#windows")
-        if windows is None:
-            windows = plan_windows(cat, args.since, args.until)
-            state[f"{cat}#windows"] = windows
-        units += [(cat, window_key(cat, windows, w), w) for w in windows]
-
-    for cat, key, (w_since, w_until) in units:
-        start = int(state.get(key, 0))
-        while True:
-            papers, total = fetch_page(cat, w_since, w_until, start)
-            time.sleep(ARXIV_DELAY)
-            if total is not None:
-                state[f"{key}#total"] = total
-            if not papers:
-                break
-
-            seen_total += len(papers)
-            ids = [p["arxiv_id"] for p in papers]
-
-            if args.dry_run:
-                fresh = ids
-            elif args.encode_only:
-                fresh = ids
-                vecs = encoder.encode([
-                    f"{p['title'][:256]} {p['abstract'][:1024]}" for p in papers])
-                d, s = vecs[0]
-                nz = sum(len(sp) for _dv, sp in vecs) / len(vecs)
-                norm = sum(x * x for x in d) ** 0.5
-                print(f"    encoded {len(vecs)}: dense dim={len(d)} "
-                      f"norm={norm:.4f} | sparse avg {nz:.0f} terms/doc",
-                      flush=True)
-            else:
-                have = turso_existing(turso_url, turso_tok, ids)
-                fresh = [i for i in ids if i not in have]
-                todo = [p for p in papers if p["arxiv_id"] in set(fresh)]
-                if todo:
-                    vecs = encoder.encode([
-                        f"{p['title'][:256]} {p['abstract'][:1024]}" for p in todo])
-                    upserter.upsert(todo, vecs)
-
-            new_total += len(fresh)
-            start += len(papers)
-            state[key] = start
-            os.makedirs(os.path.dirname(args.state) or ".", exist_ok=True)
-            json.dump(state, open(args.state, "w"))
-
-            el = time.time() - t0
-            print(f"  {key:<16} {start:>6}/{total:<7} seen={seen_total:,} "
-                  f"new={new_total:,}  {el/60:.1f} min", flush=True)
-
-            if start >= total or (args.limit and seen_total >= args.limit):
-                break
-        if args.limit and seen_total >= args.limit:
-            break
+    licence_tally: dict[str, int] = {}
+
+    def handle_page(papers: list[dict]) -> int:
+        ids = [p["arxiv_id"] for p in papers]
+        if args.source == "oai":
+            for p in papers:
+                code = p["licence"] or ("unrecognised" if p["licence_url"] else "none")
+                licence_tally[code] = licence_tally.get(code, 0) + 1
+        if args.dry_run:
+            return len(ids)
+        if args.encode_only:
+            vecs = encoder.encode([f"{p['title'][:256]} {p['abstract'][:1024]}" for p in papers])
+            d, _s = vecs[0]
+            nz = sum(len(sp) for _dv, sp in vecs) / len(vecs)
+            print(f"    encoded {len(vecs)}: dense dim={len(d)} norm={sum(x * x for x in d) ** 0.5:.4f}"
+                  f" | sparse avg {nz:.0f} terms/doc", flush=True)
+            return len(ids)
+        have = turso_existing(turso_url, turso_tok, ids)
+        todo = [p for p in papers if p["arxiv_id"] not in have]
+        if todo:
+            vecs = encoder.encode([f"{p['title'][:256]} {p['abstract'][:1024]}" for p in todo])
+            upserter.upsert(todo, vecs)               # their licences ride in the same Turso batch
+        # Papers stored before licences were kept get theirs when OAI lists them again.
+        upserter.write_licences([p for p in papers if p["arxiv_id"] in have])
+        return len(todo)
+
+    def save_state() -> None:
+        os.makedirs(os.path.dirname(args.state) or ".", exist_ok=True)
+        json.dump(state, open(args.state, "w"))
+
+    if args.source == "oai":
+        # OAI dates are inclusive datestamps; --until is exclusive like the search
+        # path. Papers are "new" from the month before --since (weekend and
+        # announcement lag), judged by id; re-runs skip stored papers anyway.
+        from datetime import date, timedelta
+        oai_until = (date.fromisoformat(args.until) - timedelta(days=1)).isoformat()
+        first = date.fromisoformat(args.since).replace(day=1) - timedelta(days=1)
+        seen_total, new_total = harvest(cats, args.since, oai_until, first.strftime("%y%m"), state,
+                                        handle_page, save_state, log=lambda m: print(m, flush=True))
+    else:
+        seen_total, new_total = walk(cats, args.since, args.until, state, handle_page, save_state,
+                                     limit=args.limit, log=lambda m: print(m, flush=True))
 
     print(f"\ndone: examined {seen_total:,}, new {new_total:,} "
           f"in {(time.time()-t0)/60:.1f} min")
+    if licence_tally:
+        print("licences: " + ", ".join(f"{k}={v:,}" for k, v in
+                                        sorted(licence_tally.items(), key=lambda kv: -kv[1])))
     if args.limit:
         return 0
+    if args.source == "oai":
+        return report_oai_coverage(cats, state)
     return report_coverage(cats, state)
 
 

scripts/ingest_backends.py

@@ -28,9 +28,13 @@ from __future__ import annotations
 import hashlib
 import json
 import os
+import sys
 import urllib.error
 import urllib.request
 
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from net_deadline import deadline  # noqa: E402
+
 BGE_MODEL = os.getenv("BGE_M3_MODEL", "BAAI/bge-m3")
 # Same pin as app/config.BGE_M3_PINS: new papers must be encoded with the weights
 # that encoded the rest of the corpus.
@@ -130,9 +134,35 @@ class Encoder:
         return results
 
 
+# Licences live in their own table: `INSERT OR REPLACE INTO papers` rewrites
+# whole rows, so a licence column there would vanish on any re-ingest (trap T5).
+# Written only by the OAI path; the search API gives no licence.
+LICENCE_DDL = ("CREATE TABLE IF NOT EXISTS paper_licences (arxiv_id TEXT PRIMARY KEY, "
+               "licence TEXT NOT NULL, licence_url TEXT, source TEXT, updated_at TEXT)")
+
+
+def licence_stmts(papers: list[dict]) -> list[dict]:
+    """Upserts for the papers whose record named a licence. A row that already
+    says the same is left alone, so daily re-harvests write nothing."""
+    return [{
+        "sql": ("INSERT INTO paper_licences (arxiv_id, licence, licence_url, source, updated_at) "
+                "VALUES (?, ?, ?, 'oai', datetime('now')) "
+                "ON CONFLICT(arxiv_id) DO UPDATE SET licence = excluded.licence, "
+                "licence_url = excluded.licence_url, source = excluded.source, "
+                "updated_at = excluded.updated_at "
+                "WHERE paper_licences.licence IS NOT excluded.licence "
+                "OR paper_licences.licence_url IS NOT excluded.licence_url"),
+        "args": [{"type": "text", "value": p["arxiv_id"]},
+                 {"type": "text", "value": p.get("licence") or ""},
+                 {"type": "text", "value": p["licence_url"]}],
+    } for p in papers if p.get("licence_url")]
+
+
 class Upserter:
     """Writes to Qdrant, Zilliz and Turso."""
 
+    _licence_table = False                  # created on the run's first licence write
+
     def __init__(self):
         self.qurl = os.environ.get("QDRANT_RECENT_URL", "").rstrip("/")
         self.qkey = os.environ.get("QDRANT_RECENT_API_KEY", "")
@@ -146,7 +176,7 @@ class Upserter:
         self.ztok = os.environ.get("ZILLIZ_TOKEN", "")
         if WRITE_ZILLIZ and not (self.zuri and self.ztok):
             raise SystemExit("INGEST_WRITE_ZILLIZ=1 needs ZILLIZ_URI and ZILLIZ_TOKEN")
-        self.turl = os.environ["TURSO_URL"].rstrip("/")
+        self.turl = os.environ["TURSO_URL"].replace("libsql://", "https://", 1).rstrip("/")
         self.ttok = os.environ["TURSO_DB_TOKEN"]
 
     # ── Qdrant ───────────────────────────────────────────────────────────
@@ -156,7 +186,7 @@ class Upserter:
             f"{self.qurl}{path}", data=data, method=method,
             headers={"api-key": self.qkey, "Content-Type": "application/json"})
         try:
-            with urllib.request.urlopen(req, timeout=timeout) as r:
+            with deadline(timeout + 30, path), urllib.request.urlopen(req, timeout=timeout) as r:
                 return json.loads(r.read())
         except urllib.error.HTTPError as e:
             # A missing point id is a 404, which urllib raises rather than
@@ -173,7 +203,7 @@ class Upserter:
             f"{self.zuri}{path}", data=json.dumps(body).encode(),
             headers={"Authorization": f"Bearer {self.ztok}",
                      "Content-Type": "application/json"})
-        with urllib.request.urlopen(req, timeout=timeout) as r:
+        with deadline(timeout + 30, path), urllib.request.urlopen(req, timeout=timeout) as r:
             return json.loads(r.read())
 
     # ── Turso ────────────────────────────────────────────────────────────
@@ -185,13 +215,26 @@ class Upserter:
             f"{self.turl}/v2/pipeline", data=payload,
             headers={"Authorization": f"Bearer {self.ttok}",
                      "Content-Type": "application/json"})
-        with urllib.request.urlopen(req, timeout=timeout) as r:
+        with deadline(timeout + 30, "Turso pipeline"), urllib.request.urlopen(req, timeout=timeout) as r:
             data = json.loads(r.read())
         for res in data.get("results", []):
             if res.get("type") == "error":
                 raise RuntimeError(str(res.get("error"))[:200])
 
+    def _with_licences(self, stmts: list[dict], papers: list[dict]) -> None:
+        """Turso `stmts` plus the licence rows for `papers`, in one pipeline."""
+        lic = licence_stmts(papers)
+        if lic and not self._licence_table:
+            lic.insert(0, {"sql": LICENCE_DDL})
+        if stmts or lic:
+            self._t(stmts + lic)
+        self._licence_table = self._licence_table or bool(lic)
+
     # ── Public ──────────────────────��────────────────────────────────────
+    def write_licences(self, papers: list[dict]) -> None:
+        """Licence rows for papers already stored."""
+        self._with_licences([], papers)
+
     def qdrant_existing(self, ids: list[str]) -> set[str]:
         """arxiv_ids of `ids` that already have a point in the collection.
 
@@ -250,4 +293,6 @@ class Upserter:
                          {"type": "text", "value": topic_label(p["categories"])},
                          {"type": "text", "value": p["update_date"]}],
             })
-        self._t(stmts)
+        # Licence rows after the paper rows: Turso is the resume marker, and a
+        # failed licence write is redone when the re-run finds the paper stored.
+        self._with_licences(stmts, papers)

scripts/net_deadline.py

@@ -0,0 +1,40 @@
+"""A hard wall-clock limit for one blocking network call.
+
+Socket timeouts bound each individual read, not the request: a server that
+trickles bytes, or a library call made without a timeout, can block forever.
+On 2026-10-01 a daily-refresh ingest sat 20 minutes at 0% CPU in an SSL read
+(_buffered_readline -> SSLSocket.read -> poll) while arXiv answered other
+requests in 0.2 s. SIGALRM interrupts a blocked syscall, and the handler's
+exception aborts the call instead of letting it resume.
+
+Main thread only (where the ingest makes all its requests); elsewhere it is a
+no-op rather than an error.
+"""
+from __future__ import annotations
+
+import contextlib
+import signal
+import threading
+
+
+class DeadlineExceeded(TimeoutError):
+    pass
+
+
+@contextlib.contextmanager
+def deadline(seconds: float, what: str = "network call"):
+    if (seconds <= 0 or not hasattr(signal, "SIGALRM")
+            or threading.current_thread() is not threading.main_thread()):
+        yield
+        return
+
+    def _expire(_signum, _frame):
+        raise DeadlineExceeded(f"{what} exceeded {seconds:.0f}s")
+
+    previous = signal.signal(signal.SIGALRM, _expire)
+    signal.setitimer(signal.ITIMER_REAL, seconds)
+    try:
+        yield
+    finally:
+        signal.setitimer(signal.ITIMER_REAL, 0)
+        signal.signal(signal.SIGALRM, previous)

tests/test_daily_refresh.py

@@ -55,3 +55,46 @@ def test_workflow_is_gated_serialised_and_passes_every_required_secret():
     assert re.search(r"(?m)^  contents: read$", text)
     for key in dr.REQUIRED_ENV + ("S2_API_KEY",):
         assert re.search(rf"(?m)^          {key}: \$\{{\{{ secrets\.{key} \}}\}}$", text), key
+    # a catch-up day reaches the script through env, never pasted into the shell
+    assert "TODAY: ${{ inputs.today }}" in text and '${TODAY:+--today "$TODAY"}' in text
+    assert "inputs.today }} \\" not in text
+    # catch-up runs can raise the ingest limit; the job ceiling sits above it
+    assert "INGEST_TIMEOUT_MIN: ${{ inputs.ingest_timeout_min || '60' }}" in text
+    assert '--ingest-timeout-min "$INGEST_TIMEOUT_MIN"' in text
+    assert re.search(r"(?m)^    timeout-minutes: 300$", text)
+
+
+def test_ingest_that_overruns_fails_the_run_with_its_output(monkeypatch, capsys):
+    import subprocess
+    import pytest
+
+    def overrun(*a, **k):
+        raise subprocess.TimeoutExpired(cmd="ingest", timeout=60, output="partial log", stderr="Thread 0x1 stack")
+
+    monkeypatch.setattr(dr.subprocess, "run", overrun)
+    with pytest.raises(SystemExit, match="exceeded 1 min"):
+        dr.stage_ingest("2026-09-28", "2026-10-02", timeout_s=60)
+    assert "Thread 0x1 stack" in capsys.readouterr().out
+
+
+def test_the_daily_ingest_uses_the_oai_harvest(monkeypatch):
+    import subprocess
+    seen = []
+
+    def run(cmd, **k):
+        seen.append(cmd)
+        return subprocess.CompletedProcess(cmd, 0, "done: examined 1,204, new 873 in 9.1 min\n", "")
+
+    monkeypatch.setattr(dr.subprocess, "run", run)
+    assert dr.stage_ingest("2026-10-17", "2026-10-21")[0] == 873
+    cmd = seen[0]
+    assert cmd[cmd.index("--source") + 1] == "oai"
+    assert cmd[cmd.index("--since") + 1] == "2026-10-17" and cmd[cmd.index("--until") + 1] == "2026-10-21"
+
+
+def test_licence_coverage_counts_new_papers_with_a_licence(monkeypatch):
+    monkeypatch.setenv("FIGURE_LICENCES", "CC-BY-4.0,CC0-1.0")
+    got = dr.licence_coverage(["CC-BY-4.0", "ARXIV-NONEXCLUSIVE-1.0", "CC-BY-NC-SA-4.0", None, "CC0-1.0"])
+    assert got["new_papers"] == 5 and got["with_licence"] == 4 and got["with_licence_share"] == 0.8
+    assert got["open"] == 3 and got["allow_figures"] == 2 and got["allow_figures_share"] == 0.4
+    assert dr.licence_coverage([])["with_licence_share"] is None      # a quiet day is not an error

tests/test_ingest_safety.py

@@ -80,21 +80,60 @@ def test_failed_fetch_reports_unknown_total(monkeypatch):
     assert ingest_arxiv.fetch_page("cs.CV", "2026-08-01", "2026-08-02", 0) == ([], None)
 
 
-def test_windows_split_until_each_fits_one_query(monkeypatch):
-    # 11,092 papers over 64 days, spread evenly: two halves of ~5.5k fit.
-    def probe(cat, since, until, start):
-        from datetime import date
-        days = (date.fromisoformat(until) - date.fromisoformat(since)).days
-        return [], round(11092 * days / 64)
-    w = ingest_arxiv.plan_windows("cs.AI", "2026-07-29", "2026-10-01", probe=probe)
-    assert w == [["2026-07-29", "2026-08-30"], ["2026-08-30", "2026-10-01"]]
+class FakeArxiv:
+    """Papers spread evenly over days; fails pages that pass arXiv's 10k cap."""
+    def __init__(self, per_day: dict, fail=()):
+        self.per_day, self.fail, self.calls = per_day, set(fail), []
+
+    def __call__(self, cat, since, until, start):
+        from datetime import date, timedelta
+        self.calls.append((cat, since, until, start))
+        if cat in self.fail:
+            return [], None
+        a, b = date.fromisoformat(since), date.fromisoformat(until)
+        ids = [f"{cat}:{(a + timedelta(d)).isoformat()}:{i}"
+               for d in range((b - a).days) for i in range(self.per_day[cat])]
+        if start + 200 > 10_000 and len(ids) > 10_000:
+            return [], None                                   # the real API's 500
+        return [{"arxiv_id": x} for x in ids[start:start + 200]], len(ids)
+
+
+def _walk(fake, cats, state=None):
+    state = {} if state is None else state
+    handled = []
+    seen, new = ingest_arxiv.walk(cats, "2026-07-29", "2026-10-01", state,
+                                  lambda ps: handled.extend(p["arxiv_id"] for p in ps) or len(ps),
+                                  lambda: None, fetch=fake, delay=0, log=lambda m: None)
+    return state, handled, seen, new
+
+
+def test_small_category_costs_only_its_own_pages():
+    fake = FakeArxiv({"cs.RO": 10})                           # 640 papers, 4 pages
+    state, handled, seen, new = _walk(fake, ["cs.RO"])
+    assert len(fake.calls) == 4 and seen == new == 640 == len(set(handled))
+    assert ingest_arxiv.report_coverage(["cs.RO"], state) == 0
+
+
+def test_oversized_window_splits_before_processing_and_covers_everything():
+    fake = FakeArxiv({"cs.AI": 175})                          # 11,200 papers in 64 days
+    state, handled, seen, _ = _walk(fake, ["cs.AI"])
+    assert state["cs.AI#windows"] == [["2026-07-29", "2026-08-30"], ["2026-08-30", "2026-10-01"]]
+    assert seen == len(handled) == len(set(handled)) == 11_200   # nothing twice, nothing lost
+    assert ingest_arxiv.report_coverage(["cs.AI"], state) == 0
+
+
+def test_failed_requests_leave_a_coverage_gap():
+    fake = FakeArxiv({"cs.CL": 5, "cs.CV": 5}, fail={"cs.CL"})
+    state, _h, _s, _n = _walk(fake, ["cs.CL", "cs.CV"])
+    assert ingest_arxiv.report_coverage(["cs.CL", "cs.CV"], state) == 3
 
 
-def test_small_or_failed_probe_keeps_one_window():
-    assert ingest_arxiv.plan_windows("cs.RO", "2026-07-29", "2026-10-01",
-                                     probe=lambda *a: ([], 3273)) == [["2026-07-29", "2026-10-01"]]
-    assert ingest_arxiv.plan_windows("cs.RO", "2026-07-29", "2026-10-01",
-                                     probe=lambda *a: ([], None)) == [["2026-07-29", "2026-10-01"]]
+def test_resumed_run_keeps_the_recorded_split():
+    fake = FakeArxiv({"cs.AI": 175})
+    state, _h, _s, _n = _walk(fake, ["cs.AI"])
+    fake2 = FakeArxiv({"cs.AI": 175})
+    _st, handled, seen, _n = _walk(fake2, ["cs.AI"], state=state)
+    assert seen == 0 and all(c[3] > 0 for c in fake2.calls)      # resumes at the end, no re-split
 
 
 def test_coverage_checks_every_window(capsys):
@@ -114,3 +153,98 @@ def test_point_ids_are_stable_disjoint_and_fit_uint64():
     assert len(set(ids)) == len(ids)                      # 300k ids, no collision
     assert min(ids) >= 1 << 62 > 300_000                  # never meets sequential ids
     assert max(ids) < 1 << 63                             # Qdrant ids are unsigned 64-bit
+
+
+def test_rate_limits_back_off_for_minutes_and_honour_retry_after():
+    import urllib.error
+    def http(code, retry_after=None):
+        headers = {"Retry-After": retry_after} if retry_after else {}
+        return urllib.error.HTTPError("u", code, "x", headers, None)
+    assert ingest_arxiv.retry_wait(http(429), 0) == 30
+    assert ingest_arxiv.retry_wait(http(429), 2) == 120
+    assert ingest_arxiv.retry_wait(http(429), 5) == 300           # capped
+    assert ingest_arxiv.retry_wait(http(503, "45"), 0) == 45      # server's own advice
+    assert ingest_arxiv.retry_wait(TimeoutError("x"), 0) == 5     # transient: retry soon
+
+
+def test_fetch_page_retries_a_rate_limit_then_succeeds(monkeypatch):
+    import io, urllib.error
+    calls, waits = [], []
+    feed = (b'<feed xmlns="http://www.w3.org/2005/Atom" xmlns:opensearch="http://a9.com/-/spec/opensearch/1.1/">'
+            b'<opensearch:totalResults>0</opensearch:totalResults></feed>')
+
+    def urlopen(url, timeout):
+        calls.append(url)
+        if len(calls) < 3:
+            raise urllib.error.HTTPError(url, 429, "Too Many Requests", {}, None)
+        return io.BytesIO(feed)
+
+    monkeypatch.setattr(ingest_arxiv.urllib.request, "urlopen", urlopen)
+    monkeypatch.setattr(ingest_arxiv.time, "sleep", waits.append)
+    assert ingest_arxiv.fetch_page("cs.RO", "2026-09-28", "2026-10-02", 0) == ([], 0)
+    assert waits == [30, 60] and len(calls) == 3
+
+
+# ── licences ─────────────────────────────────────────────────────────────────
+
+def _licensed(aid, url):
+    p = _papers(aid)[0]
+    return {**p, "licence_url": url, "licence": "CC-BY-4.0" if "creativecommons" in url else ""}
+
+
+def test_licence_rows_ride_in_the_same_batch_as_the_paper_rows(monkeypatch):
+    up, calls = _upserter(monkeypatch, present=set())
+    papers = [_licensed("2610.00001", "http://creativecommons.org/licenses/by/4.0/"), _papers("2610.00002")[0]]
+    up.upsert(papers, [([0.1] * 4, {})] * 2)
+    (batch,) = calls["t"]
+    sqls = [s["sql"] for s in batch]
+    assert [s.split()[0:3] for s in sqls[:2]] == [["INSERT", "OR", "REPLACE"]] * 2      # papers first
+    assert sqls[2] == ingest_backends.LICENCE_DDL                                        # then the table, once
+    assert sqls[3].startswith("INSERT INTO paper_licences") and "ON CONFLICT(arxiv_id) DO UPDATE" in sqls[3]
+    assert [a["value"] for a in batch[3]["args"]] == ["2610.00001", "CC-BY-4.0",
+                                                      "http://creativecommons.org/licenses/by/4.0/"]
+    assert len(batch) == 4                                     # the paper without a licence gets no row
+
+    up.upsert([_licensed("2610.00003", "http://creativecommons.org/licenses/by/4.0/")], [([0.1] * 4, {})])
+    assert ingest_backends.LICENCE_DDL not in [s["sql"] for s in calls["t"][1]]      # created once per run
+
+
+def test_stored_papers_get_licence_rows_and_nothing_else(monkeypatch):
+    up, calls = _upserter(monkeypatch, present=set())
+    up._licence_table = True
+    up.write_licences(_papers("2610.00004"))                   # no licence in the record: no call at all
+    assert calls["t"] == []
+    up.write_licences([_licensed("2610.00005", "http://arxiv.org/licenses/nonexclusive-distrib/1.0/")])
+    (batch,) = calls["t"]
+    assert len(batch) == 1 and batch[0]["sql"].startswith("INSERT INTO paper_licences")
+    # an unchanged row is left alone, so daily re-harvests write nothing
+    assert "WHERE paper_licences.licence IS NOT excluded.licence" in batch[0]["sql"]
+
+
+def test_the_licence_upsert_updates_only_rows_that_changed(tmp_path):
+    import sqlite3
+    db = sqlite3.connect(tmp_path / "t.db")
+    db.execute(ingest_backends.LICENCE_DDL)
+
+    def write(url, code):
+        (s,) = ingest_backends.licence_stmts([{"arxiv_id": "2610.00006", "licence_url": url, "licence": code}])
+        db.execute(s["sql"], [a["value"] for a in s["args"]])
+        return db.total_changes
+
+    first = write("http://arxiv.org/licenses/nonexclusive-distrib/1.0/", "ARXIV-NONEXCLUSIVE-1.0")
+    assert write("http://arxiv.org/licenses/nonexclusive-distrib/1.0/", "ARXIV-NONEXCLUSIVE-1.0") == first
+    write("http://creativecommons.org/licenses/by/4.0/", "CC-BY-4.0")
+    assert db.execute("SELECT licence, source FROM paper_licences").fetchall() == [("CC-BY-4.0", "oai")]
+
+
+@pytest.mark.parametrize("url", ["libsql://db-org.turso.io", "https://db-org.turso.io/", "https://db-org.turso.io"])
+def test_turso_urls_work_in_either_form(url, monkeypatch):
+    """Turso hands out libsql:// URLs; urllib only opens https://."""
+    assert ingest_arxiv._pipeline_url(url) == "https://db-org.turso.io/v2/pipeline"
+    monkeypatch.setenv("QDRANT_RECENT_URL", "https://recent.example")
+    monkeypatch.setenv("QDRANT_RECENT_API_KEY", "k")
+    monkeypatch.setenv("QDRANT_URL", "https://primary.example")
+    monkeypatch.setenv("TURSO_URL", url)
+    monkeypatch.setenv("TURSO_DB_TOKEN", "t")
+    monkeypatch.setattr(ingest_backends, "WRITE_ZILLIZ", False)
+    assert ingest_backends.Upserter().turl == "https://db-org.turso.io"

tests/test_net_deadline.py

@@ -0,0 +1,67 @@
+"""scripts/net_deadline.py interrupts calls that socket timeouts cannot."""
+import socket
+import sys
+import threading
+import time
+import urllib.request
+from pathlib import Path
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
+from net_deadline import DeadlineExceeded, deadline  # noqa: E402
+
+
+@pytest.fixture
+def silent_server():
+    """Accepts connections and never sends a byte: the hang seen on 2026-10-01."""
+    srv = socket.socket()
+    srv.bind(("127.0.0.1", 0))
+    srv.listen(5)
+    conns = []
+    stop = threading.Event()
+
+    def run():
+        srv.settimeout(0.2)
+        while not stop.is_set():
+            try:
+                conns.append(srv.accept()[0])
+            except OSError:
+                pass
+
+    t = threading.Thread(target=run, daemon=True)
+    t.start()
+    yield f"http://127.0.0.1:{srv.getsockname()[1]}/"
+    stop.set()
+    t.join(1)
+    for c in conns:
+        c.close()
+    srv.close()
+
+
+def test_deadline_interrupts_a_read_the_socket_timeout_would_allow(silent_server):
+    t0 = time.monotonic()
+    with pytest.raises(DeadlineExceeded):
+        with deadline(1.0, "test"), urllib.request.urlopen(silent_server, timeout=100) as r:
+            r.read()
+    assert time.monotonic() - t0 < 5
+
+
+def test_no_stray_alarm_after_a_call_that_finishes():
+    with deadline(0.3, "fast"):
+        pass
+    time.sleep(0.5)          # an alarm left armed would raise here
+
+
+def test_off_the_main_thread_it_is_a_no_op():
+    out = []
+
+    def worker():
+        with deadline(0.1, "thread"):
+            time.sleep(0.3)
+        out.append("ok")
+
+    t = threading.Thread(target=worker)
+    t.start()
+    t.join()
+    assert out == ["ok"]

tests/test_oai_harvest.py

@@ -0,0 +1,138 @@
+"""OAI-PMH harvesting in scripts/ingest_arxiv.py."""
+import io
+import sys
+import xml.etree.ElementTree as ET
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
+import ingest_arxiv as ia  # noqa: E402
+
+PAGE = """<OAI-PMH xmlns="http://www.openarchives.org/OAI/2.0/"><ListRecords>
+<record><header><identifier>oai:arXiv.org:2609.02991</identifier><datestamp>2026-09-29</datestamp></header>
+<metadata><arXiv xmlns="http://arxiv.org/OAI/arXiv/"><id>2609.02991</id><created>2026-09-02</created>
+<authors><author><keyname>Robert</keyname><forenames>Christian P.</forenames></author>
+<author><keyname>Smith</keyname><forenames>Ann</forenames><suffix>Jr</suffix></author></authors>
+<title>Uncertainty sets
+   for robust learning</title><categories>cs.LG cs.NA math.NA stat.ML</categories>
+<abstract>  We study $ϕ$-divergence uncertainty sets and   function approximation in depth.  </abstract>
+</arXiv></metadata></record>
+<record><header status="deleted"><identifier>oai:arXiv.org:2609.00001</identifier><datestamp>2026-09-29</datestamp></header></record>
+<resumptionToken completeListSize="2" cursor="0"></resumptionToken>
+</ListRecords></OAI-PMH>"""
+
+
+def test_record_is_normalised_like_the_search_api():
+    root = ET.fromstring(PAGE)
+    recs = list(root.iter("{http://www.openarchives.org/OAI/2.0/}record"))
+    p = ia.parse_oai_record(recs[0])
+    assert p["arxiv_id"] == "2609.02991"
+    assert p["title"] == "Uncertainty sets for robust learning"
+    assert "$\u03c6$-divergence" in p["abstract"] and "\u03d5" not in p["abstract"]   # NFKC
+    assert p["abstract"].startswith("We study") and "  " not in p["abstract"]
+    assert p["categories"] == "cs.LG math.NA stat.ML"       # cs.NA alias dropped, order kept
+    assert p["primary_topic"] == "cs.LG"
+    assert p["authors"] == "Christian P. Robert, Ann Smith Jr"
+    assert p["update_date"] == "2026-09-02"
+    assert ia.parse_oai_record(recs[1]) is None             # deleted
+
+
+def test_alias_collapsing_keeps_the_canonical_position():
+    assert ia.canonical_categories("eess.AS cs.AI cs.SD cs.SY eess.SP eess.SY") == "eess.AS cs.AI cs.SD eess.SP eess.SY"
+    assert ia.canonical_categories("cs.IT cs.DC math.IT") == "cs.IT cs.DC"
+    assert ia.canonical_categories("math.IT cs.LG") == "cs.IT cs.LG"      # alias alone maps in place
+
+
+def test_sets_cover_every_configured_category():
+    sets = {ia.oai_set(c) for c in ia.DEFAULT_CATEGORIES}
+    assert sets == {"cs", "stat", "math", "q-bio", "econ", "eess", "physics"}
+
+
+def _paper(aid, cats):
+    return {"arxiv_id": aid, "title": "t" * 6, "authors": "a", "abstract": "x" * 30,
+            "categories": cats, "primary_topic": cats.split()[0], "update_date": "2026-09-29"}
+
+
+def test_harvest_filters_pages_through_and_marks_complete_sets():
+    pages = {
+        ("cs", None): ([_paper("2609.00010", "cs.CL"), _paper("2609.00011", "cs.OS"),     # cs.OS not configured
+                        _paper("1810.04449", "cs.LG"), _paper("2512.00001", "cs.LG")],     # old ids
+                       "tok1", 5, 4),
+        ("cs", "tok1"): ([_paper("2609.00012", "cs.CV")], None, 5, 1),
+        ("stat", None): ([_paper("2609.00020", "stat.ML math.ST")], None, 1, 1),
+    }
+    fake = lambda s, since, until, tok: pages[(s, tok)]
+    handled, state = [], {}
+    seen, new = ia.harvest(["cs.CL", "cs.CV", "stat.ML"], "2026-09-28", "2026-09-30", "2608", state,
+                           lambda ps: handled.extend(p["arxiv_id"] for p in ps) or len(ps),
+                           lambda: None, fetch=fake, delay=0, log=lambda m: None)
+    assert handled == ["2609.00010", "2609.00012", "2609.00020"] and seen == new == 3
+    assert state == {"oai:cs": "done", "oai:stat": "done"}
+    assert ia.report_oai_coverage(["cs.CL", "stat.ML"], state) == 0
+
+
+def test_short_or_failed_sets_fail_coverage():
+    pages = {("cs", None): ([], None, 9, 4),          # arXiv promised 9 records, 4 arrived
+             ("stat", None): ([], None, None, None)}  # request failed
+    state = {}
+    ia.harvest(["cs.CL", "stat.ML"], "2026-09-28", "2026-09-30", "2608", state, lambda ps: 0,
+               lambda: None, fetch=lambda s, a, b, t: pages[(s, t)], delay=0, log=lambda m: None)
+    assert state == {"oai:cs": "short 4/9", "oai:stat": "failed"}
+    assert ia.report_oai_coverage(["cs.CL", "stat.ML"], state) == 3
+
+
+def test_a_day_with_no_records_is_complete(monkeypatch):
+    body = (b'<OAI-PMH xmlns="http://www.openarchives.org/OAI/2.0/">'
+            b'<error code="noRecordsMatch">no records</error></OAI-PMH>')
+    monkeypatch.setattr(ia.urllib.request, "urlopen", lambda url, timeout: io.BytesIO(body))
+    assert ia.fetch_oai_page("econ", "2026-09-28", "2026-09-30", None) == ([], None, 0, 0)
+    state = {}
+    ia.harvest(["econ.TH"], "2026-09-28", "2026-09-30", "2608", state, lambda ps: 0, lambda: None,
+               delay=0, log=lambda m: None)
+    assert state == {"oai:econ": "done"}
+
+
+def test_parsing_the_sample_page_via_fetch(monkeypatch):
+    monkeypatch.setattr(ia.urllib.request, "urlopen", lambda url, timeout: io.BytesIO(PAGE.encode()))
+    papers, token, size, records = ia.fetch_oai_page("cs", "2026-09-28", "2026-09-30", None)
+    assert [p["arxiv_id"] for p in papers] == ["2609.02991"] and token is None and size == 2 and records == 2
+
+
+def test_requests_identify_themselves_and_bad_requests_are_not_retried(monkeypatch):
+    import urllib.error
+    seen, waits = [], []
+
+    def urlopen(req, timeout):
+        seen.append(req.get_header("User-agent"))
+        raise urllib.error.HTTPError(req.full_url, 400, "Bad Request", {}, None)
+
+    monkeypatch.setattr(ia.urllib.request, "urlopen", urlopen)
+    monkeypatch.setattr(ia.time, "sleep", waits.append)
+    assert ia.fetch_oai_page("cs", "2026-09-28", "2026-09-30", None) == ([], None, None, None)
+    assert seen == [ia.USER_AGENT] and waits == []          # one attempt, no back-off
+    assert "Python-urllib" not in ia.USER_AGENT
+    assert ia._permanent(urllib.error.HTTPError("u", 404, "x", {}, None))
+    assert not ia._permanent(urllib.error.HTTPError("u", 429, "x", {}, None))
+    assert not ia._permanent(urllib.error.HTTPError("u", 406, "x", {}, None))   # was a temporary block
+    assert ia.retry_wait(urllib.error.HTTPError("u", 406, "x", {}, None), 0) == 30
+    assert not ia._permanent(TimeoutError("x"))
+
+
+def _record_with_licence(licence: str | None) -> ET.Element:
+    tag = f"<license>{licence}</license>" if licence is not None else ""
+    return ET.fromstring(
+        '<record xmlns="http://www.openarchives.org/OAI/2.0/"><header><identifier>oai:arXiv.org:2610.00042'
+        '</identifier></header><metadata><arXiv xmlns="http://arxiv.org/OAI/arXiv/"><id>2610.00042</id>'
+        '<created>2026-10-19</created><authors><author><keyname>Lee</keyname></author></authors>'
+        '<title>Figures with licences</title><categories>cs.CV</categories>'
+        f'<abstract>An abstract long enough to keep the record.</abstract>{tag}</arXiv></metadata></record>')
+
+
+def test_the_licence_is_read_from_the_record():
+    cc = ia.parse_oai_record(_record_with_licence("http://creativecommons.org/licenses/by/4.0/"))
+    assert cc["licence"] == "CC-BY-4.0" and cc["licence_url"] == "http://creativecommons.org/licenses/by/4.0/"
+    default = ia.parse_oai_record(_record_with_licence(" http://arxiv.org/licenses/nonexclusive-distrib/1.0/ "))
+    assert default["licence"] == "ARXIV-NONEXCLUSIVE-1.0"
+    odd = ia.parse_oai_record(_record_with_licence("https://example.org/my-licence"))
+    assert odd["licence"] == "" and odd["licence_url"] == "https://example.org/my-licence"   # kept to remap later
+    none = ia.parse_oai_record(list(ET.fromstring(PAGE).iter("{http://www.openarchives.org/OAI/2.0/}record"))[0])
+    assert none["licence"] == "" and none["licence_url"] == ""