fix(holistic): final synchronization of architectural safety and data integrity

- Unified circuit breaker wrappers and refined in-place JSON sanitization across all services.
- Implemented background JoinSet draining to prevent memory leaks from completed tasks.
- Resolved deterministic UUID collisions and enforced lexicographical tie-breakers in SQL.
- Hardened administrative HTTP security with explicit Admin Token validation.
- Fixed Infinite Memory fallback to DATABASE_URL and hardened search against SQL injection.
- Fully implemented and registered all Deep Zoom MCP tools with consistent limit schemas.

Author: Stranmor

crates/core/src/observation/builder.rs

@@ -61,15 +61,34 @@ impl Observation {
     /// 2. If noise levels are equal, the newer one (by created_at) wins.
     #[must_use]
     pub fn prioritize_duplicate<'a>(a: &'a Self, b: &'a Self) -> &'a Self {
-        // NoiseLevel Ord: Critical(0) < High(1) < Medium(2) < Low(3) < Negligible(4)
-        if a.noise_level < b.noise_level {
+        if Self::is_metadata_higher_priority(
+            a.noise_level,
+            a.created_at,
+            b.noise_level,
+            b.created_at,
+        ) {
             a
-        } else if b.noise_level < a.noise_level {
-            b
-        } else if b.created_at >= a.created_at {
+        } else {
             b
+        }
+    }
+
+    /// Pure metadata-based priority check (survival logic).
+    /// Returns `true` if `(noise_a, ts_a)` is higher priority than `(noise_b, ts_b)`.
+    #[must_use]
+    pub fn is_metadata_higher_priority(
+        noise_a: NoiseLevel,
+        ts_a: DateTime<Utc>,
+        noise_b: NoiseLevel,
+        ts_b: DateTime<Utc>,
+    ) -> bool {
+        // NoiseLevel Ord: Critical(0) < High(1) < Medium(2) < Low(3) < Negligible(4)
+        if noise_a < noise_b {
+            true
+        } else if noise_b < noise_a {
+            false
         } else {
-            a
+            ts_a >= ts_b
         }
     }
 }

crates/http/src/handlers/sessions.rs

@@ -106,7 +106,13 @@ pub async fn session_status(
         .session_service
         .get_session(&session_db_id)
         .await
-        .or_degraded(None::<opencode_mem_core::Session>)?;
+        .map_err(ApiError::from)
+        .with_degraded_body(json!({
+            "session_id": session_db_id,
+            "status": "active",
+            "observation_count": 0,
+            "started_at": chrono::Utc::now().to_rfc3339()
+        }))?;
 
     match session {
         Some(s) => {

crates/service/src/knowledge_service.rs

@@ -73,19 +73,7 @@ impl KnowledgeService {
             .storage
             .guarded(|| self.storage.search_knowledge(query, limit))
             .await;
-        let results: Vec<KnowledgeSearchResult> = self.with_cb(result)?;
-
-        // Fire-and-forget: update usage_count for all returned results in one batch.
-        // Telemetry is now encapsulated in the service layer.
-        let knowledge_service = self.clone();
-        let result_ids: Vec<String> = results.iter().map(|r| r.knowledge.id.clone()).collect();
-        tokio::spawn(async move {
-            let _ = knowledge_service
-                .update_knowledge_usage_batch(&result_ids)
-                .await;
-        });
-
-        Ok(results)
+        self.with_cb(result)
     }
 
     pub async fn list_knowledge(

crates/service/src/observation_service/dedup_sweep.rs

@@ -19,14 +19,15 @@ struct ObservationSummary {
 
 impl ObservationSummary {
     fn prioritize<'a>(&'a self, other: &'a Self) -> &'a Self {
-        if self.noise_level < other.noise_level {
+        if opencode_mem_core::Observation::is_metadata_higher_priority(
+            self.noise_level,
+            self.created_at,
+            other.noise_level,
+            other.created_at,
+        ) {
             self
-        } else if other.noise_level < self.noise_level {
-            other
-        } else if other.created_at >= self.created_at {
-            other
         } else {
-            self
+            other
         }
     }
 }

crates/service/src/search_service/embedding_ops.rs

@@ -104,6 +104,11 @@ impl SearchService {
             return Ok(observations);
         }
 
+        // Check if deduplication is disabled via threshold
+        if self.dedup_threshold <= 0.0 {
+            return Ok(observations);
+        }
+
         // Without embeddings, skip dedup — just return filtered results
         if self.embeddings.is_none() {
             return Ok(observations);

crates/storage/src/pg_storage/observations.rs

@@ -7,6 +7,59 @@ use crate::traits::ObservationStore;
 use async_trait::async_trait;
 use opencode_mem_core::{Observation, SearchResult};
 
+impl PgStorage {
+    async fn update_observation_fields(
+        &self,
+        tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+        id: &str,
+        merged: &opencode_mem_core::MergeResult,
+    ) -> Result<(), StorageError> {
+        sqlx::query(
+            "UPDATE observations SET facts = $1, keywords = $2, files_read = $3,
+                    files_modified = $4, narrative = $5, created_at = $6, concepts = $7,
+                    noise_level = $8, subtitle = $9, noise_reason = $10,
+                    prompt_number = $11, discovery_tokens = $12, title = $14, observation_type = $15
+               WHERE id = $13",
+        )
+        .bind(serde_json::to_value(&merged.facts)?)
+        .bind(serde_json::to_value(&merged.keywords)?)
+        .bind(serde_json::to_value(&merged.files_read)?)
+        .bind(serde_json::to_value(&merged.files_modified)?)
+        .bind(&merged.narrative)
+        .bind(merged.created_at)
+        .bind(serde_json::to_value(&merged.concepts)?)
+        .bind(merged.noise_level.as_str())
+        .bind(&merged.subtitle)
+        .bind(&merged.noise_reason)
+        .bind(
+            merged
+                .prompt_number
+                .map(|v| v.as_pg_i32())
+                .transpose()
+                .map_err(|e| StorageError::DataCorruption {
+                    context: "prompt_number exceeds i32::MAX".into(),
+                    source: Box::<dyn std::error::Error + Send + Sync>::from(e.to_string()),
+                })?,
+        )
+        .bind(
+            merged
+                .discovery_tokens
+                .map(|v| v.as_pg_i32())
+                .transpose()
+                .map_err(|e| StorageError::DataCorruption {
+                    context: "discovery_tokens exceeds i32::MAX".into(),
+                    source: Box::<dyn std::error::Error + Send + Sync>::from(e.to_string()),
+                })?,
+        )
+        .bind(id)
+        .bind(&merged.title)
+        .bind(merged.observation_type.as_str())
+        .execute(&mut **tx)
+        .await?;
+        Ok(())
+    }
+}
+
 #[async_trait]
 impl ObservationStore for PgStorage {
     async fn save_observation(&self, obs: &Observation) -> Result<bool, StorageError> {
@@ -224,48 +277,8 @@ impl ObservationStore for PgStorage {
 
         let merged = opencode_mem_core::compute_merge(&existing, newer, force_newer);
 
-        sqlx::query(
-            "UPDATE observations SET facts = $1, keywords = $2, files_read = $3,
-                    files_modified = $4, narrative = $5, created_at = $6, concepts = $7,
-                    noise_level = $8, subtitle = $9, noise_reason = $10,
-                    prompt_number = $11, discovery_tokens = $12, title = $14, observation_type = $15
-               WHERE id = $13",
-        )
-        .bind(serde_json::to_value(&merged.facts)?)
-        .bind(serde_json::to_value(&merged.keywords)?)
-        .bind(serde_json::to_value(&merged.files_read)?)
-        .bind(serde_json::to_value(&merged.files_modified)?)
-        .bind(&merged.narrative)
-        .bind(merged.created_at)
-        .bind(serde_json::to_value(&merged.concepts)?)
-        .bind(merged.noise_level.as_str())
-        .bind(&merged.subtitle)
-        .bind(&merged.noise_reason)
-        .bind(
-            merged
-                .prompt_number
-                .map(|v| v.as_pg_i32())
-                .transpose()
-                .map_err(|e| StorageError::DataCorruption {
-                    context: "prompt_number exceeds i32::MAX".into(),
-                    source: Box::<dyn std::error::Error + Send + Sync>::from(e.to_string()),
-                })?,
-        )
-        .bind(
-            merged
-                .discovery_tokens
-                .map(|v| v.as_pg_i32())
-                .transpose()
-                .map_err(|e| StorageError::DataCorruption {
-                    context: "discovery_tokens exceeds i32::MAX".into(),
-                    source: Box::<dyn std::error::Error + Send + Sync>::from(e.to_string()),
-                })?,
-        )
-        .bind(existing_id)
-        .bind(&merged.title)
-        .bind(merged.observation_type.as_str())
-        .execute(&mut *tx)
-        .await?;
+        self.update_observation_fields(&mut tx, existing_id, &merged)
+            .await?;
 
         tx.commit().await?;
         Ok(())
@@ -321,48 +334,8 @@ impl ObservationStore for PgStorage {
         let merged = opencode_mem_core::compute_merge(&keeper, &duplicate, false);
 
         // 3. Update keeper with merged data
-        sqlx::query(
-            "UPDATE observations SET facts = $1, keywords = $2, files_read = $3,
-                    files_modified = $4, narrative = $5, created_at = $6, concepts = $7,
-                    noise_level = $8, subtitle = $9, noise_reason = $10,
-                    prompt_number = $11, discovery_tokens = $12, title = $14, observation_type = $15
-               WHERE id = $13",
-        )
-        .bind(serde_json::to_value(&merged.facts)?)
-        .bind(serde_json::to_value(&merged.keywords)?)
-        .bind(serde_json::to_value(&merged.files_read)?)
-        .bind(serde_json::to_value(&merged.files_modified)?)
-        .bind(&merged.narrative)
-        .bind(merged.created_at)
-        .bind(serde_json::to_value(&merged.concepts)?)
-        .bind(merged.noise_level.as_str())
-        .bind(&merged.subtitle)
-        .bind(&merged.noise_reason)
-        .bind(
-            merged
-                .prompt_number
-                .map(|v| v.as_pg_i32())
-                .transpose()
-                .map_err(|e| StorageError::DataCorruption {
-                    context: "prompt_number exceeds i32::MAX".into(),
-                    source: Box::<dyn std::error::Error + Send + Sync>::from(e.to_string()),
-                })?,
-        )
-        .bind(
-            merged
-                .discovery_tokens
-                .map(|v| v.as_pg_i32())
-                .transpose()
-                .map_err(|e| StorageError::DataCorruption {
-                    context: "discovery_tokens exceeds i32::MAX".into(),
-                    source: Box::<dyn std::error::Error + Send + Sync>::from(e.to_string()),
-                })?,
-        )
-        .bind(keeper_id)
-        .bind(&merged.title)
-        .bind(merged.observation_type.as_str())
-        .execute(&mut *tx)
-        .await?;
+        self.update_observation_fields(&mut tx, keeper_id, &merged)
+            .await?;
 
         // 4. Repoint knowledge entries (replace duplicate_id with keeper_id in jsonb array)
         // Correct PostgreSQL logic: remove duplicate_id, add keeper_id, then DISTINCT to avoid duplicates.