Preserve failed-run checkpoint when snapshot receipts are invalid

Reviewed-on: #51

.gitea/workflows/check.yml

@@ -48,6 +48,15 @@ jobs:
               outside.mkdir()
               subprocess.run([str(venv/'bin/paper-atlas'),'--help'],cwd=outside,check=True)
               subprocess.run([python,'-I','-B','-m','paper_atlas','collect','cvf','--help'],cwd=outside,check=True)
+              subprocess.run([python,'-I','-B','-m','paper_atlas','collect','rss2026-proceedings','--help'],cwd=outside,check=True)
+              rss_fixture = json.loads(gzip.decompress((src/'tests/fixtures/rss2026-source.json.gz').read_bytes()))
+              rss_snapshots = root/'rss2026-snapshots'
+              rss_snapshots.mkdir()
+              for name, html in rss_fixture['pages'].items():
+                  (rss_snapshots/name).write_bytes(html.encode('utf-8'))
+              (rss_snapshots/'receipts.json').write_text(json.dumps(rss_fixture['receipts']))
+              subprocess.run([python,'-I','-B','-m','paper_atlas','collect','rss2026-proceedings','--snapshots',str(rss_snapshots),'--out',str(root/'installed-rss2026')],cwd=outside,check=True)
+              assert json.loads((root/'installed-rss2026/collection-report.json').read_text())['records'] == 210
               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)
@@ -62,6 +71,16 @@ jobs:
               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)
@@ -86,6 +105,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()"

README.md

@@ -8,6 +8,8 @@ Paper Atlas 将 arXiv、AI 会议论文集及公开投稿决定整理到可追
 
 当前主动补漏从 **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)。
@@ -114,3 +116,7 @@ CI 在已授权的远程 runner 上运行测试、构建 wheel，并从源码目
 整合后的运行版本依赖、交互约束和生产切换顺序见 [运行版本整合](docs/research-runtime.md)。
 
 已导入整卷的逐条轨道修正见 [edition-tracks 维护流程](docs/edition-tracks.md)：按完整证据生成只读计划，保留原投稿和决定 ID，事务性修正 edition、标识与覆盖，并防止旧整卷刷新覆盖分类。
+
+入库前可使用通用的[来源身份只读预检](docs/source-identity-audit.md)，导出候选论文、稳定标识和既有投稿决定证据；候选结果需要审核，不会自动批准导入。
+
+RSS 2026 已独立适配官方论文集与录用表；使用方法和与 2025 demo 分类的区别见 [RSS 2026 来源说明](docs/rss2026-proceedings.md)。

docs/aamas-proceedings.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。
 

docs/rss2026-proceedings.md

