From 4b492a12526f328697978a0e83d2967a638153bc Mon Sep 17 00:00:00 2001 From: Quintin Botes Date: Mon, 19 Oct 2026 19:41:40 +0200 Subject: [PATCH 01/11] The inference scheduler queues a base again only when it has no job waiting (#1304) * The inference scheduler queues a base again only when it has no job waiting Signed-off-by: Quintin Botes * The inference scheduling test runs in a binary of its own, since it queues every due base in the database Signed-off-by: Quintin Botes --------- Signed-off-by: Quintin Botes Co-authored-by: WaylandYang <145302500+WaylandYang@users.noreply.github.com> --- crates/utopia-server/src/main.rs | 17 +----- crates/utopia-store/src/reasoning.rs | 20 +++++++ .../tests/a_due_base_is_queued_once.rs | 52 +++++++++++++++++++ crates/utopia-store/tests/store/main.rs | 3 ++ 4 files changed, 77 insertions(+), 15 deletions(-) create mode 100644 crates/utopia-store/tests/a_due_base_is_queued_once.rs diff --git a/crates/utopia-server/src/main.rs b/crates/utopia-server/src/main.rs index 0e30b6697..6617ea075 100644 --- a/crates/utopia-server/src/main.rs +++ b/crates/utopia-server/src/main.rs @@ -260,21 +260,8 @@ async fn main() -> anyhow::Result<()> { tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); loop { tick.tick().await; - match utopia_store::reasoning::due_for_inference(&infer_state.pool).await { - Ok(due) => { - for kb_id in due { - if let Err(e) = utopia_store::jobs::enqueue( - &infer_state.pool, - "materialize_inferences", - serde_json::json!({ "kb_id": kb_id }), - ) - .await - { - tracing::warn!(kb_id = %kb_id, error = %e, "推理任务入队失败"); - } - } - } - Err(e) => tracing::warn!(error = %e, "扫描到期推理失败"), + if let Err(e) = utopia_store::reasoning::queue_due_inferences(&infer_state.pool).await { + tracing::warn!(error = %e, "扫描到期推理失败"); } } }); diff --git a/crates/utopia-store/src/reasoning.rs b/crates/utopia-store/src/reasoning.rs index 48374d4f8..607b3f5bb 100644 --- a/crates/utopia-store/src/reasoning.rs +++ b/crates/utopia-store/src/reasoning.rs @@ -2783,6 +2783,26 @@ pub async fn due_for_inference(pool: &PgPool) -> AppResult> { .await?) } +/// 定时扫描的一拍：到期的库各排一份推理任务，已经排着一份的不再排。 +/// +/// 只跳过排着的就够（#1304）：处理器先 `mark_inference_ran` 再推，任务一开始跑这个库就 +/// 不再到期；还到期的只有任务排着、`last_inference_at` 还没动的那段，而那段正是这里挡住的。 +/// 一个库排不进去不拦着别的库 +pub async fn queue_due_inferences(pool: &PgPool) -> AppResult<()> { + for kb_id in due_for_inference(pool).await? { + if let Err(e) = crate::jobs::enqueue_unless_queued( + pool, + "materialize_inferences", + json!({ "kb_id": kb_id }), + ) + .await + { + tracing::warn!(kb_id = %kb_id, error = %e, "推理任务入队失败"); + } + } + Ok(()) +} + /// 记下这一轮推完的时间。 /// /// **推完就记，哪怕什么都没变**：这一列答的是「上次看过没有」，不是「上次改过 diff --git a/crates/utopia-store/tests/a_due_base_is_queued_once.rs b/crates/utopia-store/tests/a_due_base_is_queued_once.rs new file mode 100644 index 000000000..ac220f378 --- /dev/null +++ b/crates/utopia-store/tests/a_due_base_is_queued_once.rs @@ -0,0 +1,52 @@ +//! #1304：到期的库在它的推理任务还排着时只排一份。 +//! +//! `queue_due_inferences` 扫的是全库到期的库，同时跑的测试刚建的库也会被它排上任务，所以 +//! 这个测试自成一个二进制（为什么见 `store/main.rs`），收尾删掉它排下的**每一份** +//! `materialize_inferences`，不只是自己那个库的。 + +use sqlx::PgPool; +use utopia_store::reasoning::queue_due_inferences; +use uuid::Uuid; + +#[tokio::test] +async fn a_due_base_is_queued_once_while_its_job_waits() -> anyhow::Result<()> { + let Some(url) = utopia_store::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + let (org, ws, kb) = (Uuid::now_v7(), Uuid::now_v7(), Uuid::now_v7()); + sqlx::raw_sql(&format!( + "INSERT INTO organizations(id,name) VALUES ('{org}','inference-schedule'); + INSERT INTO workspaces(id,org_id,name) VALUES ('{ws}','{org}','inference-schedule'); + INSERT INTO knowledge_bases(id,workspace_id,name,materialize_inferences) + VALUES ('{kb}','{ws}','inference-schedule',true);" + )) + .execute(&pool) + .await?; + let before: i64 = sqlx::query_scalar("SELECT coalesce(max(id), 0) FROM jobs") + .fetch_one(&pool) + .await?; + + // 两拍之间没人认领：从没推过的库两拍都算到期 + queue_due_inferences(&pool).await?; + queue_due_inferences(&pool).await?; + let queued: i64 = sqlx::query_scalar( + "SELECT count(*) FROM jobs + WHERE kind = 'materialize_inferences' AND status = 'queued' AND payload->>'kb_id' = $1", + ) + .bind(kb.to_string()) + .fetch_one(&pool) + .await?; + + sqlx::query("DELETE FROM jobs WHERE kind = 'materialize_inferences' AND id > $1") + .bind(before) + .execute(&pool) + .await?; + sqlx::query("DELETE FROM organizations WHERE id = $1") + .bind(org) + .execute(&pool) + .await?; + pool.close().await; + assert_eq!(queued, 1); + Ok(()) +} diff --git a/crates/utopia-store/tests/store/main.rs b/crates/utopia-store/tests/store/main.rs index 45cfd7e11..7bf02f198 100644 --- a/crates/utopia-store/tests/store/main.rs +++ b/crates/utopia-store/tests/store/main.rs @@ -22,6 +22,9 @@ //! 过滤——修正行、重复行都按全库找，会改到别的测试正在物化、改写的行。 //! - `a_row_a_rule_implied_follows_its_statements`：把迁移 0101 原样跑一遍，同样不按 kb //! 过滤——它按来源重算全库类型化行的见证，会改到别的测试正在物化的行。 +//! - `a_due_base_is_queued_once`：`queue_due_inferences` 扫全库到期的库。0050 起新建的库开着 +//! 推理又从没推过，别的测试刚建的库全算到期，会被它排上推理任务——数自己库任务的测试 +//! （比如 lib 单元测试里的 `phrase_bindings_delivery_tests`）就会多数出一个。 //! //! 新加测试默认放这个目录；只有碰上面那类全局状态时才放顶层，并在这里补一行为什么。 mod a_batch_decides_like_a_person; From b34e237980c943a2e003dc75ed11ded364a19c7c Mon Sep 17 00:00:00 2001 From: Quintin Botes Date: Mon, 19 Oct 2026 19:46:55 +0200 Subject: [PATCH 02/11] Raising a model's concurrency limit admits the tasks already waiting on it (#1305) * Raising a model's concurrency limit admits the tasks already waiting on it Signed-off-by: Quintin Botes * A changed model limit applies when it is saved, and raising one only adds permits Signed-off-by: Quintin Botes --------- Signed-off-by: Quintin Botes Co-authored-by: WaylandYang <145302500+WaylandYang@users.noreply.github.com> --- crates/utopia-server/src/api/admin_routes.rs | 10 +- crates/utopia-server/src/llm_util.rs | 96 ++++++++++++++---- crates/utopia-server/src/llm_util_tests.rs | 100 +++++++++++++++++++ crates/utopia-store/src/model_limits.rs | 3 +- 4 files changed, 188 insertions(+), 21 deletions(-) create mode 100644 crates/utopia-server/src/llm_util_tests.rs diff --git a/crates/utopia-server/src/api/admin_routes.rs b/crates/utopia-server/src/api/admin_routes.rs index ebbf2843d..68fcdbe0a 100644 --- a/crates/utopia-server/src/api/admin_routes.rs +++ b/crates/utopia-server/src/api/admin_routes.rs @@ -79,10 +79,10 @@ pub async fn put_deployment( .worker_concurrency .store(n as usize, std::sync::atomic::Ordering::Relaxed); } - // 按模型的限额即时生效：闸门每次调用前读库，发现限额变了就换一把新信号量 if let Some(l) = &req.default_ontology_lang { utopia_store::access::set_default_ontology_lang(&state.pool, l).await?; } + let limits_changed = req.default_model_concurrency.is_some() || req.model_limit.is_some(); if let Some(n) = req.default_model_concurrency { utopia_store::model_limits::set_default(&state.pool, n).await?; } @@ -90,6 +90,14 @@ pub async fn put_deployment( utopia_store::model_limits::set(&state.pool, &m.base_url, &m.model, m.max_concurrent) .await?; } + // 按模型的限额在这里就地生效：闸门调大放行、调小收紧，不等下一个任务来取许可时才读 + // ——那时可能所有 worker 都等在闸门上，唯一在飞的还是一次复读（#1302）。限额已经存下了， + // 调闸门失败只记一笔：下一次取许可照样读到新限额 + if limits_changed { + if let Err(e) = state.model_gates.refresh(&state.pool).await { + tracing::warn!(error = %e, "限额已保存，闸门要等下一次取许可才按新限额调"); + } + } Ok(Json(json!({ "ok": true }))) } diff --git a/crates/utopia-server/src/llm_util.rs b/crates/utopia-server/src/llm_util.rs index 448c71252..15428c630 100644 --- a/crates/utopia-server/src/llm_util.rs +++ b/crates/utopia-server/src/llm_util.rs @@ -81,24 +81,84 @@ pub fn embed_client(s: &LlmSettings) -> Option { )) } -/// 按模型的信号量注册表。限额变了就换一把新的——旧的在飞许可自然跑完， -/// 换的瞬间可能短暂超出新限额，可接受；换来的是"改完即时生效"而不必做 -/// 缓存失效，也不必和 tokio Semaphore 不能缩容的限制搏斗。 +/// 按模型的信号量注册表：每个模型一把，限额变了就地调整（#1302）。 +/// +/// 原来限额一变就换一把新的。等在旧的那把上的任务不会挪过去，每个都要拿旧限额再做 +/// 完一次调用才换到新的上——抽取一次调用几分钟，八个 worker 排在一张许可后面，限额 +/// 从 1 调到 4 之后好几个小时还是一次只跑一个，看起来就是「改了不生效」。 +/// +/// 调高用 `add_permits`，已经在等的立刻放进来。调低时空着的许可当场收走 +/// （`forget_permits`）；正被在飞调用占着的记成欠账，由等着的任务拿到时收掉。没用 +/// 「另起一个任务把差额 acquire 下来再 forget」：信号量先来先得，那个任务排在已经在等的 +/// 任务后面，还回来的许可先发给它们，它们全在调低之后的限额之上进去了，它才拿到。 +/// +/// 限额在两处读：取许可时，和管理员改了限额时（[`ModelGates::refresh`]）。只靠前一处 +/// 的话，所有 worker 都等在闸门上、唯一在飞的又是一次复读的调用时，没人来读新限额 #[derive(Default)] pub struct ModelGates { - inner: std::sync::Mutex)>>, + inner: std::sync::Mutex>, +} + +struct Gate { + sem: Arc, + limit: usize, + /// 调低之后还该收走、当时正被在飞调用占着的许可数 + owed: usize, } impl ModelGates { - fn gate(&self, key: &str, limit: usize) -> Arc { - let mut m = self.inner.lock().unwrap(); - match m.get(key) { - Some((n, sem)) if *n == limit => sem.clone(), - _ => { - let sem = Arc::new(Semaphore::new(limit)); - m.insert(key.to_string(), (limit, sem.clone())); - sem + async fn admit( + &self, + base_url: &str, + model: &str, + limit: usize, + ) -> Option { + let key = (base_url.to_string(), model.to_string()); + let sem = self.resize(&key, limit); + loop { + let permit = sem.clone().acquire_owned().await.ok()?; + if !self.repay(&key) { + return Some(permit); } + permit.forget(); + } + } + + /// 按库里的限额把认识的每个模型调一遍：管理员改了限额就调，等着的任务当场放行或收紧 + pub async fn refresh(&self, pool: &sqlx::PgPool) -> utopia_core::AppResult<()> { + let keys: Vec<(String, String)> = self.inner.lock().unwrap().keys().cloned().collect(); + for key in keys { + let limit = utopia_store::model_limits::limit_for(pool, &key.0, &key.1).await?; + self.resize(&key, limit); + } + Ok(()) + } + + fn resize(&self, key: &(String, String), limit: usize) -> Arc { + let mut gates = self.inner.lock().unwrap(); + let gate = gates.entry(key.clone()).or_insert_with(|| Gate { + sem: Arc::new(Semaphore::new(limit)), + limit, + owed: 0, + }); + if limit > gate.limit { + gate.sem.add_permits(limit - gate.limit); + } else if limit < gate.limit { + let lower = gate.limit - limit; + gate.owed += lower - gate.sem.forget_permits(lower); + } + gate.limit = limit; + gate.sem.clone() + } + + /// 还欠着许可就该收掉刚拿到的这张 + fn repay(&self, key: &(String, String)) -> bool { + match self.inner.lock().unwrap().get_mut(key) { + Some(gate) if gate.owed > 0 => { + gate.owed -= 1; + true + } + _ => false, } } } @@ -119,13 +179,7 @@ pub async fn acquire( let limit = utopia_store::model_limits::limit_for(&state.pool, base_url, model) .await .ok()?; - let key = format!("{base_url}|{model}"); - state - .model_gates - .gate(&key, limit) - .acquire_owned() - .await - .ok() + state.model_gates.admit(base_url, model, limit).await } /// `acquire` 的便捷形式：直接从工作区设置取 chat 模型的身份。 @@ -139,3 +193,7 @@ pub async fn acquire_embed(state: &AppState, s: &LlmSettings) -> Option 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, admin) = (Uuid::now_v7(), Uuid::now_v7()); + let base = format!("http://gate-{}", Uuid::now_v7()); + let dir = tempfile::tempdir()?; + let cfg = utopia_core::config::AppConfig { + data_dir: dir.path().to_string_lossy().into_owned(), + ..Default::default() + }; + let state = crate::state::AppState::new( + pool.clone(), + &cfg, + Arc::new(utopia_search::SearchIndex::open( + &dir.path().join("search"), + )?), + "test-only".into(), + ); + sqlx::raw_sql(&format!( + "INSERT INTO organizations(id,name) VALUES('{org}','model-gate-test'); + INSERT INTO users(id,org_id,email,password_hash,display_name,is_admin) + VALUES('{admin}','{org}','{admin}@example.test','unused','Admin',TRUE);" + )) + .execute(&pool) + .await?; + utopia_store::model_limits::set(&pool, &base, "m", Some(1)).await?; + let result = async { + let _in_flight = super::acquire(&state, &base, "m").await; + let (s, b) = (state.clone(), base.clone()); + let mut waiting = tokio::spawn(async move { super::acquire(&s, &b, "m").await.is_some() }); + let parked = timeout(SOON, &mut waiting).await.is_err(); + + // 这个接口每次都写 open_registration，原样送回去，部署设置那一行不变 + let open = utopia_store::access::open_registration(&pool).await?; + let body = json!({ + "open_registration": open, + "model_limit": {"base_url": base, "model": "m", "max_concurrent": 2}, + }); + let request = Request::builder() + .method("PUT") + .uri("/api/v1/admin/deployment") + .header( + "authorization", + format!("Bearer {}", crate::auth::issue_token(&state, admin)?), + ) + .header("content-type", "application/json") + .body(Body::from(body.to_string()))?; + let status = crate::api::router(state.clone(), &Default::default()) + .oneshot(request) + .await? + .status(); + let admitted = timeout(SOON, &mut waiting).await.is_ok(); + anyhow::ensure!( + status == StatusCode::OK && parked && admitted, + "status: {status}, parked: {parked}, admitted: {admitted}" + ); + anyhow::Ok(()) + } + .await; + utopia_store::model_limits::set(&pool, &base, "m", None).await?; + sqlx::query("DELETE FROM organizations WHERE id=$1") + .bind(org) + .execute(&pool) + .await?; + result +} + +#[tokio::test] +async fn lowering_the_limit_takes_effect_as_in_flight_calls_return() { + let gates = Arc::new(ModelGates::default()); + let first = gates.admit("u", "m", 2).await.unwrap(); + let second = gates.admit("u", "m", 2).await.unwrap(); + let g = gates.clone(); + let mut waiting = tokio::spawn(async move { g.admit("u", "m", 1).await }); + + // 两个在飞降到一个：还回来的第一张被收走，第二个调用还没完就仍满额 + drop(first); + assert!(timeout(SOON, &mut waiting).await.is_err()); + drop(second); + assert!(timeout(SOON, &mut waiting).await.is_ok()); +} diff --git a/crates/utopia-store/src/model_limits.rs b/crates/utopia-store/src/model_limits.rs index 2f30b1dfc..01d3b956c 100644 --- a/crates/utopia-store/src/model_limits.rs +++ b/crates/utopia-store/src/model_limits.rs @@ -13,7 +13,8 @@ use utopia_core::AppResult; /// 这个模型允许多少并发。没有专属配置就走部署缺省。 /// /// 每次 LLM 调用前查一次——相对于一次动辄二十几秒的调用，这个查询的成本可以 -/// 忽略，换来的是"管理员改完即时生效"，不必做缓存失效。 +/// 忽略。闸门手里也存着一份：管理员改限额时 `ModelGates::refresh` 再查一遍、当场 +/// 调闸门（#1302），等在闸门上的任务不必等下一次调用来读。 pub async fn limit_for(pool: &PgPool, base_url: &str, model: &str) -> AppResult { let row: Option<(i32,)> = sqlx::query_as( "SELECT max_concurrent FROM model_concurrency WHERE base_url = $1 AND model = $2", From dd99aedf87519f238192b4c006cb265cb897ad13 Mon Sep 17 00:00:00 2001 From: Szabo Bogdan Date: Mon, 19 Oct 2026 19:57:05 +0200 Subject: [PATCH 03/11] ADR 0071: a source may declare its names identities (#1319) Signed-off-by: Bogdan Szabo Co-authored-by: Claude Opus 5.5 Co-authored-by: WaylandYang <145302500+WaylandYang@users.noreply.github.com> --- ...ay-push-statements-in-the-open-contract.md | 1 + ...source-may-declare-its-names-identities.md | 59 +++++++++++++++++++ docs/decisions/README.md | 1 + 3 files changed, 61 insertions(+) create mode 100644 docs/decisions/0071-a-source-may-declare-its-names-identities.md 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 d1c15b3ba..ecccb4b72 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 @@ -94,3 +94,4 @@ Content-Type: application/json - 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. - 2026-10-15 ([#1242](https://github.com/deeplethe/utopia/issues/1242)): a pushed signature whose phrase equals exactly one declared relation, not a state, that admits both of its ends is bound by the ontology when the document's extraction finishes, and that extraction queues `materialize`, so a base with no chat model types what it is pushed. It binds the signature as [0044](0044-the-ontology-is-a-view-over-what-documents-say.md)'s revision of decision 3 describes, reviewable like any binding; it is not a typed write, decisions 3 and 5 stand, and no other pushed phrase binds before alignment reads it. - 2026-10-16 ([#1267](https://github.com/deeplethe/utopia/issues/1267)): decision 7 keeps a re-pushed statement's identity, but its role words stayed first-write-wins, so a re-push with a new qualifier value kept the old one; [0070](0070-a-role-word-speaks-while-its-chunk-is-live.md) proposes that a role word is recorded against its evidence chunk and speaks while that chunk is live. +- 2026-10-19 ([#1293](https://github.com/deeplethe/utopia/issues/1293)): proposed in [0071](0071-a-source-may-declare-its-names-identities.md): a `statements` source whose config says `{"names": "identity"}` attaches its named entities by exact name and gives their names no vectors. Decision 3's "same path" stands for every other source. diff --git a/docs/decisions/0071-a-source-may-declare-its-names-identities.md b/docs/decisions/0071-a-source-may-declare-its-names-identities.md new file mode 100644 index 000000000..da59eb112 --- /dev/null +++ b/docs/decisions/0071-a-source-may-declare-its-names-identities.md @@ -0,0 +1,59 @@ +# 0071 · A source may declare its names identities + +- **Status**: Proposed · 2026-10-19 · [#1293](https://github.com/deeplethe/utopia/issues/1293) · open: all six decisions +- **Written**: 2026-10-19 (conventions in the [README](README.md)) +- **Related**: [0054](0054-a-source-may-push-statements-in-the-open-contract.md) decision 3 sends pushed statements down the same resolution path as prose, and this record narrows that for one kind of source; [0041](0041-a-name-is-a-claim-about-an-entity.md) owns name facts and name-vector recall; [#1248](https://github.com/deeplethe/utopia/issues/1248) meets the same problem from the merge side; [#1296](https://github.com/deeplethe/utopia/issues/1296) measured what the recall costs; [#1308](https://github.com/deeplethe/utopia/pull/1308) is the disjointness fix that stands on its own + +> A backend mirrors a repository into a base and pushes `owner/repo:src/a.ts#parseDate`, `owner/repo:src/a.test.ts::adds two`, `owner/repo:specs/x/spec.md#s0003`. Each name is an identity the client already knows to be unique. Every one still goes through write-time resolution. Because they share the `owner/repo:` prefix, their name vectors sit close together, and the duplicates queue fills with pairs that can never be the same thing. + +## The problem + +0054 decision 3 is right for most pushes. A client that pushes human-style names, like `cup-7` and `kitchen table`, gains from name-vector recall exactly as prose does. What is true for some producers is narrower: *this source's* names are identities. For such a source, resolution today does three things it already knows the answer to: + +- **Name vectors** (`propose_by_name_vector`). Up to eight `name_vector|` pairs per mention, plus one `name_vectors::nearest` query per named entity per pushed document, even after an exact match (#1296). +- **The single-candidate split** (`resolve_by_name`, cosine below `SIM_ATTACH`). The same file named in two payloads with different context becomes two entities with one name. +- **Being a target.** Its names' vectors are near-neighbours for every other mention in the base, so prose mentions propose pairs against them too. + +## Decisions + +**1. The switch lives on the source.** + +A `statements` source whose config says `{"names": "identity"}`, next to the existing `{"extract": false}`, declares that every name it pushes identifies one thing. The default is unchanged, and every other source keeps 0054 decision 3 as it stands. The producer is the one that knows its names are identities, and one base can hold an identity source beside prose sources, so the switch is per source rather than per base. #1248 decides the merge side with the same flag. + +**2. An identity name attaches to its exact name, whatever the profile.** + +A named entity from an identity source attaches to the entity whose name equals it case-insensitively, in the declared class or with no class, and is created only when there is none. The profile similarity is not consulted, so the single-candidate split never happens for an identity source: the same name in two payloads is one entity. When the exact name already has two holders, which should be rare for identity names, the mention takes today's review path. Choosing between them would need a provenance lookup that this record does not add. + +**3. A name that exists only under another class is created in the declared class.** + +The source's class is part of its identity claim, so the entity is created in that class. The type-drift review stays, unless the two classes are declared disjoint (0016 B3): an exact name under another class is most likely a duplicate whose class was guessed, and it deserves one look. + +**4. An identity name has no name vector.** + +A mention from an identity source runs no `name_vectors::nearest` query and proposes no `name_vector|` pair. Its name facts get no vector either: `name_vectors::pending` skips a name fact that has evidence from an identity source (through `fact_evidence.document_id` to the document's source). So identity entities neither propose pairs nor become targets of one. A prose mention of the exact name still attaches by exact match (channel 1 of 0041 decision 3), so nothing that an exact name finds is lost. + +**5. Identity evidence wins.** + +One name fact can carry evidence from both an identity source and a prose document, because facts dedupe on subject, predicate and object. Such a fact gets no vector. The identity source has vouched for the name, and a prose mention still reaches it by exact match. This changes recall for prose: a prose mention of a *variant* of that name (an abbreviation, another script) no longer finds the entity through its vector, and the code PR covers that with the test `a_name_with_identity_evidence_gets_no_vector_even_when_prose_names_it_too`. + +**6. Switching the flag.** + +- **On:** the update that sets the flag also deletes the stored vectors of name facts that now have identity evidence, in the same transaction. Leaving them in place and skipping them would put an evidence join into every `nearest` query, which is the per-mention cost #1296 measured. +- **Off:** nothing is deleted. Those name facts become pending again, and the next extraction in the base backfills their vectors through `pending`, as it does for any name whose vector is missing. Recall for them returns from then on. Pairs that were never proposed while the flag was on are not proposed in retrospect. + +## Not doing + +- **Containment for identity names.** `containment_reviews` still runs for them. #1104's next cut rewrites that predicate, and whether an identity source skips it is a separate decision after that. +- **The single-candidate split for other sources.** Decision 2 removes it for identity sources only. Prose keeps today's split. + +## Dead ends + +- **Every `statements` push is an identity claim.** Too broad: 0054 sends pushes through the same path on purpose, and human-style names find real variants through their vectors. +- **Skip the vector channel after any single exact match.** That is #1296 item 1. Applied to every mention it changes prose resolution, and `the_same_literal_name_is_not_proposed_twice` asserts the opposite. +- **A per-base switch.** One base can mix an identity source with prose sources, and only the producer knows which names are identities. + +## Tests the code PR adds + +- `an_identity_source_adds_no_name_vectors_and_asks_no_neighbours`: a push from an identity source leaves no `name_vector|` rows and runs no `nearest` query. +- `a_name_with_identity_evidence_gets_no_vector_even_when_prose_names_it_too` (decision 5). +- Turning the flag on deletes the identity names' vectors; turning it off makes them pending again (decision 6). diff --git a/docs/decisions/README.md b/docs/decisions/README.md index f4ef89cb4..727d60be0 100644 --- a/docs/decisions/README.md +++ b/docs/decisions/README.md @@ -55,6 +55,7 @@ By domain; the domains are the files of [../design/](../design/README.md). **Sta | | Record | Status | Overtaken | |---|---|---|---| | 0041 | [A name is a claim about an entity](0041-a-name-is-a-claim-about-an-entity.md) | In progress | | +| 0071 | [A source may declare its names identities](0071-a-source-may-declare-its-names-identities.md) | Proposed | | ### [time](../design/time.md) From cd14e9578ae5b79563f75bac03b568b39ae405cc Mon Sep 17 00:00:00 2001 From: Szabo Bogdan Date: Mon, 19 Oct 2026 20:02:24 +0200 Subject: [PATCH 04/11] A token reads a base's graph overview and neighbourhood (#1264) Signed-off-by: Bogdan Szabo Co-authored-by: Claude Opus 5.5 Co-authored-by: WaylandYang <145302500+WaylandYang@users.noreply.github.com> --- crates/utopia-server/src/api/graph_routes.rs | 8 +++---- .../utopia-server/src/api/token_read_tests.rs | 11 +++++----- crates/utopia-server/src/auth.rs | 3 ++- ...ty-from-the-person-scope-from-the-token.md | 21 ++++++++++++------- docs/design/access-and-audit.md | 6 +++--- 5 files changed, 28 insertions(+), 21 deletions(-) diff --git a/crates/utopia-server/src/api/graph_routes.rs b/crates/utopia-server/src/api/graph_routes.rs index 3116474a2..8eb3e165f 100644 --- a/crates/utopia-server/src/api/graph_routes.rs +++ b/crates/utopia-server/src/api/graph_routes.rs @@ -71,11 +71,11 @@ pub struct OverviewQuery { pub async fn overview( State(state): State, - AuthUser(user): AuthUser, + reader: SessionOrToken, Path(kb_id): Path, Query(q): Query, ) -> ApiResult> { - require_kb(&state, &user, kb_id, Role::Viewer).await?; + require_reader_kb(&state, &reader, kb_id).await?; let at = parse_instant("at", q.at.as_deref())?; let as_of = parse_instant("as_of", q.as_of.as_deref())?; // 画多少个是渲染的事，库里有多少是知识库的事——两个数都回，界面才说得出 @@ -105,11 +105,11 @@ pub struct NeighborhoodQuery { pub async fn neighborhood( State(state): State, - AuthUser(user): AuthUser, + reader: SessionOrToken, Path(kb_id): Path, Query(q): Query, ) -> ApiResult> { - require_kb(&state, &user, kb_id, Role::Viewer).await?; + require_reader_kb(&state, &reader, kb_id).await?; let at = parse_instant("at", q.at.as_deref())?; let as_of = parse_instant("as_of", q.as_of.as_deref())?; let (nodes, edges) = utopia_store::graph::neighborhood( diff --git a/crates/utopia-server/src/api/token_read_tests.rs b/crates/utopia-server/src/api/token_read_tests.rs index 1855c84f1..b9d52d525 100644 --- a/crates/utopia-server/src/api/token_read_tests.rs +++ b/crates/utopia-server/src/api/token_read_tests.rs @@ -1,4 +1,4 @@ -//! 个人令牌读实体（0014 决定 7）：搜实体、实体详情、认知历史、事实证据四条读路由 +//! 个人令牌读图（0014 决定 7）：搜实体、实体详情、认知历史、事实证据、图总览、邻域六条读路由 //! 收任何一枚活着的令牌。角色照旧是上限，`kb_ids` 排除了的库答 404；改实体仍只认会话。 use crate::state::AppState; use axum::body::Body; @@ -81,13 +81,15 @@ impl Fixture { .1 } - fn read_paths(&self) -> [String; 4] { + fn read_paths(&self) -> [String; 6] { let (kb, entity, fact) = (self.kb, self.entity, self.fact); [ format!("/kbs/{kb}/entities?q=Zhang"), format!("/kbs/{kb}/entities/{entity}"), format!("/kbs/{kb}/entities/{entity}/history"), format!("/kbs/{kb}/facts/{fact}/evidence"), + format!("/kbs/{kb}/graph/overview"), + format!("/kbs/{kb}/graph/neighborhood?entity={entity}"), ] } @@ -109,8 +111,7 @@ impl Fixture { } #[tokio::test] -async fn a_read_token_of_a_base_viewer_reads_its_entities_facts_and_history() -> anyhow::Result<()> -{ +async fn a_read_token_of_a_base_viewer_gets_200_on_every_graph_read() -> anyhow::Result<()> { let Some(f) = Fixture::new().await? else { return Ok(()); }; @@ -128,7 +129,7 @@ async fn a_read_token_of_a_base_viewer_reads_its_entities_facts_and_history() -> } #[tokio::test] -async fn a_token_limited_to_another_base_gets_404_on_every_entity_read() -> anyhow::Result<()> { +async fn a_token_limited_to_another_base_gets_404_on_every_graph_read() -> anyhow::Result<()> { let Some(f) = Fixture::new().await? else { return Ok(()); }; diff --git a/crates/utopia-server/src/auth.rs b/crates/utopia-server/src/auth.rs index 5a7889866..f091f4f91 100644 --- a/crates/utopia-server/src/auth.rs +++ b/crates/utopia-server/src/auth.rs @@ -176,7 +176,8 @@ impl FromRequestParts for AuthUser { /// /// **不替换 `AuthUser`。** `AuthUser` 把每个 Bearer 都当 JWT 解，令牌在那几百条 /// 路由上一律是 401——那是有意的：哪条路由准令牌进来，要在 0014 里说出理由。 -/// 用这个提取器的只有文档原件（0052）、「把一个库搭起来」那几条和读实体那几条（决定 7）。 +/// 用这个提取器的只有文档原件（0052）、「把一个库搭起来」那几条和读图那几条：搜实体、 +/// 实体详情、认知历史、事实证据、图总览、邻域（决定 7）。 /// /// 令牌只收窄：`pat` 在的时候，路由先问 [`Self::require_write`] 或 /// [`Self::reaches`]，角色照旧由 `require_kb` 与工作区角色判。 diff --git a/docs/decisions/0014-identity-from-the-person-scope-from-the-token.md b/docs/decisions/0014-identity-from-the-person-scope-from-the-token.md index 58cff2554..850e9015b 100644 --- a/docs/decisions/0014-identity-from-the-person-scope-from-the-token.md +++ b/docs/decisions/0014-identity-from-the-person-scope-from-the-token.md @@ -87,14 +87,18 @@ push documents in. 7. **A token reads the graph** (added 2026-10-16, #1262). A service that mirrors into a base reads the base back to check what landed, and a session JWT is the wrong credential for a backend - (decision 6). Four reads now also take any live token, acting as its owner: searching a base's + (decision 6). Six reads now also take any live token, acting as its owner: searching a base's entities (`GET /kbs/{id}/entities`), an entity's detail (`GET /kbs/{id}/entities/{eid}`), its - record-axis history (`…/history`), and a fact's evidence (`GET /kbs/{id}/facts/{fid}/evidence`). - Search finds the entity by name; the detail gives its facts, names and derivations; the - history says when we changed our mind; the evidence ties a fact to its sentence. Read is the - lowest scope, so any token qualifies; the owner's role is still the ceiling and Viewer - suffices; a token whose `kb_ids` exclude the base gets 404, as for document content (0052). - Changing an entity on the same path (`PATCH`) still takes a session. + record-axis history (`…/history`), a fact's evidence (`GET /kbs/{id}/facts/{fid}/evidence`), + the graph overview (`GET /kbs/{id}/graph/overview`) and an entity's neighbourhood + (`GET /kbs/{id}/graph/neighborhood`). Search finds the entity by name; the detail gives its + facts, names and derivations; the history says when we changed our mind; the evidence ties a + fact to its sentence; the overview gives the most connected entities with the base's totals, + so a client sees the shape of what it pushed; the neighbourhood gives the entities around one, + so it can follow what a statement touches, and one started from another base's entity comes + back empty. Read is the lowest scope, so any token qualifies; the owner's role is still the + ceiling and Viewer suffices; a token whose `kb_ids` exclude the base gets 404, as for document + content (0052). Changing an entity on the same path (`PATCH`) still takes a session. ## Dead ends @@ -121,7 +125,8 @@ push documents in. push-only as well; a backend mirroring several kinds of documents pushes them through one. A token now creates `api` sources too. - 2026-10-16: decision 7 opens four graph reads to any token; decision 6's "only these routes - and document content" now also counts them. + and document content" now also counts them. 2026-10-19: the graph overview and neighbourhood + join them. ## Open questions diff --git a/docs/design/access-and-audit.md b/docs/design/access-and-audit.md index 70770528c..86e533c38 100644 --- a/docs/design/access-and-audit.md +++ b/docs/design/access-and-audit.md @@ -19,9 +19,9 @@ writes an audit row `mcp.tool_called`; the tokens page shows the plaintext once snippet [0014]. A `write` token also sets up a base as its owner: creating a base and a `statements` or `api` source, reading the source's push key, importing an ontology and deciding a kind word, with the owner's role unchanged as the ceiling; any token lists a workspace's bases; a token limited by `kb_ids` -creates no base. These routes, document content and four graph reads (entity search, detail, -history, fact evidence; Viewer suffices, `kb_ids` answers 404) take `SessionOrToken`; apart from -MCP, every other route is session-only [0014 d6, d7]. `ingest_token` on a source is a different kind: plaintext, push-only [0014]. +creates no base. These routes, document content and six graph reads (entity search, detail, +history, fact evidence, overview, neighbourhood; Viewer suffices, `kb_ids` answers 404) take +`SessionOrToken`; apart from MCP, every other route is session-only [0014 d6, d7]. `ingest_token` on a source is a different kind: plaintext, push-only [0014]. **Another app can trade its user's fresh ID token for a session.** `POST /auth/oidc/exchange` accepts an ID token from the configured issuer whose audience is in From 86b1af8b59372b120e5ad55dbc18ca7d91222db1 Mon Sep 17 00:00:00 2001 From: Szabo Bogdan Date: Mon, 19 Oct 2026 20:08:00 +0200 Subject: [PATCH 05/11] Finding a base's name facts without vectors reads only its name facts (#1318) Signed-off-by: Bogdan Szabo Co-authored-by: Claude Opus 5.5 Co-authored-by: WaylandYang <145302500+WaylandYang@users.noreply.github.com> --- crates/utopia-cli/src/main.rs | 2 +- crates/utopia-store/src/name_vectors.rs | 41 ++++++----- .../src/name_vectors_plan_tests.rs | 70 +++++++++++++++++++ .../a_similar_name_is_proposed_not_merged.rs | 36 ++++++++++ ..._a_name_fact_is_found_by_its_predicate.sql | 10 +++ 5 files changed, 142 insertions(+), 17 deletions(-) create mode 100644 crates/utopia-store/src/name_vectors_plan_tests.rs create mode 100644 migrations/0115_a_name_fact_is_found_by_its_predicate.sql diff --git a/crates/utopia-cli/src/main.rs b/crates/utopia-cli/src/main.rs index c88fc1b7c..2d09f0936 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 = 102; +const CURRENT_SCHEMA_VERSION: u32 = 103; fn main() -> anyhow::Result<()> { dotenvy::dotenv().ok(); diff --git a/crates/utopia-store/src/name_vectors.rs b/crates/utopia-store/src/name_vectors.rs index 22bfa3cbd..c61f08ab9 100644 --- a/crates/utopia-store/src/name_vectors.rs +++ b/crates/utopia-store/src/name_vectors.rs @@ -25,6 +25,21 @@ pub const TOP_K: usize = 8; /// 0.4 上下的余弦——所以这条线比画像的 `SIM_ATTACH` 高。同样是待测量的临时值 pub const SIM_FLOOR: f32 = 0.60; +/// `known_as` 的 id 用子查询取，不要改回 JOIN：子查询先算出一个值，`predicate_id = 它` 才成了 +/// `facts_live_predicate_idx`（0115）的索引条件，扫描只碰这个库的名字事实，且仍按 id 走， +/// `ORDER BY ... LIMIT` 不用排序。写成 JOIN，规划器沿 `facts_kb_id_key` 把整个库按 id 读一遍， +/// 谓词只能逐行筛（#1296：20 万条事实的库上读 20 万行，这样读 2 万行） +const PENDING_SQL: &str = "SELECT f.id, f.subject_id, f.object_value->>'value' + FROM facts f + LEFT JOIN name_vectors v ON v.fact_id = f.id + WHERE f.kb_id = $1 + AND f.predicate_id = (SELECT nr.id FROM relation_types nr + WHERE nr.kb_id = $1 AND nr.builtin AND nr.key = $2) + AND f.invalidated_at IS NULL AND f.object_value IS NOT NULL + AND v.fact_id IS NULL + ORDER BY f.id + LIMIT $3"; + /// 还没有向量的名字事实：现行的、`known_as` 上的、`name_vectors` 里没有它的。 /// 按写入先后取（`facts.id` 是 uuid v7，按它排即按写入排；这张表没有 created_at—— /// 端到端跑出来的：按一个不存在的列排，补向量每篇都静默失败），一次取一批 @@ -33,22 +48,12 @@ pub async fn pending( kb_id: Uuid, limit: i64, ) -> AppResult> { - let rows: Vec<(Uuid, Uuid, String)> = sqlx::query_as( - "SELECT f.id, f.subject_id, f.object_value->>'value' - FROM facts f - JOIN relation_types nr ON nr.id = f.predicate_id - LEFT JOIN name_vectors v ON v.fact_id = f.id - WHERE f.kb_id = $1 AND nr.kb_id = $1 AND nr.builtin AND nr.key = $2 - AND f.invalidated_at IS NULL AND f.object_value IS NOT NULL - AND v.fact_id IS NULL - ORDER BY f.id - LIMIT $3", - ) - .bind(kb_id) - .bind(crate::names::KNOWN_AS) - .bind(limit) - .fetch_all(pool) - .await?; + let rows: Vec<(Uuid, Uuid, String)> = sqlx::query_as(PENDING_SQL) + .bind(kb_id) + .bind(crate::names::KNOWN_AS) + .bind(limit) + .fetch_all(pool) + .await?; Ok(rows) } @@ -135,3 +140,7 @@ pub async fn nearest(pool: &PgPool, kb_id: Uuid, query: &[f32], k: usize) -> App .await?; Ok(rows) } + +#[cfg(test)] +#[path = "name_vectors_plan_tests.rs"] +mod plan_tests; diff --git a/crates/utopia-store/src/name_vectors_plan_tests.rs b/crates/utopia-store/src/name_vectors_plan_tests.rs new file mode 100644 index 000000000..651522f77 --- /dev/null +++ b/crates/utopia-store/src/name_vectors_plan_tests.rs @@ -0,0 +1,70 @@ +//! 补向量每篇文档都跑一次 `pending`（#1296）：它得以谓词为索引条件走 `facts_live_predicate_idx`， +//! 只读这个库的名字事实、按 id 顺序读，不用排序。改回 JOIN 时规划器会沿 `facts_kb_id_key` 读整个库。 +//! +//! 规划器按统计选索引：库是空的时候几个索引代价一样，选哪个看运气（别的测试同时在写 facts， +//! 选择就会翻）。所以这里在回滚的事务里造一个两千条事实、十分之一是名字的库并 ANALYZE， +//! 比的是有数据时它走哪条路。 +use super::PENDING_SQL; +use sqlx::PgPool; +use uuid::Uuid; + +#[tokio::test] +async fn pending_reads_the_name_facts_through_the_predicate_index() -> anyhow::Result<()> { + let Some(url) = crate::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + let (org, ws, kb) = (Uuid::now_v7(), Uuid::now_v7(), Uuid::now_v7()); + let (named, other, thing) = (Uuid::now_v7(), Uuid::now_v7(), Uuid::now_v7()); + let mut tx = pool.begin().await?; + for (sql, ids) in [ + ("SET LOCAL enable_seqscan = off", vec![]), + ( + "INSERT INTO organizations (id, name) VALUES ($1, 'plan')", + vec![org], + ), + ( + "INSERT INTO workspaces (id, org_id, name) VALUES ($1, $2, 'plan')", + vec![ws, org], + ), + ( + "INSERT INTO knowledge_bases (id, workspace_id, name) VALUES ($1, $2, 'plan')", + vec![kb, ws], + ), + ( + "INSERT INTO relation_types (id, kb_id, key, label, builtin) + VALUES ($1, $3, 'known_as', 'known as', true), ($2, $3, 'mentions', 'mentions', false)", + vec![named, other, kb], + ), + ( + "INSERT INTO entities (id, kb_id, canonical_name) VALUES ($1, $2, 'thing')", + vec![thing, kb], + ), + ( + "INSERT INTO facts (id, kb_id, subject_id, predicate_id, object_value) + SELECT gen_random_uuid(), $1, $2, CASE WHEN n % 10 = 0 THEN $3 ELSE $4 END, + jsonb_build_object('value', 'v' || n) + FROM generate_series(1, 2000) n", + vec![kb, thing, named, other], + ), + ("ANALYZE facts", vec![]), + ] { + let mut q = sqlx::query(sql); + for id in ids { + q = q.bind(id); + } + q.execute(&mut *tx).await?; + } + let plan: Vec = sqlx::query_scalar(&format!("EXPLAIN {PENDING_SQL}")) + .bind(kb) + .bind(crate::names::KNOWN_AS) + .bind(64_i64) + .fetch_all(&mut *tx) + .await?; + tx.rollback().await?; + + let plan = plan.join("\n"); + assert!(plan.contains("facts_live_predicate_idx"), "{plan}"); + assert!(!plan.contains("Sort"), "{plan}"); + Ok(()) +} diff --git a/crates/utopia-store/tests/store/a_similar_name_is_proposed_not_merged.rs b/crates/utopia-store/tests/store/a_similar_name_is_proposed_not_merged.rs index b2c6dd361..09b3a74c8 100644 --- a/crates/utopia-store/tests/store/a_similar_name_is_proposed_not_merged.rs +++ b/crates/utopia-store/tests/store/a_similar_name_is_proposed_not_merged.rs @@ -333,6 +333,42 @@ async fn a_new_name_is_pending_until_its_vector_is_set() -> anyhow::Result<()> { run } +/// 上一篇文档没补上的名字，下一篇照样补：`pending` 取整个库里没有向量的名字，先记的在前， +/// 不只是这篇刚记的。有向量的名字不在里面（#1296 改这条查询时守住这次重试） +#[tokio::test] +async fn a_name_left_without_a_vector_earlier_is_pending_before_a_new_one() -> anyhow::Result<()> { + let Some(url) = utopia_store::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + let f = seed(&pool, "name-vector-backlog").await?; + let run = async { + let mut recorded = Vec::new(); + for name in ["远洋重工", "深蓝船舶"] { + let entity = Uuid::now_v7(); + sqlx::query( + "INSERT INTO entities (id, kb_id, type_id, canonical_name) VALUES ($1, $2, $3, $4)", + ) + .bind(entity) + .bind(f.kb) + .bind(f.device) + .bind(name) + .execute(&pool) + .await?; + let fact = names::record(&pool, f.kb, entity, name, None, None) + .await? + .expect("a name fact"); + recorded.push((fact, entity, name.to_string())); + } + + assert_eq!(name_vectors::pending(&pool, f.kb, 100).await?, recorded); + Ok::<_, anyhow::Error>(()) + } + .await; + teardown(&pool, &f).await?; + run +} + /// 判过「不是一个」的一对，实体再被提到时不再排（#1104）。名字向量通道每次提到都跑， /// 而审核表只对 pending 的行去重：从前裁决器分开过的一对，下一篇文档一提到就又是一条 /// 新的 pending。还没判的那一对不受影响，照常提议 diff --git a/migrations/0115_a_name_fact_is_found_by_its_predicate.sql b/migrations/0115_a_name_fact_is_found_by_its_predicate.sql new file mode 100644 index 000000000..d7d9ec602 --- /dev/null +++ b/migrations/0115_a_name_fact_is_found_by_its_predicate.sql @@ -0,0 +1,10 @@ +-- 按谓词找现行的名字事实（#1296）。 +-- +-- 补向量每篇文档都跑一次 `name_vectors::pending`：现行的、`known_as` 上的、还没有向量的名字 +-- 事实，按 id 取一批。从前没有带谓词的索引，它只能沿 `facts_kb_id_key` 把整个库的事实按 id +-- 读一遍再筛谓词——20 万条事实的库里每篇文档读 20 万行，为的是剩下的三五个名字。 +-- +-- `id` 列让索引按谓词定位之后仍按 id 排好，`ORDER BY f.id LIMIT` 顺着走、不用排序，只读这个库 +-- 的名字事实。谓词在查询里是一个子查询的值，见 `pending` 上的说明。 +CREATE INDEX IF NOT EXISTS facts_live_predicate_idx ON facts (kb_id, predicate_id, id) + WHERE invalidated_at IS NULL; From b6099e42458831167354c6a78edfbc51a400f2ce Mon Sep 17 00:00:00 2001 From: Szabo Bogdan Date: Mon, 19 Oct 2026 20:13:14 +0200 Subject: [PATCH 06/11] Containment matches a name's % and _ literally instead of as wildcards (#1320) Signed-off-by: Bogdan Szabo Co-authored-by: Claude Opus 5.5 Co-authored-by: WaylandYang <145302500+WaylandYang@users.noreply.github.com> --- crates/utopia-store/src/resolution.rs | 21 +++++-- .../a_similar_name_is_proposed_not_merged.rs | 58 ++++++++++++++++--- 2 files changed, 68 insertions(+), 11 deletions(-) diff --git a/crates/utopia-store/src/resolution.rs b/crates/utopia-store/src/resolution.rs index 5cd4502b8..d51ea1b15 100644 --- a/crates/utopia-store/src/resolution.rs +++ b/crates/utopia-store/src/resolution.rs @@ -721,6 +721,13 @@ const MAX_CONTAIN_REVIEWS: usize = 4; /// 只取 4 行的话可能 4 行全是互斥类型，真正那一对反而被 LIMIT 切掉。 const CONTAIN_SCAN_LIMIT: i64 = 16; +/// 名字当 LIKE 的字面子串用：不转义的话名字里的 `_` 会匹配任意一个字（#1293） +fn like_escaped(name: &str) -> String { + name.replace('\\', "\\\\") + .replace('%', "\\%") + .replace('_', "\\_") +} + /// 新建实体之后，找出**名字互相包含**的既有实体，作为审阅候选。 /// /// 中文商业文本在一篇之内就会全称转简称（「星云科技上海研究院」→「上海研究院」）， @@ -792,8 +799,8 @@ async fn containment_reviews( AND lower(e.canonical_name) <> $3 AND ( (char_length(e.canonical_name) >= $4 - AND (lower(e.canonical_name) LIKE '%' || $3 || '%' - OR $3 LIKE '%' || lower(e.canonical_name) || '%')) + AND (lower(e.canonical_name) LIKE '%' || $6 || '%' + OR $3 LIKE '%' || replace(replace(replace(lower(e.canonical_name), '\\', '\\\\'), '%', '\\%'), '_', '\\_') || '%')) -- **别的名字也要参与召回，否则每合并一次就少一条桥。** -- -- 合并把名字事实搬到存活者身上：Holmes 并入 Sherlock Holmes 之后， @@ -810,8 +817,8 @@ async fn containment_reviews( AND nr.builtin AND nr.key = 'known_as' AND nf.invalidated_at IS NULL AND char_length(nf.object_value->>'value') >= $4 - AND (lower(nf.object_value->>'value') LIKE '%' || $3 || '%' - OR $3 LIKE '%' || lower(nf.object_value->>'value') || '%')) + AND (lower(nf.object_value->>'value') LIKE '%' || $6 || '%' + OR $3 LIKE '%' || replace(replace(replace(lower(nf.object_value->>'value'), '\\', '\\\\'), '%', '\\%'), '_', '\\_') || '%')) ) ORDER BY char_length(e.canonical_name) LIMIT $5", @@ -821,6 +828,7 @@ async fn containment_reviews( .bind(&lower) .bind(MIN_CONTAIN_CHARS) .bind(CONTAIN_SCAN_LIMIT) + .bind(like_escaped(&lower)) .fetch_all(pool) .await?; @@ -2535,6 +2543,11 @@ pub async fn unadopt_types(pool: &PgPool, kb_id: Uuid, batch_id: Uuid) -> AppRes mod tests { use super::*; + #[test] + fn like_escaped_turns_backslash_percent_and_underscore_into_literals() { + assert_eq!(like_escaped(r"a\b%c_d"), r"a\\b\%c\_d"); + } + #[test] fn normalize_fullwidth_and_whitespace() { assert_eq!(normalize_name("　ＡＣＭＥ Corp　"), "ACME Corp"); diff --git a/crates/utopia-store/tests/store/a_similar_name_is_proposed_not_merged.rs b/crates/utopia-store/tests/store/a_similar_name_is_proposed_not_merged.rs index 09b3a74c8..f013c07b1 100644 --- a/crates/utopia-store/tests/store/a_similar_name_is_proposed_not_merged.rs +++ b/crates/utopia-store/tests/store/a_similar_name_is_proposed_not_merged.rs @@ -102,6 +102,14 @@ fn vector_reviews(r: &utopia_store::resolution::Resolution) -> Vec<(Uuid, f32)> .collect() } +fn contained(r: &utopia_store::resolution::Resolution) -> Vec { + r.reviews + .iter() + .filter(|v| v.reason.starts_with("contains|")) + .map(|v| v.other_id) + .collect() +} + /// 相近的名字：新建实体，给裁决器排一对，绝不静默归并 #[tokio::test] async fn a_near_name_creates_an_entity_and_proposes_a_pair() -> anyhow::Result<()> { @@ -483,13 +491,6 @@ async fn names_that_carry_different_numbers_are_not_proposed() -> anyhow::Result ); // 包含关系：`migration 39` 是 `migration 395` 的子串，但不是它的简称 - let contained = |r: &utopia_store::resolution::Resolution| { - r.reviews - .iter() - .filter(|v| v.reason.starts_with("contains|")) - .map(|v| v.other_id) - .collect::>() - }; let long = resolve("migration 395", None).await?; let short = resolve("migration 39", None).await?; assert!(short.created); @@ -512,3 +513,46 @@ async fn names_that_carry_different_numbers_are_not_proposed() -> anyhow::Result teardown(&pool, &f).await?; run } + +/// 名字里的 `_` 按字面比，不当通配符（#1293）：`src/a_b.ts` 不包含在 `src/axb.ts::adds` 里， +/// 反过来 `a_b.ts` 也不在后者里；真包含的 `src/a_b.ts` 照常配对 +#[tokio::test] +async fn an_underscore_in_a_name_is_not_a_wildcard() -> anyhow::Result<()> { + let Some(url) = utopia_store::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + let f = seed(&pool, "contain-underscore").await?; + let run = async { + let resolve = |name: &'static str| { + utopia_store::resolution::resolve_mention( + &pool, + f.kb, + Some(f.device), + name, + None, + None, + None, + &[], + ) + }; + let file = resolve("src/a_b.ts").await?; + let other = resolve("src/axb.ts::adds").await?; + assert!( + !contained(&other).contains(&file.entity_id), + "库里名字的 _ 不匹配 x：{:?}", + other.reviews + ); + let brief = resolve("a_b.ts").await?; + assert_eq!( + contained(&brief), + vec![file.entity_id], + "mention 的 _ 不匹配 x，真包含的照常配对：{:?}", + brief.reviews + ); + Ok::<_, anyhow::Error>(()) + } + .await; + teardown(&pool, &f).await?; + run +} From 8750760e6f2aa3046038d555cab6552a8fb12f29 Mon Sep 17 00:00:00 2001 From: Nash Aschenbath Date: Tue, 20 Oct 2026 02:18:54 +0800 Subject: [PATCH 07/11] fix: keep Markdown name variants on the same resolution paths (#1299) Signed-off-by: Aschenbath Co-authored-by: Aschenbath Co-authored-by: WaylandYang <145302500+WaylandYang@users.noreply.github.com> --- crates/utopia-server/src/extraction.rs | 4 +- crates/utopia-server/src/extraction_open.rs | 12 +- crates/utopia-store/src/names.rs | 3 +- crates/utopia-store/src/resolution.rs | 108 ++++++++++++++++-- .../a_similar_name_is_proposed_not_merged.rs | 50 +++++--- .../store/an_undeclared_name_is_looked_up.rs | 98 ++++++++++++++++ 6 files changed, 234 insertions(+), 41 deletions(-) diff --git a/crates/utopia-server/src/extraction.rs b/crates/utopia-server/src/extraction.rs index 7894eb6de..6b7ae44fb 100644 --- a/crates/utopia-server/src/extraction.rs +++ b/crates/utopia-server/src/extraction.rs @@ -274,7 +274,7 @@ pub(crate) async fn resolve_handle( needs_adjudication: &mut bool, human_reviews_found: &mut bool, ) -> anyhow::Result { - let normalized = utopia_store::resolution::normalize_name(name).to_lowercase(); + let normalized = utopia_store::resolution::name_key(name); let response_excluded = response_claims .get(&normalized) .cloned() @@ -584,7 +584,7 @@ mod tests { &pool, kb, Some(person), - "Zhang Wei", + "**Zhang Wei**", None, None, None, diff --git a/crates/utopia-server/src/extraction_open.rs b/crates/utopia-server/src/extraction_open.rs index 0a4fa9484..6dc1aee38 100644 --- a/crates/utopia-server/src/extraction_open.rs +++ b/crates/utopia-server/src/extraction_open.rs @@ -34,6 +34,7 @@ use utopia_store::extraction_drops::reason; use utopia_store::graph::FactObject; use utopia_store::graph::Validity; use utopia_store::pending::{Outcome, Proposal}; +pub(crate) use utopia_store::resolution::name_key; use uuid::Uuid; /// 一块调 + 解析完后的中间产物，apply 半步读它落库（#588 cut 1）。 @@ -298,7 +299,7 @@ async fn apply_open_extraction( ) .await; } - touched_names.insert(utopia_store::resolution::normalize_name(name).to_lowercase()); + touched_names.insert(name_key(name)); if !doc_entities.iter().any(|(x, _, _)| *x == id) { doc_entities.push((id, kind.to_string(), name.to_string())); known_by_name.entry(key.clone()).or_insert(id); @@ -661,7 +662,7 @@ async fn apply_open_extraction( } // 一个名字不会同时是两样东西的名字：这一块或本文档前面已经把它认成 // 另一个实体的，不记 - let key = utopia_store::resolution::normalize_name(name).to_lowercase(); + let key = name_key(name); let claimed = handled_by_name .get(&key) .is_some_and(|ids| ids.iter().any(|x| *x != id)); @@ -895,13 +896,6 @@ async fn embed_pending_names( Ok(items.len()) } -pub(crate) fn name_key(name: &str) -> String { - name.split_whitespace() - .collect::>() - .join(" ") - .to_lowercase() -} - /// 陈述里写的名字 → 实体。先看这一块列出的，再看本文档前面认下的（提示词里的清单）。 /// /// **被描述的东西用到才建。** 模型会把段落里每个名词短语都列进 `e`：实测一个 25 篇的库里 diff --git a/crates/utopia-store/src/names.rs b/crates/utopia-store/src/names.rs index 687e9694a..63d04f0e5 100644 --- a/crates/utopia-store/src/names.rs +++ b/crates/utopia-store/src/names.rs @@ -13,7 +13,7 @@ //! 一条「事实」，不能冒充证据，也不能把计数撑大。 use crate::graph::{add_evidence, insert_value_fact, Validity}; -use crate::resolution::normalize_name; +use crate::resolution::{fold_markdown_markers, normalize_name}; use sqlx::PgPool; use std::collections::HashMap; use utopia_core::models::NameView; @@ -78,6 +78,7 @@ pub async fn record( attested_at: Option>, ) -> AppResult> { let name = normalize_name(name); + let name = fold_markdown_markers(&name); if name.is_empty() { return Ok(None); } diff --git a/crates/utopia-store/src/resolution.rs b/crates/utopia-store/src/resolution.rs index d51ea1b15..2262834aa 100644 --- a/crates/utopia-store/src/resolution.rs +++ b/crates/utopia-store/src/resolution.rs @@ -50,6 +50,54 @@ pub fn normalize_name(raw: &str) -> String { mapped.split_whitespace().collect::>().join(" ") } +/// Fold unambiguous Markdown formatting in names; evidence keeps the original text. +/// Input should already have passed through [`normalize_name`]. Only whole-name +/// `**...**` / `__...__` emphasis is removed; empty, nested, mismatched or +/// whitespace-padded markup is kept. +/// `__identifier__` stays literal (Unicode letters/digits and underscores count). +/// Single backticks are removed only around a slash path or a dotted filename +/// containing a letter: only letters, digits, `/`, `.`, `_` and `-` are allowed, +/// with at least one alphanumeric character and no trailing dot. This deliberately +/// leaves code, bare identifiers, numeric versions and paths with spaces alone. +pub fn fold_markdown_markers(name: &str) -> &str { + for marker in ["**", "__", "`"] { + let Some(inner) = name + .strip_prefix(marker) + .and_then(|s| s.strip_suffix(marker)) + else { + continue; + }; + if inner.is_empty() + || inner.trim() != inner + || inner.contains(['*', '`']) + || (marker != "`" + && (inner.starts_with('_') || inner.ends_with('_') || inner.contains("__"))) + { + return name; + } + let fold = match marker { + "__" => !inner.chars().all(|c| c.is_alphanumeric() || c == '_'), + "`" => { + (inner.contains('/') + || (inner.contains('.') && inner.chars().any(char::is_alphabetic))) + && inner.chars().any(char::is_alphanumeric) + && !inner.ends_with('.') + && inner + .chars() + .all(|c| c.is_alphanumeric() || matches!(c, '/' | '.' | '_' | '-')) + } + _ => true, + }; + return if fold { inner } else { name }; + } + name +} + +/// Shared identity key for recall and document-level name claims. +pub fn name_key(raw: &str) -> String { + fold_markdown_markers(&normalize_name(raw)).to_lowercase() +} + /// 泛用后缀词表：中文直接拼在词干后；英文按独立词算、首尾两种语序都认 /// （"Phoenix Project" / "Project Phoenix"）。只影响召回，判定仍走画像相似度。 const GENERIC_SUFFIXES_CJK: &[&str] = &["项目", "公司", "集团", "部门", "团队"]; @@ -89,10 +137,12 @@ pub fn name_stem(name: &str) -> Option { None } +/// 输入应已过 [`normalize_name`]。 /// mention 的召回键集合（全 lower）：本名 + 词干 + 词干的泛用后缀增广。 /// 增广覆盖反方向（库内是"星尘项目"、mention 只说"星尘"）；词干含 CJK 拼中文 -/// 尾缀，否则拼英文词（两种语序）。键数 ≤10，走 (kb,type,lower(name)) 索引多点查。 -pub fn recall_keys(name: &str) -> Vec { +/// 尾缀，否则拼英文词（两种语序）。键数 ≤11，走 (kb,type,lower(name)) 索引多点查。 +pub fn recall_keys(normalized: &str) -> Vec { + let name = fold_markdown_markers(normalized); let lower = name.to_lowercase(); let base = name_stem(name).unwrap_or_else(|| lower.clone()); let mut keys = vec![lower]; @@ -101,6 +151,8 @@ pub fn recall_keys(name: &str) -> Vec { keys.push(k); } } + // Keep exact recall of names already stored with formatting as well. + add(&mut keys, normalized.to_lowercase()); add(&mut keys, base.clone()); if base.chars().any(|c| ('\u{4E00}'..='\u{9FFF}').contains(&c)) { for suf in GENERIC_SUFFIXES_CJK { @@ -320,7 +372,7 @@ async fn propose_by_name_vector( exclude: &[Uuid], r: &mut Resolution, ) -> AppResult<()> { - let mention_name = normalize_name(raw_name).to_lowercase(); + let mention_name = name_key(raw_name); let mention_family = match type_id { Some(t) => type_label(pool, t) .await? @@ -398,10 +450,12 @@ async fn resolve_by_name( // and recallable on later calls, but this mention cannot attach to them. exclude: &[Uuid], ) -> AppResult { - let name = normalize_name(raw_name); + let normalized = normalize_name(raw_name); // 召回键 = 本名 + 泛用后缀词干及其增广（"星尘"↔"星尘项目"互为候选）。 // 只扩召回，归并与否仍由下方画像相似度分层定夺。 - let keys = recall_keys(&name); + let keys = recall_keys(&normalized); + // Store the same spelling whichever form is mentioned first. + let name = fold_markdown_markers(&normalized); // 名字召回读名字事实（0041 决定 1）：本名、简称、曾用名都是 `known_as` 上的一条。 // 度数不数名字事实——名字不是一条「关于它的事」，数进去每个实体都凭空多一 // 度数要等真有几个候选得比时才数：每个候选一条计数子查询，大库上很贵（#1296） @@ -430,7 +484,7 @@ async fn resolve_by_name( if candidates.is_empty() { // 同类型无候选 ≠ 新名字：类型标签会漂（同一团队被抽成 organization/project/ // concept），先查其它类型下的同名实体，按类型对的互斥强度分流。 - return resolve_type_drift(pool, kb_id, type_id, &name, &keys, context, exclude).await; + return resolve_type_drift(pool, kb_id, type_id, name, &keys, context, exclude).await; } let Some(ctx) = context else { @@ -517,7 +571,7 @@ async fn resolve_by_name( } } let (id, created) = - create_entity(pool, kb_id, type_id, &name, context, &weighed).await?; + create_entity(pool, kb_id, type_id, name, context, &weighed).await?; if !created { // 并行的另一份文档刚建好它：用它的，审核对也是它排的 return Ok(Resolution { @@ -526,7 +580,7 @@ async fn resolve_by_name( reviews: Vec::new(), }); } - refresh_disambiguators(pool, kb_id, &name).await?; + refresh_disambiguators(pool, kb_id, name).await?; let reviews = [(best, sim), (runner, r_sim)] .into_iter() .map(|(c, s)| ReviewRequest { @@ -594,7 +648,7 @@ async fn resolve_by_name( }); } } - let (id, created) = create_entity(pool, kb_id, type_id, &name, context, &weighed).await?; + let (id, created) = create_entity(pool, kb_id, type_id, name, context, &weighed).await?; if !created { // 并行的另一份文档刚建好它：用它的，审核对也是它排的 return Ok(Resolution { @@ -603,7 +657,7 @@ async fn resolve_by_name( reviews: Vec::new(), }); } - refresh_disambiguators(pool, kb_id, &name).await?; + refresh_disambiguators(pool, kb_id, name).await?; let mut reviews = best_scored .filter(|(_, sim)| *sim >= SIM_NEW) .map(|(c, sim)| { @@ -617,7 +671,7 @@ async fn resolve_by_name( .unwrap_or_default(); // 名字互相包含的既有实体：等值召回看不见它们（前缀枚举不完）， // 于是简称会静默变成第二个实体。只入队，不合并 - reviews.extend(containment_reviews(pool, kb_id, type_id, &name, id, context).await?); + reviews.extend(containment_reviews(pool, kb_id, type_id, name, id, context).await?); Ok(Resolution { entity_id: id, created: true, @@ -2555,6 +2609,38 @@ mod tests { assert_eq!(normalize_name("张 三"), "张 三"); } + #[test] + fn markdown_folding_is_conservative() { + for (raw, expected) in [ + ("**Pump 1**", "Pump 1"), + ("__Pump 1__", "Pump 1"), + ("__init__", "__init__"), + ("`docs/setup.md`", "docs/setup.md"), + ("`setup.md`", "setup.md"), + ("`acme`", "`acme`"), + ("`main()`", "`main()`"), + ("`1.2`", "`1.2`"), + ("**Pump 1*", "**Pump 1*"), + ("***Pump 1***", "***Pump 1***"), + ("** Pump 1 **", "** Pump 1 **"), + ("****", "****"), + ] { + assert_eq!(fold_markdown_markers(raw), expected, "{raw:?}"); + } + } + + #[test] + fn recall_keys_fold_markdown_before_stemming_and_keep_exact_spelling() { + let normalized = normalize_name("　＊＊Ｐｈｏｅｎｉｘ　 Project＊＊ "); + let keys = recall_keys(&normalized); + assert_eq!(keys[0], name_key(&normalized)); + for key in recall_keys("Phoenix Project") { + assert!(keys.contains(&key), "must recall {key:?}"); + } + assert!(keys.contains(&"**phoenix project**".to_string())); + assert!(keys.len() <= 11); + } + #[test] fn stem_generic_suffixes() { assert_eq!(name_stem("星尘项目").as_deref(), Some("星尘")); diff --git a/crates/utopia-store/tests/store/a_similar_name_is_proposed_not_merged.rs b/crates/utopia-store/tests/store/a_similar_name_is_proposed_not_merged.rs index f013c07b1..e23e1b864 100644 --- a/crates/utopia-store/tests/store/a_similar_name_is_proposed_not_merged.rs +++ b/crates/utopia-store/tests/store/a_similar_name_is_proposed_not_merged.rs @@ -275,26 +275,40 @@ async fn the_same_literal_name_is_not_proposed_twice() -> anyhow::Result<()> { let f = seed(&pool, "name-vector-literal").await?; let run = async { let same: Vec = vec![1.0, 0.0, 0.0]; - // 没有上下文向量：v1 路径归并到唯一的字面候选 - let r = utopia_store::resolution::resolve_mention( - &pool, - f.kb, - Some(f.device), - "海洋探测器1号", - None, - Some(&same), - None, - &[], + // A second entity with the same name stays a channel-1 decision, including + // when the mention carries Markdown. Neither candidate is excluded here. + sqlx::query( + "UPDATE entities SET canonical_name = '海洋探测器1号', type_id = $2 WHERE id = $1", ) + .bind(f.captain) + .bind(f.device) + .execute(&pool) .await?; - assert_eq!(r.entity_id, f.probe, "字面相等走通道 1"); - // 夹具里 海洋探测队长 的向量与之相同、大类又拦不住，它照常被提议；要看的只是 - // 已归并到的 海洋探测器1号 自己不会再被排一对 - assert!( - !vector_reviews(&r).iter().any(|(id, _)| *id == f.probe), - "已经归并到它，不再对它排 name_vector：{:?}", - r.reviews - ); + sqlx::query( + "UPDATE facts SET object_value = '{\"value\":\"海洋探测器1号\"}' WHERE subject_id = $1", + ) + .bind(f.captain) + .execute(&pool) + .await?; + for mention in ["海洋探测器1号", "**海洋探测器1号**"] { + let r = utopia_store::resolution::resolve_mention( + &pool, + f.kb, + Some(f.device), + mention, + None, + Some(&same), + None, + &[], + ) + .await?; + assert!(!r.created); + assert!( + vector_reviews(&r).is_empty(), + "a namesake must not be reconsidered by name vector: {:?}", + r.reviews + ); + } Ok::<_, anyhow::Error>(()) } .await; diff --git a/crates/utopia-store/tests/store/an_undeclared_name_is_looked_up.rs b/crates/utopia-store/tests/store/an_undeclared_name_is_looked_up.rs index d58949ca3..e373a791e 100644 --- a/crates/utopia-store/tests/store/an_undeclared_name_is_looked_up.rs +++ b/crates/utopia-store/tests/store/an_undeclared_name_is_looked_up.rs @@ -105,3 +105,101 @@ async fn a_known_name_is_found_by_any_spelling_and_an_unknown_one_is_not() -> an teardown(&pool, &f).await?; run } + +#[tokio::test] +async fn markdown_mentions_recall_names_in_either_order() -> anyhow::Result<()> { + let Some(url) = utopia_store::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + let f = seed(&pool).await?; + let run = async { + // Open-graph extraction resolves names before they have an ontology type. + sqlx::query("UPDATE entities SET type_id = NULL WHERE id = $1") + .bind(f.acme) + .execute(&pool) + .await?; + utopia_store::names::record(&pool, f.kb, f.acme, "`docs/setup.md`", None, None).await?; + for mention in [ + "**Acme Corporation**", + "__Acme Corporation__", + "**ACME**", + "docs/setup.md", + "`docs/setup.md`", + ] { + assert_eq!( + utopia_store::resolution::existing_by_name(&pool, f.kb, mention).await?, + Some(f.acme), + "{mention} must recall the existing name" + ); + let resolved = utopia_store::resolution::resolve_mention( + &pool, + f.kb, + None, + mention, + None, + None, + None, + &[], + ) + .await?; + assert_eq!(resolved.entity_id, f.acme); + assert!(!resolved.created); + assert!(resolved.reviews.is_empty()); + } + let canonical: String = + sqlx::query_scalar("SELECT canonical_name FROM entities WHERE id = $1") + .bind(f.acme) + .fetch_one(&pool) + .await?; + assert_eq!(canonical, "Acme Corporation"); + + // Marked-first names must remain recallable by later plain mentions. + let written = "**Pump 2**"; + let created = utopia_store::resolution::resolve_mention( + &pool, + f.kb, + None, + written, + None, + None, + None, + &[], + ) + .await?; + assert!(created.created); + let canonical: String = + sqlx::query_scalar("SELECT canonical_name FROM entities WHERE id = $1") + .bind(created.entity_id) + .fetch_one(&pool) + .await?; + assert_eq!(canonical, "Pump 2"); + let recorded: String = sqlx::query_scalar( + "SELECT object_value->>'value' FROM facts + WHERE subject_id = $1 AND predicate_id = $2", + ) + .bind(created.entity_id) + .bind(utopia_store::names::ensure_known_as(&pool, f.kb).await?) + .fetch_one(&pool) + .await?; + assert_eq!(recorded, "Pump 2"); + let recalled = utopia_store::resolution::resolve_mention( + &pool, + f.kb, + None, + "Pump 2", + None, + None, + None, + &[], + ) + .await?; + assert_eq!(recalled.entity_id, created.entity_id); + assert!(!recalled.created); + assert!(recalled.reviews.is_empty()); + anyhow::Ok(()) + } + .await; + teardown(&pool, &f).await?; + run +} From 0daaed0048e630ab125d021a482fe3739ee7c30a Mon Sep 17 00:00:00 2001 From: Nash Aschenbath Date: Tue, 20 Oct 2026 02:24:32 +0800 Subject: [PATCH 08/11] fix: stop alignment examples from supplying unrelated item IDs (#1323) Signed-off-by: Aschenbath Co-authored-by: Aschenbath Co-authored-by: WaylandYang <145302500+WaylandYang@users.noreply.github.com> --- crates/utopia-extract/src/align.rs | 12 +++++++----- crates/utopia-extract/src/phrase_align.rs | 14 ++++++++++---- crates/utopia-server/src/phrase_alignment.rs | 1 + 3 files changed, 18 insertions(+), 9 deletions(-) diff --git a/crates/utopia-extract/src/align.rs b/crates/utopia-extract/src/align.rs index fec250436..e73fe5ebc 100644 --- a/crates/utopia-extract/src/align.rs +++ b/crates/utopia-extract/src/align.rs @@ -67,10 +67,12 @@ the keys of its candidate classes. The classes themselves are listed once under each with its key, its label and its definition. For each item, answer with the key of the one \ candidate class that every thing of this kind is an instance of, or null.\n\ \n\ -Output exactly one JSON object and nothing else, one pair per item:\n\ -{\"b\": [[12, \"organization\"], [13, null]]}\n\ -The first element of a pair is the item's id; the second is the chosen key, copied exactly as \ -listed under that item, or null.\n\ +Output exactly one JSON object and nothing else, one pair per item. Format (uppercase words \ +and the ellipsis are placeholders, not literal output):\n\ +{\"b\": [[ITEM_ID, CLASS_KEY_OR_NULL], ...]}\n\ +Copy the exact integer after \"Item\" in the user message into that item's ITEM_ID slot. \ +Never invent an id or renumber the items. Answer every supplied id exactly once. \ +The second element is the chosen key, copied exactly as listed under that item, or null.\n\ \n\ 1. Choose a candidate only when every thing of this kind is an instance of that class by the \ class's definition. The class may be broader than the word: a bakery is an organization, a \ @@ -636,7 +638,7 @@ mod tests { let user = &msgs[1].content; assert!(system.contains("every thing of this kind is an instance of that class")); - assert!(system.contains("{\"b\": [[12, \"organization\"], [13, null]]}")); + assert!(system.contains("{\"b\": [[ITEM_ID, CLASS_KEY_OR_NULL], ...]}")); assert!(system.contains("never merely related")); assert!(system.contains("Never invent a key")); diff --git a/crates/utopia-extract/src/phrase_align.rs b/crates/utopia-extract/src/phrase_align.rs index 03ae9b016..69eab3ae7 100644 --- a/crates/utopia-extract/src/phrase_align.rs +++ b/crates/utopia-extract/src/phrase_align.rs @@ -126,7 +126,9 @@ impl Direction { } } -/// 系统消息。例子是中性的，不出自任何测量语料；规则 1 的「每一条」是整条路的判据 +/// 系统消息。例子是中性的，不出自任何测量语料；规则 1 的「每一条」是整条路的判据。 +/// Reply slots use placeholders: a local model copied the old example's 12/13/14 +/// into a batch numbered 0/1/2, leaving every signature unanswered (#1321). const PHRASE_SYSTEM: &str = "\ You bind the relation phrases documents use to the properties of a knowledge base's ontology. \ Each numbered item is one signature: a phrase as the documents wrote it, the class of the thing \ @@ -144,9 +146,13 @@ states by that property's definition, and the direction: \"forward\" when the st subject is the property's subject, \"reverse\" when the statement's object is; or null. For a \ state, also say which end of it a statement dated at a single moment marks (rule 6).\n\ \n\ -Output exactly one JSON object and nothing else, one entry per item:\n\ -{\"b\": [[12, \"headquartered_in\", \"forward\", \"none\"], [13, \"member_of\", \"forward\", \"start\"], \ -[14, null, null, null]]}\n\ +Output exactly one JSON object and nothing else, one entry per item. Format (uppercase words \ +and the ellipsis are placeholders, not literal output):\n\ +{\"b\": [[ITEM_ID, PROPERTY_KEY_OR_NULL, DIRECTION_OR_NULL, MARKS_OR_NULL], ...]}\n\ +Copy the exact integer after \"Item\" in the user message into that item's ITEM_ID slot. \ +Never invent an id or renumber the items. Answer every supplied id exactly once. \ +The other slots contain the chosen property's key, direction and marks, as described below; \ +when no property fits, all three are null.\n\ \n\ 1. Choose a property only when each of the statements under this signature states that \ property, by the definition as given. The property may be broader than the phrase: \"opened a \ diff --git a/crates/utopia-server/src/phrase_alignment.rs b/crates/utopia-server/src/phrase_alignment.rs index bdce28481..049027225 100644 --- a/crates/utopia-server/src/phrase_alignment.rs +++ b/crates/utopia-server/src/phrase_alignment.rs @@ -961,6 +961,7 @@ async fn align_phrases_locked( %kb_id, pass, items = items.len(), + item_ids = ?items.iter().map(|item| item.id).collect::>(), malformed, finish_reason = ?reply.finish_reason, chars = reply.text.chars().count(), From fa6cd6cc13e391f73d3a38f0b497bb8c053b811a Mon Sep 17 00:00:00 2001 From: Quintin Botes Date: Mon, 19 Oct 2026 20:30:56 +0200 Subject: [PATCH 09/11] A reply that keeps repeating itself comes back cut instead of running to the token ceiling (#1306) * A reply that keeps repeating itself comes back cut instead of running to the token ceiling Signed-off-by: Quintin Botes * Only a whole-row loop is cut, and the cut is reported as repetition rather than the token ceiling Signed-off-by: Quintin Botes --------- Signed-off-by: Quintin Botes Co-authored-by: WaylandYang <145302500+WaylandYang@users.noreply.github.com> --- crates/utopia-llm/src/lib.rs | 96 +++++++++++++++++++- crates/utopia-server/src/extraction_open.rs | 58 ++++++++---- crates/utopia-server/src/phrase_alignment.rs | 1 + crates/utopia-server/src/type_alignment.rs | 1 + docs/design/extraction.md | 14 +++ 5 files changed, 150 insertions(+), 20 deletions(-) diff --git a/crates/utopia-llm/src/lib.rs b/crates/utopia-llm/src/lib.rs index c046e93af..c311f72f4 100644 --- a/crates/utopia-llm/src/lib.rs +++ b/crates/utopia-llm/src/lib.rs @@ -386,6 +386,9 @@ pub struct Reply { pub finish_reason: Option, /// 端点报的用量（最后一帧）。不报就是 `None`——账上不编数字 pub usage: Option, + /// 回复在复读，是我们停了读（#1301），不是端点收的尾。与 `finish_reason` 分开记： + /// 记成 `length` 的话丢弃行会说「撞上了 token 上限」，看的人就去查那个上限了 + pub repeating: bool, } /// 一次调用的 token 用量，端点自己报的 @@ -510,6 +513,28 @@ fn strip_reasoning(reply: &str) -> &str { } } +/// 回复的尾巴是不是同一段在复读（#1301）。 +/// +/// 温度 0 的小模型会掉进复读：同一行陈述一遍遍写下去，直到撞上 +/// [`MAX_COMPLETION_TOKENS`]。本地 qwen3-8b 上约 3% 的抽取调用如此，每次占一个槽 +/// 一个多小时，吃掉六成的生成量。上限本身不能降（开推理的端点上它是思考预算）， +/// 所以看尾巴：同一段（至少 30 字节）一字不差连着出现五次，就不是在答题了—— +/// 真实的陈述行彼此不同。按字节比，多字节字符切在哪都不影响。 +/// +/// 只算**最短周期**也有 30 字节的尾巴：每条陈述以原文引文开头，一段 150 字节以上的 +/// 短周期串照样能拆成 30 字节以上的段——目录行 200 个点的引导线、一排二十个空 +/// ``，不加这一条就可能被当成复读截断（看每 256 字节的检查落在哪），引文之后 +/// 的全丢了。真复读是整行在重复；短于 30 字节的行复读时这里不截（一夜 57 次截断里有 +/// 4 次是这种），宁可让它跑到上限，也不截断引文 +fn repeats_at_tail(text: &[u8]) -> bool { + const REPEATS: usize = 5; + (30..=600.min(text.len() / REPEATS)).any(|span| { + let tail = &text[text.len() - span * REPEATS..]; + tail.chunks(span).all(|piece| piece == &tail[..span]) + && !(1..30).any(|shift| tail[shift..] == tail[..tail.len() - shift]) + }) +} + impl LlmClient { pub fn new(base_url: &str, api_key: Option<&str>, model: &str) -> Self { Self::with_timeouts(base_url, api_key, model, CONNECT_TIMEOUT, READ_TIMEOUT) @@ -698,7 +723,8 @@ impl LlmClient { let (mut saw_frame, mut ended) = (false, false); let mut finish_reason: Option = None; let mut usage: Option = None; - while let Some(part) = bytes.next().await { + let (mut checked, mut repeating) = (0, false); + 'read: while let Some(part) = bytes.next().await { let part = part.map_err(Unreachable)?; // 网络片段可能断在 UTF-8 字符中间，等完整 SSE 帧到齐再解码。 buf.push(&part); @@ -712,6 +738,24 @@ impl LlmClient { &mut finish_reason, &mut usage, ); + // 复读了就不再读，回复记成 `repeating`：抽取当它是半截的，前面完整的陈述照收， + // 不再等它写到 65,536 个 token。丢掉响应流，连接一关，端点那边也就停了。按帧看 + // 而不是按网络片段看：一个片段可能装着整段复读。每多 256 字节看一次 + if !ended && answer.len() >= checked + 256 { + checked = answer.len(); + if repeats_at_tail(answer.as_bytes()) { + tracing::warn!( + model = %self.model, + call = %call_label(messages), + chars = answer.len(), + "回复在复读，按截断收下" + ); + repeating = true; + ended = true; + buf = SseBuffer::default(); + break 'read; + } + } } } // **收尾那一帧也算**：末尾不跟空行的实现有的是，丢掉它就是丢掉答案的尾巴 @@ -749,6 +793,7 @@ impl LlmClient { text: strip_reasoning(&answer).to_string(), finish_reason, usage, + repeating, }) } @@ -1998,6 +2043,55 @@ mod tests { } } + fn content_frame(text: &str) -> String { + format!( + "data: {}\n\n", + json!({"choices": [{"delta": {"content": text}}]}) + ) + } + + async fn streamed_reply(sse: &str) -> Reply { + let (addr, server) = an_http_response("200 OK", "text/event-stream", sse).await; + let user = ChatMessage { + role: "user".into(), + content: "hi".into(), + }; + let reply = client_at(addr) + .chat_at_streaming(&[user], Some(0.0)) + .await + .unwrap(); + server.await.unwrap(); + reply + } + + /// 复读的回复按截断收下（#1301）：这条流没有收尾，没有护栏就是「流断在半路」 + #[tokio::test] + async fn a_reply_that_keeps_repeating_comes_back_cut() { + let row = r#"[null,"a","is on","b",null,{},null,null],"#; + let sse = + content_frame(r#"{"e":[["a","thing",true]],"s":["#) + &content_frame(row).repeat(200); + let reply = streamed_reply(&sse).await; + assert!(reply.repeating); + assert!(!reply.hit_token_ceiling(), "the endpoint never said length"); + assert!( + reply.text.len() < row.len() * 20, + "read {} bytes past the loop", + reply.text.len() + ); + } + + /// 引文里一条 200 个点的目录引导线不是复读：周期是一个点，不是一整行 + #[tokio::test] + async fn a_quoted_dot_leader_comes_back_whole() { + let sse = content_frame(r#"{"e":[["Introduction","section",true]],"s":[["Introduction "#) + + &content_frame(&".".repeat(200)) + + &content_frame(r#" 3","Introduction","is on page",null,"3",{},null,null]],"n":[]}"#) + + "data: [DONE]\n\n"; + let reply = streamed_reply(&sse).await; + assert!(!reply.repeating); + assert!(reply.text.ends_with(r#""n":[]}"#), "{}", reply.text); + } + #[tokio::test] async fn a_raw_stream_cut_before_its_finish_is_an_error() { use futures_util::TryStreamExt; diff --git a/crates/utopia-server/src/extraction_open.rs b/crates/utopia-server/src/extraction_open.rs index 6dc1aee38..bac9f8ea1 100644 --- a/crates/utopia-server/src/extraction_open.rs +++ b/crates/utopia-server/src/extraction_open.rs @@ -45,6 +45,14 @@ struct ChunkParsed { name_vecs: HashMap>, } +/// 回复为什么是半截的：端点撞上了 token 上限（#760），或护栏看见它在复读、停了读（#1301）。 +/// 两种都当半截处理，丢弃行要说清是哪一种——复读的去查上限，是找错了地方 +#[derive(Debug, Clone, Copy)] +enum Cut { + Ceiling, + Repeating, +} + /// 一个分块的前半步：调模型、解析回复、算名字向量（#588 cut 1）。这半步不写图， /// 所以分块之间可以不按顺序做——并发（cut 2）动的就是它；后半步 `apply_open_extraction` /// 要按分块顺序。`pushed`（0054）的块本身就是契约，不调模型，直接解析。 @@ -72,8 +80,8 @@ async fn call_and_parse_one_chunk( unextracted: &mut Vec<(i32, String)>, ) -> anyhow::Result> { // 推送来的陈述：块就是契约，解析它而不是问模型（0054）。下面从解析起一步不变 - let (reply_text, cut_by_ceiling) = if pushed { - (chunk.text.clone(), false) + let (reply_text, cut) = if pushed { + (chunk.text.clone(), None) } else { let (settings, client) = match (settings, client) { (Some(s), Some(c)) => (s, c), @@ -109,27 +117,35 @@ async fn call_and_parse_one_chunk( } }; tracing::debug!(%document_id, seq = chunk.seq, reply = %reply.text, "开放抽取的原始回复"); - // 端点说它是撞上 token 上限停的。解析器只看得见 JSON 少了尾巴，看不见 - // 少的原因，所以这句话得从回复里带过来（#760） - let cut_by_ceiling = reply.hit_token_ceiling(); - (reply.text, cut_by_ceiling) + // 解析器只看得见 JSON 少了尾巴，看不见少的原因，所以原因得从回复里带过来（#760） + let cut = if reply.repeating { + Some(Cut::Repeating) + } else if reply.hit_token_ceiling() { + Some(Cut::Ceiling) + } else { + None + }; + (reply.text, cut) }; let mut extraction = match utopia_extract::open::parse_open_response(&reply_text) { Ok(x) => x, Err(e) => { - tracing::warn!(%document_id, seq = chunk.seq, error = %e, hit_token_ceiling = cut_by_ceiling, "开放抽取回复解析失败，跳过该分块"); - // 一整块没进图，而原因不同：撞上上限是「答案太长」，改得动； - // 别的解析失败是「回复不合结构」。两种都写成同一句话就分不开了 - let (why, detail) = if cut_by_ceiling { - ( + tracing::warn!(%document_id, seq = chunk.seq, error = %e, hit_token_ceiling = matches!(cut, Some(Cut::Ceiling)), repeating = matches!(cut, Some(Cut::Repeating)), "开放抽取回复解析失败，跳过该分块"); + // 一整块没进图，而原因不同：撞上上限是「答案太长」，改得动；复读是模型卡在一行上； + // 别的解析失败是「回复不合结构」。写成同一句话就分不开了 + let (why, detail) = match cut { + Some(Cut::Ceiling) => ( "response hit the token ceiling and could not be parsed; this chunk did not enter the graph", format!("#{}: hit the token ceiling; {e}", chunk.seq), - ) - } else { - ( + ), + Some(Cut::Repeating) => ( + "response started repeating itself and could not be parsed; this chunk did not enter the graph", + format!("#{}: started repeating itself; {e}", chunk.seq), + ), + None => ( "response could not be parsed; this chunk did not enter the graph", format!("#{}: {e}", chunk.seq), - ) + ), }; drop_signal( state, @@ -164,12 +180,16 @@ async fn call_and_parse_one_chunk( kb_id, document_id, reason::TRUNCATED_REPLY, - if cut_by_ceiling { - "the open reply hit the token ceiling; kept up to the last complete item" - } else { + match cut { + Some(Cut::Ceiling) => { + "the open reply hit the token ceiling; kept up to the last complete item" + } + Some(Cut::Repeating) => { + "the reply started repeating itself; kept up to the last complete item" + } // 回复正常结束而整体不合法：模型写坏了括号。逐条读出来的都收了，写坏的 // 条数在下面那条信号里 - "the open reply was not valid JSON as a whole; read item by item" + None => "the open reply was not valid JSON as a whole; read item by item", }, None, ) diff --git a/crates/utopia-server/src/phrase_alignment.rs b/crates/utopia-server/src/phrase_alignment.rs index 049027225..81fb8b452 100644 --- a/crates/utopia-server/src/phrase_alignment.rs +++ b/crates/utopia-server/src/phrase_alignment.rs @@ -964,6 +964,7 @@ async fn align_phrases_locked( item_ids = ?items.iter().map(|item| item.id).collect::>(), malformed, finish_reason = ?reply.finish_reason, + repeating = reply.repeating, chars = reply.text.chars().count(), reply = %snippet(&reply.text), "短语对齐回复读不出一条，这一批留到下次" diff --git a/crates/utopia-server/src/type_alignment.rs b/crates/utopia-server/src/type_alignment.rs index 4d0da08ae..7052acff6 100644 --- a/crates/utopia-server/src/type_alignment.rs +++ b/crates/utopia-server/src/type_alignment.rs @@ -319,6 +319,7 @@ async fn align_types_locked( items = items.len(), malformed, finish_reason = ?reply.finish_reason, + repeating = reply.repeating, chars = reply.text.chars().count(), reply = %snippet(&reply.text), "类别词对齐回复读不出一项，这一批留到下次" diff --git a/docs/design/extraction.md b/docs/design/extraction.md index 1371e02ec..0928cb38e 100644 --- a/docs/design/extraction.md +++ b/docs/design/extraction.md @@ -105,6 +105,20 @@ token ceiling* instead of only *could not be parsed* — and the two causes stay is too long is a number to change, a reply that does not fit the shape is not. Across all bases in the e2e database `truncated_reply` has fired 10 times, and those rows keep every complete item. +**Where a looping answer stops** [#1301]. At temperature 0 a small local model sometimes falls into +a loop and writes the same row until the ceiling: on qwen3-8b about 3% of extraction calls, each +holding a slot for over an hour and together about 60% of the tokens generated. The ceiling stays +where it is, for the reasons above. Instead the streamed reply is read until the same span of 30 to +600 bytes fills its tail five times in a row, and only if that run's shortest period is also at +least 30 bytes: every statement starts with a verbatim quote, and a quoted dot leader or a row of +empty table cells repeats with a short period. A loop of rows shorter than 30 bytes is left to run +to the ceiling (4 of 57 cuts in one night of a local deployment) rather than risk cutting a quote. +The reply comes back marked `repeating`, apart from `finish_reason`, so it is never counted as the +ceiling. Its drop row reads *started repeating itself* under the same `truncated_reply` code +(`chunk_unextracted` when nothing before the loop parses), and it keeps every complete item before +the loop. The guard sits in the streaming call, so the alignment, errata, implication, time and +ontology calls that stream get it too. + **Server checks, each a drop reason in `extraction_drops`.** Every quote must occur in the chunk (`quote_not_in_chunk`); every name in its quote (`name_not_in_text`); a time mention only when its words occur in the statement's own quote (`time_not_in_quote`), because the model otherwise attaches From 33c7c66fbc784e6ae00c6487778195641a9fe30d Mon Sep 17 00:00:00 2001 From: Quintin Botes Date: Mon, 19 Oct 2026 21:00:34 +0200 Subject: [PATCH 10/11] Intake jobs are claimed before model work, so a sync no longer waits behind the extraction backlog (#1324) Signed-off-by: Quintin Botes Co-authored-by: WaylandYang <145302500+WaylandYang@users.noreply.github.com> --- crates/utopia-cli/src/main.rs | 2 +- crates/utopia-store/src/jobs.rs | 9 +++++- crates/utopia-store/src/jobs_claim_tests.rs | 30 +++++++++++++++++++ ...ake_jobs_are_claimed_before_model_work.sql | 20 +++++++++++++ 4 files changed, 59 insertions(+), 2 deletions(-) create mode 100644 crates/utopia-store/src/jobs_claim_tests.rs create mode 100644 migrations/0116_intake_jobs_are_claimed_before_model_work.sql diff --git a/crates/utopia-cli/src/main.rs b/crates/utopia-cli/src/main.rs index 2d09f0936..0d41dcf34 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 = 103; +const CURRENT_SCHEMA_VERSION: u32 = 104; fn main() -> anyhow::Result<()> { dotenvy::dotenv().ok(); diff --git a/crates/utopia-store/src/jobs.rs b/crates/utopia-store/src/jobs.rs index 0ca39fc20..f151e1bb8 100644 --- a/crates/utopia-store/src/jobs.rs +++ b/crates/utopia-store/src/jobs.rs @@ -371,6 +371,9 @@ pub async fn status_in_kb(pool: &PgPool, kb_id: Uuid, id: i64) -> AppResult AppResult> { let job = sqlx::query_as( "UPDATE jobs SET status = 'running', locked_at = now(), @@ -378,7 +381,7 @@ async fn claim_one(pool: &PgPool) -> AppResult> { WHERE id = ( SELECT id FROM jobs WHERE status = 'queued' AND run_at <= now() - ORDER BY run_at + ORDER BY job_priority(kind), run_at FOR UPDATE SKIP LOCKED LIMIT 1 ) @@ -675,6 +678,10 @@ where #[path = "jobs_writeback_tests.rs"] mod writeback_tests; +#[cfg(test)] +#[path = "jobs_claim_tests.rs"] +mod claim_tests; + #[cfg(test)] mod tests { use super::{deferred_retry_secs, retry_delay}; diff --git a/crates/utopia-store/src/jobs_claim_tests.rs b/crates/utopia-store/src/jobs_claim_tests.rs new file mode 100644 index 000000000..f0b323200 --- /dev/null +++ b/crates/utopia-store/src/jobs_claim_tests.rs @@ -0,0 +1,30 @@ +//! 认领顺序（#1300）：不调对话模型的任务排在模型任务前面，哪怕它来得晚。 +//! +//! `claim_one` 认领的是全库的任务，所以这两条放在一百年前，排在库里别的一切前面。 + +use super::claim_one; +use sqlx::PgPool; + +#[tokio::test] +async fn a_later_sync_is_claimed_before_an_earlier_extraction() -> anyhow::Result<()> { + let Some(url) = crate::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + let ids: Vec = sqlx::query_scalar( + r#"INSERT INTO jobs (kind, payload, run_at) VALUES + ('extract_document', '{"test": "claim-order"}', now() - interval '101 years'), + ('sync_source', '{"test": "claim-order"}', now() - interval '100 years') + RETURNING id"#, + ) + .fetch_all(&pool) + .await?; + + let claimed = claim_one(&pool).await?.map(|job| job.kind); + sqlx::query("DELETE FROM jobs WHERE id = ANY($1)") + .bind(&ids) + .execute(&pool) + .await?; + assert_eq!(claimed.as_deref(), Some("sync_source")); + Ok(()) +} diff --git a/migrations/0116_intake_jobs_are_claimed_before_model_work.sql b/migrations/0116_intake_jobs_are_claimed_before_model_work.sql new file mode 100644 index 000000000..d59f7458d --- /dev/null +++ b/migrations/0116_intake_jobs_are_claimed_before_model_work.sql @@ -0,0 +1,20 @@ +-- 不调对话模型的任务先认领（#1300）。 +-- +-- 认领原来只按 run_at，所有种类排一条队。几百篇文档等抽取的时候，整点的 sync_source、 +-- 新同步进来那批文档的 process_document 都要等前面每一个更早的 extract_document 跑完； +-- 本地模型上那是好几天——来源十个小时没同步，423 篇新文档一直 pending、搜不到。 +-- +-- 优先级从 kind 算出来，排任务的地方一处不用改：不调对话模型的是 0，其余是 1。 +-- embed_ontology、memory_ingest 等的是嵌入闸门，process_document 可能调 OCR 或转写， +-- 都不等对话模型。名单只在这个函数里，认领的 ORDER BY 和下面的索引都读它；要改名单， +-- 在新的迁移里 CREATE OR REPLACE 它并重建索引。没有加一列：jobs 留着全部跑完的任务， +-- 加存储列要在启动迁移时重写整张表 +CREATE FUNCTION job_priority(kind text) RETURNS smallint +LANGUAGE sql IMMUTABLE PARALLEL SAFE AS $$ + SELECT CASE WHEN kind IN ('sync_source', 'process_document', 'hydrate_rss_entry', + 'embed_ontology', 'memory_ingest', 'build_vector_index', + 'materialize_inferences', 'noop') + THEN 0::smallint ELSE 1::smallint END +$$; + +CREATE INDEX jobs_priority_claim_idx ON jobs (job_priority(kind), run_at) WHERE status = 'queued'; From d62e880b15ba80de983c83abff056c4107aef9ec Mon Sep 17 00:00:00 2001 From: Quintin Botes Date: Mon, 19 Oct 2026 21:06:09 +0200 Subject: [PATCH 11/11] A base can follow a GitHub repository's markdown files, fetching only the files that changed since the last sync (#1325) Signed-off-by: Quintin Botes Co-authored-by: WaylandYang <145302500+WaylandYang@users.noreply.github.com> --- Cargo.lock | 2 + Cargo.toml | 2 + crates/utopia-core/src/models.rs | 1 + crates/utopia-server/Cargo.toml | 2 + crates/utopia-server/src/github_issues.rs | 57 ++++--- crates/utopia-server/src/github_repo.rs | 117 ++++++++++++++ .../src/github_repo_sync_tests.rs | 147 ++++++++++++++++++ crates/utopia-server/src/ingest_sources.rs | 105 +++++++++++++ crates/utopia-server/src/main.rs | 1 + .../tests/fixtures/github_repo.json | 19 +++ .../store/a_viewer_never_sees_a_credential.rs | 4 + docs/design/sources.md | 4 +- web/src/i18n/en.ts | 6 + web/src/i18n/zh.ts | 6 + web/src/pages/Library.tsx | 49 ++++-- web/src/pages/SourcesRail.tsx | 2 + web/src/sourceKinds.ts | 1 + 17 files changed, 494 insertions(+), 31 deletions(-) create mode 100644 crates/utopia-server/src/github_repo.rs create mode 100644 crates/utopia-server/src/github_repo_sync_tests.rs create mode 100644 crates/utopia-server/tests/fixtures/github_repo.json diff --git a/Cargo.lock b/Cargo.lock index 123cae836..23de9c05d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5514,6 +5514,7 @@ dependencies = [ "feed-rs", "flate2", "futures-util", + "glob", "jsonwebtoken", "object_store", "oxrdf", @@ -5525,6 +5526,7 @@ dependencies = [ "rig-core", "serde", "serde_json", + "sha1", "sha2 0.11.0", "sqlparser", "sqlx", diff --git a/Cargo.toml b/Cargo.toml index fc8457022..fc843bd7f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -78,6 +78,8 @@ feed-rs = "2" cron = "0.12" infer = "0.22" sha2 = "0.11" +glob = "0.3" +sha1 = "0.10" dom_smoothie = "0.16" dom_query = "0.26" htmd = "0.5" diff --git a/crates/utopia-core/src/models.rs b/crates/utopia-core/src/models.rs index 4bc79d376..a32be0210 100644 --- a/crates/utopia-core/src/models.rs +++ b/crates/utopia-core/src/models.rs @@ -271,6 +271,7 @@ pub enum SourceKind { Url, Rss, GithubIssues, + GithubRepo, JiraIssues, S3, AzureBlob, diff --git a/crates/utopia-server/Cargo.toml b/crates/utopia-server/Cargo.toml index 6cf9f297c..e7f24d44c 100644 --- a/crates/utopia-server/Cargo.toml +++ b/crates/utopia-server/Cargo.toml @@ -17,6 +17,8 @@ utopia-search.workspace = true utopia-llm.workspace = true utopia-extract = { path = "../utopia-extract" } sha2.workspace = true +glob.workspace = true +sha1.workspace = true futures-util.workspace = true async-stream.workspace = true tokio.workspace = true diff --git a/crates/utopia-server/src/github_issues.rs b/crates/utopia-server/src/github_issues.rs index dffe3e661..0b7254662 100644 --- a/crates/utopia-server/src/github_issues.rs +++ b/crates/utopia-server/src/github_issues.rs @@ -209,6 +209,41 @@ pub fn sort_events(mut events: Vec) -> Vec { events } +/// 发一个带令牌的 GET 并判状态。github_issues 与 github_repo 共用这一条， +/// 免得限流提示在两个来源里各写一份、慢慢漂移。 +pub(crate) async fn get_checked( + http: &reqwest::Client, + url: reqwest::Url, + auth: Option<&str>, + accept: Option<&str>, +) -> anyhow::Result { + let mut req = http.get(url); + if let Some(a) = auth { + req = req.header(reqwest::header::AUTHORIZATION, a); + } + if let Some(a) = accept { + req = req.header(reqwest::header::ACCEPT, a); + } + let resp = req.send().await?; + let status = resp.status(); + if !status.is_success() { + // 限流说清楚是限流：未认证时每小时 60 次，一个中等仓库一次同步就能吃光。 + // 报成通用 HTTP 错会让人去查网络，而正确的动作是配一个令牌 + let remaining = resp + .headers() + .get("x-ratelimit-remaining") + .and_then(|v| v.to_str().ok()) + .unwrap_or("?"); + if status == reqwest::StatusCode::FORBIDDEN && remaining == "0" { + anyhow::bail!( + "GitHub 限流：本小时配额已用尽。未认证时每小时 60 次，配一个令牌可提到 5000" + ); + } + anyhow::bail!("HTTP {status} from GitHub"); + } + Ok(resp) +} + /// 分页取一个端点，直到空页或触顶。 pub async fn fetch_all Deserialize<'de>>( http: &reqwest::Client, @@ -229,27 +264,7 @@ pub async fn fetch_all Deserialize<'de>>( q.append_pair("per_page", &PER_PAGE.to_string()); q.append_pair("page", &page.to_string()); } - let mut req = http.get(url); - if let Some(a) = auth { - req = req.header(reqwest::header::AUTHORIZATION, a); - } - let resp = req.send().await?; - let status = resp.status(); - if !status.is_success() { - // 限流说清楚是限流：未认证时每小时 60 次，一个中等仓库一次同步就能吃光。 - // 报成通用 HTTP 错会让人去查网络，而正确的动作是配一个令牌 - let remaining = resp - .headers() - .get("x-ratelimit-remaining") - .and_then(|v| v.to_str().ok()) - .unwrap_or("?"); - if status == reqwest::StatusCode::FORBIDDEN && remaining == "0" { - anyhow::bail!( - "GitHub 限流：本小时配额已用尽。未认证时每小时 60 次，配一个令牌可提到 5000" - ); - } - anyhow::bail!("HTTP {status} from GitHub"); - } + let resp = get_checked(http, url, auth, None).await?; let batch: Vec = resp.json().await?; let n = batch.len(); out.extend(batch); diff --git a/crates/utopia-server/src/github_repo.rs b/crates/utopia-server/src/github_repo.rs new file mode 100644 index 000000000..b3d7d5c43 --- /dev/null +++ b/crates/utopia-server/src/github_repo.rs @@ -0,0 +1,117 @@ +//! GitHub 仓库文件来源：默认分支头的一棵文件树，一条路径一篇文档。 +//! 同步在 `ingest_sources`；这里只放不联网的那一半，于是单测得动。 + +use serde::Deserialize; +use sha1::Digest as _; +use std::collections::HashSet; + +#[derive(Debug, Clone, Deserialize)] +pub struct TreeEntry { + pub path: String, + pub mode: String, + #[serde(rename = "type")] + pub kind: String, + pub sha: String, + pub size: Option, +} + +#[derive(Debug, Deserialize)] +pub struct TreeResponse { + pub tree: Vec, + #[serde(default)] + pub truncated: bool, +} + +/// 哪条路径收、收多大的文件。只收 markdown 是为了不把整个代码仓当语料倒进来。 +pub struct FileFilter { + include: Vec, + exclude: Vec, + max_file_bytes: i64, +} + +impl FileFilter { + pub fn from_config(config: &serde_json::Value) -> anyhow::Result { + // 配置给了这一列就整列替换默认值 + let compile = |key: &str, defaults: &[&str]| -> anyhow::Result> { + let list: Vec<&str> = match config[key].as_array() { + Some(a) => a.iter().filter_map(|v| v.as_str()).collect(), + None => defaults.to_vec(), + }; + list.into_iter() + .map(|p| { + glob::Pattern::new(p) + .map_err(|e| anyhow::anyhow!("config.{key} has an invalid glob {p:?}: {e}")) + }) + .collect() + }; + Ok(Self { + include: compile("include", &["**/*.md", "**/*.markdown", "**/*.mdx"])?, + exclude: compile( + "exclude", + &["**/.*/**", "**/node_modules/**", "**/vendor/**"], + )?, + max_file_bytes: config["max_file_bytes"].as_i64().unwrap_or(1 << 20), + }) + } + + fn eligible(&self, e: &TreeEntry) -> bool { + let opts = glob::MatchOptions { + case_sensitive: true, + require_literal_separator: true, + require_literal_leading_dot: false, + }; + // 子模块是 commit、目录是 tree，符号链接指向别处——都不是文件本体 + e.kind == "blob" + && e.mode != "120000" + && e.size.is_some_and(|s| s <= self.max_file_bytes) + && self.include.iter().any(|p| p.matches_with(&e.path, opts)) + && !self.exclude.iter().any(|p| p.matches_with(&e.path, opts)) + } +} + +/// 按路径排序、同一 blob 只留第一条：相同文件在几个路径出现时，文档不该在路径之间横跳。 +pub fn choose(entries: Vec, filter: &FileFilter) -> Vec { + let mut entries: Vec<_> = entries.into_iter().filter(|e| filter.eligible(e)).collect(); + entries.sort_by(|a, b| a.path.cmp(&b.path)); + let mut seen = HashSet::new(); + entries.retain(|e| seen.insert(e.sha.clone())); + entries +} + +/// git 里一个 blob 的 SHA-1（先喂 `blob <字节数>\0` 再喂正文）。树里给的就是这个值， +/// 拿已存的字节算一遍就知道没变，不必再下文件。 +pub fn git_blob_sha(bytes: &[u8]) -> String { + let mut h = sha1::Sha1::new(); + h.update(format!("blob {}\0", bytes.len()).as_bytes()); + h.update(bytes); + h.finalize().iter().map(|b| format!("{b:02x}")).collect() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn the_default_globs_take_markdown_outside_hidden_and_vendored_folders() { + let f = FileFilter::from_config(&serde_json::json!({})).unwrap(); + let entry = |path: &str, mode: &str| TreeEntry { + path: path.into(), + mode: mode.into(), + kind: "blob".into(), + sha: "s".into(), + size: Some(10), + }; + for path in ["README.md", "docs/a/b.mdx"] { + assert!(f.eligible(&entry(path, "100644")), "{path}"); + } + for path in [ + ".github/PULL_REQUEST_TEMPLATE.md", + "web/node_modules/x/README.md", + "vendor/a.md", + "Cargo.toml", + ] { + assert!(!f.eligible(&entry(path, "100644")), "{path}"); + } + assert!(!f.eligible(&entry("link.md", "120000"))); + } +} diff --git a/crates/utopia-server/src/github_repo_sync_tests.rs b/crates/utopia-server/src/github_repo_sync_tests.rs new file mode 100644 index 000000000..9ce35c92a --- /dev/null +++ b/crates/utopia-server/src/github_repo_sync_tests.rs @@ -0,0 +1,147 @@ +//! 用录下的真实 GitHub 响应跑仓库同步：筛选、同一 blob 只收一处、没变不再取、 +//! 树截断时不判「不在来源中」。 +use super::sync_github_repo_with_client; +use serde_json::{json, Value}; +use std::sync::{Arc, Mutex}; +use uuid::Uuid; +use wiremock::{matchers::method, Mock, MockServer, Request, ResponseTemplate}; + +const PACKS: &str = "crates/utopia-server/packs/README.md"; +const TESTDATA: &str = "crates/utopia-server/testdata/README.md"; + +/// 录下的树再加一个 testdata/README.md 的副本。阶段 1、2 去掉 packs，阶段 1 还标了截断 +fn tree(fixture: &Value, stage: u8) -> Value { + let mut entries = fixture["tree"]["tree"].as_array().unwrap().clone(); + let mut copy = entries + .iter() + .find(|e| e["path"] == TESTDATA) + .unwrap() + .clone(); + copy["path"] = json!("docs/copy/README.md"); + entries.push(copy); + if stage > 0 { + entries.retain(|e| e["path"] != PACKS); + } + json!({ "tree": entries, "truncated": stage == 1 }) +} + +#[tokio::test] +async fn a_repository_sync_takes_each_changed_file_once() -> anyhow::Result<()> { + let Some(url) = utopia_store::test_db::url() else { + return Ok(()); + }; + let pool = sqlx::PgPool::connect(&url).await?; + let (org, ws, kb, sid) = ( + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + ); + let result = async { + // 2000 字节的上限把 2920 字节的 SECURITY.md 挡在外面 + sqlx::raw_sql(&format!( + r#"INSERT INTO organizations(id,name) VALUES('{org}','repo-test'); + INSERT INTO workspaces(id,org_id,name) VALUES('{ws}','{org}','repo-test'); + INSERT INTO knowledge_bases(id,workspace_id,name) VALUES('{kb}','{ws}','repo-test'); + INSERT INTO sources(id,kb_id,kind,name,config) + VALUES('{sid}','{kb}','github_repo','fixture', + '{{"repo":"deeplethe/utopia","max_file_bytes":2000}}');"# + )) + .execute(&pool) + .await?; + let source = utopia_store::sources::get(&pool, sid).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 state = crate::state::AppState::new(pool.clone(), &cfg, search, "test-only".into()); + + let fixture: Value = + serde_json::from_str(include_str!("../tests/fixtures/github_repo.json"))?; + let server = MockServer::start().await; + let stage = Arc::new(Mutex::new(0u8)); + let (current, fx) = (stage.clone(), fixture.clone()); + Mock::given(method("GET")) + .respond_with(move |request: &Request| { + let path = request.url.path(); + let body = if path.ends_with("/git/trees/HEAD") { + tree(&fx, *current.lock().unwrap()) + } else if path.ends_with("/commits") { + let (_, of) = request + .url + .query_pairs() + .find(|(k, _)| k == "path") + .unwrap(); + fx["commits"][&*of].clone() + } else if let Some((_, sha)) = path.split_once("/git/blobs/") { + return ResponseTemplate::new(200) + .set_body_string(fx["blobs"][sha].as_str().unwrap()); + } else { + return ResponseTemplate::new(404); + }; + ResponseTemplate::new(200).set_body_json(body) + }) + .mount(&server) + .await; + let (http, uri) = (reqwest::Client::builder().no_proxy().build()?, server.uri()); + let sync = || sync_github_repo_with_client(&state, &source, &http, &uri); + let doc = |path: &str| { + let key = format!("github:deeplethe/utopia:{path}"); + let pool = pool.clone(); + async move { utopia_store::documents::find_by_external_key(&pool, sid, &key).await } + }; + + let first = sync().await?; + for path in [PACKS, TESTDATA] { + let date = fixture["commits"][path][0]["commit"]["committer"]["date"].as_str(); + let time = doc(path).await?.and_then(|d| d.doc_time); + anyhow::ensure!( + time == date.and_then(|d| d.parse().ok()), + "{path}：{time:?}" + ); + } + for path in [ + "SECURITY.md", + ".github/PULL_REQUEST_TEMPLATE.md", + "docs/copy/README.md", + ] { + anyhow::ensure!(doc(path).await?.is_none(), "{path} 不该收"); + } + anyhow::ensure!(first.created == 2, "第一轮建了 {}", first.created); + + let seen = server.received_requests().await.unwrap().len(); + let second = sync().await?; + let refetched = server.received_requests().await.unwrap()[seen..] + .iter() + .filter(|r| !r.url.path().ends_with("/git/trees/HEAD")) + .count(); + anyhow::ensure!( + second.total() == 0 && refetched == 0, + "没变又取了 {refetched} 次" + ); + + *stage.lock().unwrap() = 1; + sync().await?; + let missing = || async { anyhow::Ok(doc(PACKS).await?.unwrap().missing_since) }; + anyhow::ensure!(missing().await?.is_none(), "截断的树不该判消失"); + *stage.lock().unwrap() = 2; + sync().await?; + anyhow::ensure!(missing().await?.is_some(), "完整的树里没了才判消失"); + anyhow::ensure!(doc(TESTDATA).await?.unwrap().missing_since.is_none()); + anyhow::Ok(()) + } + .await; + sqlx::query("DELETE FROM knowledge_bases WHERE id=$1") + .bind(kb) + .execute(&pool) + .await?; + sqlx::query("DELETE FROM organizations WHERE id=$1") + .bind(org) + .execute(&pool) + .await?; + result +} diff --git a/crates/utopia-server/src/ingest_sources.rs b/crates/utopia-server/src/ingest_sources.rs index 17469be09..26e64209d 100644 --- a/crates/utopia-server/src/ingest_sources.rs +++ b/crates/utopia-server/src/ingest_sources.rs @@ -83,6 +83,7 @@ pub async fn sync_source(state: &AppState, source_id: Uuid) -> anyhow::Result<() Some(SourceKind::Rss) => sync_rss(state, &source).await, Some(SourceKind::Custom) => sync_custom(state, &source, since).await, Some(SourceKind::GithubIssues) => sync_github_issues(state, &source, since).await, + Some(SourceKind::GithubRepo) => sync_github_repo(state, &source).await, Some(SourceKind::JiraIssues) => sync_jira_issues(state, &source, since).await, Some(SourceKind::S3 | SourceKind::AzureBlob | SourceKind::Gcs) => { sync_object_storage(state, &source).await @@ -886,6 +887,106 @@ async fn sync_github_issues_with_client( Ok(stats) } +/// GitHub 仓库文件：一次递归取默认分支的整棵树，一条路径一篇文档。 +/// 树里的 blob SHA 就判得出「没变」，变了的才逐条取提交日期与原文。 +async fn sync_github_repo(state: &AppState, source: &Source) -> anyhow::Result { + // 与 github_issues 同一个口径：固定的公网主机按内容级严格度收 + let url = reqwest::Url::parse("https://api.github.com/")?; + let http = crate::http_fetch::client_for( + &url, + crate::http_fetch::Reach::Content, + crate::http_fetch::Limits::default(), + ) + .await?; + sync_github_repo_with_client(state, source, &http, "https://api.github.com").await +} + +// HTTP 接缝让回归测试走同一条同步与落库路径，生产仍固定访问 GitHub 公网 API。 +async fn sync_github_repo_with_client( + state: &AppState, + source: &Source, + http: &reqwest::Client, + api_base: &str, +) -> anyhow::Result { + use crate::{github_issues::get_checked, github_repo as gr}; + let repo = source.config["repo"] + .as_str() + .map(str::trim) + .filter(|s| !s.is_empty()) + .ok_or_else(|| anyhow::anyhow!("github_repo source is missing config.repo (owner/name)"))?; + if repo.split('/').count() != 2 { + anyhow::bail!("config.repo should look like owner/name, got {repo:?}"); + } + let auth = source.config["auth_header"] + .as_str() + .map(str::trim) + .filter(|s| !s.is_empty()); + let filter = gr::FileFilter::from_config(&source.config)?; + let base = format!("{api_base}/repos/{repo}"); + let tree_url = reqwest::Url::parse(&format!("{base}/git/trees/HEAD?recursive=1"))?; + let tree: gr::TreeResponse = get_checked(http, tree_url, auth, None) + .await? + .json() + .await?; + let (truncated, chosen) = (tree.truncated, gr::choose(tree.tree, &filter)); + let keys: Vec = chosen + .iter() + .map(|e| format!("github:{repo}:{}", e.path)) + .collect(); + let mut stats = SyncStats::default(); + for (entry, key) in chosen.iter().zip(&keys) { + // 没变的整条跳过：已落库的字节算出同一个 git SHA，就不必再花两次请求 + let old = + utopia_store::documents::find_by_external_key(&state.pool, source.id, key).await?; + let unchanged = match old { + Some(d) if Some(d.size_bytes) == entry.size => state + .blob + .get(&d.sha256) + .await + .is_ok_and(|b| gr::git_blob_sha(&b) == entry.sha), + _ => false, + }; + if unchanged { + continue; + } + // 文档日期取最后一次动过这条路径的提交，只在新出现或变了时问 + let mut cu = reqwest::Url::parse(&format!("{base}/commits"))?; + cu.query_pairs_mut() + .append_pair("path", &entry.path) + .append_pair("per_page", "1"); + let commits: serde_json::Value = get_checked(http, cu, auth, None).await?.json().await?; + let doc_time = commits[0]["commit"]["committer"]["date"] + .as_str() + .and_then(|d| d.parse().ok()); + let bu = reqwest::Url::parse(&format!("{base}/git/blobs/{}", entry.sha))?; + let bytes = get_checked(http, bu, auth, Some("application/vnd.github.raw+json")) + .await? + .bytes() + .await?; + let action = ingest_item( + state, + source.kb_id, + source.id, + key, + &entry.path, + "text/markdown", + &bytes, + doc_time, + ) + .await?; + stats.absorb(action); + } + if truncated { + // 树不全就不判「谁消失了」——缺席可能只是还没列到 + tracing::warn!(%repo, "GitHub 文件树被截断，本轮只摄入列出的文件，不做不在来源中对账"); + } else { + utopia_store::documents::reconcile_missing(&state.pool, source.id, &keys) + .await + .map_err(|e| anyhow::anyhow!(e))?; + } + Ok(stats) +} + /// Jira 工单：一张工单一篇文档，正文里带**字段级**的变更史。 /// /// 比 GitHub 那条路省：`search?expand=changelog` 一次调用就带回工单本体、 @@ -1311,3 +1412,7 @@ mod source_filename_tests; #[cfg(test)] #[path = "github_sync_tests.rs"] mod github_sync_tests; + +#[cfg(test)] +#[path = "github_repo_sync_tests.rs"] +mod github_repo_sync_tests; diff --git a/crates/utopia-server/src/main.rs b/crates/utopia-server/src/main.rs index 6617ea075..3ad724e30 100644 --- a/crates/utopia-server/src/main.rs +++ b/crates/utopia-server/src/main.rs @@ -10,6 +10,7 @@ mod error; mod extraction; mod extraction_open; mod github_issues; +mod github_repo; mod governance; mod http_fetch; mod implication; diff --git a/crates/utopia-server/tests/fixtures/github_repo.json b/crates/utopia-server/tests/fixtures/github_repo.json new file mode 100644 index 000000000..c420115e9 --- /dev/null +++ b/crates/utopia-server/tests/fixtures/github_repo.json @@ -0,0 +1,19 @@ +{ + "note": "取自 deeplethe/utopia 的真实 GitHub 响应（树 cda3747），条目和字段已裁到我们读的那些。blobs 是 Accept: application/vnd.github.raw+json 取回的原文；commits 是 commits?path=…&per_page=1 的第一条。", + "tree": {"sha": "cda374771577b2d42ed257312864ecf8e067bfbd", "truncated": false, "tree": [ + {"path": ".github/PULL_REQUEST_TEMPLATE.md", "mode": "100644", "type": "blob", "sha": "9b05a587001713b657e8cc51bb94f3a14add1dee", "size": 1067}, + {"path": "Cargo.toml", "mode": "100644", "type": "blob", "sha": "5b65de568922150141e12599d393e21db58eb6e3", "size": 4030}, + {"path": "SECURITY.md", "mode": "100644", "type": "blob", "sha": "274cff38310d7322ef84f27174e0344435ef5afc", "size": 2920}, + {"path": "crates/utopia-server/packs/README.md", "mode": "100644", "type": "blob", "sha": "69dbb3020bce31ed843185f1dd24af44184a5781", "size": 1364}, + {"path": "crates/utopia-server/testdata/README.md", "mode": "100644", "type": "blob", "sha": "2460624379264fed241f0986cd7dda7a5d86c104", "size": 1026}, + {"path": "docs", "mode": "040000", "type": "tree", "sha": "c2e443619c8b74e52530a296f1282b77c3e4f954"} + ]}, + "commits": { + "crates/utopia-server/packs/README.md": [{"sha": "92db693c3e4763f03551fcb1144684336ccd83f5", "commit": {"committer": {"date": "2026-08-30T01:17:00Z"}}}], + "crates/utopia-server/testdata/README.md": [{"sha": "798fb2476f4ec8eacf3d6220657f27333ad50d94", "commit": {"committer": {"date": "2026-09-30T10:04:58Z"}}}] + }, + "blobs": { + "69dbb3020bce31ed843185f1dd24af44184a5781": "# 预制本体包\n\n来源文件按原样保存（gzip 压缩），由 `src/ontology_packs.rs` 用 `include_bytes!` 内嵌进二进制。\n\n**为什么内嵌而不是运行时下载**：README 承诺整套系统可以运行在完全离线的内网环境。\n运行时抓取会让这句话失效，也让构建不可复现（上游随时会发新版）。\n\n**为什么压缩**：未压缩合计 1.7 MB，压缩后 316 KB。这是个公开仓库，clone 体积是真实成本。\n`flate2` 本来就在依赖树里，解压是三行。\n\n| 文件 | 来源 | 许可 | 抓取日 |\n|---|---|---|---|\n| `schema-org.ttl.gz` | https://schema.org/version/latest/schemaorg-current-https.ttl | CC BY-SA 3.0 | 2026-08-30 |\n| `w3c-org.ttl.gz` | https://www.w3.org/ns/org.ttl | W3C Document License | 2026-08-30 |\n| `prov-o.ttl.gz` | https://www.w3.org/ns/prov.ttl | W3C Document License | 2026-08-30 |\n| `foaf.rdf.gz` | http://xmlns.com/foaf/spec/index.rdf | CC BY 1.0 | 2026-08-30 |\n| `iof-core.rdf.gz` | https://spec.industrialontologies.org/ontology/core/Core/ | MIT | 2026-08-30 |\n\n## 更新一个包\n\n重新抓取、`gzip -9c` 覆盖、更新 `ontology_packs.rs` 里的计数与本文件的抓取日。\n**不要改动原文内容**——投影只覆盖当下能消费的部分，原文保真是 [0001](../../../docs/decisions/0001-ontology-import-and-governance.md) 判据 1。\n", + "2460624379264fed241f0986cd7dda7a5d86c104": "# OCR page fixture\n\n`ark-ocr-pages.pdf` is a synthetic three-page PDF created for OCR acceptance tests.\nPages 1 and 3 contain rasterized Chinese and English text, dates, amounts, a small\ntable, an instruction printed as source text, and an unlabelled line chart. Page 2\nis blank. There is no text layer. All content is authored for this test and carries\nno third-party or private data.\n\nThe fixture tests real Poppler rendering, original page numbering after a blank\npage, and restarting a failed pass from page 1. Mock replies test those contracts;\nthey do not measure a model's recognition accuracy.\n\nWith `ARK_API_KEY` set, run the manual Agent Plan acceptance test:\n\n```sh\ncargo test -p utopia-server live_agent_plan_reads_scans -- --ignored --nocapture\n```\n\nThis sends the probe and this fixture to the real API. Optional\n`ARK_OCR_TEST_PDF` adds a public scan, and `ARK_OCR_TEST_OUTPUT` saves its text and\nprovenance without credentials. `ARK_OCR_TEST_BASE_URL` and\n`ARK_OCR_TEST_MODEL` override the Agent Plan defaults.\n" + } +} diff --git a/crates/utopia-store/tests/store/a_viewer_never_sees_a_credential.rs b/crates/utopia-store/tests/store/a_viewer_never_sees_a_credential.rs index 269f9feb9..68674ec1b 100644 --- a/crates/utopia-store/tests/store/a_viewer_never_sees_a_credential.rs +++ b/crates/utopia-store/tests/store/a_viewer_never_sees_a_credential.rs @@ -56,6 +56,10 @@ async fn a_viewer_never_sees_a_credential() -> anyhow::Result<()> { "github_issues", serde_json::json!({ "repo": "o/r", "auth_header": "Bearer g" }), ), + ( + "github_repo", + serde_json::json!({ "repo": "o/r", "auth_header": "Bearer r" }), + ), ( "jira_issues", serde_json::json!({ "base_url": "https://j.test", "project": "P", "auth_header": "Basic j" }), diff --git a/docs/design/sources.md b/docs/design/sources.md index bbc34da19..d543cec15 100644 --- a/docs/design/sources.md +++ b/docs/design/sources.md @@ -19,7 +19,9 @@ document date, so an upload-dated document counts as undated for attestation and web list derive from; credentials never leave the server (`SOURCE_SECRET_KEYS`) and a blank field keeps the stored value; a truncated round says so [0013]. A ticket is one document whose body carries its history as dated declarative sentences and a `## History` list; GitHub fetches events -per ticket; timestamps are written to the second [0013]. +per ticket; timestamps are written to the second [0013]. A GitHub repository file is one document +keyed by its path and dated by the last commit that touched it; a file whose blob SHA has not +changed is not fetched again, and identical files at several paths go in once [#1303]. **Statements.** A `statements` source takes the open extraction contract itself (`e`, `s`, `n`) on `POST /sources/{id}/statements`, with the `api` push's identity, versions, tombstones and run history; the payload is the document and its single chunk, and extraction parses the chunk instead of prompting a model, then runs the unchanged path, so a pushed statement is an open statement with the item as its evidence (null offsets) and reaches the typed graph only through alignment. The contract has no slot for a property or a class and unknown keys are refused; a second push under the same identity marks the earlier statements stale rather than closing them; tables belong on a mount, not here [0054]. diff --git a/web/src/i18n/en.ts b/web/src/i18n/en.ts index a7109fcb7..71284011c 100644 --- a/web/src/i18n/en.ts +++ b/web/src/i18n/en.ts @@ -640,6 +640,7 @@ export const en = { statements: "Statements", custom: "Custom", github_issues: "GitHub issues", + github_repo: "GitHub repository files", jira_issues: "Jira issues", s3: "S3 / MinIO", azure_blob: "Azure Blob", @@ -659,6 +660,9 @@ export const en = { "Syncs a repository's issues. **Each ticket, together with its state history**, becomes " + "one document — when it was opened, closed, relabelled, reassigned, all dated. " + "Without a token GitHub allows only 60 requests an hour.", + github_repo: + "Reads markdown files off a repository's default branch. Each file becomes one " + + "document dated by its last commit; unchanged files are not fetched again.", s3: "Reads documents out of an S3 bucket, or anything speaking the same protocol " + "(MinIO, Ceph, R2). **Leave the endpoint empty for AWS**; fill it in for a " + @@ -756,6 +760,8 @@ export const en = { notionQueryField: "Search term (optional — empty takes every shared page)", tokenField: "GitHub token (optional, never shown again)", includePullRequests: "Treat pull requests as tickets too", + includeGlobsField: "Include globs (comma-separated, optional)", + excludeGlobsField: "Exclude globs (comma-separated, optional)", interval: "Sync schedule", intervalManual: "Manual only", intervalEvery: (m: number) => diff --git a/web/src/i18n/zh.ts b/web/src/i18n/zh.ts index 9222d4edc..e74856650 100644 --- a/web/src/i18n/zh.ts +++ b/web/src/i18n/zh.ts @@ -579,6 +579,7 @@ export const zh: Strings = { statements: "陈述", custom: "自定义", github_issues: "GitHub 工单", + github_repo: "GitHub 仓库文件", jira_issues: "Jira 工单", s3: "S3 / MinIO", azure_blob: "Azure Blob", @@ -597,6 +598,9 @@ export const zh: Strings = { "同步一个仓库的工单。**每张工单连同它的状态变更史**成为一篇文档——" + "何时开出、何时关闭、标签与负责人怎么变的，都带着日期。未配令牌时" + "每小时只有 60 次请求配额。", + github_repo: + "读取一个仓库默认分支上的 markdown 文件。每个文件成为一篇文档，" + + "日期取最后一次改动它的提交；没变的文件不会重抓。", s3: "从 S3 桶里读文档，或者任何说同一套协议的东西（MinIO、Ceph、R2）。" + "**端点留空就是 AWS**，填了就是自建。每个对象成为一篇文档，" + @@ -689,6 +693,8 @@ export const zh: Strings = { notionQueryField: "搜索词（可选——留空则取全部已分享页面）", tokenField: "GitHub 令牌（可选，不再显示）", includePullRequests: "把 PR 也当工单收进来", + includeGlobsField: "包含的 glob（逗号分隔，可选）", + excludeGlobsField: "排除的 glob（逗号分隔，可选）", interval: "同步计划", intervalManual: "仅手动", intervalEvery: (m: number) => diff --git a/web/src/pages/Library.tsx b/web/src/pages/Library.tsx index 3fe65b452..40e314a03 100644 --- a/web/src/pages/Library.tsx +++ b/web/src/pages/Library.tsx @@ -1360,6 +1360,8 @@ function SourceModal({ // PR 在 GitHub 的模型里也是工单。默认不收——问「工单」要的是工单； // 但有些仓库的决策记录实际写在 PR 描述里，所以给个开关 const [includePrs, setIncludePrs] = useState(false); + const [includeGlobs, setIncludeGlobs] = useState(""); + const [excludeGlobs, setExcludeGlobs] = useState(""); const [schedule, setSchedule] = useState({ sync_interval_minutes: null, sync_cron: null, @@ -1370,6 +1372,7 @@ function SourceModal({ kind === "rss" || kind === "custom" || kind === "github_issues" || + kind === "github_repo" || kind === "jira_issues" || kind === "s3" || kind === "azure_blob" || @@ -1379,6 +1382,11 @@ function SourceModal({ const create = useMutation({ mutationFn: () => { + // 留空就不发，后端按默认的 markdown 规则筛 + const globs = (key: string, value: string) => { + const list = value.split(",").map((g) => g.trim()).filter(Boolean); + return list.length ? { [key]: list } : {}; + }; const config = kind === "url" ? { urls: urls.split("\n").map((u) => u.trim()).filter(Boolean) } @@ -1389,11 +1397,13 @@ function SourceModal({ endpoint: endpoint.trim(), ...(authHeader.trim() ? { auth_header: authHeader.trim() } : {}), } - : kind === "github_issues" + : kind === "github_issues" || kind === "github_repo" ? { repo: repo.trim(), ...(authHeader.trim() ? { auth_header: authHeader.trim() } : {}), - ...(includePrs ? { include_pull_requests: true } : {}), + ...(kind === "github_issues" + ? includePrs ? { include_pull_requests: true } : {} + : { ...globs("include", includeGlobs), ...globs("exclude", excludeGlobs) }), } : kind === "jira_issues" ? { @@ -1799,7 +1809,7 @@ function SourceModal({ )} )} - {kind === "github_issues" && ( + {(kind === "github_issues" || kind === "github_repo") && ( <> {field( S.library.repoField, @@ -1818,12 +1828,33 @@ function SourceModal({ onChange={(e) => setAuthHeader(e.target.value)} />, )} - setIncludePrs(v)} - label={S.library.includePullRequests} - /> + {kind === "github_issues" ? ( + setIncludePrs(v)} + label={S.library.includePullRequests} + /> + ) : ( + <> + {field( + S.library.includeGlobsField, + setIncludeGlobs(e.target.value)} + />, + )} + {field( + S.library.excludeGlobsField, + setExcludeGlobs(e.target.value)} + />, + )} + + )} )} diff --git a/web/src/pages/SourcesRail.tsx b/web/src/pages/SourcesRail.tsx index 9929bb0b0..e881dfbad 100644 --- a/web/src/pages/SourcesRail.tsx +++ b/web/src/pages/SourcesRail.tsx @@ -72,6 +72,7 @@ export const KIND_ICON: Record = { statements: Braces, custom: Puzzle, github_issues: CircleDot, + github_repo: FileText, jira_issues: SquareKanban, s3: HardDrive, azure_blob: Cloud, @@ -88,6 +89,7 @@ export const SYNCING_KINDS = new Set([ "rss", "custom", "github_issues", + "github_repo", "jira_issues", "s3", "azure_blob", diff --git a/web/src/sourceKinds.ts b/web/src/sourceKinds.ts index c6e30af41..35c82f18b 100644 --- a/web/src/sourceKinds.ts +++ b/web/src/sourceKinds.ts @@ -13,6 +13,7 @@ export const CREATABLE_SOURCE_KINDS = [ "url", "rss", "github_issues", + "github_repo", "jira_issues", "s3", "azure_blob",