Hold untitled public assertions and retain explicit withdrawals

Reviewed-on: #16

.gitea/workflows/check.yml

@@ -52,6 +52,7 @@ jobs:
               subprocess.run([python,'-I','-B','-m','paper_atlas','import','records','--help'],cwd=outside,check=True)
               subprocess.run([python,'-I','-B','-m','paper_atlas','import','arxiv','--help'],cwd=outside,check=True)
               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','-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/data-sources.md

@@ -93,6 +93,8 @@ paper-atlas audit public-openreview --db /data/paper-index/papers.sqlite \
 
 输出按会议、届次及 main、position、datasets_benchmarks 分文件。workshop 默认进入单独的 held 清单；有明确 workshop 名称时可通过 `--include-workshops` 单列采集，不合并到主会。Retraction、未映射的轨道或负面标签也保留在 `held-records.jsonl`，不强改为 Reject 或 Withdrawn。当前决定表没有独立 retracted 状态，因此撤回发表记录仍需后续专门建模。
 
+`parser_version=2` 还会将明确负面标签但标题缺失、空白或 null 的记录放入 `held-records.jsonl`，保留原始状态、forum ID、固定来源版本和行号；不制造标题或导入不可识别的论文，也不让这一条缺失阻断其他可用记录。held 文件哈希与各原因数量保存在 inventory；非文本标题、错误身份、冲突届次和哈希不符仍使整批失败。旧年 ICLR 镜像的明确 `Withdraw` 标签归为 `withdrawn`，`Workshop`、`Poster Workshop`、`Oral Workshop` 均不作为拒稿。采集文件与导入主库仍是两个独立步骤。
+
 该来源明确标记为 `curated_public_status_assertion`：保留源 commit、文件哈希、行号、原标签和 forum ID，不声称已核验原始 PC 签名。只保存标题、摘要、公开作者名、关键词、领域、索引与决定来源，不缓存镜像中的个人资料、邮箱、性别、评审内容或 PDF。所有输出清单均为 partial；盘点中的零条记录仅描述这个固定镜像。
 
 入库前审计核验文件哈希、数量、轨道和来源声明，并用真实 forum ID 对比同届同轨道已有投稿。存在不同决定时阻止整批导入，交由来源和时序核查；另一会议的后续录用属于独立投稿历史，不构成该次拒稿的冲突。同一 forum 的镜像不新增投稿事件，但论文实体的跨平台去重仍须在全量结束后核验。审计通过的报告包含带 SHA-256 的导入队列；应在单写入维护锁下及时串行导入，主库变化后重新审计。

docs/operations.md

@@ -38,7 +38,11 @@ SQLite 允许文本包含 `U+0000`（NUL），PostgreSQL `text` 不允许。迁
 
 论文身份匹配、跨来源合并、投稿/决定来源和标签维护复用同一套领域逻辑。PostgreSQL 使用原生 RETURNING 和 identity 序列分配 ID，继承迁移时设定的历史 ID 高水位；失败和冲突尝试可以留下序列空洞，不能把最大 ID 当作行数。绑定文本按 `nul-soh-v1` 编码，结果解码，GIN 随论文写入自动更新。
 
-会议文件的数据变化在同一事务内完成，manifest 数量不符等错误会回滚整份文件，`ingest_runs` 另保留 failed、已读数和零写入数；身份回填整个队列一次提交。arXiv 每 500 行提交数据和 checkpoint，异常只回滚尚未提交部分，重试必须仍为同一压缩包。大批次开始前核对独立审计清单、当前主库冲突和磁盘空间；完成后对数量、实际来源/决定关联及检索做数据验收，代码测试成功不代表生产导入成功。
+会议文件的数据变化在同一事务内完成，manifest 数量不符等错误会回滚整份文件，`ingest_runs` 另保留 failed、已处理数和零写入数；身份回填整个队列一次提交。arXiv 每 500 行提交数据和 checkpoint，异常只回滚尚未提交部分，重试必须仍为同一压缩包。大批次开始前核对独立审计清单、当前主库冲突和磁盘空间；完成后对数量、实际来源/决定关联及检索做数据验收，代码测试成功不代表生产导入成功。
+
+`audit public-openreview --pg-service ...` 在只读快照中核验两个受支持公开负面决定来源的固定 revision、逐文件哈希、manifest、原始出处标记和既有同届投稿。最新决定排序与查询端一致：观察时间、决定日期（为空时回退观察时间）、ID。实际导入持有写锁后再次复核该投稿的最新状态，阻止过期预检或弱来源标签覆盖冲突决定；冲突文件整体回滚并等待进一步证据核查，其他会议录用不构成同一次投稿的冲突。
+
+会议导入按原始字节流计算 SHA-256，在提交前与 manifest 提供的 `file_sha256` 比较；不符则回滚整份文件。成功批次的 checkpoint 会记录实际读取内容的哈希，便于复核导入时使用的文件版本。
 
 代码以固定 commit 部署，已经启动的任务继续使用其原版本。目录重构不要求迁移数据库，也不更改已有 JSONL 和 manifest 格式。
 