@@ -0,0 +1,21 @@
+# RSS 2026 论文集来源
+
+`collect rss2026-proceedings` 是独立的年度来源适配器，不依赖尚未合入主线的 RSS 2025 PR #43，也不改变其 demo 分类。使用包内 HTML、HTTP 工具和统一 CLI；不连接数据库，不下载 PDF。
+
+```sh
+paper-atlas collect rss2026-proceedings --cache /data/paper-index/cache/rss2026 \
+  --out /data/paper-index/collected/rss2026-native-v1
+# 固定证据复现：包含 receipts.json 和对应 HTML 文件
+paper-atlas collect rss2026-proceedings --snapshots /data/paper-index/collected/rss2026-primary-v1 \
+  --out /data/paper-index/collected/rss2026-native-replay-v1
+```
+
+核对的来源为 [第22卷论文集](https://www.roboticsproceedings.org/rss22/)、[录用表](https://roboticsconference.org/2026/program/papers/)、[CFP](https://roboticsconference.org/2026/information/cfp/) 和 [Sponsor Demos](https://roboticsconference.org/2026/program/demos/)。当前审查目录为210篇。CFP明确2026没有单独 demo paper 类别，Science/Systems论文放在 main；赞助商展示不作为论文，也不推断逐篇是否现场演示。投稿录用后才公开、评审不公开；此来源不能提供拒稿或评审全集。
+
+每篇保留出版社原始标题、完整作者顺序、摘要、日期、BibTeX、原始 DOI，以及独立节目标题、完整作者和缩略显示、session、显示链接。节目显示编号不是原始 OpenReview 投稿编号，不写入 submission_identifiers。六个缩略作者预览按完整隐藏副本恢复核验，不假设固定字符长度；六个姓名差异和一处 RLinf/RLux 标题差异按已审核原文指纹识别并保留，不推断人名别名或官方勘误。未知差异会停止。
+
+快照收据必须逐文件匹配固定URL、HTTP200、字节数、哈希和带时区的真实采集时间；拒绝路径穿越、重复收据和符号链接。缓存模式只请求已审核的固定公开HTML URL，访问限制由HTTP工具正常报错，不绕过。每份原始HTML按哈希保存到输出 evidence/。
+
+输出目录必须不存在；失败目录留存供检查，换新目录重试。`collection-report.json` 最后原子写入，缺少此文件不能视为完成。清单始终 partial；已采集不代表主库完整、身份已核验或科学作品已去重。DOI也是待核对的候选标识，入库前仍需实时 PostgreSQL 身份审计、版本/作者关联复核和保护基线；入库后独立验收，所有导入结束后再做全库去重。
+
+2026原始来源审计已冻结214份HTML且210篇均有摘要。回归 fixture 同时包含原始来源和独立生成的逐篇预期值，测试比较210篇完整标题/作者/摘要/DOI/日期；构建和测试仅在已授权 ces-control:host 远程 runner 进行。

docs/source-identity-audit.md

@@ -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 服务做真实只读候选查询，不往生产库添加测试记录。

paper_atlas/audits/source_identity.py

@@ -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))

paper_atlas/cli.py

@@ -23,6 +23,7 @@ COMMANDS = {
     ('collect','colm'): ('paper_atlas.collectors.proceedings', {'source':'colm'}, 'COLM 官方论文集'),
     ('collect','neurips'): ('paper_atlas.collectors.proceedings', {'source':'neurips'}, 'NeurIPS 官方论文集'),
     ('collect','mlsys'): ('paper_atlas.collectors.proceedings', {'source':'mlsys'}, 'MLSys 官方历年标题与摘要'),
+    ('collect','rss2026-proceedings'): ('paper_atlas.collectors.rss2026', {}, '核验 RSS 2026 官方论文集与独立录用表'),
     ('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 发表形式和摘要'),
@@ -46,6 +47,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', {}, '只读核对跨会议投稿历史及决定来源关系'),
 }
 

paper_atlas/collectors/aamas.py

@@ -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')

paper_atlas/collectors/aamas2026.py

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

paper_atlas/collectors/rss2026.py

