Correct early ACL source candidates and record official index limits

Reviewed-on: #35

.gitea/workflows/check.yml

@@ -48,7 +48,11 @@ jobs:
               outside.mkdir()
               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','collect','www-archive','--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','edition-tracks','--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 +64,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,12 @@ paper-atlas collect colm --cache /data/paper-index/raw/http-cache \
 CI 在已授权的远程 runner 上运行测试、构建 wheel，并从源码目录外验证安装后的命令及随包资源。数据库全量去重和覆盖率是独立的数据验收步骤，不因代码测试通过而自动完成。
 
 开发通过工作分支和 PR 交付，由用户手动合并。
+
+早期WWW档案的可重放元数据采集、论文/海报分轨与待核查条目见 [WWW档案采集](docs/www-archive.md)。
+历史联合会议的既有届次归属可通过 [`maintain joint-editions`](docs/joint-editions.md) 先只读规划，再按计划哈希执行；保留论文与投稿决定来源，不重导入。
+同年多届的 `event_key`、显式迁移、旧文件刷新与查询边界见 [会议事件身份](docs/event-identity.md)。
+研究类型与自定义标签可通过 [`annotate`](docs/annotations.md) 显式初始化、添加和撤回；两者独立查询，保留证据及完整更正历史，不从会议轨道推断研究类型。
+
+整合后的运行版本依赖、交互约束和生产切换顺序见 [运行版本整合](docs/research-runtime.md)。
+
+已导入整卷的逐条轨道修正见 [edition-tracks 维护流程](docs/edition-tracks.md)：按完整证据生成只读计划，保留原投稿和决定 ID，事务性修正 edition、标识与覆盖，并防止旧整卷刷新覆盖分类。

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

@@ -190,3 +190,7 @@ WWW早期逐来源核查见 `www-early-proceedings-2026-11-08.json`：1995春季
 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 覆盖率。
+
+2026年11月9日的[当前历史年度缺口核对](historical-gaps-2026-11-09.md)单列135个历史会议—年份、14个2026年度及85个经范围修正的初步候选来源年度；它使用当前生产edition/别名快照，保留原固定快照不变。年度有记录不等于逐篇、逐track或公开拒稿历史完整。
+
+[早期NLP来源范围复核](historical-source-scope-2026-11-09.md)保留3份官方网页的响应哈希，排除4个COLING卷对ACL缺口的误匹配，并区分会议举办记录、目录可获取性与论文级覆盖；尚未导入或关闭这些年度缺口。

docs/coverage/historical-gaps-2026-11-09.json

@@ -0,0 +1,867 @@
+{
+  "snapshot_at": "2026-11-08 16:34:40.741334+00:00",
+  "scope": "Official first-pass conference-year census vs current populated editions and real aliases. Missing year means no imported attempt in that year, not missing paper count or proof all other years complete. Includes announced/nonarchival exceptions below.",
+  "venues": [
+    {
+      "venue": "NeurIPS",
+      "expected_years": 40,
+      "observed_expected_years": 39,
+      "missing_years": [
+        2026
+      ],
+      "missing_count": 1,
+      "canonical_matches": [
+        "NeurIPS"
+      ]
+    },
+    {
+      "venue": "ICML",
+      "expected_years": 43,
+      "observed_expected_years": 14,
+      "missing_years": [
+        1980,
+        1983,
+        1985,
+        1987,
+        1988,
+        1989,
+        1990,
+        1991,
+        1992,
+        1993,
+        1994,
+        1995,
+        1996,
+        1997,
+        1998,
+        1999,
+        2000,
+        2001,
+        2002,
+        2003,
+        2004,
+        2005,
+        2006,
+        2007,
+        2008,
+        2009,
+        2010,
+        2011,
+        2012
+      ],
+      "missing_count": 29,
+      "canonical_matches": [
+        "ICML"
+      ]
+    },
+    {
+      "venue": "ICLR",
+      "expected_years": 14,
+      "observed_expected_years": 11,
+      "missing_years": [
+        2014,
+        2015,
+        2016
+      ],
+      "missing_count": 3,
+      "canonical_matches": [
+        "ICLR"
+      ]
+    },
+    {
+      "venue": "COLM",
+      "expected_years": 3,
+      "observed_expected_years": 3,
+      "missing_years": [],
+      "missing_count": 0,
+      "canonical_matches": [
+        "COLM"
+      ]
+    },
+    {
+      "venue": "COLT",
+      "expected_years": 39,
+      "observed_expected_years": 39,
+      "missing_years": [],
+      "missing_count": 0,
+      "canonical_matches": [
+        "COLT"
+      ]
+    },
+    {
+      "venue": "AISTATS",
+      "expected_years": 29,
+      "observed_expected_years": 25,
+      "missing_years": [
+        1985,
+        1989,
+        1991,
+        1993
+      ],
+      "missing_count": 4,
+      "canonical_matches": [
+        "AISTATS"
+      ]
+    },
+    {
+      "venue": "UAI",
+      "expected_years": 42,
+      "observed_expected_years": 42,
+      "missing_years": [],
+      "missing_count": 0,
+      "canonical_matches": [
+        "UAI"
+      ]
+    },
+    {
+      "venue": "CLeaR",
+      "expected_years": 5,
+      "observed_expected_years": 5,
+      "missing_years": [],
+      "missing_count": 0,
+      "canonical_matches": [
+        "CLEAR"
+      ]
+    },
+    {
+      "venue": "ACL",
+      "expected_years": 64,
+      "observed_expected_years": 48,
+      "missing_years": [
+        1963,
+        1964,
+        1965,
+        1966,
+        1967,
+        1968,
+        1969,
+        1970,
+        1971,
+        1972,
+        1973,
+        1974,
+        1975,
+        1976,
+        1977,
+        1978
+      ],
+      "missing_count": 16,
+      "canonical_matches": [
+        "ACL",
+        "ACL+COLING",
+        "ACL+EACL"
+      ]
+    },
+    {
+      "venue": "EMNLP",
+      "expected_years": 31,
+      "observed_expected_years": 31,
+      "missing_years": [],
+      "missing_count": 0,
+      "canonical_matches": [
+        "CoNLL+EMNLP",
+        "EMNLP"
+      ]
+    },
+    {
+      "venue": "NAACL",
+      "expected_years": 18,
+      "observed_expected_years": 18,
+      "missing_years": [],
+      "missing_count": 0,
+      "canonical_matches": [
+        "NAACL"
+      ]
+    },
+    {
+      "venue": "EACL",
+      "expected_years": 19,
+      "observed_expected_years": 19,
+      "missing_years": [],
+      "missing_count": 0,
+      "canonical_matches": [
+        "ACL+EACL",
+        "EACL"
+      ]
+    },
+    {
+      "venue": "COLING",
+      "expected_years": 31,
+      "observed_expected_years": 28,
+      "missing_years": [
+        1971,
+        1976,
+        1978
+      ],
+      "missing_count": 3,
+      "canonical_matches": [
+        "ACL+COLING",
+        "COLING",
+        "COLING+LREC"
+      ]
+    },
+    {
+      "venue": "CoNLL",
+      "expected_years": 30,
+      "observed_expected_years": 30,
+      "missing_years": [],
+      "missing_count": 0,
+      "canonical_matches": [
+        "CoNLL",
+        "CoNLL+EMNLP"
+      ]
+    },
+    {
+      "venue": "CVPR",
+      "expected_years": 39,
+      "observed_expected_years": 14,
+      "missing_years": [
+        1983,
+        1985,
+        1986,
+        1988,
+        1989,
+        1991,
+        1992,
+        1993,
+        1994,
+        1996,
+        1997,
+        1998,
+        1999,
+        2000,
+        2001,
+        2003,
+        2004,
+        2005,
+        2006,
+        2007,
+        2008,
+        2009,
+        2010,
+        2011,
+        2012
+      ],
+      "missing_count": 25,
+      "canonical_matches": [
+        "CVPR"
+      ]
+    },
+    {
+      "venue": "ICCV",
+      "expected_years": 20,
+      "observed_expected_years": 7,
+      "missing_years": [
+        1987,
+        1988,
+        1990,
+        1993,
+        1995,
+        1998,
+        1999,
+        2001,
+        2003,
+        2005,
+        2007,
+        2009,
+        2011
+      ],
+      "missing_count": 13,
+      "canonical_matches": [
+        "ICCV"
+      ]
+    },
+    {
+      "venue": "ECCV",
+      "expected_years": 19,
+      "observed_expected_years": 4,
+      "missing_years": [
+        1990,
+        1992,
+        1994,
+        1996,
+        1998,
+        2000,
+        2002,
+        2004,
+        2006,
+        2008,
+        2010,
+        2012,
+        2014,
+        2016,
+        2026
+      ],
+      "missing_count": 15,
+      "canonical_matches": [
+        "ECCV"
+      ]
+    },
+    {
+      "venue": "AAAI",
+      "expected_years": 40,
+      "observed_expected_years": 40,
+      "missing_years": [],
+      "missing_count": 0,
+      "canonical_matches": [
+        "AAAI",
+        "AAAI+IAAI"
+      ]
+    },
+    {
+      "venue": "IJCAI",
+      "expected_years": 35,
+      "observed_expected_years": 35,
+      "missing_years": [],
+      "missing_count": 0,
+      "canonical_matches": [
+        "ECAI+IJCAI",
+        "IJCAI"
+      ]
+    },
+    {
+      "venue": "ECAI",
+      "expected_years": 29,
+      "observed_expected_years": 27,
+      "missing_years": [
+        1974,
+        1980
+      ],
+      "missing_count": 2,
+      "canonical_matches": [
+        "ECAI",
+        "ECAI+IJCAI"
+      ]
+    },
+    {
+      "venue": "KDD",
+      "expected_years": 32,
+      "observed_expected_years": 32,
+      "missing_years": [],
+      "missing_count": 0,
+      "canonical_matches": [
+        "KDD"
+      ]
+    },
+    {
+      "venue": "SIGIR",
+      "expected_years": 50,
+      "observed_expected_years": 50,
+      "missing_years": [],
+      "missing_count": 0,
+      "canonical_matches": [
+        "SIGIR"
+      ]
+    },
+    {
+      "venue": "WWW",
+      "expected_years": 33,
+      "observed_expected_years": 27,
+      "missing_years": [
+        1994,
+        1996,
+        1997,
+        1998,
+        1999,
+        2000
+      ],
+      "missing_count": 6,
+      "canonical_matches": [
+        "WWW"
+      ]
+    },
+    {
+      "venue": "WSDM",
+      "expected_years": 19,
+      "observed_expected_years": 19,
+      "missing_years": [],
+      "missing_count": 0,
+      "canonical_matches": [
+        "WSDM"
+      ]
+    },
+    {
+      "venue": "RecSys",
+      "expected_years": 20,
+      "observed_expected_years": 19,
+      "missing_years": [
+        2026
+      ],
+      "missing_count": 1,
+      "canonical_matches": [
+        "RecSys"
+      ]
+    },
+    {
+      "venue": "CIKM",
+      "expected_years": 35,
+      "observed_expected_years": 34,
+      "missing_years": [
+        2026
+      ],
+      "missing_count": 1,
+      "canonical_matches": [
+        "CIKM"
+      ]
+    },
+    {
+      "venue": "ICDM",
+      "expected_years": 26,
+      "observed_expected_years": 25,
+      "missing_years": [
+        2026
+      ],
+      "missing_count": 1,
+      "canonical_matches": [
+        "ICDM"
+      ]
+    },
+    {
+      "venue": "SDM",
+      "expected_years": 26,
+      "observed_expected_years": 25,
+      "missing_years": [
+        2026
+      ],
+      "missing_count": 1,
+      "canonical_matches": [
+        "SDM"
+      ]
+    },
+    {
+      "venue": "ECML-PKDD",
+      "expected_years": 38,
+      "observed_expected_years": 33,
+      "missing_years": [
+        1986,
+        1987,
+        1988,
+        1989,
+        1991
+      ],
+      "missing_count": 5,
+      "canonical_matches": [
+        "ECML",
+        "ECML-PKDD",
+        "PKDD"
+      ]
+    },
+    {
+      "venue": "ICRA",
+      "expected_years": 43,
+      "observed_expected_years": 43,
+      "missing_years": [],
+      "missing_count": 0,
+      "canonical_matches": [
+        "ICRA"
+      ]
+    },
+    {
+      "venue": "IROS",
+      "expected_years": 39,
+      "observed_expected_years": 35,
+      "missing_years": [
+        1988,
+        1989,
+        1990,
+        2026
+      ],
+      "missing_count": 4,
+      "canonical_matches": [
+        "IROS"
+      ]
+    },
+    {
+      "venue": "RSS",
+      "expected_years": 22,
+      "observed_expected_years": 20,
+      "missing_years": [
+        2025,
+        2026
+      ],
+      "missing_count": 2,
+      "canonical_matches": [
+        "RSS"
+      ]
+    },
+    {
+      "venue": "CoRL",
+      "expected_years": 10,
+      "observed_expected_years": 9,
+      "missing_years": [
+        2026
+      ],
+      "missing_count": 1,
+      "canonical_matches": [
+        "CORL"
+      ]
+    },
+    {
+      "venue": "ICAPS",
+      "expected_years": 37,
+      "observed_expected_years": 28,
+      "missing_years": [
+        1990,
+        1991,
+        1992,
+        1993,
+        1995,
+        1997,
+        1999,
+        2001,
+        2026
+      ],
+      "missing_count": 9,
+      "canonical_matches": [
+        "AIPS",
+        "ICAPS"
+      ]
+    },
+    {
+      "venue": "KR",
+      "expected_years": 23,
+      "observed_expected_years": 23,
+      "missing_years": [],
+      "missing_count": 0,
+      "canonical_matches": [
+        "KR"
+      ]
+    },
+    {
+      "venue": "AAMAS",
+      "expected_years": 25,
+      "observed_expected_years": 22,
+      "missing_years": [
+        2012,
+        2025,
+        2026
+      ],
+      "missing_count": 3,
+      "canonical_matches": [
+        "AAMAS"
+      ]
+    },
+    {
+      "venue": "ICASSP",
+      "expected_years": 51,
+      "observed_expected_years": 49,
+      "missing_years": [
+        1977,
+        2026
+      ],
+      "missing_count": 2,
+      "canonical_matches": [
+        "ICASSP"
+      ]
+    },
+    {
+      "venue": "INTERSPEECH",
+      "expected_years": 39,
+      "observed_expected_years": 39,
+      "missing_years": [],
+      "missing_count": 0,
+      "canonical_matches": [
+        "ECST",
+        "EUROSPEECH",
+        "ICSLP",
+        "INTERSPEECH"
+      ]
+    },
+    {
+      "venue": "ISMIR",
+      "expected_years": 27,
+      "observed_expected_years": 26,
+      "missing_years": [
+        2026
+      ],
+      "missing_count": 1,
+      "canonical_matches": [
+        "ISMIR"
+      ]
+    },
+    {
+      "venue": "MICCAI",
+      "expected_years": 29,
+      "observed_expected_years": 29,
+      "missing_years": [],
+      "missing_count": 0,
+      "canonical_matches": [
+        "MICCAI"
+      ]
+    },
+    {
+      "venue": "FAccT",
+      "expected_years": 9,
+      "observed_expected_years": 9,
+      "missing_years": [],
+      "missing_count": 0,
+      "canonical_matches": [
+        "FAccT"
+      ]
+    },
+    {
+      "venue": "MLSys",
+      "expected_years": 9,
+      "observed_expected_years": 8,
+      "missing_years": [
+        2018
+      ],
+      "missing_count": 1,
+      "canonical_matches": [
+        "MLSys"
+      ]
+    },
+    {
+      "venue": "MM",
+      "expected_years": 34,
+      "observed_expected_years": 33,
+      "missing_years": [
+        2026
+      ],
+      "missing_count": 1,
+      "canonical_matches": [
+        "MM"
+      ]
+    }
+  ],
+  "expected_venue_years": 1266,
+  "missing_venue_years": 150,
+  "venues_with_missing_years": 25,
+  "known_nonpublication_exceptions": [
+    {
+      "venue": "NeurIPS",
+      "year": 2026,
+      "reason": "announced;not in proceedings index at census"
+    },
+    {
+      "venue": "CoRL",
+      "year": 2026,
+      "reason": "not in publisher proceedings index at census"
+    },
+    {
+      "venue": "MLSys",
+      "year": 2018,
+      "reason": "first non-archival SysML meeting; not a missing formal publication volume"
+    }
+  ],
+  "historical_missing_years_excluding_nonarchival_2018": 135,
+  "historical_venues_with_missing_years": 16,
+  "historical_gaps": [
+    {
+      "venue": "ICML",
+      "years": [
+        1980,
+        1983,
+        1985,
+        1987,
+        1988,
+        1989,
+        1990,
+        1991,
+        1992,
+        1993,
+        1994,
+        1995,
+        1996,
+        1997,
+        1998,
+        1999,
+        2000,
+        2001,
+        2002,
+        2003,
+        2004,
+        2005,
+        2006,
+        2007,
+        2008,
+        2009,
+        2010,
+        2011,
+        2012
+      ]
+    },
+    {
+      "venue": "ICLR",
+      "years": [
+        2014,
+        2015,
+        2016
+      ]
+    },
+    {
+      "venue": "AISTATS",
+      "years": [
+        1985,
+        1989,
+        1991,
+        1993
+      ]
+    },
+    {
+      "venue": "ACL",
+      "years": [
+        1963,
+        1964,
+        1965,
+        1966,
+        1967,
+        1968,
+        1969,
+        1970,
+        1971,
+        1972,
+        1973,
+        1974,
+        1975,
+        1976,
+        1977,
+        1978
+      ]
+    },
+    {
+      "venue": "COLING",
+      "years": [
+        1971,
+        1976,
+        1978
+      ]
+    },
+    {
+      "venue": "CVPR",
+      "years": [
+        1983,
+        1985,
+        1986,
+        1988,
+        1989,
+        1991,
+        1992,
+        1993,
+        1994,
+        1996,
+        1997,
+        1998,
+        1999,
+        2000,
+        2001,
+        2003,
+        2004,
+        2005,
+        2006,
+        2007,
+        2008,
+        2009,
+        2010,
+        2011,
+        2012
+      ]
+    },
+    {
+      "venue": "ICCV",
+      "years": [
+        1987,
+        1988,
+        1990,
+        1993,
+        1995,
+        1998,
+        1999,
+        2001,
+        2003,
+        2005,
+        2007,
+        2009,
+        2011
+      ]
+    },
+    {
+      "venue": "ECCV",
+      "years": [
+        1990,
+        1992,
+        1994,
+        1996,
+        1998,
+        2000,
+        2002,
+        2004,
+        2006,
+        2008,
+        2010,
+        2012,
+        2014,
+        2016
+      ]
+    },
+    {
+      "venue": "ECAI",
+      "years": [
+        1974,
+        1980
+      ]
+    },
+    {
+      "venue": "WWW",
+      "years": [
+        1994,
+        1996,
+        1997,
+        1998,
+        1999,
+        2000
+      ]
+    },
+    {
+      "venue": "ECML-PKDD",
+      "years": [
+        1986,
+        1987,
+        1988,
+        1989,
+        1991
+      ]
+    },
+    {
+      "venue": "IROS",
+      "years": [
+        1988,
+        1989,
+        1990
+      ]
+    },
+    {
+      "venue": "RSS",
+      "years": [
+        2025
+      ]
+    },
+    {
+      "venue": "ICAPS",
+      "years": [
+        1990,
+        1991,
+        1992,
+        1993,
+        1995,
+        1997,
+        1999,
+        2001
+      ]
+    },
+    {
+      "venue": "AAMAS",
+      "years": [
+        2012,
+        2025
+      ]
+    },
+    {
+      "venue": "ICASSP",
+      "years": [
+        1977
+      ]
+    }
+  ],
+  "current_2026_gaps": 14
+}

