From c214b29d6dae8892fee60f225717ebce082f52a4 Mon Sep 17 00:00:00 2001 From: WaylandYang Date: Mon, 14 Sep 2026 14:09:15 +0800 Subject: [PATCH 1/3] Facts and conflicts are governed like duplicates Co-Authored-By: Claude Opus 5 Signed-off-by: WaylandYang --- crates/utopia-core/src/models.rs | 4 + crates/utopia-extract/src/lib.rs | 1 + crates/utopia-extract/src/queue_agent.rs | 337 ++++++++ crates/utopia-server/src/api/review_routes.rs | 14 + crates/utopia-server/src/extraction.rs | 16 +- crates/utopia-server/src/governance.rs | 9 + crates/utopia-server/src/main.rs | 1 + crates/utopia-server/src/queue_agent.rs | 804 ++++++++++++++++++ crates/utopia-server/src/queue_agent_tests.rs | 459 ++++++++++ crates/utopia-store/src/governance.rs | 47 +- crates/utopia-store/src/graph.rs | 13 +- crates/utopia-store/src/lib.rs | 1 + crates/utopia-store/src/queue_agent.rs | 267 ++++++ crates/utopia-store/src/temporal.rs | 150 +++- ...ance-reads-the-ledger-before-it-decides.md | 2 +- .../0043-every-review-queue-is-governed.md | 69 ++ docs/decisions/README.md | 1 + migrations/0058_every_queue_is_governed.sql | 21 + web/src/api.ts | 23 +- web/src/i18n/en.ts | 12 +- web/src/i18n/zh.ts | 12 +- web/src/pages/Review.tsx | 56 +- 22 files changed, 2275 insertions(+), 44 deletions(-) create mode 100644 crates/utopia-extract/src/queue_agent.rs create mode 100644 crates/utopia-server/src/queue_agent.rs create mode 100644 crates/utopia-server/src/queue_agent_tests.rs create mode 100644 crates/utopia-store/src/queue_agent.rs create mode 100644 docs/decisions/0043-every-review-queue-is-governed.md create mode 100644 migrations/0058_every_queue_is_governed.sql diff --git a/crates/utopia-core/src/models.rs b/crates/utopia-core/src/models.rs index 8dd446f54..a5f577fd5 100644 --- a/crates/utopia-core/src/models.rs +++ b/crates/utopia-core/src/models.rs @@ -1443,6 +1443,10 @@ pub struct AgentDecisionView { pub trace: serde_json::Value, /// 循环里花的模型调用次数 pub calls: i32, + /// 重复对以外的一档（0043）：做决定那一刻这一项的样子，给人读 + pub summary: Option, + /// 这一步的参数与撤回要用的东西（0043） + pub detail: serde_json::Value, pub created_at: DateTime, pub decided_at: Option>, pub decided_by_name: Option, diff --git a/crates/utopia-extract/src/lib.rs b/crates/utopia-extract/src/lib.rs index 713bd86c3..ff0cea263 100644 --- a/crates/utopia-extract/src/lib.rs +++ b/crates/utopia-extract/src/lib.rs @@ -6,6 +6,7 @@ use serde::Deserialize; use utopia_llm::ChatMessage; pub mod governor; +pub mod queue_agent; pub mod normalize; pub use normalize::{drop_quotes_from_opening, normalize_facts, Normalization}; diff --git a/crates/utopia-extract/src/queue_agent.rs b/crates/utopia-extract/src/queue_agent.rs new file mode 100644 index 000000000..fc51908a9 --- /dev/null +++ b/crates/utopia-extract/src/queue_agent.rs @@ -0,0 +1,337 @@ +//! 审核台其余几档交给 agent 时的提示词与回复解析（0043）。 +//! +//! 与重复对的裁决（`build_adjudication_messages`）同一个形状：一次一批带编号的项，每项 +//! 附上人在这个库里做过的同类决定，回一个 JSON。模型只给出路与一句理由；它给的日期、 +//! 引文由服务端去核对在不在证据里——结构检查在服务端，语义在这里。 + +use serde::Deserialize; +use utopia_llm::ChatMessage; + +/// 一条证据，给模型看的样子 +#[derive(Debug, Clone)] +pub struct EvidenceLine { + pub document: String, + /// 文档自己的日期，YYYY-MM-DD + pub dated: Option, + pub quote: Option, +} + +/// 一条事实，给模型看的样子 +#[derive(Debug, Clone)] +pub struct FactCard { + pub subject: String, + pub predicate: String, + pub object: String, + /// 起止，已经写成 YYYY[-MM[-DD]] / unknown / 空 + pub from: Option, + pub to: Option, + pub confidence: f32, + pub evidence: Vec, +} + +/// 事实那两档里的一项 +#[derive(Debug, Clone)] +pub struct FactQuestion { + pub fact: FactCard, + /// 证据全在文档的旧版本上；`current` 是文档现在提到主语的那几段 + pub stale: bool, + pub current: Vec, + pub precedents: Vec, +} + +/// 冲突那一档里的一项 +#[derive(Debug, Clone)] +pub struct ConflictQuestion { + /// no_time | simultaneous | low_confidence + pub reason: String, + pub old: FactCard, + pub new: FactCard, + /// 同一持有者、同一关系上的其他值 + pub neighbours: Vec, + pub precedents: Vec, +} + +/// 模型对一项的回答 +#[derive(Debug, Clone, Deserialize)] +pub struct QueueVerdict { + pub i: usize, + pub action: String, + #[serde(default)] + pub confidence: Option, + #[serde(default)] + pub why: Option, + /// 冲突：close_old 闭合在哪天、retime_new 改成从哪天起，YYYY[-MM[-DD]] + #[serde(default)] + pub date: Option, + /// 事实：证据过期时，文档现在那段里说出这件事的原话 + #[serde(default)] + pub quote: Option, +} + +#[derive(Debug, Deserialize)] +struct Reply { + #[serde(default)] + verdicts: Vec, +} + +pub fn parse_verdicts(raw: &str) -> anyhow::Result> { + let json = crate::json_block(raw)?; + let reply: Reply = serde_json::from_str(&json) + .map_err(|e| anyhow::anyhow!("Failed to parse queue verdicts: {e}"))?; + Ok(reply.verdicts) +} + +const PRECEDENTS_NOTE: &str = "Some items carry precedents: decisions people made in this same \ +knowledge base on items of this kind. Treat them as how the owners of this base want such cases \ +judged, and follow one whose ground holds here; a precedent never overrides what the evidence says."; + +fn card(f: &FactCard) -> String { + let when = match (&f.from, &f.to) { + (None, None) => String::new(), + (from, to) => format!( + " [{} → {}]", + from.as_deref().unwrap_or("?"), + to.as_deref().unwrap_or("now") + ), + }; + let evidence = if f.evidence.is_empty() { + " (no evidence)".to_string() + } else { + f.evidence + .iter() + .map(|e| { + format!( + " - {}{}: \"{}\"", + e.document, + e.dated + .as_deref() + .map(|d| format!(" (dated {d})")) + .unwrap_or_default(), + e.quote.as_deref().unwrap_or("(no quote)") + ) + }) + .collect::>() + .join("\n") + }; + format!( + "{} · {} · {}{when} (confidence {:.2})\n evidence:\n{evidence}", + f.subject, f.predicate, f.object, f.confidence + ) +} + +fn precedent_lines(p: &[String]) -> String { + if p.is_empty() { + String::new() + } else { + format!( + " precedents (decided by people in this base):\n{}\n", + p.iter() + .map(|l| format!(" - {l}")) + .collect::>() + .join("\n") + ) + } +} + +/// 低置信与证据过期的事实：证据说没说这件事 +pub fn fact_messages(items: &[FactQuestion]) -> Vec { + let system = format!( + "You review facts a knowledge graph extracted from documents. For each numbered fact, \ + decide from its evidence whether the fact is what the text says.\n\ + \n\ + Actions:\n\ + - \"confirm\": the evidence states this fact — the same subject, relation, value and \ + dates. A value written in a table, a list or an amendment's new column is stated.\n\ + - \"reject\": the evidence does not say it, says something else, or attaches it to \ + another subject.\n\ + - \"unsure\": the evidence genuinely points both ways; say what a person should check.\n\ + \n\ + A fact marked STALE has evidence only in an older version of its document. Judge it \ + against the CURRENT text shown with it: confirm only when the current text still states \ + it, and then put in \"quote\" the exact words of the current text that state it; reject \ + when the current text says otherwise or no longer says it.\n\ + \n\ + {PRECEDENTS_NOTE}\n\ + \n\ + Output exactly one JSON object and nothing else:\n\ + {{\"verdicts\":[{{\"i\":0,\"action\":\"confirm|reject|unsure\",\"confidence\":0.9,\ + \"why\":\"one sentence\",\"quote\":\"only for a stale fact you confirm\"}}]}}\n\ + \n\ + Rules:\n\ + 1. One verdict per fact, using the fact's number as \"i\".\n\ + 2. confidence in 0~1 is how sure you are of the action.\n\ + 3. \"why\" is one short sentence naming what in the evidence decided it." + ); + let mut user = String::new(); + for (i, q) in items.iter().enumerate() { + let current = if q.stale { + let text = if q.current.is_empty() { + " (the current version no longer mentions the subject)".to_string() + } else { + q.current + .iter() + .map(|t| format!(" \"\"\"{t}\"\"\"")) + .collect::>() + .join("\n") + }; + format!(" STALE. current text:\n{text}\n") + } else { + String::new() + }; + user.push_str(&format!( + "Fact {i}: {}\n{current}{}\n", + card(&q.fact), + precedent_lines(&q.precedents) + )); + } + vec![ + ChatMessage { + role: "system".into(), + content: system, + }, + ChatMessage { + role: "user".into(), + content: user, + }, + ] +} + +/// 时态冲突：两个值在同一刻都成立、引擎排不开 +pub fn conflict_messages(items: &[ConflictQuestion]) -> Vec { + let system = format!( + "You resolve conflicts in a knowledge graph's timelines. Each numbered conflict is two \ + values of a relation that holds one value at a time, both holding at once, which the \ + engine could not order by itself. The reason says why: \"simultaneous\" (both start \ + on the same date), \"no_time\" (one has no date at all), \"low_confidence\" (the later \ + value was extracted with too little confidence to take over).\n\ + \n\ + Actions:\n\ + - \"close_old\": the new value took over from the old one. Give \"date\" when the old \ + value ended on a day other than the new value's start, taken from the evidence \ + (a document's date or a date its quote states).\n\ + - \"retime_new\": the new value is right but its start date is wrong; give \"date\", the \ + day it took effect according to its evidence (a document's date or a date its quote \ + states). The old value will then end there.\n\ + - \"keep_both\": both hold at once and neither replaces the other.\n\ + - \"reject_new\": the new fact misreads its evidence.\n\ + - \"unsure\": the evidence genuinely points both ways; say what a person should check.\n\ + \n\ + {PRECEDENTS_NOTE}\n\ + \n\ + Output exactly one JSON object and nothing else:\n\ + {{\"verdicts\":[{{\"i\":0,\"action\":\"close_old|retime_new|keep_both|reject_new|unsure\",\ + \"confidence\":0.9,\"why\":\"one sentence\",\"date\":\"YYYY-MM-DD when the action needs one\"}}]}}\n\ + \n\ + Rules:\n\ + 1. One verdict per conflict, using its number as \"i\".\n\ + 2. confidence in 0~1 is how sure you are of the action.\n\ + 3. A date must come from the evidence shown; never infer one." + ); + let mut user = String::new(); + for (i, q) in items.iter().enumerate() { + let neighbours = if q.neighbours.is_empty() { + String::new() + } else { + format!( + " other values on this relation:\n{}\n", + q.neighbours + .iter() + .map(|n| format!(" - {}", card(n).replace('\n', "\n "))) + .collect::>() + .join("\n") + ) + }; + user.push_str(&format!( + "Conflict {i} ({}):\n OLD: {}\n NEW: {}\n{neighbours}{}\n", + q.reason, + card(&q.old).replace('\n', "\n "), + card(&q.new).replace('\n', "\n "), + precedent_lines(&q.precedents) + )); + } + vec![ + ChatMessage { + role: "system".into(), + content: system, + }, + ChatMessage { + role: "user".into(), + content: user, + }, + ] +} + +#[cfg(test)] +mod tests { + use super::*; + + fn fact(value: &str, from: Option<&str>, confidence: f32) -> FactCard { + FactCard { + subject: "HQ Lease".into(), + predicate: "option deadline".into(), + object: value.into(), + from: from.map(str::to_string), + to: None, + confidence, + evidence: vec![EvidenceLine { + document: "sixth-amendment.html".into(), + dated: Some("2020-03-17".into()), + quote: Some("(Phase 2 Exercise Deadline) | April 14, 2020".into()), + }], + } + } + + #[test] + fn a_fact_question_shows_its_evidence_and_the_current_text_when_stale() { + let msgs = fact_messages(&[ + FactQuestion { + fact: fact("2020-04-14", Some("2020-03-17"), 0.7), + stale: false, + current: vec![], + precedents: vec!["fact.confirm: HQ Lease option deadline 2020-03-17".into()], + }, + FactQuestion { + fact: fact("2020-05-26", None, 0.9), + stale: true, + current: vec!["The Phase 2 Exercise Deadline is May 26, 2020.".into()], + precedents: vec![], + }, + ]); + let user = &msgs[1].content; + assert!(user.contains( + "Fact 0: HQ Lease · option deadline · 2020-04-14 [2020-03-17 → now] (confidence 0.70)" + )); + assert!(user.contains("sixth-amendment.html (dated 2020-03-17): \"(Phase 2 Exercise Deadline) | April 14, 2020\"")); + assert!(user.contains("fact.confirm: HQ Lease option deadline 2020-03-17")); + assert!(user.contains("Fact 1:")); + assert!(user.contains("STALE. current text:")); + assert!(msgs[0].content.contains("\"confirm\"")); + } + + #[test] + fn a_conflict_question_shows_both_sides_and_their_neighbours() { + let msgs = conflict_messages(&[ConflictQuestion { + reason: "low_confidence".into(), + old: fact("2020-03-17", Some("2020-02-18"), 0.9), + new: fact("2020-04-14", Some("2020-03-17"), 0.7), + neighbours: vec![fact("2020-05-26", Some("2020-04-14"), 0.9)], + precedents: vec![], + }]); + let user = &msgs[1].content; + assert!(user.contains("Conflict 0 (low_confidence):")); + assert!(user.contains("OLD: HQ Lease · option deadline · 2020-03-17")); + assert!(user.contains("NEW: HQ Lease · option deadline · 2020-04-14")); + assert!(user.contains("other values on this relation:")); + assert!(msgs[0].content.contains("\"retime_new\"")); + } + + #[test] + fn verdicts_parse_with_their_optional_fields() { + let raw = "```json\n{\"verdicts\":[{\"i\":0,\"action\":\"close_old\",\"confidence\":0.9,\"why\":\"x\",\"date\":\"2020-08-13\"},{\"i\":1,\"action\":\"unsure\"}]}\n```"; + let v = parse_verdicts(raw).unwrap(); + assert_eq!(v.len(), 2); + assert_eq!(v[0].date.as_deref(), Some("2020-08-13")); + assert_eq!(v[1].confidence, None); + assert_eq!(v[1].quote, None); + } +} diff --git a/crates/utopia-server/src/api/review_routes.rs b/crates/utopia-server/src/api/review_routes.rs index a3b11427f..aa5be48e3 100644 --- a/crates/utopia-server/src/api/review_routes.rs +++ b/crates/utopia-server/src/api/review_routes.rs @@ -378,6 +378,20 @@ pub async fn agent_answer( ) -> ApiResult> { require_kb(&state, &user, kb_id, Role::Editor).await?; let d = utopia_store::governance::get(&state.pool, kb_id, decision_id).await?; + // 事实与冲突两档的建议（0043）：各有各的出路 + if d.target_kind != "review" { + crate::queue_agent::answer( + &state, + kb_id, + &d, + &body.action, + user.id, + body.rationale.as_deref(), + ) + .await?; + state.emit_review(kb_id); + return Ok(Json(json!({ "ok": true }))); + } let (l, r) = ( d.left.clone().unwrap_or_default(), d.right.clone().unwrap_or_default(), diff --git a/crates/utopia-server/src/extraction.rs b/crates/utopia-server/src/extraction.rs index e54a12f71..ad70acc15 100644 --- a/crates/utopia-server/src/extraction.rs +++ b/crates/utopia-server/src/extraction.rs @@ -2858,14 +2858,14 @@ async fn run(state: &AppState, document_id: Uuid, proposer: Proposer) -> anyhow: // 治理开着（0025）排的是治理任务：灰区对与直接转人工的对都从那条先进先出的 // 队列走，先读台账再裁；同库已排着的不重复 if kb.governance { - if needs_adjudication || human_reviews_found { - utopia_store::jobs::enqueue_unless_queued( - &state.pool, - "govern", - serde_json::json!({ "kb_id": doc.kb_id }), - ) - .await?; - } + // 重复对之外，低置信的事实与时态冲突也归它（0043）：抽完一篇就排一个， + // 什么都没等着的话任务看一眼就结束 + utopia_store::jobs::enqueue_unless_queued( + &state.pool, + "govern", + serde_json::json!({ "kb_id": doc.kb_id }), + ) + .await?; } else if needs_adjudication { // 同库已排着的不重复——与下面的 resolve_types 一样。一批文档同时抽完 // 会各排一个，而它们读到的是同一批待裁项 diff --git a/crates/utopia-server/src/governance.rs b/crates/utopia-server/src/governance.rs index 3e4253822..e25683239 100644 --- a/crates/utopia-server/src/governance.rs +++ b/crates/utopia-server/src/governance.rs @@ -126,6 +126,15 @@ pub async fn govern(state: &AppState, kb_id: Uuid) -> anyhow::Result<()> { } state.emit_review(kb_id); let more = outcome?; + // 重复对之后是其余几档（0043）：合并先定下来，事实与冲突看到的才是合并之后的图 + let queues = crate::queue_agent::Ctx { + state, + kb_id, + run_id: ctx.run_id, + client: &client, + settings: &settings, + }; + let more = crate::queue_agent::run(&queues).await? || more; // 轮数用完还有积压：再排一个，下一轮从队头接着走 if more { diff --git a/crates/utopia-server/src/main.rs b/crates/utopia-server/src/main.rs index 53bdc7304..f9f02c572 100644 --- a/crates/utopia-server/src/main.rs +++ b/crates/utopia-server/src/main.rs @@ -26,6 +26,7 @@ mod pack_alignment; mod pipeline; mod predicate_match; mod query_engine; +mod queue_agent; mod rdf; mod retrieval; mod rss_full_content; diff --git a/crates/utopia-server/src/queue_agent.rs b/crates/utopia-server/src/queue_agent.rs new file mode 100644 index 000000000..ca3884c1e --- /dev/null +++ b/crates/utopia-server/src/queue_agent.rs @@ -0,0 +1,804 @@ +//! 审核台其余几档交给 agent（0043）：低置信与证据过期的事实、时态冲突。 +//! +//! 与重复对的治理（`governance`）同一个开关、同一个任务、同一张决定表、同一根保险丝， +//! 走同样的路：先读这个库里人在这一档上怎么定的，一批带编号的项问一次模型，把握够 +//! （`AUTO_CONF`）就走人裁决时的那条路动手，不够就留一条建议等人。 +//! +//! **动手的每一步都撤得回**，撤回要用的东西记在决定行的 `detail` 里：确认一条事实记下原来 +//! 的置信度、补上的那条证据；驳回一条事实靠恢复它；关上旧值、改新值的起点记下改写出来的 +//! 行与原行。人撤回一次，保险丝照重复对那样数。 +//! +//! 模型给的日期与引文**由这里核对**，不照单全收：日期得是证据里看得到的——文档自己的 +//! 日期、事实的起点、或引文里写着的那一天；过期事实的引文得原样出现在文档现在的那段里。 +//! 核对不过的，有把握也只留建议。 + +use crate::llm_util; +use crate::state::AppState; +use chrono::{DateTime, Datelike, Utc}; +use serde_json::{json, Value}; +use utopia_core::models::{AgentDecisionView, LlmSettings}; +use utopia_extract::queue_agent::{ + self as prompts, ConflictQuestion, EvidenceLine, FactCard, FactQuestion, QueueVerdict, +}; +use utopia_llm::LlmClient; +use utopia_store::governance::{self as gov, NewDecision, Target, AUTO_CONF}; +use utopia_store::queue_agent::{self as queues, Evidence, FactRow}; +use uuid::Uuid; + +/// 一次模型调用带几项 +const BATCH: i64 = 8; +/// 一个任务里每一档最多问几批，之后再排一个接着走 +const MAX_BATCHES: usize = 10; +/// 先例带几条 +const PRECEDENTS: i64 = 8; + +pub(crate) struct Ctx<'a> { + pub state: &'a AppState, + pub kb_id: Uuid, + pub run_id: Uuid, + pub client: &'a LlmClient, + pub settings: &'a Option, +} + +/// 走一遍事实与冲突两档。回 true = 批数用完还有积压 +pub(crate) async fn run(ctx: &Ctx<'_>) -> anyhow::Result { + let pool = &ctx.state.pool; + for _ in 0..MAX_BATCHES { + if !utopia_store::kbs::get(pool, ctx.kb_id).await?.governance { + return Ok(false); + } + let items = queues::fact_queue(pool, ctx.kb_id, BATCH).await?; + if items.is_empty() { + break; + } + facts(ctx, items).await?; + ctx.state.emit_review(ctx.kb_id); + } + for _ in 0..MAX_BATCHES { + if !utopia_store::kbs::get(pool, ctx.kb_id).await?.governance { + return Ok(false); + } + let items = queues::conflict_queue(pool, ctx.kb_id, BATCH).await?; + if items.is_empty() { + break; + } + conflicts(ctx, items).await?; + ctx.state.emit_review(ctx.kb_id); + } + queues::has_backlog(pool, ctx.kb_id) + .await + .map_err(Into::into) +} + +async fn ask( + ctx: &Ctx<'_>, + messages: &[utopia_llm::ChatMessage], +) -> anyhow::Result> { + let reply = { + let _permit = match ctx.settings.as_ref() { + Some(s) => llm_util::acquire_chat(ctx.state, s).await, + None => None, + }; + ctx.client.chat(messages).await? + }; + prompts::parse_verdicts(&reply) +} + +fn day(t: DateTime, precision: Option<&str>) -> String { + match precision { + Some("year") => t.format("%Y").to_string(), + Some("month") => t.format("%Y-%m").to_string(), + _ => t.format("%Y-%m-%d").to_string(), + } +} + +fn card(f: &FactRow, evidence: &[Evidence]) -> FactCard { + FactCard { + subject: f.subject.clone(), + predicate: f.predicate.clone().unwrap_or_default(), + object: f.object.clone().unwrap_or_default(), + from: f + .valid_from + .map(|t| day(t, f.valid_from_precision.as_deref())), + to: match (f.valid_to, f.valid_to_precision.as_deref()) { + (Some(t), p) => Some(day(t, p)), + (None, Some("unknown")) => Some("ended, date unknown".into()), + _ => None, + }, + confidence: f.confidence, + evidence: evidence + .iter() + .map(|e| EvidenceLine { + document: e.document.clone(), + dated: e.dated.map(|t| t.format("%Y-%m-%d").to_string()), + quote: e.quote.clone(), + }) + .collect(), + } +} + +/// 给人读的一行：做决定那一刻这一项的样子 +fn summary(f: &FactRow) -> String { + let c = card(f, &[]); + let when = match (&c.from, &c.to) { + (None, None) => String::new(), + (from, to) => format!( + " [{} → {}]", + from.as_deref().unwrap_or("?"), + to.as_deref().unwrap_or("now") + ), + }; + format!("{} · {} · {}{when}", c.subject, c.predicate, c.object) +} + +fn normalized(s: &str) -> String { + s.split_whitespace() + .collect::>() + .join(" ") + .to_lowercase() +} + +/// 模型说的日期在不在证据里：事实的起点、文档自己的日期，或引文里写着的那一天。 +/// 回核对过的日期与精度 +fn date_in_evidence( + said: Option<&str>, + fact: &FactRow, + evidence: &[Evidence], +) -> Option<(DateTime, &'static str)> { + let (at, precision) = utopia_extract::read_time(said?)?; + let same_day = |t: DateTime| t.date_naive() == at.date_naive(); + if fact.valid_from.is_some_and(same_day) + || evidence.iter().any(|e| e.dated.is_some_and(same_day)) + { + return Some((at, precision)); + } + let d = at.date_naive(); + let written = [ + d.format("%Y-%m-%d").to_string(), + d.format("%B %-d, %Y").to_string(), + d.format("%-d %B %Y").to_string(), + d.format("%b %-d, %Y").to_string(), + format!("{}年{}月{}日", d.year(), d.month(), d.day()), + ]; + evidence + .iter() + .filter_map(|e| e.quote.as_deref()) + .map(normalized) + .any(|q| written.iter().any(|w| q.contains(&normalized(w)))) + .then_some((at, precision)) +} + +fn verdict_for(verdicts: &[QueueVerdict], i: usize) -> Option<&QueueVerdict> { + verdicts.iter().find(|v| v.i == i) +} + +/// 一次动手的结果：动成了记在 detail 里，没动成（核对没过、行已经变了）说为什么 +enum Done { + Applied(Value), + Held(String), +} + +async fn facts(ctx: &Ctx<'_>, items: Vec) -> anyhow::Result<()> { + let pool = &ctx.state.pool; + let precedents = queues::precedents( + pool, + ctx.kb_id, + &["fact.confirm", "fact.reject"], + PRECEDENTS, + ) + .await?; + let mut evidence = Vec::with_capacity(items.len()); + let mut current = Vec::with_capacity(items.len()); + for f in &items { + evidence.push(queues::evidence(pool, f.id).await?); + current.push(if f.stale { + queues::current_text(pool, f.id, &f.subject).await? + } else { + Vec::new() + }); + } + let questions: Vec = items + .iter() + .enumerate() + .map(|(i, f)| FactQuestion { + fact: card(f, &evidence[i]), + stale: f.stale, + current: current[i].iter().map(|(_, t)| t.clone()).collect(), + precedents: precedents.clone(), + }) + .collect(); + let verdicts = ask(ctx, &prompts::fact_messages(&questions)).await?; + + for (i, f) in items.iter().enumerate() { + let v = verdict_for(&verdicts, i); + let action = v + .map(|v| v.action.as_str()) + .filter(|a| matches!(*a, "confirm" | "reject")) + .unwrap_or("unsure"); + let conf = v.and_then(|v| v.confidence).unwrap_or(0.0).clamp(0.0, 1.0); + let why = v.and_then(|v| v.why.clone()); + let params = json!({ "quote": v.and_then(|v| v.quote.clone()) }); + let done = if action != "unsure" && conf >= AUTO_CONF { + Some(perform_fact(ctx.state, ctx.kb_id, f, action, ¶ms, ¤t[i]).await?) + } else { + None + }; + settle_new( + ctx, + "fact", + f.id, + &summary(f), + action, + conf, + why, + params, + done, + "fact", + ) + .await?; + } + Ok(()) +} + +/// 事实那两档的一个出路，走人裁决时的那条路 +async fn perform_fact( + state: &AppState, + kb_id: Uuid, + f: &FactRow, + action: &str, + params: &Value, + current: &[(Uuid, String)], +) -> anyhow::Result { + let pool = &state.pool; + match action { + "confirm" if f.stale => { + // 过期的事实：文档现在那段里原样说着它，补上这条证据，它就不再是过期的 + let Some(quote) = params["quote"].as_str().filter(|q| !q.trim().is_empty()) else { + return Ok(Done::Held("no quote from the current text".into())); + }; + let Some((chunk_id, _)) = current + .iter() + .find(|(_, text)| normalized(text).contains(&normalized(quote))) + else { + return Ok(Done::Held("the quote is not in the current text".into())); + }; + utopia_store::graph::add_evidence(pool, f.id, *chunk_id, Some(quote), None).await?; + let fact_id = f.id; + Ok(Done::Applied( + json!({ "evidence": { "fact_id": fact_id, "chunk_id": chunk_id } }), + )) + } + "confirm" => match utopia_store::temporal::set_confidence(pool, kb_id, f.id, 1.0).await? { + Some(prior) => Ok(Done::Applied(json!({ "prior_confidence": prior }))), + None => Ok(Done::Held( + "the fact changed before it could be confirmed".into(), + )), + }, + "reject" => { + if utopia_store::temporal::retract(pool, kb_id, f.id).await? { + Ok(Done::Applied(json!({ "retracted": f.id }))) + } else { + Ok(Done::Held( + "the fact changed before it could be rejected".into(), + )) + } + } + other => Ok(Done::Held(format!("no action {other} for a fact"))), + } +} + +async fn conflicts(ctx: &Ctx<'_>, items: Vec) -> anyhow::Result<()> { + let pool = &ctx.state.pool; + let precedents = queues::precedents( + pool, + ctx.kb_id, + &[ + "conflict.close_old", + "conflict.keep_both", + "conflict.reject_new", + ], + PRECEDENTS, + ) + .await?; + struct Loaded { + row: queues::ConflictRow, + old: FactRow, + new: FactRow, + new_evidence: Vec, + question: ConflictQuestion, + } + let mut loaded = Vec::new(); + for row in items { + let (Some(old), Some(new)) = ( + queues::fact(pool, ctx.kb_id, row.old_fact_id).await?, + queues::fact(pool, ctx.kb_id, row.new_fact_id).await?, + ) else { + continue; + }; + let old_evidence = queues::evidence(pool, old.id).await?; + let new_evidence = queues::evidence(pool, new.id).await?; + let neighbours = queues::neighbours(pool, ctx.kb_id, old.id).await?; + let mut cards = Vec::new(); + for n in neighbours.iter().filter(|n| n.id != new.id) { + cards.push(card(n, &queues::evidence(pool, n.id).await?)); + } + let question = ConflictQuestion { + reason: row.reason.clone(), + old: card(&old, &old_evidence), + new: card(&new, &new_evidence), + neighbours: cards, + precedents: precedents.clone(), + }; + loaded.push(Loaded { + row, + old, + new, + new_evidence, + question, + }); + } + if loaded.is_empty() { + return Ok(()); + } + let questions: Vec = loaded.iter().map(|l| l.question.clone()).collect(); + let verdicts = ask(ctx, &prompts::conflict_messages(&questions)).await?; + + for (i, l) in loaded.iter().enumerate() { + let v = verdict_for(&verdicts, i); + let action = v + .map(|v| v.action.as_str()) + .filter(|a| matches!(*a, "close_old" | "retime_new" | "keep_both" | "reject_new")) + .unwrap_or("unsure"); + let conf = v.and_then(|v| v.confidence).unwrap_or(0.0).clamp(0.0, 1.0); + let why = v.and_then(|v| v.why.clone()); + let params = json!({ "date": v.and_then(|v| v.date.clone()) }); + let done = if action != "unsure" && conf >= AUTO_CONF { + Some( + perform_conflict( + ctx.state, + ctx.kb_id, + &l.row, + &l.new, + &l.new_evidence, + action, + ¶ms, + ) + .await?, + ) + } else { + None + }; + let text = format!( + "{} ({}) · old {} · new {}", + l.old.predicate.clone().unwrap_or_default(), + l.row.reason, + summary(&l.old), + summary(&l.new) + ); + settle_new( + ctx, "conflict", l.row.id, &text, action, conf, why, params, done, "conflict", + ) + .await?; + } + Ok(()) +} + +/// 冲突的一个出路，走人裁决时的那条路 +async fn perform_conflict( + state: &AppState, + kb_id: Uuid, + row: &queues::ConflictRow, + new: &FactRow, + new_evidence: &[Evidence], + action: &str, + params: &Value, +) -> anyhow::Result { + let pool = &state.pool; + let said = params["date"].as_str(); + match action { + "close_old" => { + let at = match said { + Some(_) => match date_in_evidence(said, new, new_evidence) { + Some(at) => Some(at), + None => { + return Ok(Done::Held( + "the date is not in the new fact's evidence".into(), + )) + } + }, + None => None, + }; + if at.is_none() && new.valid_from.is_none() { + return Ok(Done::Held("the new value has no start to close at".into())); + } + let corrected = utopia_store::temporal::resolve_conflict( + pool, + kb_id, + row.id, + "close", + at.map(|(t, _)| t), + at.map(|(_, p)| p).unwrap_or("day"), + ) + .await?; + Ok(Done::Applied( + json!({ "corrected": corrected, "original": row.old_fact_id }), + )) + } + "retime_new" => { + let Some((from, precision)) = date_in_evidence(said, new, new_evidence) else { + return Ok(Done::Held( + "the date is not in the new fact's evidence".into(), + )); + }; + let validity = utopia_store::graph::Validity { + from: Some(from), + from_precision: Some(precision), + to: new.valid_to, + to_precision: new.valid_to_precision.as_deref().map(|p| match p { + "year" => "year", + "month" => "month", + "unknown" => "unknown", + _ => "day", + }), + attested_at: None, + }; + let Some(corrected) = + utopia_store::temporal::correct_interval(pool, new.id, validity).await? + else { + return Ok(Done::Held( + "the new fact changed before it could be retimed".into(), + )); + }; + utopia_store::temporal::reconcile_moved_facts(pool, kb_id, &[corrected]).await?; + Ok(Done::Applied( + json!({ "corrected": corrected, "original": new.id }), + )) + } + "keep_both" => { + utopia_store::temporal::resolve_conflict(pool, kb_id, row.id, "keep", None, "day") + .await?; + Ok(Done::Applied(json!({}))) + } + "reject_new" => { + utopia_store::temporal::resolve_conflict( + pool, + kb_id, + row.id, + "reject_new", + None, + "day", + ) + .await?; + Ok(Done::Applied(json!({ "retracted": row.new_fact_id }))) + } + other => Ok(Done::Held(format!("no action {other} for a conflict"))), + } +} + +/// 记一笔 agent 的决定，动了手的记台账 +#[allow(clippy::too_many_arguments)] +async fn settle_new( + ctx: &Ctx<'_>, + kind: &str, + target_id: Uuid, + summary: &str, + action: &str, + conf: f32, + why: Option, + params: Value, + done: Option, + audit_kind: &str, +) -> anyhow::Result<()> { + let pool = &ctx.state.pool; + let (status, undo, held) = match done { + Some(Done::Applied(undo)) => ("applied", undo, None), + Some(Done::Held(reason)) => ("proposed", json!({}), Some(reason)), + None => ("proposed", json!({}), None), + }; + let reason = match (&why, &held) { + (Some(w), Some(h)) => Some(format!("held for a person: {h}; {w}")), + (None, Some(h)) => Some(format!("held for a person: {h}")), + (w, None) => w.clone(), + }; + let id = gov::record_for( + pool, + ctx.kb_id, + NewDecision { + run_id: ctx.run_id, + target_id, + action, + confidence: conf, + reason: reason.as_deref(), + precedents: json!([]), + status, + merge_id: None, + question: None, + trace: json!([]), + calls: 0, + }, + Target { + kind, + summary: Some(summary), + detail: json!({ "params": params, "undo": undo }), + }, + ) + .await?; + if status == "applied" { + let _ = utopia_store::audit::record_opt( + pool, + Some(ctx.kb_id), + None, + &audit_action(kind, action), + audit_kind, + Some(target_id), + json!({ "summary": summary, "confidence": conf, "via": "governor", "decision": id, + "why": why }), + ) + .await; + } + Ok(()) +} + +fn audit_action(kind: &str, action: &str) -> String { + match (kind, action) { + ("fact", a) => format!("fact.{a}"), + ("conflict", "retime_new") => "fact.time_corrected".into(), + ("conflict", a) => format!("conflict.{a}"), + (k, a) => format!("{k}.{a}"), + } +} + +/// 人回答事实或冲突上的一条 agent 决定：接受或改判一条建议、撤回一次动手。 +/// 回 `Ok(())` 之后调用方刷新审核台 +pub(crate) async fn answer( + state: &AppState, + kb_id: Uuid, + d: &AgentDecisionView, + action: &str, + user_id: Uuid, + rationale: Option<&str>, +) -> utopia_core::AppResult<()> { + let pool = &state.pool; + let invalid = |msg: String| utopia_core::AppError::invalid("agent_answer", msg); + match (d.status.as_str(), action) { + ("applied", "revert") => { + undo(state, kb_id, d).await?; + gov::settle(pool, kb_id, d.id, "reverted", user_id).await?; + let _ = utopia_store::audit::record( + pool, + Some(kb_id), + user_id, + "agent.revert", + &d.target_kind, + Some(d.target_id), + json!({ "agent_decision": d.id, "agent_action": d.action, + "summary": d.summary, "why": rationale }), + ) + .await; + crate::governance::fuse(state, kb_id).await; + Ok(()) + } + ("proposed", act) => { + let allowed: &[&str] = match d.target_kind.as_str() { + "fact" => &["confirm", "reject"], + "conflict" => &["close_old", "retime_new", "keep_both", "reject_new"], + other => return Err(invalid(format!("no answers for {other} decisions here"))), + }; + if !allowed.contains(&act) { + return Err(invalid(format!( + "{act} is not an answer to a {} decision", + d.target_kind + ))); + } + let params = d.detail.get("params").cloned().unwrap_or(json!({})); + let done = match d.target_kind.as_str() { + "fact" => { + let f = queues::fact(pool, kb_id, d.target_id) + .await? + .ok_or(utopia_core::AppError::NotFound)?; + let current = if f.stale { + queues::current_text(pool, f.id, &f.subject).await? + } else { + Vec::new() + }; + perform_fact(state, kb_id, &f, act, ¶ms, ¤t) + .await + .map_err(|e| invalid(e.to_string()))? + } + _ => { + let row = queues::conflict_queue(pool, kb_id, i64::MAX) + .await? + .into_iter() + .find(|c| c.id == d.target_id); + let row = match row { + Some(r) => r, + None => sqlx::query_as( + "SELECT id, reason, old_fact_id, new_fact_id FROM fact_conflicts + WHERE id = $1 AND kb_id = $2 AND status = 'open'", + ) + .bind(d.target_id) + .bind(kb_id) + .fetch_optional(pool) + .await? + .ok_or(utopia_core::AppError::NotFound)?, + }; + let new = queues::fact(pool, kb_id, row.new_fact_id) + .await? + .ok_or(utopia_core::AppError::NotFound)?; + let evidence = queues::evidence(pool, new.id).await?; + perform_conflict(state, kb_id, &row, &new, &evidence, act, ¶ms) + .await + .map_err(|e| invalid(e.to_string()))? + } + }; + if let Done::Held(why) = done { + return Err(invalid(why)); + } + let status = if act == d.action { + "accepted" + } else { + "overridden" + }; + gov::settle(pool, kb_id, d.id, status, user_id).await?; + let _ = utopia_store::audit::record( + pool, + Some(kb_id), + user_id, + &audit_action(&d.target_kind, act), + &d.target_kind, + Some(d.target_id), + json!({ "summary": d.summary, "agent_decision": d.id, "agent_action": d.action, + "why": rationale }), + ) + .await; + Ok(()) + } + (status, act) => Err(invalid(format!( + "cannot {act} an agent decision that is {status}" + ))), + } +} + +/// 撤回 agent 动过的一次手，按 detail 里记下的东西 +async fn undo(state: &AppState, kb_id: Uuid, d: &AgentDecisionView) -> utopia_core::AppResult<()> { + let pool = &state.pool; + let u = d.detail.get("undo").cloned().unwrap_or(json!({})); + let uuid = |k: &str| { + u.get(k) + .and_then(|v| v.as_str()) + .and_then(|s| s.parse::().ok()) + }; + match (d.target_kind.as_str(), d.action.as_str()) { + ("fact", "confirm") => { + if let Some(prior) = u.get("prior_confidence").and_then(|v| v.as_f64()) { + utopia_store::temporal::set_confidence(pool, kb_id, d.target_id, prior as f32) + .await?; + } + if let Some(ev) = u.get("evidence") { + let fact = ev + .get("fact_id") + .and_then(|v| v.as_str()) + .and_then(|s| s.parse::().ok()); + let chunk = ev + .get("chunk_id") + .and_then(|v| v.as_str()) + .and_then(|s| s.parse::().ok()); + if let (Some(fact), Some(chunk)) = (fact, chunk) { + sqlx::query("DELETE FROM fact_evidence WHERE fact_id = $1 AND chunk_id = $2") + .bind(fact) + .bind(chunk) + .execute(pool) + .await?; + utopia_store::temporal::reconcile_moved_facts(pool, kb_id, &[fact]).await?; + } + } + } + ("fact", "reject") => { + utopia_store::temporal::restore(pool, kb_id, d.target_id).await?; + } + ("conflict", "close_old" | "retime_new") => { + if let (Some(corrected), Some(original)) = (uuid("corrected"), uuid("original")) { + if utopia_store::temporal::undo_rewrite(pool, kb_id, corrected, original).await? { + // 原行回来了，它与邻居之间那道题也回来：重新「来到」时间线上，冲突照记 + utopia_store::temporal::reconcile_moved_facts(pool, kb_id, &[original]).await?; + } + } + utopia_store::temporal::reopen_conflict(pool, kb_id, d.target_id).await?; + } + ("conflict", "keep_both") => { + utopia_store::temporal::reopen_conflict(pool, kb_id, d.target_id).await?; + } + ("conflict", "reject_new") => { + if let Some(fact) = uuid("retracted") { + utopia_store::temporal::restore(pool, kb_id, fact).await?; + } + utopia_store::temporal::reopen_conflict(pool, kb_id, d.target_id).await?; + } + (kind, action) => { + return Err(utopia_core::AppError::invalid( + "agent_answer", + format!("cannot revert {kind} {action}"), + )) + } + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn at(s: &str) -> DateTime { + format!("{s}T00:00:00Z").parse().unwrap() + } + + fn row(from: Option<&str>) -> FactRow { + FactRow { + id: Uuid::nil(), + subject: "HQ Lease".into(), + predicate: Some("landlord".into()), + object: Some("BBHQ1".into()), + valid_from: from.map(at), + valid_from_precision: from.map(|_| "day".into()), + valid_to: None, + valid_to_precision: None, + confidence: 0.9, + stale: false, + } + } + + fn ev(dated: Option<&str>, quote: &str) -> Evidence { + Evidence { + chunk_id: Uuid::nil(), + document: "eleventh-amendment.html".into(), + dated: dated.map(at), + quote: Some(quote.into()), + } + } + + /// 模型给的日期要在证据里：文档的日期、事实的起点，或引文写着的那一天 + #[test] + fn a_date_is_taken_only_when_the_evidence_shows_it() { + let evidence = [ev( + Some("2020-08-13"), + "BBHQ1, LLC succeeded HPBB1 as landlord on July 31, 2020", + )]; + let f = row(Some("2016-05-16")); + assert!( + date_in_evidence(Some("2020-08-13"), &f, &evidence).is_some(), + "文档的日期" + ); + assert!( + date_in_evidence(Some("2016-05-16"), &f, &evidence).is_some(), + "事实的起点" + ); + assert!( + date_in_evidence(Some("2020-07-31"), &f, &evidence).is_some(), + "引文里写着" + ); + assert!( + date_in_evidence(Some("July 31, 2020"), &f, &evidence).is_some(), + "写法不同也认" + ); + assert!( + date_in_evidence(Some("2020-09-01"), &f, &evidence).is_none(), + "证据里没有" + ); + assert!(date_in_evidence(None, &f, &evidence).is_none()); + assert!(date_in_evidence(Some("soon"), &f, &evidence).is_none()); + } + + #[test] + fn an_action_lands_in_the_ledger_under_the_queue_it_came_from() { + assert_eq!(audit_action("fact", "confirm"), "fact.confirm"); + assert_eq!(audit_action("conflict", "close_old"), "conflict.close_old"); + assert_eq!( + audit_action("conflict", "retime_new"), + "fact.time_corrected" + ); + } +} + +#[cfg(test)] +#[path = "queue_agent_tests.rs"] +mod queue_agent_tests; diff --git a/crates/utopia-server/src/queue_agent_tests.rs b/crates/utopia-server/src/queue_agent_tests.rs new file mode 100644 index 000000000..dd8b2ddd5 --- /dev/null +++ b/crates/utopia-server/src/queue_agent_tests.rs @@ -0,0 +1,459 @@ +//! 事实与冲突两档交给 agent（0043），走整条路：假模型按脚本回话，agent 动手，人再撤回。 +//! +//! 租约的形状：第五份补充协议的截止日（0.9），第六份的新截止日模型给了 0.7——不许它 +//! 接替，队列里挂着一条低置信事实与一对低置信冲突。另有一条原文根本没说的事实，和两个 +//! 同一天开始的租户（后来的那个，原文写着它哪天才成为租户）。 +//! +//! 1. agent 确认第六份的截止日：置信度升到 1.0，第五份的截止日关在它开始时，那对冲突撤下。 +//! 2. agent 驳回原文没说的那条。 +//! 3. agent 把后来那个租户的起点改成原文写的那天：先前那个租户关在那一天。 +//! 4. 人把三步都撤回：截止日回到 0.7、第五份重新开着；驳回的事实回来；租户的起点回到原样， +//! 那对「同一天开始」的冲突重新挂上。撤回到第二次，保险丝把开关关掉。 +//! +//! 没有 `UTOPIA_DATABASE_URL` 时跳过而不是失败。自建自拆，绝不碰已有的库。 + +use super::*; +use std::sync::Arc; +use utopia_store::graph::Validity; +use utopia_store::temporal::Uniqueness; +use wiremock::{ + matchers::{method, path}, + Mock, MockServer, Request, Respond, ResponseTemplate, +}; + +/// 按提示词里的那一句回话：审事实的一批、裁冲突的一批 +#[derive(Clone)] +struct Scripted; + +impl Respond for Scripted { + fn respond(&self, request: &Request) -> ResponseTemplate { + let body: Value = request.body_json().expect("chat request is JSON"); + let system = body["messages"][0]["content"].as_str().unwrap_or_default(); + let user = body["messages"][1]["content"].as_str().unwrap_or_default(); + // 每一项按它在批里的编号答：认得出是哪一项才答，其余 unsure + let mut verdicts = Vec::new(); + for (i, block) in user + .split("\n\n") + .filter(|b| !b.trim().is_empty()) + .enumerate() + { + let v = if system.contains("You review facts") { + if block.contains("2020-04-14") { + json!({ "i": i, "action": "confirm", "confidence": 0.95, + "why": "the sixth amendment's table states it" }) + } else if block.contains("Tower 9") { + json!({ "i": i, "action": "reject", "confidence": 0.9, + "why": "the quote does not mention it" }) + } else { + json!({ "i": i, "action": "unsure", "confidence": 0.2 }) + } + } else if block.contains("B Corp") { + json!({ "i": i, "action": "retime_new", "confidence": 0.9, + "why": "B Corp became the tenant on June 1, 2021", "date": "2021-06-01" }) + } else { + json!({ "i": i, "action": "unsure", "confidence": 0.2 }) + }; + verdicts.push(v); + } + let content = json!({ "verdicts": verdicts }).to_string(); + ResponseTemplate::new(200).set_body_json(json!({ + "choices": [{ "message": { "role": "assistant", "content": content } }] + })) + } +} + +struct Fx { + state: AppState, + pool: sqlx::PgPool, + org: Uuid, + kb: Uuid, + user: Uuid, + lease: Uuid, + deadline: Uuid, + tenant: Uuid, + _server: MockServer, + dir: std::path::PathBuf, +} + +fn t(day: &str) -> DateTime { + format!("{day}T00:00:00Z").parse().unwrap() +} + +async fn fixture() -> anyhow::Result> { + let Some(url) = utopia_store::test_db::url() else { + return Ok(None); + }; + let pool = sqlx::PgPool::connect(&url).await?; + let (org, ws, kb, user) = ( + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + ); + let (etype, deadline, tenant, lease) = ( + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + ); + sqlx::query("INSERT INTO organizations(id,name) VALUES($1,'queue-agent-test')") + .bind(org) + .execute(&pool) + .await?; + sqlx::query("INSERT INTO workspaces(id,org_id,name) VALUES($1,$2,'queue-agent-test')") + .bind(ws) + .bind(org) + .execute(&pool) + .await?; + sqlx::query( + "INSERT INTO knowledge_bases(id,workspace_id,name,governance) VALUES($1,$2,'queue-agent-test',TRUE)", + ) + .bind(kb) + .bind(ws) + .execute(&pool) + .await?; + sqlx::query( + "INSERT INTO users(id,org_id,email,password_hash,display_name,is_admin) + VALUES($1,$2,$3,'','Queue Agent',TRUE)", + ) + .bind(user) + .bind(org) + .bind(format!("queue-agent-{user}@test.local")) + .execute(&pool) + .await?; + sqlx::query( + "INSERT INTO entity_types (id, kb_id, key, label) VALUES ($1, $2, 'lease', 'Lease')", + ) + .bind(etype) + .bind(kb) + .execute(&pool) + .await?; + for (id, key) in [(deadline, "option_deadline"), (tenant, "tenant_name")] { + sqlx::query( + "INSERT INTO relation_types (id, kb_id, key, label, kind, datatype, temporal, functional) + VALUES ($1, $2, $3, $3, 'attribute', 'text', 'state', TRUE)", + ) + .bind(id) + .bind(kb) + .bind(key) + .execute(&pool) + .await?; + } + sqlx::query( + "INSERT INTO entities (id, kb_id, type_id, canonical_name) VALUES ($1, $2, $3, 'HQ Lease')", + ) + .bind(lease) + .bind(kb) + .bind(etype) + .execute(&pool) + .await?; + + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/chat/completions")) + .respond_with(Scripted) + .mount(&server) + .await; + utopia_store::settings::upsert( + &pool, + ws, + Some(&server.uri()), + None, + Some("scripted-chat"), + None, + None, + None, + None, + ) + .await?; + let dir = std::env::temp_dir().join(format!("utopia-queue-agent-{kb}")); + let cfg = utopia_core::config::AppConfig { + data_dir: dir.to_string_lossy().into_owned(), + ..Default::default() + }; + let search = Arc::new(utopia_search::SearchIndex::open(&dir.join("search"))?); + let state = AppState::new(pool.clone(), &cfg, search, "test-only".into()); + Ok(Some(Fx { + state, + pool, + org, + kb, + user, + lease, + deadline, + tenant, + _server: server, + dir, + })) +} + +impl Fx { + /// 抽取的写法：一份自带日期的文档、一段引文、落库、写证据、对账 + async fn observe( + &self, + predicate: Uuid, + value: &str, + from: &str, + doc_day: &str, + quote: &str, + confidence: f32, + ) -> anyhow::Result { + let (d, c) = (Uuid::now_v7(), Uuid::now_v7()); + sqlx::query( + "INSERT INTO documents (id, kb_id, filename, sha256, doc_time, doc_time_source) + VALUES ($1, $2, $3, $3, $4, 'content')", + ) + .bind(d) + .bind(self.kb) + .bind(format!("doc-{doc_day}-{d}.html")) + .bind(t(doc_day)) + .execute(&self.pool) + .await?; + sqlx::query( + "INSERT INTO chunks (id, kb_id, document_id, seq, text) VALUES ($1, $2, $3, 0, $4)", + ) + .bind(c) + .bind(self.kb) + .bind(d) + .bind(quote) + .execute(&self.pool) + .await?; + let validity = Validity::starting(Some(t(from)), Some("day")).attested(Some(t(doc_day))); + let object = json!({ "value": value }); + let (id, _) = utopia_store::graph::insert_value_fact( + &self.pool, + self.kb, + self.lease, + Some(predicate), + &object, + validity, + confidence, + ) + .await?; + utopia_store::graph::add_evidence(&self.pool, id, c, Some(quote), None).await?; + utopia_store::temporal::reconcile_new_fact( + &self.pool, + self.kb, + id, + self.lease, + predicate, + None, + Some(&object), + Uniqueness::SubjectSide, + validity, + confidence, + ) + .await?; + Ok(id) + } + + /// (值, 起点, 终点, 置信度)，按起点排 + async fn timeline( + &self, + predicate: Uuid, + ) -> anyhow::Result, f32)>> { + Ok(sqlx::query_as( + "SELECT object_value ->> 'value', to_char(valid_from, 'YYYY-MM-DD'), + to_char(valid_to, 'YYYY-MM-DD'), confidence + FROM facts WHERE kb_id = $1 AND predicate_id = $2 AND invalidated_at IS NULL + ORDER BY valid_from, object_value ->> 'value'", + ) + .bind(self.kb) + .bind(predicate) + .fetch_all(&self.pool) + .await?) + } + + async fn open_conflicts(&self) -> anyhow::Result> { + Ok(sqlx::query_scalar( + "SELECT reason FROM fact_conflicts WHERE kb_id = $1 AND status = 'open' ORDER BY reason", + ) + .bind(self.kb) + .fetch_all(&self.pool) + .await?) + } + + async fn govern(&self) -> anyhow::Result<()> { + crate::governance::govern(&self.state, self.kb).await + } + + async fn cleanup(self) -> anyhow::Result<()> { + sqlx::query("DELETE FROM knowledge_bases WHERE id = $1") + .bind(self.kb) + .execute(&self.pool) + .await?; + sqlx::query("DELETE FROM users WHERE id = $1") + .bind(self.user) + .execute(&self.pool) + .await?; + sqlx::query("DELETE FROM organizations WHERE id = $1") + .bind(self.org) + .execute(&self.pool) + .await?; + let _ = std::fs::remove_dir_all(&self.dir); + Ok(()) + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn the_agent_settles_facts_and_conflicts_and_a_person_can_take_it_back() -> anyhow::Result<()> +{ + let Some(fx) = fixture().await? else { + return Ok(()); + }; + fx.observe( + fx.deadline, + "2020-03-17", + "2020-02-18", + "2020-02-18", + "(Phase 2 Exercise Deadline) | March 17, 2020", + 0.9, + ) + .await?; + fx.observe( + fx.deadline, + "2020-04-14", + "2020-03-17", + "2020-03-17", + "(Phase 2 Exercise Deadline) | April 14, 2020", + 0.7, + ) + .await?; + fx.observe( + fx.tenant, + "Tower 9", + "2019-01-01", + "2019-01-01", + "The landlord shall maintain the parking deck.", + 0.5, + ) + .await?; + fx.observe( + fx.tenant, + "A Corp", + "2020-01-01", + "2020-01-01", + "A Corp is the tenant under the lease.", + 0.9, + ) + .await?; + fx.observe( + fx.tenant, + "B Corp", + "2020-01-01", + "2021-06-01", + "B Corp became the tenant on June 1, 2021.", + 0.9, + ) + .await?; + let before = fx.open_conflicts().await?; + assert!(before.contains(&"low_confidence".to_string()), "{before:?}"); + assert!(before.contains(&"simultaneous".to_string()), "{before:?}"); + + fx.govern().await?; + + let deadlines = fx.timeline(fx.deadline).await?; + assert_eq!( + deadlines, + vec![ + ( + "2020-03-17".into(), + "2020-02-18".into(), + Some("2020-03-17".into()), + 0.9 + ), + ("2020-04-14".into(), "2020-03-17".into(), None, 1.0), + ], + "确认之后第六份接替了第五份" + ); + let tenants = fx.timeline(fx.tenant).await?; + assert!( + tenants.iter().all(|r| r.0 != "Tower 9"), + "原文没说的那条驳回了：{tenants:?}" + ); + assert_eq!( + tenants, + vec![ + ( + "A Corp".into(), + "2020-01-01".into(), + Some("2021-06-01".into()), + 0.9 + ), + ("B Corp".into(), "2021-06-01".into(), None, 0.9), + ], + "B Corp 的起点改成原文写的那天，A Corp 关在那一天" + ); + assert_eq!(fx.open_conflicts().await?, Vec::::new()); + + let decisions: Vec<(Uuid, String, String, String, Option)> = sqlx::query_as( + "SELECT id, target_kind, action, status, summary FROM agent_decisions + WHERE kb_id = $1 AND status = 'applied' ORDER BY created_at", + ) + .bind(fx.kb) + .fetch_all(&fx.pool) + .await?; + let applied: Vec<(&str, &str)> = decisions + .iter() + .map(|d| (d.1.as_str(), d.2.as_str())) + .collect(); + assert_eq!( + applied, + vec![ + ("fact", "confirm"), + ("fact", "reject"), + ("conflict", "retime_new") + ], + "{decisions:?}" + ); + assert!( + decisions.iter().all(|d| d.4.is_some()), + "每一笔都留了给人读的一行" + ); + + // 人把三笔都撤回 + for (id, ..) in &decisions { + let d = gov::get(&fx.pool, fx.kb, *id).await?; + answer( + &fx.state, + fx.kb, + &d, + "revert", + fx.user, + Some("checking by hand"), + ) + .await?; + } + let deadlines = fx.timeline(fx.deadline).await?; + assert_eq!( + deadlines, + vec![ + ("2020-03-17".into(), "2020-02-18".into(), None, 0.9), + ("2020-04-14".into(), "2020-03-17".into(), None, 0.7), + ], + "撤回确认：回到 0.7，第五份重新开着" + ); + let tenants = fx.timeline(fx.tenant).await?; + assert!( + tenants.iter().any(|r| r.0 == "Tower 9"), + "撤回驳回：那条回来了" + ); + assert!( + tenants + .iter() + .any(|r| r.0 == "B Corp" && r.1 == "2020-01-01"), + "撤回改起点：回到原样：{tenants:?}" + ); + let after = fx.open_conflicts().await?; + assert!( + after.contains(&"simultaneous".to_string()), + "那对同一天开始的冲突挂回来了：{after:?}" + ); + let governance: bool = + sqlx::query_scalar("SELECT governance FROM knowledge_bases WHERE id = $1") + .bind(fx.kb) + .fetch_one(&fx.pool) + .await?; + assert!(!governance, "撤回到第二次，保险丝关了开关"); + fx.cleanup().await +} diff --git a/crates/utopia-store/src/governance.rs b/crates/utopia-store/src/governance.rs index 6476316c8..1d2567d98 100644 --- a/crates/utopia-store/src/governance.rs +++ b/crates/utopia-store/src/governance.rs @@ -683,13 +683,40 @@ pub struct NewDecision<'a> { pub calls: i32, } +/// 重复对以外的一档（0043）：哪一档、做决定那一刻这一项的样子、这一步的参数与撤回要用的东西 +pub struct Target<'a> { + /// review | fact | conflict + pub kind: &'a str, + pub summary: Option<&'a str>, + pub detail: serde_json::Value, +} + +impl Target<'_> { + fn review() -> Self { + Target { + kind: "review", + summary: None, + detail: serde_json::json!({}), + } + } +} + pub async fn record(pool: &PgPool, kb_id: Uuid, d: NewDecision<'_>) -> AppResult { + record_for(pool, kb_id, d, Target::review()).await +} + +pub async fn record_for( + pool: &PgPool, + kb_id: Uuid, + d: NewDecision<'_>, + target: Target<'_>, +) -> AppResult { let id = Uuid::now_v7(); sqlx::query( "INSERT INTO agent_decisions (id, kb_id, run_id, target_kind, target_id, action, confidence, reason, - precedents, status, merge_id, question, trace, calls) - VALUES ($1, $2, $3, 'review', $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)", + precedents, status, merge_id, question, trace, calls, summary, detail) + VALUES ($1, $2, $3, $14, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $15, $16)", ) .bind(id) .bind(kb_id) @@ -704,6 +731,9 @@ pub async fn record(pool: &PgPool, kb_id: Uuid, d: NewDecision<'_>) -> AppResult .bind(d.question) .bind(d.trace) .bind(d.calls) + .bind(target.kind) + .bind(target.summary) + .bind(target.detail) .execute(pool) .await?; Ok(id) @@ -821,6 +851,7 @@ pub async fn namesakes( const VIEW: &str = "SELECT d.id, d.run_id, d.target_kind, d.target_id, d.action, d.confidence, d.reason, d.precedents, d.status, d.merge_id, d.question, d.trace, d.calls, + d.summary, d.detail, d.created_at, d.decided_at, u.display_name AS decided_by_name, a.canonical_name AS \"left\", b.canonical_name AS \"right\" @@ -1066,9 +1097,9 @@ pub async fn settle_by_merge( Ok(id) } -/// 开关开着、队列里还有 agent 没看过的对的库：定时扫描用 +/// 开关开着、还有 agent 没看过的项的库：定时扫描用。重复对之外，事实与冲突两档也算（0043） pub async fn due(pool: &PgPool) -> AppResult> { - let ids = sqlx::query_scalar(&format!( + let mut ids: Vec = sqlx::query_scalar(&format!( "SELECT kb.id FROM knowledge_bases kb WHERE kb.governance AND EXISTS ( SELECT 1 FROM resolution_reviews rr @@ -1076,6 +1107,14 @@ pub async fn due(pool: &PgPool) -> AppResult> { )) .fetch_all(pool) .await?; + let governed: Vec = sqlx::query_scalar("SELECT id FROM knowledge_bases WHERE governance") + .fetch_all(pool) + .await?; + for kb_id in governed { + if !ids.contains(&kb_id) && crate::queue_agent::has_backlog(pool, kb_id).await? { + ids.push(kb_id); + } + } Ok(ids) } diff --git a/crates/utopia-store/src/graph.rs b/crates/utopia-store/src/graph.rs index d3230cdf4..a9ed03b73 100644 --- a/crates/utopia-store/src/graph.rs +++ b/crates/utopia-store/src/graph.rs @@ -1490,14 +1490,11 @@ pub async fn stale_facts( /// 人工确认低置信事实：置信度提到 1.0。 pub async fn confirm_fact(pool: &PgPool, kb_id: Uuid, fact_id: Uuid) -> AppResult<()> { - let res = sqlx::query( - "UPDATE facts SET confidence = 1.0 WHERE id = $1 AND kb_id = $2 AND invalidated_at IS NULL", - ) - .bind(fact_id) - .bind(kb_id) - .execute(pool) - .await?; - if res.rows_affected() == 0 { + // 置信度够了，它就能接替前任：时间线跟着重算（0043） + if crate::temporal::set_confidence(pool, kb_id, fact_id, 1.0) + .await? + .is_none() + { return Err(AppError::NotFound); } // 从前这里还有一段：确认 `mapped_to` 事实时把同 (概念, 源) 的旧映射作废。 diff --git a/crates/utopia-store/src/lib.rs b/crates/utopia-store/src/lib.rs index e25e32038..044d8ac2e 100644 --- a/crates/utopia-store/src/lib.rs +++ b/crates/utopia-store/src/lib.rs @@ -27,6 +27,7 @@ pub mod ontology; pub mod palette; pub mod paths; pub mod pending; +pub mod queue_agent; pub mod reasoning; pub mod record_axis; pub mod resolution; diff --git a/crates/utopia-store/src/queue_agent.rs b/crates/utopia-store/src/queue_agent.rs new file mode 100644 index 000000000..9ac5209a7 --- /dev/null +++ b/crates/utopia-store/src/queue_agent.rs @@ -0,0 +1,267 @@ +//! 审核台其余几档交给 agent（0043）时要取的数：等着的项、它们的证据、原文现在怎么写、 +//! 人在这一档上做过的决定。只取数，不做决定——决定在 `utopia_server::queue_agent`。 +//! +//! 取的都是**还没被 agent 看过**的项：agent 看过、说了什么的，要么已经动手（那一项多半 +//! 已经离开了队列），要么留了建议等人。行被时间线重算改写过、换了 id 的，顺着 supersedes +//! 往上找，前身被看过就算看过——不然每重算一次，同一条事实就再问一遍模型。 + +use chrono::{DateTime, Utc}; +use sqlx::PgPool; +use utopia_core::AppResult; +use uuid::Uuid; + +/// agent 在这一行或它的前身上已经有过一笔决定（任何状态）。别名固定用 `f` +const LOOKED_AT: &str = "EXISTS ( + WITH RECURSIVE lineage(id) AS ( + SELECT f.id + UNION SELECT p.supersedes FROM facts p JOIN lineage l ON p.id = l.id + WHERE p.supersedes IS NOT NULL) + SELECT 1 FROM agent_decisions d + WHERE d.target_kind = 'fact' AND d.target_id IN (SELECT id FROM lineage))"; + +/// 一条事实，给 agent 看的样子 +#[derive(Debug, Clone, sqlx::FromRow)] +pub struct FactRow { + pub id: Uuid, + pub subject: String, + pub predicate: Option, + /// 实体宾语的名字，或字面值的原样 + pub object: Option, + pub valid_from: Option>, + pub valid_from_precision: Option, + pub valid_to: Option>, + pub valid_to_precision: Option, + pub confidence: f32, + /// 证据全在旧版本的分块上（「待确认」那一档） + pub stale: bool, +} + +const FACT_COLUMNS: &str = "f.id, s.canonical_name AS subject, + COALESCE(r.label, fact_surface_predicate(f.id)) AS predicate, + COALESCE(o.canonical_name, f.object_value ->> 'summary', f.object_value ->> 'value', + f.object_value #>> '{}') AS object, + f.valid_from, f.valid_from_precision, f.valid_to, f.valid_to_precision, f.confidence"; + +/// 低置信与证据过期两档里还没被 agent 看过的事实，先来先看。派生事实不在其中（它们不是 +/// 抽出来的，没有原文可核） +pub async fn fact_queue(pool: &PgPool, kb_id: Uuid, limit: i64) -> AppResult> { + let stale = crate::review::UNCONFIRMED_FACT; + let rows = sqlx::query_as(&format!( + "SELECT {FACT_COLUMNS}, ({stale}) AS stale + FROM facts f + JOIN entities s ON s.id = f.subject_id + LEFT JOIN relation_types r ON r.id = f.predicate_id + LEFT JOIN entities o ON o.id = f.object_id + WHERE f.kb_id = $1 AND f.invalidated_at IS NULL AND f.derived_by_rule IS NULL + AND (f.confidence < $2 OR ({stale})) + AND NOT {LOOKED_AT} + ORDER BY f.recorded_at, f.id + LIMIT $3" + )) + .bind(kb_id) + .bind(crate::review::LOW_CONFIDENCE_BELOW) + .bind(limit) + .fetch_all(pool) + .await?; + Ok(rows) +} + +/// 一条事实，按 id 取（不论死活） +pub async fn fact(pool: &PgPool, kb_id: Uuid, fact_id: Uuid) -> AppResult> { + let stale = crate::review::UNCONFIRMED_FACT; + Ok(sqlx::query_as(&format!( + "SELECT {FACT_COLUMNS}, ({stale}) AS stale + FROM facts f + JOIN entities s ON s.id = f.subject_id + LEFT JOIN relation_types r ON r.id = f.predicate_id + LEFT JOIN entities o ON o.id = f.object_id + WHERE f.kb_id = $1 AND f.id = $2" + )) + .bind(kb_id) + .bind(fact_id) + .fetch_optional(pool) + .await?) +} + +/// 同一持有者、同一谓词上现存的其他行：冲突两边在时间线上的邻居 +pub async fn neighbours(pool: &PgPool, kb_id: Uuid, fact_id: Uuid) -> AppResult> { + let stale = crate::review::UNCONFIRMED_FACT; + Ok(sqlx::query_as(&format!( + "SELECT {FACT_COLUMNS}, ({stale}) AS stale + FROM facts f + JOIN facts me ON me.id = $2 + JOIN entities s ON s.id = f.subject_id + LEFT JOIN relation_types r ON r.id = f.predicate_id + LEFT JOIN entities o ON o.id = f.object_id + WHERE f.kb_id = $1 AND f.invalidated_at IS NULL AND f.id <> me.id + AND f.subject_id = me.subject_id AND f.predicate_id = me.predicate_id + ORDER BY f.valid_from NULLS LAST, f.recorded_at + LIMIT 12" + )) + .bind(kb_id) + .bind(fact_id) + .fetch_all(pool) + .await?) +} + +/// 一条证据：哪份文档、那份文档自己的日期、引文 +#[derive(Debug, Clone, sqlx::FromRow)] +pub struct Evidence { + pub chunk_id: Uuid, + pub document: String, + /// 文档自己的日期（正文或来源给的），上传时刻不算 + pub dated: Option>, + pub quote: Option, +} + +pub async fn evidence(pool: &PgPool, fact_id: Uuid) -> AppResult> { + Ok(sqlx::query_as( + "SELECT fe.chunk_id, d.filename AS document, + CASE WHEN d.doc_time_source IN ('content', 'source') THEN d.doc_time END AS dated, + fe.quote + FROM fact_evidence fe + JOIN documents d ON d.id = fe.document_id + WHERE fe.fact_id = $1 AND d.deleted_at IS NULL + ORDER BY d.doc_time NULLS LAST, fe.chunk_id + LIMIT 4", + ) + .bind(fact_id) + .fetch_all(pool) + .await?) +} + +/// 证据过期的事实：它的文档**现在**的分块里提到主语的那几块。旧版本说过的话，新版本 +/// 还说不说，得看新版本 +pub async fn current_text( + pool: &PgPool, + fact_id: Uuid, + subject: &str, +) -> AppResult> { + Ok(sqlx::query_as( + "SELECT c.id, left(c.text, 1500) + FROM chunks c + WHERE c.superseded_at IS NULL + AND c.document_id IN (SELECT fe.document_id FROM fact_evidence fe WHERE fe.fact_id = $1) + AND strpos(lower(c.text), lower($2)) > 0 + ORDER BY c.seq + LIMIT 2", + ) + .bind(fact_id) + .bind(subject) + .fetch_all(pool) + .await?) +} + +/// 一条等着的冲突 +#[derive(Debug, Clone, sqlx::FromRow)] +pub struct ConflictRow { + pub id: Uuid, + pub reason: String, + pub old_fact_id: Uuid, + pub new_fact_id: Uuid, +} + +/// 开着、还没被 agent 看过的冲突，先来先看 +pub async fn conflict_queue(pool: &PgPool, kb_id: Uuid, limit: i64) -> AppResult> { + Ok(sqlx::query_as( + "SELECT c.id, c.reason, c.old_fact_id, c.new_fact_id + FROM fact_conflicts c + WHERE c.kb_id = $1 AND c.status = 'open' + AND NOT EXISTS (SELECT 1 FROM agent_decisions d + WHERE d.target_kind = 'conflict' AND d.target_id = c.id) + ORDER BY c.created_at, c.id + LIMIT $2", + ) + .bind(kb_id) + .bind(limit) + .fetch_all(pool) + .await?) +} + +/// 这个库里的人在这几种决定上最近怎么做的：每条一行，给模型当先例（0025 的约定：只认 +/// 人写的行，机器自己的决定不是先例） +pub async fn precedents( + pool: &PgPool, + kb_id: Uuid, + actions: &[&str], + limit: i64, +) -> AppResult> { + let rows: Vec<(String, serde_json::Value)> = sqlx::query_as( + "SELECT action, detail FROM audit_events + WHERE kb_id = $1 AND actor_id IS NOT NULL AND action = ANY($2) + ORDER BY created_at DESC + LIMIT $3", + ) + .bind(kb_id) + .bind(actions) + .bind(limit) + .fetch_all(pool) + .await?; + Ok(rows + .into_iter() + .map(|(action, d)| render_precedent(&action, &d)) + .collect()) +} + +/// 台账上一条人的决定写成一行 +pub fn render_precedent(action: &str, d: &serde_json::Value) -> String { + let text = |k: &str| d.get(k).and_then(|v| v.as_str()).unwrap_or(""); + let body = if action.starts_with("conflict.") { + format!( + "{} {} {} vs {} {} {}", + text("old_subject"), + text("predicate"), + text("old_object"), + text("new_subject"), + text("predicate"), + text("new_object") + ) + } else { + format!( + "{} {} {}", + text("subject"), + text("predicate"), + text("object") + ) + }; + let why = d + .get("why") + .and_then(|v| v.as_str()) + .map(|w| format!(" (because: {w})")) + .unwrap_or_default(); + format!( + "{action}: {}{why}", + body.split_whitespace().collect::>().join(" ") + ) +} + +/// 还有没被 agent 看过的事实或冲突（定时扫描用） +pub async fn has_backlog(pool: &PgPool, kb_id: Uuid) -> AppResult { + Ok(!fact_queue(pool, kb_id, 1).await?.is_empty() + || !conflict_queue(pool, kb_id, 1).await?.is_empty()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_precedent_reads_as_one_line_with_its_reason() { + let fact = serde_json::json!({ + "subject": "HQ Lease", "predicate": "option deadline", "object": "2020-04-14", + "why": "the sixth amendment's table states it" + }); + assert_eq!( + render_precedent("fact.confirm", &fact), + "fact.confirm: HQ Lease option deadline 2020-04-14 (because: the sixth amendment's table states it)" + ); + let conflict = serde_json::json!({ + "predicate": "landlord", "old_subject": "HQ Lease", "old_object": "HPBB1", + "new_subject": "HQ Lease", "new_object": "BBHQ1" + }); + assert_eq!( + render_precedent("conflict.close_old", &conflict), + "conflict.close_old: HQ Lease landlord HPBB1 vs HQ Lease landlord BBHQ1" + ); + } +} diff --git a/crates/utopia-store/src/temporal.rs b/crates/utopia-store/src/temporal.rs index 0cfedad30..7bda0c201 100644 --- a/crates/utopia-store/src/temporal.rs +++ b/crates/utopia-store/src/temporal.rs @@ -563,13 +563,43 @@ async fn tidy( report.corrected.push(corrected); } } + let mut still_held = HashSet::new(); for (row, later) in held { let row = rewritten.get(&row).copied().unwrap_or(row); let later = rewritten.get(&later).copied().unwrap_or(later); + still_held.insert((row, later)); if record_conflict_tx(tx, kb_id, row, later, "low_confidence").await? { report.conflicts += 1; } } + // 置信度不够、交给人的那一对，后任后来被人或 agent 确认过（0043）：这道题没了， + // 撤下——不撤的话它挂在队列里，问的是一件引擎已经排好的事 + let on_timeline: Vec = rows + .iter() + .map(|r| r.id) + .chain(rewritten.values().copied()) + .collect(); + let open: Vec<(Uuid, Uuid, Uuid)> = sqlx::query_as( + "SELECT c.id, c.old_fact_id, c.new_fact_id FROM fact_conflicts c + WHERE c.kb_id = $1 AND c.status = 'open' AND c.reason = 'low_confidence' + AND c.new_fact_id = ANY($2)", + ) + .bind(kb_id) + .bind(&on_timeline) + .fetch_all(&mut **tx) + .await?; + for (conflict, old, new) in open { + if !still_held.contains(&(old, new)) { + sqlx::query( + "UPDATE fact_conflicts SET status = 'withdrawn', resolution = NULL, + resolved_at = now() + WHERE id = $1 AND status = 'open'", + ) + .bind(conflict) + .execute(&mut **tx) + .await?; + } + } Ok(()) } @@ -671,6 +701,117 @@ pub async fn reconcile_predicate( reconcile_facts(pool, kb_id, &ids).await } +/// 改一条事实的置信度，时间线跟着重算：置信度决定它能不能接替前任（0022），人或 agent +/// 确认了一条低置信的值，前任就该关在它开始时。先锁时间线再改（见模块头）。 +/// 返回原来的置信度；行不在或已作废返回 `None` +pub async fn set_confidence( + pool: &PgPool, + kb_id: Uuid, + fact_id: Uuid, + confidence: f32, +) -> AppResult> { + let mut tx = pool.begin().await?; + let timelines = timelines_of(&mut *tx, kb_id, &[fact_id], None).await?; + lock_timelines(&mut tx, kb_id, &timelines).await?; + let prior: Option = sqlx::query_scalar( + "SELECT confidence FROM facts + WHERE id = $1 AND kb_id = $2 AND invalidated_at IS NULL FOR UPDATE", + ) + .bind(fact_id) + .bind(kb_id) + .fetch_optional(&mut *tx) + .await?; + if prior.is_some() { + sqlx::query("UPDATE facts SET confidence = $2 WHERE id = $1") + .bind(fact_id) + .bind(confidence) + .execute(&mut *tx) + .await?; + tidy_timelines_tx(&mut tx, kb_id, &timelines).await?; + } + tx.commit().await?; + Ok(prior) +} + +/// 撤回一次撤掉：作废了的事实回来，时间线按它回来之后的样子重算。agent 驳回一条事实之后， +/// 人撤回那个决定走这里。返回有没有救回一行 +pub async fn restore(pool: &PgPool, kb_id: Uuid, fact_id: Uuid) -> AppResult { + let mut tx = pool.begin().await?; + let timelines = timelines_of(&mut *tx, kb_id, &[fact_id], None).await?; + lock_timelines(&mut tx, kb_id, &timelines).await?; + let restored = sqlx::query( + "UPDATE facts SET invalidated_at = NULL + WHERE id = $1 AND kb_id = $2 AND invalidated_at IS NOT NULL", + ) + .bind(fact_id) + .bind(kb_id) + .execute(&mut *tx) + .await? + .rows_affected() + > 0; + if restored { + tidy_timelines_tx(&mut tx, kb_id, &timelines).await?; + } + tx.commit().await?; + Ok(restored) +} + +/// 撤回一次改写：改写出来的那一行作废，被它取代的原行回来，时间线重算。agent 关上一条 +/// 事实、改过一条事实的起点，人撤回时走这里。改写出来的行后来又被改写过的（换过终点、 +/// 被人改过），不再撤：那之后的变化有它自己的依据。返回有没有撤成 +pub async fn undo_rewrite( + pool: &PgPool, + kb_id: Uuid, + corrected: Uuid, + original: Uuid, +) -> AppResult { + let mut tx = pool.begin().await?; + let timelines = timelines_of(&mut *tx, kb_id, &[corrected, original], None).await?; + lock_timelines(&mut tx, kb_id, &timelines).await?; + let still: Option = sqlx::query_scalar( + "SELECT id FROM facts + WHERE id = $1 AND kb_id = $2 AND supersedes = $3 AND invalidated_at IS NULL + FOR UPDATE", + ) + .bind(corrected) + .bind(kb_id) + .bind(original) + .fetch_optional(&mut *tx) + .await?; + if still.is_none() { + tx.commit().await?; + return Ok(false); + } + sqlx::query("UPDATE facts SET invalidated_at = now() WHERE id = $1") + .bind(corrected) + .execute(&mut *tx) + .await?; + sqlx::query("UPDATE facts SET invalidated_at = NULL WHERE id = $1") + .bind(original) + .execute(&mut *tx) + .await?; + tidy_timelines_tx(&mut tx, kb_id, &timelines).await?; + tx.commit().await?; + Ok(true) +} + +/// 一条裁过的冲突重新开着：撤回裁决时用。两边都还活着才开得回来 +pub async fn reopen_conflict(pool: &PgPool, kb_id: Uuid, conflict_id: Uuid) -> AppResult { + let n = sqlx::query( + "UPDATE fact_conflicts c SET status = 'open', resolution = NULL, resolved_at = NULL + WHERE c.id = $1 AND c.kb_id = $2 AND c.status <> 'open' + AND NOT EXISTS (SELECT 1 FROM facts f + WHERE f.id IN (c.old_fact_id, c.new_fact_id) + AND f.invalidated_at IS NOT NULL)", + ) + .bind(conflict_id) + .bind(kb_id) + .execute(pool) + .await? + .rows_affected(); + Ok(n > 0) +} + /// 撤掉一条事实（人判它是抽取错误）：它从来不在，时间线按剩下的行重算——关在它开始时的 /// 前任重新接上。先锁时间线再作废（见模块头）。返回有没有撤掉一行；已经作废的不算 pub async fn retract(pool: &PgPool, kb_id: Uuid, fact_id: Uuid) -> AppResult { @@ -1082,7 +1223,7 @@ pub async fn list_conflicts( } /// 人工裁决：close（旧事实闭合于 close_at 或新事实起点）/ keep（并存不矛盾）/ -/// reject_new（新事实是抽取错误，作废）。 +/// reject_new（新事实是抽取错误，作废）。回 close 改写出来的那一行（撤回要用） /// 一条待裁决的冲突：旧事实、新事实、新事实的起点及其精度 type ConflictRow = (Uuid, Uuid, Option>, Option); @@ -1093,7 +1234,7 @@ pub async fn resolve_conflict( resolution: &str, close_at: Option>, close_at_precision: &str, -) -> AppResult<()> { +) -> AppResult> { let row: Option = sqlx::query_as( "SELECT c.old_fact_id, c.new_fact_id, fn_.valid_from, fn_.valid_from_precision FROM fact_conflicts c JOIN facts fn_ ON fn_.id = c.new_fact_id @@ -1107,6 +1248,7 @@ pub async fn resolve_conflict( return Err(utopia_core::AppError::NotFound); }; + let mut corrected = None; let stored = match resolution { "close" => { // 闭合点带着它的精度走：人给了日期就用人给的精度，没给就闭合在新事实的 @@ -1124,7 +1266,7 @@ pub async fn resolve_conflict( new_from_precision.as_deref().unwrap_or("day"), ), }; - close_superseded(pool, old_fact_id, at, precision).await?; + corrected = close_superseded(pool, old_fact_id, at, precision).await?; "closed" } "keep" => "kept_both", @@ -1157,7 +1299,7 @@ pub async fn resolve_conflict( .bind(stored) .execute(pool) .await?; - Ok(()) + Ok(corrected) } /// 一条谓词的一端挂着**两个以上开放值**的持有者——唯一性没声明（或声明来晚了） diff --git a/docs/decisions/0025-governance-reads-the-ledger-before-it-decides.md b/docs/decisions/0025-governance-reads-the-ledger-before-it-decides.md index 529696226..355bc1196 100644 --- a/docs/decisions/0025-governance-reads-the-ledger-before-it-decides.md +++ b/docs/decisions/0025-governance-reads-the-ledger-before-it-decides.md @@ -111,7 +111,7 @@ Duplicates that the agent proposed on carry the reason code `proposed`; pairs it ## Open questions -- **Other queues.** Conflicts have three human actions to learn from (`conflict.close_old`, `keep_both`, `reject_new`); unconfirmed and low-confidence facts have `fact.confirm` and `fact.reject`. Nods (`pending_facts`, 0015) stay out: the person who said it nods. +- **Other queues.** Conflicts have three human actions to learn from (`conflict.close_old`, `keep_both`, `reject_new`); unconfirmed and low-confidence facts have `fact.confirm` and `fact.reject`. Nods (`pending_facts`, 0015) stay out: the person who said it nods. Answered by [0043](0043-every-review-queue-is-governed.md) (2026-09-14): facts and conflicts are governed; violations, defects and mappings follow. - **Cross-base precedent.** A workspace's other bases may hold the same names. Kept to one base until someone asks. - **A budget for the batch.** The loop has a daily budget (decision 8); the batch does not, and since #458 a round makes up to three batch calls at once. A job runs twenty clusters and re-enqueues itself while backlog remains, which is what the adjudicator always did. Whether a base needs a cap on that too is a question for a large corpus. - **Tools the loop does not have yet.** The graph around a side beyond its direct facts, the entity's disambiguator history, a full-text search of the corpus. Add one when a deferred question keeps asking for it. diff --git a/docs/decisions/0043-every-review-queue-is-governed.md b/docs/decisions/0043-every-review-queue-is-governed.md new file mode 100644 index 000000000..453a644b0 --- /dev/null +++ b/docs/decisions/0043-every-review-queue-is-governed.md @@ -0,0 +1,69 @@ +# 0043 · Every review queue is governed + +- **Status**: cut 1 implemented · migration 0058 widens `agent_decisions` to `fact` and `conflict` and adds `summary` and `detail`; `queue_agent` in the server walks low-confidence and stale facts, then temporal conflicts, after the duplicate rounds of the same `govern` job; every applied action is revertible from the Agent queue · violations, ontology defects and concept mappings are cut 2 +- **Written**: 2026-09-14 (conventions in the [README](README.md)) +- **Related**: [0025](0025-governance-reads-the-ledger-before-it-decides.md) (the governor this extends; its open question "Other queues"), [0026](0026-a-decision-records-why.md) (a person's why becomes a precedent), [0027](0027-an-automatic-merge-is-gated-by-what-it-can-undo.md) (act on what can be undone), [0022](0022-an-unknown-date-is-not-an-open-one.md) (the temporal engine whose conflicts this settles), [0015](0015-recording-a-sentence-is-not-asserting-a-fact.md) (nods, which stay with people), #695 + +> On the Blackbaud lease bench the timeline held a deadline back because the model had marked the amendment's new date 0.7. A conflict waited for a person, and the as-of question came back with two answers. The duplicate queue had an agent; the conflict queue and the low-confidence queue had nobody but a person who never came. + +## Why a decision is needed + +0025 gave one queue, duplicate pairs, to an agent that reads the ledger before it decides. The others kept waiting: temporal conflicts, low-confidence facts, facts whose evidence a new version replaced, axiom violations, ontology defects, concept mappings. On a base nobody curates they only grow, and what they hold back is real. A low-confidence successor may not take over from its predecessor (0022), so the graph answers with both. Decided on 2026-09-14: every queue is governed, except nods. + +## Decisions + +### 1. One governor, every queue but nods + +The `govern` job runs the duplicate rounds of 0025 first, then the fact queues, then conflicts. The order is causal. A merge changes which facts share a timeline, and a confirmed fact withdraws the conflict it was held in (decision 5). The same switch starts and stops it, the same fuse (0025 decision 9) counts reverts of any kind, and the same hourly scan finds bases with backlog. Extraction enqueues the job after every document on a governed base, since conflicts and low-confidence facts come out of extraction, not only duplicate pairs. + +Nods (`pending_facts`, 0015) stay out: a remembered sentence becomes a fact when the person who said it nods, and an agent nodding for them would turn a remark into an assertion. 0025 said the same. + +### 2. Each queue keeps its own actions, and they are the people's + +Facts: `confirm` or `reject`. Conflicts: `close_old`, `retime_new`, `keep_both` or `reject_new`. Each action calls the store function a person's click calls (`temporal::set_confidence`, `temporal::retract`, `temporal::resolve_conflict`, `temporal::correct_interval`), so an agent's decision and a person's leave the graph in the same shape. `retime_new` is the one action people reach through the interval editor, not the conflict card. It exists because the commonest simultaneous conflict on contract chains is a successor whose start the model took from the wrong date. + +A decision is a row in `agent_decisions` with `target_kind` saying which queue. `summary` holds what the item looked like when decided: a fact or a conflict has no pair of names to join later, and its rows get rewritten. `detail` holds the action's parameters and what undoing it needs. + +### 3. The model decides; the server checks what can be checked + +One call per batch of eight, each item with its evidence (document, the document's own date, the quote) and the recent human decisions of that queue as precedents. The rules on precedents are those of 0025 and 0026. The model returns an action, a confidence and a sentence; at `AUTO_CONF` it is applied, below it is a proposal. Nothing the model says about the world is taken on trust where the server can check it: + +- A date for `close_old` or `retime_new` must be in the evidence: the fact's start, a document's own date, or a day the quote writes out. +- A stale fact is confirmed by quoting the current version of its document, and the quote must appear there. The quote is then attached as evidence, which is what takes the fact out of the stale queue. + +A check that fails turns a confident verdict into a proposal that says why it was held. + +### 4. Every action can be taken back + +An applied decision records its undo in `detail`: + +| Action | Undo | +|---|---| +| Confirm a low-confidence fact | Restore its prior confidence | +| Confirm a stale fact | Remove the attached evidence | +| Reject a fact | Restore the fact (`temporal::restore`) | +| `close_old`, `retime_new` | Invalidate the rewritten row and restore the original (`temporal::undo_rewrite`); the original arrives again, so its conflict is recorded again | +| `keep_both` | Reopen the conflict | +| `reject_new` | Restore the fact and reopen the conflict | + +A person answers from the Agent queue: accept or override a proposal, or revert an applied decision. An override runs the person's action through the same path and becomes a precedent. + +### 5. Confidence is part of the timeline + +Confirming a fact changes whether it may take over, so `set_confidence` locks the fact's timelines, changes the value and recomputes them, as a retraction does. A `low_confidence` conflict whose pair the recompute no longer holds is withdrawn. The question it asked, "may this doubtful value take over", has been answered. + +## What a reader sees + +Agent rows for facts and conflicts carry their summary in place of two names. A proposal is answered with that queue's own actions, and an applied decision offers Revert whatever it did. The ledger rows are the queue's own actions (`fact.confirm`, `conflict.close_old`, …) with no actor, like the duplicate governor's. + +## Dead ends + +- **One generic "approve / dismiss" action for every queue.** The queues differ in what a decision does to the graph: closing a value and keeping both are different graphs. A generic approve would have to map back onto each queue's actions anyway, and the model would decide without knowing which one it was choosing. +- **Letting the model set the close date freely.** It is the one thing a successor's evidence can be checked against, and the commonest mistake it would make is the one the conflict came from. +- **Resolving the held conflict when a fact is confirmed.** The confirmation doesn't know which conflicts it answers; the recompute does, because it is the thing that held the pair. + +## Open questions + +- **Cut 2: violations, ontology defects, concept mappings.** A violation's `fact_retracted` and `fact_closed` fit this shape. `axiom_relaxed` changes the ontology, and `fixed` for a defect means a person changed it; both need a gate before an agent applies them. +- **The second look.** The fact and conflict queues have no tool loop yet (0025 decision 5); the batch sees the evidence directly. Add one when proposals keep asking for something the batch could not see. +- **The interface.** The Agent queue shows these rows with minimal changes; the queue cards themselves don't yet show the agent's proposal inline, as duplicate cards do. diff --git a/docs/decisions/README.md b/docs/decisions/README.md index 40ea635bf..5bf9dba7b 100644 --- a/docs/decisions/README.md +++ b/docs/decisions/README.md @@ -66,6 +66,7 @@ The test for writing one: if someone (including us) looks at a piece of code in | 0040 | [A chunk says where its words came from](0040-a-chunk-says-where-its-words-came-from.md) | Proposed · nothing built · a chunk carries an **origin** (`stated`, `ocr`, `transcribed`, `described`) and the model that produced it, and an **anchor** back into the original bytes (a page and region, a time range, an image inside the file) — decided before any media reader, because a transcript stored without its times can never be tied to the recording again · the packer never mixes origins in a chunk · facts from a description enter below `AUTO_CLOSE_MIN_CONFIDENCE`, so a misread chart opens a conflict instead of closing a correct fact · per-modality model settings, media reading as a resumable job, origin and anchor in the API, MCP and RDF · cuts: ledger shape, scans via Docling, recordings, descriptions, video | | 0041 | [A name is a claim about an entity](0041-a-name-is-a-claim-about-an-entity.md) | Cut 0 built (identity bench) · cut 1 implemented (#670): names are value facts on `known_as`, the extractor reports other names, a shared name goes to the adjudicator; forward/reverse F1 0.43/0.54 → 0.68/0.68 · cuts 2–4 (name vectors and neighbours, evidence decides, re-evaluation) not started | | 0042 | [The chat loop is a runner with hooks](0042-the-chat-loop-is-a-runner-with-hooks.md) | Implemented (#548) · the loop is rig's runner and every policy is a hook with a typed result · the wire stays `LlmClient` behind `RigModel` · a turn cannot end before a tool has run, `no_evidence_needed` is the exit for questions not about the base · RAG fallback only on a 400/422 to the first request with tools · an empty reply is asked again once · the skip rate is the model's (DeepSeek-V3 1–3 of 12, Qwen2.5-72B 0) and recorded, not prevented | +| 0043 | [Every review queue is governed](0043-every-review-queue-is-governed.md) | Cut 1 implemented · the governor walks low-confidence and stale facts and temporal conflicts after its duplicate rounds, with each queue's own actions, dates and quotes checked against evidence, every applied action revertible · violations, defects and mappings are cut 2; nods stay with people | ## Not a decision record diff --git a/migrations/0058_every_queue_is_governed.sql b/migrations/0058_every_queue_is_governed.sql new file mode 100644 index 000000000..071808c4d --- /dev/null +++ b/migrations/0058_every_queue_is_governed.sql @@ -0,0 +1,21 @@ +-- 审核台的每一档都交给 agent（0043）：从前只有重复对（0025），冲突、低置信、证据过期的 +-- 事实都等人。记忆点头（pending_facts，0015）照旧留给人——那是说话的人自己点头。 +-- +-- `agent_decisions` 一张表装所有档：`target_kind` 说是哪一档的哪一行，`action` 是那一档 +-- 自己的出路。`summary` 是做决定那一刻这一项长什么样，写给人看——冲突与事实没有「两个 +-- 名字」可以拼，而行后来会被改写、换 id，到时候再去拼就对不上了。`detail` 放这一步的 +-- 参数（闭合在哪天、改成哪天、看的是哪段原文）和撤回要用的东西（改写出来的行、原来的 +-- 置信度、补上的那条证据）：agent 动过手的每一步都得撤得回，撤回靠的就是这一列。 +ALTER TABLE agent_decisions DROP CONSTRAINT agent_decisions_target_kind_check; +ALTER TABLE agent_decisions ADD CONSTRAINT agent_decisions_target_kind_check + CHECK (target_kind IN ('review', 'fact', 'conflict')); + +ALTER TABLE agent_decisions DROP CONSTRAINT agent_decisions_action_check; +ALTER TABLE agent_decisions ADD CONSTRAINT agent_decisions_action_check + CHECK (action IN ('merge', 'keep', 'unsure', + 'confirm', 'reject', + 'close_old', 'retime_new', 'keep_both', 'reject_new')); + +ALTER TABLE agent_decisions + ADD COLUMN summary TEXT, + ADD COLUMN detail JSONB NOT NULL DEFAULT '{}'; diff --git a/web/src/api.ts b/web/src/api.ts index 498b96dc5..6b565808c 100644 --- a/web/src/api.ts +++ b/web/src/api.ts @@ -570,13 +570,26 @@ export interface ReviewItem { proposal: ReviewProposal | null; } -/** agent 的一笔（0025）：看了哪一对、想怎么办、凭什么、人怎么答的 */ +/** agent 在各档上的出路（0025 重复对；0043 事实与冲突） */ +export type AgentAction = + | "merge" + | "keep" + | "unsure" + | "confirm" + | "reject" + | "close_old" + | "retime_new" + | "keep_both" + | "reject_new"; + +/** agent 的一笔（0025 / 0043）：看了哪一项、想怎么办、凭什么、人怎么答的 */ export interface AgentDecision { id: string; run_id: string; - target_kind: "review"; + /** review = 重复对；fact = 低置信或证据过期的事实；conflict = 时态冲突 */ + target_kind: "review" | "fact" | "conflict"; target_id: string; - action: "merge" | "keep" | "unsure"; + action: AgentAction; confidence: number; reason: string | null; /** 它被给看的先例：同对 / 同名 / 撤回各一条一条，类型对的习惯是一条汇总 */ @@ -594,6 +607,8 @@ export interface AgentDecision { decided_by_name: string | null; left: string | null; right: string | null; + /** 事实与冲突（0043）：做决定那一刻这一项的样子 */ + summary: string | null; } export type AgentPrecedent = @@ -2306,7 +2321,7 @@ export const api = { agentAnswer: ( kbId: string, decisionId: string, - action: "merge" | "keep" | "revert", + action: AgentAction | "revert", rationale?: string, ) => request<{ ok: boolean }>(`/api/v1/kbs/${kbId}/review/agent/${decisionId}`, { diff --git a/web/src/i18n/en.ts b/web/src/i18n/en.ts index 4868dff05..4ea12168a 100644 --- a/web/src/i18n/en.ts +++ b/web/src/i18n/en.ts @@ -1719,7 +1719,17 @@ export const en = { agentHint: "What the agent proposed or decided for this base, from the decisions people made here before. Answering here is your decision, and it becomes precedent for the next look.", agentEmpty: "The agent has not looked at anything yet.", - agentActions: { merge: "Merge", keep: "Keep apart", unsure: "Unsure" } as Record, + agentActions: { + merge: "Merge", + keep: "Keep apart", + unsure: "Unsure", + confirm: "Confirm", + reject: "Reject", + close_old: "Close the old value", + retime_new: "Move the new start", + keep_both: "Keep both", + reject_new: "Reject the new value", + } as Record, agentStatus: { proposed: "Proposed", applied: "Applied", diff --git a/web/src/i18n/zh.ts b/web/src/i18n/zh.ts index 8fa8fd8ec..74d8ba213 100644 --- a/web/src/i18n/zh.ts +++ b/web/src/i18n/zh.ts @@ -1502,7 +1502,17 @@ export const zh: Strings = { agentHint: "agent 依据这个库里人以前的决定，提议或裁决了什么。在这里回答就是你的决定，也会成为它下一次的先例。", agentEmpty: "agent 还没看过任何一对。", - agentActions: { merge: "合并", keep: "分开", unsure: "说不准" } as Record, + agentActions: { + merge: "合并", + keep: "分开", + unsure: "说不准", + confirm: "确认", + reject: "驳回", + close_old: "关上旧值", + retime_new: "改新值的起点", + keep_both: "两个都留着", + reject_new: "驳回新值", + } as Record, agentStatus: { proposed: "建议", applied: "已裁", diff --git a/web/src/pages/Review.tsx b/web/src/pages/Review.tsx index 26bcd6ac1..8764a5f96 100644 --- a/web/src/pages/Review.tsx +++ b/web/src/pages/Review.tsx @@ -4,6 +4,7 @@ import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; import { useNavigate, useSearch } from "@tanstack/react-router"; import { api, + type AgentAction, type AgentDecision, type AgentPrecedent, type AxiomViolation, @@ -494,8 +495,31 @@ const AGENT_ACTION_TONE: Record = { merge: "violet", keep: "neutral", unsure: "warn", + confirm: "success", + reject: "danger", + close_old: "info", + retime_new: "info", + keep_both: "neutral", + reject_new: "danger", }; +/** 一笔建议能怎么答（0043）：每一档用它自己的出路。关上旧值、改新值起点要带日期， + * 只能照 agent 给的日期接受，所以只在它自己提议时出现 */ +function answersFor(d: AgentDecision): AgentAction[] { + switch (d.target_kind) { + case "fact": + return ["reject", "confirm"]; + case "conflict": + return [ + "keep_both", + "reject_new", + ...(d.action === "close_old" || d.action === "retime_new" ? [d.action] : []), + ]; + default: + return ["keep", "merge"]; + } +} + const AGENT_STATUS_TONE: Record = { proposed: "warn", applied: "info", @@ -527,7 +551,7 @@ function AgentRow({ }: { d: AgentDecision; busy: boolean; - onAnswer: (action: "merge" | "keep" | "revert", rationale?: string) => void; + onAnswer: (action: AgentAction | "revert", rationale?: string) => void; }) { const [open, setOpen] = useState(false); const [why, setWhy] = useState(""); @@ -539,7 +563,7 @@ function AgentRow({
{S.review.agentActions[d.action]} - {d.left ?? "?"} ≟ {d.right ?? "?"} + {d.target_kind === "review" ? `${d.left ?? "?"} ≟ ${d.right ?? "?"}` : (d.summary ?? "?")} {Math.round(d.confidence * 100)}% @@ -603,17 +627,23 @@ function AgentRow({ onChange={(e) => setWhy(e.target.value)} /> )} - {d.status === "proposed" && ( - <> - - - - )} - {d.status === "applied" && d.action === "merge" && ( + ))} + {d.status === "applied" && (d.action === "merge" || d.target_kind !== "review") && ( @@ -1137,7 +1167,7 @@ export function Review() { rationale, }: { id: string; - action: "merge" | "keep" | "revert"; + action: AgentAction | "revert"; rationale?: string; }) => api.agentAnswer(kb!.id, id, action, rationale), onError: (e) => toast.error((e as Error).message), From 664d3ddbf5822bd2bfc2979fb2a3b3ef5978a0a7 Mon Sep 17 00:00:00 2001 From: WaylandYang Date: Mon, 14 Sep 2026 14:56:38 +0800 Subject: [PATCH 2/3] A fact is judged over its dates Co-Authored-By: Claude Opus 5 Signed-off-by: WaylandYang --- crates/utopia-extract/src/queue_agent.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/crates/utopia-extract/src/queue_agent.rs b/crates/utopia-extract/src/queue_agent.rs index fc51908a9..7e59cae3f 100644 --- a/crates/utopia-extract/src/queue_agent.rs +++ b/crates/utopia-extract/src/queue_agent.rs @@ -141,7 +141,9 @@ pub fn fact_messages(items: &[FactQuestion]) -> Vec { \n\ Actions:\n\ - \"confirm\": the evidence states this fact — the same subject, relation, value and \ - dates. A value written in a table, a list or an amendment's new column is stated.\n\ + dates. A value written in a table, a list or an amendment's new column is stated. \ + A fact is judged over its dates: text that ends a value on a date (deleted, \ + terminated, replaced) states that the value held until then.\n\ - \"reject\": the evidence does not say it, says something else, or attaches it to \ another subject.\n\ - \"unsure\": the evidence genuinely points both ways; say what a person should check.\n\ From b02f58b58fbdd3101b9e0acaee0f13557f211310 Mon Sep 17 00:00:00 2001 From: WaylandYang Date: Mon, 14 Sep 2026 19:56:37 +0800 Subject: [PATCH 3/3] One govern run holds a base at a time Co-Authored-By: Claude Opus 5 Signed-off-by: WaylandYang --- crates/utopia-server/src/governance.rs | 28 ++++++++-- crates/utopia-store/src/governance.rs | 55 +++++++++++++++++++ .../tests/one_govern_run_at_a_time.rs | 51 +++++++++++++++++ 3 files changed, 130 insertions(+), 4 deletions(-) create mode 100644 crates/utopia-store/tests/one_govern_run_at_a_time.rs diff --git a/crates/utopia-server/src/governance.rs b/crates/utopia-server/src/governance.rs index e25683239..a3c6017e8 100644 --- a/crates/utopia-server/src/governance.rs +++ b/crates/utopia-server/src/governance.rs @@ -110,12 +110,32 @@ pub async fn govern(state: &AppState, kb_id: Uuid) -> anyhow::Result<()> { tracing::info!(%kb_id, "治理：没有配聊天模型，队列原地等"); return Ok(()); }; + // 同库已经有一个在跑：挂回队列等它跑完（不烧重试次数），别并排裁同一批 + let Some(run) = gov::claim_run(&state.pool, kb_id).await? else { + return Err( + anyhow::Error::msg("another govern run holds this base").context( + utopia_core::Deferred::new(std::time::Duration::from_secs(30)), + ), + ); + }; + let outcome = govern_held(state, kb_id, &client, &settings).await; + run.release().await; + outcome +} + +/// 拿到库锁之后的一次治理：重复对的几轮，再是其余几档 +async fn govern_held( + state: &AppState, + kb_id: Uuid, + client: &LlmClient, + settings: &Option, +) -> anyhow::Result<()> { let ctx = Ctx { state, kb_id, run_id: Uuid::now_v7(), - client: &client, - settings: &settings, + client, + settings, }; // 上一次任务半路留下的锁先放掉；跑完（不管怎么结束的）再放一次 @@ -131,8 +151,8 @@ pub async fn govern(state: &AppState, kb_id: Uuid) -> anyhow::Result<()> { state, kb_id, run_id: ctx.run_id, - client: &client, - settings: &settings, + client, + settings, }; let more = crate::queue_agent::run(&queues).await? || more; diff --git a/crates/utopia-store/src/governance.rs b/crates/utopia-store/src/governance.rs index 1d2567d98..b867c6641 100644 --- a/crates/utopia-store/src/governance.rs +++ b/crates/utopia-store/src/governance.rs @@ -926,6 +926,61 @@ pub async fn lock(pool: &PgPool, kb_id: Uuid, ids: &[Uuid]) -> AppResult<()> { Ok(()) } +/// 一个库同一时刻只跑一个治理任务的凭据。 +/// +/// 任务开头的 [`release_locks`] 假定没有别的任务在裁：两个任务并排跑，后开始的那个一上来 +/// 就把前一个正裁着的对放掉，两边裁同一批、写重复的决定（`agent_decisions_open_idx` +/// 冲突），模型调用也翻倍。抽完一篇就排一个治理任务（0043）之后，一批文档陆续抽完， +/// 同一个库能并排跑上十个。 +/// +/// 会话级咨询锁挂在一条专用连接上。正常结束走 [`RunGuard::release`] 放锁、连接回池； +/// 任务半路被丢掉（出错、取消）时 `Drop` 把连接从池里摘下来关掉，锁跟着连接一起没—— +/// 带着锁回池，这个库的治理就再也跑不起来 +pub struct RunGuard { + conn: Option>, + kb_id: Uuid, +} + +fn run_key(kb_id: Uuid) -> String { + format!("govern:{kb_id}") +} + +/// 拿这个库的治理锁；已经有任务拿着就返回 None，不等 +pub async fn claim_run(pool: &PgPool, kb_id: Uuid) -> AppResult> { + let mut conn = pool.acquire().await?; + let (held,): (bool,) = sqlx::query_as("SELECT pg_try_advisory_lock(hashtextextended($1, 0))") + .bind(run_key(kb_id)) + .fetch_one(&mut *conn) + .await?; + Ok(held.then_some(RunGuard { + conn: Some(conn), + kb_id, + })) +} + +impl RunGuard { + pub async fn release(mut self) { + let Some(mut conn) = self.conn.take() else { + return; + }; + let unlocked = sqlx::query("SELECT pg_advisory_unlock(hashtextextended($1, 0))") + .bind(run_key(self.kb_id)) + .execute(&mut *conn) + .await; + if unlocked.is_err() { + drop(conn.detach()); + } + } +} + +impl Drop for RunGuard { + fn drop(&mut self) { + if let Some(conn) = self.conn.take() { + drop(conn.detach()); + } + } +} + /// 任务开始与结束时放开所有锁：一轮中途出错、开关关掉，都不能把对锁死 pub async fn release_locks(pool: &PgPool, kb_id: Uuid) -> AppResult { let n = sqlx::query( diff --git a/crates/utopia-store/tests/one_govern_run_at_a_time.rs b/crates/utopia-store/tests/one_govern_run_at_a_time.rs new file mode 100644 index 000000000..2ce80318a --- /dev/null +++ b/crates/utopia-store/tests/one_govern_run_at_a_time.rs @@ -0,0 +1,51 @@ +//! 一个库同一时刻只跑一个治理任务（0043）。后开始的任务开头会放掉「正在裁」的标记， +//! 两个并排跑就会裁同一批、写重复的决定。锁不碰库里任何行，只要一条连接。 +//! +//! 没有 `UTOPIA_DATABASE_URL` 时跳过而不是失败（见 `utopia_store::test_db`）。 + +use sqlx::PgPool; +use utopia_store::governance::claim_run; +use uuid::Uuid; + +#[tokio::test] +async fn a_second_run_on_the_same_base_waits_and_a_dropped_run_lets_go() -> anyhow::Result<()> { + let Some(url) = utopia_store::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + let (kb, other) = (Uuid::now_v7(), Uuid::now_v7()); + + let first = claim_run(&pool, kb) + .await? + .expect("the first run takes the base"); + assert!( + claim_run(&pool, kb).await?.is_none(), + "a second run on the same base waits" + ); + let elsewhere = claim_run(&pool, other) + .await? + .expect("another base is not held"); + + first.release().await; + let again = claim_run(&pool, kb) + .await? + .expect("a released base can be taken again"); + + // 任务半路被丢掉：锁跟着连接一起走，不会带着锁回池 + drop(again); + let mut taken = None; + for _ in 0..50 { + taken = claim_run(&pool, kb).await?; + if taken.is_some() { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } + assert!(taken.is_some(), "a dropped run lets the base go"); + + elsewhere.release().await; + if let Some(t) = taken { + t.release().await; + } + Ok(()) +}