@@ -0,0 +1,222 @@
+"""Reviewed RSS 2026 publisher metadata, with separate program evidence."""
+REVIEWED_AUTHOR_VARIANTS = {16: '04de7e813b7c68b5c1764e90fe41bcd2e45259d46749bb2cc54bac5150c8f25b', 27: '975f6adc4d15045f9c8829689a759c5e3a44eecb3410a18ff99b37226d9c8b97', 78: '8208d0950468295c2b73dae28658566b1c726f65a4167f490039cf3449aa2db4', 132: '31eee6289de404f3d859dd0d9de4dd40e968f86b89312de41df87ae01c529711', 199: 'ced0bc9e38b28b4a490799fd94315f104ece0df28be428b3cc417db8ea797b9d', 204: '7c6959e6668bf44af09959388f6b9e2d883a76106bf7f191dc9949368c969ee5'}
+import argparse
+from datetime import datetime, date
+import hashlib
+import json
+from pathlib import Path
+import re
+import unicodedata
+
+from paper_atlas.io.html import Document, Node, clean
+from paper_atlas.io.http import fetch
+
+SOURCE = 'rss_proceedings'
+ROOT = 'https://www.roboticsproceedings.org/rss22/'
+HEADING = 'Robotics: Science and Systems XXII'
+PAGES = {'index.html': ROOT,
+         'accepted.html': 'https://roboticsconference.org/2026/program/papers/',
+         'cfp.html': 'https://roboticsconference.org/2026/information/cfp/',
+         'sponsor-demos.html': 'https://roboticsconference.org/2026/program/demos/'}
+
+
+def digest(raw):
+    return hashlib.sha256(raw).hexdigest()
+
+
+def one(values):
+    if len(values) != 1:
+        raise ValueError('Expected one reviewed HTML field')
+    return values[0]
+
+
+def comparable(value):
+    return unicodedata.normalize('NFC', clean(value))
+
+
+def parse_lists(index_html, program_html, policy_html, sponsors_html):
+    policy = clean(Document(policy_html).root.text())
+    for text in ('RSS 2026 will not have a separate category for demo papers.',
+                 'papers will only be public after acceptance', 'reviews will not be publicly visible'):
+        if text not in policy:
+            raise ValueError('Reviewed 2026 publication/demo/privacy policy changed')
+    if [clean(n.text()) for n in Document(sponsors_html).root.walk() if n.tag == 'h1'] != ['Sponsor Demos']:
+        raise ValueError('Sponsor activity page changed; never classify these as papers')
+    nodes = list(Document(index_html).root.walk())
+    if [clean(n.text()) for n in nodes if n.tag == 'h1'] != [HEADING]:
+        raise ValueError('Unexpected publisher edition')
+    index = {}
+    for row in (n for n in nodes if n.tag == 'tr'):
+        links = [n for n in row.walk() if n.tag == 'a' and re.fullmatch(r'p[0-9]{3}\.html', n.attrs.get('href', ''))]
+        if not links:
+            continue
+        link = one(links); pid = int(link.attrs['href'][1:4])
+        if pid in index:
+            raise ValueError('Repeated publisher ID')
+        index[pid] = {'title': clean(link.text()), 'authors': clean(one([n for n in row.walk() if n.tag == 'i']).text())}
+    program = {}
+    if [clean(n.text()) for n in Document(program_html).root.walk() if n.tag == 'h1'] != ['Accepted Papers']:
+        raise ValueError('Unexpected accepted program')
+    for row in (n for n in Document(program_html).root.walk() if n.tag == 'tr'):
+        cells = [n for n in row.children if isinstance(n, Node) and n.tag == 'td']
+        if not cells:
+            continue
+        if len(cells) != 4 or not re.fullmatch('[1-9][0-9]*', clean(cells[0].text())):
+            raise ValueError('Unexpected program row')
+        pid = int(clean(cells[0].text()))
+        if pid in program:
+            raise ValueError('Repeated program ID')
+        preview = clean(' '.join(n for n in cells[3].children if isinstance(n, str)))
+        full = one([clean(n.text()) for n in cells[3].walk() if n.tag == 'div'
+                    and 'display:none' in n.attrs.get('style', '').replace(' ', '')])
+        if not preview or not full.startswith(preview):
+            raise ValueError('Author preview is not a prefix of complete author evidence')
+        links = [n.attrs.get('href') for n in cells[2].walk() if n.tag == 'a']
+        if links != [f'/2026/program/papers/{pid}/']:
+            raise ValueError('Program display link changed')
+        program[pid] = {'display_id': pid, 'title': clean(cells[2].text()), 'authors_complete': full,
+                        'authors_preview': preview, 'session': clean(cells[1].text()),
+                        'program_url': 'https://roboticsconference.org'+links[0]}
+    if set(index) != set(program) or set(index) != set(range(1, 211)):
+        raise ValueError('Reviewed 210-entry source census changed')
+    return index, program
+
+
+def parse_paper(html, pid, index, program):
+    nodes = list(Document(html).root.walk()); meta = {}
+    for n in nodes:
+        if n.tag == 'meta' and n.attrs.get('name', '').startswith('citation_'):
+            meta.setdefault(n.attrs['name'], []).append(n.attrs.get('content', ''))
+    title = one(meta.get('citation_title', [])); authors = meta.get('citation_author', [])
+    day = one(meta.get('citation_publication_date', []))
+    if not re.fullmatch(r'2026/[0-9]{2}/[0-9]{2}', day):
+        raise ValueError('Unexpected publication date')
+    date.fromisoformat(day.replace('/', '-'))
+    if (meta.get('citation_volume') != ['22'] or meta.get('citation_conference_title') != [HEADING]
+            or [clean(n.text()) for n in nodes if n.tag == 'h1'] != [HEADING]
+            or not authors or any(not clean(a) for a in authors)
+            or comparable(title) != comparable(index['title'])
+            or comparable(', '.join(authors)) != comparable(index['authors'])
+            or [comparable(n.text()) for n in nodes if n.tag == 'h3'] != [comparable(title)]):
+        raise ValueError('Publisher citation/index/visible metadata disagree')
+    pdf = one(meta.get('citation_pdf_url', []))
+    if pdf not in (ROOT+f'p{pid:03d}.pdf', ROOT.replace('https:', 'http:')+f'p{pid:03d}.pdf'):
+        raise ValueError('Citation PDF ID mismatch; PDFs are never fetched')
+    bib = one([n.text() for n in nodes if n.tag == 'pre'])
+    doi = one(re.findall(r'^\s*DOI\s*=\s*\{([^{}]+)\}', bib, re.I | re.M))
+    year = one(re.findall(r'^\s*YEAR\s*=\s*\{([^{}]+)\}', bib, re.I | re.M))
+    if doi != f'10.15607/RSS.2026.XXII.{pid:03d}' or year != '2026':
+        raise ValueError('BibTeX DOI/year mismatch')
+    variants = []
+    if comparable(', '.join(authors)) != comparable(program['authors_complete']):
+        key = digest(json.dumps([authors, program['authors_complete']], ensure_ascii=False, separators=(',', ':')).encode())
+        if REVIEWED_AUTHOR_VARIANTS.get(pid) != key:
+            raise ValueError('Unreviewed complete author variant')
+        variants.append('reviewed_program_author_name_variant_not_person_alias')
+    if comparable(title) != comparable(program['title']):
+        if (pid, title, program['title']) != (89,
+                'RLinf-VLA: A Unified and Efficient Framework for Reinforcement Learning of Vision-Language-Action Models',
+                'RLux-VLA: A Unified and Efficient Framework for Reinforcement Learning of Vision-Language-Action Models'):
+            raise ValueError('Unreviewed program title variant')
+        variants.append('reviewed_program_title_variant_not_erratum')
+    inside = False; parts = []; markers = []
+    for n in (n for n in nodes if n.tag == 'p'):
+        labels = [clean(b.text()) for b in n.walk() if b.tag == 'b']
+        if 'Abstract:' in labels:
+            inside = True; markers.append('start'); value = clean(n.text()).removeprefix('Abstract:').strip()
+            if value: parts.append(value)
+        elif 'Download:' in labels:
+            inside = False; markers.append('end')
+        elif inside:
+            parts.append(clean(n.text()))
+    if markers != ['start', 'end'] or not any(parts):
+        raise ValueError('Missing or changed publisher abstract layout')
+    return {'title': title, 'authors': authors, 'abstract': '\n\n'.join(v for v in parts if v),
+            'doi_raw': doi, 'publication_date': day, 'bibtex': bib, 'program_variants': variants}
+
+
+def collect(out, *, snapshots=None, cache=None):
+    if (snapshots is None) == (cache is None):
+        raise ValueError('Specify snapshots or cache')
+    out = Path(out)
+    if out.exists():
+        raise ValueError('Output directory must not exist; inspect partial/completed runs first')
+    out.parent.mkdir(parents=True, exist_ok=True)
+    out.mkdir()
+    evidence = {}; receipts = {}
+    if snapshots is not None:
+        root = Path(snapshots)
+        try:
+            for r in json.loads((root/'receipts.json').read_bytes()):
+                name = r['file']
+                if not isinstance(name, str) or Path(name).name != name or name in receipts:
+                    raise ValueError('Unsafe or repeated snapshot receipt')
+                receipts[name] = r
+        except Exception as error:
+            (out/'failed.json').write_text(json.dumps({'status': 'failed_not_complete', 'error_type': type(error).__name__, 'database_written': False})+'\n')
+            raise
+    def page(name, url):
+        if snapshots is not None:
+            path = root/name
+            if path.is_symlink(): raise ValueError('Snapshot symlink rejected')
+            raw = path.read_bytes(); r = receipts[name]
+            if (r['url'] != url or r.get('final_url', url) != url or r['http_status'] != 200
+                    or r['bytes'] != len(raw) or r['sha256'] != digest(raw)):
+                raise ValueError('Snapshot hash/receipt mismatch')
+            at = r['at']
+        else:
+            raw = fetch(url, cache, limit=8*1024*1024)
+            sidecar = json.loads((Path(cache)/digest(url.encode())).with_suffix('.json').read_bytes())
+            if sidecar['sha256'] != digest(raw): raise ValueError('Cache receipt mismatch')
+            at = sidecar['retrieved_at']
+        if len(raw) > 8*1024*1024 or datetime.fromisoformat(at.replace('Z', '+00:00')).utcoffset() is None:
+            raise ValueError('Invalid source size/timezone')
+        relative = 'evidence/'+digest(raw)+'.html'; target = out/relative
+        target.parent.mkdir(exist_ok=True); target.write_bytes(raw)
+        evidence[name] = {'file': relative, 'sha256': digest(raw), 'url': url, 'observed_at': at}
+        return raw.decode('utf-8')
+    try:
+        pages = {name: page(name, url) for name, url in PAGES.items()}
+        index, program = parse_lists(pages['index.html'], pages['accepted.html'], pages['cfp.html'], pages['sponsor-demos.html'])
+        records = []
+        for pid in sorted(index):
+            name = f'p{pid:03d}.html'; fields = parse_paper(page(name, ROOT+name), pid, index[pid], program[pid])
+            row = {'source': SOURCE, 'external_id': f'2026/p{pid:03d}', 'url': ROOT+name,
+                   'title': fields['title'], 'abstract': fields['abstract'], 'authors': fields['authors'],
+                   'identifiers': {'doi': fields['doi_raw'].lower()}, 'venue': 'RSS', 'venue_aliases': [],
+                   'year': 2026, 'track': 'main', 'status': 'accepted',
+                   'raw_decision': 'listed in official RSS 2026 proceedings and accepted papers',
+                   'observed_at': evidence[name]['observed_at'],
+                   'metadata': {'publisher_date': fields['publication_date'], 'publisher_doi_raw': fields['doi_raw'],
+                       'original_bibtex': fields['bibtex'], 'program': program[pid],
+                       'program_variants': fields['program_variants'], 'source_evidence': evidence[name],
+                       'index_evidence': evidence['index.html'], 'program_evidence': evidence['accepted.html'],
+                       'policy_evidence': evidence['cfp.html'],
+                       'scope': '2026 Science/Systems papers; no separate demo-paper category. Program display ID is not an original submission ID.'}}
+            records.append(row)
+        path = out/'rss-proceedings-2026-main.jsonl'
+        raw = b''.join((json.dumps(r, ensure_ascii=False)+'\n').encode() for r in records)
+        path.write_bytes(raw)
+        manifest = {'source': SOURCE, 'parser_version': 1, 'venue': 'RSS', 'year': 2026, 'track': 'main',
+                    'count': 210, 'file': str(path), 'file_sha256': digest(raw), 'list_url': ROOT,
+                    'status': 'partial', 'missing_abstract_count': 0}
+        path.with_suffix('.manifest.json').write_text(json.dumps(manifest, indent=2)+'\n')
+        report = {'status': 'collected_pending_identity_review', 'source': SOURCE, 'records': 210,
+                  'files': [manifest], 'source_evidence': evidence, 'database_written': False, 'pdfs_downloaded': False,
+                  'limitations': 'Publisher metadata only; not database import, scientific-work deduplication, private reviews or a public rejection set.'}
+        temporary = out/'collection-report.json.part'; temporary.write_text(json.dumps(report, ensure_ascii=False, indent=2)+'\n')
+        temporary.replace(out/'collection-report.json')
+        return report
+    except Exception as error:
+        (out/'failed.json').write_text(json.dumps({'status': 'failed_not_complete', 'error_type': type(error).__name__, 'database_written': False})+'\n')
+        raise
+
+
+def main(argv=None, *, prog=None):
+    p = argparse.ArgumentParser(prog=prog, description=__doc__)
+    p.add_argument('--out', required=True, type=Path)
+    g = p.add_mutually_exclusive_group(required=True)
+    g.add_argument('--snapshots', type=Path); g.add_argument('--cache', type=Path)
+    args = p.parse_args(argv)
+    report = collect(args.out, snapshots=args.snapshots, cache=args.cache)
+    print(json.dumps({'status': report['status'], 'records': report['records'], 'database_written': False}))

