Scope incremental parents to events and verify integrated write safeguards

README.md

@@ -90,3 +90,5 @@ CI 在已授权的远程 runner 上运行测试、构建 wheel，并从源码目
 历史联合会议的既有届次归属可通过 [`maintain joint-editions`](docs/joint-editions.md) 先只读规划，再按计划哈希执行；保留论文与投稿决定来源，不重导入。
 同年多届的 `event_key`、显式迁移、旧文件刷新与查询边界见 [会议事件身份](docs/event-identity.md)。
 研究类型与自定义标签可通过 [`annotate`](docs/annotations.md) 显式初始化、添加和撤回；两者独立查询，保留证据及完整更正历史，不从会议轨道推断研究类型。
+
+整合后的运行版本依赖、交互约束和生产切换顺序见 [运行版本整合](docs/research-runtime.md)。

docs/event-identity.md

@@ -17,7 +17,7 @@ PGSERVICEFILE=/etc/paper-atlas/pg_service.conf paper-atlas maintain event-identi
   --pg-service foundry-research --pg-schema research --out /data/paper-index/event-plan.json
 ```
 
-检查计划的数据库身份、原 edition IDs、业务表数量、目标来源与历史指纹。精确计数在同一快照中按主键分页，每页最多5000行，只读取主键；不使用估计行数。CLI向stderr持续报告计数进度，避免大库长时间无输出。已有未标记 WWW1995 edition 必须逐投稿都有固定 DBLP Fall 卷证据，才会只为原 edition 填入 `www-1995-fall`；缺证或混合卷会拒绝计划。本文命令不表示生产已经迁移。
+检查计划的数据库身份、原 edition IDs、业务表数量、目标来源与历史指纹。计数也包含已显式启用的可选注解表。精确计数在同一快照中按主键分页，每页最多5000行，只读取主键；不使用估计行数。CLI向stderr持续报告计数进度，避免大库长时间无输出。已有未标记 WWW1995 edition 必须逐投稿都有固定 DBLP Fall 卷证据，才会只为原 edition 填入 `www-1995-fall`；缺证或混合卷会拒绝计划。本文命令不表示生产已经迁移。
 
 应用时传入同一文件及命令输出的文件 SHA256：
 

docs/research-runtime.md

@@ -0,0 +1,44 @@
+# Integrated research runtime
+
+This branch composes the independently reviewed identity guards (PR25), incremental
+coverage (PR27), joint-edition refresh guard (PR29), event identity (PR30), and optional
+annotation support (PR28). It starts from main 1b81d1a and preserves their commits;
+it is an explicit integration dependency, not evidence that those PRs have merged.
+Users merge through Gitea manually. Review this combined diff against main, including
+its conflict resolutions, before using it as the maintained runtime.
+
+The shared Store preserves stable source/identifier anchors before title matching,
+rejects ambiguous unanchored matches, resolves pinned IJCAI/ECAI volumes before event
+lookup, and selects same-year events by event_key. Incremental manifests for a named
+or ambiguous event must state event_key; their parent count and optimistic update
+are checked against that exact event, not the first venue/year/track match.
+
+Query filters compose event_key, status, tags and research type. An earlier rejected
+attempt and a later known acceptance remain distinct; neither annotations nor track
+names infer a research type. Installing this code does not initialize annotation_events.
+Event plans count that optional table when present, alongside the original business
+rows. Annotation initialization and event identity upgrade remain explicit commands
+with their own transactional checks. Neither command imports historical records.
+
+## Production sequencing
+
+1. Verify the exact integrated commit on the approved ces-control runner, including
+   the real PostgreSQL interaction tests and wheel/installed CLI checks.
+2. Stage that exact source in a separate release directory and verify package hashes;
+   leave the current runtime and frozen SQLite backup intact.
+3. Inspect live writers and coordinate through the existing database writer lock.
+   Generate a fresh read-only event plan using that integrated release. Check its
+   database/schema identity, all business counts and preserved history hashes. A plan
+   from before further data or optional-schema changes must not be reused.
+4. Apply only that reviewed file with its actual SHA256, in one transaction. Independently
+   verify the constraint, original edition IDs and fields, assigned Fall source, source
+   and history bytes, ordinary queries, full-text queries and foundry health.
+5. Use the verified integrated release for subsequent writes. Older writers lacking
+   event identity support must not resume against the upgraded schema. Annotation
+   initialization requires a separate explicit operation and acceptance.
+
+These are operational requirements, not claims of completed production deployment.
+The production event-key migration and optional annotation activation have not occurred
+at preparation time. Existing migration/import/repair acceptance remains valid; do not
+rerun those jobs to install source code. Cross-feature tests use disposable fixtures
+and PostgreSQL schemas, never the production research tables.

paper_atlas/ingestion/event_identity.py

@@ -9,7 +9,7 @@ import sys
 
 from paper_atlas.db.backend import database_arguments, open_read_connection, require_sqlite_writable
 from paper_atlas.db.events import has_event_key, pinned_www_fall
-from paper_atlas.postgres.schema import TABLES, quote
+from paper_atlas.postgres.schema import TABLES, OPTIONAL_TABLES, quote
 
 MARKER = 'edition_event_identity_v1:'
 
@@ -27,7 +27,11 @@ def exact_counts(c, *, page_size=5000, progress=None):
     if not isinstance(page_size, int) or page_size < 1:
         raise ValueError('Positive count page size required')
     result = {}
-    for table in TABLES:
+    if getattr(c, 'postgres', False):
+        optional = [table for table in OPTIONAL_TABLES if c.execute("SELECT 1 FROM information_schema.tables WHERE table_schema=current_schema() AND table_name=?", (table,)).fetchone()]
+    else:
+        optional = [table for table in OPTIONAL_TABLES if c.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (table,)).fetchone()]
+    for table in (*TABLES, *optional):
         if table == 'schema_migrations':
             continue
         if getattr(c, 'postgres', False):

paper_atlas/ingestion/records.py

@@ -42,9 +42,23 @@ def source_record_count(c, source, edition):
 
 
 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()
+    from paper_atlas.db.events import has_event_key, validate_event_key
+    key = validate_event_key(manifest.get('event_key'))
+    enabled = has_event_key(c)
+    if key and not enabled:
+        raise ValueError('Event identity migration required for incremental coverage')
+    field = 'e.event_key' if enabled else "''"
+    rows = c.execute('SELECT e.id,'+field+""" 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'])).fetchall()
+    if key is None:
+        if len(rows) > 1 or any(r[1] for r in rows) or (manifest['venue'] == 'WWW' and manifest['year'] in (1994,1995)):
+            raise ValueError('Explicit manifest event_key required for incremental coverage')
+    else:
+        rows = [r for r in rows if r[1] == key]
+    if len(rows) > 1:
+        raise ValueError('Ambiguous incremental coverage event')
+    edition = rows[0] if rows else None
     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 []

tests/test_research_runtime.py

@@ -0,0 +1,154 @@
+"""Interactions between event identity and previously independent write safeguards."""
+from contextlib import contextmanager, redirect_stdout
+import hashlib
+import io
+import json
+import os
+from pathlib import Path
+import tempfile
+import unittest
+import uuid
+
+from paper_atlas import querying
+from paper_atlas.collectors.dblp import XML_SHA
+from paper_atlas.curation.annotations import initialize, assert_annotation, retract_annotation, observations
+from paper_atlas.db.backend import open_store
+from paper_atlas.ingestion.event_identity import plan, upgrade
+from paper_atlas.ingestion.records import ingest_file
+from paper_atlas.postgres.client import connect
+from paper_atlas.postgres.migration import migrate
+from paper_atlas.postgres.schema import quote
+from tests.test_event_identity import legacy, add
+
+
+class RuntimeCases:
+    @contextmanager
+    def store(self):
+        s = open_store(**self.target)
+        try:
+            yield s
+        finally:
+            s.conn.close()
+
+    def query(self, *args):
+        target = (['--db', str(self.target['db'])] if 'db' in self.target else
+                  ['--pg-service', self.target['pg_service'], '--pg-schema', self.target['pg_schema']])
+        with redirect_stdout(io.StringIO()) as out:
+            querying.main([*target, *map(str,args)])
+        return json.loads(out.getvalue())
+
+    def assertion(self, c, paper, request, **changes):
+        args = dict(paper=paper, kind='research_type', namespace='custom:method', value='empirical',
+                    request_id=request, evidence_key='review:'+request, actor='reviewer', reason='Reviewed metadata')
+        args.update(changes)
+        return assert_annotation(c, **args)
+
+    def test_event_upgrade_preserves_annotation_history_and_combined_queries(self):
+        with self.store() as s:
+            initialize(s.conn)
+            event = self.assertion(s.conn,1,'first')
+            retract_annotation(s.conn,event,request_id='retract',actor='reviewer',reason='Corrected')
+            self.assertion(s.conn,1,'second')
+            original = observations(s.conn,1,history=True)
+            s.conn.commit()
+        proposal = plan(**self.target)
+        self.assertEqual(proposal['before']['counts']['annotation_events'],3)
+        result = upgrade(proposal, **self.target)
+        self.assertEqual(result['after']['counts'],proposal['before']['counts'])
+        with self.store() as s:
+            self.assertEqual(observations(s.conn,1,history=True),original)
+            paper,_,_ = add(s,'spring','www-1995-spring','rejected')
+            self.assertEqual(paper,1)
+            s.conn.commit()
+        row = self.query('--venue','WWW','--event-key','www-1995-spring','--status','rejected',
+                         '--research-type','custom:method=empirical','--paper',1,'--annotation-history')[0]
+        self.assertEqual([a['event_key'] for a in row['known_acceptances']],['www-1995-fall'])
+        self.assertEqual({a['latest_status'] for a in row['submissions']},{'accepted','rejected'})
+        self.assertEqual(len(row['annotation_history']),3)
+        self.assertEqual(self.query('--event-key','www-1995-fall','--status','rejected'),[])
+
+    def collected(self, name, event, external, *, incremental=False, unscoped=False):
+        row = dict(source='fixture',external_id=external,url='https://example.org/'+external,
+                   title='Independent contribution '+external,authors=[],abstract='Known abstract',
+                   identifiers={'doi':'10.1234/'+external},venue='WWW',year=1995,track='main',
+                   event_key=event,status='accepted')
+        p = self.root/(name+'.jsonl');p.write_text(json.dumps(row)+'\n')
+        manifest = {k:row[k] for k in ('source','venue','year','track','event_key')}
+        manifest.update(list_url='https://example.org/'+event,count=1,status='partial',
+                        file_sha256=hashlib.sha256(p.read_bytes()).hexdigest())
+        if incremental:
+            manifest.update(coverage_mode='incremental',coverage_previous_count=1,
+                            coverage_previous_expected_count=1,coverage_expected_count=2)
+        if unscoped:
+            manifest.pop('event_key')
+        p.with_suffix('.manifest.json').write_text(json.dumps(manifest))
+        return p
+
+    def ingest(self, path):
+        with redirect_stdout(io.StringIO()):
+            ingest_file(self.target.get('db'),path,**{k:v for k,v in self.target.items() if k != 'db'})
+
+    def test_incremental_parent_is_scoped_to_exact_event(self):
+        upgrade(plan(**self.target), **self.target)
+        self.ingest(self.collected('spring','www-1995-spring','spring-one'))
+        self.ingest(self.collected('fall','www-1995-fall','fall-one'))
+        self.ingest(self.collected('append','www-1995-fall','fall-two',incremental=True))
+        with self.store() as s:
+            rows = s.conn.execute('''SELECT e.event_key,es.expected_count,es.imported_count FROM edition_sources es
+                JOIN editions e ON e.id=es.edition_id JOIN sources src ON src.id=es.source_id
+                WHERE src.name='fixture' ORDER BY e.event_key''').fetchall()
+            self.assertEqual([tuple(r) for r in rows],[('www-1995-fall',2,2),('www-1995-spring',1,1)])
+        with self.assertRaisesRegex(ValueError,'event_key'):
+            self.ingest(self.collected('unscoped','www-1995-spring','unwanted',incremental=True,unscoped=True))
+        with self.assertRaisesRegex(ValueError,'parent changed'):
+            self.ingest(self.collected('stale','www-1995-fall','stale',incremental=True))
+        with self.store() as s:
+            self.assertEqual(s.conn.execute("SELECT count(*) FROM source_records r JOIN sources src ON src.id=r.source_id WHERE src.name='fixture'").fetchone()[0],3)
+
+    def test_joint_refresh_keeps_identity_annotations_and_event_query(self):
+        upgrade(plan(**self.target), **self.target)
+        metadata = dict(snapshot_sha256=XML_SHA,proceedings_key='conf/ijcai/2022',
+                        bibliographic_year=2022,bibliography={'booktitle':['IJCAI']})
+        with self.store() as s:
+            initialize(s.conn)
+            paper,record = s.record('dblp','conf/ijcai/Joint22','https://example.org/joint',
+                                   'Joint paper',None,[],{'doi':'10.1234/joint'},metadata)
+            eid = s.decision(paper,record,'IJCAI',2022,'main','conf/ijcai/Joint22','https://example.org/joint','accepted','Accept')
+            self.assertion(s.conn,paper,'joint')
+            s.conn.commit()
+        with self.store() as s:
+            again,r = s.record('dblp','conf/ijcai/Joint22','https://example.org/joint',
+                               'Corrected joint title',None,[],{'doi':'10.1234/joint'},metadata)
+            self.assertEqual(again,paper)
+            self.assertEqual(s.decision(again,r,'IJCAI',2022,'main','conf/ijcai/Joint22','https://example.org/joint','accepted','Accept'),eid)
+            self.assertEqual(s.conn.execute("SELECT count(*) FROM editions e JOIN venues v ON v.id=e.venue_id WHERE v.acronym='IJCAI' AND e.year=2022").fetchone()[0],0)
+            s.conn.commit()
+        for venue in ['IJCAI','ECAI']:
+            rows = self.query('--venue',venue,'--year',2022,'--event-key','',
+                              '--research-type','custom:method=empirical','--paper',paper)
+            self.assertEqual(len(rows),1)
+            self.assertEqual(len(rows[0]['submissions']),1)
+            self.assertEqual(rows[0]['submissions'][0]['acronym'],'ECAI+IJCAI')
+
+
+class SQLiteRuntimeTests(RuntimeCases, unittest.TestCase):
+    def setUp(self):
+        t=tempfile.TemporaryDirectory();self.addCleanup(t.cleanup);self.root=Path(t.name)
+        db=self.root/'source.sqlite';legacy(db);self.target={'db':db}
+
+
+@unittest.skipUnless(os.environ.get('PAPER_ATLAS_TEST_PG_SERVICE'),'Requires approved remote PostgreSQL service')
+class PostgreSQLRuntimeTests(RuntimeCases, unittest.TestCase):
+    def setUp(self):
+        t=tempfile.TemporaryDirectory();self.addCleanup(t.cleanup);self.root=Path(t.name)
+        db=self.root/'source.sqlite';legacy(db)
+        self.service=os.environ['PAPER_ATLAS_TEST_PG_SERVICE'];self.schema='pa_runtime_'+uuid.uuid4().hex[:12]
+        self.addCleanup(self.cleanup)
+        with redirect_stdout(io.StringIO()):
+            migrate(db,self.service,self.root/'migration.json',schema=self.schema,min_free_gib=0)
+        self.target={'pg_service':self.service,'pg_schema':self.schema}
+
+    def cleanup(self):
+        with connect(self.service) as c:
+            for name, in c.execute('SELECT nspname FROM pg_namespace WHERE starts_with(nspname,%s)',(self.schema,)):
+                c.execute('DROP SCHEMA '+quote(name)+' CASCADE')