Recover omitted DBLP discovery families from pinned snapshots

Reviewed-on: #42

.gitea/workflows/check.yml

@@ -50,6 +50,8 @@ 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','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)
               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','maintain','postgres','--help'],cwd=outside,check=True)

README.md

@@ -68,7 +68,9 @@ paper-atlas collect colm --cache /data/paper-index/raw/http-cache \
 
 EWSL、EPS、ECP、EWSP 的[具体前身卷与完整快照要求](docs/coverage/historical-predecessors-2026-11-09.md)单独核验，保留原始系列及事件/出版年份；旧候选文件缺少 ECP/EWSP 时会明确提示，避免静默遗漏。
 
-AAMAS 的[官方论文集采集器](docs/aamas-proceedings.md)目前核验了 2012 版式，可从缓存或已校验快照采集标题、作者及页面提供的摘要；发表形式单独保存，全文/扩展摘要中未明确的投稿 track 保持待明确，不根据会话主题推定主会轨道。
+AAMAS 的[官方论文集采集器](docs/aamas-proceedings.md)支持已核验的 2012、2025 版式，可从缓存或已校验快照采集元数据。2025 年目录含 476 条非 keynote 出版物，保留完整作者和机构，HTML 未提供摘要时留空；发表形式与投稿 track 分开，不根据主题推定主会轨道或研究类型。
+
+旧 DBLP 候选文件可能漏掉新命名空间。`collect dblp-family` 可从已校验的固定压缩快照和完整目录中[恢复指定 family 的原始候选](docs/dblp-family-recovery.md)，例如 `ifaamas`；恢复文件保留卷、keynote 和 workshop 等原始条目，需另做来源范围与身份核验。
 
 `paper-atlas collect --help` 查看来源；`paper-atlas import --help` 查看导入方式。完整批次步骤见[使用指南](docs/usage.md)。
 

docs/aamas-proceedings.md

@@ -1,6 +1,14 @@
 # AAMAS 官方论文集元数据
 
-`paper_atlas.collectors.aamas` 是按需运行的采集适配器，通过统一入口 `collect aamas-proceedings` 使用。目前只支持已独立核验的 **AAMAS 2012** HTML 版式；其他年份需先核对官方目录和逐篇结构，再增加适配，不能仅替换年份。
+`paper_atlas.collectors.aamas` 是按需运行的采集适配器，通过统一入口 `collect aamas-proceedings` 使用。支持已独立核验的 **AAMAS 2012、2025** HTML 版式；其他年份需先核对官方目录和逐篇结构，再增加适配，不能仅替换年份。2025 解析单独封装在 `paper_atlas.collectors.aamas2025`。
+
+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。
+
+作者按逐篇 BR 行解析，斜体机构原文另存，保留 Unicode、顺序及完整名单。HTML 没有提供摘要或逐篇 DOI，故 `abstract=null`、标识字典为空；只保存 PDF 链接，不下载研究 PDF。不能把 Research Paper Track 的所有条目推成 main，不能把目录缺席推为 rejected/withdrawn，也不能把出版格式标签当研究类型。
+
+2025 用法：`paper-atlas collect aamas-proceedings --year 2025 --snapshots /data/review/aamas2025 --out /data/paper-index/collected/aamas-2025`；在线模式改用 `--cache /data/paper-index/cache/aamas2025`，只请求上述两个 HTML 页面。快照目录包含原始 `index.html`、`contents.html` 和 `receipts.json`，凭据必须列出各原始 URL、文件、字节数、SHA256、`http_status=200` 和带时区 `at`。已恢复的凭据需保留 `at_precision=filesystem_write_time_recovered` 等说明，不能伪造 HTTP 头或请求时间。2026-11-09 保存的两个页面通过哈希校验；独立录用清单 URL 返回 403，失败保留，不绕过或依赖其完成采集。
+
+2025 命令同样使用目录锁、临时目录、最后写入采集报告。五个 JSONL/manifest 均保持 `partial`，记录摘要缺失数量；它不连接数据库。首次采集需独立核验全部标题/作者/栏目和 source IDs，随后做 PostgreSQL 身份预检、必要的逐条版本审核、原生导入与独立验收。若失败且没有完成报告，应先检查原始响应和错误，再使用新输出目录重跑，不重放已提交批次。代码验证、原始目录盘点和采集成功均不代表已导入生产或最终去重完成。
 
 ## 来源与范围
 

docs/dblp-family-recovery.md

