Expose independent event keys in integrated coverage reports

# Conflicts:
#	paper_atlas/cli.py
#	paper_atlas/curation/__init__.py

.gitea/workflows/check.yml

@@ -49,6 +49,8 @@ jobs:
               subprocess.run([str(venv/'bin/paper-atlas'),'--help'],cwd=outside,check=True)
               subprocess.run([python,'-I','-B','-m','paper_atlas','collect','cvf','--help'],cwd=outside,check=True)
               subprocess.run([python,'-I','-B','-m','paper_atlas','maintain','postgres','--help'],cwd=outside,check=True)
+              subprocess.run([python,'-I','-B','-m','paper_atlas','maintain','joint-editions','--help'],cwd=outside,check=True)
+              subprocess.run([python,'-I','-B','-m','paper_atlas','maintain','event-identity','--help'],cwd=outside,check=True)
               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)
@@ -60,5 +62,7 @@ jobs:
               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)
               subprocess.run([python,'-I','-B','-m','paper_atlas','query','--db','installed.sqlite','--stats'],cwd=outside,check=True)
+              subprocess.run([python,'-I','-B','-m','paper_atlas','annotate','init','--db','installed.sqlite'],cwd=outside,check=True)
+              subprocess.run([python,'-I','-B','-m','paper_atlas','annotate','list','--db','installed.sqlite','--paper','1'],cwd=outside,check=True)
               subprocess.run([python,'-I','-B','-m','paper_atlas','audit','histories','--db','installed.sqlite','--examples','0'],cwd=outside,check=True)
           PY

README.md

@@ -87,3 +87,9 @@ paper-atlas collect colm --cache /data/paper-index/raw/http-cache \
 CI 在已授权的远程 runner 上运行测试、构建 wheel，并从源码目录外验证安装后的命令及随包资源。数据库全量去重和覆盖率是独立的数据验收步骤，不因代码测试通过而自动完成。
 
 开发通过工作分支和 PR 交付，由用户手动合并。
+
+历史联合会议的既有届次归属可通过 [`maintain joint-editions`](docs/joint-editions.md) 先只读规划，再按计划哈希执行；保留论文与投稿决定来源，不重导入。
+同年多届的 `event_key`、显式迁移、旧文件刷新与查询边界见 [会议事件身份](docs/event-identity.md)。
+研究类型与自定义标签可通过 [`annotate`](docs/annotations.md) 显式初始化、添加和撤回；两者独立查询，保留证据及完整更正历史，不从会议轨道推断研究类型。
+
+整合后的运行版本依赖、交互约束和生产切换顺序见 [运行版本整合](docs/research-runtime.md)。

docs/annotations.md

@@ -0,0 +1,107 @@
+# Research types and custom annotations
+
+Research type describes a claimed characteristic of the work (for example a survey,
+benchmark, empirical study or replication). It is independent of topic, paper length,
+conference track, presentation format and acceptance status. No type is inferred from
+those fields. Multiple source-specific or human classifications may coexist; no
+cross-platform vocabulary mapping or automatic classifier is provided.
+
+`annotation_events` stores both kinds with an explicit `kind` discriminator:
+`research_type` and `tag` are separate domains, even when namespace/value strings
+happen to coincide. Existing platform labels remain in `tags` / `paper_tags`.
+Each assertion carries namespace, value, actor, reason, evidence key, optional source
+record and evidence URL, and the time it was recorded. This is observation time, not
+paper publication time or a conference decision date.
+
+Values are not forced into a global taxonomy: `custom:method=empirical` and
+`platform:declared_type=empirical` retain separate provenance. An explicit platform
+assertion API requires a source record belonging to that paper and matching the
+namespace prefix. The manual CLI only accepts `custom:*`, and cannot retract platform
+assertions. No source type has been automatically assigned to existing papers.
+
+## Initialize explicitly
+
+On the published PostgreSQL dataset, using its protected service file:
+
+```sh
+PGSERVICEFILE=/etc/paper-atlas/pg_service.conf paper-atlas annotate init \
+  --pg-service foundry-research --pg-schema research
+```
+
+This creates one optional table, its indexes, and an upgrade marker in one transaction,
+using the existing dataset writer lock. It does not copy data, change existing papers
+or decisions, or create a PostgreSQL instance. Repeating it is safe; an unexplained
+pre-existing table/marker mismatch fails closed. Readers and ordinary imports do not
+install the extension. Legacy datasets remain readable. A SQLite **sample** uses
+`--db sample.sqlite`; the cutover SQLite backup remains protected against writes.
+
+## Add and retract evidence
+
+Use the canonical paper ID returned by `query`. For a reviewed classification:
+
+```sh
+paper-atlas annotate assert --pg-service foundry-research --pg-schema research \
+  --paper "$PAPER_ID" --kind research_type --namespace custom:method \
+  --value empirical --evidence-key manual:abstract-review \
+  --actor reviewer --reason 'Classified from the explicitly described study design' \
+  --request-id "$REQUEST_ID"
+```
+
+`$REQUEST_ID` must be a new, stable identifier for this action (for example a UUID).
+Retry with the same request ID and identical content after an interrupted response;
+it returns the original event ID. Reusing that ID with changed content is rejected.
+A topic annotation uses `--kind tag --namespace custom:topic --value robotics` instead.
+Optional `--source-record` and `--evidence-url` attach additional evidence; the source
+record must belong to this paper. URLs are retained as evidence and are not fetched.
+
+To withdraw one particular assertion, use its returned event ID:
+
+```sh
+paper-atlas annotate retract --pg-service foundry-research --pg-schema research \
+  --event "$EVENT_ID" --actor reviewer --reason 'Classification corrected after review' \
+  --request-id "$RETRACTION_REQUEST_ID"
+```
+
+Retraction adds an event referencing that assertion, preserving its original content.
+It does not remove another contributor's assertion of the same label, platform tags,
+source metadata or submission/decision history. A retracted assertion can be asserted
+again with a new request ID and evidence; it is a new observation. The API does not
+update or delete past annotation events. Database administrators still control direct
+SQL access. Work reviewed as the same paper may subsequently be merged by the existing
+identity workflow: annotation IDs, original paper IDs, evidence and retraction links
+are retained while the canonical paper reference follows the surviving entity.
+
+## Query and analyze
+
+```sh
+paper-atlas query --pg-service foundry-research --pg-schema research \
+  --research-type custom:method=empirical
+paper-atlas query --pg-service foundry-research --pg-schema research \
+  --tag custom:topic=robotics
+paper-atlas query --pg-service foundry-research --pg-schema research \
+  --paper "$PAPER_ID" --annotation-history
+paper-atlas annotate list --pg-service foundry-research --pg-schema research \
+  --paper "$PAPER_ID" --history
+```
+
+`--tag` includes original platform labels and active annotation tags. It never matches
+research types. `--research-type` matches active type assertions only. Paper details
+return separate `tags` (existing provenance), `annotation_tags`, and `research_types`;
+`--annotation-history` additionally includes retractions. A missing extension yields
+empty annotation results rather than changing the database.
+
+For direct SQL analysis, active research-type evidence is:
+
+```sql
+SELECT a.* FROM research.annotation_events a
+WHERE a.kind = 'research_type' AND a.retracts_id IS NULL
+  AND NOT EXISTS (SELECT 1 FROM research.annotation_events r WHERE r.retracts_id = a.id);
+```
+
+Several independent assertions can support one label; count distinct papers/labels
+when appropriate rather than counting evidence rows as papers. Namespaces stay
+separate. Standard paper queries decode the PostgreSQL text codec; direct SQL consumers
+must honor the dataset's documented codec, including NUL/SOH values. SQLite-to-PostgreSQL
+migration recognizes an installed annotation extension and verifies its full row hashes,
+constraints and identity sequence along with the original business tables. An old
+21-table snapshot stays a 21-table migration; no synthetic annotations are generated.

docs/coverage/README.md

