From 4b309853e7f495fb1b49c1a7bb18b94835203656 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=8D=97=E6=85=B6=E9=BA=9F?= Date: Sun, 20 Sep 2026 18:40:59 +0800 Subject: [PATCH 1/2] Provenance never points across a knowledge base MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Install a schema-level invariant: every reference an export can resolve must join rows that live in the same knowledge base. A column foreign key proves the target exists, not that it is the same KB's — the exporter would otherwise mint local IRIs naming foreign rows, or silently drop vocabulary references that resolve to nothing. Three layers: a precondition scan that refuses the migration on a dirty ledger, per-edge same-KB enforcement (composite (kb_id, ref) foreign keys where the row carries its own kb_id, row triggers where the kb authority is a parent row, deferred keys on same-table self-references so COPY and multi-row inserts are judged at commit), and kb-ownership immutability on every owned table. Signed-off-by: 南慶麟 --- crates/utopia-cli/src/main.rs | 2 +- .../a_dirty_ledger_stops_the_migration.rs | 205 +++++ ...a_forward_reference_is_judged_at_commit.rs | 371 +++++++++ ...gration_0070_runs_under_any_search_path.rs | 781 ++++++++++++++++++ ...e_never_points_across_a_knowledge_base.sql | 769 +++++++++++++++++ 5 files changed, 2127 insertions(+), 1 deletion(-) create mode 100644 crates/utopia-store/tests/a_dirty_ledger_stops_the_migration.rs create mode 100644 crates/utopia-store/tests/a_forward_reference_is_judged_at_commit.rs create mode 100644 crates/utopia-store/tests/migration_0070_runs_under_any_search_path.rs create mode 100644 migrations/0070_provenance_never_points_across_a_knowledge_base.sql diff --git a/crates/utopia-cli/src/main.rs b/crates/utopia-cli/src/main.rs index dfef6b479..20c5f9220 100644 --- a/crates/utopia-cli/src/main.rs +++ b/crates/utopia-cli/src/main.rs @@ -80,7 +80,7 @@ struct ManifestDataDir { /// whose `schema_version` is greater than this (forward-incompatible) and /// warns when older. Kept as a constant — bumping is a deliberate decision, /// not a side effect of a code change. -const CURRENT_SCHEMA_VERSION: u32 = 69; +const CURRENT_SCHEMA_VERSION: u32 = 70; fn main() -> anyhow::Result<()> { dotenvy::dotenv().ok(); diff --git a/crates/utopia-store/tests/a_dirty_ledger_stops_the_migration.rs b/crates/utopia-store/tests/a_dirty_ledger_stops_the_migration.rs new file mode 100644 index 000000000..a10c0fc03 --- /dev/null +++ b/crates/utopia-store/tests/a_dirty_ledger_stops_the_migration.rs @@ -0,0 +1,205 @@ +//! 0070 的 §0 前置检查：往一个**已经有**跨库坏行的库上装 +//! 「出处不许跨库」的不变量，迁移必须确定性中止——报出是哪条边、坏了几行， +//! 而不是装上之后替坏数据背书。 +//! +//! 做法：开一个隔离库，按迁移文件顺序把 ≤0069 的逐个跑掉（每个一个事务， +//! 与 sqlx::migrate 同一语义），手工塞进一条 0070 之前合法、之后非法的行， +//! 再跑 0070 本体： +//! - 干净的 0069 库 → 0070 成功，触发器在场； +//! - 脏的 0069 库 → 0070 报错且报对边名，整个迁移随事务回滚——触发器一个不留。 +//! +//! 建库失败（角色没权限的环境）与没有 UTOPIA_DATABASE_URL 一样处理：跳过。 + +use sqlx::{Acquire, PgPool}; +use uuid::Uuid; + +/// 维护库的地址：`…/utopia` → `…/postgres` +fn admin_url() -> Option { + let url = utopia_store::test_db::url()?; + let (head, _) = url.rsplit_once('/')?; + Some(format!("{head}/postgres")) +} + +/// 按文件顺序跑 ≤ `through` 的迁移，各自一个事务（与 sqlx::migrate 同一形状） +async fn migrate_to(pool: &PgPool, through: i64) -> anyhow::Result<()> { + let migrator = sqlx::migrate!("../../migrations"); + let mut conn = pool.acquire().await?; + for m in migrator.iter().filter(|m| m.version <= through) { + let mut tx = conn.begin().await?; + sqlx::raw_sql(&m.sql).execute(&mut *tx).await?; + tx.commit().await?; + } + Ok(()) +} + +async fn migration_70(pool: &PgPool) -> Result<(), sqlx::Error> { + let migrator = sqlx::migrate!("../../migrations"); + let m = migrator + .iter() + .find(|m| m.version == 70) + .expect("0070 必须在迁移集里"); + let mut conn = pool.acquire().await?; + let mut tx = conn.begin().await?; + let r = sqlx::raw_sql(&m.sql).execute(&mut *tx).await; + match r { + Ok(_) => tx.commit().await, + Err(e) => { + let _ = tx.rollback().await; + Err(e) + } + } +} + +/// 隔离库：建 → 迁到 0069 → 返回（库名, 连接池）。失败就地跳过 +async fn scratch(suffix: &str) -> Option<(String, PgPool)> { + let admin = admin_url()?; + let admin_pool = PgPool::connect(&admin).await.ok()?; + let name = format!("xkb70_{}_{}", suffix, Uuid::now_v7().simple()); + let created = sqlx::query(&format!("CREATE DATABASE {name}")) + .execute(&admin_pool) + .await; + if created.is_err() { + eprintln!("跳过：建不了隔离库（角色没有 CREATEDB）"); + admin_pool.close().await; + return None; + } + let url = utopia_store::test_db::url()?; + let (head, _) = url.rsplit_once('/')?; + let pool = PgPool::connect(&format!("{head}/{name}")).await.ok()?; + if let Err(e) = migrate_to(&pool, 69).await { + eprintln!("跳过：迁到 0069 失败（迁移链自身的问题）: {e}"); + drop_scratch(&name).await; + return None; + } + Some((name, pool)) +} + +async fn drop_scratch(name: &str) { + if let Some(admin) = admin_url() { + if let Ok(pool) = PgPool::connect(&admin).await { + let _ = sqlx::query(&format!("DROP DATABASE IF EXISTS {name} WITH (FORCE)")) + .execute(&pool) + .await; + pool.close().await; + } + } +} + +/// 0070 之前合法的最小坏账：库 A 的文档+段落+实体+事实，库 B 的实体与段落—— +/// 证据行把 A 的事实配到 B 的段落上 +async fn seed_dirty(pool: &PgPool) -> anyhow::Result<()> { + let (org, ws, a, b) = ( + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + ); + sqlx::query("INSERT INTO organizations (id, name) VALUES ($1, 'm70-test')") + .bind(org) + .execute(pool) + .await?; + sqlx::query("INSERT INTO workspaces (id, org_id, name) VALUES ($1, $2, 'm70-test')") + .bind(ws) + .bind(org) + .execute(pool) + .await?; + for kb in [a, b] { + sqlx::query( + "INSERT INTO knowledge_bases (id, workspace_id, name) VALUES ($1, $2, 'm70-test')", + ) + .bind(kb) + .bind(ws) + .execute(pool) + .await?; + } + let (doc_a, doc_b) = (Uuid::now_v7(), Uuid::now_v7()); + let (chunk_b, ent_a, ent_b) = (Uuid::now_v7(), Uuid::now_v7(), Uuid::now_v7()); + for (id, kb) in [(ent_a, a), (ent_b, b)] { + sqlx::query("INSERT INTO entities (id, kb_id, canonical_name) VALUES ($1, $2, 'e')") + .bind(id) + .bind(kb) + .execute(pool) + .await?; + } + for (id, kb, name) in [(doc_a, a, "a.md"), (doc_b, b, "b.md")] { + sqlx::query( + "INSERT INTO documents (id, kb_id, filename, sha256, status, external_key) + VALUES ($1, $2, $3, $4, 'ready', $5)", + ) + .bind(id) + .bind(kb) + .bind(name) + .bind(format!("sha-{name}")) + .bind(format!("file:///{name}")) + .execute(pool) + .await?; + } + // B 库自己的段落——0070 之前把它配到 A 的事实上，什么都不会拦 + sqlx::query( + "INSERT INTO chunks (id, kb_id, document_id, seq, text) VALUES ($1, $2, $3, 0, 'x')", + ) + .bind(chunk_b) + .bind(b) + .bind(doc_b) + .execute(pool) + .await?; + let fact_a = Uuid::now_v7(); + sqlx::query( + "INSERT INTO facts (id, kb_id, subject_id, object_id, confidence) + VALUES ($1, $2, $3, $4, 0.9)", + ) + .bind(fact_a) + .bind(a) + .bind(ent_a) + .bind(ent_a) + .execute(pool) + .await?; + // 坏行：A 的事实配 B 的段落 + sqlx::query("INSERT INTO fact_evidence (fact_id, chunk_id) VALUES ($1, $2)") + .bind(fact_a) + .bind(chunk_b) + .execute(pool) + .await?; + Ok(()) +} + +#[tokio::test] +async fn a_clean_ledger_takes_the_invariant() -> anyhow::Result<()> { + let Some((name, pool)) = scratch("clean").await else { + return Ok(()); + }; + let r = migration_70(&pool).await; + assert!(r.is_ok(), "干净的 0069 库必须装得上 0070: {r:?}"); + let n: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM pg_trigger WHERE tgname = 'fact_evidence_same_kb'", + ) + .fetch_one(&pool) + .await?; + assert_eq!(n, 1, "触发器要真的装上"); + pool.close().await; + drop_scratch(&name).await; + Ok(()) +} + +#[tokio::test] +async fn a_dirty_ledger_stops_the_migration_atomically() -> anyhow::Result<()> { + let Some((name, pool)) = scratch("dirty").await else { + return Ok(()); + }; + seed_dirty(&pool).await?; + let r = migration_70(&pool).await; + let msg = format!("{r:?}"); + assert!(r.is_err(), "脏库上跑 0070 必须中止"); + assert!( + msg.contains("cross-KB references already present") || msg.contains("evidence.chunk"), + "中止要报出坏在哪条边: {msg}" + ); + let n: i64 = + sqlx::query_scalar("SELECT COUNT(*) FROM pg_trigger WHERE tgname LIKE '%same_kb%'") + .fetch_one(&pool) + .await?; + assert_eq!(n, 0, "中止的迁移不许留下半个不变量"); + pool.close().await; + drop_scratch(&name).await; + Ok(()) +} diff --git a/crates/utopia-store/tests/a_forward_reference_is_judged_at_commit.rs b/crates/utopia-store/tests/a_forward_reference_is_judged_at_commit.rs new file mode 100644 index 000000000..01e1f0a2a --- /dev/null +++ b/crates/utopia-store/tests/a_forward_reference_is_judged_at_commit.rs @@ -0,0 +1,371 @@ +//! 同表自指的前向引用在**提交边界**上判（0070 §1b 递延复合外键）。 +//! +//! `facts.supersedes`、`facts.from_statement_id`、`relation_types.inverse_of`、 +//! `relation_types.sub_property_of` 都指着同表的行——恢复/批量装载时目标可能 +//! 在本语句之后才落盘。`(kb_id, ref)` 复合外键把「存在且同库」并成一条约束， +//! `DEFERRABLE INITIALLY DEFERRED` 让它在提交时重估，那时整批都在。 +//! +//! 写入形状分三种，各自的墙不一样： +//! - **多行 INSERT / COPY**：一条语句一批行——递延外键在提交边界看整批， +//! 跨库的过不了，同库的前向链进得来； +//! - **顺序语句**：递延外键同样把判断留到提交——同事务里「先插引用、 +//! 后插目标」现在合法，目标始终不到的提交时被拦； +//! - **replica 会话**（pg_restore --disable-triggers 的形状）：用户触发器 +//! 全关，但外键是内部约束触发器，照样查——同库判定在这一层也在场。 + +use sqlx::{Acquire, PgPool}; +use uuid::Uuid; + +struct Fixture { + org: Uuid, + a: Uuid, + b: Uuid, + ent_a: Uuid, + ent_b: Uuid, +} + +/// 两个库各一件最小零件：实体——supersedes 的合法写法也要它们 +async fn seed(pool: &PgPool) -> anyhow::Result { + let (org, ws, a, b) = ( + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + ); + sqlx::query("INSERT INTO organizations (id, name) VALUES ($1, 'fwdref-test')") + .bind(org) + .execute(pool) + .await?; + sqlx::query("INSERT INTO workspaces (id, org_id, name) VALUES ($1, $2, 'fwdref-test')") + .bind(ws) + .bind(org) + .execute(pool) + .await?; + for kb in [a, b] { + sqlx::query( + "INSERT INTO knowledge_bases (id, workspace_id, name) VALUES ($1, $2, 'fwdref-test')", + ) + .bind(kb) + .bind(ws) + .execute(pool) + .await?; + } + let (ent_a, ent_b) = (Uuid::now_v7(), Uuid::now_v7()); + for (id, kb) in [(ent_a, a), (ent_b, b)] { + sqlx::query("INSERT INTO entities (id, kb_id, canonical_name) VALUES ($1, $2, 'e')") + .bind(id) + .bind(kb) + .execute(pool) + .await?; + } + Ok(Fixture { + org, + a, + b, + ent_a, + ent_b, + }) +} + +async fn cleanup(pool: &PgPool, f: &Fixture) -> anyhow::Result<()> { + for kb in [f.a, f.b] { + sqlx::query("DELETE FROM knowledge_bases WHERE id = $1") + .bind(kb) + .execute(pool) + .await?; + } + sqlx::query("DELETE FROM organizations WHERE id = $1") + .bind(f.org) + .execute(pool) + .await?; + Ok(()) +} + +/// 多行 INSERT：引用行在前、目标行在后——FK 在语句末放行，BEFORE 逐行看时 +/// 目标还不在。**提交边界上的递延约束**是抓住它的地方：跨库的过不了， +/// 同库的照样落 +#[tokio::test] +async fn a_multi_row_insert_is_judged_at_commit() -> anyhow::Result<()> { + let Some(url) = utopia_store::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + utopia_store::db::migrate(&pool).await?; + let f = seed(&pool).await?; + + // 引用行在前、别库目标行在后：BEFORE 看不见目标，递延约束看得见整批 + let (newer, older) = (Uuid::now_v7(), Uuid::now_v7()); + let r = sqlx::query( + "INSERT INTO facts (id, kb_id, subject_id, object_id, confidence, supersedes) + VALUES ($1, $2, $4, $4, 0.9, $3), ($3, $5, $6, $6, 0.9, NULL)", + ) + .bind(newer) + .bind(f.a) + .bind(older) + .bind(f.ent_a) + .bind(f.b) // 目标行落在别库 + .bind(f.ent_b) + .execute(&pool) + .await; + assert!(r.is_err(), "多行 INSERT 里的跨库 supersedes 必须被拒"); + + // 同库的同样写法：两行同库，一条语句——合法 + let (newer2, older2) = (Uuid::now_v7(), Uuid::now_v7()); + sqlx::query( + "INSERT INTO facts (id, kb_id, subject_id, object_id, confidence, supersedes) + VALUES ($1, $2, $4, $4, 0.9, $3), ($3, $2, $4, $4, 0.9, NULL)", + ) + .bind(newer2) + .bind(f.a) + .bind(older2) + .bind(f.ent_a) + .execute(&pool) + .await?; + + cleanup(&pool, &f).await +} + +/// COPY 是恢复灌库的形状：先放行后校验只在语句提交边界做一次。别库目标 +/// 排在本批后面也过不了那道闸；同库的前向链照样进得来 +#[tokio::test] +async fn a_copy_batch_is_judged_at_commit() -> anyhow::Result<()> { + let Some(url) = utopia_store::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + utopia_store::db::migrate(&pool).await?; + let f = seed(&pool).await?; + + // 跨库：newer 在前、older 在后，older 属 B 库——提交时被拦 + let (newer, older) = (Uuid::now_v7(), Uuid::now_v7()); + let mut conn = pool.acquire().await?; + let mut tx = conn.begin().await?; + let mut copy = tx + .copy_in_raw( + "COPY public.facts (id, kb_id, subject_id, object_id, confidence, supersedes) FROM stdin", + ) + .await?; + copy.send( + format!( + "{newer}\t{a}\t{ent_a}\t{ent_a}\t0.9\t{older}\n{older}\t{b}\t{ent_b}\t{ent_b}\t0.9\t\\N\n", + a = f.a, + b = f.b, + ent_a = f.ent_a, + ent_b = f.ent_b, + ) + .into_bytes(), + ) + .await?; + copy.finish().await?; + let r = tx.commit().await; + assert!(r.is_err(), "COPY 批里的跨库 supersedes 必须在提交时被拦下"); + + // 同库：同样的前向顺序，整条链合法 + let (newer2, older2) = (Uuid::now_v7(), Uuid::now_v7()); + let mut tx = conn.begin().await?; + let mut copy = tx + .copy_in_raw( + "COPY public.facts (id, kb_id, subject_id, object_id, confidence, supersedes) FROM stdin", + ) + .await?; + copy.send( + format!( + "{newer2}\t{a}\t{ent_a}\t{ent_a}\t0.9\t{older2}\n{older2}\t{a}\t{ent_a}\t{ent_a}\t0.9\t\\N\n", + a = f.a, + ent_a = f.ent_a, + ) + .into_bytes(), + ) + .await?; + copy.finish().await?; + tx.commit().await?; + drop(conn); + + cleanup(&pool, &f).await +} + +/// 顺序写的前向引用同样归到提交边界：递延外键让「先插引用、后插目标」 +/// 在事务里合法，目标始终不到的写法在 COMMIT 被拦下 +#[tokio::test] +async fn a_sequential_forward_reference_is_judged_at_commit() -> anyhow::Result<()> { + let Some(url) = utopia_store::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + utopia_store::db::migrate(&pool).await?; + let f = seed(&pool).await?; + + // 目标始终不到：语句放行、提交时被拦 + let (newer, older) = (Uuid::now_v7(), Uuid::now_v7()); + let mut conn = pool.acquire().await?; + let mut tx = conn.begin().await?; + sqlx::query( + "INSERT INTO facts (id, kb_id, subject_id, object_id, confidence, supersedes) + VALUES ($1, $2, $3, $3, 0.9, $4)", + ) + .bind(newer) + .bind(f.a) + .bind(f.ent_a) + .bind(older) + .execute(&mut *tx) + .await?; + let r = tx.commit().await; + assert!(r.is_err(), "目标始终不到的 supersedes 顺序写，提交时必须报"); + + // 同事务里目标后到：整条链在提交时齐了——合法 + let (newer2, older2) = (Uuid::now_v7(), Uuid::now_v7()); + let mut tx = conn.begin().await?; + sqlx::query( + "INSERT INTO facts (id, kb_id, subject_id, object_id, confidence, supersedes) + VALUES ($1, $2, $3, $3, 0.9, $4)", + ) + .bind(newer2) + .bind(f.a) + .bind(f.ent_a) + .bind(older2) + .execute(&mut *tx) + .await?; + sqlx::query( + "INSERT INTO facts (id, kb_id, subject_id, object_id, confidence) + VALUES ($1, $2, $3, $3, 0.9)", + ) + .bind(older2) + .bind(f.a) + .bind(f.ent_a) + .execute(&mut *tx) + .await?; + tx.commit().await?; + drop(conn); + + cleanup(&pool, &f).await +} + +/// UPDATE 装上的跨库 supersedes：目标已存在，BEFORE 逐行检查看得见它—— +/// 当场就报；真漏过去的那一层，递延约束在提交时兜底 +#[tokio::test] +async fn an_update_to_a_foreign_supersedes_fails() -> anyhow::Result<()> { + let Some(url) = utopia_store::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + utopia_store::db::migrate(&pool).await?; + let f = seed(&pool).await?; + + let (fa, fb) = (Uuid::now_v7(), Uuid::now_v7()); + sqlx::query( + "INSERT INTO facts (id, kb_id, subject_id, object_id, confidence) + VALUES ($1, $2, $3, $3, 0.9), ($4, $5, $6, $6, 0.9)", + ) + .bind(fa) + .bind(f.a) + .bind(f.ent_a) + .bind(fb) + .bind(f.b) + .bind(f.ent_b) + .execute(&pool) + .await?; + + let mut conn = pool.acquire().await?; + let mut tx = conn.begin().await?; + let upd = sqlx::query("UPDATE facts SET supersedes = $2 WHERE id = $1") + .bind(fa) + .bind(fb) + .execute(&mut *tx) + .await; + if upd.is_ok() { + let r = tx.commit().await; + assert!(r.is_err(), "UPDATE 装上的跨库 supersedes 最迟在提交时要报"); + } + + cleanup(&pool, &f).await +} + +/// from_statement_id 是同一张表上的第二条自指边：陈述行在批尾、类型化事实 +/// 在批头的跨库写法，提交时一样被拦 +#[tokio::test] +async fn a_from_statement_is_judged_at_commit() -> anyhow::Result<()> { + let Some(url) = utopia_store::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + utopia_store::db::migrate(&pool).await?; + let f = seed(&pool).await?; + + // 类型化事实在前、别库陈述在后：BEFORE 看不见目标，递延约束看得见整批 + let (typed, stmt) = (Uuid::now_v7(), Uuid::now_v7()); + let r = sqlx::query( + "INSERT INTO facts (id, kb_id, subject_id, object_id, confidence, + layer, phrase, from_statement_id) + VALUES ($1, $2, $4, $4, 0.9, 'typed', NULL, $3), + ($3, $5, $6, $6, 0.9, 'open', 'joined', NULL)", + ) + .bind(typed) + .bind(f.a) + .bind(stmt) + .bind(f.ent_a) + .bind(f.b) // 陈述行落在别库 + .bind(f.ent_b) + .execute(&pool) + .await; + assert!( + r.is_err(), + "多行 INSERT 里的跨库 from_statement_id 必须被拒" + ); + + // 同库的同样写法：类型化事实在前、同库陈述在后——合法 + let (typed2, stmt2) = (Uuid::now_v7(), Uuid::now_v7()); + sqlx::query( + "INSERT INTO facts (id, kb_id, subject_id, object_id, confidence, + layer, phrase, from_statement_id) + VALUES ($1, $2, $4, $4, 0.9, 'typed', NULL, $3), + ($3, $2, $4, $4, 0.9, 'open', 'joined', NULL)", + ) + .bind(typed2) + .bind(f.a) + .bind(stmt2) + .bind(f.ent_a) + .execute(&pool) + .await?; + + cleanup(&pool, &f).await +} + +/// 关系的同表自指同一条边界：inverse_of / sub_property_of 引用行在前、 +/// 目标行在后——别库的在提交时被拦，同库的放行 +#[tokio::test] +async fn relation_self_links_are_judged_at_commit() -> anyhow::Result<()> { + let Some(url) = utopia_store::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + utopia_store::db::migrate(&pool).await?; + let f = seed(&pool).await?; + + // inverse_of：一条语句里 A 库的谓词在前、B 库的目标在后——提交时报 + let (inv, tgt) = (Uuid::now_v7(), Uuid::now_v7()); + let r = sqlx::query( + "INSERT INTO relation_types (id, kb_id, key, label, inverse_of) + VALUES ($1, $2, 'inv', 'inv', $3), ($3, $4, 'tgt', 'tgt', NULL)", + ) + .bind(inv) + .bind(f.a) + .bind(tgt) + .bind(f.b) + .execute(&pool) + .await; + assert!(r.is_err(), "多行 INSERT 里的跨库 inverse_of 必须被拒"); + + // sub_property_of 同库、目标行在后：合法 + let (child, parent) = (Uuid::now_v7(), Uuid::now_v7()); + sqlx::query( + "INSERT INTO relation_types (id, kb_id, key, label, sub_property_of) + VALUES ($1, $2, 'child', 'child', $3), ($3, $2, 'parent', 'parent', NULL)", + ) + .bind(child) + .bind(f.a) + .bind(parent) + .execute(&pool) + .await?; + + cleanup(&pool, &f).await +} diff --git a/crates/utopia-store/tests/migration_0070_runs_under_any_search_path.rs b/crates/utopia-store/tests/migration_0070_runs_under_any_search_path.rs new file mode 100644 index 000000000..ba362e284 --- /dev/null +++ b/crates/utopia-store/tests/migration_0070_runs_under_any_search_path.rs @@ -0,0 +1,781 @@ +//! 0070 的 DDL 不许依赖会话的 search_path：`CREATE FUNCTION`、 +//! `CREATE TRIGGER ... ON`、`EXECUTE FUNCTION` 全部限定到 `public.*`—— +//! pg_restore 会把会话 search_path 置空再灌数据，首位被占住的会话也不能 +//! 把函数建到别的 schema 去。落在别处的触发器等于没有触发器。 +//! +//! 三种会话下逐个验证： +//! - 正常 search_path（默认）→ 装上； +//! - `SET LOCAL search_path = ''` → 同样装上，函数落在 public； +//! - `SET LOCAL search_path = 'decoy'`（先在首位摆上同名干扰物）→ +//! 照样装进 public，干扰物一个不被调用。 +//! +//! 没有 `UTOPIA_DATABASE_URL` 时跳过；设了地址却连不上、或建不了库——**失败**， +//! 不是跳过：地址都给了还说「没库」是假话，那个绿色等于这条检查没跑过。 + +use sqlx::{Acquire, PgPool}; +use std::collections::{HashMap, HashSet}; +use uuid::Uuid; + +fn admin_url() -> Option { + let url = utopia_store::test_db::url()?; + let (head, _) = url.rsplit_once('/')?; + Some(format!("{head}/postgres")) +} + +/// 按文件顺序跑 ≤ `through` 的迁移，各自一个事务（与 sqlx::migrate 同一形状） +async fn migrate_to(pool: &PgPool, through: i64) -> anyhow::Result<()> { + let migrator = sqlx::migrate!("../../migrations"); + let mut conn = pool.acquire().await?; + for m in migrator.iter().filter(|m| m.version <= through) { + let mut tx = conn.begin().await?; + sqlx::raw_sql(&m.sql).execute(&mut *tx).await?; + tx.commit().await?; + } + Ok(()) +} + +/// 在 `search_path` 为 `path` 的事务里跑 0070 本体 +async fn migration_70_under(pool: &PgPool, path: &str) -> Result<(), sqlx::Error> { + let migrator = sqlx::migrate!("../../migrations"); + let m = migrator + .iter() + .find(|m| m.version == 70) + .expect("0070 必须在迁移集里"); + let mut conn = pool.acquire().await?; + let mut tx = conn.begin().await?; + sqlx::query(&format!("SET LOCAL search_path = {path}")) + .execute(&mut *tx) + .await?; + let r = sqlx::raw_sql(&m.sql).execute(&mut *tx).await; + match r { + Ok(_) => tx.commit().await, + Err(e) => { + let _ = tx.rollback().await; + Err(e) + } + } +} + +/// 隔离库：建 → 迁到 0069 → 返回（库名, 连接池）。 +/// 跳过只有一种情形：**根本没设** `UTOPIA_DATABASE_URL`。设了地址连不上、 +/// 建不了库、迁移链跑不动，全都以错误返回——给了地址却拿不到库，这次检查 +/// 就是没有执行过，不该被记成绿色 +async fn scratch(suffix: &str) -> anyhow::Result> { + let Some(admin) = admin_url() else { + return Ok(None); + }; + let admin_pool = PgPool::connect(&admin).await?; + let name = format!("xkb70sp_{}_{}", suffix, Uuid::now_v7().simple()); + sqlx::query(&format!("CREATE DATABASE {name}")) + .execute(&admin_pool) + .await?; + admin_pool.close().await; + let Some(url) = utopia_store::test_db::url() else { + drop_scratch(&name).await; + return Ok(None); + }; + let (head, _) = url.rsplit_once('/').expect("UTOPIA_DATABASE_URL 缺库名段"); + let pool = PgPool::connect(&format!("{head}/{name}")).await?; + if let Err(e) = migrate_to(&pool, 69).await { + pool.close().await; + drop_scratch(&name).await; + return Err(e); + } + Ok(Some((name, pool))) +} + +async fn drop_scratch(name: &str) { + if let Some(admin) = admin_url() { + if let Ok(pool) = PgPool::connect(&admin).await { + let _ = sqlx::query(&format!("DROP DATABASE IF EXISTS {name} WITH (FORCE)")) + .execute(&pool) + .await; + pool.close().await; + } + } +} + +/// 在隔离库上跑 `f`——不管成功、失败还是断言 panic，库都清掉再走： +/// 测失败的证据不许靠运维去捡烂尾库 +async fn with_scratch(suffix: &str, f: impl FnOnce(PgPool) -> Fut) -> anyhow::Result<()> +where + Fut: std::future::Future>, +{ + use futures_util::FutureExt; + use std::panic::AssertUnwindSafe; + let Some((name, pool)) = scratch(suffix).await? else { + return Ok(()); + }; + // catch_unwind：断言 panic 落在 Err 上，清库照常走到再重抛 + let r = AssertUnwindSafe(f(pool.clone())).catch_unwind().await; + pool.close().await; + drop_scratch(&name).await; + match r { + Ok(inner) => inner, + Err(p) => std::panic::resume_unwind(p), + } +} + +/// 装完后要点名的几件东西：声明式边 26 条复合外键、父行作证边 10 个触发器、 +/// 库不可过户 12 个触发器、支撑唯一约束 8 条，且所有新函数都在 public schema。 +/// 同表自指边的提交边界检查由 DEFERRABLE INITIALLY DEFERRED 复合外键承担 +/// （supersedes / from_statement_id / inverse_of / sub_property_of） +async fn assert_installed(pool: &PgPool) -> anyhow::Result<()> { + let fns: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM pg_proc p JOIN pg_namespace n ON n.oid = p.pronamespace + WHERE n.nspname = 'public' AND p.proname IN ( + 'fact_evidence_stays_inside_its_kb', + 'derivation_premise_stays_inside_its_kb','qualifier_stays_inside_its_facts_kb', + 'type_parent_stays_inside_its_kb', + 'relation_scope_stays_inside_its_kb', + 'relation_qualifier_stays_inside_the_kb', + 'rule_condition_refs_stay_inside_the_kb','kb_ownership_is_not_reassigned', + 'typed_source_stays_inside_its_kb', + 'squalifier_stays_inside_its_facts_kb')", + ) + .fetch_one(pool) + .await?; + assert_eq!(fns, 10, "全部触发器函数都必须落在 public schema"); + + // 复合外键：每条源行自带 kb_id 的边一条，26 条 + let fks: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM pg_constraint WHERE contype = 'f' AND conname IN ( + 'chunks_document_same_kb', + 'facts_subject_same_kb','facts_object_same_kb','facts_predicate_same_kb', + 'facts_supersedes_same_kb','facts_from_statement_same_kb', + 'derived_facts_subject_same_kb','derived_facts_object_same_kb', + 'derived_facts_predicate_same_kb','derived_facts_rule_same_kb', + 'derived_facts_attribute_rule_same_kb', + 'entities_type_same_kb', + 'entity_type_disjoint_a_same_kb','entity_type_disjoint_b_same_kb', + 'relation_types_inverse_same_kb','relation_types_sub_property_same_kb', + 'rules_predicate_same_kb', + 'attribute_rules_subject_type_same_kb','attribute_rules_conclude_type_same_kb', + 'attribute_rules_conclude_predicate_same_kb', + 'time_mentions_fact_same_kb','time_mentions_chunk_same_kb', + 'type_bindings_type_same_kb', + 'phrase_bindings_subject_type_same_kb','phrase_bindings_object_type_same_kb', + 'phrase_bindings_relation_type_same_kb')", + ) + .fetch_one(pool) + .await?; + assert_eq!(fks, 26, "每条声明式边一个复合外键，一条都不能少"); + + // 支撑唯一约束：被引表一家一个 (kb_id, id) + let uniques: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM pg_constraint WHERE contype = 'u' AND conname IN ( + 'documents_kb_id_key','chunks_kb_id_key','entities_kb_id_key', + 'entity_types_kb_id_key','relation_types_kb_id_key','facts_kb_id_key', + 'rules_kb_id_key','attribute_rules_kb_id_key')", + ) + .fetch_one(pool) + .await?; + assert_eq!(uniques, 8, "每个被引表一条 (kb_id, id) 唯一约束"); + + // 同表自指边真的递延：condeferrable 与 condeferred 两个位都立着 + let deferred: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM pg_constraint + WHERE contype = 'f' AND condeferrable AND condeferred + AND conname IN ('facts_supersedes_same_kb', + 'facts_from_statement_same_kb', + 'relation_types_inverse_same_kb', + 'relation_types_sub_property_same_kb')", + ) + .fetch_one(pool) + .await?; + assert_eq!(deferred, 4, "同表自指边的提交边界检查必须在场"); + + // 父行作证的边与库不可过户：触发器计数（FK 自带的内部触发器不算） + let trg: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM pg_trigger WHERE NOT tgisinternal AND tgname IN ( + 'fact_evidence_same_kb','fact_derivations_same_kb', + 'fact_qualifiers_same_kb','entity_type_parents_same_kb', + 'relation_type_domains_same_kb','relation_type_ranges_same_kb', + 'relation_type_qualifiers_same_kb','attribute_rule_conditions_same_kb', + 'typed_fact_sources_same_kb','statement_qualifiers_same_kb', + 'facts_keep_their_kb','derived_facts_keep_their_kb','documents_keep_their_kb', + 'entities_keep_their_kb','entity_types_keep_their_kb','relation_types_keep_their_kb', + 'rules_keep_their_kb','attribute_rules_keep_their_kb', + 'entity_type_disjoint_keep_their_kb', + 'time_mentions_keep_their_kb','type_bindings_keep_their_kb', + 'phrase_bindings_keep_their_kb')", + ) + .fetch_one(pool) + .await?; + assert_eq!(trg, 22, "父行作证的边与不可过户，一条都不能少"); + Ok(()) +} + +// ===================================================================== +// 完备性守卫:0070 的覆盖面不靠「数过的约束名/触发器名」维持——那种数法 +// 会在有人往账本里加了一条引用列、却没人回来改 0070 的时候保持全绿。 +// 这里反过来:从 pg_catalog 现数责任面里的每一条列级引用边,逐条归类, +// 归不进任何一类就红;登记表里有而 catalog 里没有,也是红。 +// ===================================================================== + +/// 0070 的责任面:语义账本的表——导出逐边解析引用的那些。 +/// 这是范围声明而不是覆盖断言:一张新表要进账本,得先把表名登记进来, +/// 它的引用边才轮到逐条归类;面外表(审计、暂存、运维)不归这条不变量管。 +const LEDGER_TABLES: &[&str] = &[ + // 自带 kb_id 的语义行 + "attribute_rules", + "chunks", + "derived_facts", + "documents", + "entities", + "entity_type_disjoint", + "entity_types", + "facts", + "phrase_bindings", + "relation_types", + "rules", + "time_mentions", + "type_bindings", + // 行自己没有 kb_id、kb 权威由父行作证的连接行 + "attribute_rule_conditions", + "entity_type_parents", + "fact_derivations", + "fact_evidence", + "fact_qualifiers", + "relation_type_domains", + "relation_type_qualifiers", + "relation_type_ranges", + "statement_qualifiers", + "typed_fact_sources", +]; + +/// 父行作证表:(表, owner 列, owner 表)。owner 边给出该行的 kb 权威, +/// 它本身不是「同库语义引用」,自动归 EXPLICIT_NON_SCOPE;其余指向带 +/// kb_id 表的边都必须登记到 TRIGGER_EDGES。 +const OWNER_ROWS: &[(&str, &str, &str)] = &[ + ("attribute_rule_conditions", "rule_id", "attribute_rules"), + ("entity_type_parents", "child_id", "entity_types"), + ("fact_derivations", "derived_fact_id", "derived_facts"), + ("fact_evidence", "fact_id", "facts"), + ("fact_qualifiers", "fact_id", "facts"), + ( + "relation_type_domains", + "relation_type_id", + "relation_types", + ), + ( + "relation_type_qualifiers", + "relation_type_id", + "relation_types", + ), + ("relation_type_ranges", "relation_type_id", "relation_types"), + ("statement_qualifiers", "fact_id", "facts"), + ("typed_fact_sources", "fact_id", "facts"), +]; + +/// 触发器边登记:(表, 列, 触发器名)。登记是列级的——「这张表装着触发器」 +/// 不蕴含「新加的列被它检查」:除登记外,还核对列落在该触发器的 +/// UPDATE OF 清单里。表上新增一条指向带 kb_id 表的引用列而不登记, +/// 就归不进任何一类。 +const TRIGGER_EDGES: &[(&str, &str, &str)] = &[ + ( + "attribute_rule_conditions", + "predicate_id", + "attribute_rule_conditions_same_kb", + ), + ( + "entity_type_parents", + "parent_id", + "entity_type_parents_same_kb", + ), + ( + "fact_derivations", + "premise_derived_id", + "fact_derivations_same_kb", + ), + ( + "fact_derivations", + "premise_fact_id", + "fact_derivations_same_kb", + ), + ("fact_evidence", "chunk_id", "fact_evidence_same_kb"), + ("fact_evidence", "document_id", "fact_evidence_same_kb"), + ("fact_qualifiers", "entity_id", "fact_qualifiers_same_kb"), + ( + "fact_qualifiers", + "qualifier_type_id", + "fact_qualifiers_same_kb", + ), + ( + "relation_type_domains", + "entity_type_id", + "relation_type_domains_same_kb", + ), + ( + "relation_type_qualifiers", + "qualifier_type_id", + "relation_type_qualifiers_same_kb", + ), + ( + "relation_type_ranges", + "entity_type_id", + "relation_type_ranges_same_kb", + ), + ( + "statement_qualifiers", + "entity_id", + "statement_qualifiers_same_kb", + ), + ( + "typed_fact_sources", + "statement_id", + "typed_fact_sources_same_kb", + ), +]; + +/// 面内但不属同库语义引用的边——一条一句话理由。owner 边不列在这里, +/// 由 OWNER_ROWS 自动豁免。 +const NON_SCOPE: &[(&str, &str, &str)] = &[ + // 采集来源归属:导出从不把这条引用解析成 IRI + ( + "documents", + "source_id", + "ingest attribution, not an export-resolved reference", + ), + // 合并簿记:merged_into 非空的行不进导出,这条指针永不进 IRI + ( + "entities", + "merged_into", + "merge bookkeeping — merged rows never reach export", + ), +]; + +/// pg_catalog 里数出来的一条列级引用边。`pairs` 是同一约束按 +/// conkey/confkey 序数对齐出的全部列对——一条复合外键给出多行边, +/// 它们共享同一份 pairs。 +struct RefEdge { + src_table: String, + src_col: String, + tgt_table: String, + tgt_col: String, + conname: String, + deferrable: bool, + deferred: bool, + pairs: Vec<(String, String)>, +} + +impl RefEdge { + fn label(&self) -> String { + format!( + "{}.{}\u{2192}{}.{}", + self.src_table, self.src_col, self.tgt_table, self.tgt_col + ) + } +} + +/// 从 pg_catalog 现数 public schema 的全部外键列对——不读迁移源码, +/// conkey/confkey 按序数对齐,不靠数组位置猜。 +async fn catalog_edges(pool: &PgPool) -> anyhow::Result> { + let rows = sqlx::query_as::<_, (i64, String, String, String, String, String, bool, bool)>( + "SELECT c.oid::bigint, + src.relname, sa.attname, tgt.relname, ta.attname, + c.conname, c.condeferrable, c.condeferred + FROM pg_constraint c + JOIN pg_class src ON src.oid = c.conrelid + JOIN pg_namespace sn + ON sn.oid = src.relnamespace AND sn.nspname = 'public' + JOIN pg_class tgt ON tgt.oid = c.confrelid + JOIN pg_namespace tn + ON tn.oid = tgt.relnamespace AND tn.nspname = 'public' + JOIN unnest(c.conkey) WITH ORDINALITY AS ck(attnum, ord) ON true + JOIN pg_attribute sa + ON sa.attrelid = c.conrelid AND sa.attnum = ck.attnum + JOIN pg_attribute ta + ON ta.attrelid = c.confrelid AND ta.attnum = c.confkey[ck.ord] + WHERE c.contype = 'f' + ORDER BY c.oid, ck.ord", + ) + .fetch_all(pool) + .await?; + let mut pairs: HashMap> = HashMap::new(); + for (oid, _, sc, _, tc, _, _, _) in &rows { + pairs + .entry(*oid) + .or_default() + .push((sc.clone(), tc.clone())); + } + Ok(rows + .into_iter() + .map(|(oid, st, sc, tt, tc, cn, dfr, dfd)| RefEdge { + pairs: pairs.get(&oid).cloned().unwrap_or_default(), + src_table: st, + src_col: sc, + tgt_table: tt, + tgt_col: tc, + conname: cn, + deferrable: dfr, + deferred: dfd, + }) + .collect()) +} + +/// 带 kb_id 列的表 = 有库归属的行。 +async fn kb_scoped_tables(pool: &PgPool) -> anyhow::Result> { + Ok(sqlx::query_scalar::<_, String>( + "SELECT c.relname FROM pg_class c + JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE n.nspname = 'public' AND c.relkind = 'r' + AND EXISTS (SELECT 1 FROM pg_attribute a + WHERE a.attrelid = c.oid AND a.attname = 'kb_id' + AND a.attnum > 0 AND NOT a.attisdropped)", + ) + .fetch_all(pool) + .await? + .into_iter() + .collect()) +} + +/// 用户触发器 -> 它 UPDATE OF 盯着的列名集合(tgattr)。 +async fn trigger_watch_lists( + pool: &PgPool, +) -> anyhow::Result>> { + let rows = sqlx::query_as::<_, (String, String, Vec)>( + "SELECT cls.relname, t.tgname, COALESCE(w.cols, '{}'::text[]) + FROM pg_trigger t + JOIN pg_class cls ON cls.oid = t.tgrelid + JOIN pg_namespace n ON n.oid = cls.relnamespace AND n.nspname = 'public' + LEFT JOIN LATERAL ( + SELECT array_agg(a.attname) AS cols + FROM pg_attribute a + WHERE a.attrelid = t.tgrelid + AND a.attnum = ANY (string_to_array(t.tgattr::text, ' ')::smallint[]) + ) w ON true + WHERE NOT t.tgisinternal", + ) + .fetch_all(pool) + .await?; + Ok(rows + .into_iter() + .map(|(t, g, c)| ((t, g), c.into_iter().collect())) + .collect()) +} + +/// 身上装着 0070 机制的表(同库约束/触发器、不可过户触发器、支撑唯一约束)—— +/// 用来与 LEDGER_TABLES 互证:机制落在面外、或面内表什么机制都没有,都算漂移。 +async fn mechanism_tables(pool: &PgPool) -> anyhow::Result> { + Ok(sqlx::query_scalar::<_, String>( + "SELECT cls.relname FROM pg_constraint c + JOIN pg_class cls ON cls.oid = c.conrelid + JOIN pg_namespace n ON n.oid = cls.relnamespace AND n.nspname = 'public' + WHERE c.conname::text ~ '(_same_kb|_kb_id_key)$' + UNION + SELECT cls.relname FROM pg_trigger t + JOIN pg_class cls ON cls.oid = t.tgrelid + JOIN pg_namespace n ON n.oid = cls.relnamespace AND n.nspname = 'public' + WHERE NOT t.tgisinternal + AND t.tgname::text ~ '(_same_kb|_keep_their_kb)$'", + ) + .fetch_all(pool) + .await? + .into_iter() + .collect()) +} + +/// 逐条归类的结果。declarative / trigger_covered / non_scope 之外 +/// 任何一桶非空,覆盖就是不完备。 +#[derive(Debug, Default)] +struct Coverage { + declarative: Vec, + trigger_covered: Vec, + non_scope: Vec, + /// owner-derived 表上 catalog 有、登记没有的边 + unknown: Vec, + /// 登记/豁免/owner 声明在 catalog 里对不上的边 + stale_registry: Vec, + /// kb 自持行上没有合法复合外键的边 + unprotected_direct: Vec, + /// 登记过、但触发器没装上或没盯该列的边 + unprotected_trigger: Vec, + /// 责任面与已装机制对不上 + surface_drift: Vec, +} + +impl Coverage { + /// 完备 = 没有未归类的边、没有腐掉的登记、面与机制互证得上, + /// 且两类已覆盖边的数量与迁移记录的 26/13 一致。 + fn assert_complete(&self) -> anyhow::Result<()> { + let mut problems = String::new(); + let mut dump = |name: &str, xs: &[String]| { + for x in xs { + problems.push_str(&format!(" {name}: {x}\n")); + } + }; + dump("UNKNOWN", &self.unknown); + dump("STALE_REGISTRY", &self.stale_registry); + dump("UNPROTECTED_DIRECT", &self.unprotected_direct); + dump("UNPROTECTED_TRIGGER", &self.unprotected_trigger); + dump("SURFACE_DRIFT", &self.surface_drift); + if self.declarative.len() != 26 { + problems.push_str(&format!( + " DECLARATIVE_EDGES = {} (expected 26)\n", + self.declarative.len() + )); + } + if self.trigger_covered.len() != 13 { + problems.push_str(&format!( + " TRIGGER_EDGES = {} (expected 13)\n", + self.trigger_covered.len() + )); + } + if problems.is_empty() { + Ok(()) + } else { + Err(anyhow::anyhow!( + "catalog-derived coverage is incomplete:\n{problems}" + )) + } + } +} + +/// 每条相关边归到恰好一类。相关 = 源表在责任面、引用列不是 kb_id +/// 本身、目标表带 kb_id——kb_id→knowledge_bases 这类容器边被 +/// `src_col != kb_id` 自然挡在面外,复合外键里的 kb_id→kb_id 那一腿 +/// 也只是绑定机制、不是一条独立语义引用。 +fn classify( + edges: &[RefEdge], + kb_scoped: &HashSet, + triggers: &HashMap<(String, String), HashSet>, + mechanism: &HashSet, +) -> Coverage { + let ledger: HashSet<&str> = LEDGER_TABLES.iter().copied().collect(); + let owner: HashMap<&str, (&str, &str)> = + OWNER_ROWS.iter().map(|(t, c, p)| (*t, (*c, *p))).collect(); + let registry: HashMap<(&str, &str), &str> = TRIGGER_EDGES + .iter() + .map(|(t, c, g)| ((*t, *c), *g)) + .collect(); + let exemptions: HashMap<(&str, &str), &str> = + NON_SCOPE.iter().map(|(t, c, r)| ((*t, *c), *r)).collect(); + let mut cov = Coverage::default(); + + for t in mechanism.iter().filter(|t| !ledger.contains(t.as_str())) { + cov.surface_drift.push(format!( + "{t}: carries 0070 mechanism but is outside the declared surface" + )); + } + for t in ledger.iter().filter(|t| !mechanism.contains(**t)) { + cov.surface_drift.push(format!( + "{t}: declared in the surface but carries no 0070 mechanism" + )); + } + for (t, _, _) in OWNER_ROWS { + if kb_scoped.contains(*t) { + cov.surface_drift.push(format!( + "{t}: owner-derived row grew its own kb_id — family changed" + )); + } + } + for t in ledger.iter().filter(|t| !owner.contains_key(**t)) { + if !kb_scoped.contains(*t) { + cov.surface_drift.push(format!( + "{t}: direct-kb surface table lost its kb_id column" + )); + } + } + + // 登记三个方向的腐化:owner 边、触发器边、豁免边在 catalog 里都得还在 + for (t, col, parent) in OWNER_ROWS { + if !edges + .iter() + .any(|e| e.src_table == *t && e.src_col == *col && e.tgt_table == *parent) + { + cov.stale_registry.push(format!( + "{t}.{col}\u{2192}{parent}: declared owner edge absent from catalog" + )); + } + } + for (t, c, g) in TRIGGER_EDGES { + if !edges.iter().any(|e| e.src_table == *t && e.src_col == *c) { + cov.stale_registry.push(format!( + "{t}.{c} ({g}): registered trigger edge absent from catalog" + )); + } + } + for (t, c, _) in NON_SCOPE { + if !edges.iter().any(|e| e.src_table == *t && e.src_col == *c) { + cov.stale_registry + .push(format!("{t}.{c}: exempted edge absent from catalog")); + } + } + + for e in edges.iter().filter(|e| { + ledger.contains(e.src_table.as_str()) + && e.src_col != "kb_id" + && kb_scoped.contains(e.tgt_table.as_str()) + }) { + let label = e.label(); + if let Some((owner_col, owner_table)) = owner.get(e.src_table.as_str()) { + if e.src_col == *owner_col { + if e.tgt_table == *owner_table { + cov.non_scope.push(format!("{label} (owner edge)")); + } else { + cov.stale_registry.push(format!( + "{label}: owner column now points at {}, not {owner_table}", + e.tgt_table + )); + } + } else if let Some(tg) = registry.get(&(e.src_table.as_str(), e.src_col.as_str())) { + match triggers.get(&(e.src_table.clone(), (*tg).to_string())) { + Some(cols) if cols.contains(&e.src_col) => cov.trigger_covered.push(label), + Some(_) => cov + .unprotected_trigger + .push(format!("{label}: trigger {tg} does not watch this column")), + None => cov + .unprotected_trigger + .push(format!("{label}: trigger {tg} is not installed")), + } + } else { + cov.unknown.push(label); + } + } else if let Some(reason) = exemptions.get(&(e.src_table.as_str(), e.src_col.as_str())) { + cov.non_scope.push(format!("{label} ({reason})")); + } else { + // 直接 kb 边:同一条约束内必须同时绑住 (kb_id→kb_id) 与 (ref→id); + // 单列 FOREIGN KEY (ref) REFERENCES t(id) 不算同库覆盖 + let composite = e.tgt_col == "id" + && e.pairs.len() == 2 + && e.pairs.iter().any(|(s, t)| s == "kb_id" && t == "kb_id"); + if !composite { + cov.unprotected_direct.push(format!( + "{label} via {} — no composite (kb_id, ref) \u{2192} (kb_id, id)", + e.conname + )); + } else if e.src_table == e.tgt_table && !(e.deferrable && e.deferred) { + cov.unprotected_direct.push(format!( + "{label} via {} — same-table self reference must be \ + DEFERRABLE INITIALLY DEFERRED", + e.conname + )); + } else { + cov.declarative.push(label); + } + } + } + cov +} + +async fn classify_reference_edges(pool: &PgPool) -> anyhow::Result { + let (edges, kb_scoped, triggers, mechanism) = tokio::try_join!( + catalog_edges(pool), + kb_scoped_tables(pool), + trigger_watch_lists(pool), + mechanism_tables(pool), + )?; + Ok(classify(&edges, &kb_scoped, &triggers, &mechanism)) +} + +#[tokio::test] +async fn every_reference_edge_on_the_ledger_is_classified() -> anyhow::Result<()> { + with_scratch("cover", |pool| async move { + migration_70_under(&pool, "public").await?; + classify_reference_edges(&pool).await?.assert_complete() + }) + .await +} + +/// 漂移探针 A:kb 自持行上新增一条单列外键(不配 kb_id)。 +/// 「26 个名字还在」挡不住它——catalog 会把它数出来,归不进任何一类。 +#[tokio::test] +async fn a_new_reference_on_a_kb_owned_row_fails_the_guard() -> anyhow::Result<()> { + with_scratch("driftd", |pool| async move { + migration_70_under(&pool, "public").await?; + sqlx::query( + "ALTER TABLE public.time_mentions + ADD COLUMN probe_ref uuid REFERENCES public.relation_types(id)", + ) + .execute(&pool) + .await?; + let cov = classify_reference_edges(&pool).await?; + assert!( + cov.unprotected_direct + .iter() + .any(|e| e.starts_with("time_mentions.probe_ref\u{2192}")), + "新加的单列引用必须落进 UNPROTECTED_DIRECT: {cov:?}" + ); + assert!(cov.assert_complete().is_err()); + Ok(()) + }) + .await +} + +/// 漂移探针 B:父行作证表上新增一条指向 kb 表的引用列,但不去碰它的 +/// 触发器。「表上有触发器」不许让这条新列显得已被覆盖——登记是列级的。 +#[tokio::test] +async fn a_new_reference_on_an_owner_derived_row_fails_the_guard() -> anyhow::Result<()> { + with_scratch("driftt", |pool| async move { + migration_70_under(&pool, "public").await?; + sqlx::query( + "ALTER TABLE public.fact_evidence + ADD COLUMN probe_rel uuid REFERENCES public.relation_types(id)", + ) + .execute(&pool) + .await?; + let cov = classify_reference_edges(&pool).await?; + assert!( + cov.unknown + .iter() + .any(|e| e.starts_with("fact_evidence.probe_rel\u{2192}")), + "未登记的新边必须落进 UNKNOWN: {cov:?}" + ); + assert!(cov.assert_complete().is_err()); + Ok(()) + }) + .await +} + +#[tokio::test] +async fn the_migration_installs_under_an_empty_search_path() -> anyhow::Result<()> { + with_scratch("empty", |pool| async move { + let r = migration_70_under(&pool, "''").await; + assert!(r.is_ok(), "search_path 为空时装得上 0070: {r:?}"); + assert_installed(&pool).await + }) + .await +} + +#[tokio::test] +async fn the_migration_installs_under_a_hostile_search_path() -> anyhow::Result<()> { + with_scratch("evil", |pool| async move { + // 首位摆个干扰 schema：同名的 entities 表与同名函数——裸名解析会先看到它。 + // 限定到 public.* 的 DDL 不该理会它 + sqlx::query("CREATE SCHEMA decoy").execute(&pool).await?; + sqlx::query("CREATE TABLE decoy.entities (id uuid, kb_id uuid, merged_into uuid)") + .execute(&pool) + .await?; + sqlx::query( + "CREATE FUNCTION decoy.kb_ownership_is_not_reassigned() RETURNS trigger + LANGUAGE plpgsql AS $$ BEGIN RETURN NULL; END; $$", + ) + .execute(&pool) + .await?; + + let r = migration_70_under(&pool, "decoy, public").await; + assert!(r.is_ok(), "首位被占的 search_path 下也装得上 0070: {r:?}"); + assert_installed(&pool).await?; + // 干扰物原样留着：一次都没被选中 + let decoy_fn: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM pg_proc p JOIN pg_namespace n ON n.oid = p.pronamespace + WHERE n.nspname = 'decoy' AND p.proname = 'kb_ownership_is_not_reassigned'", + ) + .fetch_one(&pool) + .await?; + assert_eq!(decoy_fn, 1, "decoy 函数不该被覆盖也不该被删掉"); + Ok(()) + }) + .await +} + +#[tokio::test] +async fn the_migration_installs_under_a_normal_search_path() -> anyhow::Result<()> { + with_scratch("norm", |pool| async move { + let r = migration_70_under(&pool, "public").await; + assert!(r.is_ok(), "正常 search_path 下装得上 0070: {r:?}"); + assert_installed(&pool).await + }) + .await +} diff --git a/migrations/0070_provenance_never_points_across_a_knowledge_base.sql b/migrations/0070_provenance_never_points_across_a_knowledge_base.sql new file mode 100644 index 000000000..deff0e3d7 --- /dev/null +++ b/migrations/0070_provenance_never_points_across_a_knowledge_base.sql @@ -0,0 +1,769 @@ +-- 出处链不许跨库。 +-- +-- 引用完整性只保证「指着的行存在」，不保证「指着的东西在同一个库」。导出把 +-- 引用对象的 id 铸进**本库**的 IRI（urn:utopia:kb:A:fact:{B 库的事实}）， +-- 一份看着完整、实则指着不存在之物的文件就这么出去了；而能被导出器解析的 +-- 引用（谓词、属性类型、实体类型、父类、domain/range）落到别库时更安静—— +-- 查表落空，那一截语义**不声不响地消失**。 +-- +-- 所以不变量是一条，不是一条边：**出处与归属语义上的每一个引用，两端必须 +-- 同属一个库。** 实现按「这条边的库归属谁来作证」分两层： +-- +-- §1b 声明式（26 条边）：引用行**自己带 kb_id**，且以它为准——复合外键 +-- `(kb_id, ref_id) REFERENCES target (kb_id, id)` 把「存在且同库」合成 +-- 一条约束。RI 检查在内核层跑，比行级 PL/pgSQL 调用便宜；`session_ +-- replication_role = replica` 的批量装载（pg_restore --disable-triggers +-- 的形状）也绕不过它——那一层用户触发器全体静默，约束触发器照查。 +-- 同表自指（supersedes / from_statement_id / inverse_of / +-- sub_property_of）用 `DEFERRABLE INITIALLY DEFERRED`：目标行可能在本 +-- 语句之后才落盘，**提交边界**重估时整批都在——顺带替掉了原先的递延 +-- 约束触发器。inverse_of / sub_property_of 的 ON DELETE SET NULL 带 +-- 列清单（`SET NULL (inverse_of)`，PG15+）：复合键的 SET NULL 会把 +-- kb_id 一起置空，必须限定到引用列。 +-- §1c 触发器（13 条边）：库归属**不在引用行自己手上**的边—— +-- fact_evidence / fact_qualifiers / typed_fact_sources / +-- statement_qualifiers 跟所属 fact 的库，fact_derivations 跟派生事实的 +-- 库，entity_type_parents 跟子类的库，relation_type_domains/_ranges/ +-- _qualifiers 跟关系的库，attribute_rule_conditions 跟规则的库。这些行 +-- 没有 kb_id 列；加一列反正规化的 kb_id 让复合外键够得着，等于造出第二 +-- 条「列值必须等于父行 kb_id」的不变量再拿触发器去守它——用触发器直接 +-- 查父行的库，比那份冗余+同步便宜。 +-- +-- 所有者逐条列出（★ = 声明式，○ = 触发器）： +-- +-- 所有者行 引用列 → 被引表 +-- fact_evidence ○ chunk_id → chunks · ○ document_id → documents +-- （以所属 fact 的 kb 为准） +-- chunks ★ document_id → documents（以 chunk 自己的 kb_id 为准） +-- fact_derivations ○ premise_fact_id → facts · ○ premise_derived_id → +-- derived_facts（以 derived_fact 的 kb 为准） +-- fact_qualifiers ○ qualifier_type_id → relation_types · ○ entity_id → +-- entities（以所属 fact 的 kb 为准） +-- facts ★ subject_id/object_id → entities · ★ predicate_id → +-- relation_types · ★ supersedes/from_statement_id → facts +-- （同表自指，递延） +-- derived_facts ★ subject_id/object_id → entities · ★ predicate_id → +-- relation_types · ★ rule_id → rules · ★ attribute_rule_id → +-- attribute_rules +-- entities ★ type_id → entity_types +-- entity_type_parents ○ parent_id → entity_types（以 child 的 kb 为准） +-- entity_type_disjoint ★ a_id/b_id → entity_types（以行自己的 kb_id 为准） +-- relation_type_domains / _ranges ○ entity_type_id → entity_types +-- （以 relation 的 kb 为准） +-- relation_type_qualifiers ○ qualifier_type_id → relation_types +-- （以 relation 的 kb 为准） +-- relation_types ★ inverse_of / sub_property_of → relation_types +-- （同表自指，递延） +-- rules ★ predicate_id → relation_types +-- attribute_rules ★ subject_type_id / conclude_type_id → entity_types · +-- ★ conclude_predicate_id → relation_types +-- attribute_rule_conditions ○ predicate_id → relation_types +-- （归属按**所属规则**的库判——条件行自己没有 kb 列。 +-- rule_id → attribute_rules 是普通外键：不存在的规则 +-- 装不进来，规则删了条件跟着删） +-- typed_fact_sources ○ statement_id → facts（以所属 fact 的 kb 为准） +-- statement_qualifiers ○ entity_id → entities（以所属 fact 的 kb 为准） +-- time_mentions ★ fact_id → facts · ★ chunk_id → chunks +-- （以行自己的 kb_id 为准） +-- type_bindings ★ type_id → entity_types（以行自己的 kb_id 为准） +-- phrase_bindings ★ subject_type_id / object_type_id → entity_types · +-- ★ relation_type_id → relation_types（以行自己的 kb_id 为准） +-- +-- 例外：attribute_rules.conclude_expr 与算式 operand 里 `attr` 叶子嵌着的 +-- 谓词引用。jsonb 列装不下外键，行级触发器去逐棵 JSON 树拆，等于把求值侧 +-- 的表达式语法再抄一遍——这类引用的执行层是**导出侧校验**（export.rs： +-- 取数时按库挡、序列化时按词汇表解析，越库/悬空/连 uuid 都解析不出的 +-- 一律拒导）。本迁移管的是列级引用。 +-- +-- 三层防线，各管一段： +-- §0 前置检查 —— 迁移本身先数一遍存量：库里已有越界行就**整体中止**， +-- 报出是哪条边、坏了几行。装上不变量却对已违反它的账本报喜， +-- 等于替坏数据背书。不修数据的人不该拿到「迁移成功」。 +-- 这段查询同时是**运维审计**：升级前拿它在真实库上跑一遍， +-- 就知道这次升级会不会在半截停下。 +-- §1 复合外键 + 触发器 —— 挡在一切写入路径的下游（原生 API、手写 SQL、 +-- 还没写出来的那些路径）。复合键同时管住「改引用列」与「改 +-- kb_id 过户」（被引行的 kb_id 动了，键就悬空）。 +-- §2 kb 不可过户 —— 把已被引用的行挪到别的库，等于把指着它的行一次全变 +-- 坏行。复合外键只护住**被指着**的行；§1c 那些跟父行库走的 +-- 边，父行 kb 一动就脱锚——所以不可过户是全量保留的。 +-- +-- §1c 的函数体全部**限定到 public schema 且钉死 search_path**：pg_restore 会 +-- 把会话 search_path 置空再灌数据，裸表名在那里解析不到任何东西——数据恢复 +-- 会炸在半截，留下一个说不清的残库。正确性不许依赖环境。 +-- 同一个理由，本文件的 DDL 标识符也一律 `public.*` 限定：迁移在 +-- `SET search_path=''` 或首位被恶意 schema 占住的会话里跑，都不能把函数、 +-- 触发器和约束建错地方——一个落在别处的约束等于没有约束。 +-- +-- 触发器与复合外键只管落在它们之后的写；存量坏行由 §0 挡在迁移门口，由导出 +-- 侧的体检（provenance_integrity）与逐页校验拦在序列化之前。 + +-- ===================================================================== +-- §0 前置检查：存量越界行 → 整份中止 +-- ===================================================================== +DO $$ +DECLARE + report text; +BEGIN + SELECT string_agg(edge || ' x' || n, '; ' ORDER BY edge) INTO report + FROM ( + SELECT edge, COUNT(*) AS n FROM ( + SELECT 'evidence.chunk' AS edge, f.kb_id AS owner_kb, c.kb_id AS ref_kb + FROM public.fact_evidence e + JOIN public.facts f ON f.id = e.fact_id + LEFT JOIN public.chunks c ON c.id = e.chunk_id + UNION ALL + SELECT 'evidence.document', f.kb_id, d.kb_id + FROM public.fact_evidence e + JOIN public.facts f ON f.id = e.fact_id + LEFT JOIN public.documents d ON d.id = e.document_id + WHERE e.document_id IS NOT NULL + UNION ALL + SELECT 'chunk.document', c.kb_id, d.kb_id + FROM public.chunks c + LEFT JOIN public.documents d ON d.id = c.document_id + UNION ALL + SELECT 'derivation.premise_fact', d.kb_id, p.kb_id + FROM public.fact_derivations fd + JOIN public.derived_facts d ON d.id = fd.derived_fact_id + LEFT JOIN public.facts p ON p.id = fd.premise_fact_id + WHERE fd.premise_fact_id IS NOT NULL + UNION ALL + SELECT 'derivation.premise_derived', d.kb_id, p.kb_id + FROM public.fact_derivations fd + JOIN public.derived_facts d ON d.id = fd.derived_fact_id + LEFT JOIN public.derived_facts p ON p.id = fd.premise_derived_id + WHERE fd.premise_derived_id IS NOT NULL + UNION ALL + SELECT 'qualifier.type', f.kb_id, r.kb_id + FROM public.fact_qualifiers q + JOIN public.facts f ON f.id = q.fact_id + LEFT JOIN public.relation_types r ON r.id = q.qualifier_type_id + UNION ALL + SELECT 'qualifier.entity', f.kb_id, e.kb_id + FROM public.fact_qualifiers q + JOIN public.facts f ON f.id = q.fact_id + LEFT JOIN public.entities e ON e.id = q.entity_id + WHERE q.entity_id IS NOT NULL + UNION ALL + SELECT 'fact.subject', f.kb_id, s.kb_id + FROM public.facts f + LEFT JOIN public.entities s ON s.id = f.subject_id + UNION ALL + SELECT 'fact.object', f.kb_id, o.kb_id + FROM public.facts f + LEFT JOIN public.entities o ON o.id = f.object_id + WHERE f.object_id IS NOT NULL + UNION ALL + SELECT 'fact.predicate', f.kb_id, r.kb_id + FROM public.facts f + LEFT JOIN public.relation_types r ON r.id = f.predicate_id + WHERE f.predicate_id IS NOT NULL + UNION ALL + SELECT 'fact.supersedes', f.kb_id, s.kb_id + FROM public.facts f + LEFT JOIN public.facts s ON s.id = f.supersedes + WHERE f.supersedes IS NOT NULL + UNION ALL + SELECT 'fact.from_statement', f.kb_id, s.kb_id + FROM public.facts f + LEFT JOIN public.facts s ON s.id = f.from_statement_id + WHERE f.from_statement_id IS NOT NULL + UNION ALL + SELECT 'derived.subject', d.kb_id, s.kb_id + FROM public.derived_facts d + LEFT JOIN public.entities s ON s.id = d.subject_id + UNION ALL + SELECT 'derived.object', d.kb_id, o.kb_id + FROM public.derived_facts d + LEFT JOIN public.entities o ON o.id = d.object_id + WHERE d.object_id IS NOT NULL + UNION ALL + SELECT 'derived.predicate', d.kb_id, r.kb_id + FROM public.derived_facts d + LEFT JOIN public.relation_types r ON r.id = d.predicate_id + UNION ALL + SELECT 'derived.rule', d.kb_id, r.kb_id + FROM public.derived_facts d + LEFT JOIN public.rules r ON r.id = d.rule_id + WHERE d.rule_id IS NOT NULL + UNION ALL + SELECT 'derived.attribute_rule', d.kb_id, r.kb_id + FROM public.derived_facts d + LEFT JOIN public.attribute_rules r ON r.id = d.attribute_rule_id + WHERE d.attribute_rule_id IS NOT NULL + UNION ALL + SELECT 'entity.type', e.kb_id, t.kb_id + FROM public.entities e + LEFT JOIN public.entity_types t ON t.id = e.type_id + WHERE e.type_id IS NOT NULL + UNION ALL + SELECT 'class.parent', c.kb_id, p.kb_id + FROM public.entity_type_parents x + JOIN public.entity_types c ON c.id = x.child_id + LEFT JOIN public.entity_types p ON p.id = x.parent_id + UNION ALL + SELECT 'class.disjoint', dd.kb_id, a.kb_id + FROM public.entity_type_disjoint dd + LEFT JOIN public.entity_types a ON a.id = dd.a_id + UNION ALL + SELECT 'class.disjoint', dd.kb_id, b.kb_id + FROM public.entity_type_disjoint dd + LEFT JOIN public.entity_types b ON b.id = dd.b_id + UNION ALL + SELECT 'relation.domain', r.kb_id, t.kb_id + FROM public.relation_type_domains x + JOIN public.relation_types r ON r.id = x.relation_type_id + LEFT JOIN public.entity_types t ON t.id = x.entity_type_id + UNION ALL + SELECT 'relation.range', r.kb_id, t.kb_id + FROM public.relation_type_ranges x + JOIN public.relation_types r ON r.id = x.relation_type_id + LEFT JOIN public.entity_types t ON t.id = x.entity_type_id + UNION ALL + SELECT 'relation.qualifier', r.kb_id, q.kb_id + FROM public.relation_type_qualifiers x + JOIN public.relation_types r ON r.id = x.relation_type_id + LEFT JOIN public.relation_types q ON q.id = x.qualifier_type_id + UNION ALL + SELECT 'relation.inverse', r.kb_id, t.kb_id + FROM public.relation_types r + LEFT JOIN public.relation_types t ON t.id = r.inverse_of + WHERE r.inverse_of IS NOT NULL + UNION ALL + SELECT 'relation.sub_property', r.kb_id, t.kb_id + FROM public.relation_types r + LEFT JOIN public.relation_types t ON t.id = r.sub_property_of + WHERE r.sub_property_of IS NOT NULL + UNION ALL + SELECT 'rule.predicate', u.kb_id, p.kb_id + FROM public.rules u + LEFT JOIN public.relation_types p ON p.id = u.predicate_id + UNION ALL + SELECT 'arule.subject_type', a.kb_id, t.kb_id + FROM public.attribute_rules a + LEFT JOIN public.entity_types t ON t.id = a.subject_type_id + UNION ALL + SELECT 'arule.conclude_type', a.kb_id, t.kb_id + FROM public.attribute_rules a + LEFT JOIN public.entity_types t ON t.id = a.conclude_type_id + WHERE a.conclude_type_id IS NOT NULL + UNION ALL + SELECT 'arule.conclude_predicate', a.kb_id, p.kb_id + FROM public.attribute_rules a + LEFT JOIN public.relation_types p ON p.id = a.conclude_predicate_id + WHERE a.conclude_predicate_id IS NOT NULL + UNION ALL + -- 条件行自己没有 kb 列：归属按所属规则的库判 + SELECT 'condition.predicate', a.kb_id, p.kb_id + FROM public.attribute_rule_conditions c + JOIN public.attribute_rules a ON a.id = c.rule_id + LEFT JOIN public.relation_types p ON p.id = c.predicate_id + UNION ALL + -- 来源边行自己没有 kb 列：归属按所属 fact 的库判 + SELECT 'factsource.statement', f.kb_id, s.kb_id + FROM public.typed_fact_sources ts + JOIN public.facts f ON f.id = ts.fact_id + LEFT JOIN public.facts s ON s.id = ts.statement_id + UNION ALL + -- 开放陈述的属性行自己没有 kb 列：归属按所属 fact 的库判 + SELECT 'squalifier.entity', f.kb_id, e.kb_id + FROM public.statement_qualifiers q + JOIN public.facts f ON f.id = q.fact_id + LEFT JOIN public.entities e ON e.id = q.entity_id + WHERE q.entity_id IS NOT NULL + UNION ALL + SELECT 'timemention.fact', t.kb_id, f.kb_id + FROM public.time_mentions t + LEFT JOIN public.facts f ON f.id = t.fact_id + UNION ALL + SELECT 'timemention.chunk', t.kb_id, c.kb_id + FROM public.time_mentions t + LEFT JOIN public.chunks c ON c.id = t.chunk_id + UNION ALL + SELECT 'binding.type', b.kb_id, t.kb_id + FROM public.type_bindings b + LEFT JOIN public.entity_types t ON t.id = b.type_id + WHERE b.type_id IS NOT NULL + UNION ALL + SELECT 'pbinding.subject_type', b.kb_id, t.kb_id + FROM public.phrase_bindings b + LEFT JOIN public.entity_types t ON t.id = b.subject_type_id + WHERE b.subject_type_id IS NOT NULL + UNION ALL + SELECT 'pbinding.object_type', b.kb_id, t.kb_id + FROM public.phrase_bindings b + LEFT JOIN public.entity_types t ON t.id = b.object_type_id + WHERE b.object_type_id IS NOT NULL + UNION ALL + SELECT 'pbinding.relation', b.kb_id, r.kb_id + FROM public.phrase_bindings b + LEFT JOIN public.relation_types r ON r.id = b.relation_type_id + WHERE b.relation_type_id IS NOT NULL + ) refs + WHERE ref_kb IS DISTINCT FROM owner_kb + GROUP BY edge + ) bad; + IF report IS NOT NULL THEN + RAISE EXCEPTION 'cross-KB references already present (%) — repair the ledger before this invariant can be installed', report + USING ERRCODE = 'integrity_constraint_violation'; + END IF; +END; +$$; + +-- ===================================================================== +-- §1a 复合键的支撑唯一约束：每个被引表一个 (kb_id, id) +-- ===================================================================== +-- 复合外键要求被引列上有恰好匹配的唯一约束；主键只有 (id)，不够宽。 +ALTER TABLE public.documents + ADD CONSTRAINT documents_kb_id_key UNIQUE (kb_id, id); +ALTER TABLE public.chunks + ADD CONSTRAINT chunks_kb_id_key UNIQUE (kb_id, id); +ALTER TABLE public.entities + ADD CONSTRAINT entities_kb_id_key UNIQUE (kb_id, id); +ALTER TABLE public.entity_types + ADD CONSTRAINT entity_types_kb_id_key UNIQUE (kb_id, id); +ALTER TABLE public.relation_types + ADD CONSTRAINT relation_types_kb_id_key UNIQUE (kb_id, id); +ALTER TABLE public.facts + ADD CONSTRAINT facts_kb_id_key UNIQUE (kb_id, id); +ALTER TABLE public.rules + ADD CONSTRAINT rules_kb_id_key UNIQUE (kb_id, id); +ALTER TABLE public.attribute_rules + ADD CONSTRAINT attribute_rules_kb_id_key UNIQUE (kb_id, id); + +-- ===================================================================== +-- §1b 声明式边：源行自己带 kb_id 的引用 → 复合外键（26 条） +-- ===================================================================== +-- 每条换掉原来的单列外键：`(kb_id, ref)` 同时证明「存在」与「同库」，一次 +-- 内核层点查顶掉原来「FK 查存在 + 触发器查同库」的两次。删除行为与原外键 +-- 逐条对齐。 +ALTER TABLE public.chunks + DROP CONSTRAINT chunks_document_id_fkey, + ADD CONSTRAINT chunks_document_same_kb + FOREIGN KEY (kb_id, document_id) REFERENCES public.documents (kb_id, id) + ON DELETE CASCADE; + +ALTER TABLE public.facts + DROP CONSTRAINT facts_subject_id_fkey, + DROP CONSTRAINT facts_object_id_fkey, + DROP CONSTRAINT facts_predicate_id_fkey, + DROP CONSTRAINT facts_supersedes_fkey, + DROP CONSTRAINT facts_from_statement_id_fkey, + ADD CONSTRAINT facts_subject_same_kb + FOREIGN KEY (kb_id, subject_id) REFERENCES public.entities (kb_id, id) + ON DELETE CASCADE, + ADD CONSTRAINT facts_object_same_kb + FOREIGN KEY (kb_id, object_id) REFERENCES public.entities (kb_id, id) + ON DELETE CASCADE, + ADD CONSTRAINT facts_predicate_same_kb + FOREIGN KEY (kb_id, predicate_id) REFERENCES public.relation_types (kb_id, id) + ON DELETE CASCADE, + -- 同表自指：目标行可能在本语句之后才落（COPY/多行 INSERT/同事务顺序 + -- 插入）——递延到提交边界重估，那时整批都在 + ADD CONSTRAINT facts_supersedes_same_kb + FOREIGN KEY (kb_id, supersedes) REFERENCES public.facts (kb_id, id) + DEFERRABLE INITIALLY DEFERRED, + ADD CONSTRAINT facts_from_statement_same_kb + FOREIGN KEY (kb_id, from_statement_id) REFERENCES public.facts (kb_id, id) + ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED; + +ALTER TABLE public.derived_facts + DROP CONSTRAINT derived_facts_subject_id_fkey, + DROP CONSTRAINT derived_facts_object_id_fkey, + DROP CONSTRAINT derived_facts_predicate_id_fkey, + DROP CONSTRAINT derived_facts_rule_id_fkey, + DROP CONSTRAINT derived_facts_attribute_rule_id_fkey, + ADD CONSTRAINT derived_facts_subject_same_kb + FOREIGN KEY (kb_id, subject_id) REFERENCES public.entities (kb_id, id) + ON DELETE CASCADE, + ADD CONSTRAINT derived_facts_object_same_kb + FOREIGN KEY (kb_id, object_id) REFERENCES public.entities (kb_id, id) + ON DELETE CASCADE, + ADD CONSTRAINT derived_facts_predicate_same_kb + FOREIGN KEY (kb_id, predicate_id) REFERENCES public.relation_types (kb_id, id) + ON DELETE CASCADE, + ADD CONSTRAINT derived_facts_rule_same_kb + FOREIGN KEY (kb_id, rule_id) REFERENCES public.rules (kb_id, id), + ADD CONSTRAINT derived_facts_attribute_rule_same_kb + FOREIGN KEY (kb_id, attribute_rule_id) REFERENCES public.attribute_rules (kb_id, id) + ON DELETE CASCADE; + +ALTER TABLE public.entities + DROP CONSTRAINT entities_type_id_fkey, + ADD CONSTRAINT entities_type_same_kb + FOREIGN KEY (kb_id, type_id) REFERENCES public.entity_types (kb_id, id) + ON DELETE RESTRICT; + +ALTER TABLE public.entity_type_disjoint + DROP CONSTRAINT entity_type_disjoint_a_id_fkey, + DROP CONSTRAINT entity_type_disjoint_b_id_fkey, + ADD CONSTRAINT entity_type_disjoint_a_same_kb + FOREIGN KEY (kb_id, a_id) REFERENCES public.entity_types (kb_id, id) + ON DELETE CASCADE, + ADD CONSTRAINT entity_type_disjoint_b_same_kb + FOREIGN KEY (kb_id, b_id) REFERENCES public.entity_types (kb_id, id) + ON DELETE CASCADE; + +ALTER TABLE public.relation_types + DROP CONSTRAINT relation_types_inverse_of_fkey, + DROP CONSTRAINT relation_types_sub_property_of_fkey, + -- 同表自指 → 递延；SET NULL 限定到引用列——不限定会把 kb_id 一起置空， + -- 撞上 NOT NULL 变成「删目标行直接报错」（PG15+ 的列清单语法） + ADD CONSTRAINT relation_types_inverse_same_kb + FOREIGN KEY (kb_id, inverse_of) REFERENCES public.relation_types (kb_id, id) + ON DELETE SET NULL (inverse_of) DEFERRABLE INITIALLY DEFERRED, + ADD CONSTRAINT relation_types_sub_property_same_kb + FOREIGN KEY (kb_id, sub_property_of) REFERENCES public.relation_types (kb_id, id) + ON DELETE SET NULL (sub_property_of) DEFERRABLE INITIALLY DEFERRED; + +ALTER TABLE public.rules + DROP CONSTRAINT rules_predicate_id_fkey, + ADD CONSTRAINT rules_predicate_same_kb + FOREIGN KEY (kb_id, predicate_id) REFERENCES public.relation_types (kb_id, id) + ON DELETE CASCADE; + +ALTER TABLE public.attribute_rules + DROP CONSTRAINT attribute_rules_subject_type_id_fkey, + DROP CONSTRAINT attribute_rules_conclude_type_id_fkey, + DROP CONSTRAINT attribute_rules_conclude_predicate_id_fkey, + ADD CONSTRAINT attribute_rules_subject_type_same_kb + FOREIGN KEY (kb_id, subject_type_id) REFERENCES public.entity_types (kb_id, id) + ON DELETE CASCADE, + ADD CONSTRAINT attribute_rules_conclude_type_same_kb + FOREIGN KEY (kb_id, conclude_type_id) REFERENCES public.entity_types (kb_id, id) + ON DELETE CASCADE, + ADD CONSTRAINT attribute_rules_conclude_predicate_same_kb + FOREIGN KEY (kb_id, conclude_predicate_id) REFERENCES public.relation_types (kb_id, id) + ON DELETE CASCADE; + +ALTER TABLE public.time_mentions + DROP CONSTRAINT time_mentions_fact_id_fkey, + DROP CONSTRAINT time_mentions_chunk_id_fkey, + ADD CONSTRAINT time_mentions_fact_same_kb + FOREIGN KEY (kb_id, fact_id) REFERENCES public.facts (kb_id, id) + ON DELETE CASCADE, + ADD CONSTRAINT time_mentions_chunk_same_kb + FOREIGN KEY (kb_id, chunk_id) REFERENCES public.chunks (kb_id, id) + ON DELETE CASCADE; + +ALTER TABLE public.type_bindings + DROP CONSTRAINT type_bindings_type_id_fkey, + ADD CONSTRAINT type_bindings_type_same_kb + FOREIGN KEY (kb_id, type_id) REFERENCES public.entity_types (kb_id, id) + ON DELETE CASCADE; + +ALTER TABLE public.phrase_bindings + DROP CONSTRAINT phrase_bindings_subject_type_id_fkey, + DROP CONSTRAINT phrase_bindings_object_type_id_fkey, + DROP CONSTRAINT phrase_bindings_relation_type_id_fkey, + ADD CONSTRAINT phrase_bindings_subject_type_same_kb + FOREIGN KEY (kb_id, subject_type_id) REFERENCES public.entity_types (kb_id, id) + ON DELETE CASCADE, + ADD CONSTRAINT phrase_bindings_object_type_same_kb + FOREIGN KEY (kb_id, object_type_id) REFERENCES public.entity_types (kb_id, id) + ON DELETE CASCADE, + ADD CONSTRAINT phrase_bindings_relation_type_same_kb + FOREIGN KEY (kb_id, relation_type_id) REFERENCES public.relation_types (kb_id, id) + ON DELETE CASCADE; + +-- ===================================================================== +-- §1c 触发器边：库归属在父行手上、引用行没有 kb_id 的 13 条 +-- ===================================================================== + +-- 证据行：事实、段落、冗余文档指针必须同属一个库。 +-- 父行不存在交给外键报错；这里只管「都存在，却不在同一个库」。 +CREATE FUNCTION public.fact_evidence_stays_inside_its_kb() RETURNS trigger +LANGUAGE plpgsql SET search_path = pg_catalog AS $$ +DECLARE + fact_kb uuid; + ref_kb uuid; +BEGIN + SELECT kb_id INTO fact_kb FROM public.facts WHERE id = NEW.fact_id; + SELECT kb_id INTO ref_kb FROM public.chunks WHERE id = NEW.chunk_id; + IF fact_kb IS NOT NULL AND ref_kb IS NOT NULL AND ref_kb <> fact_kb THEN + RAISE EXCEPTION 'fact_evidence cannot pair fact % with chunk % across knowledge bases', + NEW.fact_id, NEW.chunk_id; + END IF; + IF NEW.document_id IS NOT NULL AND fact_kb IS NOT NULL THEN + SELECT kb_id INTO ref_kb FROM public.documents WHERE id = NEW.document_id; + IF ref_kb IS NOT NULL AND ref_kb <> fact_kb THEN + RAISE EXCEPTION 'fact_evidence cannot point at document % across knowledge bases', + NEW.document_id; + END IF; + END IF; + RETURN NEW; +END; +$$; + +-- `ON CONFLICT DO UPDATE` 只改 quote/proposed_predicate，不在列清单里， +-- 常规的证据合并路径不会唤醒它 +CREATE TRIGGER fact_evidence_same_kb + BEFORE INSERT OR UPDATE OF fact_id, chunk_id, document_id ON public.fact_evidence + FOR EACH ROW EXECUTE FUNCTION public.fact_evidence_stays_inside_its_kb(); + +-- 证明树：前提（断言或派生）必须与结论同属一个库。前提在导出里铸成 +-- prov:used → fact:/derived: IRI——别库的前提挂上本库的结论，伪造的就是身份。 +CREATE FUNCTION public.derivation_premise_stays_inside_its_kb() RETURNS trigger +LANGUAGE plpgsql SET search_path = pg_catalog AS $$ +DECLARE + derived_kb uuid; + ref_kb uuid; +BEGIN + SELECT kb_id INTO derived_kb FROM public.derived_facts WHERE id = NEW.derived_fact_id; + IF derived_kb IS NULL THEN + RETURN NEW; + END IF; + IF NEW.premise_fact_id IS NOT NULL THEN + SELECT kb_id INTO ref_kb FROM public.facts WHERE id = NEW.premise_fact_id; + ELSE + SELECT kb_id INTO ref_kb FROM public.derived_facts WHERE id = NEW.premise_derived_id; + END IF; + IF ref_kb IS NOT NULL AND ref_kb <> derived_kb THEN + RAISE EXCEPTION 'derivation premise % cannot live outside the derived fact''s knowledge base', + COALESCE(NEW.premise_fact_id, NEW.premise_derived_id); + END IF; + RETURN NEW; +END; +$$; + +CREATE TRIGGER fact_derivations_same_kb + BEFORE INSERT OR UPDATE OF derived_fact_id, premise_fact_id, premise_derived_id + ON public.fact_derivations + FOR EACH ROW EXECUTE FUNCTION public.derivation_premise_stays_inside_its_kb(); + +-- 边上的属性：属性类型必须在本库词汇表里可解析（别库的类型在导出里会被 +-- 静默跳过——坏行不是消失，是当场报错）；实体值不许把别库实体铸进本库 IRI。 +CREATE FUNCTION public.qualifier_stays_inside_its_facts_kb() RETURNS trigger +LANGUAGE plpgsql SET search_path = pg_catalog AS $$ +DECLARE + fact_kb uuid; + ref_kb uuid; +BEGIN + SELECT kb_id INTO fact_kb FROM public.facts WHERE id = NEW.fact_id; + IF fact_kb IS NULL THEN + RETURN NEW; + END IF; + SELECT kb_id INTO ref_kb FROM public.relation_types WHERE id = NEW.qualifier_type_id; + IF ref_kb IS NOT NULL AND ref_kb <> fact_kb THEN + RAISE EXCEPTION 'qualifier type % cannot live outside the fact''s knowledge base', + NEW.qualifier_type_id; + END IF; + IF NEW.entity_id IS NOT NULL THEN + SELECT kb_id INTO ref_kb FROM public.entities WHERE id = NEW.entity_id; + IF ref_kb IS NOT NULL AND ref_kb <> fact_kb THEN + RAISE EXCEPTION 'qualifier entity % cannot live outside the fact''s knowledge base', + NEW.entity_id; + END IF; + END IF; + RETURN NEW; +END; +$$; + +CREATE TRIGGER fact_qualifiers_same_kb + BEFORE INSERT OR UPDATE OF fact_id, qualifier_type_id, entity_id ON public.fact_qualifiers + FOR EACH ROW EXECUTE FUNCTION public.qualifier_stays_inside_its_facts_kb(); + +-- 类层级：父类必须与子类同库。 +CREATE FUNCTION public.type_parent_stays_inside_its_kb() RETURNS trigger +LANGUAGE plpgsql SET search_path = pg_catalog AS $$ +DECLARE + child_kb uuid; + ref_kb uuid; +BEGIN + SELECT kb_id INTO child_kb FROM public.entity_types WHERE id = NEW.child_id; + SELECT kb_id INTO ref_kb FROM public.entity_types WHERE id = NEW.parent_id; + IF child_kb IS NOT NULL AND ref_kb IS NOT NULL AND ref_kb <> child_kb THEN + RAISE EXCEPTION 'class % cannot parent % across knowledge bases', + NEW.parent_id, NEW.child_id; + END IF; + RETURN NEW; +END; +$$; + +CREATE TRIGGER entity_type_parents_same_kb + BEFORE INSERT OR UPDATE OF child_id, parent_id ON public.entity_type_parents + FOR EACH ROW EXECUTE FUNCTION public.type_parent_stays_inside_its_kb(); + +-- 关系的 domain/range：类必须与关系同库。两张表同形，共用一个函数。 +CREATE FUNCTION public.relation_scope_stays_inside_its_kb() RETURNS trigger +LANGUAGE plpgsql SET search_path = pg_catalog AS $$ +DECLARE + rel_kb uuid; + ref_kb uuid; +BEGIN + SELECT kb_id INTO rel_kb FROM public.relation_types WHERE id = NEW.relation_type_id; + SELECT kb_id INTO ref_kb FROM public.entity_types WHERE id = NEW.entity_type_id; + IF rel_kb IS NOT NULL AND ref_kb IS NOT NULL AND ref_kb <> rel_kb THEN + RAISE EXCEPTION '% cannot name class % outside the relation''s knowledge base', + TG_TABLE_NAME, NEW.entity_type_id; + END IF; + RETURN NEW; +END; +$$; + +CREATE TRIGGER relation_type_domains_same_kb + BEFORE INSERT OR UPDATE OF relation_type_id, entity_type_id ON public.relation_type_domains + FOR EACH ROW EXECUTE FUNCTION public.relation_scope_stays_inside_its_kb(); + +CREATE TRIGGER relation_type_ranges_same_kb + BEFORE INSERT OR UPDATE OF relation_type_id, entity_type_id ON public.relation_type_ranges + FOR EACH ROW EXECUTE FUNCTION public.relation_scope_stays_inside_its_kb(); + +-- 关系挂的边属性声明：qualifier 必须是**同一个库**里的行（形态校验—— +-- 必须是 kind='attribute'——在 store 层，这里是归属层）。 +CREATE FUNCTION public.relation_qualifier_stays_inside_the_kb() RETURNS trigger +LANGUAGE plpgsql SET search_path = pg_catalog AS $$ +DECLARE + rel_kb uuid; + ref_kb uuid; +BEGIN + SELECT kb_id INTO rel_kb FROM public.relation_types WHERE id = NEW.relation_type_id; + SELECT kb_id INTO ref_kb FROM public.relation_types WHERE id = NEW.qualifier_type_id; + IF rel_kb IS NOT NULL AND ref_kb IS NOT NULL AND ref_kb <> rel_kb THEN + RAISE EXCEPTION 'relation qualifier % cannot live outside the relation''s knowledge base', + NEW.qualifier_type_id; + END IF; + RETURN NEW; +END; +$$; + +CREATE TRIGGER relation_type_qualifiers_same_kb + BEFORE INSERT OR UPDATE OF relation_type_id, qualifier_type_id ON public.relation_type_qualifiers + FOR EACH ROW EXECUTE FUNCTION public.relation_qualifier_stays_inside_the_kb(); + +-- 规则条件读的谓词：条件行没有自己的 kb 列，归属按所属规则的库判—— +-- 导出把 predicate_id 铸成本库谓词 IRI，别库的谓词会被词汇表查空。 +-- rule_id 是普通外键：不存在的规则装不进来（串行写时点检查就够—— +-- 谓词与规则都必须在 INSERT 时已存在，kb 又都不可过户） +CREATE FUNCTION public.rule_condition_refs_stay_inside_the_kb() RETURNS trigger +LANGUAGE plpgsql SET search_path = pg_catalog AS $$ +DECLARE + rule_kb uuid; + ref_kb uuid; +BEGIN + SELECT kb_id INTO rule_kb FROM public.attribute_rules WHERE id = NEW.rule_id; + IF rule_kb IS NOT NULL THEN + SELECT kb_id INTO ref_kb FROM public.relation_types WHERE id = NEW.predicate_id; + IF ref_kb IS NOT NULL AND ref_kb <> rule_kb THEN + RAISE EXCEPTION 'rule condition % predicate % cannot live outside the rule''s knowledge base', + NEW.id, NEW.predicate_id; + END IF; + END IF; + RETURN NEW; +END; +$$; + +CREATE TRIGGER attribute_rule_conditions_same_kb + BEFORE INSERT OR UPDATE OF rule_id, predicate_id ON public.attribute_rule_conditions + FOR EACH ROW EXECUTE FUNCTION public.rule_condition_refs_stay_inside_the_kb(); + +-- 陈述→类型化事实的来源边：statement 必须与行主（fact）同库。 +-- 两端都是必填外键，写入时必然在场——串行写时点检查就够，kb 又都不可过户。 +CREATE FUNCTION public.typed_source_stays_inside_its_kb() RETURNS trigger +LANGUAGE plpgsql SET search_path = pg_catalog AS $$ +DECLARE + fact_kb uuid; + ref_kb uuid; +BEGIN + SELECT kb_id INTO fact_kb FROM public.facts WHERE id = NEW.fact_id; + SELECT kb_id INTO ref_kb FROM public.facts WHERE id = NEW.statement_id; + IF fact_kb IS NOT NULL AND ref_kb IS NOT NULL AND ref_kb <> fact_kb THEN + RAISE EXCEPTION 'typed fact source % cannot live outside the fact''s knowledge base', + NEW.statement_id; + END IF; + RETURN NEW; +END; +$$; + +CREATE TRIGGER typed_fact_sources_same_kb + BEFORE INSERT OR UPDATE OF fact_id, statement_id ON public.typed_fact_sources + FOR EACH ROW EXECUTE FUNCTION public.typed_source_stays_inside_its_kb(); + +-- 开放陈述的属性：实体值必须与所属事实同库（归属按 fact 的库判—— +-- 属性行自己没有 kb 列）。 +CREATE FUNCTION public.squalifier_stays_inside_its_facts_kb() RETURNS trigger +LANGUAGE plpgsql SET search_path = pg_catalog AS $$ +DECLARE + fact_kb uuid; + ref_kb uuid; +BEGIN + IF NEW.entity_id IS NULL THEN + RETURN NEW; + END IF; + SELECT kb_id INTO fact_kb FROM public.facts WHERE id = NEW.fact_id; + SELECT kb_id INTO ref_kb FROM public.entities WHERE id = NEW.entity_id; + IF fact_kb IS NOT NULL AND ref_kb IS NOT NULL AND ref_kb <> fact_kb THEN + RAISE EXCEPTION 'statement qualifier entity % cannot live outside the fact''s knowledge base', + NEW.entity_id; + END IF; + RETURN NEW; +END; +$$; + +CREATE TRIGGER statement_qualifiers_same_kb + BEFORE INSERT OR UPDATE OF fact_id, entity_id ON public.statement_qualifiers + FOR EACH ROW EXECUTE FUNCTION public.squalifier_stays_inside_its_facts_kb(); + +-- ===================================================================== +-- §2 库归属不可过户：把被引用的行挪走，等于把指着它的行一次全变坏行 +-- ===================================================================== +-- 复合外键护住的是「被指着的那一行」的 kb_id；§1c 的边拿**父行**的 kb 作证， +-- 父行 kb 一动整批脱锚，所以不可过户一条不能少。 +CREATE FUNCTION public.kb_ownership_is_not_reassigned() RETURNS trigger +LANGUAGE plpgsql SET search_path = pg_catalog AS $$ +BEGIN + IF NEW.kb_id IS DISTINCT FROM OLD.kb_id THEN + RAISE EXCEPTION 'kb ownership of % is immutable', TG_TABLE_NAME; + END IF; + RETURN NEW; +END; +$$; + +CREATE TRIGGER facts_keep_their_kb + BEFORE UPDATE OF kb_id ON public.facts + FOR EACH ROW EXECUTE FUNCTION public.kb_ownership_is_not_reassigned(); + +CREATE TRIGGER derived_facts_keep_their_kb + BEFORE UPDATE OF kb_id ON public.derived_facts + FOR EACH ROW EXECUTE FUNCTION public.kb_ownership_is_not_reassigned(); + +CREATE TRIGGER documents_keep_their_kb + BEFORE UPDATE OF kb_id ON public.documents + FOR EACH ROW EXECUTE FUNCTION public.kb_ownership_is_not_reassigned(); + +CREATE TRIGGER entities_keep_their_kb + BEFORE UPDATE OF kb_id ON public.entities + FOR EACH ROW EXECUTE FUNCTION public.kb_ownership_is_not_reassigned(); + +CREATE TRIGGER entity_types_keep_their_kb + BEFORE UPDATE OF kb_id ON public.entity_types + FOR EACH ROW EXECUTE FUNCTION public.kb_ownership_is_not_reassigned(); + +CREATE TRIGGER relation_types_keep_their_kb + BEFORE UPDATE OF kb_id ON public.relation_types + FOR EACH ROW EXECUTE FUNCTION public.kb_ownership_is_not_reassigned(); + +CREATE TRIGGER rules_keep_their_kb + BEFORE UPDATE OF kb_id ON public.rules + FOR EACH ROW EXECUTE FUNCTION public.kb_ownership_is_not_reassigned(); + +CREATE TRIGGER attribute_rules_keep_their_kb + BEFORE UPDATE OF kb_id ON public.attribute_rules + FOR EACH ROW EXECUTE FUNCTION public.kb_ownership_is_not_reassigned(); + +CREATE TRIGGER entity_type_disjoint_keep_their_kb + BEFORE UPDATE OF kb_id ON public.entity_type_disjoint + FOR EACH ROW EXECUTE FUNCTION public.kb_ownership_is_not_reassigned(); + +CREATE TRIGGER time_mentions_keep_their_kb + BEFORE UPDATE OF kb_id ON public.time_mentions + FOR EACH ROW EXECUTE FUNCTION public.kb_ownership_is_not_reassigned(); + +CREATE TRIGGER type_bindings_keep_their_kb + BEFORE UPDATE OF kb_id ON public.type_bindings + FOR EACH ROW EXECUTE FUNCTION public.kb_ownership_is_not_reassigned(); + +CREATE TRIGGER phrase_bindings_keep_their_kb + BEFORE UPDATE OF kb_id ON public.phrase_bindings + FOR EACH ROW EXECUTE FUNCTION public.kb_ownership_is_not_reassigned(); From 4097d0af38379b8d8138cd5e9e9318277bfb3f93 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=8D=97=E6=85=B6=E9=BA=9F?= Date: Sun, 20 Sep 2026 23:20:29 +0800 Subject: [PATCH 2/2] Record why provenance enforcement is a hybrid of keys and triggers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 0048 decides the mechanism per edge: composite (kb_id, ref) foreign keys where the row carries its own kb_id, row triggers where the kb authority is a parent row, deferred keys on same-table self-references, kb immutability throughout. Carries the measured write-path cost and the operational precondition-scan requirement for real deployments. Signed-off-by: 南慶麟 --- ...ferences-stay-inside-the-knowledge-base.md | 90 +++++++++++++++++++ docs/decisions/README.md | 2 + 2 files changed, 92 insertions(+) create mode 100644 docs/decisions/0048-provenance-references-stay-inside-the-knowledge-base.md diff --git a/docs/decisions/0048-provenance-references-stay-inside-the-knowledge-base.md b/docs/decisions/0048-provenance-references-stay-inside-the-knowledge-base.md new file mode 100644 index 000000000..6757b1748 --- /dev/null +++ b/docs/decisions/0048-provenance-references-stay-inside-the-knowledge-base.md @@ -0,0 +1,90 @@ +# 0048 · Provenance references stay inside the knowledge base + +- **Status**: Proposed 2026-09-20 · implemented in PR #832 (migration 0070), pending review +- **Written**: 2026-09-20 (conventions in the [README](README.md)) +- **Related**: [0009](0009-no-type-is-a-type.md)'s "NULL means undecided" is why several edges below are nullable and therefore cannot lean on `MATCH FULL`; [0002](0002-reasoning-engine.md) owns the derivation model whose premise edges are covered here. Design question opened as issue #842. + +> The exporter writes `urn:utopia:kb:A:fact:{id}` and asserts the id belongs to knowledge base A. Nothing in the schema made that true — a foreign key proves the row exists, not which base it lives in. A cross-KB `supersedes` would mint a local IRI naming a foreign fact; a cross-KB `type_id` would resolve to nothing and the statement would silently lose its class. This record decides where the same-KB invariant is enforced, and with what. + +## The invariant + +Every reference an export can resolve must join rows that live in the same knowledge base. One invariant, stated once; the question is only which mechanism proves it per edge, and at which transaction boundary. + +## Why the existing foreign keys are insufficient + +`FOREIGN KEY (ref) REFERENCES t (id)` checks one column against one key. `kb_id` is not part of the check, so `(kb A).fact → (kb B).entity` is a perfectly satisfied foreign key. The failure is silent in both directions it matters: the exporter mints an IRI that names a row in another base (a complete-looking file pointing at nothing), or a vocabulary reference resolves to a foreign row the vocabulary lookup cannot see (a slice of semantics dropped without an error). + +## The mechanism, per edge + +Thirty-nine reference edges are protected. The split is decided by a single question: **does the referencing row carry the kb authority itself?** + +| Source | Edge → target | kb authority | Mechanism | +|---|---|---|---| +| `chunks` | `document_id` → `documents` | own `kb_id` | composite FK | +| `facts` | `subject_id`, `object_id` → `entities`; `predicate_id` → `relation_types` | own `kb_id` | composite FK | +| `facts` | `supersedes`, `from_statement_id` → `facts` | own `kb_id` | composite FK, `DEFERRABLE INITIALLY DEFERRED` | +| `derived_facts` | `subject_id`, `object_id` → `entities`; `predicate_id` → `relation_types`; `rule_id` → `rules`; `attribute_rule_id` → `attribute_rules` | own `kb_id` | composite FK | +| `entities` | `type_id` → `entity_types` | own `kb_id` | composite FK | +| `entity_type_disjoint` | `a_id`, `b_id` → `entity_types` | own `kb_id` | composite FK ×2 | +| `relation_types` | `inverse_of`, `sub_property_of` → `relation_types` | own `kb_id` | composite FK, deferred, `ON DELETE SET NULL (col)` | +| `rules` | `predicate_id` → `relation_types` | own `kb_id` | composite FK | +| `attribute_rules` | `subject_type_id`, `conclude_type_id` → `entity_types`; `conclude_predicate_id` → `relation_types` | own `kb_id` | composite FK | +| `time_mentions` | `fact_id` → `facts`; `chunk_id` → `chunks` | own `kb_id` | composite FK | +| `type_bindings` | `type_id` → `entity_types` | own `kb_id` | composite FK | +| `phrase_bindings` | `subject_type_id`, `object_type_id` → `entity_types`; `relation_type_id` → `relation_types` | own `kb_id` | composite FK | +| `fact_evidence` | `chunk_id` → `chunks`; `document_id` → `documents` | owning fact's kb | trigger | +| `fact_qualifiers` | `qualifier_type_id` → `relation_types`; `entity_id` → `entities` | owning fact's kb | trigger | +| `fact_derivations` | `premise_fact_id` → `facts`; `premise_derived_id` → `derived_facts` | derived fact's kb | trigger | +| `entity_type_parents` | `parent_id` → `entity_types` | child's kb | trigger | +| `relation_type_domains`, `relation_type_ranges` | `entity_type_id` → `entity_types` | relation's kb | trigger (shared function) | +| `relation_type_qualifiers` | `qualifier_type_id` → `relation_types` | relation's kb | trigger | +| `attribute_rule_conditions` | `predicate_id` → `relation_types` | owning rule's kb | trigger | +| `typed_fact_sources` | `statement_id` → `facts` | owning fact's kb | trigger | +| `statement_qualifiers` | `entity_id` → `entities` | owning fact's kb | trigger | + +**26 edges** are declared as `FOREIGN KEY (kb_id, ref) REFERENCES t (kb_id, id)`, replacing the single-column FK that was already there — one kernel-level lookup now proves existence *and* same-base, where before it proved existence and a second check would have had to prove the rest. Eight `UNIQUE (kb_id, id)` constraints on the referenced tables back those keys (`documents`, `chunks`, `entities`, `entity_types`, `relation_types`, `facts`, `rules`, `attribute_rules`). + +**13 edges** cannot be expressed that way: their kb authority is a *parent row's* `kb_id`, and the link row has no `kb_id` column at all. A composite key cannot name `fact.kb_id` from a `fact_evidence` row. These keep row-level `BEFORE` triggers — nine functions, ten triggers — that read the parent's base and compare. + +**`kb_id` is immutable** on every owned table (one shared function, twelve `BEFORE UPDATE OF kb_id` triggers). Composite keys only guard the rows being pointed *at*; the trigger-covered edges derive their authority from a parent row, and a parent moving base would silently un-anchor every child pointing at it. Making kb reassignment impossible is what lets a point-in-time check stay correct. + +**Four same-table self-references** (`facts.supersedes`, `facts.from_statement_id`, `relation_types.inverse_of`, `relation_types.sub_property_of`) are `DEFERRABLE INITIALLY DEFERRED`. A row-level check — immediate FK or `BEFORE` trigger alike — cannot see a target row inserted later in the same transaction, and forward references are a normal restore shape (a multi-row `INSERT` or `COPY` batch lists the referring row before its target). Deferred evaluation at commit sees the whole batch; a sequential write whose target never arrives is rejected at `COMMIT`. One boundary is honest: "the edge is valid when the transaction ends", which is exactly what a restore needs and no more than what an in-order write already had to satisfy. + +## COPY, restore, and `search_path` + +`pg_restore` loads data with `COPY` and creates constraints after, so any shape that survives the migration survives a restore — and the deferred self-references are what let a same-transaction or intra-statement forward chain restore without ordering games. Two sharper edges were designed for explicitly: + +- `pg_restore --disable-triggers` and any `session_replication_role = replica` load suppress *user* triggers wholesale. Declarative foreign keys are internal constraint triggers and are **not** suppressed — so the 26 declarative edges hold even in a replica-mode load, which is a second reason to prefer them wherever the schema can say them. The 13 trigger-covered edges admit the gap and are backstopped by the export-side `provenance_integrity` check and the §0 audit query, which doubles as a post-load audit. +- The migration pins `search_path = pg_catalog` in every function body and qualifies every identifier `public.*`, because restore empties the session `search_path` and a hostile first schema must not redirect name resolution. `migration_0070_runs_under_any_search_path` installs the whole migration under a normal, an empty, and a decoy-first `search_path`. +- The coverage itself is guarded by a catalog-derived regression in the same test file: it enumerates every column-level reference inside the ledger surface from `pg_catalog` and fails when an edge resolves to no declared composite-FK, owner-derived-trigger, or explicit exclusion — so a reference column added later cannot silently slip past the invariant. + +## The precondition scan + +§0 counts existing cross-KB rows on all 39 edges and aborts the whole migration with the offending edge names and row counts if any exist. Installing an invariant over a ledger that already violates it would be signing off on bad data; whoever will not repair the ledger must not get a green migration. **This is also the operational requirement**: the scan (or its query, which is safe to run read-only) should be executed against real deployments *before* rollout, so an upgrade does not stop halfway on a base nobody had checked. The scan is the audit; the migration failing closed is the enforcement. + +## Measured cost + +Same host (macOS, Darwin 25.4.0), same Postgres 16 (`pgvector/pgvector:pg16`), same toolchain (rustc 1.98.1), two fresh databases migrated by the respective build, alternating runs. + +`bench_100k` populate phase (`UTOPIA_BENCH_DOCS=1000`, `UTOPIA_BENCH_HUB_FACTS=1000`, ~1000 documents / ~3000 chunks / ~3000 facts written through the real store functions), three runs each, alternating patched/base: + +| run | base | patched | +|---|---|---| +| 1 | 9.18s | 9.35s | +| 2 | 10.29s | 8.71s | +| 3 | 9.77s | 11.11s | +| **median** | **9.77s** | **9.35s** | + +The medians differ by −4% with fully overlapping ranges — **no measurable write-path overhead** on the populate path, which exercises the composite-FK edges (chunks→documents, facts→entities). A focused measurement covers the trigger-covered edges the populate phase never touches: 4000-row bulk `INSERT`s on a scratch fixture, warm, median of three — `fact_evidence` 37.1ms → 63.7ms, `typed_fact_sources` 34.3ms → 58.1ms, i.e. roughly **+6–7µs per row per triggered edge** (the two `SELECT` lookups the trigger performs). On a path that writes thousands of provenance rows a minute this is noise; it is recorded because a permanent write tax should carry a number, not an adjective. + +## Alternatives considered + +- **Uniform triggers everywhere** (the first cut of the migration). One mechanism for all 39 edges is simpler to describe and needs no new unique indexes, but it pays PL/pgSQL calls where a kernel-level RI check does the same work, and — the deciding point — user triggers go silent under `session_replication_role = replica` while declarative constraints do not. The replica-mode hole is real enough that uniformity was not worth it. +- **Fully declarative: add `kb_id` to the link tables.** `fact_evidence`, `fact_qualifiers`, `fact_derivations`, `entity_type_parents`, `relation_type_*`, `attribute_rule_conditions`, `typed_fact_sources`, `statement_qualifiers` could grow a `kb_id` column and take composite keys too. But that column is a denormalization of the parent's `kb_id`, and keeping it equal is a *second* invariant — which would itself need a trigger to enforce. Trading a check for a column plus the same check is the worst of both. +- **Application-layer checks.** The exporter already filters cross-KB references defensively; that is a backstop for readers, not an invariant for writers. The schema is the only layer every write path — present and future — passes through. +- **`MATCH FULL` composite keys** were considered for nullable edges and rejected: `MATCH SIMPLE` (skip the check when the reference is NULL) preserves the existing nullable-edge semantics exactly; nothing here makes a NULL reference meaningful. + +## Open questions + +- Whether maintainers prefer the hybrid split recorded here or uniform triggers (issue #842). The trigger machinery is additive — moving an edge from §1b to §1c is a one-line change in either direction, so the decision is cheap to revisit. +- Whether the eight supporting `UNIQUE (kb_id, id)` indexes should be partial indexes over live rows instead; measured cost does not currently justify the extra subtlety. diff --git a/docs/decisions/README.md b/docs/decisions/README.md index 6fc359926..72d3d1045 100644 --- a/docs/decisions/README.md +++ b/docs/decisions/README.md @@ -73,6 +73,7 @@ The test for writing one: if someone (including us) looks at a piece of code in | 0045 | [A time mention is resolved against its document](0045-a-time-mention-is-resolved-against-its-document.md) | Accepted · cuts 1 and 2 built (#740): a document is dated from its own text, each mention is interpreted by the model and computed by code, upload time is used nowhere · cuts 3 and 4 (grades replace the confidence gate, re-resolution and the anchor queue) not built · a time expression is a mention with its words and place; the model returns shape, anchor, offset and granularity and code computes the interval; a document carries its own date, calendars and anchors across chunks, never its upload time; unresolved mentions wait for an anchor; timelines close on resolution grade instead of confidence | | 0046 | [The app surface is MCP](0046-the-app-surface-is-mcp.md) | Decided, with the refused design kept. Asked for an app center: applications built on this knowledge, mounted, run in a sandbox, handed to a team. The answer is that the surface already exists — a personal token carries identity and scope, ten read tools serve chat and MCP from one place, `as_of` reaches every graph read, and a read returns `structuredContent` with stable ledger identities — so a coding agent builds on this base today in its own platform, its own language and its own sandbox. Refused here because the layer an app would read is being replaced under it (typed facts now come only from alignment), because 0016 closes open seams before cutting new ones, and because a catalog, an execution boundary and quotas are three other products. The shape is kept with the four gates it would have to hold (runs as the caller, egress only through a declared action, a declared clock, the existing queue) and the dead ends: a container runtime (withdrawn the day it was written — WeKnora's skills are human-written and assume a shell, and they pay for it), a Wasm component runtime (better on every axis including the determinism re-parse needs, still not built because the reason is priority), a service identity per app, an app as a saved conversation. Reopened by a named customer who needs a button inside the product, by the type layer settling, or after 0034 | | 0047 | [A rule may conclude a relation](0047-a-rule-may-conclude-a-relation.md) | Proposed 2026-09-20 · nothing built · A rule reads one entity and concludes about that same entity, so a threshold over a chain — a holding above 50% in a company that itself holds above 50% in another — cannot be written at all, and the query-time path walk that answers it produces no interval, no premises and nothing a queue can see. The conclusion becomes a **relation** between the subject and one entity reached across one declared relation, valid on the intersection of every premise interval including the join edge's. The concluded edge rejoins the pool `derive()` reads and the axiom pass runs once per round, coupling the two reasoners for the first time: 0021's cycle objection is answered with the **finiteness** argument [0030](0030-a-rule-may-read-what-a-rule-concluded.md) already put in place of acyclicity, rather than with a fixed ordering that would let a legitimate rule silently never fire. Reading a value across a hop is [0032](0032-a-rule-computes-what-it-concludes.md)'s decision, reused rather than re-decided. Negation, aggregation, a second hop and user-defined recursion stay out; the three caps in play are set by measurement in the PR that changes them | +| 0048 | [Provenance references stay inside the knowledge base](0048-provenance-references-stay-inside-the-knowledge-base.md) | Proposed 2026-09-20 · implemented in PR #832 (migration 0070), pending review · a column foreign key proves the target exists, not that it is the same KB's — every reference an export can resolve gets a schema-level same-KB invariant: composite `(kb_id, ref)` foreign keys on the 26 edges whose row carries its own `kb_id` (same-table self-references deferred to commit), row triggers on the 13 whose kb authority is a parent row, `kb_id` immutability on every owned table, and a precondition scan that fails the migration closed on an already-cross-KB ledger · measured populate cost within noise; mechanism question open as issue #842 | | | Record | Domain | Status | |---|---|---|---| @@ -123,6 +124,7 @@ The test for writing one: if someone (including us) looks at a piece of code in | 0045 | [A time mention is resolved against its document](0045-a-time-mention-is-resolved-against-its-document.md) | time | current | | 0046 | [The app surface is MCP](0046-the-app-surface-is-mcp.md) | chat-and-mcp | current | | 0047 | [A rule may conclude a relation](0047-a-rule-may-conclude-a-relation.md) | rules | current | +| 0048 | [Provenance references stay inside the knowledge base](0048-provenance-references-stay-inside-the-knowledge-base.md) | ledger | current | The status word is whether a later record has overtaken this one; what is built is in the record's own status line. Domains are the files of [../design/](../design/README.md), where every record is dated and the status words are defined.