diff --git a/rust/examples/bench_handlers.rs b/rust/examples/bench_handlers.rs index 091a50d..5c29ed5 100644 --- a/rust/examples/bench_handlers.rs +++ b/rust/examples/bench_handlers.rs @@ -16,10 +16,10 @@ fn fmt_duration(d: std::time::Duration) -> String { #[tokio::main] async fn main() { - let url = std::env::var("DATABASE_URL") - .expect("Set DATABASE_URL to run benchmarks"); + let url = std::env::var("DATABASE_URL").expect("Set DATABASE_URL to run benchmarks"); - let pool = cuba_memorys::db::create_pool(&url).await + let pool = cuba_memorys::db::create_pool(&url) + .await .expect("Failed to connect"); println!("\n ═══ cuba-memorys Rust Benchmark ═══\n"); @@ -57,32 +57,44 @@ async fn main() { for i in 0..iterations { let name = format!("bench_alma_{i}_{}", uuid::Uuid::new_v4()); let _ = cuba_memorys::handlers::dispatch( - &pool, "cuba_alma", + &pool, + "cuba_alma", serde_json::json!({"action": "create", "name": &name, "entity_type": "concept"}), - ).await; + ) + .await; } let elapsed = start.elapsed(); - println!(" alma::create {:>8} /call ({iterations} calls)", fmt_duration(elapsed / iterations)); + println!( + " alma::create {:>8} /call ({iterations} calls)", + fmt_duration(elapsed / iterations) + ); } // ── 3. alma::get ────────────────────────────────────────────── { let name = format!("bench_get_{}", uuid::Uuid::new_v4()); let _ = cuba_memorys::handlers::dispatch( - &pool, "cuba_alma", + &pool, + "cuba_alma", serde_json::json!({"action": "create", "name": &name, "entity_type": "concept"}), - ).await; + ) + .await; let iterations = 100; let start = Instant::now(); for _ in 0..iterations { let _ = cuba_memorys::handlers::dispatch( - &pool, "cuba_alma", + &pool, + "cuba_alma", serde_json::json!({"action": "get", "name": &name}), - ).await; + ) + .await; } let elapsed = start.elapsed(); - println!(" alma::get {:>8} /call ({iterations} calls)", fmt_duration(elapsed / iterations)); + println!( + " alma::get {:>8} /call ({iterations} calls)", + fmt_duration(elapsed / iterations) + ); } // ── 4. cronica::add ─────────────────────────────────────────── @@ -92,7 +104,8 @@ async fn main() { let start = Instant::now(); for i in 0..iterations { let _ = cuba_memorys::handlers::dispatch( - &pool, "cuba_cronica", + &pool, + "cuba_cronica", serde_json::json!({ "action": "add", "entity_name": &name, @@ -100,10 +113,14 @@ async fn main() { "observation_type": "fact", "source": "agent" }), - ).await; + ) + .await; } let elapsed = start.elapsed(); - println!(" cronica::add {:>8} /call ({iterations} calls)", fmt_duration(elapsed / iterations)); + println!( + " cronica::add {:>8} /call ({iterations} calls)", + fmt_duration(elapsed / iterations) + ); } // ── 5. faro::search (hybrid) ────────────────────────────────── @@ -117,7 +134,10 @@ async fn main() { ).await; } let elapsed = start.elapsed(); - println!(" faro::search {:>8} /call ({iterations} calls)", fmt_duration(elapsed / iterations)); + println!( + " faro::search {:>8} /call ({iterations} calls)", + fmt_duration(elapsed / iterations) + ); } // ── 6. vigia::summary ───────────────────────────────────────── @@ -126,12 +146,17 @@ async fn main() { let start = Instant::now(); for _ in 0..iterations { let _ = cuba_memorys::handlers::dispatch( - &pool, "cuba_vigia", + &pool, + "cuba_vigia", serde_json::json!({"metric": "summary"}), - ).await; + ) + .await; } let elapsed = start.elapsed(); - println!(" vigia::summary {:>8} /call ({iterations} calls)", fmt_duration(elapsed / iterations)); + println!( + " vigia::summary {:>8} /call ({iterations} calls)", + fmt_duration(elapsed / iterations) + ); } println!("\n ═══ Benchmark Complete ═══\n"); diff --git a/rust/src/cognitive/density.rs b/rust/src/cognitive/density.rs index ed99a39..fc7771c 100644 --- a/rust/src/cognitive/density.rs +++ b/rust/src/cognitive/density.rs @@ -70,6 +70,9 @@ mod tests { fn test_density_mixed() { // Non-uniform distribution: "fast" appears 5x, rest 1x each → skewed entropy let d = information_density("fast fast fast fast fast safe modern language"); - assert!(d > 0.3 && d < 0.9, "skewed distribution should be medium: got {d}"); + assert!( + d > 0.3 && d < 0.9, + "skewed distribution should be medium: got {d}" + ); } } diff --git a/rust/src/cognitive/dual_strength.rs b/rust/src/cognitive/dual_strength.rs index 5778553..cfb5670 100644 --- a/rust/src/cognitive/dual_strength.rs +++ b/rust/src/cognitive/dual_strength.rs @@ -161,7 +161,10 @@ mod tests { fn test_sac_parameters_calibrated() { // V3: α=0.5, β=0.5: at SS=0, RS=0 → ΔSS = 0.5 * 1.0 * e^0 = 0.5 let s = increment_storage(0.0, 0.0); - assert!((s - 0.5).abs() < 0.001, "SS=0,RS=0 → ΔSS should be α: got {s}"); + assert!( + (s - 0.5).abs() < 0.001, + "SS=0,RS=0 → ΔSS should be α: got {s}" + ); // At SS=0, RS=1 → ΔSS = 0.5 * 1.0 * e^(-0.5) ≈ 0.5 * 0.6065 ≈ 0.303 let s = increment_storage(0.0, 1.0); diff --git a/rust/src/cognitive/fsrs.rs b/rust/src/cognitive/fsrs.rs index 0069061..8f53483 100644 --- a/rust/src/cognitive/fsrs.rs +++ b/rust/src/cognitive/fsrs.rs @@ -74,10 +74,7 @@ pub fn update_stability( let s = current_stability.max(0.01); match rating { - 0 => { - w[11] * d.powf(-w[12]) * ((s + 1.0).powf(w[13]) - 1.0) - * (w[14] * (1.0 - r)).exp() - } + 0 => w[11] * d.powf(-w[12]) * ((s + 1.0).powf(w[13]) - 1.0) * (w[14] * (1.0 - r)).exp(), _ => { let rating_bonus = match rating { 1 => w[15], @@ -182,14 +179,20 @@ mod tests { fn test_retrievability_custom_decay_slow() { let r_slow = retrievability_with_decay(1.0, 10.0, 0.1); let r_default = retrievability(1.0, 10.0); - assert!(r_slow > r_default, "slow decay should retain more: {r_slow} vs {r_default}"); + assert!( + r_slow > r_default, + "slow decay should retain more: {r_slow} vs {r_default}" + ); } #[test] fn test_retrievability_custom_decay_fast() { let r_fast = retrievability_with_decay(1.0, 10.0, 0.8); let r_default = retrievability(1.0, 10.0); - assert!(r_fast < r_default, "fast decay should retain less: {r_fast} vs {r_default}"); + assert!( + r_fast < r_default, + "fast decay should retain less: {r_fast} vs {r_default}" + ); } #[test] @@ -208,7 +211,10 @@ mod tests { fn test_adaptive_decay_rate_bounds() { for count in [0, 1, 5, 10, 30, 50, 100, 500, 1000] { let rate = adaptive_decay_rate(count); - assert!((0.1..=0.8).contains(&rate), "out of bounds at count={count}: {rate}"); + assert!( + (0.1..=0.8).contains(&rate), + "out of bounds at count={count}: {rate}" + ); } } @@ -221,7 +227,10 @@ mod tests { #[test] fn test_update_stability_forget() { let new_s = update_stability(10.0, 5.0, 0.3, 0); - assert!(new_s < 10.0, "stability should decrease on forget, got {new_s}"); + assert!( + new_s < 10.0, + "stability should decrease on forget, got {new_s}" + ); assert!(new_s > 0.0, "stability should remain positive"); } @@ -238,7 +247,10 @@ mod tests { // High PageRank → stability multiplied up let base_s = 10.0; let s_hub = apply_topological_inertia(base_s, 0.5); - assert!(s_hub > base_s, "hub should have higher stability: {s_hub} vs {base_s}"); + assert!( + s_hub > base_s, + "hub should have higher stability: {s_hub} vs {base_s}" + ); } #[test] @@ -257,7 +269,10 @@ mod tests { // PageRank=0 → ln(1+0) = 0 → no boost let base_s = 10.0; let s = apply_topological_inertia(base_s, 0.0); - assert!((s - base_s).abs() < 1e-10, "zero PR should have no boost: {s}"); + assert!( + (s - base_s).abs() < 1e-10, + "zero PR should have no boost: {s}" + ); } #[test] diff --git a/rust/src/cognitive/hebbian.rs b/rust/src/cognitive/hebbian.rs index 6879f54..9c911a6 100644 --- a/rust/src/cognitive/hebbian.rs +++ b/rust/src/cognitive/hebbian.rs @@ -14,8 +14,7 @@ use anyhow::Result; use sqlx::PgPool; use crate::constants::{ - HEBBIAN_ACCESS_BOOST, HEBBIAN_SEARCH_BOOST, HEBBIAN_OJA_RATE, - BCM_THROTTLE_SCALE, + BCM_THROTTLE_SCALE, HEBBIAN_ACCESS_BOOST, HEBBIAN_OJA_RATE, HEBBIAN_SEARCH_BOOST, }; // ── BCM V2: Dynamic Sliding Threshold ──────────────────────────── @@ -113,7 +112,7 @@ pub async fn oja_boost(pool: &PgPool, observation_id: uuid::Uuid, positive: bool "UPDATE brain_observations SET importance = LEAST(importance + $1, 1.0), updated_at = NOW() - WHERE id = $2" + WHERE id = $2", ) .bind(HEBBIAN_OJA_RATE) .bind(observation_id) @@ -124,7 +123,7 @@ pub async fn oja_boost(pool: &PgPool, observation_id: uuid::Uuid, positive: bool "UPDATE brain_observations SET importance = GREATEST(importance * 0.8, 0.0), updated_at = NOW() - WHERE id = $2" + WHERE id = $2", ) .bind(observation_id) .execute(pool) @@ -134,12 +133,16 @@ pub async fn oja_boost(pool: &PgPool, observation_id: uuid::Uuid, positive: bool } /// Strengthen relation on traversal (Hebbian synapse strengthening). -pub async fn strengthen_relation(pool: &PgPool, from_entity: uuid::Uuid, to_entity: uuid::Uuid) -> Result<()> { +pub async fn strengthen_relation( + pool: &PgPool, + from_entity: uuid::Uuid, + to_entity: uuid::Uuid, +) -> Result<()> { sqlx::query( "UPDATE brain_relations SET strength = LEAST(strength + $1, 1.0), updated_at = NOW() - WHERE from_entity = $2 AND to_entity = $3" + WHERE from_entity = $2 AND to_entity = $3", ) .bind(HEBBIAN_OJA_RATE) .bind(from_entity) @@ -162,7 +165,7 @@ pub async fn boost_neighbors(pool: &PgPool, entity_id: uuid::Uuid) -> Result 0.004, "low activity should give decent boost: {boost}"); + assert!( + boost > 0.004, + "low activity should give decent boost: {boost}" + ); } #[test] @@ -215,7 +224,10 @@ mod tests { // Many accesses → fully self-adjusted throttle // count=100, θ=100 → ratio=1.0 → throttle = max(0.1, 1.0 - 0.8) = 0.2 let boost = bcm_throttle_dynamic(0.01, 100, 100.0); - assert!(boost > 0.001 && boost < 0.005, "high activity → moderate throttle: {boost}"); + assert!( + boost > 0.001 && boost < 0.005, + "high activity → moderate throttle: {boost}" + ); } #[test] diff --git a/rust/src/cognitive/prediction_error.rs b/rust/src/cognitive/prediction_error.rs index 0f57d70..4a2c27d 100644 --- a/rust/src/cognitive/prediction_error.rs +++ b/rust/src/cognitive/prediction_error.rs @@ -12,10 +12,7 @@ //! Create: z ≤ 1σ (bottom 84.1%) //! Handles vector space anisotropy better than static thresholds. -use crate::constants::{ - PRED_ERROR_REINFORCE, - PRED_ERROR_UPDATE, -}; +use crate::constants::{PRED_ERROR_REINFORCE, PRED_ERROR_UPDATE}; /// Action to take based on prediction error. #[derive(Debug, Clone, PartialEq)] @@ -84,9 +81,11 @@ pub fn adaptive_thresholds_zscore(recent_similarities: &[f64]) -> (f64, f64) { let n = recent_similarities.len() as f64; let mean = recent_similarities.iter().sum::() / n; - let variance = recent_similarities.iter() + let variance = recent_similarities + .iter() .map(|x| (x - mean).powi(2)) - .sum::() / n; + .sum::() + / n; let sigma = variance.sqrt(); // V5.2: Z-score thresholds — μ + Nσ diff --git a/rust/src/cognitive/spreading.rs b/rust/src/cognitive/spreading.rs index 6adf024..a1536b1 100644 --- a/rust/src/cognitive/spreading.rs +++ b/rust/src/cognitive/spreading.rs @@ -23,7 +23,7 @@ pub async fn neighbor_diffusion(pool: &PgPool) -> Result<()> { "SELECT id, importance FROM brain_entities WHERE importance > 0.5 ORDER BY importance DESC - LIMIT 20" + LIMIT 20", ) .fetch_all(pool) .await?; @@ -39,7 +39,7 @@ pub async fn neighbor_diffusion(pool: &PgPool) -> Result<()> { "SELECT CASE WHEN from_entity = $1 THEN to_entity ELSE from_entity END, strength FROM brain_relations - WHERE from_entity = $1 OR to_entity = $1" + WHERE from_entity = $1 OR to_entity = $1", ) .bind(seed_id) .fetch_all(pool) @@ -59,7 +59,7 @@ pub async fn neighbor_diffusion(pool: &PgPool) -> Result<()> { "UPDATE brain_entities SET importance = LEAST(importance + $1, 1.0), updated_at = NOW() - WHERE id = $2" + WHERE id = $2", ) .bind(weighted_boost) .bind(neighbor_id) diff --git a/rust/src/constants.rs b/rust/src/constants.rs index 9af3fdb..bbc530d 100644 --- a/rust/src/constants.rs +++ b/rust/src/constants.rs @@ -8,12 +8,12 @@ use serde_json::Value; /// Default FSRS-6 parameters (Ye 2024) — 21 parameters. pub const FSRS6_DEFAULT_PARAMS: [f64; 21] = [ - 0.40255, 1.18385, 3.173, 15.69105, // w[0..3]: initial stability - 7.1949, 0.5345, 1.4604, 0.0046, // w[4..7]: difficulty - 1.54575, 0.1192, 1.01925, 1.9395, // w[8..11]: recall - 0.11, 0.29605, 2.2698, 0.2315, // w[12..15]: forget - 2.9898, 0.51655, 0.6621, // w[16..18]: review - 0.0, 0.0, // w[19..20]: reserved + 0.40255, 1.18385, 3.173, 15.69105, // w[0..3]: initial stability + 7.1949, 0.5345, 1.4604, 0.0046, // w[4..7]: difficulty + 1.54575, 0.1192, 1.01925, 1.9395, // w[8..11]: recall + 0.11, 0.29605, 2.2698, 0.2315, // w[12..15]: forget + 2.9898, 0.51655, 0.6621, // w[16..18]: review + 0.0, 0.0, // w[19..20]: reserved ]; /// Factor for desired retention → retrievability threshold. @@ -27,7 +27,7 @@ pub const DEDUP_THRESHOLD: f64 = 0.85; /// Prediction Error Gating thresholds (V5 — Vestige-inspired). pub const PRED_ERROR_REINFORCE: f64 = 0.92; // Very similar → reinforce existing -pub const PRED_ERROR_UPDATE: f64 = 0.75; // Somewhat similar → update existing +pub const PRED_ERROR_UPDATE: f64 = 0.75; // Somewhat similar → update existing // Below PRED_ERROR_UPDATE → create new observation /// Cache configuration. @@ -42,7 +42,7 @@ pub const COMMUNITY_SUMMARY_CAP: usize = 30; /// Hebbian boost constants. pub const HEBBIAN_ACCESS_BOOST: f64 = 0.01; pub const HEBBIAN_SEARCH_BOOST: f64 = 0.02; // VF2: Testing Effect -pub const HEBBIAN_OJA_RATE: f64 = 0.05; // Oja's learning rate +pub const HEBBIAN_OJA_RATE: f64 = 0.05; // Oja's learning rate pub const HEBBIAN_MAX_IMPORTANCE: f64 = 1.0; /// BCM Metaplasticity (Bienenstock-Cooper-Munro, 1982). @@ -64,24 +64,39 @@ pub const DEFAULT_DECAY_RATE: f64 = 0.5; // Adaptive RRF_K_MIN/MAX removed per Gemini Deep Research audit 2026-03-14. /// Relation types. -pub const VALID_RELATION_TYPES: &[&str] = &[ - "uses", "causes", "implements", "depends_on", "related_to", -]; +pub const VALID_RELATION_TYPES: &[&str] = + &["uses", "causes", "implements", "depends_on", "related_to"]; /// Entity types. pub const VALID_ENTITY_TYPES: &[&str] = &[ - "concept", "project", "technology", "person", "pattern", "config", + "concept", + "project", + "technology", + "person", + "pattern", + "config", ]; /// Observation types. pub const VALID_OBSERVATION_TYPES: &[&str] = &[ - "fact", "decision", "lesson", "preference", - "error", "solution", "context", "tool_usage", "superseded", + "fact", + "decision", + "lesson", + "preference", + "error", + "solution", + "context", + "tool_usage", + "superseded", ]; /// Observation sources. pub const VALID_SOURCES: &[&str] = &[ - "agent", "error_detection", "user", "consolidation", "inference", + "agent", + "error_detection", + "user", + "consolidation", + "inference", ]; // ── Tool Definitions ───────────────────────────────────────────── @@ -89,138 +104,190 @@ pub const VALID_SOURCES: &[&str] = &[ /// Generate MCP tool definitions for tools/list response. pub fn tool_definitions() -> Vec { vec![ - tool_def("cuba_alma", "CRUD knowledge graph entities (concepts, projects, technologies, patterns, people). Auto-boosts neighbors on access. For transient info use cuba_cronica instead.", serde_json::json!({ - "type": "object", - "properties": { - "action": {"type": "string", "enum": ["create", "update", "delete", "get"], "description": "Operation to perform"}, - "name": {"type": "string", "description": "Entity name (unique identifier)"}, - "entity_type": {"type": "string", "description": "Type: concept, project, technology, person, pattern, config"}, - "new_name": {"type": "string", "description": "New name for update action"} - }, - "required": ["action", "name"] - })), - tool_def("cuba_cronica", "Attach facts/lessons/decisions to entities. Auto-creates entity if not found. Dedup gate blocks near-duplicates. Contradictions auto-supersede old facts. Use batch_add with 'observations' array for bulk writes.", serde_json::json!({ - "type": "object", - "properties": { - "action": {"type": "string", "enum": ["add", "delete", "list", "batch_add"], "description": "Operation to perform. batch_add accepts 'observations' array."}, - "entity_name": {"type": "string", "description": "Entity to attach observation to"}, - "content": {"type": "string", "description": "Observation text"}, - "observation_type": {"type": "string", "enum": ["fact", "decision", "lesson", "preference", "context", "tool_usage"], "description": "Type of observation"}, - "source": {"type": "string", "enum": ["agent", "user", "error_detection"], "description": "Who/what created this observation"} - }, - "required": ["action", "entity_name"] - })), - tool_def("cuba_faro", "Search memory BEFORE answering to ground responses. Returns grounding scores. Mode 'verify' checks claims against evidence (confidence: verified/partial/weak/unknown). Session-aware: boosts results matching active session goals.", serde_json::json!({ - "type": "object", - "properties": { - "query": {"type": "string", "description": "Search text"}, - "mode": {"type": "string", "enum": ["hybrid", "verify"], "description": "Search mode (default: hybrid). 'verify' checks if claim is grounded."}, - "scope": {"type": "string", "enum": ["all", "entities", "observations", "errors"], "description": "Where to search (default: all)"}, - "limit": {"type": "integer", "description": "Max results (default 10, max 50)"} - }, - "required": ["query"] - })), - tool_def("cuba_puente", "Create edges between entities (uses, causes, implements, depends_on, related_to). 'traverse' explores connections, 'infer' does transitive reasoning (A→B→C). Relations strengthen with use (Hebbian).", serde_json::json!({ - "type": "object", - "properties": { - "action": {"type": "string", "enum": ["create", "delete", "traverse", "infer"], "description": "Operation to perform"}, - "from_entity": {"type": "string", "description": "Source entity name"}, - "to_entity": {"type": "string", "description": "Target entity name"}, - "relation_type": {"type": "string", "description": "Relation: uses, causes, implements, depends_on, related_to"}, - "bidirectional": {"type": "boolean", "description": "If true, relation goes both ways"}, - "start_entity": {"type": "string", "description": "Start point for traverse/infer"}, - "max_depth": {"type": "integer", "description": "Max hops for traverse/infer (default 3, max 5)"} - }, - "required": ["action"] - })), - tool_def("cuba_eco", "RLHF feedback: positive boosts importance (Oja's rule), negative decreases, correct updates content.", serde_json::json!({ - "type": "object", - "properties": { - "action": {"type": "string", "enum": ["positive", "negative", "correct"], "description": "Feedback type"}, - "entity_name": {"type": "string", "description": "Target entity"}, - "observation_id": {"type": "string", "description": "Target observation UUID"}, - "correction": {"type": "string", "description": "New content (for correct action)"} - }, - "required": ["action"] - })), - tool_def("cuba_alarma", "Report errors immediately. Auto-detects patterns (≥3 similar = warning). Hebbian: similar errors get boosted for easier retrieval.", serde_json::json!({ - "type": "object", - "properties": { - "error_type": {"type": "string", "description": "Error category: TypeError, ConnectionError, etc."}, - "error_message": {"type": "string", "description": "Full error message"}, - "context": {"type": "object", "description": "Context: {file, function, stack_trace, line}"}, - "project": {"type": "string", "description": "Project name (default: 'default')"} - }, - "required": ["error_type", "error_message"] - })), - tool_def("cuba_remedio", "Mark an error as resolved with solution. Cross-references similar unresolved errors.", serde_json::json!({ - "type": "object", - "properties": { - "error_id": {"type": "string", "description": "UUID of the error to solve"}, - "solution": {"type": "string", "description": "Solution that fixed the error"} - }, - "required": ["error_id", "solution"] - })), - tool_def("cuba_expediente", "Search past errors/solutions. Use 'proposed_action' as anti-repetition guard: warns if similar approach previously failed.", serde_json::json!({ - "type": "object", - "properties": { - "query": {"type": "string", "description": "Search text for errors"}, - "project": {"type": "string", "description": "Filter by project"}, - "resolved_only": {"type": "boolean", "description": "Only return errors with solutions"}, - "proposed_action": {"type": "string", "description": "Anti-repetition: describe what you plan to do. Returns warning if similar approach failed before."} - }, - "required": ["query"] - })), - tool_def("cuba_jornada", "Track working sessions with goals and outcomes.", serde_json::json!({ - "type": "object", - "properties": { - "action": {"type": "string", "enum": ["start", "end", "list", "current"], "description": "Session action"}, - "name": {"type": "string", "description": "Session name (for start)"}, - "goals": {"type": "array", "items": {"type": "string"}, "description": "Session goals (for start)"}, - "outcome": {"type": "string", "enum": ["success", "partial", "failed", "abandoned"], "description": "Session outcome (for end)"}, - "summary": {"type": "string", "description": "What was accomplished (for end)"} - }, - "required": ["action"] - })), - tool_def("cuba_decreto", "Record and query architecture/design decisions.", serde_json::json!({ - "type": "object", - "properties": { - "action": {"type": "string", "enum": ["record", "query", "list"], "description": "Decision action"}, - "title": {"type": "string", "description": "Decision title (for record)"}, - "context": {"type": "string", "description": "Why this decision was needed"}, - "alternatives": {"type": "array", "items": {"type": "string"}, "description": "Options considered"}, - "chosen": {"type": "string", "description": "Option chosen"}, - "rationale": {"type": "string", "description": "Why this option was chosen"}, - "query": {"type": "string", "description": "Search text (for query action)"} - }, - "required": ["action"] - })), - tool_def("cuba_vigia", "Knowledge graph analytics: summary (counts + token estimate), health (staleness, entropy, DB size), drift (chi-squared on errors), communities (Leiden), bridges (betweenness centrality).", serde_json::json!({ - "type": "object", - "properties": { - "metric": {"type": "string", "enum": ["summary", "health", "drift", "communities", "bridges"], "description": "Metric to compute"} - }, - "required": ["metric"] - })), - tool_def("cuba_zafra", "Memory maintenance: decay (FSRS adaptive), prune (remove low-importance), merge (deduplicate), summarize (compress observations), pagerank (personalized importance), find_duplicates, export, stats.", serde_json::json!({ - "type": "object", - "properties": { - "action": {"type": "string", "enum": ["decay", "prune", "merge", "summarize", "stats", "pagerank", "find_duplicates", "export", "backfill"], "description": "Consolidation action"}, - "entity_name": {"type": "string", "description": "Entity to summarize (for summarize action)"}, - "compressed_summary": {"type": "string", "description": "Compressed text replacing observations (for summarize)"}, - "threshold": {"type": "number", "description": "Importance threshold for prune (default 0.1)"}, - "similarity_threshold": {"type": "number", "description": "Similarity threshold for merge (default 0.8)"} - }, - "required": ["action"] - })), - tool_def("cuba_forget", "GDPR Right to Erasure: cascading hard-delete of an entity and ALL references across observations, relations, errors, and sessions. IRREVERSIBLE. Requires confirm=true.", serde_json::json!({ - "type": "object", - "properties": { - "entity_name": {"type": "string", "description": "Entity name to erase completely"}, - "confirm": {"type": "boolean", "description": "Must be true to proceed (safety gate)"} - }, - "required": ["entity_name", "confirm"] - })), + tool_def( + "cuba_alma", + "CRUD knowledge graph entities (concepts, projects, technologies, patterns, people). Auto-boosts neighbors on access. For transient info use cuba_cronica instead.", + serde_json::json!({ + "type": "object", + "properties": { + "action": {"type": "string", "enum": ["create", "update", "delete", "get"], "description": "Operation to perform"}, + "name": {"type": "string", "description": "Entity name (unique identifier)"}, + "entity_type": {"type": "string", "description": "Type: concept, project, technology, person, pattern, config"}, + "new_name": {"type": "string", "description": "New name for update action"} + }, + "required": ["action", "name"] + }), + ), + tool_def( + "cuba_cronica", + "Attach facts/lessons/decisions to entities. Auto-creates entity if not found. Dedup gate blocks near-duplicates. Contradictions auto-supersede old facts. Use batch_add with 'observations' array for bulk writes.", + serde_json::json!({ + "type": "object", + "properties": { + "action": {"type": "string", "enum": ["add", "delete", "list", "batch_add"], "description": "Operation to perform. batch_add accepts 'observations' array."}, + "entity_name": {"type": "string", "description": "Entity to attach observation to"}, + "content": {"type": "string", "description": "Observation text"}, + "observation_type": {"type": "string", "enum": ["fact", "decision", "lesson", "preference", "context", "tool_usage"], "description": "Type of observation"}, + "source": {"type": "string", "enum": ["agent", "user", "error_detection"], "description": "Who/what created this observation"} + }, + "required": ["action", "entity_name"] + }), + ), + tool_def( + "cuba_faro", + "Search memory BEFORE answering to ground responses. Returns grounding scores. Mode 'verify' checks claims against evidence (confidence: verified/partial/weak/unknown). Session-aware: boosts results matching active session goals.", + serde_json::json!({ + "type": "object", + "properties": { + "query": {"type": "string", "description": "Search text"}, + "mode": {"type": "string", "enum": ["hybrid", "verify"], "description": "Search mode (default: hybrid). 'verify' checks if claim is grounded."}, + "scope": {"type": "string", "enum": ["all", "entities", "observations", "errors"], "description": "Where to search (default: all)"}, + "limit": {"type": "integer", "description": "Max results (default 10, max 50)"} + }, + "required": ["query"] + }), + ), + tool_def( + "cuba_puente", + "Create edges between entities (uses, causes, implements, depends_on, related_to). 'traverse' explores connections, 'infer' does transitive reasoning (A→B→C). Relations strengthen with use (Hebbian).", + serde_json::json!({ + "type": "object", + "properties": { + "action": {"type": "string", "enum": ["create", "delete", "traverse", "infer"], "description": "Operation to perform"}, + "from_entity": {"type": "string", "description": "Source entity name"}, + "to_entity": {"type": "string", "description": "Target entity name"}, + "relation_type": {"type": "string", "description": "Relation: uses, causes, implements, depends_on, related_to"}, + "bidirectional": {"type": "boolean", "description": "If true, relation goes both ways"}, + "start_entity": {"type": "string", "description": "Start point for traverse/infer"}, + "max_depth": {"type": "integer", "description": "Max hops for traverse/infer (default 3, max 5)"} + }, + "required": ["action"] + }), + ), + tool_def( + "cuba_eco", + "RLHF feedback: positive boosts importance (Oja's rule), negative decreases, correct updates content.", + serde_json::json!({ + "type": "object", + "properties": { + "action": {"type": "string", "enum": ["positive", "negative", "correct"], "description": "Feedback type"}, + "entity_name": {"type": "string", "description": "Target entity"}, + "observation_id": {"type": "string", "description": "Target observation UUID"}, + "correction": {"type": "string", "description": "New content (for correct action)"} + }, + "required": ["action"] + }), + ), + tool_def( + "cuba_alarma", + "Report errors immediately. Auto-detects patterns (≥3 similar = warning). Hebbian: similar errors get boosted for easier retrieval.", + serde_json::json!({ + "type": "object", + "properties": { + "error_type": {"type": "string", "description": "Error category: TypeError, ConnectionError, etc."}, + "error_message": {"type": "string", "description": "Full error message"}, + "context": {"type": "object", "description": "Context: {file, function, stack_trace, line}"}, + "project": {"type": "string", "description": "Project name (default: 'default')"} + }, + "required": ["error_type", "error_message"] + }), + ), + tool_def( + "cuba_remedio", + "Mark an error as resolved with solution. Cross-references similar unresolved errors.", + serde_json::json!({ + "type": "object", + "properties": { + "error_id": {"type": "string", "description": "UUID of the error to solve"}, + "solution": {"type": "string", "description": "Solution that fixed the error"} + }, + "required": ["error_id", "solution"] + }), + ), + tool_def( + "cuba_expediente", + "Search past errors/solutions. Use 'proposed_action' as anti-repetition guard: warns if similar approach previously failed.", + serde_json::json!({ + "type": "object", + "properties": { + "query": {"type": "string", "description": "Search text for errors"}, + "project": {"type": "string", "description": "Filter by project"}, + "resolved_only": {"type": "boolean", "description": "Only return errors with solutions"}, + "proposed_action": {"type": "string", "description": "Anti-repetition: describe what you plan to do. Returns warning if similar approach failed before."} + }, + "required": ["query"] + }), + ), + tool_def( + "cuba_jornada", + "Track working sessions with goals and outcomes.", + serde_json::json!({ + "type": "object", + "properties": { + "action": {"type": "string", "enum": ["start", "end", "list", "current"], "description": "Session action"}, + "name": {"type": "string", "description": "Session name (for start)"}, + "goals": {"type": "array", "items": {"type": "string"}, "description": "Session goals (for start)"}, + "outcome": {"type": "string", "enum": ["success", "partial", "failed", "abandoned"], "description": "Session outcome (for end)"}, + "summary": {"type": "string", "description": "What was accomplished (for end)"} + }, + "required": ["action"] + }), + ), + tool_def( + "cuba_decreto", + "Record and query architecture/design decisions.", + serde_json::json!({ + "type": "object", + "properties": { + "action": {"type": "string", "enum": ["record", "query", "list"], "description": "Decision action"}, + "title": {"type": "string", "description": "Decision title (for record)"}, + "context": {"type": "string", "description": "Why this decision was needed"}, + "alternatives": {"type": "array", "items": {"type": "string"}, "description": "Options considered"}, + "chosen": {"type": "string", "description": "Option chosen"}, + "rationale": {"type": "string", "description": "Why this option was chosen"}, + "query": {"type": "string", "description": "Search text (for query action)"} + }, + "required": ["action"] + }), + ), + tool_def( + "cuba_vigia", + "Knowledge graph analytics: summary (counts + token estimate), health (staleness, entropy, DB size), drift (chi-squared on errors), communities (Leiden), bridges (betweenness centrality).", + serde_json::json!({ + "type": "object", + "properties": { + "metric": {"type": "string", "enum": ["summary", "health", "drift", "communities", "bridges"], "description": "Metric to compute"} + }, + "required": ["metric"] + }), + ), + tool_def( + "cuba_zafra", + "Memory maintenance: decay (FSRS adaptive), prune (remove low-importance), merge (deduplicate), summarize (compress observations), pagerank (personalized importance), find_duplicates, export, stats.", + serde_json::json!({ + "type": "object", + "properties": { + "action": {"type": "string", "enum": ["decay", "prune", "merge", "summarize", "stats", "pagerank", "find_duplicates", "export", "backfill"], "description": "Consolidation action"}, + "entity_name": {"type": "string", "description": "Entity to summarize (for summarize action)"}, + "compressed_summary": {"type": "string", "description": "Compressed text replacing observations (for summarize)"}, + "threshold": {"type": "number", "description": "Importance threshold for prune (default 0.1)"}, + "similarity_threshold": {"type": "number", "description": "Similarity threshold for merge (default 0.8)"} + }, + "required": ["action"] + }), + ), + tool_def( + "cuba_forget", + "GDPR Right to Erasure: cascading hard-delete of an entity and ALL references across observations, relations, errors, and sessions. IRREVERSIBLE. Requires confirm=true.", + serde_json::json!({ + "type": "object", + "properties": { + "entity_name": {"type": "string", "description": "Entity name to erase completely"}, + "confirm": {"type": "boolean", "description": "Must be true to proceed (safety gate)"} + }, + "required": ["entity_name", "confirm"] + }), + ), ] } diff --git a/rust/src/db.rs b/rust/src/db.rs index 5298498..57c9eae 100644 --- a/rust/src/db.rs +++ b/rust/src/db.rs @@ -108,11 +108,10 @@ async fn init_schema(pool: &PgPool) -> Result<()> { tracing::info!("BCM theta column verified"); // Check pgvector extension - let pgvector_check: Option<(String,)> = sqlx::query_as( - "SELECT extname::text FROM pg_extension WHERE extname = 'vector'" - ) - .fetch_optional(pool) - .await?; + let pgvector_check: Option<(String,)> = + sqlx::query_as("SELECT extname::text FROM pg_extension WHERE extname = 'vector'") + .fetch_optional(pool) + .await?; if pgvector_check.is_some() { tracing::info!("pgvector extension detected"); diff --git a/rust/src/embeddings/onnx.rs b/rust/src/embeddings/onnx.rs index d017597..678c1db 100644 --- a/rust/src/embeddings/onnx.rs +++ b/rust/src/embeddings/onnx.rs @@ -7,12 +7,12 @@ //! model_quantized.onnx and tokenizer.json. //! If not set, falls back to deterministic hash-based embeddings for testing. +use crate::search::cache::TtlLruCache; use anyhow::{Context, Result}; use ort::session::Session; use ort::session::builder::GraphOptimizationLevel; use std::path::PathBuf; use std::sync::OnceLock; -use crate::search::cache::TtlLruCache; /// Embedding dimension (BGE-small-en-v1.5 uses 384-d vectors). pub const EMBEDDING_DIM: usize = 384; @@ -86,14 +86,16 @@ fn init_onnx_session(model_file: &PathBuf, model_dir: &PathBuf) -> Result<()> { .commit_from_file(model_file) .map_err(|e| anyhow::anyhow!("load model: {e}"))?; - ONNX_SESSION.set(std::sync::Mutex::new(session)) + ONNX_SESSION + .set(std::sync::Mutex::new(session)) .map_err(|_| anyhow::anyhow!("ONNX session already initialized"))?; // Load tokenizer let tokenizer_dir = if model_dir.is_dir() { model_dir.clone() } else { - model_dir.parent() + model_dir + .parent() .map(|p| p.to_path_buf()) .unwrap_or_else(|| PathBuf::from(".")) }; @@ -107,7 +109,8 @@ fn init_onnx_session(model_file: &PathBuf, model_dir: &PathBuf) -> Result<()> { max_length: 512, ..Default::default() }; - tokenizer.with_truncation(Some(truncation)) + tokenizer + .with_truncation(Some(truncation)) .map_err(|e| anyhow::anyhow!("failed to set truncation: {e}"))?; let padding = tokenizers::PaddingParams { @@ -116,10 +119,14 @@ fn init_onnx_session(model_file: &PathBuf, model_dir: &PathBuf) -> Result<()> { }; tokenizer.with_padding(Some(padding)); - TOKENIZER.set(tokenizer) + TOKENIZER + .set(tokenizer) .map_err(|_| anyhow::anyhow!("Tokenizer already initialized"))?; } else { - return Err(anyhow::anyhow!("tokenizer.json not found at {}", tokenizer_path.display())); + return Err(anyhow::anyhow!( + "tokenizer.json not found at {}", + tokenizer_path.display() + )); } Ok(()) @@ -135,17 +142,16 @@ pub async fn embed(text: &str) -> Result> { // Check cache first let cache_key = text.to_string(); if let Ok(mut cache) = get_cache().lock() - && let Some(cached) = cache.get(&cache_key) { - return Ok(cached); - } + && let Some(cached) = cache.get(&cache_key) + { + return Ok(cached); + } // FIX B2: spawn_blocking for CPU-bound work let text_owned = text.to_string(); - let embedding = tokio::task::spawn_blocking(move || { - compute_embedding(&text_owned) - }) - .await - .context("embedding task panicked")??; + let embedding = tokio::task::spawn_blocking(move || compute_embedding(&text_owned)) + .await + .context("embedding task panicked")??; // Store in cache if let Ok(mut cache) = get_cache().lock() { @@ -161,12 +167,8 @@ pub async fn embed(text: &str) -> Result> { /// deterministic hash-based fallback for testing. fn compute_embedding(text: &str) -> Result> { match get_model_status() { - ModelStatus::Loaded => { - compute_onnx_embedding(text) - } - ModelStatus::Fallback => { - compute_hash_embedding(text) - } + ModelStatus::Loaded => compute_onnx_embedding(text), + ModelStatus::Fallback => compute_hash_embedding(text), } } @@ -175,15 +177,15 @@ fn compute_embedding(text: &str) -> Result> { /// Pipeline: tokenize → Tensor::from_array → Session::run → mean pooling → L2 normalize. /// Exact parity with Python embeddings.py implementation. fn compute_onnx_embedding(text: &str) -> Result> { - let session_lock = ONNX_SESSION.get() - .context("ONNX session not initialized")?; - let mut session = session_lock.lock() + let session_lock = ONNX_SESSION.get().context("ONNX session not initialized")?; + let mut session = session_lock + .lock() .map_err(|e| anyhow::anyhow!("session lock poisoned: {e}"))?; - let tokenizer = TOKENIZER.get() - .context("Tokenizer not initialized")?; + let tokenizer = TOKENIZER.get().context("Tokenizer not initialized")?; // 1. Tokenize - let encoding = tokenizer.encode(text, true) + let encoding = tokenizer + .encode(text, true) .map_err(|e| anyhow::anyhow!("tokenization failed: {e}"))?; let ids = encoding.get_ids(); @@ -199,17 +201,14 @@ fn compute_onnx_embedding(text: &str) -> Result> { let shape = vec![1i64, seq_len as i64]; - let input_ids_tensor = ort::value::Tensor::from_array( - (shape.clone(), input_ids) - ).context("failed to create input_ids tensor")?; + let input_ids_tensor = ort::value::Tensor::from_array((shape.clone(), input_ids)) + .context("failed to create input_ids tensor")?; - let attn_mask_tensor = ort::value::Tensor::from_array( - (shape.clone(), attn_mask.clone()) - ).context("failed to create attention_mask tensor")?; + let attn_mask_tensor = ort::value::Tensor::from_array((shape.clone(), attn_mask.clone())) + .context("failed to create attention_mask tensor")?; - let type_ids_tensor = ort::value::Tensor::from_array( - (shape, type_ids) - ).context("failed to create token_type_ids tensor")?; + let type_ids_tensor = ort::value::Tensor::from_array((shape, type_ids)) + .context("failed to create token_type_ids tensor")?; // 3. Run inference — inputs! returns Vec, not Result let inputs = ort::inputs! { @@ -217,7 +216,8 @@ fn compute_onnx_embedding(text: &str) -> Result> { "attention_mask" => attn_mask_tensor, "token_type_ids" => type_ids_tensor, }; - let outputs = session.run(inputs) + let outputs = session + .run(inputs) .map_err(|e| anyhow::anyhow!("inference failed: {e}"))?; // 4. Extract token embeddings (shape: [1, seq_len, 384]) @@ -230,7 +230,9 @@ fn compute_onnx_embedding(text: &str) -> Result> { if shape.len() != 3 || shape[2] as usize != EMBEDDING_DIM { return Err(anyhow::anyhow!( "unexpected output shape: {:?}, expected [1, {}, {}]", - shape, seq_len, EMBEDDING_DIM + shape, + seq_len, + EMBEDDING_DIM )); } @@ -254,7 +256,12 @@ fn compute_onnx_embedding(text: &str) -> Result> { } // 6. L2 normalize - let norm: f32 = sum_embedding.iter().map(|x| x * x).sum::().sqrt().max(1e-9); + let norm: f32 = sum_embedding + .iter() + .map(|x| x * x) + .sum::() + .sqrt() + .max(1e-9); for v in sum_embedding.iter_mut() { *v /= norm; } @@ -273,9 +280,9 @@ fn compute_hash_embedding(text: &str) -> Result> { let words: Vec<&str> = text_lower.split_whitespace().collect(); for (i, word) in words.iter().enumerate() { - let hash = word.bytes().fold(0u32, |acc, b| { - acc.wrapping_mul(31).wrapping_add(b as u32) - }); + let hash = word + .bytes() + .fold(0u32, |acc, b| acc.wrapping_mul(31).wrapping_add(b as u32)); let idx = (hash as usize) % EMBEDDING_DIM; embedding[idx] += 1.0 / (1.0 + i as f32); } @@ -324,7 +331,10 @@ mod tests { fn test_embedding_normalized() { let emb = compute_hash_embedding("test sentence for normalization").unwrap(); let norm: f32 = emb.iter().map(|x| x * x).sum::().sqrt(); - assert!((norm - 1.0).abs() < 0.01, "should be L2 normalized: got {norm}"); + assert!( + (norm - 1.0).abs() < 0.01, + "should be L2 normalized: got {norm}" + ); } #[test] diff --git a/rust/src/graph/centrality.rs b/rust/src/graph/centrality.rs index 9814c65..ed11f06 100644 --- a/rust/src/graph/centrality.rs +++ b/rust/src/graph/centrality.rs @@ -9,17 +9,14 @@ use std::collections::{HashMap, VecDeque}; /// Compute betweenness centrality and return top entities. pub async fn compute_bridges(pool: &PgPool, top_k: usize) -> Result> { // Fetch graph - let edges: Vec<(uuid::Uuid, uuid::Uuid)> = sqlx::query_as( - "SELECT from_entity, to_entity FROM brain_relations" - ) - .fetch_all(pool) - .await?; - - let entities: Vec<(uuid::Uuid, String)> = sqlx::query_as( - "SELECT id, name FROM brain_entities" - ) - .fetch_all(pool) - .await?; + let edges: Vec<(uuid::Uuid, uuid::Uuid)> = + sqlx::query_as("SELECT from_entity, to_entity FROM brain_relations") + .fetch_all(pool) + .await?; + + let entities: Vec<(uuid::Uuid, String)> = sqlx::query_as("SELECT id, name FROM brain_entities") + .fetch_all(pool) + .await?; if entities.is_empty() || edges.is_empty() { return Ok(vec![]); @@ -89,13 +86,18 @@ pub async fn compute_bridges(pool: &PgPool, top_k: usize) -> Result 2 { ((n - 1) * (n - 2)) as f64 } else { 1.0 }; + let norm = if n > 2 { + ((n - 1) * (n - 2)) as f64 + } else { + 1.0 + }; for b in betweenness.iter_mut() { *b /= norm; } // Sort and return top_k - let mut ranked: Vec<(String, f64)> = names.into_iter() + let mut ranked: Vec<(String, f64)> = names + .into_iter() .zip(betweenness) .filter(|(_, b)| *b > 0.0) .collect(); diff --git a/rust/src/graph/community.rs b/rust/src/graph/community.rs index 8026f66..a130dbb 100644 --- a/rust/src/graph/community.rs +++ b/rust/src/graph/community.rs @@ -20,17 +20,14 @@ const MAX_LOCAL_ITERATIONS: usize = 50; /// label propagation. Guarantees internally connected communities. pub async fn detect(pool: &PgPool) -> Result)>> { // Fetch graph - let edges: Vec<(uuid::Uuid, uuid::Uuid, f64)> = sqlx::query_as( - "SELECT from_entity, to_entity, strength FROM brain_relations" - ) - .fetch_all(pool) - .await?; - - let entities: Vec<(uuid::Uuid, String)> = sqlx::query_as( - "SELECT id, name FROM brain_entities" - ) - .fetch_all(pool) - .await?; + let edges: Vec<(uuid::Uuid, uuid::Uuid, f64)> = + sqlx::query_as("SELECT from_entity, to_entity, strength FROM brain_relations") + .fetch_all(pool) + .await?; + + let entities: Vec<(uuid::Uuid, String)> = sqlx::query_as("SELECT id, name FROM brain_entities") + .fetch_all(pool) + .await?; if entities.is_empty() { return Ok(vec![]); @@ -146,8 +143,14 @@ fn leiden_phase( let mut best_community = current_community; let mut best_delta_q = 0.0; - let ki_in_current = community_weights.get(¤t_community).copied().unwrap_or(0.0); - let sigma_tot_current = community_totals.get(¤t_community).copied().unwrap_or(0.0); + let ki_in_current = community_weights + .get(¤t_community) + .copied() + .unwrap_or(0.0); + let sigma_tot_current = community_totals + .get(¤t_community) + .copied() + .unwrap_or(0.0); for (&candidate, &ki_in_candidate) in &community_weights { if candidate == current_community { @@ -270,9 +273,11 @@ mod tests { .map(|n| n.iter().map(|(_, w)| w).sum()) .collect(); - let total_weight: f64 = neighbors.iter() + let total_weight: f64 = neighbors + .iter() .flat_map(|n| n.iter().map(|(_, w)| w)) - .sum::() / 2.0; + .sum::() + / 2.0; let mut labels: Vec = (0..6).collect(); @@ -294,10 +299,10 @@ mod tests { fn test_refinement_splits_disconnected() { // Community {0,1,2,3} but 0-1 connected and 2-3 connected, no bridge let neighbors = vec![ - vec![(1, 1.0)], // 0 - vec![(0, 1.0)], // 1 - vec![(3, 1.0)], // 2 - vec![(2, 1.0)], // 3 + vec![(1, 1.0)], // 0 + vec![(0, 1.0)], // 1 + vec![(3, 1.0)], // 2 + vec![(2, 1.0)], // 3 ]; let mut labels = vec![0, 0, 0, 0]; // All in community 0 diff --git a/rust/src/graph/pagerank.rs b/rust/src/graph/pagerank.rs index 53dfbdf..03eeb2b 100644 --- a/rust/src/graph/pagerank.rs +++ b/rust/src/graph/pagerank.rs @@ -11,35 +11,20 @@ const DAMPING: f64 = 0.85; const ITERATIONS: usize = 20; const CONVERGENCE_THRESHOLD: f64 = 1e-6; -/// Compute PageRank and store results (batch UPDATE — P1 fix). +/// Compute PageRank from a list of edges. /// -/// Returns number of entities updated. -pub async fn compute_and_store(pool: &PgPool) -> Result { - // Fetch all relations with NF-IDF hub dampening (§B) - let edges: Vec<(uuid::Uuid, uuid::Uuid, f64)> = sqlx::query_as( - r#" - SELECT r.from_entity, r.to_entity, - r.strength / LN(1.0 + COALESCE(deg.degree, 1)) AS dampened_weight - FROM brain_relations r - LEFT JOIN ( - SELECT from_entity, COUNT(*) AS degree - FROM brain_relations - GROUP BY from_entity - ) deg ON r.from_entity = deg.from_entity - "# - ) - .fetch_all(pool) - .await?; - +/// Extracts the core algorithm from database I/O to improve maintainability +/// and testability. +pub fn compute_pagerank(edges: &[(uuid::Uuid, uuid::Uuid, f64)]) -> (Vec, Vec) { if edges.is_empty() { - return Ok(0); + return (Vec::new(), Vec::new()); } // Build adjacency: node_id → (outgoing nodes with weights) let mut nodes: HashMap = HashMap::new(); let mut node_list: Vec = Vec::new(); - for (from, to, _) in &edges { + for (from, to, _) in edges { for id in [from, to] { if !nodes.contains_key(id) { let idx = node_list.len(); @@ -51,14 +36,14 @@ pub async fn compute_and_store(pool: &PgPool) -> Result { let n = node_list.len(); if n == 0 { - return Ok(0); + return (Vec::new(), Vec::new()); } // Build adjacency lists let mut outgoing: Vec> = vec![vec![]; n]; let mut out_weight_sum: Vec = vec![0.0; n]; - for (from, to, weight) in &edges { + for (from, to, weight) in edges { let from_idx = nodes[from]; let to_idx = nodes[to]; outgoing[from_idx].push((to_idx, *weight)); @@ -90,7 +75,9 @@ pub async fn compute_and_store(pool: &PgPool) -> Result { } // Convergence check - let delta: f64 = ranks.iter().zip(new_ranks.iter()) + let delta: f64 = ranks + .iter() + .zip(new_ranks.iter()) .map(|(old, new)| (old - new).abs()) .sum(); @@ -102,10 +89,37 @@ pub async fn compute_and_store(pool: &PgPool) -> Result { } } - // P1 FIX: Batch UPDATE with unnest() — 1 query instead of N - let ids: Vec = node_list.clone(); - let scores: Vec = ranks; + (node_list, ranks) +} + +/// Compute PageRank and store results (batch UPDATE — P1 fix). +/// +/// Returns number of entities updated. +pub async fn compute_and_store(pool: &PgPool) -> Result { + // Fetch all relations with NF-IDF hub dampening (§B) + let edges: Vec<(uuid::Uuid, uuid::Uuid, f64)> = sqlx::query_as( + r#" + SELECT r.from_entity, r.to_entity, + r.strength / LN(1.0 + COALESCE(deg.degree, 1)) AS dampened_weight + FROM brain_relations r + LEFT JOIN ( + SELECT from_entity, COUNT(*) AS degree + FROM brain_relations + GROUP BY from_entity + ) deg ON r.from_entity = deg.from_entity + "#, + ) + .fetch_all(pool) + .await?; + + let (ids, scores) = compute_pagerank(&edges); + let n = ids.len(); + + if n == 0 { + return Ok(0); + } + // P1 FIX: Batch UPDATE with unnest() — 1 query instead of N sqlx::query( r#" UPDATE brain_entities AS e @@ -113,7 +127,7 @@ pub async fn compute_and_store(pool: &PgPool) -> Result { updated_at = NOW() FROM (SELECT UNNEST($1::uuid[]) AS id, UNNEST($2::float8[]) AS rank) AS v WHERE e.id = v.id - "# + "#, ) .bind(&ids) .bind(&scores) @@ -134,4 +148,51 @@ mod tests { assert!(ITERATIONS > 0); assert!(CONVERGENCE_THRESHOLD > 0.0); } + + #[test] + fn test_compute_pagerank_empty() { + let (ids, scores) = compute_pagerank(&[]); + assert!(ids.is_empty()); + assert!(scores.is_empty()); + } + + #[test] + fn test_compute_pagerank_simple() { + let node1 = uuid::Uuid::new_v4(); + let node2 = uuid::Uuid::new_v4(); + let node3 = uuid::Uuid::new_v4(); + + // 1 -> 2, 2 -> 3, 3 -> 1 (cycle) + let edges = vec![ + (node1, node2, 1.0), + (node2, node3, 1.0), + (node3, node1, 1.0), + ]; + + let (ids, scores) = compute_pagerank(&edges); + assert_eq!(ids.len(), 3); + assert_eq!(scores.len(), 3); + + // In a symmetric cycle, they should all have roughly equal PageRank + let expected = 1.0 / 3.0; + for score in scores { + assert!((score - expected).abs() < 1e-5); + } + } + + #[test] + fn test_compute_pagerank_dangling() { + let node1 = uuid::Uuid::new_v4(); + let node2 = uuid::Uuid::new_v4(); + + // 1 -> 2, 2 has no outgoing (dangling) + let edges = vec![(node1, node2, 1.0)]; + + let (ids, scores) = compute_pagerank(&edges); + assert_eq!(ids.len(), 2); + assert_eq!(scores.len(), 2); + + let sum: f64 = scores.iter().sum(); + assert!((sum - 1.0).abs() < 1e-5); // Should still sum to 1 + } } diff --git a/rust/src/handlers/alarma.rs b/rust/src/handlers/alarma.rs index 6a0f57a..e67a389 100644 --- a/rust/src/handlers/alarma.rs +++ b/rust/src/handlers/alarma.rs @@ -7,10 +7,22 @@ use sqlx::PgPool; use super::zafra::safe_truncate; pub async fn handle(pool: &PgPool, args: Value) -> Result { - let error_type = args.get("error_type").and_then(|v| v.as_str()).unwrap_or("Unknown"); - let error_message = args.get("error_message").and_then(|v| v.as_str()).unwrap_or(""); - let context = args.get("context").cloned().unwrap_or(Value::Object(serde_json::Map::new())); - let project = args.get("project").and_then(|v| v.as_str()).unwrap_or("default"); + let error_type = args + .get("error_type") + .and_then(|v| v.as_str()) + .unwrap_or("Unknown"); + let error_message = args + .get("error_message") + .and_then(|v| v.as_str()) + .unwrap_or(""); + let context = args + .get("context") + .cloned() + .unwrap_or(Value::Object(serde_json::Map::new())); + let project = args + .get("project") + .and_then(|v| v.as_str()) + .unwrap_or("default"); if error_message.is_empty() { anyhow::bail!("error_message is required"); @@ -41,7 +53,7 @@ pub async fn handle(pool: &PgPool, args: Value) -> Result { if similar_count.0 >= 3 { sqlx::query( "UPDATE brain_errors SET synapse_weight = LEAST(synapse_weight + 0.1, 5.0) - WHERE similarity(error_message, $1) > 0.5 AND project = $2" + WHERE similarity(error_message, $1) > 0.5 AND project = $2", ) .bind(error_message) .bind(project) @@ -57,9 +69,10 @@ pub async fn handle(pool: &PgPool, args: Value) -> Result { }); if similar_count.0 >= 3 { - response["pattern_warning"] = serde_json::json!( - format!("⚠️ Pattern detected: {} similar errors in project '{}'", similar_count.0, project) - ); + response["pattern_warning"] = serde_json::json!(format!( + "⚠️ Pattern detected: {} similar errors in project '{}'", + similar_count.0, project + )); } Ok(response) diff --git a/rust/src/handlers/alma.rs b/rust/src/handlers/alma.rs index 05d6eb0..8e2aca8 100644 --- a/rust/src/handlers/alma.rs +++ b/rust/src/handlers/alma.rs @@ -4,8 +4,8 @@ //! V10: Upsert detection — AI knows if entity was created or re-found. //! Hebbian: get/update auto-boost entity + neighbor importance. -use crate::constants::VALID_ENTITY_TYPES; use crate::cognitive::{dual_strength, hebbian}; +use crate::constants::VALID_ENTITY_TYPES; use anyhow::{Context, Result}; use serde_json::Value; use sqlx::PgPool; @@ -69,7 +69,7 @@ async fn create(pool: &PgPool, name: &str, args: &Value) -> Result { // Actually create let row: (uuid::Uuid,) = sqlx::query_as( - "INSERT INTO brain_entities (name, entity_type) VALUES ($1, $2) RETURNING id" + "INSERT INTO brain_entities (name, entity_type) VALUES ($1, $2) RETURNING id", ) .bind(name) .bind(entity_type) @@ -93,22 +93,18 @@ async fn create(pool: &PgPool, name: &str, args: &Value) -> Result { /// Update entity name. async fn update(pool: &PgPool, name: &str, args: &Value) -> Result { - let new_name = args - .get("new_name") - .and_then(|v| v.as_str()) - .unwrap_or(""); + let new_name = args.get("new_name").and_then(|v| v.as_str()).unwrap_or(""); if new_name.is_empty() || new_name.len() > 200 { anyhow::bail!("new_name must be 1-200 characters"); } - let result = sqlx::query( - "UPDATE brain_entities SET name = $1, updated_at = NOW() WHERE name = $2" - ) - .bind(new_name) - .bind(name) - .execute(pool) - .await?; + let result = + sqlx::query("UPDATE brain_entities SET name = $1, updated_at = NOW() WHERE name = $2") + .bind(new_name) + .bind(name) + .execute(pool) + .await?; if result.rows_affected() == 0 { anyhow::bail!("Entity '{name}' not found"); @@ -148,7 +144,7 @@ async fn get(pool: &PgPool, name: &str) -> Result { // Get entity (with FOR UPDATE to prevent stale reads — FIX B1 partial) let entity: Option<(uuid::Uuid, String, String, f64, i32)> = sqlx::query_as( "SELECT id, name, entity_type, importance, access_count - FROM brain_entities WHERE name = $1" + FROM brain_entities WHERE name = $1", ) .bind(name) .fetch_optional(pool) @@ -170,7 +166,7 @@ async fn get(pool: &PgPool, name: &str) -> Result { "SELECT id, content, observation_type, importance, source FROM brain_observations WHERE entity_id = $1 AND observation_type != 'superseded' - ORDER BY importance DESC, created_at DESC" + ORDER BY importance DESC, created_at DESC", ) .bind(entity_id) .fetch_all(pool) @@ -198,7 +194,7 @@ async fn get(pool: &PgPool, name: &str) -> Result { JOIN brain_entities e ON ( CASE WHEN r.from_entity = $1 THEN r.to_entity ELSE r.from_entity END = e.id ) - WHERE r.from_entity = $1 OR r.to_entity = $1" + WHERE r.from_entity = $1 OR r.to_entity = $1", ) .bind(entity_id) .fetch_all(pool) diff --git a/rust/src/handlers/cronica.rs b/rust/src/handlers/cronica.rs index d8bfafb..f4588a4 100644 --- a/rust/src/handlers/cronica.rs +++ b/rust/src/handlers/cronica.rs @@ -5,18 +5,18 @@ //! P3: Batch dedup in batch_add (1 query per batch, not N). //! V5: Prediction Error Gating — classify by similarity. -use crate::constants::{ - DEDUP_THRESHOLD, - VALID_OBSERVATION_TYPES, VALID_SOURCES, -}; use crate::cognitive::{density, prediction_error}; +use crate::constants::{DEDUP_THRESHOLD, VALID_OBSERVATION_TYPES, VALID_SOURCES}; use anyhow::{Context, Result}; use serde_json::Value; use sqlx::PgPool; pub async fn handle(pool: &PgPool, args: Value) -> Result { let action = args.get("action").and_then(|v| v.as_str()).unwrap_or(""); - let entity_name = args.get("entity_name").and_then(|v| v.as_str()).unwrap_or(""); + let entity_name = args + .get("entity_name") + .and_then(|v| v.as_str()) + .unwrap_or(""); match action { "add" => add(pool, entity_name, &args).await, @@ -38,12 +38,18 @@ async fn add(pool: &PgPool, entity_name: &str, args: &Value) -> Result { anyhow::bail!("content must be 1-10000 characters"); } - let obs_type = args.get("observation_type").and_then(|v| v.as_str()).unwrap_or("fact"); + let obs_type = args + .get("observation_type") + .and_then(|v| v.as_str()) + .unwrap_or("fact"); if !VALID_OBSERVATION_TYPES.contains(&obs_type) { anyhow::bail!("Invalid observation_type: {obs_type}"); } - let source = args.get("source").and_then(|v| v.as_str()).unwrap_or("agent"); + let source = args + .get("source") + .and_then(|v| v.as_str()) + .unwrap_or("agent"); if !VALID_SOURCES.contains(&source) { anyhow::bail!("Invalid source: {source}"); } @@ -71,7 +77,7 @@ async fn add(pool: &PgPool, entity_name: &str, args: &Value) -> Result { importance = LEAST(importance + 0.05, 1.0), access_count = access_count + 1, last_accessed = NOW() - WHERE id = $1" + WHERE id = $1", ) .bind(obs_id) .execute(pool) @@ -90,7 +96,7 @@ async fn add(pool: &PgPool, entity_name: &str, args: &Value) -> Result { // Insert observation let row: (uuid::Uuid,) = sqlx::query_as( "INSERT INTO brain_observations (entity_id, content, observation_type, source, importance) - VALUES ($1, $2, $3, $4, $5) RETURNING id" + VALUES ($1, $2, $3, $4, $5) RETURNING id", ) .bind(entity_id) .bind(content) @@ -138,7 +144,9 @@ async fn add(pool: &PgPool, entity_name: &str, args: &Value) -> Result { /// Delete an observation by ID. async fn delete_obs(pool: &PgPool, args: &Value) -> Result { - let obs_id = args.get("observation_id").or_else(|| args.get("id")) + let obs_id = args + .get("observation_id") + .or_else(|| args.get("id")) .and_then(|v| v.as_str()) .unwrap_or(""); @@ -171,7 +179,7 @@ async fn list(pool: &PgPool, entity_name: &str) -> Result { "SELECT id, content, observation_type, importance, source, access_count FROM brain_observations WHERE entity_id = $1 AND observation_type != 'superseded' - ORDER BY importance DESC, created_at DESC" + ORDER BY importance DESC, created_at DESC", ) .bind(entity_id) .fetch_all(pool) @@ -203,7 +211,8 @@ async fn list(pool: &PgPool, entity_name: &str) -> Result { /// /// Uses explicit transaction — all-or-nothing atomicity. async fn batch_add(pool: &PgPool, args: &Value) -> Result { - let observations = args.get("observations") + let observations = args + .get("observations") .and_then(|v| v.as_array()) .context("'observations' array is required for batch_add")?; @@ -225,10 +234,19 @@ async fn batch_add(pool: &PgPool, args: &Value) -> Result { let mut deduplicated = 0u32; for obs in observations { - let entity_name = obs.get("entity_name").and_then(|v| v.as_str()).unwrap_or(""); + let entity_name = obs + .get("entity_name") + .and_then(|v| v.as_str()) + .unwrap_or(""); let content = obs.get("content").and_then(|v| v.as_str()).unwrap_or(""); - let obs_type = obs.get("observation_type").and_then(|v| v.as_str()).unwrap_or("fact"); - let source = obs.get("source").and_then(|v| v.as_str()).unwrap_or("agent"); + let obs_type = obs + .get("observation_type") + .and_then(|v| v.as_str()) + .unwrap_or("fact"); + let source = obs + .get("source") + .and_then(|v| v.as_str()) + .unwrap_or("agent"); if entity_name.is_empty() || content.is_empty() { continue; @@ -284,7 +302,7 @@ async fn batch_add(pool: &PgPool, args: &Value) -> Result { importance = LEAST(importance + 0.05, 1.0), access_count = access_count + 1, last_accessed = NOW() - WHERE id = $1" + WHERE id = $1", ) .bind(obs_id) .execute(&mut *tx) @@ -294,7 +312,9 @@ async fn batch_add(pool: &PgPool, args: &Value) -> Result { } } - tx.commit().await.context("failed to commit batch_add transaction")?; + tx.commit() + .await + .context("failed to commit batch_add transaction")?; Ok(serde_json::json!({ "action": "batch_add", @@ -320,7 +340,7 @@ async fn check_dedup(pool: &PgPool, entity_id: uuid::Uuid, content: &str) -> Res "SELECT id, content, similarity(content, $2)::float8 AS sim FROM brain_observations WHERE entity_id = $1 AND similarity(content, $2) > 0.3 - ORDER BY sim DESC LIMIT 5" + ORDER BY sim DESC LIMIT 5", ) .bind(entity_id) .bind(content) @@ -336,7 +356,7 @@ async fn check_dedup(pool: &PgPool, entity_id: uuid::Uuid, content: &str) -> Res "SELECT similarity(content, $2)::float8 FROM brain_observations WHERE entity_id = $1 AND observation_type != 'superseded' - ORDER BY created_at DESC LIMIT 20" + ORDER BY created_at DESC LIMIT 20", ) .bind(entity_id) .bind(content) @@ -349,7 +369,9 @@ async fn check_dedup(pool: &PgPool, entity_id: uuid::Uuid, content: &str) -> Res for (obs_id, existing_content, sim) in &dupes { let sim = *sim; if sim > DEDUP_THRESHOLD { - return Ok(DedupResult::Duplicate(super::zafra::safe_truncate(existing_content, 80).to_string())); + return Ok(DedupResult::Duplicate( + super::zafra::safe_truncate(existing_content, 80).to_string(), + )); } // V5.1: Adaptive PE gating (Friston, Nature 2023) let action = prediction_error::adaptive_gate(sim, &recent); @@ -369,12 +391,11 @@ async fn check_dedup(pool: &PgPool, entity_id: uuid::Uuid, content: &str) -> Res /// Ensure entity exists, creating if necessary. Returns entity_id. async fn ensure_entity(pool: &PgPool, name: &str) -> Result { // Try to find existing - let existing: Option<(uuid::Uuid,)> = sqlx::query_as( - "SELECT id FROM brain_entities WHERE name = $1" - ) - .bind(name) - .fetch_optional(pool) - .await?; + let existing: Option<(uuid::Uuid,)> = + sqlx::query_as("SELECT id FROM brain_entities WHERE name = $1") + .bind(name) + .fetch_optional(pool) + .await?; if let Some((id,)) = existing { return Ok(id); @@ -385,7 +406,7 @@ async fn ensure_entity(pool: &PgPool, name: &str) -> Result { "INSERT INTO brain_entities (name, entity_type) VALUES ($1, 'concept') ON CONFLICT (name) DO UPDATE SET updated_at = NOW() - RETURNING id" + RETURNING id", ) .bind(name) .fetch_one(pool) @@ -397,12 +418,11 @@ async fn ensure_entity(pool: &PgPool, name: &str) -> Result { /// Get entity_id by name (strict — errors if not found). async fn get_entity_id(pool: &PgPool, name: &str) -> Result { - let row: Option<(uuid::Uuid,)> = sqlx::query_as( - "SELECT id FROM brain_entities WHERE name = $1" - ) - .bind(name) - .fetch_optional(pool) - .await?; + let row: Option<(uuid::Uuid,)> = + sqlx::query_as("SELECT id FROM brain_entities WHERE name = $1") + .bind(name) + .fetch_optional(pool) + .await?; row.map(|(id,)| id) .context(format!("Entity '{name}' not found")) diff --git a/rust/src/handlers/decreto.rs b/rust/src/handlers/decreto.rs index 8867af8..d505287 100644 --- a/rust/src/handlers/decreto.rs +++ b/rust/src/handlers/decreto.rs @@ -10,9 +10,15 @@ pub async fn handle(pool: &PgPool, args: Value) -> Result { match action { "record" => { - let title = args.get("title").and_then(|v| v.as_str()).unwrap_or("Untitled"); + let title = args + .get("title") + .and_then(|v| v.as_str()) + .unwrap_or("Untitled"); let context = args.get("context").and_then(|v| v.as_str()).unwrap_or(""); - let alternatives = args.get("alternatives").cloned().unwrap_or(Value::Array(vec![])); + let alternatives = args + .get("alternatives") + .cloned() + .unwrap_or(Value::Array(vec![])); let chosen = args.get("chosen").and_then(|v| v.as_str()).unwrap_or(""); let rationale = args.get("rationale").and_then(|v| v.as_str()).unwrap_or(""); @@ -24,8 +30,11 @@ pub async fn handle(pool: &PgPool, args: Value) -> Result { // Ensure entity exists sqlx::query("INSERT INTO brain_entities (name, entity_type) VALUES ($1, 'concept') ON CONFLICT (name) DO NOTHING") .bind(entity_name).execute(pool).await?; - let entity_id: (uuid::Uuid,) = sqlx::query_as("SELECT id FROM brain_entities WHERE name = $1") - .bind(entity_name).fetch_one(pool).await?; + let entity_id: (uuid::Uuid,) = + sqlx::query_as("SELECT id FROM brain_entities WHERE name = $1") + .bind(entity_name) + .fetch_one(pool) + .await?; let row: (uuid::Uuid,) = sqlx::query_as( "INSERT INTO brain_observations (entity_id, content, observation_type, source) VALUES ($1, $2, 'decision', 'agent') RETURNING id" @@ -38,8 +47,11 @@ pub async fn handle(pool: &PgPool, args: Value) -> Result { let decisions: Vec<(uuid::Uuid, String, f64)> = sqlx::query_as( "SELECT id, content, similarity(content, $1)::float8 AS sim FROM brain_observations WHERE observation_type = 'decision' AND similarity(content, $1) > 0.2 - ORDER BY sim DESC LIMIT 10" - ).bind(query).fetch_all(pool).await?; + ORDER BY sim DESC LIMIT 10", + ) + .bind(query) + .fetch_all(pool) + .await?; let results: Vec = decisions.iter().map(|(id, content, sim)| { serde_json::json!({"id": id.to_string(), "content": content, "similarity": sim}) }).collect(); @@ -49,7 +61,10 @@ pub async fn handle(pool: &PgPool, args: Value) -> Result { let decisions: Vec<(uuid::Uuid, String)> = sqlx::query_as( "SELECT id, content FROM brain_observations WHERE observation_type = 'decision' ORDER BY created_at DESC LIMIT 20" ).fetch_all(pool).await?; - let list: Vec = decisions.iter().map(|(id, c)| serde_json::json!({"id": id.to_string(), "content": c})).collect(); + let list: Vec = decisions + .iter() + .map(|(id, c)| serde_json::json!({"id": id.to_string(), "content": c})) + .collect(); Ok(serde_json::json!({"action": "list", "decisions": list, "count": list.len()})) } _ => anyhow::bail!("Invalid action: {action}"), diff --git a/rust/src/handlers/eco.rs b/rust/src/handlers/eco.rs index 4bb706b..f3752f9 100644 --- a/rust/src/handlers/eco.rs +++ b/rust/src/handlers/eco.rs @@ -23,7 +23,11 @@ pub async fn handle(pool: &PgPool, args: Value) -> Result { /// Positive RLHF: Oja's rule — importance += η * (1 - importance). /// Also updates FSRS stability on recall (V14 — Ye 2023, Karpicke & Roediger 2008). -async fn positive(pool: &PgPool, entity_name: Option<&str>, observation_id: Option<&str>) -> Result { +async fn positive( + pool: &PgPool, + entity_name: Option<&str>, + observation_id: Option<&str>, +) -> Result { let mut boosted = 0u32; if let Some(obs_id_str) = observation_id { @@ -36,7 +40,7 @@ async fn positive(pool: &PgPool, entity_name: Option<&str>, observation_id: Opti last_accessed = NOW(), stability = GREATEST(stability * 1.2, 1.0), difficulty = GREATEST(1, difficulty - 0.1) - WHERE id = $1" + WHERE id = $1", ) .bind(obs_id) .execute(pool) @@ -50,7 +54,7 @@ async fn positive(pool: &PgPool, entity_name: Option<&str>, observation_id: Opti importance = LEAST(importance + 0.05 * (1.0 - importance), 1.0), access_count = access_count + 1, updated_at = NOW() - WHERE name = $1" + WHERE name = $1", ) .bind(name) .execute(pool) @@ -67,7 +71,11 @@ async fn positive(pool: &PgPool, entity_name: Option<&str>, observation_id: Opti } /// Negative RLHF: anti-Oja — importance -= η * importance. -async fn negative(pool: &PgPool, entity_name: Option<&str>, observation_id: Option<&str>) -> Result { +async fn negative( + pool: &PgPool, + entity_name: Option<&str>, + observation_id: Option<&str>, +) -> Result { let mut decreased = 0u32; if let Some(obs_id_str) = observation_id { @@ -76,7 +84,7 @@ async fn negative(pool: &PgPool, entity_name: Option<&str>, observation_id: Opti "UPDATE brain_observations SET importance = GREATEST(importance - 0.05 * importance, 0.0), last_accessed = NOW() - WHERE id = $1" + WHERE id = $1", ) .bind(obs_id) .execute(pool) @@ -89,7 +97,7 @@ async fn negative(pool: &PgPool, entity_name: Option<&str>, observation_id: Opti "UPDATE brain_entities SET importance = GREATEST(importance - 0.05 * importance, 0.0), updated_at = NOW() - WHERE name = $1" + WHERE name = $1", ) .bind(name) .execute(pool) @@ -108,7 +116,9 @@ async fn negative(pool: &PgPool, entity_name: Option<&str>, observation_id: Opti async fn correct(pool: &PgPool, observation_id: Option<&str>, args: &Value) -> Result { let obs_id_str = observation_id.context("observation_id required for correct")?; let obs_id: uuid::Uuid = obs_id_str.parse().context("invalid observation_id")?; - let correction = args.get("correction").and_then(|v| v.as_str()) + let correction = args + .get("correction") + .and_then(|v| v.as_str()) .context("correction text is required")?; // Archive old content in previous_versions, then update diff --git a/rust/src/handlers/expediente.rs b/rust/src/handlers/expediente.rs index 7211e64..b838c42 100644 --- a/rust/src/handlers/expediente.rs +++ b/rust/src/handlers/expediente.rs @@ -39,7 +39,10 @@ impl<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow> for ErrorRow { pub async fn handle(pool: &PgPool, args: Value) -> Result { let query = args.get("query").and_then(|v| v.as_str()).unwrap_or(""); let project = args.get("project").and_then(|v| v.as_str()); - let resolved_only = args.get("resolved_only").and_then(|v| v.as_bool()).unwrap_or(false); + let resolved_only = args + .get("resolved_only") + .and_then(|v| v.as_bool()) + .unwrap_or(false); let proposed_action = args.get("proposed_action").and_then(|v| v.as_str()); if query.is_empty() { @@ -103,34 +106,39 @@ pub async fn handle(pool: &PgPool, args: Value) -> Result { }; // FIX A-002: safe_truncate prevents panic on multi-byte UTF-8 - let results: Vec = errors.iter().map(|row| { - serde_json::json!({ - "id": row.id.to_string(), - "error_type": row.error_type, - "error_message": safe_truncate(&row.error_message, 200), - "solution": row.solution, - "resolved": row.resolved, - "project": row.project, - "similarity": row.sim + let results: Vec = errors + .iter() + .map(|row| { + serde_json::json!({ + "id": row.id.to_string(), + "error_type": row.error_type, + "error_message": safe_truncate(&row.error_message, 200), + "solution": row.solution, + "resolved": row.resolved, + "project": row.project, + "similarity": row.sim + }) }) - }).collect(); + .collect(); // Anti-repetition guard - let mut response = serde_json::json!({"query": query, "results": results, "count": results.len()}); + let mut response = + serde_json::json!({"query": query, "results": results, "count": results.len()}); if let Some(action) = proposed_action { let failed_similar: Vec<(String,)> = sqlx::query_as( "SELECT error_message FROM brain_errors - WHERE resolved = false AND similarity(solution, $1) > 0.5 LIMIT 3" + WHERE resolved = false AND similarity(solution, $1) > 0.5 LIMIT 3", ) .bind(action) .fetch_all(pool) .await?; if !failed_similar.is_empty() { - response["anti_repetition_warning"] = serde_json::json!( - format!("⚠️ Similar approach failed {} time(s) before", failed_similar.len()) - ); + response["anti_repetition_warning"] = serde_json::json!(format!( + "⚠️ Similar approach failed {} time(s) before", + failed_similar.len() + )); } } diff --git a/rust/src/handlers/faro.rs b/rust/src/handlers/faro.rs index 4e42c9c..fe76da3 100644 --- a/rust/src/handlers/faro.rs +++ b/rust/src/handlers/faro.rs @@ -24,14 +24,19 @@ pub async fn handle(pool: &PgPool, args: Value) -> Result { anyhow::bail!("query is required"); } - let mode = args.get("mode").and_then(|v| v.as_str()).unwrap_or("hybrid"); + let mode = args + .get("mode") + .and_then(|v| v.as_str()) + .unwrap_or("hybrid"); let scope = args.get("scope").and_then(|v| v.as_str()).unwrap_or("all"); - let limit = args.get("limit") + let limit = args + .get("limit") .and_then(|v| v.as_i64()) .unwrap_or(DEFAULT_LIMIT) .min(MAX_LIMIT); - let max_tokens = args.get("max_tokens") + let max_tokens = args + .get("max_tokens") .and_then(|v| v.as_i64()) .unwrap_or(DEFAULT_MAX_TOKENS); @@ -43,7 +48,13 @@ pub async fn handle(pool: &PgPool, args: Value) -> Result { } /// §A V2: Weighted RRF Entropy Routing — 3 ranges (Elastic 2025). -async fn hybrid_search(pool: &PgPool, query: &str, scope: &str, limit: i64, max_tokens: i64) -> Result { +async fn hybrid_search( + pool: &PgPool, + query: &str, + scope: &str, + limit: i64, + max_tokens: i64, +) -> Result { // §A V2: 3-range entropy routing (keyword / mixed / semantic) let query_entropy = compute_query_entropy(query); let (text_weight, vector_weight) = entropy_weights(query_entropy); @@ -62,7 +73,11 @@ async fn hybrid_search(pool: &PgPool, query: &str, scope: &str, limit: i64, max_ // Add text results with RRF rank score for (rank, result) in text_results.iter().enumerate() { - let id = result.get("id").and_then(|v| v.as_str()).unwrap_or("").to_string(); + let id = result + .get("id") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(); let rrf_score = text_weight / (rrf_k + rank as f64 + 1.0); fused_scores.insert(id, (rrf_score, result.clone())); } @@ -70,7 +85,11 @@ async fn hybrid_search(pool: &PgPool, query: &str, scope: &str, limit: i64, max_ // Add vector results (V8: only if available) if let Ok(vec_results) = vector_results { for (rank, result) in vec_results.iter().enumerate() { - let id = result.get("id").and_then(|v| v.as_str()).unwrap_or("").to_string(); + let id = result + .get("id") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(); let rrf_score = vector_weight / (rrf_k + rank as f64 + 1.0); fused_scores .entry(id.clone()) @@ -88,17 +107,20 @@ async fn hybrid_search(pool: &PgPool, query: &str, scope: &str, limit: i64, max_ results.truncate(limit as usize); // VF2: Testing Effect — boost retrieval_strength on matched observations - let matched_obs_ids: Vec = results.iter() + let matched_obs_ids: Vec = results + .iter() .filter_map(|(_, _, r)| { - r.get("id").and_then(|v| v.as_str()) + r.get("id") + .and_then(|v| v.as_str()) .and_then(|s| s.parse::().ok()) }) .collect(); if !matched_obs_ids.is_empty() - && let Err(e) = dual_strength::on_search_match(pool, &matched_obs_ids).await { - tracing::warn!(error = %e, "failed to apply Testing Effect boost"); - } + && let Err(e) = dual_strength::on_search_match(pool, &matched_obs_ids).await + { + tracing::warn!(error = %e, "failed to apply Testing Effect boost"); + } // Session awareness: check active session and boost matching results let session_boost = get_session_goals(pool).await.unwrap_or_default(); @@ -135,17 +157,21 @@ async fn hybrid_search(pool: &PgPool, query: &str, scope: &str, limit: i64, max_ // FIX R-005: Check budget BEFORE subtraction to prevent i64 underflow let mut token_budget = max_tokens; - results_json = results_json.into_iter().take_while(|r| { - let content_len = r.get("content") - .and_then(|v| v.as_str()) - .map(|s| s.len() as i64 / 4) - .unwrap_or(20); - if token_budget < content_len { - return false; - } - token_budget -= content_len; - true - }).collect(); + results_json = results_json + .into_iter() + .take_while(|r| { + let content_len = r + .get("content") + .and_then(|v| v.as_str()) + .map(|s| s.len() as i64 / 4) + .unwrap_or(20); + if token_budget < content_len { + return false; + } + token_budget -= content_len; + true + }) + .collect(); Ok(serde_json::json!({ "mode": "hybrid", @@ -172,7 +198,7 @@ async fn verify_claim(pool: &PgPool, claim: &str) -> Result { WHERE similarity(content, $1) > 0.3 AND observation_type != 'superseded' ORDER BY sim DESC - LIMIT 10" + LIMIT 10", ) .bind(claim) .fetch_all(pool) @@ -253,7 +279,8 @@ async fn text_search(pool: &PgPool, query: &str, scope: &str, limit: i64) -> Res "score": score }) }, - ).await?; + ) + .await?; results.extend(entities); } @@ -283,7 +310,8 @@ async fn text_search(pool: &PgPool, query: &str, scope: &str, limit: i64) -> Res "score": score }) }, - ).await?; + ) + .await?; results.extend(observations); } @@ -310,7 +338,8 @@ async fn text_search(pool: &PgPool, query: &str, scope: &str, limit: i64) -> Res "score": score }) }, - ).await?; + ) + .await?; results.extend(errors); } @@ -341,7 +370,7 @@ async fn vector_search(pool: &PgPool, query: &str, _scope: &str, limit: i64) -> WHERE o.embedding IS NOT NULL AND o.observation_type != 'superseded' ORDER BY o.embedding <=> $1::vector - LIMIT $2" + LIMIT $2", ) .bind(pgvector::Vector::from(embedding)) .bind(limit) @@ -410,7 +439,7 @@ fn compute_query_entropy(query: &str) -> f64 { /// Get active session goals for session-aware boosting. async fn get_session_goals(pool: &PgPool) -> Result> { let row: Option<(serde_json::Value,)> = sqlx::query_as( - "SELECT goals FROM brain_sessions WHERE ended_at IS NULL ORDER BY started_at DESC LIMIT 1" + "SELECT goals FROM brain_sessions WHERE ended_at IS NULL ORDER BY started_at DESC LIMIT 1", ) .fetch_optional(pool) .await?; @@ -428,15 +457,12 @@ async fn get_session_goals(pool: &PgPool) -> Result> { /// For each top result that has an entity name, we query its related entities /// to provide graph context. This helps the AI understand the broader /// knowledge structure around search matches. -async fn enrich_graphrag( - pool: &PgPool, - results: &[(String, f64, Value)], - top_k: usize, -) -> Value { +async fn enrich_graphrag(pool: &PgPool, results: &[(String, f64, Value)], top_k: usize) -> Value { let mut context: Vec = Vec::new(); for (_, _, result) in results.iter().take(top_k) { - let entity_name = result.get("entity_name") + let entity_name = result + .get("entity_name") .or_else(|| result.get("name")) .and_then(|v| v.as_str()); diff --git a/rust/src/handlers/forget.rs b/rust/src/handlers/forget.rs index cb5fbfe..1c59ed0 100644 --- a/rust/src/handlers/forget.rs +++ b/rust/src/handlers/forget.rs @@ -46,7 +46,7 @@ pub async fn handle(pool: &PgPool, args: Value) -> Result { WHERE error_message ILIKE '%' || $1 || '%' OR context::text ILIKE '%' || $1 || '%' RETURNING 1 - ) SELECT COUNT(*) FROM deleted" + ) SELECT COUNT(*) FROM deleted", ) .bind(entity_name) .fetch_one(&mut *tx) @@ -61,7 +61,7 @@ pub async fn handle(pool: &PgPool, args: Value) -> Result { OR session_name ILIKE '%' || $1 || '%' OR summary ILIKE '%' || $1 || '%' RETURNING 1 - ) SELECT COUNT(*) FROM deleted" + ) SELECT COUNT(*) FROM deleted", ) .bind(entity_name) .fetch_one(&mut *tx) diff --git a/rust/src/handlers/jornada.rs b/rust/src/handlers/jornada.rs index 3660c3c..ac94f1c 100644 --- a/rust/src/handlers/jornada.rs +++ b/rust/src/handlers/jornada.rs @@ -9,27 +9,44 @@ pub async fn handle(pool: &PgPool, args: Value) -> Result { match action { "start" => { - let name = args.get("name").and_then(|v| v.as_str()).unwrap_or("unnamed"); + let name = args + .get("name") + .and_then(|v| v.as_str()) + .unwrap_or("unnamed"); let goals = args.get("goals").cloned().unwrap_or(Value::Array(vec![])); let row: (uuid::Uuid,) = sqlx::query_as( - "INSERT INTO brain_sessions (session_name, goals) VALUES ($1, $2) RETURNING id" - ).bind(name).bind(&goals).fetch_one(pool).await.context("failed to start session")?; - Ok(serde_json::json!({"action": "started", "session": {"id": row.0.to_string(), "session_name": name, "started_at": chrono::Utc::now().to_rfc3339()}})) + "INSERT INTO brain_sessions (session_name, goals) VALUES ($1, $2) RETURNING id", + ) + .bind(name) + .bind(&goals) + .fetch_one(pool) + .await + .context("failed to start session")?; + Ok( + serde_json::json!({"action": "started", "session": {"id": row.0.to_string(), "session_name": name, "started_at": chrono::Utc::now().to_rfc3339()}}), + ) } "end" => { - let outcome = args.get("outcome").and_then(|v| v.as_str()).unwrap_or("success"); + let outcome = args + .get("outcome") + .and_then(|v| v.as_str()) + .unwrap_or("success"); let summary = args.get("summary").and_then(|v| v.as_str()).unwrap_or(""); let result = sqlx::query( "UPDATE brain_sessions SET ended_at = NOW(), outcome = $1, summary = $2 WHERE id = (SELECT id FROM brain_sessions WHERE ended_at IS NULL ORDER BY started_at DESC LIMIT 1)" ).bind(outcome).bind(summary).execute(pool).await?; - Ok(serde_json::json!({"action": "ended", "outcome": outcome, "updated": result.rows_affected() > 0})) + Ok( + serde_json::json!({"action": "ended", "outcome": outcome, "updated": result.rows_affected() > 0}), + ) } "current" => { let session: Option<(uuid::Uuid, Option, Value)> = sqlx::query_as( "SELECT id, session_name, goals FROM brain_sessions WHERE ended_at IS NULL ORDER BY started_at DESC LIMIT 1" ).fetch_optional(pool).await?; match session { - Some((id, name, goals)) => Ok(serde_json::json!({"action": "current", "session": {"id": id.to_string(), "name": name, "goals": goals}})), + Some((id, name, goals)) => Ok( + serde_json::json!({"action": "current", "session": {"id": id.to_string(), "name": name, "goals": goals}}), + ), None => Ok(serde_json::json!({"action": "current", "session": null})), } } diff --git a/rust/src/handlers/mod.rs b/rust/src/handlers/mod.rs index b117264..3e8ef55 100644 --- a/rust/src/handlers/mod.rs +++ b/rust/src/handlers/mod.rs @@ -7,8 +7,8 @@ use anyhow::Result; use serde_json::Value; use sqlx::PgPool; -pub mod alma; pub mod alarma; +pub mod alma; pub mod cronica; pub mod decreto; pub mod eco; diff --git a/rust/src/handlers/puente.rs b/rust/src/handlers/puente.rs index 198b392..c5e1eb8 100644 --- a/rust/src/handlers/puente.rs +++ b/rust/src/handlers/puente.rs @@ -23,10 +23,19 @@ pub async fn handle(pool: &PgPool, args: Value) -> Result { /// Create a relation between entities. async fn create(pool: &PgPool, args: &Value) -> Result { - let from = args.get("from_entity").and_then(|v| v.as_str()).unwrap_or(""); + let from = args + .get("from_entity") + .and_then(|v| v.as_str()) + .unwrap_or(""); let to = args.get("to_entity").and_then(|v| v.as_str()).unwrap_or(""); - let rel_type = args.get("relation_type").and_then(|v| v.as_str()).unwrap_or("related_to"); - let bidirectional = args.get("bidirectional").and_then(|v| v.as_bool()).unwrap_or(false); + let rel_type = args + .get("relation_type") + .and_then(|v| v.as_str()) + .unwrap_or("related_to"); + let bidirectional = args + .get("bidirectional") + .and_then(|v| v.as_bool()) + .unwrap_or(false); if from.is_empty() || to.is_empty() { anyhow::bail!("from_entity and to_entity are required"); @@ -54,7 +63,7 @@ async fn create(pool: &PgPool, args: &Value) -> Result { ON CONFLICT (from_entity, to_entity, relation_type) DO UPDATE SET strength = LEAST(brain_relations.strength + 0.1, 1.0), last_traversed = NOW() - RETURNING (xmax = 0) AS is_insert" + RETURNING (xmax = 0) AS is_insert", ) .bind(from_id) .bind(to_id) @@ -78,7 +87,7 @@ async fn create(pool: &PgPool, args: &Value) -> Result { VALUES ($1, $2, $3, true) ON CONFLICT (from_entity, to_entity, relation_type) DO UPDATE SET strength = LEAST(brain_relations.strength + 0.1, 1.0), - last_traversed = NOW()" + last_traversed = NOW()", ) .bind(to_id) .bind(from_id) @@ -98,16 +107,22 @@ async fn create(pool: &PgPool, args: &Value) -> Result { /// Delete a relation. async fn delete(pool: &PgPool, args: &Value) -> Result { - let from = args.get("from_entity").and_then(|v| v.as_str()).unwrap_or(""); + let from = args + .get("from_entity") + .and_then(|v| v.as_str()) + .unwrap_or(""); let to = args.get("to_entity").and_then(|v| v.as_str()).unwrap_or(""); - let rel_type = args.get("relation_type").and_then(|v| v.as_str()).unwrap_or(""); + let rel_type = args + .get("relation_type") + .and_then(|v| v.as_str()) + .unwrap_or(""); let from_id = get_entity_id(pool, from).await?; let to_id = get_entity_id(pool, to).await?; let result = sqlx::query( "DELETE FROM brain_relations - WHERE from_entity = $1 AND to_entity = $2 AND relation_type = $3" + WHERE from_entity = $1 AND to_entity = $2 AND relation_type = $3", ) .bind(from_id) .bind(to_id) @@ -123,8 +138,15 @@ async fn delete(pool: &PgPool, args: &Value) -> Result { /// Traverse graph from a starting entity using CTE. async fn traverse(pool: &PgPool, args: &Value) -> Result { - let start = args.get("start_entity").and_then(|v| v.as_str()).unwrap_or(""); - let max_depth = args.get("max_depth").and_then(|v| v.as_i64()).unwrap_or(3).min(5); + let start = args + .get("start_entity") + .and_then(|v| v.as_str()) + .unwrap_or(""); + let max_depth = args + .get("max_depth") + .and_then(|v| v.as_i64()) + .unwrap_or(3) + .min(5); if start.is_empty() { anyhow::bail!("start_entity is required"); @@ -163,7 +185,7 @@ async fn traverse(pool: &PgPool, args: &Value) -> Result { FROM graph_walk ORDER BY depth, strength DESC LIMIT 50 - "# + "#, ) .bind(start_id) .bind(max_depth) @@ -175,7 +197,7 @@ async fn traverse(pool: &PgPool, args: &Value) -> Result { "UPDATE brain_relations SET strength = LEAST(strength + 0.02, 1.0), last_traversed = NOW() - WHERE from_entity = $1" + WHERE from_entity = $1", ) .bind(start_id) .execute(pool) @@ -204,8 +226,15 @@ async fn traverse(pool: &PgPool, args: &Value) -> Result { /// Infer transitive connections (A→B→C). async fn infer(pool: &PgPool, args: &Value) -> Result { - let start = args.get("start_entity").and_then(|v| v.as_str()).unwrap_or(""); - let max_depth = args.get("max_depth").and_then(|v| v.as_i64()).unwrap_or(3).min(5); + let start = args + .get("start_entity") + .and_then(|v| v.as_str()) + .unwrap_or(""); + let max_depth = args + .get("max_depth") + .and_then(|v| v.as_i64()) + .unwrap_or(3) + .min(5); if start.is_empty() { anyhow::bail!("start_entity is required"); @@ -240,7 +269,7 @@ async fn infer(pool: &PgPool, args: &Value) -> Result { WHERE tc.depth > 1 ORDER BY tc.path_strength DESC LIMIT 20 - "# + "#, ) .bind(start_id) .bind(max_depth) @@ -268,12 +297,11 @@ async fn infer(pool: &PgPool, args: &Value) -> Result { /// Get entity_id by name. async fn get_entity_id(pool: &PgPool, name: &str) -> Result { - let row: Option<(uuid::Uuid,)> = sqlx::query_as( - "SELECT id FROM brain_entities WHERE name = $1" - ) - .bind(name) - .fetch_optional(pool) - .await?; + let row: Option<(uuid::Uuid,)> = + sqlx::query_as("SELECT id FROM brain_entities WHERE name = $1") + .bind(name) + .fetch_optional(pool) + .await?; row.map(|(id,)| id) .context(format!("Entity '{name}' not found")) diff --git a/rust/src/handlers/remedio.rs b/rust/src/handlers/remedio.rs index 4181234..d1b831a 100644 --- a/rust/src/handlers/remedio.rs +++ b/rust/src/handlers/remedio.rs @@ -20,7 +20,7 @@ pub async fn handle(pool: &PgPool, args: Value) -> Result { // Mark error as resolved let result = sqlx::query( - "UPDATE brain_errors SET solution = $2, resolved = true, resolved_at = NOW() WHERE id = $1" + "UPDATE brain_errors SET solution = $2, resolved = true, resolved_at = NOW() WHERE id = $1", ) .bind(error_id) .bind(solution) @@ -35,7 +35,7 @@ pub async fn handle(pool: &PgPool, args: Value) -> Result { let similar: Vec<(uuid::Uuid, String)> = sqlx::query_as( "SELECT e2.id, e2.error_message FROM brain_errors e1 JOIN brain_errors e2 ON similarity(e1.error_message, e2.error_message) > 0.5 - WHERE e1.id = $1 AND e2.resolved = false AND e2.id != $1 LIMIT 5" + WHERE e1.id = $1 AND e2.resolved = false AND e2.id != $1 LIMIT 5", ) .bind(error_id) .fetch_all(pool) diff --git a/rust/src/handlers/vigia.rs b/rust/src/handlers/vigia.rs index 70014d2..35a26b9 100644 --- a/rust/src/handlers/vigia.rs +++ b/rust/src/handlers/vigia.rs @@ -5,7 +5,10 @@ use serde_json::Value; use sqlx::PgPool; pub async fn handle(pool: &PgPool, args: Value) -> Result { - let metric = args.get("metric").and_then(|v| v.as_str()).unwrap_or("summary"); + let metric = args + .get("metric") + .and_then(|v| v.as_str()) + .unwrap_or("summary"); match metric { "summary" => summary(pool).await, @@ -18,11 +21,23 @@ pub async fn handle(pool: &PgPool, args: Value) -> Result { } async fn summary(pool: &PgPool) -> Result { - let entities: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM brain_entities").fetch_one(pool).await?; - let observations: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM brain_observations WHERE observation_type != 'superseded'").fetch_one(pool).await?; - let relations: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM brain_relations").fetch_one(pool).await?; - let errors: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM brain_errors").fetch_one(pool).await?; - let sessions: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM brain_sessions").fetch_one(pool).await?; + let entities: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM brain_entities") + .fetch_one(pool) + .await?; + let observations: (i64,) = sqlx::query_as( + "SELECT COUNT(*) FROM brain_observations WHERE observation_type != 'superseded'", + ) + .fetch_one(pool) + .await?; + let relations: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM brain_relations") + .fetch_one(pool) + .await?; + let errors: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM brain_errors") + .fetch_one(pool) + .await?; + let sessions: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM brain_sessions") + .fetch_one(pool) + .await?; let token_estimate = observations.0 * 50; Ok(serde_json::json!({ @@ -37,26 +52,32 @@ async fn summary(pool: &PgPool) -> Result { } async fn health(pool: &PgPool) -> Result { - let avg_importance: (Option,) = sqlx::query_as( - "SELECT AVG(importance)::float8 FROM brain_entities" - ).fetch_one(pool).await?; + let avg_importance: (Option,) = + sqlx::query_as("SELECT AVG(importance)::float8 FROM brain_entities") + .fetch_one(pool) + .await?; let stale_count: (i64,) = sqlx::query_as( - "SELECT COUNT(*) FROM brain_observations WHERE last_accessed < NOW() - INTERVAL '30 days'" - ).fetch_one(pool).await?; + "SELECT COUNT(*) FROM brain_observations WHERE last_accessed < NOW() - INTERVAL '30 days'", + ) + .fetch_one(pool) + .await?; - let unused_entities: (i64,) = sqlx::query_as( - "SELECT COUNT(*) FROM brain_entities WHERE access_count = 0" - ).fetch_one(pool).await?; + let unused_entities: (i64,) = + sqlx::query_as("SELECT COUNT(*) FROM brain_entities WHERE access_count = 0") + .fetch_one(pool) + .await?; - let db_size: (String,) = sqlx::query_as( - "SELECT pg_size_pretty(pg_database_size(current_database()))" - ).fetch_one(pool).await?; + let db_size: (String,) = + sqlx::query_as("SELECT pg_size_pretty(pg_database_size(current_database()))") + .fetch_one(pool) + .await?; // Entropy diversity — Shannon entropy of entity types - let type_counts: Vec<(String, i64)> = sqlx::query_as( - "SELECT entity_type, COUNT(*) FROM brain_entities GROUP BY entity_type" - ).fetch_all(pool).await?; + let type_counts: Vec<(String, i64)> = + sqlx::query_as("SELECT entity_type, COUNT(*) FROM brain_entities GROUP BY entity_type") + .fetch_all(pool) + .await?; let entity_entropy = compute_entropy(&type_counts); let max_entity_entropy = if type_counts.is_empty() { @@ -67,8 +88,10 @@ async fn health(pool: &PgPool) -> Result { // Observation type entropy let obs_counts: Vec<(String, i64)> = sqlx::query_as( - "SELECT observation_type, COUNT(*) FROM brain_observations GROUP BY observation_type" - ).fetch_all(pool).await?; + "SELECT observation_type, COUNT(*) FROM brain_observations GROUP BY observation_type", + ) + .fetch_all(pool) + .await?; let obs_entropy = compute_entropy(&obs_counts); let max_obs_entropy = if obs_counts.is_empty() { 1.0 @@ -85,8 +108,10 @@ async fn health(pool: &PgPool) -> Result { (SELECT COUNT(*) FROM brain_errors WHERE resolved), \ (SELECT COUNT(*) FROM brain_errors), \ (SELECT AVG(EXTRACT(EPOCH FROM (resolved_at - created_at)))::float8 \ - FROM brain_errors WHERE resolved AND resolved_at IS NOT NULL)" - ).fetch_one(pool).await?; + FROM brain_errors WHERE resolved AND resolved_at IS NOT NULL)", + ) + .fetch_one(pool) + .await?; let resolution_rate = if err_stats.1 > 0 { err_stats.0 as f64 / err_stats.1 as f64 @@ -118,34 +143,39 @@ async fn drift(pool: &PgPool) -> Result { let recent: Vec<(String, i64)> = sqlx::query_as( "SELECT error_type, COUNT(*) FROM brain_errors \ WHERE created_at > NOW() - INTERVAL '7 days' \ - GROUP BY error_type" - ).fetch_all(pool).await?; + GROUP BY error_type", + ) + .fetch_all(pool) + .await?; let historical: Vec<(String, f64)> = sqlx::query_as( "SELECT error_type, COUNT(*)::float8 / 4.0 FROM brain_errors \ WHERE created_at BETWEEN NOW() - INTERVAL '37 days' AND NOW() - INTERVAL '7 days' \ - GROUP BY error_type" - ).fetch_all(pool).await?; + GROUP BY error_type", + ) + .fetch_all(pool) + .await?; // Chi-squared test - let hist_map: std::collections::HashMap = - historical.into_iter().collect(); + let hist_map: std::collections::HashMap = historical.into_iter().collect(); let mut chi_squared = 0.0; let mut categories = 0u32; for (error_type, observed) in &recent { if let Some(&expected) = hist_map.get(error_type) - && expected > 0.0 { - chi_squared += (*observed as f64 - expected).powi(2) / expected; - categories += 1; - } + && expected > 0.0 + { + chi_squared += (*observed as f64 - expected).powi(2) / expected; + categories += 1; + } } let df = (categories as i32 - 1).max(1); // Simplified p-value approximation (Wilson-Hilferty for chi-squared CDF) let p_value = chi2_survival(chi_squared, df as f64); - let drift_data: Vec = recent.iter() + let drift_data: Vec = recent + .iter() .map(|(t, c)| serde_json::json!({"error_type": t, "count": c})) .collect(); @@ -163,13 +193,16 @@ async fn drift(pool: &PgPool) -> Result { async fn communities(pool: &PgPool) -> Result { match crate::graph::community::detect(pool).await { Ok(communities) => { - let community_json: Vec = communities.iter().map(|(id, members)| { - serde_json::json!({ - "community_id": id, - "size": members.len(), - "members": members + let community_json: Vec = communities + .iter() + .map(|(id, members)| { + serde_json::json!({ + "community_id": id, + "size": members.len(), + "members": members + }) }) - }).collect(); + .collect(); Ok(serde_json::json!({ "metric": "communities", "algorithm": "leiden", @@ -181,9 +214,12 @@ async fn communities(pool: &PgPool) -> Result { tracing::warn!(error = %e, "Leiden community detection failed, using fallback"); let components: Vec<(String, i64)> = sqlx::query_as( "SELECT e.entity_type, COUNT(*) FROM brain_entities e \ - GROUP BY e.entity_type ORDER BY COUNT(*) DESC" - ).fetch_all(pool).await?; - let communities: Vec = components.iter() + GROUP BY e.entity_type ORDER BY COUNT(*) DESC", + ) + .fetch_all(pool) + .await?; + let communities: Vec = components + .iter() .map(|(t, c)| serde_json::json!({"type": t, "size": c})) .collect(); Ok(serde_json::json!({ @@ -199,12 +235,15 @@ async fn communities(pool: &PgPool) -> Result { async fn bridges(pool: &PgPool) -> Result { match crate::graph::centrality::compute_bridges(pool, 10).await { Ok(ranked) => { - let bridge_list: Vec = ranked.iter().map(|(name, centrality)| { - serde_json::json!({ - "entity": name, - "centrality": (centrality * 10000.0).round() / 10000.0 + let bridge_list: Vec = ranked + .iter() + .map(|(name, centrality)| { + serde_json::json!({ + "entity": name, + "centrality": (centrality * 10000.0).round() / 10000.0 + }) }) - }).collect(); + .collect(); Ok(serde_json::json!({ "metric": "bridges", "algorithm": "brandes_betweenness", @@ -218,9 +257,12 @@ async fn bridges(pool: &PgPool) -> Result { "SELECT e.name, COUNT(r.id) as connection_count FROM brain_entities e LEFT JOIN brain_relations r ON e.id = r.from_entity OR e.id = r.to_entity GROUP BY e.name HAVING COUNT(r.id) > 2 - ORDER BY connection_count DESC LIMIT 10" - ).fetch_all(pool).await?; - let bridge_list: Vec = bridges.iter() + ORDER BY connection_count DESC LIMIT 10", + ) + .fetch_all(pool) + .await?; + let bridge_list: Vec = bridges + .iter() .map(|(n, c)| serde_json::json!({"entity": n, "connections": c})) .collect(); Ok(serde_json::json!({ @@ -256,18 +298,16 @@ fn chi2_survival(x: f64, df: f64) -> f64 { if x <= 0.0 || df <= 0.0 { return 1.0; } - let z = ((x / df).powf(1.0 / 3.0) - (1.0 - 2.0 / (9.0 * df))) - / (2.0 / (9.0 * df)).sqrt(); + let z = ((x / df).powf(1.0 / 3.0) - (1.0 - 2.0 / (9.0 * df))) / (2.0 / (9.0 * df)).sqrt(); 0.5 * erfc_approx(z / std::f64::consts::SQRT_2) } /// Approximate complementary error function (Abramowitz & Stegun 7.1.26). fn erfc_approx(x: f64) -> f64 { let t = 1.0 / (1.0 + 0.3275911 * x.abs()); - let poly = t * (0.254829592 - + t * (-0.284496736 - + t * (1.421413741 - + t * (-1.453152027 + t * 1.061405429)))); + let poly = t + * (0.254829592 + + t * (-0.284496736 + t * (1.421413741 + t * (-1.453152027 + t * 1.061405429)))); let result = poly * (-x * x).exp(); if x >= 0.0 { result } else { 2.0 - result } } diff --git a/rust/src/handlers/zafra.rs b/rust/src/handlers/zafra.rs index 1b0266a..9e41bd1 100644 --- a/rust/src/handlers/zafra.rs +++ b/rust/src/handlers/zafra.rs @@ -17,13 +17,21 @@ pub async fn handle(pool: &PgPool, args: Value) -> Result { Ok(serde_json::json!({"action": "decay", "decayed": decayed})) } "prune" => { - let threshold = args.get("threshold").and_then(|v| v.as_f64()).unwrap_or(0.1); + let threshold = args + .get("threshold") + .and_then(|v| v.as_f64()) + .unwrap_or(0.1); let result = sqlx::query("DELETE FROM brain_observations WHERE importance < $1 AND observation_type NOT IN ('decision', 'lesson')") .bind(threshold).execute(pool).await?; - Ok(serde_json::json!({"action": "prune", "pruned": result.rows_affected(), "threshold": threshold})) + Ok( + serde_json::json!({"action": "prune", "pruned": result.rows_affected(), "threshold": threshold}), + ) } "merge" => { - let sim_threshold = args.get("similarity_threshold").and_then(|v| v.as_f64()).unwrap_or(0.8); + let sim_threshold = args + .get("similarity_threshold") + .and_then(|v| v.as_f64()) + .unwrap_or(0.8); // P2 FIX: Batch merge — find duplicates in one query, merge in batch let dupes: Vec<(uuid::Uuid, uuid::Uuid, f64)> = sqlx::query_as( "SELECT a.id, b.id, similarity(a.content, b.content)::float8 AS sim @@ -33,41 +41,72 @@ pub async fn handle(pool: &PgPool, args: Value) -> Result { ).bind(sim_threshold).fetch_all(pool).await?; // FIX R-002: Atomic transaction — all-or-nothing merge - let mut tx = pool.begin().await.context("failed to begin merge transaction")?; + let mut tx = pool + .begin() + .await + .context("failed to begin merge transaction")?; let mut merged = 0u32; for (keep_id, remove_id, _) in &dupes { - sqlx::query("UPDATE brain_observations SET observation_type = 'superseded' WHERE id = $1").bind(remove_id).execute(&mut *tx).await?; + sqlx::query( + "UPDATE brain_observations SET observation_type = 'superseded' WHERE id = $1", + ) + .bind(remove_id) + .execute(&mut *tx) + .await?; sqlx::query("UPDATE brain_observations SET importance = LEAST(importance + 0.05, 1.0) WHERE id = $1").bind(keep_id).execute(&mut *tx).await?; merged += 1; } - tx.commit().await.context("failed to commit merge transaction")?; + tx.commit() + .await + .context("failed to commit merge transaction")?; Ok(serde_json::json!({"action": "merge", "merged": merged, "threshold": sim_threshold})) } "stats" => { - let entities: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM brain_entities").fetch_one(pool).await?; - let observations: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM brain_observations").fetch_one(pool).await?; - let superseded: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM brain_observations WHERE observation_type = 'superseded'").fetch_one(pool).await?; - Ok(serde_json::json!({"action": "stats", "entities": entities.0, "observations": observations.0, "superseded": superseded.0, "active": observations.0 - superseded.0})) + let entities: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM brain_entities") + .fetch_one(pool) + .await?; + let observations: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM brain_observations") + .fetch_one(pool) + .await?; + let superseded: (i64,) = sqlx::query_as( + "SELECT COUNT(*) FROM brain_observations WHERE observation_type = 'superseded'", + ) + .fetch_one(pool) + .await?; + Ok( + serde_json::json!({"action": "stats", "entities": entities.0, "observations": observations.0, "superseded": superseded.0, "active": observations.0 - superseded.0}), + ) } "pagerank" => { let ranked = crate::graph::pagerank::compute_and_store(pool).await?; Ok(serde_json::json!({"action": "pagerank", "updated": ranked})) } "summarize" => { - let entity_name = args.get("entity_name").and_then(|v| v.as_str()).unwrap_or(""); - let summary = args.get("compressed_summary").and_then(|v| v.as_str()).unwrap_or(""); + let entity_name = args + .get("entity_name") + .and_then(|v| v.as_str()) + .unwrap_or(""); + let summary = args + .get("compressed_summary") + .and_then(|v| v.as_str()) + .unwrap_or(""); if entity_name.is_empty() || summary.is_empty() { anyhow::bail!("entity_name and compressed_summary are required"); } - let entity_id: (uuid::Uuid,) = sqlx::query_as("SELECT id FROM brain_entities WHERE name = $1") - .bind(entity_name).fetch_one(pool).await?; + let entity_id: (uuid::Uuid,) = + sqlx::query_as("SELECT id FROM brain_entities WHERE name = $1") + .bind(entity_name) + .fetch_one(pool) + .await?; // Mark old observations as superseded let marked = sqlx::query("UPDATE brain_observations SET observation_type = 'superseded' WHERE entity_id = $1 AND observation_type != 'superseded'") .bind(entity_id.0).execute(pool).await?; // Insert new summary sqlx::query("INSERT INTO brain_observations (entity_id, content, observation_type, source) VALUES ($1, $2, 'fact', 'consolidation')") .bind(entity_id.0).bind(summary).execute(pool).await?; - Ok(serde_json::json!({"action": "summarize", "entity": entity_name, "superseded": marked.rows_affected()})) + Ok( + serde_json::json!({"action": "summarize", "entity": entity_name, "superseded": marked.rows_affected()}), + ) } "find_duplicates" => { let dupes: Vec<(String, String, f64)> = sqlx::query_as( @@ -77,17 +116,26 @@ pub async fn handle(pool: &PgPool, args: Value) -> Result { ORDER BY sim DESC LIMIT 20" ).fetch_all(pool).await?; // FIX R-001: safe_truncate prevents panic on multi-byte UTF-8 - let results: Vec = dupes.iter().map(|(a, b, s)| serde_json::json!({ - "content_a": safe_truncate(a, 100), - "content_b": safe_truncate(b, 100), - "similarity": s - })).collect(); - Ok(serde_json::json!({"action": "find_duplicates", "duplicates": results, "count": results.len()})) + let results: Vec = dupes + .iter() + .map(|(a, b, s)| { + serde_json::json!({ + "content_a": safe_truncate(a, 100), + "content_b": safe_truncate(b, 100), + "similarity": s + }) + }) + .collect(); + Ok( + serde_json::json!({"action": "find_duplicates", "duplicates": results, "count": results.len()}), + ) } "export" => { let entities: Vec<(uuid::Uuid, String, String, f64)> = sqlx::query_as("SELECT id, name, entity_type, importance FROM brain_entities ORDER BY importance DESC LIMIT 500").fetch_all(pool).await?; let ent_json: Vec = entities.iter().map(|(id, n, t, i)| serde_json::json!({"id": id.to_string(), "name": n, "type": t, "importance": i})).collect(); - Ok(serde_json::json!({"action": "export", "entities": ent_json, "count": ent_json.len()})) + Ok( + serde_json::json!({"action": "export", "entities": ent_json, "count": ent_json.len()}), + ) } "backfill" => { // Backfill missing Dual-Strength columns with defaults diff --git a/rust/src/main.rs b/rust/src/main.rs index a9bb63f..ac84654 100644 --- a/rust/src/main.rs +++ b/rust/src/main.rs @@ -21,10 +21,7 @@ async fn main() { ) .init(); - tracing::info!( - version = env!("CARGO_PKG_VERSION"), - "cuba-memorys starting" - ); + tracing::info!(version = env!("CARGO_PKG_VERSION"), "cuba-memorys starting"); // Graceful shutdown on SIGTERM/SIGINT let shutdown = async { diff --git a/rust/src/protocol.rs b/rust/src/protocol.rs index b16ff8c..3229da8 100644 --- a/rust/src/protocol.rs +++ b/rust/src/protocol.rs @@ -223,10 +223,7 @@ async fn rem_daemon(pool: PgPool) { // Run consolidation in a spawned task (I/O-bound: DB queries + graph ops) let pool_clone = pool.clone(); - let result = tokio::spawn(async move { - run_rem_consolidation(&pool_clone).await - }) - .await; + let result = tokio::spawn(async move { run_rem_consolidation(&pool_clone).await }).await; match result { Ok(Ok(())) => tracing::info!("REM sleep cycle completed"), @@ -259,7 +256,7 @@ async fn run_rem_consolidation(pool: &PgPool) -> Result<()> { // Protect entities accessed during active session (last 8h) sqlx::query_scalar( "SELECT DISTINCT entity_id FROM brain_observations - WHERE created_at > NOW() - INTERVAL '8 hours'" + WHERE created_at > NOW() - INTERVAL '8 hours'", ) .fetch_all(pool) .await? diff --git a/rust/src/search/cache.rs b/rust/src/search/cache.rs index 833a53c..9003c25 100644 --- a/rust/src/search/cache.rs +++ b/rust/src/search/cache.rs @@ -51,10 +51,13 @@ impl TtlLruCache { /// Insert value into cache. pub fn put(&mut self, key: String, value: V) { - self.inner.put(key, CacheEntry { - value, - inserted_at: Instant::now(), - }); + self.inner.put( + key, + CacheEntry { + value, + inserted_at: Instant::now(), + }, + ); } /// Get cache statistics. @@ -69,7 +72,9 @@ impl TtlLruCache { /// Evict expired entries proactively. pub fn evict_expired(&mut self) { - let keys_to_remove: Vec = self.inner.iter() + let keys_to_remove: Vec = self + .inner + .iter() .filter(|(_, entry)| entry.inserted_at.elapsed() >= self.ttl) .map(|(key, _)| key.clone()) .collect(); diff --git a/rust/src/search/confidence.rs b/rust/src/search/confidence.rs index c6cbf8b..cc40cae 100644 --- a/rust/src/search/confidence.rs +++ b/rust/src/search/confidence.rs @@ -6,10 +6,7 @@ /// /// Returns (confidence, level) where level is one of: /// "verified" (>0.7), "partial" (>0.5), "weak" (>0.3), "unknown" (<0.3). -pub fn compute_grounding( - similarities: &[f64], - sources: &[&str], -) -> (f64, &'static str) { +pub fn compute_grounding(similarities: &[f64], sources: &[&str]) -> (f64, &'static str) { if similarities.is_empty() { return (0.0, "unknown"); } @@ -24,8 +21,8 @@ pub fn compute_grounding( // Weights sum to 1.0: 0.45 + 0.25 + 0.20 + 0.10 = 1.00 let coverage = (count as f64 / 10.0).min(1.0); - let confidence = (max_sim * 0.45 + avg_sim * 0.25 + coverage * 0.20 + diversity * 0.10) - .min(1.0); + let confidence = + (max_sim * 0.45 + avg_sim * 0.25 + coverage * 0.20 + diversity * 0.10).min(1.0); let level = if confidence > 0.7 { "verified" @@ -63,10 +60,7 @@ mod tests { #[test] fn test_weak_evidence() { - let (conf, _level) = compute_grounding( - &[0.35], - &["agent"], - ); + let (conf, _level) = compute_grounding(&[0.35], &["agent"]); assert!(conf < 0.5, "single weak match: got {conf}"); } } diff --git a/rust/src/search/rrf.rs b/rust/src/search/rrf.rs index fa5409a..1268887 100644 --- a/rust/src/search/rrf.rs +++ b/rust/src/search/rrf.rs @@ -57,7 +57,9 @@ pub fn fuse( for (rank, result) in results.iter().enumerate() { let rrf_score = weight / (RRF_K + rank as f64 + 1.0); *scores.entry(result.id.clone()).or_default() += rrf_score; - items.entry(result.id.clone()).or_insert_with(|| result.clone()); + items + .entry(result.id.clone()) + .or_insert_with(|| result.clone()); } } @@ -73,9 +75,9 @@ pub fn fuse( let mut unique: Vec = Vec::new(); for (id, score) in sorted { if let Some(mut item) = items.remove(&id) { - let is_dup = unique.iter().any(|existing| { - text_overlap(&item.content, &existing.content) > dedup_threshold - }); + let is_dup = unique + .iter() + .any(|existing| text_overlap(&item.content, &existing.content) > dedup_threshold); if !is_dup { item.score = score; unique.push(item); @@ -113,7 +115,10 @@ mod tests { #[test] fn test_query_entropy_repetitive() { let e = query_entropy("hello hello hello"); - assert!(e < 0.01, "repetitive query should have near-zero entropy: got {e}"); + assert!( + e < 0.01, + "repetitive query should have near-zero entropy: got {e}" + ); } #[test] @@ -129,12 +134,32 @@ mod tests { #[test] fn test_rrf_fusion_basic() { let signal1 = vec![ - RankedResult { id: "a".into(), content: "alpha".into(), score: 0.0, source: "text".into() }, - RankedResult { id: "b".into(), content: "beta".into(), score: 0.0, source: "text".into() }, + RankedResult { + id: "a".into(), + content: "alpha".into(), + score: 0.0, + source: "text".into(), + }, + RankedResult { + id: "b".into(), + content: "beta".into(), + score: 0.0, + source: "text".into(), + }, ]; let signal2 = vec![ - RankedResult { id: "b".into(), content: "beta".into(), score: 0.0, source: "vec".into() }, - RankedResult { id: "c".into(), content: "gamma".into(), score: 0.0, source: "vec".into() }, + RankedResult { + id: "b".into(), + content: "beta".into(), + score: 0.0, + source: "vec".into(), + }, + RankedResult { + id: "c".into(), + content: "gamma".into(), + score: 0.0, + source: "vec".into(), + }, ]; let fused = fuse(&[(signal1, 0.5), (signal2, 0.5)], 0.75); @@ -146,16 +171,21 @@ mod tests { #[test] fn test_rrf_k60_deterministic() { // V4: k=60 always, deterministic scores - let signal = vec![ - RankedResult { id: "a".into(), content: "alpha".into(), score: 0.0, source: "text".into() }, - ]; + let signal = vec![RankedResult { + id: "a".into(), + content: "alpha".into(), + score: 0.0, + source: "text".into(), + }]; let fused = fuse(&[(signal, 1.0)], 0.75); // Score should be exactly 1.0 / (60.0 + 0 + 1.0) = 1/61 let expected = 1.0 / 61.0; assert!( (fused[0].score - expected).abs() < 1e-10, - "k=60 fixed: expected {} got {}", expected, fused[0].score + "k=60 fixed: expected {} got {}", + expected, + fused[0].score ); } } diff --git a/rust/tests/integration.rs b/rust/tests/integration.rs index 199ba68..f5f0e51 100644 --- a/rust/tests/integration.rs +++ b/rust/tests/integration.rs @@ -21,9 +21,10 @@ fn unique_name(prefix: &str) -> String { #[tokio::test] #[ignore] async fn test_all_integration() { - let url = std::env::var("DATABASE_URL") - .expect("DATABASE_URL env var required for integration tests"); - let pool = cuba_memorys::db::create_pool(&url).await + let url = + std::env::var("DATABASE_URL").expect("DATABASE_URL env var required for integration tests"); + let pool = cuba_memorys::db::create_pool(&url) + .await .expect("Failed to connect to test database"); // ── 1. Schema validation ────────────────────────────────────── @@ -32,25 +33,36 @@ async fn test_all_integration() { let tables: Vec<(String,)> = sqlx::query_as( "SELECT table_name::text FROM information_schema.tables WHERE table_schema = 'public' AND table_name LIKE 'brain_%' - ORDER BY table_name" + ORDER BY table_name", ) .fetch_all(&pool) .await .expect("Failed to query tables"); let names: Vec<&str> = tables.iter().map(|(n,)| n.as_str()).collect(); - for required in &["brain_entities", "brain_observations", "brain_relations", "brain_errors", "brain_sessions"] { + for required in &[ + "brain_entities", + "brain_observations", + "brain_relations", + "brain_errors", + "brain_sessions", + ] { assert!(names.contains(required), "Missing table: {required}"); } - println!(" ✓ All required tables exist ({} brain_* tables)", names.len()); + println!( + " ✓ All required tables exist ({} brain_* tables)", + names.len() + ); } // ── 2. pgvector extension ───────────────────────────────────── println!(" [2/7] pgvector extension..."); { - let row: (bool,) = sqlx::query_as( - "SELECT EXISTS(SELECT 1 FROM pg_extension WHERE extname = 'vector')" - ).fetch_one(&pool).await.unwrap(); + let row: (bool,) = + sqlx::query_as("SELECT EXISTS(SELECT 1 FROM pg_extension WHERE extname = 'vector')") + .fetch_one(&pool) + .await + .unwrap(); assert!(row.0, "pgvector extension required"); println!(" ✓ pgvector extension detected"); } @@ -60,15 +72,24 @@ async fn test_all_integration() { { let name = unique_name("alma"); let result = cuba_memorys::handlers::dispatch( - &pool, "cuba_alma", + &pool, + "cuba_alma", json!({ "action": "create", "name": &name, "entity_type": "concept" }), - ).await.unwrap(); - assert!(result.get("content").is_some(), "create should return content"); + ) + .await + .unwrap(); + assert!( + result.get("content").is_some(), + "create should return content" + ); let result = cuba_memorys::handlers::dispatch( - &pool, "cuba_alma", + &pool, + "cuba_alma", json!({ "action": "get", "name": &name }), - ).await.unwrap(); + ) + .await + .unwrap(); assert!(result.get("content").is_some(), "get should return content"); println!(" ✓ alma create + get OK (entity: {name})"); } @@ -78,7 +99,8 @@ async fn test_all_integration() { { let name = unique_name("cronica"); let result = cuba_memorys::handlers::dispatch( - &pool, "cuba_cronica", + &pool, + "cuba_cronica", json!({ "action": "add", "entity_name": &name, @@ -86,14 +108,22 @@ async fn test_all_integration() { "observation_type": "fact", "source": "agent" }), - ).await.unwrap(); + ) + .await + .unwrap(); assert!(result.get("content").is_some(), "add should return content"); let result = cuba_memorys::handlers::dispatch( - &pool, "cuba_cronica", + &pool, + "cuba_cronica", json!({ "action": "list", "entity_name": &name }), - ).await.unwrap(); - assert!(result.get("content").is_some(), "list should return content"); + ) + .await + .unwrap(); + assert!( + result.get("content").is_some(), + "list should return content" + ); println!(" ✓ cronica add + list OK (entity: {name})"); } @@ -112,10 +142,16 @@ async fn test_all_integration() { ).await; let result = cuba_memorys::handlers::dispatch( - &pool, "cuba_faro", + &pool, + "cuba_faro", json!({ "query": "rust programming safety", "mode": "hybrid", "limit": 5 }), - ).await.unwrap(); - assert!(result.get("content").is_some(), "faro should return content"); + ) + .await + .unwrap(); + assert!( + result.get("content").is_some(), + "faro should return content" + ); println!(" ✓ faro search OK"); } @@ -123,36 +159,50 @@ async fn test_all_integration() { println!(" [6/7] jornada lifecycle..."); { let result = cuba_memorys::handlers::dispatch( - &pool, "cuba_jornada", + &pool, + "cuba_jornada", json!({ "action": "start", "name": &unique_name("session"), "goals": ["test session management"] }), - ).await.unwrap(); - assert!(result.get("content").is_some(), "start should return content"); - - let result = cuba_memorys::handlers::dispatch( - &pool, "cuba_jornada", - json!({ "action": "current" }), - ).await.unwrap(); - assert!(result.get("content").is_some(), "current should return content"); + ) + .await + .unwrap(); + assert!( + result.get("content").is_some(), + "start should return content" + ); + + let result = + cuba_memorys::handlers::dispatch(&pool, "cuba_jornada", json!({ "action": "current" })) + .await + .unwrap(); + assert!( + result.get("content").is_some(), + "current should return content" + ); let _ = cuba_memorys::handlers::dispatch( - &pool, "cuba_jornada", + &pool, + "cuba_jornada", json!({ "action": "end", "outcome": "success", "summary": "Integration test done" }), - ).await; + ) + .await; println!(" ✓ jornada start + current + end OK"); } // ── 7. vigia summary ────────────────────────────────────────── println!(" [7/7] vigia summary..."); { - let result = cuba_memorys::handlers::dispatch( - &pool, "cuba_vigia", - json!({ "metric": "summary" }), - ).await.unwrap(); - assert!(result.get("content").is_some(), "vigia should return content"); + let result = + cuba_memorys::handlers::dispatch(&pool, "cuba_vigia", json!({ "metric": "summary" })) + .await + .unwrap(); + assert!( + result.get("content").is_some(), + "vigia should return content" + ); println!(" ✓ vigia summary OK"); } diff --git a/rust/tests/smoke_test.rs b/rust/tests/smoke_test.rs index a495a8c..a850ce4 100644 --- a/rust/tests/smoke_test.rs +++ b/rust/tests/smoke_test.rs @@ -3,22 +3,37 @@ //! Validates JSON-RPC message format, tool definitions, and //! protocol invariants without requiring a live database. -use serde_json::{json, Value}; +use serde_json::{Value, json}; /// Verify all 12 tools are defined in constants. #[test] fn test_all_tools_defined() { let tools: Vec = cuba_memorys::constants::tool_definitions(); - assert_eq!(tools.len(), 13, "Expected 13 MCP tools, got {}", tools.len()); + assert_eq!( + tools.len(), + 13, + "Expected 13 MCP tools, got {}", + tools.len() + ); - let tool_names: Vec<&str> = tools.iter() + let tool_names: Vec<&str> = tools + .iter() .filter_map(|t: &Value| t.get("name").and_then(|n: &Value| n.as_str())) .collect(); let expected = [ - "cuba_alma", "cuba_cronica", "cuba_faro", "cuba_puente", "cuba_eco", - "cuba_alarma", "cuba_remedio", "cuba_expediente", "cuba_jornada", - "cuba_decreto", "cuba_vigia", "cuba_zafra", + "cuba_alma", + "cuba_cronica", + "cuba_faro", + "cuba_puente", + "cuba_eco", + "cuba_alarma", + "cuba_remedio", + "cuba_expediente", + "cuba_jornada", + "cuba_decreto", + "cuba_vigia", + "cuba_zafra", ]; for name in &expected { @@ -32,11 +47,20 @@ fn test_tool_schema_structure() { let tools: Vec = cuba_memorys::constants::tool_definitions(); for tool in &tools { - let name = tool.get("name").and_then(|n: &Value| n.as_str()).unwrap_or("???"); + let name = tool + .get("name") + .and_then(|n: &Value| n.as_str()) + .unwrap_or("???"); assert!(tool.get("name").is_some(), "{name}: missing 'name'"); - assert!(tool.get("description").is_some(), "{name}: missing 'description'"); - assert!(tool.get("inputSchema").is_some(), "{name}: missing 'inputSchema'"); + assert!( + tool.get("description").is_some(), + "{name}: missing 'description'" + ); + assert!( + tool.get("inputSchema").is_some(), + "{name}: missing 'inputSchema'" + ); let schema = tool.get("inputSchema").unwrap(); assert_eq!( @@ -134,9 +158,18 @@ fn test_threshold_invariants() { #[test] fn test_handler_dispatch_coverage() { let tool_names = [ - "cuba_alma", "cuba_cronica", "cuba_faro", "cuba_puente", "cuba_eco", - "cuba_alarma", "cuba_remedio", "cuba_expediente", "cuba_jornada", - "cuba_decreto", "cuba_vigia", "cuba_zafra", + "cuba_alma", + "cuba_cronica", + "cuba_faro", + "cuba_puente", + "cuba_eco", + "cuba_alarma", + "cuba_remedio", + "cuba_expediente", + "cuba_jornada", + "cuba_decreto", + "cuba_vigia", + "cuba_zafra", ]; for name in &tool_names { @@ -151,14 +184,23 @@ fn test_schema_sql_content() { let schema = include_str!("../src/schema.sql"); assert!(!schema.is_empty()); - for table in &["brain_entities", "brain_observations", "brain_relations", "brain_errors", "brain_sessions"] { + for table in &[ + "brain_entities", + "brain_observations", + "brain_relations", + "brain_errors", + "brain_sessions", + ] { assert!(schema.contains(table), "Missing table: {table}"); } assert!(schema.contains("vector"), "Missing pgvector"); assert!(schema.contains("pg_trgm"), "Missing pg_trgm"); assert!(schema.contains("storage_strength"), "Missing Dual-Strength"); - assert!(schema.contains("retrieval_strength"), "Missing Dual-Strength"); + assert!( + schema.contains("retrieval_strength"), + "Missing Dual-Strength" + ); } /// Verify cognitive module constants are valid.