fix: address 5 code review issues from vulnerability audit

1. Admin auth: allow loopback access when OPENCODE_MEM_ADMIN_TOKEN is
   unset (dev convenience), require token for all IPs when set (prod)
2. CLI serve: update curl shutdown instruction with Content-Type header
   and JSON body required after CSRF fix
3. MCP SPOT violation: extract shared MaintenanceServices scheduler to
   service crate, used by both HTTP cron and MCP background loop —
   MCP now runs all 7 maintenance tasks instead of only compression
4. Pipeline dead code: fetch all unsummarized events per run (10K limit)
   instead of 100-event batches that split buckets into overlapping
   summaries; compress_events_chunked is now reachable for >200 events
5. Circuit breaker bypass: route run_full_compression and
   run_compression_pipeline through CB so DB failures are tracked

Author: Stranmor

crates/cli/src/commands/mcp.rs

@@ -3,34 +3,24 @@ use opencode_mem_core::AppConfig;
 use opencode_mem_embeddings::LazyEmbeddingService;
 use opencode_mem_llm::LlmClient;
 use opencode_mem_mcp::run_mcp_server;
+use opencode_mem_service::maintenance::{MaintenanceServices, run_maintenance_tick};
 use opencode_mem_service::{
-    InfiniteMemoryService, KnowledgeService, ObservationService, SearchService, SessionService,
+    InfiniteMemoryService, KnowledgeService, ObservationService, QueueService, SearchService,
+    SessionService,
 };
 use std::sync::Arc;
 use tokio::sync::broadcast;
 
