Add verified PostgreSQL migration and shared research queries

Merge pull request '补齐 ACM MM、MLSys 范围并接入官方摘要与发表身份' (#12 ) from codex/multimedia-mlsys into main
Reviewed-on: #12
2026-11-07 02:09:19 +08:00 · 2026-11-06 16:15:16 +00:00 · 2026-11-06 16:15:09 +00:00 · 2026-11-06 16:15:04 +00:00 · 2026-11-06 22:58:50 +08:00 · 2026-11-06 22:49:41 +08:00
30 changed files with 1359 additions and 30 deletions
@@ -30,11 +30,16 @@ jobs:
              if len(candidates)!=1:
                  raise SystemExit('Unexpected repository archive layout')
              src = candidates[0].parent
-              subprocess.run(['python3','-B','-m','unittest','discover','-s','tests','-t','.','-v'],cwd=src,check=True)
              # Build and installation checks run only on the approved remote runner.
              venv = root / 'venv'
              subprocess.run(['python3','-B','-m','venv','--system-site-packages',str(venv)],check=True)
              python = str(venv / 'bin/python')
+              subprocess.run([python,'-B','-m','pip','install','--only-binary=:all:','--no-compile','psycopg[binary]==3.3.8'],check=True)
+              testenv = os.environ.copy()
+              service_file = pathlib.Path('/etc/paper-atlas/pg_service.conf')
+              if service_file.is_file():
+                  testenv.update(PGSERVICEFILE=str(service_file),PAPER_ATLAS_TEST_PG_SERVICE='foundry-research')
+              subprocess.run([python,'-B','-m','unittest','discover','-s','tests','-t','.','-v'],cwd=src,env=testenv,check=True)
              wheels = root / 'wheels'
              subprocess.run([python,'-B','-m','pip','wheel','--no-deps','--wheel-dir',str(wheels),str(src)],check=True)
              wheel, = wheels.glob('paper_atlas-*.whl')
@@ -43,7 +48,8 @@ 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 len(json.loads(DEFAULT_REGISTRY.read_text())['venues'])==41; s=Store('installed.sqlite'); s.conn.close()"
+              subprocess.run([python,'-I','-B','-m','paper_atlas','maintain','postgres','--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()"
              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
@@ -2,9 +2,9 @@

 **长期维护、按需运行的论文元数据与公开投稿历史工具。**

-Paper Atlas 将 arXiv、AI 会议论文集及公开投稿决定整理到可追溯的 SQLite 数据库中，用于查询和后续研究分析。它按批次运行，不需要常驻 Web 服务；采集标题、摘要、标识、标签及决定证据，不下载论文全文。
+Paper Atlas 将 arXiv、AI 会议论文集及公开投稿决定整理到可追溯的数据库中，用于查询和后续研究分析。生产数据使用 PostgreSQL，支持从既有 SQLite 索引完整迁移；SQLite 保留离线核验和小型样本用途。它按批次运行，不需要常驻 Web 服务；采集标题、摘要、标识、标签及决定证据，不下载论文全文。

-项目仍在建设中。候选范围包含 41 个会议，尚未完成全部历史数据和最终去重复核。实际进度以数据库、来源清单和覆盖报告为准，而不是仓库文件数量。
+项目仍在建设中。候选范围包含 43 个会议（含 ACM MM、MLSys），尚未完成全部历史数据和最终去重复核。实际进度以数据库、来源清单和覆盖报告为准，而不是仓库文件数量。

 ## 项目结构

@@ -14,6 +14,7 @@ paper_atlas/
  collectors/            各来源采集与目录盘点
  ingestion/             导入、身份回填与单写入协调
  db/                    数据模型、身份解析与 SQL schema
+  postgres/              PostgreSQL 迁移、原生全文索引和只读连接
  audits/                来源核验和覆盖报告
  io/                    网络、快照下载、HTML 与 Git 元数据读取
  resources/             候选会议目录
@@ -30,7 +31,7 @@ pyproject.toml          包元数据、依赖和命令注册

 ```sh
 python3 -m venv .venv
-.venv/bin/python -m pip install .
+.venv/bin/python -m pip install '.[postgres]'
 source .venv/bin/activate
 paper-atlas --help
 ```
@@ -41,13 +42,13 @@ paper-atlas --help

 ```sh
 # 查看当前入库量；不同平台记录数不等于独立论文数
-paper-atlas query --db /data/paper-index/papers.sqlite --stats
+PGSERVICEFILE=/etc/paper-atlas/pg_service.conf paper-atlas query --pg-service foundry-research --stats

 # 查询标题与摘要
-paper-atlas query --db /data/paper-index/papers.sqlite --query '"language model"'
+PGSERVICEFILE=/etc/paper-atlas/pg_service.conf paper-atlas query --pg-service foundry-research --query '"language model"'

 # 查看一篇论文的来源、标签和所有已知投稿历史
-paper-atlas query --db /data/paper-index/papers.sqlite --paper 123
+PGSERVICEFILE=/etc/paper-atlas/pg_service.conf paper-atlas query --pg-service foundry-research --paper 123

 # 列出全部目标会议，包括尚未入库的项目
 paper-atlas coverage --db /data/paper-index/papers.sqlite \
@@ -60,6 +61,8 @@ paper-atlas collect colm --cache /data/paper-index/raw/http-cache \

 `paper-atlas collect --help` 查看来源；`paper-atlas import --help` 查看导入方式。完整批次步骤见[使用指南](docs/usage.md)。

+共享实例名为 `foundry-postgres`，数据位于 `foundry_data` 库的 `research` schema；其他项目可按权限复用同一组表。库名通过 libpq service 配置、schema 通过 search_path 或 `--pg-schema` 指定，不写死在查询中。PostgreSQL 凭据放在服务器权限为 `0600` 的 libpq service 文件中；命令行只使用服务名称。`paper-atlas maintain postgres --help` 提供迁移入口：先完整核验 21 张业务表，再发布目标 schema。SQLite FTS5 内部表由 PostgreSQL GIN 索引替代。切换后 SQLite 为备份，不能继续用旧导入器写入它；后续生产导入须适配 PostgreSQL，不能把 SQLite 更新当作生产更新。
+
 ## 数据如何组织

 - **论文和投稿分开。** 一篇论文可以多次投稿；某次被拒、后来在其他会议录用，两段历史都保留。
@@ -16,6 +16,8 @@
 - [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/)
@@ -29,6 +31,7 @@
 | 机器学习、理论 | 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 |
@@ -38,6 +41,23 @@

 这些是领域覆盖候选，不构成统一排名，也不能据此声称已全部入库。逐年目录以真实举办届次和正式出版清单为准：双年会议不补造不存在的年份；主会、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
@@ -1,5 +1,31 @@
 # 运行维护与数据核验

+## PostgreSQL 迁移
+
+共享实例的容器名为 `foundry-postgres`，底层镜像为 `postgres:17-alpine`；Compose 服务名仍为 `postgres`。数据库和 schema 是可配置的逻辑边界，表按研究数据域组织，不按调用项目复制。其他项目可获授 `research` 的 USAGE 及必要表权限，复用相同论文 ID 和表。未来换库只更新 libpq service；换 schema 可更新角色 search_path 或查询的 `--pg-schema`。
+
+复用已确认的现有实例，在共享 `foundry_data` 数据库按数据域使用 `research` schema 和专用角色，不修改其他业务库。安装、测试和构建在已批准的 `ces-control:host` 执行环境运行。libpq service 文件由服务器提供，权限必须为 `0600`，不得放进仓库或带密码的命令行参数。
+
+冻结 SQLite 写入，确认维护进程实际结束、WAL 已清空；采集到独立文件不修改索引。迁移持有源维护锁和目标 advisory lock，还须确保没有绕过协调器的旧版手工写入。
+
+```sh
+export PGSERVICEFILE=/etc/paper-atlas/pg_service.conf
+paper-atlas maintain postgres \
+  --db /data/paper-index/papers.sqlite --pg-service foundry-research \
+  --storage-path /var/lib/docker/volumes/endoleap-dev_postgres/_data \
+  --min-free-gib 8 --out /data/paper-index/collected/postgres-migration.json
+```
+
+全部 21 张业务表保留原 ID、原始 JSON 文本、NULL、来源、投稿、决定、标签出处和合并记录。`sqlite_sequence` 用于重设 PostgreSQL identity 序列，并考虑已合并的历史 ID，避免复用；SQLite FTS5 shadow 表根据完整标题/摘要重建为原生 GIN 索引，不逐字复制引擎文件。
+
+先 COPY 到独立暂存 schema，每张表提交和记录哈希。中断后以同一命令、同一报告路径恢复，必须仍是**完全相同的未变更快照**。不同输入、已发布目标、未清空 WAL、新的未知业务表都会拒绝覆盖；恢复只会清空本次运行尚未完成且尚未发布的暂存表。保留错误现场，不盲目从头复制。
+
+发布前逐表按主键完整回读，比较数量及逐行 SHA-256；建立并验证所有主键、唯一约束、CHECK 和外键，重设序列、建立 GIN 并 ANALYZE。全部通过后原子重命名为 `research` schema，报告才为 `complete`。`copying`、`prepared`、部分表通过或 CI 通过均不代表生产迁移完成，也不替代原任务最后的实体去重和会议覆盖审计。
+
+发布后核对 PostgreSQL 的统计、全文检索、论文详情，以及 reject 后另会 accept 的历史。确认后保留 SQLite 只读备份，并在旁保存 `<SQLite路径>.postgresql-cutover.json`，记录报告路径、源哈希、目标 service/schema；新版本旧写入入口见此标记会拒绝写入。标记不含密码。后续导入适配属于下一阶段，不可继续更新 SQLite 后声称 PostgreSQL 已同步。
+
+远程集成测试通过 `PAPER_ATLAS_TEST_PG_SERVICE` 显式指定，使用随机前缀 schema 并仅清理自己的 fixture；没有该配置时会跳过，应另外报告“未运行”，不能用 SQLite 测试代替 PostgreSQL 验证。
+
 代码以固定 commit 部署，已经启动的任务继续使用其原版本。目录重构不要求迁移数据库，也不更改已有 JSONL 和 manifest 格式。

 ## arXiv 原始编号与类别证据
@@ -20,6 +46,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. 每次采集先核对来源、许可、可访问性、空闲空间和现有运行批次。保持低并发，遵守来源限流、重试与人机验证要求。
@@ -2,6 +2,22 @@

 先在已确认的远程环境安装项目，使用 `paper-atlas`。从源码运行时，将命令前缀替换为 `python3 -B -m paper_atlas`，其余参数相同。数据目录位于仓库外。

+## PostgreSQL 查询
+
+在远程环境安装 `.[postgres]`，通过权限为 `0600` 的 libpq service 文件连接，不把密码放入 URL、命令行或仓库。
+
+```sh
+export PGSERVICEFILE=/etc/paper-atlas/pg_service.conf
+paper-atlas query --pg-service foundry-research --stats
+paper-atlas query --pg-service foundry-research --query '"language model"' --tag arxiv:category=cs.CL
+paper-atlas query --pg-service foundry-research --status rejected --outcome accepted --limit 20
+paper-atlas query --pg-service foundry-research --paper 123
+```
+
+使用只读一致性事务，支持论文 ID、会议别名、年份、轨道、标签、最新决定和跨会议录用筛选。全文检索使用 PostgreSQL `websearch_to_tsquery('english', ...)`，支持引号短语、`OR` 和减号排除；与 SQLite FTS5 的语法及分词结果不完全相同。JSON 来源列保留原始文本，SQL 分析时可用 `metadata_json::jsonb`、`authors_json::jsonb`。
+
+迁移与切换见[运行维护](operations.md#postgresql-迁移)。下述 SQLite 导入命令仅用于未切换的索引或独立样本；切换后的生产源拒绝旧写入。尚未适配的覆盖审计可只读检查冻结备份，不能当作 PostgreSQL 当前状态。
+
 ## 1. 下载固定快照

 `paper-atlas download --help` 说明大小、校验和与续传参数。各来源的固定版本与具体命令见[数据源说明](data-sources.md)。下载成功只表示原始文件已核验。
@@ -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()
@@ -18,6 +18,7 @@ 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 历史发表记录'),
@@ -28,9 +29,11 @@ COMMANDS = {
    ('maintain','backfill'): ('paper_atlas.ingestion.backfill', {}, '回填既有投稿的来源身份'),
    ('maintain','arxiv-tags'): ('paper_atlas.ingestion.arxiv_tags', {}, '恢复逐来源记录的 arXiv 类别证据'),
    ('maintain','pmlr-tracks'): ('paper_atlas.ingestion.pmlr_tracks', {}, '核对并迁移既有 PMLR 投稿轨道'),
+    ('maintain','postgres'): ('paper_atlas.postgres.migration', {}, '迁移 SQLite 完整业务数据并核验 PostgreSQL'),
    ('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', {}, '核对公开负面决定与已有投稿历史'),
 }

@@ -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'.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 mm mlsys'.split())


 def records(archive,dtd_bytes):
@@ -17,6 +17,7 @@ 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)'
@@ -36,8 +37,10 @@ def material_reason(title,authors,key,year,track):


 def identifiers(fields):
-    dois=set();arxiv=set();forums=set()
+    dois=set();arxiv=set();forums=set();mlsys=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('/')
@@ -57,9 +60,10 @@ 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={}
-    for scheme,values in (('doi',publications),('arxiv',arxiv),('openreview',forums)):
+    groups=(('doi',publications),('arxiv',arxiv),('openreview',forums),('publication_mlsys',mlsys))
+    for scheme,values in groups:
        if len(values)==1:own[scheme]=next(iter(values))
-    ambiguous={scheme:sorted(values) for scheme,values in (('doi',publications),('arxiv',arxiv),('openreview',forums)) if len(values)>1}
+    ambiguous={scheme:sorted(values) for scheme,values in groups if len(values)>1}
    return own,ambiguous


@@ -84,6 +88,8 @@ 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,
@@ -110,6 +116,12 @@ 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)
@@ -33,6 +33,8 @@ 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'])],
 }


@@ -45,6 +47,7 @@ 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)
@@ -67,6 +70,7 @@ 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'
@@ -92,5 +96,11 @@ 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'
-    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
+    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
@@ -1,4 +1,4 @@
-"""Collect COLM/NeurIPS official abstract pages without downloading PDFs."""
+"""Collect COLM, NeurIPS and MLSys official metadata without downloading PDFs."""
 import argparse
 import json
 import re
@@ -8,8 +8,11 @@ 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/'}
+ROOTS={'colm':'https://proceedings.colm.cc/','neurips':'https://proceedings.neurips.cc/',
+       'mlsys':'https://proceedings.mlsys.org/'}
+VENUES={'colm':'COLM','neurips':'NeurIPS','mlsys':'MLSys'}


 from paper_atlas.io.html import Document,Node,clean
@@ -72,6 +75,22 @@ 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=[]
@@ -90,8 +109,11 @@ 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='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'}}
+    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


 def discover(source,cache):
@@ -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('neurips' if family == 'nips' else family)
+    venue = allowed.get({'nips': 'neurips', 'acmmm': 'mm'}.get(family, family))
    if venue is None:
        return None, 'unmapped_venue'
    title = row.get('title')
@@ -0,0 +1,8 @@
+"""Prevent legacy writers from diverging after an explicit PostgreSQL cutover."""
+from pathlib import Path
+
+
+def require_sqlite_writable(path):
+    marker = Path(str(Path(path).resolve()) + '.postgresql-cutover.json')
+    if marker.exists():
+        raise RuntimeError('This SQLite source was cut over to PostgreSQL; use the configured PostgreSQL database. Marker: ' + str(marker))
@@ -6,6 +6,7 @@ import sqlite3
 import unicodedata
 from pathlib import Path
 from paper_atlas.db.submissions import attach_submission,migrate_sources,write_decision
+from paper_atlas.db.backend import require_sqlite_writable


 def normalized(value):
@@ -29,6 +30,7 @@ def author_keys(authors):

 class Store:
    def __init__(self,path):
+        require_sqlite_writable(path)
        self.conn = sqlite3.connect(path,timeout=60)
        self.conn.executescript(files('paper_atlas.db').joinpath('schema.sql').read_text())
        self.conn.execute('PRAGMA synchronous=NORMAL')
@@ -1,9 +1,10 @@
 """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','cvf','ecva','dblp'}
+PUBLICATIONS={'pmlr','acl_anthology','colm','neurips','mlsys','cvf','ecva','dblp'}


 def timestamp(value=None):
@@ -15,11 +16,26 @@ 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'])})
@@ -30,6 +46,9 @@ 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))
@@ -18,6 +18,8 @@ def save(path,state):

 def repair(db,out,resume=False,batch_size=1000):
    db=Path(db).resolve();out=Path(out).resolve()
+    from paper_atlas.db.backend import require_sqlite_writable
+    require_sqlite_writable(db)
    if not db.is_file():raise ValueError('Existing database required')
    if batch_size<1 or batch_size>10000:raise ValueError('Batch size must be between 1 and 10000')
    out.parent.mkdir(parents=True,exist_ok=True)
@@ -15,6 +15,7 @@ from pathlib import Path
 from paper_atlas.ingestion.records import ingest_file
 from paper_atlas.db.store import Store
 from paper_atlas.ingestion.backfill import backfill,queue_files
+from paper_atlas.db.backend import require_sqlite_writable


 UNIT='paper-atlas-arxiv-import.service'
@@ -25,6 +26,7 @@ def property_value(name):


 def merge_locked(db,archive,files,backfill_queue=None):
+    require_sqlite_writable(db)
    # Validate inputs before stopping the existing writer.
    for filename in files:
        path=Path(filename)
@@ -123,6 +123,9 @@ def plan(c, scopes):

 def reclassify(db, inventories, out, apply=False):
    db, out = Path(db).resolve(), Path(out).resolve()
+    if apply:
+        from paper_atlas.db.backend import require_sqlite_writable
+        require_sqlite_writable(db)
    if not db.is_file() or out == db:
        raise ValueError('Existing database and distinct report path required')
    scopes = load_inventories(inventories)
@@ -0,0 +1,13 @@
+"""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]
@@ -0,0 +1 @@
+"""PostgreSQL migration and read access for the paper index."""
@@ -0,0 +1,83 @@
+"""Libpq service connections; credentials never belong in CLI arguments."""
+from collections.abc import Mapping
+import re
+
+SEARCH_VECTOR = "to_tsvector('english'::regconfig,coalesce(title,'') || ' ' || coalesce(abstract,''))"
+
+
+def driver():
+    try:
+        import psycopg
+    except ImportError as error:
+        raise RuntimeError('Install paper-atlas[postgres] on the approved remote runner') from error
+    return psycopg
+
+
+def connect(service, *, readonly=False):
+    if not re.fullmatch(r'[A-Za-z0-9_-]+', service):
+        raise ValueError('Use a libpq service name, never a password or connection URL')
+    connection = driver().connect(service=service, connect_timeout=15, autocommit=True,
+                                  application_name='paper-atlas')
+    connection.execute("SET timezone='UTC'")
+    connection.execute("SET lock_timeout='5s'")
+    connection.execute("SET work_mem='2MB'")
+    connection.execute("SET maintenance_work_mem='16MB'")
+    connection.execute('SET max_parallel_workers_per_gather=0')
+    connection.execute('SET max_parallel_maintenance_workers=0')
+    if readonly:
+        connection.execute('BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY')
+    return connection
+
+
+class Row(Mapping):
+    """The small mapping/positional interface used by existing query output."""
+    def __init__(self, columns, values):
+        self.columns, self.values = columns, values
+
+    def __getitem__(self, key):
+        return self.values[key if isinstance(key, int) else self.columns.index(key)]
+
+    def __iter__(self):
+        return iter(self.columns)
+
+    def __len__(self):
+        return len(self.columns)
+
+
+def row_factory(cursor):
+    columns = [column.name for column in cursor.description] if cursor.description else []
+    return lambda values: Row(columns, values)
+
+
+class ReadConnection:
+    postgres = True
+
+    def __init__(self, service, schema=None):
+        if schema is not None and not re.fullmatch(r'[a-z][a-z0-9_]{0,35}', schema):
+            raise ValueError('Invalid schema name')
+        self.connection = connect(service, readonly=True)
+        from paper_atlas.postgres.schema import quote
+        if schema is not None:
+            self.connection.execute('SET LOCAL search_path TO ' + quote(schema) + ',pg_catalog')
+        self.connection.row_factory = row_factory
+        if not self.has_table('papers'):
+            self.close()
+            raise RuntimeError('PostgreSQL publication is not ready in the configured search_path')
+
+    def execute(self, query, params=()):
+        # All caller SQL is static; user values are always bound parameters.
+        if 'sqlite_master' in query:
+            match = re.search(r"name='([a-z_]+)'", query)
+            if not match or match[1] not in ('decision_sources', 'submission_sources', 'venue_aliases'):
+                raise ValueError('Unsupported legacy table lookup')
+            return self.connection.execute('''SELECT 1 FROM information_schema.tables
+                WHERE table_schema=current_schema() AND table_name=%s''',
+                (match[1],))
+        return self.connection.execute(query.replace('?', '%s'), params)
+
+    def has_table(self, name):
+        return bool(self.connection.execute('''SELECT 1 FROM information_schema.tables
+            WHERE table_schema=current_schema() AND table_name=%s''', (name,)).fetchone())
+
+    def close(self):
+        self.connection.close()
@@ -0,0 +1,232 @@
+"""Resumable, unpublished COPY followed by complete ordered content verification."""
+import argparse
+from contextlib import closing
+import fcntl
+import hashlib
+import json
+from pathlib import Path
+import re
+import shutil
+import sqlite3
+import uuid
+import zlib
+
+from paper_atlas.db.submissions import timestamp
+from paper_atlas.postgres.client import connect, SEARCH_VECTOR
+from paper_atlas.postgres.schema import TABLES, inspect, fingerprint, quote, select_sql, table_sql
+
+
+def digest_file(path):
+    digest = hashlib.sha256()
+    with path.open('rb') as stream:
+        while block := stream.read(8 * 1024 ** 2):
+            digest.update(block)
+    return digest.hexdigest()
+
+
+def update_hash(digest, row):
+    # Raw JSON columns remain text: whitespace, escapes and source evidence survive.
+    data = json.dumps(tuple(row), ensure_ascii=False, separators=(',', ':')).encode()
+    digest.update(len(data).to_bytes(8, 'big')); digest.update(data)
+
+
+def save(path, result):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    part = path.with_suffix(path.suffix + '.part')
+    part.write_text(json.dumps(result, ensure_ascii=False, indent=2) + '\n')
+    part.replace(path)
+
+
+def free_space(path, minimum):
+    if path and shutil.disk_usage(path).free < minimum * 1024 ** 3:
+        raise RuntimeError('Migration storage headroom fell below the configured minimum')
+
+
+def qualify(schema, table):
+    return quote(schema) + '.' + quote(table)
+
+
+def add_constraints(pg, stage, plan):
+    def constraint(table, name, clause):
+        exists = pg.execute('''SELECT 1 FROM pg_constraint co JOIN pg_class cl ON cl.oid=co.conrelid
+            JOIN pg_namespace ns ON ns.oid=cl.relnamespace WHERE ns.nspname=%s AND cl.relname=%s AND co.conname=%s''',
+                            (stage, table, name)).fetchone()
+        if not exists:
+            pg.execute('ALTER TABLE ' + qualify(stage, table) + ' ADD CONSTRAINT ' + quote(name) + ' ' + clause)
+
+    for table, spec in plan.items():
+        constraint(table, table + '_pkey', 'PRIMARY KEY (' + ','.join(map(quote, spec['pk'])) + ')')
+        for number, index in enumerate(spec['indexes']):
+            if index['origin'] == 'u':
+                constraint(table, table + '_unique_' + str(number), 'UNIQUE (' + ','.join(quote(n) for n, desc in index['terms']) + ')')
+            else:
+                terms = ','.join(quote(n) + (' DESC' if desc else '') for n, desc in index['terms'])
+                pg.execute('CREATE ' + ('UNIQUE ' if index['unique'] else '') + 'INDEX IF NOT EXISTS ' +
+                           quote(index['name']) + ' ON ' + qualify(stage, table) + '(' + terms + ')')
+    for table, spec in plan.items():
+        for number, fk in enumerate(spec['foreign_keys']):
+            clause = ('FOREIGN KEY (' + ','.join(map(quote, fk['columns'])) + ') REFERENCES ' +
+                      qualify(stage, fk['table']) + '(' + ','.join(map(quote, fk['references'])) + ') NOT VALID')
+            constraint(table, table + '_fk_' + str(number), clause)
+        for number, check in enumerate(spec['checks']):
+            constraint(table, table + '_check_' + str(number), 'CHECK (' + check + ') NOT VALID')
+    constraints = pg.execute('''SELECT cl.relname,co.conname FROM pg_constraint co
+        JOIN pg_class cl ON cl.oid=co.conrelid JOIN pg_namespace ns ON ns.oid=cl.relnamespace
+        WHERE ns.nspname=%s AND NOT co.convalidated ORDER BY cl.relname,co.conname''', (stage,)).fetchall()
+    for table, name in constraints:
+        pg.execute('ALTER TABLE ' + qualify(stage, table) + ' VALIDATE CONSTRAINT ' + quote(name))
+
+
+def sequences(pg, source, stage, plan):
+    old = dict(source.execute('SELECT name,seq FROM sqlite_sequence'))
+    history = {'papers': ('paper_merges', 'former_paper_id'),
+               'submissions': ('submission_merges', 'former_submission_id'),
+               'decisions': ('decision_merges', 'former_decision_id')}
+    result = {}
+    for table, spec in plan.items():
+        if spec['pk'] != ['id']:
+            continue
+        value = max(source.execute('SELECT coalesce(max(id),0) FROM ' + quote(table)).fetchone()[0], old.get(table, 0))
+        if table in history:
+            audit, column = history[table]
+            value = max(value, source.execute('SELECT coalesce(max(' + quote(column) + '),0) FROM ' + quote(audit)).fetchone()[0])
+        sequence = pg.execute('SELECT pg_get_serial_sequence(%s,%s)', (qualify(stage, table), 'id')).fetchone()[0]
+        pg.execute('SELECT setval(%s,%s,%s)', (sequence, max(value, 1), value > 0))
+        result[table] = value
+    return result
+
+
+def migrate(db, service, out, *, schema='research', storage_path=None, min_free_gib=8):
+    if not re.fullmatch(r'[a-z][a-z0-9_]{0,35}', schema):
+        raise ValueError('Use a short lowercase target schema name')
+    db, out = Path(db).resolve(), Path(out).resolve()
+    if not db.is_file() or out == db:
+        raise ValueError('Existing source and a distinct report are required')
+    if out.exists() and json.loads(out.read_text()).get('status') == 'complete':
+        raise ValueError('Completed report must be preserved')
+    free_space(storage_path, min_free_gib)
+    control = schema + '_migration'
+    report = {'status': 'preflight', 'source': str(db), 'service': service, 'schema': schema,
+              'started_at': timestamp(), 'tables': {}, 'scope': 'All 21 business tables; FTS5 internals rebuilt as native PostgreSQL GIN.'}
+    with Path(str(db) + '.maintenance.lock').open('a') as lock:
+        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        if Path(str(db) + '-wal').exists() and Path(str(db) + '-wal').stat().st_size:
+            raise ValueError('Freeze writers and checkpoint SQLite WAL before migrating')
+        with closing(sqlite3.connect(db.as_uri() + '?mode=ro', uri=True)) as source, closing(connect(service)) as pg:
+            source.execute('PRAGMA query_only=ON'); source.execute('PRAGMA cache_size=-16384'); source.execute('BEGIN')
+            plan = inspect(source)
+            stat = db.stat()
+            identity = {'file_sha256': digest_file(db), 'bytes': stat.st_size, 'mtime_ns': stat.st_mtime_ns,
+                        'schema_sha256': fingerprint(plan)}
+            report['source_identity'] = identity
+            advisory = zlib.crc32(schema.encode()) & 0x7fffffff
+            if not pg.execute('SELECT pg_try_advisory_lock(%s,%s)', (615173, advisory)).fetchone()[0]:
+                raise RuntimeError('Another migration holds the target lock')
+            if pg.execute('SELECT 1 FROM pg_namespace WHERE nspname=%s', (schema,)).fetchone():
+                raise ValueError('Target schema already exists; published data will never be overwritten')
+            pg.execute('CREATE SCHEMA IF NOT EXISTS ' + quote(control))
+            pg.execute('CREATE TABLE IF NOT EXISTS ' + qualify(control, 'run') +
+                       '(id INTEGER PRIMARY KEY CHECK(id=1), source_json TEXT NOT NULL, stage TEXT NOT NULL, status TEXT NOT NULL)')
+            pg.execute('CREATE TABLE IF NOT EXISTS ' + qualify(control, 'tables') +
+                       '(name TEXT PRIMARY KEY, rows BIGINT NOT NULL, sha256 TEXT NOT NULL, verified BOOLEAN NOT NULL DEFAULT false)')
+            run = pg.execute('SELECT source_json,stage,status FROM ' + qualify(control, 'run') + ' WHERE id=1').fetchone()
+            if run:
+                if json.loads(run[0]) != identity or run[2] == 'complete':
+                    raise ValueError('Resume requires the exact same unpublished SQLite snapshot')
+                stage = run[1]
+            else:
+                stage = schema + '_stage_' + uuid.uuid4().hex[:12]
+                with pg.transaction():
+                    pg.execute('CREATE SCHEMA ' + quote(stage))
+                    for table, spec in plan.items():
+                        pg.execute(table_sql(stage, table, spec))
+                    pg.execute('INSERT INTO ' + qualify(control, 'run') + ' VALUES(1,%s,%s,%s)',
+                               (json.dumps(identity, sort_keys=True), stage, 'copying'))
+            report.update(status='copying', stage=stage); save(out, report)
+            try:
+                for table, spec in plan.items():
+                    free_space(storage_path, min_free_gib)
+                    previous = pg.execute('SELECT rows,sha256 FROM ' + qualify(control, 'tables') + ' WHERE name=%s', (table,)).fetchone()
+                    if previous:
+                        rows, digest = previous
+                    else:
+                        digest = hashlib.sha256(); rows = 0
+                        with pg.transaction():
+                            # Only this run's unpublished, incomplete table can be reset.
+                            pg.execute('TRUNCATE ' + qualify(stage, table))
+                            columns = ','.join(quote(c['name']) for c in spec['columns'])
+                            with pg.cursor().copy('COPY ' + qualify(stage, table) + '(' + columns + ') FROM STDIN') as copy:
+                                for row in source.execute(select_sql(None, table, spec)):
+                                    copy.write_row(row); update_hash(digest, row); rows += 1
+                                    if rows % 10000 == 0:
+                                        free_space(storage_path, min_free_gib)
+                                    if rows % 100000 == 0:
+                                        print(json.dumps({'stage': 'copying', 'table': table, 'rows': rows}), flush=True)
+                            digest = digest.hexdigest()
+                            pg.execute('INSERT INTO ' + qualify(control, 'tables') + '(name,rows,sha256) VALUES(%s,%s,%s)', (table, rows, digest))
+                    report['tables'][table] = {'source_rows': rows, 'source_sha256': digest}
+                    save(out, report)
+                    print(json.dumps({'stage': 'copied', 'table': table, 'rows': rows}), flush=True)
+                report['status'] = 'validating_constraints'; save(out, report)
+                add_constraints(pg, stage, plan)
+                report['sequences'] = sequences(pg, source, stage, plan)
+                report['status'] = 'verifying_contents'; save(out, report)
+                for table, spec in plan.items():
+                    digest = hashlib.sha256(); rows = 0
+                    with pg.transaction(), pg.cursor(name='verify_' + table) as cursor:
+                        cursor.itersize = 512
+                        cursor.execute(select_sql(stage, table, spec))
+                        for row in cursor:
+                            update_hash(digest, row); rows += 1
+                    entry = report['tables'][table]
+                    entry.update(target_rows=rows, target_sha256=digest.hexdigest())
+                    if rows != entry['source_rows'] or digest.hexdigest() != entry['source_sha256']:
+                        raise ValueError('Complete content verification failed for ' + table)
+                    pg.execute('UPDATE ' + qualify(control, 'tables') + ' SET verified=true WHERE name=%s', (table,))
+                    save(out, report)
+                    print(json.dumps({'stage': 'verified', 'table': table, 'rows': rows}), flush=True)
+                report['status'] = 'building_search_index'; save(out, report)
+                free_space(storage_path, min_free_gib)
+                pg.execute('CREATE INDEX IF NOT EXISTS papers_search_gin ON ' + qualify(stage, 'papers') + ' USING gin (' + SEARCH_VECTOR + ')')
+                search = pg.execute('''SELECT am.amname,i.indisvalid,i.indisready,t.relname
+                    FROM pg_index i JOIN pg_class ix ON ix.oid=i.indexrelid
+                    JOIN pg_class t ON t.oid=i.indrelid JOIN pg_namespace ns ON ns.oid=ix.relnamespace
+                    JOIN pg_am am ON am.oid=ix.relam WHERE ns.nspname=%s AND ix.relname='papers_search_gin' ''', (stage,)).fetchone()
+                if search != ('gin', True, True, 'papers'):
+                    raise ValueError('Native full-text index is not valid and ready')
+                for table in TABLES:
+                    pg.execute('ANALYZE ' + qualify(stage, table))
+                invalid = pg.execute('''SELECT count(*) FROM pg_constraint co JOIN pg_class cl ON cl.oid=co.conrelid
+                    JOIN pg_namespace ns ON ns.oid=cl.relnamespace WHERE ns.nspname=%s AND NOT co.convalidated''', (stage,)).fetchone()[0]
+                if invalid:
+                    raise ValueError('Unvalidated constraints remain')
+                if (db.stat().st_size, db.stat().st_mtime_ns) != (identity['bytes'], identity['mtime_ns']):
+                    raise ValueError('Source changed during migration')
+                if Path(str(db) + '-wal').exists() and Path(str(db) + '-wal').stat().st_size:
+                    raise ValueError('A source writer committed during migration')
+                report.update(status='prepared', foreign_keys='all_validated', checks='all_validated', native_search_index=True)
+                save(out, report)
+                with pg.transaction():
+                    pg.execute('ALTER SCHEMA ' + quote(stage) + ' RENAME TO ' + quote(schema))
+                    pg.execute('UPDATE ' + qualify(control, 'run') + " SET status='complete' WHERE id=1")
+                report.update(status='complete', finished_at=timestamp())
+                save(out, report)
+                return report
+            except BaseException as error:
+                report.update(status='error', error_type=type(error).__name__, error=str(error), failed_at=timestamp())
+                save(out, report)
+                raise
+
+
+def main(argv=None, *, prog=None):
+    parser = argparse.ArgumentParser(prog=prog)
+    parser.add_argument('--db', required=True, help='Frozen SQLite source')
+    parser.add_argument('--pg-service', required=True, help='Protected libpq service name')
+    parser.add_argument('--schema', default='research')
+    parser.add_argument('--out', required=True)
+    parser.add_argument('--storage-path', help='Target filesystem path for free-space checks')
+    parser.add_argument('--min-free-gib', type=float, default=8)
+    args = parser.parse_args(argv)
+    result = migrate(args.db, args.pg_service, args.out, schema=args.schema,
+                     storage_path=args.storage_path, min_free_gib=args.min_free_gib)
+    print(json.dumps(result, ensure_ascii=False), flush=True)
@@ -0,0 +1,120 @@
+"""Inspect the complete business schema; rebuild SQLite engine internals natively."""
+import hashlib
+import json
+import re
+
+TABLES = ('sources', 'papers', 'identifiers', 'source_records', 'venues', 'editions',
+          'venue_aliases', 'edition_sources', 'submissions', 'decisions',
+          'submission_sources', 'submission_identifiers', 'decision_sources',
+          'submission_merges', 'decision_merges', 'schema_migrations', 'tags',
+          'paper_tags', 'match_candidates', 'paper_merges', 'ingest_runs')
+CHECKS = {
+    'venues': ["venue_type IN ('conference','journal','review_platform')"],
+    'edition_sources': ["status IN ('pending','running','complete','partial','blocked','unavailable','not_yet_published')"],
+    'decisions': ["status IN ('accepted','rejected','desk_rejected','withdrawn','under_review','unknown')"],
+    'match_candidates': ["status IN ('pending','merged','distinct')"],
+}
+TYPES = {'INTEGER': 'BIGINT', 'TEXT': 'TEXT COLLATE "C"', 'REAL': 'DOUBLE PRECISION'}
+
+
+def checks_in(ddl):
+    result = []
+    for match in re.finditer(r'\bCHECK\s*\(', ddl, re.I):
+        start = position = match.end()
+        depth, quoted = 1, False
+        while position < len(ddl) and depth:
+            char = ddl[position]
+            if char == "'":
+                if quoted and position + 1 < len(ddl) and ddl[position + 1] == "'":
+                    position += 2; continue
+                quoted = not quoted
+            elif not quoted:
+                depth += (char == '(') - (char == ')')
+            position += 1
+        if depth:
+            raise ValueError('Unclosed CHECK constraint')
+        result.append(ddl[start:position - 1])
+    return result
+
+
+def normalize_check(expression):
+    # Normalize SQL layout/keywords, retaining case and spaces inside literals.
+    tokens = re.findall(r"'(?:[^']|'')*'|[^']+", expression)
+    return ''.join(token if token.startswith("'") else re.sub(r'\s+', '', token).lower() for token in tokens)
+
+
+def quote(name):
+    return '"' + name.replace('"', '""') + '"'
+
+
+def inspect(connection):
+    business = {name for schema, name, kind, *_ in connection.execute('PRAGMA table_list')
+                if schema == 'main' and kind == 'table' and not name.startswith('sqlite_')}
+    if business != set(TABLES):
+        raise ValueError('Unexpected business tables: ' + repr(sorted(business ^ set(TABLES))))
+    result = {}
+    for table in TABLES:
+        columns = [{'name': r[1], 'type': r[2].upper(), 'notnull': bool(r[3]),
+                    'default': r[4], 'pk': r[5]}
+                   for r in connection.execute('PRAGMA table_info(' + quote(table) + ')')]
+        if any(c['type'] not in TYPES for c in columns):
+            raise ValueError('Unsupported storage type in ' + table)
+        pk = [c['name'] for c in sorted(columns, key=lambda c: c['pk']) if c['pk']]
+        if not pk:
+            raise ValueError('A deterministic primary key is required for ' + table)
+        indexes = []
+        for row in connection.execute('PRAGMA index_list(' + quote(table) + ')'):
+            if row[3] == 'pk':
+                continue
+            if row[4]:
+                raise ValueError('Partial indexes require explicit migration support')
+            terms = [(r[2], bool(r[3]), r[4]) for r in connection.execute('PRAGMA index_xinfo(' + quote(row[1]) + ')') if r[5]]
+            if any(name is None or coll != 'BINARY' for name, desc, coll in terms):
+                raise ValueError('Unsupported expression or collation in ' + table)
+            indexes.append({'name': row[1], 'unique': bool(row[2]), 'origin': row[3],
+                            'terms': [(name, desc) for name, desc, coll in terms]})
+        fks = {}
+        for row in connection.execute('PRAGMA foreign_key_list(' + quote(table) + ')'):
+            if row[5:7] != ('NO ACTION', 'NO ACTION'):
+                raise ValueError('Unsupported foreign key actions')
+            fk = fks.setdefault(row[0], {'table': row[2], 'columns': [], 'references': []})
+            fk['columns'].append(row[3]); fk['references'].append(row[4])
+        ddl = connection.execute("SELECT sql FROM sqlite_schema WHERE type='table' AND name=?", (table,)).fetchone()[0]
+        checks = CHECKS.get(table, [])
+        if sorted(map(normalize_check, checks_in(ddl))) != sorted(map(normalize_check, checks)):
+            raise ValueError('Unrecognized CHECK constraint in ' + table)
+        result[table] = {'columns': columns, 'pk': pk, 'indexes': indexes,
+                         'foreign_keys': list(fks.values()), 'checks': checks}
+    return result
+
+
+def fingerprint(plan):
+    return hashlib.sha256(json.dumps(plan, sort_keys=True, separators=(',', ':')).encode()).hexdigest()
+
+
+def default_sql(value):
+    if value is None:
+        return ''
+    if value.upper() == 'CURRENT_TIMESTAMP':
+        return " DEFAULT to_char(CURRENT_TIMESTAMP AT TIME ZONE 'UTC','YYYY-MM-DD\"T\"HH24:MI:SS.MS\"Z\"')"
+    if re.fullmatch(r'-?\d+|\'(?:[^\']|\'\')*\'', value):
+        return ' DEFAULT ' + value
+    raise ValueError('Unsupported column default')
+
+
+def table_sql(schema, table, spec):
+    columns = []
+    for column in spec['columns']:
+        generated = column['name'] == 'id' and spec['pk'] == ['id'] and column['type'] == 'INTEGER'
+        value = quote(column['name']) + ' ' + TYPES[column['type']]
+        value += ' GENERATED BY DEFAULT AS IDENTITY' if generated else default_sql(column['default'])
+        if column['notnull']:
+            value += ' NOT NULL'
+        columns.append(value)
+    return 'CREATE TABLE ' + quote(schema) + '.' + quote(table) + '(' + ','.join(columns) + ')'
+
+
+def select_sql(schema, table, spec):
+    prefix = quote(schema) + '.' if schema else ''
+    return ('SELECT ' + ','.join(quote(c['name']) for c in spec['columns']) +
+            ' FROM ' + prefix + quote(table) + ' ORDER BY ' + ','.join(quote(c) for c in spec['pk']))
@@ -23,8 +23,11 @@ def known_acceptances(conn,paper):

 def main(argv=None, *, prog=None):
    parser = argparse.ArgumentParser(prog=prog)
-    parser.add_argument('--db', required=True)
-    parser.add_argument('--query', help='SQLite FTS5 query over title and abstract')
+    database = parser.add_mutually_exclusive_group(required=True)
+    database.add_argument('--db', help='SQLite file')
+    database.add_argument('--pg-service', help='Protected libpq service; credentials stay outside CLI arguments')
+    parser.add_argument('--pg-schema', help='Optional PostgreSQL schema override; otherwise use the configured search_path')
+    parser.add_argument('--query', help='Title/abstract search: FTS5 for SQLite, web-search syntax for PostgreSQL')
    parser.add_argument('--paper',type=int,help='Paper ID; include sources, tags and submission history')
    parser.add_argument('--tag', help='Filter as namespace=value, e.g. arxiv:category=cs.AI')
    parser.add_argument('--venue', help='Conference acronym')
@@ -35,9 +38,15 @@ def main(argv=None, *, prog=None):
    parser.add_argument('--limit', type=int, default=20)
    parser.add_argument('--stats', action='store_true')
    args = parser.parse_args(argv)
-    conn = sqlite3.connect(Path(args.db).resolve().as_uri()+'?mode=ro', uri=True)
-    conn.row_factory = sqlite3.Row
-    conn.execute('BEGIN')  # One consistent read snapshot while ingestion continues.
+    if args.pg_schema and not args.pg_service:
+        parser.error('--pg-schema requires --pg-service')
+    if args.pg_service:
+        from paper_atlas.postgres.client import ReadConnection
+        conn = ReadConnection(args.pg_service, schema=args.pg_schema) if args.pg_schema else ReadConnection(args.pg_service)
+    else:
+        conn = sqlite3.connect(Path(args.db).resolve().as_uri()+'?mode=ro', uri=True)
+        conn.row_factory = sqlite3.Row
+        conn.execute('BEGIN')  # One consistent read snapshot while ingestion continues.
    if args.stats:
        result = {name:conn.execute('SELECT count(*) FROM '+name).fetchone()[0] for name in ('papers','source_records','submissions','decisions','tags','match_candidates')}
        result['records_by_source'] = [dict(r) for r in conn.execute('SELECT s.name,count(r.id) AS imported_records FROM sources s LEFT JOIN source_records r ON r.source_id=s.id GROUP BY s.id ORDER BY s.name')]
@@ -53,7 +62,12 @@ def main(argv=None, *, prog=None):
        if args.paper is not None:
            where.append('p.id=?'); params.append(args.paper)
        if args.query:
-            where.append('p.id IN (SELECT rowid FROM papers_fts WHERE papers_fts MATCH ?)'); params.append(args.query)
+            if args.pg_service:
+                from paper_atlas.postgres.client import SEARCH_VECTOR
+                where.append('p.id IN (SELECT id FROM papers WHERE '+SEARCH_VECTOR+" @@ websearch_to_tsquery('english'::regconfig,?))")
+            else:
+                where.append('p.id IN (SELECT rowid FROM papers_fts WHERE papers_fts MATCH ?)')
+            params.append(args.query)
        if args.tag:
            namespace,value = args.tag.split('=',1)
            where.append('EXISTS(SELECT 1 FROM paper_tags pt JOIN tags t ON t.id=pt.tag_id WHERE pt.paper_id=p.id AND t.namespace=? AND t.value=?)'); params.extend([namespace,value])
@@ -63,9 +77,9 @@ def main(argv=None, *, prog=None):
            clause = ['1=1']
            if args.venue:
                if conn.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='venue_aliases'").fetchone():
-                    clause.append('(v.acronym=? COLLATE NOCASE OR EXISTS(SELECT 1 FROM venue_aliases va WHERE va.venue_id=v.id AND va.alias=? COLLATE NOCASE))'); params.extend([args.venue,args.venue])
+                    clause.append('(lower(v.acronym)=lower(?) OR EXISTS(SELECT 1 FROM venue_aliases va WHERE va.venue_id=v.id AND lower(va.alias)=lower(?)))'); params.extend([args.venue,args.venue])
                else:
-                    clause.append('v.acronym=? COLLATE NOCASE');params.append(args.venue)
+                    clause.append('lower(v.acronym)=lower(?)');params.append(args.venue)
            if args.track:
                clause.append('e.track=?');params.append(args.track)
            if args.year:
@@ -12,7 +12,9 @@
    "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"
+    "iclr_program": "https://iclr.cc/virtual/2026/papers.html",
+    "mlsys": "https://proceedings.mlsys.org/",
+    "acmmm": "https://www.acmmm.org/"
  },
  "venues": [
    {
@@ -643,6 +645,37 @@
        "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": [
@@ -10,6 +10,9 @@ readme = "README.md"
 requires-python = ">=3.12"
 dependencies = ["PyYAML==6.0.3"]

+[project.optional-dependencies]
+postgres = ["psycopg[binary]==3.3.8"]
+
 [project.scripts]
 paper-atlas = "paper_atlas.cli:main"

@@ -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()
@@ -18,7 +18,10 @@ class CliTests(unittest.TestCase):

    def test_unified_import_and_query_use_shipped_schema_and_registry(self):
        from paper_atlas.config import DEFAULT_REGISTRY
-        self.assertEqual(len(json.loads(DEFAULT_REGISTRY.read_text())['venues']),41)
+        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)))
        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')
@@ -0,0 +1,166 @@
+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()
@@ -0,0 +1,212 @@
+import contextlib
+from contextlib import closing
+import io
+import json
+import os
+from pathlib import Path
+import sqlite3
+import tempfile
+import unittest
+from unittest.mock import patch
+import uuid
+
+from paper_atlas.db.store import Store
+from paper_atlas.postgres import schema
+from paper_atlas.postgres.client import ReadConnection, Row, connect
+from paper_atlas.postgres.migration import migrate
+from paper_atlas import querying
+
+
+class SchemaTests(unittest.TestCase):
+    def setUp(self):
+        self.temporary = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temporary.cleanup)
+        self.path = Path(self.temporary.name) / 'source.sqlite'
+        self.store = Store(self.path)
+        self.addCleanup(self.store.conn.close)
+
+    def test_every_business_table_is_included_and_engine_tables_are_excluded(self):
+        plan = schema.inspect(self.store.conn)
+        self.assertEqual(set(plan), set(schema.TABLES))
+        self.assertEqual(len(plan), 21)
+        self.assertNotIn('papers_fts', plan)
+        self.assertEqual(plan['paper_tags']['pk'], ['paper_id', 'tag_id', 'evidence_key'])
+        self.assertEqual(len(plan['source_records']['foreign_keys']), 2)
+        sql = schema.table_sql('stage', 'papers', plan['papers'])
+        self.assertIn('GENERATED BY DEFAULT AS IDENTITY', sql)
+        self.assertIn('TEXT COLLATE "C"', sql)
+
+    def test_unknown_tables_and_check_constraints_fail_before_copy(self):
+        self.store.conn.execute('CREATE TABLE extra(id INTEGER PRIMARY KEY)')
+        with self.assertRaisesRegex(ValueError, 'Unexpected business'):
+            schema.inspect(self.store.conn)
+        self.store.conn.execute('DROP TABLE extra')
+        self.store.conn.execute('ALTER TABLE papers ADD COLUMN extra TEXT CHECK(extra IS NULL)')
+        with self.assertRaisesRegex(ValueError, 'CHECK'):
+            schema.inspect(self.store.conn)
+
+    def test_cutover_stops_legacy_writers_but_keeps_read_access(self):
+        self.store.conn.commit(); self.store.conn.close()
+        Path(str(self.path) + '.postgresql-cutover.json').write_text('{"status":"active"}')
+        with self.assertRaisesRegex(RuntimeError, 'cut over'):
+            Store(self.path)
+        with sqlite3.connect(self.path.as_uri() + '?mode=ro', uri=True) as c:
+            self.assertEqual(c.execute('SELECT count(*) FROM papers').fetchone()[0], 0)
+
+    def test_rows_support_mapping_and_positional_query_access(self):
+        row = Row(['id', 'title'], (7, 'A'))
+        self.assertEqual(row[0], row['id'])
+        self.assertEqual(dict(row), {'id': 7, 'title': 'A'})
+
+    def test_connection_services_cannot_be_password_urls(self):
+        with self.assertRaisesRegex(ValueError, 'service name'):
+            connect('postgresql://user:secret@host/db')
+
+    def test_checks_compare_literal_values_not_only_constraint_counts(self):
+        clauses = schema.checks_in("CREATE TABLE t(x TEXT CHECK(x IN ('a','b)')),y TEXT CHECK(y IS NULL))")
+        self.assertEqual(clauses, ["x IN ('a','b)')", 'y IS NULL'])
+        self.assertEqual(schema.normalize_check("x IN ( 'a' )"), schema.normalize_check("x in('a')"))
+        self.assertNotEqual(schema.normalize_check("x IN ('A')"), schema.normalize_check("x IN ('a')"))
+
+
+@unittest.skipUnless(os.environ.get('PAPER_ATLAS_TEST_PG_SERVICE'), 'Requires an explicitly configured remote PostgreSQL test service')
+class PostgreSQLIntegrationTests(unittest.TestCase):
+    def setUp(self):
+        self.service = os.environ['PAPER_ATLAS_TEST_PG_SERVICE']
+        self.target = 'pa_test_' + uuid.uuid4().hex[:14]
+        self.temporary = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temporary.cleanup)
+        self.db = Path(self.temporary.name) / 'source.sqlite'
+        self.out = Path(self.temporary.name) / 'report.json'
+        store = Store(self.db)
+        paper, record = store.record('arxiv', '2401.00001', 'https://arxiv.org/abs/2401.00001',
+            'Quiet graph models — 研究', 'Graph models learn reinforcement policies.\nQuoted "text", tabs\tand Unicode α.',
+            ['Jane Doe', 'John Smith'], {'arxiv': '2401.00001'}, {'nested': {'a': ['α', None, 1]}, 'raw': '\\n'})
+        store.record_tags(paper, record, 'arxiv', [{'namespace': 'arxiv:category', 'value': 'cs.AI'}])
+        store.decision(paper, record, 'ICLR', 2024, 'main', 'forum-reject', 'https://example.org/reject',
+                       'rejected', 'Reject', identities=[{'scheme': 'openreview', 'value': 'forum-reject'}], observed_at='2024-01-01T00:00:00Z')
+        same, second = store.record('official', '2025/paper', 'https://example.org/accepted',
+            'Quiet graph models — 研究', 'Graph models learn reinforcement policies.', ['Jane Doe', 'John Smith'], {'arxiv': '2401.00001'})
+        self.assertEqual(same, paper)
+        store.decision(paper, second, 'COLM', 2025, 'main', 'forum-accept', 'https://example.org/accept',
+                       'accepted', 'Accept', ['Conference on Language Modeling'], observed_at='2025-01-01T00:00:00Z')
+        store.record('arxiv', '2401.00002', 'https://arxiv.org/abs/2401.00002', 'Empty abstract', None, [])
+        tag = store.conn.execute("INSERT INTO tags(namespace,value) VALUES('custom:topic','manual')").lastrowid
+        store.conn.execute('INSERT INTO paper_tags VALUES(?,?,NULL,?)', (paper, tag, 'manual:one'))
+        store.conn.execute("INSERT INTO paper_merges(former_paper_id,surviving_paper_id,reason) VALUES(900,?,'fixture')", (paper,))
+        store.conn.execute("INSERT INTO submission_merges VALUES(5000,1,'fixture',CURRENT_TIMESTAMP)")
+        store.conn.execute("INSERT INTO decision_merges VALUES(6000,1,'fixture',CURRENT_TIMESTAMP)")
+        store.conn.execute("UPDATE sqlite_sequence SET seq=1000 WHERE name='papers'")
+        store.conn.commit(); store.conn.close()
+        self.addCleanup(self.cleanup_schemas)
+
+    def cleanup_schemas(self):
+        with connect(self.service) as pg:
+            names = pg.execute('SELECT nspname FROM pg_namespace WHERE starts_with(nspname,%s)', (self.target,)).fetchall()
+            for name, in names:
+                pg.execute('DROP SCHEMA ' + schema.quote(name) + ' CASCADE')
+
+    def migrate(self):
+        with contextlib.redirect_stdout(io.StringIO()):
+            return migrate(self.db, self.service, self.out, schema=self.target, min_free_gib=0)
+
+    def published(self):
+        with connect(self.service) as pg:
+            return bool(pg.execute('SELECT 1 FROM pg_namespace WHERE nspname=%s', (self.target,)).fetchone())
+
+    def test_full_content_history_tags_sequences_and_readonly_queries(self):
+        report = self.migrate()
+        self.assertEqual(report['status'], 'complete')
+        self.assertEqual(len(report['tables']), 21)
+        for entry in report['tables'].values():
+            self.assertEqual(entry['source_rows'], entry['target_rows'])
+            self.assertEqual(entry['source_sha256'], entry['target_sha256'])
+        with connect(self.service) as pg, pg.transaction(force_rollback=True):
+            prefix = schema.quote(self.target) + '.'
+            value = pg.execute('INSERT INTO ' + prefix + 'papers(title,title_key) VALUES(%s,%s) RETURNING id', ('new', 'new')).fetchone()[0]
+            self.assertGreater(value, 1000)
+            record = pg.execute('INSERT INTO ' + prefix + "source_records(source_id,external_id,paper_id,url,title,abstract,authors_json,metadata_json) SELECT source_id,'new',%s,url,title,abstract,authors_json,metadata_json FROM " + prefix + 'source_records LIMIT 1 RETURNING id', (value,)).fetchone()[0]
+            edition = pg.execute('SELECT id FROM ' + prefix + 'editions LIMIT 1').fetchone()[0]
+            submission = pg.execute('INSERT INTO ' + prefix + 'submissions(paper_id,edition_id,source_record_id,source_submission_id) VALUES(%s,%s,%s,%s) RETURNING id', (value, edition, record, 'new')).fetchone()[0]
+            decision = pg.execute('INSERT INTO ' + prefix + "decisions(submission_id,status,raw_value,evidence_url,observed_at) VALUES(%s,'unknown','fixture','https://example.org/new','2026-01-01T00:00:00Z') RETURNING id", (submission,)).fetchone()[0]
+            self.assertGreater(submission, 5000)
+            self.assertGreater(decision, 6000)
+        real = ReadConnection
+        with patch('paper_atlas.postgres.client.ReadConnection', lambda service: real(service, self.target)):
+            output = io.StringIO()
+            with contextlib.redirect_stdout(output):
+                querying.main(['--pg-service', self.service, '--query', 'graph models', '--status', 'rejected', '--outcome', 'accepted'])
+            results = json.loads(output.getvalue())
+            self.assertEqual(len(results), 1)
+            self.assertEqual(results[0]['known_acceptances'][0]['venue'], 'COLM')
+            output = io.StringIO()
+            with contextlib.redirect_stdout(output):
+                querying.main(['--pg-service', self.service, '--venue', 'conference on language modeling', '--tag', 'custom:topic=manual', '--paper', '1'])
+            detail = json.loads(output.getvalue())[0]
+            self.assertEqual(len(detail['submissions']), 2)
+            self.assertEqual({t['namespace'] for t in detail['tags']}, {'arxiv:category', 'custom:topic'})
+            output = io.StringIO()
+            with contextlib.redirect_stdout(output):
+                querying.main(['--pg-service', self.service, '--stats'])
+            stats = json.loads(output.getvalue())
+            self.assertEqual(stats['papers'], 2)
+            self.assertEqual(stats['papers_with_known_acceptance'], 1)
+        reader = real(self.service, self.target)
+        try:
+            import psycopg
+            with self.assertRaises(psycopg.errors.ReadOnlySqlTransaction):
+                reader.execute('DELETE FROM papers')
+        finally:
+            reader.close()
+
+    def interrupt_after_first_table(self):
+        calls = 0
+        def interrupt(*args):
+            nonlocal calls
+            calls += 1
+            if calls == 3:
+                raise RuntimeError('controlled interruption')
+        with patch('paper_atlas.postgres.migration.free_space', side_effect=interrupt):
+            with self.assertRaisesRegex(RuntimeError, 'controlled interruption'):
+                self.migrate()
+        self.assertFalse(self.published())
+
+    def test_interrupted_copy_resumes_without_duplicate_rows(self):
+        self.interrupt_after_first_table()
+        self.assertEqual(self.migrate()['status'], 'complete')
+
+    def test_resume_rejects_a_changed_source_snapshot(self):
+        self.interrupt_after_first_table()
+        with closing(sqlite3.connect(self.db)) as source, source:
+            source.execute("UPDATE papers SET abstract='changed' WHERE id=1")
+        with self.assertRaisesRegex(ValueError, 'exact same'):
+            self.migrate()
+        self.assertFalse(self.published())
+
+    def test_full_hash_detects_tampering_in_a_completed_staging_table(self):
+        self.interrupt_after_first_table()
+        with connect(self.service) as pg:
+            stage = pg.execute('SELECT stage FROM ' + schema.quote(self.target + '_migration') + '.run WHERE id=1').fetchone()[0]
+            pg.execute('UPDATE ' + schema.quote(stage) + ".sources SET homepage='tampered'")
+        with self.assertRaisesRegex(ValueError, 'content verification'):
+            self.migrate()
+        self.assertFalse(self.published())
+
+    def test_foreign_key_violation_never_publishes_target(self):
+        with closing(sqlite3.connect(self.db)) as source, source:
+            source.execute('PRAGMA foreign_keys=OFF')
+            source.execute("INSERT INTO identifiers VALUES('orphan','bad',999999)")
+        import psycopg
+        with self.assertRaises(psycopg.errors.ForeignKeyViolation):
+            self.migrate()
+        self.assertFalse(self.published())
+
+    def test_published_target_cannot_be_overwritten(self):
+        self.migrate()
+        self.out.unlink()
+        with self.assertRaisesRegex(ValueError, 'never be overwritten'):
+            self.migrate()
+
+
+if __name__ == '__main__':
+    unittest.main()
Author	SHA1	Message	Date
codex-bot	a3b677bdf2	Add verified PostgreSQL migration and shared research queries Metadata integrity checks / metadata (push) Successful in 46s Details	2026-11-07 02:09:19 +08:00
Zayden-Jung	029e73cafe	Merge pull request '补齐 ACM MM、MLSys 范围并接入官方摘要与发表身份' (#12 ) from codex/multimedia-mlsys into main Metadata integrity checks / metadata (push) Successful in 27s Details Reviewed-on: #12	2026-11-06 16:15:16 +00:00
Zayden-Jung	2e95e91f5e	Merge pull request '逐条审计 arXiv 类别完整性及来源证据' (#10 ) from codex/arxiv-tag-audit into main Metadata integrity checks / metadata (push) Successful in 28s Details Reviewed-on: #10	2026-11-06 16:15:09 +00:00
Zayden-Jung	a58b96341a	Merge pull request '修正 ICML 多轨道分类并保留既有投稿与决定历史' (#11 ) from codex/pmlr-track-scope into main Metadata integrity checks / metadata (push) Successful in 28s Details Reviewed-on: #11	2026-11-06 16:15:04 +00:00
codex-bot	9a279f2eae	Handle MLSys year-labelled proceedings and bridge retained publisher evidence Metadata integrity checks / metadata (push) Successful in 53s Details	2026-11-06 22:58:50 +08:00
codex-bot	25cc5b865b	Preserve unspecified MM tracks and validate MLSys identities at the shared boundary Metadata integrity checks / metadata (push) Successful in 41s Details	2026-11-06 22:49:41 +08:00
codex-bot	c9b0cfa857	Bound arXiv tag audit batches by primary key instead of sorting the source Metadata integrity checks / metadata (push) Successful in 43s Details	2026-11-06 22:41:43 +08:00
codex-bot	a65edc6b9d	Add ACM Multimedia scope and MLSys metadata with shared publication identities Metadata integrity checks / metadata (push) Successful in 33s Details	2026-11-06 22:39:55 +08:00
codex-bot	bfa7fe2e94	Merge remote-tracking branch 'origin/main' into codex/arxiv-tag-audit Metadata integrity checks / metadata (push) Successful in 32s Details # Conflicts: # paper_atlas/cli.py	2026-11-06 21:51:29 +08:00
codex-bot	d712a4fbed	Audit every arXiv record category and evidence owner after repair Metadata integrity checks / metadata (push) Successful in 32s Details	2026-11-06 21:43:50 +08:00
				`@@ -0,0 +1 @@`
				`"""PostgreSQL migration and read access for the paper index."""`