Bound arXiv tag audit batches by primary key instead of sorting the source

# Conflicts:
#	paper_atlas/cli.py

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

docs/operations.md

@@ -20,6 +20,20 @@ paper-atlas maintain arxiv-tags --db /data/paper-index/papers.sqlite \
 
 修复使用同一维护锁，分批提交并保存断点，保留 `custom:*` 和其他来源证据；失败后确认原因已解决，再增加 `--resume` 从上次提交位置继续。完成报告不能作为重新扫描的断点，重新核验应指定新报告。源审计和标签修复均保留 8 GiB 空闲空间，不能代替所有来源入库后的实体、投稿去重与完整性检查。
 
+修复结束后还须核对每条 arXiv 来源记录的类别集合，不能只看旧证据键是否清零。下面的只读审计比较来源 metadata 中的类别、所属论文实体和逐记录证据键，同时检查没有合法 arXiv 来源的类别行：
+
+```sh
+paper-atlas audit arxiv-tags --db /data/paper-index/papers.sqlite \
+  --repair-report /data/paper-index/collected/arxiv-tag-repair.json \
+  --out /data/paper-index/collected/arxiv-tag-coverage-audit.json
+```
+
+审计持有共享维护锁；修复或其他协调写入尚未结束时会拒绝运行。提供修复报告时，还会核对其已完成状态、数据库路径和固定来源范围。审计以一致的只读快照分批读取，记录完整扫描数量、缺失/多余证据、错误论文归属、残留旧键及少量异常样例，不修改主库。
+
+来源记录按整数主键区间读取，避免 SQLite 误选按 external_id 排列的来源索引、每批重新排序整个来源；其他来源交错在编号区间内时仍按 source_id 过滤。类别证据继续使用 source_record_id 索引批量读取。完整核验范围不因查询优化缩小。
+
+只有全量记录核对完毕且无差异时才标记 `passed`；存在差异为 `failed` 并返回非零退出码，解析或执行中断为 `error`。运行中的报告不能作为完成证明。重新核验须使用新输出路径，不能覆盖已有报告；该检查不替代最后的实体去重、外键、SQLite 与全文索引核验。
+
 ## 长期维护与更新
 
 1. 每次采集先核对来源、许可、可访问性、空闲空间和现有运行批次。保持低并发，遵守来源限流、重试与人机验证要求。

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/arxiv_tags.py

