A source closes the facts it stopped stating, in batches, with its write token

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),

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::<Value>(&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
+}

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<i64>,
+}
+
+/// 来源关上自己不再陈述的事实（RC-004）。写令牌可用：推陈述的就是一个自动化客户端，
+/// 它推完一轮接着清自己留下的待确认
+pub async fn close_unconfirmed(
+    State(state): State<AppState>,
+    caller: SessionOrToken,
+    Path((kb_id, source_id)): Path<(Uuid, Uuid)>,
+    Json(body): Json<CloseUnconfirmedBody>,
+) -> ApiResult<Json<serde_json::Value>> {
+    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<AppState>,
     AuthUser(user): AuthUser,

crates/utopia-server/src/auth.rs

@@ -182,6 +182,9 @@ impl FromRequestParts<AppState> for AuthUser {
 ///
 /// 按推送时的 `external_id` 读回一篇文档（#1276）也是文档原件，开给令牌的理由写在决定 7。
 ///
+/// 来源关上自己不再陈述的事实（`close-unconfirmed`，fork 的 RC-004）：推陈述的客户端
+/// 只有令牌。
+///
 /// 令牌只收窄：`pat` 在的时候，路由先问 [`Self::require_write`] 或
 /// [`Self::reaches`]，角色照旧由 `require_kb` 与工作区角色判。
 pub struct SessionOrToken {

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;

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

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<SourceClosing> {
+    let due: Vec<(Uuid, String, chrono::DateTime<chrono::Utc>)> = 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,
+    })
+}