docs/coverage/historical-gaps-2026-11-09.md

@@ -0,0 +1,58 @@
+# 历史年度缺口的当前核对
+
+生产只读快照：2026-11-08 16:34:40.741334 UTC（北京时间11月9日00:34），foundry_data/research。对比已核实的43份首轮官方会议年度清单，使用当前有投稿记录的edition及真实venue_aliases，不沿用旧固定快照的missing_years。
+
+- 清单共1266个会议—年份位置，当前有150个位置尚无投稿记录，涉及25个目标会议。
+- 其中2025年及以前136个；排除已明确非存档的SysML2018首届，剩**135个历史年度待补采或核实，涉及16个会议**。
+- 另有14个2026年度待核实。已知NeurIPS2026和CoRL2026在首轮核查时尚未进入论文集索引，不能一律当已公开论文遗漏；其他年度也需逐个核实发布时间。
+
+这些是“整年没有任何投稿记录”的数量，不是缺论文篇数，也不是完整性百分比。一个年份只有少数记录、只有公开拒稿或缺了某个track，仍会被计作“已有年度”；这些更细缺口及公开拒稿的逐年可获得性尚须继续核验。会议别名共享的联合届次不能当成两批独立论文，未见录用也不能推断拒稿。
+
+| 会议 | 历史缺口年份数 |
+| --- | ---: |
+| ICML | 29 |
+| CVPR | 25 |
+| ACL | 16 |
+| ECCV | 14 |
+| ICCV | 13 |
+| ICAPS | 8 |
+| WWW | 6 |
+| ECML-PKDD | 5 |
+| AISTATS | 4 |
+| ICLR、COLING、IROS | 各3 |
+| ECAI、AAMAS | 各2 |
+| RSS、ICASSP | 各1 |
+
+[完整逐年结果](historical-gaps-2026-11-09.json) 保留43目标的官方预期、现有年份数、缺失年份与规范会议匹配。原始只读生产导出保存在 `/data/paper-index/collected/historical-gap-current-20261109-v1.json`。 导出SHA256 `580cdea6dd9869adafe9bc28fb75cf3854bb290ad65aa5a600d9bc368a528b93`。年度对照清单为 `docs/coverage/inventory-2026-11-08.json`；候选来源文件绑定的输入哈希与本目录缺口JSON逐字节一致。来源清单的年份和哈希沿用相应 `*-editions-2026-11-08.json`，不是本轮重新访问所有会议网站；2026发布状态需要后续更新。
+
+## 候选来源与下一步
+
+另一次只读目录核查在现有、已校验的DBLP快照中，为135个历史年度中的**85个**匹配到初步候选论文集，另**50个**尚未按本轮规则匹配到，需要继续识别目录键或寻找官方来源。未匹配不等于会议没举办、未出版或没有公开来源。
+
+[候选卷清单](historical-source-candidates-2026-11-09.json) 保留卷key、标题、出版年份、原始bibliographic counts和出版者链接。初筛使用目标family映射内的四位卷key年份或卷标题唯一显式年份，再按具体会议身份复核；两位键、出版年份与会议年份差异、前身系列及其他目录命名可能漏匹配。原始计数可能含workshop、前言、索引或多个卷，**不能当可直接导入的录用论文数量**。
+
+后续[来源范围复核](historical-source-scope-2026-11-09.md)排除了4个误归ACL的COLING卷，因此候选年度由89修正为85、未匹配年度由46增至50。原卷及排除理由保留在JSON中；135个历史年度缺口不因此改变。ACL1963–1978与COLING1971/1976/1978在当前Anthology目录未列出，继续记为来源待核实，不能一律称作已公开但漏采的论文集。
+
+优先核查已有候选卷较集中的CVPR22年、ICCV12年、ECCV14年和ICML25年。其中ICML早期workshop名称可能属于官方主系列前身，也可能是附属workshop，须对照官方届次证据逐卷区分，不能靠标题包含workshop一刀切。所有候选须经过范围、稳定标识、重复/冲突、原始字段和覆盖预检；这一步没有导入候选卷。
+
+## 复核方法
+
+只读获取当前有记录的edition，不连接papers大文本或标签表：
+
+```sql
+WITH counts AS (
+  SELECT edition_id,count(*) AS submissions
+  FROM submissions GROUP BY edition_id
+)
+SELECT e.id,e.venue_id,v.acronym,e.year,e.track,e.event_key,counts.submissions
+FROM editions e JOIN venues v ON v.id=e.venue_id
+JOIN counts ON counts.edition_id=e.id ORDER BY e.id;
+
+SELECT venue_id,alias FROM venue_aliases ORDER BY venue_id,alias;
+```
+
+逐目标取规范会议名或别名相等的edition年份集合，与该目标edition_census.expected_years做集合差；再把2026与明确非存档的SysML2018单列。保留原始导出与校验哈希，不把每个matched year误当逐篇/逐track完整。
+
+与本清单同一阶段的WWW补采只补1995年现有poster轨道，不新增缺失年度，所以即使其15条增量已验收，135这个年度计数也不会随之下降。早期1994年的50条正式展示候选、307条混合节目候选共357条仍需核实；其中并非每条都是论文。
+
+本清单用于明确第14步剩余范围，不代表第14步完成，也不替代所有导入结束后的第16步全库去重、摘要、约束和实际查询验收。

docs/coverage/historical-source-scope-2026-11-09.md

