Migrate existing PMLR submission tracks with atomic provenance checks

Reviewed-on: #9

docs/data-sources.md

@@ -58,6 +58,36 @@ PMLR 的同一出版 ID 可能对应多份 Git 元数据，例如 CoRL 2023 的
 
 PeerRead 采集仅选择作者公开发布的 ICLR 2017 投稿元数据，固定到源仓库 commit。只依据明确的布尔 `accepted` 标签形成 accepted/rejected；空标签为 unknown，不使用其他数据分区中的 probably-rejected 推断。不会保存 reviews 或 parsed_pdfs 正文。
 
+### 其他会议的公开拒稿
+
+范围是逐会议、逐年份寻找已公开的拒稿投稿，不能只补录已发表名单。ICML、NeurIPS 部分年份由作者选择是否公开拒稿，因此公开子集不能用于推算完整拒稿数或录用率。其他会议没有在某个镜像中出现 Reject，也不代表该会议没有公开拒稿；需要继续核对官方公开政策、原平台与其他有明确标签的公开快照。
+
+`collect public-openreview` 读取 [PaperCopilot 的公开元数据镜像](https://github.com/papercopilot/openreview/tree/41ee277825d685bd78e88e48b69da6aead9a69bb)，固定版本并核验每个文件的 Git blob 哈希。默认盘点该版本截至 2026 年的非 ICLR 文件；ICLR 默认沿用独立的公开原始投稿/决定快照，需要补查时可显式指定 `--family iclr`。只保留明确的 Reject、Desk rejected、Withdrawn 标签，不根据 Submitted、Active、Pending、TBD、分数或录用名单缺席推断拒稿。
+
+```sh
+paper-atlas collect public-openreview --out /data/paper-index/collected/public-openreview-v1
+paper-atlas audit public-openreview --db /data/paper-index/papers.sqlite \
+  --inventory /data/paper-index/collected/public-openreview-v1/public-openreview-inventory.json \
+  --out /data/paper-index/collected/public-openreview-preimport.json
+```
+
+输出按会议、届次及 main、position、datasets_benchmarks 分文件。workshop 默认进入单独的 held 清单；有明确 workshop 名称时可通过 `--include-workshops` 单列采集，不合并到主会。Retraction、未映射的轨道或负面标签也保留在 `held-records.jsonl`，不强改为 Reject 或 Withdrawn。当前决定表没有独立 retracted 状态，因此撤回发表记录仍需后续专门建模。
+
+该来源明确标记为 `curated_public_status_assertion`：保留源 commit、文件哈希、行号、原标签和 forum ID，不声称已核验原始 PC 签名。只保存标题、摘要、公开作者名、关键词、领域、索引与决定来源，不缓存镜像中的个人资料、邮箱、性别、评审内容或 PDF。所有输出清单均为 partial；盘点中的零条记录仅描述这个固定镜像。
+
+入库前审计核验文件哈希、数量、轨道和来源声明，并用真实 forum ID 对比同届同轨道已有投稿。存在不同决定时阻止整批导入，交由来源和时序核查；另一会议的后续录用属于独立投稿历史，不构成该次拒稿的冲突。同一 forum 的镜像不新增投稿事件，但论文实体的跨平台去重仍须在全量结束后核验。审计通过的报告包含带 SHA-256 的导入队列；应在单写入维护锁下及时串行导入，主库变化后重新审计。
+
+ICML 2026 另有[公开原始投稿快照](https://github.com/HemantKumar01/ICML-2026-data/blob/1a1b9d973a8dfbf254f740fb8100a2fcf36757ea/data/rejected.json)，包含 214 条投稿根记录。`collect icml-public-archive` 固定该 commit、文件长度及 Git blob 哈希，逐条要求 `readers` 包含 everyone、forum 等于自身 ID、域及投稿邀请属于主会、平台 `venueid` 明确为 `ICML.cc/2026/Conference/Rejected_Submission`。不能仅根据显示文案 Submitted to ICML 2026 推断拒稿。
+
+```sh
+paper-atlas collect icml-public-archive --out /data/paper-index/collected/icml-2026-public-archive
+paper-atlas audit public-openreview --db /data/paper-index/papers.sqlite \
+  --inventory /data/paper-index/collected/icml-2026-public-archive/public-openreview-inventory.json \
+  --out /data/paper-index/collected/icml-2026-public-preimport.json
+```
+
+这批证据标记为 `public_submission_venueid`，保留逐投稿许可及快照出处；它与整理者的 Reject 标签、原始签名决定分别表述。记录修改时间不当作决定日期，观察时间采用快照上界，不采用今天的导入时间。214 是这个公开 opt-in 子集的投稿数，不能当作会议全部拒稿数，也不能直接当作去重论文实体数。该采集器同样只提取元数据与标题摘要，丢弃 authorids、个人资料、评审和 PDF 字段。
+
 ```sh
 python3 -B -m paper_atlas collect peerread --cache /data/paper-index/raw/http-cache \
   --out /data/paper-index/collected
@@ -270,3 +300,13 @@ paper-atlas collect cvf --cache /data/paper-index/raw/http-cache \
 ```
 
 parser_version 2 的清单还保存实际遍历的 index_urls、入口错误及缺失摘要页面。补采结果经过清单与身份核验后才导入，不能因重试任务启动就标记缺口已解决。
+
+## PMLR 中的 ICML 多轨道论文
+
+ICML 2024 起的一个 PMLR 卷可以同时包含研究主轨道与 Position Paper，不能把整卷都标为 `main`。采集器 v3 按轨道输出 `pmlr-v<VOLUME>-main.jsonl`、`pmlr-v<VOLUME>-position.jsonl` 及各自 manifest；卷级 `pmlr-v<VOLUME>.inventory.json` 和目录清单的 `files` 数组列出实际文件。导入队列应展开 `files`，不能假定每卷只有一个 `file`。
+
+2025、2026 年通过出版社元数据的真实 OpenReview forum ID，匹配官方完整节目单中的 `Conference` / `Position_Paper_Track` 来源组；忽略负编号生成的口头报告副本，遇到冲突、缺少身份或未匹配的发表记录停止发布结果。每条记录保存节目单 URL、SHA-256、事件 ID、关联身份及轨道证据。
+
+2024 年的官方节目单把两轨都放在 `Conference`，但提供 PMLR 论文自链接。该年仅采用组织方在 [2025 Position CFP](https://icml.cc/Conferences/2025/CallForPositionPapers) 明确链接的 [2024 Position 论文标题筛选](https://icml.cc/virtual/2024/papers.html?filter=titles&search=position%3A) 规则；同时保存规则页面的哈希。这个规则不用于其他会议或年份。官方节目单将 `guyard24a` 链接误写为 `theo24a`：只对这条已核查的映射额外要求标题精确一致、至少两名公开作者重合，不进行全库模糊匹配。
+
+导入与身份回填拒绝 ICML 2024 起 parser_version<3 或没有官方轨道证据的旧文件。已有主库中的错误轨道需要在独占维护锁内迁移原投稿事件、保留决定与来源关联；仅重新导入拆分文件会留下旧的主轨道事件，不能作为修复方式。完成迁移前，不把新文件直接导入现有库。

docs/operations.md

@@ -47,3 +47,18 @@ paper-atlas coverage --db /data/paper-index/papers.sqlite \
 manifest 和采集器状态清单用于确认文件是否齐全及是否缺摘要；`ingest_runs` 和协调器日志用于确认主库导入是否成功。官方程序、书目目录和公开投稿快照分别保留，未出现在某份清单不等于被拒稿；第三方快照也不能代表平台当前全量状态。代码测试成功不能代替最终全库去重、完整性和逐届覆盖核验。
 
 当前代码与合并状态以 [Pull Requests](https://gitea.entroia.com/Zayden-Jung/paper-atlas/pulls) 为准；CI 结果以对应提交的 Actions 检查为准。工作分支交付后等待用户手动合并。合并代码不自动部署、不自动启动采集，也不代表数据批次已完成。
+## 修正已有 PMLR 投稿的轨道
+
+先用采集器 v3 把 ICML 多轨道卷采集到新目录，保留旧文件。以下命令默认只读生成迁移计划；`--inventory` 可重复指定，以完整卷为单位核对，不能只提供 Position 子集：
+
+```sh
+paper-atlas maintain pmlr-tracks --db /data/paper-index/papers.sqlite \
+  --inventory /data/paper-index/collected/pmlr-tracks-v3/pmlr-v235.inventory.json \
+  --inventory /data/paper-index/collected/pmlr-tracks-v3/pmlr-v267.inventory.json \
+  --inventory /data/paper-index/collected/pmlr-tracks-v3/pmlr-v306.inventory.json \
+  --out /data/paper-index/collected/pmlr-track-plan.json
+```
+
+检查报告后，在同一命令添加 `--apply` 并使用新的 `--out` 路径执行迁移。计划使用共享维护锁，实际迁移使用独占锁；已有 arXiv 类别修复或其他写入运行时会拒绝启动，不会停止或重启它们。一次事务内核对整个来源集合、原始题目/摘要/作者、forum 身份、已有事件及目标冲突，更新原投稿的 edition 和身份归属，回填官方轨道出处及逐轨道覆盖数，最后复查轨道与外键后提交。论文、来源、投稿、决定的 ID 及决定观察均保留，不重新创建事件、不做实体合并、不改论文正文元数据或标签。
+
+报告记录每个迁移事件的旧 edition、原身份、完整决定观察及新轨道证据。`planned` 是未执行的计划；`complete` 才表示事务已提交；`error` 表示执行失败。若进程在提交前后中断而报告停在 `prepared`，先对照主库和报告核查，不根据日志缺失自行重启任务。新路径重跑会重新核对当前数据库，已正确的轨道不会再次迁移。共享多个来源、重复目标事件或冲突决定会停止迁移，需另行核对，不能直接删掉其中一条。

docs/usage.md

@@ -14,10 +14,15 @@ paper-atlas collect cvf --cache /data/paper-index/raw/http-cache --out /data/pap
 paper-atlas collect ecva --cache /data/paper-index/raw/http-cache --out /data/paper-index/collected
 paper-atlas catalogue dblp --help
 paper-atlas audit iclr --help
+paper-atlas collect public-openreview --help
+paper-atlas collect icml-public-archive --help
+paper-atlas audit public-openreview --help
 ```
 
 采集输出 JSONL、manifest 和逐卷或逐年 catalogue。逐项检查 blocked、partial、缺失摘要及目标会议归属。DBLP 候选目录还需核验届次和 track，不能直接当作主会录用名单。
 
+其他会议的公开拒稿使用 `collect public-openreview`，再以 `audit public-openreview` 核对既有投稿身份和决定冲突。来源版本、逐文件状态盘点和未解决记录分别保存在 inventory 与 held 清单中，具体命令见[数据源说明](data-sources.md#其他会议的公开拒稿)。这类公开子集不能当作会议完整投稿名单。
+
 ## 3. 串行导入
 
 首次 arXiv 导入使用 `paper-atlas import arxiv --db ... --archive ...`。现有数据库合并已核验的会议文件使用：

paper_atlas/audits/public_openreview.py

@@ -0,0 +1,107 @@
+"""Check curated negative assertions against existing submission decisions."""
+import argparse
+from contextlib import closing
+import hashlib
+import json
+from pathlib import Path
+import sqlite3
+
+from paper_atlas.collectors.public_openreview import SOURCE, REVISION
+from paper_atlas.collectors.icml_public_archive import SOURCE as ARCHIVE_SOURCE, REVISION as ARCHIVE_REVISION
+from paper_atlas.db.submissions import timestamp
+
+
+def audit(db, inventory, out):
+    inventory = Path(inventory).resolve()
+    result = json.loads(inventory.read_text())
+    specs = {SOURCE: (REVISION, 'curated_public_status_assertion'),
+             ARCHIVE_SOURCE: (ARCHIVE_REVISION, 'public_submission_venueid')}
+    source = result.get('source')
+    if source not in specs or result.get('revision') != specs[source][0]:
+        raise ValueError('Unsupported public source inventory')
+    revision, evidence_kind = specs[source]
+    db = Path(db).resolve()
+    if not db.is_file():
+        raise ValueError('Existing database required for identity and status audit')
+    conflicts = []
+    seen = set()
+    checked = 0
+    existing_attempts = 0
+    files = []
+    checksums = {}
+    with closing(sqlite3.connect(db.as_uri() + '?mode=ro', uri=True)) as c:
+        c.execute('BEGIN')
+        for manifest in result['files']:
+            path = Path(manifest['file'])
+            if hashlib.sha256(path.read_bytes()).hexdigest() != manifest['file_sha256']:
+                raise ValueError('Collected file differs from pinned inventory')
+            if json.loads(path.with_suffix('.manifest.json').read_text()) != manifest:
+                raise ValueError('Collected manifest differs from inventory')
+            count = 0
+            for line in path.read_text().splitlines():
+                row = json.loads(line)
+                if any(row[key] != manifest[key] for key in ('source', 'venue', 'year', 'track')):
+                    raise ValueError('Record edition differs from inventory')
+                meta = row['metadata']
+                if meta.get('evidence_kind') != evidence_kind or meta.get('source_revision') != revision:
+                    raise ValueError('Missing explicit public source provenance')
+                if meta.get('original_decision_signature_verified') is not False:
+                    raise ValueError('Curated metadata must not claim original signatures')
+                if row['status'] not in ('rejected', 'desk_rejected', 'withdrawn'):
+                    raise ValueError('Unsupported negative assertion')
+                forum = row['identifiers']['openreview']
+                if forum in seen:
+                    raise ValueError('Repeated forum in normalized inventory')
+                seen.add(forum)
+                existing = c.execute('''SELECT s.id FROM submission_identifiers i
+                    JOIN submissions s ON s.id=i.submission_id JOIN editions e ON e.id=i.edition_id
+                    JOIN venues v ON v.id=e.venue_id WHERE v.acronym=? AND e.year=? AND e.track=?
+                    AND i.scheme='openreview' AND i.value=?''',
+                    (row['venue'], row['year'], row['track'], forum)).fetchall()
+                if len(existing) > 1:
+                    raise ValueError('Multiple canonical attempts for one forum')
+                if existing:
+                    existing_attempts += 1
+                    latest = c.execute('''SELECT status FROM decisions WHERE submission_id=?
+                        ORDER BY observed_at DESC,COALESCE(decision_date,'') DESC,id DESC LIMIT 1''', existing[0]).fetchone()
+                    if latest and latest[0] not in ('unknown', 'under_review', row['status']):
+                        conflicts.append({'forum': forum, 'venue': row['venue'], 'year': row['year'],
+                                          'track': row['track'], 'submission_id': existing[0][0],
+                                          'existing_status': latest[0], 'curated_status': row['status']})
+                count += 1
+            if count != manifest['count']:
+                raise ValueError('Collected record count differs from inventory')
+            checked += count
+            files.append(str(path))
+            checksums[str(path)] = manifest['file_sha256']
+    if checked != result['normalized_records']:
+        raise ValueError('Normalized inventory total differs from checked records')
+    report = {'source': source, 'revision': revision, 'database': str(db),
+              'inventory': str(inventory), 'checked_at': timestamp(), 'records_checked': checked,
+              'existing_same_edition_attempts': existing_attempts, 'conflicts': conflicts,
+              'status': 'blocked' if conflicts else 'ready_for_serial_import',
+              'files': [] if conflicts else files, 'file_sha256': {} if conflicts else checksums,
+              'note': 'Read-only preflight. Other-conference acceptances remain separate history; '
+                      'this is not an entity deduplication or complete rejection-coverage proof.'}
+    out = Path(out)
+    out.parent.mkdir(parents=True, exist_ok=True)
+    part = out.with_suffix('.part')
+    part.write_text(json.dumps(report, ensure_ascii=False, indent=2) + '\n')
+    part.replace(out)
+    print(json.dumps(report), flush=True)
+    return report
+
+
+def main(argv=None, *, prog=None):
+    parser = argparse.ArgumentParser(prog=prog)
+    parser.add_argument('--db', required=True)
+    parser.add_argument('--inventory', required=True)
+    parser.add_argument('--out', required=True)
+    args = parser.parse_args(argv)
+    report = audit(args.db, args.inventory, args.out)
+    if report['status'] == 'blocked':
+        raise SystemExit('Conflicting existing decisions require review before import')
+
+
+if __name__ == '__main__':
+    main()

paper_atlas/cli.py

@@ -13,6 +13,8 @@ COMMANDS = {
     ('collect','acl'): ('paper_atlas.collectors.acl', {}, 'ACL Anthology XML'),
     ('collect','iclr-program'): ('paper_atlas.collectors.iclr', {}, 'ICLR 官方公开程序'),
     ('collect','iclr-snapshot'): ('paper_atlas.collectors.openreview', {}, 'ICLR 公开投稿与决定快照'),
+    ('collect','public-openreview'): ('paper_atlas.collectors.public_openreview', {}, '核验其他会议的公开拒稿及撤稿标签'),
+    ('collect','icml-public-archive'): ('paper_atlas.collectors.icml_public_archive', {}, 'ICML 2026 公开拒稿原始投稿快照'),
     ('collect','peerread'): ('paper_atlas.collectors.peerread', {}, 'PeerRead 历史决定补充'),
     ('collect','colm'): ('paper_atlas.collectors.proceedings', {'source':'colm'}, 'COLM 官方论文集'),
     ('collect','neurips'): ('paper_atlas.collectors.proceedings', {'source':'neurips'}, 'NeurIPS 官方论文集'),
@@ -25,9 +27,11 @@ COMMANDS = {
     ('maintain','merge'): ('paper_atlas.ingestion.maintenance', {}, '协调单写入合并与恢复'),
     ('maintain','backfill'): ('paper_atlas.ingestion.backfill', {}, '回填既有投稿的来源身份'),
     ('maintain','arxiv-tags'): ('paper_atlas.ingestion.arxiv_tags', {}, '恢复逐来源记录的 arXiv 类别证据'),
+    ('maintain','pmlr-tracks'): ('paper_atlas.ingestion.pmlr_tracks', {}, '核对并迁移既有 PMLR 投稿轨道'),
     ('audit','iclr'): ('paper_atlas.audits.iclr', {}, '核对 ICLR 快照与同届官方程序'),
     ('audit','iclr-legacy'): ('paper_atlas.audits.iclr_legacy', {}, '独立核验 ICLR 2020 旧目录投稿身份'),
     ('audit','arxiv'): ('paper_atlas.audits.arxiv', {}, '核验 arXiv 原始编号、重复行与 ZIP 完整性'),
+    ('audit','public-openreview'): ('paper_atlas.audits.public_openreview', {}, '核对公开负面决定与已有投稿历史'),
 }
 
 

paper_atlas/collectors/icml_public_archive.py

@@ -0,0 +1,125 @@
+"""ICML 2026 public submission roots with explicit Rejected_Submission IDs."""
+import argparse
+import fcntl
+import hashlib
+import json
+from pathlib import Path
+import re
+import shutil
+import tempfile
+
+from paper_atlas.collectors.openreview import date_ms, invitations, public, value
+from paper_atlas.collectors.public_openreview import download
+from paper_atlas.db.submissions import timestamp
+
+
+SOURCE = 'openreview_public_archive'
+REVISION = '1a1b9d973a8dfbf254f740fb8100a2fcf36757ea'
+AS_OF = '2026-08-08T12:03:26.000Z'
+URL = 'https://raw.githubusercontent.com/HemantKumar01/ICML-2026-data/' + REVISION + '/data/rejected.json'
+BLOB_SHA1 = 'cebdcddffe3eb2145d32d328db89a1626e3f85eb'
+BYTES = 1600856
+EXPECTED = 214
+DOMAIN = 'ICML.cc/2026/Conference'
+VENUEID = DOMAIN + '/Rejected_Submission'
+
+
+def normalize(note):
+    forum = note.get('id')
+    if not isinstance(forum, str) or not re.fullmatch(r'[A-Za-z0-9_-]{8,32}', forum) or note.get('forum') != forum:
+        raise ValueError('Not a public submission root with a real forum identity')
+    if not public(note) or note.get('ddate'):
+        raise ValueError('Submission is not explicitly public and live in the archive')
+    if note.get('domain') != DOMAIN or DOMAIN + '/-/Submission' not in invitations(note):
+        raise ValueError('Submission is outside the ICML 2026 conference track')
+    content = note.get('content') or {}
+    if value(content, 'venueid') != VENUEID:
+        raise ValueError('Missing explicit platform Rejected_Submission classification')
+    title, abstract = value(content, 'title'), value(content, 'abstract')
+    if not isinstance(title, str) or not title.strip():
+        raise ValueError('Submission lacks a title')
+    if abstract is not None and not isinstance(abstract, str):
+        raise ValueError('Submission abstract must be text')
+    authors = value(content, 'authors', []) or []
+    if not isinstance(authors, list):
+        raise ValueError('Unexpected public author format')
+    authors = [a.strip() for a in authors if isinstance(a, str) and a.strip() and '@' not in a
+               and not re.fullmatch(r'anonymous(?: authors?)?', a.strip(), re.I)]
+    tags = []
+    for field in ('keywords', 'primary_area'):
+        raw = value(content, field)
+        terms = raw if isinstance(raw, list) else re.split(r'[,;]', raw) if field == 'keywords' and isinstance(raw, str) else [raw]
+        tags.extend({'namespace': SOURCE + ':' + field, 'value': term.strip()}
+                    for term in terms if isinstance(term, str) and term.strip())
+    url = 'https://openreview.net/forum?id=' + forum
+    return {'source': SOURCE, 'external_id': 'icml/2026/' + forum, 'url': url,
+            'title': ' '.join(title.split()), 'abstract': abstract.strip() if abstract else None,
+            'authors': authors, 'identifiers': {'openreview': forum},
+            'submission_identifiers': [{'scheme': 'openreview', 'value': forum}],
+            'venue': 'ICML', 'year': 2026, 'track': 'main', 'status': 'rejected',
+            'raw_decision': VENUEID, 'evidence_url': url, 'observed_at': AS_OF,
+            'submitted_at': date_ms(note.get('cdate') or note.get('tcdate')), 'tags': tags,
+            'metadata': {'source_revision': REVISION, 'source_blob_sha1': BLOB_SHA1,
+                         'source_file': URL, 'original_forum': forum, 'license': note.get('license'),
+                         'venue': value(content, 'venue'), 'venueid': VENUEID,
+                         'public_readers_verified': True, 'original_invitations': invitations(note),
+                         'source_note_modified_at': date_ms(note.get('mdate')),
+                         'evidence_kind': 'public_submission_venueid',
+                         'original_decision_signature_verified': False,
+                         'source_snapshot_upper_bound': AS_OF}}
+
+
+def collect(out, fetcher=None):
+    fetcher = fetcher or download
+    out = Path(out)
+    out.mkdir(parents=True, exist_ok=True)
+    with (out / 'collect.lock').open('a') as lock:
+        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        if (out / 'public-openreview-inventory.json').exists():
+            raise ValueError('Completed inventory exists; select a new output directory')
+        if shutil.disk_usage(out).free < 8 * 1024**3:
+            raise RuntimeError('Less than 8 GiB free before collection')
+        data = fetcher(URL, BYTES)
+        blob = hashlib.sha1(b'blob ' + str(len(data)).encode() + b'\0' + data).hexdigest()
+        if len(data) != BYTES or blob != BLOB_SHA1:
+            raise ValueError('Public archive checksum mismatch')
+        archive = json.loads(data)
+        notes = archive['notes']
+        if archive.get('count') != EXPECTED or not isinstance(notes, list) or len(notes) != EXPECTED:
+            raise ValueError('Public archive inventory count mismatch')
+        records = [normalize(note) for note in notes]
+        if len({row['identifiers']['openreview'] for row in records}) != EXPECTED:
+            raise ValueError('Repeated submission forum in public archive')
+        records.sort(key=lambda row: row['external_id'])
+        with tempfile.TemporaryDirectory(prefix='.icml-public-', dir=out) as temporary:
+            stage = Path(temporary)
+            path = stage / 'public-openreview-icml-2026-main.jsonl'
+            path.write_text(''.join(json.dumps(row, ensure_ascii=False) + '\n' for row in records))
+            manifest = {'source': SOURCE, 'parser_version': 1, 'venue': 'ICML', 'year': 2026,
+                        'track': 'main', 'count': EXPECTED, 'status': 'partial', 'list_url': URL,
+                        'file': str(out / path.name), 'file_sha256': hashlib.sha256(path.read_bytes()).hexdigest(),
+                        'source_revision': REVISION, 'missing_abstract_count': sum(not row['abstract'] for row in records),
+                        'evidence_kind': 'public_submission_venueid',
+                        'note': 'Public opt-in subset; explicit platform venueid, not original signed PC notes.'}
+            path.with_suffix('.manifest.json').write_text(json.dumps(manifest, indent=2) + '\n')
+            result = {'source': SOURCE, 'revision': REVISION, 'finished_at': timestamp(),
+                      'input_rows': EXPECTED, 'normalized_records': EXPECTED, 'files': [manifest],
+                      'source_files': [{'url': URL, 'git_blob_sha1': blob, 'input_bytes': len(data)}],
+                      'note': 'Snapshot preserves per-note public visibility and licenses; no reviews, authorids or PDFs retained.'}
+            inventory = stage / 'public-openreview-inventory.json'
+            inventory.write_text(json.dumps(result, indent=2) + '\n')
+            for child in sorted(stage.iterdir(), key=lambda child: child == inventory):
+                child.replace(out / child.name)
+            print(json.dumps(result), flush=True)
+            return result
+
+
+def main(argv=None, *, prog=None):
+    parser = argparse.ArgumentParser(prog=prog)
+    parser.add_argument('--out', required=True)
+    args = parser.parse_args(argv)
+    collect(args.out)
+
+
+if __name__ == '__main__':
+    main()

paper_atlas/collectors/pmlr.py

@@ -1,5 +1,7 @@
 """Collect official PMLR metadata, never the PDFs linked by the metadata."""
 import argparse
+from collections import defaultdict
+import hashlib
 from paper_atlas.config import DEFAULT_REGISTRY
 import io
 import json
@@ -8,6 +10,7 @@ import os
 import subprocess
 import tarfile
 import time
+import tempfile
 from pathlib import Path
 
 import yaml
@@ -37,6 +40,7 @@ def venue_key(shortname):
 
 
 from paper_atlas.io.http import fetch
+from paper_atlas.collectors.pmlr_tracks import collect_track_evidence, record_track
 
 
 def metadata_texts(volume,cache):
@@ -98,57 +102,88 @@ def collect_volume(volume,cache,out,allowed=None):
     track = 'workshop' if 'workshop' in title.lower() else 'main'
     path = Path(out)/f'pmlr-v{volume}.jsonl'
     path.parent.mkdir(parents=True,exist_ok=True)
-    part = path.with_suffix('.part')
+    mixed_icml = venue == 'ICML' and year >= 2024
+    mapping, track_program = collect_track_evidence(year, cache) if mixed_icml else ({}, None)
     records = {}
     duplicate_posts = 0
-    with part.open('w') as output:
-        for member_name,content in metadata_texts(volume,cache):
-            parts = re.split(r'(?m)^---\s*$',content,maxsplit=2)
-            if len(parts)<3 or parts[0].strip():
-                raise RuntimeError('Missing YAML frontmatter: '+member_name)
-            row = load_metadata(parts[1])
-            if row.get('layout') not in ('inproceedings','article'):
-                continue
-            if not row.get('title'):
-                raise ValueError('Paper metadata has no title: '+member_name)
-            if not row.get('id'):
-                raise ValueError('Paper metadata has no publication ID: '+member_name)
-            external = f"v{volume}/{row['id']}"
-            ids = {}
-            for key in ('doi','openreview','arxiv'):
-                value = row.get(key)
-                if value:
-                    value = str(value)
-                    if key=='arxiv':
-                        value = re.sub(r'^https?://arxiv.org/(abs|pdf)/','',value).removesuffix('.pdf')
-                    if key=='openreview':
-                        value = value.split('id=')[-1].split('&')[0]
-                    ids[key] = value
-            record = {'source':'pmlr','external_id':external,'url':f"https://proceedings.mlr.press/v{volume}/{row.get('id')}.html",'title':row['title'],'abstract':row.get('abstract'),'authors':row.get('author',[]),'identifiers':ids,'venue':venue,'year':year,'track':track,'status':'accepted','list_url':f'https://proceedings.mlr.press/v{volume}/','metadata':{key:row[key] for key in ('id','volume','date','page','doi','openreview','arxiv','publisher') if key in row}}
-            version={'path':member_name,'date':row.get('date')}
-            if external in records:
-                previous=records[external]
-                if ({k:v for k,v in previous.items() if k!='metadata'} != {k:v for k,v in record.items() if k!='metadata'}
-                    or {k:v for k,v in previous['metadata'].items() if k not in ('date','source_versions')} != {k:v for k,v in record['metadata'].items() if k!='date'}):
-                    raise ValueError('Conflicting metadata posts share a publication ID: '+external)
-                previous['metadata']['source_versions'].append(version)
-                if str(row.get('date') or '')>str(previous['metadata'].get('date') or ''):
-                    previous['metadata']['date']=row['date']
-                duplicate_posts+=1
-                continue
-            record['metadata']['source_versions']=[version]
-            records[external]=record
-        for record in records.values():
-            output.write(json.dumps(record,ensure_ascii=False,default=str)+'\n')
+    for member_name,content in metadata_texts(volume,cache):
+        parts = re.split(r'(?m)^---\s*$',content,maxsplit=2)
+        if len(parts)<3 or parts[0].strip():
+            raise RuntimeError('Missing YAML frontmatter: '+member_name)
+        row = load_metadata(parts[1])
+        if row.get('layout') not in ('inproceedings','article'):
+            continue
+        if not row.get('title'):
+            raise ValueError('Paper metadata has no title: '+member_name)
+        if not row.get('id'):
+            raise ValueError('Paper metadata has no publication ID: '+member_name)
+        external = f"v{volume}/{row['id']}"
+        ids = {}
+        for key in ('doi','openreview','arxiv'):
+            value = row.get(key)
+            if value:
+                value = str(value)
+                if key=='arxiv':
+                    value = re.sub(r'^https?://arxiv.org/(abs|pdf)/','',value).removesuffix('.pdf')
+                if key=='openreview':
+                    value = value.split('id=')[-1].split('&')[0]
+                ids[key] = value
+        record = {'source':'pmlr','external_id':external,'url':f"https://proceedings.mlr.press/v{volume}/{row.get('id')}.html",'title':row['title'],'abstract':row.get('abstract'),'authors':row.get('author',[]),'identifiers':ids,'venue':venue,'year':year,'track':track,'status':'accepted','list_url':f'https://proceedings.mlr.press/v{volume}/','metadata':{key:row[key] for key in ('id','volume','date','page','doi','openreview','arxiv','publisher') if key in row}}
+        version={'path':member_name,'date':row.get('date')}
+        if external in records:
+            previous=records[external]
+            if ({k:v for k,v in previous.items() if k!='metadata'} != {k:v for k,v in record.items() if k!='metadata'}
+                or {k:v for k,v in previous['metadata'].items() if k not in ('date','source_versions')} != {k:v for k,v in record['metadata'].items() if k!='date'}):
+                raise ValueError('Conflicting metadata posts share a publication ID: '+external)
+            previous['metadata']['source_versions'].append(version)
+            if str(row.get('date') or '')>str(previous['metadata'].get('date') or ''):
+                previous['metadata']['date']=row['date']
+            duplicate_posts+=1
+            continue
+        record['metadata']['source_versions']=[version]
+        records[external]=record
     count=len(records)
     missing_abstracts=sum(not (record.get('abstract') or '').strip() for record in records.values())
     if count==0:
-        part.unlink(missing_ok=True)
         raise RuntimeError('No papers parsed; refusing an empty successful collection')
-    part.replace(path)
-    result = {'source':'pmlr','parser_version':2,'venue':venue,'year':year,'track':track,'volume':volume,'count':count,'metadata_post_count':count+duplicate_posts,'duplicate_post_count':duplicate_posts,'missing_abstract_count':missing_abstracts,'list_url':f'https://proceedings.mlr.press/v{volume}/','file':str(path),'status':'partial' if missing_abstracts else 'collected'}
-    path.with_suffix('.manifest.json').write_text(json.dumps(result,indent=2))
-    print(json.dumps(result),flush=True)
+    grouped = defaultdict(list)
+    for record in records.values():
+        if mixed_icml:
+            record_track(record, mapping, track_program)
+        grouped[record['track']].append(record)
+    # Parse and verify the entire volume before replacing any published output.
+    manifests = []
+    with tempfile.TemporaryDirectory(prefix=f'.pmlr-v{volume}-', dir=path.parent) as temporary:
+        stage = Path(temporary)
+        for current_track, rows in sorted(grouped.items()):
+            filename = f'pmlr-v{volume}-{current_track}.jsonl' if mixed_icml else path.name
+            output = stage / filename
+            output.write_text(''.join(json.dumps(row, ensure_ascii=False, default=str)+'\n' for row in rows))
+            absent = sum(not (row.get('abstract') or '').strip() for row in rows)
+            manifest = {'source': 'pmlr', 'parser_version': 3, 'venue': venue, 'year': year,
+                        'track': current_track, 'volume': volume, 'count': len(rows),
+                        'missing_abstract_count': absent, 'list_url': f'https://proceedings.mlr.press/v{volume}/',
+                        'file': str(path.parent / filename), 'file_sha256': hashlib.sha256(output.read_bytes()).hexdigest(),
+                        'status': 'partial' if absent else 'collected'}
+            if track_program:
+                manifest['track_program'] = track_program
+            output.with_suffix('.manifest.json').write_text(json.dumps(manifest, indent=2)+'\n')
+            manifests.append(manifest)
+        result = {'source': 'pmlr', 'parser_version': 3, 'venue': venue, 'year': year,
+                  'track': next(iter(grouped)) if len(grouped) == 1 else 'mixed', 'volume': volume,
+                  'count': count, 'metadata_post_count': count + duplicate_posts,
+                  'duplicate_post_count': duplicate_posts, 'missing_abstract_count': missing_abstracts,
+                  'list_url': f'https://proceedings.mlr.press/v{volume}/', 'files': manifests,
+                  'status': 'partial' if missing_abstracts else 'collected'}
+        if len(manifests) == 1:
+            result['file'] = manifests[0]['file']
+        if track_program:
+            result['track_program'] = track_program
+        inventory = stage / f'pmlr-v{volume}.inventory.json'
+        inventory.write_text(json.dumps(result, indent=2)+'\n')
+        for child in sorted(stage.iterdir(), key=lambda child: child == inventory):
+            child.replace(path.parent / child.name)
+    print(json.dumps(result), flush=True)
     return result
 
 

paper_atlas/collectors/pmlr_tracks.py

@@ -0,0 +1,140 @@
+"""Resolve mixed ICML proceedings tracks from the official paper program."""
+from collections import Counter
+import hashlib
+import json
+import re
+from urllib.parse import parse_qs, urlsplit
+
+from paper_atlas.collectors.iclr import identifiers, source_number
+from paper_atlas.db.store import author_keys, normalized
+from paper_atlas.io.html import Document, clean
+from paper_atlas.io.http import fetch
+
+
+POSITION_LIST_POLICY = 'https://icml.cc/Conferences/2025/CallForPositionPapers'
+PUBLISHER_LINK_ALIASES = {'v235/theo24a': 'v235/guyard24a'}
+
+
+def position_list_policy(data):
+    """The organizers explicitly link this title filter as the 2024 position list."""
+    for node in Document(data.decode()).root.walk():
+        if node.tag != 'a' or clean(node.text()).casefold() != 'position papers':
+            continue
+        url = node.attrs.get('href', '')
+        parsed = urlsplit(url)
+        query = parse_qs(parsed.query)
+        if (parsed.netloc in ('', 'icml.cc') and parsed.path == '/virtual/2024/papers.html'
+                and query.get('filter') == ['titles'] and query.get('search') == ['position:']):
+            return {'url': POSITION_LIST_POLICY, 'sha256': hashlib.sha256(data).hexdigest(),
+                    'position_list_url': 'https://icml.cc' + parsed.path + '?' + parsed.query}
+    raise ValueError('Official policy no longer identifies the 2024 position-paper title list')
+
+
+def program_tracks(data, year, policy=None):
+    program = json.loads(data)
+    events = program.get('results') if isinstance(program, dict) else None
+    if (not isinstance(events, list) or program.get('count') != len(events)
+            or program.get('next') or program.get('previous')):
+        raise ValueError('Official ICML program must be a complete counted event list')
+    if len({str(row['id']) for row in events}) != len(events):
+        raise ValueError('Repeated event identity in official ICML program')
+    if year == 2024 and policy is None:
+        raise ValueError('The 2024 position-list policy is required')
+    mapping = {}
+    excluded = Counter()
+    for row in events:
+        number = source_number(row)
+        if number is not None and number < 0:
+            excluded['synthetic_oral_event'] += 1
+            continue
+        parsed = urlsplit(row.get('sourceurl') or '')
+        group = parse_qs(parsed.query).get('id', [])
+        groups = {f'ICML.cc/{year}/Conference': 'main',
+                  f'ICML.cc/{year}/Position_Paper_Track': 'position'}
+        if (parsed.netloc != 'openreview.net' or parsed.path != '/group'
+                or len(group) != 1 or group[0] not in groups):
+            excluded['other_source'] += 1
+            continue
+        if not str(row.get('decision') or '').casefold().startswith('accept'):
+            raise ValueError('Non-accepted event in official ICML paper program')
+        if year == 2024:
+            # The 2024 program flattens both tracks into Conference and links
+            # PMLR publication IDs instead of forums. Use only this year-specific,
+            # organizer-endorsed list rule, never a global title heuristic.
+            link = urlsplit(row.get('paper_pdf_url') or '')
+            if link.netloc != 'proceedings.mlr.press' or not re.fullmatch(r'/v\d+/[^/]+\.html', link.path):
+                raise ValueError('2024 ICML paper has no publisher self-link')
+            key = ('pmlr', link.path.lstrip('/').removesuffix('.html'))
+            track = 'position' if re.match(r'^Position(?: Paper)?:', row.get('name') or '', re.I) else 'main'
+            kind = 'official_2024_position_list'
+        else:
+            forum = identifiers(row).get('openreview')
+            if not forum:
+                raise ValueError('ICML paper has no real OpenReview self-identity')
+            key = ('openreview', forum)
+            track, kind = groups[group[0]], 'official_openreview_track_group'
+        evidence = {'track': track, 'kind': kind, 'event_ids': [row['id']],
+                    'source_group': group[0], 'program_title': row.get('name')}
+        if key[0] == 'pmlr' and key[1] in PUBLISHER_LINK_ALIASES:
+            evidence['publisher_link_alias'] = key[1]
+            evidence['_authors'] = [author['fullname'] for author in row.get('authors', [])]
+            key = ('pmlr', PUBLISHER_LINK_ALIASES[key[1]])
+        if key in mapping:
+            if mapping[key]['track'] != track:
+                raise ValueError('One paper identity has conflicting official tracks')
+            mapping[key]['event_ids'].append(row['id'])
+        else:
+            mapping[key] = evidence
+    if not mapping:
+        raise ValueError('Official ICML program contains no supported paper identities')
+    return mapping, {'event_count': len(events), 'paper_identity_count': len(mapping),
+                     'excluded_events': dict(excluded)}
+
+
+def collect_track_evidence(year, cache):
+    url = f'https://icml.cc/static/virtual/data/icml-{year}-orals-posters.json'
+    data = fetch(url, cache)
+    policy = position_list_policy(fetch(POSITION_LIST_POLICY, cache)) if year == 2024 else None
+    mapping, inventory = program_tracks(data, year, policy)
+    provenance = {'url': url, 'sha256': hashlib.sha256(data).hexdigest(), **inventory}
+    if policy:
+        provenance['position_list_policy'] = policy
+    return mapping, provenance
+
+
+def record_track(record, mapping, provenance):
+    key = ('pmlr', record['external_id']) if record['year'] == 2024 else ('openreview', record['identifiers'].get('openreview'))
+    if key not in mapping:
+        raise ValueError('Published ICML paper is absent from the verified official track inventory: ' + record['external_id'])
+    evidence = {**mapping[key], 'identity_scheme': key[0], 'identity_value': key[1],
+                'program_url': provenance['url'], 'program_sha256': provenance['sha256']}
+    if 'publisher_link_alias' in evidence:
+        overlap = author_keys(record.get('authors', [])) & author_keys(evidence.pop('_authors', []))
+        if normalized(record['title']) != normalized(evidence['program_title']) or len(overlap) < 2:
+            raise ValueError('Reviewed publisher-link alias lacks exact title and independent author corroboration')
+        evidence['corroborating_author_count'] = len(overlap)
+    if 'position_list_policy' in provenance:
+        evidence['position_list_policy'] = provenance['position_list_policy']
+    record['track'] = evidence['track']
+    record['metadata']['track_evidence'] = evidence
+
+
+def require_track_manifest(manifest):
+    if (manifest.get('source') == 'pmlr' and str(manifest.get('venue', '')).upper() == 'ICML'
+            and int(manifest.get('year', 0)) >= 2024
+            and (manifest.get('parser_version', 0) < 3 or not manifest.get('track_program'))):
+        raise ValueError('Recollect mixed ICML PMLR proceedings with parser_version>=3 and official track evidence')
+
+
+def require_track_record(row, manifest):
+    require_track_manifest({**manifest, **{key: row.get(key) for key in ('source', 'venue', 'year')}})
+    if manifest.get('track_program'):
+        evidence = row.get('metadata', {}).get('track_evidence', {})
+        program = manifest['track_program']
+        if (evidence.get('track') != row.get('track') or evidence.get('program_url') != program.get('url')
+                or evidence.get('program_sha256') != program.get('sha256')
+                or not evidence.get('event_ids') or evidence.get('identity_scheme') not in ('pmlr', 'openreview')):
+            raise ValueError('PMLR record differs from its official track evidence')
+        own = row['external_id'] if evidence['identity_scheme'] == 'pmlr' else row.get('identifiers', {}).get('openreview')
+        if not own or evidence.get('identity_value') != own:
+            raise ValueError('PMLR track evidence belongs to a different paper identity')

paper_atlas/collectors/public_openreview.py

@@ -0,0 +1,252 @@
+"""Collect explicit public negative labels without inferring unobserved decisions.
+
+PaperCopilot is a curated mirror, not the conference's signed decision feed.
+Only metadata is retained; source profiles, reviews and PDFs are never cached.
+"""
+import argparse
+from collections import Counter, defaultdict
+import fcntl
+import hashlib
+import io
+import json
+from pathlib import Path
+import re
+import shutil
+import tempfile
+import urllib.request
+
+from paper_atlas.config import DEFAULT_REGISTRY
+from paper_atlas.db.submissions import timestamp
+
+
+SOURCE = 'papercopilot_openreview'
+REVISION = '41ee277825d685bd78e88e48b69da6aead9a69bb'
+AS_OF = '2026-10-31T01:25:52.000Z'
+REPOSITORY = 'https://github.com/papercopilot/openreview'
+RAW_ROOT = 'https://raw.githubusercontent.com/papercopilot/openreview/' + REVISION + '/'
+TREE_URL = 'https://api.github.com/repos/papercopilot/openreview/git/trees/' + REVISION + '?recursive=1'
+NEGATIVE = {'reject': 'rejected', 'rejected': 'rejected',
+            'desk reject': 'desk_rejected', 'desk rejected': 'desk_rejected',
+            'withdrawn': 'withdrawn', 'withdrawal': 'withdrawn'}
+TRACKS = {'main': 'main', 'Position': 'position',
+          'Datasets & Benchmarks': 'datasets_benchmarks'}
+WORKSHOP_REJECT_LABELS = {"aim-fm workshop @ neurips'24 reject",
+                          'coginterp @ neurips 2025 reject'}
+
+
+def download(url, limit):
+    request = urllib.request.Request(url, headers={'User-Agent': 'PaperAtlas/0.1 (public scholarly metadata)'})
+    with urllib.request.urlopen(request, timeout=90) as response:
+        if response.status != 200:
+            raise ValueError('Expected a complete HTTP 200 response')
+        result = response.read(limit + 1)
+    if len(result) > limit:
+        raise ValueError('Public metadata exceeds byte limit')
+    return result
+
+
+def negative_status(raw):
+    """A closed mapping: pending, absent, low-score and retracted are not Reject."""
+    if not isinstance(raw, str):
+        raise ValueError('Status must be text')
+    key = ' '.join(raw.casefold().split())
+    if key in NEGATIVE:
+        return NEGATIVE[key]
+    if re.fullmatch(r'(?:icml|neurips) 20\d{2} (?:conference|workshop [a-z0-9 -]+) withdrawn submission', key):
+        return 'withdrawn'
+    if key in WORKSHOP_REJECT_LABELS:
+        return 'rejected'
+    return None
+
+
+def normalize(row, family, year, provenance, include_workshops=False):
+    """Return a normalized assertion or a reason requiring separate review."""
+    raw_status = row.get('status', '')
+    status = negative_status(raw_status)
+    if status is None:
+        if re.search(r'\bretract\w*\b', raw_status, re.I):
+            return None, 'retracted'
+        if re.search(r'\b(?:reject\w*|withdraw\w*|desk)\b', raw_status, re.I):
+            return None, 'unmapped_negative_label'
+        return None, 'not_explicit_negative'
+    forum = row.get('id')
+    if not isinstance(forum, str) or not re.fullmatch(r'[A-Za-z0-9_-]{8,32}', forum) or re.match(r'20\d{2}-Oral-', forum):
+        raise ValueError('Invalid public submission forum ID')
+    track = row.get('track')
+    if not isinstance(track, str):
+        raise ValueError('Track must be explicit text')
+    normalized_track = TRACKS.get(track)
+    if normalized_track is None:
+        if re.fullmatch(r'workshop_[a-z0-9_]+', track) and include_workshops:
+            normalized_track = track
+        else:
+            return None, 'separate_workshop' if track.startswith('workshop') else 'unmapped_track'
+    prefixed = re.match(r'^(ICML|NeurIPS) (20\d{2}) ', raw_status)
+    if prefixed and (prefixed[1].casefold() != ('neurips' if family == 'nips' else family) or int(prefixed[2]) != year):
+        raise ValueError('Status label conflicts with source conference year')
+    allowed = {v['acronym'].casefold(): v['acronym'] for v in json.loads(DEFAULT_REGISTRY.read_text())['venues']}
+    venue = allowed.get('neurips' if family == 'nips' else family)
+    if venue is None:
+        return None, 'unmapped_venue'
+    title = row.get('title')
+    abstract = row.get('abstract')
+    if not isinstance(title, str) or not title.strip():
+        raise ValueError('Public submission lacks a title')
+    if abstract is not None and not isinstance(abstract, str):
+        raise ValueError('Abstract must be text or null')
+    author_text = row.get('author') or ''
+    if not isinstance(author_text, str):
+        raise ValueError('Unexpected public author format')
+    authors = [a.strip() for a in author_text.split(';') if a.strip() and '@' not in a
+               and not re.fullmatch(r'anonymous(?: authors?)?', a.strip(), re.I)]
+    tags = []
+    for field in ('keywords', 'primary_area'):
+        value = row.get(field)
+        if value is None:
+            continue
+        terms = value if isinstance(value, list) else re.split(r'[,;]', value) if field == 'keywords' and isinstance(value, str) else [value]
+        tags.extend({'namespace': SOURCE + ':' + field, 'value': term.strip()}
+                    for term in terms if isinstance(term, str) and term.strip())
+    evidence = provenance['source_file'] + '#L' + str(provenance['source_line'])
+    return {'source': SOURCE, 'external_id': f'{family}/{year}/{forum}',
+            'url': 'https://openreview.net/forum?id=' + forum,
+            'title': ' '.join(title.split()), 'abstract': abstract.strip() if abstract else None,
+            'authors': authors, 'identifiers': {'openreview': forum},
+            'submission_identifiers': [{'scheme': 'openreview', 'value': forum}],
+            'venue': venue, 'year': year, 'track': normalized_track,
+            'status': status, 'raw_decision': raw_status, 'evidence_url': evidence,
+            'observed_at': AS_OF, 'tags': tags,
+            'metadata': {**provenance, 'raw_track': track, 'raw_status': raw_status,
+                         'evidence_kind': 'curated_public_status_assertion',
+                         'original_decision_signature_verified': False,
+                         'source_snapshot_upper_bound': AS_OF}}, None
+
+
+def held_metadata(row, reason, provenance):
+    # Retractions and unresolved track scopes remain visible without becoming
+    # invented Reject/Withdrawn events in a schema that cannot represent them.
+    return {key: row.get(key) for key in ('id', 'title', 'abstract', 'track', 'status')} | {
+        'reason': reason, 'provenance': provenance}
+
+
+def collect(out, families=None, years=None, include_workshops=False, fetcher=None):
+    fetcher = fetcher or download
+    out = Path(out)
+    out.mkdir(parents=True, exist_ok=True)
+    with (out / 'collect.lock').open('a') as lock:
+        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        if (out / 'public-openreview-inventory.json').exists():
+            raise ValueError('Completed inventory exists; select a new output directory')
+        if shutil.disk_usage(out).free < 8 * 1024**3:
+            raise RuntimeError('Less than 8 GiB free before collection')
+        tree = json.loads(fetcher(TREE_URL, 8 * 1024**2))
+        if tree.get('truncated') is not False or tree.get('sha') != REVISION:
+            raise ValueError('Incomplete or unpinned public source tree')
+        selected = []
+        for item in tree['tree']:
+            match = re.fullmatch(r'venues/([a-z0-9]+)/\1((?:19|20)\d{2})\.jsonl', item['path'])
+            if item['type'] != 'blob' or not match:
+                continue
+            family, year = match[1], int(match[2])
+            if year > 2026 or (families is None and family == 'iclr'):
+                continue
+            if families is not None and family not in families or years is not None and year not in years:
+                continue
+            selected.append((item, family, year))
+        if not selected:
+            raise ValueError('No pinned source files match the request')
+        if families and set(families) - {family for _, family, _ in selected}:
+            raise ValueError('Requested venue has no selected source file')
+        if years and set(years) - {year for _, _, year in selected}:
+            raise ValueError('Requested year has no selected source file')
+        return _collect_files(out, selected, include_workshops, fetcher)
+
+
+def _collect_files(out, selected, include_workshops, fetcher):
+    summaries = []
+    groups = defaultdict(list)
+    global_forums = {}
+    with tempfile.TemporaryDirectory(prefix='.public-openreview-', dir=out) as temporary:
+        stage = Path(temporary)
+        held_path = stage / 'held-records.jsonl'
+        with held_path.open('w') as held:
+            for item, family, year in sorted(selected, key=lambda entry: entry[0]['path']):
+                size = item['size']
+                if not isinstance(size, int) or not 0 < size <= 96 * 1024**2:
+                    raise ValueError('Unexpected metadata file size')
+                data = fetcher(RAW_ROOT + item['path'], size)
+                blob = hashlib.sha1(b'blob ' + str(len(data)).encode() + b'\0' + data).hexdigest()
+                if len(data) != size or blob != item['sha']:
+                    raise ValueError('Pinned source blob checksum mismatch')
+                matrix = Counter()
+                reasons = Counter()
+                seen = set()
+                rows = 0
+                for number, line in enumerate(io.BytesIO(data), 1):
+                    if len(line) > 4 * 1024**2:
+                        raise ValueError('Oversized metadata row')
+                    row = json.loads(line)
+                    forum = row.get('id')
+                    if not isinstance(forum, str) or forum in seen:
+                        raise ValueError('Missing or duplicate source forum ID')
+                    seen.add(forum)
+                    edition = (family, year, row.get('track'))
+                    if forum in global_forums and global_forums[forum] != edition:
+                        raise ValueError('One forum appears in multiple source editions or tracks')
+                    global_forums[forum] = edition
+                    rows += 1
+                    matrix[(row.get('track'), row.get('status'))] += 1
+                    provenance = {'source_revision': REVISION, 'source_blob_sha1': blob,
+                                  'source_file': RAW_ROOT + item['path'], 'source_line': number}
+                    record, reason = normalize(row, family, year, provenance, include_workshops)
+                    if record is None:
+                        reasons[reason] += 1
+                        if reason != 'not_explicit_negative':
+                            held.write(json.dumps(held_metadata(row, reason, provenance), ensure_ascii=False) + '\n')
+                    else:
+                        groups[(family, year, record['track'])].append(record)
+                summary = {'path': item['path'], 'input_bytes': size, 'git_blob_sha1': blob,
+                           'rows': rows, 'excluded_or_held': dict(reasons),
+                           'status_matrix': [{'track': t, 'status': s, 'count': n} for (t, s), n in matrix.items()]}
+                summaries.append(summary)
+                print(json.dumps({key: summary[key] for key in ('path', 'rows', 'excluded_or_held')}), flush=True)
+        files = []
+        for (family, year, track), records in sorted(groups.items()):
+            name = f'public-openreview-{family}-{year}-{track}.jsonl'
+            path = stage / name
+            path.write_text(''.join(json.dumps(row, ensure_ascii=False) + '\n' for row in records))
+            manifest = {'source': SOURCE, 'parser_version': 1, 'venue': records[0]['venue'],
+                        'year': year, 'track': track, 'count': len(records), 'status': 'partial',
+                        'file': str(out / name), 'file_sha256': hashlib.sha256(path.read_bytes()).hexdigest(),
+                        'list_url': REPOSITORY + '/tree/' + REVISION, 'source_revision': REVISION,
+                        'missing_abstract_count': sum(not row['abstract'] for row in records),
+                        'evidence_kind': 'curated_public_status_assertion',
+                        'note': 'Public subset only; no population coverage or original decision-signature claim.'}
+            path.with_suffix('.manifest.json').write_text(json.dumps(manifest, indent=2) + '\n')
+            files.append(manifest)
+        result = {'source': SOURCE, 'revision': REVISION, 'finished_at': timestamp(),
+                  'input_rows': sum(item['rows'] for item in summaries), 'source_files': summaries,
+                  'normalized_records': sum(item['count'] for item in files), 'files': files,
+                  'include_workshops': include_workshops, 'held_file': str(out / held_path.name),
+                  'held_file_sha256': hashlib.sha256(held_path.read_bytes()).hexdigest(),
+                  'note': 'Zero explicit negative records in this mirror does not prove no public rejections exist.'}
+        inventory = stage / 'public-openreview-inventory.json'
+        inventory.write_text(json.dumps(result, ensure_ascii=False, indent=2) + '\n')
+        # Publish only after every selected source has passed its pinned hash.
+        for path in sorted(stage.iterdir(), key=lambda path: path == inventory):
+            path.replace(out / path.name)
+        return result
+
+
+def main(argv=None, *, prog=None):
+    parser = argparse.ArgumentParser(prog=prog)
+    parser.add_argument('--out', required=True)
+    parser.add_argument('--family', action='append', help='Source family, e.g. icml, nips, emnlp')
+    parser.add_argument('--year', type=int, action='append')
+    parser.add_argument('--include-workshops', action='store_true')
+    args = parser.parse_args(argv)
+    collect(args.out, args.family, args.year, args.include_workshops)
+
+
+if __name__ == '__main__':
+    main()

paper_atlas/ingestion/backfill.py

@@ -6,6 +6,7 @@ from pathlib import Path
 
 from paper_atlas.db.store import Store
 from paper_atlas.db.submissions import attach_submission,identities_for_row
+from paper_atlas.collectors.pmlr_tracks import require_track_manifest, require_track_record
 
 
 def queue_files(path):
@@ -21,6 +22,7 @@ def queue_files(path):
 def backfill(db,files):
     for filename in files:
         manifest=json.loads(Path(filename).with_suffix('.manifest.json').read_text())
+        require_track_manifest(manifest)
         if manifest['source'] in ('colm','neurips','iclr_program') and manifest.get('parser_version',0)<2:
             raise ValueError('Recollect old parser output before submission evidence backfill')
     store=Store(db);c=store.conn;count=0
@@ -35,6 +37,7 @@ def backfill(db,files):
                     seen.add(row['external_id'])
                     if any(row.get(key)!=manifest.get(key) for key in ('source','venue','year','track')):
                         raise ValueError('Evidence row differs from edition manifest')
+                    require_track_record(row, manifest)
                     record=c.execute('SELECT r.id,r.paper_id,r.metadata_json FROM source_records r JOIN sources s ON s.id=r.source_id WHERE s.name=? AND r.external_id=?',(row['source'],row['external_id'])).fetchone()
                     edition=c.execute('SELECT e.id FROM editions e JOIN venues v ON v.id=e.venue_id WHERE v.acronym=? COLLATE NOCASE AND e.year=? AND e.track=?',(row['venue'],row['year'],row['track'])).fetchone()
                     if not record or not edition:raise ValueError('Evidence backfill requires an already imported record and edition')

paper_atlas/ingestion/pmlr_tracks.py

@@ -0,0 +1,200 @@
+"""Move existing PMLR submission events to verified tracks without reimporting."""
+import argparse
+from contextlib import closing
+import fcntl
+import hashlib
+import json
+from pathlib import Path
+import sqlite3
+
+from paper_atlas.collectors.pmlr_tracks import require_track_manifest, require_track_record
+from paper_atlas.db.submissions import identities_for_row, timestamp
+
+
+def save(path, result):
+    part = path.with_suffix('.part')
+    part.write_text(json.dumps(result, ensure_ascii=False, indent=2) + '\n')
+    part.replace(path)
+
+
+def load_inventories(paths):
+    scopes, seen, forums = [], set(), set()
+    for filename in paths:
+        path = Path(filename).resolve()
+        inventory = json.loads(path.read_text())
+        require_track_manifest(inventory)
+        if inventory.get('source') != 'pmlr' or inventory.get('venue') != 'ICML' or inventory.get('year', 0) < 2024:
+            raise ValueError('Only verified modern ICML PMLR volumes may be reclassified')
+        if not inventory.get('count') or not inventory.get('files'):
+            raise ValueError('A nonempty complete publication inventory is required')
+        rows, manifests = [], []
+        tracks = set()
+        for manifest in inventory['files']:
+            require_track_manifest(manifest)
+            if any(manifest.get(key) != inventory.get(key) for key in ('source', 'venue', 'year', 'volume', 'track_program')):
+                raise ValueError('Track manifest differs from volume inventory')
+            if manifest['track'] not in ('main', 'position') or manifest['track'] in tracks:
+                raise ValueError('Invalid or repeated track in volume inventory')
+            tracks.add(manifest['track'])
+            data = Path(manifest['file']).read_bytes()
+            if hashlib.sha256(data).hexdigest() != manifest.get('file_sha256'):
+                raise ValueError('Track file differs from its verified inventory')
+            chunk = [json.loads(line) for line in data.splitlines()]
+            if not chunk or len(chunk) != manifest['count']:
+                raise ValueError('Track inventory count mismatch')
+            for row in chunk:
+                if any(row.get(key) != manifest.get(key) for key in ('source', 'venue', 'year', 'track')):
+                    raise ValueError('Record differs from its track manifest')
+                if not row['external_id'].startswith('v' + str(inventory['volume']) + '/') or row['status'] != 'accepted':
+                    raise ValueError('Record is outside the accepted publication scope')
+                if row['external_id'] in seen:
+                    raise ValueError('Repeated publication in reclassification inventory')
+                seen.add(row['external_id'])
+                require_track_record(row, manifest)
+                own_forum = row.get('identifiers', {}).get('openreview')
+                if not own_forum or (row['year'], own_forum) in forums:
+                    raise ValueError('Missing or repeated forum identity in one ICML edition')
+                forums.add((row['year'], own_forum))
+            rows.extend(chunk); manifests.append(manifest)
+        if len(rows) != inventory['count']:
+            raise ValueError('Volume count differs from its track files')
+        scopes.append({'path': str(path), 'sha256': hashlib.sha256(path.read_bytes()).hexdigest(),
+                       'inventory': inventory, 'manifests': manifests, 'rows': rows})
+    if not scopes:
+        raise ValueError('At least one complete volume inventory is required')
+    return scopes
+
+
+def plan(c, scopes):
+    updates, changes = [], []
+    for scope in scopes:
+        volume = scope['inventory']['volume']
+        source_rows = c.execute('''SELECT r.id,r.external_id FROM source_records r JOIN sources src ON src.id=r.source_id
+            WHERE src.name='pmlr' AND r.external_id LIKE ?''', ('v' + str(volume) + '/%',)).fetchall()
+        if {r[1] for r in source_rows} != {r['external_id'] for r in scope['rows']}:
+            raise ValueError('Existing PMLR volume differs from the complete audited publication inventory')
+        for row in scope['rows']:
+            record = c.execute('''SELECT r.id,r.paper_id,r.title,r.abstract,r.authors_json,r.metadata_json FROM source_records r
+                JOIN sources src ON src.id=r.source_id WHERE src.name='pmlr' AND r.external_id=?''', (row['external_id'],)).fetchone()
+            if (record[2] != row['title'] or record[3] != row.get('abstract')
+                    or json.loads(record[4]) != row.get('authors', [])):
+                raise ValueError('Reclassification must preserve existing publisher text and authors: ' + row['external_id'])
+            metadata = json.loads(record[5])
+            forum = str(metadata.get('openreview') or (metadata.get('record_identifiers') or {}).get('openreview') or '')
+            forum = forum.split('id=')[-1].split('&')[0]
+            if not forum or forum != row['identifiers'].get('openreview'):
+                raise ValueError('Publisher forum identity differs from retained source metadata')
+            events = c.execute('''SELECT s.id,s.paper_id,s.edition_id,e.venue_id,e.year,e.track,v.acronym,s.source_record_id,s.source_submission_id
+                FROM submission_sources ss JOIN submissions s ON s.id=ss.submission_id JOIN editions e ON e.id=s.edition_id
+                JOIN venues v ON v.id=e.venue_id WHERE ss.source_record_id=? AND ss.source_submission_id=?''',
+                (record[0], row['external_id'])).fetchall()
+            if len(events) != 1:
+                raise ValueError('Publication must have exactly one existing submission event')
+            event = events[0]
+            sid, paper, old_edition, venue, year, old_track, acronym, owner, external = event
+            if paper != record[1] or year != row['year'] or acronym.upper() != 'ICML' or old_track not in ('main', 'position'):
+                raise ValueError('Existing submission is outside this publication scope')
+            identities = c.execute('SELECT scheme,value FROM submission_identifiers WHERE submission_id=? ORDER BY scheme,value', (sid,)).fetchall()
+            own = identities_for_row(row)
+            if {value for scheme, value in identities if scheme == 'openreview'} - {forum}:
+                raise ValueError('Existing event has a conflicting forum identity')
+            target = c.execute('SELECT id FROM editions WHERE venue_id=? AND year=? AND track=?', (venue, year, row['track'])).fetchone()
+            if target:
+                for scheme, value in sorted(set(identities) | set(own)):
+                    collision = c.execute('SELECT submission_id FROM submission_identifiers WHERE edition_id=? AND scheme=? AND value=?', (target[0], scheme, value)).fetchone()
+                    if collision and collision[0] != sid:
+                        raise ValueError('Target track already has another event for this identity; reconcile sources before moving')
+                if c.execute('SELECT 1 FROM submissions WHERE source_record_id=? AND source_submission_id=? AND edition_id=? AND id<>?', (owner, external, target[0], sid)).fetchone():
+                    raise ValueError('Target track already has this publication event')
+            if old_track != row['track']:
+                links = c.execute('SELECT source_record_id,source_submission_id FROM submission_sources WHERE submission_id=?', (sid,)).fetchall()
+                if links != [(record[0], row['external_id'])] or owner != record[0] or external != row['external_id']:
+                    raise ValueError('Moving an event shared by multiple sources requires a separate reconciliation')
+                decisions = c.execute('SELECT id,status,raw_value,evidence_url,observed_at,decision_date FROM decisions WHERE submission_id=? ORDER BY id', (sid,)).fetchall()
+                if not decisions or any(d[1] != 'accepted' for d in decisions):
+                    raise ValueError('Moving a publication with conflicting decision history requires reconciliation')
+                changes.append({'submission_id': sid, 'source_record_id': record[0], 'external_id': row['external_id'],
+                                'paper_id': paper, 'from_edition_id': old_edition, 'from_track': old_track,
+                                'to_track': row['track'], 'year': year, 'before_identifiers': identities,
+                                'before_decisions': decisions, 'track_evidence': row['metadata']['track_evidence']})
+            updates.append((row, record, event, own))
+    return updates, changes
+
+
+def reclassify(db, inventories, out, apply=False):
+    db, out = Path(db).resolve(), Path(out).resolve()
+    if not db.is_file() or out == db:
+        raise ValueError('Existing database and distinct report path required')
+    scopes = load_inventories(inventories)
+    inputs = {Path(scope['path']) for scope in scopes}
+    inputs.update(Path(m['file']).resolve() for scope in scopes for m in scope['manifests'])
+    inputs.update(Path(m['file']).with_suffix('.manifest.json').resolve() for scope in scopes for m in scope['manifests'])
+    if out in inputs:
+        raise ValueError('Report must preserve evidence inputs')
+    out.parent.mkdir(parents=True, exist_ok=True)
+    with Path(str(db) + '.maintenance.lock').open('a') as lock, Path(str(out) + '.lock').open('a') as output_lock:
+        fcntl.flock(lock, (fcntl.LOCK_EX if apply else fcntl.LOCK_SH) | fcntl.LOCK_NB)
+        fcntl.flock(output_lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        if out.exists():
+            raise ValueError('Reclassification report exists; use a new path')
+        mode = 'rw' if apply else 'ro'
+        with closing(sqlite3.connect(db.as_uri() + '?mode=' + mode, uri=True)) as c:
+            c.execute('PRAGMA foreign_keys=ON'); c.execute('PRAGMA cache_size=-32768')
+            c.execute('BEGIN IMMEDIATE' if apply else 'BEGIN')
+            updates, changes = plan(c, scopes)
+            result = {'status': 'planned', 'db': str(db), 'started_at': timestamp(),
+                      'records_verified': len(updates), 'events_to_move': len(changes), 'changes': changes,
+                      'inputs': [{key: scope[key] for key in ('path', 'sha256')} for scope in scopes],
+                      'note': 'Preserves paper/source/submission/decision IDs and all decision observations; no entity merges or text reimport.'}
+            save(out, result)
+            if not apply:
+                return result
+            try:
+                for row, record, event, own in updates:
+                    sid, _, old_edition, venue, year, old_track, _, _, _ = event
+                    c.execute('INSERT OR IGNORE INTO editions(venue_id,year,track) VALUES(?,?,?)', (venue, year, row['track']))
+                    target = c.execute('SELECT id FROM editions WHERE venue_id=? AND year=? AND track=?', (venue, year, row['track'])).fetchone()[0]
+                    if target != old_edition:
+                        c.execute('UPDATE submission_identifiers SET edition_id=? WHERE submission_id=?', (target, sid))
+                        c.execute('UPDATE submissions SET edition_id=? WHERE id=?', (target, sid))
+                    for scheme, value in own:
+                        c.execute('INSERT OR IGNORE INTO submission_identifiers VALUES(?,?,?,?)', (target, scheme, value, sid))
+                    metadata = json.loads(record[5]); metadata.update(row['metadata'])
+                    c.execute('UPDATE source_records SET metadata_json=? WHERE id=?', (json.dumps(metadata, ensure_ascii=False), record[0]))
+                for scope in scopes:
+                    for m in scope['manifests']:
+                        edition = c.execute('SELECT e.id FROM editions e JOIN venues v ON v.id=e.venue_id WHERE v.acronym=? COLLATE NOCASE AND e.year=? AND e.track=?', ('ICML', m['year'], m['track'])).fetchone()[0]
+                        source = c.execute("SELECT id FROM sources WHERE name='pmlr'").fetchone()[0]
+                        c.execute('''INSERT INTO edition_sources(edition_id,source_id,list_url,status,expected_count,imported_count,last_success_at)
+                            VALUES(?,?,?,?,?,?,?) ON CONFLICT(edition_id,source_id,list_url) DO UPDATE SET status=excluded.status,
+                            expected_count=excluded.expected_count,imported_count=excluded.imported_count,last_success_at=excluded.last_success_at,last_error=NULL''',
+                            (edition, source, m['list_url'], 'partial' if m.get('missing_abstract_count') or m.get('status') == 'partial' else 'complete', m['count'], m['count'], timestamp()))
+                _, remaining = plan(c, scopes)
+                if remaining:
+                    raise ValueError('Post-migration track comparison still has mismatches')
+                print(json.dumps({'stage': 'checking_foreign_keys', 'events_moved': len(changes)}), flush=True)
+                if c.execute('PRAGMA foreign_key_check').fetchone():
+                    raise ValueError('Foreign key violations after track reclassification')
+                result.update(status='prepared', events_moved=len(changes))
+                save(out, result)
+                c.commit()
+            except BaseException as error:
+                c.rollback(); result.update(status='error', error=str(error)); save(out, result); raise
+            result.update(status='complete', finished_at=timestamp())
+            save(out, result)
+            return result
+
+
+def main(argv=None, *, prog=None):
+    parser = argparse.ArgumentParser(prog=prog)
+    parser.add_argument('--db', required=True)
+    parser.add_argument('--inventory', action='append', required=True)
+    parser.add_argument('--out', required=True)
+    parser.add_argument('--apply', action='store_true', help='Apply the verified migration under an exclusive maintenance lock')
+    args = parser.parse_args(argv)
+    result = reclassify(args.db, args.inventory, args.out, args.apply)
+    print(json.dumps({key: value for key, value in result.items() if key != 'changes'}), flush=True)
+
+
+if __name__ == '__main__':
+    main()

paper_atlas/ingestion/records.py

@@ -6,12 +6,14 @@ from pathlib import Path
 
 from paper_atlas.db.store import Store
 from paper_atlas.db.submissions import identities_for_row,timestamp
+from paper_atlas.collectors.pmlr_tracks import require_track_manifest, require_track_record
 
 
 def ingest_file(db,path):
     path = Path(path)
     manifest_path = path.with_suffix('.manifest.json')
     manifest = json.loads(manifest_path.read_text()) if manifest_path.exists() else {}
+    require_track_manifest(manifest)
     if manifest.get('source') in ('colm','neurips') and manifest.get('parser_version',0)<2:
         raise ValueError('Recollect proceedings with parser_version>=2 before identity matching')
     if manifest.get('source')=='iclr_program' and manifest.get('parser_version',0)<2:
@@ -36,6 +38,7 @@ def ingest_file(db,path):
                     raise ValueError('Source differs from manifest')
                 if any(key in manifest and manifest[key]!=row[key] for key in ('venue','year','track')):
                     raise ValueError('Conference edition differs from manifest')
+                require_track_record(row, manifest)
                 external = row['external_id']
                 if external in seen:
                     raise ValueError('Duplicate external ID in one collected file: '+external)

tests/test_icml_public_archive.py

@@ -0,0 +1,86 @@
+import hashlib
+import json
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from paper_atlas.collectors import icml_public_archive as archive
+from paper_atlas.audits.public_openreview import audit
+from paper_atlas.db.store import Store
+from paper_atlas.ingestion.records import ingest_file
+
+
+class ICMLPublicArchiveTests(unittest.TestCase):
+    def note(self, **changes):
+        return {'id': 'PublicRoot1', 'forum': 'PublicRoot1', 'readers': ['everyone'],
+                'domain': archive.DOMAIN, 'invitations': [archive.DOMAIN + '/-/Submission'],
+                'license': 'CC BY 4.0', 'cdate': 1769251630136,
+                'content': {'title': {'value': 'A public rejected submission'},
+                            'abstract': {'value': 'Public abstract'},
+                            'authors': {'value': ['Anonymous', 'Jane Doe']},
+                            'authorids': {'value': ['private@example.org']},
+                            'venue': {'value': 'Submitted to ICML 2026'},
+                            'venueid': {'value': archive.VENUEID},
+                            'primary_area': {'value': 'learning'},
+                            'pdf': {'value': '/private.pdf'}},
+                'details': {'reviews': ['not retained']}, **changes}
+
+    def test_explicit_platform_classification_requires_public_root_and_correct_scope(self):
+        original = self.note()
+        normalized = archive.normalize(original)
+        self.assertEqual(normalized['status'], 'rejected')
+        self.assertEqual(normalized['raw_decision'], archive.VENUEID)
+        self.assertEqual(normalized['metadata']['license'], 'CC BY 4.0')
+        self.assertEqual(normalized['metadata']['evidence_kind'], 'public_submission_venueid')
+        self.assertFalse(normalized['metadata']['original_decision_signature_verified'])
+        self.assertNotIn('private@example.org', json.dumps(normalized))
+        self.assertNotIn('not retained', json.dumps(normalized))
+        for changes in ({'readers': ['authors']}, {'forum': 'AnotherRoot'}, {'ddate': 1000},
+                        {'domain': 'ICML.cc/2026/Position_Papers'}, {'invitations': []}):
+            with self.subTest(changes=changes), self.assertRaises(ValueError):
+                archive.normalize(self.note(**changes))
+        for venueid in (None, 'ICML.cc/2026/Conference/Submission', 'ICML.cc/2026/Conference/Accepted_Submission'):
+            candidate = self.note()
+            candidate['content']['venueid'] = {'value': venueid}
+            with self.assertRaisesRegex(ValueError, 'classification'):
+                archive.normalize(candidate)
+
+    def collect_fixture(self, out, notes, count=None, corrupt=False):
+        data = json.dumps({'count': len(notes) if count is None else count, 'notes': notes}).encode()
+        blob = hashlib.sha1(b'blob ' + str(len(data)).encode() + b'\0' + data).hexdigest()
+        with patch.multiple(archive, BYTES=len(data), BLOB_SHA1=blob, EXPECTED=len(notes)), \
+             patch.object(archive.shutil, 'disk_usage', return_value=type('Space', (), {'free': 20 * 1024**3})()):
+            return archive.collect(out, fetcher=lambda url, limit: data + b'x' if corrupt else data)
+
+    def test_verified_archive_roundtrips_through_preflight_and_idempotent_import(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            result = self.collect_fixture(root / 'source', [self.note()])
+            db = root / 'db.sqlite'
+            Store(db).conn.close()
+            report = audit(db, root / 'source/public-openreview-inventory.json', root / 'audit.json')
+            self.assertEqual(report['status'], 'ready_for_serial_import')
+            self.assertEqual(report['source'], archive.SOURCE)
+            self.assertEqual(result['files'][0]['status'], 'partial')
+            for _ in range(2):
+                ingest_file(db, report['files'][0])
+            store = Store(db)
+            try:
+                self.assertEqual(store.conn.execute('SELECT count(*) FROM submissions').fetchone()[0], 1)
+                self.assertEqual(store.conn.execute('SELECT status FROM decisions').fetchall(), [('rejected',)])
+            finally:
+                store.conn.close()
+
+    def test_bad_hash_count_or_duplicate_identity_publishes_nothing(self):
+        for kwargs in ({'corrupt': True}, {'count': 99}, {}):
+            notes = [self.note(), self.note()] if not kwargs else [self.note()]
+            with self.subTest(kwargs=kwargs), tempfile.TemporaryDirectory() as tmp:
+                with self.assertRaises(ValueError):
+                    self.collect_fixture(tmp, notes, **kwargs)
+                self.assertEqual(list(Path(tmp).glob('*.jsonl')), [])
+                self.assertFalse((Path(tmp) / 'public-openreview-inventory.json').exists())
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_index_core.py

@@ -55,7 +55,7 @@ class StoreTests(unittest.TestCase):
 
     def test_manifest_count_mismatch_rolls_back_records(self):
         path=Path(self.tmp.name)/'source.jsonl'
-        path.write_text(json.dumps({'source':'pmlr','external_id':'v1/paper','url':'https://example.org/paper','title':'Paper','abstract':'Abstract','authors':[],'venue':'ICML','year':2025,'track':'main','status':'accepted'})+'\n')
+        path.write_text(json.dumps({'source':'pmlr','external_id':'v1/paper','url':'https://example.org/paper','title':'Paper','abstract':'Abstract','authors':[],'venue':'ICML','year':2023,'track':'main','status':'accepted'})+'\n')
         path.with_suffix('.manifest.json').write_text(json.dumps({'source':'pmlr','list_url':'https://example.org/list','count':2}))
         with self.assertRaisesRegex(ValueError,'differs'):
             ingest_file(self.db,path)
@@ -100,7 +100,7 @@ class StoreTests(unittest.TestCase):
 
     def test_missing_abstract_is_checked_even_when_manifest_claims_complete(self):
         path=Path(self.tmp.name)/'incomplete.jsonl'
-        row={'source':'pmlr','external_id':'v1/paper','url':'https://example.org/paper','title':'Paper','abstract':None,'authors':[],'venue':'ICML','year':2025,'track':'main','status':'accepted'}
+        row={'source':'pmlr','external_id':'v1/paper','url':'https://example.org/paper','title':'Paper','abstract':None,'authors':[],'venue':'ICML','year':2023,'track':'main','status':'accepted'}
         path.write_text(json.dumps(row)+'\n')
         path.with_suffix('.manifest.json').write_text(json.dumps({'source':'pmlr','list_url':'https://example.org/list','count':1,'status':'collected','missing_abstract_count':0}))
         ingest_file(self.db,path)
@@ -108,7 +108,7 @@ class StoreTests(unittest.TestCase):
 
     def test_manifest_edition_mismatch_rolls_back_instead_of_marking_coverage(self):
         path=Path(self.tmp.name)/'wrong-edition.jsonl'
-        row={'source':'pmlr','external_id':'v1/paper','url':'https://example.org/paper','title':'Paper','abstract':'Abstract','authors':[],'venue':'ICML','year':2025,'track':'main','status':'accepted'}
+        row={'source':'pmlr','external_id':'v1/paper','url':'https://example.org/paper','title':'Paper','abstract':'Abstract','authors':[],'venue':'ICML','year':2023,'track':'main','status':'accepted'}
         path.write_text(json.dumps(row)+'\n')
         path.with_suffix('.manifest.json').write_text(json.dumps({'source':'pmlr','list_url':'https://example.org/list','count':1,'year':2024}))
         with self.assertRaisesRegex(ValueError,'edition differs'):

tests/test_pmlr.py

@@ -25,7 +25,7 @@ class MetadataTests(unittest.TestCase):
         self.assertEqual(venue_key('ECML-PKDD'),'ECML-PKDD')
 
     def test_missing_abstract_is_reported_as_partial_collection(self):
-        config=b'shortname: ICML\nyear: 2025\nbooktitle: ICML\n'
+        config=b'shortname: ICML\nyear: 2023\nbooktitle: ICML\n'
         paper='---\nid: paper1\nlayout: inproceedings\ntitle: Paper\nabstract:\n---\n'
         with tempfile.TemporaryDirectory() as tmp,patch('paper_atlas.collectors.pmlr.fetch',return_value=config),patch('paper_atlas.collectors.pmlr.metadata_texts',return_value=iter([('paper.md',paper)])):
             result=collect_volume(999,tmp,tmp)

tests/test_pmlr_track_migration.py

@@ -0,0 +1,156 @@
+import copy
+import fcntl
+import hashlib
+import json
+from pathlib import Path
+import sqlite3
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from paper_atlas.collectors.pmlr_tracks import program_tracks, record_track
+from paper_atlas.db.store import Store
+from paper_atlas.db.submissions import identities_for_row
+from paper_atlas.ingestion.pmlr_tracks import reclassify, plan
+from paper_atlas.ingestion.records import ingest_file
+from tests.test_pmlr_tracks import event, program, record
+
+
+class PmlrMigrationTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory(); self.addCleanup(self.tmp.cleanup)
+        self.root = Path(self.tmp.name); self.db = self.root / 'papers.sqlite'
+        self.inventory = self.root / 'inventory.json'
+        self.rows = [record(title='A main research paper'),
+                     record('position1', 'v267/paper2', title='A position paper')]
+        self.proof = {'url': 'https://icml.cc/static/virtual/data/icml-2025-orals-posters.json', 'sha256': 'a' * 64}
+        mapping, _ = program_tracks(program(event(), event('position1', 'Position_Paper_Track', 2)), 2025)
+        store = Store(self.db); c = store.conn
+        for row in self.rows:
+            row.update(url='https://proceedings.mlr.press/' + row['external_id'] + '.html',
+                       list_url='https://proceedings.mlr.press/v267/', status='accepted')
+            paper, source = store.record('pmlr', row['external_id'], row['url'], row['title'], row['abstract'],
+                                         row['authors'], row['identifiers'], {'openreview': row['identifiers']['openreview']})
+            store.decision(paper, source, 'ICML', 2025, 'main', row['external_id'], row['url'],
+                           'accepted', 'listed in official proceedings', identities=identities_for_row(row))
+            record_track(row, mapping, self.proof)
+        c.execute("INSERT INTO tags(namespace,value) VALUES('custom:topic','keep me')")
+        c.execute("INSERT INTO paper_tags VALUES(1,1,NULL,'custom:manual')")
+        c.commit(); c.close()
+        self.write_inventory()
+
+    def write_inventory(self):
+        files = []
+        for row in self.rows:
+            path = self.root / (row['track'] + '.jsonl')
+            data = (json.dumps(row) + '\n').encode(); path.write_bytes(data)
+            m = {'source': 'pmlr', 'venue': 'ICML', 'year': 2025, 'volume': 267,
+                 'parser_version': 3, 'track': row['track'], 'track_program': self.proof,
+                 'count': 1, 'file': str(path), 'file_sha256': hashlib.sha256(data).hexdigest(),
+                 'missing_abstract_count': 0, 'list_url': row['list_url'], 'status': 'collected'}
+            path.with_suffix('.manifest.json').write_text(json.dumps(m)); files.append(m)
+        self.inventory.write_text(json.dumps({'source': 'pmlr', 'venue': 'ICML', 'year': 2025,
+                                              'volume': 267, 'parser_version': 3, 'track_program': self.proof,
+                                              'count': 2, 'files': files}))
+
+    def tracks(self):
+        with sqlite3.connect(self.db) as c:
+            return c.execute('SELECT s.id,e.track FROM submissions s JOIN editions e ON e.id=s.edition_id ORDER BY s.id').fetchall()
+
+    def test_plan_is_read_only_and_reports_exact_original_event(self):
+        result = reclassify(self.db, [self.inventory], self.root / 'plan.json')
+        self.assertEqual((result['status'], result['records_verified'], result['events_to_move']), ('planned', 2, 1))
+        self.assertEqual(result['changes'][0]['submission_id'], 2)
+        self.assertEqual(self.tracks(), [(1, 'main'), (2, 'main')])
+
+    def test_apply_preserves_ids_decisions_text_tags_and_is_idempotent(self):
+        with sqlite3.connect(self.db) as c:
+            before = {table: c.execute('SELECT * FROM ' + table).fetchall() for table in ('papers', 'decisions', 'decision_sources', 'submission_sources', 'paper_tags')}
+        result = reclassify(self.db, [self.inventory], self.root / 'apply.json', True)
+        self.assertEqual((result['status'], result['events_moved']), ('complete', 1))
+        self.assertEqual(self.tracks(), [(1, 'main'), (2, 'position')])
+        with sqlite3.connect(self.db) as c:
+            for table, rows in before.items():
+                self.assertEqual(c.execute('SELECT * FROM ' + table).fetchall(), rows, table)
+            self.assertEqual(c.execute('SELECT count(*) FROM submission_identifiers i JOIN submissions s ON s.id=i.submission_id WHERE i.edition_id<>s.edition_id').fetchone()[0], 0)
+            self.assertEqual(c.execute('SELECT e.track,es.imported_count FROM edition_sources es JOIN editions e ON e.id=es.edition_id ORDER BY e.track').fetchall(), [('main', 1), ('position', 1)])
+            self.assertEqual(c.execute('PRAGMA foreign_key_check').fetchall(), [])
+        again = reclassify(self.db, [self.inventory], self.root / 'again.json', True)
+        self.assertEqual(again['events_moved'], 0)
+        for m in json.loads(self.inventory.read_text())['files']:
+            ingest_file(self.db, m['file'])
+        self.assertEqual(self.tracks(), [(1, 'main'), (2, 'position')])
+        with sqlite3.connect(self.db) as c:
+            self.assertEqual(c.execute('SELECT * FROM decisions').fetchall(), before['decisions'])
+
+    def test_conflicting_target_event_stops_without_merging_history(self):
+        store = Store(self.db)
+        _, source = store.record('other', 'position1', 'https://example.org', 'Other evidence', None, [], {'openreview': 'position1'})
+        store.decision(2, source, 'ICML', 2025, 'position', 'position1', 'https://example.org', 'rejected', 'Reject', identities=[('openreview', 'position1')])
+        store.conn.commit(); store.conn.close()
+        before = self.tracks()
+        with self.assertRaisesRegex(ValueError, 'another event'):
+            reclassify(self.db, [self.inventory], self.root / 'blocked.json', True)
+        self.assertEqual(self.tracks(), before)
+
+    def test_shared_event_requires_reconciliation_and_keeps_all_sources(self):
+        store = Store(self.db)
+        _, source = store.record('mirror', 'position1', 'https://example.org', 'Mirror title', None, [], {'openreview': 'position1'})
+        store.conn.execute('INSERT INTO submission_sources VALUES(?,?,?)', (2, source, 'position1'))
+        store.conn.commit(); store.conn.close()
+        with self.assertRaisesRegex(ValueError, 'multiple sources'):
+            reclassify(self.db, [self.inventory], self.root / 'shared.json', True)
+        self.assertEqual(self.tracks(), [(1, 'main'), (2, 'main')])
+
+    def test_text_mismatch_in_last_record_prevents_any_changes(self):
+        with sqlite3.connect(self.db) as c:
+            c.execute("UPDATE source_records SET abstract='Changed elsewhere' WHERE id=2")
+        with self.assertRaisesRegex(ValueError, 'preserve existing publisher text'):
+            reclassify(self.db, [self.inventory], self.root / 'text.json', True)
+        self.assertEqual(self.tracks(), [(1, 'main'), (2, 'main')])
+
+    def test_mutation_failure_rolls_back_editions_metadata_and_identifiers(self):
+        calls = 0
+        def fail_after_updates(c, scopes):
+            nonlocal calls
+            calls += 1
+            if calls == 2:
+                raise ValueError('Injected post-migration failure')
+            return plan(c, scopes)
+        with sqlite3.connect(self.db) as c:
+            before = c.execute('SELECT * FROM source_records').fetchall()
+        with patch('paper_atlas.ingestion.pmlr_tracks.plan', side_effect=fail_after_updates):
+            with self.assertRaisesRegex(ValueError, 'Injected'):
+                reclassify(self.db, [self.inventory], self.root / 'rollback.json', True)
+        self.assertEqual(self.tracks(), [(1, 'main'), (2, 'main')])
+        with sqlite3.connect(self.db) as c:
+            self.assertEqual(c.execute('SELECT * FROM source_records').fetchall(), before)
+            self.assertEqual(c.execute("SELECT count(*) FROM editions WHERE track='position'").fetchone()[0], 0)
+        self.assertEqual(json.loads((self.root / 'rollback.json').read_text())['status'], 'error')
+
+    def test_changed_evidence_and_active_writer_are_refused(self):
+        out = self.root / 'blocked.json'
+        with Path(str(self.db) + '.maintenance.lock').open('a') as lock:
+            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            with self.assertRaises(BlockingIOError):
+                reclassify(self.db, [self.inventory], out, True)
+        self.assertFalse(out.exists())
+        path = self.root / 'position.jsonl'; path.write_text(path.read_text() + ' ')
+        with self.assertRaisesRegex(ValueError, 'differs from its verified inventory'):
+            reclassify(self.db, [self.inventory], out, True)
+        self.assertEqual(self.tracks(), [(1, 'main'), (2, 'main')])
+
+    def test_input_report_and_duplicate_forums_are_refused(self):
+        original = self.inventory.read_bytes()
+        with self.assertRaisesRegex(ValueError, 'preserve evidence'):
+            reclassify(self.db, [self.inventory], self.inventory, True)
+        self.assertEqual(self.inventory.read_bytes(), original)
+        self.rows[1]['identifiers']['openreview'] = self.rows[0]['identifiers']['openreview']
+        self.rows[1]['metadata']['track_evidence']['identity_value'] = self.rows[0]['identifiers']['openreview']
+        self.write_inventory()
+        with self.assertRaisesRegex(ValueError, 'repeated forum'):
+            reclassify(self.db, [self.inventory], self.root / 'duplicate.json', True)
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_pmlr_tracks.py

@@ -0,0 +1,148 @@
+import copy
+import json
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from paper_atlas.collectors.pmlr import collect_volume
+from paper_atlas.collectors.pmlr_tracks import program_tracks, record_track, position_list_policy
+from paper_atlas.ingestion.records import ingest_file
+from paper_atlas.ingestion.backfill import backfill
+
+
+def event(forum='realForum1', track='Conference', event_id=1, year=2025, **values):
+    return {'id': event_id, 'sourceid': event_id, 'name': 'A published paper',
+            'sourceurl': f'https://openreview.net/group?id=ICML.cc/{year}/' + track,
+            'paper_url': 'https://openreview.net/forum?id=' + forum,
+            'decision': 'Accept (Poster)', **values}
+
+
+def program(*events, **changes):
+    return json.dumps({'count': len(events), 'results': list(events), **changes}).encode()
+
+
+def record(forum='realForum1', external='v267/paper1', year=2025, title='A published paper'):
+    return {'source': 'pmlr', 'external_id': external, 'venue': 'ICML', 'year': year,
+            'title': title, 'abstract': 'Original abstract', 'authors': ['Jane Doe', 'John Smith'],
+            'identifiers': {'openreview': forum}, 'metadata': {}, 'track': 'main'}
+
+
+class PmlrTrackTests(unittest.TestCase):
+    def provenance(self):
+        return {'url': 'https://icml.cc/static/virtual/data/icml-2025-orals-posters.json', 'sha256': 'a' * 64}
+
+    def test_real_groups_not_title_prefixes_determine_modern_tracks(self):
+        mapping, inventory = program_tracks(program(
+            event(name='Position: still a main-track paper'),
+            event('position1', 'Position_Paper_Track', 2, name='No prefix needed'),
+            event('2025-Oral-fake', event_id=3, sourceid=-1),
+            event('workshop1', 'Workshop', 4)), 2025)
+        main = record(title='Position: still a main-track paper')
+        position = record('position1', 'v267/paper2', title='No prefix needed')
+        record_track(main, mapping, self.provenance()); record_track(position, mapping, self.provenance())
+        self.assertEqual((main['track'], position['track']), ('main', 'position'))
+        self.assertEqual(inventory['paper_identity_count'], 2)
+        self.assertEqual(inventory['excluded_events'], {'synthetic_oral_event': 1, 'other_source': 1})
+
+    def test_conflicting_forums_pagination_and_missing_real_ids_fail(self):
+        for data in [program(event(), event(track='Position_Paper_Track', event_id=2)),
+                     program(event(), next='page2'), program(event(), count=2),
+                     program(event(paper_url='')), program(event(decision='Reject')),
+                     program(event(), event())]:
+            with self.subTest(data=data), self.assertRaises(ValueError):
+                program_tracks(data, 2025)
+
+    def test_missing_publication_identity_has_no_main_track_fallback(self):
+        mapping, _ = program_tracks(program(event()), 2025)
+        with self.assertRaisesRegex(ValueError, 'absent'):
+            record_track(record('unknownForum'), mapping, self.provenance())
+
+    def test_2024_requires_organizer_endorsed_list_and_publisher_self_links(self):
+        html = b'<a href="/virtual/2024/papers.html?filter=titles&amp;search=position%3A">position papers</a>'
+        policy = position_list_policy(html)
+        with self.assertRaises(ValueError):
+            position_list_policy(html.replace(b'2024', b'2023'))
+        e = event(year=2024, name='Position: An explicit position', paper_url='',
+                  paper_pdf_url='https://proceedings.mlr.press/v235/paper1.html')
+        with self.assertRaisesRegex(ValueError, 'policy'):
+            program_tracks(program(e), 2024)
+        mapping, _ = program_tracks(program(e), 2024, policy)
+        row = record(external='v235/paper1', year=2024)
+        record_track(row, mapping, {**self.provenance(), 'position_list_policy': policy})
+        self.assertEqual(row['track'], 'position')
+        self.assertEqual(row['metadata']['track_evidence']['identity_scheme'], 'pmlr')
+
+    def test_reviewed_2024_link_alias_still_needs_title_and_two_authors(self):
+        e = event(year=2024, paper_url='', paper_pdf_url='https://proceedings.mlr.press/v235/theo24a.html',
+                  authors=[{'fullname': 'Jane Doe'}, {'fullname': 'John Smith'}])
+        mapping, _ = program_tracks(program(e), 2024, {'verified': True})
+        row = record(external='v235/guyard24a', year=2024)
+        record_track(row, mapping, self.provenance())
+        self.assertEqual(row['metadata']['track_evidence']['corroborating_author_count'], 2)
+        self.assertNotIn('_authors', row['metadata']['track_evidence'])
+        for changed in [dict(title='Different title'), dict(authors=['Someone Else'])]:
+            bad = record(external='v235/guyard24a', year=2024); bad.update(changed)
+            with self.assertRaisesRegex(ValueError, 'corroboration'):
+                record_track(bad, mapping, self.provenance())
+
+    def test_collector_splits_tracks_preserves_ids_and_imports_one_event_each(self):
+        config = b'shortname: ICML\nyear: 2025\nbooktitle: ICML\n'
+        template = '---\nid: {id}\nlayout: inproceedings\ntitle: {title}\nabstract: Original abstract\nopenreview: {forum}\n---\n'
+        posts = [('a.md', template.format(id='paper1', title='Main paper', forum='realForum1')),
+                 ('b.md', template.format(id='paper2', title='Position paper', forum='position1'))]
+        mapping, inventory = program_tracks(program(event(), event('position1', 'Position_Paper_Track', 2)), 2025)
+        with tempfile.TemporaryDirectory() as tmp, patch('paper_atlas.collectors.pmlr.fetch', return_value=config), \
+                patch('paper_atlas.collectors.pmlr.metadata_texts', return_value=iter(posts)), \
+                patch('paper_atlas.collectors.pmlr.collect_track_evidence', return_value=(mapping, self.provenance())):
+            result = collect_volume(267, tmp, tmp)
+            self.assertNotIn('file', result)
+            self.assertEqual(result['track'], 'mixed')
+            self.assertEqual({m['track']: m['count'] for m in result['files']}, {'main': 1, 'position': 1})
+            self.assertEqual(result['count'], 2)
+            db = Path(tmp) / 'papers.sqlite'
+            for manifest in result['files']:
+                ingest_file(db, manifest['file'])
+            import sqlite3
+            with sqlite3.connect(db) as c:
+                self.assertEqual(c.execute('SELECT count(*) FROM submissions').fetchone()[0], 2)
+                self.assertEqual(c.execute('SELECT track,count(*) FROM submissions s JOIN editions e ON s.edition_id=e.id GROUP BY track').fetchall(), [('main', 1), ('position', 1)])
+                self.assertEqual(c.execute('PRAGMA foreign_key_check').fetchall(), [])
+
+    def test_unknown_track_does_not_replace_existing_volume_output(self):
+        config = b'shortname: ICML\nyear: 2025\nbooktitle: ICML\n'
+        post = '---\nid: paper1\nlayout: inproceedings\ntitle: Paper\nopenreview: missing123\n---\n'
+        with tempfile.TemporaryDirectory() as tmp, patch('paper_atlas.collectors.pmlr.fetch', return_value=config), \
+                patch('paper_atlas.collectors.pmlr.metadata_texts', return_value=iter([('a.md', post)])), \
+                patch('paper_atlas.collectors.pmlr.collect_track_evidence', return_value=({}, self.provenance())):
+            old = Path(tmp) / 'pmlr-v267.jsonl'; old.write_text('previous collection')
+            with self.assertRaisesRegex(ValueError, 'absent'):
+                collect_volume(267, tmp, tmp)
+            self.assertEqual(old.read_text(), 'previous collection')
+            self.assertFalse((Path(tmp) / 'pmlr-v267.inventory.json').exists())
+
+    def test_old_mixed_manifest_is_rejected_before_opening_database(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / 'pmlr-v267.jsonl'; path.write_text('')
+            path.with_suffix('.manifest.json').write_text(json.dumps({'source': 'pmlr', 'venue': 'ICML', 'year': 2025, 'parser_version': 2}))
+            db = Path(tmp) / 'absent.sqlite'
+            for action in [lambda: ingest_file(db, path), lambda: backfill(db, [path])]:
+                with self.assertRaisesRegex(ValueError, 'Recollect mixed'):
+                    action()
+            self.assertFalse(db.exists())
+
+    def test_changed_track_or_own_identity_rolls_back_import(self):
+        from paper_atlas.collectors.pmlr_tracks import require_track_record
+        row = record(); mapping, _ = program_tracks(program(event()), 2025)
+        record_track(row, mapping, self.provenance())
+        manifest = {'source': 'pmlr', 'venue': 'ICML', 'year': 2025, 'parser_version': 3,
+                    'track_program': self.provenance()}
+        require_track_record(row, manifest)
+        for change in [dict(track='position'), dict(identifiers={'openreview': 'otherForum'})]:
+            bad = copy.deepcopy(row); bad.update(change)
+            with self.assertRaises(ValueError):
+                require_track_record(bad, manifest)
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_public_openreview.py

@@ -0,0 +1,157 @@
+import hashlib
+import json
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from paper_atlas.collectors.public_openreview import (REVISION, SOURCE, collect,
+                                                     negative_status, normalize)
+from paper_atlas.db.store import Store
+from paper_atlas.ingestion.records import ingest_file
+from paper_atlas.audits.public_openreview import audit
+
+
+class PublicOpenReviewTests(unittest.TestCase):
+    def row(self, **overrides):
+        return {'id': 'RealForum01', 'title': 'A public paper', 'abstract': 'An abstract',
+                'author': 'Anonymous;Jane Doe;private@example.org', 'track': 'main',
+                'status': 'Reject', 'keywords': 'planning;reasoning',
+                'profiles': {'private': 'secret'}, 'reviews': ['private review'],
+                'gender': 'irrelevant', **overrides}
+
+    def provenance(self):
+        return {'source_file': 'https://example.org/pinned.jsonl', 'source_line': 7,
+                'source_revision': REVISION, 'source_blob_sha1': 'a' * 40}
+
+    def fetcher(self, files):
+        tree = []
+        payloads = {}
+        for name, rows in files.items():
+            data = ''.join(json.dumps(row) + '\n' for row in rows).encode()
+            sha = hashlib.sha1(b'blob ' + str(len(data)).encode() + b'\0' + data).hexdigest()
+            tree.append({'path': name, 'type': 'blob', 'size': len(data), 'sha': sha})
+            payloads[name] = data
+
+        def fetch(url, limit):
+            if '?recursive=1' in url:
+                return json.dumps({'sha': REVISION, 'truncated': False, 'tree': tree}).encode()
+            return payloads[url.split('/' + REVISION + '/')[1]]
+        return fetch
+
+    def collect(self, out, files, **kwargs):
+        with patch('paper_atlas.collectors.public_openreview.shutil.disk_usage', return_value=type('Space', (), {'free': 20 * 1024**3})()):
+            return collect(out, fetcher=self.fetcher(files), **kwargs)
+
+    def test_only_explicit_decisions_and_verified_scope_become_events(self):
+        for value in ('Pending', 'Active', 'Submitted to NeurIPS 2025', 'TBD', '',
+                      'Low score', 'ICML 2025 Retracted by Authors', 'Do not reject'):
+            self.assertIsNone(negative_status(value), value)
+        self.assertEqual(negative_status('Reject'), 'rejected')
+        self.assertEqual(negative_status('Desk rejected'), 'desk_rejected')
+        self.assertEqual(negative_status('NeurIPS 2023 Conference Withdrawn Submission'), 'withdrawn')
+        for raw, expected in [('main', 'main'), ('Position', 'position'),
+                              ('Datasets & Benchmarks', 'datasets_benchmarks')]:
+            row, reason = normalize(self.row(track=raw), 'nips', 2025, self.provenance())
+            self.assertIsNone(reason)
+            self.assertEqual(row['track'], expected)
+        for raw in ('workshop', 'workshop_mlsys', ''):
+            self.assertIsNone(normalize(self.row(track=raw), 'icml', 2025, self.provenance())[0])
+        with self.assertRaisesRegex(ValueError, 'conference year'):
+            normalize(self.row(status='NeurIPS 2023 Conference Withdrawn Submission'), 'icml', 2025, self.provenance())
+
+    def test_metadata_whitelist_retains_curated_provenance_without_profiles_or_reviews(self):
+        row, _ = normalize(self.row(), 'icml', 2025, self.provenance())
+        self.assertEqual(row['authors'], ['Jane Doe'])
+        self.assertEqual(row['identifiers'], {'openreview': 'RealForum01'})
+        self.assertEqual(row['metadata']['evidence_kind'], 'curated_public_status_assertion')
+        self.assertFalse(row['metadata']['original_decision_signature_verified'])
+        self.assertEqual(row['evidence_url'], 'https://example.org/pinned.jsonl#L7')
+        for private in ('secret', 'private review', 'gender', 'private@example.org'):
+            self.assertNotIn(private, json.dumps(row))
+
+    def test_pinned_collection_separates_tracks_and_holds_retraction(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            files = {'venues/icml/icml2025.jsonl': [self.row(), self.row(id='Position01', track='Position'),
+                     self.row(id='Retracted1', status='ICML 2025 Retracted by Authors'),
+                     self.row(id='Pending001', status='Submitted to ICML 2025')]}
+            result = self.collect(tmp, files)
+            self.assertEqual(result['input_rows'], 4)
+            self.assertEqual(result['normalized_records'], 2)
+            self.assertEqual({f['track'] for f in result['files']}, {'main', 'position'})
+            held = [json.loads(line) for line in Path(result['held_file']).read_text().splitlines()]
+            self.assertEqual([(r['id'], r['reason']) for r in held], [('Retracted1', 'retracted')])
+            self.assertTrue(all(f['status'] == 'partial' for f in result['files']))
+            with self.assertRaisesRegex(ValueError, 'Completed inventory'):
+                self.collect(tmp, files)
+
+    def test_hash_failure_or_duplicate_forum_publishes_no_ingestible_files(self):
+        files = {'venues/icml/icml2025.jsonl': [self.row()]}
+        fetch = self.fetcher(files)
+        def corrupted(url, limit):
+            result = fetch(url, limit)
+            return result if '?recursive' in url else result.replace(b'Reject', b'Accept')
+        for fetcher in (corrupted, self.fetcher({'venues/icml/icml2025.jsonl': [self.row(), self.row()]})):
+            with tempfile.TemporaryDirectory() as tmp, patch('paper_atlas.collectors.public_openreview.shutil.disk_usage', return_value=type('Space', (), {'free': 20 * 1024**3})()):
+                with self.assertRaises(ValueError):
+                    collect(tmp, fetcher=fetcher)
+                self.assertEqual(list(Path(tmp).glob('*.jsonl')), [])
+                self.assertFalse((Path(tmp) / 'public-openreview-inventory.json').exists())
+
+    def test_mirror_reimport_preserves_one_attempt_and_later_acceptance_elsewhere(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            result = self.collect(root / 'source', {'venues/icml/icml2025.jsonl': [self.row(author='Jane Doe')]})
+            db = root / 'db.sqlite'
+            path = result['files'][0]['file']
+            ingest_file(db, path)
+            ingest_file(db, path)
+            store = Store(db)
+            try:
+                paper = store.conn.execute('SELECT paper_id FROM source_records').fetchone()[0]
+                store.conn.execute('INSERT INTO identifiers VALUES(?,?,?)', ('arxiv', '2501.00001', paper))
+                same, record = store.record('official_program', 'neurips-later', 'https://example.org/accepted',
+                                            'A revised public paper', 'An abstract', ['Jane Doe'],
+                                            {'openreview': 'LaterForum1', 'arxiv': '2501.00001'})
+                self.assertEqual(same, paper)
+                store.decision(paper, record, 'NeurIPS', 2026, 'main', 'neurips-later',
+                               'https://example.org/accepted', 'accepted', 'Accept',
+                               identities=[('openreview', 'LaterForum1')])
+                store.conn.commit()
+                self.assertEqual(store.conn.execute('SELECT count(*) FROM submissions').fetchone()[0], 2)
+                self.assertEqual(store.conn.execute('SELECT count(*) FROM decisions').fetchone()[0], 2)
+                self.assertEqual(set(store.conn.execute('SELECT status FROM decisions')), {('accepted',), ('rejected',)})
+                self.assertEqual(store.conn.execute('SELECT count(*) FROM papers').fetchone()[0], 1)
+                self.assertEqual(store.conn.execute('PRAGMA foreign_key_check').fetchall(), [])
+            finally:
+                store.conn.close()
+
+    def test_preimport_audit_blocks_same_attempt_conflict_but_allows_later_other_venue_acceptance(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            self.collect(root / 'source', {'venues/icml/icml2025.jsonl': [self.row(author='Jane Doe')]})
+            db = root / 'db.sqlite'
+            store = Store(db)
+            paper, record = store.record('program', 'later', 'https://example.org/later',
+                                         'A public paper', 'An abstract', ['Jane Doe'], {'openreview': 'LaterForum1'})
+            store.decision(paper, record, 'NeurIPS', 2026, 'main', 'later',
+                           'https://example.org/later', 'accepted', 'Accept', identities=[('openreview', 'LaterForum1')])
+            store.conn.commit()
+            inventory = root / 'source/public-openreview-inventory.json'
+            first = audit(db, inventory, root / 'audit.json')
+            self.assertEqual(first['status'], 'ready_for_serial_import')
+            self.assertEqual(len(first['files']), 1)
+            store.decision(paper, record, 'ICML', 2025, 'main', 'same-attempt',
+                           'https://example.org/same', 'accepted', 'Accept', identities=[('openreview', 'RealForum01')])
+            store.conn.commit()
+            try:
+                second = audit(db, inventory, root / 'audit.json')
+                self.assertEqual(second['status'], 'blocked')
+                self.assertEqual(second['files'], [])
+                self.assertEqual(second['conflicts'][0]['existing_status'], 'accepted')
+            finally:
+                store.conn.close()
+
+
+if __name__ == '__main__':
+    unittest.main()