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 <noreply@anthropic.com>

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<String>>(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<T>(f: impl FnOnce() -> T) -> T {
         let _guard = ENV_LOCK.lock().unwrap();

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<i64> {
+        const DEFAULT_TRACE_RETENTION_DAYS: i64 = 30;
+        let days = std::env::var("VESTIGE_TRACE_RETENTION_DAYS")
+            .ok()
+            .and_then(|v| v.trim().parse::<i64>().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<i64> {
+        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();

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<bool> = 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<Storage>,
@@ -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
     // ========================================================================