Preprints from any server: the upload form takes a preprint DOI, checked against Crossref and DataCite when submitted (a published article, a preprint already published, or a DOI that does not exist is refused; an unreachable registry fails open and the curation team is told), and the Persistence DOI is linked to it; the DOI route sends other servers to the upload form with the DOI; the exclusivity confirmation is stored on both routes (it never was); the acceptance email says it only when it was confirmed

crossref_deposit.py

@@ -312,24 +312,20 @@ def _license(parent: ET.Element, license_id: str) -> None:
     ref.text = url
 
 
-_ARXIV_ID = re.compile(r"^(?:arxiv:)?(?:10\.48550/arxiv\.)?(\d{4}\.\d{4,5})(?:v\d+)?$", re.I)
-
-
 def preprint_doi(submission: dict) -> Optional[str]:
-    """The DOI of the preprint a DOI-route paper was submitted from (a Zenodo
-    DOI, or arXiv's 10.48550 DOI for an arXiv id), else None. The Persistence
-    article gets its own DOI and points back to it: Crossref's versioning
-    practice gives a preprint and its published version separate DOIs, linked
-    by a relation (policy 2026-09-28; the five founding papers keep their
-    Zenodo DOIs, as stated to Crossref on 2026-07-06)."""
-    if submission.get("source") != "doi":
-        return None
-    ref = (submission.get("doi") or submission.get("source_ref") or "").strip()
-    ref = re.sub(r"^(?:https?://(?:dx\.)?doi\.org/|doi:)", "", ref, flags=re.I)
-    m = _ARXIV_ID.match(ref)
-    if m:
-        return f"10.48550/arXiv.{m.group(1)}"
-    return ref if re.match(r"^10\.\d{4,9}/\S+$", ref) else None
+    """The DOI of the preprint the paper was posted as: the DOI route's own DOI
+    (Zenodo, or arXiv's 10.48550 DOI for an arXiv id), or the upload route's
+    preprint DOI field (any server). None otherwise. The Persistence article
+    gets its own DOI and points back to it: Crossref's versioning practice
+    gives a preprint and its published version separate DOIs, linked by a
+    relation (policy 2026-09-28; the five founding papers keep their Zenodo
+    DOIs, as stated to Crossref on 2026-07-06)."""
+    import preprint_check
+    if submission.get("source") == "doi":
+        ref = submission.get("doi") or submission.get("source_ref")
+    else:
+        ref = submission.get("preprint_doi")
+    return preprint_check.normalize(ref) if ref else None
 
 
 def _relations(parent: ET.Element, submission: dict) -> None:

intake/apply_decision.py

@@ -220,7 +220,7 @@ def main(argv: list[str]) -> int:
     crossref_path = (verdict == "accept" and source in ("upload", "doi")
                      and registrar == "crossref" and not (test_mode and tier == 2))
     preprint = None
-    if crossref_path and source == "doi":
+    if crossref_path:
         import crossref_deposit as _cdp
         preprint = _cdp.preprint_doi(submission)
     if crossref_path and not state_pre.get("crossref_doi"):
@@ -284,18 +284,22 @@ def main(argv: list[str]) -> int:
                           f"Will resolve to: {state_pre.get('crossref_landing_url')}"]
             else:
                 lines.append("Crossref draft: FAILED to stage -- see audit log")
