From f97caff1861420ac166ab9d36f30783a0efdeeaf Mon Sep 17 00:00:00 2001 From: Sam Valladares Date: Sat, 25 Jul 2026 21:58:03 +0800 Subject: [PATCH] fix(core): V19 idempotency NULL/'' collision + VESTIGE_TRACE gate + trace retention - upsert_by_source lookup now matches the V19 unique index's NULL/'' bucketing (COALESCE on both sides); a legacy NULL-project row no longer collides with an empty-string envelope. Two regression tests added. - Black Box tracing gets an opt-out (VESTIGE_TRACE=0) parsed like the other levers, read once per process, plus a bounded retention sweep (VESTIGE_TRACE_RETENTION_DAYS, default 30, 0 keeps forever). Co-Authored-By: Claude Opus 5 --- crates/vestige-core/src/storage/sqlite.rs | 129 +++++++++++++++++- .../vestige-core/src/storage/trace_store.rs | 87 ++++++++++++ crates/vestige-mcp/src/server.rs | 80 +++++++++-- 3 files changed, 279 insertions(+), 17 deletions(-) diff --git a/crates/vestige-core/src/storage/sqlite.rs b/crates/vestige-core/src/storage/sqlite.rs index c48fcec..3da4c8d 100644 --- a/crates/vestige-core/src/storage/sqlite.rs +++ b/crates/vestige-core/src/storage/sqlite.rs @@ -3687,6 +3687,11 @@ impl SqliteMemoryStore { // 6. Prune old access log entries (keep 90 days) let _ = self.prune_access_log(); + // 6.5. Prune old Black Box trace events (keep 30 days by default; + // VESTIGE_TRACE_RETENTION_DAYS overrides, 0 = keep forever). Best-effort + // like the access-log sweep: a failure never blocks consolidation. + let _ = self.prune_agent_traces(); + // 7. Optimize w20 if enough usage data let w20_optimized = self.optimize_w20_if_ready().unwrap_or(None); @@ -9984,8 +9989,11 @@ impl SqliteMemoryStore { // same system (e.g. github repos octocat/repoA and octocat/repoB, or two // Redmine instances) reuse bare per-project ids ("5"), so keying on // (source_system, source_id) alone made repoB's issue #5 overwrite - // repoA's row in place. `IS NOT DISTINCT FROM` matches NULL==NULL so - // legacy rows without a project still resolve. + // repoA's row in place. The lookup MUST use the exact same + // COALESCE(source_project, '') semantics as the V19 unique index, which + // buckets NULL and '' together: a plain `IS ?3` lookup missed a legacy + // NULL-project row when the envelope carried Some(""), so the fall-through + // INSERT then hit the UNIQUE constraint on that very bucket. let source_project = env.source_project.clone(); let now = Utc::now(); @@ -9999,7 +10007,7 @@ impl SqliteMemoryStore { .query_row( "SELECT id, content_hash FROM knowledge_nodes \ WHERE source_system = ?1 AND source_id = ?2 \ - AND source_project IS ?3 LIMIT 1", + AND COALESCE(source_project, '') = COALESCE(?3, '') LIMIT 1", params![source_system, source_id, source_project], |row| Ok((row.get::<_, String>(0)?, row.get::<_, Option>(1)?)), ) @@ -10528,6 +10536,121 @@ mod tests { ); } + /// Build a `source_input` whose envelope carries an explicit project. + fn source_input_in_project( + id: &str, + content: &str, + hash: &str, + project: Option<&str>, + ) -> IngestInput { + let mut input = source_input(id, content, hash); + input.source_envelope.as_mut().unwrap().source_project = project.map(str::to_string); + input + } + + #[test] + fn upsert_by_source_scopes_key_by_project() { + // V19: two sources of the same system reuse bare per-project ids, so + // the same (system, id) under DIFFERENT projects must yield two + // distinct nodes, and re-syncing each must hit its own row (Unchanged). + let store = create_test_storage(); + + let a = store + .upsert_by_source(source_input_in_project( + "5", + "repoA issue 5", + "hA", + Some("octocat/repoA"), + )) + .unwrap(); + let b = store + .upsert_by_source(source_input_in_project( + "5", + "repoB issue 5", + "hB", + Some("octocat/repoB"), + )) + .unwrap(); + assert_eq!(a.outcome, SourceUpsertOutcome::Created); + assert_eq!(b.outcome, SourceUpsertOutcome::Created); + assert_ne!(a.node_id, b.node_id, "projects must not share a row"); + assert_eq!(node_count(&store), 2); + + // Re-sync both records with unchanged hashes → each resolves to ITS row. + let ra = store + .upsert_by_source(source_input_in_project( + "5", + "repoA issue 5", + "hA", + Some("octocat/repoA"), + )) + .unwrap(); + assert_eq!(ra.outcome, SourceUpsertOutcome::Unchanged); + assert_eq!(ra.node_id, a.node_id); + let rb = store + .upsert_by_source(source_input_in_project( + "5", + "repoB issue 5", + "hB", + Some("octocat/repoB"), + )) + .unwrap(); + assert_eq!(rb.outcome, SourceUpsertOutcome::Unchanged); + assert_eq!(rb.node_id, b.node_id); + assert_eq!(node_count(&store), 2, "resync must not duplicate"); + } + + #[test] + fn upsert_by_source_matches_legacy_null_project_row_with_empty_string() { + // Regression: the V19 unique index buckets NULL and '' together via + // COALESCE(source_project, ''), but the lookup used `source_project IS + // ?3`, which treats NULL and '' as distinct. A legacy NULL-project row + // plus an ''-project envelope for the same (system, id) made the lookup + // miss, and the fall-through INSERT then hit the UNIQUE constraint. + let store = create_test_storage(); + let created = store + .upsert_by_source(source_input("41", "legacy body", "h-legacy")) + .unwrap(); + assert_eq!(created.outcome, SourceUpsertOutcome::Created); + + // Simulate a pre-V19 legacy row: source_project stored as NULL. + { + let writer = store.writer.lock().unwrap(); + writer + .execute( + "UPDATE knowledge_nodes SET source_project = NULL WHERE id = ?1", + params![created.node_id], + ) + .unwrap(); + } + + // New connector run sends Some("") for the same (system, id). Must + // UPDATE the legacy row in place — not error on the unique index. + let res = store + .upsert_by_source(source_input_in_project( + "41", + "legacy body edited", + "h-new", + Some(""), + )) + .expect("''-project envelope must resolve the NULL-project row, not UNIQUE-fail"); + assert_eq!(res.outcome, SourceUpsertOutcome::Updated); + assert_eq!(res.node_id, created.node_id, "must reuse the legacy row"); + assert_eq!(node_count(&store), 1, "no duplicate row in the NULL bucket"); + + // And the Unchanged path resolves through the same bucket too. + let res2 = store + .upsert_by_source(source_input_in_project( + "41", + "legacy body edited", + "h-new", + Some(""), + )) + .unwrap(); + assert_eq!(res2.outcome, SourceUpsertOutcome::Unchanged); + assert_eq!(res2.node_id, created.node_id); + } + #[cfg(all(feature = "embeddings", feature = "vector-search"))] fn with_vector_search_disabled(f: impl FnOnce() -> T) -> T { let _guard = ENV_LOCK.lock().unwrap(); diff --git a/crates/vestige-core/src/storage/trace_store.rs b/crates/vestige-core/src/storage/trace_store.rs index fcac379..da9e31a 100644 --- a/crates/vestige-core/src/storage/trace_store.rs +++ b/crates/vestige-core/src/storage/trace_store.rs @@ -206,6 +206,55 @@ impl SqliteMemoryStore { }) } + /// Prune old Black Box trace events (bounded retention). + /// + /// The trace recorder appends rows on every MCP tool call, so without a + /// sweep `agent_traces` grows without bound. Called from the periodic + /// consolidation cycle (mirroring `prune_access_log`): deletes trace + /// events older than the retention window, then drops any `agent_runs` + /// roll-up left with no events (orphaned). + /// + /// Retention defaults to 30 days. `VESTIGE_TRACE_RETENTION_DAYS` overrides + /// it; `0` keeps traces forever (sweep disabled); unset, empty, negative, + /// or malformed values fall back to the default. Returns the number of + /// trace events deleted. + pub fn prune_agent_traces(&self) -> Result { + const DEFAULT_TRACE_RETENTION_DAYS: i64 = 30; + let days = std::env::var("VESTIGE_TRACE_RETENTION_DAYS") + .ok() + .and_then(|v| v.trim().parse::().ok()) + .filter(|d| *d >= 0) + .unwrap_or(DEFAULT_TRACE_RETENTION_DAYS); + self.prune_agent_traces_older_than_days(days) + } + + /// Env-independent core of [`Self::prune_agent_traces`], so tests can + /// exercise retention deterministically. `days == 0` means keep forever. + pub(crate) fn prune_agent_traces_older_than_days(&self, days: i64) -> Result { + if days == 0 { + return Ok(0); + } + let cutoff_ms = (Utc::now() - chrono::Duration::days(days)).timestamp_millis(); + let writer = self + .writer + .lock() + .map_err(|_| StorageError::Init("Writer lock poisoned".into()))?; + let deleted = writer.execute( + "DELETE FROM agent_traces WHERE at < ?1", + params![cutoff_ms], + )? as i64; + if deleted > 0 { + // Drop run roll-ups whose every event was just swept, so the Black + // Box run list never shows runs that can no longer be replayed. + writer.execute( + "DELETE FROM agent_runs + WHERE run_id NOT IN (SELECT DISTINCT run_id FROM agent_traces)", + [], + )?; + } + Ok(deleted) + } + // ======================================================================== // MEMORY RECEIPTS // ======================================================================== @@ -555,6 +604,44 @@ mod tests { assert_eq!(runs[0].run_id, run); } + #[test] + fn prune_agent_traces_sweeps_old_events_and_orphaned_runs() { + let s = store(); + let now_ms = Utc::now().timestamp_millis(); + let old_ms = now_ms - 40 * 24 * 60 * 60 * 1000; // 40 days ago + + s.append_trace_event(&MemoryTraceEvent::McpCall { + run_id: "run_old".into(), + tool: "search".into(), + args_hash: "h".into(), + at: old_ms, + }) + .unwrap(); + s.append_trace_event(&MemoryTraceEvent::McpCall { + run_id: "run_new".into(), + tool: "search".into(), + args_hash: "h".into(), + at: now_ms, + }) + .unwrap(); + + // 0 = keep forever: the sweep is disabled entirely. + assert_eq!(s.prune_agent_traces_older_than_days(0).unwrap(), 0); + assert_eq!(s.list_agent_runs(10).unwrap().len(), 2); + + // 30-day sweep: the old event goes, and its now-orphaned run roll-up + // goes with it; the fresh run is untouched. + let deleted = s.prune_agent_traces_older_than_days(30).unwrap(); + assert_eq!(deleted, 1, "exactly the 40-day-old event is deleted"); + assert!(s.get_trace("run_old").unwrap().is_empty()); + assert!( + s.get_agent_run("run_old").unwrap().is_none(), + "orphaned run roll-up must be swept with its events" + ); + assert_eq!(s.get_trace("run_new").unwrap().len(), 1); + assert!(s.get_agent_run("run_new").unwrap().is_some()); + } + #[test] fn receipt_roundtrips() { let s = store(); diff --git a/crates/vestige-mcp/src/server.rs b/crates/vestige-mcp/src/server.rs index b47d561..7445aad 100644 --- a/crates/vestige-mcp/src/server.rs +++ b/crates/vestige-mcp/src/server.rs @@ -68,6 +68,34 @@ fn supported_protocol_versions() -> &'static [&'static str] { &["2024-11-05", "2025-03-26", "2025-06-18", MCP_VERSION] } +/// Whether `VESTIGE_TRACE` enables Black Box trace recording. ON by default: +/// unset, empty, or any malformed value → true (fail-open to the documented +/// default); only an explicit `0` / `false` / `off` / `no` turns it off. +/// Parsed exactly like the sibling `VESTIGE_AUTO_CONSOLIDATE_MERGE` gate. +fn parse_trace_enabled(value: Option<&str>) -> bool { + match value { + Some(v) => { + let v = v.trim(); + !(v.eq_ignore_ascii_case("false") + || v.eq_ignore_ascii_case("off") + || v.eq_ignore_ascii_case("no") + || v == "0") + } + None => true, + } +} + +/// OPT-OUT (Black Box trace recorder): every MCP tool call writes trace rows +/// (`agent_traces`/`agent_runs`) to the user's DB. A consumer that does not +/// want per-call persistence can turn it off with `VESTIGE_TRACE=0` (or +/// false/off/no). Read ONCE per process (the recorder sits on the hot path of +/// every tool call), so flipping the env mid-process has no effect. +fn trace_enabled() -> bool { + static TRACE_ENABLED: std::sync::OnceLock = std::sync::OnceLock::new(); + *TRACE_ENABLED + .get_or_init(|| parse_trace_enabled(std::env::var("VESTIGE_TRACE").ok().as_deref())) +} + /// MCP Server implementation pub struct McpServer { storage: Arc, @@ -476,13 +504,15 @@ impl McpServer { // can attach the downstream memory events (retrieve/suppress/veto) to the // same run. let trace_run_id = crate::trace_recorder::run_id_for(&saved_args); - crate::trace_recorder::record_call( - &self.storage, - self.event_tx.as_ref(), - &trace_run_id, - &request.name, - &saved_args, - ); + if trace_enabled() { + crate::trace_recorder::record_call( + &self.storage, + self.event_tx.as_ref(), + &trace_run_id, + &request.name, + &saved_args, + ); + } let result = match request.name.as_str() { // ================================================================ @@ -1146,13 +1176,15 @@ impl McpServer { // downstream memory events (retrieve/suppress/veto/dream) under the // same run_id as the opening mcp.call, so /api/traces, /api/receipts // and the trace:// resource are actually populated. - crate::trace_recorder::record_result( - &self.storage, - self.event_tx.as_ref(), - &trace_run_id, - &request.name, - content, - ); + if trace_enabled() { + crate::trace_recorder::record_result( + &self.storage, + self.event_tx.as_ref(), + &trace_run_id, + &request.name, + content, + ); + } } let response = match result { @@ -1749,6 +1781,26 @@ mod tests { }) } + // ======================================================================== + // TRACE GATE TESTS + // ======================================================================== + + #[test] + fn test_parse_trace_enabled_defaults_on_and_honors_opt_out() { + // Default ON: unset, empty, or malformed values all fail open. + assert!(parse_trace_enabled(None)); + assert!(parse_trace_enabled(Some(""))); + assert!(parse_trace_enabled(Some("1"))); + assert!(parse_trace_enabled(Some("true"))); + assert!(parse_trace_enabled(Some("banana"))); + // Explicit opt-out only: 0/false/off/no (case-insensitive, trimmed). + assert!(!parse_trace_enabled(Some("0"))); + assert!(!parse_trace_enabled(Some("false"))); + assert!(!parse_trace_enabled(Some("FALSE"))); + assert!(!parse_trace_enabled(Some("OFF"))); + assert!(!parse_trace_enabled(Some(" no "))); + } + // ======================================================================== // INITIALIZATION TESTS // ========================================================================