@@ -0,0 +1,32 @@
+# 早期 NLP 来源范围复核
+
+核查时间：2026-11-08 18:27 UTC（北京时间11月9日）。本轮只修正来源候选和记录可获得性，没有操作生产数据库。
+
+## 跨会议候选修正
+
+先前候选清单将 `conf/coling/1965`、`1967`、`1969`、`1973` 匹配到 ACL 年会缺口。这四卷标题明确属于 COLING；[COLING 官方 Anthology 目录](https://aclanthology.org/venues/coling/)将它们列在 COLING 下。出版者写作 ACL 或目标映射包含 COLING，不足以证明某一年是 ACL 联办届次。
+
+从 ACL 候选中排除上述四卷，原始卷字段和排除理由保留在[候选JSON](historical-source-candidates-2026-11-09.json)的 `excluded_cross_venue_candidates` 中。135个历史会议—年份位置仍未关闭；有初步候选来源的年度由89修正为85，尚未匹配的由46增至50。这些数字是目录初筛结果，不是已验证可导入的论文数。
+
+联合届次仍按具体年度的明确会议身份处理，例如有官方联办证据的卷可以属于多个会议。不能一律排除跨family卷，也不能把联合卷重复导入为两批论文。
+
+## 来源待核实的早期年份
+
+| 范围 | 本轮官方来源观察 | 处理 |
+| --- | --- | --- |
+| ACL 1963–1978，共16年 | [ACL 官方 Anthology 目录](https://aclanthology.org/venues/acl/)目前从1979年列起 | 保留历史举办清单；逐年核实是否有可索引发表材料，不能据会议举办就断言漏采正式论文集，也不据目录缺失断言未出版 |
+| COLING 1971、1976、1978 | [主办委员会历史](https://ufal.mff.cuni.cz/iccl-old)列出这三届；当前Anthology COLING目录未列出 | 来源缺口继续开放，需要主办方或出版者逐篇索引；不生成占位论文或录用/拒稿记录 |
+
+这19个位置不能用其他会议的卷补齐，也不能因本轮未找到目录而删除。公开拒稿可获得性是独立核查，不从论文集或录用清单缺失推断。
+
+## 证据与复核
+
+三份原始HTML及 `receipts.json` 保存在工作证据目录 `paper-index/nlp-history-source-review-v1/`；结构化响应URL、时间、状态、字节数和完整SHA256收录于候选JSON的 `scope_review.evidence`。三次响应均为HTTP200；未获取论文全文。
+
+| 官方来源 | 响应SHA256 |
+| --- | --- |
+| [acl-venue](https://aclanthology.org/venues/acl/) | `d04c188c0769e065f18ca9c0246f70086e2f7027ce3efe5ea0f2cdc641796b72` |
+| [coling-venue](https://aclanthology.org/venues/coling/) | `8e7ec0a2915bfbccaa26b7a7558a9cf977c4aadc119794516ecebb826f61623f` |
+| [iccl-history](https://ufal.mff.cuni.cz/iccl-old) | `c5286dc51236c90685d22543ff1f7b209adcb3eedb82a7b1ac2115c8dd0169fc` |
+
+复核时先校验保存的原始响应哈希，再读取目录中的会议身份和年度。实时网页若已变化，应另存新响应和时间，不能把新内容套用旧哈希。本次不改最初会议年度清单或生产年度快照，不代表第14步覆盖验收或第16步最终全库审计完成。

docs/coverage/www-1995-posters-reviewed-2026-11-09.json

@@ -0,0 +1,201 @@
+{
+  "format": 1,
+  "source_cache_sha256": "8a65562f3490497cada2919ec77827134beabd5ec9c3690dec3625e2318aa54e",
+  "records": [
+    {
+      "entry_id": "45",
+      "individual_url": "https://archives.iw3c2.org/www3/proceedings/posters/45/index.html",
+      "individual_sha256": "83bede6afc9a7039fae2dd1f28274a394418b34f5d5632590b913617580bda1a",
+      "title": "WISE - World Wide Information System for R&D; Efforts",
+      "authors": [
+        "Fritz Loseries",
+        "Veronika Samara",
+        "Bodo Urban"
+      ],
+      "author_source": "canonical_byline",
+      "reason": "Reviewed canonical H1 and explicit authorship for this exact linked poster entry; retain abstract-heading and author-index title variants without fuzzy matching.",
+      "missing_abstract_reason": ""
+    },
+    {
+      "entry_id": "16",
+      "individual_url": "https://archives.iw3c2.org/www3/proceedings/posters/16/index.html",
+      "individual_sha256": "cf76da903eaeaa27d19857fdaa13220f6821559bf4ac22844fae10abbaa6f3f3",
+      "title": "The World Wide Web as a Tool for Collaborative Inter-Agency Projects: Creation of an Environmental Metadata Database for the Analysis of the Venice Lagoon Ecosystem.",
+      "authors": [
+        "Mirko Labbri"
+      ],
+      "author_source": "canonical_byline",
+      "reason": "Reviewed canonical H1 and explicit authorship for this exact linked poster entry; retain abstract-heading and author-index title variants without fuzzy matching.",
+      "missing_abstract_reason": ""
+    },
+    {
+      "entry_id": "17",
+      "individual_url": "https://archives.iw3c2.org/www3/proceedings/posters/17/index.html",
+      "individual_sha256": "0578fec9c82e8f2f98c4de0f1bcc44011ebb84c53353fdcc423a69cf70673bc2",
+      "title": "NetRepreneur",
+      "authors": [
+        "Kari-Pekka Turtiainen",
+        "Antti Auer"
+      ],
+      "author_source": "canonical_byline",
+      "reason": "Reviewed canonical H1 and explicit authorship for this exact linked poster entry; retain abstract-heading and author-index title variants without fuzzy matching.",
+      "missing_abstract_reason": ""
+    },
+    {
+      "entry_id": "29",
+      "individual_url": "https://archives.iw3c2.org/www3/proceedings/posters/29/index.html",
+      "individual_sha256": "624ad3f685de1163411e33e90488975d848a63243fd7ffbed0862c6fb58c4140",
+      "title": "Mosaic TueV 2.4.3: Multilingual GUI and Improvements towards User Friendliness",
+      "authors": [
+        "Mike Bretz",
+        "Uwe Koch",
+        "Paul de Bra"
+      ],
+      "author_source": "canonical_byline",
+      "reason": "Reviewed canonical H1 and explicit authorship for this exact linked poster entry; retain abstract-heading and author-index title variants without fuzzy matching.",
+      "missing_abstract_reason": ""
+    },
+    {
+      "entry_id": "13",
+      "individual_url": "https://archives.iw3c2.org/www3/proceedings/posters/13/index.html",
+      "individual_sha256": "02409321a3403a94d8362a9862f4960386ea9c344fb47f4bcdf9eae128e8b46e",
+      "title": "Using a WWW Browser as an Alternative User Interface for Interactive Applications",
+      "authors": [
+        "Frank Lonczewski"
+      ],
+      "author_source": "canonical_byline",
+      "reason": "Reviewed canonical H1 and explicit authorship for this exact linked poster entry; retain abstract-heading and author-index title variants without fuzzy matching.",
+      "missing_abstract_reason": ""
+    },
+    {
+      "entry_id": "36",
+      "individual_url": "https://archives.iw3c2.org/www3/proceedings/posters/36/index.html",
+      "individual_sha256": "ed9dada4ea5903cb8f0a4e75f2958416a6e32647fbf3530e58edd9abf26070ce",
+      "title": "HyperSource: A Hypermedia Program Development And Documentation System",
+      "authors": [
+        "Jan Oliver Borchers"
+      ],
+      "author_source": "canonical_byline",
+      "reason": "Reviewed canonical H1 and explicit authorship for this exact linked poster entry; retain abstract-heading and author-index title variants without fuzzy matching.",
+      "missing_abstract_reason": ""
+    },
+    {
+      "entry_id": "57",
+      "individual_url": "https://archives.iw3c2.org/www3/proceedings/posters/57/index.html",
+      "individual_sha256": "3ad4206ad28a19823172cd2845f636fe7a461c251940e9a93deb8f6484b8a984",
+      "title": "Progressive Image Transmission Capability (PROTRAC) on the World Wide Web",
+      "authors": [
+        "Vojislav Lalich-Petrich",
+        "Gaurav Bhatia",
+        "Larry Davis"
+      ],
+      "author_source": "canonical_byline",
+      "reason": "Reviewed canonical H1 and explicit authorship for this exact linked poster entry; retain abstract-heading and author-index title variants without fuzzy matching.",
+      "missing_abstract_reason": ""
+    },
+    {
+      "entry_id": "38",
+      "individual_url": "https://archives.iw3c2.org/www3/proceedings/posters/38/index.html",
+      "individual_sha256": "940370acc5e6a954b197053fe451e466a904d1f0ac82c8fe4e6ca455faf2c927",
+      "title": "Measuring the Web with Lycos",
+      "authors": [
+        "Michael L. Mauldin"
+      ],
+      "author_source": "canonical_byline",
+      "reason": "Reviewed canonical H1 and explicit authorship for this exact linked poster entry; retain abstract-heading and author-index title variants without fuzzy matching.",
+      "missing_abstract_reason": "Official abstract listing has no abstract. Canonical HTML has body sections but no explicit abstract; body is not substituted as a summary."
+    },
+    {
+      "entry_id": "05",
+      "individual_url": "https://archives.iw3c2.org/www3/proceedings/posters/05/index.html",
+      "individual_sha256": "1858be87d6e7f47476e1cb8fb4793c0fce3b8a6ff653bcfd4a35089e0520f177",
+      "title": "Virtual Places - Adding people to the web",
+      "authors": [
+        "Yosi Mass"
+      ],
+      "author_source": "author_index",
+      "reason": "Reviewed canonical H1 and explicit authorship for this exact linked poster entry; retain abstract-heading and author-index title variants without fuzzy matching.",
+      "missing_abstract_reason": "Official abstract listing and canonical HTML have no abstract. Canonical page identifies a contact and links PostScript; no full-text download. Author identity comes from author index, not inferred from contact role."
+    },
+    {
+      "entry_id": "37",
+      "individual_url": "https://archives.iw3c2.org/www3/proceedings/posters/37/index.html",
+      "individual_sha256": "b700a0a51b7a82a5f88373ae0283bb369d1cf6f455acff7d91cd7313fa8b3adf",
+      "title": "The Internet Public Library: Serving a Global Community",
+      "authors": [
+        "Susan Schweitzer"
+      ],
+      "author_source": "canonical_byline",
+      "reason": "Reviewed canonical H1 and explicit authorship for this exact linked poster entry; retain abstract-heading and author-index title variants without fuzzy matching.",
+      "missing_abstract_reason": ""
+    },
+    {
+      "entry_id": "41",
+      "individual_url": "https://archives.iw3c2.org/www3/proceedings/posters/41/index.html",
+      "individual_sha256": "397b474109f9c1240a77d0f490f2b02dd35fae93febbb7397d1b2a31759a8ec1",
+      "title": "Graphical Browsing in the WorldWideWeb and the MultiMedia Forum",
+      "authors": [
+        "Rainer Pagé",
+        "Klaus Süllow"
+      ],
+      "author_source": "canonical_byline",
+      "reason": "Reviewed canonical H1 and explicit authorship for this exact linked poster entry; retain abstract-heading and author-index title variants without fuzzy matching.",
+      "missing_abstract_reason": ""
+    },
+    {
+      "entry_id": "47",
+      "individual_url": "https://archives.iw3c2.org/www3/proceedings/posters/47/index.html",
+      "individual_sha256": "52eb4bfdd658730a222c98170b72dabf9cbcc5ffd2ada1e900b826baf2b02833",
+      "title": "Interoperating Advanced Hypermedia Systems with WWW",
+      "authors": [
+        "Jian Zhang",
+        "Martin Fruehauf",
+        "Thomas Kirste"
+      ],
+      "author_source": "canonical_byline",
+      "reason": "Reviewed canonical H1 and explicit authorship for this exact linked poster entry; retain abstract-heading and author-index title variants without fuzzy matching.",
+      "missing_abstract_reason": ""
+    },
+    {
+      "entry_id": "50",
+      "individual_url": "https://archives.iw3c2.org/www3/proceedings/posters/50/index.html",
+      "individual_sha256": "8dc8b0561bd8b1b106c5b209a973c695c3cc9df5288450d4a01fe32541476b7e",
+      "title": "Web Enabled Agile Manufacturing",
+      "authors": [
+        "S. V. Iyer",
+        "T. Singh",
+        "U. Ravaioli",
+        "R. E. DeVor",
+        "S. G. Kapoor",
+        "G. Veknatasubramian"
+      ],
+      "author_source": "canonical_byline",
+      "reason": "Reviewed canonical H1 and explicit authorship for this exact linked poster entry; retain abstract-heading and author-index title variants without fuzzy matching.",
+      "missing_abstract_reason": ""
+    },
+    {
+      "entry_id": "55",
+      "individual_url": "https://archives.iw3c2.org/www3/proceedings/posters/55/index.html",
+      "individual_sha256": "21780cd4047f513e8c9aeced5247002e27cf71719f08de85c6531b57a9e5058d",
+      "title": "Weblets: Fundamental Building Blocks for WWW Tools",
+      "authors": [
+        "Thomas J. Watt Jr."
+      ],
+      "author_source": "canonical_byline",
+      "reason": "Reviewed canonical H1 and explicit authorship for this exact linked poster entry; retain abstract-heading and author-index title variants without fuzzy matching.",
+      "missing_abstract_reason": ""
+    },
+    {
+      "entry_id": "59",
+      "individual_url": "https://archives.iw3c2.org/www3/proceedings/posters/59/index.html",
+      "individual_sha256": "fed4cc9dcaee350a12c5e4b6fa9c99be9bead25a81632512d31566400d3d0fad",
+      "title": "Usenet and the World-Wide Web",
+      "authors": [
+        "M. L. Grant"
+      ],
+      "author_source": "canonical_byline",
+      "reason": "Reviewed canonical H1 and explicit authorship for this exact linked poster entry; retain abstract-heading and author-index title variants without fuzzy matching.",
+      "missing_abstract_reason": ""
+    }
+  ]
+}

docs/edition-tracks.md

@@ -0,0 +1,45 @@
+# 按证据修正既有投稿的轨道
+
+`paper-atlas maintain edition-tracks` 用于已导入整卷论文集、后来取得逐条官方分区证据的情形。它更正原投稿的 edition 归属，保留投稿 ID、所有决定和来源关系；不会重新导入论文、合并实体、生成研究方法标签或推断拒稿。支持独立 SQLite 样本和共享 PostgreSQL；不得写入已冻结的生产 SQLite 备份。
+
+## 证据与完整范围
+
+先核实官方目录、稳定 DOI、原始来源记录和分区，保存原始页面及 SHA256。输入 JSON 为 `{"format":1,"records":[...]}`，每条必须包含：
+
+| 字段 | 含义 |
+| --- | --- |
+| `submission_id`、`source_record_id`、`external_id` | 原有投稿、主来源记录及主来源投稿标识 |
+| `source`、`doi` | 原来源名称及其 metadata.record_identifiers 中的 DOI |
+| `from_edition_id`、`year`、`venue` | 原 edition 及其真实年份和规范会议名 |
+| `venue_names` | 规范会议名加全部已有别名；必须与库中一致 |
+| `track` | 经核实的目标轨道，使用小写字母、数字、下划线；范围未明确可用 proceedings_unspecified |
+| `section` | 官方分区原文或明确的未确定说明，不推断研究方法 |
+| `evidence_url`、`evidence_sha256` | 已人工/独立核实的 HTTPS 原始证据和内容哈希 |
+
+输入必须列出每个原 edition 的**全部**投稿，包含留在原轨道的记录。工具核对身份、DOI、单来源发表观察、现有别名、完整范围以及目标冲突；不自动解释 HTML，也不把填写了 URL/哈希当成已证明内容。核查者应先独立证明每条分区对应关系。
+
+多来源共同拥有的投稿、有非 accepted 决定历史的投稿、已有分类修订、非空目标 edition、目标已有覆盖或标识等情形会停止，需要专门核对。每个原 edition 须有一个明确来源清单；其他来源或多清单覆盖需先另行协调。本功能不是任意批量移动或跨会合并工具。
+
+## 只读计划、事务执行和验收
+
+```sh
+paper-atlas maintain edition-tracks \
+  --pg-service foundry-research --pg-schema research \
+  --inventory reviewed-inventory.json --out plan.json
+
+sha256sum plan.json
+
+paper-atlas maintain edition-tracks \
+  --pg-service foundry-research --pg-schema research \
+  --plan plan.json --plan-sha256 <上一步的完整SHA256> --out applied.json
+```
+
+报告路径必须是已有目录中的新文件。预检只读。执行使用 PostgreSQL 单写入 advisory lock，或 SQLite BEGIN IMMEDIATE；整个分类修正、逐来源证据、标识、覆盖及验收在一个事务中完成。计划绑定数据库身份、全部来源文本/非修订元数据、标签、可选 annotation_events、投稿/决定/来源/标识内容及原覆盖快照；过期即停止。失败回滚数据，PostgreSQL 序列可能留下正常间隙，不能用 max(id) 当数量。
+
+目标 edition 沿用原会议、年份、event_key，不删除或重命名原 edition。只更正既有 submissions.edition_id 和对应 submission_identifiers.edition_id。原文、来源检索时间、决定 ID/状态/证据及各表来源关系不变。
+
+分类证据写入来源元数据的保留字段 `edition_scope_curation_v1`，包括原范围、目标 edition、官方分区、URL、哈希和计划摘要。普通来源导入不能伪造或替换此字段；后续 record 刷新保留修订。相同来源投稿通过旧范围调用 decision 时仍路由到修正后的 edition。旧整卷 main 文件通过 import records 刷新会报错并回滚，要求先按核实轨道重新整理，避免覆盖正确的统计。
+
+覆盖保留原清单 URL、cursor、成功时间，按实际目标分配 expected_count/imported_count 并标 partial；本工具核实分类，未核实摘要或所有来源完整性。原覆盖整行保留在计划中，不能继续将整卷数当 main 论文数。应用后校验原 edition 身份、全部受保护内容、分类、标识、修订字段和覆盖；相同计划再次执行仅在这些验收条件仍成立时返回 already_applied。
+
+生产使用前必须在已确认 runner 完成软件验证。正式数据修正后仍需独立验收，并在最终全库审计中复核；单元测试和应用回执不能代替生产验收。

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.

docs/www-archive.md

@@ -0,0 +1,111 @@
+# Early WWW archive collector
+
+`paper-atlas collect www-archive` is a maintained metadata adapter. It reads official
+IW3C2 archive HTML, keeps exact source hashes, and separates importable rows from
+program entries that still need review. It fetches no PDF, follows no paper-body or
+reference link, and makes no database connection.
+
+For a fresh source snapshot on the server:
+
+```sh
+paper-atlas collect www-archive \
+  --http-cache /data/paper-index/cache/www-archive \
+  --source-cache /data/paper-index/collected/www-archive-source-v2 \
+  --out /data/paper-index/collected/www-archive-normalized-v2
+```
+
+For reproducible offline replay of the already verified snapshot:
+
+```sh
+paper-atlas collect www-archive \
+  --source-cache /data/paper-index/collected/www-early-archive-source-v1 \
+  --source-cache-sha256 c9388ccef44486e0684c9ee9e8620a465cdc4574e6bc9ea9391d743e01a3312c \
+  --out /data/paper-index/collected/www-archive-normalized-v1
+```
+
+Use a new output directory for another batch. Existing snapshots and normalized
+batches are never overwritten. The manifest hash binds every source URL, byte count,
+response hash and cache filename. Plain files must remain directly within the cache;
+symlinks, escaping paths, duplicate URLs, non-200 responses and altered bytes fail.
+The command uses UTF-8 where valid and explicitly records an ISO-8859-1 fallback for
+older pages; original bytes remain authoritative for encoding corrections.
+
+## Evidence and outputs
+
+- WWW3 Spring 1995 papers use `event_key=www-1995-spring`, `track=main`. The author,
+  session and abstract indexes must have the same nonempty entry-ID set. The ID is
+  local to this exact event and proceedings section, never a DOI or OpenReview ID.
+- Posters remain `track=poster`. An abstract without a linked author-index entry
+  stays in review even when another index has a placeholder `#` link for it.
+- Paper URLs must agree between author and session indexes. A malformed whitespace
+  URL in the abstract list may use the exact event/entry-ID match only when both
+  independent indexes corroborate the same clean URL. The original href is retained.
+- Canonical names come from the explicitly labeled author index. The source lists
+  them alphabetically; this is **not** original byline order. Both the raw index names
+  and the full abstract-page byline/affiliation text remain in metadata. No person ID
+  or cross-source identity is inferred from a name. Conflicting titles or missing
+  abstracts stay in review. Topic keywords are not treated as an abstract.
+- WWW1 1994 formal presentations and WWW2 1994 mixed categories remain review
+  candidates. The first list contains uncertain business-track annotations and a
+  count discrepancy; the second contains panels, tutorials, products and other
+  activities. Neither presentation listing nor a missing full-text link establishes
+  a regular-paper acceptance or a rejection. Raw category/title/name claims are kept.
+
+The main and poster JSONL files have independent manifests with explicit event key,
+track, count and SHA256. `source_listing_count` records the full recognized abstract
+listing, while `held_entry_count` records excluded entries; neither is a ready-row
+count. Their status stays `partial`: a verified linked subset does
+not prove complete event coverage. `review-candidates.jsonl` contains nested
+candidates and reasons, with no importable decision status. It is excluded from
+`import-queue.json`. `report.json` accounts for ready/held records and every source
+used. Newlines delimit JSONL; Unicode U+2028/U+2029 inside metadata are not line breaks.
+The output is published only after all required source and scope checks pass.
+
+## Before importing
+
+This independent collector branch starts from main; it does not include or silently
+install the event-aware writer. Its output requires the reviewed integrated runtime
+from PR31 (or a main revision containing that implementation). The production event
+schema migration has been independently accepted, preserving existing edition1212
+and its 59 Fall submissions; that is not an import of the Spring proceedings.
+
+Inspect held entries and original author names, preflight every ready row against
+current PostgreSQL stable identifiers and candidate identities, and verify the
+source/year/track/event coverage before running `import records`. When a ready subset
+excludes held entries, preserve the independently known listing total in the existing
+incremental-coverage protocol: bind a freshly checked parent count and expected count,
+and set the full expected count from this manifest. A new source uses previous count
+0 and previous expected count null; an existing source requires its actual current
+values. These live parent values are not guessed by the offline collector. Do not automatically
+run this queue against an older writer. Preserve extended versions and independent
+submission/decision evidence, and keep unresolved identities held. A normalization
+report or passing tests alone do not establish production import, complete historical
+coverage or final global deduplication.
+
+Initial fixed-cache replay (64ced7a, CI1057): 38 main papers and38 posters ready;
+372 entries held (50 first-conference fields,307 mixed program entries,15 posters).
+Independent report44dfd070 verified source hashes,counts and author evidence, finding
+one existing exact-title candidate. That publication is explicitly linked to
+arXiv1811.12161 by its WWW1995/journal-reference declaration; arXiv1810.07232 is a
+different1997 invited chapter and remains separate despite the overlap note. This
+is a preparation observation, not evidence of completed production import or final
+global deduplication.
+
+## 以单篇官方页面核实待处理 poster
+
+作者索引漏链接、摘要目录使用短标题或标题存在拼写差异时，不放宽全局标题匹配。可以先取得单篇官方 HTML，人工核对 H1、署名区与原目录同一 entry URL，再提供带 SHA256 的逐条核查文件：
+
+```sh
+paper-atlas collect www-archive \
+  --source-cache /data/paper-index/collected/www-poster-reviewed-source-v2 \
+  --source-cache-sha256 8a65562f3490497cada2919ec77827134beabd5ec9c3690dec3625e2318aa54e \
+  --reviewed-posters docs/coverage/www-1995-posters-reviewed-2026-11-09.json \
+  --reviewed-posters-sha256 <核查文件的SHA256> \
+  --out <新的输出目录>
+```
+
+[1995 年15条poster的核查输入](coverage/www-1995-posters-reviewed-2026-11-09.json) 绑定61页来源快照：原46页和补取的15页单篇HTML。程序验证输入文件及快照哈希、单篇页面哈希、完全一致的 canonical H1、原摘要目录中的同一URL/entry ID及作者证据，不通过相似标题猜身份。规范标题采用单篇H1，摘要目录标题及作者索引标题均在 metadata 中保留。
+
+`author_source=canonical_byline` 表示名字出现在H1之后、摘要/关键词或下一节标题之前的署名区域，由逐条核查明确选出，不从全文引用或机构猜作者；保留原始缩写、顺序、重音和拼写。只有 `author_source=author_index` 时使用原作者索引顺序并逐项核对；05的单篇页面写的是联系人，作者身份使用原作者索引证据，不由联系人角色推断。核查输入是审阅后的数据声明，程序的字符串/哈希检查不代替人工识别署名语义。
+
+38和05的来源没有可明确引用的摘要。保留 `abstract=null` 和逐条缺失原因，manifest准确报告 missing_abstract_count，继续标partial；正文、联系方式及PostScript链接不当作摘要，也不下载正文。缺摘要不等于拒稿或未发表。本批输出仍包含原先已验收记录，正式导入前必须只取新增external ID，检查实时覆盖父记录并按incremental模式准备，不能直接重导入全部输出。

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', {}, '查看全部候选会议的覆盖情况'),
@@ -22,10 +23,14 @@ COMMANDS = {
     ('collect','mlsys'): ('paper_atlas.collectors.proceedings', {'source':'mlsys'}, 'MLSys 官方历年标题与摘要'),
     ('collect','cvf'): ('paper_atlas.collectors.vision', {'source':'cvf'}, 'CVPR / ICCV 官方开放目录'),
     ('collect','ecva'): ('paper_atlas.collectors.vision', {'source':'ecva'}, 'ECCV 官方开放目录'),
+    ('collect','www-archive'): ('paper_atlas.collectors.www_archive', {}, '核验早期WWW官方档案并分开待核查条目'),
     ('collect','dblp'): ('paper_atlas.collectors.dblp_records', {}, '核对并整理 DBLP 历史发表记录'),
     ('catalogue','dblp'): ('paper_atlas.collectors.dblp', {}, '盘点 DBLP 历史会议目录'),
     ('import','arxiv'): ('paper_atlas.ingestion.arxiv', {}, '导入 arXiv 元数据 ZIP'),
     ('import','records'): ('paper_atlas.ingestion.records', {}, '导入核验过的会议文件'),
+    ('maintain','edition-tracks'): ('paper_atlas.ingestion.edition_tracks', {}, '按逐条来源证据修正既有投稿轨道并保护刷新'),
+    ('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/collectors/www_archive.py

@@ -0,0 +1,340 @@
+"""Reproducible early WWW metadata; unresolved program entries stay outside imports."""
+import argparse
+from collections import defaultdict
+import hashlib
+from html.parser import HTMLParser
+import json
+from pathlib import Path
+import re
+import tempfile
+from urllib.parse import urljoin, urlsplit
+
+from paper_atlas.io.html import Document, clean
+from paper_atlas.io.http import fetch
+
+ROOT = 'https://archives.iw3c2.org/'
+SPRING = ROOT+'www3/proceedings/'
+FALL = ROOT+'www2/Proceedings/WWW2_Proceedings.html'
+SEEDS = [ROOT+'www1/PrelimProcs.html', ROOT+'www1/Proceedings.html', FALL,
+         SPRING+'proceedings.html', *[SPRING+'papers/'+x+'.html' for x in ('author','session','abstract')],
+         *[SPRING+'posters/'+x+'.html' for x in ('author','abstract')]]
+
+
+def decode(raw):
+    try:
+        return raw.decode('utf-8-sig'), 'utf-8'
+    except UnicodeDecodeError:
+        return raw.decode('iso-8859-1'), 'iso-8859-1 fallback; original bytes retained'
+
+
+def text(html):
+    return re.sub(r'[ \t\r\n\f\v]+',' ',Document(html).root.text()).strip(' \t\r\n\f\v')
+
+
+def anchors(html):
+    return [(n.attrs.get('href',''),clean(n.text())) for n in Document(html).root.walk() if n.tag=='a']
+
+
+def source_url(url):
+    p=urlsplit(url)
+    if p.scheme!='https' or p.netloc!='archives.iw3c2.org' or p.query or p.fragment or not p.path.endswith('.html'):
+        raise ValueError('Only exact official archive HTML source URLs are allowed')
+    return url
+
+
+class Sources:
+    def __init__(self, directory, expected_sha256):
+        self.root=Path(directory).resolve();manifest=self.root/'manifest.json';raw=manifest.read_bytes()
+        if hashlib.sha256(raw).hexdigest()!=expected_sha256:
+            raise ValueError('Source cache manifest checksum mismatch')
+        self.rows={};self.used=set()
+        for row in json.loads(raw)['sources']:
+            url=source_url(row['url'])
+            if url in self.rows or row.get('status')!=200:
+                raise ValueError('Duplicate or incomplete source response')
+            p=self.root/row['file']
+            if p.is_symlink() or p.parent!=self.root or not p.is_file():
+                raise ValueError('Source cache paths must be plain files directly within the cache')
+            data=p.read_bytes()
+            if len(data)>2*1024**2 or len(data)!=row['bytes'] or hashlib.sha256(data).hexdigest()!=row['sha256']:
+                raise ValueError('Source response checksum or size mismatch')
+            self.rows[url]={'raw':data,'receipt':row}
+
+    def read(self,url):
+        source_url(url)
+        if url not in self.rows:
+            raise ValueError('Required source missing from cache: '+url)
+        self.used.add(url)
+        return decode(self.rows[url]['raw'])[0]
+
+    def proof(self,urls):
+        return [{'url':url,'sha256':self.rows[url]['receipt']['sha256'],
+                 'encoding':decode(self.rows[url]['raw'])[1]} for url in sorted(set(urls))]
+
+
+def download_sources(http_cache, directory):
+    directory=Path(directory)
+    if directory.exists():
+        raise ValueError('Refusing to overwrite a source snapshot')
+    directory.mkdir(parents=True)
+    responses={url:fetch(url,http_cache,2*1024**2) for url in SEEDS}
+    html=decode(responses[FALL])[0]
+    for href,label in anchors(html):
+        if href!='Author_Inst.html' and href.endswith('.html'):
+            url=source_url(urljoin(FALL,href))
+            if not url.startswith(ROOT+'www2/Proceedings/'):
+                raise ValueError('Unexpected fall category source')
+            responses[url]=fetch(url,http_cache,2*1024**2)
+    receipts=[]
+    for i,(url,raw) in enumerate(sorted(responses.items())):
+        name='source-'+str(i).zfill(2)+'.html';(directory/name).write_bytes(raw)
+        receipts.append({'url':url,'status':200,'file':name,'bytes':len(raw),'sha256':hashlib.sha256(raw).hexdigest()})
+    manifest=directory/'manifest.json';manifest.write_text(json.dumps({'sources':receipts},indent=2)+'\n')
+    return hashlib.sha256(manifest.read_bytes()).hexdigest()
+
+
+def identity(href,track):
+    url=urljoin(SPRING+track+'/',href);p=urlsplit(url)
+    match=re.fullmatch(r'/www3/proceedings/'+track+r'/(\d+)/[^?#]+',p.path)
+    if p.scheme=='https' and p.netloc=='archives.iw3c2.org' and not p.query and not p.fragment and match:
+        return match[1],url
+    return None
+
+
+class AuthorIndex(HTMLParser):
+    def __init__(self,html,track):
+        super().__init__(convert_charrefs=True);self.track=track;self.name='';self.bold=None;self.link=None;self.link_text=[]
+        self.entries=defaultdict(lambda:{'authors':[],'urls':set(),'titles':set()});self.feed(html)
+
+    def handle_starttag(self,tag,attrs):
+        if tag=='b':self.bold=[]
+        if tag=='a':self.link=dict(attrs).get('href');self.link_text=[]
+
+    def handle_data(self,data):
+        if self.bold is not None:self.bold.append(data)
+        if self.link is not None:self.link_text.append(data)
+
+    def handle_endtag(self,tag):
+        if tag=='b' and self.bold is not None:
+            self.name=clean(''.join(self.bold));self.bold=None
+        if tag=='a' and self.link is not None:
+            own=identity(self.link,self.track)
+            if own:
+                key,url=own;entry=self.entries[key];entry['urls'].add(url);entry['titles'].add(clean(''.join(self.link_text)))
+                if not self.name:raise ValueError('Paper link has no archive author context')
+                if self.name not in entry['authors']:entry['authors'].append(self.name)
+            self.link=None
+
+
+def sections(html):
+    """Use explicit h3 boundaries; old HTML omits many p/li closing tags."""
+    result=[]
+    for chunk in re.split(r'<h3\b[^>]*>',html,flags=re.I)[1:]:
+        parts=re.split(r'</h3\s*>',chunk,maxsplit=1,flags=re.I)
+        if len(parts)!=2:raise ValueError('Unclosed archive entry heading')
+        heading,body=parts
+        body=re.split(r'<(?:hr|h2|address)\b',body,maxsplit=1,flags=re.I)[0]
+        author=re.search(r'<i\b[^>]*>(.*?)</i\s*>',body,re.I|re.S)
+        paragraphs=[]
+        for paragraph in re.findall(r'<p\b[^>]*>(.*?)(?=</p\s*>|<p\b|$)',body,re.I|re.S):
+            value=text(paragraph)
+            if value and not value.casefold().startswith('keywords:'):paragraphs.append(value)
+        result.append({'heading':heading,'title':text(heading),'abstract':'\n\n'.join(paragraphs)or None,
+                       'author_byline_raw':text(author[1])if author else None})
+    if not result:raise ValueError('No explicit archive entry headings')
+    return result
+
+
+def reviewed_posters(sources, path, expected_sha256, cache_sha256):
+    if path is None:
+        if expected_sha256:raise ValueError('Reviewed poster path required')
+        return {},None
+    raw=Path(path).read_bytes()
+    if hashlib.sha256(raw).hexdigest()!=expected_sha256:raise ValueError('Reviewed poster checksum mismatch')
+    review=json.loads(raw)
+    if set(review)!={'format','source_cache_sha256','records'} or review['format']!=1 or review['source_cache_sha256']!=cache_sha256:
+        raise ValueError('Reviewed poster snapshot binding differs')
+    if not isinstance(review['records'],list) or not 0<len(review['records'])<=100:
+        raise ValueError('Bounded nonempty poster review required')
+    result={}
+    for row in review['records']:
+        if set(row)!={'entry_id','individual_url','individual_sha256','title','authors','author_source','reason','missing_abstract_reason'}:
+            raise ValueError('Unexpected reviewed poster fields')
+        key=row['entry_id'];own=identity(row['individual_url'],'posters')
+        if not isinstance(key,str) or not own or own[0]!=key or key in result:
+            raise ValueError('Reviewed poster identity duplicated or outside canonical entry')
+        if row['author_source'] not in ('canonical_byline','author_index') or not isinstance(row['authors'],list) or not row['authors'] or any(not isinstance(a,str) or not a.strip()for a in row['authors']):
+            raise ValueError('Explicit author evidence required')
+        if len(set(row['authors']))!=len(row['authors']) or not isinstance(row['reason'],str) or not row['reason'].strip():
+            raise ValueError('Reviewed author order/reason invalid')
+        if sources.rows.get(own[1],{}).get('receipt',{}).get('sha256')!=row['individual_sha256']:
+            raise ValueError('Individual poster page checksum differs')
+        page=sources.read(own[1]);headings=re.findall(r'<h1\b[^>]*>(.*?)</h1\s*>',page,re.I|re.S)
+        if len(headings)!=1 or text(headings[0])!=row['title']:
+            raise ValueError('Reviewed title differs from exact canonical H1')
+        header=re.split(r'</h1\s*>',page,maxsplit=1,flags=re.I)[1]
+        header=re.split(r'<h[23]\b',header,maxsplit=1,flags=re.I)[0]
+        header=re.split(r'\b(?:keywords|abstract)\s*:',text(header),maxsplit=1,flags=re.I)[0]
+        if row['author_source']=='canonical_byline' and any(a.casefold() not in header.casefold()for a in row['authors']):
+            raise ValueError('Reviewed author absent from canonical byline region')
+        result[key]={**row,'canonical_byline_region_raw':header}
+    return result,hashlib.sha256(raw).hexdigest()
+
+
+def reviewed_row(sources,key,url,item,authors,review):
+    if url!=review['individual_url']:
+        raise ValueError('Reviewed canonical page differs from abstract-entry URL')
+    if authors and authors['urls']!={url}:
+        raise ValueError('Reviewed poster author-index URL disagrees')
+    if review['author_source']=='author_index':
+        names=[]
+        for name in authors['authors']if authors else []:
+            parts=name.split(',');names.append(clean(parts[1]+' '+parts[0])if len(parts)==2 else name)
+        if names!=review['authors']:raise ValueError('Reviewed author-index evidence differs')
+    missing=review['missing_abstract_reason']
+    if not isinstance(missing,str) or bool(missing.strip())==bool(item['abstract']):
+        raise ValueError('Missing abstract requires explicit reason; never invent a summary')
+    urls=[SPRING+'posters/author.html',SPRING+'posters/abstract.html',url]
+    return {'source':'www_archive','external_id':'www3:posters:'+key,'url':url,'title':review['title'],
+            'abstract':item['abstract'],'authors':review['authors'],'identifiers':{},'venue':'WWW','year':1995,
+            'track':'poster','event_key':'www-1995-spring','status':'accepted','raw_decision':'listed in official posters proceedings',
+            'metadata':{'entry_id':key,'source_evidence':sources.proof(urls),'abstract_heading_title_raw':item['title'],
+                        'author_byline_raw':item['author_byline_raw'],'author_index_raw':authors['authors']if authors else [],
+                        'author_index_titles_raw':sorted(authors['titles'])if authors else [],'abstract_href_raw':url,
+                        'author_order':review['author_source'],'reviewed_poster_identity':review,
+                        'missing_abstract_reason':missing or None},
+            'tags':[{'namespace':'www_archive:proceedings_section','value':'posters'}]}
+
+
+def spring_rows(sources,track,reviews=None):
+    base=SPRING+track+'/';author_url=base+'author.html';abstract_url=base+'abstract.html'
+    authors=AuthorIndex(sources.read(author_url),track).entries
+    listing=sections(sources.read(abstract_url));by_id={};held=[];session={}
+    if track=='papers':
+        session_url=base+'session.html'
+        for href,title in anchors(sources.read(session_url)):
+            own=identity(href,track)
+            if own:
+                key,url=own
+                if key in session:raise ValueError('Duplicate main session entry ID')
+                session[key]=url
+    for item in listing:
+        links=[identity(h,track)for h,t in anchors(item['heading'])if identity(h,track)]
+        if len(links)!=1:
+            held.append({'reason':'unverified_heading_identity','evidence':abstract_url,'candidate':item});continue
+        key,url=links[0]
+        if key in by_id:raise ValueError('Duplicate abstract entry ID')
+        by_id[key]=(url,item)
+    if not by_id:raise ValueError('No stable abstract entry IDs')
+    if track=='papers' and (set(authors)!=set(by_id) or set(session)!=set(by_id)):
+        raise ValueError('Main author/session/abstract entry ID sets disagree')
+    if set(authors)-set(by_id):raise ValueError('Author index has entries outside the abstract listing')
+    reviews=reviews or {}
+    if set(reviews)-set(by_id):raise ValueError('Reviewed poster outside abstract listing')
+    rows=[]
+    for key,(url,item) in by_id.items():
+        if key in reviews:
+            if track!='posters':raise ValueError('Reviews only apply to poster entries')
+            rows.append(reviewed_row(sources,key,url,item,authors.get(key),reviews[key]));continue
+        if key not in authors:
+            held.append({'reason':'abstract_entry_missing_linked_author_index','entry_id':key,'evidence':abstract_url,'candidate':item});continue
+        alternatives=authors[key]['urls']
+        if len(alternatives)!=1:raise ValueError('Ambiguous author-index paper URL')
+        canonical=next(iter(alternatives))
+        if track=='papers' and session[key]!=canonical:raise ValueError('Main author/session paper URLs disagree')
+        if url!=canonical and (track!='papers' or not re.search(r'\s',url)):
+            raise ValueError('Uncorroborated archive URL discrepancy')
+        if not item['abstract']:
+            held.append({'reason':'missing_abstract','entry_id':key,'evidence':abstract_url,'candidate':item});continue
+        if item['title'].casefold() not in {t.casefold() for t in authors[key]['titles']}:
+            held.append({'reason':'abstract_and_author_index_titles_disagree','entry_id':key,'evidence':abstract_url,'candidate':item,'author_index_titles':sorted(authors[key]['titles'])});continue
+        names=[]
+        for name in authors[key]['authors']:
+            parts=name.split(',')
+            names.append(clean(parts[1]+' '+parts[0])if len(parts)==2 else name)
+        proof_urls=[author_url,abstract_url]+([session_url]if track=='papers'else[])
+        rows.append({'source':'www_archive','external_id':'www3:'+track+':'+key,'url':canonical,
+                     'title':item['title'],'abstract':item['abstract'],'authors':names,'identifiers':{},
+                     'venue':'WWW','year':1995,'track':'main'if track=='papers'else'poster','event_key':'www-1995-spring',
+                     'status':'accepted','raw_decision':'listed in official '+track+' proceedings',
+                     'metadata':{'entry_id':key,'source_evidence':sources.proof(proof_urls),
+                                 'author_index_raw':authors[key]['authors'],'author_index_titles_raw':sorted(authors[key]['titles']),'author_order':'archive author-index order, not original byline rank',
+                                 'author_byline_raw':item['author_byline_raw'],'abstract_href_raw':url,
+                                 'url_corroborated_by_author_and_session':url!=canonical},
+                     'tags':[{'namespace':'www_archive:proceedings_section','value':track}]})
+    return rows,held,{'abstract_entries':len(by_id),'linked_author_entries':len(authors),'ready':len(rows),'held':len(held)}
+
+
+def early_review(sources):
+    held=[];url=ROOT+'www1/PrelimProcs.html';html=sources.read(url)
+    for ordinal,chunk in enumerate(re.split(r'<li\b[^>]*>',html,flags=re.I)[1:],1):
+        title=re.search(r'<em>Title:</em>\s*<b>(.*?)</b>',chunk,re.I|re.S)
+        if not title:raise ValueError('Unclassified first-conference entry')
+        names=re.findall(r'<em>Author:</em>(.*?)(?=<br\b)',chunk,re.I|re.S)
+        held.append({'reason':'formal_presentation_scope_and_count_need_review','event_key':'www-1994-spring',
+                     'ordinal':ordinal,'evidence':url,'candidate':{'title':text(title[1]),
+                     'authors_raw':[re.sub(r'\s*<[^>]+>','',text(n)).strip()for n in names],
+                     'keywords_raw':text(re.search(r'<em>Keywords:</em>(.*?)(?=<br\b)',chunk,re.I|re.S)[1]) if re.search(r'<em>Keywords:</em>(.*?)(?=<br\b)',chunk,re.I|re.S)else None},
+                     'source_evidence':sources.proof([url])})
+    if not held:raise ValueError('First-conference presentation listing empty')
+    category_urls={}
+    for href,label in anchors(sources.read(FALL)):
+        if href!='Author_Inst.html' and href.endswith('.html'):
+            category_url=source_url(urljoin(FALL,href))
+            if not category_url.startswith(ROOT+'www2/Proceedings/'):raise ValueError('Unexpected fall category')
+            category_urls[category_url]=label
+    if not category_urls:raise ValueError('Fall archive category listing empty')
+    for url,label in sorted(category_urls.items()):
+        html=sources.read(url)
+        for ordinal,chunk in enumerate(re.split(r'<dt\b[^>]*>',html,flags=re.I)[1:],1):
+            fields=re.split(r'<dd\b[^>]*>',chunk,maxsplit=1,flags=re.I)
+            title=text(fields[0]);raw_byline=text(re.split(r'</dl\b',fields[1],maxsplit=1,flags=re.I)[0]) if len(fields)==2 else None
+            if title:
+                held.append({'reason':'mixed_program_category_not_verified_paper_acceptance','event_key':'www-1994-fall',
+                             'category':label,'ordinal':ordinal,'evidence':url,'candidate':{'title':title,'authors_raw':raw_byline},
+                             'source_evidence':sources.proof([FALL,url])})
+    return held,{'first_conference_fields':sum(r['event_key']=='www-1994-spring'for r in held),'fall_categories':len(category_urls),
+                 'fall_program_entries':sum(r['event_key']=='www-1994-fall'for r in held)}
+
+
+def collect(source_cache,cache_sha256,out,*,reviewed=None,reviewed_sha256=None):
+    out=Path(out)
+    if out.exists():raise ValueError('Refusing to overwrite an existing normalized batch')
+    sources=Sources(source_cache,cache_sha256);groups={};held=[];counts={}
+    reviews,review_sha=reviewed_posters(sources,reviewed,reviewed_sha256,cache_sha256)
+    for track in ('papers','posters'):
+        rows,review,counts[track]=spring_rows(sources,track,reviews if track=='posters'else None);groups[track]=rows;held.extend(review)
+    review,counts['earlier']=early_review(sources);held.extend(review)
+    for item in held:
+        if 'source_evidence' not in item:item['source_evidence']=sources.proof([item['evidence']])
+    out.parent.mkdir(parents=True,exist_ok=True)
+    with tempfile.TemporaryDirectory(prefix='.www-archive-',dir=out.parent)as tmp:
+        stage=Path(tmp);queue=[]
+        for track,rows in groups.items():
+            if not rows:continue
+            filename='www-1995-spring-'+('main'if track=='papers'else'poster')+'.jsonl';raw=''.join(json.dumps(r,ensure_ascii=False)+'\n'for r in rows).encode();(stage/filename).write_bytes(raw)
+            manifest={'source':'www_archive','parser_version':2 if review_sha else 1,'venue':'WWW','year':1995,'track':rows[0]['track'],'event_key':'www-1995-spring',
+                      'list_url':SPRING+track+'/abstract.html','count':len(rows),'status':'partial','file_sha256':hashlib.sha256(raw).hexdigest(),
+                      'missing_abstract_count':sum(not bool((r['abstract']or '').strip())for r in rows),'source_listing_count':counts[track]['abstract_entries'],'held_entry_count':counts[track]['held'],'source_cache_sha256':cache_sha256,'scope':'Verified linked archive entries; held entries and other event tracks excluded, not complete event coverage.'}
+            (stage/filename).with_suffix('.manifest.json').write_text(json.dumps(manifest,indent=2)+'\n');queue.append(str(out/filename))
+        (stage/'review-candidates.jsonl').write_text(''.join(json.dumps(r,ensure_ascii=False)+'\n'for r in held))
+        report={'status':'normalized_not_imported','source_cache_sha256':cache_sha256,'counts':counts,'ready':sum(len(r)for r in groups.values()),'held':len(held),
+                'source_evidence':sources.proof(sources.used),'files':queue,'scope':'Early WWW archive only; no PDF fetching, no rejected/withdrawn inference, no database access. Event-aware importer required.'}
+        if review_sha:report['reviewed_posters_sha256']=review_sha
+        (stage/'report.json').write_text(json.dumps(report,ensure_ascii=False,indent=2)+'\n')
+        (stage/'import-queue.json').write_text(json.dumps({'files':queue},indent=2)+'\n')
+        stage.rename(out)
+    return report
+
+
+def main(argv=None,*,prog=None):
+    p=argparse.ArgumentParser(prog=prog);p.add_argument('--source-cache',required=True);p.add_argument('--source-cache-sha256');p.add_argument('--http-cache');p.add_argument('--out',required=True)
+    p.add_argument('--reviewed-posters');p.add_argument('--reviewed-posters-sha256')
+    args=p.parse_args(argv)
+    if bool(args.http_cache)==bool(args.source_cache_sha256):p.error('Choose a fresh --http-cache download or an existing --source-cache-sha256 snapshot')
+    sha=download_sources(args.http_cache,args.source_cache)if args.http_cache else args.source_cache_sha256
+    if bool(args.reviewed_posters)!=bool(args.reviewed_posters_sha256):p.error('Reviewed poster file and SHA256 must be supplied together')
+    print(json.dumps(collect(args.source_cache,sha,args.out,reviewed=args.reviewed_posters,reviewed_sha256=args.reviewed_posters_sha256),ensure_ascii=False),flush=True)
+
+
+if __name__=='__main__':main()

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/edition_scopes.py

@@ -0,0 +1,38 @@
+"""Retain reviewed event scopes when a source refresh repeats a whole-volume label."""
+import json
+
+KEY = 'edition_scope_curation_v1'
+
+
+def preserve(metadata, previous):
+    old = json.loads(previous).get(KEY) if previous else None
+    if KEY in metadata and metadata[KEY] != old:
+        raise ValueError('Source import cannot supply or replace reviewed edition scope')
+    if old is not None:
+        metadata[KEY] = old
+    return metadata
+
+
+def resolve(c, record, external, venue, year, track, event_key):
+    row = c.execute('SELECT metadata_json FROM source_records WHERE id=?', (record,)).fetchone()
+    entries = json.loads(row[0]).get(KEY, []) if row else []
+    if not isinstance(entries, list):
+        raise ValueError('Invalid reviewed edition scope')
+    applicable = []
+    for entry in entries:
+        if entry['external_id'] != external:
+            continue
+        old, target = entry['from'], entry['target']
+        if int(year) != old['year'] or venue not in entry['venue_names']:
+            continue
+        if track not in (old['track'], target['track']):
+            continue
+        if event_key is not None and event_key != old['event_key']:
+            continue
+        current = c.execute('SELECT venue_id,year,track,event_key FROM editions WHERE id=?', (target['edition_id'],)).fetchone()
+        if not current or [current[i] for i in range(4)] != [target[k] for k in ('venue_id','year','track','event_key')]:
+            raise ValueError('Reviewed target edition has changed')
+        applicable.append(target['edition_id'])
+    if len(applicable) > 1:
+        raise ValueError('Ambiguous reviewed edition scope')
+    return applicable[0] if applicable else None

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))
@@ -91,6 +96,8 @@ class Store:
             if found:
                 matches.add(found[0])
         old = c.execute('SELECT paper_id,title,abstract,retrieved_at,metadata_json FROM source_records WHERE source_id=? AND external_id=?',(source_id:=self.source(source,url),external_id)).fetchone()
+        from paper_atlas.db.edition_scopes import preserve
+        metadata = preserve(dict(metadata or {}), old[4] if old else None)
         if old:
             matches.add(old[0])
         key = normalized(title)
@@ -148,15 +155,32 @@ 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
+        from paper_atlas.db.edition_scopes import resolve
+        edition = resolve(c,record,external_id,venue,year,track,event_key)
+        if edition is None:
+            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/edition_tracks.py

@@ -0,0 +1,245 @@
+"""Snapshot-bound reclassification of existing attempts, without reimport or merge."""
+import argparse
+from collections import Counter
+from contextlib import closing
+import hashlib
+import json
+from pathlib import Path
+import re
+
+from paper_atlas.db.backend import database_arguments, open_read_connection, open_store, inserted_id
+from paper_atlas.db.edition_scopes import KEY
+from paper_atlas.ingestion.event_identity import database_identity, digest, values
+
+MARKER = 'edition_tracks_v1:'
+
+
+def validate(inventory):
+    if (not isinstance(inventory, dict) or inventory.get('format') != 1
+            or not isinstance(inventory.get('records'), list) or not 0 < len(inventory['records']) <= 10000):
+        raise ValueError('Nonempty bounded reviewed inventory required')
+    ids = set(); records = set()
+    for row in inventory['records']:
+        required = {'submission_id','source_record_id','external_id','source','doi','from_edition_id',
+                    'year','venue','venue_names','track','section','evidence_url','evidence_sha256'}
+        if not isinstance(row, dict) or set(row) != required:
+            raise ValueError('Unexpected inventory record fields')
+        for field in ('submission_id','source_record_id','from_edition_id','year'):
+            if type(row[field]) is not int or row[field] <= 0:
+                raise ValueError('Positive integer inventory identity required')
+        if row['submission_id'] in ids or row['source_record_id'] in records:
+            raise ValueError('Repeated submission/source identity in inventory')
+        ids.add(row['submission_id']); records.add(row['source_record_id'])
+        for field in ('external_id','source','doi','venue','section','evidence_url'):
+            if not isinstance(row[field], str) or not row[field].strip():
+                raise ValueError('Missing inventory evidence field')
+        if (not isinstance(row['venue_names'], list) or not row['venue_names']
+                or not all(isinstance(v,str) and v for v in row['venue_names'])
+                or row['venue'] not in row['venue_names']
+                or not re.fullmatch(r'[a-z][a-z0-9_]{0,63}', row['track'])
+                or not re.fullmatch(r'[0-9a-f]{64}', row['evidence_sha256'])
+                or not row['evidence_url'].startswith('https://')):
+            raise ValueError('Invalid scope or evidence binding')
+    return inventory
+
+
+def bounded(c, sql, params):
+    rows = values(c.execute(sql+' LIMIT 100001', params))
+    if len(rows) > 100000:
+        raise ValueError('Protected scope exceeds bounded reclassification limit')
+    return rows
+
+
+def protect(c, ids):
+    marks = ','.join('?' for _ in ids)
+    subs = 'SELECT id FROM submissions WHERE id IN ('+marks+')'
+    papers = 'SELECT paper_id FROM submissions WHERE id IN ('+marks+')'
+    decisions = 'SELECT id FROM decisions WHERE submission_id IN ('+subs+')'
+    queries = {
+        'submissions': ('SELECT id,paper_id,source_record_id,source_submission_id,submitted_at FROM submissions WHERE id IN ('+marks+') ORDER BY id'),
+        'submission_identifiers': ('SELECT scheme,value,submission_id FROM submission_identifiers WHERE submission_id IN ('+subs+') ORDER BY submission_id,scheme,value'),
+        'submission_sources': ('SELECT * FROM submission_sources WHERE submission_id IN ('+subs+') ORDER BY submission_id,source_record_id,source_submission_id'),
+        'decisions': ('SELECT * FROM decisions WHERE submission_id IN ('+subs+') ORDER BY id'),
+        'decision_sources': ('SELECT * FROM decision_sources WHERE decision_id IN ('+decisions+') ORDER BY decision_id,source_record_id,source_decision_id'),
+        'papers': ('SELECT * FROM papers WHERE id IN ('+papers+') ORDER BY id'),
+        'identifiers': ('SELECT * FROM identifiers WHERE paper_id IN ('+papers+') ORDER BY scheme,value'),
+        'paper_tags': ('SELECT * FROM paper_tags WHERE paper_id IN ('+papers+') ORDER BY paper_id,tag_id,evidence_key'),
+    }
+    from paper_atlas.curation.annotations import available
+    if available(c):
+        queries['annotation_events'] = 'SELECT * FROM annotation_events WHERE paper_id IN ('+papers+') ORDER BY id'
+    result = {table: {'rows':len(rows), 'sha256':digest(rows)} for table, sql in queries.items()
+              for rows in [bounded(c,sql,ids)]}
+    rows = bounded(c, 'SELECT r.* FROM source_records r WHERE r.id IN (SELECT source_record_id FROM submission_sources WHERE submission_id IN ('+subs+')) ORDER BY r.id', ids)
+    # Only the reviewed curation field may change; every other metadata value
+    # and all raw text/IDs/retrieval times remain protected.
+    for row in rows:
+        metadata = json.loads(row[8]); metadata.pop(KEY,None); row[8] = metadata
+    result['source_records'] = {'rows':len(rows), 'sha256':digest(rows)}
+    return result
+
+
+def inspect(c, inventory):
+    validate(inventory); changes=[]; ids=tuple(sorted(r['submission_id']for r in inventory['records']))
+    original = sorted({r['from_edition_id'] for r in inventory['records']})
+    actual_ids = {r[0] for r in c.execute('SELECT id FROM submissions WHERE edition_id IN ('+','.join('?'for _ in original)+')', original)}
+    if actual_ids != set(ids):
+        raise ValueError('Inventory must cover every attempt in each original edition')
+    for row in inventory['records']:
+        current = c.execute('''SELECT s.edition_id,s.source_record_id,s.source_submission_id,s.paper_id,
+            e.venue_id,e.year,e.track,e.event_key,v.acronym,r.external_id,r.metadata_json,src.name
+            FROM submissions s JOIN editions e ON e.id=s.edition_id JOIN venues v ON v.id=e.venue_id
+            JOIN source_records r ON r.id=s.source_record_id JOIN sources src ON src.id=r.source_id WHERE s.id=?''',(row['submission_id'],)).fetchone()
+        if not current or [current[i]for i in (0,1,2,5,8,9,11)] != [row[k]for k in ('from_edition_id','source_record_id','external_id','year','venue','external_id','source')]:
+            raise ValueError('Submission identity/scope differs from inventory')
+        aliases={r[0]for r in c.execute('SELECT alias FROM venue_aliases WHERE venue_id=?',(current[4],))}|{current[8]}
+        if set(row['venue_names'])!=aliases:
+            raise ValueError('Venue names differ from current aliases')
+        metadata=json.loads(current[10])
+        if KEY in metadata:
+            raise ValueError('Existing curated scope requires explicit reconciliation')
+        if (metadata.get('record_identifiers') or {}).get('doi') != row['doi']:
+            raise ValueError('Stable DOI differs from reviewed source evidence')
+        links=values(c.execute('SELECT source_record_id,source_submission_id FROM submission_sources WHERE submission_id=?',(row['submission_id'],)))
+        if links != [[row['source_record_id'],row['external_id']]]:
+            raise ValueError('Shared-source attempt requires separate reconciliation')
+        decisions=values(c.execute('SELECT status FROM decisions WHERE submission_id=?',(row['submission_id'],)))
+        if not decisions or any(d[0]!='accepted'for d in decisions):
+            raise ValueError('Only publication observations can use this track correction')
+        target=c.execute('SELECT id FROM editions WHERE venue_id=? AND year=? AND track=? AND event_key=?',(current[4],current[5],row['track'],current[7])).fetchone()
+        if target and target[0] != current[0]:
+            if c.execute('SELECT 1 FROM submissions WHERE edition_id=? LIMIT 1',(target[0],)).fetchone():
+                raise ValueError('Nonempty target edition requires separate reconciliation')
+            if c.execute('SELECT 1 FROM submission_identifiers WHERE edition_id=? LIMIT 1',(target[0],)).fetchone():
+                raise ValueError('Target identifier scope already exists')
+            if c.execute('SELECT 1 FROM edition_sources WHERE edition_id=? LIMIT 1',(target[0],)).fetchone():
+                raise ValueError('Target coverage already exists; reconcile first')
+        if c.execute('SELECT 1 FROM submission_identifiers WHERE submission_id=? AND edition_id<>? LIMIT 1',(row['submission_id'],current[0])).fetchone():
+            raise ValueError('Existing identifier edition is inconsistent')
+        changes.append({**row,'from_track':current[6],'event_key':current[7], 'venue_id':current[4],
+                        'target_edition_id':target[0]if target else None})
+    coverage=bounded(c,'SELECT * FROM edition_sources WHERE edition_id IN ('+','.join('?'for _ in original)+') ORDER BY edition_id,source_id,list_url',original)
+    if Counter(row[0]for row in coverage)!=Counter({edition:1 for edition in original}):
+        raise ValueError('Exactly one original coverage listing per edition required')
+    return {'records':changes, 'protected':protect(c,ids), 'coverage':coverage,
+            'editions':bounded(c,'SELECT * FROM editions WHERE id IN ('+','.join('?'for _ in original)+') ORDER BY id',original)}
+
+
+def proposal(c, inventory, identity):
+    return {'format':1,'database':identity,'inventory':inventory,'before':inspect(c,inventory)}
+
+
+def verify(c, plan):
+    records=plan['before']['records']; ids=tuple(sorted(r['submission_id']for r in records))
+    if protect(c,ids)!=plan['before']['protected']:
+        raise ValueError('Protected history/text/tags/identities changed; roll back')
+    for row in records:
+        event=c.execute('''SELECT e.id,e.venue_id,e.year,e.track,e.event_key FROM submissions s JOIN editions e ON e.id=s.edition_id WHERE s.id=?''',(row['submission_id'],)).fetchone()
+        if not event or [event[i]for i in range(1,5)] != [row['venue_id'],row['year'],row['track'],row['event_key']]:
+            raise ValueError('Target track assignment failed')
+        if c.execute('SELECT 1 FROM submission_identifiers WHERE submission_id=? AND edition_id<>? LIMIT 1',(row['submission_id'],event[0])).fetchone():
+            raise ValueError('Target identifier assignment failed')
+        metadata=json.loads(c.execute('SELECT metadata_json FROM source_records WHERE id=?',(row['source_record_id'],)).fetchone()[0])
+        if row['track'] != row['from_track']:
+            if metadata.get(KEY) != [curation(row,event[0],digest(plan))]:
+                raise ValueError('Reviewed refresh protection missing or changed')
+        elif KEY in metadata:
+            raise ValueError('Unexpected scope revision on unchanged track')
+    original=sorted({r['from_edition_id']for r in records})
+    if bounded(c,'SELECT * FROM editions WHERE id IN ('+','.join('?'for _ in original)+') ORDER BY id',original)!=plan['before']['editions']:
+        raise ValueError('Original edition identities changed')
+    for old in plan['before']['coverage']:
+        edition,source,url,_,_,_,cursor,success,_=old
+        expected=Counter()
+        for row in records:
+            if row['from_edition_id']==edition:
+                target=c.execute('SELECT edition_id FROM submissions WHERE id=?',(row['submission_id'],)).fetchone()[0]
+                expected[target]+=1
+        for target,number in expected.items():
+            actual=c.execute('SELECT status,expected_count,imported_count,cursor,last_success_at,last_error FROM edition_sources WHERE edition_id=? AND source_id=? AND list_url=?',(target,source,url)).fetchone()
+            note='Track scope corrected from reviewed per-record evidence; original whole-volume coverage retained in plan '+digest(plan)
+            if actual is None or [actual[i]for i in range(6)]!=['partial',number,number,cursor,success,note]:
+                raise ValueError('Corrected coverage differs from reviewed scope')
+    return {'records_verified':len(records),'protected':plan['before']['protected']}
+
+
+def curation(row, target, signature):
+    return {'external_id':row['external_id'], 'venue_names':sorted(set(row['venue_names'])),
+            'from':{'edition_id':row['from_edition_id'],'year':row['year'],'track':row['from_track'],'event_key':row['event_key']},
+            'target':{'edition_id':target,'venue_id':row['venue_id'],'year':row['year'],'track':row['track'],'event_key':row['event_key']},
+            'section':row['section'],'evidence_url':row['evidence_url'],'evidence_sha256':row['evidence_sha256'],'plan_digest':signature}
+
+
+def apply_plan(c, plan):
+    if plan.get('format') != 1:
+        raise ValueError('Unsupported plan format')
+    signature=digest(plan);marker=MARKER+signature
+    if c.execute('SELECT 1 FROM schema_migrations WHERE name=?',(marker,)).fetchone():
+        return {'status':'already_applied','events_moved':0,**verify(c,plan)}
+    if inspect(c,plan['inventory']) != plan['before']:
+        raise ValueError('Stale track plan; create and review a fresh plan')
+    targets={}; counts=Counter(); moved=0
+    for row in plan['before']['records']:
+        key=(row['venue_id'],row['year'],row['track'],row['event_key'])
+        if key not in targets:
+            found=c.execute('SELECT id FROM editions WHERE venue_id=? AND year=? AND track=? AND event_key=?',key).fetchone()
+            targets[key]=found[0]if found else inserted_id(c,'INSERT INTO editions(venue_id,year,track,event_key) VALUES(?,?,?,?)',key)
+        target=targets[key];counts[(row['from_edition_id'],target)]+=1
+        if target != row['from_edition_id']:
+            c.execute('UPDATE submission_identifiers SET edition_id=? WHERE submission_id=?',(target,row['submission_id']))
+            if c.execute('UPDATE submissions SET edition_id=? WHERE id=? AND edition_id=?',(target,row['submission_id'],row['from_edition_id'])).rowcount!=1:
+                raise ValueError('Concurrent submission change')
+            metadata=json.loads(c.execute('SELECT metadata_json FROM source_records WHERE id=?',(row['source_record_id'],)).fetchone()[0]);metadata[KEY]=[curation(row,target,signature)]
+            c.execute('UPDATE source_records SET metadata_json=? WHERE id=?',(json.dumps(metadata,ensure_ascii=False),row['source_record_id']));moved+=1
+    # Preserve each original source URL and timestamps, but do not retain its
+    # old whole-volume count as main-track coverage. Partial is deliberate:
+    # this operation verifies scope, not abstracts or all possible sources.
+    for old in plan['before']['coverage']:
+        edition,source,url,status,expected,imported,cursor,success,error=old
+        own={r['submission_id'] for r in plan['before']['records']if r['from_edition_id']==edition}
+        source_name=c.execute('SELECT name FROM sources WHERE id=?',(source,)).fetchone()[0]
+        if {r['source']for r in plan['before']['records']if r['submission_id']in own}!={source_name}:
+            raise ValueError('Coverage includes a source outside the reviewed inventory')
+        for (origin,target),number in counts.items():
+            if origin!=edition:continue
+            note='Track scope corrected from reviewed per-record evidence; original whole-volume coverage retained in plan '+signature
+            c.execute('''INSERT INTO edition_sources(edition_id,source_id,list_url,status,expected_count,imported_count,cursor,last_success_at,last_error)
+                VALUES(?,?,?,?,?,?,?,?,?) ON CONFLICT(edition_id,source_id,list_url) DO UPDATE SET status=excluded.status,
+                expected_count=excluded.expected_count,imported_count=excluded.imported_count,last_error=excluded.last_error''',
+                (target,source,url,'partial',number,number,cursor,success,note))
+    result=verify(c,plan)
+    c.execute('INSERT INTO schema_migrations(name) VALUES(?)',(marker,))
+    return {'status':'applied','events_moved':moved,**result}
+
+
+def run(db,out,*,pg_service=None,pg_schema=None,inventory=None,plan=None,plan_sha256=None):
+    out=Path(out)
+    if out.exists() or not out.parent.is_dir():raise ValueError('New report path in existing directory required')
+    expected=None
+    if plan:
+        raw=Path(plan).read_bytes()
+        if hashlib.sha256(raw).hexdigest()!=plan_sha256:raise ValueError('Plan hash differs')
+        expected=json.loads(raw);validate(expected['inventory'])
+    elif plan_sha256:raise ValueError('Plan path required')
+    if bool(expected)==bool(inventory):raise ValueError('Choose inventory preflight or apply plan')
+    if expected is None:
+        with closing(open_read_connection(db,pg_service=pg_service,pg_schema=pg_schema))as c:
+            result=proposal(c,validate(json.loads(Path(inventory).read_bytes())),database_identity(c,db))
+    else:
+        store=open_store(db,pg_service=pg_service,pg_schema=pg_schema)
+        with closing(store.conn)as c:
+            try:
+                if expected['database']!=database_identity(c,db):raise ValueError('Plan belongs to another database')
+                if not getattr(c,'postgres',False):c.commit();c.execute('BEGIN IMMEDIATE')
+                result=apply_plan(c,expected);c.commit()
+            except BaseException:c.rollback();raise
+    out.write_text(json.dumps(result,ensure_ascii=False,indent=2)+'\n');return result
+
+
+def main(argv=None,*,prog=None):
+    p=argparse.ArgumentParser(prog=prog);database_arguments(p)
+    p.add_argument('--inventory');p.add_argument('--plan');p.add_argument('--plan-sha256');p.add_argument('--out',required=True)
+    a=p.parse_args(argv)
+    if bool(a.plan)!=bool(a.plan_sha256):p.error('--plan and --plan-sha256 must be supplied together')
+    r=run(a.db,a.out,pg_service=a.pg_service,pg_schema=a.pg_schema,inventory=a.inventory,plan=a.plan,plan_sha256=a.plan_sha256)
+    print(json.dumps({'status':r.get('status','planned'),'events_moved':r.get('events_moved',0),'out':a.out}),flush=True)

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,13 @@ 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'))
+                    from paper_atlas.db.edition_scopes import KEY
+                    scope_metadata=json.loads(c.execute('SELECT metadata_json FROM source_records WHERE id=?',(record,)).fetchone()[0])
+                    if scope_metadata.get(KEY):
+                        actual_track=c.execute('SELECT track FROM editions WHERE id=?',(edition,)).fetchone()[0]
+                        if row['track']!=actual_track:
+                            raise ValueError('Recollect this whole-volume source with reviewed per-record tracks before refresh')
                     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_edition_tracks.py

@@ -0,0 +1,159 @@
+"""Stable-attempt correction, rollback and stale whole-volume refresh on both engines."""
+from contextlib import closing, redirect_stdout
+import copy
+import hashlib
+import io
+import json
+import os
+from pathlib import Path
+import tempfile
+import unittest
+import uuid
+
+from paper_atlas.db.backend import open_store
+from paper_atlas.db.edition_scopes import KEY
+from paper_atlas.db.store import Store
+from paper_atlas.ingestion.edition_tracks import proposal,apply_plan,verify,validate,run
+from paper_atlas.ingestion.event_identity import database_identity,values
+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
+
+
+def seed(db):
+    s=Store(db); rows=[]
+    for i,track in enumerate(['main','survey','doctoral_consortium','proceedings_unspecified'],1):
+        external='entry'+str(i);doi='10.1234/entry'+str(i)
+        p,r=s.record('fixture',external,'https://example.org/'+external,'Publication '+external,
+                     'Abstract with NUL\x00 and SOH\x01',[],{'doi':doi})
+        eid=s.decision(p,r,'JOINT',2022,'main',external,'https://example.org/'+external,'accepted','published',identities=[('doi',doi)],observed_at='2022-01-01T00:00:00Z')
+        sid=s.conn.execute('SELECT id FROM submissions WHERE source_record_id=?',(r,)).fetchone()[0]
+        s.record_tags(p,r,'fixture',[{'namespace':'fixture:subject','value':'Original'}])
+        rows.append(dict(submission_id=sid,source_record_id=r,external_id=external,source='fixture',doi=doi,
+                         from_edition_id=eid,year=2022,venue='JOINT',venue_names=['JOINT'],track=track,
+                         section=track,evidence_url='https://example.org/proceedings/2022/',evidence_sha256='a'*64))
+    source=s.conn.execute("SELECT id FROM sources WHERE name='fixture'").fetchone()[0]
+    s.conn.execute("INSERT INTO edition_sources VALUES(?,?,?,'complete',4,4,'checkpoint','2022-01-01T00:00:00Z',NULL)",(eid,source,'https://example.org/volume'))
+    s.conn.commit();s.conn.close();return {'format':1,'records':rows}
+
+
+class TrackCases:
+    def store(self):return open_store(**self.target)
+
+    def plan(self,c):return proposal(c,self.inventory,database_identity(c,self.target.get('db')))
+
+    def test_move_preserves_all_history_tags_text_and_stable_identities(self):
+        with closing(self.store().conn)as c:
+            plan=self.plan(c);before=plan['before']['protected'];r=apply_plan(c,plan);c.commit()
+            self.assertEqual((r['events_moved'],r['records_verified']),(3,4))
+            self.assertEqual(r['protected'],before)
+            self.assertEqual(apply_plan(c,plan)['status'],'already_applied')
+            self.assertEqual(values(c.execute('SELECT status,expected_count,imported_count FROM edition_sources ORDER BY edition_id')),[['partial',1,1]]*4)
+            self.assertEqual(c.execute('SELECT count(*) FROM submissions').fetchone()[0],4)
+            self.assertEqual(c.execute('SELECT count(*) FROM decisions').fetchone()[0],4)
+            self.assertEqual(c.execute('SELECT count(*) FROM paper_merges').fetchone()[0],0)
+
+    def test_source_record_refresh_keeps_scope_and_original_attempt(self):
+        with closing(self.store().conn)as c:
+            plan=self.plan(c);apply_plan(c,plan);c.commit()
+        with closing(self.store().conn)as c:
+            # Use Store methods with the same writer connection as production.
+            s=object.__new__(Store);s.conn=c;s.event_identity=True
+            r=self.inventory['records'][1];metadata=json.loads(c.execute('SELECT metadata_json FROM source_records WHERE id=?',(r['source_record_id'],)).fetchone()[0]);saved=metadata[KEY]
+            p,record=s.record('fixture',r['external_id'],'https://example.org/new','Refreshed title',None,[],{'doi':r['doi']},{'new_source_field':True})
+            eid=s.decision(p,record,'JOINT',2022,'main',r['external_id'],'https://example.org/entry2','accepted','published',identities=[('doi',r['doi'])],observed_at='2022-01-01T00:00:00Z')
+            self.assertEqual(c.execute('SELECT track FROM editions WHERE id=?',(eid,)).fetchone()[0],'survey')
+            self.assertEqual(c.execute('SELECT count(*) FROM submissions').fetchone()[0],4)
+            self.assertEqual(json.loads(c.execute('SELECT metadata_json FROM source_records WHERE id=?',(record,)).fetchone()[0])[KEY],saved)
+            with self.assertRaisesRegex(ValueError,'cannot supply'):
+                s.record('fixture',r['external_id'],'https://example.org/new','Forged',None,[],metadata={KEY:[]})
+            c.rollback()
+
+    def test_stale_snapshot_and_partial_inventory_are_rejected(self):
+        with closing(self.store().conn)as c:
+            plan=self.plan(c)
+            c.execute("UPDATE papers SET title='Changed' WHERE id=(SELECT paper_id FROM submissions WHERE id=?)",(self.inventory['records'][0]['submission_id'],))
+            with self.assertRaisesRegex(ValueError,'Stale'):apply_plan(c,plan)
+            c.rollback()
+            bad=copy.deepcopy(self.inventory);bad['records'].pop()
+            with self.assertRaisesRegex(ValueError,'every attempt'):proposal(c,bad,database_identity(c,self.target.get('db')))
+            bad=copy.deepcopy(self.inventory);bad['records'][0]['doi']='10.1234/wrong'
+            with self.assertRaisesRegex(ValueError,'DOI'):proposal(c,bad,database_identity(c,self.target.get('db')))
+
+    def test_failure_after_updates_rolls_back_all_scope_and_metadata_changes(self):
+        with closing(self.store().conn)as c:
+            plan=self.plan(c)
+            # A late coverage failure must roll back earlier moved attempts.
+            source=c.execute("SELECT id FROM sources WHERE name='fixture'").fetchone()[0]
+            c.execute("INSERT INTO sources(name,homepage) VALUES('other','https://example.org')")
+            other=c.execute("SELECT id FROM sources WHERE name='other'").fetchone()[0]
+            c.execute('UPDATE edition_sources SET source_id=? WHERE source_id=?',(other,source));c.commit()
+            plan=self.plan(c)
+            try:
+                with self.assertRaisesRegex(ValueError,'outside'):apply_plan(c,plan)
+            finally:c.rollback()
+            self.assertEqual(self.plan(c)['before'],plan['before'])
+            self.assertEqual(c.execute('SELECT count(*) FROM editions').fetchone()[0],1)
+
+    def test_import_old_whole_volume_fails_without_corrupting_coverage(self):
+        with closing(self.store().conn)as c:
+            plan=self.plan(c);apply_plan(c,plan);c.commit()
+        r=self.inventory['records'][1];path=self.root/'old.jsonl'
+        row=dict(source='fixture',external_id=r['external_id'],url='https://example.org/entry2',title='Old title',authors=[],identifiers={'doi':r['doi']},venue='JOINT',year=2022,track='main',status='accepted')
+        path.write_text(json.dumps(row)+'\n');path.with_suffix('.manifest.json').write_text(json.dumps(dict(source='fixture',venue='JOINT',year=2022,track='main',count=1,list_url='https://example.org/volume',status='complete')))
+        with self.assertRaisesRegex(ValueError,'Recollect'):ingest_file(self.target.get('db'),path,**{k:v for k,v in self.target.items()if k!='db'})
+        with closing(self.store().conn)as c:verify(c,plan)
+
+    def test_nonempty_target_and_alias_drift_fail_before_moving(self):
+        with closing(self.store().conn)as c:
+            c.execute("INSERT INTO venue_aliases VALUES((SELECT id FROM venues WHERE acronym='JOINT'),'ADDED')")
+            with self.assertRaisesRegex(ValueError,'aliases'):self.plan(c)
+            c.rollback()
+            s=object.__new__(Store);s.conn=c;s.event_identity=True
+            p,r=s.record('fixture','collision','https://example.org/collision','Distinct contribution',None,[],{'doi':'10.1234/collision'})
+            s.decision(p,r,'JOINT',2022,'survey','collision','https://example.org/collision','accepted','published')
+            with self.assertRaisesRegex(ValueError,'Nonempty target'):self.plan(c)
+            c.rollback()
+            self.assertEqual(c.execute('SELECT count(*) FROM submissions').fetchone()[0],4)
+
+    def test_repeated_plan_detects_coverage_or_scope_evidence_drift(self):
+        with closing(self.store().conn)as c:
+            plan=self.plan(c);apply_plan(c,plan);c.commit()
+            c.execute("UPDATE edition_sources SET imported_count=99")
+            with self.assertRaisesRegex(ValueError,'coverage'):apply_plan(c,plan)
+            c.rollback()
+            record=self.inventory['records'][0]['source_record_id']
+            m=json.loads(c.execute('SELECT metadata_json FROM source_records WHERE id=?',(record,)).fetchone()[0]);m[KEY]=[]
+            c.execute('UPDATE source_records SET metadata_json=? WHERE id=?',(json.dumps(m),record))
+            with self.assertRaisesRegex(ValueError,'unchanged track'):apply_plan(c,plan)
+            c.rollback()
+
+    def test_cli_hash_binding_preflight_and_duplicate_inventory(self):
+        inventory=self.root/'inventory.json';inventory.write_text(json.dumps(self.inventory))
+        planfile=self.root/'plan.json';p=run(self.target.get('db'),planfile,inventory=inventory,**{k:v for k,v in self.target.items()if k!='db'})
+        with self.assertRaisesRegex(ValueError,'hash'):run(self.target.get('db'),self.root/'bad.json',plan=planfile,plan_sha256='0'*64,**{k:v for k,v in self.target.items()if k!='db'})
+        report=run(self.target.get('db'),self.root/'applied.json',plan=planfile,plan_sha256=hashlib.sha256(planfile.read_bytes()).hexdigest(),**{k:v for k,v in self.target.items()if k!='db'})
+        self.assertEqual(report['events_moved'],3)
+        bad=copy.deepcopy(self.inventory);bad['records'].append(bad['records'][0])
+        with self.assertRaisesRegex(ValueError,'Repeated'):validate(bad)
+
+
+class SQLiteTrackTests(TrackCases,unittest.TestCase):
+    def setUp(self):
+        t=tempfile.TemporaryDirectory();self.addCleanup(t.cleanup);self.root=Path(t.name);db=self.root/'source.sqlite'
+        self.inventory=seed(db);self.target={'db':db}
+
+
+@unittest.skipUnless(os.environ.get('PAPER_ATLAS_TEST_PG_SERVICE'),'Requires approved remote PostgreSQL service')
+class PostgreSQLTrackTests(TrackCases,unittest.TestCase):
+    def setUp(self):
+        t=tempfile.TemporaryDirectory();self.addCleanup(t.cleanup);self.root=Path(t.name);db=self.root/'source.sqlite';self.inventory=seed(db)
+        self.service=os.environ['PAPER_ATLAS_TEST_PG_SERVICE'];self.schema='pa_tracks_'+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_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()

tests/test_www_archive.py

@@ -0,0 +1,155 @@
+import hashlib
+import json
+from pathlib import Path
+import tempfile
+import unittest
+
+from paper_atlas.collectors.www_archive import (ROOT, SPRING, FALL, Sources, collect, decode,
+                                                early_review, spring_rows)
+
+
+class ArchiveTests(unittest.TestCase):
+    def setUp(self):
+        t=tempfile.TemporaryDirectory();self.addCleanup(t.cleanup);self.root=Path(t.name);self.cache=self.root/'cache';self.cache.mkdir()
+        self.pages={ROOT+'www1/PrelimProcs.html':'''<h1>Formal presentations</h1><ol><li><em>Keywords:</em> BUSINESS TRACK?<br><em>Title:</em><b> First contribution</b><br><em>Author:</em>Jane Doe &lt;jane@example.org&gt;<br><li><em>Title:</em><b>Unlinked contribution</b><br><em>Author:</em>John Roe<br></ol>''',
+                    FALL:'<h1>Conference Tracks</h1><ol><li><a href="Agents.html">Agents</a><li><a href="Tutor.html">Tutorials</a></ol>',
+                    ROOT+'www2/Proceedings/Agents.html':'<h3>Panel</h3><dl><dt><a href="a.html">Linked talk</a><dd>Jane Doe<dt>Unlinked talk</a><dd>John Roe</dl>',
+                    ROOT+'www2/Proceedings/Tutor.html':'<h1>Tutorials</h1><dl><dt>Training session<dd>Teacher</dl>'}
+        for track in ['papers','posters']:
+            self.pages[SPRING+track+'/author.html']='<b>Doe, Jane</b><ul><li><a href="01/paper.html">Same contribution</a></li></ul><b>Roe, John</b><a href="01/paper.html">Same contribution</a>'
+            self.pages[SPRING+track+'/abstract.html']='''<h1>Abstracts</h1><a href="#one">Same contribution</a><h3><a href="01/paper.html">Same contribution</a></h3><i>Author: Jane Doe, John Roe<br>Institute</i><p>Actual abstract with <i>inline emphasis</i> and a separator\u2028 preserved.</p><hr><h2>Another session</h2>'''
+        self.pages[SPRING+'papers/session.html']='<a href="01/paper.html">Same contribution</a>'
+
+    def sources(self):
+        rows=[]
+        for i,(url,html) in enumerate(sorted(self.pages.items())):
+            raw=html.encode()if isinstance(html,str)else html;name='source-'+str(i)+'.html';(self.cache/name).write_bytes(raw)
+            rows.append(dict(url=url,status=200,file=name,bytes=len(raw),sha256=hashlib.sha256(raw).hexdigest()))
+        p=self.cache/'manifest.json';p.write_text(json.dumps({'sources':rows}));sha=hashlib.sha256(p.read_bytes()).hexdigest()
+        return Sources(self.cache,sha),sha
+
+    def test_offline_batch_separates_tracks_review_and_importable_rows(self):
+        _,sha=self.sources();out=self.root/'batch';report=collect(self.cache,sha,out)
+        self.assertEqual(report['ready'],2);self.assertEqual(report['held'],5)
+        queue=json.loads((out/'import-queue.json').read_text())['files'];self.assertEqual(len(queue),2)
+        for name in queue:
+            p=Path(name);rows=[json.loads(l)for l in p.read_text().split('\n') if l];self.assertEqual(len(rows),1)
+            row=rows[0];self.assertEqual(row['event_key'],'www-1995-spring');self.assertEqual(row['authors'],['Jane Doe','John Roe'])
+            self.assertIn(row['track'],['main','poster']);self.assertIn('inline emphasis',row['abstract']);self.assertNotIn('Another session',row['abstract'])
+            self.assertIn('\u2028',row['abstract']);manifest=json.loads(p.with_suffix('.manifest.json').read_text());self.assertEqual(manifest['file_sha256'],hashlib.sha256(p.read_bytes()).hexdigest());self.assertEqual(manifest['status'],'partial')
+        held=[json.loads(l)for l in(out/'review-candidates.jsonl').read_text().split('\n') if l]
+        self.assertTrue(all('status'not in r and 'source_evidence'in r for r in held))
+        self.assertFalse(any('review-candidates'in p for p in queue))
+
+    def test_partial_manifest_preserves_full_listing_size_separately_from_ready_count(self):
+        self.pages[SPRING+'posters/abstract.html']+='<h3><a href="02/other.html">Unlinked poster</a></h3><i>Author: Other Person</i><p>Actual abstract</p>'
+        _,sha=self.sources();out=self.root/'batch';collect(self.cache,sha,out)
+        manifest=json.loads((out/'www-1995-spring-poster.manifest.json').read_text())
+        self.assertEqual((manifest['count'],manifest['source_listing_count'],manifest['held_entry_count']),(1,2,1))
+        self.assertNotIn('coverage_previous_count',manifest)  # Live parent must be independently preflighted.
+
+    def test_malformed_href_requires_exact_id_and_two_clean_index_anchors(self):
+        self.pages[SPRING+'papers/abstract.html']=self.pages[SPRING+'papers/abstract.html'].replace('01/paper.html','01/        paper.html')
+        sources,_=self.sources();rows,held,_=spring_rows(sources,'papers');self.assertEqual(held,[])
+        self.assertEqual(rows[0]['url'],SPRING+'papers/01/paper.html');self.assertTrue(rows[0]['metadata']['url_corroborated_by_author_and_session'])
+        self.assertIn('        ',rows[0]['metadata']['abstract_href_raw'])
+
+    def test_main_index_disagreement_fails_without_publishing_batch(self):
+        self.pages[SPRING+'papers/session.html']='<a href="02/other.html">Other work</a>';_,sha=self.sources();out=self.root/'batch'
+        with self.assertRaisesRegex(ValueError,'ID sets disagree'):collect(self.cache,sha,out)
+        self.assertFalse(out.exists())
+
+    def test_same_entry_id_with_conflicting_clean_url_is_rejected(self):
+        self.pages[SPRING+'papers/session.html']='<a href="01/other.html">Same contribution</a>';sources,_=self.sources()
+        with self.assertRaisesRegex(ValueError,'URLs disagree'):spring_rows(sources,'papers')
+
+    def test_poster_without_linked_author_entry_is_held_even_with_abstract(self):
+        self.pages[SPRING+'posters/author.html']='<b>Doe, Jane</b><a href="#">Same contribution</a>';sources,_=self.sources();rows,held,counts=spring_rows(sources,'posters')
+        self.assertEqual(rows,[]);self.assertEqual(counts['abstract_entries'],1);self.assertEqual(held[0]['reason'],'abstract_entry_missing_linked_author_index')
+
+    def test_unlinked_and_panel_entries_never_become_acceptance_or_rejection(self):
+        sources,_=self.sources();held,counts=early_review(sources);self.assertEqual(counts,{'first_conference_fields':2,'fall_categories':2,'fall_program_entries':3})
+        self.assertTrue(any(r['candidate']['title']=='Unlinked talk'for r in held));self.assertTrue(all('status'not in r for r in held))
+        self.assertEqual(held[0]['candidate']['authors_raw'],['Jane Doe'])
+
+    def test_title_disagreement_goes_to_review_with_source_claims(self):
+        self.pages[SPRING+'papers/abstract.html']=self.pages[SPRING+'papers/abstract.html'].replace('Same contribution','Different contribution');sources,_=self.sources();rows,held,_=spring_rows(sources,'papers')
+        self.assertEqual(rows,[]);self.assertEqual(held[0]['reason'],'abstract_and_author_index_titles_disagree')
+
+    def test_cache_tampering_and_manifest_hash_mismatch_fail_closed(self):
+        _,sha=self.sources()
+        with self.assertRaisesRegex(ValueError,'manifest checksum'):Sources(self.cache,'0'*64)
+        p=next(self.cache.glob('source-*.html'));p.write_bytes(p.read_bytes()+b'x')
+        with self.assertRaisesRegex(ValueError,'checksum or size'):Sources(self.cache,sha)
+
+    def test_cache_cannot_read_paths_outside_snapshot(self):
+        _,sha=self.sources();p=self.cache/'manifest.json';d=json.loads(p.read_text());d['sources'][0]['file']='../outside.html';p.write_text(json.dumps(d))
+        with self.assertRaisesRegex(ValueError,'plain files'):Sources(self.cache,hashlib.sha256(p.read_bytes()).hexdigest())
+
+    def test_keywords_only_is_missing_abstract_not_invented_summary(self):
+        self.pages[SPRING+'papers/abstract.html']='<h3><a href="01/paper.html">Same contribution</a></h3><i>Author: Jane Doe</i><p>Keywords: networking, hypertext</p>';sources,_=self.sources();rows,held,_=spring_rows(sources,'papers')
+        self.assertEqual(rows,[]);self.assertEqual(held[0]['reason'],'missing_abstract')
+
+    def test_old_encoding_and_duplicate_abstract_identity(self):
+        self.assertEqual(decode(b'M\xfcller')[0],'Müller')
+        self.pages[SPRING+'papers/abstract.html']*=2;sources,_=self.sources()
+        with self.assertRaisesRegex(ValueError,'Duplicate abstract'):spring_rows(sources,'papers')
+
+    def test_existing_output_is_never_overwritten(self):
+        _,sha=self.sources();out=self.root/'batch';out.mkdir();(out/'keep').write_text('original')
+        with self.assertRaisesRegex(ValueError,'overwrite'):collect(self.cache,sha,out)
+        self.assertEqual((out/'keep').read_text(),'original')
+
+    def review(self,*,authors=None,author_source='canonical_byline',missing='',page=None):
+        url=SPRING+'posters/01/paper.html'
+        self.pages[url]=page or '<h1>Canonical full title</h1><p>Jane Doe, John Roe</p><p>Keywords: web</p><h2>Abstract</h2><p>Body</p>'
+        _,sha=self.sources();raw=self.pages[url].encode()if isinstance(self.pages[url],str)else self.pages[url]
+        row=dict(entry_id='01',individual_url=url,individual_sha256=hashlib.sha256(raw).hexdigest(),
+                 title='Canonical full title',authors=authors or ['Jane Doe','John Roe'],author_source=author_source,
+                 reason='Reviewed exact canonical entry and its explicit authorship; retain original title variants',missing_abstract_reason=missing)
+        p=self.root/'review.json';p.write_text(json.dumps({'format':1,'source_cache_sha256':sha,'records':[row]}));return sha,p,hashlib.sha256(p.read_bytes()).hexdigest()
+
+    def test_reviewed_variant_keeps_primary_title_and_all_old_title_claims(self):
+        sha,p,h=self.review();out=self.root/'reviewed';r=collect(self.cache,sha,out,reviewed=p,reviewed_sha256=h)
+        row=json.loads((out/'www-1995-spring-poster.jsonl').read_text());self.assertEqual(row['title'],'Canonical full title')
+        self.assertEqual(row['metadata']['abstract_heading_title_raw'],'Same contribution');self.assertEqual(row['metadata']['author_index_titles_raw'],['Same contribution'])
+        self.assertEqual(len(row['metadata']['source_evidence']),3);self.assertEqual(r['reviewed_posters_sha256'],h)
+        self.assertEqual(row['external_id'],'www3:posters:01');self.assertEqual(row['status'],'accepted')
+
+    def test_canonical_byline_can_resolve_unlinked_author_index_without_guessing(self):
+        self.pages[SPRING+'posters/author.html']='<b>Doe, Jane</b><a href="#">Broken index link</a>'
+        sha,p,h=self.review();out=self.root/'reviewed';collect(self.cache,sha,out,reviewed=p,reviewed_sha256=h)
+        row=json.loads((out/'www-1995-spring-poster.jsonl').read_text());self.assertEqual(row['authors'],['Jane Doe','John Roe'])
+        self.assertEqual(row['metadata']['author_index_raw'],[]);self.assertEqual(row['metadata']['author_order'],'canonical_byline')
+
+    def test_reviewed_missing_abstract_is_explicit_partial_not_body_summary(self):
+        self.pages[SPRING+'posters/abstract.html']='<h3><a href="01/paper.html">Same contribution</a></h3><i>Author: Jane Doe</i><p>Keywords: web</p>'
+        sha,p,h=self.review(missing='Neither listing nor canonical page supplies an explicit abstract')
+        out=self.root/'reviewed';collect(self.cache,sha,out,reviewed=p,reviewed_sha256=h)
+        row=json.loads((out/'www-1995-spring-poster.jsonl').read_text());m=json.loads((out/'www-1995-spring-poster.manifest.json').read_text())
+        self.assertIsNone(row['abstract']);self.assertEqual((m['missing_abstract_count'],m['status']),(1,'partial'))
+        sha,p,h=self.review()
+        with self.assertRaisesRegex(ValueError,'Missing abstract'):collect(self.cache,sha,self.root/'bad',reviewed=p,reviewed_sha256=h)
+        self.assertFalse((self.root/'bad').exists())
+
+    def test_reviewed_names_must_be_in_byline_region_not_citations_or_contact_guess(self):
+        sha,p,h=self.review(authors=['Unrelated Person'],page='<h1>Canonical full title</h1><p>Jane Doe</p><h2>Abstract</h2><p>Unrelated Person cited here</p>')
+        with self.assertRaisesRegex(ValueError,'byline region'):collect(self.cache,sha,self.root/'bad',reviewed=p,reviewed_sha256=h)
+        sha,p,h=self.review(author_source='author_index',authors=['Contact Person'])
+        with self.assertRaisesRegex(ValueError,'author-index'):collect(self.cache,sha,self.root/'bad',reviewed=p,reviewed_sha256=h)
+
+    def test_review_hash_cache_binding_and_canonical_h1_are_all_required(self):
+        sha,p,h=self.review()
+        with self.assertRaisesRegex(ValueError,'checksum'):collect(self.cache,sha,self.root/'bad',reviewed=p,reviewed_sha256='0'*64)
+        d=json.loads(p.read_text());d['source_cache_sha256']='0'*64;p.write_text(json.dumps(d));h=hashlib.sha256(p.read_bytes()).hexdigest()
+        with self.assertRaisesRegex(ValueError,'binding'):collect(self.cache,sha,self.root/'bad',reviewed=p,reviewed_sha256=h)
+        d['source_cache_sha256']=sha;d['records'][0]['title']='Wrong title';p.write_text(json.dumps(d));h=hashlib.sha256(p.read_bytes()).hexdigest()
+        with self.assertRaisesRegex(ValueError,'canonical H1'):collect(self.cache,sha,self.root/'bad',reviewed=p,reviewed_sha256=h)
+
+    def test_unlisted_review_entry_and_changed_primary_digest_fail_closed(self):
+        sha,p,h=self.review();d=json.loads(p.read_text());d['records'][0]['individual_sha256']='0'*64;p.write_text(json.dumps(d));h=hashlib.sha256(p.read_bytes()).hexdigest()
+        with self.assertRaisesRegex(ValueError,'page checksum'):collect(self.cache,sha,self.root/'bad',reviewed=p,reviewed_sha256=h)
+        self.pages[SPRING+'posters/abstract.html']=self.pages[SPRING+'posters/abstract.html'].replace('01/paper.html','02/other.html')
+        self.pages[SPRING+'posters/author.html']=self.pages[SPRING+'posters/author.html'].replace('01/paper.html','02/other.html')
+        sha,p,h=self.review()
+        with self.assertRaisesRegex(ValueError,'outside abstract'):collect(self.cache,sha,self.root/'bad',reviewed=p,reviewed_sha256=h)