runtime: Implement lock-wrapped mutation trace coordination

Expose coordinate() as the runtime boundary for driving mutation-cursor processing against a real worktree while preventing concurrent same-worktree observations from racing.

Resolve the Git directory, acquire a bounded WorktreeLock, derive WorktreeId from checkout identity, and delegate to the existing snapshot/protocol/CAS pipeline; add serialization coverage and update the runtime documentation and plan.

Plan: mutation-cursor-runtime-coordinator (T05)

Co-authored-by: SCE <sce@crocoder.dev>

Author: davidabram

cli/src/services/mutation_trace/runtime/coordinator.rs

@@ -1,17 +1,24 @@
+use std::path::Path;
+use std::time::Duration;
+
 use anyhow::Result;
 use uuid::Uuid;
 
 use crate::services::agent_trace_db::repository::RepositoryAgentTraceDb;
+use crate::services::checkout::{get_or_create_checkout_id, resolve_git_dir};
 use crate::services::mutation_trace::protocol;
 use crate::services::mutation_trace::store::{CasResult, DurableTransition, MutationTraceStore};
 use crate::services::mutation_trace::types::{
     self, ActorKind, AttemptId, Boundary, EventId, MutationEvent, ScopeId, TreeId, WorktreeId,
 };
 
 use super::git_snapshot::GitSnapshotService;
+use super::worktree_lock::{WorktreeLock, WorktreeLockError};
 
 const MAX_CAS_RETRY_ATTEMPTS: u32 = 5;
 
+const WORKTREE_LOCK_TIMEOUT: Duration = Duration::from_secs(10);
+
 #[derive(Clone, Debug)]
 pub enum RuntimeBoundary {
     Start {
@@ -104,6 +111,27 @@ impl SnapshotCapture for GitSnapshotService {
     }
 }
 
+pub fn coordinate(
+    repository_root: &Path,
+    db: &RepositoryAgentTraceDb,
+    boundary: &RuntimeBoundary,
+) -> Result<CoordinateOutcome, CoordinateError> {
+    let git_dir = resolve_git_dir(repository_root).map_err(CoordinateError::Other)?;
+
+    let _lock = WorktreeLock::acquire(&git_dir, WORKTREE_LOCK_TIMEOUT).map_err(lock_acquisition)?;
+
+    let checkout_id = get_or_create_checkout_id(&git_dir).map_err(CoordinateError::Other)?;
+    let worktree_id = WorktreeId(checkout_id);
+
+    let snapshot = GitSnapshotService::new(repository_root).map_err(CoordinateError::Other)?;
+
+    coordinate_boundary(db, &snapshot, &worktree_id, boundary)
+}
+
+fn lock_acquisition(error: WorktreeLockError) -> CoordinateError {
+    CoordinateError::LockAcquisition(anyhow::Error::new(error))
+}
+
 fn coordinate_boundary<C: SnapshotCapture>(
     db: &RepositoryAgentTraceDb,
     capture: &C,
@@ -315,7 +343,10 @@ where
 mod tests {
     use std::cell::{Cell, RefCell};
     use std::collections::VecDeque;
+    use std::process::Command;
     use std::sync::atomic::{AtomicU64, Ordering};
+    use std::sync::mpsc;
+    use std::thread;
 
     use super::*;
     use crate::services::mutation_trace::store::encode_revision;
@@ -345,6 +376,38 @@ mod tests {
         }
     }
 
+    fn unique_test_repo(label: &str) -> std::path::PathBuf {
+        let id = NEXT_TEST_DB_ID.fetch_add(1, Ordering::Relaxed);
+        std::env::temp_dir().join(format!(
+            "sce-mutation-trace-coordinator-repo-{label}-{}-{id}",
+            std::process::id()
+        ))
+    }
+
+    fn init_repo(repo_root: &std::path::Path) {
+        std::fs::create_dir_all(repo_root).expect("repo root should be created");
+        run_git(repo_root, &["init", "--quiet"]);
+        run_git(repo_root, &["config", "user.email", "test@example.com"]);
+        run_git(repo_root, &["config", "user.name", "Test"]);
+    }
+
+    fn run_git(repo_root: &std::path::Path, args: &[&str]) {
+        let output = Command::new("git")
+            .args(args)
+            .current_dir(repo_root)
+            .output()
+            .expect("git command should spawn");
+        assert!(
+            output.status.success(),
+            "git {args:?} failed: {}",
+            String::from_utf8_lossy(&output.stderr)
+        );
+    }
+
+    fn remove_test_repo(repo_root: &std::path::Path) {
+        let _ = std::fs::remove_dir_all(repo_root);
+    }
+
     enum FakeOutcome {
         Succeed(TreeId),
         Fail(String),
@@ -1228,4 +1291,57 @@ mod tests {
 
         remove_test_db(&db_path);
     }
+
+    #[test]
+    fn two_threads_on_the_same_worktree_serialize() {
+        let repo_root = unique_test_repo("t05-serialize");
+        init_repo(&repo_root);
+        let db_path = repo_root.join("agent-trace.db");
+
+        let git_dir = resolve_git_dir(&repo_root).expect("git dir should resolve");
+        let held = WorktreeLock::acquire(&git_dir, Duration::from_secs(5))
+            .expect("the test should acquire the worktree lock before the worker runs");
+
+        let (started_tx, started_rx) = mpsc::channel();
+        let (done_tx, done_rx) = mpsc::channel();
+        let repo_root_clone = repo_root.clone();
+        let db_path_clone = db_path.clone();
+        let worker = thread::spawn(move || {
+            let db = RepositoryAgentTraceDb::new_at(&db_path_clone).expect("worker db should open");
+            started_tx
+                .send(())
+                .expect("started signal channel should still be open");
+            let outcome = coordinate(&repo_root_clone, &db, &RuntimeBoundary::Flush);
+            done_tx
+                .send(())
+                .expect("done signal channel should still be open");
+            outcome
+        });
+
+        started_rx
+            .recv_timeout(Duration::from_secs(5))
+            .expect("worker thread should reach the coordinate() call");
+
+        assert!(
+            done_rx.recv_timeout(Duration::from_millis(500)).is_err(),
+            "coordinate() must not enter its critical section while another holder owns the worktree lock"
+        );
+
+        drop(held);
+
+        done_rx
+            .recv_timeout(Duration::from_secs(5))
+            .expect("coordinate() should proceed once the worktree lock is released");
+
+        let outcome = worker
+            .join()
+            .expect("worker thread should not panic")
+            .expect("coordinate() should succeed once it can acquire the worktree lock");
+        assert_eq!(
+            outcome.revision, 0,
+            "the worker's first observation flush should not advance the revision"
+        );
+
+        remove_test_repo(&repo_root);
+    }
 }