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

# Conflicts:
#	paper_atlas/cli.py

.gitea/workflows/check.yml

@@ -43,7 +43,7 @@ jobs:
               outside.mkdir()
               subprocess.run([str(venv/'bin/paper-atlas'),'--help'],cwd=outside,check=True)
               subprocess.run([python,'-I','-B','-m','paper_atlas','collect','cvf','--help'],cwd=outside,check=True)
-              smoke = "from paper_atlas.db.store import Store; from paper_atlas.config import DEFAULT_REGISTRY; import json; assert {'COLM','MM','MLSys'} <= {v['acronym'] for v in json.loads(DEFAULT_REGISTRY.read_text())['venues']}; s=Store('installed.sqlite'); s.conn.close()"
+              smoke = "from paper_atlas.db.store import Store; from paper_atlas.config import DEFAULT_REGISTRY; import json; assert len(json.loads(DEFAULT_REGISTRY.read_text())['venues'])==41; s=Store('installed.sqlite'); s.conn.close()"
               subprocess.run([python,'-I','-B','-c',smoke],cwd=outside,check=True)
               subprocess.run([python,'-I','-B','-m','paper_atlas','query','--db','installed.sqlite','--stats'],cwd=outside,check=True)
           PY

README.md

@@ -4,7 +4,7 @@
 
 Paper Atlas 将 arXiv、AI 会议论文集及公开投稿决定整理到可追溯的 SQLite 数据库中，用于查询和后续研究分析。它按批次运行，不需要常驻 Web 服务；采集标题、摘要、标识、标签及决定证据，不下载论文全文。
 
-项目仍在建设中。候选范围包含 43 个会议（含 ACM MM、MLSys），尚未完成全部历史数据和最终去重复核。实际进度以数据库、来源清单和覆盖报告为准，而不是仓库文件数量。
+项目仍在建设中。候选范围包含 41 个会议，尚未完成全部历史数据和最终去重复核。实际进度以数据库、来源清单和覆盖报告为准，而不是仓库文件数量。
 
 ## 项目结构
 

docs/data-sources.md

