Add read-only source identity candidate audit with evidence shards

Merge pull request '适配AAMAS2026官方目录并保留撤稿与撤回证据' (#49 ) from codex/aamas2026 into main
Reviewed-on: #49
2026-11-10 02:55:27 +08:00 · 2026-11-09 18:49:37 +00:00 · 2026-11-09 18:49:29 +00:00 · 2026-11-10 02:18:38 +08:00 · 2026-11-10 01:12:45 +08:00 · 2026-11-10 01:09:25 +08:00
28 changed files with 3689 additions and 7 deletions
@@ -12,7 +12,7 @@ jobs:
      - name: Verify metadata storage and query commands
        run: |
          python3 -B - <<'PY'
-          import io, os, pathlib, re, subprocess, tarfile, tempfile, urllib.request
+          import gzip, io, json, os, pathlib, re, subprocess, tarfile, tempfile, urllib.request
          sha = os.environ['ARCHIVE_SHA']
          if not re.fullmatch(r'[0-9a-fA-F]{40,64}', sha):
              raise SystemExit('Invalid source commit')
@@ -50,9 +50,42 @@ jobs:
              subprocess.run([python,'-I','-B','-m','paper_atlas','collect','cvf','--help'],cwd=outside,check=True)
              subprocess.run([python,'-I','-B','-m','paper_atlas','collect','www-archive','--help'],cwd=outside,check=True)
              subprocess.run([python,'-I','-B','-m','paper_atlas','collect','aamas-proceedings','--help'],cwd=outside,check=True)
+              subprocess.run([python,'-I','-B','-m','paper_atlas','collect','neurips-program','--help'],cwd=outside,check=True)
+              neurips_snapshots = root/'neurips2026-program-snapshots'
+              neurips_snapshots.mkdir()
+              (neurips_snapshots/'program-json.response').write_bytes(gzip.decompress((src/'tests/fixtures/neurips2026-program.json.gz').read_bytes()))
+              (neurips_snapshots/'receipts.json').write_bytes((src/'tests/fixtures/neurips2026-receipts.json').read_bytes())
+              subprocess.run([python,'-I','-B','-m','paper_atlas','collect','neurips-program','--snapshots',str(neurips_snapshots),'--out',str(root/'installed-neurips2026-program')],cwd=outside,check=True)
+              neurips_report = json.loads((root/'installed-neurips2026-program/neurips-program-2026.json').read_text())
+              assert neurips_report['records'] == 9004 and len(neurips_report['held_presentations']) == 139
+              assert len(neurips_report['exact_duplicate_event_rows']) == 1 and neurips_report['folded_extra_presentation_rows'] == 132
+              assert not neurips_report['database_written'] and not neurips_report['pdfs_downloaded']
+              subprocess.run([python,'-I','-B','-m','paper_atlas','collect','dblp-family','--help'],cwd=outside,check=True)
              subprocess.run([python,'-I','-B','-m','paper_atlas','collect','aamas-proceedings','--year','2025','--snapshots',str(src/'tests/fixtures/aamas2025'),'--out',str(root/'installed-aamas2025')],cwd=outside,check=True)
+              aamas2026 = json.loads(gzip.decompress((src/'tests/fixtures/aamas2026-fixture.json.gz').read_bytes()))
+              aamas2026_snapshots = root/'aamas2026-snapshots'
+              aamas2026_snapshots.mkdir()
+              (aamas2026_snapshots/'contents.html').write_text(aamas2026['contents'])
+              (aamas2026_snapshots/'receipts.json').write_text(json.dumps(aamas2026['receipts']))
+              subprocess.run([python,'-I','-B','-m','paper_atlas','collect','aamas-proceedings','--year','2026','--snapshots',str(aamas2026_snapshots),'--out',str(root/'installed-aamas2026')],cwd=outside,check=True)
+              aamas2026_report = json.loads((root/'installed-aamas2026/aamas-proceedings-2026.json').read_text())
+              assert aamas2026_report['records'] == 639 and len(aamas2026_report['held_publications']) == 1
+              assert aamas2026_report['status_counts'] == {'accepted': 638, 'withdrawn': 1}
+              assert not aamas2026_report['database_written'] and not aamas2026_report['pdfs_downloaded']
              subprocess.run([python,'-I','-B','-m','paper_atlas','collect','iclr-archive','--help'],cwd=outside,check=True)
              subprocess.run([python,'-I','-B','-m','paper_atlas','collect','acl','--help'],cwd=outside,check=True)
+              subprocess.run([python,'-I','-B','-m','paper_atlas','collect','springer-selected','--help'],cwd=outside,check=True)
+              springer = json.loads(gzip.decompress((src/'tests/fixtures/springer-aistats1993-fixture.json.gz').read_bytes()))
+              for group in ('toc','chapters'):
+                  folder = root/'springer-snapshots'/group
+                  folder.mkdir(parents=True)
+                  for name, content in springer[group].items():
+                      (folder/name).write_text(content)
+              subprocess.run([python,'-I','-B','-m','paper_atlas','collect','springer-selected','--toc-snapshots',str(root/'springer-snapshots/toc'),'--chapter-snapshots',str(root/'springer-snapshots/chapters'),'--out',str(root/'installed-springer')],cwd=outside,check=True)
+              springer_report = json.loads((root/'installed-springer/collection-report.json').read_bytes())
+              assert springer_report['records'] == 49 and not springer_report['database_written']
+              springer_manifest, = springer_report['files']
+              assert springer_manifest['status'] == 'partial' and springer_manifest['missing_abstract_count'] == 0
              subprocess.run([python,'-I','-B','-m','paper_atlas','maintain','postgres','--help'],cwd=outside,check=True)
              subprocess.run([python,'-I','-B','-m','paper_atlas','maintain','edition-tracks','--help'],cwd=outside,check=True)
              subprocess.run([python,'-I','-B','-m','paper_atlas','maintain','joint-editions','--help'],cwd=outside,check=True)
@@ -63,6 +96,7 @@ jobs:
              subprocess.run([python,'-I','-B','-m','paper_atlas','audit','public-openreview','--help'],cwd=outside,check=True)
              subprocess.run([python,'-I','-B','-m','paper_atlas','coverage','--help'],cwd=outside,check=True)
              subprocess.run([python,'-I','-B','-m','paper_atlas','audit','histories','--help'],cwd=outside,check=True)
+              subprocess.run([python,'-I','-B','-m','paper_atlas','audit','source-identities','--help'],cwd=outside,check=True)
              subprocess.run([python,'-I','-B','-m','paper_atlas','curate','identifiers','--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()"
@@ -6,8 +6,14 @@ Paper Atlas 将 arXiv、AI 会议论文集及公开投稿决定整理到可追

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

+当前主动补漏从 **2020 年**起；2000 年起已发现的名单、缺口与待办保存在[补漏恢复清单](docs/coverage/backfill-handoff-2026-11-10.md)，下次从这里接续。已有旧数据保留，范围排除不算补齐。
+
+AAMAS 2026 的官方 HTML 目录由 `collect aamas-proceedings --year 2026` 适配，区分 track 和发表形式，keynote 排除、撤稿证据单独保留；[采集与恢复说明](docs/aamas-proceedings.md)。采集不写数据库，仍需身份核验和增量导入。
+
 历史 ICLR 2014–2016 官方录用页由 `collect iclr-archive` 适配，保留主会/workshop及重复展示证据；运行和身份核验边界见 [历史档案说明](docs/coverage/iclr-archive-2014-2016.md)。

+`collect springer-selected` 核验限定的会后选集，分开保存会议年份、出版年份及选集事件身份。目前支持 AISTATS 1993 对应的 1994 年出版选集，原始会议决定仍须单独核查；配置、证据和恢复见 [会后选集说明](docs/springer-selected-volumes.md)。
+
 ## 项目结构

 ACL 1963 的 AMTCL 年会摘要与 1978 的 TINLAP-2 来源关系均按精确卷适配，保留原系列、出版形式和不明逐篇角色；不将 EarlyMT 全系列映射到 ACL，1965 的替代关系也不转成全局别名。来源、排除条目和恢复边界见 [ACL 早期来源核查](docs/coverage/acl-early-sources-2026-11-09.md)。
@@ -64,12 +70,16 @@ paper-atlas collect colm --cache /data/paper-index/raw/http-cache \
  --out /data/paper-index/collected
 ```

+NeurIPS 2026 的[官方节目采集器](docs/neurips-program.md)按 main、evaluations/datasets、position 分开保存，合并同一 forum 的口头与海报展示，保留期刊等未决关系；运行 `paper-atlas collect neurips-program`。它只生成元数据文件，身份核验和生产导入另行执行。
+
 早期主系列的Workshop名称按[ICML/IROS精确届次证据](docs/coverage/historical-primary-series-2026-11-09.md)识别，保留原名与范围证据；代码支持不代表已补齐生产数据。

 EWSL、EPS、ECP、EWSP 的[具体前身卷与完整快照要求](docs/coverage/historical-predecessors-2026-11-09.md)单独核验，保留原始系列及事件/出版年份；旧候选文件缺少 ECP/EWSP 时会明确提示，避免静默遗漏。

 AAMAS 的[官方论文集采集器](docs/aamas-proceedings.md)支持已核验的 2012、2025 版式，可从缓存或已校验快照采集元数据。2025 年目录含 476 条非 keynote 出版物，保留完整作者和机构，HTML 未提供摘要时留空；发表形式与投稿 track 分开，不根据主题推定主会轨道或研究类型。

+旧 DBLP 候选文件可能漏掉新命名空间。`collect dblp-family` 可从已校验的固定压缩快照和完整目录中[恢复指定 family 的原始候选](docs/dblp-family-recovery.md)，例如 `ifaamas`；恢复文件保留卷、keynote 和 workshop 等原始条目，需另做来源范围与身份核验。
+
 `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` 写入主库，整个写入会话只允许一个协调写入者，会议文件失败时整体回滚并保留失败批次记录。
@@ -106,3 +116,5 @@ CI 在已授权的远程 runner 上运行测试、构建 wheel，并从源码目
 整合后的运行版本依赖、交互约束和生产切换顺序见 [运行版本整合](docs/research-runtime.md)。

 已导入整卷的逐条轨道修正见 [edition-tracks 维护流程](docs/edition-tracks.md)：按完整证据生成只读计划，保留原投稿和决定 ID，事务性修正 edition、标识与覆盖，并防止旧整卷刷新覆盖分类。
+
+入库前可使用通用的[来源身份只读预检](docs/source-identity-audit.md)，导出候选论文、稳定标识和既有投稿决定证据；候选结果需要审核，不会自动批准导入。
@@ -1,6 +1,31 @@
 # AAMAS 官方论文集元数据

-`paper_atlas.collectors.aamas` 是按需运行的采集适配器，通过统一入口 `collect aamas-proceedings` 使用。支持已独立核验的 **AAMAS 2012、2025** HTML 版式；其他年份需先核对官方目录和逐篇结构，再增加适配，不能仅替换年份。2025 解析单独封装在 `paper_atlas.collectors.aamas2025`。
+`paper_atlas.collectors.aamas` 是按需运行的采集适配器，通过统一入口 `collect aamas-proceedings` 使用。支持已独立核验的 **AAMAS 2012、2025、2026** HTML 版式；其他年份需先核对官方目录和逐篇结构，再增加适配，不能仅替换年份。2025、2026 解析分别封装在 `paper_atlas.collectors.aamas2025`、`paper_atlas.collectors.aamas2026`。
+
+
+## AAMAS 2026
+
+2026 只读取[官方 HTML 目录](https://www.ifaamas.org/Proceedings/aamas2026/forms/contents.htm)，不下载 PDF。冻结目录 SHA256 `ce2bc7fc5da340930343901506a9ce37c0334d4068f07546431c9deee3f13de9`，643 个条目：keynote 3、research 338、extended abstract 185、AAAI 19、Blue Sky 17、doctoral 34、demo 44、JAAMAS 3。
+
+排除 3 个 keynote；research 的 p238 明确标注 `Retracted June 8, 2026`，完整题名、作者、栏目、日期、原文和来源回执保存在采集报告 `held_publications` 中，**不会输出 accepted 或 withdrawn 投稿记录**。当前决定模型没有独立 retracted 状态，后续须单独建模和核验；原始证据不能删除。任何新增或改变的撤稿标记都会停止采集，等待人工核查。
+
+另一个 extended abstract p3178 明确标注 `Withdrawn June 8, 2026`，输出 `withdrawn`，在 metadata 保留日期 `2026-06-08` 和日精度；原网页没有时区/时刻，不能伪造决定 timestamp，不补造其此前的接受决定或 PDF 链接。因此 639 个输出条目是 **638 accepted、1 withdrawn**，按六个 track 分组：`proceedings_unspecified` 522、`aaai_transfer` 19、`blue_sky` 17、`doctoral` 34、`demo` 44、`jaamas` 3。全文/扩展摘要是发表形式，不是研究类型；research 栏目不能证明每篇原始 main/special 投稿轨道。JAAMAS 保留期刊展示语义，不制造另一次独立审稿录用。
+
+[AAAI 转投详细政策](https://cyprusconferences.org/aamas2026/call-for-papers-aaai-track/) 要求基于 AAAI-26 rejected 稿件；[总览](https://cyprusconferences.org/aamas2026/calls/) 却写 AAAI-25，须保留冲突。采集器保留 AAAI 栏目归属，不据此生成原 AAAI 投稿 ID、年份或逐篇拒稿决定。目录不是原始决定 note，未来关联仍需逐篇证据。
+
+```sh
+paper-atlas collect aamas-proceedings --year 2026 \
+  --snapshots /data/paper-index/collected/aamas2026-primary-evidence-v1 \
+  --out /data/paper-index/collected/aamas2026-normalized-v1
+```
+
+在线模式将 `--snapshots` 换成 `--cache /data/paper-index/cache/aamas2026`。冻结模式 `receipts.json` 的 `receipts` 至少包含官方目录 URL、相对 `file`、`bytes`、`sha256`、`status=200` 和带时区 `finished_at`。拒绝越界路径、越界符号链接、失败状态、哈希/大小不符的页面。这里只读取目录对应回执，其他已冻结政策网页继续保留。
+
+输出六组 JSONL/manifest、原始 HTML 与最后发布的 `aamas-proceedings-2026.json`；统一入口沿用输出目录锁和暂存目录。manifest 全部为 `partial`，639 条 HTML 摘要缺失如实记录。发布编号使用年份和起始页，随机 PDF 文档 ID 单独保留，不能沿用 2025 的 pN.pdf 规则。
+
+首次部署先核对来源哈希和生产状态，再在新输出目录运行采集、独立比对 643 条标题/作者/栏目分区，随后做 PostgreSQL 身份预检与增量导入。采集、远程 CI 通过或目录含某个年份均不代表生产已导入、摘要齐全或最终去重完成。较早年份不自动启动补采，当前主动补漏范围按用户要求从 2020 年起。
+
+## AAMAS 2025

 2025 版式使用[官方首页](https://www.ifaamas.org/Proceedings/aamas2025/)和[出版目录](https://www.ifaamas.org/Proceedings/aamas2025/forms/contents.htm)。原始目录共 479 条：3 keynote、260 research paper、148 extended abstract、12 Blue Sky、25 doctoral、22 Demo 栏目、9 JAAMAS。排除 keynote，输出 476 条；Demo 栏目中的 p3000 是明确标题的 COMP24 竞赛报告，发表年份仍为 2025，格式记 `competition_report`、投稿 track 留作 `proceedings_unspecified`，保留原 Demo 栏目证据。其余输出按 track 分组为 409 unspecified、12 blue_sky、25 doctoral、21 demo、9 jaamas。

@@ -1,5 +1,7 @@
 # 会议覆盖与公开决定核查

+**当前工作范围：2020 年起主动补漏。** 2000 年起已发现资料保留，恢复工作从 [补漏范围与恢复清单（2026-11-10）](backfill-handoff-2026-11-10.md) 开始；下文旧日期快照不改写，较早年份排除不算补齐。
+
 这是 2026-11-08 的首份核查基线，**完整逐年政策普查和最终去重仍未完成**。收录 43 个目标，不能把“有记录”当作“完整覆盖”。结构化清单见 [inventory-2026-11-08.json](inventory-2026-11-08.json)。清单中的 `policy_checks` 均限定所列年份和轨道；空数组表示尚未核实，绝不表示没有公开拒稿。

 ## 快照与复核
@@ -0,0 +1,52 @@
+# 补漏范围与恢复清单（2026-11-10）
+
+用户确定：**后续主动补漏从 2020 年起**。2000 年起已经发现的名单、缺口、资料组合和身份核查记录保留在项目中，供下次接着做；已有旧数据继续保留。停止补采的年份标为范围外或延期，不算作已补齐。
+
+本页是下一次工作的入口，结构化记录见 [backfill-handoff-2026-11-10.json](backfill-handoff-2026-11-10.json)。它保存 43 个会议的 2000–2026 已观察年份、缺年、官方目录文件入口、近期待办与恢复证据。旧的日期快照不改写，也不能当实时数据库。
+
+## 当前范围和年份差集
+
+依据生产只读快照 `2026-11-09 16:26:41.688509+00:00`，原全历史口径为 1266 个目标年份槽位：1208 已观察、43 个历史缺年、14 个 2026 缺年，以及 1 个 SysML 2018 非存档例外。原 135 个历史缺口中有 92 个通过数据补入解决；其余 43 个现在停止补采，不能因此宣称全部解决。
+
+- **2020–2025**：没有整年完全缺记录的目标槽位，但逐篇、track、摘要、公开负面决定仍不完整。
+- **2026**：NeurIPS、ECCV、RecSys、CIKM、ICDM、SDM、IROS、RSS、CoRL、ICAPS、AAMAS、ICASSP、ISMIR、ACM MM，共 14 个槽位。须逐个核查是否已经公开出版；不能将尚未发布的资料说成漏采。
+- **2000–2019 延期线索**：WWW 2000、ICAPS 2001 两个整年缺口；官方历届记录分别见 [WWW](www-editions-2026-11-08.json)、[ICAPS](icaps-editions-2026-11-08.json)。SysML 2018 是非存档首届，见 [MLSys](mlsys-editions-2026-11-08.json)，不能制造正式出版记录。
+- **2000 年前**：剩余 41 个历史缺年停止补采。旧来源文件继续保留作证据，不再作为当前工作队列。
+
+以上只判断目标会议年份是否至少有一条投稿记录，绝不等于目录完整、独立论文数量或最终去重验收。
+
+## 近期资料与恢复位置
+
+服务器采集根目录为 `/data/paper-index/collected`，本地证据备份为 `/Users/zzyang/Documents/Codex/2026-11-06/paper-index`。这些目录提供大体积原始文件，仓库 JSON 提供路径、状态和可用哈希；恢复前先验证文件存在、哈希和生产实际状态。
+
+| 资料 | 当前状态 | 下一步与证据入口 |
+| --- | --- | --- |
+| AAMAS 2026 | 官方目录 643 条，3 个 keynote 排除；640 个非 keynote 中含 1 个明确撤稿。尚未导入 | `aamas2026-primary-evidence-v1/` 保存官方 HTML、回执、逐项目录、审计和校验清单。八个标题差异/撤稿前缀条目需核查。先扩展正式 collector 并远程验证，再做新 PostgreSQL 身份预检。 |
+| CVPR 2026 Findings | 941 条摘要已核验，尚未导入 | `cvf-2026-verified/recovery-audit.json`；原 404 与 RVLF/FVLF 路径更正证据保留。先核对 manifests、track 和实体关系。 |
+| AAMAS 2025 | 476 个非 keynote 来源中，240 已验收、236 待身份核查 | `aamas2025-import225-v1/` 与 `aamas2025-import15-v1/` 是已消费批次，禁止重放。15 条最近验收报告哈希见 JSON。余下资料从当前身份候选与新只读基线继续。 |
+| RSS 2025 | 163 条目录中 162 已验收，`2025/p117` 待核查 | 保留 spinal/RMCC 关系问题；查看工作状态键 `rss2025_held7_primary_evidence_v1`，不能重跑已消费的 6/88/54/14 批次。 |
+| 近期公开负面决定首批 | 1625 条已验收：1624 rejected、1 withdrawn | `public-negative-postgresql-import-postcheck-v1.json`；仅该批次验收，不是公开拒稿全集，不要重放。 |
+
+AAMAS 2026 官方目录分区：research 338、extended abstracts 185、AAAI 19、Blue Sky 17、doctoral 34、demo 44、JAAMAS 3，另有 keynote 3。JAAMAS 是期刊展示关系，不能凭目录再造一次独立审稿录用。官方撤稿条目在 research 分区，不能把全目录都标为当前 accepted。
+
+AAAI 转投政策有年份冲突：[详细征稿页](https://cyprusconferences.org/aamas2026/call-for-papers-aaai-track/) 指 AAAI-26 rejected，[总览](https://cyprusconferences.org/aamas2026/calls/) 写 AAAI-25。保留两份证据；19 个 AAAI track 条目不能直接转成 19 条有原始 note 的拒稿决定。原投稿 ID、逐篇决定和版本关系仍需核实。
+
+## 2000–2019 已发现但延期的身份与名单工作
+
+- **AAMAS 2012**：官方 334 条，319 已独立验收、15 个身份条目保留待查。验收报告 `aamas2012-import-v1/postgresql-postcheck-v1.json`，哈希在 JSON；已验收 319 条及被替代的 316 条计划均不得重放。来源布局与用法见 [AAMAS 适配说明](../aamas-proceedings.md)。
+- **UAI 2008**：官方 programme 与归档资料确认了四篇独立论文，但部分原始 DBLP 键的标题、作者、页码、出版链接对应仍有冲突。JSON 保留四组候选、arXiv ID、官方题名及归档报告哈希。之前仅根据页码推断题名/作者错误的结论已撤回，未应用纠正；恢复时不得据此覆盖或合并。
+- **其他名单**：本目录全部 `*-editions-2026-11-08.json` 保留各会议官方年份及来源；[历史缺口](historical-gaps-2026-11-09.md)、[来源候选](historical-source-candidates-2026-11-09.json)、[来源范围](historical-source-scope-2026-11-09.md)、[主系列](historical-primary-series-2026-11-09.md)、[前身会议](historical-predecessors-2026-11-09.md)、[ICLR 档案](iclr-archive-2014-2016.md)继续留档。这些旧报告的“缺失”可能已被后续导入解决，恢复时必须对照当前状态，不能直接当可执行队列。
+
+## 公开拒稿资料不能遗漏
+
+结构化清单保留最近已查的政策 URL、限定年份和可用性结论。优先继续 UAI 2024–2026 的 opt-in 公开拒稿来源、EMNLP 2023 的公开集合范围及其他目标的逐年核查；平台使用 OpenReview 不能证明 rejected 公开。访问受限如实记录，不绕过 Challenge。
+
+保留 rejected、desk rejected、withdrawn、retracted、resubmit 的区别；评审建议不等于最终决定。COLM 2026 的不公开政策只适用于已核实年份，不能推广到其他会议。PaperCopilot 与原 OpenReview、镜像间有重叠，候选计数不能相加。论文先被 A 拒绝再被 B 录用时，两次投稿、决定、来源与证据全部保留。
+
+## 下次启动顺序
+
+1. 读取本页和 JSON，再核实服务器工作状态、明确的 systemd invocation 与终态报告。权威工作状态 `/Users/zzyang/Documents/Codex/2026-11-06/paper-index/work-state.json` 中部分旧字段会过期；以最新验收和实际查询为准。
+2. 当前主动范围为 2020 起，2000–2019 只恢复线索，不自动启动其采集。先检查新来源是否已导入，验收过的队列禁止重放。
+3. 使用现有 `foundry-postgres` 的 `foundry_data.research`。当前已验证 writer 为提交 `714f745bd5c3fbc742be1efae0b31ab37a55e21f` 的固定 release；恢复时重新确认，不使用旧 `/data/paper-index/code` 替代。服务配置仅由服务器受保护文件读取，不复制密码。
+4. 对待导入资料做新只读身份预检，处理稳定标识、短标题、作者/版本差异、track 与跨会历史，再生成小范围增量队列。每批独立验收，SQLite 只读备份保持保护。
+5. 编译、测试或 wheel 构建仅在已确认 `ces-control:host` runner 执行。代码按工作分支交付 PR，等待用户手动合并。范围内导入结束后仍须全库去重、近似标题/作者冲突、来源-年份-track、摘要、约束、实际全文查询验收；不能用单批导入或 CI 成功替代。
@@ -0,0 +1,37 @@
+# DBLP family 候选恢复
+
+DBLP 的会议命名空间会变化。旧候选过滤器包含 `atal`、`aamas`，遗漏了 `ifaamas`；2026-11-01 固定快照的完整目录实际已有 `conf/ifaamas/2025` 和 `conf/ifaamas/2026`。新目录盘点已把 `ifaamas` 纳入广泛候选范围，已有快照也可按需恢复，无需重新下载或重建完整目录。
+
+## 运行
+
+在已有元数据的服务器运行：
+
+```sh
+paper-atlas collect dblp-family \
+  --archive /data/paper-index/raw/dblp-2026-11-01.xml.gz \
+  --dtd /data/paper-index/raw/dblp-2023-06-28.dtd \
+  --catalogue /data/paper-index/collected/dblp/dblp-catalogue.sqlite \
+  --family ifaamas \
+  --out /data/paper-index/collected/ifaamas-recovered
+```
+
+`--family` 可以重复，但仅接受项目明确列出的候选 family。`--out` 必须尚不存在。同名输出由文件锁防止并发恢复；任务运行中应观测原任务，不能因为日志间隔长就重复启动。完整扫描会定期打印进度。
+
+命令重新验证固定压缩包 SHA256、DTD MD5 与大小，并只读打开目录。目录旁的 `dblp-ai-candidates.manifest.json` 必须属于同一个快照和 DTD，包含完整记录类型计数。旧候选文件不作为完整性依据，也不会被改写。
+
+恢复扫描使用项目原有 XML 流式解析器，仅允许已校验的本地 DTD，不联网读取 XML 实体。最终会比对完整快照的记录总数、逐 family 数量和全部选中卷的原始元数据。全部一致后才原子发布输出目录；失败时不发布半成品，也不覆盖既有文件。排查明确终止的失败任务后，可用尚不存在的输出目录重新运行。
+
+## 输出与后续流程
+
+- `dblp-family-candidates.jsonl`：原始 `inproceedings` 和 `proceedings`，保留书目 key、原始作者、标题、年份、卷引用、出版链接等字段。
+- `recovery-manifest.json`：快照、DTD、目录凭据哈希，完整扫描数量、逐 family 数量、选中卷、旧候选文件的已知数量及输出哈希。状态为 `recovered_raw_candidates_pending_scope_review`。
+
+恢复结果没有新增规范化的 `venue`、`track` 或决定标签，也不是 `import records` 的输入队列。`ifaamas` 的整卷可能混有研究论文、扩展摘要、keynote、博士生、演示和 JAAMAS 展示。目录过滤修复不会把这些条目统一标成 main；规范化的范围映射仍需第一方逐栏目证据。
+
+后续先用官方目录核对标题、页码、作者及发表形式，再查数据库稳定标识和候选实体，保留每次投稿、决定和标签来源。只有审核后的新增分区才能原生导入并独立验收；已验收的来源记录不能重放。单个年份有记录不代表逐篇、track、摘要或公开拒稿覆盖完整。
+
+## 2026-11-09 核查实例
+
+对既有固定快照的独立原始元数据恢复发现 `ifaamas` 有 1,122 条出版物候选及 2 卷：2025 年 479 条、2026 年 643 条。2025 年的 479 条标题与起始页均对应官方目录，其中 3 条 keynote 应排除，余下 476 条出版物需逐项处理。该数字包含已经验收的 225 条来源，不是 1,122 篇可新增论文。
+
+此实例说明过滤遗漏的影响，不代表恢复命令已完成生产导入。2026 年仍需核验官方出版清单、栏目、身份及逐次决定；缺少录用记录不能推断为拒稿。最终全库去重、覆盖和查询审计继续单独进行。
@@ -0,0 +1,56 @@
+# NeurIPS 官方节目元数据
+
+`paper-atlas collect neurips-program` 是按需运行的采集器，当前核验范围是 2026 年的[官方 orals/posters JSON](https://neurips.cc/static/virtual/data/neurips-2026-orals-posters.json)。它只请求这个固定公开元数据文件，不连接数据库，不请求 OpenReview API，不下载 PDF、视频或个人资料页。模块位于 `paper_atlas.collectors.neurips_program`，通过统一 CLI 使用。
+
+## 已核验快照
+
+2026-11-09T17:57:40Z 冻结的原始响应共 20,447,021 字节，SHA256 `879293a863057fa05370443b3b5ddee12a10eeb37c810abc6c1d53f71f545997`。原始 `count=9276` 与结果数组长度一致，`next`、`previous` 均为空；这只证明此份公开文件已完整取得，不是全部会议出版物或公开拒稿全集。
+
+按官方逐条 `sourceurl` 中的明确投稿来源分开处理：
+
+| 原始 OpenReview group | 数据库 track | 不同 forum 数 |
+| --- | --- | ---: |
+| `NeurIPS.cc/2026/Conference` | `main` | 7896 |
+| `NeurIPS.cc/2026/Evaluations_and_Datasets_Track` | `evaluations_datasets` | 971 |
+| `NeurIPS.cc/2026/Position_Paper_Track` | `position` | 137 |
+| 合计 | | 9004 |
+
+只接受已核验的显式 `Accept`、`Accept (poster/spotlight/oral)` 标签，保留原始大小写和文本。未知来源、决定、展示类型、可见性或无法验证的 forum 链接进入 `held_presentations`，不推成 accepted 或 rejected。空标题、缺少作者、相同 event ID 不同内容、同一 forum 的题名/作者/来源/决定冲突会停止采集。
+
+原 9276 行中，event 154866 有一条内容完全相同的重复行，单独保留其原始行号；另外 132 个口头展示与同 forum 的海报合并为一条投稿候选，完整保留两种展示及时间、场次、原始带符号 `sourceid`。负数 `sourceid` 不能被解释为拒稿或另一次投稿，也不转换为全局论文标识。
+
+其余 139 行单独留档：87 个期刊展示、36 个 Bridging the Future 展示、16 个未提供明确投稿来源的条目。这些不因缺少链接而被推为拒稿，也不能被冒充为独立主会审稿录用。后续须核实原期刊发表、认证、复现或其他展示关系。
+
+9004 条是不同 forum 的来源记录，不是已经新增 9004 篇独立科学论文。不同 forum 即使题名和作者相同也保留各自投稿；跨会及版本关系另行核验。题名、完整作者顺序和机构取自原始 JSON，口头与海报不是研究类型；展示标签独立使用 `neurips_program:presentation_format` namespace。节目中的未来展示时间不冒充决定时间。
+
+## 运行与恢复
+
+在线模式只请求一个固定元数据 URL，并复用受限大小的响应缓存：
+
+```sh
+paper-atlas collect neurips-program \
+  --cache /data/paper-index/cache/neurips2026-program \
+  --out /data/paper-index/collected/neurips2026-program-YYYYMMDD
+```
+
+冻结模式只读 `receipts.json` 对应的原始文件：
+
+```sh
+paper-atlas collect neurips-program \
+  --snapshots /data/paper-index/collected/neurips2026-program-probe-v1 \
+  --out /data/paper-index/collected/neurips2026-program-normalized-v1
+```
+
+快照回执包含精确 `url`、相对 `file`、`bytes`、`sha256`、`status=200` 和带时区的 `finished_at`。拒绝越界路径/符号链接、重复回执、失败响应、大小或哈希不符。在线模式沿用项目 HTTP 缓存和限速，验证回执时间后保存原始 JSON。输出目录有互斥锁，完成报告最后发布；已有完成报告时要求新建输出目录，避免覆盖后误重放。
+
+输出三个 track 的 JSONL/manifest、原始 JSON 证据和 `neurips-program-2026.json`。manifest 始终为 `partial`；原始节目 JSON 没有摘要，9004 条 `abstract=null` 和缺失原因如实保存，不能用别篇摘要填充。命令终端只输出数量与路径，完整待处理展示与原始行号在报告中。
+
+恢复时先校验文件和回执，使用同一完整快照生成新输出；若已有完成报告或导入回执，先核实实际来源绑定和 checkpoint，不重放已消费队列。生成文件后还需 PostgreSQL 身份预检、关联审核、增量队列、原生导入和独立验收；源文件成功、CI 通过均不能替代生产导入或最终全库去重。
+
+## 目录差异与公开拒稿限制
+
+同期[官网 Downloads 页面](https://neurips.cc/Downloads/2026)的活动总数不是本 JSON 的论文总数。逐个 poster ID/题名对照后，网页另有 event 155165（Lemon）未出现在本 JSON 中，须单独取得并审核元数据，不能补造其 track 或 forum。16 个未提供来源的条目在 JSON 中使用通用 session URL，而网页有各自 poster URL；原值保留，不能因 URL 映射差异制造新论文。
+
+[2026 Main Track Handbook](https://neurips.cc/Conferences/2026/MainTrackHandbook)明确允许 rejected 作者选择公开。当前无账号公开 notes API 请求返回 `ChallengeRequiredError`、HTTP 403，失败原文和回执保留，不绕过验证；因此不能把本 accepted 节目文件当作拒稿数据源或据此推断拒稿是否存在。官网下载表单的 JSON 导出也曾超时，未取得导出内容，已改用独立公开静态节目文件并保留失败证据，不重试超时导出。
+
+当前主动补漏范围从 2020 年起。期刊/展示关系、网页独有条目、摘要和公开负面决定仍有待办；冻结资料和代码适配不代表完整会议覆盖。
@@ -0,0 +1,21 @@
+# 来源身份只读预检
+
+`audit source-identities` 在 PostgreSQL 的同一 `REPEATABLE READ READ ONLY` 快照中，检查规范化 JSONL 的来源绑定、稳定标识及精确标题候选，并导出候选论文的原始来源和投稿决定历史。用于入库前评审，不执行导入、合并或标识修改。
+
+```sh
+PGSERVICEFILE=/etc/paper-atlas/pg_service.conf paper-atlas audit source-identities \
+  --pg-service foundry-research --pg-schema research \
+  --files /data/paper-index/collected/example/main.jsonl \
+          /data/paper-index/collected/example/position.jsonl \
+  --out /data/paper-index/collected/example-identity-preflight-v1
+```
+
+每个 JSONL 必须有同名 `.manifest.json`，包含 `source`、`venue`、`year`、`track`、`count` 和 `file_sha256`。逐行校验届次、显式决定状态、标识类型和跨文件来源主键唯一性；文件哈希和数量必须相符。只读取调用者指定的文件，不跟随 manifest 中的绝对 `file` 路径；原始行哈希包含换行符。原始 JSONL 和 manifest 应与报告一起保留。
+
+输出目录必须不存在，防止复用旧报告或覆盖证据。失败会保留 `failed.json` 及已完成分片，不能把部分输出当成功；排查后使用新的输出目录重新核查。成功报告 `postgresql-preflight.json` 最后原子写入，记录输入哈希、数据库快照和每条候选。实体详情存为 `entities/<分片>/<paper_id>.json.gz`，报告中列出压缩文件的 SHA-256；包含完整 paper、标识和限量原始来源、投稿、决定及出处。每篇最多保留 100 条来源、200 次投稿，每次投稿最多 200 条决定，每组决定或投稿出处最多 100 条，所有截断均显式标记。精确标题最多取 21 个候选；达到上限即标记，不能据此推算总数。这里没有计算全表精确数量。
+
+候选分类用于安排审核：既有来源记录需要核对并避免重放，多个绑定冲突必须暂缓；单个 DOI、arXiv 或 OpenReview 绑定也可能错误，必须核对原文、标题、作者和历史。作者比较保留完整规范化姓名的重复次数，但姓名相同不能证明是同一个人。未找到精确标题或标识不能证明是新作品，仍需近似标题、版本和跨会历史核查。
+
+所有条目固定为 `not_approved_for_import`，批准数量为 0。不要直接把此报告当成 curation 计划或导入队列。后续获批导入前应重新检查实时来源绑定；来源摘要缺失不得覆盖已有摘要，旧拒稿与新录用应作为不同投稿经历保留。本命令不能代替所有导入结束后的全库去重、覆盖和约束验收。
+
+构建与测试仅在已确认的 `ces-control:host` 远程 runner 执行。集成测试使用已有 PostgreSQL 服务做真实只读候选查询，不往生产库添加测试记录。
@@ -0,0 +1,36 @@
+# 会后选集的元数据采集
+
+`collect springer-selected` 读取已冻结、带 HTTP 回执的公开 HTML，不访问 PDF、不写业务库。目前仅支持核验过的 *Selecting Models from Data: Artificial Intelligence and Statistics IV*，书籍 DOI 为 `10.1007/978-1-4612-2660-4`。增加其他卷前必须另行核对会议关系、目录、年份和字段格式。
+
+[会议官方旧站](https://www.gatsby.ucl.ac.uk/aistats/workshops.html)区分现场分发的 Preliminary Papers 与之后修订、再次评审的选集。[出版社卷说明](https://link.springer.com/book/10.1007/978-1-4612-2660-4)将本书关联到 1993 年 1 月的第四届 workshop，书籍出版年份为 1994。三页目录共有 49 个章节 DOI；分区 Front Matter 不作为章节论文导入。
+
+规范记录保留：
+
+- 会议年份 1993、独立 track `selected_volume` 与事件标识 `aistats-1993-selected-volume-1994`；出版年份单独记为 1994。
+- 章节本身的 DOI、作者原始引用写法及顺序、姓名首字母、标题、摘要、页码和逐页原始证据。不扩写作者首字母。
+- 明确提供本章 `publication_doi` 投稿出版标识，不把它转移到旧的原始会议评审记录，也不因短标题相同而强制合并实体。
+- `springer_selected_volume:publication_format` 标签与独立来源 namespace。
+- `accepted` 仅表示章节已列入这本出版选集，原始 1993 投稿评审决定、决定日期和全部录用清单均未知。后续原会议 main、拒稿或撤稿证据可以独立保留。
+
+不能据此宣称会议全部录用、全部投稿或公开拒稿覆盖完整，也不能把出版社机器添加的关键词视为作者标签。未命中选集不等于被拒。
+
+## 按需运行与恢复
+
+输入分别是卷目录/旧站说明和章节页的快照目录。每个目录的 `receipts.json` 包含 `responses` 列表，每条保留 `url`、`final_url`、`file`、`http_status`、`headers`、`bytes`、`sha256`、`at`。需要三页目录、官方旧站说明及全部 49 个章节页面。
+
+```sh
+paper-atlas collect springer-selected \
+  --toc-snapshots /data/paper-index/collected/aistats1993-book-evidence-v1 \
+  --chapter-snapshots /data/paper-index/collected/aistats1993-chapter-evidence-v1 \
+  --out /data/paper-index/collected/aistats1993-source-v1
+```
+
+采集器验证回执和内容哈希、文件路径、HTTP 类型、目录/章节 DOI 和标题、作者格式、出版年份及页码。Springer 对公开元数据返回的同路径 `cookies_not_supported` 诊断参数会原样保留；不接受其他域、路径、分页变化或未知查询参数。官方旧站说明固定哈希为 `6db205b9dbc82b4a096556efd2aa67e065cad8e0bd6351176b1962ffd1fca179`，变化时须重新审核。
+
+全部文件先核验，输出使用排他锁和暂存目录，完成报告最后发布。已有输出拒绝覆盖；发现不完整或已消费文件，应先查运行状态与报告，另建明确版本目录，不能重放生产导入。
+
+输出 JSONL 配套 manifest 始终为 `partial`。收集完成后还必须做 PostgreSQL 稳定标识/标题/作者身份预检、排除已导入记录、准备限定增量文件、保存旧数据保护基线、原生导入和独立验收。49 条来源不代表 49 个新增论文实体，全部来源处理后仍须全库去重审计。
+
+## 固定测试资料
+
+`tests/fixtures/springer-aistats1993-fixture.json.gz` 保存 4 份范围/目录 HTML、49 份章节 HTML、原始请求回执及独立原始字段审计的预期元数据，SHA256 为 `94028fd37080f9d8027e5790e379cc151ee6eab17ed0d1dcc0f59e6e75f9d4d0`。摘要按 HTML 段落边界分隔；未修改原始 HTML 或回执。它是压缩的公开元数据测试资料，不包含论文 PDF。测试覆盖全部真实章节字段、缺失和漂移输入、来源安全边界、幂等导入及旧拒稿历史保留。构建和测试只在已授权远程 runner 执行。
@@ -0,0 +1,252 @@
+"""Read-only, evidence-preserving identity candidates for normalized source files.
+
+A candidate is never permission to import or merge. Absence of an exact match is
+never proof of a new work. Run again against a fresh snapshot before curation.
+"""
+import argparse
+from collections import Counter
+import gzip
+import hashlib
+import json
+from pathlib import Path
+
+from paper_atlas.db.store import author_keys, normalized
+from paper_atlas.db.submissions import timestamp
+from paper_atlas.postgres.client import ReadConnection
+
+STATUSES = {'accepted', 'rejected', 'desk_rejected', 'withdrawn', 'under_review', 'unknown'}
+
+
+def digest(data):
+    return hashlib.sha256(data).hexdigest()
+
+
+def authors_multiset(authors):
+    # author_keys intentionally removes duplicates; identity review must retain them.
+    return Counter(key for author in authors for key in author_keys([author]))
+
+
+def load_inputs(files):
+    records, inputs, seen = [], [], set()
+    for filename in files:
+        path = Path(filename).resolve()
+        manifest_path = path.with_suffix('.manifest.json')
+        raw_manifest = manifest_path.read_bytes()
+        manifest = json.loads(raw_manifest)
+        hasher, count = hashlib.sha256(), 0
+        with path.open('rb') as stream:
+            for line_number, line in enumerate(stream, 1):
+                hasher.update(line)
+                row = json.loads(line)
+                for key in ('source', 'external_id', 'title', 'venue', 'track'):
+                    if not isinstance(row.get(key), str) or not row[key].strip():
+                        raise ValueError(f'Invalid {key}: {path.name}:{line_number}')
+                if type(row.get('year')) is not int or row.get('status') not in STATUSES:
+                    raise ValueError('Invalid year or explicit source status')
+                for key in ('source', 'venue', 'year', 'track'):
+                    if row[key] != manifest.get(key):
+                        raise ValueError(f'Manifest scope mismatch: {key}')
+                authors = row.get('authors')
+                if not isinstance(authors, list) or any(not isinstance(a, (str, list, dict)) for a in authors):
+                    raise ValueError('Invalid authors')
+                identifiers = row.get('identifiers', {})
+                if not isinstance(identifiers, dict) or any(
+                        not isinstance(k, str) or not k.strip() or not isinstance(v, str) or not v.strip()
+                        for k, v in identifiers.items()):
+                    raise ValueError('Identifiers must contain singular nonempty strings')
+                key = (row['source'], row['external_id'])
+                if key in seen:
+                    raise ValueError('Duplicate source/external_id across inputs')
+                seen.add(key)
+                record = {k: row[k] for k in ('source', 'external_id', 'title', 'authors',
+                                              'venue', 'year', 'track', 'status')}
+                record.update(identifiers=identifiers, file=str(path), line=line_number,
+                              raw_line_sha256=digest(line), source_record_hits=[],
+                              identifier_hits=[], title_candidate_ids=[])
+                records.append(record)
+                count += 1
+        if not count or count != manifest.get('count') or hasher.hexdigest() != manifest.get('file_sha256'):
+            raise ValueError('Input count/hash mismatch or empty input')
+        inputs.append({'file': str(path), 'file_sha256': hasher.hexdigest(), 'count': count,
+                       'manifest_sha256': digest(raw_manifest), 'manifest': manifest})
+    if not records:
+        raise ValueError('At least one nonempty input is required')
+    return records, inputs
+
+
+def _lookup(connection, values, columns, lateral):
+    """Bound scalar VALUES allow the existing text codec to encode each key."""
+    width = len(columns)
+    for start in range(0, len(values), 100):
+        batch = values[start:start + 100]
+        sql = 'WITH keys(' + ','.join(columns) + ') AS (VALUES ' + ','.join(
+            '(' + ','.join(['?'] * width) + ')' for _ in batch) + ') SELECT keys.n,hit.* FROM keys JOIN LATERAL (' + lateral + ') hit ON true'
+        for row in connection.execute(sql, tuple(value for item in batch for value in item)):
+            yield dict(row)
+
+
+def find_candidates(connection, records):
+    source_keys = [(i, row['source'], row['external_id']) for i, row in enumerate(records)]
+    for hit in _lookup(connection, source_keys, ('n', 'source', 'external_id'), '''
+            SELECT r.id,r.paper_id FROM sources s JOIN source_records r ON r.source_id=s.id
+            WHERE s.name=keys.source AND r.external_id=keys.external_id'''):
+        records[hit.pop('n')]['source_record_hits'].append(hit)
+    identifiers = []
+    for i, row in enumerate(records):
+        pairs = {(row['source'], row['external_id']), *row['identifiers'].items()}
+        identifiers.extend((i, scheme, value) for scheme, value in sorted(pairs))
+    for hit in _lookup(connection, identifiers, ('n', 'scheme', 'value'), '''
+            SELECT scheme,value,paper_id FROM identifiers
+            WHERE scheme=keys.scheme AND value=keys.value'''):
+        records[hit.pop('n')]['identifier_hits'].append(hit)
+    title_keys = [(i, normalized(row['title'])) for i, row in enumerate(records)]
+    for hit in _lookup(connection, title_keys, ('n', 'title_key'), '''
+            SELECT id AS paper_id FROM papers WHERE title_key=keys.title_key ORDER BY id LIMIT 21'''):
+        records[hit['n']]['title_candidate_ids'].append(hit['paper_id'])
+
+
+def entity_packet(connection, paper_id):
+    paper = connection.execute('SELECT * FROM papers WHERE id=?', (paper_id,)).fetchone()
+    if paper is None:
+        raise RuntimeError('Candidate disappeared from read-only snapshot')
+    sources = [dict(r) for r in connection.execute('''SELECT r.*,s.name AS source
+        FROM source_records r JOIN sources s ON s.id=r.source_id
+        WHERE r.paper_id=? ORDER BY r.id LIMIT 101''', (paper_id,))]
+    attempts = [dict(r) for r in connection.execute('''SELECT s.*,v.acronym AS venue,e.year,e.track,e.event_key
+        FROM submissions s JOIN editions e ON e.id=s.edition_id JOIN venues v ON v.id=e.venue_id
+        WHERE s.paper_id=? ORDER BY s.id LIMIT 201''', (paper_id,))]
+    packet = {'paper': dict(paper), 'identifiers': [dict(r) for r in connection.execute(
+        'SELECT * FROM identifiers WHERE paper_id=? ORDER BY scheme,value', (paper_id,))],
+        'source_records': sources[:100], 'source_records_truncated': len(sources) > 100,
+        'attempts': attempts[:200], 'attempts_truncated': len(attempts) > 200}
+    for attempt in packet['attempts']:
+        observations = [dict(r) for r in connection.execute(
+            'SELECT * FROM decisions WHERE submission_id=? ORDER BY id LIMIT 201', (attempt['id'],))]
+        attempt['decisions'], attempt['decisions_truncated'] = observations[:200], len(observations) > 200
+        provenance = [dict(r) for r in connection.execute('''SELECT ss.*,r.external_id,s.name AS source
+            FROM submission_sources ss JOIN source_records r ON r.id=ss.source_record_id
+            JOIN sources s ON s.id=r.source_id WHERE ss.submission_id=?
+            ORDER BY ss.source_record_id,ss.source_submission_id LIMIT 101''', (attempt['id'],))]
+        attempt['sources'], attempt['sources_truncated'] = provenance[:100], len(provenance) > 100
+        for decision in attempt['decisions']:
+            provenance = [dict(r) for r in connection.execute('''SELECT ds.*,r.external_id,s.name AS source
+                FROM decision_sources ds JOIN source_records r ON r.id=ds.source_record_id
+                JOIN sources s ON s.id=r.source_id WHERE ds.decision_id=?
+                ORDER BY ds.source_record_id,ds.source_decision_id LIMIT 101''', (decision['id'],))]
+            decision['sources'], decision['sources_truncated'] = provenance[:100], len(provenance) > 100
+    return packet
+
+
+def classify(record, entities):
+    anchors = {r['paper_id'] for r in record['source_record_hits'] + record['identifier_hits']}
+    titles = record['title_candidate_ids']
+    flags = []
+    own_authors = authors_multiset(record['authors'])
+    if not own_authors:
+        flags.append('missing_source_authors')
+    if any(n > 1 for n in own_authors.values()):
+        flags.append('duplicate_normalized_author_names')
+    if len(normalized(record['title'])) < 25:
+        flags.append('short_title')
+    comparisons = []
+    for pid in sorted(anchors | set(titles)):
+        entity = entities[pid]
+        other = authors_multiset(entity['authors'])
+        comparisons.append({'paper_id': pid,
+            'normalized_title_equal': normalized(record['title']) == normalized(entity['title']),
+            'nonempty_author_multiset_equal': bool(own_authors) and own_authors == other,
+            'author_name_overlap': sorted(own_authors.keys() & other.keys())})
+    if len(anchors) > 1:
+        kind = 'conflicting_existing_bindings_hold'
+    elif record['source_record_hits']:
+        kind = 'existing_source_record_reconcile_no_replay'
+    elif anchors:
+        kind = 'existing_identifier_binding_requires_review'
+    elif len(titles) >= 21:
+        kind = 'title_candidate_limit_reached_hold'
+    elif not titles:
+        kind = 'no_exact_match_not_proof_of_novelty'
+    elif len(titles) > 1:
+        kind = 'multiple_exact_title_entities_hold'
+    elif comparisons[0]['nonempty_author_multiset_equal']:
+        kind = 'one_exact_title_author_multiset_requires_review'
+    elif comparisons[0]['author_name_overlap']:
+        kind = 'partial_author_overlap_requires_review'
+    else:
+        kind = 'no_author_agreement_hold'
+    return {'kind': kind, 'flags': flags, 'comparisons': comparisons,
+            'title_candidates_truncated': len(titles) >= 21,
+            'decision': 'not_approved_for_import'}
+
+
+def _atomic_json(path, value):
+    temporary = path.with_suffix(path.suffix + '.part')
+    with temporary.open('w', encoding='utf-8') as stream:
+        json.dump(value, stream, ensure_ascii=False, indent=2)
+        stream.write('\n')
+    temporary.replace(path)
+
+
+def audit(files, out, *, pg_service, pg_schema):
+    # Reject input corruption before opening a database connection or reserving output.
+    records, inputs = load_inputs(files)
+    out = Path(out)
+    out.mkdir(parents=True, exist_ok=False)
+    (out / 'audit.lock').write_text(timestamp() + '\n')
+    connection = None
+    try:
+        connection = ReadConnection(pg_service, pg_schema)
+        connection.execute("SET LOCAL statement_timeout='60s'")
+        if connection.execute('SHOW transaction_read_only').fetchone()[0] != 'on':
+            raise RuntimeError('A read-only transaction is required')
+        snapshot = dict(connection.execute('''SELECT current_database() AS database,current_schema() AS schema,
+            transaction_timestamp()::text AS transaction_at,pg_current_snapshot()::text AS snapshot''').fetchone())
+        find_candidates(connection, records)
+        ids = sorted({hit['paper_id'] for row in records for hit in row['source_record_hits'] + row['identifier_hits']} |
+                     {pid for row in records for pid in row['title_candidate_ids']})
+        entities, index = {}, {}
+        for number, pid in enumerate(ids, 1):
+            packet = entity_packet(connection, pid)
+            relative = Path('entities') / str(pid // 10000) / f'{pid}.json.gz'
+            target = out / relative
+            target.parent.mkdir(parents=True, exist_ok=True)
+            raw = gzip.compress(json.dumps(packet, ensure_ascii=False).encode(), mtime=0)
+            target.write_bytes(raw)
+            index[str(pid)] = {'file': relative.as_posix(), 'sha256': digest(raw)}
+            entities[pid] = {'title': packet['paper']['title'],
+                             'authors': json.loads(packet['paper']['authors_json'])}
+            if number % 250 == 0 or number == len(ids):
+                print(json.dumps({'phase': 'entity_evidence', 'completed': number, 'total': len(ids)}), flush=True)
+        kinds = Counter()
+        for row in records:
+            row.update(classify(row, entities))
+            kinds[row['kind']] += 1
+        report = {'status': 'completed_read_only_candidates_not_import_approval',
+                  'started_snapshot': snapshot, 'completed_at': timestamp(), 'transaction_read_only': True,
+                  'database_written': False, 'approved_for_import': 0, 'records': len(records),
+                  'kind_counts': dict(kinds), 'inputs': inputs, 'entities': index, 'entries': records,
+                  'limitations': ['Exact keys only; no match does not establish a new work.',
+                      'Name agreement does not prove person or scientific-work identity.',
+                      'Bound identifiers can be incorrect; review titles, authors and original evidence.',
+                      'Source/history limits are explicit in compressed entity packets.',
+                      'Not a final whole-database deduplication or coverage audit.']}
+        _atomic_json(out / 'postgresql-preflight.json', report)
+        return report
+    except Exception as error:
+        _atomic_json(out / 'failed.json', {'status': 'failed_no_import_approval',
+                     'error_type': type(error).__name__, 'database_written': False})
+        raise
+    finally:
+        if connection is not None:
+            connection.close()
+
+
+def main(argv=None, *, prog=None):
+    parser = argparse.ArgumentParser(prog=prog, description=__doc__)
+    parser.add_argument('--files', nargs='+', required=True, type=Path)
+    parser.add_argument('--pg-service', required=True)
+    parser.add_argument('--pg-schema', required=True)
+    parser.add_argument('--out', required=True, type=Path)
+    args = parser.parse_args(argv)
+    report = audit(args.files, args.out, pg_service=args.pg_service, pg_schema=args.pg_schema)
+    print(json.dumps({k: report[k] for k in ('status', 'records', 'kind_counts', 'approved_for_import')}, ensure_ascii=False))
@@ -13,6 +13,7 @@ COMMANDS = {
    ('download',): ('paper_atlas.io.snapshots', {}, '下载并校验固定版本快照'),
    ('collect','pmlr'): ('paper_atlas.collectors.pmlr', {}, 'PMLR 官方元数据'),
    ('collect','acl'): ('paper_atlas.collectors.acl', {}, 'ACL Anthology XML'),
+    ('collect','springer-selected'): ('paper_atlas.collectors.springer_selected', {}, '核验限定的会后选集章节与原会议关系'),
    ('collect','iclr-archive'): ('paper_atlas.collectors.iclr_archive', {}, '核对ICLR 2014–2016官方历史录用档案'),
    ('collect','iclr-program'): ('paper_atlas.collectors.iclr', {}, 'ICLR 官方公开程序'),
    ('collect','iclr-snapshot'): ('paper_atlas.collectors.openreview', {}, 'ICLR 公开投稿与决定快照'),
@@ -25,8 +26,10 @@ COMMANDS = {
    ('collect','cvf'): ('paper_atlas.collectors.vision', {'source':'cvf'}, 'CVPR / ICCV 官方开放目录'),
    ('collect','ecva'): ('paper_atlas.collectors.vision', {'source':'ecva'}, 'ECCV 官方开放目录'),
    ('collect','aamas-proceedings'): ('paper_atlas.collectors.aamas', {}, '按官方目录区分 AAMAS 发表形式和摘要'),
+    ('collect','neurips-program'): ('paper_atlas.collectors.neurips_program', {}, '按公开节目来源区分 NeurIPS 2026 投稿与重复展示'),
    ('collect','www-archive'): ('paper_atlas.collectors.www_archive', {}, '核验早期WWW官方档案并分开待核查条目'),
    ('collect','dblp'): ('paper_atlas.collectors.dblp_records', {}, '核对并整理 DBLP 历史发表记录'),
+    ('collect','dblp-family'): ('paper_atlas.collectors.dblp_recovery', {}, '从固定快照恢复指定 DBLP family 的原始候选'),
    ('catalogue','dblp'): ('paper_atlas.collectors.dblp', {}, '盘点 DBLP 历史会议目录'),
    ('import','arxiv'): ('paper_atlas.ingestion.arxiv', {}, '导入 arXiv 元数据 ZIP'),
    ('import','records'): ('paper_atlas.ingestion.records', {}, '导入核验过的会议文件'),
@@ -43,6 +46,7 @@ COMMANDS = {
    ('audit','arxiv'): ('paper_atlas.audits.arxiv', {}, '核验 arXiv 原始编号、重复行与 ZIP 完整性'),
    ('audit','arxiv-tags'): ('paper_atlas.audits.arxiv_tags', {}, '逐记录核验 arXiv 类别及出处完整性'),
    ('audit','public-openreview'): ('paper_atlas.audits.public_openreview', {}, '核对公开负面决定与已有投稿历史'),
+    ('audit','source-identities'): ('paper_atlas.audits.source_identity', {}, '只读核对来源文件的身份候选与既有投稿历史'),
    ('audit','histories'): ('paper_atlas.audits.histories', {}, '只读核对跨会议投稿历史及决定来源关系'),
 }

@@ -136,6 +136,9 @@ def read_pages(cache=None, snapshots=None):


 def _collect(year, out, *, cache=None, snapshots=None):
+    if year == 2026:
+        from paper_atlas.collectors.aamas2026 import collect as collect_2026
+        return collect_2026(out, cache=cache, snapshots=snapshots)
    if year == 2025:
        from paper_atlas.collectors.aamas2025 import collect as collect_2025
        return collect_2025(out, cache=cache, snapshots=snapshots)
@@ -193,8 +196,8 @@ def _collect(year, out, *, cache=None, snapshots=None):


 def collect(year, out, *, cache=None, snapshots=None):
-    if year not in (2012, 2025):
-        raise ValueError('Only independently reviewed AAMAS 2012/2025 layouts are supported')
+    if year not in (2012, 2025, 2026):
+        raise ValueError('Only independently reviewed AAMAS 2012/2025/2026 layouts are supported')
    out = Path(out)
    out.mkdir(parents=True, exist_ok=True)
    with (out/'aamas.collect.lock').open('a') as lock:
@@ -204,7 +207,7 @@ def collect(year, out, *, cache=None, snapshots=None):

 def main(argv=None, *, prog=None):
    parser = argparse.ArgumentParser(prog=prog)
-    parser.add_argument('--year', type=int, choices=[2012, 2025], required=True)
+    parser.add_argument('--year', type=int, choices=[2012, 2025, 2026], required=True)
    parser.add_argument('--out', required=True)
    source = parser.add_mutually_exclusive_group(required=True)
    source.add_argument('--cache')
@@ -0,0 +1,219 @@
+"""Reviewed AAMAS 2026 HTML metadata; retractions stay outside accepted queues."""
+from collections import Counter, defaultdict
+import hashlib
+import json
+from pathlib import Path
+import re
+import tempfile
+
+from paper_atlas.collectors.aamas2025 import one
+from paper_atlas.db.submissions import timestamp
+from paper_atlas.io.html import Document, Node, clean
+from paper_atlas.io.http import fetch
+
+ROOT = 'https://www.ifaamas.org/Proceedings/aamas2026/'
+URL = ROOT+'forms/contents.htm'
+TITLE = 'AAMAS 2026: Proceedings of the 2026 International Conference on Autonomous Agents and Multiagent Systems'
+SECTIONS = {
+    'K': ('Keynote Talks', 'keynote', 'keynote', 3),
+    'R': ('Research Paper Track', 'full_paper', 'proceedings_unspecified', 338),
+    '1': ('Extended Abstracts', 'extended_abstract', 'proceedings_unspecified', 185),
+    'A': ('AAAI Track', 'full_paper', 'aaai_transfer', 19),
+    'B': ('Blue Sky Ideas Track', 'blue_sky_paper', 'blue_sky', 17),
+    'C': ('Doctoral Consortium', 'doctoral_abstract', 'doctoral', 34),
+    'D': ('Demonstration Track', 'demo', 'demo', 44),
+    'J': ('JAAMAS Track', 'journal_presentation', 'jaamas', 3),
+}
+PDF = re.compile(r'\.\./pdfs/([A-Z]{4}[0-9]{4})\.pdf')
+RETRACTION_TITLE = 'Unifying Model-Free Efficiency and Model-Based Representations via Latent Dynamics'
+WITHDRAWAL_TITLE = 'Cognitive Heterogeneity and Behavioral Biases in Multi-Stage Supply Chains: Evidence from LLM-Based Agent Simulations'
+
+
+def parse_authors(paragraph):
+    font = one([n for n in paragraph.walk() if n.tag == 'font' and
+                n.attrs.get('color', '').lower() == '#222222'], 'ordered author block')
+    authors, affiliations = [], []
+    name, affiliation = '', ''
+    def flush():
+        nonlocal name, affiliation
+        if clean(name) or clean(affiliation):
+            if not clean(name):
+                raise ValueError('Affiliation without author')
+            authors.append(clean(name)); affiliations.append(clean(affiliation))
+        name, affiliation = '', ''
+    for child in font.children:
+        if isinstance(child, str):
+            name += child
+        elif child.tag == 'br':
+            flush()
+        elif child.tag == 'i':
+            affiliation += child.text()
+        else:
+            raise ValueError('Unreviewed author markup: '+child.tag)
+    flush()
+    if not authors:
+        raise ValueError('Empty author list')
+    return authors, affiliations
+
+
+def parse_contents(html, evidence):
+    doc = Document(html)
+    if [clean(n.text()) for n in doc.root.walk() if n.tag == 'title'] != [TITLE]:
+        raise ValueError('Unexpected AAMAS proceedings title/year')
+    section = None
+    headers, records, excluded, held = [], [], [], []
+    counts, seen_pages, seen_pdf_ids = Counter(), set(), set()
+    for node in doc.root.walk():
+        if node.tag != 'p' or any(n is not node and n.tag == 'p' for n in node.walk()):
+            continue
+        anchors = [n.attrs['name'] for n in node.walk() if n.tag == 'a' and n.attrs.get('name') in SECTIONS]
+        if anchors:
+            section = one(anchors, 'section anchor')
+            if clean(node.text()) != SECTIONS[section][0]:
+                raise ValueError('Publisher section heading differs from reviewed layout')
+            headers.append(section)
+        page_fonts = [n for n in node.walk() if n.tag == 'font' and
+                      n.attrs.get('color', '').lower() == '#444444']
+        if not page_fonts:
+            continue
+        page_text = clean(one(page_fonts, 'page marker').text())
+        page_match = re.fullmatch(r'\(Page ([1-9][0-9]*)\)', page_text)
+        if not page_match or section not in SECTIONS:
+            raise ValueError('Invalid page or paper outside reviewed sections')
+        page = int(page_match[1])
+        if page in seen_pages:
+            raise ValueError('Repeated publisher page ID')
+        seen_pages.add(page)
+        title_node = one([n for n in node.walk() if n.tag == 'strong'], 'paper title')
+        title = clean(title_node.text())
+        if not title:
+            raise ValueError('Empty paper title')
+        authors, affiliations = parse_authors(node)
+        link = one([n for n in node.walk() if n.tag == 'a' and 'href' in n.attrs], 'paper link')
+        href = link.attrs['href']
+        title_nodes = {id(n) for n in title_node.walk()}
+        retractions = [clean(n.text()) for n in link.walk() if n.tag == 'i' and id(n) not in title_nodes]
+        heading, publication_format, track, _ = SECTIONS[section]
+        counts[section] += 1
+        metadata = {'publisher_page': page, 'section_url': URL, 'section_anchor': section,
+                    'section_heading': heading, 'publication_format': publication_format,
+                    'author_affiliations': affiliations, 'source_evidence': evidence,
+                    'decision_date_precision': 'not_supplied',
+                    'missing_abstract_reason': 'Publisher HTML supplies no abstract; research PDF not downloaded.',
+                    'publication_evidence': 'Published proceedings entry, not an original decision note.'}
+        withdrawn = retractions == ['Withdrawn June 8, 2026']
+        if withdrawn:
+            if (section, page, title, href) != ('1', 3178, WITHDRAWAL_TITLE, ''):
+                raise ValueError('Unreviewed withdrawal notice')
+            metadata.update({'publication_state': 'withdrawn',
+                             'withdrawal_date': '2026-06-08',
+                             'raw_publication_state': retractions[0],
+                             'decision_date_precision': 'day',
+                             'publication_evidence': 'Explicit dated publisher withdrawal notice, not an original review decision note.'})
+        elif retractions:
+            if (section, page, title, href, retractions) != ('R', 238, RETRACTION_TITLE, '', ['Retracted June 8, 2026']):
+                raise ValueError('Unreviewed retraction or publication status markup')
+            held.append({'source': 'aamas_proceedings', 'external_id': '2026/p238',
+                         'title': title, 'authors': authors, 'year': 2026, 'track': track,
+                         'url': URL+'#R', 'abstract': None, 'identifiers': {},
+                         'observed_at': evidence['observed_at'], 'metadata': metadata,
+                         'publication_state': 'retracted', 'retraction_date': '2026-06-08',
+                         'raw_publication_state': retractions[0],
+                         'hold_reason': 'Explicit retraction; not an accepted/withdrawn submission assertion. Dedicated decision modeling and identity review required.'})
+            continue
+        if not withdrawn:
+            match = PDF.fullmatch(href)
+            if not match or match[1] in seen_pdf_ids:
+                raise ValueError('Invalid or repeated publisher PDF ID')
+            seen_pdf_ids.add(match[1])
+            metadata.update({'publisher_document_id': match[1], 'paper_url': ROOT+'pdfs/'+match[1]+'.pdf'})
+        if section == 'K':
+            excluded.append({'publisher_page': page, 'title': title, 'authors': authors,
+                             'metadata': metadata, 'reason': 'Publisher keynote, not a research submission.'})
+            continue
+        metadata['track_scope_note'] = ('Published format known; individual main/special submission track not established.'
+                                        if track == 'proceedings_unspecified' else 'Explicit publisher section.')
+        if section == 'A':
+            metadata['prior_submission_note'] = 'AAAI transfer section does not supply original submission IDs or individual prior decision notes; none inferred.'
+        if section == 'J':
+            metadata['track_scope_note'] = 'Journal presentation; not another independent journal review or research-track acceptance.'
+        records.append({'source': 'aamas_proceedings', 'external_id': '2026/p'+str(page),
+                        'url': URL+'#'+section, 'title': title, 'abstract': None,
+                        'authors': authors, 'identifiers': {}, 'venue': 'AAMAS', 'venue_aliases': [],
+                        'year': 2026, 'track': track, 'status': 'withdrawn' if withdrawn else 'accepted',
+                        'observed_at': evidence['observed_at'],
+                        'raw_decision': retractions[0] if withdrawn else 'listed in official AAMAS 2026 published proceedings '+heading,
+                        'metadata': metadata,
+                        'tags': [{'namespace': 'aamas_proceedings:publication_format', 'value': publication_format}]})
+    if headers != list(SECTIONS) or dict(counts) != {k: v[3] for k, v in SECTIONS.items()}:
+        raise ValueError('Publisher section inventory differs from reviewed 2026 layout')
+    if (len(records), len(excluded), len(held)) != (639, 3, 1):
+        raise ValueError('Unexpected accepted/keynote/retraction partition')
+    if Counter(r['status'] for r in records) != {'accepted': 638, 'withdrawn': 1}:
+        raise ValueError('Unexpected publisher publication-state partition')
+    return records, excluded, held
+
+
+def read_page(cache, snapshots):
+    if bool(cache) == bool(snapshots):
+        raise ValueError('Choose exactly one cache or frozen snapshot directory')
+    if snapshots:
+        base = Path(snapshots).resolve()
+        frozen = json.loads((base/'receipts.json').read_text())['receipts']
+        r = one([r for r in frozen if r['url'] == URL], 'frozen response receipt')
+        filename = Path(r['file'])
+        if (filename.is_absolute() or '..' in filename.parts or
+                not (base/filename).resolve().is_relative_to(base) or r.get('status') != 200):
+            raise ValueError('Unsafe or unsuccessful publisher response')
+        data = (base/filename).read_bytes()
+        if len(data) > 8*1024**2 or len(data) != r['bytes'] or hashlib.sha256(data).hexdigest() != r['sha256']:
+            raise ValueError('Frozen publisher response checksum/size mismatch')
+        observed = r['finished_at']
+    else:
+        data = fetch(URL, cache, 8*1024**2, delay=1)
+        r = json.loads((Path(cache)/hashlib.sha256(URL.encode()).hexdigest()).with_suffix('.json').read_text())
+        observed = r['retrieved_at']
+    timestamp(observed)
+    return data, {'url': URL, 'sha256': hashlib.sha256(data).hexdigest(), 'bytes': len(data),
+                  'file': 'evidence/'+hashlib.sha256(data).hexdigest()+'.html',
+                  'observed_at': observed, 'http_status': 200, 'response_receipt': r}
+
+
+def collect(out, *, cache=None, snapshots=None):
+    data, evidence = read_page(cache, snapshots)
+    records, excluded, held = parse_contents(data.decode('utf-8'), evidence)
+    grouped = defaultdict(list)
+    for row in records:
+        grouped[row['track']].append(row)
+    out = Path(out); out.mkdir(parents=True, exist_ok=True); files = []
+    with tempfile.TemporaryDirectory(prefix='.aamas2026-', dir=out) as temporary:
+        stage = Path(temporary); (stage/'evidence').mkdir()
+        (stage/evidence['file']).write_bytes(data)
+        for track, group in sorted(grouped.items()):
+            p = stage/f'aamas-proceedings-2026-{track}.jsonl'
+            p.write_text(''.join(json.dumps(r, ensure_ascii=False)+'\n' for r in group))
+            manifest = {'source': 'aamas_proceedings', 'parser_version': 3, 'venue': 'AAMAS', 'year': 2026,
+                        'track': track, 'count': len(group), 'file': str(out/p.name),
+                        'file_sha256': hashlib.sha256(p.read_bytes()).hexdigest(), 'list_url': URL,
+                        'status': 'partial', 'missing_abstract_count': len(group),
+                        'status_counts': dict(Counter(r['status'] for r in group)),
+                        'publication_format_counts': dict(Counter(r['metadata']['publication_format'] for r in group)),
+                        'note': 'Reviewed TOC partition excluding keynote/retracted entries; identity and full submission history remain unverified.'}
+            p.with_suffix('.manifest.json').write_text(json.dumps(manifest, ensure_ascii=False, indent=2)+'\n')
+            files.append(manifest)
+        result = {'source': 'aamas_proceedings', 'year': 2026, 'finished_at': timestamp(),
+                  'status': 'collected_pending_identity_review', 'records': len(records), 'files': files,
+                  'excluded_publications': excluded, 'held_publications': held,
+                  'status_counts': dict(Counter(r['status'] for r in records)),
+                  'source_receipts': [evidence], 'database_written': False, 'pdfs_downloaded': False,
+                  'note': 'Retraction evidence held separately; publication is not an original decision note, and absence is not rejection.'}
+        report = 'aamas-proceedings-2026.json'
+        (stage/report).write_text(json.dumps(result, ensure_ascii=False, indent=2)+'\n')
+        (out/'evidence').mkdir(exist_ok=True)
+        for p in (stage/'evidence').iterdir():
+            p.replace(out/'evidence'/p.name)
+        for p in stage.iterdir():
+            if p.is_file() and p.name != report:
+                p.replace(out/p.name)
+        (stage/report).replace(out/report)
+    return result
@@ -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 ecp ewsp kr atal aamas icassp interspeech eurospeech icslp ismir miccai fat facct lrec mm mlsys'.split())
+FAMILIES=set('nips icml iclr colm colt aistats uai clear acl emnlp naacl eacl coling conll cvpr iccv eccv aaai ijcai ecai kdd sigir www wsdm recsys cikm icdm sdm ecml pkdd ecmlpkdd icra iros rss corl aips icaps ecp ewsp kr atal aamas ifaamas icassp interspeech eurospeech icslp ismir miccai fat facct lrec mm mlsys'.split())


 def records(archive,dtd_bytes):
@@ -0,0 +1,156 @@
+"""Recover raw family candidates from a pinned snapshot and its existing catalogue.
+
+The old candidate file is not an exhaustive source: earlier discovery filters
+can omit an entire namespace. This command checks the complete snapshot and
+catalogue, without assigning venues, tracks, acceptance or paper identities.
+"""
+import argparse
+from collections import Counter
+from contextlib import closing
+import fcntl
+import json
+from pathlib import Path
+import re
+import sqlite3
+import tempfile
+
+from paper_atlas.collectors.dblp import DTD_MD5, FAMILIES, RELEASE, XML_SHA, family, records
+from paper_atlas.db.submissions import timestamp
+from paper_atlas.io.snapshots import digest_file
+
+
+def catalogue_evidence(catalogue, families, expected_sha, expected_dtd_md5):
+    manifest_path = catalogue.with_name('dblp-ai-candidates.manifest.json')
+    manifest = json.loads(manifest_path.read_text())
+    if (manifest.get('snapshot_sha256') != expected_sha
+            or manifest.get('dtd_md5') != expected_dtd_md5):
+        raise ValueError('Catalogue belongs to a different snapshot or DTD')
+    totals = manifest.get('record_types', {})
+    if (set(totals) != {'proceedings', 'inproceedings'}
+            or any(type(n) is not int or n < 0 for n in totals.values())
+            or type(manifest.get('records_scanned')) is not int
+            or sum(totals.values()) != manifest['records_scanned']):
+        raise ValueError('Incomplete catalogue record totals')
+    counts = Counter()
+    volumes = {}
+    # The catalogue is evidence, never a writable recovery destination.
+    with closing(sqlite3.connect(catalogue.resolve().as_uri() + '?mode=ro', uri=True)) as c:
+        for name in families:
+            found = c.execute('SELECT record_type,count FROM family_counts WHERE family=?',
+                              (name,)).fetchall()
+            if not found or not any(n for _, n in found):
+                raise ValueError('Family absent from complete catalogue: ' + name)
+            for kind, count in found:
+                if kind not in totals or type(count) is not int or count < 0:
+                    raise ValueError('Invalid catalogue family count')
+                counts[(name, kind)] = count
+            for key, raw in c.execute('SELECT key,metadata_json FROM proceedings WHERE family=?',
+                                      (name,)):
+                row = json.loads(raw)
+                if (family(key) != name or row.get('key') != key
+                        or row.get('record_type') != 'proceedings'):
+                    raise ValueError('Invalid catalogue proceedings evidence')
+                volumes[key] = row
+    if len(volumes) != sum(counts[(name, 'proceedings')] for name in families):
+        raise ValueError('Catalogue proceedings count differs from volume records')
+    return manifest, digest_file(manifest_path)['sha256'], counts, volumes
+
+
+def recover(archive, dtd, catalogue, out, families, *, expected_sha=XML_SHA,
+            expected_dtd_md5=DTD_MD5):
+    families = list(families)
+    if (not families or any(not isinstance(name, str)
+                           or not re.fullmatch(r'[a-z][a-z0-9-]*', name)
+                           or name not in FAMILIES for name in families)):
+        raise ValueError('Select an explicitly supported discovery family')
+    families = sorted(set(families))
+    archive, dtd, catalogue = Path(archive), Path(dtd), Path(catalogue)
+    output = Path(out).absolute()
+    if output.exists() or output.is_symlink():
+        raise ValueError('Recovery output must not already exist')
+    output.parent.mkdir(parents=True, exist_ok=True)
+    lock_path = output.parent / ('.' + output.name + '.recovery.lock')
+    with lock_path.open('a') as lock:
+        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        if output.exists() or output.is_symlink():
+            raise ValueError('Recovery output appeared before scan')
+        actual = digest_file(archive)
+        if actual['sha256'] != expected_sha:
+            raise ValueError('DBLP snapshot checksum mismatch')
+        if dtd.stat().st_size > 128 * 1024 or digest_file(dtd)['md5'] != expected_dtd_md5:
+            raise ValueError('DBLP DTD checksum mismatch or size limit')
+        manifest, manifest_sha, expected, volumes = catalogue_evidence(
+            catalogue, families, expected_sha, expected_dtd_md5)
+        totals, found = Counter(), Counter()
+        seen_keys, found_volumes = set(), set()
+        with tempfile.TemporaryDirectory(prefix='.dblp-family-', dir=output.parent) as tmp:
+            stage = Path(tmp) / 'recovered'
+            stage.mkdir()
+            target = stage / 'dblp-family-candidates.jsonl'
+            with target.open('w') as stream:
+                for row in records(archive, dtd.read_bytes()):
+                    totals[row['record_type']] += 1
+                    name = family(row['key'])
+                    if name in families:
+                        if row['key'] in seen_keys:
+                            raise ValueError('Duplicate raw bibliographic key')
+                        seen_keys.add(row['key'])
+                        found[(name, row['record_type'])] += 1
+                        if row['record_type'] == 'proceedings':
+                            if volumes.get(row['key']) != row:
+                                raise ValueError('Raw proceedings differs from catalogue evidence')
+                            found_volumes.add(row['key'])
+                        stream.write(json.dumps(row, ensure_ascii=False) + '\n')
+                    scanned = sum(totals.values())
+                    if scanned % 250000 == 0:
+                        print(json.dumps({'status': 'scanning_raw_family_candidates',
+                                          'records_scanned': scanned,
+                                          'recovered_records': sum(found.values())}), flush=True)
+            if (totals != Counter(manifest['record_types'])
+                    or sum(totals.values()) != manifest['records_scanned']):
+                raise ValueError('Complete snapshot totals differ from catalogue')
+            if found != expected or found_volumes != volumes.keys():
+                raise ValueError('Recovered family totals differ from catalogue')
+            report = {
+                'format_version': 1, 'source': 'dblp',
+                'status': 'recovered_raw_candidates_pending_scope_review',
+                'finished_at': timestamp(), 'source_release': RELEASE,
+                'snapshot_sha256': actual['sha256'], 'dtd_md5': expected_dtd_md5,
+                'catalogue_manifest_sha256': manifest_sha, 'families': families,
+                'records_scanned': sum(totals.values()), 'record_types': dict(totals),
+                'recovered_records': sum(found.values()),
+                'recovered_publications': sum(found[(n, 'inproceedings')] for n in families),
+                'recovered_volumes': len(found_volumes),
+                'family_counts': [{'family': name, 'record_type': kind, 'count': count}
+                                  for (name, kind), count in sorted(found.items())],
+                'old_candidate_family_counts': {
+                    name: manifest.get('candidate_family_counts', {}).get(name, 0)
+                    for name in families},
+                'catalogue_volumes': [volumes[k] for k in sorted(volumes)],
+                'file': str(output / target.name),
+                'file_sha256': digest_file(target)['sha256'], 'database_written': False,
+                'note': 'Raw bibliographic candidates, including volume records and possible '
+                        'keynotes/workshops. No venue, track, acceptance or work identity '
+                        'inferred. Requires primary-source scope and identity review; '
+                        'not a normalized import queue. Existing imported sources must '
+                        'not be replayed.',
+            }
+            # Deliberately not an import-records sidecar manifest.
+            (stage / 'recovery-manifest.json').write_text(
+                json.dumps(report, ensure_ascii=False, indent=2) + '\n')
+            if output.exists() or output.is_symlink():
+                raise ValueError('Recovery output appeared during scan')
+            stage.rename(output)
+    return report
+
+
+def main(argv=None, *, prog=None):
+    parser = argparse.ArgumentParser(prog=prog, description=__doc__)
+    parser.add_argument('--archive', required=True, type=Path)
+    parser.add_argument('--dtd', required=True, type=Path)
+    parser.add_argument('--catalogue', required=True, type=Path)
+    parser.add_argument('--family', required=True, action='append', choices=sorted(FAMILIES))
+    parser.add_argument('--out', required=True, type=Path)
+    args = parser.parse_args(argv)
+    print(json.dumps(recover(args.archive, args.dtd, args.catalogue, args.out, args.family),
+                     ensure_ascii=False))
@@ -0,0 +1,218 @@
+"""Official NeurIPS program metadata, with presentations separate from submissions."""
+import argparse
+from collections import Counter, defaultdict
+import fcntl
+import hashlib
+import json
+from pathlib import Path
+import re
+import tempfile
+from urllib.parse import parse_qs, urlsplit
+
+from paper_atlas.db.submissions import timestamp
+from paper_atlas.io.http import fetch
+
+URL = 'https://neurips.cc/static/virtual/data/neurips-2026-orals-posters.json'
+GROUPS = {
+    'https://openreview.net/group?id=NeurIPS.cc/2026/Conference': 'main',
+    'https://openreview.net/group?id=NeurIPS.cc/2026/Evaluations_and_Datasets_Track': 'evaluations_datasets',
+    'https://openreview.net/group?id=NeurIPS.cc/2026/Position_Paper_Track': 'position',
+}
+ACCEPTED = {'Accept', 'Accept (poster)', 'Accept (spotlight)', 'Accept (oral)',
+            'Accept (Spotlight)', 'Accept (Oral)'}
+LIMIT = 64*1024**2
+
+
+def forum_id(url):
+    if not isinstance(url, str):
+        return None
+    try:
+        parsed = urlsplit(url)
+    except ValueError:
+        return None
+    query = parse_qs(parsed.query, keep_blank_values=True)
+    if (parsed.scheme != 'https' or parsed.netloc != 'openreview.net' or
+            parsed.path != '/forum' or parsed.fragment or set(query) != {'id'} or
+            len(query['id']) != 1 or not re.fullmatch(r'[A-Za-z0-9_-]{8,32}', query['id'][0])):
+        return None
+    return query['id'][0]
+
+
+def parse_program(data, evidence):
+    root = json.loads(data)
+    if (not isinstance(root, dict) or not {'count', 'next', 'previous', 'results'} <= set(root) or
+            not isinstance(root.get('results'), list) or
+            type(root.get('count')) is not int or root['count'] != len(root['results']) or
+            root.get('next') is not None or root.get('previous') is not None):
+        raise ValueError('Expected one complete, nonpaginated public program snapshot')
+    if not root['results']:
+        raise ValueError('Empty program snapshot')
+    events, grouped, held, duplicates = {}, defaultdict(list), [], []
+    for index, event in enumerate(root['results']):
+        if not isinstance(event, dict) or type(event.get('id')) is not int or event['id'] <= 0:
+            raise ValueError('Missing stable event ID')
+        eid = event['id']
+        if eid in events:
+            first_index, first = events[eid]
+            if first != event:
+                raise ValueError('Conflicting payloads share one publisher event ID')
+            duplicates.append({'event_id': eid, 'first_row': first_index, 'duplicate_row': index})
+            continue
+        events[eid] = index, event
+        origin = event.get('sourceurl')
+        reason = None
+        if not isinstance(origin, str) or origin not in GROUPS:
+            reason = ('journal_presentation_requires_original_publication_relation'
+                      if isinstance(origin, str) and origin.startswith('journal-track/') else
+                      'bridging_future_scope_requires_review' if origin == 'neurips2026-bridging-the-future' else
+                      'unreviewed_or_missing_submission_origin')
+        elif event.get('visible') is not True:
+            reason = 'program_visibility_requires_review'
+        elif not isinstance(event.get('decision'), str) or event['decision'] not in ACCEPTED:
+            reason = 'unreviewed_decision_label'
+        elif event.get('event_type') not in ('Poster', 'Oral'):
+            reason = 'unreviewed_presentation_type'
+        forum = forum_id(event.get('paper_url'))
+        if reason is None and forum is None:
+            reason = 'missing_or_invalid_original_submission_link'
+        if reason:
+            held.append({'row': index, 'event_id': eid, 'reason': reason, 'raw_event': event})
+            continue
+        if (not isinstance(event.get('name'), str) or not event['name'].strip() or
+                not isinstance(event.get('authors'), list) or not event['authors'] or
+                any(not isinstance(a, dict) or not isinstance(a.get('fullname'), str) or
+                    not a['fullname'].strip() for a in event['authors'])):
+            raise ValueError('Empty title or incomplete ordered author list')
+        if not isinstance(event.get('virtualsite_url'), str) or not re.fullmatch(
+                r'/virtual/2026/(?:poster|oral)/'+str(eid), event['virtualsite_url']):
+            raise ValueError('Publisher event URL differs from year/type/ID')
+        grouped[forum].append((index, event))
+    records = []
+    folded = 0
+    for forum, observations in sorted(grouped.items()):
+        # Signed sourceid values are preserved verbatim in the presentation
+        # evidence; they are not independent submissions or scientific works.
+        signatures = {json.dumps([e['name'], e['authors'], e['sourceurl'], e['decision']],
+                                 ensure_ascii=False, sort_keys=True) for _, e in observations}
+        if len(signatures) != 1:
+            raise ValueError('Title, authors, origin or decision disagree for one forum')
+        posters = [e for _, e in observations if e['event_type'] == 'Poster']
+        event = posters[0] if posters else observations[0][1]
+        track = GROUPS[event['sourceurl']]
+        presentations = []
+        for index, e in observations:
+            presentations.append({'row': index, 'event_id': e['id'], 'uid': e.get('uid'),
+                                  'sourceid': e.get('sourceid'), 'event_type': e['event_type'],
+                                  'eventtype': e.get('eventtype'), 'program_url': e['virtualsite_url'],
+                                  'session': e.get('session'), 'starttime': e.get('starttime'),
+                                  'endtime': e.get('endtime'), 'parent_id': e.get('parent_id'),
+                                  'raw_decision': e['decision']})
+        formats = sorted({e['event_type'] for _, e in observations})
+        records.append({
+            'source': 'neurips_program', 'external_id': '2026/'+forum,
+            'url': event['paper_url'], 'title': event['name'], 'abstract': None,
+            'authors': [a['fullname'] for a in event['authors']],
+            'identifiers': {'openreview': forum}, 'venue': 'NeurIPS', 'venue_aliases': [],
+            'year': 2026, 'track': track, 'status': 'accepted',
+            'observed_at': evidence['observed_at'], 'raw_decision': event['decision'],
+            'metadata': {'source_group_url': event['sourceurl'], 'forum_id': forum,
+                         'author_affiliations': [a.get('institution') for a in event['authors']],
+                         'presentations': presentations, 'source_evidence': evidence,
+                         'publication_evidence': 'Explicit accepted label in official conference program; not an original decision note or proof of proceedings publication.',
+                         'decision_date_precision': 'not_supplied',
+                         'missing_abstract_reason': 'Program JSON does not contain abstracts; paper HTML/PDF not fetched by this command.',
+                         'identity_scope_note': 'One record per original forum, not one scientific work per event. Other venues/versions require independent identity review.'},
+            'tags': [{'namespace': 'neurips_program:presentation_format', 'value': value}
+                     for value in formats]})
+        folded += len(observations)-1
+    if not records:
+        raise ValueError('No reviewed submission origins found')
+    if root['count'] != len(records)+len(held)+len(duplicates)+folded:
+        raise ValueError('Program partition failed')
+    return records, held, duplicates, folded
+
+
+def read_program(cache, snapshots):
+    if bool(cache) == bool(snapshots):
+        raise ValueError('Choose exactly one cache or frozen snapshot directory')
+    if snapshots:
+        base = Path(snapshots).resolve()
+        matches = [r for r in json.loads((base/'receipts.json').read_text())['receipts'] if r['url'] == URL]
+        if len(matches) != 1:
+            raise ValueError('Expected one official program receipt')
+        r = matches[0]; filename = Path(r['file'])
+        if (filename.is_absolute() or '..' in filename.parts or
+                not (base/filename).resolve().is_relative_to(base) or r.get('status') != 200):
+            raise ValueError('Unsafe or unsuccessful frozen program response')
+        data = (base/filename).read_bytes()
+        if len(data) > LIMIT or len(data) != r['bytes'] or hashlib.sha256(data).hexdigest() != r['sha256']:
+            raise ValueError('Frozen program checksum/size mismatch')
+        observed = r['finished_at']
+    else:
+        data = fetch(URL, cache, LIMIT, delay=1)
+        r = json.loads((Path(cache)/hashlib.sha256(URL.encode()).hexdigest()).with_suffix('.json').read_text())
+        observed = r['retrieved_at']
+    timestamp(observed)
+    return data, {'url': URL, 'sha256': hashlib.sha256(data).hexdigest(), 'bytes': len(data),
+                  'file': 'evidence/'+hashlib.sha256(data).hexdigest()+'.json',
+                  'observed_at': observed, 'http_status': 200, 'response_receipt': r}
+
+
+def collect(out, *, cache=None, snapshots=None):
+    out = Path(out); out.mkdir(parents=True, exist_ok=True)
+    report_name = 'neurips-program-2026.json'
+    with (out/'neurips-program.collect.lock').open('a') as lock:
+        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        if (out/report_name).exists():
+            raise ValueError('Completed collection exists; use a new directory and reconcile before refresh')
+        data, evidence = read_program(cache, snapshots)
+        records, held, duplicates, folded = parse_program(data, evidence)
+        groups = defaultdict(list)
+        for r in records:
+            groups[r['track']].append(r)
+        files = []
+        with tempfile.TemporaryDirectory(prefix='.neurips-program-', dir=out) as temporary:
+            stage = Path(temporary); (stage/'evidence').mkdir()
+            (stage/evidence['file']).write_bytes(data)
+            for track, group in sorted(groups.items()):
+                p = stage/f'neurips-program-2026-{track}.jsonl'
+                p.write_text(''.join(json.dumps(r, ensure_ascii=False)+'\n' for r in group))
+                manifest = {'source': 'neurips_program', 'parser_version': 1, 'venue': 'NeurIPS',
+                            'year': 2026, 'track': track, 'count': len(group), 'file': str(out/p.name),
+                            'file_sha256': hashlib.sha256(p.read_bytes()).hexdigest(), 'list_url': URL,
+                            'status': 'partial', 'missing_abstract_count': len(group),
+                            'note': 'Reviewed forum-level program partition; not full current proceedings, abstracts, public rejection coverage or scientific-work dedup.'}
+                p.with_suffix('.manifest.json').write_text(json.dumps(manifest, ensure_ascii=False, indent=2)+'\n')
+                files.append(manifest)
+            result = {'source': 'neurips_program', 'year': 2026, 'finished_at': timestamp(),
+                      'status': 'collected_pending_identity_review', 'records': len(records), 'files': files,
+                      'held_presentations': held, 'exact_duplicate_event_rows': duplicates,
+                      'folded_extra_presentation_rows': folded, 'source_receipts': [evidence],
+                      'database_written': False, 'pdfs_downloaded': False,
+                      'note': 'Public program only; rejected opt-in availability needs a separate source. Journals, bridging and missing origins remain held; absence is not rejection.'}
+            (stage/report_name).write_text(json.dumps(result, ensure_ascii=False, indent=2)+'\n')
+            (out/'evidence').mkdir(exist_ok=True)
+            for p in (stage/'evidence').iterdir():
+                p.replace(out/'evidence'/p.name)
+            for p in stage.iterdir():
+                if p.is_file() and p.name != report_name:
+                    p.replace(out/p.name)
+            (stage/report_name).replace(out/report_name)
+        return result
+
+
+def main(argv=None, *, prog=None):
+    parser = argparse.ArgumentParser(prog=prog)
+    parser.add_argument('--out', required=True)
+    source = parser.add_mutually_exclusive_group(required=True)
+    source.add_argument('--cache')
+    source.add_argument('--snapshots')
+    args = parser.parse_args(argv)
+    result = collect(args.out, cache=args.cache, snapshots=args.snapshots)
+    print(json.dumps({'source': result['source'], 'year': 2026, 'status': result['status'],
+                      'records': result['records'], 'files': result['files'],
+                      'held_presentations': len(result['held_presentations']),
+                      'exact_duplicate_event_rows': len(result['exact_duplicate_event_rows']),
+                      'folded_extra_presentation_rows': result['folded_extra_presentation_rows'],
+                      'report_file': str(Path(args.out)/'neurips-program-2026.json'),
+                      'database_written': False, 'pdfs_downloaded': False}, ensure_ascii=False))
@@ -0,0 +1,231 @@
+"""Reviewed postconference selected volumes from frozen public HTML metadata."""
+import argparse
+from collections import defaultdict
+import fcntl
+import hashlib
+import json
+from pathlib import Path
+import re
+import tempfile
+from urllib.parse import parse_qs, urlsplit
+
+from paper_atlas.db.submissions import timestamp
+from paper_atlas.io.html import Document, clean
+
+SOURCE = 'springer_selected_volume'
+BOOK = '10.1007/978-1-4612-2660-4'
+BOOK_URL = 'https://link.springer.com/book/'+BOOK
+HISTORY_URL = 'https://www.gatsby.ucl.ac.uk/aistats/workshops.html'
+HISTORY_SHA = '6db205b9dbc82b4a096556efd2aa67e065cad8e0bd6351176b1962ffd1fca179'
+SCOPE = {
+    'venue': 'AISTATS', 'year': 1993, 'track': 'selected_volume',
+    'event_key': 'aistats-1993-selected-volume-1994',
+    'publication_year': 1994, 'book_doi': BOOK,
+    'book_title': 'Selecting Models from Data', 'expected_chapters': 49,
+    'relationship': 'Selected workshop papers were revised and rereviewed for later publication; this is not the original complete workshop decision list.',
+}
+
+
+def snapshot_index(root):
+    root = Path(root).resolve()
+    receipts = json.loads((root/'receipts.json').read_bytes())['responses']
+    index = {}
+    for receipt in receipts:
+        url = receipt['url']
+        if url in index:
+            raise ValueError('Duplicate snapshot URL')
+        index[url] = (root, receipt)
+    return index
+
+
+def read_page(index, url):
+    if url not in index:
+        raise ValueError('Missing required public HTML snapshot: '+url)
+    root, receipt = index[url]
+    filename = Path(receipt['file'])
+    path = root/filename
+    content_type = {k.lower():v for k,v in receipt.get('headers', {}).items()}.get('content-type', '')
+    requested, final = urlsplit(url), urlsplit(receipt.get('final_url', ''))
+    query = parse_qs(final.query, keep_blank_values=True)
+    original_query = parse_qs(requested.query, keep_blank_values=True)
+    # Springer returns the public metadata with an explicit cookie diagnostic.
+    # Retain that final URL verbatim; accept no other redirect or changed page.
+    if query != original_query and requested.netloc == 'link.springer.com':
+        if (query.pop('error', None) != ['cookies_not_supported']
+                or not re.fullmatch(r'[0-9a-f-]{36}', ''.join(query.pop('code', [])))):
+            raise ValueError('Unreviewed Springer redirect query')
+    if (filename.is_absolute() or '..' in filename.parts or not path.resolve().is_relative_to(root)
+            or receipt.get('http_status') != 200
+            or (final.scheme, final.netloc, final.path) != (requested.scheme, requested.netloc, requested.path)
+            or final.fragment or query != original_query
+            or 'text/html' not in content_type.lower()):
+        raise ValueError('Unsafe, redirected or unsuccessful public HTML snapshot')
+    raw = path.read_bytes()
+    digest = hashlib.sha256(raw).hexdigest()
+    if len(raw) > 2_000_000 or len(raw) != receipt['bytes'] or digest != receipt['sha256']:
+        raise ValueError('Snapshot checksum or size differs')
+    evidence = {'url':url, 'sha256':digest, 'bytes':len(raw),
+                'file':'evidence/'+digest+'.html', 'http_status':200,
+                'observed_at':timestamp(receipt['at']), 'response_receipt':receipt}
+    return raw, evidence
+
+
+def parse_toc(html, *, excluded_materials=None):
+    entries = {}
+    for node in Document(html).root.walk():
+        if node.tag != 'li' or node.attrs.get('data-test') != 'chapter':
+            continue
+        links = [n for n in node.walk() if n.tag == 'a' and re.fullmatch(
+            r'/chapter/'+re.escape(BOOK)+r'_[1-9]\d*', n.attrs.get('href', ''))]
+        if not links:
+            headings = [clean(n.text()) for n in node.walk() if n.tag in ('h3','h4')]
+            frontmatter = [n.attrs.get('href','') for n in node.walk()
+                           if n.tag == 'a' and n.attrs.get('data-test') == 'front-matter-pdf']
+            if headings == ['Front Matter'] and len(frontmatter) == 1:
+                if excluded_materials is not None:
+                    excluded_materials.append({'title':'Front Matter', 'url':frontmatter[0],
+                                               'reason':'Section front matter without a chapter DOI; not an authored chapter.'})
+                continue
+        if len(links) != 1:
+            raise ValueError('Chapter entry lacks one reviewed-book DOI')
+        link, = links
+        doi = link.attrs['href'].removeprefix('/chapter/')
+        if doi in entries or not clean(link.text()):
+            raise ValueError('Duplicate DOI or missing chapter title')
+        entries[doi] = clean(link.text())
+    return entries
+
+
+def parse_chapter(html, doi):
+    nodes = list(Document(html).root.walk())
+    meta = defaultdict(list)
+    for node in nodes:
+        if node.tag == 'meta' and node.attrs.get('name', '').startswith('citation_'):
+            meta[node.attrs['name']].append(node.attrs.get('content', ''))
+    def one(key):
+        values = meta[key]
+        if len(values) != 1 or not values[0]:
+            raise ValueError('Missing or ambiguous citation field: '+key)
+        return values[0]
+    if (one('citation_doi') != doi or one('citation_inbook_title') != SCOPE['book_title']
+            or one('citation_publication_date') != str(SCOPE['publication_year'])):
+        raise ValueError('Chapter DOI, book or publication year differs from reviewed scope')
+    raw_authors = meta['citation_author']
+    if not raw_authors:
+        raise ValueError('Missing named chapter authors')
+    authors = []
+    for raw in raw_authors:
+        if raw.count(',') != 1:
+            raise ValueError('Unreviewed citation author format')
+        family, given = [clean(s) for s in raw.split(',', 1)]
+        if not family or not given:
+            raise ValueError('Incomplete citation author')
+        authors.append({'given':given, 'family':family})
+    abstract_nodes = [n for n in nodes if n.attrs.get('id') == 'Abs1-content']
+    if len(abstract_nodes) > 1:
+        raise ValueError('Ambiguous abstract section')
+    abstract = clean(abstract_nodes[0].text()) if abstract_nodes else None
+    first, last = one('citation_firstpage'), one('citation_lastpage')
+    if not first.isdigit() or not last.isdigit() or int(first) > int(last):
+        raise ValueError('Invalid chapter pages')
+    return {'doi':doi, 'title':one('citation_title'), 'authors':authors,
+            'authors_citation_raw':list(raw_authors), 'abstract':abstract or None,
+            'first_page':first, 'last_page':last}
+
+
+def record_for(chapter, evidence, scope_evidence):
+    return {
+        'source':SOURCE, 'external_id':chapter['doi'],
+        'url':'https://link.springer.com/chapter/'+chapter['doi'],
+        'title':chapter['title'], 'abstract':chapter['abstract'], 'authors':chapter['authors'],
+        'identifiers':{'doi':chapter['doi']},
+        'submission_identifiers':[{'scheme':'publication_doi','value':chapter['doi']}],
+        **{k:SCOPE[k] for k in ('venue','year','track','event_key')},
+        'status':'accepted', 'observed_at':evidence['observed_at'],
+        'raw_decision':'listed in the 1994 selected postconference volume; original 1993 review decision not supplied',
+        'tags':[{'namespace':SOURCE+':publication_format','value':'selected_volume_chapter'}],
+        'metadata':{
+            'publication_format':'selected_volume_chapter', 'bibliographic_year':1994,
+            'related_workshop_year':1993, 'book_doi':BOOK, 'book_title':SCOPE['book_title'],
+            'year_basis':'Reviewed workshop year; bibliographic publication year is separately retained.',
+            'first_page':chapter['first_page'], 'last_page':chapter['last_page'],
+            'authors_citation_raw':chapter['authors_citation_raw'],
+            'publication_evidence':'Published selected-volume chapter, not an original workshop decision note.',
+            'scope_relationship':SCOPE['relationship'], 'decision_date_precision':'not_supplied',
+            'source_evidence':evidence, 'scope_evidence':scope_evidence,
+        },
+    }
+
+
+def collect(out, *, toc_snapshots, chapter_snapshots):
+    toc_index, chapter_index = snapshot_index(toc_snapshots), snapshot_index(chapter_snapshots)
+    frozen = {}
+    def read(index, url):
+        raw, evidence = read_page(index, url)
+        frozen[evidence['file']] = raw
+        return raw.decode('utf-8'), evidence
+    history, history_evidence = read(toc_index, HISTORY_URL)
+    if hashlib.sha256(history.encode()).hexdigest() != HISTORY_SHA:
+        raise ValueError('Reviewed selected-volume relationship evidence changed')
+    scope_evidence = [history_evidence]
+    titles, excluded = {}, []
+    for page in range(1, 4):
+        html, evidence = read(toc_index, BOOK_URL+('?page='+str(page) if page > 1 else ''))
+        scope_evidence.append(evidence)
+        entries = parse_toc(html, excluded_materials=excluded)
+        if titles.keys() & entries.keys():
+            raise ValueError('DOI repeated across TOC pages')
+        titles.update(entries)
+    expected = {BOOK+'_'+str(i) for i in range(1, 50)}
+    if titles.keys() != expected:
+        raise ValueError('Reviewed 49-chapter inventory changed')
+    chapter_urls = {'https://link.springer.com/chapter/'+doi for doi in expected}
+    if chapter_index.keys() != chapter_urls:
+        raise ValueError('Missing or extraneous chapter snapshot')
+    records = []
+    for doi in sorted(expected, key=lambda s:int(s.rsplit('_',1)[1])):
+        html, evidence = read(chapter_index, 'https://link.springer.com/chapter/'+doi)
+        chapter = parse_chapter(html, doi)
+        if chapter['title'] != titles[doi]:
+            raise ValueError('Chapter title disagrees with official TOC')
+        records.append(record_for(chapter, evidence, scope_evidence))
+    out = Path(out)
+    out.mkdir(parents=True, exist_ok=True)
+    with (out/'springer-selected.collect.lock').open('a') as lock:
+        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        if any(p.name != 'springer-selected.collect.lock' for p in out.iterdir()):
+            raise FileExistsError('Existing output must be inspected, not overwritten')
+        with tempfile.TemporaryDirectory(prefix='.springer-selected-', dir=out) as temporary:
+            stage = Path(temporary)
+            (stage/'evidence').mkdir()
+            for filename, raw in frozen.items():
+                (stage/filename).write_bytes(raw)
+            path = stage/'springer-aistats-1993-selected-volume.jsonl'
+            path.write_text(''.join(json.dumps(r,ensure_ascii=False)+'\n' for r in records))
+            manifest = {'source':SOURCE, 'parser_version':1,
+                        **{k:SCOPE[k] for k in ('venue','year','track','event_key')},
+                        'count':len(records), 'file':str(out/path.name),
+                        'file_sha256':hashlib.sha256(path.read_bytes()).hexdigest(),
+                        'list_url':BOOK_URL, 'status':'partial',
+                        'excluded_materials':excluded,
+                        'missing_abstract_count':sum(not r['abstract'] for r in records)}
+            path.with_suffix('.manifest.json').write_text(json.dumps(manifest,indent=2)+'\n')
+            report = {'status':'collected_pending_identity_review', 'records':len(records),
+                      'files':[manifest], 'reviewed_scope':SCOPE, 'database_written':False,
+                      'pdfs_downloaded':False, 'scope_evidence':scope_evidence}
+            (stage/'collection-report.json').write_text(json.dumps(report,ensure_ascii=False,indent=2)+'\n')
+            for item in stage.iterdir():
+                if item.name != 'collection-report.json':
+                    item.replace(out/item.name)
+            (stage/'collection-report.json').replace(out/'collection-report.json')
+    return report
+
+
+def main(argv=None, *, prog=None):
+    parser = argparse.ArgumentParser(prog=prog)
+    parser.add_argument('--toc-snapshots', required=True)
+    parser.add_argument('--chapter-snapshots', required=True)
+    parser.add_argument('--out', required=True)
+    args = parser.parse_args(argv)
+    print(json.dumps(collect(args.out, toc_snapshots=args.toc_snapshots,
+                             chapter_snapshots=args.chapter_snapshots),ensure_ascii=False))
@@ -0,0 +1,15 @@
+{
+  "receipts": [
+    {
+      "url": "https://neurips.cc/static/virtual/data/neurips-2026-orals-posters.json",
+      "final_url": "https://neurips.cc/static/virtual/data/neurips-2026-orals-posters.json",
+      "status": 200,
+      "content_type": "application/json",
+      "file": "program-json.response",
+      "bytes": 20447021,
+      "sha256": "879293a863057fa05370443b3b5ddee12a10eeb37c810abc6c1d53f71f545997",
+      "started_at": "2026-11-09T17:57:23.825509+00:00",
+      "finished_at": "2026-11-09T17:57:40.125580+00:00"
+    }
+  ]
+}
@@ -0,0 +1,167 @@
+import gzip
+import hashlib
+import json
+from pathlib import Path
+import tempfile
+import sqlite3
+import unittest
+from unittest.mock import patch
+
+from paper_atlas.collectors import aamas, aamas2026
+from paper_atlas.ingestion.records import ingest_file
+
+
+FIXTURE = json.loads(gzip.decompress((Path(__file__).parent/'fixtures/aamas2026-fixture.json.gz').read_bytes()))
+EVIDENCE = {'url': aamas2026.URL, 'observed_at': FIXTURE['receipts']['receipts'][0]['finished_at']}
+
+
+class Aamas2026Tests(unittest.TestCase):
+    def parse(self, html=None):
+        return aamas2026.parse_contents(FIXTURE['contents'] if html is None else html, EVIDENCE)
+
+    def snapshots(self, folder):
+        folder.mkdir()
+        (folder/'contents.html').write_text(FIXTURE['contents'])
+        (folder/'receipts.json').write_text(json.dumps(FIXTURE['receipts']))
+        return folder
+
+    def test_every_publisher_title_author_affiliation_and_page_preserved(self):
+        rows, excluded, held = self.parse()
+        self.assertEqual((len(rows), len(excluded), len(held)), (639, 3, 1))
+        expected = {r['publisher_page']: r for r in FIXTURE['expected_entries']}
+        self.assertEqual(len(expected), 643)
+        all_rows = rows+excluded+held
+        self.assertEqual({r['metadata']['publisher_page'] for r in all_rows}, set(expected))
+        for r in all_rows:
+            e = expected[r['metadata']['publisher_page']]
+            self.assertEqual(r['title'], e['title'])
+            self.assertEqual(r['authors'], e['authors'])
+            self.assertEqual(r['metadata']['author_affiliations'], e['affiliations'])
+            self.assertEqual(r['metadata']['section_heading'], e['section'])
+        for r in rows:
+            self.assertIsNone(r['abstract'])
+            self.assertEqual(r['identifiers'], {})
+            if r['external_id'] != '2026/p3178':
+                self.assertEqual(r['status'], 'accepted')
+                self.assertNotIn('decision_date', r)
+
+    def test_explicit_withdrawal_preserves_status_date_and_no_pdf(self):
+        rows, _, _ = self.parse()
+        row, = [r for r in rows if r['external_id'] == '2026/p3178']
+        self.assertEqual(row['status'], 'withdrawn')
+        self.assertEqual(row['metadata']['withdrawal_date'], '2026-06-08')
+        self.assertNotIn('decision_date', row)
+        self.assertEqual(row['raw_decision'], 'Withdrawn June 8, 2026')
+        self.assertNotIn('paper_url', row['metadata'])
+        self.assertEqual(row['metadata']['publication_state'], 'withdrawn')
+        self.assertEqual(sum(r['status'] == 'accepted' for r in rows), 638)
+
+    def test_withdrawal_ingestion_retains_date_precision_without_invented_acceptance(self):
+        rows, _, _ = self.parse()
+        row, = [r for r in rows if r['external_id'] == '2026/p3178']
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp); path = root/'withdrawn.jsonl'; db = root/'sample.sqlite'
+            path.write_text(json.dumps(row, ensure_ascii=False)+'\n')
+            path.with_suffix('.manifest.json').write_text(json.dumps({
+                'source': 'aamas_proceedings', 'list_url': aamas2026.URL,
+                'venue': 'AAMAS', 'year': 2026, 'track': row['track'],
+                'count': 1, 'status': 'partial', 'missing_abstract_count': 1,
+                'file_sha256': hashlib.sha256(path.read_bytes()).hexdigest()}))
+            ingest_file(db, path)
+            with sqlite3.connect(db) as c:
+                self.assertEqual(c.execute('SELECT status,raw_value,decision_date FROM decisions').fetchall(),
+                                 [('withdrawn', 'Withdrawn June 8, 2026', None)])
+                metadata = json.loads(c.execute('SELECT metadata_json FROM source_records').fetchone()[0])
+                self.assertEqual(metadata['withdrawal_date'], '2026-06-08')
+                self.assertEqual(metadata['decision_date_precision'], 'day')
+                self.assertEqual(c.execute('PRAGMA foreign_key_check').fetchall(), [])
+
+    def test_retraction_is_not_accepted_or_withdrawn_and_retains_date(self):
+        rows, _, held = self.parse()
+        self.assertNotIn('2026/p238', {r['external_id'] for r in rows})
+        h, = held
+        self.assertEqual(h['publication_state'], 'retracted')
+        self.assertEqual(h['retraction_date'], '2026-06-08')
+        self.assertEqual(h['raw_publication_state'], 'Retracted June 8, 2026')
+        self.assertNotIn('status', h)
+        self.assertNotIn('paper_url', h['metadata'])
+        self.assertEqual(h['metadata']['source_evidence'], EVIDENCE)
+
+    def test_tracks_formats_and_prior_decisions_are_not_conflated(self):
+        rows, _, _ = self.parse()
+        counts = {t: sum(r['track'] == t for r in rows) for t in {r['track'] for r in rows}}
+        self.assertEqual(counts, {'proceedings_unspecified': 522, 'aaai_transfer': 19,
+                                 'blue_sky': 17, 'doctoral': 34, 'demo': 44, 'jaamas': 3})
+        for row in rows:
+            self.assertEqual(row['tags'][0]['namespace'], 'aamas_proceedings:publication_format')
+            if row['track'] == 'aaai_transfer':
+                self.assertIn('none inferred', row['metadata']['prior_submission_note'])
+                self.assertNotIn('decision_events', row)
+            if row['track'] == 'jaamas':
+                self.assertEqual(row['metadata']['publication_format'], 'journal_presentation')
+                self.assertIn('not another independent', row['metadata']['track_scope_note'])
+
+    def test_refuses_changed_year_section_page_link_author_and_retraction(self):
+        html = FIXTURE['contents']
+        replacements = [('AAMAS 2026:', 'AAMAS 2027:'),
+                        ('name="R" id="R"></a><font color="#010101">Research Paper Track',
+                         'name="R" id="R"></a><font color="#010101">Workshop Track'),
+                        ('(Page 4)', '(Page 1)'),
+                        ('../pdfs/HQQZ1937.pdf', 'https://other.example/paper.pdf'),
+                        ('Michael Wooldridge <i>', '<i>'),
+                        ('Retracted June 8, 2026', 'Withdrawn June 8, 2026')]
+        for old, new in replacements:
+            self.assertIn(old, html)
+            with self.subTest(change=old), self.assertRaises(ValueError):
+                self.parse(html.replace(old, new, 1))
+
+    def test_frozen_collection_is_partial_with_full_held_evidence(self):
+        with tempfile.TemporaryDirectory() as tmp, patch.object(aamas2026, 'fetch', side_effect=AssertionError('No HTTP')):
+            root = Path(tmp)
+            report = aamas.collect(2026, root/'out', snapshots=self.snapshots(root/'source'))
+            self.assertEqual(report['records'], 639)
+            self.assertEqual(report['status_counts'], {'accepted': 638, 'withdrawn': 1})
+            self.assertEqual(len(report['files']), 6)
+            self.assertEqual(len(report['held_publications']), 1)
+            self.assertFalse(report['database_written'])
+            self.assertFalse(report['pdfs_downloaded'])
+            for m in report['files']:
+                self.assertEqual(m['status'], 'partial')
+                self.assertEqual(m['missing_abstract_count'], m['count'])
+                self.assertEqual(hashlib.sha256(Path(m['file']).read_bytes()).hexdigest(), m['file_sha256'])
+            self.assertTrue((root/'out/aamas-proceedings-2026.json').is_file())
+
+    def test_frozen_response_rejects_tamper_unsafe_path_status_and_symlink(self):
+        for mode in ('tamper', 'path', 'status', 'symlink'):
+            with self.subTest(mode=mode), tempfile.TemporaryDirectory() as tmp:
+                root = Path(tmp); source = self.snapshots(root/'source')
+                r = json.loads((source/'receipts.json').read_text())
+                if mode == 'tamper':
+                    (source/'contents.html').write_text('bad')
+                elif mode == 'path':
+                    r['receipts'][0]['file'] = '../outside.html'
+                elif mode == 'status':
+                    r['receipts'][0]['status'] = 403
+                else:
+                    (root/'outside.html').write_text(FIXTURE['contents'])
+                    (source/'contents.html').unlink()
+                    (source/'contents.html').symlink_to(root/'outside.html')
+                (source/'receipts.json').write_text(json.dumps(r))
+                with self.assertRaises(ValueError):
+                    aamas.collect(2026, root/'out', snapshots=source)
+                self.assertFalse((root/'out/aamas-proceedings-2026.json').exists())
+
+    def test_live_mode_fetches_only_html_and_preserves_receipt(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp); cache = root/'cache'; cache.mkdir()
+            receipt = {'retrieved_at': EVIDENCE['observed_at'], 'url': aamas2026.URL}
+            (cache/hashlib.sha256(aamas2026.URL.encode()).hexdigest()).with_suffix('.json').write_text(json.dumps(receipt))
+            with patch.object(aamas2026, 'fetch', return_value=FIXTURE['contents'].encode()) as fetch:
+                report = aamas.collect(2026, root/'out', cache=cache)
+            fetch.assert_called_once_with(aamas2026.URL, cache, 8*1024**2, delay=1)
+            self.assertEqual(report['source_receipts'][0]['response_receipt'], receipt)
+
+    def test_frozen_fixture_is_original_hashed_response(self):
+        r, = FIXTURE['receipts']['receipts']
+        self.assertEqual(hashlib.sha256(FIXTURE['contents'].encode()).hexdigest(), r['sha256'])
+        self.assertEqual(r['bytes'], len(FIXTURE['contents'].encode()))
@@ -99,7 +99,7 @@ class AamasProceedingsTests(unittest.TestCase):
                aamas.collect(2012, root/'out', snapshots=source)
            self.assertFalse((root/'out/aamas-proceedings-2012.json').exists())
            with self.assertRaises(ValueError):
-                aamas.collect(2026, root/'future', snapshots=source)
+                aamas.collect(2027, root/'future', snapshots=source)
            self.assertFalse((root/'future').exists())

    def test_ids_colliding_across_sections_do_not_produce_a_completed_collection(self):
@@ -0,0 +1,141 @@
+import gzip
+import hashlib
+import json
+from pathlib import Path
+import sqlite3
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from paper_atlas.collectors.dblp import catalogue
+from paper_atlas.collectors.dblp_recovery import recover
+from paper_atlas.collectors.dblp_scope import scope
+from paper_atlas.ingestion.records import ingest_file
+
+
+class DblpRecoveryTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.root = Path(self.tmp.name)
+        self.dtd = self.root / 'dblp.dtd'
+        self.dtd.write_bytes(b'<!ENTITY eacute "&#233;">')
+        self.archive = self.root / 'dblp.xml.gz'
+        # These include a keynote and a workshop, neither a main-paper inference.
+        body = '''<proceedings key="conf/ifaamas/2025"><title>AAMAS 2025</title><year>2025</year></proceedings>
+<inproceedings key="conf/ifaamas/keynote"><author>Ren&eacute;</author><title>Keynote: Agents</title><year>2025</year><booktitle>AAMAS</booktitle><crossref>conf/ifaamas/2025</crossref></inproceedings>
+<inproceedings key="conf/ifaamas/paper"><author>Alice Doe</author><title>Research <i>agents</i></title><year>2025</year><booktitle>AAMAS</booktitle><crossref>conf/ifaamas/2025</crossref><ee>https://dl.acm.org/doi/10.5555/3709347.123</ee></inproceedings>
+<proceedings key="conf/ifaamas/2025w"><title>AAMAS Workshop</title><year>2025</year></proceedings>
+<inproceedings key="conf/ifaamas/workshop"><title>Workshop paper</title><year>2025</year><booktitle>AAMAS Workshop</booktitle><crossref>conf/ifaamas/2025w</crossref></inproceedings>
+<inproceedings key="conf/unrelated/other"><title>Other family</title><year>2025</year></inproceedings>'''
+        with gzip.open(self.archive, 'wb') as stream:
+            stream.write(('<?xml version="1.0"?><!DOCTYPE dblp SYSTEM "dblp.dtd"><dblp>'
+                          + body + '</dblp>').encode())
+        self.sha = hashlib.sha256(self.archive.read_bytes()).hexdigest()
+        self.md5 = hashlib.md5(self.dtd.read_bytes()).hexdigest()
+        self.manifest = catalogue(self.archive, self.dtd, self.root / 'catalogue',
+                                  self.sha, self.md5)
+        self.catalogue = Path(self.manifest['catalogue'])
+        self.manifest_path = self.catalogue.with_name('dblp-ai-candidates.manifest.json')
+
+    def run_recovery(self, **kwargs):
+        options = dict(expected_sha=self.sha, expected_dtd_md5=self.md5)
+        options.update(kwargs)
+        return recover(self.archive, self.dtd, self.catalogue, self.root / 'recovered',
+                       ['ifaamas'], **options)
+
+    def test_fresh_discovery_and_recovery_from_old_candidate_omission(self):
+        self.assertEqual(self.manifest['candidate_family_counts']['ifaamas'], 3)
+        # Simulate an older filter without altering its complete catalogue.
+        manifest = dict(self.manifest)
+        manifest['candidate_family_counts'] = {}
+        self.manifest_path.write_text(json.dumps(manifest))
+        candidate = Path(manifest['file'])
+        candidate.write_text('old incomplete candidate file\n')
+        before = {p: p.read_bytes() for p in [self.archive, self.dtd, self.catalogue,
+                                             self.manifest_path, candidate]}
+        with patch('urllib.request.urlopen', side_effect=AssertionError('No network')):
+            report = self.run_recovery()
+        rows = [json.loads(line) for line in Path(report['file']).read_text().splitlines()]
+        self.assertEqual(report['recovered_publications'], 3)
+        self.assertEqual(report['recovered_volumes'], 2)
+        self.assertEqual(report['recovered_records'], 5)
+        self.assertEqual(report['records_scanned'], 6)
+        self.assertEqual(report['old_candidate_family_counts'], {'ifaamas': 0})
+        self.assertFalse(report['database_written'])
+        self.assertEqual(next(r for r in rows if r['key'].endswith('/keynote'))['fields']['author'], ['René'])
+        self.assertTrue(any(r['key'].endswith('/workshop') for r in rows))
+        self.assertTrue(all(not ({'source', 'status', 'venue', 'track'} & r.keys()) for r in rows))
+        publication = next(r for r in rows if r['key'].endswith('/paper'))
+        volume = next(r for r in rows if r['key'] == 'conf/ifaamas/2025')
+        self.assertEqual(scope(publication, volume), (None, 'outside_target_family'))
+        self.assertEqual({p: p.read_bytes() for p in before}, before)
+        self.assertEqual(hashlib.sha256(Path(report['file']).read_bytes()).hexdigest(),
+                         report['file_sha256'])
+        # Raw recovery files cannot accidentally enter the normalized writer.
+        with patch('paper_atlas.ingestion.records.open_store') as writer:
+            with self.assertRaisesRegex(ValueError, 'source manifest'):
+                ingest_file(self.root / 'never.sqlite', report['file'])
+            writer.assert_not_called()
+        self.assertFalse((self.root / 'never.sqlite').exists())
+
+    def test_checksum_mismatch_does_not_publish(self):
+        for options in [{'expected_sha': '0' * 64}, {'expected_dtd_md5': '0' * 32}]:
+            with self.subTest(options=options), self.assertRaisesRegex(ValueError, 'checksum'):
+                self.run_recovery(**options)
+            self.assertFalse((self.root / 'recovered').exists())
+
+    def test_wrong_catalogue_snapshot_and_incomplete_totals_are_rejected(self):
+        for changes in [{'snapshot_sha256': '0' * 64}, {'dtd_md5': '0' * 32},
+                        {'records_scanned': 99}, {'record_types': {'proceedings': 2}}]:
+            manifest = {**self.manifest, **changes}
+            self.manifest_path.write_text(json.dumps(manifest))
+            with self.subTest(changes=changes), self.assertRaises(ValueError):
+                self.run_recovery()
+            self.assertFalse((self.root / 'recovered').exists())
+
+    def test_changed_family_count_does_not_publish_partial_result(self):
+        with sqlite3.connect(self.catalogue) as c:
+            c.execute("UPDATE family_counts SET count=count+1 WHERE family='ifaamas' AND record_type='inproceedings'")
+        with self.assertRaisesRegex(ValueError, 'family totals'):
+            self.run_recovery()
+        self.assertFalse((self.root / 'recovered').exists())
+
+    def test_changed_volume_metadata_is_detected(self):
+        with sqlite3.connect(self.catalogue) as c:
+            row = json.loads(c.execute("SELECT metadata_json FROM proceedings WHERE key='conf/ifaamas/2025'").fetchone()[0])
+            row['fields']['title'] = ['Wrong volume']
+            c.execute("UPDATE proceedings SET metadata_json=? WHERE key='conf/ifaamas/2025'", (json.dumps(row),))
+        with self.assertRaisesRegex(ValueError, 'proceedings differs'):
+            self.run_recovery()
+        self.assertFalse((self.root / 'recovered').exists())
+
+    def test_truncated_gzip_never_publishes(self):
+        self.archive.write_bytes(self.archive.read_bytes()[:-8])
+        revised_sha = hashlib.sha256(self.archive.read_bytes()).hexdigest()
+        self.manifest_path.write_text(json.dumps({**self.manifest, 'snapshot_sha256': revised_sha}))
+        with self.assertRaises((EOFError, OSError)):
+            self.run_recovery(expected_sha=revised_sha)
+        self.assertFalse((self.root / 'recovered').exists())
+
+    def test_existing_output_is_preserved_and_unknown_family_is_rejected(self):
+        out = self.root / 'recovered'
+        out.mkdir()
+        sentinel = out / 'keep'
+        sentinel.write_text('existing')
+        with self.assertRaisesRegex(ValueError, 'already exist'):
+            self.run_recovery()
+        self.assertEqual(sentinel.read_text(), 'existing')
+        for names in [[], ['../ifaamas'], ['not-a-supported-family'], [None]]:
+            with self.subTest(names=names), self.assertRaises(ValueError):
+                recover(self.archive, self.dtd, self.catalogue, self.root / 'other', names)
+
+    def test_requested_family_missing_from_catalogue_is_explicit(self):
+        with self.assertRaisesRegex(ValueError, 'absent from complete catalogue'):
+            recover(self.archive, self.dtd, self.catalogue, self.root / 'recovered',
+                    ['ifaamas', 'icml'], expected_sha=self.sha, expected_dtd_md5=self.md5)
+        self.assertFalse((self.root / 'recovered').exists())
+
+
+if __name__ == '__main__':
+    unittest.main()
@@ -0,0 +1,174 @@
+import copy
+from collections import Counter
+import gzip
+import hashlib
+import json
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+from urllib.parse import parse_qs, urlsplit
+
+from paper_atlas.collectors import neurips_program as program
+
+FIXTURES = Path(__file__).parent/'fixtures'
+RAW = gzip.decompress((FIXTURES/'neurips2026-program.json.gz').read_bytes())
+RECEIPTS = json.loads((FIXTURES/'neurips2026-receipts.json').read_text())
+EVIDENCE = {'url': program.URL, 'observed_at': RECEIPTS['receipts'][0]['finished_at']}
+
+
+class NeuripsProgramTests(unittest.TestCase):
+    def sample(self):
+        rows = json.loads(RAW)['results']
+        oral = next(r for r in rows if r['id'] == 154866)
+        poster = next(r for r in rows if r['paper_url'] == oral['paper_url'] and r['event_type'] == 'Poster')
+        return copy.deepcopy(poster), copy.deepcopy(oral)
+
+    def payload(self, rows):
+        return json.dumps({'count': len(rows), 'next': None, 'previous': None, 'results': rows})
+
+    def snapshots(self, folder):
+        folder.mkdir()
+        (folder/'program-json.response').write_bytes(RAW)
+        (folder/'receipts.json').write_text(json.dumps(RECEIPTS))
+        return folder
+
+    def test_full_snapshot_tracks_and_all_titles_authors_affiliations(self):
+        rows, held, duplicates, folded = program.parse_program(RAW, EVIDENCE)
+        self.assertEqual((len(rows), len(held), len(duplicates), folded), (9004, 139, 1, 132))
+        self.assertEqual(Counter(r['track'] for r in rows),
+                         {'main': 7896, 'evaluations_datasets': 971, 'position': 137})
+        expected = {}
+        for e in json.loads(RAW)['results']:
+            if e['sourceurl'] in program.GROUPS and e['event_type'] == 'Poster':
+                forum = parse_qs(urlsplit(e['paper_url']).query)['id'][0]
+                expected[forum] = e
+        self.assertEqual(len(expected), 9004)
+        for r in rows:
+            e = expected[r['identifiers']['openreview']]
+            self.assertEqual(r['title'], e['name'])
+            self.assertEqual(r['authors'], [a['fullname'] for a in e['authors']])
+            self.assertEqual(r['metadata']['author_affiliations'], [a.get('institution') for a in e['authors']])
+            self.assertEqual(r['raw_decision'], e['decision'])
+            self.assertEqual(r['status'], 'accepted')
+            self.assertNotIn('decision_date', r)
+            self.assertIsNone(r['abstract'])
+            self.assertEqual(r['year'], 2026)
+        self.assertEqual(duplicates[0]['event_id'], 154866)
+
+    def test_same_forum_folds_presentations_and_preserves_signed_source_ids(self):
+        poster, oral = self.sample()
+        rows, held, duplicates, folded = program.parse_program(self.payload([oral, poster, oral]), EVIDENCE)
+        row, = rows
+        self.assertEqual((len(held), len(duplicates), folded), (0, 1, 1))
+        self.assertEqual({p['sourceid'] for p in row['metadata']['presentations']},
+                         {poster['sourceid'], oral['sourceid']})
+        self.assertLess(oral['sourceid'], 0)
+        self.assertEqual({t['value'] for t in row['tags']}, {'Poster', 'Oral'})
+        self.assertTrue(all(t['namespace'] == 'neurips_program:presentation_format' for t in row['tags']))
+
+    def test_identical_title_and_authors_in_distinct_forums_remain_distinct(self):
+        poster, _ = self.sample(); other = copy.deepcopy(poster)
+        other.update(id=999999, uid='0'*32, sourceid=999999, paper_url='https://openreview.net/forum?id=otherForum7',
+                     virtualsite_url='/virtual/2026/poster/999999')
+        rows, _, _, _ = program.parse_program(self.payload([poster, other]), EVIDENCE)
+        self.assertEqual(len(rows), 2)
+        self.assertEqual(len({r['external_id'] for r in rows}), 2)
+
+    def test_conflicting_event_ids_or_same_forum_bibliography_are_rejected(self):
+        poster, oral = self.sample()
+        for field in ('name', 'authors', 'sourceurl', 'decision'):
+            changed = copy.deepcopy(oral)
+            if field == 'authors':
+                changed['authors'][0]['fullname'] = 'Different Author'
+            elif field == 'sourceurl':
+                changed[field] = 'https://openreview.net/group?id=NeurIPS.cc/2026/Position_Paper_Track'
+            elif field == 'decision':
+                changed[field] = 'Accept (poster)'
+            else:
+                changed[field] += ' changed'
+            with self.subTest(field=field), self.assertRaises(ValueError):
+                program.parse_program(self.payload([poster, changed]), EVIDENCE)
+        changed = copy.deepcopy(poster); changed['name'] += ' changed'
+        with self.assertRaises(ValueError):
+            program.parse_program(self.payload([poster, changed]), EVIDENCE)
+
+    def test_unreviewed_origin_decision_visibility_and_link_are_held(self):
+        poster, _ = self.sample()
+        mutations = [{'sourceurl': 'journal-track/NeurIPS_2026-Posters_TMLR.xlsx'},
+                     {'sourceurl': 'neurips2026-bridging-the-future'},
+                     {'sourceurl': None}, {'sourceurl': 'https://openreview.net/group?id=NeurIPS.cc/2027/Conference'},
+                     {'decision': 'Reject'}, {'visible': False},
+                     {'paper_url': 'https://other.example/forum?id=anotherID7'}]
+        for mutation in mutations:
+            changed = copy.deepcopy(poster); changed.update(id=999999, virtualsite_url='/virtual/2026/poster/999999')
+            changed.update(mutation)
+            rows, held, _, _ = program.parse_program(self.payload([poster, changed]), EVIDENCE)
+            self.assertEqual((len(rows), len(held)), (1, 1))
+            self.assertEqual(held[0]['raw_event'], changed)
+
+    def test_rejects_truncation_pagination_invalid_url_and_missing_authors(self):
+        poster, _ = self.sample()
+        raw = json.loads(self.payload([poster]))
+        bad = [dict(raw, count=2), dict(raw, next='https://example.com/next'),
+               {k: v for k, v in raw.items() if k != 'previous'}]
+        for payload in bad:
+            with self.assertRaises(ValueError):
+                program.parse_program(json.dumps(payload), EVIDENCE)
+        for mutation in [{'authors': []}, {'virtualsite_url': '/virtual/2025/poster/154865'}, {'id': True}]:
+            changed = copy.deepcopy(poster); changed.update(mutation)
+            with self.assertRaises(ValueError):
+                program.parse_program(self.payload([changed]), EVIDENCE)
+
+    def test_frozen_collection_is_partial_and_no_http_or_database_write(self):
+        with tempfile.TemporaryDirectory() as tmp, patch.object(program, 'fetch', side_effect=AssertionError('No HTTP')):
+            root = Path(tmp); source = self.snapshots(root/'source')
+            report = program.collect(root/'out', snapshots=source)
+            self.assertEqual(report['records'], 9004)
+            self.assertEqual(len(report['files']), 3)
+            self.assertFalse(report['database_written'])
+            self.assertFalse(report['pdfs_downloaded'])
+            for m in report['files']:
+                self.assertEqual(m['status'], 'partial')
+                self.assertEqual(m['missing_abstract_count'], m['count'])
+                self.assertEqual(hashlib.sha256(Path(m['file']).read_bytes()).hexdigest(), m['file_sha256'])
+            with self.assertRaises(ValueError):
+                program.collect(root/'out', snapshots=source)
+
+    def test_frozen_receipt_tamper_path_symlink_status_and_duplicate(self):
+        for mode in ('tamper', 'path', 'symlink', 'status', 'duplicate'):
+            with self.subTest(mode=mode), tempfile.TemporaryDirectory() as tmp:
+                root = Path(tmp); source = self.snapshots(root/'source')
+                receipts = copy.deepcopy(RECEIPTS)
+                if mode == 'tamper':
+                    (source/'program-json.response').write_text('bad')
+                elif mode == 'path':
+                    receipts['receipts'][0]['file'] = '../outside.json'
+                elif mode == 'symlink':
+                    (root/'outside.json').write_bytes(RAW)
+                    (source/'program-json.response').unlink()
+                    (source/'program-json.response').symlink_to(root/'outside.json')
+                elif mode == 'status':
+                    receipts['receipts'][0]['status'] = 403
+                else:
+                    receipts['receipts'].append(copy.deepcopy(receipts['receipts'][0]))
+                (source/'receipts.json').write_text(json.dumps(receipts))
+                with self.assertRaises(ValueError):
+                    program.collect(root/'out', snapshots=source)
+                self.assertFalse((root/'out/neurips-program-2026.json').exists())
+
+    def test_online_mode_only_requests_fixed_program_metadata(self):
+        poster, _ = self.sample(); raw = self.payload([poster]).encode()
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp); cache = root/'cache'; cache.mkdir()
+            receipt = {'retrieved_at': EVIDENCE['observed_at'], 'url': program.URL}
+            (cache/hashlib.sha256(program.URL.encode()).hexdigest()).with_suffix('.json').write_text(json.dumps(receipt))
+            with patch.object(program, 'fetch', return_value=raw) as fetch:
+                report = program.collect(root/'out', cache=cache)
+            fetch.assert_called_once_with(program.URL, cache, program.LIMIT, delay=1)
+            self.assertEqual(report['records'], 1)
+
+    def test_frozen_fixture_preserves_actual_response_bytes(self):
+        receipt, = RECEIPTS['receipts']
+        self.assertEqual(hashlib.sha256(RAW).hexdigest(), receipt['sha256'])
+        self.assertEqual(len(RAW), receipt['bytes'])
@@ -0,0 +1,148 @@
+import gzip
+import hashlib
+import json
+import os
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from paper_atlas.audits.source_identity import audit, classify, load_inputs
+from paper_atlas.postgres.client import ReadConnection
+
+
+def record(**updates):
+    row = dict(source='reviewed_program', external_id='2026/forum-a',
+               title='A sufficiently specific scientific paper title', authors=['Jane Doe', 'John Smith'],
+               venue='NeurIPS', year=2026, track='main', status='accepted', identifiers={},
+               source_record_hits=[], identifier_hits=[], title_candidate_ids=[])
+    row.update(updates)
+    return row
+
+
+def write_input(root, rows):
+    path = root / 'source.jsonl'
+    raw = b''.join((json.dumps(row) + '\n').encode() for row in rows)
+    path.write_bytes(raw)
+    manifest = {k: rows[0][k] for k in ('source', 'venue', 'year', 'track')}
+    manifest.update(count=len(rows), file_sha256=hashlib.sha256(raw).hexdigest(),
+                    file='/ignored/untrusted/location.jsonl', status='partial')
+    path.with_suffix('.manifest.json').write_text(json.dumps(manifest))
+    return path
+
+
+class CandidateTests(unittest.TestCase):
+    def test_duplicate_names_do_not_equal_distinct_author_multiset(self):
+        row = record(authors=['Jane Doe', 'Jane Doe'], title_candidate_ids=[7])
+        result = classify(row, {7: {'title': row['title'], 'authors': ['Jane Doe']}})
+        self.assertEqual(result['kind'], 'partial_author_overlap_requires_review')
+        self.assertIn('duplicate_normalized_author_names', result['flags'])
+        self.assertFalse(result['comparisons'][0]['nonempty_author_multiset_equal'])
+
+    def test_arxiv_author_representation_and_order_match_but_never_approve(self):
+        row = record(title_candidate_ids=[7])
+        result = classify(row, {7: {'title': row['title'], 'authors': [['Smith', 'John', ''], ['Doe', 'Jane', '']]}})
+        self.assertEqual(result['kind'], 'one_exact_title_author_multiset_requires_review')
+        self.assertEqual(result['decision'], 'not_approved_for_import')
+
+    def test_binding_conflict_overrides_existing_source(self):
+        row = record(source_record_hits=[{'paper_id': 7}], identifier_hits=[{'paper_id': 8}])
+        entities = {pid: {'title': row['title'], 'authors': row['authors']} for pid in (7, 8)}
+        self.assertEqual(classify(row, entities)['kind'], 'conflicting_existing_bindings_hold')
+
+    def test_wrong_doi_binding_retains_author_title_disagreement(self):
+        row = record(identifier_hits=[{'paper_id': 7}])
+        result = classify(row, {7: {'title': 'Completely different work', 'authors': ['Somebody Else']}})
+        self.assertEqual(result['kind'], 'existing_identifier_binding_requires_review')
+        self.assertFalse(result['comparisons'][0]['normalized_title_equal'])
+        self.assertEqual(result['comparisons'][0]['author_name_overlap'], [])
+
+    def test_no_hit_is_not_novelty_and_missing_authors_never_agree(self):
+        self.assertEqual(classify(record(), {})['kind'], 'no_exact_match_not_proof_of_novelty')
+        row = record(authors=[], title_candidate_ids=[7])
+        result = classify(row, {7: {'title': row['title'], 'authors': []}})
+        self.assertEqual(result['kind'], 'no_author_agreement_hold')
+
+    def test_title_candidate_cap_is_explicit(self):
+        row = record(title_candidate_ids=list(range(21)))
+        entities = {pid: {'title': row['title'], 'authors': row['authors']} for pid in range(21)}
+        self.assertTrue(classify(row, entities)['title_candidates_truncated'])
+        self.assertEqual(classify(row, entities)['kind'], 'title_candidate_limit_reached_hold')
+
+
+class InputTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.root = Path(self.temp.name)
+
+    def test_raw_line_hash_and_manifest_path_not_followed(self):
+        path = write_input(self.root, [record()])
+        rows, inputs = load_inputs([path])
+        self.assertEqual(rows[0]['raw_line_sha256'], hashlib.sha256(path.read_bytes()).hexdigest())
+        self.assertEqual(inputs[0]['count'], 1)
+
+    def test_hash_tamper_rejected_before_database_connection(self):
+        path = write_input(self.root, [record()])
+        path.write_bytes(path.read_bytes().replace(b'Jane Doe', b'Jane Roe'))
+        with patch('paper_atlas.audits.source_identity.ReadConnection') as connection:
+            with self.assertRaisesRegex(ValueError, 'hash mismatch'):
+                audit([path], self.root / 'out', pg_service='test', pg_schema='research')
+            connection.assert_not_called()
+
+    def test_manifest_scope_and_count_tamper_and_duplicate_sources(self):
+        path = write_input(self.root, [record()])
+        manifest_path = path.with_suffix('.manifest.json')
+        manifest = json.loads(manifest_path.read_text())
+        for key, value in [('year', 2025), ('count', 2)]:
+            altered = dict(manifest, **{key: value})
+            manifest_path.write_text(json.dumps(altered))
+            with self.assertRaises(ValueError):
+                load_inputs([path])
+        manifest_path.write_text(json.dumps(manifest))
+        with self.assertRaisesRegex(ValueError, 'Duplicate source'):
+            load_inputs([path, path])
+
+    def test_existing_output_never_replayed_and_failure_not_completion(self):
+        path = write_input(self.root, [record()])
+        out = self.root / 'out'
+        with patch('paper_atlas.audits.source_identity.ReadConnection', side_effect=RuntimeError('unavailable')):
+            with self.assertRaises(RuntimeError):
+                audit([path], out, pg_service='test', pg_schema='research')
+        self.assertTrue((out / 'failed.json').exists())
+        self.assertFalse((out / 'postgresql-preflight.json').exists())
+        with self.assertRaises(FileExistsError):
+            audit([path], out, pg_service='test', pg_schema='research')
+
+
+@unittest.skipUnless(os.environ.get('PAPER_ATLAS_TEST_PG_SERVICE'), 'Requires the authorized remote PostgreSQL service')
+class ReadOnlyIntegrationTests(unittest.TestCase):
+    def test_existing_binding_full_evidence_and_no_import_approval(self):
+        service = os.environ['PAPER_ATLAS_TEST_PG_SERVICE']
+        connection = ReadConnection(service, 'research')
+        try:
+            self.assertEqual(connection.execute('SHOW transaction_read_only').fetchone()[0], 'on')
+            hit = connection.execute("SELECT value,paper_id FROM identifiers WHERE scheme='arxiv' ORDER BY value LIMIT 1").fetchone()
+            self.assertIsNotNone(hit)
+            paper = connection.execute('SELECT title,authors_json FROM papers WHERE id=?', (hit['paper_id'],)).fetchone()
+        finally:
+            connection.close()
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            path = write_input(root, [record(title=paper['title'], authors=json.loads(paper['authors_json']),
+                                            identifiers={'arxiv': hit['value']})])
+            report = audit([path], root / 'out', pg_service=service, pg_schema='research')
+            self.assertTrue(report['transaction_read_only'])
+            self.assertFalse(report['database_written'])
+            self.assertEqual(report['approved_for_import'], 0)
+            row, = report['entries']
+            self.assertIn(hit['paper_id'], [r['paper_id'] for r in row['identifier_hits']])
+            shard = report['entities'][str(hit['paper_id'])]
+            raw = (root / 'out' / shard['file']).read_bytes()
+            self.assertEqual(hashlib.sha256(raw).hexdigest(), shard['sha256'])
+            packet = json.loads(gzip.decompress(raw))
+            self.assertEqual(packet['paper']['id'], hit['paper_id'])
+            self.assertEqual(packet['paper']['title'], paper['title'])
+            self.assertEqual(packet['paper']['authors_json'], paper['authors_json'])
+            self.assertIn('attempts', packet)
+            self.assertTrue((root / 'out/postgresql-preflight.json').is_file())
@@ -0,0 +1,160 @@
+"""Real 49-chapter snapshots compared with a separately frozen metadata audit."""
+import copy
+import gzip
+import hashlib
+import json
+from pathlib import Path
+import sqlite3
+import tempfile
+import unittest
+
+from paper_atlas.collectors.springer_selected import (
+    BOOK, SOURCE, collect, parse_chapter, parse_toc, read_page, snapshot_index,
+)
+from paper_atlas.db.store import Store
+from paper_atlas.ingestion.records import ingest_file
+
+FIXTURE = Path(__file__).parent/'fixtures/springer-aistats1993-fixture.json.gz'
+
+
+class SpringerSelectedTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        raw = FIXTURE.read_bytes()
+        if hashlib.sha256(raw).hexdigest() != '94028fd37080f9d8027e5790e379cc151ee6eab17ed0d1dcc0f59e6e75f9d4d0':
+            raise AssertionError('Fixed source fixture differs')
+        cls.fixture = json.loads(gzip.decompress(raw))
+
+    def snapshots(self, root):
+        for group in ['toc','chapters']:
+            folder = root/group
+            folder.mkdir()
+            for name, text in self.fixture[group].items():
+                (folder/name).write_text(text)
+        return {'toc_snapshots':root/'toc','chapter_snapshots':root/'chapters'}
+
+    def test_all49_original_fields_against_independent_audit(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            report = collect(root/'out', **self.snapshots(root))
+            manifest, = report['files']
+            rows = [json.loads(s) for s in Path(manifest['file']).read_text().splitlines()]
+            self.assertEqual(len(rows), 49)
+            self.assertEqual(manifest['status'], 'partial')
+            self.assertEqual(manifest['missing_abstract_count'], 0)
+            self.assertTrue(manifest['excluded_materials'])
+            self.assertTrue(all(x['title'] == 'Front Matter' for x in manifest['excluded_materials']))
+            for row, expected in zip(rows, self.fixture['expected']):
+                for field in ['title','abstract']:
+                    self.assertEqual(row[field], expected[field])
+                self.assertEqual(row['identifiers'], {'doi':expected['doi']})
+                self.assertEqual(row['submission_identifiers'], [{'scheme':'publication_doi','value':expected['doi']}])
+                self.assertEqual(row['external_id'], expected['doi'])
+                self.assertEqual(row['metadata']['authors_citation_raw'], expected['authors_citation_raw'])
+                self.assertEqual(row['authors'], [{'family':s.split(',',1)[0].strip(), 'given':s.split(',',1)[1].strip()} for s in expected['authors_citation_raw']])
+                self.assertEqual(row['metadata']['first_page'], expected['first_page'])
+                self.assertEqual(row['metadata']['last_page'], expected['last_page'])
+                self.assertEqual((row['year'],row['metadata']['bibliographic_year']), (1993,1994))
+                self.assertEqual(row['track'], 'selected_volume')
+                self.assertEqual(row['event_key'], 'aistats-1993-selected-volume-1994')
+                self.assertNotIn('decision_date', row)
+                self.assertIn('original 1993 review decision not supplied', row['raw_decision'])
+                for e in [row['metadata']['source_evidence'], *row['metadata']['scope_evidence']]:
+                    self.assertEqual(hashlib.sha256((root/'out'/e['file']).read_bytes()).hexdigest(), e['sha256'])
+            self.assertFalse(report['database_written'])
+            with self.assertRaises(FileExistsError):
+                collect(root/'out', toc_snapshots=root/'toc', chapter_snapshots=root/'chapters')
+
+    def test_native_history_namespaces_and_idempotent_import(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            report = collect(root/'out', **self.snapshots(root))
+            manifest, = report['files']
+            first = json.loads(Path(manifest['file']).read_text().splitlines()[0])
+            db = root/'sample.sqlite'
+            store = Store(db)
+            pid, rid = store.record('prior_public','earlier','https://example.org/earlier',first['title'],first['abstract'],first['authors'])
+            store.decision(pid,rid,'AISTATS',1993,'main','earlier','https://example.org/earlier','rejected','original public rejection')
+            # Explicitly verified work identity; the short title alone must not
+            # force a merge or attach this publication DOI to the old attempt.
+            store.conn.execute('INSERT INTO identifiers VALUES(?,?,?)', ('doi',first['identifiers']['doi'],pid))
+            store.conn.commit()
+            old = store.conn.execute('SELECT * FROM decisions').fetchall()
+            store.conn.close()
+            ingest_file(db, manifest['file'])
+            ingest_file(db, manifest['file'])
+            with sqlite3.connect(db) as c:
+                self.assertEqual(c.execute('SELECT * FROM decisions WHERE status=\'rejected\'').fetchall(), [tuple(r) for r in old])
+                self.assertEqual(c.execute('SELECT count(*) FROM papers').fetchone()[0], 49)
+                self.assertEqual(c.execute('SELECT count(*) FROM submissions').fetchone()[0], 50)
+                self.assertEqual(c.execute('SELECT count(*) FROM decisions').fetchone()[0], 50)
+                self.assertEqual(c.execute('SELECT e.year,e.track,e.event_key FROM editions e ORDER BY e.id').fetchall(), [(1993,'main',''),(1993,'selected_volume','aistats-1993-selected-volume-1994')])
+                self.assertEqual(c.execute('SELECT count(*) FROM paper_tags pt JOIN tags t ON t.id=pt.tag_id WHERE t.namespace=?', (SOURCE+':publication_format',)).fetchone()[0], 49)
+                self.assertEqual(c.execute('SELECT count(*) FROM submission_identifiers WHERE scheme=\'publication_doi\'').fetchone()[0], 49)
+                self.assertIsNone(c.execute("SELECT 1 FROM submission_identifiers si JOIN submissions s ON s.id=si.submission_id JOIN editions e ON e.id=s.edition_id WHERE e.track='main'").fetchone())
+
+    def test_short_unanchored_title_keeps_identity_unconfirmed(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            store = Store(Path(tmp)/'identity.sqlite')
+            title = 'Statistical strategy: step 1'
+            authors = [{'given':'D. J.','family':'Hand'}]
+            old, _ = store.record('prior_public','earlier','https://example.org/old',title,'Same abstract',authors)
+            new, _ = store.record(SOURCE,BOOK+'_1','https://link.springer.com/chapter/'+BOOK+'_1',title,'Same abstract',authors,{'doi':BOOK+'_1'})
+            self.assertNotEqual(old, new)
+            self.assertEqual(store.conn.execute('SELECT count(*) FROM match_candidates WHERE method=\'exact_title_unconfirmed\'').fetchone()[0], 1)
+            store.conn.close()
+
+    def test_receipt_hash_path_redirect_and_content_type_guards(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            self.snapshots(root)
+            index = snapshot_index(root/'chapters')
+            url = 'https://link.springer.com/chapter/'+BOOK+'_1'
+            self.assertTrue(read_page(index,url)[0])
+            for key, value in [('sha256','0'*64),('file','../escape.html'),('final_url','https://example.org/chapter'),('http_status',403),('headers',{'Content-Type':'application/pdf'})]:
+                changed = copy.deepcopy(index)
+                changed[url][1][key] = value
+                with self.subTest(key=key), self.assertRaises((ValueError,KeyError)):
+                    read_page(changed,url)
+            changed = copy.deepcopy(index)
+            changed[url][1]['final_url'] = url+'?page=2'
+            with self.assertRaises(ValueError):
+                read_page(changed,url)
+
+    def test_missing_or_duplicate_snapshot_stops_collection(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            args = self.snapshots(root)
+            path = root/'chapters/receipts.json'
+            d = json.loads(path.read_bytes())
+            original = copy.deepcopy(d)
+            d['responses'].pop()
+            path.write_text(json.dumps(d))
+            with self.assertRaises(ValueError):
+                collect(root/'out', **args)
+            self.assertFalse((root/'out').exists())
+            original['responses'].append(original['responses'][0])
+            path.write_text(json.dumps(original))
+            with self.assertRaises(ValueError):
+                snapshot_index(root/'chapters')
+
+    def test_book_doi_year_and_duplicate_citation_fields_reject_drift(self):
+        html = self.fixture['chapters']['chapter-01.html']
+        for key, value in [('citation_doi',BOOK+'_2'),('citation_inbook_title','Another Book'),('citation_publication_date','1993')]:
+            import re
+            changed, count = re.subn(r'(<meta[^>]*name="'+key+r'"[^>]*content=")[^"]*', lambda m:m[1]+value, html)
+            self.assertEqual(count, 1)
+            with self.subTest(key=key), self.assertRaises(ValueError):
+                parse_chapter(changed, BOOK+'_1')
+        with self.assertRaises(ValueError):
+            parse_chapter(html+'<meta name="citation_doi" content="'+BOOK+'_1">', BOOK+'_1')
+
+    def test_missing_abstract_is_explicit_and_does_not_use_other_page_text(self):
+        html = self.fixture['chapters']['chapter-01.html'].replace('id="Abs1-content"','id="not-an-abstract"')
+        self.assertIsNone(parse_chapter(html, BOOK+'_1')['abstract'])
+
+    def test_toc_ignores_frontmatter_but_rejects_duplicate_publications(self):
+        html = self.fixture['toc']['springer-toc1.html']
+        self.assertEqual(len(parse_toc(html)), 17)
+        with self.assertRaises(ValueError):
+            parse_toc(html+html)
Author	SHA1	Message	Date
codex-bot	fe3e2502a5	Add read-only source identity candidate audit with evidence shards Metadata integrity checks / metadata (push) Successful in 2m42s Details	2026-11-10 02:55:27 +08:00
Zayden-Jung	b675a38b97	Merge pull request '适配AAMAS2026官方目录并保留撤稿与撤回证据' (#49 ) from codex/aamas2026 into main Metadata integrity checks / metadata (push) Successful in 2m45s Details Reviewed-on: #49	2026-11-09 18:49:37 +00:00
Zayden-Jung	cb875b8eef	Merge pull request '封装NeurIPS2026节目采集，分离投稿与重复展示' (#50 ) from codex/neurips-program into main Metadata integrity checks / metadata (push) Successful in 2m34s Details Reviewed-on: #50	2026-11-09 18:49:29 +00:00
codex-bot	6df904e6d8	Collect official NeurIPS 2026 program with forum-level presentation deduplication Metadata integrity checks / metadata (push) Successful in 2m40s Details	2026-11-10 02:18:38 +08:00
codex-bot	57a557c66b	Merge current main and preserve scope and AAMAS documentation entries Metadata integrity checks / metadata (push) Successful in 2m27s Details	2026-11-10 01:12:45 +08:00
codex-bot	5b5e606f0c	Correct independent affiliation inventory and target section heading mutation Metadata integrity checks / metadata (push) Successful in 2m22s Details	2026-11-10 01:09:25 +08:00
Zayden-Jung	3ff9ee9a94	Merge pull request '限定2020起补漏，保留2000年以来来源与恢复清单' (#48 ) from codex/coverage-handoff-2020 into main Metadata integrity checks / metadata (push) Successful in 2m18s Details Reviewed-on: #48	2026-11-09 17:05:50 +00:00
codex-bot	9f84fd7caf	Separate italic title content from publisher status notices Metadata integrity checks / metadata (push) Failing after 2m6s Details	2026-11-10 01:04:46 +08:00
codex-bot	bf05d3cdc8	Keep day-only withdrawal evidence through native ingestion Metadata integrity checks / metadata (push) Failing after 2m6s Details	2026-11-10 01:02:54 +08:00
codex-bot	930df2a4d5	Preserve explicit withdrawal and byte-exact AAMAS fixture evidence Metadata integrity checks / metadata (push) Failing after 2m4s Details	2026-11-10 01:00:58 +08:00
codex-bot	37380e1d3b	Collect reviewed AAMAS 2026 tracks and hold explicit retraction evidence Metadata integrity checks / metadata (push) Failing after 2m10s Details	2026-11-10 00:57:09 +08:00
codex-bot	25ed4559ca	docs: set 2020 backfill scope and preserve recovery inventory since 2000 Metadata integrity checks / metadata (push) Successful in 2m9s Details	2026-11-10 00:50:15 +08:00
Zayden-Jung	e0ac62464a	Merge pull request '封装有来源核验的 Springer 会后选集采集器' (#47 ) from codex/springer-selected into main Metadata integrity checks / metadata (push) Successful in 2m22s Details Reviewed-on: #47	2026-11-09 16:47:08 +00:00
Zayden-Jung	e37fa46e42	Merge pull request '补入精确的 ACL 1963 AMTCL 年会摘要卷' (#46 ) from codex/acl1963 into main Metadata integrity checks / metadata (push) Successful in 2m9s Details Reviewed-on: #46	2026-11-09 16:47:02 +00:00
Zayden-Jung	70463c0250	Merge pull request '修复DBLP family候选漏采并提供快照恢复命令' (#45 ) from codex/dblp-ifaamas-discovery into main Metadata integrity checks / metadata (push) Successful in 2m12s Details Reviewed-on: #45	2026-11-09 16:46:56 +00:00
codex-bot	5c78a36fd5	Scope chapter publication identities and test explicit work anchors Metadata integrity checks / metadata (push) Successful in 2m20s Details	2026-11-10 00:04:35 +08:00
codex-bot	7facde6c94	Handle section front matter and preserve abstract paragraph boundaries Metadata integrity checks / metadata (push) Failing after 2m3s Details	2026-11-10 00:00:30 +08:00
codex-bot	c06981b244	Collect reviewed Springer postconference selected volumes Metadata integrity checks / metadata (push) Failing after 2m4s Details	2026-11-09 23:51:25 +08:00
codex-bot	fe0f08a550	Recover omitted DBLP discovery families from pinned snapshots Metadata integrity checks / metadata (push) Successful in 2m11s Details	2026-11-09 22:28:50 +08:00