@@ -188,5 +188,3 @@ MM 2026 的[官方 CFP](https://2026.acmmm.org/site/cfp-guidelines.html)说明
 WWW早期逐来源核查见 `www-early-proceedings-2026-11-08.json`：1995春季论文三目录均为38个相同条目ID，海报摘要53个条目与作者链接47个条目不齐；1994春季50个Title字段与主页49场报告说法待逐项核对。上述为源目录观察，不是已导入或全部录用数量；主会、海报、教程等继续独立核验。
 
 WWW事件标识生产迁移已独立验收（报告SHA `f5dbb8a8`）：保留1399个原届次字段及IDs、59个Fall投稿和完整全库历史哈希。服务器已缓存46个官方HTML来源（约223KB），只作为后续维护式采集器的输入；尚未补导早期WWW论文，也未启用可选注解表。
-
-IJCAI 联办届次的逐条后续核查见 [2018/2022 分区与目录差集](ijcai-sections-2026-11-08.md) 和 [稳定 ID 映射](ijcai-sections-2026-11-08.json)。已定位 345 条非 main 分类及 2022 官方 HTML 缺失但官方文件仍存在的一条记录；生产分类尚未修正，不能把整卷数量作为 main 覆盖率。

docs/coverage/ijcai-sections-2026-11-08.md

@@ -1,40 +0,0 @@
-# IJCAI 2018、2022 逐条目录与 track 核查
-
-2026-11-08 的只读生产快照中，edition 867（2018）和 871（2022）已正确属于 IJCAI/ECAI 联办会议，但整本论文集的所有投稿仍归在 `main`。本核查用官方目录的逐条 DOI 和分区定位问题，不修改生产数据。
-
-| 年份 | 库中投稿 | 官方目录条目 | DOI 一对一对应 | 官方 Main track | 明确非 main | 当前目录缺项 |
-| --- | ---: | ---: | ---: | ---: | ---: | ---: |
-| 2018 | 870 | 870 | 870 | 709 | 161 | 0 |
-| 2022 | 863 | 862 | 862 | 678 | 184 | 1 |
-
-2018 官方分区为 Main track 709、Invited Speakers 3、Evolution of the contours of AI 6、Sister Conferences Best Papers 46、Survey track 16、Journal track 24、Early Career 19、Doctoral Consortium 17、Demos 30。
-
-2022 官方分区为 Main Track 678、Special Track on AI, the Arts and Creativity 17、Special Track on AI for Good 33、Sister Conferences Best Papers 26、Survey Track 38、Journal Track 19、Early Career 8、Doctoral Consortium 25、Demo Track 18。Special Track 保留独立范围，不因同在正式论文集就合进 main。
-
-## 862/863 差异已定位，不能删一条凑数
-
-数据库独有条目：`conf/ijcai/Fang22`，题名 **Fixed-Budget Pure Exploration in Multinomial Logit Bandits**，DOI `10.24963/ijcai.2022/409`，页码 2951–2957。submission 333599、source_record 3578445、paper 3442268。
-
-[当前官方总目录](https://ijcai.org/proceedings/2022/) 没有这一条，详情页 `/proceedings/2022/409` 本次 HTTP 404；但[官方论文文件](https://ijcai.org/proceedings/2022/0409.pdf) HEAD 返回 HTTP 200、application/pdf、153657 字节。浏览器读取的官方文件标题、作者 Boli Fang、IJCAI-22 页脚及首页 2951 与数据库对应，共七页。服务器没有下载保存 PDF 正文。
-
-因此这是已定位的“当前 HTML 目录缺项但官方文件仍存在”的来源差异。没有证据证明撤稿、拒稿或数据库多收了一篇；保留原条目与发表证据。其当前官方 track 仍待明确，不根据相邻编号擅自归 main。
-
-## 可复核证据
-
-同目录 JSON 保存两份官方 HTML 的 URL、HTTP 状态、时间、字节数和 SHA256，以及生产只读导出 SHA256。服务器快照目录 `/data/paper-index/collected/ijcai-section-audit-v1/` 保留 HTML、来源回执、原始数据库来源导出和 409 HTTP 回执。
-
-JSON 的 `years[].sections[].records` 每条为 `[submission_id, source_record_id, doi_suffix]`；DOI 为 `10.24963/ijcai.<year>/<doi_suffix>`。这些是现有稳定 ID，不是待插入的新投稿。全部 1732 个当前目录条目均对应唯一数据库来源 DOI，目录与数据库均无重复 DOI，官方目录独有条目为零。
-
-标题仅作辅助核验：NFKC、casefold、去除 Unicode 非 word 字符后，2018 全部一致；2022 有一个拼写差异，DOI `/552` 的 DBLP 标题将 Revisiting、Re-Construction 拼为 Revisting、Re-Constrcution。原始标题均保留，未因此创建或合并论文。
-
-## 未完成事项及修正边界
-
-本次交付只完成来源核查和逐条分区映射，**345 条非 main 的生产分类尚未修正**，第 14 步尚未完成。下一步需要可复用的、带快照前提和验收报告的分区修正流程：
-
-- 保持原 submission、source_record、paper、decision 的 ID 和证据内容；纠正 edition 归属，不制造“新一次投稿”。
-- 同步处理依赖 edition 的 submission_identifiers、覆盖记录与联合会议别名；不能只移动 submissions 一个字段。
-- 保存官方分区文字、URL、源文件 SHA 和原始分类，防止以后 DBLP 整卷刷新把精确 track 覆盖回 main。
-- 对 409 保持单独待核实状态，不推断撤稿、拒稿或当前 main 归属。
-- 修正后独立核验完整投稿/决定历史、稳定标识、各 track 行数及实际查询。研究方法类型不能由 track 自动生成。
-
-这是 2018、2022 两届的完整目录差集和分区证据，不代表其他年份已逐条验收，也不替代所有导入结束后的全库去重。

docs/event-identity.md

@@ -0,0 +1,44 @@
+# 同年多届会议的事件身份
+
+`venue`、`year`、`track` 描述系列、年份和投稿轨道；`event_key` 区分同系列、同年、同轨道的实际会议事件。例如 WWW 1995 的春季与秋季会议分别使用 `www-1995-spring` 和 `www-1995-fall`，两者仍是 `main`。相同论文可以在两届有不同投稿与决定；论文去重不会把这两次投稿合成一次。
+
+普通单届数据继续使用空键。新输入在同一组已有具名事件或已知同年多届时必须提供明确键；不能任选第一个 edition。已有来源刷新可依据自己的原投稿关系回到原事件，不能被显式改到另一事件。精确固定 DBLP 快照、WWW Fall 书目名、1995 年及 `www1995f.html` 卷共同支持旧秋季文件的归属；不按标题猜年份或季节。
+
+规范 JSONL 行及 manifest 都可提供 `event_key`，非空值限制为最多80个小写字母、数字及 `._:-`，首字符为字母或数字。manifest 提供键时，每行必须一致；一份文件仍仅对应一个事件及轨道。跨行冲突会回滚整份业务导入。身份回填同样按事件选择，不能借机把来源移到另一届。
+
+## 新样本与既有数据
+
+新建 SQLite 样本的 `editions` 唯一键为 `(venue_id, year, track, event_key)`。已有 SQLite/PostgreSQL 不会因普通读写命令自动升级。其普通单届导入继续可用；WWW1994/1995或带事件键的导入会要求先迁移。冻结的 SQLite 生产备份仍受 cutover 标记保护，不能写入或升级。
+
+先在 PostgreSQL 生成只读计划：
+
+```sh
+PGSERVICEFILE=/etc/paper-atlas/pg_service.conf paper-atlas maintain event-identity \
+  --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`；缺证或混合卷会拒绝计划。本文命令不表示生产已经迁移。
+
+应用时传入同一文件及命令输出的文件 SHA256：
+
+```sh
+PGSERVICEFILE=/etc/paper-atlas/pg_service.conf paper-atlas maintain event-identity \
+  --pg-service foundry-research --pg-schema research \
+  --apply-plan /data/paper-index/event-plan.json --plan-sha256 PLAN_FILE_SHA256 \
+  --out /data/paper-index/event-applied.json
+```
+
+原生 PostgreSQL writer 使用现有 schema advisory lock，并在事务中锁定 editions，核对计划后添加列及替换唯一约束。SQLite 样本使用独占写事务，重建父表时保留 IDs，并在提交前检查外键；不级联删除子记录。额外列、未知唯一约束或 SQLite 额外索引会拒绝自动升级，须单独评估。未改变原版四列的排序内容哈希、任何业务表数量或目标投稿/决定/来源内容，才可提交；失败回滚。迁移标记与 DDL 同事务。原计划立即重试为 `already_applied`；若后续业务已经变化，应重新生成计划，不重放旧计划。
+
+ECML 与 PKDD 的联合举办前事件已有不同 venue/edition，保持原身份即可。这项迁移不会把它们合并，也不会创建新的论文、投稿或决定。WWW 春季及1994两届的逐篇采集、来源核验和导入仍须另做。
+
+## 查询与验收
+
+```sh
+paper-atlas query --pg-service foundry-research --pg-schema research \
+  --venue WWW --year 1995 --track main --event-key www-1995-fall
+```
+
+详细投稿历史及 `known_acceptances` 返回 `event_key`。`--event-key` 可与 `--status` 一起限定同一事件中的最新已知决定。未迁移的旧库仍可普通查询，其事件键为空；不能声称已经区分季节。`audit histories` 的新库事件输出也保留键。
+
+远程测试应覆盖同一作品的两届独立投稿、拒稿后另一届录用、原 IDs 与 Unicode/NUL 来源内容保留、旧文件刷新、歧义及错误键拒绝、整文件回滚、明确事件的身份回填、旧计划失效、重试与真实 PostgreSQL 迁移。生产应用后还须独立核验约束、来源和历史、全文索引及 foundry-core 健康；单元测试不替代生产或最终全库去重验收。

docs/joint-editions.md

@@ -0,0 +1,67 @@
+# Repairing joint conference membership
+
+EURAI's [official history](https://www.eurai.org/ecai/history) identifies ECAI 2018
+and 2022 as joint IJCAI/ECAI conferences. The pinned DBLP volumes
+`conf/ijcai/2018` and `conf/ijcai/2022` use `IJCAI` as their booktitle and title.
+Existing records can therefore already be present under IJCAI while an ECAI query
+shows a missing year. Adding ECAI as a global alias to the ordinary IJCAI venue
+would incorrectly include all other IJCAI years.
+
+The existing canonical `ECAI+IJCAI` venue has both member aliases. This maintenance
+command changes only the venue reference of the two existing `main` edition rows.
+Edition IDs, submission IDs, decisions, source text, authors, identifiers, tags,
+source links and coverage records stay intact. It does not merge or create papers
+or submissions, resolve per-paper tracks, or turn reviews into negative decisions.
+The DBLP collector applies the joint mapping only to these two exact main volume
+keys; original booktitle/title remain in source metadata with the membership evidence.
+The shared ingestion entry point also resolves these exact pinned source volumes before
+creating an edition, so refreshing an older normalized file cannot recreate a separate
+IJCAI edition. Other sources, years, workshop volumes and additional tracks retain their existing rules.
+
+## Review a read-only plan
+
+Use a runtime containing this repair command. The configured PostgreSQL writer lock
+is used only for the apply operation; do not replace a newer ingestion runtime with
+an older main-based feature branch.
+
+```sh
+PGSERVICEFILE=/etc/paper-atlas/pg_service.conf paper-atlas maintain joint-editions \
+  --pg-service foundry-research --pg-schema research --out joint-plan.json
+```
+
+The preflight requires the existing joint venue and exactly its two member aliases,
+one nonempty unambiguous edition per year, and matching pinned DBLP volume evidence
+for every existing submission. A colliding target edition, missing provenance or
+unexpected source scope fails closed. The plan binds current database/schema, IDs,
+counts and full fingerprints of affected papers, identifiers, source records, tags,
+submissions, decisions and their provenance, and coverage rows. Observation time is
+recorded separately from conference year. These are scoped checks, not a final global
+identity or conference completeness audit.
+
+## Apply the exact reviewed plan
+
+Compute the file's SHA256 after review and pass it explicitly:
+
+```sh
+PGSERVICEFILE=/etc/paper-atlas/pg_service.conf paper-atlas maintain joint-editions \
+  --pg-service foundry-research --pg-schema research --plan joint-plan.json \
+  --plan-sha256 "$PLAN_SHA256" --out joint-applied.json
+```
+
+The plan hash is checked before opening a writer. Current evidence is checked again
+under the dataset writer lock; both edition updates and an audit marker commit in one
+transaction. Protected row hashes are checked again before commit. Changed evidence
+requires a fresh reviewed plan. Existing output reports are preserved; choose a new
+path for another run. SQLite samples use `--db sample.sqlite`; the frozen production
+backup remains protected.
+
+If a response is lost after commit, rerun with the same plan and a new output path.
+When protected evidence is unchanged and both editions already point to the joint
+venue, the result is `already_joint` with zero updates. It does not replay imports or
+regenerate source records. Keep the plan, apply report and independent postcheck
+alongside collection evidence. The audit marker is `joint_ijcai_ecai_2018_2022_v1`.
+
+Publication counts can differ between the DBLP volume and the current publisher
+page. This membership repair preserves all existing rows rather than deleting entries
+to force an aggregate match; section-level main/invited/other classification and
+publication completeness require separate per-paper evidence.

docs/research-runtime.md

@@ -0,0 +1,49 @@
+# 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.
+
+Coverage output also returns event_key for each independent edition ID. Two same-year
+main events keep their own status/source counts; venue paper totals remain distinct
+work counts rather than sums across events. Legacy datasets without the event column
+return an empty event key and remain read-only compatible.

paper_atlas/audits/coverage.py

@@ -45,6 +45,8 @@ def report(db, registry, *, pg_service=None, pg_schema=None):
             database, schema = identity[0], identity[1]
         else:
             database, schema = str(Path(db).resolve()), None
+        from paper_atlas.db.events import has_event_key
+        event = 'e.event_key' if has_event_key(connection) else "''"
         latest = _status_counts(connection, latest=True)
         observations = _status_counts(connection, latest=False)
         # Aggregate the paper join once for the whole snapshot. Repeating it
@@ -73,9 +75,9 @@ def report(db, registry, *, pg_service=None, pg_schema=None):
             params = (acronym, acronym)
             editions = []
             for row in connection.execute('''SELECT e.id,v.id AS venue_id,v.acronym AS canonical_venue,
-                    e.year,e.track FROM editions e JOIN venues v ON v.id=e.venue_id
+                    e.year,e.track,''' + event + ''' AS event_key FROM editions e JOIN venues v ON v.id=e.venue_id
                     WHERE ''' + scope + '''
-                    ORDER BY e.year,e.track,v.acronym''', params):
+                    ORDER BY e.year,e.track,event_key,v.acronym,e.id''', params):
                 edition = dict(row)
                 edition.update(edition_counts.get(row['id'], {
                     'submissions': 0, 'papers': 0, 'papers_without_abstract_text': 0}))

paper_atlas/audits/histories.py

@@ -124,7 +124,11 @@ def report(db=None, *, pg_service=None, pg_schema=None, examples=20):
                 sample_attempts.append(rows)
 
         current = []
-        for raw in connection.execute(ATTEMPTS):
+        from paper_atlas.db.events import has_event_key
+        attempts_sql = ATTEMPTS
+        if has_event_key(connection):
+            attempts_sql = attempts_sql.replace('e.year,e.track,', 'e.year,e.track,e.event_key,', 1).replace('e.year,e.track\n', 'e.year,e.track,e.event_key\n')
+        for raw in connection.execute(attempts_sql):
             row = dict(raw)
             if current and row['paper_id'] != current[0]['paper_id']:
                 consume(current)

paper_atlas/cli.py

@@ -6,6 +6,7 @@ from paper_atlas import __version__
 
 
 COMMANDS = {
+    ('annotate',): ('paper_atlas.curation.annotations', {}, '管理独立的研究类型与自定义标签证据'),
     ('curate','identifiers'): ('paper_atlas.curation.identifiers', {}, '按证据补充稳定标识并生成独立待导入文件'),
     ('query',): ('paper_atlas.querying', {}, '查询论文、标签和投稿历史'),
     ('coverage',): ('paper_atlas.audits.coverage', {}, '查看全部候选会议的覆盖情况'),
@@ -26,6 +27,8 @@ COMMANDS = {
     ('catalogue','dblp'): ('paper_atlas.collectors.dblp', {}, '盘点 DBLP 历史会议目录'),
     ('import','arxiv'): ('paper_atlas.ingestion.arxiv', {}, '导入 arXiv 元数据 ZIP'),
     ('import','records'): ('paper_atlas.ingestion.records', {}, '导入核验过的会议文件'),
+    ('maintain','joint-editions'): ('paper_atlas.ingestion.joint_editions', {}, '按已核验计划修正既有联合会议届次归属'),
+    ('maintain','event-identity'): ('paper_atlas.ingestion.event_identity', {}, '按核验计划增加同年会议事件身份'),
     ('maintain','merge'): ('paper_atlas.ingestion.maintenance', {}, '协调单写入合并与恢复'),
     ('maintain','backfill'): ('paper_atlas.ingestion.backfill', {}, '回填既有投稿的来源身份'),
     ('maintain','arxiv-tags'): ('paper_atlas.ingestion.arxiv_tags', {}, '恢复逐来源记录的 arXiv 类别证据'),

paper_atlas/collectors/dblp_records.py

@@ -88,6 +88,8 @@ def normalize_row(row,volumes,targets,include_workshops=False,through_year=2026)
               'scope_rule':edition['rule'],'author_names_raw':fields.get('author',[]),'publtype':row.get('publtype'),
               'bibliography':{key:value for key,value in fields.items() if key in ('booktitle','year','ee','url','pages','volume','number','publisher','series','isbn')},
               'ambiguous_identifiers':ambiguous,'publication_evidence':'inproceedings bibliography; not an original decision note'}
+    if edition.get('joint_edition_evidence'):
+        metadata['joint_edition_evidence'] = edition['joint_edition_evidence']
     if edition.get('track_scope_note'):
         metadata['track_scope_note']=edition['track_scope_note']
     if edition.get('track_scope_evidence_url'):

paper_atlas/collectors/dblp_scope.py

@@ -131,8 +131,15 @@ def scope(row,volume,through_year=2026):
     elif suffix.casefold() in ('poster','posters') and venue not in ('ICLR','ACL','EMNLP','NAACL','EACL','COLING','CoNLL'):track='poster'
     if venue=='AAAI+IAAI':track='proceedings_unspecified'
     if venue=='AAMAS' and year<2002:return None,'predecessor_not_aamas'
+    joint_evidence = None
+    if (family == 'ijcai' and base == 'IJCAI' and track == 'main'
+            and year in (2018, 2022) and volume['key'] == 'conf/ijcai/' + str(year)):
+        venue, aliases = 'ECAI+IJCAI', ['ECAI', 'IJCAI']
+        joint_evidence = 'https://www.eurai.org/ecai/history'
     result = {'venue':venue,'venue_aliases':aliases,'year':year,'track':track,'bibliographic_year':int(bibliographic[0]),
               'rule':family+':'+base,'proceedings_title':title,'edition_year_basis':year_basis}
+    if joint_evidence:
+        result['joint_edition_evidence'] = joint_evidence
     if historical:
         result['track_scope_note'] = historical[1]
         result['track_scope_evidence_url'] = historical[2]

paper_atlas/curation/__init__.py

@@ -1 +1 @@
-"""Offline, evidence-bound preparation of metadata for separate ingestion."""
+"""Evidence-bound identity preparation and independent type/tag annotations."""

paper_atlas/curation/annotations.py

@@ -0,0 +1,218 @@
+"""Append-only tag/type assertions and explicit retractions; no inferred labels."""
+import argparse
+from contextlib import closing
+from importlib.resources import files
+import json
+import re
+from urllib.parse import urlparse
+
+from paper_atlas.db.backend import database_arguments, inserted_id, open_read_connection, open_store
+from paper_atlas.db.submissions import timestamp
+
+TABLE = 'annotation_events'
+MIGRATION = 'annotation_events_v1'
+COLUMNS = ('id', 'request_id', 'paper_id', 'original_paper_id', 'kind', 'namespace',
+           'value', 'source_record_id', 'evidence_key', 'evidence_url', 'reason',
+           'actor', 'observed_at', 'retracts_id')
+ACTIVE = '''a.retracts_id IS NULL AND NOT EXISTS
+    (SELECT 1 FROM annotation_events removed WHERE removed.retracts_id=a.id)'''
+
+
+def available(conn):
+    if getattr(conn, 'postgres', False):
+        if hasattr(conn, 'has_table'):
+            return conn.has_table(TABLE)
+        return bool(conn.execute("SELECT 1 FROM information_schema.tables WHERE table_schema=current_schema() AND table_name=?", (TABLE,)).fetchone())
+    return bool(conn.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (TABLE,)).fetchone())
+
+
+def _begin_write(conn):
+    if not getattr(conn, 'postgres', False) and not conn.in_transaction:
+        conn.execute('BEGIN IMMEDIATE')
+
+
+def initialize(conn):
+    """Explicit transactional schema extension; does not rescan or rewrite papers."""
+    _begin_write(conn)
+    exists = available(conn)
+    marker = conn.execute('SELECT 1 FROM schema_migrations WHERE name=?', (MIGRATION,)).fetchone()
+    if exists != bool(marker):
+        raise RuntimeError('Annotation table/upgrade marker mismatch; inspect before changing schema')
+    if exists:
+        result = conn.execute('SELECT * FROM annotation_events LIMIT 0')
+        description = result.cursor.description if getattr(conn, 'postgres', False) else result.description
+        if tuple(d[0] for d in description) != COLUMNS:
+            raise RuntimeError('Unexpected annotation schema')
+        return False
+    ddl = files('paper_atlas.db').joinpath('annotations.sql').read_text()
+    if getattr(conn, 'postgres', False):
+        ddl = re.sub(r'\bINTEGER\b', 'BIGINT', ddl)
+        ddl = re.sub(r'\bTEXT\b', 'TEXT COLLATE "C"', ddl)
+        ddl = ddl.replace('id BIGINT PRIMARY KEY', 'id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY', 1)
+    for statement in ddl.split(';'):
+        if statement.strip():
+            conn.execute(statement)
+    conn.execute('INSERT INTO schema_migrations(name) VALUES(?)', (MIGRATION,))
+    return True
+
+
+def _text(value, name, limit=512):
+    if not isinstance(value, str) or not value.strip() or len(value) > limit:
+        raise ValueError(name + ' must be nonempty text of at most ' + str(limit) + ' characters')
+    return value
+
+
+def _id(value, name):
+    if type(value) is not int or value <= 0:
+        raise ValueError(name + ' must be a positive integer')
+    return value
+
+
+def _url(value):
+    if value is not None:
+        _text(value, 'evidence_url', 4096)
+        parsed = urlparse(value)
+        if parsed.scheme not in ('https', 'http') or not parsed.hostname or parsed.username or parsed.password:
+            raise ValueError('evidence_url must be an HTTP(S) URL without credentials')
+    return value
+
+
+def _require(conn):
+    if not available(conn) or not conn.execute('SELECT 1 FROM schema_migrations WHERE name=?', (MIGRATION,)).fetchone():
+        raise RuntimeError('Run paper-atlas annotate init explicitly before adding annotations')
+
+
+def _row(row):
+    return dict(zip(COLUMNS, (row[i] for i in range(len(COLUMNS)))))
+
+
+def _replay(conn, request_id, payload):
+    old = conn.execute('SELECT * FROM annotation_events WHERE request_id=?', (request_id,)).fetchone()
+    if old is None:
+        return None
+    old = _row(old)
+    if any(old[k] != value for k, value in payload.items()):
+        raise ValueError('request_id already identifies different annotation content')
+    return old['id']
+
+
+def assert_annotation(conn, paper, kind, namespace, value, *, request_id, evidence_key,
+                      reason, actor, evidence_url=None, source_record_id=None):
+    _begin_write(conn)
+    _require(conn)
+    paper = _id(paper, 'paper')
+    if kind not in ('tag', 'research_type'):
+        raise ValueError('kind must be tag or research_type')
+    _text(namespace, 'namespace', 128)
+    if not re.fullmatch(r'[a-z][a-z0-9_]*:[a-z][a-z0-9_.-]*', namespace):
+        raise ValueError('namespace must have an explicit source:scope')
+    payload = dict(original_paper_id=paper, kind=kind, namespace=namespace,
+                   value=_text(value, 'value'), request_id=_text(request_id, 'request_id', 128),
+                   source_record_id=source_record_id, evidence_key=_text(evidence_key, 'evidence_key', 256),
+                   reason=_text(reason, 'reason', 8192), actor=_text(actor, 'actor', 256),
+                   evidence_url=_url(evidence_url), retracts_id=None)
+    if source_record_id is not None:
+        _id(source_record_id, 'source_record_id')
+    replay = _replay(conn, request_id, payload)
+    if replay is not None:
+        return replay
+    if not conn.execute('SELECT 1 FROM papers WHERE id=?', (paper,)).fetchone():
+        raise ValueError('Existing canonical paper required; no automatic identity merge')
+    prefix = namespace.split(':', 1)[0]
+    if prefix != 'custom' and source_record_id is None:
+        raise ValueError('Platform annotations require their own source record')
+    if source_record_id is not None:
+        source = conn.execute('''SELECT r.paper_id,s.name FROM source_records r
+            JOIN sources s ON s.id=r.source_id WHERE r.id=?''', (source_record_id,)).fetchone()
+        if not source or source[0] != paper or (prefix != 'custom' and prefix != source[1]):
+            raise ValueError('Annotation source record/namespace must belong to this paper and platform')
+    payload.update(paper_id=paper, observed_at=timestamp())
+    fields = tuple(payload)
+    return inserted_id(conn, 'INSERT INTO annotation_events(' + ','.join(fields) + ') VALUES(' +
+                       ','.join('?' for _ in fields) + ')', tuple(payload[k] for k in fields))
+
+
+def retract_annotation(conn, event, *, request_id, reason, actor, evidence_url=None, custom_only=False):
+    _begin_write(conn)
+    _require(conn)
+    event = _id(event, 'event')
+    payload = dict(request_id=_text(request_id, 'request_id', 128), retracts_id=event,
+                   reason=_text(reason, 'reason', 8192), actor=_text(actor, 'actor', 256), evidence_url=_url(evidence_url))
+    target = conn.execute('SELECT * FROM annotation_events WHERE id=?', (event,)).fetchone()
+    if target is None or target[-1] is not None:
+        raise ValueError('Retraction must reference an existing assertion')
+    target = _row(target)
+    if custom_only and not target['namespace'].startswith('custom:'):
+        raise ValueError('Manual CLI cannot retract platform evidence')
+    replay = _replay(conn, request_id, payload)
+    if replay is not None:
+        return replay
+    if conn.execute('SELECT 1 FROM annotation_events WHERE retracts_id=?', (event,)).fetchone():
+        raise ValueError('Assertion already retracted; its history is retained')
+    payload.update({k: target[k] for k in ('paper_id', 'original_paper_id', 'kind', 'namespace', 'value', 'source_record_id', 'evidence_key')})
+    payload['observed_at'] = timestamp()
+    fields = tuple(payload)
+    return inserted_id(conn, 'INSERT INTO annotation_events(' + ','.join(fields) + ') VALUES(' +
+                       ','.join('?' for _ in fields) + ')', tuple(payload[k] for k in fields))
+
+
+def observations(conn, paper, *, kind=None, history=False):
+    if not available(conn):
+        return []
+    where = ['a.paper_id=?']; params = [_id(paper, 'paper')]
+    if not history:
+        where.append(ACTIVE)
+    if kind is not None:
+        if kind not in ('tag', 'research_type'):
+            raise ValueError('Invalid annotation kind')
+        where.append('a.kind=?'); params.append(kind)
+    return [_row(r) for r in conn.execute('SELECT a.* FROM annotation_events a WHERE ' +
+                                         ' AND '.join(where) + ' ORDER BY a.id', params)]
+
+
+def main(argv=None, *, prog=None):
+    parser = argparse.ArgumentParser(prog=prog, description='显式初始化、追加或撤回带证据的自定义标签和研究类型')
+    actions = parser.add_subparsers(dest='action', required=True)
+    for action in ('init', 'assert', 'retract', 'list'):
+        p = actions.add_parser(action)
+        database_arguments(p)
+        if action in ('assert', 'retract'):
+            for field in ('request-id', 'reason', 'actor'):
+                p.add_argument('--' + field, required=True)
+            p.add_argument('--evidence-url')
+        if action in ('assert', 'list'):
+            p.add_argument('--paper', type=int, required=True)
+            p.add_argument('--kind', choices=('tag', 'research_type'), required=action == 'assert')
+        if action == 'assert':
+            for field in ('namespace', 'value', 'evidence-key'):
+                p.add_argument('--' + field, required=True)
+            p.add_argument('--source-record', type=int)
+        if action == 'retract':
+            p.add_argument('--event', type=int, required=True)
+        if action == 'list':
+            p.add_argument('--history', action='store_true')
+    args = parser.parse_args(argv)
+    if args.action == 'list':
+        with closing(open_read_connection(args.db, pg_service=args.pg_service, pg_schema=args.pg_schema)) as conn:
+            result = observations(conn, args.paper, kind=args.kind, history=args.history)
+    else:
+        if args.action == 'assert' and not args.namespace.startswith('custom:'):
+            parser.error('Manual CLI requires custom:* namespace; platform assertions use validated source APIs')
+        with closing(open_store(args.db, pg_service=args.pg_service, pg_schema=args.pg_schema).conn) as conn:
+            try:
+                if not getattr(conn, 'postgres', False):
+                    conn.commit()
+                    conn.execute('BEGIN IMMEDIATE')
+                if args.action == 'init':
+                    result = {'initialized': initialize(conn)}
+                else:
+                    kwargs = dict(request_id=args.request_id, reason=args.reason, actor=args.actor, evidence_url=args.evidence_url)
+                    event = (assert_annotation(conn, args.paper, args.kind, args.namespace, args.value,
+                              evidence_key=args.evidence_key, source_record_id=args.source_record, **kwargs)
+                             if args.action == 'assert' else retract_annotation(conn, args.event, custom_only=True, **kwargs))
+                    result = {'event_id': event}
+                conn.commit()
+            except BaseException:
+                conn.rollback()
+                raise
+    print(json.dumps(result, ensure_ascii=False, indent=2))

paper_atlas/db/annotations.sql

@@ -0,0 +1,15 @@
+CREATE TABLE annotation_events (
+ id INTEGER PRIMARY KEY,
+ request_id TEXT NOT NULL UNIQUE,
+ paper_id INTEGER NOT NULL REFERENCES papers(id),
+ original_paper_id INTEGER NOT NULL,
+ kind TEXT NOT NULL CHECK(kind IN ('tag','research_type')),
+ namespace TEXT NOT NULL, value TEXT NOT NULL,
+ source_record_id INTEGER REFERENCES source_records(id),
+ evidence_key TEXT NOT NULL, evidence_url TEXT,
+ reason TEXT NOT NULL, actor TEXT NOT NULL, observed_at TEXT NOT NULL,
+ retracts_id INTEGER UNIQUE REFERENCES annotation_events(id),
+ CHECK(retracts_id IS NULL OR retracts_id<>id)
+);
+CREATE INDEX annotation_events_paper ON annotation_events(paper_id,kind,id);
+CREATE INDEX annotation_events_label ON annotation_events(kind,namespace,value,paper_id);

paper_atlas/db/events.py

@@ -0,0 +1,71 @@
+"""Stable event identity within a venue/year/track, independent of paper identity."""
+import json
+import re
+
+
+def has_event_key(c):
+    if getattr(c, 'postgres', False):
+        return bool(c.execute("SELECT 1 FROM information_schema.columns WHERE table_schema=current_schema() AND table_name='editions' AND column_name='event_key'").fetchone())
+    return any(row[1] == 'event_key' for row in c.execute('PRAGMA table_info(editions)'))
+
+
+def validate_event_key(value):
+    if value is not None and (not isinstance(value, str) or
+            (value != '' and not re.fullmatch(r'[a-z0-9][a-z0-9._:-]{0,79}', value))):
+        raise ValueError('Invalid conference event_key')
+    return value
+
+
+def pinned_www_fall(source, metadata):
+    from paper_atlas.collectors.dblp import XML_SHA
+    b = metadata.get('bibliography') or {}
+    urls = b.get('url') or []
+    return (source == 'dblp' and metadata.get('snapshot_sha256') == XML_SHA
+            and metadata.get('bibliographic_year') == 1995
+            and str(metadata.get('proceedings_key', '')).startswith('conf/www/')
+            and b.get('booktitle') == ['WWW (Fall)'] and len(urls) == 1
+            and urls[0].split('#')[0] == 'db/conf/www/www1995f.html')
+
+
+def find_edition(store, venue_id, venue, year, track, record, external_id,
+                 event_key=None, *, create=True):
+    c = store.conn
+    validate_event_key(event_key)
+    enabled = has_event_key(c) if getattr(store, 'event_identity', None) is None else store.event_identity
+    known_multi = venue.upper() == 'WWW' and int(year) in (1994, 1995)
+    source = c.execute('SELECT src.name,r.metadata_json FROM source_records r JOIN sources src ON src.id=r.source_id WHERE r.id=?', (record,)).fetchone()
+    inferred = ('www-1995-fall' if known_multi and int(year) == 1995 and source
+                and pinned_www_fall(source[0], json.loads(source[1])) else None)
+    if inferred:
+        if event_key is not None and event_key != inferred:
+            raise ValueError('event_key conflicts with pinned WWW Fall source evidence')
+        event_key = inferred
+    if not enabled:
+        if event_key or known_multi:
+            raise ValueError('Explicit event identity migration required before this conference import')
+        row = c.execute('SELECT id FROM editions WHERE venue_id=? AND year=? AND track=?', (venue_id, year, track)).fetchone()
+        if row or not create:
+            return row[0] if row else None
+        c.execute('INSERT INTO editions(venue_id,year,track) VALUES(?,?,?) ON CONFLICT DO NOTHING', (venue_id, year, track))
+        return c.execute('SELECT id FROM editions WHERE venue_id=? AND year=? AND track=?', (venue_id, year, track)).fetchone()[0]
+    anchored = c.execute('''SELECT DISTINCT e.id,e.event_key FROM editions e JOIN submissions s ON s.edition_id=e.id
+        JOIN submission_sources ss ON ss.submission_id=s.id WHERE e.venue_id=? AND e.year=? AND e.track=?
+        AND ss.source_record_id=? AND ss.source_submission_id=?''', (venue_id, year, track, record, external_id)).fetchall()
+    if len(anchored) > 1:
+        raise ValueError('Source record already has ambiguous conference events')
+    if anchored:
+        if event_key is not None and event_key != anchored[0][1]:
+            raise ValueError('Cannot move an existing source to another conference event')
+        event_key = anchored[0][1]
+    if event_key is None:
+        rows = c.execute('SELECT event_key FROM editions WHERE venue_id=? AND year=? AND track=?', (venue_id, year, track)).fetchall()
+        if len(rows) > 1 or any(row[0] for row in rows):
+            raise ValueError('Explicit event_key required; conference event is ambiguous')
+        event_key = ''
+    if known_multi and not event_key:
+        raise ValueError('Explicit event_key required for same-year WWW events')
+    row = c.execute('SELECT id FROM editions WHERE venue_id=? AND year=? AND track=? AND event_key=?', (venue_id, year, track, event_key)).fetchone()
+    if row or not create:
+        return row[0] if row else None
+    c.execute('INSERT INTO editions(venue_id,year,track,event_key) VALUES(?,?,?,?) ON CONFLICT DO NOTHING', (venue_id, year, track, event_key))
+    return c.execute('SELECT id FROM editions WHERE venue_id=? AND year=? AND track=? AND event_key=?', (venue_id, year, track, event_key)).fetchone()[0]

paper_atlas/db/schema.sql

@@ -29,7 +29,8 @@ CREATE TABLE IF NOT EXISTS venues (
 CREATE TABLE IF NOT EXISTS editions (
  id INTEGER PRIMARY KEY, venue_id INTEGER NOT NULL REFERENCES venues(id),
  year INTEGER NOT NULL, track TEXT NOT NULL DEFAULT 'main',
- UNIQUE(venue_id,year,track)
+ event_key TEXT NOT NULL DEFAULT '',
+ UNIQUE(venue_id,year,track,event_key)
 );
 CREATE TABLE IF NOT EXISTS venue_aliases (
  venue_id INTEGER NOT NULL REFERENCES venues(id), alias TEXT NOT NULL,

paper_atlas/db/store.py

@@ -36,6 +36,8 @@ class Store:
         self.conn.execute('PRAGMA synchronous=NORMAL')
         self.conn.execute('PRAGMA cache_size=-32768')
         migrate_sources(self.conn)
+        from paper_atlas.db.events import has_event_key
+        self.event_identity = has_event_key(self.conn)
 
     def source(self,name,url):
         self.conn.execute('INSERT INTO sources(name,homepage) VALUES(?,?) ON CONFLICT DO NOTHING',(name,url))
@@ -55,6 +57,9 @@ class Store:
         c.execute('UPDATE identifiers SET paper_id=? WHERE paper_id=?',(target,other))
         c.execute('UPDATE source_records SET paper_id=? WHERE paper_id=?',(target,other))
         c.execute('UPDATE submissions SET paper_id=? WHERE paper_id=?',(target,other))
+        from paper_atlas.curation.annotations import available
+        if available(c):
+            c.execute('UPDATE annotation_events SET paper_id=? WHERE paper_id=?', (target,other))
         c.execute('INSERT INTO paper_tags SELECT ?,tag_id,source_record_id,evidence_key FROM paper_tags WHERE paper_id=? ON CONFLICT DO NOTHING',(target,other))
         c.execute('DELETE FROM paper_tags WHERE paper_id=?',(other,))
         c.execute('UPDATE paper_merges SET surviving_paper_id=? WHERE surviving_paper_id=?',(target,other))
@@ -148,15 +153,29 @@ 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,guard_status=False):
+    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,event_key=None):
         c = self.conn
+        # An already collected file may still call these exact pinned volumes
+        # IJCAI. Resolve membership before finding/creating an edition so a
+        # later refresh cannot split the repaired joint event again.
+        if venue == 'IJCAI' and year in (2018,2022) and track == 'main':
+            from paper_atlas.collectors.dblp import XML_SHA
+            evidence=c.execute('SELECT src.name,r.metadata_json FROM source_records r JOIN sources src ON src.id=r.source_id WHERE r.id=?',(record,)).fetchone()
+            if evidence and evidence[0] == 'dblp':
+                metadata=json.loads(evidence[1])
+                if (metadata.get('snapshot_sha256') == XML_SHA
+                        and metadata.get('proceedings_key') == 'conf/ijcai/'+str(year)
+                        and metadata.get('bibliographic_year') == year
+                        and (metadata.get('bibliography') or {}).get('booktitle') in (['IJCAI'],['IJCAI/ECAI'])):
+                    venue='ECAI+IJCAI'
+                    aliases=sorted(set(aliases or []) | {'ECAI','IJCAI'})
         existing=c.execute('SELECT id FROM venues WHERE lower(acronym)=lower(?)',(venue,)).fetchone()
         if existing:venue_id=existing[0]
         else:venue_id=inserted_id(c,"INSERT INTO venues(acronym,name,field,venue_type) VALUES(?,?,?,'conference')",(venue,venue,'unclassified'))
         for alias in aliases or []:
             c.execute('INSERT INTO venue_aliases VALUES(?,?) ON CONFLICT DO NOTHING',(venue_id,alias))
-        c.execute('INSERT INTO editions(venue_id,year,track) VALUES(?,?,?) ON CONFLICT DO NOTHING',(venue_id,int(year),track))
-        edition = c.execute('SELECT id FROM editions WHERE venue_id=? AND year=? AND track=?',(venue_id,int(year),track)).fetchone()[0]
+        from paper_atlas.db.events import find_edition
+        edition = find_edition(self,venue_id,venue,int(year),track,record,external_id,event_key)
         # 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 [])

paper_atlas/ingestion/backfill.py

@@ -37,9 +37,14 @@ def backfill(db,files, *, pg_service=None, pg_schema=None):
                     seen.add(row['external_id'])
                     if any(row.get(key)!=manifest.get(key) for key in ('source','venue','year','track')):
                         raise ValueError('Evidence row differs from edition manifest')
+                    if 'event_key' in manifest and row.get('event_key')!=manifest['event_key']:
+                        raise ValueError('Event identity differs from manifest')
                     require_track_record(row, manifest)
                     record=c.execute('SELECT r.id,r.paper_id,r.metadata_json FROM source_records r JOIN sources s ON s.id=r.source_id WHERE s.name=? AND r.external_id=?',(row['source'],row['external_id'])).fetchone()
-                    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=?',(row['venue'],row['year'],row['track'])).fetchone()
+                    from paper_atlas.db.events import find_edition
+                    venue=c.execute('SELECT id FROM venues WHERE lower(acronym)=lower(?)',(row['venue'],)).fetchone()
+                    eid=find_edition(store,venue[0],row['venue'],row['year'],row['track'],record[0],row['external_id'],row.get('event_key'),create=False) if record and venue else None
+                    edition=(eid,) if eid is not None else None
                     if not record or not edition:raise ValueError('Evidence backfill requires an already imported record and edition')
                     if not c.execute('SELECT 1 FROM submission_sources ss JOIN submissions s ON s.id=ss.submission_id WHERE ss.source_record_id=? AND ss.source_submission_id=? AND s.edition_id=?',(record[0],row['external_id'],edition[0])).fetchone():
                         raise ValueError('No existing submission to backfill')

paper_atlas/ingestion/event_identity.py

@@ -0,0 +1,221 @@
+"""Explicit, plan-bound edition event-key upgrade; never reimport submissions."""
+import argparse
+from contextlib import closing
+import hashlib
+import json
+from pathlib import Path
+import sqlite3
+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, OPTIONAL_TABLES, quote
+
+MARKER = 'edition_event_identity_v1:'
+
+
+def digest(value):
+    return hashlib.sha256(json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(',', ':')).encode()).hexdigest()
+
+
+def values(rows):
+    return [[r[i] for i in range(len(r))] for r in rows]
+
+
+def exact_counts(c, *, page_size=5000, progress=None):
+    """Count every row in the same snapshot through bounded primary-key pages."""
+    if not isinstance(page_size, int) or page_size < 1:
+        raise ValueError('Positive count page size required')
+    result = {}
+    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):
+            columns = [r[0] for r in c.execute("SELECT a.attname FROM pg_constraint pc JOIN pg_attribute a ON a.attrelid=pc.conrelid AND a.attnum=ANY(pc.conkey) WHERE pc.conrelid=?::regclass AND pc.contype='p' ORDER BY array_position(pc.conkey,a.attnum)", (table,))]
+        else:
+            columns = [r[1] for r in sorted(c.execute('PRAGMA table_info('+quote(table)+')'), key=lambda r:r[5]) if r[5]]
+        if not columns:
+            raise ValueError('Primary key required for exact count: '+table)
+        fields = ','.join(quote(column) for column in columns)
+        last = None; count = 0; pages = 0
+        if progress:
+            progress({'table':table, 'rows_counted':0, 'complete':False})
+        while True:
+            where = (' WHERE ('+fields+') > ('+','.join('?' for _ in columns)+')') if last is not None else ''
+            batch = values(c.execute('SELECT '+fields+' FROM '+quote(table)+where+' ORDER BY '+fields+' LIMIT ?', (*last, page_size) if last is not None else (page_size,)).fetchall())
+            count += len(batch); pages += 1
+            complete = len(batch) < page_size
+            if progress and (complete or pages % 20 == 0):
+                progress({'table':table, 'rows_counted':count, 'complete':complete})
+            if complete:
+                break
+            last = batch[-1]
+        result[table] = count
+    return result
+
+
+def inspect(c, *, progress=None):
+    enabled = has_event_key(c)
+    if getattr(c, 'postgres', False):
+        columns = [r[0] for r in c.execute("SELECT column_name FROM information_schema.columns WHERE table_schema=current_schema() AND table_name='editions' ORDER BY ordinal_position")]
+        unique = [r[0] for r in c.execute("SELECT pg_get_constraintdef(oid) FROM pg_constraint WHERE conrelid='editions'::regclass AND contype='u'")]
+        expected_unique = ['UNIQUE (venue_id, year, track, event_key)' if enabled else 'UNIQUE (venue_id, year, track)']
+    else:
+        columns = [r[1] for r in c.execute('PRAGMA table_info(editions)')]
+        indexes = list(c.execute('PRAGMA index_list(editions)'))
+        unique = [tuple(r[2] for r in c.execute('PRAGMA index_info('+quote(index[1])+')')) for index in indexes if index[2]]
+        expected_unique = [('venue_id','year','track','event_key') if enabled else ('venue_id','year','track')]
+        if len(indexes) != 1:
+            raise ValueError('Unexpected edition indexes; require a reviewed migration')
+    if columns != ['id','venue_id','year','track'] + (['event_key'] if enabled else []) or unique != expected_unique:
+        raise ValueError('Unexpected edition schema; require a reviewed migration')
+    rows = values(c.execute('SELECT id,venue_id,year,track FROM editions ORDER BY id'))
+    keys = values(c.execute('SELECT id,event_key FROM editions ORDER BY id')) if enabled else [[r[0], ''] for r in rows]
+    counts = exact_counts(c, progress=progress)
+    targets = values(c.execute("SELECT e.id FROM editions e JOIN venues v ON v.id=e.venue_id WHERE v.acronym='WWW' AND e.year=1995 AND e.track='main' ORDER BY e.id"))
+    assignments = []
+    for (eid,) in targets:
+        current_key = dict(keys)[eid]
+        if current_key:
+            continue
+        evidence = values(c.execute('''SELECT s.id,src.name,r.id,r.external_id,r.metadata_json FROM submissions s
+            JOIN submission_sources ss ON ss.submission_id=s.id JOIN source_records r ON r.id=ss.source_record_id
+            JOIN sources src ON src.id=r.source_id WHERE s.edition_id=? ORDER BY s.id,r.id''', (eid,)))
+        count = c.execute('SELECT count(*) FROM submissions WHERE edition_id=?', (eid,)).fetchone()[0]
+        supported = {r[0] for r in evidence if pinned_www_fall(r[1], json.loads(r[4]))}
+        if not count or len(supported) != count:
+            raise ValueError('Existing unlabelled WWW1995 edition lacks complete pinned Fall evidence')
+        assignments.append({'edition_id':eid, 'event_key':'www-1995-fall', 'submissions':count,
+                            'source_evidence_sha256':digest(evidence)})
+    # Protect all rows in the affected attempt history, including non-DBLP evidence.
+    target_ids = [r[0] for r in targets]
+    protected = {}
+    for table,sql in {
+        'submissions': 'SELECT s.* FROM submissions s WHERE s.edition_id IN ({}) ORDER BY s.id',
+        'decisions': 'SELECT d.* FROM decisions d JOIN submissions s ON s.id=d.submission_id WHERE s.edition_id IN ({}) ORDER BY d.id',
+        'submission_sources': 'SELECT ss.* FROM submission_sources ss JOIN submissions s ON s.id=ss.submission_id WHERE s.edition_id IN ({}) ORDER BY ss.submission_id,ss.source_record_id,ss.source_submission_id',
+        'decision_sources': 'SELECT ds.* FROM decision_sources ds JOIN decisions d ON d.id=ds.decision_id JOIN submissions s ON s.id=d.submission_id WHERE s.edition_id IN ({}) ORDER BY ds.decision_id,ds.source_record_id,ds.source_decision_id',
+        'source_records': 'SELECT DISTINCT r.* FROM source_records r JOIN submission_sources ss ON ss.source_record_id=r.id JOIN submissions s ON s.id=ss.submission_id WHERE s.edition_id IN ({}) ORDER BY r.id',
+    }.items():
+        records = values(c.execute(sql.format(','.join('?' for _ in target_ids)), target_ids)) if target_ids else []
+        protected[table] = {'count':len(records), 'sha256':digest(records)}
+    return {'event_column':enabled, 'editions_sha256':digest(rows), 'edition_ids':[r[0] for r in rows],
+            'event_keys':keys, 'counts':counts, 'assignments':assignments, 'protected':protected}
+
+
+def database_identity(c, db=None):
+    if getattr(c, 'postgres', False):
+        r = c.execute('SELECT current_database(),current_schema()').fetchone()
+        return {'engine':'postgres', 'database':r[0], 'schema':r[1]}
+    return {'engine':'sqlite', 'path':str(Path(db).resolve())}
+
+
+def plan(db=None, *, pg_service=None, pg_schema=None, progress=None):
+    with closing(open_read_connection(db, pg_service=pg_service, pg_schema=pg_schema)) as c:
+        return {'format':1, 'database':database_identity(c, db), 'before':inspect(c, progress=progress)}
+
+
+def apply_plan(c, proposal, *, progress=None):
+    before = proposal['before']
+    if proposal.get('format') != 1:
+        raise ValueError('Unsupported event identity plan')
+    if getattr(c, 'postgres', False):
+        c.execute('LOCK TABLE editions IN ACCESS EXCLUSIVE MODE')
+    current = inspect(c, progress=progress)
+    signature = digest(proposal)
+    marker = MARKER + signature
+    if c.execute('SELECT 1 FROM schema_migrations WHERE name=?', (marker,)).fetchone():
+        expected_keys = dict(before['event_keys'])
+        expected_keys.update({a['edition_id']:a['event_key'] for a in before['assignments']})
+        if (not current['event_column'] or current['editions_sha256'] != before['editions_sha256']
+                or current['protected'] != before['protected'] or current['counts'] != before['counts']
+                or dict(current['event_keys']) != expected_keys):
+            raise ValueError('Applied plan no longer matches current data; generate a new plan')
+        return {'status':'already_applied', 'updated_editions':0, 'after':current}
+    if current != before:
+        raise ValueError('Event identity plan is stale; generate a new plan')
+    if not current['event_column']:
+        if getattr(c, 'postgres', False):
+            constraints = c.execute("SELECT conname,pg_get_constraintdef(oid) FROM pg_constraint WHERE conrelid='editions'::regclass AND contype='u'").fetchall()
+            matches = [name for name,definition in constraints if definition == 'UNIQUE (venue_id, year, track)']
+            if len(matches) != 1 or len(constraints) != 1:
+                raise ValueError('Unexpected legacy edition uniqueness')
+            c.execute("ALTER TABLE editions ADD COLUMN event_key TEXT COLLATE \"C\" NOT NULL DEFAULT ''")
+            c.execute('ALTER TABLE editions DROP CONSTRAINT '+quote(matches[0]))
+            c.execute('ALTER TABLE editions ADD CONSTRAINT editions_event_identity_unique UNIQUE(venue_id,year,track,event_key)')
+        else:
+            unique = [tuple(r[2] for r in c.execute('PRAGMA index_info('+quote(index[1])+')'))
+                      for index in c.execute('PRAGMA index_list(editions)') if index[2]]
+            if unique != [('venue_id','year','track')]:
+                raise ValueError('Unexpected legacy edition uniqueness')
+            c.execute("CREATE TABLE editions_event_upgrade(id INTEGER PRIMARY KEY,venue_id INTEGER NOT NULL REFERENCES venues(id),year INTEGER NOT NULL,track TEXT NOT NULL DEFAULT 'main',event_key TEXT NOT NULL DEFAULT '',UNIQUE(venue_id,year,track,event_key))")
+            c.execute('INSERT INTO editions_event_upgrade(id,venue_id,year,track) SELECT id,venue_id,year,track FROM editions')
+            c.execute('DROP TABLE editions')
+            c.execute('ALTER TABLE editions_event_upgrade RENAME TO editions')
+    for item in before['assignments']:
+        result = c.execute("UPDATE editions SET event_key=? WHERE id=? AND event_key=''", (item['event_key'],item['edition_id']))
+        if result.rowcount != 1:
+            raise ValueError('Edition assignment changed during upgrade')
+    c.execute('INSERT INTO schema_migrations(name) VALUES(?)', (marker,))
+    after = inspect(c, progress=progress)
+    expected_keys = dict(before['event_keys'])
+    expected_keys.update({a['edition_id']:a['event_key'] for a in before['assignments']})
+    if (after['editions_sha256'] != before['editions_sha256'] or after['protected'] != before['protected']
+            or after['counts'] != before['counts'] or dict(after['event_keys']) != expected_keys):
+        raise ValueError('Event upgrade changed protected data')
+    if not getattr(c, 'postgres', False) and c.execute('PRAGMA foreign_key_check').fetchall():
+        raise ValueError('Foreign key violation after event upgrade')
+    return {'status':'applied', 'updated_editions':len(before['assignments']), 'after':after}
+
+
+def upgrade(proposal, db=None, *, pg_service=None, pg_schema=None, progress=None):
+    if pg_service:
+        from paper_atlas.postgres.writer import WriteConnection
+        with closing(WriteConnection(pg_service, pg_schema)) as c:
+            if proposal['database'] != database_identity(c):
+                raise ValueError('Plan targets a different database')
+            try:
+                result = apply_plan(c, proposal, progress=progress); c.commit(); return result
+            except BaseException:
+                c.rollback(); raise
+    require_sqlite_writable(db)
+    if not Path(db).is_file():
+        raise ValueError('Existing SQLite sample required')
+    with closing(sqlite3.connect(db, timeout=30)) as c:
+        if proposal['database'] != database_identity(c, db):
+            raise ValueError('Plan targets a different database')
+        c.execute('PRAGMA foreign_keys=OFF')  # Outside the transaction; no cascading deletes.
+        try:
+            c.execute('BEGIN IMMEDIATE')
+            result = apply_plan(c, proposal, progress=progress); c.commit(); return result
+        except BaseException:
+            c.rollback(); raise
+        finally:
+            c.execute('PRAGMA foreign_keys=ON')
+
+
+def main(argv=None, *, prog=None):
+    p = argparse.ArgumentParser(prog=prog); database_arguments(p)
+    p.add_argument('--out', required=True)
+    p.add_argument('--apply-plan'); p.add_argument('--plan-sha256')
+    args = p.parse_args(argv)
+    if bool(args.apply_plan) != bool(args.plan_sha256):
+        p.error('--apply-plan and --plan-sha256 must be supplied together')
+    target = Path(args.out)
+    if target.exists():
+        raise ValueError('Refusing to overwrite an existing plan/report')
+    progress = lambda event: print(json.dumps({'phase':'exact_row_count', **event}), file=sys.stderr, flush=True)
+    if args.apply_plan:
+        raw = Path(args.apply_plan).read_bytes()
+        if hashlib.sha256(raw).hexdigest() != args.plan_sha256:
+            raise ValueError('Plan file SHA256 differs from approved bytes')
+        result = upgrade(json.loads(raw), args.db, pg_service=args.pg_service, pg_schema=args.pg_schema, progress=progress)
+    else:
+        result = plan(args.db, pg_service=args.pg_service, pg_schema=args.pg_schema, progress=progress)
+    target.write_text(json.dumps(result, ensure_ascii=False, indent=2)+'\n')
+    print(json.dumps({'out':str(target), 'sha256':hashlib.sha256(target.read_bytes()).hexdigest(),
+                      'status':result.get('status','planned')}), flush=True)

paper_atlas/ingestion/joint_editions.py

@@ -0,0 +1,204 @@
+"""Evidence-bound repair of two existing IJCAI/ECAI edition memberships."""
+import argparse
+from contextlib import closing
+import hashlib
+import json
+from pathlib import Path
+
+from paper_atlas.db.backend import database_arguments, open_read_connection, open_store
+from paper_atlas.db.submissions import timestamp
+from paper_atlas.collectors.dblp import XML_SHA
+
+YEARS = (2018, 2022)
+TARGET = 'ECAI+IJCAI'
+MEMBERS = {'ECAI', 'IJCAI'}
+EVIDENCE = 'https://www.eurai.org/ecai/history'
+MARKER = 'joint_ijcai_ecai_2018_2022_v1'
+
+
+def _digest(c, sql, params=()):
+    sha = hashlib.sha256(); count = 0
+    for row in c.execute(sql, params):
+        sha.update((json.dumps([row[i] for i in range(len(row))], ensure_ascii=False,
+                              separators=(',', ':')) + '\n').encode())
+        count += 1
+        if count > 100000:
+            raise ValueError('Joint repair evidence scope exceeded bounded row limit')
+    return {'rows': count, 'sha256': sha.hexdigest()}
+
+
+def fingerprints(c, ids):
+    marks = ','.join('?' for _ in ids)
+    submissions = 'SELECT id FROM submissions WHERE edition_id IN (' + marks + ')'
+    papers = 'SELECT paper_id FROM submissions WHERE edition_id IN (' + marks + ')'
+    decisions = 'SELECT id FROM decisions WHERE submission_id IN (' + submissions + ')'
+    checks = {
+        'papers': ('id IN (' + papers + ')', 'id'),
+        'identifiers': ('paper_id IN (' + papers + ')', 'scheme,value'),
+        'source_records': ('paper_id IN (' + papers + ')', 'id'),
+        'paper_tags': ('paper_id IN (' + papers + ')', 'paper_id,tag_id,evidence_key'),
+        'submissions': ('edition_id IN (' + marks + ')', 'id'),
+        'decisions': ('submission_id IN (' + submissions + ')', 'id'),
+        'submission_sources': ('submission_id IN (' + submissions + ')', 'submission_id,source_record_id,source_submission_id'),
+        'submission_identifiers': ('edition_id IN (' + marks + ')', 'edition_id,scheme,value'),
+        'decision_sources': ('decision_id IN (' + decisions + ')', 'decision_id,source_record_id,source_decision_id'),
+        'edition_sources': ('edition_id IN (' + marks + ')', 'edition_id,source_id,list_url'),
+    }
+    return {table: _digest(c, 'SELECT * FROM ' + table + ' WHERE ' + where + ' ORDER BY ' + order, ids)
+            for table, (where, order) in checks.items()}
+
+
+def inspect(c):
+    target = c.execute('SELECT id FROM venues WHERE acronym=?', (TARGET,)).fetchall()
+    if len(target) != 1:
+        raise ValueError('Exactly one existing canonical joint venue required')
+    target = target[0][0]
+    aliases = {r[0] for r in c.execute('SELECT alias FROM venue_aliases WHERE venue_id=?', (target,))}
+    if aliases != MEMBERS:
+        raise ValueError('Joint target must already have exactly ECAI and IJCAI aliases')
+    editions = []
+    for year in YEARS:
+        rows = c.execute('''SELECT e.id,e.venue_id,v.acronym FROM editions e JOIN venues v ON v.id=e.venue_id
+            WHERE v.acronym IN ('IJCAI','ECAI+IJCAI') AND e.year=? AND e.track='main' ''', (year,)).fetchall()
+        if len(rows) != 1:
+            raise ValueError('Expected one existing edition per year; missing or colliding editions require separate review')
+        edition, venue, acronym = (rows[0][i] for i in range(3))
+        count = c.execute('SELECT count(*) FROM submissions WHERE edition_id=?', (edition,)).fetchone()[0]
+        if not 0 < count <= 5000:
+            raise ValueError('Existing edition must have a nonempty bounded submission scope')
+        # Every existing submission must be supported by its own pinned DBLP
+        # volume record. No title/author matching, new attempts or text writes.
+        evidence = c.execute('''SELECT s.id,r.external_id,r.metadata_json FROM submissions s
+            JOIN submission_sources ss ON ss.submission_id=s.id
+            JOIN source_records r ON r.id=ss.source_record_id JOIN sources src ON src.id=r.source_id
+            WHERE s.edition_id=? AND src.name='dblp' ORDER BY s.id,r.id''', (edition,)).fetchall()
+        seen = set()
+        for row in evidence:
+            sid, external, metadata = (row[i] for i in range(3))
+            meta = json.loads(metadata)
+            bibliography = meta.get('bibliography') or {}
+            if (meta.get('snapshot_sha256') != XML_SHA
+                    or meta.get('proceedings_key') != 'conf/ijcai/' + str(year)
+                    or meta.get('bibliographic_year') != year
+                    or bibliography.get('booktitle') not in (['IJCAI'], ['IJCAI/ECAI'])
+                    or not external.startswith('conf/ijcai/')):
+                raise ValueError('Source evidence outside the exact verified IJCAI/ECAI volume')
+            seen.add(sid)
+        if len(seen) != count:
+            raise ValueError('Every submission must have matching original volume evidence')
+        editions.append({'id': edition, 'year': year, 'track': 'main', 'from_venue_id': venue,
+                         'from_venue': acronym, 'target_venue_id': target, 'submissions': count,
+                         'evidence_url': EVIDENCE, 'proceedings_key': 'conf/ijcai/' + str(year)})
+    ids = tuple(e['id'] for e in editions)
+    return {'format_version': 1, 'kind': MARKER, 'checked_at': timestamp(), 'editions': editions,
+            'content_fingerprints': fingerprints(c, ids), 'target_aliases': sorted(aliases),
+            'status': 'already_joint' if all(e['from_venue_id'] == target for e in editions) else 'ready',
+            'scope': 'Only these two existing edition venue references; no source text, paper, submission, decision, track, identifier or tag change.'}
+
+
+def validate_plan(expected):
+    if not isinstance(expected, dict):
+        raise ValueError('Repair plan must be a JSON object')
+    if set(expected) != {'format_version','kind','checked_at','editions','content_fingerprints','target_aliases','status','scope','database'}:
+        raise ValueError('Unexpected repair plan fields')
+    if (type(expected['format_version']) is not int or expected['format_version'] != 1
+            or expected['kind'] != MARKER or expected['status'] not in ('ready','already_joint')):
+        raise ValueError('Unsupported repair plan')
+    if (not isinstance(expected['database'],dict) or not isinstance(expected['content_fingerprints'],dict)
+            or not isinstance(expected['editions'],list) or len(expected['editions']) != 2
+            or expected['target_aliases'] != sorted(MEMBERS)):
+        raise ValueError('Invalid repair plan structure')
+    for edition, year in zip(expected['editions'], YEARS):
+        if not isinstance(edition,dict) or edition.get('year') != year:
+            raise ValueError('Repair plan must name the two exact ordered years')
+        for key in ('id','year','from_venue_id','target_venue_id','submissions'):
+            if type(edition.get(key)) is not int or edition[key] <= 0:
+                raise ValueError('Repair plan IDs/counts must be positive integers')
+
+
+def apply_plan(c, expected):
+    validate_plan(expected)
+    actual = inspect(c)
+    if actual['content_fingerprints'] != expected['content_fingerprints'] or actual['target_aliases'] != expected['target_aliases']:
+        raise ValueError('Stale repair plan: protected content or aliases changed')
+    if len(expected['editions']) != 2:
+        raise ValueError('Exactly two editions required')
+    for old, now in zip(expected['editions'], actual['editions']):
+        if set(old) != set(now):
+            raise ValueError('Unexpected edition fields')
+        for key in ('id','year','track','target_venue_id','submissions','evidence_url','proceedings_key'):
+            if old[key] != now[key]:
+                raise ValueError('Stale repair plan: edition differs')
+        if now['from_venue_id'] != old['from_venue_id'] and now['from_venue_id'] != now['target_venue_id']:
+            raise ValueError('Edition membership differs from plan')
+    if actual['status'] == 'already_joint':
+        return {**actual, 'status':'already_joint', 'updated_editions':0}
+    for edition in actual['editions']:
+        if edition['from_venue_id'] == edition['target_venue_id']:
+            continue
+        changed = c.execute('''UPDATE editions SET venue_id=? WHERE id=? AND venue_id=? AND year=? AND track=?''',
+                            (edition['target_venue_id'],edition['id'],edition['from_venue_id'],edition['year'],edition['track']))
+        if changed.rowcount != 1:
+            raise ValueError('Concurrent edition change; entire repair must roll back')
+    after = inspect(c)
+    if after['status'] != 'already_joint' or after['content_fingerprints'] != actual['content_fingerprints']:
+        raise ValueError('Postcheck changed protected content; entire repair must roll back')
+    c.execute('INSERT INTO schema_migrations(name) VALUES(?) ON CONFLICT DO NOTHING', (MARKER,))
+    return {**after, 'status':'applied', 'updated_editions':sum(e['from_venue_id'] != e['target_venue_id'] for e in actual['editions'])}
+
+
+def database_identity(c, db):
+    if getattr(c, 'postgres', False):
+        row = c.execute('SELECT current_database(),current_schema()').fetchone()
+        return {'engine':'postgresql','database':row[0],'schema':row[1]}
+    return {'engine':'sqlite','path':str(Path(db).resolve())}
+
+
+def run(db, out, *, pg_service=None, pg_schema=None, plan=None, plan_sha256=None):
+    out = Path(out).resolve()
+    if not out.parent.is_dir():
+        raise ValueError('Report parent directory must already exist')
+    if out.exists():
+        raise ValueError('Preserve existing evidence report; choose a new output path')
+    expected = None
+    if plan:
+        raw = Path(plan).read_bytes()
+        if hashlib.sha256(raw).hexdigest() != plan_sha256:
+            raise ValueError('Repair plan hash differs before opening database')
+        expected = json.loads(raw)
+        validate_plan(expected)
+    elif plan_sha256:
+        raise ValueError('Plan hash requires an existing plan')
+    if db and out == Path(db).resolve():
+        raise ValueError('Report must not overwrite database')
+    writer = open_store(db, pg_service=pg_service, pg_schema=pg_schema) if expected is not None else None
+    c = writer.conn if writer else open_read_connection(db, pg_service=pg_service, pg_schema=pg_schema)
+    with closing(c):
+        try:
+            if expected is not None and not getattr(c, 'postgres', False):
+                c.commit(); c.execute('BEGIN IMMEDIATE')
+            identity = database_identity(c, db)
+            if expected is not None and expected['database'] != identity:
+                raise ValueError('Plan belongs to another database/schema')
+            report = apply_plan(c, expected) if expected is not None else inspect(c)
+            report['database'] = identity
+            if expected is not None:
+                report['plan_sha256'] = plan_sha256
+            if expected is not None:
+                c.commit()
+        except BaseException:
+            if expected is not None: c.rollback()
+            raise
+    out.write_text(json.dumps(report,ensure_ascii=False,indent=2)+'\n')
+    print(json.dumps({'status':report['status'],'editions':len(report['editions']),
+                      'updated_editions':report.get('updated_editions',0),'report':str(out)}))
+    return report
+
+
+def main(argv=None, *, prog=None):
+    parser = argparse.ArgumentParser(prog=prog); database_arguments(parser)
+    parser.add_argument('--out', required=True)
+    parser.add_argument('--plan', help='Apply an already reviewed plan; omitted means read-only preflight')
+    parser.add_argument('--plan-sha256', help='Required exact plan SHA256 when applying')
+    args = parser.parse_args(argv)
+    run(args.db,args.out,pg_service=args.pg_service,pg_schema=args.pg_schema,plan=args.plan,plan_sha256=args.plan_sha256)

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 []
@@ -94,9 +108,11 @@ def ingest_file(db,path, *, pg_service=None, pg_schema=None):
                 row = json.loads(line)
                 if row['source'] != source_name:
                     raise ValueError('Source differs from manifest')
-                if any(key in manifest and manifest[key]!=row[key] for key in ('venue','year','track')):
+                if any(key in manifest and manifest[key]!=row.get(key) for key in ('venue','year','track','event_key')):
                     raise ValueError('Conference edition differs from manifest')
                 require_track_record(row, manifest)
+                from paper_atlas.db.events import validate_event_key
+                validate_event_key(row.get('event_key'))
                 external = row['external_id']
                 if external in seen:
                     raise ValueError('Duplicate external ID in one collected file: '+external)
@@ -117,7 +133,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'),guard_status=source_name in (PUBLIC_SOURCE,ARCHIVE_SOURCE))
+                        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),event_key=row.get('event_key'))
                     editions.add(edition)
                 else:
                     raise ValueError('This importer requires an explicit supported decision status')

paper_atlas/postgres/migration.py

@@ -143,7 +143,7 @@ def migrate(db, service, out, *, schema='research', storage_path=None, min_free_
     free_space(storage_path, min_free_gib)
     control = schema + '_migration'
     report = {'status': 'preflight', 'source': str(db), 'service': service, 'schema': schema,
-              'started_at': timestamp(), 'tables': {}, 'scope': 'All 21 business tables; FTS5 internals rebuilt as native PostgreSQL GIN.'}
+              'started_at': timestamp(), 'tables': {}, 'scope': 'All baseline business tables and recognized installed extensions; FTS5 rebuilt as native PostgreSQL GIN.'}
     with Path(str(db) + '.maintenance.lock').open('a') as lock:
         fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
         if Path(str(db) + '-wal').exists() and Path(str(db) + '-wal').stat().st_size:
@@ -240,7 +240,7 @@ def migrate(db, service, out, *, schema='research', storage_path=None, min_free_
                     JOIN pg_am am ON am.oid=ix.relam WHERE ns.nspname=%s AND ix.relname='papers_search_gin' ''', (stage,)).fetchone()
                 if search != ('gin', True, True, 'papers'):
                     raise ValueError('Native full-text index is not valid and ready')
-                for table in TABLES:
+                for table in plan:
                     pg.execute('ANALYZE ' + qualify(stage, table))
                 invalid = pg.execute('''SELECT count(*) FROM pg_constraint co JOIN pg_class cl ON cl.oid=co.conrelid
                     JOIN pg_namespace ns ON ns.oid=cl.relnamespace WHERE ns.nspname=%s AND NOT co.convalidated''', (stage,)).fetchone()[0]

paper_atlas/postgres/schema.py

@@ -8,7 +8,9 @@ TABLES = ('sources', 'papers', 'identifiers', 'source_records', 'venues', 'editi
           'submission_sources', 'submission_identifiers', 'decision_sources',
           'submission_merges', 'decision_merges', 'schema_migrations', 'tags',
           'paper_tags', 'match_candidates', 'paper_merges', 'ingest_runs')
+OPTIONAL_TABLES = ('annotation_events',)
 CHECKS = {
+    'annotation_events': ["kind IN ('tag','research_type')", 'retracts_id IS NULL OR retracts_id<>id'],
     'venues': ["venue_type IN ('conference','journal','review_platform')"],
     'edition_sources': ["status IN ('pending','running','complete','partial','blocked','unavailable','not_yet_published')"],
     'decisions': ["status IN ('accepted','rejected','desk_rejected','withdrawn','under_review','unknown')"],
@@ -50,10 +52,10 @@ def quote(name):
 def inspect(connection):
     business = {name for schema, name, kind, *_ in connection.execute('PRAGMA table_list')
                 if schema == 'main' and kind == 'table' and not name.startswith('sqlite_')}
-    if business != set(TABLES):
+    if not set(TABLES) <= business or business - set(TABLES) - set(OPTIONAL_TABLES):
         raise ValueError('Unexpected business tables: ' + repr(sorted(business ^ set(TABLES))))
     result = {}
-    for table in TABLES:
+    for table in (*TABLES, *(name for name in OPTIONAL_TABLES if name in business)):
         columns = [{'name': r[1], 'type': r[2].upper(), 'notnull': bool(r[3]),
                     'default': r[4], 'pk': r[5]}
                    for r in connection.execute('PRAGMA table_info(' + quote(table) + ')')]

paper_atlas/postgres/writer.py

@@ -85,3 +85,5 @@ class WriteConnection:
 class PostgresStore(Store):
     def __init__(self, service, schema=None):
         self.conn = WriteConnection(service, schema)
+        from paper_atlas.db.events import has_event_key
+        self.event_identity = has_event_key(self.conn)

paper_atlas/querying.py

@@ -14,7 +14,9 @@ ACCEPTED_PAPERS = "SELECT s.paper_id FROM submissions s JOIN decisions d ON d.su
 
 
 def known_acceptances(conn,paper):
-    return [dict(row) for row in conn.execute('''SELECT s.id AS submission_id,v.acronym AS venue,e.year,e.track,
+    from paper_atlas.db.events import has_event_key
+    event = 'e.event_key' if has_event_key(conn) else "''"
+    return [dict(row) for row in conn.execute('''SELECT s.id AS submission_id,v.acronym AS venue,e.year,e.track,'''+event+''' AS event_key,
         d.id AS decision_id,d.decision_date,d.observed_at,d.evidence_url
         FROM submissions s JOIN editions e ON e.id=s.edition_id JOIN venues v ON v.id=e.venue_id
         JOIN decisions d ON d.submission_id=s.id WHERE s.paper_id=? AND d.status='accepted' AND '''+LATEST_DECISION+
@@ -29,10 +31,13 @@ def main(argv=None, *, prog=None):
     parser.add_argument('--pg-schema', help='Optional PostgreSQL schema override; otherwise use the configured search_path')
     parser.add_argument('--query', help='Title/abstract search: FTS5 for SQLite, web-search syntax for PostgreSQL')
     parser.add_argument('--paper',type=int,help='Paper ID; include sources, tags and submission history')
+    parser.add_argument('--research-type', help='Independent type filter as namespace=value; no type inferred from track')
+    parser.add_argument('--annotation-history', action='store_true', help='Include retraction evidence for --paper')
     parser.add_argument('--tag', help='Filter as namespace=value, e.g. arxiv:category=cs.AI')
     parser.add_argument('--venue', help='Conference acronym')
     parser.add_argument('--year', type=int)
     parser.add_argument('--track',help='Exact track, e.g. main, findings, workshop')
+    parser.add_argument('--event-key',help='Exact event within a venue/year/track')
     parser.add_argument('--status', choices=['accepted','rejected','desk_rejected','withdrawn','under_review','unknown'])
     parser.add_argument('--outcome',choices=['accepted','no_known_acceptance'],help='跨会议的已知录用情况；未查到录用不等于从未录用')
     parser.add_argument('--limit', type=int, default=20)
@@ -47,6 +52,13 @@ def main(argv=None, *, prog=None):
         conn = sqlite3.connect(Path(args.db).resolve().as_uri()+'?mode=ro', uri=True)
         conn.row_factory = sqlite3.Row
         conn.execute('BEGIN')  # One consistent read snapshot while ingestion continues.
+    from paper_atlas.db.events import has_event_key
+    event = 'e.event_key' if has_event_key(conn) else "''"
+    if args.event_key is not None and event == "''":
+        conn.close()
+        raise ValueError('Event identity migration required for event-key queries')
+    from paper_atlas.curation.annotations import available, observations, ACTIVE
+    has_annotations = available(conn)
     if args.stats:
         result = {name:conn.execute('SELECT count(*) FROM '+name).fetchone()[0] for name in ('papers','source_records','submissions','decisions','tags','match_candidates')}
         result['records_by_source'] = [dict(r) for r in conn.execute('SELECT s.name,count(r.id) AS imported_records FROM sources s LEFT JOIN source_records r ON r.source_id=s.id GROUP BY s.id ORDER BY s.name')]
@@ -69,17 +81,35 @@ def main(argv=None, *, prog=None):
                 where.append('p.id IN (SELECT rowid FROM papers_fts WHERE papers_fts MATCH ?)')
             params.append(args.query)
         if args.tag:
+            if '=' not in args.tag:
+                parser.error('--tag requires namespace=value')
             namespace,value = args.tag.split('=',1)
-            where.append('EXISTS(SELECT 1 FROM paper_tags pt JOIN tags t ON t.id=pt.tag_id WHERE pt.paper_id=p.id AND t.namespace=? AND t.value=?)'); params.extend([namespace,value])
+            platform = 'EXISTS(SELECT 1 FROM paper_tags pt JOIN tags t ON t.id=pt.tag_id WHERE pt.paper_id=p.id AND t.namespace=? AND t.value=?)'
+            params.extend([namespace,value])
+            if has_annotations:
+                platform = '(' + platform + ' OR EXISTS(SELECT 1 FROM annotation_events a WHERE a.paper_id=p.id AND a.kind=\'tag\' AND a.namespace=? AND a.value=? AND ' + ACTIVE + '))'
+                params.extend([namespace,value])
+            where.append(platform)
+        if args.research_type:
+            if '=' not in args.research_type:
+                parser.error('--research-type requires namespace=value')
+            namespace,value = args.research_type.split('=',1)
+            if has_annotations:
+                where.append('EXISTS(SELECT 1 FROM annotation_events a WHERE a.paper_id=p.id AND a.kind=\'research_type\' AND a.namespace=? AND a.value=? AND ' + ACTIVE + ')')
+                params.extend([namespace,value])
+            else:
+                where.append('1=0')
         if args.outcome:
             where.append('p.id '+('IN' if args.outcome=='accepted' else 'NOT IN')+' ('+ACCEPTED_PAPERS+')')
-        if args.venue or args.year or args.status or args.track:
+        if args.venue or args.year or args.status or args.track or args.event_key is not None:
             clause = ['1=1']
             if args.venue:
                 if conn.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='venue_aliases'").fetchone():
                     clause.append('(lower(v.acronym)=lower(?) OR EXISTS(SELECT 1 FROM venue_aliases va WHERE va.venue_id=v.id AND lower(va.alias)=lower(?)))'); params.extend([args.venue,args.venue])
                 else:
                     clause.append('lower(v.acronym)=lower(?)');params.append(args.venue)
+            if args.event_key is not None:
+                clause.append('e.event_key=?');params.append(args.event_key)
             if args.track:
                 clause.append('e.track=?');params.append(args.track)
             if args.year:
@@ -98,10 +128,14 @@ def main(argv=None, *, prog=None):
             paper['known_acceptances']=known_acceptances(conn,row['id'])
             paper['publication_outcome']='accepted' if paper['known_acceptances'] else 'no_known_acceptance'
             if args.paper is not None:
+                paper['research_types'] = observations(conn, row['id'], kind='research_type')
+                paper['annotation_tags'] = observations(conn, row['id'], kind='tag')
+                if args.annotation_history:
+                    paper['annotation_history'] = observations(conn, row['id'], history=True)
                 paper['tags'] = [dict(r) for r in conn.execute('SELECT t.namespace,t.value,pt.evidence_key FROM paper_tags pt JOIN tags t ON t.id=pt.tag_id WHERE pt.paper_id=?',(row['id'],))]
                 paper['sources'] = [dict(r) for r in conn.execute('SELECT s.name,r.external_id,r.url,r.title,r.abstract,r.retrieved_at FROM source_records r JOIN sources s ON s.id=r.source_id WHERE r.paper_id=?',(row['id'],))]
                 submissions = []
-                for sub in conn.execute('SELECT s.id,s.source_submission_id,s.submitted_at,v.acronym,e.year,e.track FROM submissions s JOIN editions e ON e.id=s.edition_id JOIN venues v ON v.id=e.venue_id WHERE s.paper_id=? ORDER BY e.year,s.id',(row['id'],)):
+                for sub in conn.execute('SELECT s.id,s.source_submission_id,s.submitted_at,v.acronym,e.year,e.track,'+event+' AS event_key FROM submissions s JOIN editions e ON e.id=s.edition_id JOIN venues v ON v.id=e.venue_id WHERE s.paper_id=? ORDER BY e.year,s.id',(row['id'],)):
                     submission = dict(sub)
                     submission['decisions'] = []
                     for d in conn.execute('SELECT id,status,raw_value,evidence_url,observed_at,decision_date FROM decisions WHERE submission_id=? ORDER BY observed_at,COALESCE(decision_date,observed_at),id',(sub['id'],)):

pyproject.toml

@@ -21,5 +21,5 @@ include = ["paper_atlas*"]
 exclude = ["tests*"]
 
 [tool.setuptools.package-data]
-"paper_atlas.db" = ["schema.sql"]
+"paper_atlas.db" = ["schema.sql", "annotations.sql"]
 "paper_atlas.resources" = ["venues.json"]

tests/test_annotations.py

@@ -0,0 +1,227 @@
+from contextlib import closing, redirect_stdout
+import io
+import json
+import os
+from pathlib import Path
+import tempfile
+import unittest
+import uuid
+
+from paper_atlas import querying
+from paper_atlas.curation.annotations import (assert_annotation, available, initialize, main,
+                                             observations, retract_annotation)
+from paper_atlas.db.backend import open_store
+from paper_atlas.db.store import Store
+from paper_atlas.postgres import schema
+from paper_atlas.postgres.client import connect
+from paper_atlas.postgres.migration import migrate
+
+
+class AnnotationTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory(); self.addCleanup(self.tmp.cleanup)
+        self.db = Path(self.tmp.name) / 'sample.sqlite'
+        self.store = Store(self.db); self.addCleanup(self.store.conn.close)
+        self.c = self.store.conn
+        self.paper, self.record = self.store.record('fixture', 'one', 'https://example.org/one', 'First research', None, [])
+        self.other, self.other_record = self.store.record('fixture', 'two', 'https://example.org/two', 'Other research', None, [])
+        self.store.record_tags(self.paper, self.record, 'fixture', [{'namespace':'fixture:topic','value':'original'}])
+        self.c.commit()
+
+    def add(self, **changes):
+        args = dict(paper=self.paper, kind='tag', namespace='custom:topic', value='reading',
+                    request_id='request-one', evidence_key='manual:reading', reason='Reviewed metadata', actor='reviewer')
+        args.update(changes)
+        return assert_annotation(self.c, **args)
+
+    def query(self, *args):
+        self.c.commit()
+        with redirect_stdout(io.StringIO()) as output:
+            querying.main(['--db', str(self.db), *map(str, args)])
+        return json.loads(output.getvalue())
+
+    def test_read_does_not_install_extension_and_write_requires_explicit_init(self):
+        self.assertFalse(available(self.c))
+        self.assertEqual(observations(self.c, self.paper), [])
+        with self.assertRaisesRegex(RuntimeError, 'init explicitly'):
+            self.add()
+        self.assertEqual(self.query('--research-type', 'custom:method=empirical'), [])
+        self.assertFalse(available(self.c))
+        self.assertEqual(len(schema.inspect(self.c)), 21)
+
+    def test_schema_upgrade_is_transactional_and_marker_mismatch_fails_closed(self):
+        self.assertTrue(initialize(self.c)); self.c.rollback()
+        self.assertFalse(available(self.c))
+        self.c.execute("INSERT INTO schema_migrations(name) VALUES('annotation_events_v1')")
+        with self.assertRaisesRegex(RuntimeError, 'mismatch'):
+            initialize(self.c)
+        self.c.rollback()
+        self.assertTrue(initialize(self.c)); self.assertFalse(initialize(self.c))
+        self.assertEqual(len(schema.inspect(self.c)), 22)
+
+    def test_unmarked_table_cannot_receive_assertions(self):
+        initialize(self.c)
+        self.c.execute("DELETE FROM schema_migrations WHERE name='annotation_events_v1'")
+        with self.assertRaisesRegex(RuntimeError, 'init explicitly'):
+            self.add()
+        self.assertEqual(self.c.execute('SELECT count(*) FROM annotation_events').fetchone()[0], 0)
+
+    def test_types_tags_and_platform_labels_are_independent_and_queryable(self):
+        initialize(self.c)
+        self.add()
+        self.add(kind='research_type', namespace='custom:topic', request_id='type', value='reading')
+        self.assertEqual(len(self.query('--tag', 'custom:topic=reading')), 1)
+        self.assertEqual(len(self.query('--research-type', 'custom:topic=reading')), 1)
+        self.assertEqual(len(self.query('--tag', 'fixture:topic=original')), 1)
+        detail = self.query('--paper', self.paper)[0]
+        self.assertEqual(len(detail['annotation_tags']), 1)
+        self.assertEqual(len(detail['research_types']), 1)
+        self.assertNotIn('annotation_history', detail)
+        self.assertEqual(detail['tags'][0]['namespace'], 'fixture:topic')
+
+    def test_repeated_request_is_idempotent_and_changed_content_fails(self):
+        initialize(self.c); event = self.add()
+        self.assertEqual(self.add(), event)
+        for change in ({'value':'different'}, {'kind':'research_type'}, {'reason':'changed'}, {'paper':self.other}):
+            with self.assertRaisesRegex(ValueError, 'different annotation'):
+                self.add(**change)
+        self.assertEqual(self.c.execute('SELECT count(*) FROM annotation_events').fetchone()[0], 1)
+
+    def test_retraction_keeps_original_and_other_assertions_of_same_label(self):
+        initialize(self.c); first = self.add(); second = self.add(request_id='second', evidence_key='independent')
+        kwargs = dict(request_id='retract', reason='No longer relevant', actor='reviewer')
+        removed = retract_annotation(self.c, first, **kwargs)
+        self.assertEqual(retract_annotation(self.c, first, **kwargs), removed)
+        self.assertEqual([r['id'] for r in observations(self.c, self.paper)], [second])
+        self.assertEqual(len(self.query('--paper', self.paper, '--annotation-history')[0]['annotation_history']), 3)
+        with self.assertRaisesRegex(ValueError, 'already retracted'):
+            retract_annotation(self.c, first, **{**kwargs,'request_id':'duplicate'})
+        with self.assertRaisesRegex(ValueError, 'existing assertion'):
+            retract_annotation(self.c, removed, **{**kwargs,'request_id':'nested'})
+
+    def test_source_assertions_require_matching_paper_and_platform(self):
+        initialize(self.c)
+        for change in ({'namespace':'fixture:type'}, {'namespace':'other:type','source_record_id':self.record},
+                       {'namespace':'fixture:type','source_record_id':self.other_record}):
+            with self.assertRaises(ValueError):
+                self.add(**change)
+        event = self.add(kind='research_type', namespace='fixture:declared_type', source_record_id=self.record)
+        with self.assertRaisesRegex(ValueError, 'Manual CLI'):
+            retract_annotation(self.c, event, request_id='cannot', reason='x', actor='y', custom_only=True)
+        self.assertEqual(observations(self.c, self.paper)[0]['source_record_id'], self.record)
+
+    def test_platform_tag_refresh_preserves_both_custom_kinds(self):
+        initialize(self.c); self.add(); self.add(kind='research_type', request_id='type')
+        self.store.record_tags(self.paper, self.record, 'fixture', [{'namespace':'fixture:topic','value':'refreshed'}])
+        self.assertEqual(len(observations(self.c, self.paper)), 2)
+        self.assertEqual(len(self.query('--tag','fixture:topic=refreshed')), 1)
+        self.assertEqual(self.query('--tag','fixture:topic=original'), [])
+
+    def test_merge_preserves_retractions_evidence_and_original_paper_ids(self):
+        initialize(self.c); event = self.add(source_record_id=self.record)
+        removed = retract_annotation(self.c, event, request_id='withdraw', reason='corrected', actor='reviewer')
+        active = self.add(request_id='active', kind='research_type', value='survey')
+        self.store.merge(self.other, self.paper, 'Independently reviewed identity')
+        rows = observations(self.c, self.other, history=True)
+        self.assertEqual([r['id'] for r in rows], [event, removed, active])
+        self.assertEqual({r['original_paper_id'] for r in rows}, {self.paper})
+        self.assertEqual({r['paper_id'] for r in rows}, {self.other})
+        self.assertEqual([r['id'] for r in observations(self.c, self.other)], [active])
+        self.assertEqual(self.add(source_record_id=self.record), event)  # exact retry survives canonical merge
+        self.assertEqual(self.c.execute('PRAGMA foreign_key_check').fetchall(), [])
+
+    def test_invalid_input_does_not_create_events(self):
+        initialize(self.c)
+        for change in ({'paper':True}, {'paper':999999}, {'kind':'main'}, {'namespace':'arxiv'},
+                       {'namespace':'custom:topic;DROP'}, {'value':' '}, {'reason':''},
+                       {'actor':''}, {'request_id':''}, {'evidence_key':''}, {'source_record_id':True},
+                       {'evidence_url':'file:///etc/passwd'}, {'evidence_url':'https://user:secret@example.org/'}):
+            with self.subTest(change=change), self.assertRaises(ValueError):
+                self.add(**change)
+        self.assertEqual(self.c.execute('SELECT count(*) FROM annotation_events').fetchone()[0], 0)
+
+    def test_cli_manual_scope_and_complete_assert_retract_lifecycle(self):
+        with redirect_stdout(io.StringIO()):
+            main(['init','--db',str(self.db)])
+            main(['assert','--db',str(self.db),'--paper',str(self.paper),'--kind','research_type',
+                  '--namespace','custom:method','--value','empirical','--evidence-key','manual:review',
+                  '--request-id','cli-one','--reason','Read abstract','--actor','reviewer'])
+        self.assertEqual(len(self.query('--research-type','custom:method=empirical')), 1)
+        event = observations(self.c, self.paper)[0]['id']
+        with redirect_stdout(io.StringIO()):
+            main(['retract','--db',str(self.db),'--event',str(event),'--request-id','cli-two',
+                  '--reason','Corrected classification','--actor','reviewer'])
+        self.assertEqual(self.query('--research-type','custom:method=empirical'), [])
+        with redirect_stdout(io.StringIO()) as out:
+            main(['list','--db',str(self.db),'--paper',str(self.paper),'--history'])
+        self.assertEqual(len(json.loads(out.getvalue())), 2)
+
+
+@unittest.skipUnless(os.environ.get('PAPER_ATLAS_TEST_PG_SERVICE'), 'Requires approved remote PostgreSQL service')
+class PostgreSQLAnnotationTests(unittest.TestCase):
+    def setUp(self):
+        self.service = os.environ['PAPER_ATLAS_TEST_PG_SERVICE']
+        self.target = 'pa_anno_' + uuid.uuid4().hex[:12]
+        self.tmp = tempfile.TemporaryDirectory(); self.addCleanup(self.tmp.cleanup)
+        self.db = Path(self.tmp.name) / 'source.sqlite'
+        self.s = Store(self.db)
+        self.paper, self.record = self.s.record('fixture','one','https://example.org','Paper',None,[])
+        self.s.conn.commit()
+        self.addCleanup(self.s.conn.close); self.addCleanup(self.cleanup)
+
+    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.target,)):
+                c.execute('DROP SCHEMA ' + schema.quote(name) + ' CASCADE')
+
+    def migrate(self):
+        self.s.conn.commit()
+        self.s.conn.close()  # Freeze/checkpoint the fixture, just like a real source migration.
+        with redirect_stdout(io.StringIO()):
+            return migrate(self.db, self.service, Path(self.tmp.name)/'report.json', schema=self.target, min_free_gib=0)
+
+    def args(self, **changes):
+        a = dict(paper=self.paper,kind='research_type',namespace='custom:method',value='survey\x00and\x01',
+                 request_id='native',evidence_key='manual',reason='Reviewed\x00evidence',actor='reviewer')
+        a.update(changes); return a
+
+    def test_native_schema_upgrade_roundtrip_retraction_and_queries(self):
+        self.migrate()
+        with closing(open_store(pg_service=self.service, pg_schema=self.target).conn) as c:
+            self.assertTrue(initialize(c)); self.assertFalse(initialize(c))
+            event = assert_annotation(c, **self.args()); c.commit()
+            self.assertEqual(assert_annotation(c, **self.args()), event)
+            self.assertEqual(observations(c,self.paper)[0]['value'], 'survey\x00and\x01')
+            c.commit()
+        with redirect_stdout(io.StringIO()) as out:
+            querying.main(['--pg-service',self.service,'--pg-schema',self.target,'--paper',str(self.paper)])
+        self.assertEqual(json.loads(out.getvalue())[0]['research_types'][0]['id'], event)
+        with closing(open_store(pg_service=self.service,pg_schema=self.target).conn) as c:
+            retract_annotation(c,event,request_id='removed',reason='Changed',actor='reviewer');c.commit()
+            self.assertEqual(observations(c,self.paper), [])
+            self.assertEqual(len(observations(c,self.paper,history=True)), 2)
+
+    def test_initialized_sqlite_extension_migrates_all_rows_hashes_and_identity(self):
+        initialize(self.s.conn)
+        event = assert_annotation(self.s.conn, **self.args())
+        retract_annotation(self.s.conn,event,request_id='retracted',reason='Corrected',actor='reviewer')
+        report = self.migrate()
+        self.assertEqual(len(report['tables']),22)
+        self.assertEqual(report['tables']['annotation_events']['source_rows'],2)
+        self.assertEqual(report['tables']['annotation_events']['source_sha256'],report['tables']['annotation_events']['target_sha256'])
+        with closing(open_store(pg_service=self.service,pg_schema=self.target).conn) as c:
+            new = assert_annotation(c, **self.args(request_id='after-migration'))
+            self.assertGreater(new,2); c.commit()
+            self.assertEqual(len(observations(c,self.paper,history=True)),3)
+
+    def test_native_upgrade_rollback_keeps_legacy_schema_usable(self):
+        self.migrate()
+        with closing(open_store(pg_service=self.service,pg_schema=self.target).conn) as c:
+            initialize(c); c.rollback()
+            self.assertFalse(available(c))
+            self.assertEqual(c.execute('SELECT count(*) FROM papers').fetchone()[0],1)
+            self.assertTrue(initialize(c))
+            with self.assertRaises(ValueError):
+                assert_annotation(c, **self.args(namespace='wrong:type',source_record_id=self.record))
+            self.assertEqual(c.execute('SELECT count(*) FROM annotation_events').fetchone()[0],0)
+            c.rollback()

tests/test_event_identity.py

@@ -0,0 +1,202 @@
+from contextlib import closing, redirect_stdout, contextmanager
+import io
+import json
+import os
+from pathlib import Path
+import sqlite3
+import tempfile
+import unittest
+import uuid
+from importlib.resources import files
+
+from paper_atlas.db.store import Store
+from paper_atlas.db.backend import open_store
+from paper_atlas.db.events import has_event_key
+from paper_atlas.collectors.dblp import XML_SHA
+from paper_atlas.ingestion.event_identity import plan, upgrade, exact_counts
+from paper_atlas.ingestion.records import ingest_file
+from paper_atlas.ingestion.backfill import backfill
+from paper_atlas.postgres.client import connect
+from paper_atlas.postgres.schema import quote
+from paper_atlas.postgres.migration import migrate
+from paper_atlas import querying
+
+
+@contextmanager
+def opened_store(**kwargs):
+    store=open_store(**kwargs)
+    try:
+        yield store
+    finally:
+        store.conn.close()
+
+
+def legacy(db):
+    sql = files('paper_atlas.db').joinpath('schema.sql').read_text().replace(" event_key TEXT NOT NULL DEFAULT '',\n", '').replace('UNIQUE(venue_id,year,track,event_key)', 'UNIQUE(venue_id,year,track)')
+    c = sqlite3.connect(db); c.executescript(sql); c.close()
+    s = Store(db)
+    p,r = s.record('dblp','conf/www/Fall95','https://example.org/fall','Same paper in two events',
+                    'Abstract\x00with\x01text',['A Person'],{'arxiv':'2501.00001'})
+    s.record_tags(p,r,'dblp',[{'namespace':'dblp:booktitle','value':'WWW (Fall)'}])
+    s.decision(p,r,'Legacy',1995,'main','conf/www/Fall95','https://example.org/accept','accepted','Accept')
+    s.conn.execute("UPDATE venues SET acronym='WWW',name='WWW'")
+    meta={'snapshot_sha256':XML_SHA,'proceedings_key':'conf/www/1995f','bibliographic_year':1995,
+          'bibliography':{'booktitle':['WWW (Fall)'],'url':['db/conf/www/www1995f.html#Fall95']}}
+    s.conn.execute('UPDATE source_records SET metadata_json=?',(json.dumps(meta),));s.conn.commit();s.conn.close()
+    return p,r
+
+
+def add(store, external, event=None, status='accepted', venue='WWW', year=1995):
+    p,r = store.record('fixture',external,'https://example.org/'+external,'Same paper in two events',
+                       'Abstract\x00with\x01text',['A Person'],{'arxiv':'2501.00001'})
+    edition = store.decision(p,r,venue,year,'main',external,'https://example.org/'+status,status,status,event_key=event)
+    return p,r,edition
+
+
+def file(root, name, rows, event=None):
+    p=root/(name+'.jsonl');p.write_text(''.join(json.dumps(r)+'\n'for r in rows))
+    manifest={'source':'fixture','venue':'WWW','year':1995,'track':'main','count':len(rows),'list_url':'https://example.org/'+name}
+    if event is not None:manifest['event_key']=event
+    p.with_suffix('.manifest.json').write_text(json.dumps(manifest));return p
+
+
+class EventIdentityTests(unittest.TestCase):
+    def setUp(self):
+        t=tempfile.TemporaryDirectory();self.addCleanup(t.cleanup);self.root=Path(t.name);self.db=self.root/'sample.sqlite'
+
+    def test_same_paper_distinct_events_keep_both_attempts_and_decisions(self):
+        s=Store(self.db);self.addCleanup(s.conn.close)
+        paper,_,spring=add(s,'spring','www-1995-spring','rejected');_,_,fall=add(s,'fall','www-1995-fall')
+        s.conn.commit();self.assertNotEqual(spring,fall)
+        with redirect_stdout(io.StringIO())as out:querying.main(['--db',str(self.db),'--paper',str(paper)])
+        row=json.loads(out.getvalue())[0];self.assertEqual({a['event_key']for a in row['submissions']},{'www-1995-spring','www-1995-fall'})
+        self.assertEqual([a['event_key']for a in row['known_acceptances']],['www-1995-fall'])
+        self.assertEqual({a['latest_status']for a in row['submissions']},{'rejected','accepted'})
+        with redirect_stdout(io.StringIO())as out:querying.main(['--db',str(self.db),'--venue','WWW','--event-key','www-1995-spring','--status','rejected'])
+        self.assertEqual(len(json.loads(out.getvalue())),1)
+
+    def test_explicit_migration_preserves_ids_history_and_source_bytes(self):
+        paper,record=legacy(self.db);proposal=plan(self.db);self.assertFalse(proposal['before']['event_column'])
+        result=upgrade(proposal,self.db);self.assertEqual(result['updated_editions'],1)
+        self.assertEqual(result['after']['protected'],proposal['before']['protected'])
+        self.assertEqual(result['after']['editions_sha256'],proposal['before']['editions_sha256'])
+        self.assertEqual(upgrade(proposal,self.db)['status'],'already_applied')
+        s=Store(self.db);self.addCleanup(s.conn.close)
+        self.assertEqual(s.conn.execute('SELECT id,event_key FROM editions').fetchall(),[(1,'www-1995-fall')])
+        # An old file without event_key still has its original source anchor.
+        self.assertEqual(s.decision(paper,record,'WWW',1995,'main','conf/www/Fall95','https://example.org/accept','accepted','Accept'),1)
+        self.assertEqual(s.conn.execute('SELECT count(*) FROM submissions').fetchone()[0],1)
+        self.assertEqual(s.conn.execute('PRAGMA foreign_key_check').fetchall(),[])
+
+    def test_unverified_legacy_volume_is_not_automatically_labelled(self):
+        legacy(self.db)
+        with sqlite3.connect(self.db)as c:c.execute("UPDATE source_records SET metadata_json='{}'")
+        with self.assertRaisesRegex(ValueError,'complete pinned Fall'):plan(self.db)
+        with sqlite3.connect(self.db)as c:self.assertFalse(has_event_key(c))
+
+    def test_stale_plan_rolls_back_before_schema_change(self):
+        legacy(self.db);proposal=plan(self.db)
+        with sqlite3.connect(self.db)as c:c.execute("UPDATE editions SET track='unspecified'")
+        with self.assertRaisesRegex(ValueError,'stale'):upgrade(proposal,self.db)
+        with sqlite3.connect(self.db)as c:self.assertFalse(has_event_key(c));self.assertEqual(c.execute('SELECT track FROM editions').fetchone()[0],'unspecified')
+
+    def test_unscoped_source_and_wrong_refresh_event_fail(self):
+        s=Store(self.db);self.addCleanup(s.conn.close);_,r,e=add(s,'first','www-1995-fall');s.conn.commit()
+        p,r2=s.record('fixture','unscoped','https://example.org/u','Other work',None,[])
+        with self.assertRaisesRegex(ValueError,'event_key'):s.decision(p,r2,'WWW',1995,'main','unscoped','https://example.org/u','accepted','Accept')
+        with self.assertRaisesRegex(ValueError,'Cannot move'):s.decision(p,r,'WWW',1995,'main','first','https://example.org/u','accepted','Accept',event_key='www-1995-spring')
+        self.assertEqual(s.conn.execute('SELECT count(*) FROM editions').fetchone()[0],1)
+
+    def test_pinned_fall_cannot_claim_spring(self):
+        p,r=legacy(self.db);upgrade(plan(self.db),self.db);s=Store(self.db);self.addCleanup(s.conn.close)
+        with self.assertRaisesRegex(ValueError,'pinned WWW Fall'):s.decision(p,r,'WWW',1995,'main','conf/www/Fall95','https://example.org/u','accepted','Accept',event_key='www-1995-spring')
+
+    def test_manifest_event_mismatch_rolls_back_the_whole_file(self):
+        s=Store(self.db);s.conn.close()
+        base={'source':'fixture','venue':'WWW','year':1995,'track':'main','status':'accepted','title':'Input work','url':'https://example.org/p','authors':[]}
+        path=file(self.root,'bad',[dict(base,external_id='a',event_key='www-1995-spring'),dict(base,external_id='b',event_key='www-1995-fall')],'www-1995-spring')
+        with self.assertRaisesRegex(ValueError,'edition differs'):ingest_file(self.db,path)
+        with sqlite3.connect(self.db)as c:
+            self.assertEqual(c.execute('SELECT count(*) FROM papers').fetchone()[0],0)
+            self.assertEqual(c.execute('SELECT count(*) FROM editions').fetchone()[0],0)
+
+    def test_backfill_selects_the_explicit_event_and_preserves_attempts(self):
+        s=Store(self.db);add(s,'spring','www-1995-spring','rejected');add(s,'fall','www-1995-fall');s.conn.commit();s.conn.close()
+        row={'source':'fixture','external_id':'fall','venue':'WWW','year':1995,'track':'main','event_key':'www-1995-fall','status':'accepted','identifiers':{'openreview':'forumFall'}}
+        path=file(self.root,'backfill',[row],'www-1995-fall')
+        with redirect_stdout(io.StringIO()):result=backfill(self.db,[path])
+        self.assertEqual(result['submissions_after'],2)
+        with sqlite3.connect(self.db)as c:
+            self.assertEqual(c.execute("SELECT e.event_key FROM submission_identifiers si JOIN submissions s ON s.id=si.submission_id JOIN editions e ON e.id=s.edition_id WHERE si.scheme='openreview' AND si.value='forumFall'").fetchone()[0],'www-1995-fall')
+
+    def test_exact_paged_counts_cover_composite_key_boundaries(self):
+        s=Store(self.db);self.addCleanup(s.conn.close)
+        for i in range(7):add(s,'entry'+str(i),'www-1995-fall')
+        s.conn.commit();events=[];counts=exact_counts(s.conn,page_size=2,progress=events.append)
+        self.assertEqual(counts['source_records'],7)
+        self.assertEqual(counts['identifiers'],8)
+        self.assertEqual(counts['submission_sources'],7)
+        self.assertEqual([e['rows_counted']for e in events if e['table']=='identifiers'and e['complete']],[8])
+
+    def test_legacy_single_event_writes_and_reads_remain_available(self):
+        legacy(self.db);s=Store(self.db);self.addCleanup(s.conn.close)
+        add(s,'ordinary',venue='COLM',year=2025);s.conn.commit()
+        with redirect_stdout(io.StringIO())as out:querying.main(['--db',str(self.db),'--venue','COLM'])
+        self.assertEqual(json.loads(out.getvalue())[0]['known_acceptances'][-1]['event_key'],'')
+        with self.assertRaisesRegex(ValueError,'migration required'):
+            querying.main(['--db',str(self.db),'--event-key','www-1995-fall'])
+
+    def test_unexpected_extra_edition_column_requires_review(self):
+        legacy(self.db)
+        with sqlite3.connect(self.db)as c:c.execute('ALTER TABLE editions ADD COLUMN custom_data TEXT')
+        with self.assertRaisesRegex(ValueError,'Unexpected edition schema'):plan(self.db)
+
+    def test_file_spanning_two_events_without_manifest_key_rolls_back(self):
+        Store(self.db).conn.close()
+        base={'source':'fixture','venue':'WWW','year':1995,'track':'main','status':'accepted','title':'Input work','url':'https://example.org/p','authors':[]}
+        path=file(self.root,'two',[dict(base,external_id='a',event_key='www-1995-spring'),dict(base,external_id='b',event_key='www-1995-fall')])
+        with self.assertRaisesRegex(ValueError,'exactly one'):ingest_file(self.db,path)
+        with sqlite3.connect(self.db)as c:self.assertEqual(c.execute('SELECT count(*) FROM submissions').fetchone()[0],0)
+
+    def test_cutover_sqlite_backup_remains_protected(self):
+        legacy(self.db);proposal=plan(self.db);Path(str(self.db.resolve())+'.postgresql-cutover.json').write_text('{}')
+        with self.assertRaisesRegex(RuntimeError,'cut over'):upgrade(proposal,self.db)
+
+
+@unittest.skipUnless(os.environ.get('PAPER_ATLAS_TEST_PG_SERVICE'),'Requires approved remote PostgreSQL test service')
+class PostgreSQLEventIdentityTests(unittest.TestCase):
+    def setUp(self):
+        self.service=os.environ['PAPER_ATLAS_TEST_PG_SERVICE'];self.schema='pa_event_'+uuid.uuid4().hex[:12]
+        t=tempfile.TemporaryDirectory();self.addCleanup(t.cleanup);self.root=Path(t.name);self.db=self.root/'seed.sqlite';legacy(self.db)
+        self.addCleanup(self.cleanup)
+        with redirect_stdout(io.StringIO()):migrate(self.db,self.service,self.root/'migrate.json',schema=self.schema,min_free_gib=0)
+
+    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,)).fetchall():c.execute('DROP SCHEMA '+quote(name)+' CASCADE')
+
+    def test_native_upgrade_preserves_ids_and_codec_then_accepts_separate_event(self):
+        proposal=plan(pg_service=self.service,pg_schema=self.schema)
+        result=upgrade(proposal,pg_service=self.service,pg_schema=self.schema)
+        self.assertEqual(result['after']['protected'],proposal['before']['protected'])
+        self.assertEqual(upgrade(proposal,pg_service=self.service,pg_schema=self.schema)['status'],'already_applied')
+        with opened_store(pg_service=self.service,pg_schema=self.schema)as s:
+            paper,_,spring=add(s,'spring','www-1995-spring','rejected');s.conn.commit()
+            self.assertNotEqual(spring,1)
+            self.assertEqual(s.conn.execute('SELECT abstract FROM papers WHERE id=?',(paper,)).fetchone()[0],'Abstract\x00with\x01text')
+        with redirect_stdout(io.StringIO())as out:querying.main(['--pg-service',self.service,'--pg-schema',self.schema,'--paper',str(paper)])
+        row=json.loads(out.getvalue())[0];self.assertEqual(len(row['submissions']),2);self.assertEqual(len(row['known_acceptances']),1)
+
+    def test_native_paged_counts_preserve_encoded_composite_key_order(self):
+        with opened_store(pg_service=self.service,pg_schema=self.schema)as s:
+            for value in ['a\x00b','a\x01b','a\x02b','z']:
+                s.conn.execute('INSERT INTO identifiers(scheme,value,paper_id) VALUES(?,?,?)',('custom',value,1))
+            s.conn.commit();counts=exact_counts(s.conn,page_size=2)
+            self.assertEqual(counts['identifiers'],6)
+            self.assertEqual(counts['submissions'],1)
+
+    def test_stale_native_plan_is_rejected_and_lock_released(self):
+        proposal=plan(pg_service=self.service,pg_schema=self.schema)
+        with opened_store(pg_service=self.service,pg_schema=self.schema)as s:s.conn.execute("UPDATE editions SET track='unspecified'");s.conn.commit()
+        with self.assertRaisesRegex(ValueError,'stale'):upgrade(proposal,pg_service=self.service,pg_schema=self.schema)
+        with opened_store(pg_service=self.service,pg_schema=self.schema)as s:self.assertFalse(has_event_key(s.conn))

tests/test_joint_editions.py

@@ -0,0 +1,187 @@
+from contextlib import closing, redirect_stdout
+import hashlib
+import io
+import json
+import os
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+import uuid
+
+from paper_atlas.collectors.dblp import XML_SHA
+from paper_atlas.collectors.dblp_scope import scope
+from paper_atlas.collectors.dblp_records import normalize_row
+from paper_atlas.db.store import Store
+from paper_atlas.db.backend import open_store
+from paper_atlas.ingestion.joint_editions import (TARGET, MARKER, inspect, apply_plan, run)
+from paper_atlas.postgres.client import connect
+from paper_atlas.postgres.migration import migrate
+from paper_atlas.postgres.schema import quote
+from paper_atlas import querying
+
+
+def seed(store):
+    c = store.conn
+    c.execute("INSERT INTO venues(acronym,name,field,venue_type) VALUES(?,?,'reasoning','conference')", (TARGET,TARGET))
+    target = c.execute('SELECT id FROM venues WHERE acronym=?',(TARGET,)).fetchone()[0]
+    for alias in ['ECAI','IJCAI']:
+        c.execute('INSERT INTO venue_aliases VALUES(?,?)',(target,alias))
+    for year in [2018,2019,2022]:
+        metadata = {'snapshot_sha256':XML_SHA,'proceedings_key':'conf/ijcai/'+str(year),
+                    'bibliographic_year':year,'bibliography':{'booktitle':['IJCAI']}}
+        p,r=store.record('dblp','conf/ijcai/fixture'+str(year),'https://dblp.org/rec/fixture',
+                         'Original publication '+str(year),'Abstract\x00with\x01data',[],
+                         {'doi':'10.24963/ijcai.'+str(year)+'/1'}, {})
+        store.record_tags(p,r,'dblp',[{'namespace':'dblp:booktitle','value':'IJCAI'}])
+        store.decision(p,r,'IJCAI',year,'main','fixture'+str(year),'https://example.org/decision','accepted','listed')
+        stored=json.loads(c.execute('SELECT metadata_json FROM source_records WHERE id=?',(r,)).fetchone()[0]);stored.update(metadata)
+        c.execute('UPDATE source_records SET metadata_json=? WHERE id=?',(json.dumps(stored),r))
+    c.commit()
+
+
+def plan(c):
+    return {**inspect(c),'database':{'engine':'sqlite','path':'fixture'}}
+
+
+class JointEditionTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp=tempfile.TemporaryDirectory(); self.addCleanup(self.tmp.cleanup)
+        self.db=Path(self.tmp.name)/'sample.sqlite'
+        self.store=Store(self.db); self.addCleanup(self.store.conn.close)
+        seed(self.store); self.c=self.store.conn
+
+    def test_apply_preserves_every_protected_row_and_edition_id(self):
+        before=plan(self.c); ids=[r['id']for r in before['editions']]
+        result=apply_plan(self.c,before);self.c.commit()
+        self.assertEqual(result['updated_editions'],2)
+        self.assertEqual(result['content_fingerprints'],before['content_fingerprints'])
+        self.assertEqual([r['id']for r in result['editions']],ids)
+        untouched=self.c.execute('SELECT v.acronym FROM editions e JOIN venues v ON v.id=e.venue_id WHERE e.year=2019').fetchone()[0]
+        self.assertEqual(untouched,'IJCAI')
+        self.assertEqual(self.c.execute('PRAGMA foreign_key_check').fetchall(),[])
+        self.assertEqual(self.c.execute('SELECT count(*) FROM schema_migrations WHERE name=?',(MARKER,)).fetchone()[0],1)
+        with redirect_stdout(io.StringIO())as out:
+            querying.main(['--db',str(self.db),'--venue','ECAI'])
+        self.assertEqual(len(json.loads(out.getvalue())),2)
+        self.assertEqual(apply_plan(self.c,before)['updated_editions'],0)
+
+    def test_refresh_of_old_normalized_source_does_not_recreate_ordinary_edition(self):
+        before=plan(self.c);apply_plan(self.c,before)
+        for year in [2018,2022]:
+            row=self.c.execute('''SELECT r.paper_id,r.id,d.observed_at FROM source_records r
+                JOIN submissions s ON s.source_record_id=r.id JOIN decisions d ON d.submission_id=s.id
+                WHERE r.external_id=?''',('conf/ijcai/fixture'+str(year),)).fetchone()
+            self.store.decision(row[0],row[1],'IJCAI',year,'main','fixture'+str(year),
+                                'https://example.org/decision','accepted','listed',observed_at=row[2])
+        self.assertEqual(inspect(self.c)['content_fingerprints'],before['content_fingerprints'])
+        self.assertEqual(self.c.execute('SELECT count(*) FROM editions').fetchone()[0],3)
+
+    def test_any_changed_source_text_rejects_stale_plan_before_membership_updates(self):
+        before=plan(self.c)
+        self.c.execute("UPDATE source_records SET title='New evidence' WHERE external_id='conf/ijcai/fixture2018'")
+        with self.assertRaisesRegex(ValueError,'Stale repair plan'):
+            apply_plan(self.c,before)
+        self.assertEqual(inspect(self.c)['status'],'ready')
+
+    def test_conflicting_target_edition_is_never_merged_or_deleted(self):
+        target=self.c.execute('SELECT id FROM venues WHERE acronym=?',(TARGET,)).fetchone()[0]
+        self.c.execute("INSERT INTO editions(venue_id,year,track) VALUES(?,2018,'main')",(target,))
+        with self.assertRaisesRegex(ValueError,'colliding'):
+            inspect(self.c)
+        self.assertEqual(self.c.execute('SELECT count(*) FROM editions WHERE year=2018').fetchone()[0],2)
+
+    def test_wrong_source_volume_and_snapshot_fail_closed(self):
+        original=self.c.execute("SELECT metadata_json FROM source_records WHERE external_id='conf/ijcai/fixture2018'").fetchone()[0]
+        for key,value in [('proceedings_key','conf/ijcai/2018w'),('snapshot_sha256','wrong'),('bibliographic_year',2017),('bibliography',{'booktitle':['Other']})]:
+            m=json.loads(original);m[key]=value
+            self.c.execute("UPDATE source_records SET metadata_json=? WHERE external_id='conf/ijcai/fixture2018'",(json.dumps(m),))
+            with self.subTest(key=key),self.assertRaisesRegex(ValueError,'Source evidence'):
+                inspect(self.c)
+
+    def test_missing_source_support_and_wrong_aliases_are_not_guessed(self):
+        self.c.execute("DELETE FROM venue_aliases WHERE alias='ECAI'")
+        with self.assertRaisesRegex(ValueError,'aliases'):
+            inspect(self.c)
+        self.c.rollback()
+        self.c.execute('DELETE FROM submission_sources WHERE source_record_id=(SELECT id FROM source_records WHERE external_id=?)',('conf/ijcai/fixture2022',))
+        with self.assertRaisesRegex(ValueError,'Every submission'):
+            inspect(self.c)
+
+    def test_cli_plan_readonly_hash_binding_identity_and_repeat_apply(self):
+        out=Path(self.tmp.name)/'plan.json'
+        with redirect_stdout(io.StringIO()):run(self.db,out)
+        self.assertEqual(inspect(self.c)['status'],'ready')
+        raw=out.read_bytes();sha=hashlib.sha256(raw).hexdigest()
+        with patch('paper_atlas.ingestion.joint_editions.open_store')as writer:
+            with self.assertRaisesRegex(ValueError,'hash differs'):
+                run(self.db,Path(self.tmp.name)/'bad.json',plan=out,plan_sha256='0'*64)
+            writer.assert_not_called()
+        with redirect_stdout(io.StringIO()):
+            result=run(self.db,Path(self.tmp.name)/'applied.json',plan=out,plan_sha256=sha)
+            again=run(self.db,Path(self.tmp.name)/'retry.json',plan=out,plan_sha256=sha)
+        self.assertEqual(result['updated_editions'],2);self.assertEqual(again['updated_editions'],0)
+        self.assertEqual(result['plan_sha256'],sha);self.assertEqual(again['plan_sha256'],sha)
+        altered=json.loads(raw);altered['database']['path']='another.sqlite'
+        bad=Path(self.tmp.name)/'wrong-db.json';bad.write_text(json.dumps(altered))
+        with self.assertRaisesRegex(ValueError,'another database'):
+            run(self.db,Path(self.tmp.name)/'wrong-result.json',plan=bad,plan_sha256=hashlib.sha256(bad.read_bytes()).hexdigest())
+
+    def test_invalid_empty_plan_fails_before_opening_writer(self):
+        f=Path(self.tmp.name)/'empty.json';f.write_text('{}')
+        with patch('paper_atlas.ingestion.joint_editions.open_store')as writer:
+            with self.assertRaises(ValueError):
+                run(self.db,Path(self.tmp.name)/'empty-result.json',plan=f,plan_sha256=hashlib.sha256(f.read_bytes()).hexdigest())
+            writer.assert_not_called()
+
+    def test_collector_only_changes_the_two_exact_main_volume_keys(self):
+        for year in [2018,2019,2022]:
+            volume={'key':'conf/ijcai/'+str(year),'fields':{'title':['IJCAI '+str(year)],'year':[str(year)]}}
+            row={'record_type':'inproceedings','key':'conf/ijcai/One'+str(year),'fields':{'booktitle':['IJCAI'],
+                 'year':[str(year)],'crossref':[volume['key']],'title':['Original research title'],
+                 'author':['Jane Doe'],'ee':['https://doi.org/10.24963/ijcai.'+str(year)+'/1']}}
+            result,error=normalize_row(row,{volume['key']:volume},{'IJCAI','ECAI'})
+            self.assertIsNone(error)
+            self.assertEqual(result['venue'],TARGET if year in(2018,2022)else'IJCAI')
+            self.assertEqual(result['metadata']['bibliography']['booktitle'],['IJCAI'])
+            self.assertEqual(result['title'],'Original research title')
+            if year in(2018,2022):
+                self.assertIn('joint_edition_evidence',result['metadata'])
+                row['fields']['booktitle']=['IJCAI (Workshop)']
+                scoped,_=scope(row,volume)
+                self.assertEqual(scoped['venue'],'IJCAI')
+                self.assertEqual(scoped['track'],'workshop')
+
+
+@unittest.skipUnless(os.environ.get('PAPER_ATLAS_TEST_PG_SERVICE'),'Requires approved remote PostgreSQL service')
+class PostgreSQLJointTests(unittest.TestCase):
+    def setUp(self):
+        self.service=os.environ['PAPER_ATLAS_TEST_PG_SERVICE'];self.target='pa_joint_'+uuid.uuid4().hex[:12]
+        self.tmp=tempfile.TemporaryDirectory();self.addCleanup(self.tmp.cleanup)
+        self.db=Path(self.tmp.name)/'sample.sqlite';s=Store(self.db);seed(s);s.conn.close()
+        self.addCleanup(self.cleanup)
+        with redirect_stdout(io.StringIO()):migrate(self.db,self.service,Path(self.tmp.name)/'migration.json',schema=self.target,min_free_gib=0)
+
+    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.target,)).fetchall():
+                c.execute('DROP SCHEMA '+quote(name)+' CASCADE')
+
+    def test_native_repair_is_idempotent_keeps_codec_and_provenance(self):
+        p=Path(self.tmp.name)/'plan.json'
+        with redirect_stdout(io.StringIO()):
+            before=run(None,p,pg_service=self.service,pg_schema=self.target)
+            after=run(None,Path(self.tmp.name)/'applied.json',pg_service=self.service,pg_schema=self.target,plan=p,plan_sha256=hashlib.sha256(p.read_bytes()).hexdigest())
+        self.assertEqual(after['content_fingerprints'],before['content_fingerprints'])
+        with closing(open_store(pg_service=self.service,pg_schema=self.target).conn)as c:
+            self.assertEqual(apply_plan(c,before)['updated_editions'],0)
+            self.assertEqual(c.execute('SELECT abstract FROM papers ORDER BY id LIMIT 1').fetchone()[0],'Abstract\x00with\x01data')
+
+    def test_native_stale_source_plan_preserves_both_original_memberships(self):
+        with closing(open_store(pg_service=self.service,pg_schema=self.target).conn)as c:
+            before=plan(c)
+            c.execute("UPDATE source_records SET title='Changed' WHERE external_id=?",('conf/ijcai/fixture2022',));c.commit()
+            with self.assertRaisesRegex(ValueError,'Stale repair plan'):
+                apply_plan(c,before)
+            c.rollback()
+            self.assertEqual(inspect(c)['status'],'ready')

tests/test_research_runtime.py

@@ -0,0 +1,172 @@
+"""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.audits.coverage import report as coverage_report
+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_coverage_labels_both_same_year_events_and_keeps_legacy_readable(self):
+        registry=self.root/'registry.json'
+        registry.write_text(json.dumps({'scope_note':'Fixture only','venues':[{'acronym':'WWW','field':'web'}]}))
+        def snapshot():
+            return coverage_report(self.target.get('db'),registry,**{k:v for k,v in self.target.items() if k!='db'})['venues'][0]
+        self.assertEqual([e['event_key']for e in snapshot()['editions']],[''])
+        upgrade(plan(**self.target),**self.target)
+        with self.store() as s:
+            add(s,'spring','www-1995-spring','rejected');s.conn.commit()
+        result=snapshot();events={e['event_key']:e for e in result['editions']}
+        self.assertEqual(set(events),{'www-1995-fall','www-1995-spring'})
+        self.assertEqual((result['imported_submissions'],result['distinct_papers']),(2,1))
+        self.assertEqual(events['www-1995-fall']['latest_observed_status_counts']['accepted'],1)
+        self.assertEqual(events['www-1995-spring']['latest_observed_status_counts']['rejected'],1)
+        self.assertEqual(events['www-1995-fall']['source_record_counts'][0]['source'],'dblp')
+        self.assertEqual(events['www-1995-spring']['source_record_counts'][0]['source'],'fixture')
+
+    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')

tests/test_submission_evidence.py

@@ -133,7 +133,7 @@ class SubmissionTests(unittest.TestCase):
         c.execute("INSERT INTO papers(id,title,title_key) VALUES(1,'Paper','paper')")
         c.execute("INSERT INTO source_records VALUES(1,1,'paper',1,'https://example.org/p','Paper','Abstract','[]','{}',CURRENT_TIMESTAMP)")
         c.execute("INSERT INTO venues VALUES(1,'ICLR','ICLR','machine_learning','conference')")
-        c.execute("INSERT INTO editions VALUES(1,1,2026,'main')")
+        c.execute("INSERT INTO editions(id,venue_id,year,track) VALUES(1,1,2026,'main')")
         c.execute("INSERT INTO submissions VALUES(1,1,1,1,'paper',NULL)")
         c.execute("INSERT INTO decisions VALUES(1,1,'accepted','Accept','https://example.org/decision','2026-01-01T00:00:00Z',NULL)")
         c.commit();c.close()