docs/usage.md

@@ -41,6 +41,16 @@ paper-atlas audit public-openreview --help
 
 其他会议的公开拒稿使用 `collect public-openreview`，再以 `audit public-openreview` 核对既有投稿身份和决定冲突。来源版本、逐文件状态盘点和未解决记录分别保存在 inventory 与 held 清单中，具体命令见[数据源说明](data-sources.md#其他会议的公开拒稿)。这类公开子集不能当作会议完整投稿名单。
 
+主库切换后，预检须直接连接当前 PostgreSQL，不能用冻结的 SQLite 备份推断当前决定：
+
+```sh
+paper-atlas audit public-openreview --pg-service foundry-research \
+  --inventory /data/paper-index/collected/public-openreview-v1/public-openreview-inventory.json \
+  --out /data/paper-index/collected/public-negative-pg-preflight.json
+```
+
+预检使用只读一致性快照，报告会记录实际数据库、schema、文件哈希和同届次/轨道的决定冲突；其他会议的录用保留为独立历史。预检不预留写锁，实际导入这两类公开负面决定时会在写事务中再次复核同一次投稿的最新状态；过期预检报告不能覆盖已经变化的决定。
+
 ## 3. 串行导入
 
 PostgreSQL 主库逐文件导入已经核验的资料：

paper_atlas/audits/public_openreview.py

@@ -4,14 +4,14 @@ from contextlib import closing
 import hashlib
 import json
 from pathlib import Path
-import sqlite3
 
 from paper_atlas.collectors.public_openreview import SOURCE, REVISION
 from paper_atlas.collectors.icml_public_archive import SOURCE as ARCHIVE_SOURCE, REVISION as ARCHIVE_REVISION
-from paper_atlas.db.submissions import timestamp
+from paper_atlas.db.submissions import timestamp, latest_status
+from paper_atlas.db.backend import database_arguments, open_read_connection
 
 
-def audit(db, inventory, out):
+def audit(db, inventory, out, *, pg_service=None, pg_schema=None):
     inventory = Path(inventory).resolve()
     result = json.loads(inventory.read_text())
     specs = {SOURCE: (REVISION, 'curated_public_status_assertion'),
@@ -20,17 +20,18 @@ def audit(db, inventory, out):
     if source not in specs or result.get('revision') != specs[source][0]:
         raise ValueError('Unsupported public source inventory')
     revision, evidence_kind = specs[source]
-    db = Path(db).resolve()
-    if not db.is_file():
-        raise ValueError('Existing database required for identity and status audit')
     conflicts = []
     seen = set()
     checked = 0
     existing_attempts = 0
     files = []
     checksums = {}
-    with closing(sqlite3.connect(db.as_uri() + '?mode=ro', uri=True)) as c:
+    with closing(open_read_connection(db, pg_service=pg_service, pg_schema=pg_schema)) as c:
-        c.execute('BEGIN')
+        if pg_service:
+            identity = c.execute('SELECT current_database(),current_schema()').fetchone()
+            database, actual_schema = identity[0], identity[1]
+        else:
+            database, actual_schema = str(Path(db).resolve()), None
         for manifest in result['files']:
             path = Path(manifest['file'])
             if hashlib.sha256(path.read_bytes()).hexdigest() != manifest['file_sha256']:
@@ -62,12 +63,11 @@ def audit(db, inventory, out):
                     raise ValueError('Multiple canonical attempts for one forum')
                 if existing:
                     existing_attempts += 1
-                    latest = c.execute('''SELECT status FROM decisions WHERE submission_id=?
+                    latest = latest_status(c, existing[0][0])
-                        ORDER BY observed_at DESC,COALESCE(decision_date,'') DESC,id DESC LIMIT 1''', existing[0]).fetchone()
+                    if latest and latest not in ('unknown', 'under_review', row['status']):
-                    if latest and latest[0] not in ('unknown', 'under_review', row['status']):
                         conflicts.append({'forum': forum, 'venue': row['venue'], 'year': row['year'],
                                           'track': row['track'], 'submission_id': existing[0][0],
-                                          'existing_status': latest[0], 'curated_status': row['status']})
+                                          'existing_status': latest, 'curated_status': row['status']})
                 count += 1
             if count != manifest['count']:
                 raise ValueError('Collected record count differs from inventory')
