diff --git a/crates/utopia-core/src/models.rs b/crates/utopia-core/src/models.rs index 08251fef9..43efef659 100644 --- a/crates/utopia-core/src/models.rs +++ b/crates/utopia-core/src/models.rs @@ -278,6 +278,8 @@ pub enum SourceKind { Webdav, Notion, Api, + /// 推送的不是文档而是陈述本身（0054）：请求体就是开放抽取契约，抽取不问模型 + Statements, Custom, /// 每个库自带的记忆来源，不可建不可删（0015） Memory, diff --git a/crates/utopia-server/src/api/mod.rs b/crates/utopia-server/src/api/mod.rs index 01fd55a71..6a96a5094 100644 --- a/crates/utopia-server/src/api/mod.rs +++ b/crates/utopia-server/src/api/mod.rs @@ -475,6 +475,15 @@ pub fn router(state: AppState, cfg: &AppConfig) -> Router { .route("/kbs/{id}/ingest", post(sources_routes::ingest)) // api 来源推送：来源专属密钥认证（Bearer），无会话 .route("/sources/{source_id}/ingest", post(sources_routes::push)) + // 推陈述而不是推文档（0054）：请求体就是开放抽取契约，抽取不问模型 + // 请求体在验钥匙之前就已读完，所以这里压一个远小于缺省 2 MiB 的上限； + // 门口的 64 KiB 判定仍是 422，这个上限只挡明显不是一份载荷的东西 + .route( + "/sources/{source_id}/statements", + post(sources_routes::push_statements).layer(DefaultBodyLimit::max( + 4 * sources_routes::STATEMENTS_MAX_BYTES, + )), + ) .route( "/kbs/{id}/sources/{source_id}/token", get(sources_routes::get_token), diff --git a/crates/utopia-server/src/api/sources_routes.rs b/crates/utopia-server/src/api/sources_routes.rs index fd2725c58..7b0e94670 100644 --- a/crates/utopia-server/src/api/sources_routes.rs +++ b/crates/utopia-server/src/api/sources_routes.rs @@ -115,7 +115,7 @@ pub async fn create( } // api 来源：生成专属推送密钥（此后可随时经 get_token 查看） let mut ingest_token: Option = None; - if source.kind == "api" { + if has_push_token(&source) { let token = new_ingest_token(); utopia_store::sources::set_ingest_token(&state.pool, source.id, &token).await?; ingest_token = Some(token); @@ -136,7 +136,13 @@ pub async fn create( )) } -/// 查看 api 来源的推送密钥（Editor；列表响应从不携带，查看走这里）。 +/// 有推送密钥的来源：`api` 推文档，`statements` 推陈述（0054）。两种的密钥同一套 +/// 生成、查看、轮换 +fn has_push_token(source: &utopia_core::models::Source) -> bool { + matches!(source.kind.as_str(), "api" | "statements") +} + +/// 查看推送密钥（Editor；列表响应从不携带，查看走这里）。 pub async fn get_token( State(state): State, AuthUser(user): AuthUser, @@ -144,13 +150,13 @@ pub async fn get_token( ) -> ApiResult> { require_kb(&state, &user, kb_id, Role::Editor).await?; let source = source_in_kb(&state, kb_id, source_id).await?; - if source.kind != "api" { + if !has_push_token(&source) { return Err(utopia_core::AppError::NotFound.into()); } Ok(Json(json!({ "ingest_token": source.ingest_token }))) } -/// 轮换 api 来源的推送密钥：旧密钥立即失效。 +/// 轮换推送密钥：旧密钥立即失效。 pub async fn rotate_token( State(state): State, AuthUser(user): AuthUser, @@ -158,7 +164,7 @@ pub async fn rotate_token( ) -> ApiResult> { require_kb(&state, &user, kb_id, Role::Editor).await?; let source = source_in_kb(&state, kb_id, source_id).await?; - if source.kind != "api" { + if !has_push_token(&source) { return Err(utopia_core::AppError::NotFound.into()); } let token = new_ingest_token(); @@ -548,6 +554,232 @@ pub async fn push( } } +/// 推陈述（0054）：请求体就是开放抽取契约（`e` / `s` / `n`），外加 `api` 推送的那层信封。 +/// +/// 和 `push` 同一把钥匙、同一套身份语义（`statements:{external_id}`）、同一份 run 记录； +/// 不同的只有两点，都在门口定死：**载荷照原样成为文档，一整块**，抽取时按契约解析而不问 +/// 模型；**信封与契约之外的任何键都拒收**——契约里本来就没有属性、类或谓词的格子， +/// 一个写了 `predicate` 的调用方应当在这里得到 422，而不是在图里找不到它以为写进去的类型事实。 +pub async fn push_statements( + State(state): State, + Path(source_id): Path, + headers: HeaderMap, + bytes: axum::body::Bytes, +) -> ApiResult> { + let source = utopia_store::sources::get(&state.pool, source_id).await?; + if source.kind != "statements" { + return Err(utopia_core::AppError::NotFound.into()); + } + let token = headers + .get("authorization") + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.strip_prefix("Bearer ")) + .map(str::trim) + .filter(|s| !s.is_empty()) + .ok_or(utopia_core::AppError::Unauthorized)?; + if !push_key_matches(source.ingest_token.as_deref(), token) { + return Err(utopia_core::AppError::Unauthorized.into()); + } + + let run = utopia_store::sources::start_run(&state.pool, source.id).await?; + match handle_statements_push(&state, &source, &bytes).await { + Ok(action) => { + let (created, updated) = match action { + crate::ingest_sources::IngestAction::Created => (1, 0), + crate::ingest_sources::IngestAction::Updated + | crate::ingest_sources::IngestAction::Moved => (0, 1), + crate::ingest_sources::IngestAction::Unchanged + | crate::ingest_sources::IngestAction::Tombstoned => (0, 0), + }; + utopia_store::sources::finish_run(&state.pool, run, source.id, None, created, updated) + .await?; + utopia_store::sources::finish_sync(&state.pool, source.id, None, created).await?; + state.emit_source(source.kb_id); + Ok(Json(json!({ "action": action_str(action) }))) + } + Err(err) => { + let msg = err.message().to_string(); + utopia_store::sources::finish_run(&state.pool, run, source.id, Some(&msg), 0, 0) + .await?; + if let PushError::Failed(_) = err { + utopia_store::sources::finish_sync(&state.pool, source.id, Some(&msg), 0).await?; + } + state.emit_source(source.kb_id); + Err(utopia_core::AppError::Validation(msg).into()) + } + } +} + +/// 一次推送最多多少字节、多少条陈述。第一刀的上限，不是契约：一块就是一份载荷， +/// 这两个数只是不让一份载荷大到审核卡片和证据视图没法呈现 +pub(crate) const STATEMENTS_MAX_BYTES: usize = 64 * 1024; +const STATEMENTS_MAX_ITEMS: usize = 200; +/// 信封里允许的键。契约的三个数组之外，只有 `api` 推送也有的那三个 +const STATEMENTS_ENVELOPE: [&str; 6] = ["external_id", "doc_time", "deleted", "e", "s", "n"]; + +#[derive(serde::Deserialize)] +struct StatementsBody { + external_id: String, + #[serde(default)] + doc_time: Option>, + #[serde(default)] + deleted: bool, + #[serde(default)] + e: serde_json::Value, + #[serde(default)] + s: serde_json::Value, + #[serde(default)] + n: serde_json::Value, +} + +/// 门口的校验：形状对不对、有没有契约之外的键、每条陈述的引文格是不是空的。 +/// 通过就把 `{e, s, n}` 按契约重新序列化成文档正文——存的是我们自己写出来的那份， +/// 不是调用方发来的字节，于是文档里没有信封、没有多余空白，块就是契约本身 +fn validate_statements_payload(raw: &[u8]) -> Result<(StatementsBody, Option), String> { + if raw.len() > STATEMENTS_MAX_BYTES { + return Err(format!( + "payload is {} bytes; the limit is {STATEMENTS_MAX_BYTES}", + raw.len() + )); + } + let value: serde_json::Value = + serde_json::from_slice(raw).map_err(|e| format!("Invalid JSON payload: {e}"))?; + let Some(map) = value.as_object() else { + return Err("the payload must be a JSON object".into()); + }; + if let Some(extra) = map + .keys() + .find(|k| !STATEMENTS_ENVELOPE.contains(&k.as_str())) + { + return Err(format!( + "unknown key {extra:?}: the contract has no slot for it (allowed: external_id, doc_time, deleted, e, s, n)" + )); + } + let body: StatementsBody = + serde_json::from_value(value).map_err(|e| format!("Invalid payload: {e}"))?; + if body.external_id.trim().is_empty() { + return Err("external_id is required".into()); + } + if body.deleted { + return Ok((body, None)); + } + let arrays = [("e", &body.e), ("s", &body.s), ("n", &body.n)]; + for (key, v) in arrays { + if !v.is_array() { + return Err(format!("{key} must be an array")); + } + } + let statements = body.s.as_array().expect("checked above"); + if statements.is_empty() { + return Err("s must hold at least one statement".into()); + } + if statements.len() > STATEMENTS_MAX_ITEMS { + return Err(format!( + "{} statements; the limit is {STATEMENTS_MAX_ITEMS} per push", + statements.len() + )); + } + for (i, item) in statements.iter().enumerate() { + let Some(arr) = item.as_array() else { + return Err(format!("s[{i}] must be an array of eight slots")); + }; + if arr.len() != 8 { + return Err(format!( + "s[{i}] has {} slots; the contract has eight", + arr.len() + )); + } + if !arr[0].is_null() { + return Err(format!( + "s[{i}][0] (quote) must be null: the item is its own evidence" + )); + } + if !arr[5].is_null() && !arr[5].is_object() { + return Err(format!("s[{i}][5] (qualifiers) must be an object or null")); + } + } + for (i, item) in body.n.as_array().expect("checked above").iter().enumerate() { + let Some(arr) = item.as_array() else { + return Err(format!("n[{i}] must be an array")); + }; + if arr.len() > 2 && !arr[2].is_null() { + return Err(format!("n[{i}][2] (quote) must be null")); + } + } + // 只有契约能通过 parse_open_response：这一步在门口跑一遍，抽取时不会再有别的答案 + let content = serde_json::to_string_pretty(&json!({ "e": body.e, "s": body.s, "n": body.n })) + .map_err(|e| format!("cannot serialise the contract: {e}"))?; + let parsed = utopia_extract::open::parse_open_response(&content) + .map_err(|e| format!("the contract does not parse: {e}"))?; + if parsed.skipped > 0 { + return Err(format!( + "{} item(s) are malformed for the contract (e: [name, kind, named]; s: [null, subject, phrase, object, value, qualifiers, when, ended]; n: [entity, name, null])", + parsed.skipped + )); + } + if parsed.statements.is_empty() { + return Err("no statement survived parsing".into()); + } + // 主语和别名所属的东西必须在 `e` 里（0054 决定 5）：抽取时找不到的会作为 UNKNOWN_REF + // 静默丢掉，而门口的职责就是不让调用方以为它写进去了。折叠规则与抽取的 `name_key` + // 同一条：只许空白和大小写不同。宾语不在此列——它落成字面值，陈述照落 + let listed: std::collections::HashSet = parsed + .entities + .iter() + .map(|e| crate::extraction_open::name_key(&e.name)) + .collect(); + for (i, st) in parsed.statements.iter().enumerate() { + if !listed.contains(&crate::extraction_open::name_key(&st.subject)) { + return Err(format!( + "s[{i}][1] (subject) {:?} is not a thing listed in e", + st.subject + )); + } + } + for (i, n) in parsed.names.iter().enumerate() { + if !listed.contains(&crate::extraction_open::name_key(&n.entity)) { + return Err(format!( + "n[{i}][0] (entity) {:?} is not a thing listed in e", + n.entity + )); + } + } + Ok((body, Some(content))) +} + +async fn handle_statements_push( + state: &AppState, + source: &utopia_core::models::Source, + bytes: &[u8], +) -> Result { + let (body, content) = validate_statements_payload(bytes).map_err(PushError::Rejected)?; + let identity = body.external_id.trim().to_string(); + let key = format!("statements:{identity}"); + let Some(content) = content else { + utopia_store::documents::mark_missing_keys(&state.pool, source.id, &[key]) + .await + .map_err(|e| PushError::Failed(e.to_string()))?; + return Ok(crate::ingest_sources::IngestAction::Tombstoned); + }; + let filename = format!("{identity}.json"); + let action = crate::ingest_sources::ingest_item( + state, + source.kb_id, + source.id, + &key, + &filename, + "application/json", + content.as_bytes(), + body.doc_time, + ) + .await + .map_err(|e| PushError::Failed(e.to_string()))?; + utopia_store::documents::clear_missing_keys(&state.pool, source.id, &[key]) + .await + .map_err(|e| PushError::Failed(e.to_string()))?; + Ok(action) +} + /// 来源级全量重抽（增量语义）：该来源下所有 ready 文档重新过一遍抽取。 /// 走正常管道——实体消解、事实去重、时态冲突照常，既有人工决策全部保留。 pub async fn re_extract( @@ -588,6 +820,10 @@ pub async fn re_extract( Ok(Json(json!({ "queued": ids.len() }))) } +#[cfg(test)] +#[path = "sources_statements_tests.rs"] +mod statements_tests; + #[cfg(test)] mod tests { use super::{keep_secrets, validate_rss_config}; diff --git a/crates/utopia-server/src/api/sources_statements_tests.rs b/crates/utopia-server/src/api/sources_statements_tests.rs new file mode 100644 index 000000000..3c6cd29ad --- /dev/null +++ b/crates/utopia-server/src/api/sources_statements_tests.rs @@ -0,0 +1,531 @@ +//! 推陈述的来源（0054）：请求体就是开放抽取契约，门口拒绝契约之外的键，通过的载荷 +//! 整份成一块，抽取按契约解析而**不问模型**——夹具故意不配对话模型，证明这条路不需要它。 +//! 连库的部分没有 `UTOPIA_DATABASE_URL` 就跳过（同 documents_routes_tests）。 + +use axum::body::{to_bytes, Body}; +use axum::http::{Request, StatusCode}; +use serde_json::{json, Value}; +use std::sync::Arc; +use tower::ServiceExt; +use utopia_core::models::Proposer; +use utopia_store::documents; +use uuid::Uuid; + +/// 门口的校验不连库，任何环境都跑 +#[test] +fn the_door_refuses_what_the_contract_has_no_slot_for() { + let ok = json!({ + "external_id": "obs-1", + "e": [["cup-7", "cup", true], ["kitchen table", "table", true]], + "s": [[null, "cup-7", "is on", "kitchen table", null, {}, "08:14:03", null]], + "n": [] + }); + let (_, content) = super::validate_statements_payload(ok.to_string().as_bytes()) + .expect("a well-formed contract passes"); + let content = content.expect("a non-tombstone yields the document text"); + // 存的是我们重新序列化的那份：没有信封，能被抽取用的同一个解析器读回 + assert!(!content.contains("external_id")); + let parsed = utopia_extract::open::parse_open_response(&content).unwrap(); + assert_eq!(parsed.statements.len(), 1); + assert_eq!(parsed.statements[0].phrase, "is on"); + + let refuse = |body: Value, needle: &str| { + let err = super::validate_statements_payload(body.to_string().as_bytes()) + .err() + .unwrap_or_else(|| panic!("{body} must be refused")); + assert!(err.contains(needle), "{err:?} should mention {needle:?}"); + }; + // 契约里没有属性的格子：一个 `predicate` 键在门口就拦下，而不是静默忽略 + let mut typed = ok.clone(); + typed["predicate"] = json!("located_in"); + refuse(typed, "unknown key"); + // 引文格必须为空：条目自己就是证据 + let mut quoted = ok.clone(); + quoted["s"][0][0] = json!("cup-7 is on the kitchen table"); + refuse(quoted, "quote"); + // 八格少一格不是截断，是形状错 + let mut short = ok.clone(); + short["s"][0] = json!([null, "cup-7", "is on", "kitchen table"]); + refuse(short, "eight"); + // 没有身份就没有更新语义 + let mut anon = ok.clone(); + anon["external_id"] = json!(" "); + refuse(anon, "external_id"); + // 空陈述数组：什么都推不进图，直说 + let mut empty = ok.clone(); + empty["s"] = json!([]); + refuse(empty, "at least one"); + // 主语没在 `e` 里：抽取会把它作为 UNKNOWN_REF 静默丢掉，门口就得说不 + let mut stray = ok.clone(); + stray["s"][0][1] = json!("cup-8"); + refuse(stray, "not a thing listed in e"); + // 只差空白和大小写的算同一个名字（与抽取的 `name_key` 同一条折叠规则） + let mut folded = ok.clone(); + folded["s"][0][1] = json!(" Cup-7 "); + super::validate_statements_payload(folded.to_string().as_bytes()) + .expect("whitespace and case do not make a different thing"); + // 别名所属的东西也一样；别名的引文格同样必须为空 + let mut alias = ok.clone(); + alias["n"] = json!([["mug-7", "the cup", null]]); + refuse(alias, "n[0][0]"); + let mut alias_quoted = ok.clone(); + alias_quoted["n"] = json!([["cup-7", "the cup", "the cup sat there"]]); + refuse(alias_quoted, "n[0][2]"); + // 条数和字节数的上限：第一刀的限制，超过直说而不是截断 + let mut many = ok.clone(); + many["s"] = json!(vec![ok["s"][0].clone(); 201]); + refuse(many, "limit is 200"); + let mut fat = ok.clone(); + fat["s"][0][5] = json!({ "note": "x".repeat(64 * 1024) }); + refuse(fat, "limit is 65536"); +} + +struct Fixture { + pool: sqlx::PgPool, + state: crate::state::AppState, + app: axum::Router, + org: Uuid, + kb: Uuid, + source: Uuid, + api_source: Uuid, + token: String, + api_token: String, + /// 编辑者的会话令牌：查看和轮换密钥走会话认证，不走推送密钥 + session: String, + _dir: tempfile::TempDir, +} + +impl Fixture { + async fn new() -> anyhow::Result> { + let Some(url) = utopia_store::test_db::url() else { + return Ok(None); + }; + let pool = sqlx::PgPool::connect(&url).await?; + utopia_store::db::migrate(&pool).await?; + let (org, ws, kb, user, source, api_source) = ( + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + ); + // Only locally generated UUIDs are interpolated into fixture SQL. + sqlx::raw_sql(&format!( + "INSERT INTO organizations(id,name) VALUES ('{org}','statements-push-test'); + INSERT INTO workspaces(id,org_id,name) VALUES ('{ws}','{org}','statements-push-test'); + INSERT INTO users(id,org_id,email,display_name,password_hash) + VALUES ('{user}','{org}','{user}@statements.test','statements-test','unused'); + INSERT INTO knowledge_bases(id,workspace_id,name) VALUES ('{kb}','{ws}','observations'); + INSERT INTO kb_members(kb_id,user_id,role) VALUES ('{kb}','{user}','editor'); + INSERT INTO sources(id,kb_id,kind,name) VALUES + ('{source}','{kb}','statements','robot-1'), + ('{api_source}','{kb}','api','api-source');" + )) + .execute(&pool) + .await?; + // 故意**不**配对话模型：这条路不需要它 + let token = super::new_ingest_token(); + utopia_store::sources::set_ingest_token(&pool, source, &token).await?; + let api_token = super::new_ingest_token(); + utopia_store::sources::set_ingest_token(&pool, api_source, &api_token).await?; + let dir = tempfile::tempdir()?; + let cfg = utopia_core::config::AppConfig { + data_dir: dir.path().to_string_lossy().into_owned(), + ..Default::default() + }; + let search = Arc::new(utopia_search::SearchIndex::open( + &dir.path().join("search"), + )?); + let state = crate::state::AppState::new(pool.clone(), &cfg, search, "test-only".into()); + let session = crate::auth::issue_token(&state, user)?; + let app = super::super::router(state.clone(), &cfg); + Ok(Some(Self { + pool, + state, + app, + org, + kb, + source, + api_source, + token, + api_token, + session, + _dir: dir, + })) + } + + async fn push( + &self, + source: Uuid, + token: &str, + body: &Value, + ) -> anyhow::Result<(StatusCode, Value)> { + self.push_raw(source, Some(token), body.to_string().into_bytes()) + .await + } + + /// 不带 Authorization 头（`None`）或推原始字节：门口的 401 和字节上限要从 HTTP 这一侧看 + async fn push_raw( + &self, + source: Uuid, + token: Option<&str>, + body: Vec, + ) -> anyhow::Result<(StatusCode, Value)> { + let mut request = Request::post(format!("/api/v1/sources/{source}/statements")) + .header("Content-Type", "application/json"); + if let Some(token) = token { + request = request.header("Authorization", format!("Bearer {token}")); + } + let response = self + .app + .clone() + .oneshot(request.body(Body::from(body))?) + .await?; + let status = response.status(); + let bytes = to_bytes(response.into_body(), 1 << 20).await?; + let value: Value = serde_json::from_slice(&bytes).unwrap_or(Value::Null); + Ok((status, value)) + } + + /// 以编辑者会话调一个来源接口（查看 / 轮换密钥） + async fn as_editor(&self, method: &str, path: &str) -> anyhow::Result<(StatusCode, Value)> { + let response = self + .app + .clone() + .oneshot( + Request::builder() + .method(method) + .uri(format!("/api/v1/kbs/{}/sources/{path}", self.kb)) + .header("Authorization", format!("Bearer {}", self.session)) + .body(Body::empty())?, + ) + .await?; + let status = response.status(); + let bytes = to_bytes(response.into_body(), 1 << 20).await?; + let value: Value = serde_json::from_slice(&bytes).unwrap_or(Value::Null); + Ok((status, value)) + } + + async fn cleanup(self) -> anyhow::Result<()> { + sqlx::query("DELETE FROM organizations WHERE id = $1") + .bind(self.org) + .execute(&self.pool) + .await?; + self.pool.close().await; + Ok(()) + } +} + +fn observation(when: &str, place: &str) -> Value { + json!({ + "external_id": "obs-000412", + "doc_time": "2026-09-23T08:14:03Z", + "e": [["cup-7", "cup", true], [place, "table", true]], + "s": [[null, "cup-7", "is on", place, null, {}, when, null]], + "n": [] + }) +} + +/// 推一条陈述，走完处理与抽取，它就是一条开放陈述：有短语、有主宾实体、有证据行 +/// （块 = 载荷，偏移为空），而工作区没有任何对话模型 +#[tokio::test] +async fn a_pushed_statement_reaches_the_open_graph_without_a_model() -> anyhow::Result<()> { + let Some(f) = Fixture::new().await? else { + return Ok(()); + }; + let (status, body) = f + .push( + f.source, + &f.token, + &observation("08:14:03", "kitchen table"), + ) + .await?; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["action"], "created"); + + let doc = documents::find_by_external_key(&f.pool, f.source, "statements:obs-000412") + .await? + .expect("the push created a document under its identity"); + assert_eq!(doc.mime, "application/json"); + + crate::pipeline::process_document(&f.state, doc.id).await?; + let chunks: Vec<(String,)> = + sqlx::query_as("SELECT text FROM chunks WHERE document_id = $1 AND superseded_at IS NULL") + .bind(doc.id) + .fetch_all(&f.pool) + .await?; + assert_eq!( + chunks.len(), + 1, + "the payload is one chunk, not a budgeted split" + ); + utopia_extract::open::parse_open_response(&chunks[0].0) + .expect("the chunk is the contract verbatim"); + + crate::extraction::extract_document( + &f.state, + doc.id, + Proposer { + user_id: None, + token_id: None, + }, + ) + .await?; + let (status,): (String,) = sqlx::query_as("SELECT graph_status FROM documents WHERE id = $1") + .bind(doc.id) + .fetch_one(&f.pool) + .await?; + assert_ne!(status, "failed", "extraction must not need a chat model"); + + let facts: Vec<(Uuid, String)> = sqlx::query_as( + "SELECT id, phrase FROM facts + WHERE kb_id = $1 AND layer = 'open' AND invalidated_at IS NULL", + ) + .bind(f.kb) + .fetch_all(&f.pool) + .await?; + assert_eq!(facts.len(), 1, "{facts:?}"); + assert_eq!(facts[0].1, "is on"); + let (chunk_matches, quote_null, offsets_null): (bool, bool, bool) = sqlx::query_as( + "SELECT chunk_id = (SELECT id FROM chunks WHERE document_id = $2 AND superseded_at IS NULL), + quote IS NULL, quote_start IS NULL AND quote_end IS NULL + FROM fact_evidence WHERE fact_id = $1", + ) + .bind(facts[0].0) + .bind(doc.id) + .fetch_one(&f.pool) + .await?; + assert!(chunk_matches, "the evidence is the payload's own chunk"); + assert!( + quote_null && offsets_null, + "the item is its own evidence: no quote, no offsets" + ); + let (entities,): (i64,) = sqlx::query_as( + "SELECT count(*) FROM entities + WHERE kb_id = $1 AND canonical_name IN ('cup-7', 'kitchen table')", + ) + .bind(f.kb) + .fetch_one(&f.pool) + .await?; + assert_eq!( + entities, 2, + "both things are entities with the pushed names" + ); + // 时间词落成提及（0054 决定 3）：没有引文时在载荷自己里找，而不是走引文路退出 + let mentions: Vec<(String, String)> = + sqlx::query_as("SELECT text, role FROM time_mentions WHERE fact_id = $1") + .bind(facts[0].0) + .fetch_all(&f.pool) + .await?; + assert_eq!( + mentions, + vec![("08:14:03".to_string(), "when".to_string())], + "the pushed `when` is a time mention on the fact" + ); + f.cleanup().await +} + +/// 同一身份再推一份新内容是更新：原地替换并记版本，和 `api` 推送一个语义 +#[tokio::test] +async fn a_second_push_under_the_same_identity_updates_in_place() -> anyhow::Result<()> { + let Some(f) = Fixture::new().await? else { + return Ok(()); + }; + let (status, body) = f + .push( + f.source, + &f.token, + &observation("08:14:03", "kitchen table"), + ) + .await?; + assert_eq!(status, StatusCode::OK, "{body}"); + let (status, body) = f + .push( + f.source, + &f.token, + &observation("08:14:03", "kitchen table"), + ) + .await?; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["action"], "unchanged", "same content is a no-op"); + let (status, body) = f + .push(f.source, &f.token, &observation("08:20:00", "counter")) + .await?; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["action"], "updated"); + let doc = documents::find_by_external_key(&f.pool, f.source, "statements:obs-000412") + .await? + .expect("still one document under the identity"); + let (versions,): (i64,) = + sqlx::query_as("SELECT count(*) FROM document_versions WHERE document_id = $1") + .bind(doc.id) + .fetch_one(&f.pool) + .await?; + assert_eq!(versions, 2, "the update recorded a version"); + f.cleanup().await +} + +/// 门口的拒绝走到 HTTP 是 422（`AppError::Validation`，与 `api` 推送被拒时同一个码）， +/// 并且这次推送留在 run 历史里；`api` 来源不认这条路由（404），钥匙不对是 401 +#[tokio::test] +async fn the_route_answers_422_404_and_401_at_the_door() -> anyhow::Result<()> { + let Some(f) = Fixture::new().await? else { + return Ok(()); + }; + let mut typed = observation("08:14:03", "kitchen table"); + typed["class"] = json!("Cup"); + let (status, body) = f.push(f.source, &f.token, &typed).await?; + assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{body}"); + let (status, _) = f + .push( + f.api_source, + &f.api_token, + &observation("08:14:03", "kitchen table"), + ) + .await?; + assert_eq!( + status, + StatusCode::NOT_FOUND, + "an api source has no statements route" + ); + let (status, _) = f + .push( + f.source, + "utp_not-the-key", + &observation("08:14:03", "kitchen table"), + ) + .await?; + assert_eq!(status, StatusCode::UNAUTHORIZED); + let (status, _) = f + .push_raw( + f.source, + None, + observation("08:14:03", "kitchen table") + .to_string() + .into_bytes(), + ) + .await?; + assert_eq!(status, StatusCode::UNAUTHORIZED, "no header is no key"); + // 64 KiB 上限从 HTTP 这一侧看仍是 422 带说明，不是路由层的 413 + let mut fat = observation("08:14:03", "kitchen table"); + fat["s"][0][5] = json!({ "note": "x".repeat(64 * 1024) }); + let (status, body) = f.push(f.source, &f.token, &fat).await?; + assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{body}"); + assert!( + body.to_string().contains("limit is 65536"), + "the refusal names the limit: {body}" + ); + f.cleanup().await +} + +/// 墓碑与复活，和 `api` 推送同一语义：`deleted: true` 给该身份打 "Not in source" 标记而不删； +/// 同一身份再推内容就把标记清掉。没见过的身份打墓碑是空操作，照样回 marked_missing +#[tokio::test] +async fn a_tombstone_marks_the_item_missing_and_a_new_push_revives_it() -> anyhow::Result<()> { + let Some(f) = Fixture::new().await? else { + return Ok(()); + }; + let (status, body) = f + .push( + f.source, + &f.token, + &observation("08:14:03", "kitchen table"), + ) + .await?; + assert_eq!(status, StatusCode::OK, "{body}"); + let (status, body) = f + .push( + f.source, + &f.token, + &json!({ "external_id": "obs-000412", "deleted": true }), + ) + .await?; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["action"], "marked_missing"); + let doc = documents::find_by_external_key(&f.pool, f.source, "statements:obs-000412") + .await? + .expect("a tombstone marks, it does not delete"); + let missing: (bool, bool) = sqlx::query_as( + "SELECT missing_since IS NOT NULL, deleted_at IS NULL FROM documents WHERE id = $1", + ) + .bind(doc.id) + .fetch_one(&f.pool) + .await?; + assert_eq!(missing, (true, true), "marked missing, still present"); + // 复活：同一身份、同样内容——文档没变（unchanged），但标记清掉了 + let (status, body) = f + .push( + f.source, + &f.token, + &observation("08:14:03", "kitchen table"), + ) + .await?; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["action"], "unchanged"); + let (revived,): (bool,) = + sqlx::query_as("SELECT missing_since IS NULL FROM documents WHERE id = $1") + .bind(doc.id) + .fetch_one(&f.pool) + .await?; + assert!(revived, "a new push under the identity clears the marker"); + // 没见过的身份：打不到任何文档，也不算错 + let (status, body) = f + .push( + f.source, + &f.token, + &json!({ "external_id": "never-pushed", "deleted": true }), + ) + .await?; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["action"], "marked_missing"); + assert!( + documents::find_by_external_key(&f.pool, f.source, "statements:never-pushed") + .await? + .is_none(), + "a tombstone never creates a document" + ); + f.cleanup().await +} + +/// 密钥的查看和轮换对 `statements` 来源和 `api` 来源一样可用：创建时给过一次的密钥 +/// 之后还查得到，轮换后旧密钥立刻失效、新密钥能推。端到端跑服务时抓到的：这两个接口 +/// 原来只认 `api` +#[tokio::test] +async fn the_push_token_can_be_viewed_and_rotated_like_an_api_source() -> anyhow::Result<()> { + let Some(f) = Fixture::new().await? else { + return Ok(()); + }; + let (status, body) = f.as_editor("GET", &format!("{}/token", f.source)).await?; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!( + body["ingest_token"], f.token, + "the token given at creation is viewable" + ); + let (status, body) = f + .as_editor("POST", &format!("{}/rotate-token", f.source)) + .await?; + assert_eq!(status, StatusCode::OK, "{body}"); + let rotated = body["ingest_token"] + .as_str() + .expect("a new token") + .to_string(); + assert_ne!(rotated, f.token); + let (status, _) = f + .push( + f.source, + &f.token, + &observation("08:14:03", "kitchen table"), + ) + .await?; + assert_eq!(status, StatusCode::UNAUTHORIZED, "the old token is dead"); + let (status, body) = f + .push( + f.source, + &rotated, + &observation("08:14:03", "kitchen table"), + ) + .await?; + assert_eq!(status, StatusCode::OK, "{body}"); + f.cleanup().await +} diff --git a/crates/utopia-server/src/extraction.rs b/crates/utopia-server/src/extraction.rs index e111adbda..fcd234983 100644 --- a/crates/utopia-server/src/extraction.rs +++ b/crates/utopia-server/src/extraction.rs @@ -341,11 +341,24 @@ async fn run(state: &AppState, document_id: Uuid, proposer: Proposer) -> anyhow: return Ok(()); } let kb = utopia_store::kbs::get(&state.pool, doc.kb_id).await?; - let settings = utopia_store::settings::get(&state.pool, kb.workspace_id) + // 推送来的陈述（0054）：块就是契约，抽取按契约解析、不问模型，没配对话模型也照抽 + let pushed = crate::pipeline::source_kind(state, doc.source_id) .await? - .ok_or_else(|| anyhow::anyhow!("Chat model not configured; cannot extract"))?; - let client = llm_util::chat_client(&settings) - .ok_or_else(|| anyhow::anyhow!("Chat model not configured; cannot extract"))?; + .as_deref() + == Some("statements"); + // settings 有就传：推送路径不问对话模型，但名字向量的嵌入模型（#877）仍从它来 + let settings = utopia_store::settings::get(&state.pool, kb.workspace_id).await?; + let client = if pushed { + None + } else { + let settings = settings + .as_ref() + .ok_or_else(|| anyhow::anyhow!("Chat model not configured; cannot extract"))?; + Some( + llm_util::chat_client(settings) + .ok_or_else(|| anyhow::anyhow!("Chat model not configured; cannot extract"))?, + ) + }; // 所有权凭证：重抽会自增 epoch，任务据此察觉自己已被接管（见 `run_open` 的分块循环） let my_epoch = utopia_store::documents::extract_epoch(&state.pool, document_id).await?; @@ -353,7 +366,15 @@ async fn run(state: &AppState, document_id: Uuid, proposer: Proposer) -> anyhow: state.emit_document(doc.kb_id, document_id); let await_nod = utopia_store::memory::is_memory_document(&state.pool, document_id).await?; crate::extraction_open::run_open( - state, &doc, &kb, &settings, &client, my_epoch, proposer, await_nod, + state, + &doc, + &kb, + settings.as_ref(), + client.as_ref(), + my_epoch, + proposer, + await_nod, + pushed, ) .await } diff --git a/crates/utopia-server/src/extraction_open.rs b/crates/utopia-server/src/extraction_open.rs index 39988677a..c70157d9e 100644 --- a/crates/utopia-server/src/extraction_open.rs +++ b/crates/utopia-server/src/extraction_open.rs @@ -174,7 +174,7 @@ async fn embed_pending_names( Ok(items.len()) } -fn name_key(name: &str) -> String { +pub(crate) fn name_key(name: &str) -> String { name.split_whitespace() .collect::>() .join(" ") @@ -222,17 +222,20 @@ async fn place( } /// `await_nod`：这是记忆日志（0015）——陈述不直接落库，原样进待确认表，人点头时才成为开放陈述。 -/// `proposer`：那句话是谁、经哪枚令牌说的（0026），随待确认项一起记 +/// `proposer`：那句话是谁、经哪枚令牌说的（0026），随待确认项一起记。 +/// `pushed`：块本身就是契约（0054 的 `statements` 来源）——不建提示词、不问模型，直接解析； +/// 这时 `client`（对话模型）为 None；`settings` 有就照传，名字向量的嵌入模型从它来。其余一步不变 #[allow(clippy::too_many_arguments)] pub(crate) async fn run_open( state: &AppState, doc: &Document, kb: &KnowledgeBase, - settings: &LlmSettings, - client: &utopia_llm::LlmClient, + settings: Option<&LlmSettings>, + client: Option<&utopia_llm::LlmClient>, my_epoch: i32, proposer: Proposer, await_nod: bool, + pushed: bool, ) -> anyhow::Result<()> { let pool = &state.pool; let document_id = doc.id; @@ -267,7 +270,7 @@ pub(crate) async fn run_open( // 名字向量的嵌入客户端（0041 决定 3 通道 2）。没配嵌入模型就是 None：召回退回 // 字面相等，抽取照常 - let embed = crate::llm_util::embed_client(settings); + let embed = settings.and_then(crate::llm_util::embed_client); for chunk in chunks.iter() { // 被接管则安静退场（重抽自增 epoch）：检查放在调用模型之前 if utopia_store::documents::extract_epoch(pool, document_id).await? != my_epoch { @@ -288,39 +291,53 @@ pub(crate) async fn run_open( .as_ref() .filter(|(id, _)| *id != chunk.id) .map(|(_, text)| text.as_str()); - let messages = - utopia_extract::open::build_open_messages(&doc.filename, &known, opening, &chunk.text); - // 温度 0：照抄原文的活不该靠采样。端点缺省 1.0 时同一块两次回复密度差三倍 - let reply = match chat_retrying_rate_limits_at( - state, - settings, - client, - &messages, - Some(0.0), - ) - .await - { - Ok(r) => r, - Err(e) => { - tracing::warn!(%document_id, seq = chunk.seq, error = %e, "开放抽取调用失败，跳过该分块"); - drop_signal( - state, - kb_id, - document_id, - reason::CHUNK_UNEXTRACTED, - "调用失败，这一块没有进图", - Some(&format!("#{}：{e}", chunk.seq)), - ) - .await; - unextracted.push((chunk.seq, format!("调用失败：{e}"))); - continue; - } + // 推送来的陈述：块就是契约，解析它而不是问模型（0054）。下面从解析起一步不变 + let (reply_text, cut_by_ceiling) = if pushed { + (chunk.text.clone(), false) + } else { + let (settings, client) = match (settings, client) { + (Some(s), Some(c)) => (s, c), + _ => anyhow::bail!("Chat model not configured; cannot extract"), + }; + let messages = utopia_extract::open::build_open_messages( + &doc.filename, + &known, + opening, + &chunk.text, + ); + // 温度 0：照抄原文的活不该靠采样。端点缺省 1.0 时同一块两次回复密度差三倍 + let reply = match chat_retrying_rate_limits_at( + state, + settings, + client, + &messages, + Some(0.0), + ) + .await + { + Ok(r) => r, + Err(e) => { + tracing::warn!(%document_id, seq = chunk.seq, error = %e, "开放抽取调用失败，跳过该分块"); + drop_signal( + state, + kb_id, + document_id, + reason::CHUNK_UNEXTRACTED, + "调用失败，这一块没有进图", + Some(&format!("#{}：{e}", chunk.seq)), + ) + .await; + unextracted.push((chunk.seq, format!("调用失败：{e}"))); + continue; + } + }; + tracing::debug!(%document_id, seq = chunk.seq, reply = %reply.text, "开放抽取的原始回复"); + // 端点说它是撞上 token 上限停的。解析器只看得见 JSON 少了尾巴，看不见 + // 少的原因，所以这句话得从回复里带过来（#760） + let cut_by_ceiling = reply.hit_token_ceiling(); + (reply.text, cut_by_ceiling) }; - tracing::debug!(%document_id, seq = chunk.seq, reply = %reply.text, "开放抽取的原始回复"); - // 端点说它是撞上 token 上限停的。解析器只看得见 JSON 少了尾巴，看不见 - // 少的原因，所以这句话得从回复里带过来（#760） - let cut_by_ceiling = reply.hit_token_ceiling(); - let extraction = match utopia_extract::open::parse_open_response(&reply.text) { + let extraction = match utopia_extract::open::parse_open_response(&reply_text) { Ok(x) => x, Err(e) => { tracing::warn!(%document_id, seq = chunk.seq, error = %e, hit_token_ceiling = cut_by_ceiling, "开放抽取回复解析失败，跳过该分块"); @@ -383,8 +400,8 @@ pub(crate) async fn run_open( // 名字向量（0041 决定 3 通道 2）：这一块里有名字的东西，名字字符串各算一条， // 消解时拿它在同库的名字向量里找近邻。一块一批；算不出来（端点抖了）不拦抽取， // 只是这一块少一条召回通道 - let name_vecs: HashMap> = match &embed { - Some(client) => { + let name_vecs: HashMap> = match (settings, &embed) { + (Some(settings), Some(client)) => { let mut wanted: Vec<(String, String)> = Vec::new(); let mut seen: HashSet = HashSet::new(); for e in &extraction.entities { @@ -395,7 +412,7 @@ pub(crate) async fn run_open( } embed_names(state, settings, client, &wanted).await } - None => HashMap::new(), + _ => HashMap::new(), }; // ---- 东西：有名字的走身份消解，被描述的建成没有名字事实的实体 ---- @@ -658,7 +675,14 @@ pub(crate) async fn run_open( .filter_map(|(role, words)| Some((role, words?.trim()))) .filter(|(_, w)| !w.is_empty()) { - match locate_time(&chunk.text, quote, words) { + // 推送来的陈述没有引文：条目自己就是证据（0054 决定 4），这一块就是这一份 + // 载荷，时间词在块里找。走 `locate_time` 会在 `quote?` 上退出，起止就都丢了 + let located = if pushed { + locate(&chunk.text, words).map(|(start, _)| start) + } else { + locate_time(&chunk.text, quote, words) + }; + match located { Some(start) => time_words.push((words, start, role)), None => { drop_signal( @@ -888,7 +912,7 @@ pub(crate) async fn run_open( // 名字向量：这篇新写的名字事实，向量补上（0041 决定 3 通道 2）。算不出来只记日志—— // 文档已经抽完了，不能因为召回的辅助数据没算而把它标成 failed - if let Some(client) = &embed { + if let (Some(settings), Some(client)) = (settings, &embed) { match embed_pending_names(state, settings, client, kb_id).await { Ok(n) if n > 0 => tracing::info!(%document_id, names = n, "名字向量已补"), Ok(_) => {} diff --git a/crates/utopia-server/src/ingest_sources.rs b/crates/utopia-server/src/ingest_sources.rs index ab34d1928..2040b9961 100644 --- a/crates/utopia-server/src/ingest_sources.rs +++ b/crates/utopia-server/src/ingest_sources.rs @@ -90,9 +90,13 @@ pub async fn sync_source(state: &AppState, source_id: Uuid) -> anyhow::Result<() Some(SourceKind::Webdav) => sync_webdav(state, &source).await, Some(SourceKind::Notion) => sync_notion(state, &source).await, // 被动容器：folder / api / memory / upload 没有拉取语义 - Some(SourceKind::Folder | SourceKind::Api | SourceKind::Memory | SourceKind::Upload) => { - Ok(SyncStats::default()) - } + Some( + SourceKind::Folder + | SourceKind::Api + | SourceKind::Statements + | SourceKind::Memory + | SourceKind::Upload, + ) => Ok(SyncStats::default()), None => Err(anyhow::anyhow!("unknown source kind `{}`", source.kind)), }; diff --git a/crates/utopia-server/src/pipeline.rs b/crates/utopia-server/src/pipeline.rs index 04b58f41e..fe9e62009 100644 --- a/crates/utopia-server/src/pipeline.rs +++ b/crates/utopia-server/src/pipeline.rs @@ -9,6 +9,19 @@ use utopia_llm::LlmClient; use uuid::Uuid; /// 这份文档的来源要不要抽取。没有来源的文档（直接上传、记忆片段）照旧抽。 +/// 文档所属来源的种类；没有来源（上传）为 None。抽取那边也要问同一个问题（0054） +pub(crate) async fn source_kind( + state: &AppState, + source_id: Option, +) -> anyhow::Result> { + let Some(id) = source_id else { + return Ok(None); + }; + Ok(Some( + utopia_store::sources::get(&state.pool, id).await?.kind, + )) +} + async fn source_extracts(state: &AppState, source_id: Option) -> anyhow::Result { let Some(id) = source_id else { return Ok(true); @@ -107,6 +120,8 @@ async fn run(state: &AppState, document_id: Uuid) -> anyhow::Result<()> { .await?; let kb_row = utopia_store::kbs::get(&state.pool, doc.kb_id).await?; let settings = utopia_store::settings::get(&state.pool, kb_row.workspace_id).await?; + let pushed_statements = + source_kind(state, doc.source_id).await?.as_deref() == Some("statements"); // 2. 分块 + 入库 let (text, pieces) = match parsed { @@ -115,7 +130,20 @@ async fn run(state: &AppState, document_id: Uuid) -> anyhow::Result<()> { // 那条路共用 `utopia_core::without_nul`（#665）。剥必须在算长度、分块之前：之后的 // text_len、分块偏移、全文索引、嵌入读的都是这一份，彼此才对得上 let text = utopia_core::without_nul(&parsed.text).into_owned(); - let pieces = utopia_ingest::chunk_with_budget(&text, state.chunk_tokens); + // 推送来的陈述（0054）：载荷就是契约，整份是一块。分块预算是给模型的注意力 + // 定的，这条路没有模型读；切开了契约就解析不回来 + let pieces = if pushed_statements { + vec![utopia_ingest::ChunkPiece { + seq: 0, + char_start: 0, + char_end: text.chars().count() as i32, + heading: None, + provenance: utopia_ingest::Provenance::stated(), + text: text.clone(), + }] + } else { + utopia_ingest::chunk_with_budget(&text, state.chunk_tokens) + }; (text, pieces) } // 没有文本层的扫描件、图片：工作区配了版面识别服务就交给它读（0040 第二刀）， @@ -197,8 +225,9 @@ async fn run(state: &AppState, document_id: Uuid) -> anyhow::Result<()> { return Ok(()); } - // 两段式：索引就绪后，若配置了对话模型则排队图谱抽取（不阻塞可搜可问） - if settings.as_ref().is_some_and(|s| s.chat_ready()) { + // 两段式：索引就绪后，若配置了对话模型则排队图谱抽取（不阻塞可搜可问）。 + // 推送来的陈述不问模型（0054），没配也排 + if pushed_statements || settings.as_ref().is_some_and(|s| s.chat_ready()) { utopia_store::documents::set_graph_status(&state.pool, document_id, "queued").await?; utopia_store::jobs::enqueue( &state.pool, diff --git a/docs/decisions/0054-a-source-may-push-statements-in-the-open-contract.md b/docs/decisions/0054-a-source-may-push-statements-in-the-open-contract.md new file mode 100644 index 000000000..54326b98f --- /dev/null +++ b/docs/decisions/0054-a-source-may-push-statements-in-the-open-contract.md @@ -0,0 +1,90 @@ +# 0054 · A source may push statements in the open contract + +- **Status**: proposed · cut 1 in this record's PR: the `statements` source kind, `POST /sources/{id}/statements`, deterministic extraction, the Library entry · no schema change +- **Written**: 2026-09-23 +- **Related**: [0044](0044-the-ontology-is-a-view-over-what-documents-say.md) owns the contract this reuses and the rule that typed facts come only from alignment; [0001](0001-extraction.md) is why every statement has evidence; [0022](0022-a-fact-has-two-clocks.md) is the two clocks a pushed item lands on; [0036](0036-exploration-aligns-a-schema-to-the-ontology.md) is where structured *state* lives, which this record leaves alone; [0015](0015-recording-a-sentence-is-not-asserting-a-fact.md) is why a person's `remember` needs a nod and a source's document does not; #875 is the case that surfaced it. + +> A robot's perception stack, an ERP's event bus, a sensor gateway: each already holds `{thing, relation, value, when}`. Today the only way in is to spell that into prose, push it as a document, chunk it, and pay a model call to read the prose back into a statement. The typed value becomes a sentence and then a guess at the sentence; the round trip is slow, non-deterministic and costs a model call per chunk for input that was never ambiguous. The base can read a table row without a model (#744). It cannot read a row that arrives on its own. + +## What the ground already gives, and what it withholds + +Four parts are reusable as they stand: + +- **A push interface with identity.** `POST /sources/{id}/ingest` on an `api` source: a per-source bearer key, an `external_id` that makes a second push an update in place with a version recorded, a tombstone, and a run row per call (`ingest_item_with_outcome`). +- **The open contract and its parser.** `utopia_extract::open` defines the compact reply (`e` things, `s` statements, `n` names) and `parse_open_response` reads it without any reference to the model that produced it. +- **Everything after the parse.** `extraction_open::run_open` resolves names to entities, builds described things, records names as facts, writes each statement as an open fact with its evidence located in the chunk, keeps time words verbatim, hangs qualifiers on the edge, and counts what it dropped and why. None of it knows where the reply came from. +- **A deterministic extractor for tables.** A table row is read as statements about its row's thing with the column heading as the phrase, with no model in the loop (#744). + +What it withholds: a way in that skips the model. Every pushed byte is read as prose, chunked on a token budget sized for a model's attention (`BUDGET_TOKENS = 300`), and handed to a chat endpoint that must be configured before anything reaches the graph. + +## Decisions + +**1. The body is the contract.** + +A `statements` source accepts the open extraction shape verbatim: `e`, `s` and `n` arrays with the same positions the model is asked to fill, wrapped in the same envelope `api` pushes use (`external_id`, `doc_time`, `deleted`). There is no second representation. A client writes what the extractor would have written; the parser that reads it is the parser that reads the model. The reason is 0032's and 0044's: a representation beside the one that runs is a second source of truth, and this one would drift the day the contract changed. + +**2. The payload is the document, in one piece.** + +The `{e, s, n}` object is stored as the document's content and as its single chunk, verbatim. Identity, versions, tombstones and the run history are exactly the `api` source's; the document appears in the Library under its source like any other. The chunker is not consulted: its budget exists so that a model reads a passage it can hold, and no model reads this. + +**3. No model, the same path.** + +For a document under a `statements` source, extraction parses the chunk instead of prompting for it, then continues unchanged: identity resolution, described things, name facts, evidence, time mentions, qualifiers, drop signals. The job runs whether or not a chat model is configured. A pushed statement is therefore an open statement in every respect a document's is, and reaches the typed graph the same way: through alignment (0044 cut 2), never before it. + +**4. The item is its own evidence.** + +A pushed statement carries no quote: the passage that states it is the item itself. Its evidence row names the chunk and the phrase and has null offsets, which is what `fact_evidence` already means by "the quote was not located" (0061). Names are recorded when they appear in the payload text, which for a well-formed payload is always. + +**5. There is no slot for a type.** + +The contract has positions for a phrase, a subject, an object or a value, qualifiers and time words. It has none for a property, a class or a predicate id, and this record adds none. A payload with keys outside the envelope and the contract is refused at the door, not silently ignored, so that a client cannot believe it wrote a typed fact. For the same reason a statement whose subject, or a name whose thing, is not listed in `e` is refused at the door too: extraction would drop it silently as an unknown reference, and the client would believe it landed. An object not listed in `e` is not refused; it lands as a literal value, as it does for a model's reply. + +**6. Events, not state.** + +A `statements` push says that something was the case at a time. A table that *is* the current state of a system belongs on a mount and is read at query time (0036); pushing its rows as statements would copy state into the ledger and then let the two drift. The guide says this in one sentence, because the first person to try will try with a table. + +**7. An update marks, it does not close.** + +A second push under the same `external_id` supersedes the earlier chunk. The statements that stood on it become stale under the existing rule (`documents::delete`'s comment: "没再提 ≠ 不成立"): they are handed to review, not invalidated. Closing an interval because a later observation contradicts it is the temporal engine's and alignment's job, on the typed layer, and this record does not reach into it. For the robotics case in #875 this is the honest answer: "the object was on the table" stays true of the earlier moment; what changes is what still holds now, and that is a typed question. + +## API + +``` +POST /api/v1/sources/{source_id}/statements +Authorization: Bearer +Content-Type: application/json +``` + +```json +{ + "external_id": "obs-000412", + "doc_time": "2026-09-23T08:14:03Z", + "e": [["cup-7", "cup", true], ["kitchen table", "table", true]], + "s": [[null, "cup-7", "is on", "kitchen table", null, {}, "08:14:03", null]], + "n": [] +} +``` + +- `external_id` is required and is the identity (`statements:{external_id}`); a second push with new content updates in place and records a version; `deleted: true` tombstones it. +- `doc_time` is the observation's own time and lands on the world axis; push time is the record axis (0022). Without it the item is undated, as an upload is. +- Each `s` item is `[quote, subject, phrase, object, value, qualifiers, when, ended]`; `quote` must be `null`. Each `e` item is `[name, kind word, named]`; each `n` item is `[entity name, other name, quote]` with `quote` null. +- Keys other than `external_id`, `doc_time`, `deleted`, `e`, `s`, `n` are refused with 422, as is a subject or an `n` entity not listed in `e`. A body over 64 KiB or with more than 200 statements is refused with 422; those are cut-1 limits, not contracts. +- The response is the `api` push's: `{"action": "created" | "updated" | "unchanged" | "marked_missing"}`. + +## Not doing + +- A batch endpoint. Identity, versions and runs are per item; a client that has a hundred observations makes a hundred calls, as `api` clients do. +- A typed write, a `predicate_id`, a `class` field, or any promise that a pushed statement binds before alignment reads it. +- A table importer. Tables are 0036's. +- Synthesising a quote so that offsets are non-null. The item is the passage; an offset into it would say nothing. +- A confidence per item. Every statement enters at 1.0, as a document's do; a pushed observation with an uncertainty is a qualifier (`{"confidence": "0.72"}`) the way any document's hedge is, until a record decides otherwise. + +## Phasing + +1. This PR: the kind, the route, single-chunk storage, deterministic extraction, the Library entry with the token dialog, the guide section, tests for the door and for a pushed statement reaching the open graph with evidence. +2. After 0044 cut 2 lands: measure that pushed statements bind under the same signatures as extracted ones, on a corpus where the same events are both pushed and described in prose. + +## Open questions + +- Whether a statement with no offsets should look any different on a Review card. Today it does not. +- Whether `when` should accept an RFC 3339 instant directly rather than time words, once the `instant` precision on the roadmap exists (0045). diff --git a/docs/decisions/README.md b/docs/decisions/README.md index 2631abb91..2f5f274b9 100644 --- a/docs/decisions/README.md +++ b/docs/decisions/README.md @@ -79,6 +79,7 @@ The test for writing one: if someone (including us) looks at a piece of code in | 0051 | [A human phrase decision carries its materialization work](0051-a-human-phrase-decision-carries-its-materialization-work.md) | Proposed · decision and materialization delivery; shared refactors and real regressions only | | 0052 | [Document content is a read contract over the retained ledger](0052-document-content-is-a-read-contract.md) | Proposed 2026-09-21 · implemented in #860 · two Viewer-level reads serve the retained originals the export already names by digest: `/documents/{id}/content[?version=N]` and `/documents/{id}/versions`; the handler locks the document and its ledger row through the blob read, purge answers 410, a ledger-referenced missing blob is a 500 invariant failure, a session or a scoped PAT may read, ingest tokens may not | 0053 | [A phrase decision records the inputs it considered](0053-a-phrase-decision-records-the-inputs-it-considered.md) | Implemented 2026-09-23 · a decision stores a fingerprint of the ancestor closures and admitted candidates it saw; stale means the fingerprint of the current inputs differs, which is what timestamps could not see (#807, #795): inheritance, parent edges, edits during the request; no-candidate and overflow become recorded outcomes; requeue reads live signatures only, so orphaned rows stop looping +| 0054 | [A source may push statements in the open contract](0054-a-source-may-push-statements-in-the-open-contract.md) | Proposed 2026-09-23 · cut 1 in its PR · a `statements` source accepts the open extraction contract (`e`/`s`/`n`) verbatim on `POST /sources/{id}/statements` with the `api` push's identity, versions and tombstones; the payload is stored as one chunk and extraction parses it instead of prompting a model, then runs the unchanged path, so a pushed statement is an open statement and reaches the typed graph only through alignment; there is no slot for a property or class; an update marks earlier statements stale, it does not close them; tables stay on the mount (0036) | | Record | Domain | Status | |---|---|---|---| @@ -135,6 +136,7 @@ The test for writing one: if someone (including us) looks at a piece of code in | 0051 | [A human phrase decision carries its materialization work](0051-a-human-phrase-decision-carries-its-materialization-work.md) | ontology | proposed | | 0052 | [Document content is a read contract over the retained ledger](0052-document-content-is-a-read-contract.md) | sources | current | | 0053 | [A phrase decision records the inputs it considered](0053-a-phrase-decision-records-the-inputs-it-considered.md) | ontology | current | +| 0054 | [A source may push statements in the open contract](0054-a-source-may-push-statements-in-the-open-contract.md) | sources | proposed | The status word is whether a later record has overtaken this one; what is built is in the record's own status line. Domains are the files of [../design/](../design/README.md), where every record is dated and the status words are defined. diff --git a/docs/design/sources.md b/docs/design/sources.md index 272c2dabe..8685f5a91 100644 --- a/docs/design/sources.md +++ b/docs/design/sources.md @@ -21,6 +21,8 @@ keeps the stored value; a truncated round says so [0013]. A ticket is one docume carries its history as dated declarative sentences and a `## History` list; GitHub fetches events per ticket; timestamps are written to the second [0013]. +**Statements.** A `statements` source takes the open extraction contract itself (`e`, `s`, `n`) on `POST /sources/{id}/statements`, with the `api` push's identity, versions, tombstones and run history; the payload is the document and its single chunk, and extraction parses the chunk instead of prompting a model, then runs the unchanged path, so a pushed statement is an open statement with the item as its evidence (null offsets) and reaches the typed graph only through alignment. The contract has no slot for a property or a class and unknown keys are refused; a second push under the same identity marks the earlier statements stale rather than closing them; tables belong on a mount, not here [0054]. + **RSS.** Observations are not documents: one table with baseline, candidate and no_source rows; jobs own attempts, documents own identity; an entry without a GUID or an article link is skipped; purge and reappearance are fenced by database time; one Readability extractor serves linked pages diff --git a/web/src/docs/ingest.md b/web/src/docs/ingest.md index c6e1351df..0304072f3 100644 --- a/web/src/docs/ingest.md +++ b/web/src/docs/ingest.md @@ -4,12 +4,13 @@ Utopia pulls or receives documents through **sources**. Two source kinds speak J ## Choosing between them -| | Custom (pull) | API (push) | -|---|---|---| -| Who initiates | Utopia, on a schedule | Your service, any time | -| Auth | Optional header you configure | Per-source Bearer token | -| Fits | Feeds, exports, periodic snapshots | Event-driven systems, scripts, CI | -| Deletion signal | `deleted` array in the response | `deleted: true` in a push | +| | Custom (pull) | API (push) | Statements (push) | +|---|---|---|---| +| Who initiates | Utopia, on a schedule | Your service, any time | Your service, any time | +| Auth | Optional header you configure | Per-source Bearer token | Per-source Bearer token | +| What you send | Items with text | Documents with text | Statements already in the extraction contract; no model reads them | +| Fits | Feeds, exports, periodic snapshots | Event-driven systems, scripts, CI | Sensors, event buses, anything that already knows `{thing, relation, value, when}` | +| Deletion signal | `deleted` array in the response | `deleted: true` in a push | `deleted: true` in a push | --- @@ -119,9 +120,49 @@ curl -X POST "https://utopia.example.com/api/v1/sources/01a0…/ingest" \ --- +## Statements source — the structured push interface + +Create a **Statements** source; it gets its own push token like an API source. Use it when your system already holds the statement and would otherwise have to write it out as prose for a model to read back. Nothing here calls a model: the body is stored as the document and read by the same parser that reads the extractor's reply, so a pushed statement is an open statement in every respect a document's is. It reaches the typed graph the same way, through alignment, never before. + +``` +POST {your-utopia-base}/api/v1/sources/{source_id}/statements +Authorization: Bearer utp_… +Content-Type: application/json +``` + +```json +{ + "external_id": "obs-000412", + "doc_time": "2026-09-23T08:14:03Z", + "e": [["cup-7", "cup", true], ["kitchen table", "table", true]], + "s": [[null, "cup-7", "is on", "kitchen table", null, {}, "08:14:03", null]], + "n": [] +} +``` + +| Field | Required | Meaning | +|---|---|---| +| `external_id` | yes | Stable identity. Same identity + new content → update in place, with a version recorded. | +| `doc_time` | no | RFC 3339; the observation's own time. Without it the item is undated. | +| `e` | yes | Things: `[name, kind word, named]`. `named` is `true` for a name, `false` for a description. | +| `s` | yes | Statements: `[quote, subject, phrase, object, value, qualifiers, when, ended]`. `quote` must be `null`; `subject` and `object` name things listed in `e`; give `object` or `value`, not both; `qualifiers` is an object keyed by your own role words; `when` / `ended` are time words as you would write them. | +| `n` | yes | Other names: `[entity name, other name, quote]`, `quote` null. May be empty. | +| `deleted` | no | `true` marks the identified item "Not in source". | + +Any other key is refused with `422`, so a mistaken `predicate` or `class` field cannot pass as a typed fact: the contract has no slot for one. A `subject`, or the entity of an `n` row, that is not listed in `e` is refused the same way, rather than dropped later in silence. A body over 64 KiB or with more than 200 statements is refused too. + +Two things to know before wiring a system to it: + +- **Send events, not state.** A row that *is* the current state of a system belongs on a mounted database, read at query time. Pushing it here copies state into the ledger and the two drift. +- **A new push under the same identity does not close the old statement.** The earlier one becomes stale and goes to review; deciding that an interval ended is done on the typed layer, by alignment and the temporal engine, not by the push. + +The response is the API push's: `{ "action": "created" }` · `updated` · `unchanged` · `marked_missing`. + +--- + ## Shared semantics -- **Identity, not filenames.** Documents are tracked by `custom:{id}` / `api:{external_id}` keys. Renames are recognized as moves; content changes update the same document. +- **Identity, not filenames.** Documents are tracked by `custom:{id}` / `api:{external_id}` / `statements:{external_id}` keys. Renames are recognized as moves; content changes update the same document. - **Updates keep history.** Every content change records a version; earlier extracted knowledge keeps its provenance. - **Deletion is a marker.** Tombstones set a "Not in source" flag; the Library shows a cleanup action, and a human confirms actual deletion. - **`doc_time` drives the time axis.** Documents without it fall back to their ingestion time — real timestamps make the temporal graph meaningfully better. diff --git a/web/src/i18n/en.ts b/web/src/i18n/en.ts index e06f4f731..f8aed7854 100644 --- a/web/src/i18n/en.ts +++ b/web/src/i18n/en.ts @@ -586,6 +586,7 @@ export const en = { url: "URLs", rss: "RSS feed", api: "API", + statements: "Statements", custom: "Custom", github_issues: "GitHub issues", jira_issues: "Jira issues", @@ -626,6 +627,8 @@ export const en = { "and it appears here. Dated by when the page was last edited, which is the page's " + "own clock rather than ours.", api: "External systems push JSON documents here, authenticated with this source's own token.", + statements: + "Your system pushes statements already in the extraction contract; no model reads them. Send events, not the state of a table.", custom: "Polls a URL you control on a schedule — your service returns JSON items and Utopia keeps them in sync.", memory: diff --git a/web/src/i18n/zh.ts b/web/src/i18n/zh.ts index 77ea854ae..8119ee46b 100644 --- a/web/src/i18n/zh.ts +++ b/web/src/i18n/zh.ts @@ -535,6 +535,7 @@ export const zh: Strings = { url: "网页", rss: "RSS 订阅", api: "API", + statements: "陈述", custom: "自定义", github_issues: "GitHub 工单", jira_issues: "Jira 工单", @@ -572,6 +573,8 @@ export const zh: Strings = { "同步一个 Notion 集成能看见的页面——把页面分享给集成，它就会出现在这里。" + "日期取页面最后一次编辑的时刻，那是页面自己的时钟，不是我们抓它的时刻。", api: "外部系统把 JSON 文档推送到这里，用这个来源自己的令牌认证。", + statements: + "外部系统把已经是抽取契约形状的陈述推送到这里，不经模型；发事件，别发整张表的状态。", custom: "按计划轮询一个你控制的 URL——你的服务返回 JSON 条目，Utopia 保持同步。", memory: diff --git a/web/src/pages/Library.tsx b/web/src/pages/Library.tsx index 3a8a666f3..30bed581d 100644 --- a/web/src/pages/Library.tsx +++ b/web/src/pages/Library.tsx @@ -922,7 +922,8 @@ function SourceBar({ onToken: () => void; }) { const isPull = SYNCING_KINDS.has(source.kind); - const isApi = source.kind === "api"; + // 推送类来源：api 推文档，statements 推陈述（0054）；界面上同一套状态、令牌与指南 + const isApi = source.kind === "api" || source.kind === "statements"; const busy = source.last_sync_status === "running" || source.last_sync_status === "queued"; // 历史数据的 config 可能是 jsonb null（缺省 Value::Null 落库所致）——防御性兜底 const cfg = source.config ?? {}; @@ -1002,7 +1003,7 @@ function SourceBar({ )}
{/* 集成型来源（custom 拉取 / api 推送）：接口文档随手可达 */} - {(source.kind === "custom" || source.kind === "api") && ( + {(source.kind === "custom" || isApi) && ( )} {/* History 对拉取型与 api 推送型都开放：推送失败（格式错等）也记 run */} - {(isPull || source.kind === "api") && ( + {(isPull || isApi) && ( /* 激活态用反色（与弹窗类型 tab、图标选中同一语汇），一眼可辨 */