From dde278f9b9ab135e38e77f666e27ae54bc4f4916 Mon Sep 17 00:00:00 2001 From: Bogdan Szabo Date: Sun, 25 Oct 2026 15:10:27 +0100 Subject: [PATCH 1/2] A role word is recorded against the chunk that says it, so a re-push can say something new Signed-off-by: Bogdan Szabo --- crates/utopia-cli/src/main.rs | 2 +- crates/utopia-server/src/api/mcp_tests.rs | 5 +- crates/utopia-server/src/extraction_open.rs | 1 + .../src/export_provenance_integrity.sql | 8 + crates/utopia-store/src/graph.rs | 16 +- crates/utopia-store/src/materialize.rs | 4 +- crates/utopia-store/src/pending.rs | 3 +- .../a_bound_statement_becomes_a_typed_fact.rs | 13 +- ...ole_word_moves_onto_its_evidence_chunks.rs | 289 ++++++++++++++++++ ...pen_statement_keeps_the_documents_words.rs | 37 ++- .../exported_references_never_cross_a_kb.rs | 31 +- crates/utopia-store/tests/store/main.rs | 1 + ...gration_0070_runs_under_any_search_path.rs | 16 +- ...ferences-stay-inside-the-knowledge-base.md | 8 +- ...ole-word-speaks-while-its-chunk-is-live.md | 2 +- docs/decisions/README.md | 2 +- ...ole_word_is_recorded_against_its_chunk.sql | 125 ++++++++ 17 files changed, 518 insertions(+), 45 deletions(-) create mode 100644 crates/utopia-store/tests/store/a_role_word_moves_onto_its_evidence_chunks.rs create mode 100644 migrations/0121_a_role_word_is_recorded_against_its_chunk.sql diff --git a/crates/utopia-cli/src/main.rs b/crates/utopia-cli/src/main.rs index 913dcf3e0..617df6e55 100644 --- a/crates/utopia-cli/src/main.rs +++ b/crates/utopia-cli/src/main.rs @@ -82,7 +82,7 @@ struct ManifestDataDir { /// not a side effect of a code change. // 是迁移文件的**个数**，不是最大的编号（守卫 `schema_version_policy_compares_against_current` // 按个数比）：编号有空缺时两者不同——0071 由一个开放 PR 占着，0072 先落，个数是 71 -const CURRENT_SCHEMA_VERSION: u32 = 110; +const CURRENT_SCHEMA_VERSION: u32 = 111; #[tokio::main] async fn main() -> anyhow::Result<()> { diff --git a/crates/utopia-server/src/api/mcp_tests.rs b/crates/utopia-server/src/api/mcp_tests.rs index ca6f3e280..d92916da3 100644 --- a/crates/utopia-server/src/api/mcp_tests.rs +++ b/crates/utopia-server/src/api/mcp_tests.rs @@ -1008,10 +1008,11 @@ async fn entity_facts_reads_back_a_statements_role_words() -> anyhow::Result<()> return Ok(()); }; sqlx::query( - "INSERT INTO statement_qualifiers(fact_id,role,value) - VALUES ($1,'lines','{\"value\":\"10-42\"}'::jsonb)", + "INSERT INTO statement_qualifiers(fact_id,role,value,chunk_id) + VALUES ($1,'lines','{\"value\":\"10-42\"}'::jsonb,$2)", ) .bind(f.corrected) + .bind(f.chunk) .execute(&f.state.pool) .await?; let result = f diff --git a/crates/utopia-server/src/extraction_open.rs b/crates/utopia-server/src/extraction_open.rs index b328583b6..e9cf5cb84 100644 --- a/crates/utopia-server/src/extraction_open.rs +++ b/crates/utopia-server/src/extraction_open.rs @@ -665,6 +665,7 @@ async fn apply_open_extraction( utopia_store::graph::add_statement_qualifier( pool, fact_id, + chunk.id, role, value.as_ref(), *entity, diff --git a/crates/utopia-store/src/export_provenance_integrity.sql b/crates/utopia-store/src/export_provenance_integrity.sql index c4c894832..3132ece48 100644 --- a/crates/utopia-store/src/export_provenance_integrity.sql +++ b/crates/utopia-store/src/export_provenance_integrity.sql @@ -266,6 +266,14 @@ SELECT edge, kind, COUNT(*) AS rows FROM ( LEFT JOIN entities e ON e.id = q.entity_id WHERE f.kb_id = $1 AND q.entity_id IS NOT NULL UNION ALL + -- @edge statement_qualifiers.chunk_id -> chunks.id + -- 角色词记在哪一块上：归属同样按所属 fact 的库判 + SELECT 'squalifier.chunk', 'cross_kb', c.kb_id IS DISTINCT FROM f.kb_id + FROM statement_qualifiers q + JOIN facts f ON f.id = q.fact_id + LEFT JOIN chunks c ON c.id = q.chunk_id + WHERE f.kb_id = $1 + UNION ALL -- @edge time_mentions.fact_id -> facts.id -- 时间提及以行自己的 kb 归属：指的事实与段落都必须同库 SELECT 'timemention.fact', 'cross_kb', f.kb_id IS DISTINCT FROM t.kb_id diff --git a/crates/utopia-store/src/graph.rs b/crates/utopia-store/src/graph.rs index a68aef9fd..02632074f 100644 --- a/crates/utopia-store/src/graph.rs +++ b/crates/utopia-store/src/graph.rs @@ -199,12 +199,12 @@ pub async fn fact_qualifiers_for( Ok(out) } -/// 往一条开放陈述上写一个按角色词记的属性。**先写者留着**：同一条陈述同一个角色词 -/// 再听到一次不覆盖（与 [`upsert_fact_qualifier`] 的 Conflict 一条规矩——两次观察 -/// 不一致该另立一行加一条冲突，这一刀还没走到那一步）。值与实体必须恰好给一个 +/// 往一条开放陈述上写一个按角色词记的属性，记在说它的那一块上（0070 决定 1）。 +/// 同一块再说一次不写新行；另一块说的另记一行，旧行不动。值与实体必须恰好给一个 pub async fn add_statement_qualifier( pool: &PgPool, fact_id: Uuid, + chunk_id: Uuid, role: &str, value: Option<&serde_json::Value>, entity_id: Option, @@ -223,19 +223,21 @@ pub async fn add_statement_qualifier( )); } sqlx::query( - "INSERT INTO statement_qualifiers (fact_id, role, value, entity_id) - VALUES ($1, $2, $3, $4) ON CONFLICT (fact_id, role) DO NOTHING", + "INSERT INTO statement_qualifiers (fact_id, role, value, entity_id, chunk_id) + VALUES ($1, $2, $3, $4, $5) ON CONFLICT (fact_id, role, chunk_id) DO NOTHING", ) .bind(fact_id) .bind(role) .bind(value) .bind(entity_id) + .bind(chunk_id) .execute(pool) .await?; Ok(()) } -/// 一批开放陈述各自的角色词属性，按事实 id 取回；实体那一格连上显示名 +/// 一批开放陈述各自的角色词属性，按事实 id 取回；实体那一格连上显示名。 +/// 同一个值记在几块上只返回一次 pub async fn statement_qualifiers_for( pool: &PgPool, fact_ids: &[Uuid], @@ -253,7 +255,7 @@ pub async fn statement_qualifiers_for( entity_name: Option, } let rows: Vec = sqlx::query_as( - "SELECT q.fact_id, q.role, q.value, q.entity_id, e.canonical_name AS entity_name + "SELECT DISTINCT q.fact_id, q.role, q.value, q.entity_id, e.canonical_name AS entity_name FROM statement_qualifiers q LEFT JOIN entities e ON e.id = q.entity_id WHERE q.fact_id = ANY($1) diff --git a/crates/utopia-store/src/materialize.rs b/crates/utopia-store/src/materialize.rs index c2d263cee..7937b3fa3 100644 --- a/crates/utopia-store/src/materialize.rs +++ b/crates/utopia-store/src/materialize.rs @@ -424,8 +424,8 @@ async fn materialize_in_tx( .execute(&mut **tx) .await?; sqlx::query( - "INSERT INTO statement_qualifiers (fact_id, role, value, entity_id) - SELECT $1, role, value, entity_id FROM statement_qualifiers WHERE fact_id = $2 + "INSERT INTO statement_qualifiers (fact_id, role, value, entity_id, chunk_id) + SELECT $1, role, value, entity_id, chunk_id FROM statement_qualifiers WHERE fact_id = $2 ON CONFLICT DO NOTHING", ) .bind(fact) diff --git a/crates/utopia-store/src/pending.rs b/crates/utopia-store/src/pending.rs index 8d581902d..03f402c41 100644 --- a/crates/utopia-store/src/pending.rs +++ b/crates/utopia-store/src/pending.rs @@ -499,7 +499,8 @@ async fn confirm_open( } else { q.get("value") }; - crate::graph::add_statement_qualifier(pool, fact_id, role, value, entity_id).await?; + crate::graph::add_statement_qualifier(pool, fact_id, v.chunk_id, role, value, entity_id) + .await?; } for t in json_items(v.time_words.as_ref()) { let (Some(text), Some(char_start)) = ( diff --git a/crates/utopia-store/tests/store/a_bound_statement_becomes_a_typed_fact.rs b/crates/utopia-store/tests/store/a_bound_statement_becomes_a_typed_fact.rs index c53a3dd55..2f20afc8e 100644 --- a/crates/utopia-store/tests/store/a_bound_statement_becomes_a_typed_fact.rs +++ b/crates/utopia-store/tests/store/a_bound_statement_becomes_a_typed_fact.rs @@ -133,21 +133,26 @@ async fn a_bound_statement_becomes_a_typed_fact() -> anyhow::Result<()> { .bind(doc) .execute(&pool) .await?; - anyhow::Ok(()) + anyhow::Ok(chunk) } }; - evidence_of(stated, "bakery.txt").await?; + let stated_chunk = evidence_of(stated, "bakery.txt").await?; evidence_of(again, "bakery-2.txt").await?; + let planned_chunk = evidence_of(planned, "bakery-plan.txt").await?; sqlx::query( - "INSERT INTO statement_qualifiers (fact_id, role, value) VALUES ($1, 'since', '\"2019\"'::jsonb)", + "INSERT INTO statement_qualifiers (fact_id, role, value, chunk_id) + VALUES ($1, 'since', '\"2019\"'::jsonb, $2)", ) .bind(stated) + .bind(stated_chunk) .execute(&pool) .await?; sqlx::query( - "INSERT INTO statement_qualifiers (fact_id, role, value) VALUES ($1, 'mood', '\"will\"'::jsonb)", + "INSERT INTO statement_qualifiers (fact_id, role, value, chunk_id) + VALUES ($1, 'mood', '\"will\"'::jsonb, $2)", ) .bind(planned) + .bind(planned_chunk) .execute(&pool) .await?; diff --git a/crates/utopia-store/tests/store/a_role_word_moves_onto_its_evidence_chunks.rs b/crates/utopia-store/tests/store/a_role_word_moves_onto_its_evidence_chunks.rs new file mode 100644 index 000000000..ce8188f40 --- /dev/null +++ b/crates/utopia-store/tests/store/a_role_word_moves_onto_its_evidence_chunks.rs @@ -0,0 +1,289 @@ +//! 0121：角色词搬到它事实的每一块证据上（0070 决定 1，#1267）。 +//! +//! 开一个隔离库迁到 0120，种下旧形状的行，再跑 0121： +//! - 每块证据（被取代的也算）都拿到陈述的角色词，读出来仍是一个值； +//! - 类型化行只在说过这个词、且值与它相同的来源陈述的块上拿到它； +//! - 已失效、没有证据的事实（purge 留下的）的角色词被删掉； +//! - 活着的事实有角色词却没有证据，迁移整份中止。 +//! +//! 建库失败（角色没权限的环境）与没有 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")) +} + +/// 按文件顺序跑 `from..=through` 的迁移，各自一个事务（与 sqlx::migrate 同一形状） +async fn migrate(pool: &PgPool, from: i64, through: i64) -> Result<(), sqlx::Error> { + let migrator = sqlx::migrate!("../../migrations"); + let mut conn = pool.acquire().await?; + for m in migrator + .iter() + .filter(|m| (from..=through).contains(&m.version)) + { + let mut tx = conn.begin().await?; + sqlx::raw_sql(&m.sql).execute(&mut *tx).await?; + tx.commit().await?; + } + Ok(()) +} + +/// 隔离库：建 → 迁到 0120 → 返回（库名, 连接池）。失败就地跳过 +async fn scratch(suffix: &str) -> Option<(String, PgPool)> { + let admin_pool = PgPool::connect(&admin_url()?).await.ok()?; + let name = format!("sq121_{}_{}", suffix, Uuid::now_v7().simple()); + let created = sqlx::query(&format!("CREATE DATABASE {name}")) + .execute(&admin_pool) + .await; + admin_pool.close().await; + if created.is_err() { + eprintln!("跳过：建不了隔离库（角色没有 CREATEDB）"); + 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(&pool, 0, 120).await { + eprintln!("跳过：迁到 0120 失败（迁移链自身的问题）: {e}"); + drop_scratch(&name).await; + return None; + } + Some((name, pool)) +} + +async fn drop_scratch(name: &str) { + let Some(admin) = admin_url() else { return }; + 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; + } +} + +struct Ledger { + kb: Uuid, + entity: Uuid, + superseded_chunk: Uuid, + live_chunk: Uuid, +} + +/// 一个库、一个实体、一份文档的两块：旧的那块已被取代 +async fn seed_ledger(pool: &PgPool) -> anyhow::Result { + let (org, ws, kb, entity, doc) = ( + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + ); + let (superseded_chunk, live_chunk) = (Uuid::now_v7(), Uuid::now_v7()); + sqlx::query("INSERT INTO organizations (id, name) VALUES ($1, 'm121-test')") + .bind(org) + .execute(pool) + .await?; + sqlx::query("INSERT INTO workspaces (id, org_id, name) VALUES ($1, $2, 'm121-test')") + .bind(ws) + .bind(org) + .execute(pool) + .await?; + sqlx::query( + "INSERT INTO knowledge_bases (id, workspace_id, name) VALUES ($1, $2, 'm121-test')", + ) + .bind(kb) + .bind(ws) + .execute(pool) + .await?; + sqlx::query("INSERT INTO entities (id, kb_id, canonical_name) VALUES ($1, $2, 'parse.rs')") + .bind(entity) + .bind(kb) + .execute(pool) + .await?; + sqlx::query( + "INSERT INTO documents (id, kb_id, filename, sha256, status) + VALUES ($1, $2, 'ci.json', 'm121', 'ready')", + ) + .bind(doc) + .bind(kb) + .execute(pool) + .await?; + sqlx::query( + "INSERT INTO chunks (id, kb_id, document_id, seq, text, superseded_at) + VALUES ($1, $3, $4, 0, 'lines 40-52', now()), ($2, $3, $4, 1, 'lines 58-70', NULL)", + ) + .bind(superseded_chunk) + .bind(live_chunk) + .bind(kb) + .bind(doc) + .execute(pool) + .await?; + Ok(Ledger { + kb, + entity, + superseded_chunk, + live_chunk, + }) +} + +/// 一条事实；`open` 是开放陈述，否则是类型化行 +async fn fact(pool: &PgPool, ledger: &Ledger, open: bool, live: bool) -> anyhow::Result { + let fact = Uuid::now_v7(); + sqlx::query( + "INSERT INTO facts (id, kb_id, subject_id, object_id, confidence, invalidated_at, layer, phrase) + VALUES ($1, $2, $3, $3, 0.9, CASE WHEN $4 THEN NULL ELSE now() END, + CASE WHEN $5 THEN 'open' ELSE 'typed' END, CASE WHEN $5 THEN 'covers' END)", + ) + .bind(fact) + .bind(ledger.kb) + .bind(ledger.entity) + .bind(live) + .bind(open) + .execute(pool) + .await?; + Ok(fact) +} + +/// 按 0120 的形状写一个 `lines` 角色词（没有 chunk_id） +async fn says_lines(pool: &PgPool, fact: Uuid) -> anyhow::Result<()> { + says_lines_as(pool, fact, "40-52").await +} + +async fn says_lines_as(pool: &PgPool, fact: Uuid, lines: &str) -> anyhow::Result<()> { + sqlx::query("INSERT INTO statement_qualifiers (fact_id, role, value) VALUES ($1, 'lines', $2)") + .bind(fact) + .bind(serde_json::json!(lines)) + .execute(pool) + .await?; + Ok(()) +} + +async fn evidence(pool: &PgPool, fact: Uuid, chunks: &[Uuid]) -> anyhow::Result<()> { + for chunk in chunks { + sqlx::query("INSERT INTO fact_evidence (fact_id, chunk_id) VALUES ($1, $2)") + .bind(fact) + .bind(chunk) + .execute(pool) + .await?; + } + Ok(()) +} + +async fn role_word_chunks(pool: &PgPool, facts: Vec) -> anyhow::Result> { + let rows = sqlx::query_as( + "SELECT fact_id, chunk_id FROM statement_qualifiers + WHERE fact_id = ANY($1) ORDER BY fact_id, chunk_id", + ) + .bind(facts) + .fetch_all(pool) + .await?; + Ok(rows) +} + +#[tokio::test] +async fn role_words_move_onto_every_evidence_chunk_and_leave_a_purged_fact() -> anyhow::Result<()> { + let Some((name, pool)) = scratch("chunks").await else { + return Ok(()); + }; + let ledger = seed_ledger(&pool).await?; + let stated = fact(&pool, &ledger, true, true).await?; + says_lines(&pool, stated).await?; + evidence(&pool, stated, &[ledger.superseded_chunk, ledger.live_chunk]).await?; + let purged = fact(&pool, &ledger, true, false).await?; + says_lines(&pool, purged).await?; + + migrate(&pool, 121, 121).await?; + + let mut expected = vec![ + (stated, ledger.superseded_chunk), + (stated, ledger.live_chunk), + ]; + expected.sort(); + assert_eq!( + role_word_chunks(&pool, vec![stated, purged]).await?, + expected, + "每块证据一行，没有证据的已失效事实一行不留" + ); + let read = utopia_store::graph::statement_qualifiers_for(&pool, &[stated]).await?; + assert_eq!( + serde_json::to_value(&read[&stated])?, + serde_json::json!([{ "role": "lines", "value": "40-52", "entity_id": null, "entity_name": null }]) + ); + pool.close().await; + drop_scratch(&name).await; + Ok(()) +} + +#[tokio::test] +async fn a_typed_row_takes_a_role_word_only_on_the_chunk_of_a_statement_that_said_the_same_value( +) -> anyhow::Result<()> { + let Some((name, pool)) = scratch("typed").await else { + return Ok(()); + }; + let ledger = seed_ledger(&pool).await?; + let (said, silent, disagreed) = ( + fact(&pool, &ledger, true, true).await?, + fact(&pool, &ledger, true, true).await?, + fact(&pool, &ledger, true, true).await?, + ); + says_lines(&pool, said).await?; + evidence(&pool, said, &[ledger.superseded_chunk]).await?; + evidence(&pool, silent, &[ledger.live_chunk]).await?; + // 说了另一个值的来源：它的块不替类型化行说话，类型化行仍只读出原来的值 + says_lines_as(&pool, disagreed, "58-70").await?; + evidence(&pool, disagreed, &[ledger.live_chunk]).await?; + let typed = fact(&pool, &ledger, false, true).await?; + says_lines(&pool, typed).await?; + evidence(&pool, typed, &[ledger.superseded_chunk, ledger.live_chunk]).await?; + sqlx::query("INSERT INTO typed_fact_sources (fact_id, statement_id) VALUES ($1, $2), ($1, $3), ($1, $4)", + ) + .bind(typed) + .bind(said) + .bind(silent) + .bind(disagreed) + .execute(&pool) + .await?; + // 来源已经不在的类型化行：角色词照旧抄到它的证据上，读出来不少 + let sourceless = fact(&pool, &ledger, false, true).await?; + says_lines(&pool, sourceless).await?; + evidence(&pool, sourceless, &[ledger.live_chunk]).await?; + + migrate(&pool, 121, 121).await?; + + let mut expected = vec![ + (typed, ledger.superseded_chunk), + (sourceless, ledger.live_chunk), + ]; + expected.sort(); + assert_eq!( + role_word_chunks(&pool, vec![typed, sourceless]).await?, + expected + ); + pool.close().await; + drop_scratch(&name).await; + Ok(()) +} + +#[tokio::test] +async fn a_live_fact_with_role_words_and_no_evidence_stops_the_migration() -> anyhow::Result<()> { + let Some((name, pool)) = scratch("orphan").await else { + return Ok(()); + }; + let ledger = seed_ledger(&pool).await?; + let orphan = fact(&pool, &ledger, true, true).await?; + says_lines(&pool, orphan).await?; + + let err = migrate(&pool, 121, 121).await.unwrap_err().to_string(); + + assert!( + err.contains("live facts with role words and no evidence (x1)"), + "{err}" + ); + pool.close().await; + drop_scratch(&name).await; + Ok(()) +} diff --git a/crates/utopia-store/tests/store/an_open_statement_keeps_the_documents_words.rs b/crates/utopia-store/tests/store/an_open_statement_keeps_the_documents_words.rs index ddc0b1020..46983265b 100644 --- a/crates/utopia-store/tests/store/an_open_statement_keeps_the_documents_words.rs +++ b/crates/utopia-store/tests/store/an_open_statement_keeps_the_documents_words.rs @@ -147,6 +147,7 @@ async fn an_open_statement_shows_under_its_phrase_and_reuses_its_row() -> anyhow graph::add_statement_qualifier( &pool, fact, + f.chunk_a, "amount", Some(&serde_json::json!({ "value": "$1.2 billion" })), None, @@ -318,27 +319,39 @@ async fn an_open_statement_shows_under_its_phrase_and_reuses_its_row() -> anyhow Some(serde_json::json!({ "value": "$1.2 billion" })) ); assert!(q[0].entity_id.is_none()); - // 先写者留着 - graph::add_statement_qualifier( - &pool, - fact, - "amount", - Some(&serde_json::json!({ "value": "$2 billion" })), - None, + // 角色词记在说它的那一块上：另一块的值另记一行，同一块再说一次不添行 + for (chunk, amount) in [(f.chunk_b, "$2 billion"), (f.chunk_a, "$1.2 billion")] { + graph::add_statement_qualifier( + &pool, + fact, + chunk, + "amount", + Some(&serde_json::json!({ "value": amount })), + None, + ) + .await?; + } + let amounts: Vec<(Uuid, serde_json::Value)> = sqlx::query_as( + "SELECT chunk_id, value FROM statement_qualifiers + WHERE fact_id = $1 AND role = 'amount' ORDER BY chunk_id", ) + .bind(fact) + .fetch_all(&pool) .await?; - let q = &graph::statement_qualifiers_for(&pool, &[fact]).await?[&fact]; assert_eq!( - q[0].value, - Some(serde_json::json!({ "value": "$1.2 billion" })) + amounts, + vec![ + (f.chunk_a, serde_json::json!({ "value": "$1.2 billion" })), + (f.chunk_b, serde_json::json!({ "value": "$2 billion" })), + ] ); // 值与实体必须恰好一个 assert!( - graph::add_statement_qualifier(&pool, fact, "buyer", None, None) + graph::add_statement_qualifier(&pool, fact, f.chunk_a, "buyer", None, None) .await .is_err() ); - graph::add_statement_qualifier(&pool, fact, "buyer", None, Some(f.acme)).await?; + graph::add_statement_qualifier(&pool, fact, f.chunk_a, "buyer", None, Some(f.acme)).await?; let q = &graph::statement_qualifiers_for(&pool, &[fact]).await?[&fact]; let buyer = q.iter().find(|x| x.role == "buyer").unwrap(); assert_eq!(buyer.entity_id, Some(f.acme)); diff --git a/crates/utopia-store/tests/store/exported_references_never_cross_a_kb.rs b/crates/utopia-store/tests/store/exported_references_never_cross_a_kb.rs index 281a1c1a9..481523ce5 100644 --- a/crates/utopia-store/tests/store/exported_references_never_cross_a_kb.rs +++ b/crates/utopia-store/tests/store/exported_references_never_cross_a_kb.rs @@ -379,14 +379,25 @@ async fn new_cross_kb_writes_are_rejected_on_every_exported_edge() -> anyhow::Re // —— 开放陈述的属性（statement_qualifiers，0061）：行没有 kb 列，归属按 // 所属 fact 的库判；别库的实体值会被铸进本库 IRI let err = sqlx::query( - "INSERT INTO statement_qualifiers (fact_id, role, entity_id) - VALUES ($1, 'as', $2)", + "INSERT INTO statement_qualifiers (fact_id, role, entity_id, chunk_id) + VALUES ($1, 'as', $2, $3)", ) .bind(f.fact_a) .bind(f.ent_b) + .bind(f.chunk_a) .execute(&pool) .await; assert!(err.is_err(), "squalifier.entity→foreign 必须被拒"); + // 角色词记在别库的块上（0070 决定 1） + let err = sqlx::query( + "INSERT INTO statement_qualifiers (fact_id, role, value, chunk_id) + VALUES ($1, 'lines', '\"1\"'::jsonb, $2)", + ) + .bind(f.fact_a) + .bind(f.chunk_b) + .execute(&pool) + .await; + assert!(err.is_err(), "squalifier.chunk→foreign 必须被拒"); // —— 时间提及（0061/0064）：提及自己的 kb 必须与它指的事实、段落同属一库 let err = sqlx::query( @@ -989,13 +1000,22 @@ async fn malformed_rows_fail_every_exported_edge_closed() -> anyhow::Result<()> .bind(f.fact_b) .execute(&mut *tx) .await?; - // 开放陈述的属性值指着别库实体 + // 开放陈述的属性值指着别库实体；另一个角色词记在别库的块上 sqlx::query( - "INSERT INTO statement_qualifiers (fact_id, role, entity_id) - VALUES ($1, 'as', $2)", + "INSERT INTO statement_qualifiers (fact_id, role, entity_id, chunk_id) + VALUES ($1, 'as', $2, $3)", ) .bind(f.fact_a) .bind(f.ent_b) + .bind(f.chunk_a) + .execute(&mut *tx) + .await?; + sqlx::query( + "INSERT INTO statement_qualifiers (fact_id, role, value, chunk_id) + VALUES ($1, 'lines', '\"1\"'::jsonb, $2)", + ) + .bind(f.fact_a) + .bind(f.chunk_b) .execute(&mut *tx) .await?; // 时间提及两种坏法：归属在 A 却指着 B 的段落（体检扫得见——行按自己的 @@ -1466,6 +1486,7 @@ async fn malformed_rows_fail_every_exported_edge_closed() -> anyhow::Result<()> "condition.predicate", "factsource.statement", "squalifier.entity", + "squalifier.chunk", "timemention.fact", "timemention.chunk", "binding.type", diff --git a/crates/utopia-store/tests/store/main.rs b/crates/utopia-store/tests/store/main.rs index 0f6ef7a36..bc6e39235 100644 --- a/crates/utopia-store/tests/store/main.rs +++ b/crates/utopia-store/tests/store/main.rs @@ -84,6 +84,7 @@ mod a_retired_account; mod a_retraction_leaves_the_graph; mod a_review_has_a_summary; mod a_review_page_is_assembled_in_batches; +mod a_role_word_moves_onto_its_evidence_chunks; mod a_rule_computes_what_it_concludes; mod a_rule_concludes_a_type; mod a_rule_reads_what_a_rule_concluded; diff --git a/crates/utopia-store/tests/store/migration_0070_runs_under_any_search_path.rs b/crates/utopia-store/tests/store/migration_0070_runs_under_any_search_path.rs index 7949f1145..21cfcc790 100644 --- a/crates/utopia-store/tests/store/migration_0070_runs_under_any_search_path.rs +++ b/crates/utopia-store/tests/store/migration_0070_runs_under_any_search_path.rs @@ -428,6 +428,11 @@ const TRIGGER_EDGES: &[(&str, &str, &str)] = &[ "entity_type_id", "relation_type_ranges_same_kb", ), + ( + "statement_qualifiers", + "chunk_id", + "statement_qualifiers_same_kb", + ), ( "statement_qualifiers", "entity_id", @@ -619,7 +624,7 @@ struct Coverage { impl Coverage { /// 完备 = 没有未归类的边、没有腐掉的登记、面与机制互证得上, /// 每条受保护边都进了导出体检、体检里没有腐掉的分支, - /// 且两类已覆盖边的数量与当前迁移链的 44/16 一致。 + /// 且两类已覆盖边的数量与当前迁移链的 44/17 一致。 fn assert_complete(&self) -> anyhow::Result<()> { let mut problems = String::new(); let mut dump = |name: &str, xs: &[String]| { @@ -641,9 +646,10 @@ impl Coverage { self.declarative.len() )); } - if self.trigger_covered.len() != 16 { + // 0121 adds the chunk a role word is recorded against. + if self.trigger_covered.len() != 17 { problems.push_str(&format!( - " TRIGGER_EDGES = {} (expected 16)\n", + " TRIGGER_EDGES = {} (expected 17)\n", self.trigger_covered.len() )); } @@ -1062,8 +1068,8 @@ async fn a_protected_edge_missing_from_preflight_fails_the_guard() -> anyhow::Re #[test] fn the_preflight_check_notices_a_dropped_or_stray_edge() -> anyhow::Result<()> { let surface = preflight_surface()?; - // 同一份文件两个方向各数一次:60 条结构边 + 5 条 merged 过滤检查 - assert_eq!(surface.edges.len(), 60, "preflight 必须覆盖全部保护边"); + // 同一份文件两个方向各数一次:61 条结构边 + 5 条 merged 过滤检查 + assert_eq!(surface.edges.len(), 61, "preflight 必须覆盖全部保护边"); assert_eq!( surface.filters, [ 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 index fc1a24ff4..6496b486e 100644 --- a/docs/decisions/0048-provenance-references-stay-inside-the-knowledge-base.md +++ b/docs/decisions/0048-provenance-references-stay-inside-the-knowledge-base.md @@ -16,7 +16,7 @@ Every reference an export can resolve must join rows that live in the same knowl ## 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?** +Forty reference edges are protected (thirty-nine at migration 0070; 0121 added `statement_qualifiers.chunk_id`). The split is decided by a single question: **does the referencing row carry the kb authority itself?** | Source | Edge → target | kb authority | Mechanism | |---|---|---|---| @@ -40,11 +40,11 @@ Thirty-nine reference edges are protected. The split is decided by a single ques | `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 | +| `statement_qualifiers` | `entity_id` → `entities`; `chunk_id` → `chunks` | 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. +**14 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. @@ -54,7 +54,7 @@ Thirty-nine reference edges are protected. The split is decided by a single ques `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. +- `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 14 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 same guard also reads the exact preflight scan the export runs (`export_provenance_integrity.sql`) and fails when a protected edge has no scan branch — or a scan branch no longer names a protected edge — so schema protection and export preflight cannot drift apart. diff --git a/docs/decisions/0070-a-role-word-speaks-while-its-chunk-is-live.md b/docs/decisions/0070-a-role-word-speaks-while-its-chunk-is-live.md index 7cd32a6ac..5d50f2617 100644 --- a/docs/decisions/0070-a-role-word-speaks-while-its-chunk-is-live.md +++ b/docs/decisions/0070-a-role-word-speaks-while-its-chunk-is-live.md @@ -1,6 +1,6 @@ # 0070 · A role word speaks while its chunk is live -- **Status**: Proposed · 2026-10-16 · [#1267](https://github.com/deeplethe/utopia/issues/1267) · open: all five decisions +- **Status**: In progress · 2026-10-25 · [#1267](https://github.com/deeplethe/utopia/issues/1267) · decision 1 built (migration 0121) · open: decisions 2 to 4 - **Written**: 2026-10-16 (conventions in the [README](README.md)) - **Related**: [0037](0037-a-relation-carries-its-own-attributes.md) decided that a qualifier is not part of a fact's identity and that the first value stays; [0044](0044-the-ontology-is-a-view-over-what-documents-say.md) owns the open statement and the role words on it; [0054](0054-a-source-may-push-statements-in-the-open-contract.md) decision 7 is the re-push that supersedes a chunk; migration `0061` created `statement_qualifiers`; [0048](0048-provenance-references-stay-inside-the-knowledge-base.md) gets one more edge to keep inside the base; [0019](0019-the-second-clock-can-be-rewound.md) is the record axis `as_of` reads on; [#1267](https://github.com/deeplethe/utopia/issues/1267), [#1272](https://github.com/deeplethe/utopia/issues/1272) diff --git a/docs/decisions/README.md b/docs/decisions/README.md index 564febbf8..10a3051e0 100644 --- a/docs/decisions/README.md +++ b/docs/decisions/README.md @@ -75,7 +75,7 @@ By domain; the domains are the files of [../design/](../design/README.md). **Sta | 0010 | [An unnamed relation stays empty](0010-no-relation-is-no-relation.md) | Implemented | partly (by 0044) | | 0037 | [A relation carries its own attributes](0037-a-relation-carries-its-own-attributes.md) | In progress | partly (by 0044) | | 0048 | [Provenance references stay inside the knowledge base](0048-provenance-references-stay-inside-the-knowledge-base.md) | Implemented | | -| 0070 | [A role word speaks while its chunk is live](0070-a-role-word-speaks-while-its-chunk-is-live.md) | Proposed | | +| 0070 | [A role word speaks while its chunk is live](0070-a-role-word-speaks-while-its-chunk-is-live.md) | In progress | | ### [governance](../design/governance.md) diff --git a/migrations/0121_a_role_word_is_recorded_against_its_chunk.sql b/migrations/0121_a_role_word_is_recorded_against_its_chunk.sql new file mode 100644 index 000000000..8b48c822d --- /dev/null +++ b/migrations/0121_a_role_word_is_recorded_against_its_chunk.sql @@ -0,0 +1,125 @@ +-- 角色词记在说它的那一块上（0070 决定 1，#1267）。 +-- +-- `statement_qualifiers` 从前按 (fact, role) 只留一行、先写者留着：同一个 external_id 重推、 +-- 值变了，新值被 `DO NOTHING` 丢掉。现在每条角色词带上说它的 `chunk_id`，主键变成 +-- (fact, role, chunk)，新推的块写新行，旧行不动。 +-- +-- 回填分三步，先陈述、后由陈述推出的行： +-- 1. 开放陈述：每个 (fact, role) 的现值抄到它的每一块证据上，被取代的块也抄。 +-- 2. 类型化行与隐含行：从来源陈述（`typed_fact_sources`、`implied_fact_sources`）第 1 步 +-- 写下的逐块行里，挑出与这一行自己的值和实体都相同的，抄下它们落在这一行证据上的块。 +-- 于是另一条来源陈述的块不会替没说过这个词的陈述说话，说了别的值的来源也不会。 +-- 3. 剩下的（来源已经不在、或没有一条来源给过同样的值）照第 1 步抄到这一行的每一块证据上， +-- 今天读得到的角色词一个不丢。 +-- 每个 (fact, role) 只留下原来那一个值，所以迁移之后读出来的和之前一样。几条来源陈述对同一个 +-- 角色词给了不同的值时，类型化行仍只读它原来的值；别的值由 0070 决定 3 带过来。 +-- +-- 没有证据的事实：`documents::purge` 删掉块和证据、留下已失效的事实和它的角色词。这些行 +-- 没有块可挂，删掉并报数。**活着的**事实有角色词却没有证据，是账本坏了，整份中止。 +-- +-- `chunk_id` 是一条新的出处边（0048）：触发器同时查块与事实同库，导出体检扫它。 + +-- §0 前置检查：活着的事实有角色词却没有证据 → 中止 +DO $$ +DECLARE + orphans bigint; +BEGIN + SELECT COUNT(DISTINCT q.fact_id) INTO orphans + FROM public.statement_qualifiers q + JOIN public.facts f ON f.id = q.fact_id + WHERE f.invalidated_at IS NULL + AND NOT EXISTS (SELECT 1 FROM public.fact_evidence e WHERE e.fact_id = q.fact_id); + IF orphans > 0 THEN + RAISE EXCEPTION 'live facts with role words and no evidence (x%) — repair the ledger before role words can move onto chunks', orphans; + END IF; +END; +$$; + +DROP TRIGGER statement_qualifiers_same_kb ON public.statement_qualifiers; + +ALTER TABLE public.statement_qualifiers + ADD COLUMN chunk_id uuid, + DROP CONSTRAINT statement_qualifiers_pkey; + +INSERT INTO public.statement_qualifiers (fact_id, role, value, entity_id, chunk_id) +SELECT q.fact_id, q.role, q.value, q.entity_id, e.chunk_id + FROM public.statement_qualifiers q + JOIN public.facts f ON f.id = q.fact_id AND f.layer = 'open' + JOIN public.fact_evidence e ON e.fact_id = q.fact_id + WHERE q.chunk_id IS NULL; + +-- 两条来源陈述在同一块上说了同一个值时只留一行 +INSERT INTO public.statement_qualifiers (fact_id, role, value, entity_id, chunk_id) +SELECT DISTINCT q.fact_id, q.role, q.value, q.entity_id, s.chunk_id + FROM public.statement_qualifiers q + JOIN (SELECT fact_id, statement_id FROM public.typed_fact_sources + UNION + SELECT fact_id, statement_id FROM public.implied_fact_sources + WHERE statement_id IS NOT NULL) src ON src.fact_id = q.fact_id + JOIN public.statement_qualifiers s + ON s.fact_id = src.statement_id AND s.role = q.role AND s.chunk_id IS NOT NULL + AND s.value IS NOT DISTINCT FROM q.value AND s.entity_id IS NOT DISTINCT FROM q.entity_id + JOIN public.fact_evidence e ON e.fact_id = q.fact_id AND e.chunk_id = s.chunk_id + WHERE q.chunk_id IS NULL; + +INSERT INTO public.statement_qualifiers (fact_id, role, value, entity_id, chunk_id) +SELECT q.fact_id, q.role, q.value, q.entity_id, e.chunk_id + FROM public.statement_qualifiers q + JOIN public.fact_evidence e ON e.fact_id = q.fact_id + WHERE q.chunk_id IS NULL + AND NOT EXISTS (SELECT 1 FROM public.statement_qualifiers n + WHERE n.fact_id = q.fact_id AND n.role = q.role AND n.chunk_id IS NOT NULL); + +DO $$ +DECLARE + dropped bigint; +BEGIN + DELETE FROM public.statement_qualifiers q + WHERE q.chunk_id IS NULL + AND NOT EXISTS (SELECT 1 FROM public.fact_evidence e WHERE e.fact_id = q.fact_id); + GET DIAGNOSTICS dropped = ROW_COUNT; + RAISE NOTICE 'dropped % role words on facts with no evidence', dropped; +END; +$$; + +DELETE FROM public.statement_qualifiers WHERE chunk_id IS NULL; + +ALTER TABLE public.statement_qualifiers + ALTER COLUMN chunk_id SET NOT NULL, + ADD CONSTRAINT statement_qualifiers_chunk_id_fkey + FOREIGN KEY (chunk_id) REFERENCES public.chunks(id) ON DELETE CASCADE, + ADD PRIMARY KEY (fact_id, role, chunk_id); + +-- 删块时级联要按块找 +CREATE INDEX statement_qualifiers_chunk_fk_idx ON public.statement_qualifiers (chunk_id); + +-- 实体值与块都必须与所属事实同库 +CREATE OR REPLACE 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 + 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.chunks WHERE id = NEW.chunk_id; + IF ref_kb IS NOT NULL AND ref_kb <> fact_kb THEN + RAISE EXCEPTION 'statement qualifier chunk % cannot live outside the fact''s knowledge base', + NEW.chunk_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 'statement qualifier entity % cannot live outside the fact''s knowledge base', + NEW.entity_id; + END IF; + END IF; + RETURN NEW; +END; +$$; + +CREATE TRIGGER statement_qualifiers_same_kb + BEFORE INSERT OR UPDATE OF fact_id, entity_id, chunk_id ON public.statement_qualifiers + FOR EACH ROW EXECUTE FUNCTION public.squalifier_stays_inside_its_facts_kb(); From ace51a9d4b9f714d9679fc364311e1473e8d3a2c Mon Sep 17 00:00:00 2001 From: Bogdan Szabo Date: Sun, 25 Oct 2026 16:14:52 +0100 Subject: [PATCH 2/2] A role word speaks while its chunk is live, so a re-push that changes a value reads back only the new one Signed-off-by: Bogdan Szabo --- crates/utopia-core/src/models.rs | 2 + crates/utopia-server/src/api/mcp_tests.rs | 3 +- .../src/api/sources_statements_tests.rs | 36 ++ crates/utopia-store/src/graph.rs | 39 ++- crates/utopia-store/src/materialize.rs | 21 +- crates/utopia-store/src/record_axis.rs | 55 +++ .../a_bound_statement_becomes_a_typed_fact.rs | 12 +- ...ole_word_moves_onto_its_evidence_chunks.rs | 5 +- ...pen_statement_keeps_the_documents_words.rs | 323 +++++++++++++++++- ...ole-word-speaks-while-its-chunk-is-live.md | 2 +- docs/design/ledger.md | 2 +- web/src/api.ts | 2 + web/src/docs/mcp.md | 4 +- 13 files changed, 473 insertions(+), 33 deletions(-) diff --git a/crates/utopia-core/src/models.rs b/crates/utopia-core/src/models.rs index a5036d33d..fc2bf6834 100644 --- a/crates/utopia-core/src/models.rs +++ b/crates/utopia-core/src/models.rs @@ -676,6 +676,8 @@ pub struct StatementQualifier { pub value: Option, pub entity_id: Option, pub entity_name: Option, + /// 说这个值的那几块（0070）：两块活着的块说法不同，各自一条 + pub chunk_ids: Vec, } /// 一次 OWL 导入的记录。原文按内容寻址存在 blob 里，这行只是账。 diff --git a/crates/utopia-server/src/api/mcp_tests.rs b/crates/utopia-server/src/api/mcp_tests.rs index d92916da3..08ef1921b 100644 --- a/crates/utopia-server/src/api/mcp_tests.rs +++ b/crates/utopia-server/src/api/mcp_tests.rs @@ -1028,7 +1028,8 @@ async fn entity_facts_reads_back_a_statements_role_words() -> anyhow::Result<()> .unwrap(); assert_eq!( fact["statement_qualifiers"], - json!([{"role":"lines","value":{"value":"10-42"},"entity_id":null,"entity_name":null}]) + json!([{"role":"lines","value":{"value":"10-42"},"entity_id":null,"entity_name":null, + "chunk_ids":[f.chunk]}]) ); f.clean().await } diff --git a/crates/utopia-server/src/api/sources_statements_tests.rs b/crates/utopia-server/src/api/sources_statements_tests.rs index ce1085d7f..44daaf5f7 100644 --- a/crates/utopia-server/src/api/sources_statements_tests.rs +++ b/crates/utopia-server/src/api/sources_statements_tests.rs @@ -437,6 +437,42 @@ async fn a_second_push_under_the_same_identity_updates_in_place() -> anyhow::Res f.cleanup().await } +/// 0070 的例子：再推一次把 `lines` 从 40-52 改成 58-70，读回来只有新的范围 +#[tokio::test] +async fn a_re_push_that_moves_lines_to_58_70_reads_back_only_58_70() -> anyhow::Result<()> { + let Some(f) = Fixture::new().await? else { + return Ok(()); + }; + for lines in ["40-52", "58-70"] { + let mut body = observation("08:14:03", "kitchen table"); + body["s"][0][5] = json!({ "lines": lines }); + let (status, body) = f.push(f.source, &f.token, &body).await?; + assert_eq!(status, StatusCode::OK, "{body}"); + let doc = documents::find_by_external_key(&f.pool, f.source, "statements:obs-000412") + .await? + .expect("the push has a document"); + crate::pipeline::process_document(&f.state, doc.id).await?; + let proposer = Proposer { + user_id: None, + token_id: None, + }; + crate::extraction::extract_document(&f.state, doc.id, proposer).await?; + } + let (cup,): (Uuid,) = + sqlx::query_as("SELECT id FROM entities WHERE kb_id = $1 AND canonical_name = 'cup-7'") + .bind(f.kb) + .fetch_one(&f.pool) + .await?; + let (_, facts) = utopia_store::graph::entity_detail(&f.pool, f.kb, cup, None, None).await?; + let said: Vec<_> = facts + .iter() + .flat_map(|x| &x.statement_qualifiers) + .map(|q| (q.role.as_str(), q.value.clone())) + .collect(); + assert_eq!(said, vec![("lines", Some(json!("58-70")))]); + f.cleanup().await +} + /// 门口的拒绝走到 HTTP 是 422（`AppError::Validation`，与 `api` 推送被拒时同一个码）， /// 并且这次推送留在 run 历史里；`api` 来源不认这条路由（404），钥匙不对是 401 #[tokio::test] diff --git a/crates/utopia-store/src/graph.rs b/crates/utopia-store/src/graph.rs index 02632074f..825205cc5 100644 --- a/crates/utopia-store/src/graph.rs +++ b/crates/utopia-store/src/graph.rs @@ -237,10 +237,11 @@ pub async fn add_statement_qualifier( } /// 一批开放陈述各自的角色词属性，按事实 id 取回；实体那一格连上显示名。 -/// 同一个值记在几块上只返回一次 +/// 只取 `as_of` 时刻说话的行（0070 决定 2、3）；同一个值记在几块上返回一次，带上那几块 pub async fn statement_qualifiers_for( pool: &PgPool, fact_ids: &[Uuid], + as_of: Option>, ) -> AppResult>> { let mut out: HashMap> = HashMap::new(); if fact_ids.is_empty() { @@ -253,23 +254,37 @@ pub async fn statement_qualifiers_for( value: Option, entity_id: Option, entity_name: Option, + chunk_ids: Vec, } - let rows: Vec = sqlx::query_as( - "SELECT DISTINCT q.fact_id, q.role, q.value, q.entity_id, e.canonical_name AS entity_name + let as_of_param = as_of.map(|_| 2); + let sql = format!( + "WITH rw_f AS (SELECT rw_ee.fact_id, {ends} FROM {evidence} + WHERE rw_ee.fact_id = ANY($1) GROUP BY rw_ee.fact_id) + SELECT q.fact_id, q.role, q.value, q.entity_id, e.canonical_name AS entity_name, + array_agg(q.chunk_id ORDER BY q.chunk_id) AS chunk_ids FROM statement_qualifiers q + JOIN rw_f ON rw_f.fact_id = q.fact_id + JOIN chunks rw_c ON rw_c.id = q.chunk_id + JOIN documents rw_d ON rw_d.id = rw_c.document_id LEFT JOIN entities e ON e.id = q.entity_id - WHERE q.fact_id = ANY($1) - ORDER BY q.fact_id, q.role", - ) - .bind(fact_ids) - .fetch_all(pool) - .await?; - for r in rows { + WHERE q.fact_id = ANY($1) AND {speaks} + GROUP BY q.fact_id, q.role, q.value, q.entity_id, e.canonical_name + ORDER BY q.fact_id, q.role, chunk_ids", + ends = crate::record_axis::role_word_fact_ends(as_of_param), + evidence = crate::record_axis::ROLE_WORD_EVIDENCE, + speaks = crate::record_axis::role_word_speaks_given("rw_c", "rw_d", "rw_f", as_of_param), + ); + let mut query = sqlx::query_as::<_, Row>(&sql).bind(fact_ids); + if let Some(at) = as_of { + query = query.bind(at); + } + for r in query.fetch_all(pool).await? { out.entry(r.fact_id).or_default().push(StatementQualifier { role: r.role, value: r.value, entity_id: r.entity_id, entity_name: r.entity_name, + chunk_ids: r.chunk_ids, }); } Ok(out) @@ -1519,7 +1534,7 @@ async fn edges_among( let ids: Vec = edges.iter().map(|x| x.id).collect(); let (mut typed, mut worded) = tokio::try_join!( fact_qualifiers_for(pool, &ids), - statement_qualifiers_for(pool, &ids) + statement_qualifiers_for(pool, &ids, as_of) )?; for x in edges.iter_mut() { x.qualifiers = typed.remove(&x.id).unwrap_or_default(); @@ -1868,7 +1883,7 @@ pub async fn entity_detail( let ids: Vec = facts.iter().map(|x| x.id).collect(); let (mut typed, mut worded) = tokio::try_join!( fact_qualifiers_for(pool, &ids), - statement_qualifiers_for(pool, &ids) + statement_qualifiers_for(pool, &ids, as_of) )?; for x in facts.iter_mut() { x.qualifiers = typed.remove(&x.id).unwrap_or_default(); diff --git a/crates/utopia-store/src/materialize.rs b/crates/utopia-store/src/materialize.rs index 7937b3fa3..f0ede80c7 100644 --- a/crates/utopia-store/src/materialize.rs +++ b/crates/utopia-store/src/materialize.rs @@ -208,8 +208,8 @@ async fn materialize_in_tx( AND b.relation_type_id = t.predicate_id AND ((b.direction = 'forward' AND t.subject_id = s.subject_id) OR (b.direction = 'reverse' AND t.subject_id = s.object_id)) - AND NOT EXISTS (SELECT 1 FROM statement_qualifiers q - WHERE q.fact_id = s.id AND q.role = 'mood'))" + AND {NO_MOOD})", + NO_MOOD = no_mood_spoken(), )) .persistent(false) .bind(kb_id) @@ -274,8 +274,7 @@ async fn materialize_in_tx( WHERE s.kb_id = $1 AND s.layer = 'open' AND s.invalidated_at IS NULL AND b.status = 'bound' AND (b.direction = 'forward' OR s.object_id IS NOT NULL) - AND NOT EXISTS (SELECT 1 FROM statement_qualifiers q - WHERE q.fact_id = s.id AND q.role = 'mood') + AND {NO_MOOD} -- 勘误撤过的（陈述, 属性）不再算（0044 决定 7）：撤销要站得住 AND NOT EXISTS (SELECT 1 FROM errata_actions ea WHERE ea.statement_id = s.id AND ea.predicate_id = b.relation_type_id @@ -283,7 +282,8 @@ async fn materialize_in_tx( AND NOT EXISTS (SELECT 1 FROM typed_fact_sources src JOIN facts t ON t.id = src.fact_id WHERE src.statement_id = s.id AND t.invalidated_at IS NULL AND t.predicate_id = b.relation_type_id) - ORDER BY s.id" + ORDER BY s.id", + NO_MOOD = no_mood_spoken(), )) .persistent(false) .bind(kb_id) @@ -556,7 +556,7 @@ async fn imply_in_tx( AND pr.reading = r.reading AND pr.phrase = {TEXT} WHERE r.kb_id = $1 AND r.status = 'approved' AND r.trigger = 'phrase' AND {RULE_MATCH} - AND NOT EXISTS (SELECT 1 FROM statement_qualifiers q WHERE q.fact_id = s.id AND q.role = 'mood') + AND {NO_MOOD} AND (r.reading IS NULL OR pr.entity_id IS NOT NULL OR pr.value IS NOT NULL) AND NOT EXISTS (SELECT 1 FROM errata_actions ea WHERE ea.statement_id = s.id AND ea.predicate_id = r.conclude_property_id @@ -566,6 +566,7 @@ async fn imply_in_tx( ORDER BY s.id", TEXT = crate::implication_rules::object_text_sql(), RULE_MATCH = crate::implication_rules::RULE_MATCH, + NO_MOOD = no_mood_spoken(), )) .persistent(false) .bind(kb_id) @@ -829,6 +830,14 @@ fn attribute_value(raw: &Value, datatype: Option<&str>, unit: Option<&str>) -> O Some(out) } +/// 陈述 `s` 没有一个说着话的 `mood`（0070 决定 2）：重推丢掉了 mood，旧块上的 mood 就不再拦着 +fn no_mood_spoken() -> String { + format!( + "NOT EXISTS (SELECT 1 FROM statement_qualifiers q WHERE q.fact_id = s.id AND q.role = 'mood' AND {})", + crate::record_axis::role_word_speaks_at("q", None) + ) +} + /// 类别词的归一（同 type_bindings）：这里要在 SQL 里对上实体的 specific_type fn kind_word_sql(col: &str) -> String { format!("lower(btrim(regexp_replace({col}, '\\s+', ' ', 'g')))") diff --git a/crates/utopia-store/src/record_axis.rs b/crates/utopia-store/src/record_axis.rs index 01387f075..acfaad570 100644 --- a/crates/utopia-store/src/record_axis.rs +++ b/crates/utopia-store/src/record_axis.rs @@ -102,6 +102,61 @@ pub fn chunk_live_at(alias: &str, as_of: Option) -> String { } } +/// 一条事实的证据，选成 `rw_ee`（证据）、`rw_ec`（块）、`rw_ed`（文档），供 [`role_word_fact_ends`] 聚合 +pub const ROLE_WORD_EVIDENCE: &str = "fact_evidence rw_ee + JOIN chunks rw_ec ON rw_ec.id = rw_ee.chunk_id + JOIN documents rw_ed ON rw_ed.id = rw_ec.document_id"; + +/// 按事实聚合 [`ROLE_WORD_EVIDENCE`]：T 时刻有没有活块（`any_live`），T 时刻已有的块里 +/// 最晚的终点（`last_end`）。块的终点是 `superseded_at` 与文档 `deleted_at` 里早的那个 +pub fn role_word_fact_ends(as_of: Option) -> String { + format!( + "bool_or({} AND {}) AS any_live, + max(least(rw_ec.superseded_at, rw_ed.deleted_at)) FILTER (WHERE {}) AS last_end", + chunk_live_at("rw_ec", as_of), + document_live_at("rw_ed", as_of), + chunk_existed_at("rw_ec", as_of), + ) +} + +fn chunk_existed_at(alias: &str, as_of: Option) -> String { + match as_of { + None => "true".into(), + Some(param) => format!("{alias}.created_at <= coalesce(${param}, now())"), + } +} + +/// 角色词在 T 时刻替它的事实说话吗（0070 决定 2、3），给定它那块 `c`、那块的文档 `d` +/// 和它事实的 [`role_word_fact_ends`] 一行 `f`。 +/// +/// 它那块与那块的文档在 T 时刻都还活着，就说话；活着的几块值不同，都说话。 +/// 事实在 T 时刻一块活证据都没有时，最后到头的那几块说话（同一时刻到头的都算）， +/// 只看 T 时刻已有的块。每条事实的证据只聚合一次：重推几百次的陈述不按行重扫 +pub fn role_word_speaks_given(c: &str, d: &str, f: &str, as_of: Option) -> String { + format!( + "(({c_live} AND {d_live}) + OR (NOT {f}.any_live AND {existed} + AND least({c}.superseded_at, {d}.deleted_at) = {f}.last_end))", + c_live = chunk_live_at(c, as_of), + d_live = document_live_at(d, as_of), + existed = chunk_existed_at(c, as_of), + ) +} + +/// `statement_qualifiers`：这一行 `q` 在 T 时刻说话吗，[`role_word_speaks_given`] 的单行形状。 +/// 别名都带 `rw_` 前缀，不和调用方的撞 +pub fn role_word_speaks_at(q: &str, as_of: Option) -> String { + format!( + "EXISTS (SELECT 1 FROM chunks rw_c JOIN documents rw_d ON rw_d.id = rw_c.document_id + CROSS JOIN LATERAL (SELECT {ends} FROM {evidence} + WHERE rw_ee.fact_id = {q}.fact_id) rw_f + WHERE rw_c.id = {q}.chunk_id AND {speaks})", + ends = role_word_fact_ends(as_of), + evidence = ROLE_WORD_EVIDENCE, + speaks = role_word_speaks_given("rw_c", "rw_d", "rw_f", as_of), + ) +} + /// `entity_merges`：这次合并在 T 时刻**生效着**吗（0019 第二刀 / #336）。 /// /// 实体身上没有记录轴——`merged_into` 只说合并发生过，不说何时。时刻在这张表上， diff --git a/crates/utopia-store/tests/store/a_bound_statement_becomes_a_typed_fact.rs b/crates/utopia-store/tests/store/a_bound_statement_becomes_a_typed_fact.rs index 2f20afc8e..bc01f4d1f 100644 --- a/crates/utopia-store/tests/store/a_bound_statement_becomes_a_typed_fact.rs +++ b/crates/utopia-store/tests/store/a_bound_statement_becomes_a_typed_fact.rs @@ -5,7 +5,8 @@ //! place) 绑到 headquartered_in：算出一条类型化行，谓词是它，证据与限定抄过来， //! `from_statement_id` 指回陈述；再跑一遍什么都不动；带 mood 的陈述不算；另一份文档 //! 的同一句并进同一行（两条来源、两条证据）；作废其中一条陈述，行还在、来源少一条； -//! 绑定翻成 reverse，旧行作废、新行主宾对调；绑定改成 none，行作废、不再补。 +//! 绑定翻成 reverse，旧行作废、新行主宾对调；绑定改成 none，行作废、不再补； +//! 带 mood 的陈述重推、新块不再说 mood，就成一行。 //! //! 没有 `UTOPIA_DATABASE_URL` 时跳过而不是失败。自建自拆，绝不碰已有的库。 @@ -265,6 +266,15 @@ async fn a_bound_statement_becomes_a_typed_fact() -> anyhow::Result<()> { .await?; assert_eq!(materialize(&pool, kb).await?, Outcome { retired: 1, added: 0, merged: 0, implied: 0, corrected: 0, conflicts: 0 }); assert_eq!(utopia_store::materialize::count(&pool, kb).await?, 0); + + // 带 mood 的那条重推、新块不再说 mood：旧块上的 mood 不再拦着，它成一行（0070 决定 2） + sqlx::query("UPDATE chunks SET superseded_at = now() WHERE id = $1") + .bind(planned_chunk) + .execute(&pool) + .await?; + evidence_of(planned, "bakery-plan-2.txt").await?; + assert_eq!(materialize(&pool, kb).await?.added, 1); + assert_eq!(utopia_store::materialize::count(&pool, kb).await?, 1); anyhow::Ok(()) } .await; diff --git a/crates/utopia-store/tests/store/a_role_word_moves_onto_its_evidence_chunks.rs b/crates/utopia-store/tests/store/a_role_word_moves_onto_its_evidence_chunks.rs index ce8188f40..3360cd3a8 100644 --- a/crates/utopia-store/tests/store/a_role_word_moves_onto_its_evidence_chunks.rs +++ b/crates/utopia-store/tests/store/a_role_word_moves_onto_its_evidence_chunks.rs @@ -208,10 +208,11 @@ async fn role_words_move_onto_every_evidence_chunk_and_leave_a_purged_fact() -> expected, "每块证据一行，没有证据的已失效事实一行不留" ); - let read = utopia_store::graph::statement_qualifiers_for(&pool, &[stated]).await?; + let read = utopia_store::graph::statement_qualifiers_for(&pool, &[stated], None).await?; assert_eq!( serde_json::to_value(&read[&stated])?, - serde_json::json!([{ "role": "lines", "value": "40-52", "entity_id": null, "entity_name": null }]) + serde_json::json!([{ "role": "lines", "value": "40-52", "entity_id": null, + "entity_name": null, "chunk_ids": [ledger.live_chunk] }]) ); pool.close().await; drop_scratch(&name).await; diff --git a/crates/utopia-store/tests/store/an_open_statement_keeps_the_documents_words.rs b/crates/utopia-store/tests/store/an_open_statement_keeps_the_documents_words.rs index 46983265b..34b292cf5 100644 --- a/crates/utopia-store/tests/store/an_open_statement_keeps_the_documents_words.rs +++ b/crates/utopia-store/tests/store/an_open_statement_keeps_the_documents_words.rs @@ -4,6 +4,7 @@ //! 空谓词的读路径（图、实体面板、邻域、路径、导出）要按短语显示它；同一句话再听到 //! 一次是同一行、证据累积；它的短语永远不出现在等着被采纳的说法里；角色词属性与 //! 时间词各自取得回来；证据带着引文在块里的偏移；库拒绝没有短语的开放行。 +//! 角色词只在它那块活着时说话，没有活块时最后被取代的那块说话（0070）。 //! //! 没有 `UTOPIA_DATABASE_URL` 时跳过而不是失败。自建自拆，绝不碰已有的库。 @@ -24,7 +25,7 @@ struct Fixture { chunk_b: Uuid, } -async fn seed(pool: &PgPool) -> anyhow::Result { +async fn seed(pool: &PgPool, org_name: &str) -> anyhow::Result { let (org, ws, kb) = (Uuid::now_v7(), Uuid::now_v7(), Uuid::now_v7()); let etype = Uuid::now_v7(); let (acme, beta) = (Uuid::now_v7(), Uuid::now_v7()); @@ -36,19 +37,19 @@ async fn seed(pool: &PgPool) -> anyhow::Result { ); sqlx::query("INSERT INTO organizations (id, name) VALUES ($1, $2)") .bind(org) - .bind(ORG) + .bind(org_name) .execute(pool) .await?; sqlx::query("INSERT INTO workspaces (id, org_id, name) VALUES ($1, $2, $3)") .bind(ws) .bind(org) - .bind(ORG) + .bind(org_name) .execute(pool) .await?; sqlx::query("INSERT INTO knowledge_bases (id, workspace_id, name) VALUES ($1, $2, $3)") .bind(kb) .bind(ws) - .bind(ORG) + .bind(org_name) .execute(pool) .await?; sqlx::query("INSERT INTO entity_types (id, kb_id, key, label) VALUES ($1, $2, 'org', '组织')") @@ -119,7 +120,7 @@ async fn an_open_statement_shows_under_its_phrase_and_reuses_its_row() -> anyhow .bind(ORG) .execute(&pool) .await?; - let f = seed(&pool).await?; + let f = seed(&pool, ORG).await?; let run = async { // 1. 落一条开放陈述：Acme —acquired→ Beta，短语照写、谓词为空、没有 valid_* @@ -199,7 +200,8 @@ async fn an_open_statement_shows_under_its_phrase_and_reuses_its_row() -> anyhow let amount = |q: &[utopia_core::models::StatementQualifier]| { serde_json::to_value(q).unwrap() == serde_json::json!([{ "role": "amount", - "value": { "value": "$1.2 billion" }, "entity_id": null, "entity_name": null }]) + "value": { "value": "$1.2 billion" }, "entity_id": null, "entity_name": null, + "chunk_ids": [f.chunk_a] }]) }; assert!( amount(&e.statement_qualifiers), @@ -310,7 +312,7 @@ async fn an_open_statement_shows_under_its_phrase_and_reuses_its_row() -> anyhow assert!(created && other != fact); // 4. 角色词属性与时间词各自取得回来 - let qualifiers = graph::statement_qualifiers_for(&pool, &[fact]).await?; + let qualifiers = graph::statement_qualifiers_for(&pool, &[fact], None).await?; let q = &qualifiers[&fact]; assert_eq!(q.len(), 1); assert_eq!(q[0].role, "amount"); @@ -352,7 +354,7 @@ async fn an_open_statement_shows_under_its_phrase_and_reuses_its_row() -> anyhow .is_err() ); graph::add_statement_qualifier(&pool, fact, f.chunk_a, "buyer", None, Some(f.acme)).await?; - let q = &graph::statement_qualifiers_for(&pool, &[fact]).await?[&fact]; + let q = &graph::statement_qualifiers_for(&pool, &[fact], None).await?[&fact]; let buyer = q.iter().find(|x| x.role == "buyer").unwrap(); assert_eq!(buyer.entity_id, Some(f.acme)); assert_eq!(buyer.entity_name.as_deref(), Some("Acme")); @@ -417,3 +419,308 @@ async fn an_open_statement_shows_under_its_phrase_and_reuses_its_row() -> anyhow .await?; run } + +/// 一条陈述，两块都是它的证据；每块记下自己说的角色词 +async fn stated(pool: &PgPool, f: &Fixture, words: &[(Uuid, &str, &str)]) -> anyhow::Result { + let (fact, _) = graph::insert_open_statement( + pool, + f.kb, + f.acme, + "covers", + FactObject::Entity(f.beta), + None, + 0.9, + ) + .await?; + for chunk in [f.chunk_a, f.chunk_b] { + graph::add_evidence(pool, fact, chunk, None, None).await?; + } + for (chunk, role, value) in words { + graph::add_statement_qualifier( + pool, + fact, + *chunk, + role, + Some(&serde_json::json!(value)), + None, + ) + .await?; + } + Ok(fact) +} + +async fn supersede(pool: &PgPool, chunk: Uuid, at: &str) -> anyhow::Result<()> { + sqlx::query("UPDATE chunks SET superseded_at = $2::timestamptz WHERE id = $1") + .bind(chunk) + .bind(at) + .execute(pool) + .await?; + Ok(()) +} + +/// 现在说话的角色词：(角色, 值, 说它的块) +async fn speaking( + pool: &PgPool, + fact: Uuid, +) -> anyhow::Result)>> { + let read = graph::statement_qualifiers_for(pool, &[fact], None).await?; + Ok(read + .get(&fact) + .into_iter() + .flatten() + .map(|q| { + ( + q.role.clone(), + q.value.clone().unwrap_or_default(), + q.chunk_ids.clone(), + ) + }) + .collect()) +} + +/// 每个测试自己的组织：建库、跑完、拆掉 +async fn with_fixture(org_name: &str, run: F) -> anyhow::Result<()> +where + F: FnOnce(PgPool, Fixture) -> Fut, + Fut: std::future::Future>, +{ + let Some(url) = utopia_store::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + let clean = |pool: PgPool| async move { + sqlx::query("DELETE FROM organizations WHERE name = $1") + .bind(org_name) + .execute(&pool) + .await + }; + clean(pool.clone()).await?; + let f = seed(&pool, org_name).await?; + let outcome = run(pool.clone(), f).await; + clean(pool).await?; + outcome +} + +fn lines(value: &str, chunk: Uuid) -> (String, serde_json::Value, Vec) { + ("lines".into(), serde_json::json!(value), vec![chunk]) +} + +#[tokio::test] +async fn a_superseded_chunk_stops_speaking_once_a_live_chunk_says_58_70() -> anyhow::Result<()> { + with_fixture("role-word-superseded-test", |pool, f| async move { + let fact = stated( + &pool, + &f, + &[(f.chunk_a, "lines", "40-52"), (f.chunk_b, "lines", "58-70")], + ) + .await?; + supersede(&pool, f.chunk_a, "2026-03-01").await?; + assert_eq!( + speaking(&pool, fact).await?, + vec![lines("58-70", f.chunk_b)] + ); + Ok(()) + }) + .await +} + +#[tokio::test] +async fn two_live_chunks_that_disagree_both_speak_with_their_chunk() -> anyhow::Result<()> { + with_fixture("role-word-disagree-test", |pool, f| async move { + let fact = stated( + &pool, + &f, + &[(f.chunk_a, "lines", "40-52"), (f.chunk_b, "lines", "58-70")], + ) + .await?; + assert_eq!( + speaking(&pool, fact).await?, + vec![lines("40-52", f.chunk_a), lines("58-70", f.chunk_b)] + ); + Ok(()) + }) + .await +} + +#[tokio::test] +async fn a_status_the_live_chunk_does_not_repeat_is_gone() -> anyhow::Result<()> { + with_fixture("role-word-dropped-test", |pool, f| async move { + let fact = stated( + &pool, + &f, + &[ + (f.chunk_a, "lines", "40-52"), + (f.chunk_a, "status", "failing"), + (f.chunk_b, "lines", "58-70"), + ], + ) + .await?; + supersede(&pool, f.chunk_a, "2026-03-01").await?; + assert_eq!( + speaking(&pool, fact).await?, + vec![lines("58-70", f.chunk_b)] + ); + Ok(()) + }) + .await +} + +#[tokio::test] +async fn a_stale_statement_shows_58_70_from_its_last_superseded_chunk() -> anyhow::Result<()> { + with_fixture("role-word-stale-test", |pool, f| async move { + let fact = stated( + &pool, + &f, + &[(f.chunk_a, "lines", "40-52"), (f.chunk_b, "lines", "58-70")], + ) + .await?; + supersede(&pool, f.chunk_a, "2026-03-01").await?; + supersede(&pool, f.chunk_b, "2026-04-01").await?; + assert_eq!( + speaking(&pool, fact).await?, + vec![lines("58-70", f.chunk_b)] + ); + Ok(()) + }) + .await +} + +#[tokio::test] +async fn a_deleted_documents_chunk_is_dated_by_its_deleted_at() -> anyhow::Result<()> { + with_fixture("role-word-deleted-test", |pool, f| async move { + let fact = stated( + &pool, + &f, + &[(f.chunk_a, "lines", "40-52"), (f.chunk_b, "lines", "58-70")], + ) + .await?; + supersede(&pool, f.chunk_a, "2026-03-01").await?; + sqlx::query("UPDATE documents SET deleted_at = now() WHERE kb_id = $1") + .bind(f.kb) + .execute(&pool) + .await?; + assert_eq!( + speaking(&pool, fact).await?, + vec![lines("58-70", f.chunk_b)] + ); + Ok(()) + }) + .await +} + +#[tokio::test] +async fn a_restored_document_lets_its_live_chunk_speak_58_70_alone_again() -> anyhow::Result<()> { + with_fixture("role-word-restored-test", |pool, f| async move { + let fact = stated( + &pool, + &f, + &[(f.chunk_a, "lines", "40-52"), (f.chunk_b, "lines", "58-70")], + ) + .await?; + supersede(&pool, f.chunk_a, "2026-03-01").await?; + let deleted_at = |at: Option<&'static str>| { + sqlx::query("UPDATE documents SET deleted_at = $2::timestamptz WHERE kb_id = $1") + .bind(f.kb) + .bind(at) + .execute(&pool) + }; + // 删在取代之前：两块同一刻到头，都说话 + deleted_at(Some("2026-02-01")).await?; + assert_eq!( + speaking(&pool, fact).await?, + vec![lines("40-52", f.chunk_a), lines("58-70", f.chunk_b)] + ); + deleted_at(None).await?; + assert_eq!( + speaking(&pool, fact).await?, + vec![lines("58-70", f.chunk_b)] + ); + Ok(()) + }) + .await +} + +#[tokio::test] +async fn a_stale_read_as_of_february_shows_40_52_before_the_latest_chunk_existed( +) -> anyhow::Result<()> { + with_fixture("role-word-stale-as-of-test", |pool, f| async move { + let fact = stated( + &pool, + &f, + &[(f.chunk_a, "lines", "40-52"), (f.chunk_b, "lines", "58-70")], + ) + .await?; + sqlx::raw_sql(&format!( + "UPDATE documents SET created_at = '2026-01-01' WHERE kb_id = '{kb}'; + UPDATE chunks SET created_at = '2026-01-01', superseded_at = '2026-02-01' + WHERE id = '{a}'; + UPDATE chunks SET created_at = '2026-03-01', superseded_at = '2026-04-01' + WHERE id = '{b}';", + kb = f.kb, + a = f.chunk_a, + b = f.chunk_b, + )) + .execute(&pool) + .await?; + let as_of = "2026-02-15T00:00:00Z".parse::>()?; + let read = graph::statement_qualifiers_for(&pool, &[fact], Some(as_of)).await?; + assert_eq!( + read[&fact] + .iter() + .map(|q| q.value.clone()) + .collect::>(), + vec![Some(serde_json::json!("40-52"))] + ); + Ok(()) + }) + .await +} + +#[tokio::test] +async fn a_read_as_of_february_shows_40_52_said_before_the_march_push() -> anyhow::Result<()> { + with_fixture("role-word-as-of-test", |pool, f| async move { + let fact = stated( + &pool, + &f, + &[(f.chunk_a, "lines", "40-52"), (f.chunk_b, "lines", "58-70")], + ) + .await?; + sqlx::raw_sql(&format!( + "UPDATE entities SET created_at = '2026-01-01' WHERE kb_id = '{kb}'; + UPDATE documents SET created_at = '2026-01-01' WHERE kb_id = '{kb}'; + UPDATE facts SET recorded_at = '2026-01-01' WHERE id = '{fact}'; + UPDATE chunks SET created_at = '2026-01-01', superseded_at = '2026-03-01' + WHERE id = '{a}'; + UPDATE chunks SET created_at = '2026-03-01' WHERE id = '{b}';", + kb = f.kb, + a = f.chunk_a, + b = f.chunk_b, + )) + .execute(&pool) + .await?; + let read_at = |as_of: Option<&str>| { + let pool = pool.clone(); + let as_of = as_of.map(|t| t.parse::>().unwrap()); + async move { + let (_, facts) = graph::entity_detail(&pool, f.kb, f.acme, None, as_of).await?; + let row = facts + .into_iter() + .find(|x| x.id == fact) + .expect("the statement"); + anyhow::Ok( + row.statement_qualifiers + .into_iter() + .map(|q| q.value) + .collect::>(), + ) + } + }; + assert_eq!( + read_at(Some("2026-02-01T00:00:00Z")).await?, + vec![Some(serde_json::json!("40-52"))] + ); + assert_eq!(read_at(None).await?, vec![Some(serde_json::json!("58-70"))]); + Ok(()) + }) + .await +} diff --git a/docs/decisions/0070-a-role-word-speaks-while-its-chunk-is-live.md b/docs/decisions/0070-a-role-word-speaks-while-its-chunk-is-live.md index 5d50f2617..6f122fb85 100644 --- a/docs/decisions/0070-a-role-word-speaks-while-its-chunk-is-live.md +++ b/docs/decisions/0070-a-role-word-speaks-while-its-chunk-is-live.md @@ -1,6 +1,6 @@ # 0070 · A role word speaks while its chunk is live -- **Status**: In progress · 2026-10-25 · [#1267](https://github.com/deeplethe/utopia/issues/1267) · decision 1 built (migration 0121) · open: decisions 2 to 4 +- **Status**: In progress · 2026-10-25 · [#1267](https://github.com/deeplethe/utopia/issues/1267) · decision 1 built (migration 0121), decisions 2 and 3 built (`record_axis::role_word_speaks_at`, the three `mood` checks included) · open: decision 4 - **Written**: 2026-10-16 (conventions in the [README](README.md)) - **Related**: [0037](0037-a-relation-carries-its-own-attributes.md) decided that a qualifier is not part of a fact's identity and that the first value stays; [0044](0044-the-ontology-is-a-view-over-what-documents-say.md) owns the open statement and the role words on it; [0054](0054-a-source-may-push-statements-in-the-open-contract.md) decision 7 is the re-push that supersedes a chunk; migration `0061` created `statement_qualifiers`; [0048](0048-provenance-references-stay-inside-the-knowledge-base.md) gets one more edge to keep inside the base; [0019](0019-the-second-clock-can-be-rewound.md) is the record axis `as_of` reads on; [#1267](https://github.com/deeplethe/utopia/issues/1267), [#1272](https://github.com/deeplethe/utopia/issues/1272) diff --git a/docs/design/ledger.md b/docs/design/ledger.md index 34b908222..bfa9d5d71 100644 --- a/docs/design/ledger.md +++ b/docs/design/ledger.md @@ -36,7 +36,7 @@ declared attribute for typed rows, converted by the attribute's datatype, the un text (`$4 billion` becomes 4000000000 with unit `$`) and the declared unit only when the text carries none; a qualifier is not part of the edge's identity; every superseding writer copies qualifiers with the evidence [0037, #731]. Entity detail, graph edges and the MCP `entity_facts` -tool return both: typed ones in `qualifiers`, role words in `statement_qualifiers` [#1267]. +tool return both: typed ones in `qualifiers`, role words in `statement_qualifiers`, each with the live chunks that say it [#1267, 0070]. `time_mentions (fact, chunk, text, char_start)` hold verbatim time words [#731]. Names are value facts on `known_as` [0041]. `entities.description` marks a described, unnamed thing [#731]. diff --git a/web/src/api.ts b/web/src/api.ts index 65c5d15c0..0d94044a5 100644 --- a/web/src/api.ts +++ b/web/src/api.ts @@ -1123,6 +1123,8 @@ export interface StatementQualifier { value: unknown | null; entity_id: string | null; entity_name: string | null; + /** 说这个值的那几块（0070）：两块活着的块说法不同，各自一条 */ + chunk_ids: string[]; } export interface EntityFact { diff --git a/web/src/docs/mcp.md b/web/src/docs/mcp.md index 3f2eded6b..e8958e7e1 100644 --- a/web/src/docs/mcp.md +++ b/web/src/docs/mcp.md @@ -143,7 +143,9 @@ unaccepted predicate name; it does **not** mean the fact is derived. `qualifier_type_id`, `key`, `label`, raw `value`, `entity_id`, and `entity_name`. `statement_qualifiers[]` holds the document's own role words for the statement (such as `amount` or `lines`), which are not declared attributes: each has `role`, -raw `value`, `entity_id`, and `entity_name`. +raw `value`, `entity_id`, `entity_name`, and the `chunk_ids` that say it. Only +chunks live at the read's time speak, so a re-pushed value replaces the old one; +two live chunks that disagree give two entries. A `derived_facts[]` entry contains `id`, `subject_id`, `subject`, `predicate_id`, `predicate`, `object_id`, `object`, raw `object_value`, `rule`, `rule_name`,