diff --git a/crates/utopia-cli/src/main.rs b/crates/utopia-cli/src/main.rs index 0d41dcf34..8f0fe0364 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 = 104; +const CURRENT_SCHEMA_VERSION: u32 = 105; fn main() -> anyhow::Result<()> { dotenvy::dotenv().ok(); diff --git a/crates/utopia-core/src/models.rs b/crates/utopia-core/src/models.rs index a32be0210..1054757ba 100644 --- a/crates/utopia-core/src/models.rs +++ b/crates/utopia-core/src/models.rs @@ -1134,6 +1134,8 @@ pub struct KnowledgeBase { /// 治理开关(0025,**缺省开**,见 0050):开着,govern 任务按先进先出过等人的 /// 重复对,先读台账里人的先例再裁;关掉,任务在两簇之间看到就停 pub governance: bool, + /// Opt-in confirmation of well-grounded exploration definitions (0011, #504). + pub auto_confirm_mappings: bool, /// 这次打开治理的时刻;保险丝只数它之后的撤回(0025 决定 9)。 /// 新库生下来就开着治理,这一格于是等于建库的时刻(0050) pub governance_since: Option>, @@ -1265,6 +1267,9 @@ pub struct ConceptMapping { /// 人从零写的口径记写的人;探索提的为空(#562)。`decided_by` 分不出这件事—— /// 探索提的经人确认之后同样有 decided_by pub written_by: Option, + pub decided_by: Option, + pub confidence: Option, + pub rationale: Option, } /// 一处公理违规,配好展示所需的三元组文本(见 `axiom_violations`)。 diff --git a/crates/utopia-server/src/api/kbs.rs b/crates/utopia-server/src/api/kbs.rs index 9937d5070..ca1daed99 100644 --- a/crates/utopia-server/src/api/kbs.rs +++ b/crates/utopia-server/src/api/kbs.rs @@ -51,6 +51,8 @@ pub struct UpdateKbReq { /// 打开那一刻就排一轮,关掉后任务在两簇之间看到就停。见 docs/decisions/0025 #[serde(default)] pub governance: Option, + #[serde(default)] + pub auto_confirm_mappings: Option, /// 人写的数据约定(「测试单不算数」这类 schema 里没有的规则),问数与探索的 /// 提示词都读它。探索生成的描述在另一个字段,PATCH 不了。见 #570 #[serde(default)] @@ -125,6 +127,7 @@ pub async fn create( None, None, None, + None, ) .await?; } @@ -193,7 +196,29 @@ pub async fn update( Path(id): Path, Json(req): Json, ) -> ApiResult> { - kb_with_role(&state, &user, id, Role::Admin).await?; + // Editors may opt in; a mixed request still needs the usual Admin role. + let setting_only = req.auto_confirm_mappings.is_some() + && req.name.is_none() + && req.description.is_none() + && req.visibility.is_none() + && req.auto_extend_ontology.is_none() + && req.ontology_lang.is_none() + && req.materialize_inferences.is_none() + && req.inference_interval_minutes.is_none() + && req.auto_type_resolution.is_none() + && req.governance.is_none() + && req.data_conventions.is_none(); + kb_with_role( + &state, + &user, + id, + if setting_only { + Role::Editor + } else { + Role::Admin + }, + ) + .await?; let kb = utopia_store::kbs::update( &state.pool, id, @@ -207,6 +232,7 @@ pub async fn update( req.auto_type_resolution, req.governance, req.data_conventions.as_deref().map(str::trim), + req.auto_confirm_mappings, ) .await?; // 打开开关就自动开始处理:排一轮,同库已排着的不重复 @@ -227,6 +253,7 @@ pub async fn update( "name": req.name, "visibility": req.visibility, "governance": req.governance, + "auto_confirm_mappings": req.auto_confirm_mappings, "data_conventions": req.data_conventions.is_some(), }), ) @@ -460,3 +487,82 @@ pub async fn list_packs(AuthUser(_): AuthUser) -> ApiResult anyhow::Result<()> { + let Some(url) = utopia_store::test_db::url() else { + return Ok(()); + }; + let pool = sqlx::PgPool::connect(&url).await?; + utopia_store::db::migrate(&pool).await?; + let [org, ws, kb, other, admin, editor, viewer] = std::array::from_fn(|_| Uuid::now_v7()); + sqlx::raw_sql(&format!( + "INSERT INTO organizations(id,name) VALUES ('{org}','auto-mapping-route'); + INSERT INTO workspaces(id,org_id,name) VALUES ('{ws}','{org}','auto-mapping-route'); + INSERT INTO knowledge_bases(id,workspace_id,name,visibility) VALUES + ('{kb}','{ws}','mappings','restricted'), ('{other}','{ws}','other','restricted'); + INSERT INTO users(id,org_id,email,display_name,password_hash) VALUES + ('{admin}','{org}','{admin}@map.test','Admin','unused'), + ('{editor}','{org}','{editor}@map.test','Editor','unused'), + ('{viewer}','{org}','{viewer}@map.test','Viewer','unused'); + INSERT INTO kb_members(kb_id,user_id,role) VALUES + ('{kb}','{admin}','admin'), ('{kb}','{editor}','editor'), ('{kb}','{viewer}','viewer');" + )).execute(&pool).await?; + let result = async { + 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()); + let patch = |user, base, body: serde_json::Value| { + let state = state.clone(); + let cfg = cfg.clone(); + async move { + let token = crate::auth::issue_token(&state, user)?; + let response = crate::api::router(state, &cfg).oneshot( + Request::patch(format!("/api/v1/kbs/{base}")) + .header("authorization", format!("Bearer {token}")) + .header("content-type", "application/json") + .body(Body::from(body.to_string()))? + ).await?; + let status = response.status(); + let body: serde_json::Value = serde_json::from_slice(&to_bytes(response.into_body(), 1024*1024).await?)?; + anyhow::Ok((status, body)) + } + }; + assert!(!utopia_store::kbs::get(&pool, kb).await?.auto_confirm_mappings); + let (status, body) = patch(editor, kb, json!({"auto_confirm_mappings": true})).await?; + assert_eq!(status, StatusCode::OK); + assert_eq!(body["auto_confirm_mappings"], true); + assert_eq!(patch(viewer, kb, json!({"auto_confirm_mappings": false})).await?.0, StatusCode::FORBIDDEN); + assert_eq!(patch(editor, other, json!({"auto_confirm_mappings": true})).await?.0, StatusCode::NOT_FOUND); + assert_eq!(patch(editor, kb, json!({"auto_confirm_mappings": false, "name": "Changed"})).await?.0, StatusCode::FORBIDDEN); + assert!(utopia_store::kbs::get(&pool, kb).await?.auto_confirm_mappings); + assert_eq!(patch(admin, kb, json!({"description": "Changed"})).await?.0, StatusCode::OK); + assert!(utopia_store::kbs::get(&pool, kb).await?.auto_confirm_mappings); + assert_eq!(patch(editor, kb, json!({"auto_confirm_mappings": false})).await?.0, StatusCode::OK); + let events: Vec = sqlx::query_scalar( + "SELECT detail FROM audit_events WHERE kb_id = $1 AND actor_id = $2 AND action = 'kb.updated' ORDER BY created_at" + ).bind(kb).bind(editor).fetch_all(&pool).await?; + assert_eq!(events.len(), 2); + assert_eq!(events[0]["auto_confirm_mappings"], true); + assert_eq!(events[1]["auto_confirm_mappings"], false); + anyhow::Ok(()) + }.await; + sqlx::query("DELETE FROM knowledge_bases WHERE id = ANY($1)") + .bind(&[kb, other][..]) + .execute(&pool) + .await?; + result + } +} diff --git a/crates/utopia-server/src/mappings.rs b/crates/utopia-server/src/mappings.rs index c10c5814e..e9c816d89 100644 --- a/crates/utopia-server/src/mappings.rs +++ b/crates/utopia-server/src/mappings.rs @@ -1,13 +1,13 @@ //! 问数语义映射的 Agentic 探索:读挂载源的 schema + KB 既有概念,让 LLM 提议 //! "业务概念(Metric/Dimension 实体)→ 数据资产定义" 的映射。 //! -//! 提议写进 `concept_mappings`(status = proposed)→ Review 页自成一档, -//! Confirm / Reject 之后 status 变 confirmed,问数只读确认过的那些。 +//! 提议写进 `concept_mappings`,数据映射页确认或拒绝;问数只读已确认的定义。 +//! 库可选择自动确认有置信度与理由支撑的定义(0011,#504,默认关)。 //! //! **从前它是一条 0.6 置信的 `mapped_to` 事实**,借「低置信事实」那一档露面。 //! 搬出来的理由见 0011:它不是关于世界的断言,是配置——而「确认」这个动作 //! 当时是 `UPDATE facts SET confidence = 1.0`,原地改一张不许原地改的表。 -//! agent 只提议,口径生效权在人——与消解"宁分勿合"同一哲学。 +//! 人的决定与修改不会被后续探索覆盖;自动确认保留模型的理由与审计。 use crate::llm_util; use crate::state::AppState; @@ -311,9 +311,11 @@ async fn explore(state: &AppState, kb_id: Uuid, run: Uuid) -> anyhow::Result<()> \"definition\": {{\"table\": \"schema.table\", \"expr\": \"SQL expression\", \ \"sql\": \"full SELECT if joins are needed (optional)\", \"unit\": \"optional\"}}, \ \"summary\": \"one line: source + expression, shown to reviewers\", \ - \"rationale\": \"why this mapping, citing column comments\"}}\n\ + \"confidence\": 0.0, \"rationale\": \"why this mapping, citing column comments\"}}\n\ Metrics are aggregatable quantities (use sum/count/avg in expr); dimensions are \ - group-by columns. Propose at most {cap}, only well-grounded ones.", + group-by columns. Confidence is a number from 0 to 1 measuring whether the exact \ + definition follows the schema and stated conventions; reduce it when business meaning \ + is uncertain. Propose at most {cap}, only well-grounded ones.", if existing_names.is_empty() { "(none)".into() } else { @@ -428,7 +430,7 @@ async fn explore(state: &AppState, kb_id: Uuid, run: Uuid) -> anyhow::Result<()> if let Some(t) = s("table") { covered.insert(t); } - let (_, written) = utopia_store::mappings::propose( + let (_, written) = utopia_store::mappings::propose_with_evidence( &state.pool, kb_id, resolved.entity_id, @@ -440,6 +442,8 @@ async fn explore(state: &AppState, kb_id: Uuid, run: Uuid) -> anyhow::Result<()> // summary 是给人看的那句:Review 列表与问数 prompt 都靠它 p["summary"].as_str().or(def["summary"].as_str()), def["derived"].as_bool().unwrap_or(false), + p["confidence"].as_f64(), + p["rationale"].as_str(), ) .await?; if written { @@ -619,4 +623,120 @@ mod tests { // **0 条上限会把「连不上」变成「模型什么都没提」**,两件事又混在一起了 assert_eq!(proposal_cap(0), 12); } + #[tokio::test] + async fn exploration_keeps_the_models_evidence_and_confirms_only_after_opt_in( + ) -> anyhow::Result<()> { + use serde_json::json; + use std::sync::Arc; + use uuid::Uuid; + use wiremock::{ + matchers::{body_string_contains, method, path}, + Mock, MockServer, ResponseTemplate, + }; + let Some(url) = utopia_store::test_db::url() else { + return Ok(()); + }; + let pool = sqlx::PgPool::connect(&url).await?; + utopia_store::db::migrate(&pool).await?; + let [org, ws, kb, user] = std::array::from_fn(|_| Uuid::now_v7()); + sqlx::raw_sql(&format!( + "INSERT INTO organizations(id,name) VALUES ('{org}','mapping-exploration'); + INSERT INTO workspaces(id,org_id,name) VALUES ('{ws}','{org}','mapping-exploration'); + INSERT INTO knowledge_bases(id,workspace_id,name) VALUES ('{kb}','{ws}','mapping-exploration'); + INSERT INTO users(id,org_id,email,display_name,password_hash) + VALUES ('{user}','{org}','{user}@mapping.test','Editor','unused');" + )).execute(&pool).await?; + let model = MockServer::start().await; + let proposals = json!([{"name":"Source rows", "kind":"metric", "source":"test", + "definition":{"table":"public.entities", "expr":"COUNT(*)"}, + "confidence":0.8, "rationale":"Count each entity row"}]) + .to_string(); + Mock::given(method("POST")) + .and(path("/chat/completions")) + .and(body_string_contains("ONLY a JSON array")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "choices":[{"message":{"role":"assistant","content":proposals}}] + }))) + .mount(&model) + .await; + Mock::given(method("POST")) + .and(path("/chat/completions")) + .and(body_string_contains("ONLY a JSON object")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "choices":[{"message":{"role":"assistant","content":"{}"}}] + }))) + .mount(&model) + .await; + let source = utopia_store::datasources::create( + &pool, + &format!("mapping-{}", kb.simple()), + "postgres", + &url, + user, + ) + .await?; + let result = async { + sqlx::query("INSERT INTO kb_data_sources(kb_id,data_source_id) VALUES ($1,$2)") + .bind(kb) + .bind(source) + .execute(&pool) + .await?; + utopia_store::settings::upsert( + &pool, + ws, + Some(&model.uri()), + None, + Some("scripted"), + None, + None, + None, + None, + ) + .await?; + 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(), + ); + super::explore_mappings(&state, kb).await?; + let pending = utopia_store::mappings::proposed(&pool, kb, 10, 0) + .await? + .remove(0); + assert_eq!(pending.confidence, Some(0.8)); + assert_eq!(pending.rationale.as_deref(), Some("Count each entity row")); + sqlx::query("UPDATE knowledge_bases SET auto_confirm_mappings = true WHERE id = $1") + .bind(kb) + .execute(&pool) + .await?; + super::explore_mappings(&state, kb).await?; + let effective = utopia_store::mappings::confirmed(&pool, kb, 10) + .await? + .remove(0); + assert_eq!(effective.id, pending.id); + assert_eq!(effective.decided_by, None); + let requests = model.received_requests().await.unwrap(); + let prompt = requests + .iter() + .find(|r| String::from_utf8_lossy(&r.body).contains("ONLY a JSON array")) + .unwrap(); + assert!(String::from_utf8_lossy(&prompt.body).contains("confidence")); + assert_eq!(requests.len(), 4); + anyhow::Ok(()) + } + .await; + sqlx::query("DELETE FROM knowledge_bases WHERE id = $1") + .bind(kb) + .execute(&pool) + .await?; + utopia_store::datasources::delete(&pool, source).await?; + result + } } diff --git a/crates/utopia-store/src/kbs.rs b/crates/utopia-store/src/kbs.rs index af2baec83..bb43ab98c 100644 --- a/crates/utopia-store/src/kbs.rs +++ b/crates/utopia-store/src/kbs.rs @@ -110,6 +110,7 @@ pub async fn update( auto_type_resolution: Option, governance: Option, data_conventions: Option<&str>, + auto_confirm_mappings: Option, ) -> AppResult { // 改语言不回头重写已有的类——它们已经是这个库的数据,可能有人手工调过。 // 这一列往后管的是**新**描述(自动扩本体、AI 建议)写成什么语言 @@ -151,6 +152,7 @@ pub async fn update( ELSE governance_since END, -- 人写的约定;清空要送空串,送 null 等于不改(与 description 同一约定) data_conventions = COALESCE($11, data_conventions), + auto_confirm_mappings = COALESCE($12, auto_confirm_mappings), updated_at = now() WHERE id = $1 RETURNING *", ) @@ -165,6 +167,7 @@ pub async fn update( .bind(auto_type_resolution) .bind(governance) .bind(data_conventions) + .bind(auto_confirm_mappings) .fetch_optional(pool) .await? .ok_or(AppError::NotFound) diff --git a/crates/utopia-store/src/mappings.rs b/crates/utopia-store/src/mappings.rs index 1f5f5e207..99fbc4e13 100644 --- a/crates/utopia-store/src/mappings.rs +++ b/crates/utopia-store/src/mappings.rs @@ -33,35 +33,65 @@ pub async fn propose( summary: Option<&str>, derived: bool, ) -> AppResult<(Uuid, bool)> { - // **`DO UPDATE ... WHERE` 不满足时 `RETURNING` 一行都不返回。** - // - // 这是 Postgres 的实情而不是直觉:条件挡住更新,那一行就不算被这条语句 - // 动过,于是也不出现在 RETURNING 里。测试当场撞上——第二次提议一条已被 - // 拒绝的映射,`fetch_one` 报「no rows returned」。 - // - // 所以把 id 单独查出来:更新与否是一回事,「这条映射是哪一行」是另一回事。 - let existing: Option<(Uuid,)> = sqlx::query_as( - "SELECT id FROM concept_mappings - WHERE kb_id = $1 AND concept_id = $2 AND source = $3", + propose_with_evidence( + pool, kb_id, concept_id, source, table_name, expr, sql, unit, summary, derived, None, None, + ) + .await +} + +/// Re-evaluate the opt-in on every exploration write. An automatic definition can improve; +/// a human decision (even an edit of a pending proposal) cannot be replaced by exploration. +#[allow(clippy::too_many_arguments)] +pub async fn propose_with_evidence( + pool: &PgPool, + kb_id: Uuid, + concept_id: Uuid, + source: &str, + table_name: Option<&str>, + expr: Option<&str>, + sql: Option<&str>, + unit: Option<&str>, + summary: Option<&str>, + derived: bool, + confidence: Option, + rationale: Option<&str>, +) -> AppResult<(Uuid, bool)> { + let confidence = confidence.filter(|c| c.is_finite() && (0.0..=1.0).contains(c)); + let rationale = rationale.map(str::trim).filter(|s| !s.is_empty()); + let mut tx = pool.begin().await?; + // A switch changed during a long model call must apply to its eventual write. SHARE + // also orders this write against a concurrent opt-out without serializing explorations. + let enabled: bool = sqlx::query_scalar( + "SELECT auto_confirm_mappings FROM knowledge_bases WHERE id = $1 FOR SHARE", ) .bind(kb_id) - .bind(concept_id) - .bind(source) - .fetch_optional(pool) + .fetch_one(&mut *tx) .await?; - let id = existing.map(|(i,)| i).unwrap_or_else(Uuid::now_v7); - let written = sqlx::query( + let automatic = enabled + && confidence.is_some_and(|c| c >= 0.8) + && rationale.is_some() + && [expr, sql] + .into_iter() + .flatten() + .any(|s| !s.trim().is_empty()); + let status = if automatic { "confirmed" } else { "proposed" }; + let written: Option = sqlx::query_scalar( "INSERT INTO concept_mappings - (id, kb_id, concept_id, source, table_name, expr, sql, unit, summary, derived) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) + (id, kb_id, concept_id, source, table_name, expr, sql, unit, summary, derived, + confidence, rationale, status, decided_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, + CASE WHEN $13 = 'confirmed' THEN now() END) ON CONFLICT (kb_id, concept_id, source) DO UPDATE SET table_name = EXCLUDED.table_name, expr = EXCLUDED.expr, sql = EXCLUDED.sql, unit = EXCLUDED.unit, summary = EXCLUDED.summary, derived = EXCLUDED.derived, - updated_at = now() - WHERE concept_mappings.status = 'proposed'", + confidence = EXCLUDED.confidence, rationale = EXCLUDED.rationale, + status = EXCLUDED.status, decided_at = EXCLUDED.decided_at, updated_at = now() + WHERE concept_mappings.decided_by IS NULL + AND concept_mappings.status IN ('proposed', 'confirmed') + RETURNING id", ) - .bind(id) + .bind(Uuid::now_v7()) .bind(kb_id) .bind(concept_id) .bind(source) @@ -71,12 +101,45 @@ pub async fn propose( .bind(unit) .bind(summary) .bind(derived) - .execute(pool) - .await? - .rows_affected(); - // 插入或刷新了算写入;撞上已确认 / 已拒绝的那一行,WHERE 挡下更新,写入数为零—— - // 调用方靠这一位区分「进了待看」与「决定还在」 - Ok((id, written > 0)) + .bind(confidence) + .bind(rationale) + .bind(status) + .fetch_optional(&mut *tx) + .await?; + // RETURNING supplies the actual row id after concurrent inserts. A protected row + // returns nothing, so read its id separately without reporting a successful write. + let id = match written { + Some(id) => id, + None => sqlx::query_scalar( + "SELECT id FROM concept_mappings WHERE kb_id = $1 AND concept_id = $2 AND source = $3", + ) + .bind(kb_id) + .bind(concept_id) + .bind(source) + .fetch_one(&mut *tx) + .await?, + }; + if written.is_some() && automatic { + // Automation changes the effective definition: losing this event must roll back + // that change, rather than leave an effective definition nobody can account for. + sqlx::query( + "INSERT INTO audit_events (id, kb_id, actor_id, action, target_kind, target_id, detail) + VALUES ($1, $2, NULL, 'mapping.auto_confirmed', 'mapping', $3, $4)", + ) + .bind(Uuid::now_v7()) + .bind(kb_id) + .bind(id) + .bind(serde_json::json!({ + "concept_id": concept_id, "source": source, "confidence": confidence, + "rationale": rationale, "threshold": 0.8, + "definition": { "table": table_name, "expr": expr, "sql": sql, + "unit": unit, "summary": summary, "derived": derived } + })) + .execute(&mut *tx) + .await?; + } + tx.commit().await?; + Ok((id, written.is_some())) } /// 人从零写一条口径(#562)。**落下来就是确认的**:写的人就是表态的人, @@ -87,7 +150,7 @@ pub async fn propose( /// /// 同一个 (概念, 源) 已经有一条时报冲突,而不是悄悄覆盖:那一条可能是探索提的、 /// 人已经确认过的,覆盖等于抹掉一次表态。要改用 `revise`。 -/// 探索反过来也盖不掉这条:`propose` 只刷新 `proposed` 的行。 +/// 探索反过来也盖不掉这条:`propose` 只刷新没有人工决定的行。 #[allow(clippy::too_many_arguments)] pub async fn create( pool: &PgPool, @@ -140,7 +203,8 @@ pub async fn proposed( ) -> AppResult> { Ok(sqlx::query_as( "SELECT m.id, m.concept_id, e.canonical_name AS concept_name, m.source, - m.table_name, m.expr, m.sql, m.unit, m.summary, m.derived, m.status, m.written_by + m.table_name, m.expr, m.sql, m.unit, m.summary, m.derived, m.status, m.written_by, + m.decided_by, m.confidence, m.rationale FROM concept_mappings m JOIN entities e ON e.id = m.concept_id WHERE m.kb_id = $1 AND m.status = 'proposed' @@ -154,12 +218,13 @@ pub async fn proposed( .await?) } -/// 人确认过的。问数把它们注进 system prompt——**只用确认过的口径**, +/// 已生效的(人工或自动确认)。问数把它们注进 system prompt——**只用确认过的口径**, /// 而不是每次从 schema 猜。 pub async fn confirmed(pool: &PgPool, kb_id: Uuid, limit: i64) -> AppResult> { Ok(sqlx::query_as( "SELECT m.id, m.concept_id, e.canonical_name AS concept_name, m.source, - m.table_name, m.expr, m.sql, m.unit, m.summary, m.derived, m.status, m.written_by + m.table_name, m.expr, m.sql, m.unit, m.summary, m.derived, m.status, m.written_by, + m.decided_by, m.confidence, m.rationale FROM concept_mappings m JOIN entities e ON e.id = m.concept_id WHERE m.kb_id = $1 AND m.status = 'confirmed' @@ -221,7 +286,7 @@ pub async fn revise( let mut tx = pool.begin().await?; let before: Option<(serde_json::Value,)> = sqlx::query_as( "SELECT to_jsonb(m) - 'id' - 'kb_id' FROM concept_mappings m - WHERE m.id = $2 AND m.kb_id = $1", + WHERE m.id = $2 AND m.kb_id = $1 FOR UPDATE", ) .bind(kb_id) .bind(mapping_id) @@ -244,7 +309,8 @@ pub async fn revise( sqlx::query( "UPDATE concept_mappings SET table_name = $3, expr = $4, sql = $5, unit = $6, summary = $7, - derived = $8, updated_at = now() + derived = $8, decided_by = $9, decided_at = now(), + confidence = NULL, rationale = NULL, updated_at = now() WHERE id = $2 AND kb_id = $1", ) .bind(kb_id) @@ -255,6 +321,7 @@ pub async fn revise( .bind(unit) .bind(summary) .bind(derived) + .bind(actor) .execute(&mut *tx) .await?; tx.commit().await?; @@ -284,7 +351,8 @@ pub async fn page( OR m.table_name ILIKE '%' || $3 || '%')"; let rows: Vec = sqlx::query_as(&format!( "SELECT m.id, m.concept_id, e.canonical_name AS concept_name, m.source, - m.table_name, m.expr, m.sql, m.unit, m.summary, m.derived, m.status, m.written_by + m.table_name, m.expr, m.sql, m.unit, m.summary, m.derived, m.status, m.written_by, + m.decided_by, m.confidence, m.rationale FROM concept_mappings m JOIN entities e ON e.id = m.concept_id {WHERE} @@ -497,7 +565,8 @@ pub async fn confirmed_texts(pool: &PgPool, kb_id: Uuid) -> AppResult AppResult> { let rows: Vec = sqlx::query_as( "SELECT m.id, m.concept_id, e.canonical_name AS concept_name, m.source, - m.table_name, m.expr, m.sql, m.unit, m.summary, m.derived, m.status, m.written_by + m.table_name, m.expr, m.sql, m.unit, m.summary, m.derived, m.status, m.written_by, + m.decided_by, m.confidence, m.rationale FROM concept_mappings m JOIN entities e ON e.id = m.concept_id WHERE m.kb_id = $1 AND m.id = ANY($2)", ) diff --git a/crates/utopia-store/tests/store/a_mapping_is_not_a_fact.rs b/crates/utopia-store/tests/store/a_mapping_is_not_a_fact.rs index 7f65fde4f..ccb0cd52a 100644 --- a/crates/utopia-store/tests/store/a_mapping_is_not_a_fact.rs +++ b/crates/utopia-store/tests/store/a_mapping_is_not_a_fact.rs @@ -187,3 +187,130 @@ async fn a_rejected_mapping_does_not_come_back() -> anyhow::Result<()> { .await?; run } + +#[tokio::test] +async fn automatic_definitions_refresh_until_a_person_decides() -> anyhow::Result<()> { + let Some(url) = utopia_store::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + utopia_store::db::migrate(&pool).await?; + let (org, kb, ent) = fixture(&pool).await?; + let run = async { + let user = Uuid::now_v7(); + sqlx::query("INSERT INTO users (id, org_id, email, display_name, password_hash) + VALUES ($1, $2, $1 || '@auto-map.test', 'Editor', 'unused')") + .bind(user).bind(org).execute(&pool).await?; + let p = |confidence, rationale, expr| utopia_store::mappings::propose_with_evidence( + &pool, kb, ent, "warehouse", Some("orders"), expr, None, None, + Some("Revenue"), false, confidence, rationale); + assert!(!utopia_store::kbs::get(&pool, kb).await?.auto_confirm_mappings); + let (id, _) = p(Some(1.0), Some("The schema states revenue"), Some("SUM(amount)")).await?; + assert_eq!(utopia_store::mappings::proposed(&pool, kb, 10, 0).await?.len(), 1); + sqlx::query("UPDATE knowledge_bases SET auto_confirm_mappings = true WHERE id = $1") + .bind(kb).execute(&pool).await?; + // Neither missing/invalid evidence nor a table name on its own is enough to + // turn configuration on. Boundary 0.8 is accepted in the next write. + for (conf, reason, expr) in [ + (None, Some("Grounded"), Some("SUM(amount)")), + (Some(0.79), Some("Grounded"), Some("SUM(amount)")), + (Some(f64::NAN), Some("Grounded"), Some("SUM(amount)")), + (Some(1.1), Some("Grounded"), Some("SUM(amount)")), + (Some(0.9), Some(" "), Some("SUM(amount)")), + (Some(0.9), Some("Grounded"), None), + ] { + p(conf, reason, expr).await?; + assert!(utopia_store::mappings::confirmed(&pool, kb, 10).await?.is_empty()); + } + // The trigger is scoped to this fixture's base, so other tests can run alongside it. + let fault = format!("auto_mapping_fault_{}", kb.simple()); + sqlx::raw_sql(&format!( + "CREATE FUNCTION {fault}() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN + IF NEW.kb_id = '{kb}' AND NEW.action = 'mapping.auto_confirmed' THEN + RAISE EXCEPTION 'injected mapping audit failure'; END IF; RETURN NEW; END $$; + CREATE TRIGGER {fault} BEFORE INSERT ON audit_events + FOR EACH ROW EXECUTE FUNCTION {fault}();" + )).execute(&pool).await?; + let failed = p(Some(0.9), Some("Grounded"), Some("SUM(amount)")).await; + sqlx::raw_sql(&format!("DROP TRIGGER {fault} ON audit_events; DROP FUNCTION {fault}();")) + .execute(&pool).await?; + assert!(failed.is_err()); + assert!(utopia_store::mappings::confirmed(&pool, kb, 10).await?.is_empty()); + let pending = utopia_store::mappings::proposed(&pool, kb, 10, 0).await?.remove(0); + assert_eq!(pending.expr, None, "failed auto write must restore the preceding proposal"); + assert_eq!(p(Some(0.8), Some(" The schema states revenue "), Some("SUM(amount)")).await?, (id, true)); + let auto = utopia_store::mappings::confirmed(&pool, kb, 10).await?.remove(0); + assert_eq!(auto.confidence, Some(0.8)); + assert_eq!(auto.rationale.as_deref(), Some("The schema states revenue")); + assert_eq!(auto.decided_by, None); + assert_eq!(utopia_store::mappings::page(&pool, kb, None, None, 10, 0).await?.0[0].rationale, auto.rationale); + assert_eq!(utopia_store::mappings::by_ids(&pool, kb, &[id]).await?[0].confidence, Some(0.8)); + let events: Vec<(Option, serde_json::Value)> = sqlx::query_as( + "SELECT actor_id, detail FROM audit_events WHERE target_id = $1 AND action = 'mapping.auto_confirmed'" + ).bind(id).fetch_all(&pool).await?; + assert_eq!(events.len(), 1); + assert_eq!(events[0].0, None); + assert_eq!(events[0].1["definition"]["expr"], "SUM(amount)"); + assert_eq!(p(Some(0.9), Some("Updated schema"), Some("SUM(net_amount)")).await?, (id, true)); + assert_eq!(utopia_store::mappings::confirmed(&pool, kb, 10).await?[0].expr.as_deref(), Some("SUM(net_amount)")); + p(Some(0.7), Some("Uncertain"), Some("SUM(amount)")).await?; + assert!(utopia_store::mappings::confirmed(&pool, kb, 10).await?.is_empty()); + p(Some(0.9), Some("Grounded"), Some("SUM(amount)")).await?; + sqlx::query("UPDATE knowledge_bases SET auto_confirm_mappings = false WHERE id = $1") + .bind(kb).execute(&pool).await?; + assert_eq!(utopia_store::mappings::confirmed(&pool, kb, 10).await?.len(), 1); + p(Some(0.9), Some("Grounded"), Some("SUM(amount)")).await?; + assert!(utopia_store::mappings::confirmed(&pool, kb, 10).await?.is_empty()); + sqlx::query("UPDATE knowledge_bases SET auto_confirm_mappings = true WHERE id = $1") + .bind(kb).execute(&pool).await?; + // Editing a pending proposal also takes ownership. Its status remains pending. + utopia_store::mappings::revise(&pool, kb, id, Some("orders"), Some("SUM(human_amount)"), + None, None, None, false, user).await?; + assert_eq!(p(Some(1.0), Some("Grounded"), Some("SUM(amount)")).await?, (id, false)); + let human = utopia_store::mappings::proposed(&pool, kb, 10, 0).await?.remove(0); + assert_eq!(human.expr.as_deref(), Some("SUM(human_amount)")); + assert_eq!(human.decided_by, Some(user)); + assert_eq!(human.confidence, None, "model evidence described the preceding definition"); + assert_eq!(human.rationale, None); + let revisions = utopia_store::mappings::revisions(&pool, kb, id).await?; + assert_eq!(revisions.len(), 1); + assert_eq!(revisions[0].before["decided_by"], serde_json::Value::Null); + for status in ["confirmed", "rejected"] { + utopia_store::mappings::decide(&pool, kb, id, status, user).await?; + assert_eq!(p(Some(1.0), Some("Grounded"), Some("SUM(amount)")).await?, (id, false)); + } + anyhow::Ok(()) + }.await; + sqlx::query("DELETE FROM knowledge_bases WHERE id = $1") + .bind(kb) + .execute(&pool) + .await?; + run +} + +#[tokio::test] +async fn concurrent_automatic_proposals_return_the_stored_id() -> anyhow::Result<()> { + let Some(url) = utopia_store::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + let (_, kb, ent) = fixture(&pool).await?; + let run = async { + sqlx::query("UPDATE knowledge_bases SET auto_confirm_mappings = true WHERE id = $1") + .bind(kb).execute(&pool).await?; + let p = || utopia_store::mappings::propose_with_evidence(&pool, kb, ent, "warehouse", + None, Some("COUNT(*)"), None, None, None, false, Some(0.9), Some("Count the rows")); + let (a, b) = tokio::join!(p(), p()); + let (a, b) = (a?, b?); + assert_eq!(a, b); + let ids: Vec = sqlx::query_scalar("SELECT target_id FROM audit_events WHERE kb_id = $1 AND action = 'mapping.auto_confirmed'") + .bind(kb).fetch_all(&pool).await?; + assert_eq!(ids, vec![a.0, a.0]); + anyhow::Ok(()) + }.await; + sqlx::query("DELETE FROM knowledge_bases WHERE id = $1") + .bind(kb) + .execute(&pool) + .await?; + run +} diff --git a/crates/utopia-store/tests/store/exploration_describes_the_data.rs b/crates/utopia-store/tests/store/exploration_describes_the_data.rs index 20d564eac..fdf1dcc30 100644 --- a/crates/utopia-store/tests/store/exploration_describes_the_data.rs +++ b/crates/utopia-store/tests/store/exploration_describes_the_data.rs @@ -51,6 +51,7 @@ async fn the_description_and_the_conventions_do_not_overwrite_each_other() -> an None, None, Some("Amounts are in cents.\nis_test = 1 is excluded."), + None, ) .await?; // 探索写描述与问题 @@ -101,6 +102,7 @@ async fn the_description_and_the_conventions_do_not_overwrite_each_other() -> an None, None, Some("cents only"), + None, ) .await?; let got = utopia_store::kbs::get(&pool, kb).await?; @@ -119,6 +121,7 @@ async fn the_description_and_the_conventions_do_not_overwrite_each_other() -> an None, None, no, + None, ) .await?; assert_eq!( diff --git a/docs/decisions/0011-a-mapping-is-not-a-fact.md b/docs/decisions/0011-a-mapping-is-not-a-fact.md index ab1d03134..36f4efbac 100644 --- a/docs/decisions/0011-a-mapping-is-not-a-fact.md +++ b/docs/decisions/0011-a-mapping-is-not-a-fact.md @@ -70,6 +70,27 @@ rebuilding all three. ## Revisions +- 2026-10-20 (#504): a base may opt into automatic confirmation of exploration's + definitions with `auto_confirm_mappings`, default off and independent of governance. + An Editor may change this setting; other base settings keep their existing Admin gate. + Exploration keeps `confidence` (0–1) and `rationale` separately from status. A score + of at least 0.8, a nonempty rationale and an expression or SQL definition are required; + missing or invalid evidence leaves the proposal waiting for a person. This threshold + is an opt-in gate, not a measured guarantee of correctness or a reason to enable it by + default. No live-model accuracy improvement is claimed. + A confirmed row with no `decided_by` is automatic. Exploration may refresh it and + re-evaluates the gate on each write: weaker evidence or a disabled switch returns that + refreshed definition to proposed. Turning the switch off alone does not revoke existing + definitions. Confirming, rejecting or revising a row records the person and protects it + from further exploration, including a person editing a still-pending proposal. An edit + clears model evidence from the current definition; its history snapshot keeps the evidence + for the preceding definition. + The upsert judges that protection while holding the row lock; human revision locks the + row before taking its history snapshot. Each automatic write and its system audit event + (`mapping.auto_confirmed`, including the definition and evidence) commit together, so + audit failure cannot leave an unrecorded effective definition. Readers expose the evidence + and decision actor, and the mappings page marks automatic definitions. This applies to + the existing concept definitions; table-alignment adoption in 0036 remains its own decision. - 2026-09-02: the status line said "planned" while the record arrived in the same commit as its implementation. The status line is the implementing PR's responsibility; now a README rule. diff --git a/docs/design/lakehouse-and-actions.md b/docs/design/lakehouse-and-actions.md index 368d8cbce..7ac4bb0b6 100644 --- a/docs/design/lakehouse-and-actions.md +++ b/docs/design/lakehouse-and-actions.md @@ -22,7 +22,11 @@ against a real cluster; Databricks and Snowflake are covered by protocol replays **The semantic layer.** `concept_mappings` (concept, source, table, expr, sql, unit, summary, status, revisions) is configuration outside the ledger with its own page, proposed by exploration and confirmed by a person; rerunning exploration does not erase a rejection; a Viewer reads, an -Editor revises; chat reads confirmed mappings into its prompt, capped at 30 [0011]. A definition +Editor revises. A base may opt into automatic confirmation at confidence ≥ 0.8 with a +rationale and an expression or SQL (default off, #504); automatic definitions can be refreshed +but human decisions and edits remain protected. The mappings page shows the decision origin +and evidence; automatic writes commit with their audit event. Chat reads confirmed mappings +into its prompt, capped at 30 [0011]. A definition can be written by hand (#563), and a one-page conventions document in the schema folder moved a bench from 2 to 14 of 18 with no code [0036]. The concept exploration used to create (`Metric` and `Dimension` entities with a SQL expression in a side table) is the wrong kind of thing and is to diff --git a/migrations/0117_a_base_can_opt_into_mapping_confirmation.sql b/migrations/0117_a_base_can_opt_into_mapping_confirmation.sql new file mode 100644 index 000000000..8a2368aac --- /dev/null +++ b/migrations/0117_a_base_can_opt_into_mapping_confirmation.sql @@ -0,0 +1,6 @@ +-- Confidence is evidence for a decision, not an encoding of mapping status (0011). +ALTER TABLE knowledge_bases + ADD COLUMN auto_confirm_mappings boolean NOT NULL DEFAULT false; +ALTER TABLE concept_mappings + ADD COLUMN confidence double precision CHECK (confidence >= 0 AND confidence <= 1), + ADD COLUMN rationale text; diff --git a/web/src/api.ts b/web/src/api.ts index edea88047..81214286c 100644 --- a/web/src/api.ts +++ b/web/src/api.ts @@ -129,6 +129,7 @@ export interface Kb { /** 治理开关(0025,缺省关):agent 按先进先出过等人的重复对,先读台账里人的 * 先例再裁;打开就开始,关掉就停 */ governance: boolean; + auto_confirm_mappings: boolean; /** 这次打开治理的时刻;保险丝只数它之后的撤回 */ governance_since: string | null; /** 多久重推一次(分钟)。事实持续在变,只靠手点会让派生一直是缺的 */ @@ -718,6 +719,9 @@ export interface ConceptMapping { /** 派生指标:算出来的,不是表里的列 */ derived: boolean; status: "proposed" | "confirmed" | "rejected"; + decided_by: string | null; + confidence: number | null; + rationale: string | null; } /** 一条口径改动之前的样子。**整版快照而不是差异**(0006): * 读的时候要回答的是「当时是什么」,差异得从头重放才答得出来 */ diff --git a/web/src/i18n/en.ts b/web/src/i18n/en.ts index 71284011c..97a238ffd 100644 --- a/web/src/i18n/en.ts +++ b/web/src/i18n/en.ts @@ -1925,6 +1925,9 @@ export const en = { colSource: "Source", colDefinition: "How it is computed", colStatus: "Status", + autoConfirmed: "Automatically confirmed", + confidence: (score: number) => `Confidence: ${Math.round(score * 100)}%`, + rationale: (text: string) => `Rationale: ${text}`, derivedBadge: "Derived", noDefinition: "(empty)", approve: "Confirm", @@ -2377,6 +2380,8 @@ export const en = { autoResolveTypes: "Resolve entity types after extraction", autoResolveTypesNote: "After each document is extracted, run a round of type resolution on entities the engine has not looked at yet. Only refinements within the current class are applied on their own — a re-classification across the tree still waits for you on the Ontology page. Every batch is listed there and can be undone.", + autoConfirmMappings: "Automatically confirm mapping proposals", + autoConfirmMappingsNote: "Off by default. On exploration, definitions with confidence of at least 80%, a rationale and an expression or SQL become effective immediately. A wrong definition can produce a plausible wrong answer. Human decisions and edits are kept. Turning this off leaves existing definitions in place; refreshed automatic definitions return to review.", governance: "Let the agent work the duplicates queue", governanceNote: "First in, first out. Before deciding a pair the agent reads what people in this base decided on the same names and the same kinds of pairs. It merges only where that history supports it, keeps apart on confidence, and leaves a proposal for everything else. Every decision is listed under Agent on the Review page and can be reverted. Turning this off stops the queue.", diff --git a/web/src/i18n/zh.ts b/web/src/i18n/zh.ts index e74856650..1e5147cec 100644 --- a/web/src/i18n/zh.ts +++ b/web/src/i18n/zh.ts @@ -1673,6 +1673,9 @@ export const zh: Strings = { colSource: "数据源", colDefinition: "怎么算", colStatus: "状态", + autoConfirmed: "自动确认", + confidence: (score: number) => `置信度:${Math.round(score * 100)}%`, + rationale: (text: string) => `理由:${text}`, derivedBadge: "派生", noDefinition: "(未填)", approve: "确认", @@ -2076,6 +2079,8 @@ export const zh: Strings = { autoResolveTypes: "抽取后自动消解实体类型", autoResolveTypesNote: "每篇文档抽完,对引擎还没看过的实体跑一轮类型消解。只有在现类子树里精化的才自动落地,跨轴的改判仍留在本体页等你。每一批都列在本体页,随时可以撤回。", + autoConfirmMappings: "自动确认映射提议", + autoConfirmMappingsNote: "默认关闭。探索时,置信度至少为 80%、有理由且有表达式或 SQL 的定义立即生效。错误定义可能产生看似合理的错误答案。人工决定和修改会保留。关闭后已有定义继续生效;重新探索刷新自动定义时会将其送回审核。", governance: "让 agent 处理重复项队列", governanceNote: "先进先出。裁一对之前,agent 先读这个库里的人对同样的名字、同样的类型对做过什么决定:有先例撑着才合并,有把握就分开,其余的写成建议留给你。每一笔都列在审核页的 Agent 里,可以撤回。关掉,队列就停。", diff --git a/web/src/pages/KbSettings.tsx b/web/src/pages/KbSettings.tsx index b8478e20c..d430a72f0 100644 --- a/web/src/pages/KbSettings.tsx +++ b/web/src/pages/KbSettings.tsx @@ -120,6 +120,7 @@ export function KbSettings() { const [autoResolve, setAutoResolve] = useState(false); // 治理(0025):**缺省关**——它会合并实体,还没在哪个库上量过;打开就开始,关掉就停 const [governance, setGovernance] = useState(false); + const [autoMappings, setAutoMappings] = useState(false); const [inferMins, setInferMins] = useState(60); const [ontoLang, setOntoLang] = useState<"en" | "zh">("en"); const [confirmingDelete, setConfirmingDelete] = useState(false); @@ -133,6 +134,7 @@ export function KbSettings() { setMaterialize(kb.data.materialize_inferences); setAutoResolve(kb.data.auto_type_resolution); setGovernance(kb.data.governance); + setAutoMappings(kb.data.auto_confirm_mappings); setInferMins(kb.data.inference_interval_minutes); setOntoLang(kb.data.ontology_lang); } @@ -142,6 +144,7 @@ export function KbSettings() { const saveIdentity = useKbPatch(kbId!, setError); const saveVisibility = useKbPatch(kbId!, setError); const saveAutomation = useKbPatch(kbId!, setError); + const saveMappings = useKbPatch(kbId!, setError); const saveLang = useKbPatch(kbId!, setError); const removeKb = useMutation({ @@ -388,6 +391,25 @@ export function KbSettings() { + saveMappings.mutate({ auto_confirm_mappings: autoMappings })} + > + {S.kbset.save} + + } + > + + + {/* 语料语言。**不是界面语言**——类描述逐字进抽取提示词, 读者是正在读这些文档的模型,所以它跟文档走不跟读者走 */} onDecisionPending(false), }); + const automatic = m.status === "confirmed" && m.decided_by === null; const how = howComputed(m); return (
@@ -393,6 +394,9 @@ function MappingRow({ {S.mapping.derivedBadge} )} + {automatic && ( + {S.mapping.autoConfirmed} + )} {statusLabel(m.status)} @@ -411,6 +415,13 @@ function MappingRow({

{m.summary}

)} + {m.confidence !== null && ( +

{S.mapping.confidence(m.confidence)}

+ )} + {m.rationale && ( +

{S.mapping.rationale(m.rationale)}

+ )} + {editing ? ( - {m.status === "proposed" && ( + {(m.status === "proposed" || automatic) && ( <>