-            if source == "doi":
-                lines.append(f"Preprint (linked as hasPreprint): "
-                             f"{('https://doi.org/' + preprint) if preprint else 'NOT RESOLVED -- check the relation before --live'}")
-                if not (submission.get("license") or "").lower().startswith("cc"):
-                    lines.append(f"Licence on the preprint: {submission.get('license') or 'none recorded'} "
-                                 f"-- check it before --live (the site hosts the PDF)")
-            elif state_pre.get("deposit_draft_url"):
-                lines.append(f"Zenodo archive DRAFT (unpublished): {state_pre['deposit_draft_url']}")
-            elif form.get("deposit_consent"):
-                lines.append("Zenodo archive draft: NOT staged -- see audit log")
-            else:
-                lines.append("Zenodo archive: author did not consent to deposit")
+            if preprint:
+                lines.append(f"Preprint (linked as hasPreprint): https://doi.org/{preprint}")
+                for flag in ((submission.get("preprint_meta") or {}).get("flags") or []):
+                    lines.append(f"  check the preprint: {flag}")
+            elif source == "doi":
+                lines.append("Preprint: NOT RESOLVED -- check the relation before --live")
+            if source == "doi" and not (submission.get("license") or "").lower().startswith("cc"):
+                lines.append(f"Licence on the preprint: {submission.get('license') or 'none recorded'} "
+                             f"-- check it before --live (the site hosts the PDF)")
+            if source == "upload":   # a DOI-route preprint is archived where it is
+                if state_pre.get("deposit_draft_url"):
+                    lines.append(f"Zenodo archive DRAFT (unpublished): {state_pre['deposit_draft_url']}")
+                elif form.get("deposit_consent"):
+                    lines.append("Zenodo archive draft: NOT staged -- see audit log")
+                else:
+                    lines.append("Zenodo archive: author did not consent to deposit")
             lines += ["Acceptance email: Gmail Drafts (review and send by hand).",
                       f"When ready: intake/register-doi.sh {sub_id} --live"]
             notify.send_to_curator("\n".join(lines), parse_mode=None,
@@ -419,6 +423,7 @@ def main(argv: list[str]) -> int:
         terms_version=submission.get("terms_version") or getattr(config, "TERMS_VERSION", ""),
         approval_url=approval_url, objection_deadline=objection_deadline,
         code_link_claims=code_link_claims, preprint_doi=preprint or "",
+        exclusivity_confirmed=form.get("exclusivity_acknowledged") is True,
     )
     if ok:
         # Decision emails go to Gmail Drafts (curator-applied decision path).

intake/intake_server.py

@@ -45,6 +45,7 @@ if str(_REPO_ROOT) not in sys.path:
 import config  # noqa: E402
 
 import submission_intake as ingest  # noqa: E402  — pipeline module
+import preprint_check  # noqa: E402
 import notify  # noqa: E402
 
 from . import notify_author  # local
