From 3343022ad7c10132fda3d61bbfd035c32a0f5820 Mon Sep 17 00:00:00 2001 From: dada-yan Date: Sun, 20 Sep 2026 19:29:00 +1000 Subject: [PATCH] Preserve literal objects when exporting unbound statements Signed-off-by: dada-yan --- crates/utopia-server/src/api/mcp_tests.rs | 153 ++++++++++++++++++++++ crates/utopia-server/src/rdf.rs | 103 ++++++++++++++- 2 files changed, 252 insertions(+), 4 deletions(-) diff --git a/crates/utopia-server/src/api/mcp_tests.rs b/crates/utopia-server/src/api/mcp_tests.rs index 99528a28e..18f269bb9 100644 --- a/crates/utopia-server/src/api/mcp_tests.rs +++ b/crates/utopia-server/src/api/mcp_tests.rs @@ -1632,3 +1632,156 @@ async fn rule_matches_keep_materialized_intervals_and_count_rows() -> anyhow::Re ); f.clean().await } + +// Reuse the authenticated ledger fixture so RDF exercises the same stored records +// as structured MCP reads, including evidence and retracted history. +#[tokio::test] +async fn rdf_export_preserves_unbound_literal_objects() -> anyhow::Result<()> { + use axum::body::{to_bytes, Body}; + use axum::http::{Request, StatusCode}; + use oxrdf::{vocab::rdf, Literal, Term}; + use tower::ServiceExt; + + let Some(f) = Fixture::new().await? else { + return Ok(()); + }; + async fn check(f: &Fixture) -> anyhow::Result<()> { + let value = json!({"value": "待复检"}); + let (statement, _) = utopia_store::graph::insert_open_statement( + &f.state.pool, + f.kb, + f.subject, + "状态", + utopia_store::graph::FactObject::Value(&value), + Some("2026-01-01T00:00:00Z".parse()?), + 0.9, + ) + .await?; + sqlx::query("UPDATE facts SET recorded_at='2026-02-01' WHERE id=$1") + .bind(statement) + .execute(&f.state.pool) + .await?; + sqlx::query( + "INSERT INTO fact_evidence(fact_id,chunk_id,document_id,doc_version,quote) + VALUES ($1,$2,$3,1,'设备 A 待复检')", + ) + .bind(statement) + .bind(f.chunk) + .bind(f.document) + .execute(&f.state.pool) + .await?; + let auth = utopia_store::tokens::authenticate(&f.state.pool, &f.token).await?; + let jwt = crate::auth::issue_token(&f.state, auth.user_id)?; + let app = crate::api::router(f.state.clone(), &Default::default()); + let names = crate::rdf::Names::new(f.kb, None).map_err(anyhow::Error::msg)?; + let stmt = names.fact(statement); + let mut formats = Vec::new(); + for retracted in [false, true] { + if retracted { + sqlx::query("UPDATE facts SET invalidated_at='2026-03-01' WHERE id=$1") + .bind(statement) + .execute(&f.state.pool) + .await?; + } + // Snapshot every KB-scoped business table, including queues and adoption + // records. Request audit is deliberately excluded from this read-only check. + let tables: Vec = sqlx::query_scalar( + "SELECT table_name FROM information_schema.columns + WHERE table_schema='public' AND column_name='kb_id' + AND table_name <> 'audit_events' ORDER BY table_name", + ) + .fetch_all(&f.state.pool) + .await?; + let snapshot = async { + let mut rows = Vec::new(); + for table in &tables { + let sql = format!("SELECT COALESCE(jsonb_agg(to_jsonb(t) ORDER BY to_jsonb(t)::text), '[]'::jsonb) FROM \"{}\" t WHERE kb_id=$1", table.replace('"', "\"\"")); + rows.push( + sqlx::query_scalar::<_, Value>(&sql) + .bind(f.kb) + .fetch_one(&f.state.pool) + .await?, + ); + } + Ok::<_, anyhow::Error>(rows) + }; + let before = snapshot.await?; + let extra_sql = "SELECT jsonb_build_array( + (SELECT COALESCE(jsonb_agg(to_jsonb(e) ORDER BY to_jsonb(e)::text), '[]') + FROM fact_evidence e JOIN facts f ON f.id=e.fact_id WHERE f.kb_id=$1), + (SELECT COALESCE(jsonb_agg(to_jsonb(j) ORDER BY j.id), '[]') FROM jobs j + WHERE payload->>'kb_id'=$1::text OR payload->>'document_id' IN + (SELECT id::text FROM documents WHERE kb_id=$1)))"; + let extra_before: Value = sqlx::query_scalar(extra_sql) + .bind(f.kb) + .fetch_one(&f.state.pool) + .await?; + for format in ["turtle", "jsonld"] { + let response = app + .clone() + .oneshot( + Request::builder() + .uri(format!("/api/v1/kbs/{}/export?format={format}", f.kb)) + .header("authorization", format!("Bearer {jwt}")) + .body(Body::empty())?, + ) + .await?; + anyhow::ensure!(response.status() == StatusCode::OK, "export rejected"); + let bytes = to_bytes(response.into_body(), 1024 * 1024).await?; + let format = if format == "turtle" { + oxrdfio::RdfFormat::Turtle + } else { + oxrdfio::RdfFormat::JsonLd { + profile: oxrdfio::JsonLdProfileSet::empty(), + } + }; + let quads = oxrdfio::RdfParser::from_format(format) + .for_slice(&bytes) + .collect::, _>>()?; + anyhow::ensure!( + quads.iter().any(|q| q.subject == stmt.clone().into() + && q.predicate == rdf::OBJECT + && q.object == Term::Literal(Literal::new_simple_literal("待复检"))), + "unbound statement lost its rdf:object in authenticated export" + ); + anyhow::ensure!( + !quads + .iter() + .any(|q| q.subject == stmt.clone().into() && q.predicate == rdf::PREDICATE), + "invented a bound predicate" + ); + anyhow::ensure!( + quads.iter().any(|q| q.subject == stmt.clone().into() + && q.predicate.as_str() == "http://www.w3.org/ns/prov#wasDerivedFrom" + && q.object == names.document(f.document).into()), + "lost evidence source" + ); + formats.push(quads); + } + anyhow::ensure!( + formats[formats.len() - 1] == formats[formats.len() - 2], + "formats disagree" + ); + let extra_after: Value = sqlx::query_scalar(extra_sql) + .bind(f.kb) + .fetch_one(&f.state.pool) + .await?; + anyhow::ensure!( + extra_before == extra_after, + "export changed evidence or jobs" + ); + for (table, expected) in tables.iter().zip(before) { + let sql = format!("SELECT COALESCE(jsonb_agg(to_jsonb(t) ORDER BY to_jsonb(t)::text), '[]'::jsonb) FROM \"{}\" t WHERE kb_id=$1", table.replace('"', "\"\"")); + let actual: Value = sqlx::query_scalar(&sql) + .bind(f.kb) + .fetch_one(&f.state.pool) + .await?; + anyhow::ensure!(actual == expected, "export changed {table}"); + } + } + Ok(()) + } + let result = check(&f).await; + let cleanup = f.clean().await; + result.and(cleanup) +} diff --git a/crates/utopia-server/src/rdf.rs b/crates/utopia-server/src/rdf.rs index 8be69b9b4..bec9ff7a2 100644 --- a/crates/utopia-server/src/rdf.rs +++ b/crates/utopia-server/src/rdf.rs @@ -453,10 +453,11 @@ pub fn emit_fact( let predicate = f.predicate_id.and_then(|p| vocab.relation(p)).cloned(); let object: Option = match (f.object_id, &f.object_value) { (Some(o), _) => Some(names.entity(o).into()), - (None, Some(v)) => f.predicate_id.map(|p| { - let (datatype, _) = vocab.literal_shape(p); - literal_value(v, datatype).into() - }), + (None, Some(v)) => { + // An unbound statement still has an object; only its datatype is unknown. + let datatype = f.predicate_id.and_then(|p| vocab.literal_shape(p).0); + Some(literal_value(v, datatype).into()) + } _ => None, }; @@ -801,6 +802,100 @@ mod tests { const OBJ: &str = ""; const WORKS_FOR: &str = "https://schema.org/worksFor"; + #[test] + fn unbound_literal_objects_survive_both_formats() { + for value in [ + serde_json::json!({"value": "待复检"}), + serde_json::json!({"value": "quote: \" and slash: \\"}), + serde_json::json!({"value": ""}), + serde_json::json!({"value": 0}), + serde_json::json!({"value": false}), + serde_json::json!({"value": null, "summary": "not specified"}), + serde_json::Value::Null, + ] { + let mut f = fact(5); + f.predicate_id = None; + f.surface_predicate = Some("状态".into()); + f.object_id = None; + f.object_value = Some(value.clone()); + f.documents = vec![id(20)]; + f.quotes = vec!["设备 A 待复检".into()]; + f.supersedes = Some(id(6)); + for retracted in [false, true] { + f.invalidated_at = retracted.then(|| at("2026-02-01T00:00:00Z")); + let mut sets = Vec::new(); + for format in [Format::Turtle, Format::JsonLd] { + let quads = export(format, |sink, names, vocab| { + emit_fact(sink, names, vocab, &f, at("2026-06-01T00:00:00Z")).unwrap(); + }); + assert_eq!( + objects(&quads, STMT, rdf::OBJECT.as_str()), + vec![literal_value(&value, None).to_string()], + "unbound statement lost its literal object: {value}" + ); + assert!(objects(&quads, STMT, rdf::PREDICATE.as_str()).is_empty()); + assert!(!quads.iter().any(|q| q.subject.to_string() == SUBJ)); + sets.push(quads.into_iter().collect::>()); + } + assert_eq!(sets[0], sets[1]); + } + } + } + + #[test] + fn bound_literal_datatypes_survive_both_formats() { + for (datatype, value, expected) in [ + ( + "number", + serde_json::json!(0), + Literal::new_typed_literal("0", xsd::DECIMAL), + ), + ( + "text", + serde_json::json!("待复检"), + Literal::new_simple_literal("待复检"), + ), + ( + "bool", + serde_json::json!(false), + Literal::new_typed_literal("false", xsd::BOOLEAN), + ), + ] { + let mut f = fact(5); + f.predicate_id = Some(id(4)); + f.object_id = None; + f.object_value = Some(serde_json::json!({"value": value})); + for format in [Format::Turtle, Format::JsonLd] { + let quads = export(format, |sink, names, _| { + let mut property = relation(4, "value", None, "attribute"); + property.datatype = Some(datatype.into()); + let vocab = vocabulary(names, &[], &[property]); + emit_fact(sink, names, &vocab, &f, at("2026-06-01T00:00:00Z")).unwrap(); + }); + assert_eq!( + objects(&quads, STMT, rdf::OBJECT.as_str()), + vec![expected.to_string()] + ); + assert!(quads.iter().any(|q| q.subject.to_string() == SUBJ + && q.object == Term::Literal(expected.clone()))); + } + } + } + + #[test] + fn an_absent_object_is_not_an_empty_literal() { + let mut f = fact(5); + f.predicate_id = None; + f.object_id = None; + f.object_value = None; + for format in [Format::Turtle, Format::JsonLd] { + let quads = export(format, |sink, names, vocab| { + emit_fact(sink, names, vocab, &f, at("2026-06-01T00:00:00Z")).unwrap(); + }); + assert!(objects(&quads, STMT, rdf::OBJECT.as_str()).is_empty()); + } + } + #[test] fn an_imported_class_keeps_its_own_iri() { let quads = export(Format::Turtle, |_, _, _| {});