Prepare evidence-bound identifier additions without rewriting source records

.gitea/workflows/check.yml

@@ -54,6 +54,7 @@ jobs:
               subprocess.run([python,'-I','-B','-m','paper_atlas','maintain','backfill','--help'],cwd=outside,check=True)
               subprocess.run([python,'-I','-B','-m','paper_atlas','audit','public-openreview','--help'],cwd=outside,check=True)
               subprocess.run([python,'-I','-B','-m','paper_atlas','audit','histories','--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()"
               subprocess.run([python,'-I','-B','-c',smoke],cwd=outside,check=True)

docs/operations.md

@@ -120,36 +120,3 @@ paper-atlas maintain pmlr-tracks --db /data/paper-index/papers.sqlite \
 检查报告后，在同一命令添加 `--apply` 并使用新的 `--out` 路径执行迁移。计划使用共享维护锁，实际迁移使用独占锁；已有 arXiv 类别修复或其他写入运行时会拒绝启动，不会停止或重启它们。一次事务内核对整个来源集合、原始题目/摘要/作者、forum 身份、已有事件及目标冲突，更新原投稿的 edition 和身份归属，回填官方轨道出处及逐轨道覆盖数，最后复查轨道与外键后提交。论文、来源、投稿、决定的 ID 及决定观察均保留，不重新创建事件、不做实体合并、不改论文正文元数据或标签。
 
 报告记录每个迁移事件的旧 edition、原身份、完整决定观察及新轨道证据。`planned` 是未执行的计划；`complete` 才表示事务已提交；`error` 表示执行失败。若进程在提交前后中断而报告停在 `prepared`，先对照主库和报告核查，不根据日志缺失自行重启任务。新路径重跑会重新核对当前数据库，已正确的轨道不会再次迁移。共享多个来源、重复目标事件或冲突决定会停止迁移，需另行核对，不能直接删掉其中一条。
-
-## 追加同一目录的部分记录
-
-默认 `import records` 把 manifest 当成一份来源快照，覆盖统计使用该文件的数量。
-若目录已有2条，之后仅补入1条，不能把这一小批当成完整快照，否则会把目录计数改成1。
-为已核验的部分补充批次，在 manifest 显式声明：
-
-```json
-{
-  "coverage_mode": "incremental",
-  "coverage_previous_count": 2,
-  "coverage_previous_expected_count": 2,
-  "coverage_expected_count": 4,
-  "status": "partial",
-  "count": 1
-}
-```
-
-这只是新增字段示例，manifest仍须保留source、venue、year、track、list_url、文件哈希等来源证据。
-`count`是本文件行数，`coverage_previous_count`是导入前该来源/届次的实际记录数，
-`coverage_previous_expected_count`是旧覆盖行的expected_count（未知时用null），
-`coverage_expected_count`是独立目录证据支持的整份来源预期数，不能小于已知旧预期数。
-这个例子成功后覆盖统计为预期4、已导入3；文件checkpoint的records_written仍为1。
-刷新相同记录不重复累加。初次导入部分来源可声明previous_count为0、previous_expected_count为null。
-
-只支持当前最多一个规范目录URL且旧覆盖状态为partial的来源/届次；多目录、旧状态complete、
-旧计数与实际来源记录不符，或前置计数已变化，都会中止，先核对来源范围或重新取得完整快照。
-该模式在写入前检查父状态，并在同一文件事务中按实际来源记录重新计数、检查总量界限和乐观更新覆盖行。
-失败会回滚本文件的论文、来源、决定、标签和覆盖变化，保留失败运行记录；更早已提交文件不重放。
-增量模式始终保留partial，不能靠一个小批次推断完整论文集、摘要或公开拒稿覆盖。
-
-仍需先独立核对新增记录身份与决定，按原生单写入保护串行导入，再独立核验数据与原有历史。
-这些计数检查不证明外部目录完整，也不代替最终全库去重。

docs/usage.md

@@ -2,6 +2,55 @@
 
 先在已确认的远程环境安装项目，使用 `paper-atlas`。从源码运行时，将命令前缀替换为 `python3 -B -m paper_atlas`，其余参数相同。数据目录位于仓库外。
 
+## 按证据补充稳定标识
+
+`paper-atlas curate identifiers --plan /data/review/plan.json --output /data/review/prepared`
+只生成新的 JSONL、manifest、证据副本及 `curation-report.json`，不会连接数据库。
+输出目录必须不存在，父目录必须已存在；任何输入或断言失败，都不发布本批输出。
+原始文件不变，未选中的记录逐字节保留。题名、作者、年份、页码、投稿决定、标签和既有 metadata
+均不重写，仅补充 `identifiers` 并追加 `metadata.identifier_curation` 证据链。
+
+计划格式如下（哈希占位符须替换为真实 SHA256，所有路径按命令当前目录解析）：
+
+```json
+{
+  "format_version": 1,
+  "inputs": [{
+    "file": "/data/review/uai2001.jsonl",
+    "sha256": "<JSONL 文件 SHA256>",
+    "manifest_sha256": "<同名 .manifest.json 文件 SHA256>"
+  }],
+  "assertions": [{
+    "source": "dblp",
+    "external_id": "conf/uai/example",
+    "record_sha256": "<原始物理行字节 SHA256，包含该行原有换行符>",
+    "identifiers": {"arxiv": "1408.1485"},
+    "evidence": [{
+      "file": "/data/review/primary-metadata.json",
+      "sha256": "<证据 JSON 文件 SHA256>",
+      "url": "https://example.org/primary-record",
+      "checked_at": "2026-11-08T08:00:00Z",
+      "note": "说明人工核对的题名、作者、出版版本和稳定标识证据"
+    }]
+  }]
+}
+```
+
+输入 manifest 必须包含 `source/venue/year/track/count/list_url/file_sha256`，且与逐条记录一致。
+断言必须唯一、准确命中一条记录。仅支持无版本后缀的 arXiv 编号与小写规范 DOI；
+已有同类标识不允许覆盖，相同值可以补记证据。证据只接受不超过 16 MiB 的 JSON 元数据文件，
+来源 URL 要求 HTTPS 且不含凭据，核查时间必须带时区；不下载论文 PDF。
+
+`verification_scope=input_and_evidence_integrity_only` 和 `semantic_identity_verified=false`
+明确表示哈希校验不能证明两条记录属于同一作品。请先独立核对版本关系；仅年份、页码或相同题名
+不能作为合并依据。存在来源冲突的行应继续暂缓。新 manifest 保留原来的 partial 状态及摘要缺失数，
+不会把补充编号伪装成来源覆盖或摘要补全。证据副本按 SHA256 保存在输出目录 `evidence/`，
+原计划也保存在 `plan.json`，供之后复查。
+
+文件生成后仍需另做当前 PostgreSQL 身份、投稿和决定冲突预检；导入运行时必须包含
+“稳定标识优先、多个标题候选则整文件回滚”的身份保护，再以 `import records` 独立导入并验收。
+该命令不会自动导入或合并实体，也不代表最终全库去重通过。
+
 ## PostgreSQL 查询
 
 在远程环境安装 `.[postgres]`，通过权限为 `0600` 的 libpq service 文件连接，不把密码放入 URL、命令行或仓库。

paper_atlas/cli.py

@@ -6,6 +6,7 @@ from paper_atlas import __version__
 
 
 COMMANDS = {
+    ('curate','identifiers'): ('paper_atlas.curation.identifiers', {}, '按证据补充稳定标识并生成独立待导入文件'),
     ('query',): ('paper_atlas.querying', {}, '查询论文、标签和投稿历史'),
     ('coverage',): ('paper_atlas.audits.coverage', {}, '查看全部候选会议的覆盖情况'),
     ('download',): ('paper_atlas.io.snapshots', {}, '下载并校验固定版本快照'),
@@ -48,7 +49,7 @@ def main(argv=None):
             leaf=commands.add_parser(route[0],help=help_text,add_help=False)
         else:
             if route[0] not in groups:
-                group=commands.add_parser(route[0],help={'collect':'采集来源元数据','catalogue':'盘点历史目录','import':'导入数据库','maintain':'维护数据库','audit':'核验数据'}[route[0]])
+                group=commands.add_parser(route[0],help={'collect':'采集来源元数据','catalogue':'盘点历史目录','import':'导入数据库','maintain':'维护数据库','audit':'核验数据','curate':'按证据整理元数据'}[route[0]])
                 groups[route[0]]=group.add_subparsers(dest='operation',required=True)
             leaf=groups[route[0]].add_parser(route[1],help=help_text,add_help=False)
         leaf.set_defaults(handler=(route,module,kwargs))

paper_atlas/curation/__init__.py

@@ -0,0 +1 @@
+"""Offline, evidence-bound preparation of metadata for separate ingestion."""

paper_atlas/curation/identifiers.py

@@ -0,0 +1,212 @@
+"""Add reviewed stable identifiers without rewriting bibliographic source fields.
+
+Hashes establish file integrity, not the truth of an identity assertion. Identity
+review and a fresh database preflight remain separate requirements.
+"""
+import argparse
+from datetime import datetime
+import hashlib
+import json
+from pathlib import Path
+import re
+import tempfile
+from urllib.parse import urlsplit
+
+
+SCOPE = 'input_and_evidence_integrity_only'
+MAX_METADATA_BYTES = 16 * 1024 * 1024
+
+
+def digest(data):
+    return hashlib.sha256(data).hexdigest()
+
+
+def object_keys(value, required, optional=()):
+    if not isinstance(value, dict) or not set(required) <= value.keys() or value.keys() - set(required) - set(optional):
+        raise ValueError('Missing or unsupported plan fields')
+
+
+def json_bytes(value):
+    return (json.dumps(value, ensure_ascii=False, separators=(',', ':'), allow_nan=False) + '\n').encode('utf-8')
+
+
+def decode(raw):
+    def unique(pairs):
+        result = {}
+        for key, value in pairs:
+            if key in result:
+                raise ValueError('Duplicate JSON key: ' + key)
+            result[key] = value
+        return result
+    return json.loads(raw, object_pairs_hook=unique,
+                      parse_constant=lambda value: (_ for _ in ()).throw(ValueError('Non-finite JSON value')))
+
+
+def metadata(path, expected=None):
+    with Path(path).open('rb') as stream:
+        raw = stream.read(MAX_METADATA_BYTES + 1)
+    if len(raw) > MAX_METADATA_BYTES:
+        raise ValueError('Metadata exceeds size limit')
+    if expected is not None:
+        check_hash(raw, expected)
+    return raw, decode(raw)
+
+
+def check_hash(raw, expected):
+    if not isinstance(expected, str) or not re.fullmatch('[0-9a-f]{64}', expected) or digest(raw) != expected:
+        raise ValueError('SHA256 mismatch')
+
+
+def identifier(scheme, value):
+    if not isinstance(value, str) or any(char.isspace() or ord(char) < 32 or ord(char) == 127 for char in value):
+        raise ValueError('Invalid identifier')
+    if scheme == 'arxiv':
+        pattern = r'(?:\d{2}(?:0[1-9]|1[0-2])\.\d{4,5}|[a-zA-Z][a-zA-Z.-]*/\d{2}(?:0[1-9]|1[0-2])\d{3})'
+        valid = re.fullmatch(pattern, value)
+    elif scheme == 'doi':
+        valid = re.fullmatch(r'10\.\d{4,9}/\S+', value) and value == value.lower()
+    else:
+        valid = False
+    if not valid:
+        raise ValueError('Unsupported or noncanonical identifier: ' + scheme)
+
+
+def prepare(plan_path, output):
+    plan_raw, plan = metadata(plan_path)
+    object_keys(plan, ('format_version', 'inputs', 'assertions'))
+    if type(plan['format_version']) is not int or plan['format_version'] != 1:
+        raise ValueError('Unsupported format version')
+    if not isinstance(plan['inputs'], list) or not plan['inputs'] or not isinstance(plan['assertions'], list) or not plan['assertions']:
+        raise ValueError('Inputs and assertions must be nonempty lists')
+    output = Path(output).absolute()
+    if output.exists():
+        raise ValueError('Output directory must not already exist')
+    if not output.parent.is_dir():
+        raise ValueError('Output parent directory must exist')
+    assertions = {}
+    evidence_files = {}
+    checked_evidence = {}
+    for assertion in plan['assertions']:
+        object_keys(assertion, ('source', 'external_id', 'record_sha256', 'identifiers', 'evidence'))
+        if any(not isinstance(assertion[key], str) or not assertion[key] for key in ('source', 'external_id')):
+            raise ValueError('Invalid source identity')
+        key = (assertion['source'], assertion['external_id'])
+        if key in assertions:
+            raise ValueError('Duplicate identity assertion')
+        ids = assertion['identifiers']
+        if not isinstance(ids, dict) or not ids:
+            raise ValueError('Identifiers must be a nonempty object')
+        for scheme, value in ids.items():
+            identifier(scheme, value)
+        if not isinstance(assertion['evidence'], list) or not assertion['evidence']:
+            raise ValueError('Evidence is required')
+        references = []
+        for evidence in assertion['evidence']:
+            object_keys(evidence, ('file', 'sha256', 'url', 'checked_at', 'note'))
+            if not all(isinstance(value, str) and value for value in evidence.values()):
+                raise ValueError('Invalid evidence fields')
+            url = urlsplit(evidence['url'])
+            if url.scheme != 'https' or not url.hostname or url.username or url.password:
+                raise ValueError('Evidence requires an HTTPS URL without credentials')
+            if datetime.fromisoformat(evidence['checked_at'].replace('Z', '+00:00')).utcoffset() is None:
+                raise ValueError('Evidence timestamp requires a timezone')
+            evidence_key = (str(Path(evidence['file']).absolute()), evidence['sha256'])
+            if evidence_key not in checked_evidence:
+                checked_evidence[evidence_key] = metadata(evidence['file'], evidence['sha256'])[0]
+            raw = checked_evidence[evidence_key]
+            evidence_files[evidence['sha256']] = raw
+            references.append({**evidence, 'file': 'evidence/' + evidence['sha256'] + '.json'})
+        assertions[key] = (assertion, references)
+    seen = set()
+    applied = set()
+    names = set()
+    report = {'format_version': 1, 'plan_sha256': digest(plan_raw),
+              'verification_scope': SCOPE, 'semantic_identity_verified': False,
+              'database_written': False, 'files': [], 'assertions_applied': 0, 'identifiers_added': 0}
+    with tempfile.TemporaryDirectory(prefix='.identifier-curation-', dir=output.parent) as temporary:
+        staging = Path(temporary) / 'prepared'
+        staging.mkdir()
+        (staging / 'evidence').mkdir()
+        (staging / 'plan.json').write_bytes(plan_raw)
+        for sha, raw in evidence_files.items():
+            (staging / 'evidence' / (sha + '.json')).write_bytes(raw)
+        for entry in plan['inputs']:
+            object_keys(entry, ('file', 'sha256', 'manifest_sha256'))
+            path = Path(entry['file'])
+            if path.suffix != '.jsonl' or path.name in names:
+                raise ValueError('Inputs require distinct JSONL basenames')
+            names.add(path.name)
+            _, manifest = metadata(path.with_suffix('.manifest.json'), entry['manifest_sha256'])
+            if not isinstance(manifest, dict) or not {'source','venue','year','track','count','list_url','file_sha256'} <= manifest.keys():
+                raise ValueError('Complete source manifest required')
+            if manifest['file_sha256'] != entry['sha256']:
+                raise ValueError('Manifest and plan hashes differ')
+            before = hashlib.sha256()
+            after = hashlib.sha256()
+            count = 0
+            with path.open('rb') as source, (staging / path.name).open('wb') as target:
+                for raw in source:
+                    before.update(raw)
+                    row = decode(raw)
+                    if any(row[field] != manifest[field] for field in ('source', 'venue', 'year', 'track')):
+                        raise ValueError('Record differs from source manifest')
+                    key = (row['source'], row['external_id'])
+                    if key in seen:
+                        raise ValueError('Duplicate source identity in input batch')
+                    seen.add(key)
+                    if key in assertions:
+                        assertion, references = assertions[key]
+                        check_hash(raw, assertion['record_sha256'])
+                        existing = row.get('identifiers', {})
+                        if existing is None:
+                            existing = {}
+                        if not isinstance(existing, dict):
+                            raise ValueError('Invalid existing identifiers')
+                        for scheme, value in assertion['identifiers'].items():
+                            if scheme in existing and existing[scheme] != value:
+                                raise ValueError('Existing identifier conflict')
+                        details = row.setdefault('metadata', {})
+                        if not isinstance(details, dict):
+                            raise ValueError('Invalid record metadata')
+                        history = details.setdefault('identifier_curation', [])
+                        if not isinstance(history, list):
+                            raise ValueError('Invalid curation history')
+                        history.append({'plan_sha256': digest(plan_raw), 'input_file_sha256': entry['sha256'],
+                                        'record_sha256': digest(raw), 'previous_identifiers': dict(existing),
+                                        'identifiers_asserted': assertion['identifiers'], 'evidence': references,
+                                        'verification_scope': SCOPE})
+                        report['identifiers_added'] += len(assertion['identifiers'].keys() - existing.keys())
+                        row['identifiers'] = {**existing, **assertion['identifiers']}
+                        raw = json_bytes(row)
+                        applied.add(key)
+                    target.write(raw)
+                    after.update(raw)
+                    count += 1
+            if before.hexdigest() != entry['sha256'] or not re.fullmatch('[0-9a-f]{64}', entry['sha256']):
+                raise ValueError('Input SHA256 mismatch')
+            if not count or type(manifest['count']) is not int or count != manifest['count']:
+                raise ValueError('Manifest count mismatch')
+            provenance = {'plan_sha256': digest(plan_raw), 'input_file_sha256': before.hexdigest(),
+                          'input_manifest_sha256': entry['manifest_sha256'], 'verification_scope': SCOPE}
+            prepared = {**manifest, 'file': str(output / path.name), 'file_sha256': after.hexdigest(),
+                        'identifier_curation': provenance}
+            (staging / path.with_suffix('.manifest.json').name).write_bytes(json_bytes(prepared))
+            report['files'].append({'file': str(output / path.name), 'count': count,
+                                    'input_sha256': before.hexdigest(), 'sha256': after.hexdigest()})
+        if applied != assertions.keys():
+            raise ValueError('Assertion target absent from input batch')
+        report['assertions_applied'] = len(applied)
+        (staging / 'curation-report.json').write_bytes(json_bytes(report))
+        # All files are on the same filesystem. No database or input mutation.
+        if output.exists():
+            raise ValueError('Output directory appeared during preparation')
+        staging.rename(output)
+    return report
+
+
+def main(argv=None, *, prog=None):
+    parser = argparse.ArgumentParser(prog=prog, description=__doc__)
+    parser.add_argument('--plan', required=True, type=Path)
+    parser.add_argument('--output', required=True, type=Path)
+    args = parser.parse_args(argv)
+    print(json.dumps(prepare(args.plan, args.output), ensure_ascii=False))

paper_atlas/ingestion/records.py

@@ -12,55 +12,6 @@ from paper_atlas.collectors.public_openreview import SOURCE as PUBLIC_SOURCE
 from paper_atlas.collectors.icml_public_archive import SOURCE as ARCHIVE_SOURCE
 
 
-def coverage_options(manifest):
-    mode = manifest.get('coverage_mode', 'snapshot')
-    if mode not in ('snapshot', 'incremental'):
-        raise ValueError('Unsupported coverage mode')
-    if mode == 'snapshot':
-        return None
-    required = ('venue', 'year', 'track', 'count', 'coverage_previous_count',
-                'coverage_previous_expected_count', 'coverage_expected_count')
-    if any(key not in manifest for key in required) or manifest.get('status') != 'partial':
-        raise ValueError('Incremental coverage requires a partial, explicitly scoped manifest')
-    previous = manifest['coverage_previous_count']
-    expected = manifest['coverage_expected_count']
-    prior_expected = manifest['coverage_previous_expected_count']
-    if (type(previous) is not int or previous < 0 or type(expected) is not int or expected < 1
-            or type(manifest['count']) is not int or manifest['count'] < 1
-            or expected < max(previous, manifest['count'])
-            or (prior_expected is not None and (type(prior_expected) is not int or prior_expected < 0
-                                               or expected < prior_expected))):
-        raise ValueError('Invalid incremental coverage counts')
-    return previous, prior_expected, expected
-
-
-def source_record_count(c, source, edition):
-    return c.execute('''SELECT count(DISTINCT ss.source_record_id) FROM submission_sources ss
-        JOIN submissions s ON s.id=ss.submission_id
-        JOIN source_records sr ON sr.id=ss.source_record_id
-        WHERE s.edition_id=? AND sr.source_id=?''', (edition, source)).fetchone()[0]
-
-
-def require_incremental_parent(c, source, manifest, options):
-    edition = c.execute('''SELECT e.id FROM editions e JOIN venues v ON v.id=e.venue_id
-        WHERE lower(v.acronym)=lower(?) AND e.year=? AND e.track=?''',
-        (manifest['venue'], manifest['year'], manifest['track'])).fetchone()
-    entries = c.execute('''SELECT list_url,status,expected_count,imported_count
-        FROM edition_sources WHERE edition_id=? AND source_id=?''',
-        (edition[0], source)).fetchall() if edition else []
-    previous, prior_expected, _ = options
-    if entries:
-        if len(entries) != 1 or entries[0][0] != manifest['list_url']:
-            raise ValueError('Incremental coverage requires one unambiguous source listing')
-        if tuple(entries[0][1:]) != ('partial', prior_expected, previous):
-            raise ValueError('Incremental coverage parent changed or is not partial')
-    elif previous != 0 or prior_expected is not None:
-        raise ValueError('Incremental coverage parent is absent')
-    actual = source_record_count(c, source, edition[0]) if edition else 0
-    if actual != previous:
-        raise ValueError('Incremental coverage parent differs from actual source records')
-
-
 def ingest_file(db,path, *, pg_service=None, pg_schema=None):
     path = Path(path)
     manifest_path = path.with_suffix('.manifest.json')
@@ -73,7 +24,6 @@ def ingest_file(db,path, *, pg_service=None, pg_schema=None):
     source_name = manifest.get('source')
     if not source_name:
         raise ValueError('A source manifest is required')
-    incremental = coverage_options(manifest)
     store = open_store(db, pg_service=pg_service, pg_schema=pg_schema)
     c = store.conn
     run = None
@@ -85,8 +35,6 @@ def ingest_file(db,path, *, pg_service=None, pg_schema=None):
         source_id = store.source(source_name,manifest['list_url'])
         run = inserted_id(c,'INSERT INTO ingest_runs(source_id,checkpoint_json) VALUES(?,?)',(source_id,json.dumps({'file':str(path),'expected_sha256':manifest.get('file_sha256')})))
         c.commit()
-        if incremental is not None:
-            require_incremental_parent(c, source_id, manifest, incremental)
         digest = hashlib.sha256()
         with path.open('rb') as stream:
             for line in stream:
@@ -134,25 +82,8 @@ def ingest_file(db,path, *, pg_service=None, pg_schema=None):
         if expected is not None and expected != count:
             raise ValueError(f'Manifest count {expected} differs from records {count}')
         complete=expected==count and manifest.get('status') not in ('partial','blocked') and not manifest.get('missing_abstract_count',0) and not missing_abstracts
-        imported = count
-        condition = ''
-        condition_params = ()
-        if incremental is not None:
-            previous, prior_expected, expected = incremental
-            imported = source_record_count(c, source_id, edition)
-            if imported < previous or imported > expected:
-                raise ValueError('Incremental coverage exceeds the declared source bounds')
-            complete = False
-            condition = " WHERE edition_sources.imported_count=? AND COALESCE(edition_sources.expected_count,-1)=? AND edition_sources.status='partial'"
-            condition_params = (previous, -1 if prior_expected is None else prior_expected)
-        result = c.execute('INSERT INTO edition_sources(edition_id,source_id,list_url,status,expected_count,imported_count,last_success_at) VALUES(?,?,?,?,?,?,?) ON CONFLICT(edition_id,source_id,list_url) DO UPDATE SET status=excluded.status,expected_count=excluded.expected_count,imported_count=excluded.imported_count,last_success_at=excluded.last_success_at,last_error=NULL'+condition,(edition,source_id,manifest['list_url'],'complete' if complete else 'partial',expected,imported,time.strftime('%Y-%m-%dT%H:%M:%SZ',time.gmtime()),*condition_params))
-        if incremental is not None and result.rowcount != 1:
-            raise ValueError('Incremental coverage parent changed during import')
-        checkpoint = {'file':str(path),'file_sha256':digest.hexdigest()}
-        if incremental is not None:
-            checkpoint['coverage'] = {'mode':'incremental','previous_count':previous,
-                                      'expected_count':expected,'imported_count':imported}
-        c.execute("UPDATE ingest_runs SET status='complete',finished_at=?,records_seen=?,records_written=?,checkpoint_json=? WHERE id=?",(timestamp(),count,count,json.dumps(checkpoint),run))
+        c.execute('INSERT INTO edition_sources(edition_id,source_id,list_url,status,expected_count,imported_count,last_success_at) VALUES(?,?,?,?,?,?,?) ON CONFLICT(edition_id,source_id,list_url) DO UPDATE SET status=excluded.status,expected_count=excluded.expected_count,imported_count=excluded.imported_count,last_success_at=excluded.last_success_at,last_error=NULL',(edition,source_id,manifest['list_url'],'complete' if complete else 'partial',expected,count,time.strftime('%Y-%m-%dT%H:%M:%SZ',time.gmtime())))
+        c.execute("UPDATE ingest_runs SET status='complete',finished_at=?,records_seen=?,records_written=?,checkpoint_json=? WHERE id=?",(timestamp(),count,count,json.dumps({'file':str(path),'file_sha256':digest.hexdigest()}),run))
         c.commit()
         print(json.dumps({'file':str(path),'status':'imported','records':count,'missing_abstract_count':missing_abstracts}),flush=True)
     except BaseException as exc:

tests/test_identifier_curation.py

@@ -0,0 +1,185 @@
+import copy
+import json
+from pathlib import Path
+import tempfile
+import unittest
+
+from paper_atlas.curation.identifiers import SCOPE, digest, json_bytes, prepare
+from paper_atlas.db.store import Store
+from paper_atlas.ingestion.records import ingest_file
+
+
+class IdentifierCurationTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.root = Path(self.tmp.name)
+        self.output = self.root / 'prepared'
+        self.row = {'source':'dblp','external_id':'conf/uai/example','url':'https://example.org/paper',
+                    'title':'A sufficiently long example research paper title','abstract':None,
+                    'authors':['Jane Doe','John Smith'],'venue':'UAI','year':2001,'track':'main',
+                    'status':'accepted','tags':[{'namespace':'dblp:topic','value':'version\u2029study\x00'}],
+                    'metadata':{'pages':'203-210','raw_title':'Original\u2029title'}}
+        self.evidence = self.root / 'evidence.json'
+        self.evidence.write_bytes(json_bytes({'url':'https://example.org/archive','comment':'Conference version'}))
+        self.path = self.root / 'records.jsonl'
+        self.raw = json_bytes(self.row)
+        self.path.write_bytes(self.raw)
+        self.manifest = {'source':'dblp','venue':'UAI','year':2001,'track':'main','count':1,
+                         'list_url':'https://example.org/list','file_sha256':digest(self.raw),
+                         'status':'partial','missing_abstract_count':1,'parser_version':2}
+        self.path.with_suffix('.manifest.json').write_bytes(json_bytes(self.manifest))
+        self.plan = {'format_version':1,'inputs':[self.input(self.path)],'assertions':[
+            {'source':'dblp','external_id':self.row['external_id'],'record_sha256':digest(self.raw),
+             'identifiers':{'arxiv':'1408.1485'},'evidence':[{
+                 'file':str(self.evidence),'sha256':digest(self.evidence.read_bytes()),
+                 'url':'https://example.org/archive','checked_at':'2026-11-08T08:00:00Z',
+                 'note':'Reviewed title, authors, publication and version evidence'}]}]}
+
+    def input(self, path):
+        return {'file':str(path),'sha256':digest(path.read_bytes()),
+                'manifest_sha256':digest(path.with_suffix('.manifest.json').read_bytes())}
+
+    def run_plan(self, plan=None):
+        path = self.root / 'plan.json'
+        path.write_bytes(json_bytes(self.plan if plan is None else plan))
+        return prepare(path, self.output)
+
+    def test_only_identifiers_and_provenance_change_and_raw_inputs_survive(self):
+        result = self.run_plan()
+        self.assertEqual(self.path.read_bytes(), self.raw)
+        prepared = json.loads((self.output / self.path.name).read_bytes())
+        provenance = prepared['metadata'].pop('identifier_curation')
+        self.assertEqual(prepared.pop('identifiers'), {'arxiv':'1408.1485'})
+        self.assertEqual(prepared, self.row)
+        self.assertEqual(provenance[0]['record_sha256'], digest(self.raw))
+        self.assertEqual((self.output / provenance[0]['evidence'][0]['file']).read_bytes(), self.evidence.read_bytes())
+        manifest = json.loads((self.output / 'records.manifest.json').read_bytes())
+        for key, value in self.manifest.items():
+            if key != 'file_sha256': self.assertEqual(manifest[key], value)
+        self.assertEqual(manifest['file_sha256'], digest((self.output / self.path.name).read_bytes()))
+        self.assertEqual(result['assertions_applied'], 1)
+        self.assertEqual(result['identifiers_added'], 1)
+        self.assertEqual(result['verification_scope'], SCOPE)
+        self.assertFalse(result['semantic_identity_verified'])
+        self.assertFalse(result['database_written'])
+        self.assertEqual(list(self.output.glob('*.jsonl')), [self.output / self.path.name])
+        with self.assertRaisesRegex(ValueError, 'already exist'): self.run_plan()
+
+    def test_late_invalid_file_publishes_nothing(self):
+        second = self.root / 'second.jsonl'
+        second.write_bytes(json_bytes({**self.row,'external_id':'second'}))
+        manifest = {**self.manifest,'file_sha256':digest(second.read_bytes()),'count':2}
+        second.with_suffix('.manifest.json').write_bytes(json_bytes(manifest))
+        self.plan['inputs'].append(self.input(second))
+        with self.assertRaisesRegex(ValueError, 'count mismatch'): self.run_plan()
+        self.assertFalse(self.output.exists())
+        self.assertFalse(list(self.root.glob('.identifier-curation-*')))
+        self.assertEqual(self.path.read_bytes(), self.raw)
+
+    def test_unasserted_rows_keep_exact_physical_bytes_and_doi_is_additive(self):
+        untouched = (json.dumps({**self.row,'external_id':'unselected'},ensure_ascii=False,indent=None) + '  ').encode()
+        self.path.write_bytes(self.raw + untouched)
+        self.path.with_suffix('.manifest.json').write_bytes(json_bytes({**self.manifest,
+            'file_sha256':digest(self.path.read_bytes()),'count':2,'missing_abstract_count':2}))
+        self.plan['inputs'] = [self.input(self.path)]
+        self.plan['assertions'][0]['identifiers']['doi'] = '10.1234/example'
+        result = self.run_plan()
+        with (self.output / self.path.name).open('rb') as stream:
+            first = json.loads(next(stream))
+            self.assertEqual(next(stream), untouched)
+        self.assertEqual(first['identifiers'], {'arxiv':'1408.1485','doi':'10.1234/example'})
+        self.assertEqual(result['identifiers_added'], 2)
+
+    def test_missing_duplicate_and_mismatched_assertions_fail_closed(self):
+        for mode in ('missing','duplicate','line_hash','duplicate_input'):
+            with self.subTest(mode=mode):
+                plan = copy.deepcopy(self.plan)
+                assertion = plan['assertions'][0]
+                if mode == 'missing': assertion['external_id'] = 'absent'
+                elif mode == 'duplicate': plan['assertions'].append(copy.deepcopy(assertion))
+                elif mode == 'line_hash': assertion['record_sha256'] = '0' * 64
+                else:
+                    other = self.root / 'duplicate.jsonl'
+                    other.write_bytes(self.raw)
+                    other.with_suffix('.manifest.json').write_bytes(json_bytes(self.manifest))
+                    plan['inputs'].append(self.input(other))
+                with self.assertRaises(ValueError): self.run_plan(plan)
+                self.assertFalse(self.output.exists())
+
+    def test_identifier_conflict_does_not_overwrite_and_same_value_is_auditable(self):
+        self.row['identifiers'] = {'arxiv':'cs/0307069'}
+        self.path.write_bytes(json_bytes(self.row))
+        self.manifest['file_sha256'] = digest(self.path.read_bytes())
+        self.path.with_suffix('.manifest.json').write_bytes(json_bytes(self.manifest))
+        self.plan['inputs'] = [self.input(self.path)]
+        self.plan['assertions'][0]['record_sha256'] = digest(self.path.read_bytes())
+        with self.assertRaisesRegex(ValueError, 'identifier conflict'): self.run_plan()
+        self.assertFalse(self.output.exists())
+        self.plan['assertions'][0]['identifiers']['arxiv'] = 'cs/0307069'
+        self.assertEqual(self.run_plan()['identifiers_added'], 0)
+
+    def test_unsupported_changes_identifiers_and_unbound_evidence_are_rejected(self):
+        for mode in ('title','bad_month','version','unknown_scheme','doi_case','evidence_hash','credentials','timestamp'):
+            with self.subTest(mode=mode):
+                plan = copy.deepcopy(self.plan)
+                assertion = plan['assertions'][0]
+                evidence = assertion['evidence'][0]
+                if mode == 'title': assertion['title'] = 'Rewrite forbidden'
+                elif mode == 'bad_month': assertion['identifiers'] = {'arxiv':'1413.12345'}
+                elif mode == 'version': assertion['identifiers'] = {'arxiv':'1408.1485v2'}
+                elif mode == 'unknown_scheme': assertion['identifiers'] = {'url':'https://example.org'}
+                elif mode == 'doi_case': assertion['identifiers'] = {'doi':'10.1234/UPPER'}
+                elif mode == 'evidence_hash': evidence['sha256'] = '0' * 64
+                elif mode == 'credentials': evidence['url'] = 'https://user:secret@example.org'
+                else: evidence['checked_at'] = '2026-11-08T08:00:00'
+                with self.assertRaises(ValueError): self.run_plan(plan)
+                self.assertFalse(self.output.exists())
+
+    def test_input_and_manifest_hashes_and_duplicate_json_keys_are_checked(self):
+        for mode in ('input','manifest','duplicate_json','invalid_evidence_json'):
+            with self.subTest(mode=mode):
+                plan = copy.deepcopy(self.plan)
+                if mode == 'input':
+                    plan['inputs'][0]['sha256'] = '0' * 64
+                elif mode == 'manifest':
+                    plan['inputs'][0]['manifest_sha256'] = '0' * 64
+                elif mode == 'duplicate_json':
+                    raw = self.raw[:-2] + b',"source":"dblp"}\n'
+                    self.path.write_bytes(raw)
+                    self.path.with_suffix('.manifest.json').write_bytes(json_bytes({**self.manifest,'file_sha256':digest(raw)}))
+                    plan['inputs'] = [self.input(self.path)]
+                else:
+                    self.evidence.write_bytes(b'not json')
+                    plan['assertions'][0]['evidence'][0]['sha256'] = digest(self.evidence.read_bytes())
+                with self.assertRaises(ValueError): self.run_plan(plan)
+                self.assertFalse(self.output.exists())
+                self.path.write_bytes(self.raw)
+                self.path.with_suffix('.manifest.json').write_bytes(json_bytes(self.manifest))
+
+    def test_curated_anchor_import_does_not_merge_extended_version_or_its_history(self):
+        db = self.root / 'papers.sqlite'
+        store = Store(db)
+        try:
+            original, _ = store.record('arxiv','1408.1485','https://example.org/one',self.row['title'],None,self.row['authors'],{'arxiv':'1408.1485'})
+            extended, record = store.record('arxiv','cs/0307069','https://example.org/two',self.row['title'],None,self.row['authors'],{'arxiv':'cs/0307069'})
+            store.decision(extended,record,'ICLR',2025,'main','another-attempt','https://example.org/reject','rejected','Reject')
+            store.conn.commit()
+        finally:
+            store.conn.close()
+        self.run_plan()
+        ingest_file(db,self.output / self.path.name)
+        store = Store(db)
+        try:
+            c = store.conn
+            self.assertEqual(c.execute('SELECT count(*) FROM papers').fetchone()[0], 2)
+            self.assertEqual(c.execute('SELECT paper_id FROM source_records WHERE external_id=?',(self.row['external_id'],)).fetchone()[0], original)
+            self.assertEqual(c.execute('SELECT s.paper_id FROM submissions s JOIN editions e ON e.id=s.edition_id JOIN venues v ON v.id=e.venue_id WHERE v.acronym=?',('ICLR',)).fetchone()[0], extended)
+            self.assertEqual(c.execute('SELECT count(*) FROM paper_merges').fetchone()[0], 0)
+            self.assertEqual(c.execute('PRAGMA foreign_key_check').fetchall(), [])
+        finally:
+            store.conn.close()
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_incremental_coverage.py

@@ -1,116 +0,0 @@
-from contextlib import redirect_stdout
-import hashlib
-import io
-import json
-from pathlib import Path
-import sqlite3
-import tempfile
-import unittest
-
-from paper_atlas.ingestion.records import ingest_file
-
-
-class IncrementalCoverageTests(unittest.TestCase):
-    def setUp(self):
-        self.tmp = tempfile.TemporaryDirectory()
-        self.addCleanup(self.tmp.cleanup)
-        self.root = Path(self.tmp.name)
-        self.db = self.root / 'papers.sqlite'
-
-    def file(self, name, external_ids, **options):
-        rows = [{'source':'example','external_id':key,'url':'https://example.org/'+key,
-                 'venue':'Example','year':2026,'track':'main','status':'accepted',
-                 'title':'Independent research contribution '+key,'abstract':'Known abstract',
-                 'authors':[], 'identifiers':{'doi':'10.1234/'+key}} for key in external_ids]
-        path = self.root / (name+'.jsonl')
-        path.write_bytes((''.join(json.dumps(row)+'\n' for row in rows)).encode())
-        manifest = {key:rows[0][key] for key in ('source','venue','year','track')}
-        manifest.update(list_url='https://example.org/2026', count=len(rows),status='partial',
-                        file_sha256=hashlib.sha256(path.read_bytes()).hexdigest())
-        manifest.update(options)
-        path.with_suffix('.manifest.json').write_text(json.dumps(manifest))
-        return path
-
-    def ingest(self, path):
-        with redirect_stdout(io.StringIO()):
-            ingest_file(self.db,path)
-
-    def incremental(self, previous=2, prior_expected=2, expected=4):
-        return {'coverage_mode':'incremental','coverage_previous_count':previous,
-                'coverage_previous_expected_count':prior_expected,'coverage_expected_count':expected}
-
-    def coverage(self):
-        with sqlite3.connect(self.db) as c:
-            return c.execute('SELECT status,expected_count,imported_count FROM edition_sources').fetchall()
-
-    def test_append_and_refresh_count_actual_source_records_without_erasing_earlier_batch(self):
-        self.ingest(self.file('first',['a','b']))
-        self.ingest(self.file('append',['c'],**self.incremental()))
-        self.assertEqual(self.coverage(),[('partial',4,3)])
-        self.ingest(self.file('refresh',['c'],**self.incremental(3,4,4)))
-        self.assertEqual(self.coverage(),[('partial',4,3)])
-        with sqlite3.connect(self.db) as c:
-            self.assertEqual(c.execute('SELECT count(*) FROM source_records').fetchone()[0],3)
-            self.assertEqual(c.execute('SELECT count(*) FROM submissions').fetchone()[0],3)
-            runs = c.execute('SELECT records_written,checkpoint_json FROM ingest_runs ORDER BY id').fetchall()
-            self.assertEqual([r[0] for r in runs],[2,1,1])
-            self.assertEqual(json.loads(runs[-1][1])['coverage'],
-                             {'mode':'incremental','previous_count':3,'expected_count':4,'imported_count':3})
-            self.assertEqual(c.execute('PRAGMA foreign_key_check').fetchall(),[])
-
-    def test_stale_parent_stops_before_source_writes(self):
-        self.ingest(self.file('first',['a','b']))
-        with self.assertRaisesRegex(ValueError,'parent changed'):
-            self.ingest(self.file('stale',['c'],**self.incremental(1,2,4)))
-        self.assertEqual(self.coverage(),[('partial',2,2)])
-        with sqlite3.connect(self.db) as c:
-            self.assertEqual(c.execute("SELECT count(*) FROM source_records WHERE external_id='c'").fetchone()[0],0)
-            self.assertEqual(c.execute('SELECT status FROM ingest_runs ORDER BY id DESC LIMIT 1').fetchone()[0],'failed')
-
-    def test_late_source_bound_failure_rolls_back_new_records_and_coverage(self):
-        self.ingest(self.file('first',['a','b']))
-        with self.assertRaisesRegex(ValueError,'source bounds'):
-            self.ingest(self.file('overflow',['c'],**self.incremental(2,2,2)))
-        self.assertEqual(self.coverage(),[('partial',2,2)])
-        with sqlite3.connect(self.db) as c:
-            for table in ('papers','source_records','submissions','decisions'):
-                self.assertEqual(c.execute('SELECT count(*) FROM '+table).fetchone()[0],2)
-
-    def test_multiple_listings_cannot_be_silently_aggregated(self):
-        self.ingest(self.file('one',['a']))
-        self.ingest(self.file('two',['b'],list_url='https://example.org/another-list'))
-        before=self.coverage()
-        with self.assertRaisesRegex(ValueError,'unambiguous source listing'):
-            self.ingest(self.file('ambiguous',['c'],**self.incremental(2,1,4)))
-        self.assertEqual(self.coverage(),before)
-
-    def test_untracked_source_rows_require_reconciliation(self):
-        self.ingest(self.file('first',['a','b']))
-        with sqlite3.connect(self.db) as c:
-            c.execute('UPDATE edition_sources SET imported_count=1')
-        with self.assertRaisesRegex(ValueError,'actual source records'):
-            self.ingest(self.file('stale-summary',['c'],**self.incremental(1,2,4)))
-        self.assertEqual(self.coverage(),[('partial',2,1)])
-
-    def test_first_partial_chunk_can_declare_an_independently_known_source_total(self):
-        self.ingest(self.file('first',['a'],**self.incremental(0,None,4)))
-        self.assertEqual(self.coverage(),[('partial',4,1)])
-
-    def test_complete_listing_is_not_demoted_by_a_partial_chunk(self):
-        self.ingest(self.file('complete',['a','b'],status='complete'))
-        with self.assertRaisesRegex(ValueError,'not partial'):
-            self.ingest(self.file('append',['c'],**self.incremental()))
-        self.assertEqual(self.coverage(),[('complete',2,2)])
-
-    def test_invalid_mode_counts_and_missing_scope_fail_before_database_creation(self):
-        for options in ({'coverage_mode':'unknown'}, self.incremental(True,2,4),
-                        self.incremental(2,5,4),self.incremental(-1,2,4),
-                        {'coverage_mode':'incremental'},
-                        {**self.incremental(),'status':'complete'}):
-            with self.subTest(options=options):
-                with self.assertRaises(ValueError):self.ingest(self.file('invalid',['a'],**options))
-                self.assertFalse(self.db.exists())
-
-
-if __name__=='__main__':
-    unittest.main()

tests/test_postgres_ingestion.py

@@ -87,82 +87,6 @@ class PostgreSQLIngestionTests(unittest.TestCase):
         with redirect_stdout(io.StringIO()):
             ingest_file(None, path, pg_service=self.service, pg_schema=self.schema)
 
-    def test_incremental_source_coverage_counts_prior_chunks_and_refreshes_on_postgres(self):
-        rows = [self.row(source='dblp',external=key,status='accepted',year=2001,venue='UAI')
-                for key in ('incremental1','incremental2','incremental3')]
-        listing = 'https://example.org/uai/2001'
-        def manifest(path, **options):
-            target=path.with_suffix('.manifest.json')
-            value=json.loads(target.read_text())
-            value.update(list_url=listing,status='partial',**options)
-            target.write_text(json.dumps(value))
-        first=self.file('initial',rows[:2]);manifest(first);self.ingest(first)
-        second=self.file('append',[rows[2]])
-        manifest(second,coverage_mode='incremental',coverage_previous_count=2,
-                 coverage_previous_expected_count=2,coverage_expected_count=4)
-        self.ingest(second)
-        manifest(second,coverage_mode='incremental',coverage_previous_count=3,
-                 coverage_previous_expected_count=4,coverage_expected_count=4)
-        self.ingest(second)
-        with closing(ReadConnection(self.service,self.schema)) as c:
-            coverage=c.execute('SELECT status,expected_count,imported_count FROM edition_sources WHERE list_url=?',(listing,)).fetchone()
-            self.assertEqual(tuple(coverage.values),('partial',4,3))
-            stored=c.execute('SELECT abstract FROM source_records WHERE external_id=?',('incremental3',)).fetchone()
-            self.assertEqual(stored[0],rows[2]['abstract'])
-            self.assertEqual(c.execute("SELECT count(*) FROM source_records r JOIN sources s ON s.id=r.source_id WHERE s.name='dblp'").fetchone()[0],3)
-            run=c.execute('SELECT records_written,checkpoint_json FROM ingest_runs ORDER BY id DESC LIMIT 1').fetchone()
-            self.assertEqual(run[0],1)
-            self.assertEqual(json.loads(run[1])['coverage']['imported_count'],3)
-
-    def test_incremental_postgres_bound_failure_rolls_back_rows_and_summary(self):
-        listing='https://example.org/uai/2001'
-        rows=[self.row(source='dblp',external=key,status='accepted',year=2001,venue='UAI')
-              for key in ('bound1','bound2','bound3')]
-        first=self.file('initial',rows[:2]);target=first.with_suffix('.manifest.json')
-        value=json.loads(target.read_text());value.update(list_url=listing,status='partial');target.write_text(json.dumps(value))
-        self.ingest(first)
-        with closing(ReadConnection(self.service,self.schema)) as c:
-            before={table:c.execute('SELECT count(*) FROM '+table).fetchone()[0]
-                    for table in ('papers','source_records','submissions','decisions')}
-        path=self.file('overflow',[rows[2]]);target=path.with_suffix('.manifest.json')
-        value=json.loads(target.read_text());value.update(list_url=listing,status='partial',coverage_mode='incremental',
-            coverage_previous_count=2,coverage_previous_expected_count=2,coverage_expected_count=2)
-        target.write_text(json.dumps(value))
-        with self.assertRaisesRegex(ValueError,'source bounds'):self.ingest(path)
-        with closing(ReadConnection(self.service,self.schema)) as c:
-            self.assertEqual({table:c.execute('SELECT count(*) FROM '+table).fetchone()[0] for table in before},before)
-            coverage=c.execute('SELECT status,expected_count,imported_count FROM edition_sources WHERE list_url=?',(listing,)).fetchone()
-            self.assertEqual(tuple(coverage.values),('partial',2,2))
-            self.assertEqual(c.execute('SELECT status FROM ingest_runs ORDER BY id DESC LIMIT 1').fetchone()[0],'failed')
-
-    def test_incremental_postgres_does_not_overwrite_a_concurrently_changed_parent(self):
-        listing='https://example.org/uai/2001'
-        rows=[self.row(source='dblp',external=key,status='accepted',year=2001,venue='UAI')
-              for key in ('concurrent1','concurrent2','concurrent3')]
-        first=self.file('initial',rows[:2]);target=first.with_suffix('.manifest.json')
-        value=json.loads(target.read_text());value.update(list_url=listing,status='partial');target.write_text(json.dumps(value))
-        self.ingest(first)
-        path=self.file('racing',[rows[2]]);target=path.with_suffix('.manifest.json')
-        value=json.loads(target.read_text());value.update(list_url=listing,status='partial',coverage_mode='incremental',
-            coverage_previous_count=2,coverage_previous_expected_count=2,coverage_expected_count=4)
-        target.write_text(json.dumps(value))
-        original=Store.record
-        def change_parent(store,*args,**kwargs):
-            result=original(store,*args,**kwargs)
-            # A separate committed SQL writer need not honor the native
-            # advisory lock. Its update must survive our rejected file.
-            with connect(self.service) as other:
-                other.execute('SET search_path TO '+quote(self.schema)+',pg_catalog')
-                other.execute('UPDATE edition_sources SET expected_count=99 WHERE list_url=%s',(listing,))
-            return result
-        with patch.object(Store,'record',change_parent):
-            with self.assertRaisesRegex(ValueError,'parent changed during import'):self.ingest(path)
-        with closing(ReadConnection(self.service,self.schema)) as c:
-            coverage=c.execute('SELECT status,expected_count,imported_count FROM edition_sources WHERE list_url=?',(listing,)).fetchone()
-            self.assertEqual(tuple(coverage.values),('partial',99,2))
-            self.assertEqual(c.execute('SELECT count(*) FROM source_records WHERE external_id=?',('concurrent3',)).fetchone()[0],0)
-            self.assertEqual(c.execute('SELECT status FROM ingest_runs ORDER BY id DESC LIMIT 1').fetchone()[0],'failed')
-
     def test_native_writes_keep_nul_historical_ids_and_later_acceptance(self):
         store = self.store()
         try: