diff --git a/crates/utopia-cli/src/main.rs b/crates/utopia-cli/src/main.rs index 252f51859..dbb76c53e 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 = 76; +const CURRENT_SCHEMA_VERSION: u32 = 77; fn main() -> anyhow::Result<()> { dotenvy::dotenv().ok(); diff --git a/crates/utopia-server/src/api/sources_routes.rs b/crates/utopia-server/src/api/sources_routes.rs index 7b0e94670..e0ea55bca 100644 --- a/crates/utopia-server/src/api/sources_routes.rs +++ b/crates/utopia-server/src/api/sources_routes.rs @@ -633,8 +633,8 @@ struct StatementsBody { } /// 门口的校验：形状对不对、有没有契约之外的键、每条陈述的引文格是不是空的。 -/// 通过就把 `{e, s, n}` 按契约重新序列化成文档正文——存的是我们自己写出来的那份， -/// 不是调用方发来的字节，于是文档里没有信封、没有多余空白，块就是契约本身 +/// 通过就按契约重新序列化成文档正文——存的是我们自己写出来的那份，不是调用方发来的 +/// 字节：身份、日期（有的话）和 `{e, s, n}`，键序固定、没有多余空白，块就是契约本身 fn validate_statements_payload(raw: &[u8]) -> Result<(StatementsBody, Option), String> { if raw.len() > STATEMENTS_MAX_BYTES { return Err(format!( @@ -706,8 +706,20 @@ fn validate_statements_payload(raw: &[u8]) -> Result<(StatementsBody, Option>(), + ["e", "external_id", "n", "s"], + "identity plus the three arrays, nothing else" + ); let parsed = utopia_extract::open::parse_open_response(&content).unwrap(); assert_eq!(parsed.statements.len(), 1); assert_eq!(parsed.statements[0].phrase, "is on"); @@ -529,3 +540,40 @@ async fn the_push_token_can_be_viewed_and_rotated_like_an_api_source() -> anyhow assert_eq!(status, StatusCode::OK, "{body}"); f.cleanup().await } + +/// 同一份载荷在新身份下是另一次观测（#900）：两次看到杯子在桌上就是两篇文档，各带自己的 +/// 日期。身份写在正文里，所以两篇正文不同，库里「一份内容一篇文档」的唯一性和文件型 +/// 来源那条「同内容出现在新路径 = 改名」的识别都碰不到它 +#[tokio::test] +async fn the_same_payload_under_a_new_identity_is_a_second_observation() -> anyhow::Result<()> { + let Some(f) = Fixture::new().await? else { + return Ok(()); + }; + let mut first = observation("08:14:03", "kitchen table"); + first["external_id"] = json!("obs-1"); + first["doc_time"] = json!("2026-09-23T08:14:03Z"); + let (status, body) = f.push(f.source, &f.token, &first).await?; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["action"], "created"); + let mut second = first.clone(); + second["external_id"] = json!("obs-2"); + second["doc_time"] = json!("2026-09-23T08:20:00Z"); + let (status, body) = f.push(f.source, &f.token, &second).await?; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["action"], "created", "not a move"); + for (key, when) in [ + ("statements:obs-1", "2026-09-23T08:14:03Z"), + ("statements:obs-2", "2026-09-23T08:20:00Z"), + ] { + let doc = documents::find_by_external_key(&f.pool, f.source, key) + .await? + .unwrap_or_else(|| panic!("{key} is its own document")); + assert_eq!( + doc.doc_time + .map(|t| t.to_rfc3339_opts(chrono::SecondsFormat::Secs, true)), + Some(when.to_string()), + "each observation keeps its own date" + ); + } + f.cleanup().await +} diff --git a/crates/utopia-store/src/documents.rs b/crates/utopia-store/src/documents.rs index f64637fd1..c2567557b 100644 --- a/crates/utopia-store/src/documents.rs +++ b/crates/utopia-store/src/documents.rs @@ -155,8 +155,8 @@ pub async fn create_with_version_and_processing( _ => AppError::Db(e), })?; sqlx::query( - "INSERT INTO document_versions (id, document_id, version, sha256, size_bytes) - VALUES ($1, $2, 1, $3, $4)", + "INSERT INTO document_versions (id, document_id, version, sha256, size_bytes, doc_time) + VALUES ($1, $2, 1, $3, $4, (SELECT doc_time FROM documents WHERE id = $2))", ) .bind(Uuid::now_v7()) .bind(document.id) @@ -210,10 +210,10 @@ pub async fn replace_content_and_enqueue_processing( .execute(&mut *tx) .await?; sqlx::query( - "INSERT INTO document_versions (id, document_id, version, sha256, size_bytes) + "INSERT INTO document_versions (id, document_id, version, sha256, size_bytes, doc_time) VALUES ($1, $2, (SELECT coalesce(max(version), 0) + 1 FROM document_versions WHERE document_id = $2), - $3, $4)", + $3, $4, (SELECT doc_time FROM documents WHERE id = $2))", ) .bind(Uuid::now_v7()) .bind(id) @@ -285,10 +285,10 @@ pub async fn upsert_source_document_tx( .execute(&mut **tx) .await?; sqlx::query( - "INSERT INTO document_versions (id, document_id, version, sha256, size_bytes) + "INSERT INTO document_versions (id, document_id, version, sha256, size_bytes, doc_time) VALUES ($1, $2, (SELECT coalesce(max(version), 0) + 1 FROM document_versions WHERE document_id = $2), - $3, $4)", + $3, $4, (SELECT doc_time FROM documents WHERE id = $2))", ) .bind(Uuid::now_v7()) .bind(document.id) @@ -340,10 +340,10 @@ pub async fn upsert_source_document_tx( .execute(&mut **tx) .await?; sqlx::query( - "INSERT INTO document_versions (id, document_id, version, sha256, size_bytes) + "INSERT INTO document_versions (id, document_id, version, sha256, size_bytes, doc_time) VALUES ($1, $2, (SELECT coalesce(max(version), 0) + 1 FROM document_versions WHERE document_id = $2), - $3, $4)", + $3, $4, (SELECT doc_time FROM documents WHERE id = $2))", ) .bind(Uuid::now_v7()) .bind(document.id) @@ -390,8 +390,8 @@ pub async fn upsert_source_document_tx( _ => AppError::Db(e), })?; sqlx::query( - "INSERT INTO document_versions (id, document_id, version, sha256, size_bytes) - VALUES ($1, $2, 1, $3, $4)", + "INSERT INTO document_versions (id, document_id, version, sha256, size_bytes, doc_time) + VALUES ($1, $2, 1, $3, $4, (SELECT doc_time FROM documents WHERE id = $2))", ) .bind(Uuid::now_v7()) .bind(document.id) @@ -727,10 +727,10 @@ pub async fn record_version( size_bytes: i64, ) -> AppResult<()> { sqlx::query( - "INSERT INTO document_versions (id, document_id, version, sha256, size_bytes) + "INSERT INTO document_versions (id, document_id, version, sha256, size_bytes, doc_time) VALUES ($1, $2, (SELECT coalesce(max(version), 0) + 1 FROM document_versions WHERE document_id = $2), - $3, $4)", + $3, $4, (SELECT doc_time FROM documents WHERE id = $2))", ) .bind(Uuid::now_v7()) .bind(document_id) diff --git a/crates/utopia-store/src/materialize.rs b/crates/utopia-store/src/materialize.rs index 349772b62..8f4b4daae 100644 --- a/crates/utopia-store/src/materialize.rs +++ b/crates/utopia-store/src/materialize.rs @@ -12,6 +12,11 @@ //! 重算是集合运算，跑多少遍结果一样：先删「不再成立」的来源（陈述作废了、签名不再绑着、 //! 绑到了别的属性或反了方向、行本身作废了），再作废来源全空的类型化行，最后给「该有而 //! 没有」的（陈述, 绑定）对补上——有同断言的行就并进去，没有才新建。没有模型调用。 +//! +//! 写下的行随后**对账**（#899）：写路径上抽取和点头写完一条 state 事实都会沿它的唯一性 +//! 方向重算时间线（`temporal::reconcile_new_fact`），物化出来的行是同一种观察，不该 +//! 少这一步——否则一个函数型属性的两个值各开着一段，后一次观察关不上前一次。重算 +//! 提交之后再对账：对账按时间线各自开事务、拿自己的锁，不在物化的事务里做 use sqlx::PgPool; use utopia_core::AppResult; @@ -38,6 +43,10 @@ pub struct Outcome { pub merged: u64, /// 规则算出来的隐含行（0044 决定 3 第五片），新建的 pub implied: u64, + /// 对账自动闭合而改写出来的修正行数（#899） + pub corrected: u64, + /// 对账裁不了、交给人的冲突数 + pub conflicts: u32, } /// 一条该物化的（陈述, 绑定）对，连陈述上要抄的东西。 @@ -71,8 +80,24 @@ pub async fn materialize(pool: &PgPool, kb_id: Uuid) -> AppResult { .bind(kb_id.to_string()) .execute(&mut *tx) .await?; - let outcome = materialize_in_tx(&mut tx, kb_id).await?; + let (outcome, written) = materialize_in_tx(&mut tx, kb_id).await?; tx.commit().await?; + reconcile_written(pool, kb_id, outcome, &written).await +} + +/// 这一轮写下（新建或并入）的类型化行沿各自的唯一性时间线对账（#899）。只有 state 且 +/// 声明了唯一性的谓词有时间线，`timelines_of` 自己筛；其余的行这里是空转 +async fn reconcile_written( + pool: &PgPool, + kb_id: Uuid, + mut outcome: Outcome, + written: &[Uuid], +) -> AppResult { + if !written.is_empty() { + let report = crate::temporal::reconcile_facts(pool, kb_id, written).await?; + outcome.corrected = report.corrected.len() as u64; + outcome.conflicts = report.conflicts; + } Ok(outcome) } @@ -96,15 +121,19 @@ pub async fn try_materialize(pool: &PgPool, kb_id: Uuid) -> AppResult, kb_id: Uuid, -) -> AppResult { +) -> AppResult<(Outcome, Vec)> { + // 这一轮写下的行（新建的和并入的），提交后对账 + let mut written: Vec = Vec::new(); // 1. 删不再成立的来源：陈述死了、行死了、签名没绑着、属性或方向变了、陈述带了 mood sqlx::query(&format!( "DELETE FROM typed_fact_sources src @@ -248,6 +277,7 @@ async fn materialize_in_tx( } _ => continue, }; + written.push(fact); if new { added += 1; sqlx::query("UPDATE facts SET from_statement_id = $2 WHERE id = $1 AND from_statement_id IS NULL") @@ -311,13 +341,18 @@ async fn materialize_in_tx( } // 3b. 已批准的规则算隐含行（0044 决定 3 第五片）。读数只查缓存：缓存里没有的这一轮 // 不算，`read_phrases` 填上之后再来。短语规则按陈述触发，类别词规则按实体触发 - let implied = imply_in_tx(tx, kb_id).await?; - Ok(Outcome { - retired, - added, - merged, - implied, - }) + let implied = imply_in_tx(tx, kb_id, &mut written).await?; + Ok(( + Outcome { + retired, + added, + merged, + implied, + corrected: 0, + conflicts: 0, + }, + written, + )) } #[derive(sqlx::FromRow)] @@ -341,6 +376,7 @@ struct Implied { async fn imply_in_tx( tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, kb_id: Uuid, + written: &mut Vec, ) -> AppResult { // 短语规则：签名下活着的、没 mood 的陈述；宾语是读数的答案（缓存里的实体或值）， // 没有读数时就是陈述的宾语。已经有活着的隐含行以这条陈述为来源的不再算 @@ -425,6 +461,7 @@ async fn imply_in_tx( d.confidence, ) .await?; + written.push(fact); if new { implied += 1; sqlx::query("UPDATE facts SET implied = TRUE WHERE id = $1") diff --git a/crates/utopia-store/src/materialize_delivery_tests.rs b/crates/utopia-store/src/materialize_delivery_tests.rs index a4b7c598c..5e5d6573a 100644 --- a/crates/utopia-store/src/materialize_delivery_tests.rs +++ b/crates/utopia-store/src/materialize_delivery_tests.rs @@ -27,7 +27,7 @@ async fn try_materialize(pool: &PgPool, kb_id: Uuid) -> AppResult, + from: Uuid, + to: Uuid, +) -> AppResult<()> { + sqlx::query( + "INSERT INTO typed_fact_sources (fact_id, statement_id) + SELECT $2, statement_id FROM typed_fact_sources WHERE fact_id = $1 + ON CONFLICT DO NOTHING", + ) + .bind(from) + .bind(to) + .execute(&mut **tx) + .await?; + sqlx::query( + "INSERT INTO implied_fact_sources (fact_id, rule_id, statement_id, entity_id) + SELECT $2, rule_id, statement_id, entity_id FROM implied_fact_sources WHERE fact_id = $1 + ON CONFLICT DO NOTHING", + ) + .bind(from) + .bind(to) + .execute(&mut **tx) + .await?; + Ok(()) +} + /// 作废 + 改写一行的持有者：主语或宾语换成另一个实体，其余照旧（证据、边上的属性随行）。 /// 撤回合并把合并之后才改写出来的行送回源实体时用——原地改主语，记录轴回放合并窗口时 /// 就找不到它当时挂在哪（0027 只认合并账本上的行）。返回新行 id；`None` = 已作废 @@ -1016,10 +1056,12 @@ pub async fn rehome_tx( "INSERT INTO facts (id, kb_id, subject_id, predicate_id, object_id, object_value, valid_from, valid_from_precision, valid_to, valid_to_precision, confidence, derived_by_rule, supersedes, - attested_from, attested_to, end_derived) + attested_from, attested_to, end_derived, + from_statement_id, implied) SELECT $1, kb_id, COALESCE($3, subject_id), predicate_id, COALESCE($4, object_id), object_value, valid_from, valid_from_precision, valid_to, valid_to_precision, - confidence, derived_by_rule, id, attested_from, attested_to, end_derived + confidence, derived_by_rule, id, attested_from, attested_to, end_derived, + from_statement_id, implied FROM facts WHERE id = $2", ) .bind(moved) @@ -1037,6 +1079,7 @@ pub async fn rehome_tx( .await?; copy_evidence(tx, fact_id, moved).await?; copy_qualifiers(tx, fact_id, moved).await?; + copy_materialization_links(tx, fact_id, moved).await?; Ok(Some(moved)) } 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 71c5cd08d..c53a3dd55 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 @@ -179,7 +179,7 @@ async fn a_bound_statement_becomes_a_typed_fact() -> anyhow::Result<()> { // 来源搬过来），第三条并进去（merged 1）——最后一行，三条来源，两条证据；带 mood 的 // 那条不算 let first = materialize(&pool, kb).await?; - assert_eq!(first, Outcome { retired: 0, added: 2, merged: 1, implied: 0 }); + assert_eq!(first, Outcome { retired: 0, added: 2, merged: 1, implied: 0, corrected: 0, conflicts: 0 }); let live = |pool: PgPool| async move { sqlx::query_as::<_, (Uuid, Uuid, Uuid, Uuid, Uuid, Option>, Option)>( "SELECT id, subject_id, object_id, predicate_id, from_statement_id, valid_from, @@ -238,7 +238,7 @@ async fn a_bound_statement_becomes_a_typed_fact() -> anyhow::Result<()> { .execute(&pool) .await?; // 旧行的来源全不成立了：作废 1；反向重算时裸的那条先成行、带时间的再取代它：新建 2 - assert_eq!(materialize(&pool, kb).await?, Outcome { retired: 1, added: 2, merged: 0, implied: 0 }); + assert_eq!(materialize(&pool, kb).await?, Outcome { retired: 1, added: 2, merged: 0, implied: 0, corrected: 0, conflicts: 0 }); let rows = live(pool.clone()).await?; assert_eq!(rows.len(), 1); assert_eq!((rows[0].1, rows[0].2), (port, bakery), "方向反了主宾对调"); @@ -258,7 +258,7 @@ async fn a_bound_statement_becomes_a_typed_fact() -> anyhow::Result<()> { .bind(kb) .execute(&pool) .await?; - assert_eq!(materialize(&pool, kb).await?, Outcome { retired: 1, added: 0, merged: 0, implied: 0 }); + 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); anyhow::Ok(()) } diff --git a/crates/utopia-store/tests/store/a_plan_step_follows_its_premise.rs b/crates/utopia-store/tests/store/a_plan_step_follows_its_premise.rs index e3e61b90b..3ed965b7d 100644 --- a/crates/utopia-store/tests/store/a_plan_step_follows_its_premise.rs +++ b/crates/utopia-store/tests/store/a_plan_step_follows_its_premise.rs @@ -631,7 +631,7 @@ async fn typed_desk(pool: &PgPool, f: &Fixture) -> anyhow::Result> { .await?) } -/// 两份观察各是一份文档（一次观察一个身份）。显式对账之后桌面那一段关在第二份的日期上， +/// 两份观察各是一份文档（一次观察一个身份）。物化随手对账，桌面那一段关在第二份的日期上， /// 步骤跟着退场——`POST /kbs/{id}/ontology/relation-types/{type_id}/reconcile` 就是这一步 #[tokio::test] async fn an_explicit_reconcile_closes_the_earlier_place_and_the_step_leaves() -> anyhow::Result<()> @@ -658,13 +658,15 @@ async fn an_explicit_reconcile_closes_the_earlier_place_and_the_step_leaves() -> statement(&pool, &f, c2, "shelf", T2).await?; bind_is_on(&pool, &f).await?; let typed = materialize::materialize(&pool, f.kb).await?; - assert_eq!(typed.added, 2, "{typed:?}"); + // 物化自己就对账了（#899）：桌面那一段在这里关上。显式对账仍然可用，只是没剩下 + // 要改的 + assert_eq!((typed.added, typed.corrected), (2, 1), "{typed:?}"); let report = temporal::reconcile_predicate(&pool, f.kb, f.location).await?; assert_eq!( (report.corrected.len(), report.conflicts), - (1, 0), - "{report:?}" + (0, 0), + "nothing left for the explicit reconcile: {report:?}" ); let desk = typed_desk(&pool, &f).await?; assert_eq!(desk.len(), 1); @@ -684,3 +686,142 @@ async fn an_explicit_reconcile_closes_the_earlier_place_and_the_step_leaves() -> cleanup(&pool, f.org).await?; run } + +/// #899：两份观察各是一份文档，绑定之后**物化自己**就把桌面那一段关上，不用再显式对账—— +/// 物化出来的行和抽取、点头写下的一样是新观察，写完就沿唯一性时间线重算 +#[tokio::test] +async fn a_later_bound_statement_closes_the_earlier_place() -> anyhow::Result<()> { + let Some(url) = utopia_store::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + let f = seed(&pool, "issue899-materialize-timeline").await?; + + let run = async { + let rule = step_rule( + &pool, + &f, + "S_A pick cup from desk", + f.cup, + f.sa_ready, + &[(f.location, "desk")], + ) + .await?; + let (_, c1) = document(&pool, &f, "obs-1.json", T1, "cup-7 is on desk").await?; + let (_, c2) = document(&pool, &f, "obs-2.json", T2, "cup-7 is on shelf").await?; + statement(&pool, &f, c1, "desk", T1).await?; + statement(&pool, &f, c2, "shelf", T2).await?; + bind_is_on(&pool, &f).await?; + let typed = materialize::materialize(&pool, f.kb).await?; + assert_eq!( + (typed.added, typed.corrected, typed.conflicts), + (2, 1, 0), + "{typed:?}" + ); + let desk = typed_desk(&pool, &f).await?; + assert_eq!(desk.len(), 1); + assert_eq!( + desk[0].valid_to_precision.as_deref(), + Some("unknown"), + "location is functional and a later place was materialized: the desk row is closed \ + without an explicit reconcile: {desk:?}" + ); + // 再跑一遍是空转：没有新行，也不再对账 + let again = materialize::materialize(&pool, f.kb).await?; + assert_eq!( + (again.added, again.merged, again.corrected, again.conflicts), + (0, 0, 0, 0), + "{again:?}" + ); + reasoning::materialize(&pool, f.kb).await?; + assert!(!holds_at(&pool, &f, f.a, rule, NOW, None).await?); + assert!(holds_at(&pool, &f, f.a, rule, T_MID, None).await?); + anyhow::Ok(()) + } + .await; + + cleanup(&pool, f.org).await?; + run +} + +/// #900：同一身份再推一份新内容（原地替换、记版本、`doc_time` 换成新的，新块顶替旧块）。 +/// 停在旧版上的证据按**它那一版**的日期算，所以后一次观察关得上前一段，而不是记成 +/// 「同时」的冲突 +#[tokio::test] +async fn a_same_identity_update_closes_the_earlier_place() -> anyhow::Result<()> { + let Some(url) = utopia_store::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + let f = seed(&pool, "issue900-same-identity").await?; + + let run = async { + let (doc, c1) = document(&pool, &f, "cup-7.json", T1, "cup-7 is on desk").await?; + // 直接写进库的文档没有版本行：第一版按现在的日期补上，和摄入路径写下的一样 + utopia_store::documents::record_version(&pool, doc, "cup-7-v1", 17).await?; + statement(&pool, &f, c1, "desk", T1).await?; + bind_is_on(&pool, &f).await?; + let first = materialize::materialize(&pool, f.kb).await?; + assert_eq!(first.added, 1, "{first:?}"); + + // 第二次推送：原地替换、记版本、doc_time 换成新的（ingest_item 的那一步），新块顶替旧块 + utopia_store::documents::replace_content_and_enqueue_processing( + &pool, + doc, + "cup-7.json", + "application/json", + 17, + "cup-7-v2", + Some(t(T2)), + ) + .await?; + // 那一步排下的处理任务这里不跑：抽取的结果由下面几行写出来 + sqlx::query("DELETE FROM jobs WHERE payload->>'document_id' = $1") + .bind(doc.to_string()) + .execute(&pool) + .await?; + let text = "cup-7 is on shelf".to_string(); + let piece = utopia_ingest::ChunkPiece { + seq: 0, + char_start: 0, + char_end: text.chars().count() as i32, + heading: None, + provenance: utopia_ingest::Provenance::stated(), + text: text.clone(), + }; + utopia_store::documents::replace_chunks(&pool, f.kb, doc, &[piece]).await?; + let (c2,): (Uuid,) = sqlx::query_as( + "SELECT id FROM chunks WHERE document_id = $1 AND superseded_at IS NULL", + ) + .bind(doc) + .fetch_one(&pool) + .await?; + statement(&pool, &f, c2, "shelf", T2).await?; + let second = materialize::materialize(&pool, f.kb).await?; + assert_eq!( + (second.added, second.corrected, second.conflicts), + (1, 1, 0), + "the desk row dates at version 1's time, the shelf row at version 2's: {second:?}" + ); + let desk = typed_desk(&pool, &f).await?; + assert_eq!(desk.len(), 1); + assert_eq!( + desk[0].valid_to_precision.as_deref(), + Some("unknown"), + "after a same-identity update the desk row is closed: {desk:?}" + ); + // 版本表记着各自的日期 + let dates: Vec<(i32, Option>)> = sqlx::query_as( + "SELECT version, doc_time FROM document_versions WHERE document_id = $1 ORDER BY version", + ) + .bind(doc) + .fetch_all(&pool) + .await?; + assert_eq!(dates, vec![(1, Some(t(T1))), (2, Some(t(T2)))]); + anyhow::Ok(()) + } + .await; + + cleanup(&pool, f.org).await?; + run +} diff --git a/docs/decisions/0054-a-source-may-push-statements-in-the-open-contract.md b/docs/decisions/0054-a-source-may-push-statements-in-the-open-contract.md index 54326b98f..d48e13f25 100644 --- a/docs/decisions/0054-a-source-may-push-statements-in-the-open-contract.md +++ b/docs/decisions/0054-a-source-may-push-statements-in-the-open-contract.md @@ -25,7 +25,7 @@ A `statements` source accepts the open extraction shape verbatim: `e`, `s` and ` **2. The payload is the document, in one piece.** -The `{e, s, n}` object is stored as the document's content and as its single chunk, verbatim. Identity, versions, tombstones and the run history are exactly the `api` source's; the document appears in the Library under its source like any other. The chunker is not consulted: its budget exists so that a model reads a passage it can hold, and no model reads this. +The `{e, s, n}` object is stored as the document's content and as its single chunk, verbatim, preceded by the observation's `external_id` and `doc_time` so that the document says which observation it is (revised 2026-09-25: two observations that saw the same thing are two documents, and a base keeps one document per content). Identity, versions, tombstones and the run history are exactly the `api` source's; the document appears in the Library under its source like any other. The chunker is not consulted: its budget exists so that a model reads a passage it can hold, and no model reads this. **3. No model, the same path.** @@ -65,7 +65,7 @@ Content-Type: application/json } ``` -- `external_id` is required and is the identity (`statements:{external_id}`); a second push with new content updates in place and records a version; `deleted: true` tombstones it. +- `external_id` is required and is the identity (`statements:{external_id}`); a second push with new content updates in place and records a version; `deleted: true` tombstones it. One observation, one identity: the same payload under a new `external_id` is a second observation with its own date, never a rename of the first. - `doc_time` is the observation's own time and lands on the world axis; push time is the record axis (0022). Without it the item is undated, as an upload is. - Each `s` item is `[quote, subject, phrase, object, value, qualifiers, when, ended]`; `quote` must be `null`. Each `e` item is `[name, kind word, named]`; each `n` item is `[entity name, other name, quote]` with `quote` null. - Keys other than `external_id`, `doc_time`, `deleted`, `e`, `s`, `n` are refused with 422, as is a subject or an `n` entity not listed in `e`. A body over 64 KiB or with more than 200 statements is refused with 422; those are cut-1 limits, not contracts. @@ -88,3 +88,7 @@ Content-Type: application/json - Whether a statement with no offsets should look any different on a Review card. Today it does not. - Whether `when` should accept an RFC 3339 instant directly rather than time words, once the `instant` precision on the roadmap exists (0045). + +## Revisions + +- 2026-09-25 (#899, #900): typed materialization now reconciles the rows it writes along their uniqueness timelines, as the write path always did, so a later observation of a functional attribute closes the earlier one without a manual reconcile. A document version records the `doc_time` it was pushed with, and a fact's evidence date is taken from its own version, so a same-identity update no longer makes the earlier statement look simultaneous with the later one. The stored document carries the observation's identity and date ahead of the three arrays, so the same payload under a new identity is a second document, never a rename of the first. diff --git a/migrations/0090_a_version_keeps_the_date_it_was_pushed_with.sql b/migrations/0090_a_version_keeps_the_date_it_was_pushed_with.sql new file mode 100644 index 000000000..d09286129 --- /dev/null +++ b/migrations/0090_a_version_keeps_the_date_it_was_pushed_with.sql @@ -0,0 +1,16 @@ +-- 一版文档记下它自己的日期（#900）。 +-- +-- 同一身份再推一份新内容是原地更新：`documents.doc_time` 换成新的，旧块作废、旧证据 +-- 停在旧版上。时间线给没起点的行排序用的是证据文件自带的日期——按文档当前的日期算， +-- 停在旧版上的行就和新行「同时」开始，函数型属性的前一段永远关不上（对账记成 +-- simultaneous 冲突）。版本表记下每一版推来时的日期，证据按自己那一版取日期。 +-- +-- 回填：只有当前这一版的日期是知道的（就是文档现在的日期）；更早的版本留空，取日期时 +-- 退回文档的日期，与从前一样。 +ALTER TABLE document_versions ADD COLUMN doc_time TIMESTAMPTZ; + +UPDATE document_versions v + SET doc_time = d.doc_time + FROM documents d + WHERE d.id = v.document_id + AND v.version = (SELECT max(x.version) FROM document_versions x WHERE x.document_id = v.document_id); diff --git a/web/src/docs/ingest.md b/web/src/docs/ingest.md index 0304072f3..d5cbaac44 100644 --- a/web/src/docs/ingest.md +++ b/web/src/docs/ingest.md @@ -142,7 +142,7 @@ Content-Type: application/json | Field | Required | Meaning | |---|---|---| -| `external_id` | yes | Stable identity. Same identity + new content → update in place, with a version recorded. | +| `external_id` | yes | Stable identity. Same identity + new content → update in place, with a version recorded. One observation, one identity: the same payload under a new identity is a second observation with its own `doc_time`, never a rename. | | `doc_time` | no | RFC 3339; the observation's own time. Without it the item is undated. | | `e` | yes | Things: `[name, kind word, named]`. `named` is `true` for a name, `false` for a description. | | `s` | yes | Statements: `[quote, subject, phrase, object, value, qualifiers, when, ended]`. `quote` must be `null`; `subject` and `object` name things listed in `e`; give `object` or `value`, not both; `qualifiers` is an object keyed by your own role words; `when` / `ended` are time words as you would write them. |