From 4182070e76dfa13ef7dbb526a0b6a4d155f19cd6 Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Sun, 8 Nov 2026 21:04:05 +0800 Subject: [PATCH] Scope incremental parents to events and verify integrated write safeguards --- README.md | 2 + docs/event-identity.md | 2 +- docs/research-runtime.md | 44 +++++++ paper_atlas/ingestion/event_identity.py | 8 +- paper_atlas/ingestion/records.py | 20 ++- tests/test_research_runtime.py | 154 ++++++++++++++++++++++++ 6 files changed, 224 insertions(+), 6 deletions(-) create mode 100644 docs/research-runtime.md create mode 100644 tests/test_research_runtime.py diff --git a/README.md b/README.md index fc4b55d..dd61ba5 100644 --- a/README.md +++ b/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)。 diff --git a/docs/event-identity.md b/docs/event-identity.md index 7cf6eba..8f65575 100644 --- a/docs/event-identity.md +++ b/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： diff --git a/docs/research-runtime.md b/docs/research-runtime.md new file mode 100644 index 0000000..16bfac2 --- /dev/null +++ b/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. diff --git a/paper_atlas/ingestion/event_identity.py b/paper_atlas/ingestion/event_identity.py index b5f4020..71e082d 100644 --- a/paper_atlas/ingestion/event_identity.py +++ b/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): diff --git a/paper_atlas/ingestion/records.py b/paper_atlas/ingestion/records.py index 50baa40..63898ca 100644 --- a/paper_atlas/ingestion/records.py +++ b/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 [] diff --git a/tests/test_research_runtime.py b/tests/test_research_runtime.py new file mode 100644 index 0000000..c451278 --- /dev/null +++ b/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')