@@ -0,0 +1,37 @@
+# DBLP family 候选恢复
+
+DBLP 的会议命名空间会变化。旧候选过滤器包含 `atal`、`aamas`，遗漏了 `ifaamas`；2026-11-01 固定快照的完整目录实际已有 `conf/ifaamas/2025` 和 `conf/ifaamas/2026`。新目录盘点已把 `ifaamas` 纳入广泛候选范围，已有快照也可按需恢复，无需重新下载或重建完整目录。
+
+## 运行
+
+在已有元数据的服务器运行：
+
+```sh
+paper-atlas collect dblp-family \
+  --archive /data/paper-index/raw/dblp-2026-11-01.xml.gz \
+  --dtd /data/paper-index/raw/dblp-2023-06-28.dtd \
+  --catalogue /data/paper-index/collected/dblp/dblp-catalogue.sqlite \
+  --family ifaamas \
+  --out /data/paper-index/collected/ifaamas-recovered
+```
+
+`--family` 可以重复，但仅接受项目明确列出的候选 family。`--out` 必须尚不存在。同名输出由文件锁防止并发恢复；任务运行中应观测原任务，不能因为日志间隔长就重复启动。完整扫描会定期打印进度。
+
+命令重新验证固定压缩包 SHA256、DTD MD5 与大小，并只读打开目录。目录旁的 `dblp-ai-candidates.manifest.json` 必须属于同一个快照和 DTD，包含完整记录类型计数。旧候选文件不作为完整性依据，也不会被改写。
+
+恢复扫描使用项目原有 XML 流式解析器，仅允许已校验的本地 DTD，不联网读取 XML 实体。最终会比对完整快照的记录总数、逐 family 数量和全部选中卷的原始元数据。全部一致后才原子发布输出目录；失败时不发布半成品，也不覆盖既有文件。排查明确终止的失败任务后，可用尚不存在的输出目录重新运行。
+
+## 输出与后续流程
+
+- `dblp-family-candidates.jsonl`：原始 `inproceedings` 和 `proceedings`，保留书目 key、原始作者、标题、年份、卷引用、出版链接等字段。
+- `recovery-manifest.json`：快照、DTD、目录凭据哈希，完整扫描数量、逐 family 数量、选中卷、旧候选文件的已知数量及输出哈希。状态为 `recovered_raw_candidates_pending_scope_review`。
+
+恢复结果没有新增规范化的 `venue`、`track` 或决定标签，也不是 `import records` 的输入队列。`ifaamas` 的整卷可能混有研究论文、扩展摘要、keynote、博士生、演示和 JAAMAS 展示。目录过滤修复不会把这些条目统一标成 main；规范化的范围映射仍需第一方逐栏目证据。
+
+后续先用官方目录核对标题、页码、作者及发表形式，再查数据库稳定标识和候选实体，保留每次投稿、决定和标签来源。只有审核后的新增分区才能原生导入并独立验收；已验收的来源记录不能重放。单个年份有记录不代表逐篇、track、摘要或公开拒稿覆盖完整。
+
+## 2026-11-09 核查实例
+
+对既有固定快照的独立原始元数据恢复发现 `ifaamas` 有 1,122 条出版物候选及 2 卷：2025 年 479 条、2026 年 643 条。2025 年的 479 条标题与起始页均对应官方目录，其中 3 条 keynote 应排除，余下 476 条出版物需逐项处理。该数字包含已经验收的 225 条来源，不是 1,122 篇可新增论文。
+
+此实例说明过滤遗漏的影响，不代表恢复命令已完成生产导入。2026 年仍需核验官方出版清单、栏目、身份及逐次决定；缺少录用记录不能推断为拒稿。最终全库去重、覆盖和查询审计继续单独进行。

paper_atlas/cli.py

@@ -27,6 +27,7 @@ COMMANDS = {
     ('collect','aamas-proceedings'): ('paper_atlas.collectors.aamas', {}, '按官方目录区分 AAMAS 发表形式和摘要'),
     ('collect','www-archive'): ('paper_atlas.collectors.www_archive', {}, '核验早期WWW官方档案并分开待核查条目'),
     ('collect','dblp'): ('paper_atlas.collectors.dblp_records', {}, '核对并整理 DBLP 历史发表记录'),
+    ('collect','dblp-family'): ('paper_atlas.collectors.dblp_recovery', {}, '从固定快照恢复指定 DBLP family 的原始候选'),
     ('catalogue','dblp'): ('paper_atlas.collectors.dblp', {}, '盘点 DBLP 历史会议目录'),
     ('import','arxiv'): ('paper_atlas.ingestion.arxiv', {}, '导入 arXiv 元数据 ZIP'),
     ('import','records'): ('paper_atlas.ingestion.records', {}, '导入核验过的会议文件'),

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 == 2025:
+        from paper_atlas.collectors.aamas2025 import collect as collect_2025
+        return collect_2025(out, cache=cache, snapshots=snapshots)
     if year != 2012:
         raise ValueError('Only the independently reviewed AAMAS 2012 layout is supported')
     pages, receipts = read_pages(cache, snapshots)
@@ -190,8 +193,8 @@ def _collect(year, out, *, cache=None, snapshots=None):
 
 
 def collect(year, out, *, cache=None, snapshots=None):
-    if year != 2012:
-        raise ValueError('Only the independently reviewed AAMAS 2012 layout is supported')
+    if year not in (2012, 2025):
+        raise ValueError('Only independently reviewed AAMAS 2012/2025 layouts are supported')
     out = Path(out)
     out.mkdir(parents=True, exist_ok=True)
     with (out/'aamas.collect.lock').open('a') as lock:
@@ -201,7 +204,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], required=True)
+    parser.add_argument('--year', type=int, choices=[2012, 2025], required=True)
     parser.add_argument('--out', required=True)
     source = parser.add_mutually_exclusive_group(required=True)
     source.add_argument('--cache')

paper_atlas/collectors/aamas2025.py

@@ -0,0 +1,183 @@
+"""Reviewed AAMAS 2025 publisher TOC; HTML metadata only, never paper PDFs."""
+from collections import Counter, defaultdict
+import hashlib
+import json
+from pathlib import Path
+import re
+import tempfile
+
+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/aamas2025/'
+URLS = {'index': ROOT, 'contents': ROOT+'forms/contents.htm'}
+TITLE = 'AAMAS 2025: Proceedings of the 2025 International Conference on Autonomous Agents and Multiagent Systems'
+SECTIONS = {
+    '1': ('Keynote Talks', 'keynote', 'keynote', 3),
+    '2': ('Research Paper Track', 'full_paper', 'proceedings_unspecified', 260),
+    '3': ('Extended Abstracts', 'extended_abstract', 'proceedings_unspecified', 148),
+    '4': ('Blue Sky Ideas', 'blue_sky_paper', 'blue_sky', 12),
+    '5': ('Doctoral Consortium', 'doctoral_abstract', 'doctoral', 25),
+    '6': ('Demo Track', 'demo', 'demo', 22),
+    '7': ('JAAMAS Track', 'journal_presentation', 'jaamas', 9),
+}
+COMPETITION_TITLE = '[COMP24] The Automated Negotiating Agents Competition (ANAC) 2024 Challenges and Results'
+PAPER = re.compile(r'\.\./pdfs/p([0-9]+)\.pdf')
+
+
+def one(items, label):
+    if len(items) != 1:
+        raise ValueError('Expected one '+label)
+    return items[0]
+
+
+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')
+    for paragraph in doc.root.walk():
+        if paragraph.tag != 'p' or any(n is not paragraph and n.tag == 'p' for n in paragraph.walk()):
+            continue
+        anchors = [n.attrs['name'] for n in paragraph.walk() if n.tag == 'a' and n.attrs.get('name') in SECTIONS]
+        if anchors and (len(anchors) != 1 or clean(paragraph.text()) != SECTIONS[anchors[0]][0]):
+            raise ValueError('Publisher section heading differs from reviewed layout')
+    section = None
+    headers = []
+    records, excluded, counts, seen = [], [], Counter(), set()
+    for node in doc.root.walk():
+        if node.tag == 'a' and node.attrs.get('name') in SECTIONS:
+            section = node.attrs['name']
+            headers.append(section)
+        if node.tag != 'p' or any(n is not node and n.tag == 'p' for n in node.walk()):
+            continue
+        links = [n for n in node.walk() if n.tag == 'a' and PAPER.fullmatch(n.attrs.get('href', ''))]
+        if not links:
+            continue
+        link = one(links, 'paper link per paragraph')
+        if section not in SECTIONS:
+            raise ValueError('Paper outside a reviewed publisher section')
+        page = int(PAPER.fullmatch(link.attrs['href'])[1])
+        if page in seen:
+            raise ValueError('Repeated publisher paper/page ID')
+        seen.add(page)
+        title = clean(link.text())
+        if not title:
+            raise ValueError('Empty paper title')
+        # Names are outside italicized affiliations and separated by explicit BRs.
+        font = one([n for n in node.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')
+        heading, publication_format, track, _ = SECTIONS[section]
+        counts[section] += 1
+        if section == '1':
+            excluded.append({'publisher_page': page, 'title': title, 'authors': authors, 'reason': 'publisher keynote talk, not a research-paper submission'})
+            continue
+        note = 'Explicit publisher section; section topics are not research types.'
+        if track == 'proceedings_unspecified':
+            note = 'Published format is known; individual main/special submission track is not provided.'
+        if page == 3000:
+            if section != '6' or title != COMPETITION_TITLE:
+                raise ValueError('Competition report differs from reviewed publisher entry')
+            publication_format, track = 'competition_report', 'proceedings_unspecified'
+            note = 'Publisher places this competition report in Demo Track; no individual demonstration submission is inferred.'
+        records.append({
+            'source': 'aamas_proceedings', 'external_id': '2025/p'+str(page),
+            'url': URLS['contents']+'#'+section, 'title': title, 'abstract': None,
+            'authors': authors, 'identifiers': {}, 'venue': 'AAMAS', 'venue_aliases': [],
+            'year': 2025, 'track': track, 'status': 'accepted',
+            'observed_at': evidence['observed_at'],
+            'raw_decision': 'listed in official AAMAS 2025 published proceedings '+heading,
+            'tags': [{'namespace': 'aamas_proceedings:publication_format', 'value': publication_format}],
+            'metadata': {'publisher_page': page, 'paper_url': ROOT+'pdfs/p'+str(page)+'.pdf',
+                         'section_url': URLS['contents'], 'section_anchor': section,
+                         'section_heading': heading, 'publication_format': publication_format,
+                         'author_affiliations': affiliations, 'source_evidence': evidence,
+                         'track_scope_note': note, '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.'}})
+    if headers != list(SECTIONS) or dict(counts) != {k: v[3] for k, v in SECTIONS.items()}:
+        raise ValueError('Publisher section inventory differs from reviewed 2025 layout')
+    if len(records) != 476 or len(excluded) != 3 or sum(r['metadata']['publication_format'] == 'competition_report' for r in records) != 1:
+        raise ValueError('Unexpected publication partition')
+    return records, excluded
+
+
+def read_pages(cache, snapshots):
+    if bool(cache) == bool(snapshots):
+        raise ValueError('Choose exactly one cache or frozen snapshot directory')
+    frozen = json.loads((Path(snapshots)/'receipts.json').read_text())['receipts'] if snapshots else []
+    pages = {}
+    for key, url in URLS.items():
+        if snapshots:
+            r = one([r for r in frozen if r['url'] == url], 'frozen response receipt')
+            filename = Path(r['file']); base = Path(snapshots).resolve()
+            if filename.is_absolute() or '..' in filename.parts or not (base/filename).resolve().is_relative_to(base) or r.get('http_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['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)
+        pages[key] = 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}
+    return pages
+
+
+def collect(out, *, cache=None, snapshots=None):
+    pages = read_pages(cache, snapshots)
+    if [clean(n.text()) for n in Document(pages['index'][0].decode('utf-8')).root.walk() if n.tag == 'title'] != [TITLE]:
+        raise ValueError('Unexpected AAMAS 2025 introduction')
+    records, excluded = parse_contents(pages['contents'][0].decode('utf-8'), pages['contents'][1])
+    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='.aamas2025-', dir=out) as temporary:
+        stage = Path(temporary); (stage/'evidence').mkdir()
+        for data, e in pages.values(): (stage/e['file']).write_bytes(data)
+        for track, group in sorted(grouped.items()):
+            p = stage/f'aamas-proceedings-2025-{track}.jsonl'
+            p.write_text(''.join(json.dumps(r, ensure_ascii=False)+'\n' for r in group))
+            m = {'source': 'aamas_proceedings', 'parser_version': 2, 'venue': 'AAMAS', 'year': 2025,
+                 'track': track, 'count': len(group), 'file': str(out/p.name), 'file_sha256': hashlib.sha256(p.read_bytes()).hexdigest(),
+                 'list_url': URLS['contents'], 'status': 'partial', 'missing_abstract_count': len(group),
+                 'publication_format_counts': dict(Counter(r['metadata']['publication_format'] for r in group)),
+                 'note': 'Complete published TOC partition; HTML abstracts absent, identities and submission-track scope require independent review.'}
+            p.with_suffix('.manifest.json').write_text(json.dumps(m, ensure_ascii=False, indent=2)+'\n'); files.append(m)
+        result = {'source': 'aamas_proceedings', 'year': 2025, 'finished_at': timestamp(),
+                  'status': 'collected_pending_identity_review', 'records': len(records), 'files': files,
+                  'excluded_publications': excluded, 'source_receipts': [e for _, e in pages.values()],
+                  'publication_format_counts': dict(Counter(r['metadata']['publication_format'] for r in records)),
+                  'database_written': False, 'pdfs_downloaded': False,
+                  'note': 'Neither absence from proceedings nor aggregate rejection/withdrawal counts identify an individual negative decision.'}
+        report = 'aamas-proceedings-2025.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/dblp.py

