diff --git a/crates/utopia-server/src/decay_sweep.rs b/crates/utopia-server/src/decay_sweep.rs new file mode 100644 index 000000000..fe851b082 --- /dev/null +++ b/crates/utopia-server/src/decay_sweep.rs @@ -0,0 +1,45 @@ +//! 衰减的每日扫描（RC-003）：把打开了衰减的源淘汰回它的 `max_items`。 +//! +//! 选谁由 `utopia_store::decay::plan_eviction` 定；这里照人删文档的办法删（可恢复）， +//! 同时把文档从全文索引里拿掉，并记一笔无人操作的审计。 + +use crate::state::AppState; +use serde_json::json; +use utopia_core::AppError; +use uuid::Uuid; + +pub async fn sweep_source(state: &AppState, source_id: Uuid) -> anyhow::Result<()> { + let now = chrono::Utc::now(); + let Some(eviction) = utopia_store::decay::plan_eviction(&state.pool, source_id, now).await? + else { + return Ok(()); + }; + let mut evicted = 0; + for id in &eviction.documents { + match utopia_store::documents::delete(&state.pool, eviction.kb_id, *id, None).await { + Ok(_) => {} + // 扫描进行当中被人先删了 + Err(AppError::NotFound) => continue, + Err(e) => return Err(e.into()), + } + let search = state.search.clone(); + let document_id = id.to_string(); + tokio::task::spawn_blocking(move || search.delete_document(&document_id)).await??; + evicted += 1; + } + utopia_store::audit::record_opt( + &state.pool, + Some(eviction.kb_id), + None, + "decay.evict", + "source", + Some(source_id), + json!({ "evicted": evicted, "lowest_score": eviction.lowest_score }), + ) + .await?; + Ok(()) +} + +#[cfg(test)] +#[path = "decay_sweep_tests.rs"] +mod decay_sweep_tests; diff --git a/crates/utopia-server/src/decay_sweep_tests.rs b/crates/utopia-server/src/decay_sweep_tests.rs new file mode 100644 index 000000000..518e69ac0 --- /dev/null +++ b/crates/utopia-server/src/decay_sweep_tests.rs @@ -0,0 +1,136 @@ +//! 每日扫描把超出容量的源淘汰回去（RC-003），打在真库上。 +//! +//! 要钉住的行为：分最低的文档被删掉（可恢复）、从全文索引里拿掉，留下的原样， +//! 并记一笔无人操作的审计；没超出的源什么都不动。 +//! +//! 没有 `UTOPIA_DATABASE_URL` 时跳过而不是失败。自建自拆，绝不碰已有的库。 + +use std::sync::Arc; +use uuid::Uuid; + +#[tokio::test] +async fn a_sweep_removes_the_lowest_scoring_document_from_the_base_and_the_index( +) -> anyhow::Result<()> { + let Some(url) = utopia_store::test_db::url() else { + return Ok(()); + }; + let pool = sqlx::PgPool::connect(&url).await?; + utopia_store::db::migrate(&pool).await?; + let (org, ws, kb, source) = ( + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + ); + let (weak, strong) = (Uuid::now_v7(), Uuid::now_v7()); + let (weak_chunk, strong_chunk) = (Uuid::now_v7(), Uuid::now_v7()); + sqlx::query("INSERT INTO organizations(id,name) VALUES($1,'decay-sweep-server-test')") + .bind(org) + .execute(&pool) + .await?; + sqlx::query("INSERT INTO workspaces(id,org_id,name) VALUES($1,$2,'decay-sweep-server-test')") + .bind(ws) + .bind(org) + .execute(&pool) + .await?; + sqlx::query( + "INSERT INTO knowledge_bases(id,workspace_id,name) VALUES($1,$2,'decay-sweep-server-test')", + ) + .bind(kb) + .bind(ws) + .execute(&pool) + .await?; + sqlx::query( + "INSERT INTO sources(id,kb_id,kind,name,decay) + VALUES($1,$2,'folder','f','{\"max_items\":1,\"min_age_days\":0}'::jsonb)", + ) + .bind(source) + .bind(kb) + .execute(&pool) + .await?; + for (doc, chunk, name, text) in [ + (weak, weak_chunk, "notes/a", "apple orchard harvest report"), + ( + strong, + strong_chunk, + "decision/b", + "bridge concrete inspection", + ), + ] { + sqlx::query( + "INSERT INTO documents(id,kb_id,source_id,filename,sha256,created_at) + VALUES($1,$2,$3,$4,$1::text, now() - interval '100 days')", + ) + .bind(doc) + .bind(kb) + .bind(source) + .bind(name) + .execute(&pool) + .await?; + sqlx::query("INSERT INTO chunks(id,kb_id,document_id,seq,text) VALUES($1,$2,$3,0,$4)") + .bind(chunk) + .bind(kb) + .bind(doc) + .bind(text) + .execute(&pool) + .await?; + } + + let dir = std::env::temp_dir().join(format!("utopia-decay-sweep-{kb}")); + let cfg = utopia_core::config::AppConfig { + data_dir: dir.to_string_lossy().into_owned(), + ..Default::default() + }; + let search = Arc::new(utopia_search::SearchIndex::open(&dir.join("search"))?); + for (doc, chunk, text) in [ + (weak, weak_chunk, "apple orchard harvest report"), + (strong, strong_chunk, "bridge concrete inspection"), + ] { + search.reindex_document( + &kb.to_string(), + &doc.to_string(), + &[(chunk.to_string(), text.into())], + )?; + } + let state = crate::state::AppState::new(pool.clone(), &cfg, search.clone(), "test-only".into()); + + let run = async { + super::sweep_source(&state, source).await?; + + let deleted: Vec<(Uuid, bool)> = sqlx::query_as( + "SELECT id, deleted_at IS NOT NULL FROM documents WHERE source_id = $1 ORDER BY filename", + ) + .bind(source) + .fetch_all(&pool) + .await?; + assert_eq!(deleted, vec![(strong, false), (weak, true)]); + assert!(search.search(&kb.to_string(), "apple", 5, None)?.is_empty()); + assert_eq!(search.search(&kb.to_string(), "bridge", 5, None)?.len(), 1); + let audited: i64 = sqlx::query_scalar( + "SELECT count(*) FROM audit_events WHERE target_id = $1 AND action = 'decay.evict' + AND actor_id IS NULL", + ) + .bind(source) + .fetch_one(&pool) + .await?; + assert_eq!(audited, 1); + + super::sweep_source(&state, source).await?; + let still_live: i64 = sqlx::query_scalar( + "SELECT count(*) FROM documents WHERE source_id = $1 AND deleted_at IS NULL", + ) + .bind(source) + .fetch_one(&pool) + .await?; + assert_eq!(still_live, 1, "a source at its cap loses nothing more"); + anyhow::Ok(()) + } + .await; + + sqlx::query("DELETE FROM organizations WHERE id = $1") + .bind(org) + .execute(&pool) + .await?; + let _ = std::fs::remove_dir_all(&dir); + run +} diff --git a/crates/utopia-server/src/main.rs b/crates/utopia-server/src/main.rs index 6a947a9d9..fa3909631 100644 --- a/crates/utopia-server/src/main.rs +++ b/crates/utopia-server/src/main.rs @@ -4,6 +4,7 @@ mod api; mod auth; mod blob; mod client_ctx; +mod decay_sweep; mod docs_corpus; mod errata; mod error; @@ -295,6 +296,32 @@ async fn main() -> anyhow::Result<()> { } }); + // 衰减的每日扫描（RC-003）：打开了衰减的源，活文档超过 max_items 的，排一个淘汰任务 + let decay_state = state.clone(); + tokio::spawn(async move { + let mut tick = tokio::time::interval(std::time::Duration::from_secs(86_400)); + tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + tick.tick().await; + match utopia_store::decay::sources_over_cap(&decay_state.pool).await { + Ok(sources) => { + for source_id in sources { + if let Err(e) = utopia_store::jobs::enqueue_unless_queued( + &decay_state.pool, + "decay_sweep", + serde_json::json!({ "source_id": source_id }), + ) + .await + { + tracing::warn!(source_id = %source_id, error = %e, "衰减任务入队失败"); + } + } + } + Err(e) => tracing::warn!(error = %e, "扫描超出容量的源失败"), + } + } + }); + let app = api::router(state, &cfg); let listener = tokio::net::TcpListener::bind(&cfg.bind_addr).await?; tracing::info!("Utopia 服务启动于 http://{}", cfg.bind_addr); @@ -599,6 +626,15 @@ async fn dispatch(st: &state::AppState, job: &utopia_store::jobs::Job) -> anyhow .ok_or_else(|| anyhow::anyhow!("payload 缺少 kb_id"))?; adjudication::adjudicate_entities(st, kb_id).await } + "decay_sweep" => { + let source_id: Uuid = job + .payload + .get("source_id") + .and_then(|v| v.as_str()) + .and_then(|s| s.parse().ok()) + .ok_or_else(|| anyhow::anyhow!("payload 缺少 source_id"))?; + decay_sweep::sweep_source(st, source_id).await + } "govern" => { let kb_id: Uuid = job .payload diff --git a/crates/utopia-store/src/decay.rs b/crates/utopia-store/src/decay.rs index 794d5bb66..7908e8b47 100644 --- a/crates/utopia-store/src/decay.rs +++ b/crates/utopia-store/src/decay.rs @@ -81,10 +81,11 @@ pub async fn record_retrievals(pool: &PgPool, document_ids: &[Uuid]) -> AppResul Ok(written.rows_affected()) } -/// 打分用到的一个文档：`key` 是文档名，`value` 是它的文字。 -pub struct Memory<'a> { - pub key: &'a str, - pub value: &'a str, +/// 打分用到的一个文档：`key` 是文档名，`value_len` 是它的文字有多长（UTF-16 码元数， +/// 和 Lore 的 JS 字符串长度一致；库里按字符数，非基本平面的字符会少算，只差在边界上）。 +pub struct Memory { + pub key: String, + pub value_len: usize, pub created_at: DateTime, pub last_retrieved_at: Option>, /// 0 或没有都取 [`DEFAULT_HALF_LIFE_DAYS`]。 @@ -110,22 +111,26 @@ fn decay_strength(memory: &Memory, now: DateTime) -> f64 { } fn importance_adjustments(memory: &Memory) -> i64 { - value_length_adjustment(memory.value) - + key_prefix_adjustment(memory.key) - + key_topic_adjustment(memory.key) + value_length_adjustment(memory.value_len) + + key_prefix_adjustment(&memory.key) + + key_topic_adjustment(&memory.key) + retrieval_adjustment(memory.retrieval_count) + if memory.stale { -1 } else { 0 } } -/// 长度按 UTF-16 码元数，和 Lore（JS 的 `string.length`）一致，表情符号算两个。 -fn value_length_adjustment(value: &str) -> i64 { - match value.encode_utf16().count() { +fn value_length_adjustment(value_len: usize) -> i64 { + match value_len { 0..=49 => -2, 501.. => 1, _ => 0, } } +/// 一段文字的 UTF-16 码元数，Lore 里的 `string.length`：表情符号算两个。 +pub fn utf16_len(text: &str) -> usize { + text.encode_utf16().count() +} + fn key_prefix_adjustment(key: &str) -> i64 { let prefixes = ["auto-curation/", "session-summary/"]; -(prefixes.iter().filter(|p| key.starts_with(**p)).count() as i64) @@ -144,6 +149,145 @@ fn retrieval_adjustment(retrievals: u32) -> i64 { } } +/// 一直没被检索：最近一次检索（没有就创建）距今超过 `stale_after_days` 天。陈旧是读的时候 +/// 算的，不写进库，也不碰事实的 `confidence`。 +pub fn is_stale(last_touched: DateTime, stale_after_days: u32, now: DateTime) -> bool { + now - last_touched > chrono::Duration::days(i64::from(stale_after_days)) +} + +/// 淘汰候选：一个文档和给它打分用的材料。 +pub struct Candidate { + pub document_id: Uuid, + pub memory: Memory, +} + +/// 分最低的 `count` 个，分相同的保持候选的先后（候选按创建时间从旧到新给）。 +pub fn lowest_scoring( + candidates: &[Candidate], + now: DateTime, + count: usize, +) -> Vec<(Uuid, u8)> { + let mut scored: Vec<(Uuid, u8)> = candidates + .iter() + .map(|c| (c.document_id, score_importance(&c.memory, now))) + .collect(); + scored.sort_by_key(|(_, score)| *score); + scored.truncate(count); + scored +} + +#[derive(sqlx::FromRow)] +struct CandidateRow { + id: Uuid, + filename: String, + created_at: DateTime, + last_retrieved_at: Option>, + half_life_days: Option, + retrieval_count: i32, + text_len: i64, +} + +impl CandidateRow { + fn into_candidate(self, settings: &DecaySettings, now: DateTime) -> Candidate { + let last_touched = self.last_retrieved_at.unwrap_or(self.created_at); + Candidate { + document_id: self.id, + memory: Memory { + key: self.filename, + value_len: self.text_len as usize, + created_at: self.created_at, + last_retrieved_at: self.last_retrieved_at, + half_life_days: Some( + self.half_life_days + .map_or(f64::from(settings.half_life_days), f64::from), + ), + retrieval_count: self.retrieval_count as u32, + stale: is_stale(last_touched, settings.stale_after_days, now), + }, + } + } +} + +/// 淘汰的计划：哪个源、哪些文档、其中最低的分。 +pub struct Eviction { + pub kb_id: Uuid, + pub documents: Vec, + pub lowest_score: u8, +} + +/// 打开了衰减、活文档数超过 `max_items` 的源。 +pub async fn sources_over_cap(pool: &PgPool) -> AppResult> { + let ids: Vec<(Uuid,)> = sqlx::query_as( + "SELECT s.id FROM sources s + WHERE s.decay IS NOT NULL + AND (SELECT count(*) FROM documents d + WHERE d.source_id = s.id AND d.deleted_at IS NULL) + > COALESCE((s.decay->>'max_items')::bigint, $1)", + ) + .bind(i64::from(DecaySettings::default().max_items)) + .fetch_all(pool) + .await?; + Ok(ids.into_iter().map(|(id,)| id).collect()) +} + +/// 把一个源淘汰回 `max_items`：在比 `min_age_days` 老的文档里取最旧的 `2 × 多出来的数`， +/// 打分，其中最低的 `多出来的数` 个。多取一倍是为了让分有挑的余地，而不是只拿最旧的。 +/// 源没开衰减或没超出，返回 None。 +pub async fn plan_eviction( + pool: &PgPool, + source_id: Uuid, + now: DateTime, +) -> AppResult> { + let source: Option<(Uuid, Option)> = + sqlx::query_as("SELECT kb_id, decay FROM sources WHERE id = $1") + .bind(source_id) + .fetch_optional(pool) + .await?; + let Some((kb_id, Some(decay))) = source else { + return Ok(None); + }; + let settings: DecaySettings = + serde_json::from_value(decay).map_err(|e| AppError::Other(e.into()))?; + let live: i64 = sqlx::query_scalar( + "SELECT count(*) FROM documents WHERE source_id = $1 AND deleted_at IS NULL", + ) + .bind(source_id) + .fetch_one(pool) + .await?; + let excess = live - i64::from(settings.max_items); + if excess <= 0 { + return Ok(None); + } + let rows: Vec = sqlx::query_as( + "SELECT d.id, d.filename, d.created_at, sal.last_retrieved_at, sal.half_life_days, + COALESCE(sal.retrieval_count, 0) AS retrieval_count, + COALESCE((SELECT sum(char_length(c.text)) FROM chunks c + WHERE c.document_id = d.id AND c.superseded_at IS NULL), 0)::bigint + AS text_len + FROM documents d LEFT JOIN document_salience sal ON sal.document_id = d.id + WHERE d.source_id = $1 AND d.deleted_at IS NULL AND d.created_at <= $2 + ORDER BY d.created_at LIMIT $3", + ) + .bind(source_id) + .bind(now - chrono::Duration::days(i64::from(settings.min_age_days))) + .bind(excess * 2) + .fetch_all(pool) + .await?; + let candidates: Vec = rows + .into_iter() + .map(|row| row.into_candidate(&settings, now)) + .collect(); + let chosen = lowest_scoring(&candidates, now, excess as usize); + let Some((_, lowest_score)) = chosen.first().copied() else { + return Ok(None); + }; + Ok(Some(Eviction { + kb_id, + documents: chosen.into_iter().map(|(id, _)| id).collect(), + lowest_score, + })) +} + #[cfg(test)] mod tests { use super::*; @@ -153,10 +297,10 @@ mod tests { Utc.with_ymd_and_hms(2026, 10, 21, 0, 0, 0).unwrap() } - fn memory<'a>(key: &'a str, value: &'a str, created_days_ago: i64) -> Memory<'a> { + fn memory(key: &str, value: &str, created_days_ago: i64) -> Memory { Memory { - key, - value, + key: key.into(), + value_len: utf16_len(value), created_at: now() - Duration::days(created_days_ago), last_retrieved_at: None, half_life_days: None, @@ -242,6 +386,37 @@ mod tests { ); } + fn candidate(key: &str, days_old: i64) -> Candidate { + Candidate { + document_id: Uuid::now_v7(), + memory: memory(key, &"x".repeat(100), days_old), + } + } + + #[test] + fn the_lowest_scores_go_and_ties_keep_the_oldest_first() { + let candidates = [ + candidate("notes/a", 200), + candidate("decision/db", 200), + candidate("session-summary/x", 200), + candidate("notes/b", 100), + ]; + let chosen = lowest_scoring(&candidates, now(), 2); + let ids: Vec = chosen.iter().map(|(id, _)| *id).collect(); + assert_eq!( + ids, + vec![candidates[2].document_id, candidates[0].document_id] + ); + assert!(chosen[0].1 <= chosen[1].1); + } + + #[test] + fn a_document_is_stale_only_after_the_days_have_passed() { + let touched = now() - Duration::days(30); + assert!(!is_stale(touched, 30, now())); + assert!(is_stale(touched, 29, now())); + } + #[test] fn an_empty_settings_object_is_lores_defaults() { let settings: DecaySettings = serde_json::from_str("{}").unwrap(); diff --git a/crates/utopia-store/tests/store/a_source_over_its_cap_gives_up_its_lowest_scores.rs b/crates/utopia-store/tests/store/a_source_over_its_cap_gives_up_its_lowest_scores.rs new file mode 100644 index 000000000..86420025a --- /dev/null +++ b/crates/utopia-store/tests/store/a_source_over_its_cap_gives_up_its_lowest_scores.rs @@ -0,0 +1,116 @@ +//! 衰减的淘汰计划（RC-003），打在真库上。 +//! +//! 要钉住的行为： +//! +//! - **超出 max_items 的源才被排上**，没开衰减的、没超出的不排 +//! - **只在比 min_age_days 老的文档里挑**，新的碰不得 +//! - **挑分最低的，多出几个拿几个**；最近被检索的、名字带 decision 的分高，留下（久没人读的算陈旧，各扣一分） +//! - **分相同的，旧的先走** +//! +//! 没有 `UTOPIA_DATABASE_URL` 时跳过而不是失败。自建自拆，绝不碰已有的库。 + +use chrono::{Duration, Utc}; +use sqlx::PgPool; +use utopia_store::decay::{plan_eviction, sources_over_cap}; +use uuid::Uuid; + +#[tokio::test] +async fn a_source_over_its_cap_gives_up_its_lowest_scores() -> anyhow::Result<()> { + let Some(url) = utopia_store::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + utopia_store::db::migrate(&pool).await?; + + let (org, ws, kb) = (Uuid::now_v7(), Uuid::now_v7(), Uuid::now_v7()); + let (capped, plain, roomy) = (Uuid::now_v7(), Uuid::now_v7(), Uuid::now_v7()); + sqlx::query("INSERT INTO organizations (id, name) VALUES ($1, 'decay-sweep-test')") + .bind(org) + .execute(&pool) + .await?; + sqlx::query("INSERT INTO workspaces (id, org_id, name) VALUES ($1, $2, 'decay-sweep-test')") + .bind(ws) + .bind(org) + .execute(&pool) + .await?; + sqlx::query( + "INSERT INTO knowledge_bases (id, workspace_id, name) VALUES ($1, $2, 'decay-sweep-test')", + ) + .bind(kb) + .bind(ws) + .execute(&pool) + .await?; + for (id, decay) in [ + (capped, Some(r#"{"max_items": 3, "min_age_days": 30}"#)), + (plain, None), + (roomy, Some(r#"{"max_items": 10}"#)), + ] { + sqlx::query( + "INSERT INTO sources (id, kb_id, kind, name, decay) VALUES ($1, $2, 'folder', 'f', $3::jsonb)", + ) + .bind(id) + .bind(kb) + .bind(decay) + .execute(&pool) + .await?; + } + let now = Utc::now(); + let add = |source: Uuid, name: &'static str, days_old: i64| { + let pool = pool.clone(); + async move { + let id = Uuid::now_v7(); + sqlx::query( + "INSERT INTO documents (id, kb_id, source_id, filename, sha256, status, created_at) + VALUES ($1, $2, $3, $4, $1::text, 'ready', $5)", + ) + .bind(id) + .bind(kb) + .bind(source) + .bind(name) + .bind(now - Duration::days(days_old)) + .execute(&pool) + .await?; + anyhow::Ok(id) + } + }; + + let run = async { + let old = add(capped, "notes/a", 200).await?; + let decision = add(capped, "decision/db", 100).await?; + let summary = add(capped, "session-summary/d", 150).await?; + let read = add(capped, "notes/c", 90).await?; + let young = add(capped, "notes/e", 5).await?; + add(plain, "notes/p", 400).await?; + sqlx::query( + "INSERT INTO document_salience (document_id, retrieval_count, last_retrieved_at, half_life_days) + VALUES ($1, 25, $2, 60)", + ) + .bind(read) + .bind(now - Duration::days(1)) + .execute(&pool) + .await?; + + let over = sources_over_cap(&pool).await?; + assert!(over.contains(&capped), "five documents over a cap of three"); + assert!(!over.contains(&plain), "a source that does not decay is never over a cap"); + assert!(!over.contains(&roomy)); + + let plan = plan_eviction(&pool, capped, now).await?.expect("a plan"); + assert_eq!( + (plan.kb_id, plan.documents, plan.lowest_score), + (kb, vec![old, summary], 0), + "two go, the oldest first; the topic one, the read one and the young one stay" + ); + assert!(plan_eviction(&pool, plain, now).await?.is_none()); + assert!(plan_eviction(&pool, roomy, now).await?.is_none()); + let _ = (decision, read, young); + anyhow::Ok(()) + } + .await; + + sqlx::query("DELETE FROM organizations WHERE id = $1") + .bind(org) + .execute(&pool) + .await?; + run +} diff --git a/crates/utopia-store/tests/store/main.rs b/crates/utopia-store/tests/store/main.rs index ebae77744..995ed0eb0 100644 --- a/crates/utopia-store/tests/store/main.rs +++ b/crates/utopia-store/tests/store/main.rs @@ -100,6 +100,7 @@ mod a_single_name_candidate_is_not_counted; mod a_source_decay_is_set_and_cleared_by_update; mod a_source_decays_only_when_asked; mod a_source_kind_is_listed_once; +mod a_source_over_its_cap_gives_up_its_lowest_scores; mod a_source_reaches_only_where_it_was_granted; mod a_statement_is_attested_by_what_its_document_says; mod a_stopped_chat_is_replayed_as_incomplete; diff --git a/docs/design/sources.md b/docs/design/sources.md index 1595178c9..0ffd0c7a8 100644 --- a/docs/design/sources.md +++ b/docs/design/sources.md @@ -132,6 +132,11 @@ A source may let its documents decay [RC-003](../decisions/re-cinq/RC-003-a-sour `{}` is Lore's behaviour), `null` turns it off, and leaving the field out changes nothing. A new source decays nothing. +Once a day a source that holds more live documents than its `max_items` is taken back to it: from the +documents older than `min_age_days`, the lowest scoring ones are deleted the way a person deletes them +(so they can be restored), and one `decay.evict` audit row records how many. A source at or under its +cap loses nothing, and a source that does not decay is never looked at. + ## Open questions - Whether low engine confidence on an OCR or transcript segment should cap facts [0040].