Add read-only source identity candidate audit with evidence shards

Reviewed-on: #49

.gitea/workflows/check.yml

@@ -50,6 +50,16 @@ 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()))
@@ -86,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()"

README.md

@@ -70,6 +70,8 @@ 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 时会明确提示，避免静默遗漏。
@@ -114,3 +116,5 @@ CI 在已授权的远程 runner 上运行测试、构建 wheel，并从源码目
 整合后的运行版本依赖、交互约束和生产切换顺序见 [运行版本整合](docs/research-runtime.md)。
 
 已导入整卷的逐条轨道修正见 [edition-tracks 维护流程](docs/edition-tracks.md)：按完整证据生成只读计划，保留原投稿和决定 ID，事务性修正 edition、标识与覆盖，并防止旧整卷刷新覆盖分类。
+
+入库前可使用通用的[来源身份只读预检](docs/source-identity-audit.md)，导出候选论文、稳定标识和既有投稿决定证据；候选结果需要审核，不会自动批准导入。

docs/neurips-program.md

@@ -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 年起。期刊/展示关系、网页独有条目、摘要和公开负面决定仍有待办；冻结资料和代码适配不代表完整会议覆盖。

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

@@ -26,6 +26,7 @@ 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 的原始候选'),
@@ -45,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', {}, '只读核对跨会议投稿历史及决定来源关系'),
 }
 

paper_atlas/collectors/neurips_program.py

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

tests/fixtures/neurips2026-receipts.json

@@ -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"
+    }
+  ]
+}

tests/test_neurips_program.py

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

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