@@ -20,7 +20,7 @@ DTD_MD5='96e4283e71886dec2173718957293648'
 DTD_NAMES={'dblp.dtd','dblp-2023-06-28.dtd'}
 RELEASE='https://doi.org/10.4230/dblp.xml.2026-11-01'
 # Broad discovery candidates, including historical names; not a venue mapping.
-FAMILIES=set('nips icml iclr colm colt aistats uai clear acl emnlp naacl eacl coling conll cvpr iccv eccv aaai ijcai ecai kdd sigir www wsdm recsys cikm icdm sdm ecml pkdd ecmlpkdd icra iros rss corl aips icaps ecp ewsp kr atal aamas icassp interspeech eurospeech icslp ismir miccai fat facct lrec mm mlsys'.split())
+FAMILIES=set('nips icml iclr colm colt aistats uai clear acl emnlp naacl eacl coling conll cvpr iccv eccv aaai ijcai ecai kdd sigir www wsdm recsys cikm icdm sdm ecml pkdd ecmlpkdd icra iros rss corl aips icaps ecp ewsp kr atal aamas ifaamas icassp interspeech eurospeech icslp ismir miccai fat facct lrec mm mlsys'.split())
 
 
 def records(archive,dtd_bytes):

paper_atlas/collectors/dblp_recovery.py

@@ -0,0 +1,156 @@
+"""Recover raw family candidates from a pinned snapshot and its existing catalogue.
+
+The old candidate file is not an exhaustive source: earlier discovery filters
+can omit an entire namespace. This command checks the complete snapshot and
+catalogue, without assigning venues, tracks, acceptance or paper identities.
+"""
+import argparse
+from collections import Counter
+from contextlib import closing
+import fcntl
+import json
+from pathlib import Path
+import re
+import sqlite3
+import tempfile
+
+from paper_atlas.collectors.dblp import DTD_MD5, FAMILIES, RELEASE, XML_SHA, family, records
+from paper_atlas.db.submissions import timestamp
+from paper_atlas.io.snapshots import digest_file
+
+
+def catalogue_evidence(catalogue, families, expected_sha, expected_dtd_md5):
+    manifest_path = catalogue.with_name('dblp-ai-candidates.manifest.json')
+    manifest = json.loads(manifest_path.read_text())
+    if (manifest.get('snapshot_sha256') != expected_sha
+            or manifest.get('dtd_md5') != expected_dtd_md5):
+        raise ValueError('Catalogue belongs to a different snapshot or DTD')
+    totals = manifest.get('record_types', {})
+    if (set(totals) != {'proceedings', 'inproceedings'}
+            or any(type(n) is not int or n < 0 for n in totals.values())
+            or type(manifest.get('records_scanned')) is not int
+            or sum(totals.values()) != manifest['records_scanned']):
+        raise ValueError('Incomplete catalogue record totals')
+    counts = Counter()
+    volumes = {}
+    # The catalogue is evidence, never a writable recovery destination.
+    with closing(sqlite3.connect(catalogue.resolve().as_uri() + '?mode=ro', uri=True)) as c:
+        for name in families:
+            found = c.execute('SELECT record_type,count FROM family_counts WHERE family=?',
+                              (name,)).fetchall()
+            if not found or not any(n for _, n in found):
+                raise ValueError('Family absent from complete catalogue: ' + name)
+            for kind, count in found:
+                if kind not in totals or type(count) is not int or count < 0:
+                    raise ValueError('Invalid catalogue family count')
+                counts[(name, kind)] = count
+            for key, raw in c.execute('SELECT key,metadata_json FROM proceedings WHERE family=?',
+                                      (name,)):
+                row = json.loads(raw)
+                if (family(key) != name or row.get('key') != key
+                        or row.get('record_type') != 'proceedings'):
+                    raise ValueError('Invalid catalogue proceedings evidence')
+                volumes[key] = row
+    if len(volumes) != sum(counts[(name, 'proceedings')] for name in families):
+        raise ValueError('Catalogue proceedings count differs from volume records')
+    return manifest, digest_file(manifest_path)['sha256'], counts, volumes
+
+
+def recover(archive, dtd, catalogue, out, families, *, expected_sha=XML_SHA,
+            expected_dtd_md5=DTD_MD5):
+    families = list(families)
+    if (not families or any(not isinstance(name, str)
+                           or not re.fullmatch(r'[a-z][a-z0-9-]*', name)
+                           or name not in FAMILIES for name in families)):
+        raise ValueError('Select an explicitly supported discovery family')
+    families = sorted(set(families))
+    archive, dtd, catalogue = Path(archive), Path(dtd), Path(catalogue)
+    output = Path(out).absolute()
+    if output.exists() or output.is_symlink():
+        raise ValueError('Recovery output must not already exist')
+    output.parent.mkdir(parents=True, exist_ok=True)
+    lock_path = output.parent / ('.' + output.name + '.recovery.lock')
+    with lock_path.open('a') as lock:
+        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        if output.exists() or output.is_symlink():
+            raise ValueError('Recovery output appeared before scan')
+        actual = digest_file(archive)
+        if actual['sha256'] != expected_sha:
+            raise ValueError('DBLP snapshot checksum mismatch')
+        if dtd.stat().st_size > 128 * 1024 or digest_file(dtd)['md5'] != expected_dtd_md5:
+            raise ValueError('DBLP DTD checksum mismatch or size limit')
+        manifest, manifest_sha, expected, volumes = catalogue_evidence(
+            catalogue, families, expected_sha, expected_dtd_md5)
+        totals, found = Counter(), Counter()
+        seen_keys, found_volumes = set(), set()
+        with tempfile.TemporaryDirectory(prefix='.dblp-family-', dir=output.parent) as tmp:
+            stage = Path(tmp) / 'recovered'
+            stage.mkdir()
+            target = stage / 'dblp-family-candidates.jsonl'
+            with target.open('w') as stream:
+                for row in records(archive, dtd.read_bytes()):
+                    totals[row['record_type']] += 1
+                    name = family(row['key'])
+                    if name in families:
+                        if row['key'] in seen_keys:
+                            raise ValueError('Duplicate raw bibliographic key')
+                        seen_keys.add(row['key'])
+                        found[(name, row['record_type'])] += 1
+                        if row['record_type'] == 'proceedings':
+                            if volumes.get(row['key']) != row:
+                                raise ValueError('Raw proceedings differs from catalogue evidence')
+                            found_volumes.add(row['key'])
+                        stream.write(json.dumps(row, ensure_ascii=False) + '\n')
+                    scanned = sum(totals.values())
+                    if scanned % 250000 == 0:
+                        print(json.dumps({'status': 'scanning_raw_family_candidates',
+                                          'records_scanned': scanned,
+                                          'recovered_records': sum(found.values())}), flush=True)
+            if (totals != Counter(manifest['record_types'])
+                    or sum(totals.values()) != manifest['records_scanned']):
+                raise ValueError('Complete snapshot totals differ from catalogue')
+            if found != expected or found_volumes != volumes.keys():
+                raise ValueError('Recovered family totals differ from catalogue')
+            report = {
+                'format_version': 1, 'source': 'dblp',
+                'status': 'recovered_raw_candidates_pending_scope_review',
+                'finished_at': timestamp(), 'source_release': RELEASE,
+                'snapshot_sha256': actual['sha256'], 'dtd_md5': expected_dtd_md5,
+                'catalogue_manifest_sha256': manifest_sha, 'families': families,
+                'records_scanned': sum(totals.values()), 'record_types': dict(totals),
+                'recovered_records': sum(found.values()),
+                'recovered_publications': sum(found[(n, 'inproceedings')] for n in families),
+                'recovered_volumes': len(found_volumes),
+                'family_counts': [{'family': name, 'record_type': kind, 'count': count}
+                                  for (name, kind), count in sorted(found.items())],
+                'old_candidate_family_counts': {
+                    name: manifest.get('candidate_family_counts', {}).get(name, 0)
+                    for name in families},
+                'catalogue_volumes': [volumes[k] for k in sorted(volumes)],
+                'file': str(output / target.name),
+                'file_sha256': digest_file(target)['sha256'], 'database_written': False,
+                'note': 'Raw bibliographic candidates, including volume records and possible '
+                        'keynotes/workshops. No venue, track, acceptance or work identity '
+                        'inferred. Requires primary-source scope and identity review; '
+                        'not a normalized import queue. Existing imported sources must '
+                        'not be replayed.',
+            }
+            # Deliberately not an import-records sidecar manifest.
+            (stage / 'recovery-manifest.json').write_text(
+                json.dumps(report, ensure_ascii=False, indent=2) + '\n')
+            if output.exists() or output.is_symlink():
+                raise ValueError('Recovery output appeared during scan')
+            stage.rename(output)
+    return report
+
+
+def main(argv=None, *, prog=None):
+    parser = argparse.ArgumentParser(prog=prog, description=__doc__)
+    parser.add_argument('--archive', required=True, type=Path)
+    parser.add_argument('--dtd', required=True, type=Path)
+    parser.add_argument('--catalogue', required=True, type=Path)
+    parser.add_argument('--family', required=True, action='append', choices=sorted(FAMILIES))
+    parser.add_argument('--out', required=True, type=Path)
+    args = parser.parse_args(argv)
+    print(json.dumps(recover(args.archive, args.dtd, args.catalogue, args.out, args.family),
+                     ensure_ascii=False))