@@ -0,0 +1,143 @@
+"""Compare every arXiv record's category evidence with retained source metadata."""
+import argparse
+from collections import defaultdict
+from contextlib import closing
+import fcntl
+import json
+from pathlib import Path
+import sqlite3
+
+from paper_atlas.db.submissions import timestamp
+
+
+def save(path, state):
+    part = path.with_suffix('.part')
+    part.write_text(json.dumps(state, ensure_ascii=False, indent=2) + '\n')
+    part.replace(path)
+
+
+def audit(db, out, repair_report=None, batch_size=500):
+    db, out = Path(db).resolve(), Path(out).resolve()
+    if not db.is_file():
+        raise ValueError('Existing database required')
+    if not 1 <= batch_size <= 500:
+        raise ValueError('Batch size must be between 1 and 500')
+    if out == db or repair_report and out == Path(repair_report).resolve():
+        raise ValueError('Audit output must preserve its inputs')
+    out.parent.mkdir(parents=True, exist_ok=True)
+    with Path(str(db) + '.maintenance.lock').open('a') as maintenance, Path(str(out) + '.lock').open('a') as output_lock:
+        # A partial repair is not a useful full-category audit. This also keeps
+        # cooperating maintenance writers out for the entire read-only snapshot.
+        fcntl.flock(maintenance, fcntl.LOCK_SH | fcntl.LOCK_NB)
+        fcntl.flock(output_lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        if out.exists():
+            raise ValueError('Audit report exists; select a new output path')
+        with closing(sqlite3.connect(db.as_uri() + '?mode=ro', uri=True)) as c:
+            c.execute('PRAGMA cache_size=-32768')
+            c.execute('BEGIN')
+            source = c.execute("SELECT id FROM sources WHERE name='arxiv'").fetchone()
+            if source is None:
+                raise ValueError('No imported arXiv source')
+            total, maximum = c.execute('SELECT count(*),COALESCE(max(id),0) FROM source_records WHERE source_id=?', source).fetchone()
+            if not total:
+                raise ValueError('No imported arXiv records')
+            scope = {'db': str(db), 'source_id': source[0], 'source_record_count': total,
+                     'maximum_source_record_id': maximum}
+            if repair_report:
+                repair = json.loads(Path(repair_report).read_text())
+                if repair.get('status') != 'complete' or repair.get('records_refreshed') != total:
+                    raise ValueError('Category repair has not completed this source inventory')
+                if any(repair.get(key) != value for key, value in scope.items()):
+                    raise ValueError('Repair report belongs to a different source inventory')
+            state = {**scope, 'started_at': timestamp(), 'status': 'running', 'records_checked': 0,
+                     'last_record_id': 0, 'expected_evidence': 0, 'actual_evidence': 0,
+                     'records_with_mismatches': 0, 'missing_evidence': 0, 'unexpected_evidence': 0,
+                     'legacy_evidence': 0, 'examples': [], 'repair_report': str(Path(repair_report).resolve()) if repair_report else None}
+            save(out, state)
+            try:
+                while True:
+                    # The (source_id,external_id) index orders by external ID,
+                    # so choosing it sorts the entire source again per batch.
+                    # This fixed ID interval must use the integer primary key.
+                    records = c.execute('''SELECT id,external_id,paper_id,metadata_json FROM source_records NOT INDEXED
+                        WHERE source_id=? AND id>? AND id<=? ORDER BY id LIMIT ?''',
+                        (source[0], state['last_record_id'], maximum, batch_size)).fetchall()
+                    if not records:
+                        break
+                    ids = [row[0] for row in records]
+                    actual = defaultdict(set)
+                    for record, paper, category, evidence in c.execute('''SELECT pt.source_record_id,pt.paper_id,t.value,pt.evidence_key
+                        FROM paper_tags pt JOIN tags t ON t.id=pt.tag_id WHERE t.namespace='arxiv:category'
+                        AND pt.source_record_id IN (''' + ','.join('?' for _ in ids) + ')', ids):
+                        actual[record].add((category, paper, evidence))
+                        state['legacy_evidence'] += evidence == 'arxiv:categories'
+                    for record, external, paper, payload in records:
+                        metadata = json.loads(payload)
+                        if not isinstance(metadata, dict):
+                            raise ValueError('arXiv source metadata must be an object')
+                        categories = metadata.get('categories')
+                        if categories is None:
+                            categories = ''
+                        if not isinstance(categories, str):
+                            raise ValueError('arXiv source categories must be text')
+                        evidence_key = f'arxiv:record:{record}:tags'
+                        expected = {(category, paper, evidence_key) for category in categories.split()}
+                        found = actual[record]
+                        missing, unexpected = expected - found, found - expected
+                        state['expected_evidence'] += len(expected)
+                        state['actual_evidence'] += len(found)
+                        state['missing_evidence'] += len(missing)
+                        state['unexpected_evidence'] += len(unexpected)
+                        if missing or unexpected:
+                            state['records_with_mismatches'] += 1
+                            if len(state['examples']) < 20:
+                                state['examples'].append({'record_id': record, 'external_id': external,
+                                    'paper_id': paper, 'missing': sorted(missing), 'unexpected': sorted(unexpected)})
+                        state['records_checked'] += 1
+                    state['last_record_id'] = ids[-1]
+                    if state['records_checked'] % 10000 < len(records):
+                        save(out, state)
+                        print(json.dumps({k: state[k] for k in ('status', 'records_checked', 'source_record_count', 'records_with_mismatches')}), flush=True)
+                if state['records_checked'] != total:
+                    raise ValueError('Audited record count differs from the fixed source inventory')
+                # Every category row must also have an arXiv source-record owner.
+                # The full namespace total catches orphan and wrong-source evidence
+                # that cannot appear in the per-record traversal above.
+                all_evidence = c.execute("SELECT count(*) FROM paper_tags pt JOIN tags t ON t.id=pt.tag_id WHERE t.namespace='arxiv:category'").fetchone()[0]
+                unscoped = all_evidence - state['actual_evidence']
+                if unscoped < 0:
+                    raise ValueError('Category evidence totals are inconsistent')
+                state.update(namespace_evidence=all_evidence, evidence_without_arxiv_record=unscoped)
+                if unscoped:
+                    state['unscoped_examples'] = [dict(zip(('paper_id', 'category', 'source_record_id', 'evidence_key'), row))
+                        for row in c.execute('''SELECT pt.paper_id,t.value,pt.source_record_id,pt.evidence_key
+                            FROM paper_tags pt JOIN tags t ON t.id=pt.tag_id
+                            LEFT JOIN source_records r ON r.id=pt.source_record_id
+                            WHERE t.namespace='arxiv:category' AND (r.source_id IS NULL OR r.source_id<>?) LIMIT 20''', source)]
+                passed = not state['records_with_mismatches'] and not unscoped
+                state.update(status='passed' if passed else 'failed', finished_at=timestamp(),
+                             note='Full per-record category, canonical paper and evidence-key equality. '
+                                  'Read-only; not a global paper deduplication or full database-integrity audit.')
+                save(out, state)
+            except BaseException as error:
+                state.update(status='error', error=str(error), finished_at=timestamp())
+                save(out, state)
+                raise
+    print(json.dumps(state), flush=True)
+    return state
+
+
+def main(argv=None, *, prog=None):
+    parser = argparse.ArgumentParser(prog=prog)
+    parser.add_argument('--db', required=True)
+    parser.add_argument('--out', required=True)
+    parser.add_argument('--repair-report')
+    parser.add_argument('--batch-size', type=int, default=500)
+    args = parser.parse_args(argv)
+    result = audit(args.db, args.out, args.repair_report, args.batch_size)
+    if result['status'] != 'passed':
+        raise SystemExit('arXiv category evidence audit failed; inspect the report')
+
+
+if __name__ == '__main__':
+    main()

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 官方论文集'),
@@ -28,6 +30,8 @@ COMMANDS = {
     ('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','arxiv-tags'): ('paper_atlas.audits.arxiv_tags', {}, '逐记录核验 arXiv 类别及出处完整性'),
+    ('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/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()

tests/test_arxiv_tag_audit.py

@@ -0,0 +1,138 @@
+import fcntl
+import json
+from pathlib import Path
+import tempfile
+import unittest
+import sqlite3
+from unittest.mock import patch
+
+from paper_atlas.audits.arxiv_tags import audit, main
+from paper_atlas.db.store import Store
+from paper_atlas.ingestion.arxiv import refresh_categories
+
+
+class ArxivTagAuditTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.root = Path(self.temp.name)
+        self.db = self.root / 'papers.sqlite'
+        self.out = self.root / 'audit.json'
+        self.store = Store(self.db)
+        self.addCleanup(self.store.conn.close)
+        self.records = []
+        for external in ('one', 'two'):
+            paper, record = self.store.record('arxiv', external, 'https://arxiv.org/abs/' + external,
+                'Same research work', 'Abstract', [], {'doi': '10.1234/same'},
+                {'categories': 'cs.AI cs.LG cs.AI'})
+            refresh_categories(self.store, paper, record, 'cs.AI cs.LG cs.AI')
+            self.records.append(record)
+        self.paper = paper
+        tag = self.store.conn.execute("INSERT INTO tags(namespace,value) VALUES('custom:topic','cs.AI')").lastrowid
+        self.store.conn.execute("INSERT INTO paper_tags VALUES(?,?,NULL,'manual:1')", (paper, tag))
+        self.store.conn.commit()
+
+    def test_all_merged_records_keep_their_own_categories_and_custom_tags_are_untouched(self):
+        before = self.store.conn.execute('SELECT * FROM paper_tags ORDER BY evidence_key,tag_id').fetchall()
+        result = audit(self.db, self.out, batch_size=1)
+        self.assertEqual(result['status'], 'passed')
+        self.assertEqual(result['records_checked'], 2)
+        self.assertEqual(result['expected_evidence'], 4)
+        self.assertEqual(result['actual_evidence'], 4)
+        self.assertEqual(result['evidence_without_arxiv_record'], 0)
+        self.assertEqual(self.store.conn.execute('SELECT * FROM paper_tags ORDER BY evidence_key,tag_id').fetchall(), before)
+
+    def test_zero_legacy_keys_does_not_hide_missing_categories_and_cli_fails(self):
+        self.store.conn.execute("DELETE FROM paper_tags WHERE source_record_id IS NOT NULL")
+        self.store.conn.commit()
+        with self.assertRaises(SystemExit):
+            main(['--db', str(self.db), '--out', str(self.out)])
+        result = json.loads(self.out.read_text())
+        self.assertEqual(result['status'], 'failed')
+        self.assertEqual(result['legacy_evidence'], 0)
+        self.assertEqual(result['missing_evidence'], 4)
+        self.assertEqual(result['records_with_mismatches'], 2)
+
+    def test_wrong_canonical_paper_or_stale_key_is_detected(self):
+        other, _ = self.store.record('example', 'other', 'https://example.org', 'Different work', 'AB', [])
+        self.store.conn.execute("UPDATE paper_tags SET paper_id=?,evidence_key='arxiv:categories' WHERE source_record_id=?", (other, self.records[0]))
+        self.store.conn.commit()
+        result = audit(self.db, self.out)
+        self.assertEqual(result['status'], 'failed')
+        self.assertEqual(result['records_with_mismatches'], 1)
+        self.assertEqual(result['legacy_evidence'], 2)
+        self.assertEqual(result['missing_evidence'], 2)
+        self.assertEqual(result['unexpected_evidence'], 2)
+
+    def test_wrong_source_and_unowned_category_rows_cannot_escape_the_full_audit(self):
+        paper, record = self.store.record('example', 'other', 'https://example.org', 'Other work', 'AB', [])
+        tag = self.store.conn.execute("SELECT id FROM tags WHERE namespace='arxiv:category' AND value='cs.AI'").fetchone()[0]
+        self.store.conn.execute('INSERT INTO paper_tags VALUES(?,?,?,?)', (paper, tag, record, 'wrong-source'))
+        self.store.conn.execute('INSERT INTO paper_tags VALUES(?,?,NULL,?)', (paper, tag, 'unowned'))
+        self.store.conn.commit()
+        result = audit(self.db, self.out)
+        self.assertEqual(result['status'], 'failed')
+        self.assertEqual(result['records_with_mismatches'], 0)
+        self.assertEqual(result['evidence_without_arxiv_record'], 2)
+        self.assertEqual(len(result['unscoped_examples']), 2)
+
+    def test_active_writer_and_incomplete_repair_are_rejected_before_publishing(self):
+        with Path(str(self.db) + '.maintenance.lock').open('a') as lock:
+            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            with self.assertRaises(BlockingIOError):
+                audit(self.db, self.out)
+        self.assertFalse(self.out.exists())
+        repair = self.root / 'repair.json'
+        repair.write_text(json.dumps({'status': 'running'}))
+        with self.assertRaisesRegex(ValueError, 'not completed'):
+            audit(self.db, self.out, repair)
+        self.assertFalse(self.out.exists())
+
+    def test_repair_report_scope_and_completed_audit_are_not_silently_reused(self):
+        repair = self.root / 'repair.json'
+        repair.write_text(json.dumps({'status': 'complete', 'records_refreshed': 2, 'db': str(self.db),
+            'source_id': 1, 'source_record_count': 2, 'maximum_source_record_id': 999}))
+        with self.assertRaisesRegex(ValueError, 'different source inventory'):
+            audit(self.db, self.out, repair)
+        result = audit(self.db, self.out)
+        before = self.out.read_bytes()
+        with self.assertRaisesRegex(ValueError, 'report exists'):
+            audit(self.db, self.out)
+        self.assertEqual(self.out.read_bytes(), before)
+        self.assertEqual(result['status'], 'passed')
+
+    def test_invalid_source_metadata_is_an_error_not_a_partial_pass(self):
+        self.store.conn.execute('UPDATE source_records SET metadata_json=? WHERE id=?',
+                                (json.dumps({'categories': ['invalid']}), self.records[0]))
+        self.store.conn.commit()
+        with self.assertRaisesRegex(ValueError, 'categories must be text'):
+            audit(self.db, self.out)
+        self.assertEqual(json.loads(self.out.read_text())['status'], 'error')
+
+    def test_interleaved_sources_use_bounded_primary_key_batches_without_sorting(self):
+        self.store.record('example', 'interleaved', 'https://example.org', 'Other source', 'AB', [])
+        paper, record = self.store.record('arxiv', 'three', 'https://arxiv.org/abs/three',
+                                          'A third record', 'AB', [], metadata={'categories': 'cs.AI'})
+        refresh_categories(self.store, paper, record, 'cs.AI')
+        self.store.conn.commit()
+        queries = []
+        connect = sqlite3.connect
+        def traced(*args, **kwargs):
+            connection = connect(*args, **kwargs)
+            connection.set_trace_callback(queries.append)
+            return connection
+        with patch('paper_atlas.audits.arxiv_tags.sqlite3.connect', side_effect=traced):
+            result = audit(self.db, self.out, batch_size=1)
+        self.assertEqual(result['status'], 'passed')
+        self.assertEqual(result['records_checked'], 3)
+        self.assertEqual(result['last_record_id'], record)
+        batches = [sql for sql in queries if sql.startswith('SELECT id,external_id,paper_id,metadata_json')]
+        self.assertEqual(len(batches), 4)  # Includes the bounded final empty read.
+        for sql in batches:
+            plan = ' '.join(row[3] for row in self.store.conn.execute('EXPLAIN QUERY PLAN ' + sql))
+            self.assertIn('INTEGER PRIMARY KEY', plan)
+            self.assertNotIn('TEMP B-TREE', plan)
+
+
+if __name__ == '__main__':
+    unittest.main()

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_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()