locks: add process-wide lock hold time tracking

TrackedMutex and TrackedRwLock wrappers that record hold durations
by source location using #[track_caller]. Stats written to
~/.consciousness/lock-stats.json every second, sorted by max hold time.

Re-exported as crate::Mutex so all locks are instrumented. To disable,
swap the re-export back to tokio::sync::Mutex.

Co-Authored-By: Proof of Concept <poc@bcachefs.org>

src/agent/mod.rs

@@ -141,8 +141,8 @@ pub struct Agent {
     pub app_config: crate::config::AppConfig,
     pub prompt_file: String,
     pub session_id: String,
-    pub context: tokio::sync::Mutex<ContextState>,
-    pub state: tokio::sync::Mutex<AgentState>,
+    pub context: crate::Mutex<ContextState>,
+    pub state: crate::Mutex<AgentState>,
 }
 
 /// Mutable agent state — behind its own mutex.
@@ -218,8 +218,8 @@ impl Agent {
             app_config,
             prompt_file,
             session_id,
-            context: tokio::sync::Mutex::new(context),
-            state: tokio::sync::Mutex::new(AgentState {
+            context: crate::Mutex::new(context),
+            state: crate::Mutex::new(AgentState {
                 tools: agent_tools,
                 mcp_tools: McpToolAccess::All,
                 last_prompt_tokens: 0,
@@ -255,8 +255,8 @@ impl Agent {
             app_config: self.app_config.clone(),
             prompt_file: self.prompt_file.clone(),
             session_id: self.session_id.clone(),
-            context: tokio::sync::Mutex::new(ctx),
-            state: tokio::sync::Mutex::new(AgentState {
+            context: crate::Mutex::new(ctx),
+            state: crate::Mutex::new(AgentState {
                 tools,
                 mcp_tools: McpToolAccess::None,
                 last_prompt_tokens: 0,

src/agent/tools/lsp.rs

@@ -123,7 +123,7 @@ fn find_project_root(file_path: &str) -> Option<String> {
 const IDLE_TIMEOUT_SECS: u64 = 600;
 
 use std::sync::OnceLock;
-use tokio::sync::Mutex as TokioMutex;
+use crate::Mutex as TokioMutex;
 
 struct Registry {
     configs: Vec<crate::config::LspServerConfig>,

src/agent/tools/mcp_client.rs

@@ -10,7 +10,7 @@ use serde_json::json;
 use std::sync::OnceLock;
 use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, BufWriter};
 use tokio::process::{Child, ChildStdin, ChildStdout, Command};
-use tokio::sync::Mutex as TokioMutex;
+use crate::Mutex as TokioMutex;
 
 #[derive(Debug, Clone)]
 pub struct McpTool {

src/agent/tools/memory.rs

@@ -20,7 +20,7 @@ fn get_f64(args: &serde_json::Value, name: &str) -> Result<f64> {
     args.get(name).and_then(|v| v.as_f64()).context(format!("{} is required", name))
 }
 
-async fn cached_store() -> Result<std::sync::Arc<tokio::sync::Mutex<Store>>> {
+async fn cached_store() -> Result<std::sync::Arc<crate::Mutex<Store>>> {
     Store::cached().await.map_err(|e| anyhow::anyhow!("{}", e))
 }
 

src/hippocampus/store/persist.rs

@@ -21,16 +21,16 @@ use std::path::Path;
 use std::sync::Arc;
 
 /// Process-global cached store. Reloads only when log files change.
-static CACHED_STORE: tokio::sync::OnceCell<Arc<tokio::sync::Mutex<Store>>> =
+static CACHED_STORE: tokio::sync::OnceCell<Arc<crate::Mutex<Store>>> =
     tokio::sync::OnceCell::const_new();
 
 impl Store {
     /// Get or create the process-global cached store.
     /// Reloads from disk if log files have changed since last load.
-    pub async fn cached() -> Result<Arc<tokio::sync::Mutex<Store>>, String> {
+    pub async fn cached() -> Result<Arc<crate::Mutex<Store>>, String> {
         let store = CACHED_STORE.get_or_try_init(|| async {
             let s = Store::load()?;
-            Ok::<_, String>(Arc::new(tokio::sync::Mutex::new(s)))
+            Ok::<_, String>(Arc::new(crate::Mutex::new(s)))
         }).await?;
         {
             let mut guard = store.lock().await;

src/lib.rs

@@ -1,3 +1,5 @@
+#![feature(async_fn_track_caller)]
+
 // consciousness — unified crate for memory, agents, and subconscious processes
 //
 // thought/       — shared cognitive substrate (tools, context, memory ops)
@@ -47,6 +49,16 @@ pub mod session;
 // Shared utilities
 pub mod util;
 
+// Lock hold time tracking
+pub mod locks;
+
+// Re-export tracked locks as the default — swap to tokio::sync to disable tracking
+pub use locks::TrackedMutex as Mutex;
+pub use locks::TrackedMutexGuard as MutexGuard;
+pub use locks::TrackedRwLock as RwLock;
+pub use locks::TrackedRwLockReadGuard as RwLockReadGuard;
+pub use locks::TrackedRwLockWriteGuard as RwLockWriteGuard;
+
 // CLI handlers
 pub mod cli;
 

src/locks.rs

@@ -0,0 +1,235 @@
+// Lock hold time tracking
+//
+// Wrappers around tokio::sync primitives that track how long locks are held,
+// keyed by source location. Use `lock_stats()` to get a snapshot.
+
+use std::collections::HashMap;
+use std::panic::Location;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::OnceLock;
+use std::time::Instant;
+
+use tokio::sync::{Mutex, MutexGuard, RwLock, RwLockReadGuard, RwLockWriteGuard};
+
+// ── Stats Registry ─────────────────────────────────────────────
+
+struct LocationStats {
+    count: AtomicU64,
+    total_ns: AtomicU64,
+    max_ns: AtomicU64,
+}
+
+impl LocationStats {
+    fn new() -> Self {
+        Self {
+            count: AtomicU64::new(0),
+            total_ns: AtomicU64::new(0),
+            max_ns: AtomicU64::new(0),
+        }
+    }
+
+    fn record(&self, duration_ns: u64) {
+        self.count.fetch_add(1, Ordering::Relaxed);
+        self.total_ns.fetch_add(duration_ns, Ordering::Relaxed);
+        // Update max using CAS loop
+        let mut current = self.max_ns.load(Ordering::Relaxed);
+        while duration_ns > current {
+            match self.max_ns.compare_exchange_weak(
+                current, duration_ns, Ordering::Relaxed, Ordering::Relaxed
+            ) {
+                Ok(_) => break,
+                Err(c) => current = c,
+            }
+        }
+    }
+
+    fn snapshot(&self) -> LockStats {
+        let count = self.count.load(Ordering::Relaxed);
+        let total_ns = self.total_ns.load(Ordering::Relaxed);
+        let max_ns = self.max_ns.load(Ordering::Relaxed);
+        LockStats {
+            count,
+            total_ns,
+            max_ns,
+            avg_ns: if count > 0 { total_ns / count } else { 0 },
+        }
+    }
+}
+
+/// Stats for a single lock location.
+#[derive(Clone, Debug)]
+pub struct LockStats {
+    pub count: u64,
+    pub total_ns: u64,
+    pub max_ns: u64,
+    pub avg_ns: u64,
+}
+
+type StatsMap = std::sync::Mutex<HashMap<&'static Location<'static>, LocationStats>>;
+
+fn stats_map() -> &'static StatsMap {
+    static MAP: OnceLock<StatsMap> = OnceLock::new();
+    MAP.get_or_init(|| std::sync::Mutex::new(HashMap::new()))
+}
+
+fn record_hold_time(loc: &'static Location<'static>, duration_ns: u64) {
+    let map = stats_map().lock().unwrap();
+    if let Some(stats) = map.get(&loc) {
+        stats.record(duration_ns);
+        return;
+    }
+    drop(map);
+
+    // First time seeing this location — need write access
+    let mut map = stats_map().lock().unwrap();
+    let stats = map.entry(loc).or_insert_with(LocationStats::new);
+    stats.record(duration_ns);
+}
+
+/// Get a snapshot of all lock stats, sorted by max hold time (descending).
+pub fn lock_stats() -> Vec<(String, LockStats)> {
+    let map = stats_map().lock().unwrap();
+    let mut stats: Vec<_> = map.iter()
+        .map(|(loc, s)| (format!("{}:{}", loc.file(), loc.line()), s.snapshot()))
+        .collect();
+    stats.sort_by(|a, b| b.1.max_ns.cmp(&a.1.max_ns));
+    stats
+}
+
+/// Reset all lock stats.
+pub fn reset_lock_stats() {
+    let mut map = stats_map().lock().unwrap();
+    map.clear();
+}
+
+// ── TrackedMutex ───────────────────────────────────────────────
+
+/// A Mutex wrapper that tracks hold times by caller location.
+pub struct TrackedMutex<T> {
+    inner: Mutex<T>,
+}
+
+impl<T> TrackedMutex<T> {
+    pub fn new(value: T) -> Self {
+        Self { inner: Mutex::new(value) }
+    }
+
+    #[track_caller]
+    pub async fn lock(&self) -> TrackedMutexGuard<'_, T> {
+        let location = Location::caller();
+        let guard = self.inner.lock().await;
+        TrackedMutexGuard {
+            guard,
+            acquired_at: Instant::now(),
+            location,
+        }
+    }
+
+    #[track_caller]
+    pub fn try_lock(&self) -> Result<TrackedMutexGuard<'_, T>, tokio::sync::TryLockError> {
+        let location = Location::caller();
+        let guard = self.inner.try_lock()?;
+        Ok(TrackedMutexGuard {
+            guard,
+            acquired_at: Instant::now(),
+            location,
+        })
+    }
+}
+
+pub struct TrackedMutexGuard<'a, T> {
+    guard: MutexGuard<'a, T>,
+    acquired_at: Instant,
+    location: &'static Location<'static>,
+}
+
+impl<T> Drop for TrackedMutexGuard<'_, T> {
+    fn drop(&mut self) {
+        let duration = self.acquired_at.elapsed();
+        record_hold_time(self.location, duration.as_nanos() as u64);
+    }
+}
+
+impl<T> std::ops::Deref for TrackedMutexGuard<'_, T> {
+    type Target = T;
+    fn deref(&self) -> &T { &self.guard }
+}
+
+impl<T> std::ops::DerefMut for TrackedMutexGuard<'_, T> {
+    fn deref_mut(&mut self) -> &mut T { &mut self.guard }
+}
+
+// ── TrackedRwLock ──────────────────────────────────────────────
+
+/// An RwLock wrapper that tracks hold times by caller location.
+pub struct TrackedRwLock<T> {
+    inner: RwLock<T>,
+}
+
+impl<T> TrackedRwLock<T> {
+    pub fn new(value: T) -> Self {
+        Self { inner: RwLock::new(value) }
+    }
+
+    #[track_caller]
+    pub async fn read(&self) -> TrackedRwLockReadGuard<'_, T> {
+        let location = Location::caller();
+        let guard = self.inner.read().await;
+        TrackedRwLockReadGuard {
+            guard,
+            acquired_at: Instant::now(),
+            location,
+        }
+    }
+
+    #[track_caller]
+    pub async fn write(&self) -> TrackedRwLockWriteGuard<'_, T> {
+        let location = Location::caller();
+        let guard = self.inner.write().await;
+        TrackedRwLockWriteGuard {
+            guard,
+            acquired_at: Instant::now(),
+            location,
+        }
+    }
+}
+
+pub struct TrackedRwLockReadGuard<'a, T> {
+    guard: RwLockReadGuard<'a, T>,
+    acquired_at: Instant,
+    location: &'static Location<'static>,
+}
+
+impl<T> Drop for TrackedRwLockReadGuard<'_, T> {
+    fn drop(&mut self) {
+        let duration = self.acquired_at.elapsed();
+        record_hold_time(self.location, duration.as_nanos() as u64);
+    }
+}
+
+impl<T> std::ops::Deref for TrackedRwLockReadGuard<'_, T> {
+    type Target = T;
+    fn deref(&self) -> &T { &self.guard }
+}
+
+pub struct TrackedRwLockWriteGuard<'a, T> {
+    guard: RwLockWriteGuard<'a, T>,
+    acquired_at: Instant,
+    location: &'static Location<'static>,
+}
+
+impl<T> Drop for TrackedRwLockWriteGuard<'_, T> {
+    fn drop(&mut self) {
+        let duration = self.acquired_at.elapsed();
+        record_hold_time(self.location, duration.as_nanos() as u64);
+    }
+}
+
+impl<T> std::ops::Deref for TrackedRwLockWriteGuard<'_, T> {
+    type Target = T;
+    fn deref(&self) -> &T { &self.guard }
+}
+
+impl<T> std::ops::DerefMut for TrackedRwLockWriteGuard<'_, T> {
+    fn deref_mut(&mut self) -> &mut T { &mut self.guard }
+}

src/mind/mod.rs

@@ -268,8 +268,8 @@ pub struct Mind {
     pub agent: Arc<Agent>,
     pub shared: Arc<SharedMindState>,
     pub config: SessionConfig,
-    pub subconscious: Arc<tokio::sync::Mutex<Subconscious>>,
-    pub unconscious: Arc<tokio::sync::Mutex<Unconscious>>,
+    pub subconscious: Arc<crate::Mutex<Subconscious>>,
+    pub unconscious: Arc<crate::Mutex<Unconscious>>,
     turn_tx: mpsc::Sender<(Result<TurnResult>, StreamTarget)>,
     turn_watch: tokio::sync::watch::Sender<bool>,
     /// Signals conscious activity to the unconscious loop.
@@ -309,10 +309,10 @@ impl Mind {
         sup.load_config();
         sup.ensure_running();
 
-        let subconscious = Arc::new(tokio::sync::Mutex::new(Subconscious::new()));
+        let subconscious = Arc::new(crate::Mutex::new(Subconscious::new()));
         subconscious.lock().await.init_output_tool(subconscious.clone());
 
-        let unconscious = Arc::new(tokio::sync::Mutex::new(Unconscious::new()));
+        let unconscious = Arc::new(crate::Mutex::new(Unconscious::new()));
 
         // Spawn the unconscious loop on its own task
         if !config.no_agents {
@@ -584,6 +584,28 @@ impl Mind {
         mut input_rx: tokio::sync::mpsc::UnboundedReceiver<MindCommand>,
         mut turn_rx: mpsc::Receiver<(Result<TurnResult>, StreamTarget)>,
     ) {
+        // Spawn lock stats logger
+        tokio::spawn(async {
+            let path = dirs::home_dir().unwrap_or_default()
+                .join(".consciousness/lock-stats.json");
+            let mut interval = tokio::time::interval(std::time::Duration::from_secs(1));
+            loop {
+                interval.tick().await;
+                let stats = crate::locks::lock_stats();
+                if stats.is_empty() { continue; }
+                let json: Vec<serde_json::Value> = stats.iter()
+                    .map(|(loc, s)| serde_json::json!({
+                        "location": loc,
+                        "count": s.count,
+                        "total_ms": s.total_ns as f64 / 1_000_000.0,
+                        "avg_ms": s.avg_ns as f64 / 1_000_000.0,
+                        "max_ms": s.max_ns as f64 / 1_000_000.0,
+                    }))
+                    .collect();
+                let _ = std::fs::write(&path, serde_json::to_string_pretty(&json).unwrap_or_default());
+            }
+        });
+
         let mut bg_rx = self.bg_rx.lock().unwrap().take()
             .expect("Mind::run() called twice");
         let mut sub_handle: Option<tokio::task::JoinHandle<()>> = None;

src/mind/subconscious.rs

@@ -410,7 +410,7 @@ impl Subconscious {
     /// Late-init: push the output tool onto each agent's tool list.
     /// Called after Subconscious is wrapped in Arc<Mutex<>> so the
     /// closure can capture a reference back.
-    pub fn init_output_tool(&mut self, self_arc: std::sync::Arc<tokio::sync::Mutex<Self>>) {
+    pub fn init_output_tool(&mut self, self_arc: std::sync::Arc<crate::Mutex<Self>>) {
         for agent in &mut self.agents {
             let Some(ref mut auto) = agent.auto else { continue };
             let sub = self_arc.clone();