tests/fixtures/aamas2025/index.html

@@ -0,0 +1,54 @@
+<HTML><HEAD><TITLE>AAMAS 2025: Proceedings of the 2025 International Conference on Autonomous Agents and Multiagent Systems</TITLE>
+<META http-equiv=Content-Type content="text/html; charset=utf-8"><LINK 
+href="forms/index_files/factemplate.css" type=text/css rel=stylesheet>
+<META content="MSHTML 6.00.2900.3562" name=GENERATOR></HEAD>
+<BODY bgcolor="#010101" bgproperties="fixed"><DIV style="TEXT-ALIGN: center" >
+<table width="724" border="0" align="center" cellpadding="0" cellspacing="0" valign="top">
+  <td colspan="2" align="center" valign="top" bgcolor="#FFFFFF"><img src="forms/index_files/aamas25.png" width="750" height="300" alt=""/></td>
+  </tr>
+  <tr bgcolor="#FFFFFF"> 
+    <td width="125" align="center" valign="top" bgcolor="#FFFFFF"><table width="125" border="1" align="center" cellpadding="1" cellspacing="1" bordercolor="#444444" style="margin-bottom: 2">
+        <tr> 
+          <td bgcolor="#FFFFFF"><p style="text-indent: 0; margin-left: 6; margin-right: 6; margin-top: 5; margin-bottom: 4" align="left"><A href="forms/index.htm">Main 
+              Page</A></p></td>
+        </tr>
+        <tr> 
+          <td bgcolor="#FFFFFF"><p style="text-indent: 0; margin-left: 6; margin-right: 6; margin-top: 5; margin-bottom: 4" align="left"><A href="forms/contents.htm">Table 
+              of Contents</A></p></td>
+        </tr>
+      </table></td>
+    <td width="599"><p style="text-indent: 0; margin-left: 12; margin-right: 14; margin-top: 12; margin-bottom: 12" align="left"><font color="#010101" size="4" face="Arial, Helvetica, sans-serif"><strong>Welcome 
+        to the Proceedings of AAMAS-2025</strong></font></p>
+      <p style="text-indent: 0; margin-left: 12; margin-right: 14; margin-top: 12; margin-bottom: 12;" align="left"><font color="#333333" size="2" face="Times New Roman, Times, serif">These 
+        are the proceedings of the 24th International Conference on Autonomous 
+        Agents and Multiagent Systems (AAMAS-2025). They are published by the 
+        International Foundation for Autonomous Agents and Multiagent Systems 
+    (IFAAMAS).</font></p></td>
+  </tr>
+  <tr bgcolor="#FFFFFF"> 
+    <td align="center" valign="middle" bgcolor="#FFFFFF">&nbsp;</td>
+    <td><table width="96%" border="0" cellspacing="2" cellpadding="2">
+        <tr align="left" valign="top"> 
+          <td> <p style="text-indent: 0; margin-left: 12; margin-right: 6; margin-top: 0; margin-bottom: 6;" align="left"><font color="#333333" face="Times New Roman, Times, serif"><font size="2"><strong>ISBN: 979-8-4007-1426-9</strong></font></font></p></td>
+          <td> <p style="text-indent: 0; margin-left: 12; margin-right: 6; margin-top: 0; margin-bottom: 6;" align="left"><font color="#333333" face="Times New Roman, Times, serif"></font></p></td>
+        </tr>
+        <tr> 
+          <td width="42%"><div align="left"><img src="forms/index_files/ifaamas.jpg" width="256" height="104"></div></td>
+          <td width="58%"><div align="center"><img src="forms/index_files/ACM-In-Cooperation_medium.jpg" width="285" height="86"></div></td>
+        </tr>
+      </table>
+      <p style="text-indent: 0; margin-left: 12; margin-right: 14; margin-top: 12; margin-bottom: 12;" align="left"><font color="#333333" size="2" face="Times New Roman, Times, serif">Copyright 
+        &copy; 2025 by International Foundation for Autonomous Agents and Multiagent 
+        Systems (IFAAMAS). Permission to make digital or hard copies of portions 
+        of this work for personal or classroom use is granted without fee provided 
+        that copies are not made or distributed for profit or commercial advantage 
+        and that copies bear this notice and the full citation on the first page. 
+        Copyright for components of this work owned by others than IFAAMAS must 
+        be honored. Abstracting with credit is permitted. To copy otherwise, to 
+        republish, to post on servers or to redistribute to lists, requires prior 
+    specific permission and/or a fee. </font></p></td>
+  </tr>
+</table>
+  <p>&nbsp;</p>
+
+</BODY></HTML>

