Audit public decisions against PostgreSQL and recheck imports

Merge pull request '为 PostgreSQL 主库接入元数据导入与投稿证据写入' (#15 ) from codex/postgres-ingestion into main
Reviewed-on: #15
2026-11-07 16:01:27 +08:00 · 2026-11-07 07:53:18 +00:00 · 2026-11-07 15:38:42 +08:00 · 2026-11-06 19:37:00 +00:00 · 2026-11-07 03:27:57 +08:00 · 2026-11-06 18:56:51 +00:00
41 changed files with 2603 additions and 169 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,13 @@ 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)
+              subprocess.run([python,'-I','-B','-m','paper_atlas','import','records','--help'],cwd=outside,check=True)
+              subprocess.run([python,'-I','-B','-m','paper_atlas','import','arxiv','--help'],cwd=outside,check=True)
+              subprocess.run([python,'-I','-B','-m','paper_atlas','maintain','backfill','--help'],cwd=outside,check=True)
+              subprocess.run([python,'-I','-B','-m','paper_atlas','audit','public-openreview','--help'],cwd=outside,check=True)
+              subprocess.run([python,'-I','-B','-c','from paper_atlas.postgres.writer import PostgresStore, WriteConnection'],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 为只读备份；`import records`、`import arxiv` 和 `maintain backfill` 使用 `--pg-service foundry-research` 写入主库，整个写入会话只允许一个协调写入者，会议文件失败时整体回滚并保留失败批次记录。
+
 ## 数据如何组织

 - **论文和投稿分开。** 一篇论文可以多次投稿；某次被拒、后来在其他会议录用，两段历史都保留。
@@ -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
@@ -300,3 +320,13 @@ paper-atlas collect cvf --cache /data/paper-index/raw/http-cache \
 ```

 parser_version 2 的清单还保存实际遍历的 index_urls、入口错误及缺失摘要页面。补采结果经过清单与身份核验后才导入，不能因重试任务启动就标记缺口已解决。
+
+## PMLR 中的 ICML 多轨道论文
+
+ICML 2024 起的一个 PMLR 卷可以同时包含研究主轨道与 Position Paper，不能把整卷都标为 `main`。采集器 v3 按轨道输出 `pmlr-v<VOLUME>-main.jsonl`、`pmlr-v<VOLUME>-position.jsonl` 及各自 manifest；卷级 `pmlr-v<VOLUME>.inventory.json` 和目录清单的 `files` 数组列出实际文件。导入队列应展开 `files`，不能假定每卷只有一个 `file`。
+
+2025、2026 年通过出版社元数据的真实 OpenReview forum ID，匹配官方完整节目单中的 `Conference` / `Position_Paper_Track` 来源组；忽略负编号生成的口头报告副本，遇到冲突、缺少身份或未匹配的发表记录停止发布结果。每条记录保存节目单 URL、SHA-256、事件 ID、关联身份及轨道证据。
+
+2024 年的官方节目单把两轨都放在 `Conference`，但提供 PMLR 论文自链接。该年仅采用组织方在 [2025 Position CFP](https://icml.cc/Conferences/2025/CallForPositionPapers) 明确链接的 [2024 Position 论文标题筛选](https://icml.cc/virtual/2024/papers.html?filter=titles&search=position%3A) 规则；同时保存规则页面的哈希。这个规则不用于其他会议或年份。官方节目单将 `guyard24a` 链接误写为 `theo24a`：只对这条已核查的映射额外要求标题精确一致、至少两名公开作者重合，不进行全库模糊匹配。
+
+导入与身份回填拒绝 ICML 2024 起 parser_version<3 或没有官方轨道证据的旧文件。已有主库中的错误轨道需要在独占维护锁内迁移原投稿事件、保留决定与来源关联；仅重新导入拆分文件会留下旧的主轨道事件，不能作为修复方式。完成迁移前，不把新文件直接导入现有库。
@@ -1,5 +1,49 @@
 # 运行维护与数据核验

+## 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 索引，不逐字复制引擎文件。
+
+SQLite 允许文本包含 `U+0000`（NUL），PostgreSQL `text` 不允许。迁移使用 `nul-soh-v1` 可逆编码：NUL 存为 `U+0001 U+0001`，原有 `U+0001` 存为 `U+0001 U+0002`，其他字符不变；包括字串前缀在内的二进制排序保持一致，主键、唯一约束和外键可继续验证。编码版本保存在 `<schema>_migration.run.text_codec`，发布时同时写入 schema 注释 `paper-atlas:text-codec=nul-soh-v1`，其他项目只需研究表的权限即可读取格式信息，无须访问迁移控制表。CLI 的只读连接自动还原结果并编码筛选参数；直接 SQL/ETL 调用者应使用 `paper_atlas.postgres.text.encode/decode`，不能删除或直接替换控制字符。全文检索把这些控制字符当作分隔符，不承担逐字节匹配。
+
+先 COPY 到独立暂存 schema，每张表提交和记录哈希。中断后以同一命令、同一报告路径恢复，必须仍是**完全相同的未变更快照**。不同输入、已发布目标、未清空 WAL、新的未知业务表都会拒绝覆盖；恢复只会清空本次运行尚未完成且尚未发布的暂存表。保留错误现场，不盲目从头复制。
+
+恢复旧版本未发布的暂存数据时，在一个事务中转义已提交表里原有的 `U+0001`，然后记录新编码版本；失败回滚，重复恢复不会二次编码。已经提交的表无需从 SQLite 重新复制。报告中的 `source_sha256`、`target_sha256` 比较完整还原后的原始值；`target_storage_sha256` 另记实际存储值的哈希，包含转义的表两种哈希可以不同。
+
+发布前逐表按主键完整回读，比较数量及逐行 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 验证。
+
+## PostgreSQL 增量写入
+
+`import records`、`import arxiv`、`maintain backfill` 使用 `--pg-service` 选择主库，`--pg-schema` 可覆盖配置的 search_path。写入只接受已经发布、拥有完整业务表和受支持编码标记的 schema，不自动建库或引导空表。SQLite 样本仍使用 `--db`，已切换备份继续禁止写入。
+
+写入连接按数据库/schema 获取会话级 advisory lock，文件提交、回滚以及 arXiv 分批提交均不释放；关闭连接才释放，其他写入者立即得到明确的占用错误。只读查询继续使用独立一致性快照。锁是所有调用方必须遵守的协议，其他项目不要绕过接口并发修改同一论文和投稿；[PostgreSQL 会话锁说明](https://www.postgresql.org/docs/17/explicit-locking.html#ADVISORY-LOCKS)解释了其事务边界。
+
+论文身份匹配、跨来源合并、投稿/决定来源和标签维护复用同一套领域逻辑。PostgreSQL 使用原生 RETURNING 和 identity 序列分配 ID，继承迁移时设定的历史 ID 高水位；失败和冲突尝试可以留下序列空洞，不能把最大 ID 当作行数。绑定文本按 `nul-soh-v1` 编码，结果解码，GIN 随论文写入自动更新。
+
+会议文件的数据变化在同一事务内完成，manifest 数量不符等错误会回滚整份文件，`ingest_runs` 另保留 failed、已处理数和零写入数；身份回填整个队列一次提交。arXiv 每 500 行提交数据和 checkpoint，异常只回滚尚未提交部分，重试必须仍为同一压缩包。大批次开始前核对独立审计清单、当前主库冲突和磁盘空间；完成后对数量、实际来源/决定关联及检索做数据验收，代码测试成功不代表生产导入成功。
+
+`audit public-openreview --pg-service ...` 在只读快照中核验两个受支持公开负面决定来源的固定 revision、逐文件哈希、manifest、原始出处标记和既有同届投稿。最新决定排序与查询端一致：观察时间、决定日期（为空时回退观察时间）、ID。实际导入持有写锁后再次复核该投稿的最新状态，阻止过期预检或弱来源标签覆盖冲突决定；冲突文件整体回滚并等待进一步证据核查，其他会议录用不构成同一次投稿的冲突。
+
+会议导入按原始字节流计算 SHA-256，在提交前与 manifest 提供的 `file_sha256` 比较；不符则回滚整份文件。成功批次的 checkpoint 会记录实际读取内容的哈希，便于复核导入时使用的文件版本。
+
 代码以固定 commit 部署，已经启动的任务继续使用其原版本。目录重构不要求迁移数据库，也不更改已有 JSONL 和 manifest 格式。

 ## arXiv 原始编号与类别证据
@@ -61,3 +105,18 @@ paper-atlas coverage --db /data/paper-index/papers.sqlite \
 manifest 和采集器状态清单用于确认文件是否齐全及是否缺摘要；`ingest_runs` 和协调器日志用于确认主库导入是否成功。官方程序、书目目录和公开投稿快照分别保留，未出现在某份清单不等于被拒稿；第三方快照也不能代表平台当前全量状态。代码测试成功不能代替最终全库去重、完整性和逐届覆盖核验。

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

 先在已确认的远程环境安装项目，使用 `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 来源列保留原始文本；转成 `jsonb` 前须确认原始值符合该类型的限制。
+
+CLI 自动还原迁移中的 NUL/SOH 文本转义，详情中的来源摘要和筛选参数保留原字符。直接 SQL 读取可能看到 `U+0001` 转义序列；跨项目 ETL 应按迁移元数据中的 `text_codec` 使用 `paper_atlas.postgres.text.encode/decode`。存储格式和无损哈希验收见[运行维护](operations.md#postgresql-迁移)。
+
+迁移与切换见[运行维护](operations.md#postgresql-迁移)。`import records`、`import arxiv` 和 `maintain backfill` 均支持 `--pg-service` 和可选 `--pg-schema`；与 `--db` 二选一。SQLite 导入仅用于未切换的索引或独立样本。尚未适配 PostgreSQL 的维护、覆盖审计命令仍可只读检查冻结备份，不能当作主库的当前状态。
+
 ## 1. 下载固定快照

 `paper-atlas download --help` 说明大小、校验和与续传参数。各来源的固定版本与具体命令见[数据源说明](data-sources.md)。下载成功只表示原始文件已核验。
@@ -23,12 +41,35 @@ paper-atlas audit public-openreview --help

 其他会议的公开拒稿使用 `collect public-openreview`，再以 `audit public-openreview` 核对既有投稿身份和决定冲突。来源版本、逐文件状态盘点和未解决记录分别保存在 inventory 与 held 清单中，具体命令见[数据源说明](data-sources.md#其他会议的公开拒稿)。这类公开子集不能当作会议完整投稿名单。

-## 3. 串行导入
-
-首次 arXiv 导入使用 `paper-atlas import arxiv --db ... --archive ...`。现有数据库合并已核验的会议文件使用：
+主库切换后，预检须直接连接当前 PostgreSQL，不能用冻结的 SQLite 备份推断当前决定：

 ```sh
-paper-atlas maintain merge --db /data/paper-index/papers.sqlite \
+paper-atlas audit public-openreview --pg-service foundry-research \
+  --inventory /data/paper-index/collected/public-openreview-v1/public-openreview-inventory.json \
+  --out /data/paper-index/collected/public-negative-pg-preflight.json
+```
+
+预检使用只读一致性快照，报告会记录实际数据库、schema、文件哈希和同届次/轨道的决定冲突；其他会议的录用保留为独立历史。预检不预留写锁，实际导入这两类公开负面决定时会在写事务中再次复核同一次投稿的最新状态；过期预检报告不能覆盖已经变化的决定。
+
+## 3. 串行导入
+
+PostgreSQL 主库逐文件导入已经核验的资料：
+
+```sh
+paper-atlas import records --pg-service foundry-research \
+  /data/paper-index/collected/selected-file.jsonl
+paper-atlas maintain backfill --pg-service foundry-research \
+  --queue /data/paper-index/collected/audited-identity-queue.json
+```
+
+文件和 queue 须替换为实际核验通过的路径。会议导入按文件提交，失败时该文件的论文、标签和决定更改全部回滚，失败批次日志保留；重跑会按来源标识与决定证据去重。回填使用已有源记录，只合并经独立身份佐证的投稿，不重新导入文本。所有写入者须遵守同一数据库/schema 的会话锁；其他项目也应通过相同的写入接口更新共享论文和投稿。
+
+未来需要导入新的 arXiv 元数据包时使用 `paper-atlas import arxiv --pg-service foundry-research --archive ... --storage-path ...`。每 500 行提交断点，恢复同一文件时保留已提交 ID 和标签证据；`--storage-path` 指定运行主机上要监测的文件系统，在数据库主机执行时应指向数据库卷。已完成的历史压缩包不需要重复导入。
+
+SQLite 独立样本的首次 arXiv 导入使用 `paper-atlas import arxiv --db ... --archive ...`。SQLite 协调器仍适用于未切换的索引：
+
+```sh
+paper-atlas maintain merge --db /data/paper-index/samples/example.sqlite \
  --archive /data/paper-index/raw/arxiv-metadata.zip \
  /data/paper-index/collected/selected-file.jsonl
 ```
@@ -4,14 +4,14 @@ from contextlib import closing
 import hashlib
 import json
 from pathlib import Path
-import sqlite3

 from paper_atlas.collectors.public_openreview import SOURCE, REVISION
 from paper_atlas.collectors.icml_public_archive import SOURCE as ARCHIVE_SOURCE, REVISION as ARCHIVE_REVISION
-from paper_atlas.db.submissions import timestamp
+from paper_atlas.db.submissions import timestamp, latest_status
+from paper_atlas.db.backend import database_arguments, open_read_connection


-def audit(db, inventory, out):
+def audit(db, inventory, out, *, pg_service=None, pg_schema=None):
    inventory = Path(inventory).resolve()
    result = json.loads(inventory.read_text())
    specs = {SOURCE: (REVISION, 'curated_public_status_assertion'),
@@ -20,17 +20,18 @@ def audit(db, inventory, out):
    if source not in specs or result.get('revision') != specs[source][0]:
        raise ValueError('Unsupported public source inventory')
    revision, evidence_kind = specs[source]
-    db = Path(db).resolve()
-    if not db.is_file():
-        raise ValueError('Existing database required for identity and status audit')
    conflicts = []
    seen = set()
    checked = 0
    existing_attempts = 0
    files = []
    checksums = {}
-    with closing(sqlite3.connect(db.as_uri() + '?mode=ro', uri=True)) as c:
-        c.execute('BEGIN')
+    with closing(open_read_connection(db, pg_service=pg_service, pg_schema=pg_schema)) as c:
+        if pg_service:
+            identity = c.execute('SELECT current_database(),current_schema()').fetchone()
+            database, actual_schema = identity[0], identity[1]
+        else:
+            database, actual_schema = str(Path(db).resolve()), None
        for manifest in result['files']:
            path = Path(manifest['file'])
            if hashlib.sha256(path.read_bytes()).hexdigest() != manifest['file_sha256']:
@@ -62,12 +63,11 @@ def audit(db, inventory, out):
                    raise ValueError('Multiple canonical attempts for one forum')
                if existing:
                    existing_attempts += 1
-                    latest = c.execute('''SELECT status FROM decisions WHERE submission_id=?
-                        ORDER BY observed_at DESC,COALESCE(decision_date,'') DESC,id DESC LIMIT 1''', existing[0]).fetchone()
-                    if latest and latest[0] not in ('unknown', 'under_review', row['status']):
+                    latest = latest_status(c, existing[0][0])
+                    if latest and latest not in ('unknown', 'under_review', row['status']):
                        conflicts.append({'forum': forum, 'venue': row['venue'], 'year': row['year'],
                                          'track': row['track'], 'submission_id': existing[0][0],
-                                          'existing_status': latest[0], 'curated_status': row['status']})
+                                          'existing_status': latest, 'curated_status': row['status']})
                count += 1
            if count != manifest['count']:
                raise ValueError('Collected record count differs from inventory')
@@ -76,7 +76,8 @@ def audit(db, inventory, out):
            checksums[str(path)] = manifest['file_sha256']
    if checked != result['normalized_records']:
        raise ValueError('Normalized inventory total differs from checked records')
-    report = {'source': source, 'revision': revision, 'database': str(db),
+    report = {'source': source, 'revision': revision, 'database': database,
+              'engine': 'postgresql' if pg_service else 'sqlite', 'schema': actual_schema, 'pg_service': pg_service,
              'inventory': str(inventory), 'checked_at': timestamp(), 'records_checked': checked,
              'existing_same_edition_attempts': existing_attempts, 'conflicts': conflicts,
              'status': 'blocked' if conflicts else 'ready_for_serial_import',
@@ -94,11 +95,11 @@ def audit(db, inventory, out):

 def main(argv=None, *, prog=None):
    parser = argparse.ArgumentParser(prog=prog)
-    parser.add_argument('--db', required=True)
+    database_arguments(parser)
    parser.add_argument('--inventory', required=True)
    parser.add_argument('--out', required=True)
    args = parser.parse_args(argv)
-    report = audit(args.db, args.inventory, args.out)
+    report = audit(args.db, args.inventory, args.out, pg_service=args.pg_service, pg_schema=args.pg_schema)
    if report['status'] == 'blocked':
        raise SystemExit('Conflicting existing decisions require review before import')

@@ -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 历史发表记录'),
@@ -27,6 +28,8 @@ COMMANDS = {
    ('maintain','merge'): ('paper_atlas.ingestion.maintenance', {}, '协调单写入合并与恢复'),
    ('maintain','backfill'): ('paper_atlas.ingestion.backfill', {}, '回填既有投稿的来源身份'),
    ('maintain','arxiv-tags'): ('paper_atlas.ingestion.arxiv_tags', {}, '恢复逐来源记录的 arXiv 类别证据'),
+    ('maintain','pmlr-tracks'): ('paper_atlas.ingestion.pmlr_tracks', {}, '核对并迁移既有 PMLR 投稿轨道'),
+    ('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 完整性'),
@@ -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,5 +1,7 @@
 """Collect official PMLR metadata, never the PDFs linked by the metadata."""
 import argparse
+from collections import defaultdict
+import hashlib
 from paper_atlas.config import DEFAULT_REGISTRY
 import io
 import json
@@ -8,6 +10,7 @@ import os
 import subprocess
 import tarfile
 import time
+import tempfile
 from pathlib import Path

 import yaml
@@ -37,6 +40,7 @@ def venue_key(shortname):


 from paper_atlas.io.http import fetch
+from paper_atlas.collectors.pmlr_tracks import collect_track_evidence, record_track


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


@@ -0,0 +1,140 @@
+"""Resolve mixed ICML proceedings tracks from the official paper program."""
+from collections import Counter
+import hashlib
+import json
+import re
+from urllib.parse import parse_qs, urlsplit
+
+from paper_atlas.collectors.iclr import identifiers, source_number
+from paper_atlas.db.store import author_keys, normalized
+from paper_atlas.io.html import Document, clean
+from paper_atlas.io.http import fetch
+
+
+POSITION_LIST_POLICY = 'https://icml.cc/Conferences/2025/CallForPositionPapers'
+PUBLISHER_LINK_ALIASES = {'v235/theo24a': 'v235/guyard24a'}
+
+
+def position_list_policy(data):
+    """The organizers explicitly link this title filter as the 2024 position list."""
+    for node in Document(data.decode()).root.walk():
+        if node.tag != 'a' or clean(node.text()).casefold() != 'position papers':
+            continue
+        url = node.attrs.get('href', '')
+        parsed = urlsplit(url)
+        query = parse_qs(parsed.query)
+        if (parsed.netloc in ('', 'icml.cc') and parsed.path == '/virtual/2024/papers.html'
+                and query.get('filter') == ['titles'] and query.get('search') == ['position:']):
+            return {'url': POSITION_LIST_POLICY, 'sha256': hashlib.sha256(data).hexdigest(),
+                    'position_list_url': 'https://icml.cc' + parsed.path + '?' + parsed.query}
+    raise ValueError('Official policy no longer identifies the 2024 position-paper title list')
+
+
+def program_tracks(data, year, policy=None):
+    program = json.loads(data)
+    events = program.get('results') if isinstance(program, dict) else None
+    if (not isinstance(events, list) or program.get('count') != len(events)
+            or program.get('next') or program.get('previous')):
+        raise ValueError('Official ICML program must be a complete counted event list')
+    if len({str(row['id']) for row in events}) != len(events):
+        raise ValueError('Repeated event identity in official ICML program')
+    if year == 2024 and policy is None:
+        raise ValueError('The 2024 position-list policy is required')
+    mapping = {}
+    excluded = Counter()
+    for row in events:
+        number = source_number(row)
+        if number is not None and number < 0:
+            excluded['synthetic_oral_event'] += 1
+            continue
+        parsed = urlsplit(row.get('sourceurl') or '')
+        group = parse_qs(parsed.query).get('id', [])
+        groups = {f'ICML.cc/{year}/Conference': 'main',
+                  f'ICML.cc/{year}/Position_Paper_Track': 'position'}
+        if (parsed.netloc != 'openreview.net' or parsed.path != '/group'
+                or len(group) != 1 or group[0] not in groups):
+            excluded['other_source'] += 1
+            continue
+        if not str(row.get('decision') or '').casefold().startswith('accept'):
+            raise ValueError('Non-accepted event in official ICML paper program')
+        if year == 2024:
+            # The 2024 program flattens both tracks into Conference and links
+            # PMLR publication IDs instead of forums. Use only this year-specific,
+            # organizer-endorsed list rule, never a global title heuristic.
+            link = urlsplit(row.get('paper_pdf_url') or '')
+            if link.netloc != 'proceedings.mlr.press' or not re.fullmatch(r'/v\d+/[^/]+\.html', link.path):
+                raise ValueError('2024 ICML paper has no publisher self-link')
+            key = ('pmlr', link.path.lstrip('/').removesuffix('.html'))
+            track = 'position' if re.match(r'^Position(?: Paper)?:', row.get('name') or '', re.I) else 'main'
+            kind = 'official_2024_position_list'
+        else:
+            forum = identifiers(row).get('openreview')
+            if not forum:
+                raise ValueError('ICML paper has no real OpenReview self-identity')
+            key = ('openreview', forum)
+            track, kind = groups[group[0]], 'official_openreview_track_group'
+        evidence = {'track': track, 'kind': kind, 'event_ids': [row['id']],
+                    'source_group': group[0], 'program_title': row.get('name')}
+        if key[0] == 'pmlr' and key[1] in PUBLISHER_LINK_ALIASES:
+            evidence['publisher_link_alias'] = key[1]
+            evidence['_authors'] = [author['fullname'] for author in row.get('authors', [])]
+            key = ('pmlr', PUBLISHER_LINK_ALIASES[key[1]])
+        if key in mapping:
+            if mapping[key]['track'] != track:
+                raise ValueError('One paper identity has conflicting official tracks')
+            mapping[key]['event_ids'].append(row['id'])
+        else:
+            mapping[key] = evidence
+    if not mapping:
+        raise ValueError('Official ICML program contains no supported paper identities')
+    return mapping, {'event_count': len(events), 'paper_identity_count': len(mapping),
+                     'excluded_events': dict(excluded)}
+
+
+def collect_track_evidence(year, cache):
+    url = f'https://icml.cc/static/virtual/data/icml-{year}-orals-posters.json'
+    data = fetch(url, cache)
+    policy = position_list_policy(fetch(POSITION_LIST_POLICY, cache)) if year == 2024 else None
+    mapping, inventory = program_tracks(data, year, policy)
+    provenance = {'url': url, 'sha256': hashlib.sha256(data).hexdigest(), **inventory}
+    if policy:
+        provenance['position_list_policy'] = policy
+    return mapping, provenance
+
+
+def record_track(record, mapping, provenance):
+    key = ('pmlr', record['external_id']) if record['year'] == 2024 else ('openreview', record['identifiers'].get('openreview'))
+    if key not in mapping:
+        raise ValueError('Published ICML paper is absent from the verified official track inventory: ' + record['external_id'])
+    evidence = {**mapping[key], 'identity_scheme': key[0], 'identity_value': key[1],
+                'program_url': provenance['url'], 'program_sha256': provenance['sha256']}
+    if 'publisher_link_alias' in evidence:
+        overlap = author_keys(record.get('authors', [])) & author_keys(evidence.pop('_authors', []))
+        if normalized(record['title']) != normalized(evidence['program_title']) or len(overlap) < 2:
+            raise ValueError('Reviewed publisher-link alias lacks exact title and independent author corroboration')
+        evidence['corroborating_author_count'] = len(overlap)
+    if 'position_list_policy' in provenance:
+        evidence['position_list_policy'] = provenance['position_list_policy']
+    record['track'] = evidence['track']
+    record['metadata']['track_evidence'] = evidence
+
+
+def require_track_manifest(manifest):
+    if (manifest.get('source') == 'pmlr' and str(manifest.get('venue', '')).upper() == 'ICML'
+            and int(manifest.get('year', 0)) >= 2024
+            and (manifest.get('parser_version', 0) < 3 or not manifest.get('track_program'))):
+        raise ValueError('Recollect mixed ICML PMLR proceedings with parser_version>=3 and official track evidence')
+
+
+def require_track_record(row, manifest):
+    require_track_manifest({**manifest, **{key: row.get(key) for key in ('source', 'venue', 'year')}})
+    if manifest.get('track_program'):
+        evidence = row.get('metadata', {}).get('track_evidence', {})
+        program = manifest['track_program']
+        if (evidence.get('track') != row.get('track') or evidence.get('program_url') != program.get('url')
+                or evidence.get('program_sha256') != program.get('sha256')
+                or not evidence.get('event_ids') or evidence.get('identity_scheme') not in ('pmlr', 'openreview')):
+            raise ValueError('PMLR record differs from its official track evidence')
+        own = row['external_id'] if evidence['identity_scheme'] == 'pmlr' else row.get('identifiers', {}).get('openreview')
+        if not own or evidence.get('identity_value') != own:
+            raise ValueError('PMLR track evidence belongs to a different paper identity')
@@ -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,50 @@
+"""Prevent legacy writers from diverging after an explicit PostgreSQL cutover."""
+from pathlib import Path
+
+
+def database_arguments(parser):
+    target = parser.add_mutually_exclusive_group(required=True)
+    target.add_argument('--db', help='Writable SQLite sample; never the frozen production backup')
+    target.add_argument('--pg-service', help='Protected libpq service for the published PostgreSQL dataset')
+    parser.add_argument('--pg-schema', help='Optional schema; otherwise use the configured search_path')
+
+
+def open_store(db=None, *, pg_service=None, pg_schema=None):
+    if bool(db) == bool(pg_service) or (pg_schema and not pg_service):
+        raise ValueError('Choose exactly one of SQLite db or PostgreSQL service; schema requires PostgreSQL')
+    if pg_service:
+        from paper_atlas.postgres.writer import PostgresStore
+        return PostgresStore(pg_service, pg_schema)
+    from paper_atlas.db.store import Store
+    return Store(db)
+
+
+def open_read_connection(db=None, *, pg_service=None, pg_schema=None):
+    if bool(db) == bool(pg_service) or (pg_schema and not pg_service):
+        raise ValueError('Choose exactly one of SQLite db or PostgreSQL service; schema requires PostgreSQL')
+    if pg_service:
+        from paper_atlas.postgres.client import ReadConnection
+        return ReadConnection(pg_service, pg_schema)
+    import sqlite3
+    path = Path(db).resolve()
+    if not path.is_file():
+        raise ValueError('Existing database required for read-only audit')
+    connection = sqlite3.connect(path.as_uri() + '?mode=ro', uri=True)
+    try:
+        connection.execute('BEGIN')
+    except BaseException:
+        connection.close()
+        raise
+    return connection
+
+
+def inserted_id(connection, sql, params=()):
+    if getattr(connection, 'postgres', False):
+        return connection.execute(sql + ' RETURNING id', params).fetchone()[0]
+    return connection.execute(sql, params).lastrowid
+
+
+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))
@@ -5,7 +5,8 @@ import re
 import sqlite3
 import unicodedata
 from pathlib import Path
-from paper_atlas.db.submissions import attach_submission,migrate_sources,write_decision
+from paper_atlas.db.submissions import attach_submission,migrate_sources,write_decision,latest_status
+from paper_atlas.db.backend import require_sqlite_writable, inserted_id


 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')
@@ -36,7 +38,7 @@ class Store:
        migrate_sources(self.conn)

    def source(self,name,url):
-        self.conn.execute('INSERT OR IGNORE INTO sources(name,homepage) VALUES(?,?)',(name,url))
+        self.conn.execute('INSERT INTO sources(name,homepage) VALUES(?,?) ON CONFLICT DO NOTHING',(name,url))
        return self.conn.execute('SELECT id FROM sources WHERE name=?',(name,)).fetchone()[0]

    def merge(self,target,other,reason):
@@ -53,7 +55,7 @@ class Store:
        c.execute('UPDATE identifiers SET paper_id=? WHERE paper_id=?',(target,other))
        c.execute('UPDATE source_records SET paper_id=? WHERE paper_id=?',(target,other))
        c.execute('UPDATE submissions SET paper_id=? WHERE paper_id=?',(target,other))
-        c.execute('INSERT OR IGNORE INTO paper_tags SELECT ?,tag_id,source_record_id,evidence_key FROM paper_tags WHERE paper_id=?',(target,other))
+        c.execute('INSERT INTO paper_tags SELECT ?,tag_id,source_record_id,evidence_key FROM paper_tags WHERE paper_id=? ON CONFLICT DO NOTHING',(target,other))
        c.execute('DELETE FROM paper_tags WHERE paper_id=?',(other,))
        c.execute('UPDATE paper_merges SET surviving_paper_id=? WHERE surviving_paper_id=?',(target,other))
        # Reattach pending candidate evidence without losing colliding rows.
@@ -63,7 +65,7 @@ class Store:
            left = target if left == other else left
            right = target if right == other else right
            if left != right:
-                c.execute('INSERT OR IGNORE INTO match_candidates VALUES(?,?,?,?,?,?)',(min(left,right),max(left,right),method,score,evidence,status))
+                c.execute('INSERT INTO match_candidates VALUES(?,?,?,?,?,?) ON CONFLICT DO NOTHING',(min(left,right),max(left,right),method,score,evidence,status))
        c.execute('DELETE FROM papers WHERE id=?',(other,))

    def record(self,source,external_id,url,title,abstract,authors,identifiers=None,metadata=None,published=None):
@@ -109,18 +111,19 @@ class Store:
                c.execute("UPDATE papers SET title=?,abstract=COALESCE(NULLIF(?,''),abstract),title_key=?,authors_json=CASE WHEN ?='[]' THEN authors_json ELSE ? END WHERE id=? AND title=? AND COALESCE(abstract,'')=COALESCE(?,'')",(title,abstract,key,json.dumps(authors,ensure_ascii=False),json.dumps(authors,ensure_ascii=False),paper,old[1],old_abstract))
            c.execute("UPDATE papers SET abstract=CASE WHEN abstract IS NULL OR abstract='' THEN ? ELSE abstract END WHERE id=?",(abstract,paper))
        else:
-            paper = c.execute('INSERT INTO papers(title,abstract,title_key,authors_json,first_published) VALUES(?,?,?,?,?)',(title,abstract,key,json.dumps(authors,ensure_ascii=False),published)).lastrowid
+            paper = inserted_id(c,'INSERT INTO papers(title,abstract,title_key,authors_json,first_published) VALUES(?,?,?,?,?)',(title,abstract,key,json.dumps(authors,ensure_ascii=False),published))
        for scheme,value in ids:
-            c.execute('INSERT OR IGNORE INTO identifiers VALUES(?,?,?)',(scheme,value,paper))
+            c.execute('INSERT INTO identifiers VALUES(?,?,?) ON CONFLICT DO NOTHING',(scheme,value,paper))
        metadata=dict(metadata or {});metadata['record_identifiers']=dict(ids)
        if old and not abstract:
            previous={'abstract':old[2],'observed_at':old[3]} if old[2] else json.loads(old[4]).get('last_known_abstract')
            if previous:metadata['last_known_abstract']=previous
-        c.execute('INSERT INTO source_records(source_id,external_id,paper_id,url,title,abstract,authors_json,metadata_json) VALUES(?,?,?,?,?,?,?,?) ON CONFLICT(source_id,external_id) DO UPDATE SET paper_id=excluded.paper_id,url=excluded.url,title=excluded.title,abstract=excluded.abstract,authors_json=excluded.authors_json,metadata_json=excluded.metadata_json,retrieved_at=CURRENT_TIMESTAMP',(source_id,external_id,paper,url,title,abstract,json.dumps(authors,ensure_ascii=False),json.dumps(metadata,ensure_ascii=False,default=str)))
+        from paper_atlas.db.submissions import timestamp
+        c.execute('INSERT INTO source_records(source_id,external_id,paper_id,url,title,abstract,authors_json,metadata_json,retrieved_at) VALUES(?,?,?,?,?,?,?,?,?) ON CONFLICT(source_id,external_id) DO UPDATE SET paper_id=excluded.paper_id,url=excluded.url,title=excluded.title,abstract=excluded.abstract,authors_json=excluded.authors_json,metadata_json=excluded.metadata_json,retrieved_at=excluded.retrieved_at',(source_id,external_id,paper,url,title,abstract,json.dumps(authors,ensure_ascii=False),json.dumps(metadata,ensure_ascii=False,default=str),timestamp()))
        record = c.execute('SELECT id FROM source_records WHERE source_id=? AND external_id=?',(source_id,external_id)).fetchone()[0]
        for other,_,_ in title_matches:
            if other != paper and c.execute('SELECT 1 FROM papers WHERE id=?',(other,)).fetchone():
-                c.execute('INSERT OR IGNORE INTO match_candidates VALUES(?,?,?,?,?,?)',(min(paper,other),max(paper,other),'exact_title_unconfirmed',None,json.dumps({'title':title}), 'pending'))
+                c.execute('INSERT INTO match_candidates VALUES(?,?,?,?,?,?) ON CONFLICT DO NOTHING',(min(paper,other),max(paper,other),'exact_title_unconfirmed',None,json.dumps({'title':title}), 'pending'))
        return paper,record

    def record_tags(self,paper,record,source,tags):
@@ -135,22 +138,26 @@ class Store:
        c=self.conn
        c.execute('DELETE FROM paper_tags WHERE source_record_id=? AND evidence_key=?',(record,evidence))
        for namespace,value in sorted(values):
-            c.execute('INSERT OR IGNORE INTO tags(namespace,value) VALUES(?,?)',(namespace,value))
+            c.execute('INSERT INTO tags(namespace,value) VALUES(?,?) ON CONFLICT DO NOTHING',(namespace,value))
            tag=c.execute('SELECT id FROM tags WHERE namespace=? AND value=?',(namespace,value)).fetchone()[0]
-            c.execute('INSERT OR IGNORE INTO paper_tags VALUES(?,?,?,?)',(paper,tag,record,evidence))
+            c.execute('INSERT INTO paper_tags VALUES(?,?,?,?) ON CONFLICT DO NOTHING',(paper,tag,record,evidence))

-    def decision(self,paper,record,venue,year,track,external_id,url,status,raw_value,aliases=None,identities=None,observed_at=None,decision_date=None,source_decision_id='',submitted_at=None):
+    def decision(self,paper,record,venue,year,track,external_id,url,status,raw_value,aliases=None,identities=None,observed_at=None,decision_date=None,source_decision_id='',submitted_at=None,guard_status=False):
        c = self.conn
-        existing=c.execute('SELECT id FROM venues WHERE acronym=? COLLATE NOCASE',(venue,)).fetchone()
+        existing=c.execute('SELECT id FROM venues WHERE lower(acronym)=lower(?)',(venue,)).fetchone()
        if existing:venue_id=existing[0]
-        else:venue_id=c.execute("INSERT INTO venues(acronym,name,field,venue_type) VALUES(?,?,?,'conference')",(venue,venue,'unclassified')).lastrowid
+        else:venue_id=inserted_id(c,"INSERT INTO venues(acronym,name,field,venue_type) VALUES(?,?,?,'conference')",(venue,venue,'unclassified'))
        for alias in aliases or []:
-            c.execute('INSERT OR IGNORE INTO venue_aliases VALUES(?,?)',(venue_id,alias))
-        c.execute('INSERT OR IGNORE INTO editions(venue_id,year,track) VALUES(?,?,?)',(venue_id,int(year),track))
+            c.execute('INSERT INTO venue_aliases VALUES(?,?) ON CONFLICT DO NOTHING',(venue_id,alias))
+        c.execute('INSERT INTO editions(venue_id,year,track) VALUES(?,?,?) ON CONFLICT DO NOTHING',(venue_id,int(year),track))
        edition = c.execute('SELECT id FROM editions WHERE venue_id=? AND year=? AND track=?',(venue_id,int(year),track)).fetchone()[0]
        # A previous event in this row may already have merged the input paper.
        paper=c.execute('SELECT paper_id FROM source_records WHERE id=?',(record,)).fetchone()[0]
        submission=attach_submission(self,edition,paper,record,external_id,identities or [])
+        if guard_status and status in ('rejected','desk_rejected','withdrawn'):
+            current = latest_status(c, submission)
+            if current and current not in ('unknown','under_review',status):
+                raise ValueError('Curated decision conflicts with current canonical submission status: ' + current)
        if submitted_at:
            from paper_atlas.db.submissions import timestamp
            c.execute('UPDATE submissions SET submitted_at=COALESCE(submitted_at,?) WHERE id=?',(timestamp(submitted_at),submission))
@@ -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))
@@ -39,21 +58,29 @@ def identities_for_row(row):
 def migrate_sources(c):
    if c.execute("SELECT 1 FROM schema_migrations WHERE name='submission_source_baseline_v1'").fetchone():return
    with c:
-        c.execute('INSERT OR IGNORE INTO submission_sources SELECT id,source_record_id,source_submission_id FROM submissions')
-        c.execute("INSERT OR IGNORE INTO decision_sources SELECT d.id,s.source_record_id,'' FROM decisions d JOIN submissions s ON s.id=d.submission_id")
+        c.execute('INSERT INTO submission_sources SELECT id,source_record_id,source_submission_id FROM submissions WHERE 1=1 ON CONFLICT DO NOTHING')
+        c.execute("INSERT INTO decision_sources SELECT d.id,s.source_record_id,'' FROM decisions d JOIN submissions s ON s.id=d.submission_id WHERE 1=1 ON CONFLICT DO NOTHING")
        c.execute("INSERT INTO schema_migrations(name) VALUES('submission_source_baseline_v1')")


 def next_id(c,table,audit,former):
    # Merged IDs are kept in audit; never allocate them to a later unrelated event.
+    if getattr(c, 'postgres', False):
+        return c.next_id(table)
    return max(c.execute('SELECT COALESCE(max(id),0) FROM '+table).fetchone()[0],c.execute('SELECT COALESCE(max('+former+'),0) FROM '+audit).fetchone()[0])+1


+def latest_status(c, submission):
+    row = c.execute('''SELECT status FROM decisions WHERE submission_id=?
+        ORDER BY observed_at DESC,COALESCE(decision_date,observed_at) DESC,id DESC LIMIT 1''', (submission,)).fetchone()
+    return row[0] if row else None
+
+
 def merge_submissions(c,target,other):
    if target==other:return
    editions=c.execute('SELECT id,edition_id FROM submissions WHERE id IN (?,?)',(target,other)).fetchall()
    if len(editions)!=2 or editions[0][1]!=editions[1][1]:raise ValueError('Cannot combine submissions from different editions')
-    c.execute('INSERT OR IGNORE INTO submission_sources SELECT ?,source_record_id,source_submission_id FROM submission_sources WHERE submission_id=?',(target,other))
+    c.execute('INSERT INTO submission_sources SELECT ?,source_record_id,source_submission_id FROM submission_sources WHERE submission_id=? ON CONFLICT DO NOTHING',(target,other))
    c.execute('DELETE FROM submission_sources WHERE submission_id=?',(other,))
    for decision,status,raw,url,observed,date in c.execute('SELECT id,status,raw_value,evidence_url,observed_at,decision_date FROM decisions WHERE submission_id=?',(other,)).fetchall():
        duplicate=c.execute('SELECT id,decision_date FROM decisions WHERE submission_id=? AND status=? AND raw_value=? AND evidence_url=? AND observed_at=?',(target,status,raw,url,observed)).fetchone()
@@ -61,7 +88,7 @@ def merge_submissions(c,target,other):
            surviving=duplicate[0]
            if date and duplicate[1] and date!=duplicate[1]:raise ValueError('Conflicting dates for the same decision observation')
            if date and not duplicate[1]:c.execute('UPDATE decisions SET decision_date=? WHERE id=?',(date,surviving))
-            c.execute('INSERT OR IGNORE INTO decision_sources SELECT ?,source_record_id,source_decision_id FROM decision_sources WHERE decision_id=?',(surviving,decision))
+            c.execute('INSERT INTO decision_sources SELECT ?,source_record_id,source_decision_id FROM decision_sources WHERE decision_id=? ON CONFLICT DO NOTHING',(surviving,decision))
            c.execute('DELETE FROM decision_sources WHERE decision_id=?',(decision,))
            c.execute('UPDATE decision_merges SET surviving_decision_id=? WHERE surviving_decision_id=?',(surviving,decision))
            c.execute("INSERT INTO decision_merges(former_decision_id,surviving_decision_id,reason) VALUES(?,?,'same decision observation on merged submission')",(decision,surviving))
@@ -95,9 +122,9 @@ def attach_submission(store,edition,paper,record,external_id,identities):
    else:
        submission=next_id(c,'submissions','submission_merges','former_submission_id')
        c.execute('INSERT INTO submissions(id,paper_id,edition_id,source_record_id,source_submission_id) VALUES(?,?,?,?,?)',(submission,paper,edition,record,external_id))
-    c.execute('INSERT OR IGNORE INTO submission_sources VALUES(?,?,?)',(submission,record,external_id))
+    c.execute('INSERT INTO submission_sources VALUES(?,?,?) ON CONFLICT DO NOTHING',(submission,record,external_id))
    for scheme,value in identities:
-        c.execute('INSERT OR IGNORE INTO submission_identifiers VALUES(?,?,?,?)',(edition,scheme,value,submission))
+        c.execute('INSERT INTO submission_identifiers VALUES(?,?,?,?) ON CONFLICT DO NOTHING',(edition,scheme,value,submission))
    return submission


@@ -111,5 +138,5 @@ def write_decision(store,submission,record,status,raw,url,observed_at=None,decis
    else:
        decision=next_id(c,'decisions','decision_merges','former_decision_id')
        c.execute('INSERT INTO decisions(id,submission_id,status,raw_value,evidence_url,observed_at,decision_date) VALUES(?,?,?,?,?,?,?)',(decision,submission,status,raw,url,observed,date))
-    c.execute('INSERT OR IGNORE INTO decision_sources VALUES(?,?,?)',(decision,record,source_decision_id or ''))
+    c.execute('INSERT INTO decision_sources VALUES(?,?,?) ON CONFLICT DO NOTHING',(decision,record,source_decision_id or ''))
    return decision
@@ -1,4 +1,4 @@
-"""Stream the official metadata ZIP into SQLite. No PDFs, source files or builds."""
+"""Stream the official metadata ZIP into the configured metadata database."""
 import argparse
 import json
 import time
@@ -7,7 +7,8 @@ import zipfile
 from pathlib import Path


-from paper_atlas.db.store import Store
+from paper_atlas.db.backend import database_arguments, open_store, inserted_id
+from paper_atlas.db.submissions import timestamp


 def refresh_categories(store,paper,record,categories):
@@ -17,28 +18,29 @@ def refresh_categories(store,paper,record,categories):
    store.conn.execute("DELETE FROM paper_tags WHERE source_record_id=? AND evidence_key='arxiv:categories' AND tag_id IN (SELECT id FROM tags WHERE namespace='arxiv:category')",(record,))


-def ingest(db, archive):
-    store = Store(db)
+def ingest(db, archive, *, pg_service=None, pg_schema=None, storage_path=None):
+    store = open_store(db,pg_service=pg_service,pg_schema=pg_schema)
    conn = store.conn
-    conn.execute("INSERT OR IGNORE INTO sources(name,homepage) VALUES('arxiv','https://arxiv.org')")
-    source = conn.execute("SELECT id FROM sources WHERE name='arxiv'").fetchone()[0]
-    previous = conn.execute('SELECT id,checkpoint_json FROM ingest_runs WHERE source_id=? AND status IN (\'running\',\'failed\') ORDER BY id DESC LIMIT 1', (source,)).fetchone()
-    fingerprint = {'path':str(Path(archive).resolve()), 'bytes':Path(archive).stat().st_size, 'mtime_ns':Path(archive).stat().st_mtime_ns}
-    start = 0
-    if previous:
-        checkpoint = json.loads(previous[1] or '{}')
-        if checkpoint.get('archive') == fingerprint:
-            run = previous[0]
-            start = checkpoint.get('line', 0)
-            conn.execute("UPDATE ingest_runs SET status='running',error=NULL WHERE id=?", (run,))
-        else:
-            raise RuntimeError('Pending run belongs to a different archive; resolve it explicitly')
-    else:
-        run = conn.execute('INSERT INTO ingest_runs(source_id,checkpoint_json) VALUES(?,?)', (source,json.dumps({'archive':fingerprint,'line':0}))).lastrowid
-    conn.commit()
-    line_number = start
-    last = time.monotonic()
+    run = None
    try:
+        source = store.source('arxiv','https://arxiv.org')
+        previous = conn.execute('SELECT id,checkpoint_json FROM ingest_runs WHERE source_id=? AND status IN (\'running\',\'failed\') ORDER BY id DESC LIMIT 1', (source,)).fetchone()
+        fingerprint = {'path':str(Path(archive).resolve()), 'bytes':Path(archive).stat().st_size, 'mtime_ns':Path(archive).stat().st_mtime_ns}
+        start = 0
+        if previous:
+            checkpoint = json.loads(previous[1] or '{}')
+            if checkpoint.get('archive') == fingerprint:
+                run = previous[0]
+                start = checkpoint.get('line', 0)
+                conn.execute("UPDATE ingest_runs SET status='running',error=NULL WHERE id=?", (run,))
+            else:
+                raise RuntimeError('Pending run belongs to a different archive; resolve it explicitly')
+        else:
+            run = inserted_id(conn,'INSERT INTO ingest_runs(source_id,checkpoint_json) VALUES(?,?)', (source,json.dumps({'archive':fingerprint,'line':0})))
+        conn.commit()
+        line_number = start
+        last = time.monotonic()
+        monitored = Path(storage_path) if storage_path else (Path(db).parent if db else Path(archive).parent)
        with zipfile.ZipFile(archive) as zipped:
            # Reading the member to EOF also checks its CRC.
            with zipped.open('arxiv-metadata-oai-snapshot.json') as stream:
@@ -54,20 +56,21 @@ def ingest(db, archive):
                    paper,record = store.record('arxiv',external_id,'https://arxiv.org/abs/'+external_id,title,abstract,authors,{'arxiv':external_id,'doi':row.get('doi')},metadata,(row.get('versions') or [{}])[0].get('created'))
                    refresh_categories(store,paper,record,row.get('categories') or '')
                    if line_number % 500 == 0:
-                        if shutil.disk_usage(Path(db).parent).free < 4*1024**3:
+                        if shutil.disk_usage(monitored).free < 4*1024**3:
                            raise RuntimeError('Stopped before exhausting disk: less than 4 GiB free')
                        conn.execute('UPDATE ingest_runs SET records_seen=?,records_written=?,checkpoint_json=? WHERE id=?',(line_number,line_number,json.dumps({'archive':fingerprint,'line':line_number}),run))
                        conn.commit()
                        if time.monotonic()-last >= 20:
                            print(json.dumps({'run':run,'rows':line_number}),flush=True)
                            last = time.monotonic()
-        conn.execute("UPDATE ingest_runs SET status='complete',finished_at=CURRENT_TIMESTAMP,records_seen=?,records_written=?,checkpoint_json=? WHERE id=?",(line_number,line_number,json.dumps({'archive':fingerprint,'line':line_number}),run))
+        conn.execute("UPDATE ingest_runs SET status='complete',finished_at=?,records_seen=?,records_written=?,checkpoint_json=? WHERE id=?",(timestamp(),line_number,line_number,json.dumps({'archive':fingerprint,'line':line_number}),run))
        conn.commit()
        print(json.dumps({'status':'complete','rows':line_number}),flush=True)
    except BaseException as exc:
        conn.rollback()
-        conn.execute("UPDATE ingest_runs SET status='failed',error=? WHERE id=?",(str(exc),run))
-        conn.commit()
+        if run is not None:
+            conn.execute("UPDATE ingest_runs SET status='failed',error=? WHERE id=?",(str(exc),run))
+            conn.commit()
        raise
    finally:
        conn.close()
@@ -75,10 +78,11 @@ def ingest(db, archive):

 def main(argv=None, *, prog=None):
    parser = argparse.ArgumentParser(prog=prog)
-    parser.add_argument('--db', required=True)
+    database_arguments(parser)
    parser.add_argument('--archive', required=True)
+    parser.add_argument('--storage-path', help='Local filesystem to monitor for free space; specify the database volume when running beside PostgreSQL')
    args = parser.parse_args(argv)
-    ingest(args.db,args.archive)
+    ingest(args.db,args.archive,pg_service=args.pg_service,pg_schema=args.pg_schema,storage_path=args.storage_path)


 if __name__ == '__main__':
@@ -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)
@@ -4,8 +4,9 @@ import hashlib
 import json
 from pathlib import Path

-from paper_atlas.db.store import Store
+from paper_atlas.db.backend import database_arguments, open_store
 from paper_atlas.db.submissions import attach_submission,identities_for_row
+from paper_atlas.collectors.pmlr_tracks import require_track_manifest, require_track_record


 def queue_files(path):
@@ -18,14 +19,15 @@ def queue_files(path):
    return files


-def backfill(db,files):
+def backfill(db,files, *, pg_service=None, pg_schema=None):
    for filename in files:
        manifest=json.loads(Path(filename).with_suffix('.manifest.json').read_text())
+        require_track_manifest(manifest)
        if manifest['source'] in ('colm','neurips','iclr_program') and manifest.get('parser_version',0)<2:
            raise ValueError('Recollect old parser output before submission evidence backfill')
-    store=Store(db);c=store.conn;count=0
-    before=c.execute('SELECT count(*) FROM submissions').fetchone()[0]
+    store=open_store(db,pg_service=pg_service,pg_schema=pg_schema);c=store.conn;count=0
    try:
+        before=c.execute('SELECT count(*) FROM submissions').fetchone()[0]
        for filename in files:
            path=Path(filename);manifest=json.loads(path.with_suffix('.manifest.json').read_text());seen=set()
            with path.open() as stream:
@@ -35,8 +37,9 @@ def backfill(db,files):
                    seen.add(row['external_id'])
                    if any(row.get(key)!=manifest.get(key) for key in ('source','venue','year','track')):
                        raise ValueError('Evidence row differs from edition manifest')
+                    require_track_record(row, manifest)
                    record=c.execute('SELECT r.id,r.paper_id,r.metadata_json FROM source_records r JOIN sources s ON s.id=r.source_id WHERE s.name=? AND r.external_id=?',(row['source'],row['external_id'])).fetchone()
-                    edition=c.execute('SELECT e.id FROM editions e JOIN venues v ON v.id=e.venue_id WHERE v.acronym=? COLLATE NOCASE AND e.year=? AND e.track=?',(row['venue'],row['year'],row['track'])).fetchone()
+                    edition=c.execute('SELECT e.id FROM editions e JOIN venues v ON v.id=e.venue_id WHERE lower(v.acronym)=lower(?) AND e.year=? AND e.track=?',(row['venue'],row['year'],row['track'])).fetchone()
                    if not record or not edition:raise ValueError('Evidence backfill requires an already imported record and edition')
                    if not c.execute('SELECT 1 FROM submission_sources ss JOIN submissions s ON s.id=ss.submission_id WHERE ss.source_record_id=? AND ss.source_submission_id=? AND s.edition_id=?',(record[0],row['external_id'],edition[0])).fetchone():
                        raise ValueError('No existing submission to backfill')
@@ -59,8 +62,8 @@ def backfill(db,files):


 def main(argv=None, *, prog=None):
-    parser=argparse.ArgumentParser(prog=prog);parser.add_argument('--db',required=True);parser.add_argument('--queue',required=True)
-    args=parser.parse_args(argv);backfill(args.db,queue_files(args.queue))
+    parser=argparse.ArgumentParser(prog=prog);database_arguments(parser);parser.add_argument('--queue',required=True)
+    args=parser.parse_args(argv);backfill(args.db,queue_files(args.queue),pg_service=args.pg_service,pg_schema=args.pg_schema)


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

-from paper_atlas.db.store import Store
+from paper_atlas.db.backend import database_arguments, open_store, inserted_id
 from paper_atlas.db.submissions import identities_for_row,timestamp
+from paper_atlas.collectors.pmlr_tracks import require_track_manifest, require_track_record
+from paper_atlas.collectors.public_openreview import SOURCE as PUBLIC_SOURCE
+from paper_atlas.collectors.icml_public_archive import SOURCE as ARCHIVE_SOURCE


-def ingest_file(db,path):
+def ingest_file(db,path, *, pg_service=None, pg_schema=None):
    path = Path(path)
    manifest_path = path.with_suffix('.manifest.json')
    manifest = json.loads(manifest_path.read_text()) if manifest_path.exists() else {}
+    require_track_manifest(manifest)
    if manifest.get('source') in ('colm','neurips') and manifest.get('parser_version',0)<2:
        raise ValueError('Recollect proceedings with parser_version>=2 before identity matching')
    if manifest.get('source')=='iclr_program' and manifest.get('parser_version',0)<2:
        raise ValueError('Recollect ICLR with parser_version>=2 before submission identity matching')
-    store = Store(db)
-    c = store.conn
    source_name = manifest.get('source')
    if not source_name:
        raise ValueError('A source manifest is required')
-    source_id = store.source(source_name,manifest['list_url'])
-    run = c.execute('INSERT INTO ingest_runs(source_id,checkpoint_json) VALUES(?,?)',(source_id,json.dumps({'file':str(path)}))).lastrowid
-    c.commit()
+    store = open_store(db, pg_service=pg_service, pg_schema=pg_schema)
+    c = store.conn
+    run = None
    count = 0
    missing_abstracts = 0
    seen = set()
    editions = set()
    try:
-        with path.open() as stream:
+        source_id = store.source(source_name,manifest['list_url'])
+        run = inserted_id(c,'INSERT INTO ingest_runs(source_id,checkpoint_json) VALUES(?,?)',(source_id,json.dumps({'file':str(path),'expected_sha256':manifest.get('file_sha256')})))
+        c.commit()
+        digest = hashlib.sha256()
+        with path.open('rb') as stream:
            for line in stream:
+                digest.update(line)
                row = json.loads(line)
                if row['source'] != source_name:
                    raise ValueError('Source differs from manifest')
                if any(key in manifest and manifest[key]!=row[key] for key in ('venue','year','track')):
                    raise ValueError('Conference edition differs from manifest')
+                require_track_record(row, manifest)
                external = row['external_id']
                if external in seen:
                    raise ValueError('Duplicate external ID in one collected file: '+external)
@@ -56,7 +65,7 @@ def ingest_file(db,path):
                        status=event['status']
                        if events and status is None:raise ValueError('Explicit decision event requires a status')
                        if status is not None and status not in ('accepted','rejected','desk_rejected','withdrawn','under_review','unknown'):raise ValueError('Unsupported decision event')
-                        edition=store.decision(paper,record,row['venue'],row['year'],row['track'],external,event.get('evidence_url',row['url']),status,event.get('raw_value',''),row.get('venue_aliases'),identities=identities,observed_at=event.get('observed_at') or row_observed,decision_date=event.get('decision_date'),source_decision_id=event.get('source_decision_id',''),submitted_at=row.get('submitted_at'))
+                        edition=store.decision(paper,record,row['venue'],row['year'],row['track'],external,event.get('evidence_url',row['url']),status,event.get('raw_value',''),row.get('venue_aliases'),identities=identities,observed_at=event.get('observed_at') or row_observed,decision_date=event.get('decision_date'),source_decision_id=event.get('source_decision_id',''),submitted_at=row.get('submitted_at'),guard_status=source_name in (PUBLIC_SOURCE,ARCHIVE_SOURCE))
                    editions.add(edition)
                else:
                    raise ValueError('This importer requires an explicit supported decision status')
@@ -64,6 +73,8 @@ def ingest_file(db,path):
                missing_abstracts += not bool((row.get('abstract') or '').strip())
        if not count:
            raise ValueError('Empty metadata file')
+        if manifest.get('file_sha256') and digest.hexdigest() != manifest['file_sha256']:
+            raise ValueError('Collected file hash differs from manifest')
        if len(editions) != 1:
            raise ValueError('A manifest must describe exactly one conference edition and track')
        edition = next(iter(editions))
@@ -72,13 +83,14 @@ def ingest_file(db,path):
            raise ValueError(f'Manifest count {expected} differs from records {count}')
        complete=expected==count and manifest.get('status') not in ('partial','blocked') and not manifest.get('missing_abstract_count',0) and not missing_abstracts
        c.execute('INSERT INTO edition_sources(edition_id,source_id,list_url,status,expected_count,imported_count,last_success_at) VALUES(?,?,?,?,?,?,?) ON CONFLICT(edition_id,source_id,list_url) DO UPDATE SET status=excluded.status,expected_count=excluded.expected_count,imported_count=excluded.imported_count,last_success_at=excluded.last_success_at,last_error=NULL',(edition,source_id,manifest['list_url'],'complete' if complete else 'partial',expected,count,time.strftime('%Y-%m-%dT%H:%M:%SZ',time.gmtime())))
-        c.execute("UPDATE ingest_runs SET status='complete',finished_at=CURRENT_TIMESTAMP,records_seen=?,records_written=? WHERE id=?",(count,count,run))
+        c.execute("UPDATE ingest_runs SET status='complete',finished_at=?,records_seen=?,records_written=?,checkpoint_json=? WHERE id=?",(timestamp(),count,count,json.dumps({'file':str(path),'file_sha256':digest.hexdigest()}),run))
        c.commit()
        print(json.dumps({'file':str(path),'status':'imported','records':count,'missing_abstract_count':missing_abstracts}),flush=True)
    except BaseException as exc:
        c.rollback()
-        c.execute("UPDATE ingest_runs SET status='failed',finished_at=CURRENT_TIMESTAMP,error=? WHERE id=?",(str(exc),run))
-        c.commit()
+        if run is not None:
+            c.execute("UPDATE ingest_runs SET status='failed',finished_at=?,records_seen=?,records_written=0,error=? WHERE id=?",(timestamp(),count,str(exc),run))
+            c.commit()
        raise
    finally:
        c.close()
@@ -86,11 +98,11 @@ def ingest_file(db,path):

 def main(argv=None, *, prog=None):
    parser = argparse.ArgumentParser(prog=prog)
-    parser.add_argument('--db',required=True)
+    database_arguments(parser)
    parser.add_argument('files',nargs='+')
    args = parser.parse_args(argv)
    for filename in args.files:
-        ingest_file(args.db,filename)
+        ingest_file(args.db,filename,pg_service=args.pg_service,pg_schema=args.pg_schema)


 if __name__ == '__main__':
@@ -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,96 @@
+"""Libpq service connections; credentials never belong in CLI arguments."""
+from collections.abc import Mapping
+import re
+from paper_atlas.postgres import text
+
+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, *, encoded=False):
+    columns = [column.name for column in cursor.description] if cursor.description else []
+    return lambda values: Row(columns, tuple(map(text.decode, values)) if encoded else 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')
+        if not self.has_table('papers'):
+            self.close()
+            raise RuntimeError('PostgreSQL publication is not ready in the configured search_path')
+        current = self.connection.execute('SELECT current_schema()').fetchone()[0]
+        # Schema comments are readable by other projects granted table access;
+        # consumers need no privileges on the private migration controller.
+        annotation = self.connection.execute("SELECT obj_description(oid,'pg_namespace') FROM pg_namespace WHERE nspname=%s", (current,)).fetchone()[0]
+        self.encoded = False
+        if annotation and annotation.startswith('paper-atlas:text-codec='):
+            codec = annotation.removeprefix('paper-atlas:text-codec=')
+            if codec != text.CODEC:
+                self.close()
+                raise RuntimeError('Unsupported PostgreSQL text storage codec')
+            self.encoded = codec == text.CODEC
+        self.connection.row_factory = lambda cursor: row_factory(cursor, encoded=self.encoded)
+
+    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],))
+        values = tuple(map(text.encode, params)) if self.encoded else params
+        return self.connection.execute(query.replace('?', '%s'), values)
+
+    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,279 @@
+"""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
+from paper_atlas.postgres import text
+
+
+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 upgrade_text_codec(pg, control, stage, plan):
+    """Upgrade committed legacy staging rows once; never recopy their source."""
+    changed = {}
+    with pg.transaction():
+        copied = {r[0] for r in pg.execute('SELECT name FROM ' + qualify(control, 'tables'))}
+        if not copied <= plan.keys():
+            raise ValueError('Unknown committed staging tables')
+        # An older run may have reached constraint validation. Replacing SOH
+        # can temporarily equal another legacy key even though the final mapping
+        # is bijective. Rebuild unpublished constraints/indexes after the upgrade.
+        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 co.contype IN ('f','p','u')
+            ORDER BY CASE WHEN co.contype='f' THEN 0 ELSE 1 END''', (stage,)).fetchall()
+        for table, name in constraints:
+            pg.execute('ALTER TABLE ' + qualify(stage, table) + ' DROP CONSTRAINT ' + quote(name))
+        indexes = pg.execute('''SELECT cl.relname FROM pg_index i JOIN pg_class cl ON cl.oid=i.indexrelid
+            JOIN pg_namespace ns ON ns.oid=cl.relnamespace WHERE ns.nspname=%s AND i.indisunique''', (stage,)).fetchall()
+        for name, in indexes:
+            pg.execute('DROP INDEX ' + qualify(stage, name))
+        for table in plan:
+            if table not in copied:
+                continue
+            columns = [quote(c['name']) for c in plan[table]['columns'] if c['type'] == 'TEXT']
+            if columns:
+                assignments = ','.join(c + '=replace(' + c + ',%s,%s)' for c in columns)
+                predicate = ' OR '.join('strpos(' + c + ',%s)>0' for c in columns)
+                params = [v for c in columns for v in (text.ESCAPE, text.encode(text.ESCAPE))]
+                params += [text.ESCAPE] * len(columns)
+                changed[table] = pg.execute('UPDATE ' + qualify(stage, table) + ' SET ' + assignments +
+                                            ' WHERE ' + predicate, params).rowcount
+        pg.execute('UPDATE ' + qualify(control, 'run') + ' SET text_codec=%s WHERE id=1', (text.CODEC,))
+    return changed
+
+
+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)')
+            pg.execute('ALTER TABLE ' + qualify(control, 'run') + " ADD COLUMN IF NOT EXISTS text_codec TEXT NOT NULL DEFAULT 'legacy'")
+            run = pg.execute('SELECT source_json,stage,status,text_codec 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]
+                if run[3] == 'legacy':
+                    report['legacy_text_upgrade_rows'] = upgrade_text_codec(pg, control, stage, plan)
+                elif run[3] != text.CODEC:
+                    raise ValueError('Unsupported staging text storage codec')
+            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') +
+                               '(id,source_json,stage,status,text_codec) VALUES(1,%s,%s,%s,%s)',
+                               (json.dumps(identity, sort_keys=True), stage, 'copying', text.CODEC))
+            report.update(status='copying', stage=stage, text_codec=text.CODEC,
+                          content_hash_scope='Decoded original SQLite values; storage escapes are reversible.')
+            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(tuple(map(text.encode, 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(); storage_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(storage_digest, row)
+                            update_hash(digest, tuple(map(text.decode, row))); rows += 1
+                    entry = report['tables'][table]
+                    entry.update(target_rows=rows, target_sha256=digest.hexdigest(),
+                                 target_storage_sha256=storage_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('COMMENT ON SCHEMA ' + quote(stage) + " IS 'paper-atlas:text-codec=" + text.CODEC + "'")
+                    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']))
@@ -0,0 +1,31 @@
+"""Order-preserving, reversible storage for SQLite text containing NUL.
+
+PostgreSQL text forbids U+0000. Encode it as SOH/SOH, and encode a literal
+SOH as SOH/STX. All other characters stay unchanged. The mapping preserves
+binary lexicographic order, including string prefixes and composite keys.
+"""
+
+CODEC = 'nul-soh-v1'
+ESCAPE = '\x01'
+
+
+def encode(value):
+    if isinstance(value, str):
+        return value.replace(ESCAPE, ESCAPE + '\x02').replace('\x00', ESCAPE + ESCAPE)
+    return value
+
+
+def decode(value):
+    if not isinstance(value, str) or ESCAPE not in value:
+        return value
+    result = []
+    start = 0
+    while (position := value.find(ESCAPE, start)) >= 0:
+        result.append(value[start:position])
+        following = value[position + 1:position + 2]
+        if following not in (ESCAPE, '\x02'):
+            raise ValueError('Invalid nul-soh-v1 text escape')
+        result.append('\x00' if following == ESCAPE else ESCAPE)
+        start = position + 2
+    result.append(value[start:])
+    return ''.join(result)
@@ -0,0 +1,87 @@
+"""Native transactional writes reusing the shared identity/provenance policy."""
+import re
+import zlib
+
+from paper_atlas.db.store import Store
+from paper_atlas.postgres import text
+from paper_atlas.postgres.client import connect
+from paper_atlas.postgres.schema import TABLES, quote
+
+
+class Result:
+    def __init__(self, cursor):
+        self.cursor = cursor
+
+    def fetchone(self):
+        row = self.cursor.fetchone()
+        return tuple(map(text.decode, row)) if row is not None else None
+
+    def fetchall(self):
+        return [tuple(map(text.decode, row)) for row in self.cursor.fetchall()]
+
+    def __iter__(self):
+        return (tuple(map(text.decode, row)) for row in self.cursor)
+
+    @property
+    def rowcount(self):
+        return self.cursor.rowcount
+
+
+class WriteConnection:
+    postgres = True
+
+    def __init__(self, service, schema=None):
+        self.connection = connect(service)
+        try:
+            if schema is None:
+                schema = self.connection.execute('SELECT current_schema()').fetchone()[0]
+            if not isinstance(schema, str) or not re.fullmatch(r'[a-z][a-z0-9_]{0,35}', schema):
+                raise ValueError('Invalid PostgreSQL write schema')
+            annotation = self.connection.execute("SELECT obj_description(oid,'pg_namespace') FROM pg_namespace WHERE nspname=%s", (schema,)).fetchone()
+            if not annotation or annotation[0] != 'paper-atlas:text-codec=' + text.CODEC:
+                raise RuntimeError('Writes require a published dataset with the supported text codec')
+            tables = {r[0] for r in self.connection.execute("SELECT table_name FROM information_schema.tables WHERE table_schema=%s AND table_type='BASE TABLE'", (schema,))}
+            if not set(TABLES) <= tables:
+                raise RuntimeError('Published business tables are incomplete or inaccessible')
+            if not self.connection.execute('SELECT pg_try_advisory_lock(%s,%s)', (615174, zlib.crc32(schema.encode()) & 0x7fffffff)).fetchone()[0]:
+                raise RuntimeError('Another writer holds this PostgreSQL dataset; retry after it closes')
+            self.schema = schema
+            self.connection.execute('SET search_path TO ' + quote(schema) + ',pg_catalog')
+            if not self.connection.execute("SELECT 1 FROM schema_migrations WHERE name='submission_source_baseline_v1'").fetchone():
+                raise RuntimeError('Submission provenance baseline is missing')
+            self.connection.autocommit = False
+        except BaseException:
+            self.connection.close()
+            raise
+
+    def execute(self, sql, params=()):
+        # The shared domain uses static SQL and bound parameters. It uses no
+        # SQLite-specific syntax; this adapter only binds/decodes text values.
+        return Result(self.connection.execute(sql.replace('?', '%s'), tuple(map(text.encode, params))))
+
+    def next_id(self, table):
+        if table not in ('submissions', 'decisions'):
+            raise ValueError('Unsupported explicit identity sequence')
+        return self.connection.execute("SELECT nextval(pg_get_serial_sequence(%s,'id'))", (quote(self.schema) + '.' + quote(table),)).fetchone()[0]
+
+    def commit(self):
+        self.connection.commit()
+
+    def rollback(self):
+        self.connection.rollback()
+
+    def close(self):
+        # Ends the session and releases its advisory lock, including after an
+        # error/rollback. File/chunk commits deliberately retain the writer lock.
+        self.connection.close()
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, kind, error, traceback):
+        self.rollback() if kind else self.commit()
+
+
+class PostgresStore(Store):
+    def __init__(self, service, schema=None):
+        self.conn = WriteConnection(service, schema)
@@ -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"

@@ -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')
@@ -55,7 +55,7 @@ class StoreTests(unittest.TestCase):

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

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

    def test_manifest_edition_mismatch_rolls_back_instead_of_marking_coverage(self):
        path=Path(self.tmp.name)/'wrong-edition.jsonl'
-        row={'source':'pmlr','external_id':'v1/paper','url':'https://example.org/paper','title':'Paper','abstract':'Abstract','authors':[],'venue':'ICML','year':2025,'track':'main','status':'accepted'}
+        row={'source':'pmlr','external_id':'v1/paper','url':'https://example.org/paper','title':'Paper','abstract':'Abstract','authors':[],'venue':'ICML','year':2023,'track':'main','status':'accepted'}
        path.write_text(json.dumps(row)+'\n')
        path.with_suffix('.manifest.json').write_text(json.dumps({'source':'pmlr','list_url':'https://example.org/list','count':1,'year':2024}))
        with self.assertRaisesRegex(ValueError,'edition differs'):
@@ -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()
@@ -25,7 +25,7 @@ class MetadataTests(unittest.TestCase):
        self.assertEqual(venue_key('ECML-PKDD'),'ECML-PKDD')

    def test_missing_abstract_is_reported_as_partial_collection(self):
-        config=b'shortname: ICML\nyear: 2025\nbooktitle: ICML\n'
+        config=b'shortname: ICML\nyear: 2023\nbooktitle: ICML\n'
        paper='---\nid: paper1\nlayout: inproceedings\ntitle: Paper\nabstract:\n---\n'
        with tempfile.TemporaryDirectory() as tmp,patch('paper_atlas.collectors.pmlr.fetch',return_value=config),patch('paper_atlas.collectors.pmlr.metadata_texts',return_value=iter([('paper.md',paper)])):
            result=collect_volume(999,tmp,tmp)
@@ -0,0 +1,156 @@
+import copy
+import fcntl
+import hashlib
+import json
+from pathlib import Path
+import sqlite3
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from paper_atlas.collectors.pmlr_tracks import program_tracks, record_track
+from paper_atlas.db.store import Store
+from paper_atlas.db.submissions import identities_for_row
+from paper_atlas.ingestion.pmlr_tracks import reclassify, plan
+from paper_atlas.ingestion.records import ingest_file
+from tests.test_pmlr_tracks import event, program, record
+
+
+class PmlrMigrationTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory(); self.addCleanup(self.tmp.cleanup)
+        self.root = Path(self.tmp.name); self.db = self.root / 'papers.sqlite'
+        self.inventory = self.root / 'inventory.json'
+        self.rows = [record(title='A main research paper'),
+                     record('position1', 'v267/paper2', title='A position paper')]
+        self.proof = {'url': 'https://icml.cc/static/virtual/data/icml-2025-orals-posters.json', 'sha256': 'a' * 64}
+        mapping, _ = program_tracks(program(event(), event('position1', 'Position_Paper_Track', 2)), 2025)
+        store = Store(self.db); c = store.conn
+        for row in self.rows:
+            row.update(url='https://proceedings.mlr.press/' + row['external_id'] + '.html',
+                       list_url='https://proceedings.mlr.press/v267/', status='accepted')
+            paper, source = store.record('pmlr', row['external_id'], row['url'], row['title'], row['abstract'],
+                                         row['authors'], row['identifiers'], {'openreview': row['identifiers']['openreview']})
+            store.decision(paper, source, 'ICML', 2025, 'main', row['external_id'], row['url'],
+                           'accepted', 'listed in official proceedings', identities=identities_for_row(row))
+            record_track(row, mapping, self.proof)
+        c.execute("INSERT INTO tags(namespace,value) VALUES('custom:topic','keep me')")
+        c.execute("INSERT INTO paper_tags VALUES(1,1,NULL,'custom:manual')")
+        c.commit(); c.close()
+        self.write_inventory()
+
+    def write_inventory(self):
+        files = []
+        for row in self.rows:
+            path = self.root / (row['track'] + '.jsonl')
+            data = (json.dumps(row) + '\n').encode(); path.write_bytes(data)
+            m = {'source': 'pmlr', 'venue': 'ICML', 'year': 2025, 'volume': 267,
+                 'parser_version': 3, 'track': row['track'], 'track_program': self.proof,
+                 'count': 1, 'file': str(path), 'file_sha256': hashlib.sha256(data).hexdigest(),
+                 'missing_abstract_count': 0, 'list_url': row['list_url'], 'status': 'collected'}
+            path.with_suffix('.manifest.json').write_text(json.dumps(m)); files.append(m)
+        self.inventory.write_text(json.dumps({'source': 'pmlr', 'venue': 'ICML', 'year': 2025,
+                                              'volume': 267, 'parser_version': 3, 'track_program': self.proof,
+                                              'count': 2, 'files': files}))
+
+    def tracks(self):
+        with sqlite3.connect(self.db) as c:
+            return c.execute('SELECT s.id,e.track FROM submissions s JOIN editions e ON e.id=s.edition_id ORDER BY s.id').fetchall()
+
+    def test_plan_is_read_only_and_reports_exact_original_event(self):
+        result = reclassify(self.db, [self.inventory], self.root / 'plan.json')
+        self.assertEqual((result['status'], result['records_verified'], result['events_to_move']), ('planned', 2, 1))
+        self.assertEqual(result['changes'][0]['submission_id'], 2)
+        self.assertEqual(self.tracks(), [(1, 'main'), (2, 'main')])
+
+    def test_apply_preserves_ids_decisions_text_tags_and_is_idempotent(self):
+        with sqlite3.connect(self.db) as c:
+            before = {table: c.execute('SELECT * FROM ' + table).fetchall() for table in ('papers', 'decisions', 'decision_sources', 'submission_sources', 'paper_tags')}
+        result = reclassify(self.db, [self.inventory], self.root / 'apply.json', True)
+        self.assertEqual((result['status'], result['events_moved']), ('complete', 1))
+        self.assertEqual(self.tracks(), [(1, 'main'), (2, 'position')])
+        with sqlite3.connect(self.db) as c:
+            for table, rows in before.items():
+                self.assertEqual(c.execute('SELECT * FROM ' + table).fetchall(), rows, table)
+            self.assertEqual(c.execute('SELECT count(*) FROM submission_identifiers i JOIN submissions s ON s.id=i.submission_id WHERE i.edition_id<>s.edition_id').fetchone()[0], 0)
+            self.assertEqual(c.execute('SELECT e.track,es.imported_count FROM edition_sources es JOIN editions e ON e.id=es.edition_id ORDER BY e.track').fetchall(), [('main', 1), ('position', 1)])
+            self.assertEqual(c.execute('PRAGMA foreign_key_check').fetchall(), [])
+        again = reclassify(self.db, [self.inventory], self.root / 'again.json', True)
+        self.assertEqual(again['events_moved'], 0)
+        for m in json.loads(self.inventory.read_text())['files']:
+            ingest_file(self.db, m['file'])
+        self.assertEqual(self.tracks(), [(1, 'main'), (2, 'position')])
+        with sqlite3.connect(self.db) as c:
+            self.assertEqual(c.execute('SELECT * FROM decisions').fetchall(), before['decisions'])
+
+    def test_conflicting_target_event_stops_without_merging_history(self):
+        store = Store(self.db)
+        _, source = store.record('other', 'position1', 'https://example.org', 'Other evidence', None, [], {'openreview': 'position1'})
+        store.decision(2, source, 'ICML', 2025, 'position', 'position1', 'https://example.org', 'rejected', 'Reject', identities=[('openreview', 'position1')])
+        store.conn.commit(); store.conn.close()
+        before = self.tracks()
+        with self.assertRaisesRegex(ValueError, 'another event'):
+            reclassify(self.db, [self.inventory], self.root / 'blocked.json', True)
+        self.assertEqual(self.tracks(), before)
+
+    def test_shared_event_requires_reconciliation_and_keeps_all_sources(self):
+        store = Store(self.db)
+        _, source = store.record('mirror', 'position1', 'https://example.org', 'Mirror title', None, [], {'openreview': 'position1'})
+        store.conn.execute('INSERT INTO submission_sources VALUES(?,?,?)', (2, source, 'position1'))
+        store.conn.commit(); store.conn.close()
+        with self.assertRaisesRegex(ValueError, 'multiple sources'):
+            reclassify(self.db, [self.inventory], self.root / 'shared.json', True)
+        self.assertEqual(self.tracks(), [(1, 'main'), (2, 'main')])
+
+    def test_text_mismatch_in_last_record_prevents_any_changes(self):
+        with sqlite3.connect(self.db) as c:
+            c.execute("UPDATE source_records SET abstract='Changed elsewhere' WHERE id=2")
+        with self.assertRaisesRegex(ValueError, 'preserve existing publisher text'):
+            reclassify(self.db, [self.inventory], self.root / 'text.json', True)
+        self.assertEqual(self.tracks(), [(1, 'main'), (2, 'main')])
+
+    def test_mutation_failure_rolls_back_editions_metadata_and_identifiers(self):
+        calls = 0
+        def fail_after_updates(c, scopes):
+            nonlocal calls
+            calls += 1
+            if calls == 2:
+                raise ValueError('Injected post-migration failure')
+            return plan(c, scopes)
+        with sqlite3.connect(self.db) as c:
+            before = c.execute('SELECT * FROM source_records').fetchall()
+        with patch('paper_atlas.ingestion.pmlr_tracks.plan', side_effect=fail_after_updates):
+            with self.assertRaisesRegex(ValueError, 'Injected'):
+                reclassify(self.db, [self.inventory], self.root / 'rollback.json', True)
+        self.assertEqual(self.tracks(), [(1, 'main'), (2, 'main')])
+        with sqlite3.connect(self.db) as c:
+            self.assertEqual(c.execute('SELECT * FROM source_records').fetchall(), before)
+            self.assertEqual(c.execute("SELECT count(*) FROM editions WHERE track='position'").fetchone()[0], 0)
+        self.assertEqual(json.loads((self.root / 'rollback.json').read_text())['status'], 'error')
+
+    def test_changed_evidence_and_active_writer_are_refused(self):
+        out = self.root / 'blocked.json'
+        with Path(str(self.db) + '.maintenance.lock').open('a') as lock:
+            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            with self.assertRaises(BlockingIOError):
+                reclassify(self.db, [self.inventory], out, True)
+        self.assertFalse(out.exists())
+        path = self.root / 'position.jsonl'; path.write_text(path.read_text() + ' ')
+        with self.assertRaisesRegex(ValueError, 'differs from its verified inventory'):
+            reclassify(self.db, [self.inventory], out, True)
+        self.assertEqual(self.tracks(), [(1, 'main'), (2, 'main')])
+
+    def test_input_report_and_duplicate_forums_are_refused(self):
+        original = self.inventory.read_bytes()
+        with self.assertRaisesRegex(ValueError, 'preserve evidence'):
+            reclassify(self.db, [self.inventory], self.inventory, True)
+        self.assertEqual(self.inventory.read_bytes(), original)
+        self.rows[1]['identifiers']['openreview'] = self.rows[0]['identifiers']['openreview']
+        self.rows[1]['metadata']['track_evidence']['identity_value'] = self.rows[0]['identifiers']['openreview']
+        self.write_inventory()
+        with self.assertRaisesRegex(ValueError, 'repeated forum'):
+            reclassify(self.db, [self.inventory], self.root / 'duplicate.json', True)
+
+
+if __name__ == '__main__':
+    unittest.main()
@@ -0,0 +1,148 @@
+import copy
+import json
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from paper_atlas.collectors.pmlr import collect_volume
+from paper_atlas.collectors.pmlr_tracks import program_tracks, record_track, position_list_policy
+from paper_atlas.ingestion.records import ingest_file
+from paper_atlas.ingestion.backfill import backfill
+
+
+def event(forum='realForum1', track='Conference', event_id=1, year=2025, **values):
+    return {'id': event_id, 'sourceid': event_id, 'name': 'A published paper',
+            'sourceurl': f'https://openreview.net/group?id=ICML.cc/{year}/' + track,
+            'paper_url': 'https://openreview.net/forum?id=' + forum,
+            'decision': 'Accept (Poster)', **values}
+
+
+def program(*events, **changes):
+    return json.dumps({'count': len(events), 'results': list(events), **changes}).encode()
+
+
+def record(forum='realForum1', external='v267/paper1', year=2025, title='A published paper'):
+    return {'source': 'pmlr', 'external_id': external, 'venue': 'ICML', 'year': year,
+            'title': title, 'abstract': 'Original abstract', 'authors': ['Jane Doe', 'John Smith'],
+            'identifiers': {'openreview': forum}, 'metadata': {}, 'track': 'main'}
+
+
+class PmlrTrackTests(unittest.TestCase):
+    def provenance(self):
+        return {'url': 'https://icml.cc/static/virtual/data/icml-2025-orals-posters.json', 'sha256': 'a' * 64}
+
+    def test_real_groups_not_title_prefixes_determine_modern_tracks(self):
+        mapping, inventory = program_tracks(program(
+            event(name='Position: still a main-track paper'),
+            event('position1', 'Position_Paper_Track', 2, name='No prefix needed'),
+            event('2025-Oral-fake', event_id=3, sourceid=-1),
+            event('workshop1', 'Workshop', 4)), 2025)
+        main = record(title='Position: still a main-track paper')
+        position = record('position1', 'v267/paper2', title='No prefix needed')
+        record_track(main, mapping, self.provenance()); record_track(position, mapping, self.provenance())
+        self.assertEqual((main['track'], position['track']), ('main', 'position'))
+        self.assertEqual(inventory['paper_identity_count'], 2)
+        self.assertEqual(inventory['excluded_events'], {'synthetic_oral_event': 1, 'other_source': 1})
+
+    def test_conflicting_forums_pagination_and_missing_real_ids_fail(self):
+        for data in [program(event(), event(track='Position_Paper_Track', event_id=2)),
+                     program(event(), next='page2'), program(event(), count=2),
+                     program(event(paper_url='')), program(event(decision='Reject')),
+                     program(event(), event())]:
+            with self.subTest(data=data), self.assertRaises(ValueError):
+                program_tracks(data, 2025)
+
+    def test_missing_publication_identity_has_no_main_track_fallback(self):
+        mapping, _ = program_tracks(program(event()), 2025)
+        with self.assertRaisesRegex(ValueError, 'absent'):
+            record_track(record('unknownForum'), mapping, self.provenance())
+
+    def test_2024_requires_organizer_endorsed_list_and_publisher_self_links(self):
+        html = b'<a href="/virtual/2024/papers.html?filter=titles&amp;search=position%3A">position papers</a>'
+        policy = position_list_policy(html)
+        with self.assertRaises(ValueError):
+            position_list_policy(html.replace(b'2024', b'2023'))
+        e = event(year=2024, name='Position: An explicit position', paper_url='',
+                  paper_pdf_url='https://proceedings.mlr.press/v235/paper1.html')
+        with self.assertRaisesRegex(ValueError, 'policy'):
+            program_tracks(program(e), 2024)
+        mapping, _ = program_tracks(program(e), 2024, policy)
+        row = record(external='v235/paper1', year=2024)
+        record_track(row, mapping, {**self.provenance(), 'position_list_policy': policy})
+        self.assertEqual(row['track'], 'position')
+        self.assertEqual(row['metadata']['track_evidence']['identity_scheme'], 'pmlr')
+
+    def test_reviewed_2024_link_alias_still_needs_title_and_two_authors(self):
+        e = event(year=2024, paper_url='', paper_pdf_url='https://proceedings.mlr.press/v235/theo24a.html',
+                  authors=[{'fullname': 'Jane Doe'}, {'fullname': 'John Smith'}])
+        mapping, _ = program_tracks(program(e), 2024, {'verified': True})
+        row = record(external='v235/guyard24a', year=2024)
+        record_track(row, mapping, self.provenance())
+        self.assertEqual(row['metadata']['track_evidence']['corroborating_author_count'], 2)
+        self.assertNotIn('_authors', row['metadata']['track_evidence'])
+        for changed in [dict(title='Different title'), dict(authors=['Someone Else'])]:
+            bad = record(external='v235/guyard24a', year=2024); bad.update(changed)
+            with self.assertRaisesRegex(ValueError, 'corroboration'):
+                record_track(bad, mapping, self.provenance())
+
+    def test_collector_splits_tracks_preserves_ids_and_imports_one_event_each(self):
+        config = b'shortname: ICML\nyear: 2025\nbooktitle: ICML\n'
+        template = '---\nid: {id}\nlayout: inproceedings\ntitle: {title}\nabstract: Original abstract\nopenreview: {forum}\n---\n'
+        posts = [('a.md', template.format(id='paper1', title='Main paper', forum='realForum1')),
+                 ('b.md', template.format(id='paper2', title='Position paper', forum='position1'))]
+        mapping, inventory = program_tracks(program(event(), event('position1', 'Position_Paper_Track', 2)), 2025)
+        with tempfile.TemporaryDirectory() as tmp, patch('paper_atlas.collectors.pmlr.fetch', return_value=config), \
+                patch('paper_atlas.collectors.pmlr.metadata_texts', return_value=iter(posts)), \
+                patch('paper_atlas.collectors.pmlr.collect_track_evidence', return_value=(mapping, self.provenance())):
+            result = collect_volume(267, tmp, tmp)
+            self.assertNotIn('file', result)
+            self.assertEqual(result['track'], 'mixed')
+            self.assertEqual({m['track']: m['count'] for m in result['files']}, {'main': 1, 'position': 1})
+            self.assertEqual(result['count'], 2)
+            db = Path(tmp) / 'papers.sqlite'
+            for manifest in result['files']:
+                ingest_file(db, manifest['file'])
+            import sqlite3
+            with sqlite3.connect(db) as c:
+                self.assertEqual(c.execute('SELECT count(*) FROM submissions').fetchone()[0], 2)
+                self.assertEqual(c.execute('SELECT track,count(*) FROM submissions s JOIN editions e ON s.edition_id=e.id GROUP BY track').fetchall(), [('main', 1), ('position', 1)])
+                self.assertEqual(c.execute('PRAGMA foreign_key_check').fetchall(), [])
+
+    def test_unknown_track_does_not_replace_existing_volume_output(self):
+        config = b'shortname: ICML\nyear: 2025\nbooktitle: ICML\n'
+        post = '---\nid: paper1\nlayout: inproceedings\ntitle: Paper\nopenreview: missing123\n---\n'
+        with tempfile.TemporaryDirectory() as tmp, patch('paper_atlas.collectors.pmlr.fetch', return_value=config), \
+                patch('paper_atlas.collectors.pmlr.metadata_texts', return_value=iter([('a.md', post)])), \
+                patch('paper_atlas.collectors.pmlr.collect_track_evidence', return_value=({}, self.provenance())):
+            old = Path(tmp) / 'pmlr-v267.jsonl'; old.write_text('previous collection')
+            with self.assertRaisesRegex(ValueError, 'absent'):
+                collect_volume(267, tmp, tmp)
+            self.assertEqual(old.read_text(), 'previous collection')
+            self.assertFalse((Path(tmp) / 'pmlr-v267.inventory.json').exists())
+
+    def test_old_mixed_manifest_is_rejected_before_opening_database(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / 'pmlr-v267.jsonl'; path.write_text('')
+            path.with_suffix('.manifest.json').write_text(json.dumps({'source': 'pmlr', 'venue': 'ICML', 'year': 2025, 'parser_version': 2}))
+            db = Path(tmp) / 'absent.sqlite'
+            for action in [lambda: ingest_file(db, path), lambda: backfill(db, [path])]:
+                with self.assertRaisesRegex(ValueError, 'Recollect mixed'):
+                    action()
+            self.assertFalse(db.exists())
+
+    def test_changed_track_or_own_identity_rolls_back_import(self):
+        from paper_atlas.collectors.pmlr_tracks import require_track_record
+        row = record(); mapping, _ = program_tracks(program(event()), 2025)
+        record_track(row, mapping, self.provenance())
+        manifest = {'source': 'pmlr', 'venue': 'ICML', 'year': 2025, 'parser_version': 3,
+                    'track_program': self.provenance()}
+        require_track_record(row, manifest)
+        for change in [dict(track='position'), dict(identifiers={'openreview': 'otherForum'})]:
+            bad = copy.deepcopy(row); bad.update(change)
+            with self.assertRaises(ValueError):
+                require_track_record(bad, manifest)
+
+
+if __name__ == '__main__':
+    unittest.main()
@@ -0,0 +1,309 @@
+import contextlib
+from contextlib import closing
+import io
+import itertools
+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, text
+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')"))
+
+    def test_text_codec_is_bijective_and_preserves_key_order_and_prefixes(self):
+        alphabet = ['\x00', '\x01', '\x02', 'a', 'é', '🙂']
+        values = [''.join(v) for n in range(4) for v in itertools.product(alphabet, repeat=n)]
+        for value in values:
+            encoded = text.encode(value)
+            self.assertNotIn('\x00', encoded)
+            self.assertEqual(text.decode(encoded), value)
+        self.assertEqual(sorted(map(text.encode, values)), list(map(text.encode, sorted(values))))
+        for value in [None, 7, 0.5]:
+            self.assertEqual(text.encode(value), value)
+            self.assertEqual(text.decode(value), value)
+        for value in ['\x01', 'prefix\x01x']:
+            with self.assertRaisesRegex(ValueError, 'Invalid'):
+                text.decode(value)
+
+
+@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()
+
+    def test_nul_and_literal_escape_survive_full_copy_keys_query_and_hash(self):
+        original = 'before\x00survivor; literal\x01\x01 and \x01\x02 — 研究'
+        keys = ['a\x00z', 'a\x01z', 'a\x01\x02z', 'a\x02z']
+        with closing(sqlite3.connect(self.db)) as source, source:
+            source.execute('UPDATE papers SET abstract=? WHERE id=1', (original,))
+            source.execute('UPDATE source_records SET abstract=? WHERE id=1', (original,))
+            source.executemany("INSERT INTO identifiers VALUES('control',?,1)", [(k,) for k in keys])
+            source.execute("INSERT INTO tags(namespace,value) VALUES('custom:control',?)", (original,))
+            tag = source.execute("SELECT id FROM tags WHERE namespace='custom:control'").fetchone()[0]
+            source.executemany('INSERT INTO paper_tags VALUES(1,?,NULL,?)', [(tag, k) for k in keys])
+        report = self.migrate()
+        self.assertEqual(report['text_codec'], text.CODEC)
+        for entry in report['tables'].values():
+            self.assertEqual(entry['source_sha256'], entry['target_sha256'])
+        self.assertNotEqual(report['tables']['source_records']['source_sha256'],
+                            report['tables']['source_records']['target_storage_sha256'])
+        with connect(self.service) as pg:
+            self.assertEqual(pg.execute("SELECT obj_description(oid,'pg_namespace') FROM pg_namespace WHERE nspname=%s", (self.target,)).fetchone()[0], 'paper-atlas:text-codec=' + text.CODEC)
+            raw = pg.execute('SELECT abstract FROM ' + schema.quote(self.target) + '.source_records WHERE id=1').fetchone()[0]
+            self.assertNotIn('\x00', raw)
+            self.assertEqual(text.decode(raw), original)
+        reader = ReadConnection(self.service, self.target)
+        try:
+            self.assertEqual(reader.execute('SELECT abstract FROM source_records WHERE id=?', (1,)).fetchone()[0], original)
+            self.assertEqual([r[0] for r in reader.execute("SELECT value FROM identifiers WHERE scheme='control' ORDER BY value")], sorted(keys))
+            self.assertEqual(reader.execute("SELECT value FROM identifiers WHERE scheme='control' AND value=?", (keys[0],)).fetchone()[0], keys[0])
+        finally:
+            reader.close()
+        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, '--paper', '1', '--tag', 'custom:control=' + original])
+            result = json.loads(output.getvalue())[0]
+            self.assertEqual(result['abstract'], original)
+            self.assertEqual(next(s['abstract'] for s in result['sources'] if s['external_id'] == '2401.00001'), original)
+            output = io.StringIO()
+            with contextlib.redirect_stdout(output):
+                querying.main(['--pg-service', self.service, '--query', 'survivor'])
+            self.assertEqual([r['id'] for r in json.loads(output.getvalue())], [1])
+
+    def test_old_completed_staging_table_is_upgraded_atomically_on_resume(self):
+        original = 'https://example.org/literal\x01\x02'
+        with closing(sqlite3.connect(self.db)) as source, source:
+            source.execute('UPDATE sources SET homepage=? WHERE id=1', (original,))
+        self.interrupt_after_first_table()
+        with connect(self.service) as pg:
+            control = schema.quote(self.target + '_migration')
+            stage = pg.execute('SELECT stage FROM ' + control + '.run WHERE id=1').fetchone()[0]
+            # Reproduce the old implementation: raw SOH text and no codec metadata.
+            pg.execute('UPDATE ' + schema.quote(stage) + '.sources SET homepage=%s WHERE id=1', (original,))
+            pg.execute('ALTER TABLE ' + control + '.run DROP COLUMN text_codec')
+        report = self.migrate()
+        self.assertEqual(report['legacy_text_upgrade_rows']['sources'], 1)
+        self.assertEqual(report['tables']['sources']['source_sha256'], report['tables']['sources']['target_sha256'])
+        reader = ReadConnection(self.service, self.target)
+        try:
+            self.assertEqual(reader.execute('SELECT homepage FROM sources WHERE id=1').fetchone()[0], original)
+        finally:
+            reader.close()
+
+    def test_legacy_unique_keys_can_upgrade_after_constraints_were_built(self):
+        keys = ['x\x01', 'x\x01\x02']
+        with closing(sqlite3.connect(self.db)) as source, source:
+            source.executemany("INSERT INTO identifiers VALUES('control',?,1)", [(k,) for k in keys])
+        # Emulate an old run interrupted after constraints: no storage encoding.
+        with patch('paper_atlas.postgres.text.encode', side_effect=lambda value: value), \
+             patch('paper_atlas.postgres.migration.sequences', side_effect=RuntimeError('after constraints')):
+            with self.assertRaisesRegex(RuntimeError, 'after constraints'):
+                self.migrate()
+        with connect(self.service) as pg:
+            pg.execute('ALTER TABLE ' + schema.quote(self.target + '_migration') + '.run DROP COLUMN text_codec')
+        report = self.migrate()
+        self.assertEqual(report['status'], 'complete')
+        self.assertEqual(report['tables']['identifiers']['source_sha256'], report['tables']['identifiers']['target_sha256'])
+        reader = ReadConnection(self.service, self.target)
+        try:
+            self.assertEqual([r[0] for r in reader.execute("SELECT value FROM identifiers WHERE scheme='control' ORDER BY value")], keys)
+        finally:
+            reader.close()
+
+
+if __name__ == '__main__':
+    unittest.main()
@@ -0,0 +1,298 @@
+from contextlib import closing, redirect_stdout
+import io
+import json
+import os
+from pathlib import Path
+import sqlite3
+import tempfile
+from types import SimpleNamespace
+import unittest
+from unittest.mock import patch
+import uuid
+import zipfile
+
+from paper_atlas.db.backend import open_store, open_read_connection
+from paper_atlas.db.store import Store
+from paper_atlas.ingestion import arxiv
+from paper_atlas.ingestion.backfill import backfill
+from paper_atlas.ingestion.records import ingest_file
+from paper_atlas.postgres.client import connect, ReadConnection
+from paper_atlas.postgres.migration import migrate
+from paper_atlas.postgres.schema import quote
+
+
+class BackendSelectionTests(unittest.TestCase):
+    def test_ambiguous_targets_fail_before_opening_any_database(self):
+        for kwargs in ({}, {'db': 'example', 'pg_service': 'example'}, {'db': 'example', 'pg_schema': 'research'}):
+            with self.assertRaisesRegex(ValueError, 'Choose exactly one'):
+                open_store(**kwargs)
+
+    def test_readonly_audit_does_not_create_an_absent_sqlite_database(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / 'absent.sqlite'
+            with self.assertRaisesRegex(ValueError, 'Existing database'):
+                open_read_connection(path)
+            self.assertFalse(path.exists())
+
+
+@unittest.skipUnless(os.environ.get('PAPER_ATLAS_TEST_PG_SERVICE'), 'Requires the approved remote PostgreSQL test service')
+class PostgreSQLIngestionTests(unittest.TestCase):
+    def setUp(self):
+        self.service = os.environ['PAPER_ATLAS_TEST_PG_SERVICE']
+        self.schema = 'pa_write_' + uuid.uuid4().hex[:12]
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.root = Path(self.temp.name)
+        source = self.root / 'seed.sqlite'
+        store = Store(source)
+        paper, record = store.record('arxiv', '2401.00001', 'https://arxiv.org/abs/2401.00001',
+                                     'Existing reference paper', 'Existing abstract', [])
+        store.decision(paper, record, 'ICLR', 2024, 'main', 'seed', 'https://example.org/seed', 'rejected', 'Reject')
+        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)
+        with redirect_stdout(io.StringIO()):
+            migrate(source, self.service, self.root / 'migration.json', schema=self.schema, min_free_gib=0)
+
+    def cleanup(self):
+        with connect(self.service) as pg:
+            names = pg.execute('SELECT nspname FROM pg_namespace WHERE starts_with(nspname,%s)', (self.schema,)).fetchall()
+            for name, in names:
+                pg.execute('DROP SCHEMA ' + quote(name) + ' CASCADE')
+
+    def store(self):
+        return open_store(pg_service=self.service, pg_schema=self.schema)
+
+    def row(self, source='openreview_snapshot', external='forumNew1', status='rejected', year=2025, venue='ICLR'):
+        return {'source': source, 'external_id': external, 'url': 'https://example.org/' + external,
+                'venue': venue, 'year': year, 'track': 'main', 'status': status,
+                'title': 'A newly observed shared paper', 'abstract': 'Before\x00survivor and literal\x01\x02',
+                'authors': ['Jane Doe', 'John Smith'], 'identifiers': {'arxiv': '2501.99999'},
+                'observed_at': str(year) + '-01-20T00:00:00Z'}
+
+    def file(self, name, rows, count=None):
+        path = self.root / (name + '.jsonl')
+        path.write_text(''.join(json.dumps(r, ensure_ascii=False) + '\n' for r in rows))
+        first = rows[0]
+        manifest = {key: first[key] for key in ('source', 'venue', 'year', 'track')}
+        manifest.update(parser_version=2, list_url='https://example.org/' + name,
+                        count=len(rows) if count is None else count, missing_abstract_count=0, status='complete')
+        path.with_suffix('.manifest.json').write_text(json.dumps(manifest))
+        return path
+
+    def ingest(self, path):
+        with redirect_stdout(io.StringIO()):
+            ingest_file(None, path, pg_service=self.service, pg_schema=self.schema)
+
+    def test_native_writes_keep_nul_historical_ids_and_later_acceptance(self):
+        store = self.store()
+        try:
+            paper, record = store.record('public', 'new-forum', 'https://example.org/new', 'New research identity',
+                                         'Left\x00right\x01\x02', [], {'arxiv': '2501.99999'})
+            self.assertGreater(paper, 1000)
+            store.decision(paper, record, 'ICLR', 2025, 'main', 'new-forum', 'https://example.org/reject', 'rejected', 'Reject')
+            _, second = store.record('official', 'new-pub', 'https://example.org/pub', 'Changed title', None, [], {'arxiv': '2501.99999'})
+            store.decision(paper, second, 'COLM', 2026, 'main', 'new-pub', 'https://example.org/accept', 'accepted', 'Accept')
+            store.conn.commit()
+            self.assertEqual(store.conn.execute('SELECT abstract FROM papers WHERE id=?', (paper,)).fetchone()[0], 'Left\x00right\x01\x02')
+            ids = store.conn.execute('SELECT s.id,d.id FROM submissions s JOIN decisions d ON d.submission_id=s.id WHERE s.paper_id=?', (paper,)).fetchall()
+            self.assertEqual(len(ids), 2)
+            self.assertTrue(all(sid > 5000 and did > 6000 for sid, did in ids))
+        finally:
+            store.conn.close()
+        reader = ReadConnection(self.service, self.schema)
+        try:
+            from paper_atlas.querying import known_acceptances
+            self.assertEqual(known_acceptances(reader, paper)[0]['venue'], 'COLM')
+        finally:
+            reader.close()
+
+    def test_file_import_is_idempotent_and_refreshes_only_own_tags(self):
+        row = self.row(); row['tags'] = [{'namespace': 'openreview_snapshot:area', 'value': 'theory'}]
+        path = self.file('reject', [row]); self.ingest(path)
+        store = self.store()
+        try:
+            paper = store.conn.execute("SELECT paper_id FROM identifiers WHERE scheme='arxiv' AND value='2501.99999'").fetchone()[0]
+            store.conn.execute("INSERT INTO tags(namespace,value) VALUES('custom:topic','manual') ON CONFLICT DO NOTHING")
+            tag = store.conn.execute("SELECT id FROM tags WHERE namespace='custom:topic'").fetchone()[0]
+            store.conn.execute('INSERT INTO paper_tags VALUES(?,?,NULL,?)', (paper, tag, 'manual'))
+            store.conn.commit()
+        finally:
+            store.conn.close()
+        row['tags'][0]['value'] = 'learning'; self.ingest(self.file('reject', [row]))
+        self.ingest(self.file('accept', [self.row('official', 'published', 'accepted', 2026, 'COLM')]))
+        store = self.store()
+        try:
+            self.assertEqual(store.conn.execute('SELECT count(*) FROM papers').fetchone()[0], 2)
+            self.assertEqual(store.conn.execute('SELECT count(*) FROM submissions WHERE paper_id=?', (paper,)).fetchone()[0], 2)
+            self.assertEqual(store.conn.execute('SELECT count(*) FROM decisions d JOIN submissions s ON s.id=d.submission_id WHERE s.paper_id=?', (paper,)).fetchone()[0], 2)
+            tags = set(store.conn.execute('SELECT t.namespace,t.value FROM paper_tags p JOIN tags t ON t.id=p.tag_id WHERE p.paper_id=?', (paper,)))
+            self.assertEqual(tags, {('custom:topic', 'manual'), ('openreview_snapshot:area', 'learning')})
+        finally:
+            store.conn.close()
+
+    def test_bad_manifest_rolls_back_whole_file_and_keeps_failed_run(self):
+        with self.assertRaisesRegex(ValueError, 'Manifest count'):
+            self.ingest(self.file('bad', [self.row()], count=2))
+        store = self.store()
+        try:
+            self.assertEqual(store.conn.execute('SELECT count(*) FROM papers').fetchone()[0], 1)
+            self.assertEqual(store.conn.execute('SELECT count(*) FROM source_records').fetchone()[0], 1)
+            self.assertEqual(store.conn.execute('SELECT status,records_seen,records_written FROM ingest_runs').fetchone(), ('failed', 1, 0))
+        finally:
+            store.conn.close()
+
+    def test_changed_file_hash_rolls_back_data_before_commit(self):
+        path = self.file('changed', [self.row()])
+        manifest = json.loads(path.with_suffix('.manifest.json').read_text())
+        manifest['file_sha256'] = '0' * 64
+        path.with_suffix('.manifest.json').write_text(json.dumps(manifest))
+        with self.assertRaisesRegex(ValueError, 'hash differs'):
+            self.ingest(path)
+        store = self.store()
+        try:
+            self.assertEqual(store.conn.execute('SELECT count(*) FROM papers').fetchone()[0], 1)
+            self.assertEqual(store.conn.execute('SELECT count(*) FROM source_records').fetchone()[0], 1)
+            self.assertEqual(store.conn.execute('SELECT status,records_written FROM ingest_runs').fetchone(), ('failed', 0))
+        finally:
+            store.conn.close()
+
+    def test_writer_lock_survives_commit_and_rollback_but_readers_continue(self):
+        first = self.store()
+        try:
+            first.conn.commit()
+            with self.assertRaisesRegex(RuntimeError, 'Another writer'):
+                self.store()
+            first.conn.rollback()
+            with self.assertRaisesRegex(RuntimeError, 'Another writer'):
+                self.store()
+            reader = ReadConnection(self.service, self.schema)
+            try:
+                self.assertEqual(reader.execute('SELECT count(*) FROM papers').fetchone()[0], 1)
+            finally:
+                reader.close()
+        finally:
+            first.conn.close()
+        reopened = self.store(); reopened.conn.close()
+
+    def test_unpublished_schema_is_rejected_without_bootstrapping_tables(self):
+        staging = self.schema + '_unfinished'
+        with connect(self.service) as pg:
+            pg.execute('CREATE SCHEMA ' + quote(staging))
+        with self.assertRaisesRegex(RuntimeError, 'published dataset'):
+            open_store(pg_service=self.service, pg_schema=staging)
+        with connect(self.service) as pg:
+            self.assertEqual(pg.execute('SELECT count(*) FROM information_schema.tables WHERE table_schema=%s', (staging,)).fetchone()[0], 0)
+
+    def test_negative_preflight_uses_current_pg_attempt_and_blocks_stale_import(self):
+        import hashlib
+        from paper_atlas.collectors.public_openreview import SOURCE, REVISION
+        from paper_atlas.audits.public_openreview import audit
+        row = self.row(SOURCE, 'negativeForum')
+        row['identifiers']['openreview'] = 'negativeForum'
+        row['metadata'] = {'evidence_kind': 'curated_public_status_assertion',
+                           'source_revision': REVISION, 'original_decision_signature_verified': False}
+        path = self.file('negative', [row])
+        manifest = json.loads(path.with_suffix('.manifest.json').read_text())
+        manifest.update(file=str(path), file_sha256=hashlib.sha256(path.read_bytes()).hexdigest())
+        path.with_suffix('.manifest.json').write_text(json.dumps(manifest))
+        inventory = self.root / 'negative-inventory.json'
+        inventory.write_text(json.dumps({'source': SOURCE, 'revision': REVISION,
+                                        'normalized_records': 1, 'files': [manifest]}))
+        # An acceptance in another conference does not block this rejection.
+        self.ingest(self.file('other-conference', [self.row('official', 'acceptedElsewhere', 'accepted', 2026, 'COLM')]))
+        def preflight(name):
+            with redirect_stdout(io.StringIO()):
+                return audit(None, inventory, self.root / (name + '.json'), pg_service=self.service, pg_schema=self.schema)
+        first = preflight('before')
+        self.assertEqual(first['status'], 'ready_for_serial_import')
+        self.assertEqual(first['engine'], 'postgresql')
+        self.assertEqual(first['schema'], self.schema)
+        self.ingest(path)
+        self.assertEqual(preflight('existing')['existing_same_edition_attempts'], 1)
+        store = self.store()
+        try:
+            paper, record = store.conn.execute('''SELECT r.paper_id,r.id FROM source_records r
+                JOIN sources s ON s.id=r.source_id WHERE s.name=? AND r.external_id=?''', (SOURCE, row['external_id'])).fetchone()
+            store.decision(paper, record, 'ICLR', 2025, 'main', row['external_id'],
+                           'https://example.org/formal', 'accepted', 'Accept', observed_at='2027-01-01T00:00:00Z')
+            # Equal observation time, but an explicitly earlier decision date.
+            # NULL dates must fall back to observed_at, not the empty string.
+            store.decision(paper, record, 'ICLR', 2025, 'main', row['external_id'],
+                           'https://example.org/older', 'rejected', 'Reject', observed_at='2027-01-01T00:00:00Z',
+                           decision_date='2026-12-31T00:00:00Z')
+            store.conn.commit()
+        finally:
+            store.conn.close()
+        blocked = preflight('blocked')
+        self.assertEqual(blocked['status'], 'blocked')
+        self.assertEqual(blocked['conflicts'][0]['existing_status'], 'accepted')
+        self.assertEqual(blocked['files'], [])
+        # The earlier ready report cannot authorize overwriting current status.
+        with self.assertRaisesRegex(ValueError, 'conflicts with current'):
+            self.ingest(path)
+        reader = open_read_connection(pg_service=self.service, pg_schema=self.schema)
+        try:
+            self.assertEqual(reader.execute('SHOW transaction_read_only').fetchone()[0], 'on')
+            from paper_atlas.db.submissions import latest_status
+            sid = reader.execute("SELECT submission_id FROM submission_identifiers WHERE scheme='openreview' AND value=?", ('negativeForum',)).fetchone()[0]
+            self.assertEqual(latest_status(reader, sid), 'accepted')
+        finally:
+            reader.close()
+
+    def test_backfill_merges_evidence_without_reimporting_or_losing_history(self):
+        files = []
+        for source, external, title in [('official', 'legacy', 'First bibliographic name'), ('openreview_snapshot', 'forum', 'Renamed submission')]:
+            row = self.row(source, external, 'accepted')
+            row.update(title=title, authors=[], identifiers={}, evidence_url='https://example.org/same-decision')
+            path = self.file(source, [row]); self.ingest(path)
+            row['identifiers'] = {'openreview': 'sharedForum'}
+            self.file(source, [row]); files.append(path)
+        with redirect_stdout(io.StringIO()):
+            result = backfill(None, files, pg_service=self.service, pg_schema=self.schema)
+        self.assertEqual((result['submissions_before'], result['submissions_after']), (3, 2))
+        store = self.store()
+        try:
+            self.assertEqual(store.conn.execute('SELECT count(*) FROM papers').fetchone()[0], 2)
+            self.assertEqual(store.conn.execute('SELECT count(*) FROM decision_sources').fetchone()[0], 3)
+            self.assertEqual(store.conn.execute('SELECT count(*) FROM decision_merges').fetchone()[0], 2)
+        finally:
+            store.conn.close()
+
+    def test_arxiv_chunk_resume_preserves_committed_ids_and_category_evidence(self):
+        archive = self.root / 'arxiv.zip'
+        rows = [{'id': f'2501.{i:05d}', 'title': f'Checkpoint paper {i}', 'abstract': 'Metadata abstract',
+                 'authors': [], 'categories': 'cs.AI'} for i in range(1, 502)]
+        with zipfile.ZipFile(archive, 'w') as zipped:
+            zipped.writestr('arxiv-metadata-oai-snapshot.json', ''.join(json.dumps(r) + '\n' for r in rows))
+        calls = 0
+        original = arxiv.refresh_categories
+        def interrupt(*args):
+            nonlocal calls
+            calls += 1
+            if calls == 501:
+                raise RuntimeError('controlled interruption')
+            return original(*args)
+        with patch('paper_atlas.ingestion.arxiv.shutil.disk_usage', return_value=SimpleNamespace(free=8 * 1024**3)), \
+             patch('paper_atlas.ingestion.arxiv.refresh_categories', side_effect=interrupt), redirect_stdout(io.StringIO()):
+            with self.assertRaisesRegex(RuntimeError, 'controlled interruption'):
+                arxiv.ingest(None, archive, pg_service=self.service, pg_schema=self.schema)
+        store = self.store()
+        try:
+            self.assertEqual(store.conn.execute('SELECT count(*) FROM papers').fetchone()[0], 501)
+            preserved = store.conn.execute("SELECT paper_id FROM identifiers WHERE scheme='arxiv' AND value='2501.00001'").fetchone()[0]
+        finally:
+            store.conn.close()
+        with redirect_stdout(io.StringIO()):
+            arxiv.ingest(None, archive, pg_service=self.service, pg_schema=self.schema)
+        store = self.store()
+        try:
+            self.assertEqual(store.conn.execute('SELECT count(*) FROM papers').fetchone()[0], 502)
+            self.assertEqual(store.conn.execute("SELECT paper_id FROM identifiers WHERE scheme='arxiv' AND value='2501.00001'").fetchone()[0], preserved)
+            self.assertEqual(store.conn.execute('SELECT count(*) FROM paper_tags').fetchone()[0], 501)
+            self.assertEqual(store.conn.execute('SELECT status,records_written FROM ingest_runs').fetchone(), ('complete', 501))
+        finally:
+            store.conn.close()
Author	SHA1	Message	Date
codex-bot	6fa59c4d26	Audit public decisions against PostgreSQL and recheck imports Metadata integrity checks / metadata (push) Successful in 1m9s Details	2026-11-07 16:01:27 +08:00
Zayden-Jung	dc41dfa350	Merge pull request '为 PostgreSQL 主库接入元数据导入与投稿证据写入' (#15 ) from codex/postgres-ingestion into main Metadata integrity checks / metadata (push) Successful in 1m25s Details Reviewed-on: #15	2026-11-07 07:53:18 +00:00
codex-bot	6ea150690d	Support native PostgreSQL metadata ingestion and provenance writes Metadata integrity checks / metadata (push) Successful in 1m7s Details	2026-11-07 15:38:42 +08:00
Zayden-Jung	426aa042a5	Merge pull request '无损迁移含 NUL 的来源文本并恢复 PostgreSQL 断点' (#14 ) from codex/postgres-nul into main Metadata integrity checks / metadata (push) Successful in 1m24s Details Reviewed-on: #14	2026-11-06 19:37:00 +00:00
codex-bot	4e62710bba	Preserve NUL text through PostgreSQL migration and resume Metadata integrity checks / metadata (push) Successful in 1m7s Details	2026-11-07 03:27:57 +08:00
Zayden-Jung	3f040b968f	Merge pull request '迁移 PostgreSQL 并保留完整研究数据与投稿历史' (#13 ) from codex/postgresql-migration into main Metadata integrity checks / metadata (push) Successful in 1m18s Details Reviewed-on: #13	2026-11-06 18:56:51 +00:00
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	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	4e0e3f0ce8	Migrate existing PMLR submission tracks with atomic provenance checks Metadata integrity checks / metadata (push) Successful in 44s Details	2026-11-06 22:15:56 +08:00
codex-bot	17bdaa77fb	Resolve mixed ICML PMLR tracks from verified official programs Metadata integrity checks / metadata (push) Successful in 40s Details	2026-11-06 22:12:08 +08:00
				`@@ -0,0 +1 @@`
				`"""PostgreSQL migration and read access for the paper index."""`