@@ -76,7 +76,8 @@ def audit(db, inventory, out):
             checksums[str(path)] = manifest['file_sha256']
     if checked != result['normalized_records']:
         raise ValueError('Normalized inventory total differs from checked records')
-    report = {'source': source, 'revision': revision, 'database': str(db),
+    report = {'source': source, 'revision': revision, 'database': database,
+              'engine': 'postgresql' if pg_service else 'sqlite', 'schema': actual_schema, 'pg_service': pg_service,
               'inventory': str(inventory), 'checked_at': timestamp(), 'records_checked': checked,
               'existing_same_edition_attempts': existing_attempts, 'conflicts': conflicts,
               'status': 'blocked' if conflicts else 'ready_for_serial_import',
@@ -94,11 +95,11 @@ def audit(db, inventory, out):
 
 def main(argv=None, *, prog=None):
     parser = argparse.ArgumentParser(prog=prog)
-    parser.add_argument('--db', required=True)
+    database_arguments(parser)
     parser.add_argument('--inventory', required=True)
     parser.add_argument('--out', required=True)
     args = parser.parse_args(argv)
-    report = audit(args.db, args.inventory, args.out)
+    report = audit(args.db, args.inventory, args.out, pg_service=args.pg_service, pg_schema=args.pg_schema)
     if report['status'] == 'blocked':
         raise SystemExit('Conflicting existing decisions require review before import')
 

paper_atlas/collectors/public_openreview.py

@@ -27,7 +27,7 @@ RAW_ROOT = 'https://raw.githubusercontent.com/papercopilot/openreview/' + REVISI
 TREE_URL = 'https://api.github.com/repos/papercopilot/openreview/git/trees/' + REVISION + '?recursive=1'
 NEGATIVE = {'reject': 'rejected', 'rejected': 'rejected',
             'desk reject': 'desk_rejected', 'desk rejected': 'desk_rejected',
-            'withdrawn': 'withdrawn', 'withdrawal': 'withdrawn'}
+            'withdraw': 'withdrawn', 'withdrawn': 'withdrawn', 'withdrawal': 'withdrawn'}
 TRACKS = {'main': 'main', 'Position': 'position',
           'Datasets & Benchmarks': 'datasets_benchmarks'}
 WORKSHOP_REJECT_LABELS = {"aim-fm workshop @ neurips'24 reject",
@@ -90,7 +90,9 @@ def normalize(row, family, year, provenance, include_workshops=False):
         return None, 'unmapped_venue'
     title = row.get('title')
     abstract = row.get('abstract')
-    if not isinstance(title, str) or not title.strip():
+    if title is None or (isinstance(title, str) and not title.strip()):
+        return None, 'missing_title'
+    if not isinstance(title, str):
         raise ValueError('Public submission lacks a title')
     if abstract is not None and not isinstance(abstract, str):
         raise ValueError('Abstract must be text or null')
@@ -215,7 +217,7 @@ def _collect_files(out, selected, include_workshops, fetcher):
             name = f'public-openreview-{family}-{year}-{track}.jsonl'
             path = stage / name
             path.write_text(''.join(json.dumps(row, ensure_ascii=False) + '\n' for row in records))
-            manifest = {'source': SOURCE, 'parser_version': 1, 'venue': records[0]['venue'],
+            manifest = {'source': SOURCE, 'parser_version': 2, 'venue': records[0]['venue'],
                         'year': year, 'track': track, 'count': len(records), 'status': 'partial',
                         'file': str(out / name), 'file_sha256': hashlib.sha256(path.read_bytes()).hexdigest(),
                         'list_url': REPOSITORY + '/tree/' + REVISION, 'source_revision': REVISION,

paper_atlas/db/backend.py

@@ -19,6 +19,25 @@ def open_store(db=None, *, pg_service=None, pg_schema=None):
     return Store(db)
 
 
+def open_read_connection(db=None, *, pg_service=None, pg_schema=None):
+    if bool(db) == bool(pg_service) or (pg_schema and not pg_service):
+        raise ValueError('Choose exactly one of SQLite db or PostgreSQL service; schema requires PostgreSQL')
+    if pg_service:
+        from paper_atlas.postgres.client import ReadConnection
+        return ReadConnection(pg_service, pg_schema)
+    import sqlite3
+    path = Path(db).resolve()
+    if not path.is_file():
+        raise ValueError('Existing database required for read-only audit')
+    connection = sqlite3.connect(path.as_uri() + '?mode=ro', uri=True)
+    try:
+        connection.execute('BEGIN')
+    except BaseException:
+        connection.close()
+        raise
+    return connection
+
+
 def inserted_id(connection, sql, params=()):
     if getattr(connection, 'postgres', False):
         return connection.execute(sql + ' RETURNING id', params).fetchone()[0]

paper_atlas/db/store.py

@@ -5,7 +5,7 @@ import re
 import sqlite3
 import unicodedata
 from pathlib import Path
-from paper_atlas.db.submissions import attach_submission,migrate_sources,write_decision
+from paper_atlas.db.submissions import attach_submission,migrate_sources,write_decision,latest_status
 from paper_atlas.db.backend import require_sqlite_writable, inserted_id
 
 
@@ -142,7 +142,7 @@ class Store:
             tag=c.execute('SELECT id FROM tags WHERE namespace=? AND value=?',(namespace,value)).fetchone()[0]
             c.execute('INSERT INTO paper_tags VALUES(?,?,?,?) ON CONFLICT DO NOTHING',(paper,tag,record,evidence))
 
-    def decision(self,paper,record,venue,year,track,external_id,url,status,raw_value,aliases=None,identities=None,observed_at=None,decision_date=None,source_decision_id='',submitted_at=None):
+    def decision(self,paper,record,venue,year,track,external_id,url,status,raw_value,aliases=None,identities=None,observed_at=None,decision_date=None,source_decision_id='',submitted_at=None,guard_status=False):
         c = self.conn
         existing=c.execute('SELECT id FROM venues WHERE lower(acronym)=lower(?)',(venue,)).fetchone()
         if existing:venue_id=existing[0]
@@ -154,6 +154,10 @@ class Store:
         # A previous event in this row may already have merged the input paper.
         paper=c.execute('SELECT paper_id FROM source_records WHERE id=?',(record,)).fetchone()[0]
         submission=attach_submission(self,edition,paper,record,external_id,identities or [])
+        if guard_status and status in ('rejected','desk_rejected','withdrawn'):
+            current = latest_status(c, submission)
+            if current and current not in ('unknown','under_review',status):
+                raise ValueError('Curated decision conflicts with current canonical submission status: ' + current)
         if submitted_at:
             from paper_atlas.db.submissions import timestamp
             c.execute('UPDATE submissions SET submitted_at=COALESCE(submitted_at,?) WHERE id=?',(timestamp(submitted_at),submission))

paper_atlas/db/submissions.py

@@ -70,6 +70,12 @@ def next_id(c,table,audit,former):
     return max(c.execute('SELECT COALESCE(max(id),0) FROM '+table).fetchone()[0],c.execute('SELECT COALESCE(max('+former+'),0) FROM '+audit).fetchone()[0])+1
 
 
+def latest_status(c, submission):
+    row = c.execute('''SELECT status FROM decisions WHERE submission_id=?
+        ORDER BY observed_at DESC,COALESCE(decision_date,observed_at) DESC,id DESC LIMIT 1''', (submission,)).fetchone()
+    return row[0] if row else None
+
+
 def merge_submissions(c,target,other):
     if target==other:return
     editions=c.execute('SELECT id,edition_id FROM submissions WHERE id IN (?,?)',(target,other)).fetchall()

paper_atlas/ingestion/records.py

@@ -1,5 +1,6 @@
 """Import normalized conference JSONL files with coverage evidence."""
 import argparse
+import hashlib
 import json
 import time
 from pathlib import Path
@@ -7,6 +8,8 @@ from pathlib import Path
 from paper_atlas.db.backend import database_arguments, open_store, inserted_id
 from paper_atlas.db.submissions import identities_for_row,timestamp
 from paper_atlas.collectors.pmlr_tracks import require_track_manifest, require_track_record
+from paper_atlas.collectors.public_openreview import SOURCE as PUBLIC_SOURCE
+from paper_atlas.collectors.icml_public_archive import SOURCE as ARCHIVE_SOURCE
 
 
 def ingest_file(db,path, *, pg_service=None, pg_schema=None):
@@ -30,10 +33,12 @@ def ingest_file(db,path, *, pg_service=None, pg_schema=None):
     editions = set()
     try:
         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)})))