tests/fixtures/aamas2025/receipts.json

@@ -0,0 +1,38 @@
+{
+  "status": "two_successful_official_metadata_pages_recovered_third403_not_retried",
+  "receipts": [
+    {
+      "file": "index.html",
+      "url": "https://www.ifaamas.org/Proceedings/aamas2025/",
+      "bytes": 4107,
+      "sha256": "850b7825c66d0147ac8a8bbf03a554590e4a9e8f4b5c576e043999bbd43317f4",
+      "at": "2026-11-09T12:31:49.347577+00:00",
+      "at_precision": "filesystem_write_time_recovered",
+      "http_status": 200,
+      "content_type": "text/html",
+      "receipt_recovered": true,
+      "scope": "Collectorasserted200/htmlbeforewrite;request3acceptedprogram403;noheaders/finalredirectURLpersisted;no refetch or rawbytesmutation."
+    },
+    {
+      "file": "contents.html",
+      "url": "https://www.ifaamas.org/Proceedings/aamas2025/forms/contents.htm",
+      "bytes": 370650,
+      "sha256": "732847b51972f34c721a49d8bd963a80a029a48f3ec08022c9a0d5ddc2945549",
+      "at": "2026-11-09T12:31:52.006642+00:00",
+      "at_precision": "filesystem_write_time_recovered",
+      "http_status": 200,
+      "content_type": "text/html",
+      "receipt_recovered": true,
+      "scope": "Collectorasserted200/htmlbeforewrite;request3acceptedprogram403;noheaders/finalredirectURLpersisted;no refetch or rawbytesmutation."
+    }
+  ],
+  "failed_url": "https://aamas2025.org/index.php/conference/program/accepted-papers/",
+  "not_attempted_url": "https://aamas2025.org/index.php/conference/calls/submission-instructions-main-technical-track/",
+  "original_error": {
+    "error": "HTTP Error 403: Forbidden",
+    "type": "HTTPError",
+    "downloaded": 2
+  },
+  "research_fulltext_downloaded": false,
+  "database_written": false
+}

tests/test_aamas2025.py

