feat(ingest): keep each harvested paper's licence in Turso paper_licences

parse_oai_record reads <license> into licence_url and a short code
(app.licences.code_from_url). New papers' licence rows go in the same
Turso pipeline as their papers rows; papers already stored get theirs
when OAI lists them again. Own table, never a papers column: INSERT OR
REPLACE rewrites whole papers rows (trap T5). The upsert skips rows that
already say the same, so daily re-harvests write nothing.

Dry run over 2026-10-16: 1,833 papers kept, all with a licence
(ARXIV-NONEXCLUSIVE 865, CC-BY-4.0 773, CC-BY-NC-ND 79, CC-BY-NC-SA 67,
CC-BY-SA 36, CC0 13).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

scripts/ingest_arxiv.py

@@ -57,7 +57,9 @@ 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",
@@ -328,6 +330,7 @@ def parse_oai_record(rec) -> dict | None:
                  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,
@@ -336,6 +339,8 @@ def parse_oai_record(rec) -> dict | None:
         "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),
     }
 
 
@@ -507,9 +512,14 @@ def main() -> int:
 
     seen_total = new_total = 0
     t0 = time.time()
+    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:
@@ -523,7 +533,9 @@ def main() -> int:
         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)
+            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:
@@ -545,6 +557,9 @@ def main() -> int:
 
     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":

scripts/ingest_backends.py

@@ -134,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", "")
@@ -195,7 +221,20 @@ class Upserter:
             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.
 
@@ -254,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)

tests/test_ingest_safety.py

@@ -183,3 +183,55 @@ def test_fetch_page_retries_a_rate_limit_then_succeeds(monkeypatch):
     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")]

tests/test_oai_harvest.py

@@ -115,3 +115,24 @@ def test_requests_identify_themselves_and_bad_requests_are_not_retried(monkeypat
     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"] == ""