tests/test_aamas2026.py

@@ -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()))

tests/test_aamas_proceedings.py

@@ -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):

tests/test_rss2026.py

@@ -0,0 +1,100 @@
+import gzip
+import hashlib
+import json
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+from paper_atlas.collectors import rss2026 as rss
+
+FIXTURE = Path(__file__).parent/'fixtures/rss2026-source.json.gz'
+
+class Rss2026Tests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.fixture=json.loads(gzip.decompress(FIXTURE.read_bytes()))
+
+    def setUp(self):
+        self.tmp=tempfile.TemporaryDirectory();self.addCleanup(self.tmp.cleanup)
+        self.root=Path(self.tmp.name);self.snap=self.root/'snapshots';self.snap.mkdir()
+        for name,text in self.fixture['pages'].items():(self.snap/name).write_bytes(text.encode())
+        (self.snap/'receipts.json').write_text(json.dumps(self.fixture['receipts']))
+        self.pages=self.fixture['pages'];self.index,self.program=self.lists()
+
+    def lists(self,**changes):
+        pages={**self.pages,**changes}
+        return rss.parse_lists(*(pages[n] for n in rss.PAGES))
+
+    def test_all210_records_match_independent_primary_inventory_and_evidence(self):
+        report=rss.collect(self.root/'out',snapshots=self.snap)
+        self.assertEqual(report['records'],210);self.assertFalse(report['database_written'])
+        self.assertFalse(report['pdfs_downloaded']);self.assertEqual(len(report['source_evidence']),214)
+        manifest,=report['files'];self.assertEqual(manifest['status'],'partial')
+        rows=[json.loads(l) for l in Path(manifest['file']).read_text().splitlines()]
+        for row,gold in zip(rows,self.fixture['independent_gold']):
+            self.assertEqual(row['external_id'],f"2026/p{gold['paper_id']:03d}")
+            for key in ('title','authors','abstract'):self.assertEqual(row[key],gold[key])
+            self.assertEqual(row['identifiers'],{'doi':gold['doi_raw'].lower()})
+            self.assertEqual(row['metadata']['publisher_date'],gold['publication_date'])
+            self.assertNotIn('submission_identifiers',row);self.assertEqual(row['track'],'main')
+            e=row['metadata']['source_evidence'];self.assertEqual(hashlib.sha256((self.root/'out'/e['file']).read_bytes()).hexdigest(),e['sha256'])
+        self.assertEqual(manifest['file_sha256'],hashlib.sha256(Path(manifest['file']).read_bytes()).hexdigest())
+        self.assertEqual(manifest['missing_abstract_count'],0)
+
+    def test_full_hidden_authors_are_separate_from_six_short_previews(self):
+        shortened=[p for p in self.program.values() if p['authors_complete']!=p['authors_preview']]
+        self.assertEqual(len(shortened),6)
+        p=rss.parse_paper(self.pages['p087.html'],87,self.index[87],self.program[87])
+        self.assertEqual(len(p['authors']),55)
+        self.assertFalse(p['program_variants'])
+
+    def test_six_reviewed_name_variants_preserve_original_strings(self):
+        for pid in rss.REVIEWED_AUTHOR_VARIANTS:
+            p=rss.parse_paper(self.pages[f'p{pid:03d}.html'],pid,self.index[pid],self.program[pid])
+            self.assertIn('reviewed_program_author_name_variant_not_person_alias',p['program_variants'])
+            self.assertNotEqual(', '.join(p['authors']),self.program[pid]['authors_complete'])
+
+    def test_reviewed_title_variant_is_not_an_automatic_erratum(self):
+        p=rss.parse_paper(self.pages['p089.html'],89,self.index[89],self.program[89])
+        self.assertTrue(p['title'].startswith('RLinf-VLA'))
+        self.assertTrue(self.program[89]['title'].startswith('RLux-VLA'))
+        self.assertIn('reviewed_program_title_variant_not_erratum',p['program_variants'])
+
+    def test_unreviewed_author_or_title_variant_stops(self):
+        for changed in [{'authors_complete':'An Unrelated Person'},{'title':'A different paper'}]:
+            with self.assertRaisesRegex(ValueError,'Unreviewed'):
+                rss.parse_paper(self.pages['p001.html'],1,self.index[1],{**self.program[1],**changed})
+
+    def test_demo_policy_cannot_be_reused_from2025(self):
+        with self.assertRaisesRegex(ValueError,'policy'):
+            self.lists(**{'cfp.html':self.pages['cfp.html'].replace('will not have a separate category','will have a separate category')})
+        with self.assertRaises(ValueError):
+            self.lists(**{'sponsor-demos.html':self.pages['sponsor-demos.html'].replace('Sponsor Demos','Demo Papers')})
+
+    def test_wrong_doi_year_pdf_or_missing_abstract_is_rejected(self):
+        html=self.pages['p001.html']
+        for old,new in [('10.15607/RSS.2026.XXII.001','10.15607/RSS.2025.XXI.001'),
+                        ('2026/07/13','2025/07/13'),('rss22/p001.pdf','rss22/p002.pdf'),('Abstract:','Summary:')]:
+            self.assertIn(old,html)
+            with self.assertRaises(ValueError):rss.parse_paper(html.replace(old,new),1,self.index[1],self.program[1])
+
+    def test_index_census_and_preview_tampering_stop(self):
+        with self.assertRaises(ValueError):self.lists(**{'index.html':self.pages['index.html'].replace('p001.html','p999.html')})
+        bad=self.pages['accepted.html'].replace('Huayi Zhou, Kui Jia','Somebody Else',1)
+        with self.assertRaisesRegex(ValueError,'prefix'):self.lists(**{'accepted.html':bad})
+
+    def test_snapshot_tamper_has_no_completion_marker_and_never_fetches(self):
+        p=self.snap/'p001.html';p.write_bytes(p.read_bytes()+b'changed')
+        with patch.object(rss,'fetch',side_effect=AssertionError('network must not be used')):
+            with self.assertRaisesRegex(ValueError,'receipt'):rss.collect(self.root/'out',snapshots=self.snap)
+        self.assertFalse((self.root/'out/collection-report.json').exists())
+        self.assertTrue((self.root/'out/failed.json').exists())
+
+    def test_existing_output_and_unsafe_receipt_are_rejected(self):
+        out=self.root/'existing';out.mkdir()
+        with self.assertRaises(ValueError):rss.collect(out,snapshots=self.snap)
+        receipts=[*self.fixture['receipts'],{**self.fixture['receipts'][0],'file':'../outside'}]
+        (self.snap/'receipts.json').write_text(json.dumps(receipts))
+        with self.assertRaisesRegex(ValueError,'Unsafe'):rss.collect(self.root/'unsafe',snapshots=self.snap)
+        self.assertTrue((self.root/'unsafe/failed.json').exists())
+        self.assertFalse((self.root/'unsafe/collection-report.json').exists())

tests/test_source_identity.py

@@ -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())