-fn start_mcp_background_tasks(infinite_mem: Option<Arc<InfiniteMemoryService>>) {
-    if let Some(mem) = infinite_mem {
-        tokio::spawn(async move {
-            let mut interval = tokio::time::interval(std::time::Duration::from_secs(300));
-            loop {
-                interval.tick().await;
-                match mem.run_full_compression().await {
-                    Ok((five_min, hour, day)) => {
-                        if five_min > 0 || hour > 0 || day > 0 {
-                            tracing::info!(
-                                "MCP cron: created {} 5min, {} hour, {} day summaries",
-                                five_min,
-                                hour,
-                                day,
-                            );
-                        }
-                    }
-                    Err(e) => tracing::warn!("MCP cron: infinite memory error: {e:?}"),
-                }
-            }
-        });
-    }
+fn start_mcp_background_tasks(services: Arc<MaintenanceServices>) {
+    tokio::spawn(async move {
+        let mut interval = tokio::time::interval(std::time::Duration::from_secs(5));
+        let mut loop_count: u64 = 0;
+        loop {
+            interval.tick().await;
+            loop_count = loop_count.wrapping_add(1);
+            run_maintenance_tick(&services, loop_count).await;
+        }
+    });
 }
 
 pub(crate) async fn run(config: Arc<AppConfig>) -> Result<()> {
@@ -99,7 +89,7 @@ pub(crate) async fn run(config: Arc<AppConfig>) -> Result<()> {
     let session_service = Arc::new(SessionService::new(storage.clone(), llm.clone()));
     let knowledge_service = Arc::new(KnowledgeService::new(storage.clone(), embeddings.clone()));
     let search_service = Arc::new(SearchService::new(
-        storage,
+        storage.clone(),
         embeddings,
         infinite_mem.clone(),
         config.dedup_threshold,
@@ -109,7 +99,19 @@ pub(crate) async fn run(config: Arc<AppConfig>) -> Result<()> {
 
     let pending_writes = Arc::new(opencode_mem_service::PendingWriteQueue::new());
 
-    start_mcp_background_tasks(infinite_mem.clone());
+    let queue_service = Arc::new(QueueService::new(storage, pending_writes.clone(), &config));
+
+    let maintenance = Arc::new(MaintenanceServices {
+        observation_service: observation_service.clone(),
+        session_service: session_service.clone(),
+        knowledge_service: knowledge_service.clone(),
+        search_service: search_service.clone(),
+        queue_service,
+        infinite_mem: infinite_mem.clone(),
+        config: config.clone(),
+    });
+
+    start_mcp_background_tasks(maintenance);
 
     run_mcp_server(
         infinite_mem,

crates/cli/src/commands/serve.rs

@@ -140,7 +140,9 @@ pub(crate) async fn run(port: u16, host: String, config: Arc<AppConfig>) -> Resu
                 "Port {} is already in use.\n\n\
 Another instance of opencode-mem is likely running.\n\
 To stop it, run:\n\
-  curl -X POST http://127.0.0.1:{}/api/admin/shutdown\n\
+  curl -X POST -H 'Content-Type: application/json' -d 'null' \
+http://127.0.0.1:{}/api/admin/shutdown\n\
+If OPENCODE_MEM_ADMIN_TOKEN is set, add: -H 'X-Admin-Token: <token>'\n\
 Or kill the process manually before starting a new server.",
                 port,
                 port

crates/http/src/handlers/cron.rs

@@ -2,8 +2,19 @@ use std::sync::Arc;
 use std::sync::atomic::Ordering;
 
 use crate::AppState;
+use opencode_mem_service::maintenance::{MaintenanceServices, run_maintenance_tick};
 
 pub async fn start_cron_scheduler(state: Arc<AppState>) {
+    let services = MaintenanceServices {
+        observation_service: Arc::clone(&state.observation_service),
+        session_service: Arc::clone(&state.session_service),
+        knowledge_service: Arc::clone(&state.knowledge_service),
+        search_service: Arc::clone(&state.search_service),
+        queue_service: Arc::clone(&state.queue_service),
+        infinite_mem: state.infinite_mem.clone(),
+        config: Arc::clone(&state.config),
+    };
+
     let mut interval = tokio::time::interval(std::time::Duration::from_secs(5));
     let mut shutdown_rx = state.shutdown_tx.subscribe();
     let mut loop_count: u64 = 0;
@@ -22,126 +33,6 @@ pub async fn start_cron_scheduler(state: Arc<AppState>) {
 
         loop_count = loop_count.wrapping_add(1);
 
-        if loop_count.is_multiple_of(60)
-            && let Some(ref infinite_mem) = state.infinite_mem
-        {
-            tracing::debug!("Cron: running infinite memory compression...");
-            let mem = Arc::clone(infinite_mem);
-            state.background_tasks.lock().await.spawn(async move {
-                match mem.run_full_compression().await {
-                    Ok((five_min, hour, day)) => {
-                        if five_min > 0 || hour > 0 || day > 0 {
-                            tracing::info!(
-                                "Cron: created {} 5min, {} hour, {} day summaries",
-                                five_min,
-                                hour,
-                                day,
-                            );
-                        }
-                    }
-                    Err(e) => tracing::warn!("Cron: infinite memory error: {e:?}"),
-                }
-            });
-        }
-
-        if loop_count.is_multiple_of(180) {
-            tracing::debug!("Cron: running embedding backfill...");
-            let state_clone = Arc::clone(&state);
-            state.background_tasks.lock().await.spawn(async move {
-                match state_clone.search_service.run_embedding_backfill(100).await {
-                    Ok(generated) if generated > 0 => {
-                        tracing::info!("Cron: generated {} embeddings", generated);
-                    }
-                    Ok(_) => {}
-                    Err(e) => tracing::warn!("Cron: embedding backfill failed: {}", e),
-                }
-            });
-        }
-
-        if loop_count.is_multiple_of(360) {
-            let state_clone = Arc::clone(&state);
-            state.background_tasks.lock().await.spawn(async move {
-                match state_clone.observation_service.run_dedup_sweep().await {
-                    Ok(merged) if merged > 0 => {
-                        tracing::info!(merged, "Cron: dedup sweep completed");
-                    }
-                    Ok(_) => {}
-                    Err(e) => tracing::warn!(error = %e, "Cron: dedup sweep failed"),
-                }
-            });
-        }
-
-        if loop_count.is_multiple_of(720) {
-            let state_clone = Arc::clone(&state);
-            state.background_tasks.lock().await.spawn(async move {
-                if let Err(e) = state_clone
-                    .observation_service
-                    .cleanup_old_injections()
-                    .await
-                {
-                    tracing::warn!(error = %e, "Cron: injection cleanup failed");
-                }
-            });
-        }
-
-        if loop_count.is_multiple_of(17280) {
-            let ttl_secs = state.config.dlq_ttl_secs();
-            let state_clone = Arc::clone(&state);
-            state.background_tasks.lock().await.spawn(async move {
-                match state_clone
-                    .queue_service
-                    .clear_stale_failed_messages(ttl_secs)
-                    .await
-                {
-                    Ok(deleted) if deleted > 0 => {
-                        tracing::info!(deleted, "Cron: DLQ garbage collection completed");
-                    }
-                    Ok(_) => {}
-                    Err(e) => tracing::warn!(error = %e, "Cron: DLQ garbage collection failed"),
-                }
-            });
-        }
-
-        if loop_count.is_multiple_of(2160) {
-            let state_clone = Arc::clone(&state);
-            state.background_tasks.lock().await.spawn(async move {
-                match state_clone
-                    .knowledge_service
-                    .run_confidence_lifecycle()
-                    .await
-                {
-                    Ok((decayed, archived)) if decayed > 0 || archived > 0 => {
-                        tracing::info!(
-                            decayed,
-                            archived,
-                            "Cron: knowledge confidence lifecycle completed"
-                        );
-                    }
-                    Ok(_) => {}
-                    Err(e) => {
-                        tracing::warn!(error = %e, "Cron: knowledge confidence lifecycle failed");
-                    }
-                }
-            });
-        }
-
-        if loop_count.is_multiple_of(360) {
-            let state_clone = Arc::clone(&state);
-            state.background_tasks.lock().await.spawn(async move {
-                match state_clone
-                    .session_service
-                    .generate_pending_summaries(10)
-                    .await
-                {
-                    Ok(n) if n > 0 => {
-                        tracing::info!(generated = n, "Cron: session summary generation completed");
-                    }
-                    Ok(_) => {}
-                    Err(e) => {
-                        tracing::warn!(error = %e, "Cron: session summary generation failed");
-                    }
-                }
-            });
-        }
+        run_maintenance_tick(&services, loop_count).await;
     }
 }

crates/http/src/handlers/mod.rs

@@ -12,19 +12,29 @@
     reason = "HTTP handlers are called once from router"
 )]
 
+/// Admin access control with two modes:
+/// - **Token configured** (`OPENCODE_MEM_ADMIN_TOKEN` set): require the token
+///   for ALL requests regardless of source IP. Secure for production.
+/// - **No token configured**: allow admin access from loopback IPs only
+///   (127.0.0.1, ::1). Convenient for local development where the CLI
+///   `serve` command tells users to `curl -X POST .../api/admin/shutdown`.
 pub(crate) fn check_admin_access(
-    _addr: &std::net::SocketAddr,
+    addr: &std::net::SocketAddr,
     headers: &axum::http::HeaderMap,
     config: &opencode_mem_core::AppConfig,
 ) -> bool {
-    if let Some(ref token) = config.admin_token
-        && let Some(provided) = headers.get("x-admin-token").and_then(|h| h.to_str().ok())
-        && subtle::ConstantTimeEq::ct_eq(provided.as_bytes(), token.as_bytes()).into()
-    {
-        return true;
+    if let Some(ref token) = config.admin_token {
+        // Token mode: require valid token regardless of IP
+        if let Some(provided) = headers.get("x-admin-token").and_then(|h| h.to_str().ok())
+            && subtle::ConstantTimeEq::ct_eq(provided.as_bytes(), token.as_bytes()).into()
+        {
+            return true;
+        }
+        false
+    } else {
+        // No token mode: allow loopback only
+        addr.ip().is_loopback()
     }
-
-    false
 }
 
 pub mod admin;

crates/service/src/infinite_memory_service/mod.rs

@@ -205,7 +205,22 @@ impl InfiniteMemoryService {
     }
 
     pub async fn run_compression_pipeline(&self) -> Result<u32> {
-        pipeline::run_compression_pipeline(&self.pool, &self.llm).await
+        if !self.circuit_breaker.should_allow() {
+            return Err(anyhow::anyhow!(
+                "circuit breaker open, {} seconds until probe",
+                self.circuit_breaker.seconds_until_probe()
+            ));
+        }
+        let result = pipeline::run_compression_pipeline(&self.pool, &self.llm).await;
+        match &result {
+            Ok(_) => {
+                self.circuit_breaker.record_success();
+            }
+            Err(_) => {
+                self.circuit_breaker.record_failure();
+            }
+        }
+        result
     }
 
     pub async fn create_hour_summary(
@@ -253,7 +268,22 @@ impl InfiniteMemoryService {
     }
 
     pub async fn run_full_compression(&self) -> Result<(u32, u32, u32)> {
-        pipeline::run_full_compression(&self.pool, &self.llm).await
+        if !self.circuit_breaker.should_allow() {
+            return Err(anyhow::anyhow!(
+                "circuit breaker open, {} seconds until probe",
+                self.circuit_breaker.seconds_until_probe()
+            ));
+        }
+        let result = pipeline::run_full_compression(&self.pool, &self.llm).await;
+        match &result {
+            Ok(_) => {
+                self.circuit_breaker.record_success();
+            }
+            Err(_) => {
+                self.circuit_breaker.record_failure();
+            }
+        }
+        result
     }
 
     pub async fn guarded<F, Fut, T>(&self, op_f: F) -> Result<T, StorageError>