@@ -0,0 +1,85 @@
+import hashlib
+import json
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from paper_atlas.collectors import aamas, aamas2025
+
+
+FIXTURE = Path(__file__).parent/'fixtures/aamas2025'
+
+
+class Aamas2025Tests(unittest.TestCase):
+    def parse(self, html=None):
+        return aamas2025.parse_contents(html or (FIXTURE/'contents.html').read_text(),
+            {'observed_at': '2026-11-09T12:31:50+00:00', 'url': aamas2025.URLS['contents']})
+
+    def test_real_toc_preserves_every_title_author_and_affiliation(self):
+        expected = json.loads((FIXTURE/'expected-authors.json').read_text())
+        self.assertEqual(hashlib.sha256((FIXTURE/'contents.html').read_bytes()).hexdigest(), expected['source_sha256'])
+        rows, excluded = self.parse()
+        self.assertEqual((len(rows), len(excluded)), (476, 3))
+        by_url = {r['paper_url']: r for r in expected['entries']}
+        for row in rows:
+            original = by_url[row['metadata']['paper_url']]
+            self.assertEqual(row['title'], original['title'])
+            self.assertEqual(row['authors'], [a['name'] for a in original['authors']])
+            self.assertEqual(row['metadata']['author_affiliations'], [a['affiliation'] for a in original['authors']])
+            self.assertIsNone(row['abstract'])
+            self.assertEqual(row['identifiers'], {})
+            self.assertEqual(row['status'], 'accepted')
+            self.assertNotIn('decision_date', row)
+            self.assertTrue(all(t['namespace'] == 'aamas_proceedings:publication_format' for t in row['tags']))
+
+    def test_publication_format_and_track_are_separate(self):
+        rows, excluded = self.parse()
+        groups = {k: sum(r['track'] == k for r in rows) for k in {r['track'] for r in rows}}
+        self.assertEqual(groups, {'proceedings_unspecified': 409, 'blue_sky': 12, 'doctoral': 25, 'demo': 21, 'jaamas': 9})
+        competition = next(r for r in rows if r['external_id'] == '2025/p3000')
+        self.assertEqual(competition['year'], 2025)
+        self.assertEqual(competition['metadata']['publication_format'], 'competition_report')
+        self.assertEqual(competition['metadata']['section_heading'], 'Demo Track')
+        self.assertEqual(competition['track'], 'proceedings_unspecified')
+        self.assertEqual(len({r['external_id'] for r in rows}), 476)
+        self.assertEqual([r['publisher_page'] for r in excluded], [1, 3, 4])
+
+    def test_refuses_changed_edition_inventory_or_author_markup(self):
+        original = (FIXTURE/'contents.html').read_text()
+        for html in [original.replace('AAMAS 2025:', 'AAMAS 2026:', 1),
+                     original.replace('../pdfs/p5.pdf', '../pdfs/p14.pdf', 1),
+                     original.replace('Paper Track</font>', 'Workshop Track</font>'),
+                     original.replace('Amir Abolfazli', '', 1),
+                     original.replace('[COMP24]', '[CHANGED]', 1)]:
+            with self.subTest(change=html[:60]):
+                with self.assertRaises(ValueError): self.parse(html)
+
+    def test_frozen_collection_is_partial_and_never_fetches_pdfs(self):
+        with tempfile.TemporaryDirectory() as tmp, patch.object(aamas2025, 'fetch', side_effect=AssertionError('No HTTP in frozen mode')):
+            out = Path(tmp)/'out'
+            report = aamas.collect(2025, out, snapshots=FIXTURE)
+            self.assertEqual(report['records'], 476)
+            self.assertFalse(report['database_written'])
+            self.assertFalse(report['pdfs_downloaded'])
+            self.assertEqual(len(report['files']), 5)
+            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'])
+            receipt = report['source_receipts'][0]['response_receipt']
+            self.assertEqual(receipt['at_precision'], 'filesystem_write_time_recovered')
+            self.assertTrue((out/'aamas-proceedings-2025.json').is_file())
+
+    def test_rejects_tampered_or_unsafe_frozen_response(self):
+        import shutil
+        for mode in ['tamper', 'path', 'status']:
+            with self.subTest(mode=mode), tempfile.TemporaryDirectory() as tmp:
+                root = Path(tmp); source = root/'source'; shutil.copytree(FIXTURE, source)
+                p = source/'receipts.json'; r = json.loads(p.read_text())
+                if mode == 'tamper': (source/'contents.html').write_text('bad')
+                elif mode == 'path': r['receipts'][0]['file'] = '../index.html'
+                else: r['receipts'][0]['http_status'] = 403
+                p.write_text(json.dumps(r))
+                with self.assertRaises(ValueError): aamas.collect(2025, root/'out', snapshots=source)
+                self.assertFalse((root/'out/aamas-proceedings-2025.json').exists())

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(2025, root/'future', snapshots=source)
+                aamas.collect(2026, root/'future', snapshots=source)
             self.assertFalse((root/'future').exists())
 
     def test_ids_colliding_across_sections_do_not_produce_a_completed_collection(self):

tests/test_dblp_recovery.py

