Migrate existing PMLR submission tracks with atomic provenance checks

docs/data-sources.md

@@ -300,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

@@ -20,20 +20,6 @@ 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. 每次采集先核对来源、许可、可访问性、空闲空间和现有运行批次。保持低并发，遵守来源限流、重试与人机验证要求。
@@ -61,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`，先对照主库和报告核查，不根据日志缺失自行重启任务。新路径重跑会重新核对当前数据库，已正确的轨道不会再次迁移。共享多个来源、重复目标事件或冲突决定会停止迁移，需另行核对，不能直接删掉其中一条。

paper_atlas/audits/arxiv_tags.py

@@ -1,143 +0,0 @@
-"""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/cli.py

@@ -27,10 +27,10 @@ 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','arxiv-tags'): ('paper_atlas.audits.arxiv_tags', {}, '逐记录核验 arXiv 类别及出处完整性'),
     ('audit','public-openreview'): ('paper_atlas.audits.public_openreview', {}, '核对公开负面决定与已有投稿历史'),
 }
 

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

@@ -1,138 +0,0 @@
-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_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()