+        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()
-        with path.open() as stream:
+        digest = hashlib.sha256()
+        with path.open('rb') as stream:
             for line in stream:
+                digest.update(line)
                 row = json.loads(line)
                 if row['source'] != source_name:
                     raise ValueError('Source differs from manifest')
@@ -60,7 +65,7 @@ def ingest_file(db,path, *, pg_service=None, pg_schema=None):
                         status=event['status']
                         if events and status is None:raise ValueError('Explicit decision event requires a status')
                         if status is not None and status not in ('accepted','rejected','desk_rejected','withdrawn','under_review','unknown'):raise ValueError('Unsupported decision event')
-                        edition=store.decision(paper,record,row['venue'],row['year'],row['track'],external,event.get('evidence_url',row['url']),status,event.get('raw_value',''),row.get('venue_aliases'),identities=identities,observed_at=event.get('observed_at') or row_observed,decision_date=event.get('decision_date'),source_decision_id=event.get('source_decision_id',''),submitted_at=row.get('submitted_at'))
+                        edition=store.decision(paper,record,row['venue'],row['year'],row['track'],external,event.get('evidence_url',row['url']),status,event.get('raw_value',''),row.get('venue_aliases'),identities=identities,observed_at=event.get('observed_at') or row_observed,decision_date=event.get('decision_date'),source_decision_id=event.get('source_decision_id',''),submitted_at=row.get('submitted_at'),guard_status=source_name in (PUBLIC_SOURCE,ARCHIVE_SOURCE))
                     editions.add(edition)
                 else:
                     raise ValueError('This importer requires an explicit supported decision status')
@@ -68,6 +73,8 @@ def ingest_file(db,path, *, pg_service=None, pg_schema=None):
                 missing_abstracts += not bool((row.get('abstract') or '').strip())
         if not count:
             raise ValueError('Empty metadata file')
+        if manifest.get('file_sha256') and digest.hexdigest() != manifest['file_sha256']:
+            raise ValueError('Collected file hash differs from manifest')
         if len(editions) != 1:
             raise ValueError('A manifest must describe exactly one conference edition and track')
         edition = next(iter(editions))
@@ -76,7 +83,7 @@ def ingest_file(db,path, *, pg_service=None, pg_schema=None):
             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
         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=? WHERE id=?",(timestamp(),count,count,run))
+        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_postgres_ingestion.py

@@ -11,7 +11,7 @@ from unittest.mock import patch
 import uuid
 import zipfile
 
-from paper_atlas.db.backend import open_store
+from paper_atlas.db.backend import open_store, open_read_connection
 from paper_atlas.db.store import Store
 from paper_atlas.ingestion import arxiv
 from paper_atlas.ingestion.backfill import backfill
@@ -27,6 +27,13 @@ class BackendSelectionTests(unittest.TestCase):
             with self.assertRaisesRegex(ValueError, 'Choose exactly one'):
                 open_store(**kwargs)
 