@@ -0,0 +1,141 @@
+import gzip
+import hashlib
+import json
+from pathlib import Path
+import sqlite3
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from paper_atlas.collectors.dblp import catalogue
+from paper_atlas.collectors.dblp_recovery import recover
+from paper_atlas.collectors.dblp_scope import scope
+from paper_atlas.ingestion.records import ingest_file
+
+
+class DblpRecoveryTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.root = Path(self.tmp.name)
+        self.dtd = self.root / 'dblp.dtd'
+        self.dtd.write_bytes(b'<!ENTITY eacute "&#233;">')
+        self.archive = self.root / 'dblp.xml.gz'
+        # These include a keynote and a workshop, neither a main-paper inference.
+        body = '''<proceedings key="conf/ifaamas/2025"><title>AAMAS 2025</title><year>2025</year></proceedings>
+<inproceedings key="conf/ifaamas/keynote"><author>Ren&eacute;</author><title>Keynote: Agents</title><year>2025</year><booktitle>AAMAS</booktitle><crossref>conf/ifaamas/2025</crossref></inproceedings>
+<inproceedings key="conf/ifaamas/paper"><author>Alice Doe</author><title>Research <i>agents</i></title><year>2025</year><booktitle>AAMAS</booktitle><crossref>conf/ifaamas/2025</crossref><ee>https://dl.acm.org/doi/10.5555/3709347.123</ee></inproceedings>
+<proceedings key="conf/ifaamas/2025w"><title>AAMAS Workshop</title><year>2025</year></proceedings>
+<inproceedings key="conf/ifaamas/workshop"><title>Workshop paper</title><year>2025</year><booktitle>AAMAS Workshop</booktitle><crossref>conf/ifaamas/2025w</crossref></inproceedings>
+<inproceedings key="conf/unrelated/other"><title>Other family</title><year>2025</year></inproceedings>'''
+        with gzip.open(self.archive, 'wb') as stream:
+            stream.write(('<?xml version="1.0"?><!DOCTYPE dblp SYSTEM "dblp.dtd"><dblp>'
+                          + body + '</dblp>').encode())
+        self.sha = hashlib.sha256(self.archive.read_bytes()).hexdigest()
+        self.md5 = hashlib.md5(self.dtd.read_bytes()).hexdigest()
+        self.manifest = catalogue(self.archive, self.dtd, self.root / 'catalogue',
+                                  self.sha, self.md5)
+        self.catalogue = Path(self.manifest['catalogue'])
+        self.manifest_path = self.catalogue.with_name('dblp-ai-candidates.manifest.json')
+
+    def run_recovery(self, **kwargs):
+        options = dict(expected_sha=self.sha, expected_dtd_md5=self.md5)
+        options.update(kwargs)
+        return recover(self.archive, self.dtd, self.catalogue, self.root / 'recovered',
+                       ['ifaamas'], **options)
+
+    def test_fresh_discovery_and_recovery_from_old_candidate_omission(self):
+        self.assertEqual(self.manifest['candidate_family_counts']['ifaamas'], 3)
+        # Simulate an older filter without altering its complete catalogue.
+        manifest = dict(self.manifest)
+        manifest['candidate_family_counts'] = {}
+        self.manifest_path.write_text(json.dumps(manifest))
+        candidate = Path(manifest['file'])
+        candidate.write_text('old incomplete candidate file\n')
+        before = {p: p.read_bytes() for p in [self.archive, self.dtd, self.catalogue,
+                                             self.manifest_path, candidate]}
+        with patch('urllib.request.urlopen', side_effect=AssertionError('No network')):
+            report = self.run_recovery()
+        rows = [json.loads(line) for line in Path(report['file']).read_text().splitlines()]
+        self.assertEqual(report['recovered_publications'], 3)
+        self.assertEqual(report['recovered_volumes'], 2)
+        self.assertEqual(report['recovered_records'], 5)
+        self.assertEqual(report['records_scanned'], 6)
+        self.assertEqual(report['old_candidate_family_counts'], {'ifaamas': 0})
+        self.assertFalse(report['database_written'])
+        self.assertEqual(next(r for r in rows if r['key'].endswith('/keynote'))['fields']['author'], ['René'])
+        self.assertTrue(any(r['key'].endswith('/workshop') for r in rows))
+        self.assertTrue(all(not ({'source', 'status', 'venue', 'track'} & r.keys()) for r in rows))
+        publication = next(r for r in rows if r['key'].endswith('/paper'))
+        volume = next(r for r in rows if r['key'] == 'conf/ifaamas/2025')
+        self.assertEqual(scope(publication, volume), (None, 'outside_target_family'))
+        self.assertEqual({p: p.read_bytes() for p in before}, before)
+        self.assertEqual(hashlib.sha256(Path(report['file']).read_bytes()).hexdigest(),
+                         report['file_sha256'])
+        # Raw recovery files cannot accidentally enter the normalized writer.
+        with patch('paper_atlas.ingestion.records.open_store') as writer:
+            with self.assertRaisesRegex(ValueError, 'source manifest'):
+                ingest_file(self.root / 'never.sqlite', report['file'])
+            writer.assert_not_called()
+        self.assertFalse((self.root / 'never.sqlite').exists())
+
+    def test_checksum_mismatch_does_not_publish(self):
+        for options in [{'expected_sha': '0' * 64}, {'expected_dtd_md5': '0' * 32}]:
+            with self.subTest(options=options), self.assertRaisesRegex(ValueError, 'checksum'):
+                self.run_recovery(**options)
+            self.assertFalse((self.root / 'recovered').exists())
+
+    def test_wrong_catalogue_snapshot_and_incomplete_totals_are_rejected(self):
+        for changes in [{'snapshot_sha256': '0' * 64}, {'dtd_md5': '0' * 32},
+                        {'records_scanned': 99}, {'record_types': {'proceedings': 2}}]:
+            manifest = {**self.manifest, **changes}
+            self.manifest_path.write_text(json.dumps(manifest))
+            with self.subTest(changes=changes), self.assertRaises(ValueError):
+                self.run_recovery()
+            self.assertFalse((self.root / 'recovered').exists())
+
+    def test_changed_family_count_does_not_publish_partial_result(self):
+        with sqlite3.connect(self.catalogue) as c:
+            c.execute("UPDATE family_counts SET count=count+1 WHERE family='ifaamas' AND record_type='inproceedings'")
+        with self.assertRaisesRegex(ValueError, 'family totals'):
+            self.run_recovery()
+        self.assertFalse((self.root / 'recovered').exists())
+
+    def test_changed_volume_metadata_is_detected(self):
+        with sqlite3.connect(self.catalogue) as c:
+            row = json.loads(c.execute("SELECT metadata_json FROM proceedings WHERE key='conf/ifaamas/2025'").fetchone()[0])
+            row['fields']['title'] = ['Wrong volume']
+            c.execute("UPDATE proceedings SET metadata_json=? WHERE key='conf/ifaamas/2025'", (json.dumps(row),))
+        with self.assertRaisesRegex(ValueError, 'proceedings differs'):
+            self.run_recovery()
+        self.assertFalse((self.root / 'recovered').exists())
+
+    def test_truncated_gzip_never_publishes(self):
+        self.archive.write_bytes(self.archive.read_bytes()[:-8])
+        revised_sha = hashlib.sha256(self.archive.read_bytes()).hexdigest()
+        self.manifest_path.write_text(json.dumps({**self.manifest, 'snapshot_sha256': revised_sha}))
+        with self.assertRaises((EOFError, OSError)):
+            self.run_recovery(expected_sha=revised_sha)
+        self.assertFalse((self.root / 'recovered').exists())
+
+    def test_existing_output_is_preserved_and_unknown_family_is_rejected(self):
+        out = self.root / 'recovered'
+        out.mkdir()
+        sentinel = out / 'keep'
+        sentinel.write_text('existing')
+        with self.assertRaisesRegex(ValueError, 'already exist'):
+            self.run_recovery()
+        self.assertEqual(sentinel.read_text(), 'existing')
+        for names in [[], ['../ifaamas'], ['not-a-supported-family'], [None]]:
+            with self.subTest(names=names), self.assertRaises(ValueError):
+                recover(self.archive, self.dtd, self.catalogue, self.root / 'other', names)
+
+    def test_requested_family_missing_from_catalogue_is_explicit(self):
+        with self.assertRaisesRegex(ValueError, 'absent from complete catalogue'):
+            recover(self.archive, self.dtd, self.catalogue, self.root / 'recovered',
+                    ['ifaamas', 'icml'], expected_sha=self.sha, expected_dtd_md5=self.md5)
+        self.assertFalse((self.root / 'recovered').exists())
+
+
+if __name__ == '__main__':
+    unittest.main()