@@ -745,6 +746,7 @@ async def handle_test_pipeline_submission(
 
     if has_doi:
         _validate_doi_shape(doi)
+    preprint = None if has_doi else await _preprint_from_form(form)
 
     sub_id = _allocate_test_sub_id()
     token = _test_token(tier, seed=sub_id)
@@ -877,6 +879,8 @@ async def handle_test_pipeline_submission(
         "funding": funding,
         "related_identifiers": related_identifiers,
         "code_data": code_data or {"available": None, "url": None},
+        "preprint_doi": preprint["doi"] if preprint else None,
+        "preprint_meta": preprint,
         "pdf": {
             "filename": "paper.pdf",
             "size_bytes": pdf_size,
@@ -952,6 +956,31 @@ def healthz():
     return {"ok": True, "ts": _now_iso()}
 
 
+async def _preprint_from_form(form) -> dict | None:
+    """The upload route's optional preprint DOI, from any preprint server
+    (policy 2026-09-28). The preprint keeps its DOI; the Persistence article gets
+    its own, linked to it. Refused (422) when it is not a DOI, does not exist, or
+    is a published work; a registry that cannot be reached fails open (the
+    record says unverified and the curation team checks by hand)."""
+    raw = form.get("preprint_doi")
+    raw = raw.strip() if isinstance(raw, str) else ""
+    if not raw:
+        return None
+    doi = preprint_check.normalize(raw)
+    if not doi:
+        raise HTTPException(422, {
+            "error": "preprint_doi_invalid",
+            "message": ("The preprint DOI should look like 10.xxxx/… (or an arXiv ID such as "
+                        "2401.12345). Check it, or leave the field empty."),
+        })
+    title = form.get("title")
+    try:
+        return await asyncio.to_thread(preprint_check.check, doi,
+                                       title=title if isinstance(title, str) else "")
+    except preprint_check.Refused as exc:
+        raise HTTPException(422, {"error": "preprint_doi_refused", "message": str(exc)})
+
+
 def _validate_doi_shape(doi: str) -> None:
     """Synchronous shape-only check on a DOI or arXiv reference. The actual
     resolution + PDF fetch is deferred to the worker so the handler
@@ -969,11 +998,10 @@ def _validate_doi_shape(doi: str) -> None:
         raise HTTPException(422, {
             "error": "doi_unsupported",
             "message": (
-                "DOI mode supports Zenodo (10.5281/zenodo.NNNNN) and arXiv "
-                "(10.48550/arXiv.YYMM.NNNNN, or a bare arXiv ID like "
-                "2103.12345). Other DOI sources (Crossref, DataCite, "
-                "publisher DOIs) are a known gap — for those, upload the "
-                "PDF directly using the 'Upload PDF' tab."
+                "The DOI route fetches papers from Zenodo and arXiv. For a "
+                "preprint on any other server, use the upload form: upload "
+                "the PDF and enter this DOI as your preprint DOI "
+                f"(https://icsacinstitute.org/submit/upload?preprint_doi={doi})."
             ),
         })
     raise HTTPException(422, {
@@ -1006,6 +1034,8 @@ async def api_submit(request: Request):
         "email": form.get("email"),
         "orcid": form.get("orcid", ""),
         "coi": form.get("coi", ""),
+        # Never passed before 2026-09-28, so every submission stored None.
+        "exclusivity_acknowledged": form.get("exclusivity_acknowledged"),
     })
     code_data = _parse_code_data(form.get("code_data_available"), form.get("code_data_url"))
 
@@ -1093,6 +1123,7 @@ async def api_submit(request: Request):
         # previously blew CF Pages' upstream-fetch timeout and blocked
         # the form's redirect on success).
         _validate_doi_shape(doi)
+    preprint = None if has_doi else await _preprint_from_form(form)
 
     sub_id = _allocate_sub_id()
     sub_dir = SUBMISSIONS_ROOT / sub_id
@@ -1228,6 +1259,8 @@ async def api_submit(request: Request):
         "funding": funding,
         "related_identifiers": related_identifiers,
         "code_data": code_data or {"available": None, "url": None},
+        "preprint_doi": preprint["doi"] if preprint else None,
+        "preprint_meta": preprint,
         "pdf": {
             "filename": "paper.pdf",
             "size_bytes": pdf_size,

intake/notify_author.py

@@ -310,6 +310,7 @@ def send_decision(*, to: str, sub_id: str, title: str, author_name: str,
                   objection_deadline: str = "",
                   code_link_claims: list[str] | None = None,
                   preprint_doi: str = "",
+                  exclusivity_confirmed: bool = False,
                   ) -> tuple[bool, str]:
     """Send the decision email with two PDF attachments (panel report + RQC).
 
@@ -438,6 +439,8 @@ def send_decision(*, to: str, sub_id: str, title: str, author_name: str,
         "prior_doi_line": (
             f"Your preprint keeps its own DOI ({preprint_doi}) and stays where it is. The *Persistence* "
             f"DOI is the published version, and the two are linked in their DOI records."
+            + (" We take it that the paper is not under review elsewhere, as confirmed at submission."
+               if exclusivity_confirmed else "")
             if preprint_doi else
             "We take it that this paper has no other DOI and is not under review elsewhere, "
             "as confirmed at submission."),

intake/scripts/test_preprint_doi.py

@@ -0,0 +1,215 @@
+#!/usr/bin/env python3
+"""Checks for preprint DOIs from any server (policy 2026-09-28) and the
+exclusivity confirmation on both routes. Offline: the DOI registries are
+replaced by canned records, no mail, no pings; every write lands in one
+temporary directory that is removed at the end, pass or fail. Run from the
+repo root with the venv:
+    .venv/bin/python intake/scripts/test_preprint_doi.py
+"""
+from __future__ import annotations
+
+import email as _email
+import email.policy as _policy
+import hashlib
+import hmac
+import json
+import pathlib
+import shutil
+import sys
+import tempfile
+import time
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(ROOT))
+PAPER_PDF = Path.home() / "icsac-submissions" / "ICSAC-SUB-00008" / "paper.pdf"
+TEST_ORCID = "0000-0002-1825-0097"
+
+failures: list[str] = []
+
+
+def check(cond: bool, name: str) -> None:
+    print(("  ok   " if cond else "  FAIL ") + name)
+    if not cond:
+        failures.append(name)
+
+
+# Canned registry answers, keyed by URL fragment.
+RA = {"10.20944/preprints202607.0102.v1": "Crossref", "10.1101/2024.09.14.613029": "Crossref",
+      "10.1038/nature12373": "Crossref", "10.1364/opticaopen.29459153.v1": "Crossref",
+      "10.48550/arXiv.2401.12345": "DataCite", "10.5281/zenodo.1": "DataCite",
+      "10.51094/jxiv.1": "JaLC"}
+CR = {
+    "10.20944/preprints202607.0102.v1": {"type": "posted-content", "subtype": "preprint",
+                                         "title": ["A test paper on boundaries"], "publisher": "MDPI AG"},
+    "10.1101/2024.09.14.613029": {"type": "posted-content", "subtype": "preprint",
+                                  "title": ["Something else entirely about proteins"],
+                                  "institution": [{"name": "bioRxiv"}]},
+    "10.1038/nature12373": {"type": "journal-article", "title": ["Nanometre-scale thermometry"],
+                            "container-title": ["Nature"]},
+    "10.1364/opticaopen.29459153.v1": {"type": "posted-content", "subtype": "preprint", "title": ["X"],
+                                       "relation": {"is-preprint-of": [{"id": "10.1364/OE.572415"}]}},
+}
+DC = {"10.48550/arXiv.2401.12345": {"types": {"resourceTypeGeneral": "Preprint"}, "publisher": "arXiv",
+                                    "titles": [{"title": "A test paper on boundaries"}]},
+      "10.5281/zenodo.1": {"types": {"resourceTypeGeneral": "JournalArticle"}, "publisher": {"name": "Zenodo"},
+                           "titles": [{"title": "A test paper on boundaries"}]}}
+NETWORK_DOWN = {"on": False}
+
+
+def fake_get(url: str, timeout: float):
+    import urllib.parse
+    if NETWORK_DOWN["on"]:
+        raise OSError("network down")
+    tail = urllib.parse.unquote(url.split("/ra/", 1)[-1] if "/ra/" in url else
+                                url.split("/works/", 1)[-1] if "/works/" in url else url.split("/dois/", 1)[-1])
+    if "/ra/" in url:
+        return [{"DOI": tail, "RA": RA[tail]}] if tail in RA else [{"DOI": tail, "status": "DOI does not exist"}]
+    if "/works/" in url:
+        return {"message": CR[tail]}
+    return {"data": {"attributes": DC[tail]}}
+
+
+tmp = Path(tempfile.mkdtemp(prefix="preprint-"))
+real_home = pathlib.Path.home
+try:
+    import preprint_check as pc
+    pc._get = fake_get
+
+    print("1. the check")
+    r = pc.check("10.20944/preprints202607.0102.v1", title="A test paper on boundaries")
+    check(r["verified"] and r["type"] == "posted-content/preprint" and not r["flags"], "Preprints.org preprint: accepted, clean")
+    r = pc.check("10.1101/2024.09.14.613029", title="A test paper on boundaries")
+    check(r["server"] == "bioRxiv" and any("title differs" in f for f in r["flags"]), "bioRxiv preprint with another title: accepted, flagged")
+    for d, why in (("10.1038/nature12373", "a journal article"), ("10.1364/opticaopen.29459153.v1", "a preprint already published"),
+                   ("10.9999/nope", "a DOI that does not exist")):
+        try:
+            pc.check(d)
+            check(False, f"refused: {why}")
+        except pc.Refused:
+            check(True, f"refused: {why}")
+    r = pc.check("10.48550/arXiv.2401.12345")
+    check(r["verified"] and r["type"] == "Preprint" and not r["flags"], "arXiv (DataCite Preprint): accepted, clean")
+    r = pc.check("10.5281/zenodo.1")
+    check(r["server"] == "Zenodo" and any("JournalArticle" in f for f in r["flags"]), "Zenodo filed as a journal article: accepted, flagged")
+    r = pc.check("10.51094/jxiv.1")
+    check(r["ra"] == "JaLC" and not r["verified"] and r["flags"], "another agency (JaLC): accepted, flagged")
+    NETWORK_DOWN["on"] = True
+    r = pc.check("10.20944/preprints202607.0102.v1")
+    NETWORK_DOWN["on"] = False
+    check(not r["verified"] and any("could not be reached" in f for f in r["flags"]), "registries down: fails open, flagged")
+    check(pc.normalize("https://doi.org/10.1101/2024.09.14.613029") == "10.1101/2024.09.14.613029", "resolver URL normalised")
+    check(pc.normalize("arXiv:2401.12345v3") == "10.48550/arXiv.2401.12345", "arXiv: prefix and version normalised")
+    check(pc.normalize("not a doi") is None, "garbage refused")
+
+    print("2. /api/submit (T2, signed like the proxy)")
+    import httpx
+    from fastapi.testclient import TestClient
+    from intake import intake_server as iss
+    iss.HMAC_SECRET = b"preprint-test-secret"
+    iss.TEST_ORCID_WHITELIST = frozenset({TEST_ORCID})
+    subs = tmp / "subs"
+    iss.SUBMISSIONS_ROOT = subs
+    iss.QUEUE_DIR = subs / "queue"
+    iss.COUNTER_FILE = subs / ".counter"
+    iss.TEST_SUBMISSIONS_ROOT = subs / "test"
+    iss.TEST_QUEUE_DIR = subs / "test" / "queue"
+    iss._audit_append = lambda entry, test_mode=False: None
+    _n = iter(range(1791000000, 1791000100))
+    iss._allocate_test_sub_id = lambda: f"ICSAC-SUB-TEST-{next(_n)}"
+    client = TestClient(iss.app)
+    pdf_bytes = PAPER_PDF.read_bytes()
+
+    def post(data: dict, with_pdf: bool) -> tuple[int, dict]:
+        req = httpx.Request("POST", "http://intake/api/submit", data=data,
+                            files={"pdf": ("paper.pdf", pdf_bytes, "application/pdf")} if with_pdf else None)
+        body = req.read()
+        ts = str(int(time.time()))
+        sig = hmac.new(iss.HMAC_SECRET, f"{ts}.".encode() + body, hashlib.sha256).hexdigest()
+        resp = client.post("/api/submit", content=body, headers={
+            "content-type": req.headers["content-type"], "x-icsac-signature": f"sha256={sig}",
+            "x-icsac-timestamp": ts, "x-icsac-auth-orcid": TEST_ORCID, "x-icsac-test-tier": "t2"})
+        try:
+            return resp.status_code, resp.json()
+        except Exception:
+            return resp.status_code, {"raw": resp.text[:300]}
+
+    base = {"name": "Test Author", "email": "author@example.com", "orcid": TEST_ORCID, "coi": "true",
+            "exclusivity_acknowledged": "true", "code_data_available": "no"}
+    upload = dict(base, mode="upload", deposit_consent="true", title="A test paper on boundaries",
+                  abstract="An abstract long enough to pass the fifty-character minimum for uploads.",
+                  keywords="testing", license="cc-by-4.0", resource_type="preprint", publication_date="2026-09-28",
+                  subject="", funding="", creators=json.dumps([{"name": "Test Author", "orcid": TEST_ORCID}]),
+                  related_identifiers="[]")
+
+    def record(resp: dict) -> dict:
+        sid = resp.get("sub_id") or ""
+        return json.loads((subs / "test" / sid / "submission.json").read_text()) if sid else {}
+
+    def dirs() -> int:
+        return len([p for p in (subs / "test").iterdir() if p.name.startswith("ICSAC-SUB-TEST-")]) if (subs / "test").exists() else 0
+
+    st, resp = post(dict(upload, preprint_doi="https://doi.org/10.20944/preprints202607.0102.v1"), True)
+    rec = record(resp)
+    check(st in (200, 202) and rec.get("preprint_doi") == "10.20944/preprints202607.0102.v1"
+          and (rec.get("preprint_meta") or {}).get("verified"), f"upload + Preprints.org DOI: stored, verified ({st})")
+    check(rec.get("form", {}).get("exclusivity_acknowledged") is True, "upload: the exclusivity confirmation is stored")
+    up_rec = rec
+    st, resp = post(dict(upload), True)
+    check(st in (200, 202) and record(resp).get("preprint_doi") is None, "upload without a preprint: none stored")
+    before = dirs()
+    st, resp = post(dict(upload, preprint_doi="10.1038/nature12373"), True)
+    check(st == 422 and resp.get("error") == "preprint_doi_refused" and "Nature" in json.dumps(resp),
+          f"upload + a journal article DOI: refused with the journal named ({st})")
+    check(dirs() == before, "a refused preprint leaves no submission behind")
+    st, resp = post(dict(upload, preprint_doi="banana"), True)
+    check(st == 422 and resp.get("error") == "preprint_doi_invalid", f"upload + garbage: refused ({st})")
+    st, resp = post(dict(base, mode="doi", doi="10.5281/zenodo.1"), False)
+    check(st in (200, 202) and record(resp).get("form", {}).get("exclusivity_acknowledged") is True,
+          f"DOI route: the exclusivity confirmation is stored ({st})")
+    st, resp = post(dict(base, mode="doi", doi="10.1101/2024.09.14.613029"), False)
+    check(st == 422 and "submit/upload?preprint_doi=10.1101/2024.09.14.613029" in json.dumps(resp),
+          f"DOI route + a bioRxiv DOI: sent to the upload form with the DOI ({st})")
+
+    print("3. the deposit and the emails")
+    import crossref_deposit as cd
+    check(cd.preprint_doi(up_rec) == "10.20944/preprints202607.0102.v1", "the pipeline reads the upload route's preprint")
+    sub = dict(up_rec, sub_id="ICSAC-SUB-00099")
+    xml = cd.build_deposit_xml(sub, doi="10.67697/icsac.2026.099", landing_url="https://icsacinstitute.org/publications/t",
+                               pdf_url=None, content_type="journal-article")
+    cd.validate_xml(xml)
+    check(b'relationship-type="hasPreprint"' in xml and b"10.20944/preprints202607.0102.v1" in xml,
+          "upload + preprint: the deposit validates and links the preprint")
+    from intake import notify_author as na
+    pathlib.Path.home = classmethod(lambda cls: tmp / "home")
+    common = dict(to="author@example.com", title="T", author_name="A", verdict="accept", panel_report_md="",
+                  rqc_md="", tier=2, compaction_manifest={"_failure": "test"},
+                  approval_url="https://icsacinstitute.org/approve/?t=X", objection_deadline="October 5, 2026")
+    out = tmp / "home" / "icsac-submissions" / "test" / "_outbox"
+
+    def body(sid: str) -> str:
+        p = out / f"{sid}.eml"
+        if not p.exists():
+            return ""
+        m = _email.message_from_bytes(p.read_bytes(), policy=_policy.default)
+        return "".join(x.get_payload(decode=True).decode() for x in m.walk() if x.get_content_type() == "text/plain")
+
+    na.send_decision(sub_id="ICSAC-SUB-TEST-A", source="upload", source_ref="paper.pdf",
+                     preprint_doi="10.20944/preprints202607.0102.v1", exclusivity_confirmed=True, **common)
+    t = body("ICSAC-SUB-TEST-A")
+    check("keeps its own DOI (10.20944/preprints202607.0102.v1)" in t and "not under review elsewhere, as confirmed" in t
+          and "has no other DOI" not in t, "upload + preprint: the email names the preprint and the confirmation")
+    na.send_decision(sub_id="ICSAC-SUB-TEST-B", source="doi", source_ref="10.5281/zenodo.1",
+                     preprint_doi="10.5281/zenodo.1", exclusivity_confirmed=False, **common)
+    t = body("ICSAC-SUB-TEST-B")
+    check("keeps its own DOI" in t and "as confirmed at submission" not in t,
+          "a paper whose confirmation was never recorded is not told it confirmed")
+    pathlib.Path.home = real_home
+    src = (ROOT / "intake" / "apply_decision.py").read_text()
+    check("exclusivity_confirmed=form.get(\"exclusivity_acknowledged\") is True" in src, "the accept passes the recorded confirmation")
+finally:
+    pathlib.Path.home = real_home
+    shutil.rmtree(tmp, ignore_errors=True)
+
+print(f"\n{'ALL GREEN' if not failures else f'{len(failures)} FAILURE(S)'}")
+sys.exit(1 if failures else 0)

preprint_check.py

@@ -0,0 +1,134 @@
+"""Preprint DOI check (policy 2026-09-28).
+
+A paper already posted as a preprint may be submitted: the preprint keeps its
+DOI and the Persistence article gets its own under the Institute prefix, linked
+to it (Crossref hasPreprint). Any preprint server counts. The check asks doi.org
+which registration agency holds the DOI, then reads that agency's record:
+
+  Crossref   preprints are type "posted-content", subtype "preprint", on every
+             server checked on 2026-09-28 (Preprints.org, bioRxiv/medRxiv, SSRN,
+             Research Square, ChemRxiv, TechRxiv, Authorea, the OSF family,
+             EarthArXiv, engrXiv, Cambridge Open Engage, ESSOAr, JMIR Preprints,
+             Beilstein Archives). A journal article, proceedings paper or book
+             record is REFUSED: that work is published. So is a preprint whose
+             record says it "is-preprint-of" a published version. Some prefixes
+             (SSRN, JMIR, SciELO, Wiley) hold both, so the record decides, not
+             the prefix.
+  DataCite   arXiv and Zenodo preprints are resourceTypeGeneral "Preprint";
+             other types are accepted and flagged for the curation team
+             (Zenodo authors often file a preprint as "Journal article").
+  other RAs  accepted and flagged (the record was not read).
+
+A DOI that does not exist is refused. A registry that cannot be reached fails
+OPEN: accepted, marked unverified and flagged; the curation team checks by hand.
+"""
+from __future__ import annotations
+
+import json
+import re
+import urllib.error
+import urllib.parse
+import urllib.request
+
+USER_AGENT = "ICSAC-intake/1.0 (mailto:help@icsacinstitute.org)"
+_ARXIV_ID = re.compile(r"^(?:arxiv:)?(?:10\.48550/arxiv\.)?(\d{4}\.\d{4,5})(?:v\d+)?$", re.I)
+_DOI = re.compile(r"^10\.\d{4,9}/\S+$")
+PUBLISHED_TYPES = {
+    "journal-article", "proceedings-article", "book-chapter", "book", "monograph",
+    "edited-book", "reference-book", "book-part", "book-section", "book-track",
+    "reference-entry",
+}
+
+
+class Refused(ValueError):
+    """The DOI cannot be accepted as a preprint. The message is author-facing."""
+
+
+def normalize(raw) -> str | None:
+    """A bare DOI (arXiv ids become arXiv's 10.48550 DOI, version dropped), or None."""
+    s = (raw or "").strip() if isinstance(raw, str) else ""
+    s = re.sub(r"^(?:https?://(?:dx\.)?doi\.org/|doi:\s*)", "", s, flags=re.I).strip()
+    m = _ARXIV_ID.match(s)
+    if m:
+        return f"10.48550/arXiv.{m.group(1)}"
+    return s if _DOI.match(s) and len(s) <= 300 else None
+
+
+def _get(url: str, timeout: float):
+    req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT,
+                                               "Accept": "application/json"})
+    with urllib.request.urlopen(req, timeout=timeout) as r:
+        return json.load(r)
+
+
+def _words(text) -> set[str]:
+    return set(re.findall(r"[a-z0-9]{4,}", (text or "").lower()))
+
+
+def title_differs(a, b) -> bool:
+    """True when two titles share under 40 % of their longer words."""
+    wa, wb = _words(a), _words(b)
+    if not wa or not wb:
+        return False
+    return len(wa & wb) / len(wa | wb) < 0.4
+
+
+def check(doi: str, *, title: str = "", timeout: float = 10.0) -> dict:
+    """Read the preprint's record. Returns {doi, ra, type, server, title,
+    verified, flags}; raises Refused when the DOI cannot be a preprint."""
+    out: dict = {"doi": doi, "ra": None, "type": None, "server": None, "title": None,
+                 "verified": False, "flags": []}
+    q = urllib.parse.quote(doi, safe="/")
+    try:
+        rows = _get(f"https://doi.org/ra/{q}", timeout)
+    except Exception as exc:
+        out["flags"].append(f"the DOI registries could not be reached ({type(exc).__name__}); "
+                            f"check the preprint by hand")
+        return out
+    row = rows[0] if isinstance(rows, list) and rows else {}
+    ra = row.get("RA") or ""
+    if not ra:
+        raise Refused(f"The preprint DOI {doi} does not exist. Check it, or leave the field empty.")
+    out["ra"] = ra
+    try:
+        if ra == "Crossref":
+            m = _get(f"https://api.crossref.org/works/{q}", timeout)["message"]
+            typ = m.get("type") or ""
+            sub = m.get("subtype")
+            out["type"] = f"{typ}/{sub}" if sub else typ
+            out["title"] = (m.get("title") or [None])[0]
+            inst = m.get("institution") or []
+            out["server"] = (inst[0].get("name") if inst else None) or m.get("publisher")
+            if typ in PUBLISHED_TYPES:
+                where = (m.get("container-title") or [None])[0] or m.get("publisher") or "a journal"
+                raise Refused(
+                    f"The DOI {doi} is a published {typ.replace('-', ' ')} ({where}), not a preprint. "
+                    f"ICSAC publishes work that has not been published elsewhere; a preprint of it is fine.")
+            published = (m.get("relation") or {}).get("is-preprint-of") or []
+            if published:
+                raise Refused(
+                    f"The preprint {doi} is already linked to a published version "
+                    f"({published[0].get('id')}). ICSAC publishes work that has not been published elsewhere.")
+            if typ != "posted-content":
+                out["flags"].append(f"Crossref type {typ}, not a preprint record")
+            elif sub and sub != "preprint":
+                out["flags"].append(f"posted content of subtype {sub}, not a preprint")
+        elif ra == "DataCite":
+            a = _get(f"https://api.datacite.org/dois/{q}", timeout)["data"]["attributes"]
+            rtg = (a.get("types") or {}).get("resourceTypeGeneral")
+            pub = a.get("publisher")
+            out["type"] = rtg
+            out["server"] = pub.get("name") if isinstance(pub, dict) else pub
+            out["title"] = ((a.get("titles") or [{}])[0]).get("title")
+            if rtg != "Preprint":
+                out["flags"].append(f"DataCite type {rtg}, not marked as a preprint")
+        else:
+            out["flags"].append(f"registered with {ra}; the record was not read")
+        out["verified"] = ra in ("Crossref", "DataCite")
+    except Refused:
+        raise
+    except Exception as exc:
+        out["flags"].append(f"the {ra} record could not be read ({type(exc).__name__}); check it by hand")
+    if title and out["title"] and title_differs(title, out["title"]):
+        out["flags"].append(f"the preprint's title differs: {out['title'][:150]!r}")
+    return out