+    def test_readonly_audit_does_not_create_an_absent_sqlite_database(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / 'absent.sqlite'
+            with self.assertRaisesRegex(ValueError, 'Existing database'):
+                open_read_connection(path)
+            self.assertFalse(path.exists())
+
 
 @unittest.skipUnless(os.environ.get('PAPER_ATLAS_TEST_PG_SERVICE'), 'Requires the approved remote PostgreSQL test service')
 class PostgreSQLIngestionTests(unittest.TestCase):
@@ -138,6 +145,21 @@ class PostgreSQLIngestionTests(unittest.TestCase):
         finally:
             store.conn.close()
 
+    def test_changed_file_hash_rolls_back_data_before_commit(self):
+        path = self.file('changed', [self.row()])
+        manifest = json.loads(path.with_suffix('.manifest.json').read_text())
+        manifest['file_sha256'] = '0' * 64
+        path.with_suffix('.manifest.json').write_text(json.dumps(manifest))
+        with self.assertRaisesRegex(ValueError, 'hash differs'):
+            self.ingest(path)
+        store = self.store()
+        try:
+            self.assertEqual(store.conn.execute('SELECT count(*) FROM papers').fetchone()[0], 1)
+            self.assertEqual(store.conn.execute('SELECT count(*) FROM source_records').fetchone()[0], 1)
+            self.assertEqual(store.conn.execute('SELECT status,records_written FROM ingest_runs').fetchone(), ('failed', 0))
+        finally:
+            store.conn.close()
+
     def test_writer_lock_survives_commit_and_rollback_but_readers_continue(self):
         first = self.store()
         try:
@@ -165,6 +187,62 @@ class PostgreSQLIngestionTests(unittest.TestCase):
         with connect(self.service) as pg:
             self.assertEqual(pg.execute('SELECT count(*) FROM information_schema.tables WHERE table_schema=%s', (staging,)).fetchone()[0], 0)
 
+    def test_negative_preflight_uses_current_pg_attempt_and_blocks_stale_import(self):
+        import hashlib
+        from paper_atlas.collectors.public_openreview import SOURCE, REVISION
+        from paper_atlas.audits.public_openreview import audit
+        row = self.row(SOURCE, 'negativeForum')
+        row['identifiers']['openreview'] = 'negativeForum'
+        row['metadata'] = {'evidence_kind': 'curated_public_status_assertion',
+                           'source_revision': REVISION, 'original_decision_signature_verified': False}
+        path = self.file('negative', [row])
+        manifest = json.loads(path.with_suffix('.manifest.json').read_text())
+        manifest.update(file=str(path), file_sha256=hashlib.sha256(path.read_bytes()).hexdigest())
+        path.with_suffix('.manifest.json').write_text(json.dumps(manifest))
+        inventory = self.root / 'negative-inventory.json'
+        inventory.write_text(json.dumps({'source': SOURCE, 'revision': REVISION,
+                                        'normalized_records': 1, 'files': [manifest]}))
+        # An acceptance in another conference does not block this rejection.
+        self.ingest(self.file('other-conference', [self.row('official', 'acceptedElsewhere', 'accepted', 2026, 'COLM')]))
+        def preflight(name):
+            with redirect_stdout(io.StringIO()):
+                return audit(None, inventory, self.root / (name + '.json'), pg_service=self.service, pg_schema=self.schema)
+        first = preflight('before')
+        self.assertEqual(first['status'], 'ready_for_serial_import')
+        self.assertEqual(first['engine'], 'postgresql')
+        self.assertEqual(first['schema'], self.schema)
+        self.ingest(path)
+        self.assertEqual(preflight('existing')['existing_same_edition_attempts'], 1)
+        store = self.store()
+        try:
+            paper, record = store.conn.execute('''SELECT r.paper_id,r.id FROM source_records r
+                JOIN sources s ON s.id=r.source_id WHERE s.name=? AND r.external_id=?''', (SOURCE, row['external_id'])).fetchone()
+            store.decision(paper, record, 'ICLR', 2025, 'main', row['external_id'],
+                           'https://example.org/formal', 'accepted', 'Accept', observed_at='2027-01-01T00:00:00Z')
+            # Equal observation time, but an explicitly earlier decision date.
+            # NULL dates must fall back to observed_at, not the empty string.
+            store.decision(paper, record, 'ICLR', 2025, 'main', row['external_id'],
+                           'https://example.org/older', 'rejected', 'Reject', observed_at='2027-01-01T00:00:00Z',
+                           decision_date='2026-12-31T00:00:00Z')
+            store.conn.commit()
+        finally:
+            store.conn.close()
+        blocked = preflight('blocked')
+        self.assertEqual(blocked['status'], 'blocked')
+        self.assertEqual(blocked['conflicts'][0]['existing_status'], 'accepted')
+        self.assertEqual(blocked['files'], [])
+        # The earlier ready report cannot authorize overwriting current status.
+        with self.assertRaisesRegex(ValueError, 'conflicts with current'):
+            self.ingest(path)
+        reader = open_read_connection(pg_service=self.service, pg_schema=self.schema)
+        try:
+            self.assertEqual(reader.execute('SHOW transaction_read_only').fetchone()[0], 'on')
+            from paper_atlas.db.submissions import latest_status
+            sid = reader.execute("SELECT submission_id FROM submission_identifiers WHERE scheme='openreview' AND value=?", ('negativeForum',)).fetchone()[0]
+            self.assertEqual(latest_status(reader, sid), 'accepted')
+        finally:
+            reader.close()
+
     def test_backfill_merges_evidence_without_reimporting_or_losing_history(self):
         files = []
         for source, external, title in [('official', 'legacy', 'First bibliographic name'), ('openreview_snapshot', 'forum', 'Renamed submission')]:

tests/test_public_openreview.py

@@ -45,10 +45,12 @@ class PublicOpenReviewTests(unittest.TestCase):
 
     def test_only_explicit_decisions_and_verified_scope_become_events(self):
         for value in ('Pending', 'Active', 'Submitted to NeurIPS 2025', 'TBD', '',
-                      'Low score', 'ICML 2025 Retracted by Authors', 'Do not reject'):
+                      'Low score', 'ICML 2025 Retracted by Authors', 'Do not reject',
+                      'Workshop', 'Poster Workshop', 'Oral Workshop'):
             self.assertIsNone(negative_status(value), value)
         self.assertEqual(negative_status('Reject'), 'rejected')
         self.assertEqual(negative_status('Desk rejected'), 'desk_rejected')
+        self.assertEqual(negative_status('Withdraw'), 'withdrawn')
         self.assertEqual(negative_status('NeurIPS 2023 Conference Withdrawn Submission'), 'withdrawn')
         for raw, expected in [('main', 'main'), ('Position', 'position'),
                               ('Datasets & Benchmarks', 'datasets_benchmarks')]:
@@ -98,6 +100,46 @@ class PublicOpenReviewTests(unittest.TestCase):
                 self.assertEqual(list(Path(tmp).glob('*.jsonl')), [])
                 self.assertFalse((Path(tmp) / 'public-openreview-inventory.json').exists())
 
+    def test_missing_title_is_held_without_losing_valid_negative_records(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            result = self.collect(tmp, {'venues/iclr/iclr2017.jsonl': [
+                self.row(id='SJUdkecgx', title=''),
+                self.row(id='NoTitle001', title=None),
+                self.row(id='Whitespace', title='  '),
+                self.row(id='ValidReject'),
+                self.row(id='Withdraw01', status='Withdraw'),
+                self.row(id='Workshop01', status='Workshop'),
+            ]}, families=['iclr'], years=[2017])
+            self.assertEqual(result['input_rows'], 6)
+            self.assertEqual(result['normalized_records'], 2)
+            self.assertEqual(result['source_files'][0]['excluded_or_held'],
+                             {'missing_title': 3, 'not_explicit_negative': 1})
+            held_path = Path(result['held_file'])
+            self.assertEqual(hashlib.sha256(held_path.read_bytes()).hexdigest(),
+                             result['held_file_sha256'])
+            held = [json.loads(line) for line in held_path.read_text().splitlines()]
+            self.assertEqual([row['id'] for row in held], ['SJUdkecgx', 'NoTitle001', 'Whitespace'])
+            self.assertTrue(all(row['reason'] == 'missing_title' and row['status'] == 'Reject'
+                                for row in held))
+            self.assertEqual([row['provenance']['source_line'] for row in held], [1, 2, 3])
+            self.assertNotIn('private@example.org', held_path.read_text())
+            self.assertNotIn('private review', held_path.read_text())
+            manifest = result['files'][0]
+            self.assertEqual((manifest['parser_version'], manifest['status']), (2, 'partial'))
+            rows = [json.loads(line) for line in Path(manifest['file']).read_text().splitlines()]
+            self.assertEqual([(row['identifiers']['openreview'], row['status']) for row in rows],
+                             [('ValidReject', 'rejected'), ('Withdraw01', 'withdrawn')])
+            for row in held:
+                self.assertEqual(row['provenance']['source_revision'], REVISION)
+
+    def test_nontext_title_still_fails_collection(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            with self.assertRaisesRegex(ValueError, 'lacks a title'):
+                self.collect(tmp, {'venues/iclr/iclr2017.jsonl': [self.row(title=['invalid'])]},
+                             families=['iclr'], years=[2017])
+            self.assertFalse((Path(tmp) / 'public-openreview-inventory.json').exists())
+            self.assertEqual(list(Path(tmp).glob('*.jsonl')), [])
+
     def test_mirror_reimport_preserves_one_attempt_and_later_acceptance_elsewhere(self):
         with tempfile.TemporaryDirectory() as tmp:
             root = Path(tmp)