diff --git a/memoria/crates/memoria-api/src/metrics_summary.rs b/memoria/crates/memoria-api/src/metrics_summary.rs index 78f8485..9f9324d 100644 --- a/memoria/crates/memoria-api/src/metrics_summary.rs +++ b/memoria/crates/memoria-api/src/metrics_summary.rs @@ -670,9 +670,9 @@ impl MetricsSummaryManager { let coverage_sql = format!( r#"SELECT COUNT(*) AS total_users, - SUM(CASE WHEN s.user_id IS NULL THEN 1 ELSE 0 END) AS missing_users, - SUM(CASE WHEN s.user_id IS NOT NULL AND s.has_pending = 1 THEN 1 ELSE 0 END) AS dirty_users, - SUM(CASE WHEN s.user_id IS NOT NULL AND s.has_pending = 0 THEN 1 ELSE 0 END) AS ready_users + COUNT(CASE WHEN s.user_id IS NULL THEN 1 END) AS missing_users, + COUNT(CASE WHEN s.user_id IS NOT NULL AND s.has_pending = 1 THEN 1 END) AS dirty_users, + COUNT(CASE WHEN s.user_id IS NOT NULL AND s.has_pending = 0 THEN 1 END) AS ready_users FROM mem_user_registry r LEFT JOIN ({DEDUPED_STATE_SUBQUERY}) s ON s.user_id = r.user_id WHERE r.status = 'active'"# @@ -684,7 +684,7 @@ impl MetricsSummaryManager { // Scalar families from rollups let scalar_rows = sqlx::query( - r#"SELECT r.family, COALESCE(SUM(r.value), 0) AS total + r#"SELECT r.family, CAST(COALESCE(SUM(r.value), 0) AS SIGNED) AS total FROM mem_metrics_user_rollups r INNER JOIN mem_user_registry u ON u.user_id = r.user_id WHERE u.status = 'active' AND r.bucket = '__total__' @@ -709,7 +709,7 @@ impl MetricsSummaryManager { .collect::>() .join(","); let sql = format!( - r#"SELECT r.family, r.bucket, COALESCE(SUM(r.value), 0) AS total + r#"SELECT r.family, r.bucket, CAST(COALESCE(SUM(r.value), 0) AS SIGNED) AS total FROM mem_metrics_user_rollups r INNER JOIN mem_user_registry u ON u.user_id = r.user_id WHERE u.status = 'active' AND r.family IN ({placeholders}) @@ -1540,10 +1540,14 @@ fn db_err(e: impl std::fmt::Display) -> MemoriaError { } fn optional_i64(row: &sqlx::mysql::MySqlRow, column: &str) -> i64 { - row.try_get::, _>(column) - .ok() - .flatten() - .unwrap_or(0) + match row.try_get::, _>(column) { + Ok(Some(value)) => value, + Ok(None) => 0, + Err(error) => { + warn!(column, error = %error, "failed to decode metrics aggregate"); + 0 + } + } } fn clamp_metric_age(age_secs: Option) -> Option { diff --git a/memoria/crates/memoria-api/src/routes/memory.rs b/memoria/crates/memoria-api/src/routes/memory.rs index b84bf44..becebb3 100644 --- a/memoria/crates/memoria-api/src/routes/memory.rs +++ b/memoria/crates/memoria-api/src/routes/memory.rs @@ -737,7 +737,7 @@ pub async fn get_profile( let stats_rows = if let Some(sid) = subject_id { let q = format!( "SELECT memory_type, COUNT(*) as cnt, \ - ROUND(AVG(initial_confidence), 2) as avg_conf, \ + CAST(ROUND(AVG(initial_confidence), 2) AS DOUBLE) as avg_conf, \ MIN(observed_at) as oldest, MAX(observed_at) as newest \ FROM {table} WHERE user_id = ? AND subject_id = ? AND is_active = 1 GROUP BY memory_type" ); @@ -745,7 +745,7 @@ pub async fn get_profile( } else { let q = format!( "SELECT memory_type, COUNT(*) as cnt, \ - ROUND(AVG(initial_confidence), 2) as avg_conf, \ + CAST(ROUND(AVG(initial_confidence), 2) AS DOUBLE) as avg_conf, \ MIN(observed_at) as oldest, MAX(observed_at) as newest \ FROM {table} WHERE user_id = ? AND is_active = 1 GROUP BY memory_type" ); @@ -764,7 +764,17 @@ pub async fn get_profile( let cnt: i64 = r.try_get("cnt").unwrap_or(0); by_type.insert(mt, serde_json::json!(cnt)); total += cnt; - if let Ok(c) = r.try_get::("avg_conf") { conf_sum += c * cnt as f64; conf_n += cnt; } + match r.try_get::, _>("avg_conf") { + Ok(Some(c)) => { + conf_sum += c * cnt as f64; + conf_n += cnt; + } + Ok(None) => {} + Err(error) => tracing::warn!( + error = %error, + "failed to decode profile average confidence" + ), + } if let Ok(Some(d)) = r.try_get::, _>("oldest") { let s = d.to_string(); if oldest.as_ref().is_none_or(|o| s < *o) { oldest = Some(s); } diff --git a/memoria/crates/memoria-api/tests/api_e2e.rs b/memoria/crates/memoria-api/tests/api_e2e.rs index ef1af09..0566fdb 100644 --- a/memoria/crates/memoria-api/tests/api_e2e.rs +++ b/memoria/crates/memoria-api/tests/api_e2e.rs @@ -4125,7 +4125,7 @@ async fn test_admin_trigger_governance() { #[tokio::test] async fn test_health_endpoints() { - let (base, client, _server) = spawn_server().await; + let (base, client, server) = spawn_server().await; let user = uid(); // Store some memories @@ -4139,6 +4139,15 @@ async fn test_health_endpoints() { .unwrap(); } + // Legacy or directly-written rows may contain NULL despite the column default. + // Aggregate endpoints must preserve that unknown value instead of failing to decode it. + let pool = server.user_db_pool(&user).await; + sqlx::query("UPDATE mem_memories SET initial_confidence = NULL WHERE user_id = ?") + .bind(&user) + .execute(&pool) + .await + .unwrap(); + // GET /v1/health/analyze let r = client .get(format!("{base}/v1/health/analyze")) @@ -4149,6 +4158,7 @@ async fn test_health_endpoints() { assert_eq!(r.status(), 200); let body: Value = r.json().await.unwrap(); assert!(body["semantic"]["total"].as_i64().unwrap() >= 3); + assert!(body["semantic"]["avg_confidence"].is_null()); println!("✅ health analyze: {body}"); // GET /v1/health/storage @@ -4164,6 +4174,18 @@ async fn test_health_endpoints() { assert!(body["active"].as_i64().unwrap() >= 3); println!("✅ health storage: {body}"); + // Profile statistics use a separate aggregate query and must follow the + // same all-NULL confidence semantics. + let r = client + .get(format!("{base}/v1/profiles/me")) + .header("X-User-Id", &user) + .send() + .await + .unwrap(); + assert_eq!(r.status(), 200); + let body: Value = r.json().await.unwrap(); + assert_eq!(body["stats"]["avg_confidence"].as_f64(), Some(0.0)); + // GET /v1/health/capacity let r = client .get(format!("{base}/v1/health/capacity")) diff --git a/memoria/crates/memoria-mcp/tests/feedback_e2e.rs b/memoria/crates/memoria-mcp/tests/feedback_e2e.rs index 1a37f34..6e43135 100644 --- a/memoria/crates/memoria-mcp/tests/feedback_e2e.rs +++ b/memoria/crates/memoria-mcp/tests/feedback_e2e.rs @@ -592,10 +592,10 @@ async fn test_feedback_db_verification() { let stats_row: (i64, i64, i64, i64, i64) = sqlx::query_as(&format!( "SELECT \ COUNT(*) as total, \ - SUM(CASE WHEN signal = 'useful' THEN 1 ELSE 0 END) as useful, \ - SUM(CASE WHEN signal = 'irrelevant' THEN 1 ELSE 0 END) as irrelevant, \ - SUM(CASE WHEN signal = 'outdated' THEN 1 ELSE 0 END) as outdated, \ - SUM(CASE WHEN signal = 'wrong' THEN 1 ELSE 0 END) as wrong \ + COUNT(CASE WHEN signal = 'useful' THEN 1 END) as useful, \ + COUNT(CASE WHEN signal = 'irrelevant' THEN 1 END) as irrelevant, \ + COUNT(CASE WHEN signal = 'outdated' THEN 1 END) as outdated, \ + COUNT(CASE WHEN signal = 'wrong' THEN 1 END) as wrong \ FROM {feedback_table} WHERE user_id = ?" )) .bind(&uid) diff --git a/memoria/crates/memoria-storage/src/store.rs b/memoria/crates/memoria-storage/src/store.rs index c10e49c..f5de83b 100644 --- a/memoria/crates/memoria-storage/src/store.rs +++ b/memoria/crates/memoria-storage/src/store.rs @@ -4386,10 +4386,10 @@ impl SqlMemoryStore { let row: (i64, i64, i64, i64, i64) = sqlx::query_as(&format!( "SELECT \ COUNT(*) as total, \ - COALESCE(SUM(CASE WHEN signal = 'useful' THEN 1 ELSE 0 END), 0) as useful, \ - COALESCE(SUM(CASE WHEN signal = 'irrelevant' THEN 1 ELSE 0 END), 0) as irrelevant, \ - COALESCE(SUM(CASE WHEN signal = 'outdated' THEN 1 ELSE 0 END), 0) as outdated, \ - COALESCE(SUM(CASE WHEN signal = 'wrong' THEN 1 ELSE 0 END), 0) as wrong \ + COUNT(CASE WHEN signal = 'useful' THEN 1 END) as useful, \ + COUNT(CASE WHEN signal = 'irrelevant' THEN 1 END) as irrelevant, \ + COUNT(CASE WHEN signal = 'outdated' THEN 1 END) as outdated, \ + COUNT(CASE WHEN signal = 'wrong' THEN 1 END) as wrong \ FROM {feedback_table} WHERE user_id = ?" )) .bind(user_id) @@ -4647,9 +4647,9 @@ impl SqlMemoryStore { ) -> Result { let mut conn = self.conn().await?; let memories_table = self.t("mem_memories"); - let row: (i64, Option) = sqlx::query_as(&format!( + let row: (i64, i64) = sqlx::query_as(&format!( "SELECT COUNT(*) as total_changes, \ - SUM(CASE WHEN superseded_by IS NOT NULL AND superseded_by != '' THEN 1 ELSE 0 END) as supersedes \ + COUNT(CASE WHEN superseded_by IS NOT NULL AND superseded_by != '' THEN 1 END) as supersedes \ FROM {memories_table} \ WHERE user_id = ? AND updated_at >= DATE_SUB(NOW(), INTERVAL ? HOUR)" )) @@ -4662,7 +4662,7 @@ impl SqlMemoryStore { if total == 0 { return Ok(false); } - Ok(supersedes.unwrap_or(0) as f64 / total as f64 > 0.3) + Ok(supersedes as f64 / total as f64 > 0.3) } /// Hygiene diagnostics: orphan counts and stale data that governance can clean. @@ -4807,10 +4807,11 @@ impl SqlMemoryStore { pub async fn health_analyze(&self, user_id: &str) -> Result { let mut conn = self.conn().await?; let memories_table = self.t("mem_memories"); - let rows: Vec<(String, i64, f64, i64, f64)> = sqlx::query_as(&format!( - "SELECT memory_type, COUNT(*) as total, AVG(initial_confidence) as avg_conf, \ + let rows: Vec<(String, i64, Option, i64, f64)> = sqlx::query_as(&format!( + "SELECT memory_type, COUNT(*) as total, \ + CAST(AVG(initial_confidence) AS DOUBLE) as avg_conf, \ COUNT(CASE WHEN superseded_by IS NOT NULL AND superseded_by != '' THEN 1 END) as superseded, \ - AVG(TIMESTAMPDIFF(HOUR, observed_at, NOW())) as avg_stale_h \ + CAST(AVG(TIMESTAMPDIFF(HOUR, observed_at, NOW())) AS DOUBLE) as avg_stale_h \ FROM {memories_table} WHERE user_id = ? GROUP BY memory_type" )) .bind(user_id) @@ -4847,8 +4848,8 @@ impl SqlMemoryStore { let memories_table = self.t("mem_memories"); let row: (i64, i64, f64) = sqlx::query_as(&format!( "SELECT COUNT(*) as total, \ - SUM(CASE WHEN is_active = 1 THEN 1 ELSE 0 END) as active, \ - AVG(LENGTH(content)) as avg_content_size \ + COUNT(CASE WHEN is_active = 1 THEN 1 END) as active, \ + CAST(COALESCE(AVG(LENGTH(content)), 0) AS DOUBLE) as avg_content_size \ FROM {memories_table} WHERE user_id = ?" )) .bind(user_id) @@ -5197,32 +5198,58 @@ impl SqlMemoryStore { .filter(|v| !v.is_empty()) // Some([]) → None → SQL NULL .map(vec_to_mo); - sqlx::query(&format!( + // MatrixOne 4.2 can retain a prepared parameter's NULL state across + // executions (matrixorigin/matrixone#26874). Keep nullable values out + // of bind parameters: + // each cached SQL shape then binds a value or contains a literal NULL, + // but never transitions the same parameter from NULL back to a value. + let nullable = |present| if present { "?" } else { "NULL" }; + let session_id = nullable_str(&memory.session_id); + let superseded_by = nullable_str(&memory.superseded_by); + let author_param = nullable(memory.author_id.is_some()); + let subject_param = nullable(memory.subject_id.is_some()); + let embedding_param = nullable(embedding.is_some()); + let session_param = nullable(session_id.is_some()); + let superseded_param = nullable(superseded_by.is_some()); + let sql = format!( r#"INSERT INTO {table} (memory_id, user_id, author_id, subject_id, memory_type, content, embedding, session_id, source_event_ids, extra_metadata, is_active, superseded_by, trust_tier, initial_confidence, observed_at, created_at, updated_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, ?, ?, ?, ?, ?, ?)"# - )) - .bind(&memory.memory_id) - .bind(&memory.user_id) - .bind(memory.author_id.as_deref()) - .bind(memory.subject_id.as_deref()) - .bind(memory.memory_type.to_string()) - .bind(&memory.content) - .bind(embedding) - .bind(nullable_str(&memory.session_id)) - .bind(source_event_ids) - .bind(extra_metadata) - .bind(nullable_str(&memory.superseded_by)) - .bind(memory.trust_tier.to_string()) - .bind(memory.initial_confidence as f32) - .bind(observed_at) - .bind(created_at) - .bind(now) - .execute(&self.pool) - .await - .map_err(db_err)?; + VALUES (?, ?, {author_param}, {subject_param}, ?, ?, {embedding_param}, + {session_param}, ?, ?, 1, {superseded_param}, ?, ?, ?, ?, ?)"# + ); + let mut query = sqlx::query(&sql) + .bind(&memory.memory_id) + .bind(&memory.user_id); + if let Some(author_id) = memory.author_id.as_deref() { + query = query.bind(author_id); + } + if let Some(subject_id) = memory.subject_id.as_deref() { + query = query.bind(subject_id); + } + query = query + .bind(memory.memory_type.to_string()) + .bind(&memory.content); + if let Some(embedding) = embedding { + query = query.bind(embedding); + } + if let Some(session_id) = session_id { + query = query.bind(session_id); + } + query = query.bind(source_event_ids).bind(extra_metadata); + if let Some(superseded_by) = superseded_by { + query = query.bind(superseded_by); + } + query + .bind(memory.trust_tier.to_string()) + .bind(memory.initial_confidence as f32) + .bind(observed_at) + .bind(created_at) + .bind(now) + .execute(&self.pool) + .await + .map_err(db_err)?; Ok(()) } @@ -5240,7 +5267,17 @@ impl SqlMemoryStore { for chunk in memories.chunks(50) { let placeholders = chunk .iter() - .map(|_| "(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, ?, ?, ?, ?, ?, ?)") + .map(|m| { + let nullable = |present| if present { "?" } else { "NULL" }; + format!( + "(?, ?, {}, {}, ?, ?, {}, {}, ?, ?, 1, {}, ?, ?, ?, ?, ?)", + nullable(m.author_id.is_some()), + nullable(m.subject_id.is_some()), + nullable(m.embedding.as_ref().is_some_and(|v| !v.is_empty())), + nullable(nullable_str(&m.session_id).is_some()), + nullable(nullable_str(&m.superseded_by).is_some()) + ) + }) .collect::>() .join(", "); let sql = format!( @@ -5267,18 +5304,27 @@ impl SqlMemoryStore { .as_deref() .filter(|v| !v.is_empty()) .map(vec_to_mo); + q = q.bind(m.memory_id.clone()).bind(m.user_id.clone()); + if let Some(author_id) = &m.author_id { + q = q.bind(author_id.clone()); + } + if let Some(subject_id) = &m.subject_id { + q = q.bind(subject_id.clone()); + } q = q - .bind(m.memory_id.clone()) - .bind(m.user_id.clone()) - .bind(m.author_id.clone()) - .bind(m.subject_id.clone()) .bind(m.memory_type.to_string()) - .bind(m.content.clone()) - .bind(embedding) - .bind(nullable_str(&m.session_id).map(str::to_string)) - .bind(source_event_ids) - .bind(extra_metadata) - .bind(nullable_str(&m.superseded_by).map(str::to_string)) + .bind(m.content.clone()); + if let Some(embedding) = embedding { + q = q.bind(embedding); + } + if let Some(session_id) = nullable_str(&m.session_id) { + q = q.bind(session_id.to_string()); + } + q = q.bind(source_event_ids).bind(extra_metadata); + if let Some(superseded_by) = nullable_str(&m.superseded_by) { + q = q.bind(superseded_by.to_string()); + } + q = q .bind(m.trust_tier.to_string()) .bind(m.initial_confidence as f32) .bind(observed_at) diff --git a/memoria/crates/memoria-storage/tests/store_crud.rs b/memoria/crates/memoria-storage/tests/store_crud.rs index a09fa56..1638d59 100644 --- a/memoria/crates/memoria-storage/tests/store_crud.rs +++ b/memoria/crates/memoria-storage/tests/store_crud.rs @@ -566,6 +566,88 @@ async fn test_null_optional_fields() { println!("✅ null_optional_fields: all NULLs round-trip correctly"); } +#[tokio::test] +async fn test_insert_normalizes_empty_nullable_strings() { + let (store, uid) = setup().await; + let id = format!("empty-nullable-{uid}"); + let mut memory = make_memory(&id, "empty nullable strings", &uid); + memory.session_id = Some(String::new()); + memory.superseded_by = Some(String::new()); + + store + .insert(&memory) + .await + .expect("insert empty nullable strings"); + + let stored: (Option, Option) = sqlx::query_as( + "SELECT session_id, superseded_by FROM mem_memories WHERE memory_id = ?", + ) + .bind(&id) + .fetch_one(store.pool()) + .await + .expect("read raw nullable strings"); + assert_eq!(stored, (None, None), "empty strings must be stored as NULL"); +} + +#[tokio::test] +async fn test_session_id_recovers_after_null_insert_on_same_connection() { + let (store, uid) = setup().await; + // Force all three INSERTs through one physical connection. MatrixOne + // matrixorigin/matrixone#26874 retained a prepared parameter's NULL state + // on statement reuse. + let store = store + .spawn_background_store(1) + .await + .expect("single-connection store"); + + let before_id = format!("session-before-{uid}"); + let null_id = format!("session-null-{uid}"); + let after_id = format!("session-after-{uid}"); + + let mut before = make_memory(&before_id, "session before null", &uid); + before.session_id = Some("sess-before".to_string()); + store.insert(&before).await.expect("insert before NULL"); + + let mut unscoped = make_memory(&null_id, "unscoped memory", &uid); + unscoped.session_id = None; + store.insert(&unscoped).await.expect("insert NULL session"); + + let mut after = make_memory(&after_id, "session after null", &uid); + after.session_id = Some("sess-after".to_string()); + store.insert(&after).await.expect("insert after NULL"); + + let before = store.get(&before_id).await.expect("get before").unwrap(); + let unscoped = store.get(&null_id).await.expect("get unscoped").unwrap(); + let after = store.get(&after_id).await.expect("get after").unwrap(); + + assert_eq!(before.session_id.as_deref(), Some("sess-before")); + assert!(unscoped.session_id.is_none()); + assert_eq!(after.session_id.as_deref(), Some("sess-after")); +} + +#[tokio::test] +async fn test_batch_insert_normalizes_empty_nullable_strings() { + let (store, uid) = setup().await; + let id = format!("batch-empty-nullable-{uid}"); + let mut memory = make_memory(&id, "batch empty nullable strings", &uid); + memory.session_id = Some(String::new()); + memory.superseded_by = Some(String::new()); + + store + .batch_insert_into("mem_memories", &[&memory]) + .await + .expect("batch insert empty nullable strings"); + + let stored: (Option, Option) = sqlx::query_as( + "SELECT session_id, superseded_by FROM mem_memories WHERE memory_id = ?", + ) + .bind(&id) + .fetch_one(store.pool()) + .await + .expect("read raw batch nullable strings"); + assert_eq!(stored, (None, None), "empty strings must be stored as NULL"); +} + // ── insert_entity_links batch optimization tests ───────────────────────────── #[tokio::test] diff --git a/memoria/crates/memoria-storage/tests/vector_index_ops.rs b/memoria/crates/memoria-storage/tests/vector_index_ops.rs index 0d46ad4..fa41cf4 100644 --- a/memoria/crates/memoria-storage/tests/vector_index_ops.rs +++ b/memoria/crates/memoria-storage/tests/vector_index_ops.rs @@ -638,13 +638,13 @@ async fn test_health_hygiene_does_not_misreport_entity_nodes() { } // ═══════════════════════════════════════════════════════════════════════════════ -// detect_pollution — empty result set (SUM returns NULL) +// detect_pollution — empty result set (conditional COUNT returns zero) // ═══════════════════════════════════════════════════════════════════════════════ #[tokio::test] async fn test_detect_pollution_empty_user() { let store = setup().await; - // Brand new user with zero memories — SUM(CASE...) returns NULL + // Brand new user with zero memories — conditional COUNT returns zero. let uid = uuid::Uuid::new_v4().to_string(); let result = store.detect_pollution(&uid, 24).await.unwrap(); assert!(!result, "empty user should not be flagged as polluted");