@@ -16,8 +16,6 @@
 - [ACL Anthology 权威 XML 与 API](https://aclanthology.org/info/development/)
 - [NeurIPS](https://proceedings.neurips.cc/)
 - [COLM](https://proceedings.colm.cc/)
-- [MLSys](https://proceedings.mlsys.org/)
-- [ACM Multimedia 历史目录](https://dblp.org/db/conf/mm/index.html)
 - [CVF](https://openaccess.thecvf.com/)
 - [ECVA](https://www.ecva.net/papers.php)
 - [IJCAI](https://www.ijcai.org/all_proceedings/)
@@ -31,7 +29,6 @@
 | 机器学习、理论 | NeurIPS、ICML、ICLR、COLM、COLT、AISTATS、UAI、CLeaR | OpenReview、PMLR、NeurIPS、COLM 官方论文集 |
 | 自然语言 | ACL、EMNLP、NAACL、EACL、COLING、CoNLL | ACL Anthology |
 | 视觉 | CVPR、ICCV、ECCV | CVF、ECVA |
-| 多模态与机器学习系统 | ACM MM、MLSys | ACM / DBLP、MLSys 官方论文集 |
 | 综合 AI | AAAI、IJCAI、ECAI | 官方论文集、DBLP |
 | 检索与数据 | KDD、SIGIR、WWW、WSDM、RecSys、CIKM、ICDM、SDM、ECML-PKDD | 官方出版方、DBLP，具体年份入口待逐项登记 |
 | 机器人 | ICRA、IROS、RSS、CoRL | 官方论文集、PMLR、DBLP |
@@ -41,23 +38,6 @@
 
 这些是领域覆盖候选，不构成统一排名，也不能据此声称已全部入库。逐年目录以真实举办届次和正式出版清单为准：双年会议不补造不存在的年份；主会、Findings、workshop 分开。`edition_sources` 保存实际采集状态，不能用候选名单代替覆盖率报告。
 
-## ACM MM 与 MLSys
-
-会议注册表包含 `MM`（别名 `ACMMM`、`ACM MM`）和 `MLSys`。MLSys 官方根目录目前列出 2019–2026 年的 8 卷，直接采集每篇标题、摘要、公开作者和引用元数据：
-
-```sh
-paper-atlas collect mlsys --cache /data/paper-index/raw/http-cache \
-  --out /data/paper-index/collected/mlsys
-```
-
-每年清单核对官方声明数量，缺摘要标为 partial；未知现代轨道、空目录或读取失败不能记为成功。`SysML 2019` 归于同一 MLSys 系列并保留别名。2018 年官方说明接受的投稿为 non-archival，论文集从 2019 年开始；不拼造 2018 年归档卷，也不把未出现在论文集中的投稿判为拒稿。参见 [2018 年官方说明](https://mlsys.org/Conferences/2018/)。
-
-MLSys 老式 `/paper/<year>/hash/…` 与新式 `/paper_files/paper/<year>/hash/…` 链接规范为同一个 `publication_mlsys` 标识，也支持 2023 年实际使用的 `-Abstract-mlsys2023.html` 及对应年份标签；标签年份必须与目录年份一致。DBLP 和官方摘要源共用它，避免同一发表记录成为两个论文实体或两次投稿。只从自身出版社链接提取，不从摘要参考文献提取，也不把发表标识用于合并某次拒稿尝试。共享投稿身份入口要求 MLSys 源具有有效的同年出版社身份，缺失或错年的文件不能生成投稿事件；旧版 DBLP 整理文件可利用已保留的唯一自身出版社链接补得此身份，不需要丢弃来源或依靠标题猜测。
-
-ACM MM 使用固定 DBLP 原始快照和已核验的 proceedings crossref 盘点历史，主会 `ACM Multimedia`、历史编号分卷与同系列 workshop 分开。主会卷也包含 Doctoral Symposium、Reproducibility Companion Papers、Tutorials 等章节（参见 [2025 年目录](https://dblp.org/db/conf/mm/mm2025.html)），而原始 XML 不含逐篇章节，所以这类记录暂标 `proceedings_unspecified`，不能整卷当作 regular 主轨道；后续需逐篇目录证据补齐。旧版候选 JSONL 没有 `mm` / `mlsys` 家族，不能用它假装覆盖新增会议；应从已经下载并校验的原始 XML 重新筛选这两个会议，保留完整扫描数量、卷证据与遗漏原因。DBLP 没有摘要，后续须通过 DOI、真实 forum 等独立标识关联官方或公开元数据，不能将书目导入成功表述为摘要已齐全。
-
-公开拒稿镜像的 `acmmm` 家族映射到 `MM`，只接受明确的负面状态，继续保留镜像与原始签名决定的区别。现有固定 OpenReview 镜像只含 ACM MM 2024 的录用记录，没有据此证明其他年份没有公开拒稿；[2025 年官方 CFP](https://acmmm2025.org/call-for-papers/) 仅说明审稿期间不公开，不能外推为会后永不公开。
-
 ## 官方会议元数据采集
 
 ```sh

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. 每次采集先核对来源、许可、可访问性、空闲空间和现有运行批次。保持低并发，遵守来源限流、重试与人机验证要求。

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

@@ -18,7 +18,6 @@ COMMANDS = {
     ('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 官方论文集'),
-    ('collect','mlsys'): ('paper_atlas.collectors.proceedings', {'source':'mlsys'}, 'MLSys 官方历年标题与摘要'),
     ('collect','cvf'): ('paper_atlas.collectors.vision', {'source':'cvf'}, 'CVPR / ICCV 官方开放目录'),
     ('collect','ecva'): ('paper_atlas.collectors.vision', {'source':'ecva'}, 'ECCV 官方开放目录'),
     ('collect','dblp'): ('paper_atlas.collectors.dblp_records', {}, '核对并整理 DBLP 历史发表记录'),
@@ -31,6 +30,7 @@ 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/dblp.py

@@ -20,7 +20,7 @@ DTD_MD5='96e4283e71886dec2173718957293648'
 DTD_NAMES={'dblp.dtd','dblp-2023-06-28.dtd'}
 RELEASE='https://doi.org/10.4230/dblp.xml.2026-11-01'
 # Broad discovery candidates, including historical names; not a venue mapping.
-FAMILIES=set('nips icml iclr colm colt aistats uai clear acl emnlp naacl eacl coling conll cvpr iccv eccv aaai ijcai ecai kdd sigir www wsdm recsys cikm icdm sdm ecml pkdd ecmlpkdd icra iros rss corl aips icaps kr atal aamas icassp interspeech eurospeech icslp ismir miccai fat facct lrec mm mlsys'.split())
+FAMILIES=set('nips icml iclr colm colt aistats uai clear acl emnlp naacl eacl coling conll cvpr iccv eccv aaai ijcai ecai kdd sigir www wsdm recsys cikm icdm sdm ecml pkdd ecmlpkdd icra iros rss corl aips icaps kr atal aamas icassp interspeech eurospeech icslp ismir miccai fat facct lrec'.split())
 
 
 def records(archive,dtd_bytes):

paper_atlas/collectors/dblp_records.py

@@ -17,7 +17,6 @@ from paper_atlas.collectors.dblp import XML_SHA,DTD_MD5,RELEASE,records
 from paper_atlas.collectors.dblp_scope import scope
 from paper_atlas.db.submissions import timestamp
 from paper_atlas.io.snapshots import digest_file
-from paper_atlas.io.publication_ids import mlsys_id
 
 AS_OF='2026-11-01T00:00:00.000Z'
 MATERIAL_PART=r'(?:title(?: page)?|preface|foreword|editorial|front\s?matter|back\s?matter|(?:table of )?contents|(?:author |subject )?index|contributors|(?:conference |program |organizing |steering )?committee|conference organization|reviewers)'
@@ -37,10 +36,8 @@ def material_reason(title,authors,key,year,track):
 
 
 def identifiers(fields):
-    dois=set();arxiv=set();forums=set();mlsys=set()
+    dois=set();arxiv=set();forums=set()
     for link in fields.get('ee',[]):
-        if own_mlsys := mlsys_id(link):
-            mlsys.add(own_mlsys)
         parsed=urlsplit(link);path=unquote(parsed.path)
         doi=None
         if parsed.netloc in ('doi.org','dx.doi.org','doi.ieeecomputersociety.org'):doi=path.lstrip('/')
@@ -60,10 +57,9 @@ def identifiers(fields):
         if match:arxiv.add(re.sub(r'v\d+$','',match[1]))
     publications={doi for doi in dois if not doi.startswith('10.48550/arxiv.')}
     own={}
-    groups=(('doi',publications),('arxiv',arxiv),('openreview',forums),('publication_mlsys',mlsys))
-    for scheme,values in groups:
+    for scheme,values in (('doi',publications),('arxiv',arxiv),('openreview',forums)):
         if len(values)==1:own[scheme]=next(iter(values))
-    ambiguous={scheme:sorted(values) for scheme,values in groups if len(values)>1}
+    ambiguous={scheme:sorted(values) for scheme,values in (('doi',publications),('arxiv',arxiv),('openreview',forums)) if len(values)>1}
     return own,ambiguous
 
 
@@ -88,8 +84,6 @@ def normalize_row(row,volumes,targets,include_workshops=False,through_year=2026)
               'scope_rule':edition['rule'],'author_names_raw':fields.get('author',[]),'publtype':row.get('publtype'),
               'bibliography':{key:value for key,value in fields.items() if key in ('booktitle','year','ee','url','pages','volume','number','publisher','series','isbn')},
               'ambiguous_identifiers':ambiguous,'publication_evidence':'inproceedings bibliography; not an original decision note'}
-    if edition.get('track_scope_note'):
-        metadata['track_scope_note']=edition['track_scope_note']
     flagged=bool(re.search(r'\b(?:withdrawn|retracted)\b',row.get('publtype') or '',re.I))
     result={'source':'dblp','external_id':row['key'],'url':'https://dblp.org/rec/'+row['key']+'.html',
             'title':title,'abstract':None,'authors':authors,'identifiers':own,
@@ -116,12 +110,6 @@ def collect(catalogue,out,venues=None,candidates=None,archive=None,dtd=None,incl
     allowed={row['acronym'].casefold():row['acronym'] for row in json.loads(DEFAULT_REGISTRY.read_text())['venues']}
     if venues and set(name.casefold() for name in venues)-allowed.keys():raise ValueError('Unknown registry venue')
     targets={allowed[name.casefold()] for name in venues} if venues else set(allowed.values())
-    if candidates:
-        families = manifest.get('candidate_family_counts', {})
-        missing = [venue for venue, family in [('MM', 'mm'), ('MLSys', 'mlsys')]
-                   if venue in targets and not families.get(family)]
-        if missing:
-            raise ValueError('Candidate file predates requested families ' + ', '.join(missing) + '; use the original --archive')
     out=Path(out);out.mkdir(parents=True,exist_ok=True)
     with (out/'dblp.normalize.lock').open('a') as lock:
         fcntl.flock(lock,fcntl.LOCK_EX|fcntl.LOCK_NB)

paper_atlas/collectors/dblp_scope.py

@@ -33,8 +33,6 @@ BOOKS = {
     'interspeech':[('INTERSPEECH','INTERSPEECH',[]),('EUROSPEECH','EUROSPEECH',['INTERSPEECH']),('ICSLP','ICSLP',['INTERSPEECH']),('ECST','ECST',['INTERSPEECH'])],
     'ismir':[('ISMIR','ISMIR',[])], 'miccai':[('MICCAI','MICCAI',[])],
     'fat':[('FAccT','FAccT',[]),('FAT*','FAccT',[]),('FAT','FAccT',[])],
-    'mm':[('ACM Multimedia','MM',['ACMMM','ACM MM'])],
-    'mlsys':[('MLSys','MLSys',[]),('SysML','MLSys',['SysML'])],
 }
 
 
@@ -47,7 +45,6 @@ def scope(row,volume,through_year=2026):
     if not volume:return None,'missing_proceedings'
     title=' '.join(volume['fields'].get('title',[]))
     if family=='clear2' and 'causal learning and reasoning' not in title.casefold():return None,'ambiguous_venue_name'
-    if family=='mlsys' and 'machine learning and systems' not in title.casefold():return None,'ambiguous_venue_name'
     matched=None
     for base,venue,aliases in BOOKS[family]:
         match=re.fullmatch(re.escape(base)+r'(?:\s+\((.*)\)|\s+(.*)|,\s*(Vol\.\s*\d+))?',book,re.I)
@@ -70,7 +67,6 @@ def scope(row,volume,through_year=2026):
         if year%100!=int(short[1]):return None,'unverified_edition_year'
         year_basis='unique explicit event year in proceedings title, corroborated by two-digit proceedings key'
     if year>through_year:return None,'future_edition'
-    if venue=='MLSys' and year<2019:return None,'nonarchival_predecessor'
     bibliographic=fields.get('year',[])
     if len(bibliographic)!=1 or not re.fullmatch(r'\d{4}',bibliographic[0]):return None,'ambiguous_bibliographic_year'
     if int(bibliographic[0])!=year and not re.search(r'\b'+str(year)+r'\b',title):return None,'unverified_year_difference'
@@ -96,11 +92,5 @@ def scope(row,volume,through_year=2026):
     elif suffix.casefold() in ('poster','posters') and venue not in ('ICLR','ACL','EMNLP','NAACL','EACL','COLING','CoNLL'):track='poster'
     if venue=='AAAI+IAAI':track='proceedings_unspecified'
     if venue=='AAMAS' and year<2002:return None,'predecessor_not_aamas'
-    result = {'venue':venue,'venue_aliases':aliases,'year':year,'track':track,'bibliographic_year':int(bibliographic[0]),
-              'rule':family+':'+base,'proceedings_title':title,'edition_year_basis':year_basis}
-    if venue=='MM' and track=='main':
-        # A main MM volume also includes doctoral, tutorials, companion papers
-        # and other sections. DBLP XML omits the per-paper TOC section.
-        result['track']='proceedings_unspecified'
-        result['track_scope_note']='Main proceedings volume; per-paper section is absent from the bibliography. Do not infer a regular research track.'
-    return result,None
+    return {'venue':venue,'venue_aliases':aliases,'year':year,'track':track,'bibliographic_year':int(bibliographic[0]),
+            'rule':family+':'+base,'proceedings_title':title,'edition_year_basis':year_basis},None

paper_atlas/collectors/proceedings.py

@@ -1,4 +1,4 @@
-"""Collect COLM, NeurIPS and MLSys official metadata without downloading PDFs."""
+"""Collect COLM/NeurIPS official abstract pages without downloading PDFs."""
 import argparse
 import json
 import re
@@ -8,11 +8,8 @@ from pathlib import Path
 from urllib.parse import urljoin,urlsplit
 
 from paper_atlas.io.http import fetch
-from paper_atlas.io.publication_ids import mlsys_id
 
-ROOTS={'colm':'https://proceedings.colm.cc/','neurips':'https://proceedings.neurips.cc/',
-       'mlsys':'https://proceedings.mlsys.org/'}
-VENUES={'colm':'COLM','neurips':'NeurIPS','mlsys':'MLSys'}
+ROOTS={'colm':'https://proceedings.colm.cc/','neurips':'https://proceedings.neurips.cc/'}
 
 
 from paper_atlas.io.html import Document,Node,clean
@@ -75,22 +72,6 @@ def parse_paper(html,url,source,year,index_url):
     track_text=' '.join(clean(n.text()) for n in nodes if 'paper-track' in n.attrs.get('class','').split())
     track=track_key(track_text or url)
     identifiers={}
-    if source == 'mlsys':
-        if year < 2019:
-            raise ValueError('SysML 2018 was non-archival; official proceedings begin in 2019')
-        own = mlsys_id(url)
-        if not own or int(own.split('/')[0]) != year:
-            raise ValueError('MLSys paper URL does not identify this publication year')
-        identifiers['publication_mlsys'] = own
-        if track_text:
-            explicit = {'conference': 'main', 'main track': 'main', 'research track': 'main',
-                        'industry track': 'industry', 'industrial track': 'industry',
-                        'mlsys' + str(year): 'main'}
-            if track_text.casefold() not in explicit:
-                raise ValueError('Unclassified official MLSys track: ' + track_text)
-            track = explicit[track_text.casefold()]
-        elif year >= 2025 and not url.endswith('-Abstract-Conference.html'):
-            raise ValueError('Modern MLSys paper lacks an explicit conference track')
     if meta['citation_doi']:
         identifiers['doi']=meta['citation_doi'][0]
     action_nodes=[]
@@ -109,11 +90,8 @@ def parse_paper(html,url,source,year,index_url):
             from urllib.parse import parse_qs
             forum=parse_qs(parsed.query).get('id')
             if forum: identifiers['openreview']=forum[0]
-    venue=VENUES[source]
-    result = {'source':source,'external_id':urlsplit(url).path,'url':url,'title':clean(title),'abstract':abstract,'authors':authors,'identifiers':identifiers,'venue':venue,'year':year,'track':track,'status':'accepted','raw_decision':'listed in official proceedings','list_url':index_url,'metadata':{'citation':dict(meta),'publication_state':'pre-proceedings' if 'pre-proceedings' in html.lower() else 'proceedings'}}
-    if source == 'mlsys' and year == 2019:
-        result['venue_aliases'] = ['SysML']
-    return result
+    venue='COLM' if source=='colm' else 'NeurIPS'
+    return {'source':source,'external_id':urlsplit(url).path,'url':url,'title':clean(title),'abstract':abstract,'authors':authors,'identifiers':identifiers,'venue':venue,'year':year,'track':track,'status':'accepted','raw_decision':'listed in official proceedings','list_url':index_url,'metadata':{'citation':dict(meta),'publication_state':'pre-proceedings' if 'pre-proceedings' in html.lower() else 'proceedings'}}
 
 
 def discover(source,cache):

paper_atlas/collectors/public_openreview.py

@@ -85,7 +85,7 @@ def normalize(row, family, year, provenance, include_workshops=False):
     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({'nips': 'neurips', 'acmmm': 'mm'}.get(family, family))
+    venue = allowed.get('neurips' if family == 'nips' else family)
     if venue is None:
         return None, 'unmapped_venue'
     title = row.get('title')

paper_atlas/db/submissions.py

@@ -1,10 +1,9 @@
 """Canonical submission events with independent source and decision evidence."""
 from datetime import datetime, timezone
 import re
-from paper_atlas.io.publication_ids import mlsys_id
 
 
-PUBLICATIONS={'pmlr','acl_anthology','colm','neurips','mlsys','cvf','ecva','dblp'}
+PUBLICATIONS={'pmlr','acl_anthology','colm','neurips','cvf','ecva','dblp'}
 
 
 def timestamp(value=None):
@@ -16,26 +15,11 @@ def timestamp(value=None):
 def identities_for_row(row):
     identities=list(row.get('submission_identifiers') or [])
     own=row.get('identifiers') or {}
-    if row['source']=='mlsys':
-        value=own.get('publication_mlsys')
-        if (not isinstance(value,str) or not re.fullmatch(str(row['year'])+r'/[a-f0-9]{32}',value)
-                or row['venue']!='MLSys'):
-            raise ValueError('Invalid authoritative MLSys publisher identity')
     if own.get('openreview'):
         identities.append({'scheme':'openreview','value':own['openreview']})
     if row['source'] in PUBLICATIONS and row.get('status')=='accepted' and own.get('doi'):
         doi=re.sub(r'^https?://(?:dx\.)?doi\.org/','',str(own['doi']),flags=re.I).strip().lower()
         if re.fullmatch(r'10\.\d{4,9}/\S+',doi):identities.append({'scheme':'publication_doi','value':doi})
-    publication_mlsys = own.get('publication_mlsys')
-    if not publication_mlsys and row['source']=='dblp' and row['venue']=='MLSys':
-        # Older normalized DBLP files retain their own publisher links even
-        # when a newly supported historical URL label was not yet parsed.
-        links = (row.get('metadata') or {}).get('bibliography', {}).get('ee', [])
-        candidates = {identity for link in links if (identity := mlsys_id(link))}
-        if len(candidates)==1:
-            publication_mlsys = candidates.pop()
-    if row['source'] in PUBLICATIONS and row.get('status')=='accepted' and publication_mlsys:
-        identities.append({'scheme':'publication_mlsys','value':publication_mlsys})
     meta=row.get('metadata') or {}
     if row['source']=='iclr_program' and row['venue']=='ICLR' and meta.get('program_identity_type')=='submission' and meta.get('program_submission_number_verified') is True:
         identities.append({'scheme':'iclr_number','value':str(meta['program_source_id'])})
@@ -46,9 +30,6 @@ def identities_for_row(row):
             valid=re.fullmatch(r'[A-Za-z0-9_-]+',value) and not re.match(r'20\d{2}-Oral-',value)
         elif scheme=='iclr_number':valid=re.fullmatch(r'[1-9]\d*',value) and row['venue']=='ICLR'
         elif scheme=='publication_doi':valid=re.fullmatch(r'10\.\d{4,9}/\S+',value)
-        elif scheme=='publication_mlsys':
-            valid=(re.fullmatch(str(row['year'])+r'/[a-f0-9]{32}',value)
-                   and row['venue']=='MLSys' and row['source'] in PUBLICATIONS and row.get('status')=='accepted')
         else:valid=False
         if not valid:raise ValueError('Invalid authoritative submission identity')
         result.add((scheme,value))

paper_atlas/io/publication_ids.py

@@ -1,13 +0,0 @@
-"""Stable identities from publisher-owned metadata self-links."""
-import re
-from urllib.parse import urlsplit
-
-
-def mlsys_id(url):
-    parsed = urlsplit(url)
-    if parsed.scheme not in ('https', 'http') or parsed.netloc != 'proceedings.mlsys.org':
-        return None
-    match = re.fullmatch(r'/paper(?:_files/paper)?/((?:19|20)\d{2})/hash/([a-f0-9]{32})-Abstract(?:-(?:Conference|mlsys(20\d{2})))?\.html', parsed.path)
-    if not match or match[3] and match[3] != match[1]:
-        return None
-    return match[1] + '/' + match[2]

paper_atlas/resources/venues.json

@@ -12,9 +12,7 @@
     "ijcai": "https://www.ijcai.org/all_proceedings/",
     "dblp": "https://dblp.org/db/conf/",
     "colm": "https://proceedings.colm.cc/",
-    "iclr_program": "https://iclr.cc/virtual/2026/papers.html",
-    "mlsys": "https://proceedings.mlsys.org/",
-    "acmmm": "https://www.acmmm.org/"
+    "iclr_program": "https://iclr.cc/virtual/2026/papers.html"
   },
   "venues": [
     {
@@ -645,37 +643,6 @@
         "main"
       ],
       "coverage": "pending"
-    },
-    {
-      "acronym": "MLSys",
-      "field": "ml_systems",
-      "venue_type": "conference",
-      "discovery_sources": [
-        "mlsys",
-        "dblp"
-      ],
-      "scope": "user_confirmed_related",
-      "years": "all_publicly_available",
-      "tracks": [
-        "main"
-      ],
-      "coverage": "pending"
-    },
-    {
-      "acronym": "MM",
-      "field": "multimodal",
-      "venue_type": "conference",
-      "discovery_sources": [
-        "dblp",
-        "acmmm",
-        "openreview"
-      ],
-      "scope": "user_confirmed_related",
-      "years": "all_publicly_available",
-      "tracks": [
-        "main"
-      ],
-      "coverage": "pending"
     }
   ],
   "additional_tracks": [

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_cli.py

@@ -18,10 +18,7 @@ class CliTests(unittest.TestCase):
 
     def test_unified_import_and_query_use_shipped_schema_and_registry(self):
         from paper_atlas.config import DEFAULT_REGISTRY
-        venues = json.loads(DEFAULT_REGISTRY.read_text())['venues']
-        names = [venue['acronym'] for venue in venues]
-        self.assertTrue({'COLM', 'MM', 'MLSys'} <= set(names))
-        self.assertEqual(len(names), len(set(names)))
+        self.assertEqual(len(json.loads(DEFAULT_REGISTRY.read_text())['venues']),41)
         with tempfile.TemporaryDirectory() as tmp:
             root=Path(tmp);db=root/'db.sqlite';source=root/'records.jsonl'
             source.write_text(json.dumps({'source':'example','external_id':'one','url':'https://example.org','title':'CLI paper','abstract':'Abstract','authors':['Jane Doe'],'venue':'Example','year':2026,'track':'main','status':'accepted'})+'\n')

tests/test_mlsys_multimedia.py

@@ -1,166 +0,0 @@
-import hashlib
-import json
-from contextlib import closing
-from pathlib import Path
-import sqlite3
-import tempfile
-import unittest
-from unittest.mock import patch
-
-from paper_atlas.collectors.proceedings import discover, parse_paper
-from paper_atlas.collectors.dblp import XML_SHA
-from paper_atlas.collectors.dblp_records import identifiers, normalize_row, collect
-from paper_atlas.collectors.dblp_scope import scope
-from paper_atlas.collectors.public_openreview import normalize
-from paper_atlas.db.submissions import identities_for_row
-from paper_atlas.ingestion.records import ingest_file
-from paper_atlas.io.publication_ids import mlsys_id
-
-
-HASH = '0123456789abcdef' * 2
-ROOT = 'https://proceedings.mlsys.org/'
-
-
-def page(year=2026, modern=True):
-    return ROOT + ('paper_files/paper/' if modern else 'paper/') + str(year) + '/hash/' + HASH + ('-Abstract-Conference.html' if modern else '-Abstract.html')
-
-
-def raw_row(family='mm', book='ACM Multimedia', year=2024):
-    return {'record_type': 'inproceedings', 'key': f'conf/{family}/Paper24', 'publtype': None,
-            'fields': {'title': ['A published paper'], 'author': ['Jane Doe', 'John Smith'],
-                       'booktitle': [book], 'year': [str(year)], 'crossref': [f'conf/{family}/{year}']}}
-
-
-def volume(family='mm', year=2024, title=None):
-    return {'record_type': 'proceedings', 'key': f'conf/{family}/{year}',
-            'fields': {'title': [title or f'ACM Multimedia {year}'], 'year': [str(year)]}}
-
-
-class MlsysMultimediaTests(unittest.TestCase):
-    def test_mlsys_discovery_retains_all_official_years_and_ignores_pdf_or_other_host(self):
-        html = ''.join(f'<a href="/paper_files/paper/{year}">MLSys {year}</a>' for year in range(2019, 2027))
-        html += '<a href="https://example.org/paper_files/paper/2018">Other</a><a href="/paper/2026/file/paper.pdf">PDF</a>'
-        with patch('paper_atlas.collectors.proceedings.fetch', return_value=html.encode()):
-            self.assertEqual([year for _, year in discover('mlsys', '/unused')], list(range(2019, 2027)))
-
-    def test_mlsys_metadata_has_stable_publisher_identity_and_explicit_track(self):
-        html = '<meta name="citation_title" content="A title"><meta name="citation_author" content="Doe, Jane"><span class="paper-track">Conference</span><h2>Abstract</h2><p>Real abstract.</p><h4>Name Change Policy</h4><p>Footer.</p>'
-        row = parse_paper(html, page(), 'mlsys', 2026, ROOT + 'paper_files/paper/2026')
-        self.assertEqual((row['venue'], row['track'], row['abstract'], row['authors']), ('MLSys', 'main', 'Real abstract.', ['Jane Doe']))
-        self.assertEqual(row['identifiers']['publication_mlsys'], '2026/' + HASH)
-        with self.assertRaisesRegex(ValueError, 'Unclassified'):
-            parse_paper(html.replace('Conference', 'New Unknown Track'), page(), 'mlsys', 2026, ROOT)
-        with self.assertRaisesRegex(ValueError, 'publication year'):
-            parse_paper(html, page(), 'mlsys', 2025, ROOT)
-        with self.assertRaisesRegex(ValueError, 'non-archival'):
-            parse_paper(html, page(2018), 'mlsys', 2018, ROOT)
-
-    def test_year_label_and_retained_bibliography_links_bridge_the_2023_format(self):
-        url = page(2023).replace('-Conference.html', '-mlsys2023.html')
-        self.assertEqual(mlsys_id(url), '2023/' + HASH)
-        self.assertIsNone(mlsys_id(url.replace('mlsys2023.html', 'mlsys2022.html')))
-        official = parse_paper('<h1>Updated title</h1><span class="paper-track">mlsys2023</span><p class="paper-abstract">Official abstract</p>', url, 'mlsys', 2023, ROOT)
-        self.assertEqual(official['track'], 'main')
-        legacy = {'source': 'dblp', 'external_id': 'conf/mlsys/paper23', 'url': 'https://dblp.org/rec/conf/mlsys/paper23',
-                  'venue': 'MLSys', 'year': 2023, 'track': 'main', 'status': 'accepted', 'title': 'Bibliographic title',
-                  'abstract': None, 'authors': [], 'identifiers': {}, 'metadata': {'bibliography': {'ee': [url]}}}
-        self.assertEqual(identities_for_row(legacy), [('publication_mlsys', '2023/' + HASH)])
-        self.assertEqual(identities_for_row({**legacy, 'status': 'unknown'}), [])
-        with tempfile.TemporaryDirectory() as tmp:
-            root = Path(tmp); db = root / 'papers.sqlite'
-            for row in [legacy, official]:
-                path = root / (row['source'] + '.jsonl'); path.write_text(json.dumps(row) + '\n')
-                path.with_suffix('.manifest.json').write_text(json.dumps({'source': row['source'], 'parser_version': 2, 'list_url': ROOT, 'count': 1}))
-                ingest_file(db, path)
-            with closing(sqlite3.connect(db)) as c:
-                self.assertEqual(c.execute('SELECT count(*) FROM papers').fetchone()[0], 1)
-                self.assertEqual(c.execute('SELECT count(*) FROM submissions').fetchone()[0], 1)
-                self.assertEqual(c.execute('SELECT count(*) FROM submission_sources').fetchone()[0], 2)
-
-    def test_legacy_mlsys_and_sysml_alias_preserve_same_publication_identity(self):
-        self.assertEqual(mlsys_id(page(2019, False)), mlsys_id(page(2019)))
-        self.assertIsNone(mlsys_id(page().replace('proceedings.mlsys.org', 'example.org')))
-        own, ambiguous = identifiers({'ee': [page(2019, False), page(2019)], 'cite': [page(2025)]})
-        self.assertEqual(own, {'publication_mlsys': '2019/' + HASH}); self.assertFalse(ambiguous)
-        _, ambiguous = identifiers({'ee': [page(2019), page(2019).replace(HASH, 'a' * 32)]})
-        self.assertEqual(len(ambiguous['publication_mlsys']), 2)
-        row = parse_paper('<h1>Legacy</h1><p class="paper-abstract">Abstract</p>', page(2019, False), 'mlsys', 2019, ROOT)
-        self.assertEqual(row['venue_aliases'], ['SysML'])
-        self.assertEqual(row['track'], 'main')
-
-    def test_mm_main_numbered_parts_and_workshops_remain_separate(self):
-        for book in ['ACM Multimedia', 'ACM Multimedia (1)', 'ACM Multimedia (2)']:
-            result, reason = scope(raw_row(book=book), volume())
-            self.assertIsNone(reason); self.assertEqual(result['venue'], 'MM'); self.assertEqual(result['track'], 'proceedings_unspecified')
-            self.assertIn('per-paper section', result['track_scope_note'])
-            self.assertIn('ACMMM', result['venue_aliases'])
-        self.assertEqual(scope(raw_row(book='ACM Multimedia Workshops'), volume())[0]['track'], 'workshop')
-        self.assertEqual(scope(raw_row(book='ACM Multimedia (Thematic Workshops)'), volume())[0]['track'], 'workshop')
-        self.assertIsNone(scope(raw_row(book='MMSports@MM'), volume())[0])
-        self.assertEqual(normalize_row(raw_row(book='ACM Multimedia Workshops'), {'conf/mm/2024': volume()}, {'MM'})[1], 'separate_workshop')
-
-    def test_sysml_alias_requires_machine_learning_proceedings(self):
-        row = raw_row('mlsys', 'SysML', 2019)
-        result, reason = scope(row, volume('mlsys', 2019, 'Conference on Machine Learning and Systems, SysML 2019'))
-        self.assertIsNone(reason); self.assertEqual(result['venue'], 'MLSys'); self.assertEqual(result['venue_aliases'], ['SysML'])
-        self.assertIsNone(scope(row, volume('mlsys', 2019, 'Systems Modeling Language 2019'))[0])
-
-    def test_bibliography_and_official_abstract_import_as_one_paper_and_submission(self):
-        raw = raw_row('mlsys', 'SysML', 2019); raw['fields']['ee'] = [page(2019, False)]
-        bib, reason = normalize_row(raw, {'conf/mlsys/2019': volume('mlsys', 2019, 'Machine Learning and Systems 2019')}, {'MLSys'})
-        self.assertIsNone(reason)
-        official = parse_paper('<meta name="citation_title" content="An updated title"><p class="paper-abstract">The official abstract.</p>', page(2019), 'mlsys', 2019, ROOT)
-        # Different title and absent official authors deliberately prevent a title-based bridge.
-        with tempfile.TemporaryDirectory() as tmp:
-            root = Path(tmp); db = root / 'papers.sqlite'
-            for row in [bib, official]:
-                path = root / (row['source'] + '.jsonl'); path.write_text(json.dumps(row) + '\n')
-                path.with_suffix('.manifest.json').write_text(json.dumps({'source': row['source'], 'parser_version': 2,
-                    'list_url': ROOT, 'count': 1, 'venue': 'MLSys', 'year': 2019, 'track': 'main'}))
-                ingest_file(db, path)
-            with closing(sqlite3.connect(db)) as c:
-                self.assertEqual(c.execute('SELECT count(*) FROM papers').fetchone()[0], 1)
-                self.assertEqual(c.execute('SELECT count(*) FROM submissions').fetchone()[0], 1)
-                self.assertEqual(c.execute('SELECT count(*) FROM submission_sources').fetchone()[0], 2)
-                self.assertEqual(c.execute('SELECT abstract FROM papers').fetchone()[0], 'The official abstract.')
-                self.assertEqual(c.execute('PRAGMA foreign_key_check').fetchall(), [])
-
-    def test_publisher_identity_cannot_bridge_rejection_other_year_or_venue(self):
-        row = {'source': 'mlsys', 'venue': 'MLSys', 'year': 2026, 'status': 'accepted',
-               'identifiers': {'publication_mlsys': '2026/' + HASH}}
-        self.assertEqual(identities_for_row(row), [('publication_mlsys', '2026/' + HASH)])
-        self.assertEqual(identities_for_row({**row, 'status': 'rejected'}), [])
-        with self.assertRaisesRegex(ValueError, 'Invalid authoritative'):
-            identities_for_row({**row, 'identifiers': {}})
-        for changed in [dict(year=2025), dict(venue='MM'), dict(source='curated')]:
-            bad = {**row, **changed, 'submission_identifiers': [{'scheme': 'publication_mlsys', 'value': '2026/' + HASH}]}
-            with self.assertRaisesRegex(ValueError, 'Invalid authoritative'):
-                identities_for_row(bad)
-
-    def test_explicit_mm_negative_assertion_maps_to_mm_without_inventing_labels(self):
-        provenance = {'source_file': 'https://example.org/pinned', 'source_line': 1}
-        row = {'id': 'realForum1', 'title': 'A title', 'abstract': 'An abstract', 'track': 'main', 'status': 'Reject'}
-        normalized, reason = normalize(row, 'acmmm', 2025, provenance)
-        self.assertIsNone(reason); self.assertEqual(normalized['venue'], 'MM')
-        self.assertEqual(normalized['status'], 'rejected')
-        self.assertIsNone(normalize({**row, 'status': 'Submitted'}, 'acmmm', 2025, provenance)[0])
-
-    def test_old_candidate_subset_cannot_silently_omit_new_requested_venues(self):
-        with tempfile.TemporaryDirectory() as tmp:
-            root = Path(tmp); catalogue = root / 'dblp-catalogue.sqlite'
-            with closing(sqlite3.connect(catalogue)) as c, c:
-                c.execute('CREATE TABLE proceedings(key TEXT PRIMARY KEY,metadata_json TEXT)')
-                c.execute('INSERT INTO proceedings VALUES(?,?)', ('conf/mm/2024', json.dumps(volume())))
-            candidates = root / 'candidates.jsonl'; candidates.write_text(json.dumps(raw_row()) + '\n')
-            original = candidates.read_bytes()
-            catalogue.with_name('dblp-ai-candidates.manifest.json').write_text(json.dumps({
-                'snapshot_sha256': XML_SHA, 'candidate_sha256': hashlib.sha256(original).hexdigest(),
-                'candidate_family_counts': {'aaai': 1}, 'record_types': {'proceedings': 1}}))
-            with self.assertRaisesRegex(ValueError, 'original --archive'):
-                collect(catalogue, root / 'out', ['MM'], candidates=candidates)
-            self.assertEqual(candidates.read_bytes(), original)
-            self.assertFalse((root / 'out').exists())
-
-
-if __name__ == '__main__':
-    unittest.main()