mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-27 21:08:00 +00:00
* fix(memory): graph query returns namespace data and add sync e2e tests (#344) The knowledge graph UI showed empty because graph_query(None) only queried the graph_global table, while ingestion writes to graph_namespace. Now graph_query(None) queries both tables via graph_query_all(), merging results. Changes: - Added graph_query_all() in unified graph store to query across all namespaces - MemoryClient::graph_query(None) now uses graph_query_all() instead of graph_query_global() - MemoryWorkspace passes selectedNamespace to the RPC call - Added diagnostic logging in ingestion pipeline (RelEx model availability, extraction counts) - Added debug logging in tauriCommands for unexpected response shapes - Added 2 integration tests proving document sync populates the graph (ignored by default for CI, run with --ignored) Closes #344 Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * style: remove trailing comma for Prettier compliance Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
sanil-23
co-authored by
Claude Opus 4.6
Claude Opus 4.6
parent
48d937cf6b
commit
b8ae9674b3
@@ -168,9 +168,10 @@ export function MemoryWorkspace({ onToast }: MemoryWorkspaceProps) {
|
||||
|
||||
setGraphRelationsLoading(true);
|
||||
try {
|
||||
const relations = await memoryGraphQuery();
|
||||
const relations = await memoryGraphQuery(selectedNamespace || undefined);
|
||||
setGraphRelations(relations);
|
||||
} catch {
|
||||
} catch (err) {
|
||||
console.error('[MemoryWorkspace] memoryGraphQuery failed:', err);
|
||||
setGraphRelations([]);
|
||||
} finally {
|
||||
setGraphRelationsLoading(false);
|
||||
|
||||
@@ -520,6 +520,10 @@ export async function memoryGraphQuery(
|
||||
if (Array.isArray(raw)) return raw;
|
||||
if (raw && typeof raw === 'object' && 'result' in raw && Array.isArray(raw.result))
|
||||
return raw.result;
|
||||
console.debug(
|
||||
'[memoryGraphQuery] unexpected response shape, returning empty array. Raw response:',
|
||||
raw
|
||||
);
|
||||
return [];
|
||||
}
|
||||
|
||||
|
||||
@@ -934,6 +934,13 @@ async fn parse_document(
|
||||
let chunks = UnifiedMemory::chunk_document_content(content, DEFAULT_CHUNK_TOKENS);
|
||||
let relex_runtime = relex::runtime(&config.model_name).await;
|
||||
let model_enabled = relex_runtime.is_some();
|
||||
log::debug!(
|
||||
"[memory:ingestion] parse_document title={title:?} model={} relex_available={model_enabled} \
|
||||
content_len={} chunk_count={}",
|
||||
config.model_name,
|
||||
content.len(),
|
||||
chunks.len(),
|
||||
);
|
||||
let mut accumulator = ExtractionAccumulator {
|
||||
document_title: Some(sanitize_entity_name(title)),
|
||||
primary_subject: detect_primary_subject(title),
|
||||
@@ -1585,6 +1592,15 @@ async fn parse_document(
|
||||
}).collect::<Vec<_>>(),
|
||||
});
|
||||
|
||||
log::debug!(
|
||||
"[memory:ingestion] parse_document complete title={title:?} \
|
||||
entities={} relations={} preferences={} decisions={}",
|
||||
entities.len(),
|
||||
relations.len(),
|
||||
accumulator.preferences.len(),
|
||||
accumulator.decisions.len(),
|
||||
);
|
||||
|
||||
ParsedIngestion {
|
||||
tags,
|
||||
metadata,
|
||||
|
||||
@@ -341,6 +341,9 @@ impl MemoryClient {
|
||||
}
|
||||
|
||||
/// Query relationships in the knowledge graph using optional filters.
|
||||
///
|
||||
/// When `namespace` is `None`, returns relations from **all** namespaces
|
||||
/// plus the global graph, so ingested data is always surfaced in the UI.
|
||||
pub async fn graph_query(
|
||||
&self,
|
||||
namespace: Option<&str>,
|
||||
@@ -353,7 +356,7 @@ impl MemoryClient {
|
||||
.graph_query_namespace(ns, subject, predicate)
|
||||
.await
|
||||
}
|
||||
None => self.inner.graph_query_global(subject, predicate).await,
|
||||
None => self.inner.graph_query_all(subject, predicate).await,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -129,6 +129,30 @@ impl UnifiedMemory {
|
||||
.collect::<Vec<_>>())
|
||||
}
|
||||
|
||||
/// Query all graph relations across every namespace AND global, with
|
||||
/// optional subject/predicate filters. Used when the caller passes no
|
||||
/// namespace so that ingested (namespace-scoped) data is still surfaced.
|
||||
pub async fn graph_query_all(
|
||||
&self,
|
||||
subject: Option<&str>,
|
||||
predicate: Option<&str>,
|
||||
) -> Result<Vec<serde_json::Value>, String> {
|
||||
let mut rows = self
|
||||
.graph_relations_all_namespaces(subject, predicate)
|
||||
.await?;
|
||||
rows.extend(self.graph_relations_global(subject, predicate).await?);
|
||||
rows.sort_by(|a, b| {
|
||||
b.updated_at
|
||||
.partial_cmp(&a.updated_at)
|
||||
.unwrap_or(std::cmp::Ordering::Equal)
|
||||
});
|
||||
rows.truncate(300);
|
||||
Ok(rows
|
||||
.into_iter()
|
||||
.map(Self::graph_relation_to_json)
|
||||
.collect::<Vec<_>>())
|
||||
}
|
||||
|
||||
pub async fn graph_query_namespace(
|
||||
&self,
|
||||
namespace: &str,
|
||||
@@ -241,6 +265,48 @@ impl UnifiedMemory {
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
/// Query relations from `graph_namespace` across ALL namespaces, with
|
||||
/// optional subject/predicate filters.
|
||||
pub(crate) async fn graph_relations_all_namespaces(
|
||||
&self,
|
||||
subject: Option<&str>,
|
||||
predicate: Option<&str>,
|
||||
) -> Result<Vec<GraphRelationRecord>, String> {
|
||||
let conn = self.conn.lock();
|
||||
let subject = subject.map(Self::normalize_graph_entity);
|
||||
let predicate = predicate.map(Self::normalize_graph_predicate);
|
||||
let mut stmt = conn
|
||||
.prepare(
|
||||
"SELECT namespace, subject, predicate, object, attrs_json, updated_at
|
||||
FROM graph_namespace
|
||||
WHERE (?1 IS NULL OR subject = ?1)
|
||||
AND (?2 IS NULL OR predicate = ?2)
|
||||
ORDER BY updated_at DESC
|
||||
LIMIT 300",
|
||||
)
|
||||
.map_err(|e| format!("graph_relations_all_namespaces prepare: {e}"))?;
|
||||
let mut rows = stmt
|
||||
.query(params![subject, predicate])
|
||||
.map_err(|e| format!("graph_relations_all_namespaces query: {e}"))?;
|
||||
let mut out = Vec::new();
|
||||
while let Some(row) = rows
|
||||
.next()
|
||||
.map_err(|e| format!("graph_relations_all_namespaces row: {e}"))?
|
||||
{
|
||||
let namespace: String = row.get(0).map_err(|e| e.to_string())?;
|
||||
let attrs_raw: String = row.get(4).map_err(|e| e.to_string())?;
|
||||
out.push(Self::graph_relation_from_parts(
|
||||
Some(namespace),
|
||||
row.get(1).map_err(|e| e.to_string())?,
|
||||
row.get(2).map_err(|e| e.to_string())?,
|
||||
row.get(3).map_err(|e| e.to_string())?,
|
||||
&attrs_raw,
|
||||
row.get(5).map_err(|e| e.to_string())?,
|
||||
));
|
||||
}
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
async fn graph_upsert_internal(
|
||||
&self,
|
||||
namespace: Option<&str>,
|
||||
|
||||
@@ -0,0 +1,262 @@
|
||||
//! Integration test: document ingestion → graph query pipeline.
|
||||
//!
|
||||
//! Verifies that storing a document through the memory system produces
|
||||
//! graph entities and relations that are queryable via the same APIs
|
||||
//! the UI calls.
|
||||
//!
|
||||
//! Tests are `#[ignore]` by default (slow, requires disk I/O + ingestion worker).
|
||||
//! Run explicitly:
|
||||
//! cargo test --test memory_graph_sync_e2e -- --ignored --nocapture
|
||||
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use serde_json::json;
|
||||
use tempfile::tempdir;
|
||||
|
||||
use openhuman_core::openhuman::memory::{
|
||||
embeddings::NoopEmbedding, MemoryClient, MemoryIngestionConfig, MemoryIngestionRequest,
|
||||
NamespaceDocumentInput, UnifiedMemory,
|
||||
};
|
||||
|
||||
/// Config that skips the GLiNER RelEx ONNX model (avoids ORT init issues
|
||||
/// on CI runners). Heuristic extraction still runs and will pick up
|
||||
/// structured lines like "Owner:", "Project name:", etc.
|
||||
fn ci_safe_config() -> MemoryIngestionConfig {
|
||||
MemoryIngestionConfig {
|
||||
model_name: "__test_no_model__".to_string(),
|
||||
..MemoryIngestionConfig::default()
|
||||
}
|
||||
}
|
||||
|
||||
/// A document with known entities that the heuristic extractor can find.
|
||||
/// Uses structured lines (Project name, Owner, etc.) that the parser
|
||||
/// recognises without requiring the ONNX model.
|
||||
const TEST_DOCUMENT: &str = "\
|
||||
Project name: Acme Corp
|
||||
Owner: Alice
|
||||
|
||||
Alice works at Acme Corp. Bob is the CEO of Acme Corp.
|
||||
|
||||
From: Alice <alice@acme.com>
|
||||
To: Bob <bob@acme.com>
|
||||
Subject: Q4 roadmap review
|
||||
|
||||
Hi Bob, let's review the Q4 roadmap for Acme Corp next week.
|
||||
|
||||
Decision: Ship the beta release by end of November.
|
||||
Preferred communication channel: Slack over email.
|
||||
";
|
||||
|
||||
// ── Test: full ingest_document → graph_query_namespace ─────────────────
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore] // Slow: SQLite + ingestion pipeline. Run with --ignored.
|
||||
async fn ingest_document_populates_namespace_graph() {
|
||||
let _ = env_logger::builder()
|
||||
.filter_level(log::LevelFilter::Debug)
|
||||
.is_test(true)
|
||||
.try_init();
|
||||
|
||||
let tmp = tempdir().expect("tempdir");
|
||||
let memory =
|
||||
UnifiedMemory::new(tmp.path(), Arc::new(NoopEmbedding), None).expect("UnifiedMemory::new");
|
||||
|
||||
let namespace = "test-ns";
|
||||
|
||||
let result = memory
|
||||
.ingest_document(MemoryIngestionRequest {
|
||||
document: NamespaceDocumentInput {
|
||||
namespace: namespace.to_string(),
|
||||
key: "acme-doc".to_string(),
|
||||
title: "Acme Corp team overview".to_string(),
|
||||
content: TEST_DOCUMENT.to_string(),
|
||||
source_type: "doc".to_string(),
|
||||
priority: "high".to_string(),
|
||||
tags: Vec::new(),
|
||||
metadata: json!({}),
|
||||
category: "core".to_string(),
|
||||
session_id: None,
|
||||
document_id: None,
|
||||
},
|
||||
config: ci_safe_config(),
|
||||
})
|
||||
.await
|
||||
.expect("ingest_document");
|
||||
|
||||
eprintln!("--- Ingestion result ---");
|
||||
eprintln!(" document_id: {}", result.document_id);
|
||||
eprintln!(" namespace: {}", result.namespace);
|
||||
eprintln!(" entities: {}", result.entity_count);
|
||||
eprintln!(" relations: {}", result.relation_count);
|
||||
eprintln!(" chunks: {}", result.chunk_count);
|
||||
eprintln!(" preferences: {}", result.preference_count);
|
||||
eprintln!(" decisions: {}", result.decision_count);
|
||||
|
||||
for entity in &result.entities {
|
||||
eprintln!(" entity: {} ({})", entity.name, entity.entity_type);
|
||||
}
|
||||
for relation in &result.relations {
|
||||
eprintln!(
|
||||
" relation: {} --[{}]--> {}",
|
||||
relation.subject, relation.predicate, relation.object
|
||||
);
|
||||
}
|
||||
|
||||
// ── Verify entities extracted ──
|
||||
assert!(
|
||||
result.entity_count >= 2,
|
||||
"Expected at least 2 entities from heuristic extraction, got {}",
|
||||
result.entity_count
|
||||
);
|
||||
|
||||
let entity_names: Vec<&str> = result.entities.iter().map(|e| e.name.as_str()).collect();
|
||||
eprintln!(" All entity names: {entity_names:?}");
|
||||
|
||||
// The heuristic extractor should find ALICE and ACME CORP from the
|
||||
// structured lines.
|
||||
assert!(
|
||||
entity_names.iter().any(|n| n.contains("ALICE")),
|
||||
"Expected entity 'ALICE' among: {entity_names:?}"
|
||||
);
|
||||
assert!(
|
||||
entity_names.iter().any(|n| n.contains("ACME")),
|
||||
"Expected entity containing 'ACME' among: {entity_names:?}"
|
||||
);
|
||||
|
||||
// ── Verify relations extracted ──
|
||||
assert!(
|
||||
result.relation_count >= 1,
|
||||
"Expected at least 1 relation, got {}",
|
||||
result.relation_count
|
||||
);
|
||||
|
||||
// ── Verify graph is queryable via namespace ──
|
||||
let graph_rows = memory
|
||||
.graph_query_namespace(namespace, None, None)
|
||||
.await
|
||||
.expect("graph_query_namespace");
|
||||
|
||||
eprintln!(
|
||||
"\n--- graph_query_namespace({namespace}) returned {} rows ---",
|
||||
graph_rows.len()
|
||||
);
|
||||
for row in &graph_rows {
|
||||
eprintln!(" {row}");
|
||||
}
|
||||
|
||||
assert!(
|
||||
!graph_rows.is_empty(),
|
||||
"graph_query_namespace should return relations after ingestion"
|
||||
);
|
||||
|
||||
// ── Verify graph_query_all also returns the namespace data ──
|
||||
let all_rows = memory
|
||||
.graph_query_all(None, None)
|
||||
.await
|
||||
.expect("graph_query_all");
|
||||
|
||||
eprintln!("\n--- graph_query_all returned {} rows ---", all_rows.len());
|
||||
|
||||
assert!(
|
||||
!all_rows.is_empty(),
|
||||
"graph_query_all should include namespace relations when no namespace filter is set"
|
||||
);
|
||||
|
||||
// At minimum, the all-query should contain the same rows as namespace
|
||||
assert!(
|
||||
all_rows.len() >= graph_rows.len(),
|
||||
"graph_query_all ({}) should return at least as many rows as namespace query ({})",
|
||||
all_rows.len(),
|
||||
graph_rows.len()
|
||||
);
|
||||
}
|
||||
|
||||
// ── Test: MemoryClient put_doc → background extraction → graph_query ──
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore] // Slow: background worker + 5s wait. Run with --ignored.
|
||||
async fn put_doc_background_extraction_then_graph_query() {
|
||||
let _ = env_logger::builder()
|
||||
.filter_level(log::LevelFilter::Debug)
|
||||
.is_test(true)
|
||||
.try_init();
|
||||
|
||||
let tmp = tempdir().expect("tempdir");
|
||||
let workspace_dir = tmp.path().join("workspace");
|
||||
std::fs::create_dir_all(&workspace_dir).unwrap();
|
||||
|
||||
let client = MemoryClient::from_workspace_dir(workspace_dir).expect("MemoryClient");
|
||||
|
||||
let namespace = "test-bg";
|
||||
let doc_id = client
|
||||
.put_doc(NamespaceDocumentInput {
|
||||
namespace: namespace.to_string(),
|
||||
key: "bg-test-doc".to_string(),
|
||||
title: "Background extraction test".to_string(),
|
||||
content: TEST_DOCUMENT.to_string(),
|
||||
source_type: "doc".to_string(),
|
||||
priority: "medium".to_string(),
|
||||
tags: Vec::new(),
|
||||
metadata: json!({}),
|
||||
category: "core".to_string(),
|
||||
session_id: None,
|
||||
document_id: None,
|
||||
})
|
||||
.await
|
||||
.expect("put_doc");
|
||||
|
||||
eprintln!("put_doc returned doc_id={doc_id}");
|
||||
|
||||
// Wait for the background ingestion worker to process the job.
|
||||
// The worker runs on a separate tokio task; give it time to complete.
|
||||
tokio::time::sleep(Duration::from_secs(5)).await;
|
||||
|
||||
// Query with namespace
|
||||
let ns_rows = client
|
||||
.graph_query(Some(namespace), None, None)
|
||||
.await
|
||||
.expect("graph_query with namespace");
|
||||
|
||||
eprintln!(
|
||||
"graph_query(Some({namespace})) returned {} rows",
|
||||
ns_rows.len()
|
||||
);
|
||||
|
||||
// Query without namespace (the fix: should include namespace data)
|
||||
let all_rows = client
|
||||
.graph_query(None, None, None)
|
||||
.await
|
||||
.expect("graph_query without namespace");
|
||||
|
||||
eprintln!("graph_query(None) returned {} rows", all_rows.len());
|
||||
|
||||
// The background worker uses the default config which tries to load the
|
||||
// ONNX model. On CI this may fail silently, yielding 0 relations. The
|
||||
// heuristic extractor still runs, so we usually get relations, but we
|
||||
// assert conservatively: if namespace query found rows, the all-query
|
||||
// must too.
|
||||
if !ns_rows.is_empty() {
|
||||
assert!(
|
||||
!all_rows.is_empty(),
|
||||
"graph_query(None) must return rows when graph_query(Some(ns)) does"
|
||||
);
|
||||
}
|
||||
|
||||
// Verify document was stored regardless
|
||||
let docs = client
|
||||
.list_documents(Some(namespace))
|
||||
.await
|
||||
.expect("list_documents");
|
||||
let doc_count = docs
|
||||
.get("documents")
|
||||
.and_then(|d| d.as_array())
|
||||
.map(|a| a.len())
|
||||
.unwrap_or(0);
|
||||
|
||||
eprintln!("Documents in namespace '{namespace}': {doc_count}");
|
||||
assert!(
|
||||
doc_count >= 1,
|
||||
"Expected at least 1 document after put_doc, got {doc_count}"
|
||||
);
|
||||
}
|
||||
Reference in New Issue
Block a user