From 363a574b216308939032ee866bd675046080797a Mon Sep 17 00:00:00 2001 From: Wayland Yang Date: Wed, 23 Sep 2026 20:37:22 +0800 Subject: [PATCH 1/4] Let a source push statements in the open contract instead of prose a model must read back A statements source takes the extraction contract itself (e/s/n) on POST /sources/{id}/statements with the api push's identity, versions and tombstones; the payload is one chunk and extraction parses it instead of prompting, so a pushed statement is an open statement and reaches the typed graph only through alignment. Record 0054. Co-Authored-By: Claude Fable 5.1 Signed-off-by: Wayland Yang --- crates/utopia-core/src/models.rs | 2 + crates/utopia-server/src/api/mod.rs | 5 + .../utopia-server/src/api/sources_routes.rs | 208 ++++++++++- .../src/api/sources_statements_tests.rs | 333 ++++++++++++++++++ crates/utopia-server/src/extraction.rs | 28 +- crates/utopia-server/src/extraction_open.rs | 87 +++-- crates/utopia-server/src/ingest_sources.rs | 10 +- crates/utopia-server/src/pipeline.rs | 35 +- ...ay-push-statements-in-the-open-contract.md | 90 +++++ docs/decisions/README.md | 2 + docs/design/sources.md | 2 + web/src/docs/ingest.md | 55 ++- web/src/i18n/en.ts | 3 + web/src/i18n/zh.ts | 3 + web/src/pages/Library.tsx | 11 +- web/src/pages/SourcesRail.tsx | 4 +- web/src/sourceKinds.ts | 1 + 17 files changed, 819 insertions(+), 60 deletions(-) create mode 100644 crates/utopia-server/src/api/sources_statements_tests.rs create mode 100644 docs/decisions/0054-a-source-may-push-statements-in-the-open-contract.md 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..96424d4ae 100644 --- a/crates/utopia-server/src/api/mod.rs +++ b/crates/utopia-server/src/api/mod.rs @@ -475,6 +475,11 @@ 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）：请求体就是开放抽取契约，抽取不问模型 + .route( + "/sources/{source_id}/statements", + post(sources_routes::push_statements), + ) .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 105600cc0..bd9f88d22 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 matches!(source.kind.as_str(), "api" | "statements") { let token = new_ingest_token(); utopia_store::sources::set_ingest_token(&state.pool, source.id, &token).await?; ingest_token = Some(token); @@ -539,6 +539,208 @@ 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()) + } + } +} + +/// 一次推送最多多少字节、多少条陈述。第一刀的上限，不是契约：一块就是一份载荷， +/// 这两个数只是不让一份载荷大到审核卡片和证据视图没法呈现 +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()); + } + 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( @@ -579,6 +781,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..4c6c1c3a2 --- /dev/null +++ b/crates/utopia-server/src/api/sources_statements_tests.rs @@ -0,0 +1,333 @@ +//! 推陈述的来源（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"); +} + +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, + _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 app = super::super::router(state.clone(), &cfg); + Ok(Some(Self { + pool, + state, + app, + org, + kb, + source, + api_source, + token, + api_token, + _dir: dir, + })) + } + + async fn push( + &self, + source: Uuid, + token: &str, + body: &Value, + ) -> anyhow::Result<(StatusCode, Value)> { + let response = self + .app + .clone() + .oneshot( + Request::post(format!("/api/v1/sources/{source}/statements")) + .header("Authorization", format!("Bearer {token}")) + .header("Content-Type", "application/json") + .body(Body::from(body.to_string()))?, + ) + .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" + ); + 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); + f.cleanup().await +} diff --git a/crates/utopia-server/src/extraction.rs b/crates/utopia-server/src/extraction.rs index e111adbda..05acfc7ac 100644 --- a/crates/utopia-server/src/extraction.rs +++ b/crates/utopia-server/src/extraction.rs @@ -341,11 +341,21 @@ 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"); + let settings = utopia_store::settings::get(&state.pool, kb.workspace_id).await?; + let (settings, client) = if pushed { + (None, None) + } else { + let settings = + settings.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"))?; + (Some(settings), Some(client)) + }; // 所有权凭证：重抽会自增 epoch，任务据此察觉自己已被接管（见 `run_open` 的分块循环） let my_epoch = utopia_store::documents::extract_epoch(&state.pool, document_id).await?; @@ -353,7 +363,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..48879676c 100644 --- a/crates/utopia-server/src/extraction_open.rs +++ b/crates/utopia-server/src/extraction_open.rs @@ -222,17 +222,20 @@ async fn place( } /// `await_nod`：这是记忆日志（0015）——陈述不直接落库，原样进待确认表，人点头时才成为开放陈述。 -/// `proposer`：那句话是谁、经哪枚令牌说的（0026），随待确认项一起记 +/// `proposer`：那句话是谁、经哪枚令牌说的（0026），随待确认项一起记。 +/// `pushed`：块本身就是契约（0054 的 `statements` 来源）——不建提示词、不问模型，直接解析； +/// 这时 `settings` 与 `client` 可以为 None，其余一步不变 #[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; @@ -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, "开放抽取回复解析失败，跳过该分块"); 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..df230bfe2 --- /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. + +**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. 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 3ce0ed034..33ed2a4ab 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..3afb14579 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 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 1bfe312bd..64f667e3e 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 2add82f51..c5037d1f7 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、图标选中同一语汇），一眼可辨 */