diff --git a/crates/utopia-server/src/api/mod.rs b/crates/utopia-server/src/api/mod.rs index d8d9949d9..b985f1c9f 100644 --- a/crates/utopia-server/src/api/mod.rs +++ b/crates/utopia-server/src/api/mod.rs @@ -492,6 +492,11 @@ pub fn router(state: AppState, cfg: &AppConfig) -> Router { "/kbs/{id}/sources/{source_id}/missing/cleanup", post(sources_routes::cleanup_missing), ) + // 来源关上自己不再陈述的事实（RC-004），写令牌可用 + .route( + "/kbs/{id}/sources/{source_id}/close-unconfirmed", + post(sources_routes::close_unconfirmed), + ) .route( "/documents/{id}/extractions", get(documents_routes::extractions), diff --git a/crates/utopia-server/src/api/sources_close_unconfirmed_tests.rs b/crates/utopia-server/src/api/sources_close_unconfirmed_tests.rs new file mode 100644 index 000000000..b90125cd7 --- /dev/null +++ b/crates/utopia-server/src/api/sources_close_unconfirmed_tests.rs @@ -0,0 +1,94 @@ +//! `close-unconfirmed` 收写令牌（RC-004）：读令牌 403，别的库的来源 404。 +//! 关哪些、怎么关由 store 的 `a_source_closes_its_unconfirmed_facts` 钉住。 +use crate::state::AppState; +use axum::body::{to_bytes, Body}; +use axum::http::{Request, StatusCode}; +use serde_json::{json, Value}; +use std::sync::Arc; +use tower::ServiceExt; +use uuid::Uuid; + +#[tokio::test] +async fn a_write_token_closes_through_its_own_bases_source_only() -> anyhow::Result<()> { + let Some(url) = utopia_store::test_db::url() else { + return Ok(()); + }; + let pool = sqlx::PgPool::connect(&url).await?; + let [org, ws, editor, kb, other_kb, source, foreign] = [(); 7].map(|_| Uuid::now_v7()); + sqlx::raw_sql(&format!( + "INSERT INTO organizations(id,name) VALUES ('{org}','close-unconfirmed-route'); + INSERT INTO workspaces(id,org_id,name) VALUES ('{ws}','{org}','close-unconfirmed-route'); + INSERT INTO users(id,org_id,email,display_name,password_hash) VALUES + ('{editor}','{org}','{editor}@close-unconfirmed.test','Editor','unused'); + INSERT INTO memberships(user_id,workspace_id,role) VALUES ('{editor}','{ws}','editor'); + INSERT INTO knowledge_bases(id,workspace_id,name) VALUES + ('{kb}','{ws}','mirror'), ('{other_kb}','{ws}','other'); + INSERT INTO kb_members(kb_id,user_id,role) VALUES + ('{kb}','{editor}','editor'), ('{other_kb}','{editor}','editor'); + INSERT INTO sources(id,kb_id,kind,name) VALUES + ('{source}','{kb}','statements','mirror'), ('{foreign}','{other_kb}','statements','other');" + )) + .execute(&pool) + .await?; + let dir = tempfile::tempdir()?; + let cfg = utopia_core::config::AppConfig { + data_dir: dir.path().to_string_lossy().into_owned(), + ..Default::default() + }; + let search = Arc::new(utopia_search::SearchIndex::open( + &dir.path().join("search"), + )?); + let app = super::super::router( + AppState::new(pool.clone(), &cfg, search, "test-only".into()), + &cfg, + ); + let post = |source: Uuid, token: String| { + let app = app.clone(); + async move { + let request = Request::post(format!( + "/api/v1/kbs/{kb}/sources/{source}/close-unconfirmed" + )) + .header("authorization", format!("Bearer {token}")) + .header("content-type", "application/json") + .body(Body::from(json!({ "limit": 10 }).to_string()))?; + let response = app.oneshot(request).await?; + let status = response.status(); + let bytes = to_bytes(response.into_body(), 1024 * 1024).await?; + anyhow::Ok(( + status, + serde_json::from_slice::(&bytes).unwrap_or(Value::Null), + )) + } + }; + let result = async { + let issue = |scope: &'static str| { + let pool = pool.clone(); + async move { + anyhow::Ok( + utopia_store::tokens::issue(&pool, editor, "ci", scope, None, None) + .await? + .1, + ) + } + }; + let (write, read) = (issue("write").await?, issue("read").await?); + assert_eq!( + post(source, write.clone()).await?, + ( + StatusCode::OK, + json!({ "closed": 0, "skipped": 0, "remaining": 0 }) + ) + ); + assert_eq!(post(source, read).await?.0, StatusCode::FORBIDDEN); + assert_eq!(post(foreign, write).await?.0, StatusCode::NOT_FOUND); + anyhow::Ok(()) + } + .await; + sqlx::raw_sql(&format!( + "DELETE FROM knowledge_bases WHERE id IN ('{kb}','{other_kb}'); + DELETE FROM organizations WHERE id = '{org}';" + )) + .execute(&pool) + .await?; + result +} diff --git a/crates/utopia-server/src/api/sources_routes.rs b/crates/utopia-server/src/api/sources_routes.rs index 8285b2d9b..1c6b54d1d 100644 --- a/crates/utopia-server/src/api/sources_routes.rs +++ b/crates/utopia-server/src/api/sources_routes.rs @@ -364,6 +364,63 @@ pub async fn cleanup_missing( #[path = "sources_cleanup_tests.rs"] mod cleanup_tests; +/// 一批至多关这么多条：每条是一笔带锁的作废+改写，一次请求不该占着库太久 +const CLOSE_UNCONFIRMED_MAX: i64 = 5000; + +#[derive(Deserialize)] +pub struct CloseUnconfirmedBody { + #[serde(default)] + pub limit: Option, +} + +/// 来源关上自己不再陈述的事实（RC-004）。写令牌可用：推陈述的就是一个自动化客户端， +/// 它推完一轮接着清自己留下的待确认 +pub async fn close_unconfirmed( + State(state): State, + caller: SessionOrToken, + Path((kb_id, source_id)): Path<(Uuid, Uuid)>, + Json(body): Json, +) -> ApiResult> { + caller.require_write(Some(kb_id))?; + require_kb(&state, &caller.user, kb_id, Role::Editor).await?; + source_in_kb(&state, kb_id, source_id).await?; + let limit = body.limit.unwrap_or(1000); + if !(1..=CLOSE_UNCONFIRMED_MAX).contains(&limit) { + return Err(utopia_core::AppError::invalid( + "bad_limit", + format!("limit is between 1 and {CLOSE_UNCONFIRMED_MAX}"), + ) + .into()); + } + let batch = + utopia_store::unconfirmed::close_of_source(&state.pool, kb_id, source_id, limit).await?; + let _ = utopia_store::audit::record( + &state.pool, + Some(kb_id), + caller.user.id, + "source.close_unconfirmed", + "source", + Some(source_id), + utopia_store::audit::via_token( + json!({ "source_id": source_id, "closed": batch.closed, "skipped": batch.skipped }), + caller.token_id(), + ), + ) + .await; + if batch.closed > 0 { + state.emit_review(kb_id); + } + Ok(Json(json!({ + "closed": batch.closed, + "skipped": batch.skipped, + "remaining": batch.remaining, + }))) +} + +#[cfg(test)] +#[path = "sources_close_unconfirmed_tests.rs"] +mod close_unconfirmed_tests; + pub async fn delete( State(state): State, AuthUser(user): AuthUser, diff --git a/crates/utopia-server/src/auth.rs b/crates/utopia-server/src/auth.rs index a40be2926..0e4a966f9 100644 --- a/crates/utopia-server/src/auth.rs +++ b/crates/utopia-server/src/auth.rs @@ -182,6 +182,9 @@ impl FromRequestParts for AuthUser { /// /// 按推送时的 `external_id` 读回一篇文档（#1276）也是文档原件，开给令牌的理由写在决定 7。 /// +/// 来源关上自己不再陈述的事实（`close-unconfirmed`，fork 的 RC-004）：推陈述的客户端 +/// 只有令牌。 +/// /// 令牌只收窄：`pat` 在的时候，路由先问 [`Self::require_write`] 或 /// [`Self::reaches`]，角色照旧由 `require_kb` 与工作区角色判。 pub struct SessionOrToken { diff --git a/crates/utopia-store/src/lib.rs b/crates/utopia-store/src/lib.rs index bb52d627e..99a04d739 100644 --- a/crates/utopia-store/src/lib.rs +++ b/crates/utopia-store/src/lib.rs @@ -52,6 +52,7 @@ pub mod test_db; pub mod time_mentions; pub mod tokens; pub mod type_bindings; +pub mod unconfirmed; pub mod vector_index; pub mod workspaces; pub mod world_axis; diff --git a/crates/utopia-store/src/temporal.rs b/crates/utopia-store/src/temporal.rs index 85e6d9853..1c67f9a2e 100644 --- a/crates/utopia-store/src/temporal.rs +++ b/crates/utopia-store/src/temporal.rs @@ -1091,12 +1091,15 @@ async fn rewrite_end_tx( valid_from, valid_from_precision, valid_from_grade, valid_to, valid_to_precision, confidence, supersedes, attested_from, attested_by, attested_to, end_derived, - from_statement_id, implied, corrected_ends, end_after_period) + from_statement_id, implied, layer, phrase, + corrected_ends, end_after_period) SELECT $1, kb_id, subject_id, predicate_id, object_id, object_value, valid_from, valid_from_precision, valid_from_grade, $3, $4, confidence, id, attested_from, attested_by, CASE WHEN $4::text = 'unknown' THEN COALESCE($5, now()) END, $6, - from_statement_id, implied, + -- 开放陈述关上后还是陈述（RC-004）：少了这两列，修正行成了没有谓词的类型化行， + -- 物化再也读不到它 + from_statement_id, implied, layer, phrase, -- 终点在这里被重新写下（文档说了终点，或时间线推了一个），人改过的终点就不在了； -- 起点那一半照旧 CASE corrected_ends WHEN 'both' THEN 'start' WHEN 'end' THEN NULL diff --git a/crates/utopia-store/src/unconfirmed.rs b/crates/utopia-store/src/unconfirmed.rs new file mode 100644 index 000000000..5cfd00992 --- /dev/null +++ b/crates/utopia-store/src/unconfirmed.rs @@ -0,0 +1,106 @@ +//! 一个来源关上自己不再陈述的事实（RC-004）。 +//! +//! 推陈述的来源每次重推，旧版分块被取代；不再被说出的那条事实只剩停在旧分块上的证据， +//! 落进审核的「待确认」（[`crate::review::UNCONFIRMED_FACT`]）一直不走。这里成批关上它们： +//! 终点是它最后一条证据被取代的那一刻，走 [`crate::temporal::close_superseded`]——与 +//! 审核里人点「闭合」同一条作废+改写，历史留着，不撤销也不否决。 + +use sqlx::PgPool; +use utopia_core::AppResult; +use uuid::Uuid; + +/// 一批的结果。`remaining` 是这批之后还能关的条数，全量数（由来源的文档出发，走 +/// `fact_evidence_document_idx`）；`skipped` 是起点晚于那一刻、关不上、留给人的条数 +pub struct SourceClosing { + pub closed: u64, + pub skipped: i64, + pub remaining: i64, +} + +/// 来源 `$2` 的待确认事实，库 `$1`：证据全出自这个来源的文档、全被取代；不含名字事实 +/// （`known_as` 是 Utopia 自己写的）；带陈述来源的类型化行不算——它跟着陈述走，陈述关了 +/// 物化把它按关上的陈述重建。`inverted`：起点（按它的精度截断）晚于终点， +/// 与 `close_superseded` 拒绝 `interval_inverted` 是同一个比较 +fn candidates() -> String { + format!( + "WITH ev AS ( + SELECT fe.fact_id, date_trunc('second', max(c.superseded_at), 'UTC') AS ended + FROM documents d + JOIN fact_evidence fe ON fe.document_id = d.id + JOIN chunks c ON c.id = fe.chunk_id + WHERE d.kb_id = $1 AND d.source_id = $2 + GROUP BY fe.fact_id + HAVING bool_and(c.superseded_at IS NOT NULL)) + SELECT f.id, f.layer, ev.ended, + COALESCE(date_trunc(CASE WHEN f.valid_from_precision IN + ('year','month','day','hour','minute','second') + THEN f.valid_from_precision ELSE 'microseconds' END, + f.valid_from, 'UTC') > ev.ended, false) AS inverted + FROM ev JOIN facts f ON f.id = ev.fact_id + WHERE f.kb_id = $1 AND f.invalidated_at IS NULL AND {unconfirmed} AND {not_a_name} + AND NOT EXISTS (SELECT 1 FROM fact_evidence fe + JOIN chunks c ON c.id = fe.chunk_id + JOIN documents d ON d.id = c.document_id + WHERE fe.fact_id = f.id AND d.source_id IS DISTINCT FROM $2) + AND (f.layer = 'open' + OR NOT EXISTS (SELECT 1 FROM typed_fact_sources s WHERE s.fact_id = f.id))", + unconfirmed = crate::review::UNCONFIRMED_FACT, + not_a_name = crate::names::not_a_name("f"), + ) +} + +/// 关上至多 `limit` 条，开放陈述在前。关了陈述就排一次类型化重算（与否决陈述同一个做法， +/// `graph::reject_fact`）：旧陈述作废，它物化出的行在重算里退下，再按关上的陈述重建 +pub async fn close_of_source( + pool: &PgPool, + kb_id: Uuid, + source_id: Uuid, + limit: i64, +) -> AppResult { + let due: Vec<(Uuid, String, chrono::DateTime)> = sqlx::query_as(&format!( + "SELECT id, layer, ended FROM ({}) c WHERE NOT inverted + ORDER BY layer = 'typed', id LIMIT $3", + candidates() + )) + .persistent(false) + .bind(kb_id) + .bind(source_id) + .bind(limit) + .fetch_all(pool) + .await?; + let mut closed = 0; + let mut statements = false; + for (fact_id, layer, ended) in &due { + // None = 期间被别人作废或关了，不算这批的 + if crate::temporal::close_superseded(pool, *fact_id, *ended, "second") + .await? + .is_some() + { + closed += 1; + statements |= layer == "open"; + } + } + if statements { + crate::jobs::enqueue_unless_queued( + pool, + crate::phrase_bindings::MATERIALIZE_KIND, + serde_json::json!({ "kb_id": kb_id }), + ) + .await?; + } + let (remaining, skipped): (i64, i64) = sqlx::query_as(&format!( + "SELECT count(*) FILTER (WHERE NOT inverted), count(*) FILTER (WHERE inverted) + FROM ({}) c", + candidates() + )) + .persistent(false) + .bind(kb_id) + .bind(source_id) + .fetch_one(pool) + .await?; + Ok(SourceClosing { + closed, + skipped, + remaining, + }) +} diff --git a/crates/utopia-store/tests/store/a_source_closes_its_unconfirmed_facts.rs b/crates/utopia-store/tests/store/a_source_closes_its_unconfirmed_facts.rs new file mode 100644 index 000000000..49975b034 --- /dev/null +++ b/crates/utopia-store/tests/store/a_source_closes_its_unconfirmed_facts.rs @@ -0,0 +1,132 @@ +//! 一个来源关上自己不再陈述的事实（RC-004）：证据全停在被取代的分块上、证据全出自这个 +//! 来源的现行事实，按最后一条证据被取代的那一刻闭合，走作废+改写，历史留着。 +//! 还有活证据的、带着别的来源证据的、名字事实、起点在那一刻之后的，都不动。 + +use chrono::{DateTime, Utc}; +use sqlx::PgPool; +use utopia_store::{materialize, phrase_bindings, unconfirmed}; +use uuid::Uuid; + +#[tokio::test] +async fn a_source_closes_only_the_facts_it_alone_stopped_stating() -> anyhow::Result<()> { + let Some(url) = utopia_store::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + let [org, ws, kb, class, ada, bea, cy, port, oslo, based_in, known_as, binding] = + std::array::from_fn(|_| Uuid::now_v7()); + let [source, other, doc, other_doc, old, live, elsewhere] = + std::array::from_fn(|_| Uuid::now_v7()); + let [statement, still_stated, shared, name, late, lone] = + std::array::from_fn(|_| Uuid::now_v7()); + let ended: DateTime = "2025-06-01T12:34:56Z".parse()?; + let run = async { + sqlx::raw_sql(&format!( + "INSERT INTO organizations(id,name) VALUES ('{org}','close-unconfirmed'); + INSERT INTO workspaces(id,org_id,name) VALUES ('{ws}','{org}','close-unconfirmed'); + INSERT INTO knowledge_bases(id,workspace_id,name) VALUES ('{kb}','{ws}','close-unconfirmed'); + INSERT INTO entity_types(id,kb_id,key,label) VALUES ('{class}','{kb}','thing','Thing'); + INSERT INTO entities(id,kb_id,type_id,canonical_name) VALUES + ('{ada}','{kb}','{class}','Ada'), ('{bea}','{kb}','{class}','Bea'), + ('{cy}','{kb}','{class}','Cy'), ('{port}','{kb}','{class}','Port Ellen'), + ('{oslo}','{kb}','{class}','Oslo'); + INSERT INTO relation_types (id,kb_id,key,label,temporal) + VALUES ('{based_in}','{kb}','based_in','based in','state'); + INSERT INTO relation_types (id,kb_id,key,label,kind,datatype,temporal,builtin) + VALUES ('{known_as}','{kb}','known_as','known as','attribute','text','state',true); + INSERT INTO phrase_bindings + (id,kb_id,phrase,subject_type_id,object_type_id,object_is_value, + relation_type_id,direction,status) + VALUES ('{binding}','{kb}','is based in','{class}','{class}',false, + '{based_in}','forward','bound'); + INSERT INTO sources(id,kb_id,kind,name) VALUES + ('{source}','{kb}','statements','mirror'), ('{other}','{kb}','statements','other'); + INSERT INTO documents(id,kb_id,filename,sha256,source_id) VALUES + ('{doc}','{kb}','a.md','sha-a','{source}'), + ('{other_doc}','{kb}','b.md','sha-b','{other}'); + INSERT INTO chunks(id,kb_id,document_id,seq,text,doc_version,superseded_at) VALUES + ('{old}','{kb}','{doc}',0,'v1','1','2025-06-01T12:34:56.789Z'), + ('{live}','{kb}','{doc}',1,'v2','2',NULL), + ('{elsewhere}','{kb}','{other_doc}',0,'b','1','2025-06-01T12:00:00Z'); + INSERT INTO facts (id,kb_id,subject_id,object_id,layer,phrase,confidence, + valid_from,valid_from_precision) VALUES + ('{statement}','{kb}','{ada}','{port}','open','is based in',0.9,'2020-01-01','day'), + ('{still_stated}','{kb}','{bea}','{port}','open','is based in',0.9,NULL,NULL), + ('{shared}','{kb}','{cy}','{port}','open','is based in',0.9,NULL,NULL); + INSERT INTO facts (id,kb_id,subject_id,predicate_id,object_id,object_value,confidence, + valid_from,valid_from_precision) VALUES + ('{name}','{kb}','{ada}','{known_as}',NULL,'{{\"value\":\"Ada\"}}',0.9,NULL,NULL), + ('{late}','{kb}','{bea}','{based_in}','{oslo}',NULL,0.9,'2026-01-01','day'), + ('{lone}','{kb}','{cy}','{based_in}','{oslo}',NULL,0.9,NULL,NULL); + INSERT INTO fact_evidence (fact_id,chunk_id,document_id,quote) VALUES + ('{statement}','{old}','{doc}','q'), ('{still_stated}','{old}','{doc}','q'), + ('{still_stated}','{live}','{doc}','q'), ('{shared}','{old}','{doc}','q'), + ('{shared}','{elsewhere}','{other_doc}','q'), ('{name}','{old}','{doc}','q'), + ('{late}','{old}','{doc}','q'), ('{lone}','{old}','{doc}','q');" + )) + .execute(&pool) + .await?; + // 陈述先物化出类型化行：它跟着陈述走，不由这里关 + materialize::materialize(&pool, kb).await?; + + let first = unconfirmed::close_of_source(&pool, kb, source, 1).await?; + assert_eq!( + (first.closed, first.skipped, first.remaining), + (1, 1, 1), + "开放陈述先关；起点在终点之后的算跳过；还剩那条孤零零的类型化行" + ); + let queued: i64 = sqlx::query_scalar( + "SELECT count(*) FROM jobs WHERE kind = $1 AND payload->>'kb_id' = $2 AND status = 'queued'", + ) + .bind(phrase_bindings::MATERIALIZE_KIND) + .bind(kb.to_string()) + .fetch_one(&pool) + .await?; + assert_eq!(queued, 1, "关了陈述就排一次类型化重算"); + let second = unconfirmed::close_of_source(&pool, kb, source, 1000).await?; + assert_eq!((second.closed, second.skipped, second.remaining), (1, 1, 0)); + + let successor = |fact: Uuid| { + let pool = pool.clone(); + async move { + sqlx::query_as::<_, (Option>, Option, String)>( + "SELECT n.valid_to, n.valid_to_precision, n.layer FROM facts n + JOIN facts o ON o.id = n.supersedes + WHERE n.supersedes = $1 AND n.invalidated_at IS NULL + AND o.invalidated_at IS NOT NULL", + ) + .bind(fact) + .fetch_optional(&pool) + .await + } + }; + let closed_at = |layer: &str| Some((Some(ended), Some("second".into()), layer.into())); + assert_eq!(successor(statement).await?, closed_at("open"), "陈述关上后还是陈述"); + assert_eq!(successor(lone).await?, closed_at("typed")); + for untouched in [still_stated, shared, name, late] { + assert_eq!(successor(untouched).await?, None); + } + + materialize::materialize(&pool, kb).await?; + let typed_end: Vec>> = sqlx::query_scalar( + "SELECT valid_to FROM facts WHERE kb_id = $1 AND layer = 'typed' AND subject_id = $2 + AND predicate_id = $3 AND invalidated_at IS NULL", + ) + .bind(kb) + .bind(ada) + .bind(based_in) + .fetch_all(&pool) + .await?; + assert_eq!(typed_end, vec![Some(ended)], "类型化行按关上的陈述重建"); + anyhow::Ok(()) + } + .await; + sqlx::raw_sql(&format!( + "DELETE FROM jobs WHERE payload->>'kb_id' = '{kb}'; + DELETE FROM knowledge_bases WHERE id = '{kb}'; + DELETE FROM organizations WHERE id = '{org}';" + )) + .execute(&pool) + .await?; + run +} diff --git a/crates/utopia-store/tests/store/main.rs b/crates/utopia-store/tests/store/main.rs index 4184eba40..b5b891238 100644 --- a/crates/utopia-store/tests/store/main.rs +++ b/crates/utopia-store/tests/store/main.rs @@ -97,6 +97,7 @@ mod a_signature_holds_on_every_path; mod a_signature_vector_lasts_while_its_text_does; mod a_similar_name_is_proposed_not_merged; mod a_single_name_candidate_is_not_counted; +mod a_source_closes_its_unconfirmed_facts; mod a_source_decay_is_set_and_cleared_by_update; mod a_source_decays_only_when_asked; mod a_source_kind_is_listed_once; diff --git a/docs/decisions/re-cinq/RC-004-a-source-closes-its-unconfirmed-facts.md b/docs/decisions/re-cinq/RC-004-a-source-closes-its-unconfirmed-facts.md new file mode 100644 index 000000000..ed32bc9f8 --- /dev/null +++ b/docs/decisions/re-cinq/RC-004-a-source-closes-its-unconfirmed-facts.md @@ -0,0 +1,25 @@ +# RC-004 · A source closes its unconfirmed facts + +> re:cinq fork only. Status: accepted 2026-10-24. Upstream: not yet proposed. + +## Context + +A `statements` source ([0054](../0054-a-source-may-push-statements-in-the-open-contract.md)) pushes the same items again and again. When an item stops stating a fact, because a value changed or a relation went away, the old fact keeps only evidence on superseded chunks. The Review page lists it as unconfirmed (`review::UNCONFIRMED_FACT`) until a person closes it, one at a time, with a session. Our bases hold about 1.65 million of them, and the system that knows they ended holds only a write token. + +## Decision + +**`POST /kbs/{kb}/sources/{source}/close-unconfirmed` closes a batch.** It needs a write token or a session, and the editor role on the base. A source of another base is `404`. `limit` is 1 to 5000, 1000 by default. + +**It closes only what the source alone stopped stating.** A fact qualifies when it is unconfirmed, every chunk that attests it belongs to a document of this source, and it is not a `known_as` name fact, which Utopia writes itself. A typed row with a statement source (`typed_fact_sources`) is left out: it follows its statements. + +**A close is the Review page's close.** Each fact goes through `temporal::close_superseded`, ending at the latest `superseded_at` of its evidence at `second` precision. The old row is invalidated and the closed row supersedes it, with its evidence. Open statements go first. When any was closed, a typed recompute is queued, the same way a rejected statement queues one. The recompute retires the typed row of the old statement and builds it again from the closed one, with its end. A fact whose start is after that end cannot be closed this way. The query leaves it out and counts it as `skipped`, so it cannot fill every batch. + +**A closed statement stays a statement.** `rewrite_end_tx` now copies `layer` and `phrase`. Before this change, closing an open statement wrote a typed row with no predicate, and the recompute never saw it again. + +**One audit row per batch** (`source.close_unconfirmed` with `source_id`, `closed`, `skipped` and the token). The answer is `{closed, skipped, remaining}`. `remaining` is a full count of what a further call would close. It runs the same query from the source's documents through `fact_evidence_document_idx`. + +## Consequences + +- The pusher can empty its own part of the Review queue. History is kept: nothing is rejected or deleted. +- A fact attested by two sources stays in the queue for a person. +- `rehome_tx` and the time correction in `temporal.rs` still write their rows without `layer` and `phrase`. They are left for another change. diff --git a/web/src/docs/ingest.md b/web/src/docs/ingest.md index 35bc59479..b5b7c4f2b 100644 --- a/web/src/docs/ingest.md +++ b/web/src/docs/ingest.md @@ -162,6 +162,20 @@ Two things to know before wiring a system to it: The response is the API push's: `{ "action": "created" }` · `updated` · `unchanged` · `marked_missing` · `duplicate` (with `duplicate_of`). It is never `moved`: the stored document carries its `external_id`, so two identities never share content. +### Closing what a source stopped stating + +The stale statements a source leaves behind can be closed in batches by the system that pushes them: + +``` +POST {your-utopia-base}/api/v1/kbs/{kb_id}/sources/{source_id}/close-unconfirmed +Authorization: Bearer utp_pat_… +Content-Type: application/json + +{ "limit": 1000 } +``` + +It closes up to `limit` (1–5000, default 1000) facts that are still open and that this source alone attested, in versions it has since replaced. Each one ends at the moment its last attesting version was replaced, to the second, the same way closing it on the Review page does: the old row stays in history. Facts with evidence from another source, Utopia's own name facts, and typed facts that follow an open statement are left alone; the typed facts are rebuilt closed by the recompute this queues. The credential is a write token (or session) of an editor of the base. The answer is `{ "closed": n, "skipped": n, "remaining": n }`: `skipped` counts facts that start after that moment and are left for a person, and `remaining` counts what a further call would close. Call it until `remaining` is `0`. + --- ## Reading a pushed document back