Audit public decisions against PostgreSQL and recheck imports

Reviewed-on: #15

.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/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:
-        c.execute('BEGIN')
+    with closing(open_read_connection(db, pg_service=pg_service, pg_schema=pg_schema)) as c:
+        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=?
-                        ORDER BY observed_at DESC,COALESCE(decision_date,'') DESC,id DESC LIMIT 1''', existing[0]).fetchone()
-                    if latest and latest[0] not in ('unknown', 'under_review', row['status']):
+                    latest = latest_status(c, existing[0][0])
+                    if latest and latest 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/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')]: