feat(file-lock): add read/write/directory_add lock modes with compatibility matrix [fn-110.7]

- Migration v3: composite PK (file_path, task_id) enables shared locks
- LockMode enum with compatibility: read+read, read+dir_add, dir_add+dir_add OK; write exclusive
- CLI: `flowctl lock --mode read|write|directory_add` (default: write)
- check_locks() returns Vec<LockEntry> for multi-holder visibility
- Task spec format: `**Files (write):**` and `**Files (read):**`
- Merge conflict recovery: abort → classify → rebase-first → retry or restart
- git rerere.enabled true in orchestrator setup
- Retry storm protection: >50% retry triggers fresh wave
- 13 lock-specific tests passing

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: z23cc

flowctl/crates/flowctl-cli/src/commands/db_shim.rs

@@ -14,7 +14,7 @@
 
 use std::path::{Path, PathBuf};
 
-pub use flowctl_db::{DbError, GapRow, ReindexResult};
+pub use flowctl_db::{DbError, GapRow, LockEntry, LockMode, ReindexResult};
 pub use flowctl_db::metrics::{
     Bottleneck, DoraMetrics, EpicStats, Summary, TokenBreakdown, WeeklyTrend,
 };
@@ -304,9 +304,9 @@ impl FileLockRepo {
         Self(conn.inner())
     }
 
-    pub fn acquire(&self, file_path: &str, task_id: &str) -> Result<(), DbError> {
+    pub fn acquire(&self, file_path: &str, task_id: &str, mode: &LockMode) -> Result<(), DbError> {
         block_on(
-            flowctl_db::FileLockRepo::new(self.0.clone()).acquire(file_path, task_id),
+            flowctl_db::FileLockRepo::new(self.0.clone()).acquire(file_path, task_id, mode),
         )
     }
 
@@ -322,17 +322,21 @@ impl FileLockRepo {
         block_on(flowctl_db::FileLockRepo::new(self.0.clone()).check(file_path))
     }
 
+    pub fn check_locks(&self, file_path: &str) -> Result<Vec<LockEntry>, DbError> {
+        block_on(flowctl_db::FileLockRepo::new(self.0.clone()).check_locks(file_path))
+    }
+
     pub fn heartbeat(&self, task_id: &str) -> Result<u64, DbError> {
         block_on(flowctl_db::FileLockRepo::new(self.0.clone()).heartbeat(task_id))
     }
 
-    /// List all active locks: (file_path, task_id, locked_at).
-    pub fn list_all(&self) -> Result<Vec<(String, String, String)>, DbError> {
+    /// List all active locks: (file_path, task_id, locked_at, lock_mode).
+    pub fn list_all(&self) -> Result<Vec<(String, String, String, String)>, DbError> {
         let inner = self.0.clone();
         block_on(async move {
             let mut rows = inner
                 .query(
-                    "SELECT file_path, task_id, locked_at FROM file_locks ORDER BY file_path",
+                    "SELECT file_path, task_id, locked_at, lock_mode FROM file_locks ORDER BY file_path",
                     (),
                 )
                 .await?;
@@ -342,6 +346,7 @@ impl FileLockRepo {
                     row.get::<String>(0)?,
                     row.get::<String>(1)?,
                     row.get::<String>(2)?,
+                    row.get::<String>(3)?,
                 ));
             }
             Ok(out)

flowctl/crates/flowctl-cli/src/commands/query.rs

@@ -598,32 +598,31 @@ pub fn cmd_files(json_mode: bool, epic: String) {
 // ── Lock commands (Teams mode) ─────────────────────────────────────
 
 
-pub fn cmd_lock(json: bool, task: String, files: String) {
+pub fn cmd_lock(json: bool, task: String, files: String, mode: String) {
     let _flow_dir = ensure_flow_exists();
 
     let file_list: Vec<&str> = files.split(',').map(str::trim).filter(|s| !s.is_empty()).collect();
     if file_list.is_empty() {
         error_exit("No files specified for locking.");
     }
 
+    let lock_mode = crate::commands::db_shim::LockMode::from_str(&mode)
+        .unwrap_or_else(|e| error_exit(&format!("Invalid lock mode: {}", e)));
+
     let conn = require_db();
     let repo = crate::commands::db_shim::FileLockRepo::new(&conn);
 
     let mut locked = Vec::new();
     let mut already_locked = Vec::new();
 
     for file in &file_list {
-        match repo.acquire(file, &task) {
+        match repo.acquire(file, &task, &lock_mode) {
             Ok(()) => locked.push(file.to_string()),
-            Err(crate::commands::db_shim::DbError::Constraint(_)) => {
+            Err(crate::commands::db_shim::DbError::Constraint(msg)) => {
                 // Already locked — find out by whom
-                let owner = repo.check(file).ok().flatten().unwrap_or_else(|| "unknown".to_string());
-                if owner == task {
-                    // Re-locking own file is fine, treat as locked
-                    locked.push(file.to_string());
-                } else {
-                    already_locked.push(json!({"file": file, "owner": owner}));
-                }
+                let entries = repo.check_locks(file).ok().unwrap_or_default();
+                let owners: Vec<String> = entries.iter().map(|e| format!("{}({})", e.task_id, e.lock_mode.as_str())).collect();
+                already_locked.push(json!({"file": file, "owners": owners, "detail": msg}));
             }
             Err(e) => {
                 error_exit(&format!("Failed to lock {}: {}", file, e));
@@ -636,16 +635,17 @@ pub fn cmd_lock(json: bool, task: String, files: String) {
             "locked": locked,
             "already_locked": already_locked,
             "task": task,
+            "mode": mode,
         }));
     } else {
         if !locked.is_empty() {
-            println!("Locked {} file(s) for task {}", locked.len(), task);
+            println!("Locked {} file(s) for task {} (mode: {})", locked.len(), task, mode);
         }
         for al in &already_locked {
             println!(
-                "Already locked: {} (owner: {})",
+                "Already locked: {} (owners: {})",
                 al["file"].as_str().unwrap_or(""),
-                al["owner"].as_str().unwrap_or("")
+                al["owners"],
             );
         }
     }
@@ -703,19 +703,24 @@ pub fn cmd_lock_check(json: bool, file: Option<String>) {
 
     match file {
         Some(f) => {
-            match repo.check(&f) {
-                Ok(Some(owner)) => {
+            match repo.check_locks(&f) {
+                Ok(entries) if !entries.is_empty() => {
+                    let lock_info: Vec<serde_json::Value> = entries.iter().map(|e| json!({
+                        "task_id": e.task_id,
+                        "mode": e.lock_mode.as_str(),
+                    })).collect();
                     if json {
                         json_output(json!({
                             "file": f,
                             "locked": true,
-                            "owner": owner,
+                            "locks": lock_info,
                         }));
                     } else {
-                        println!("{}: locked by {}", f, owner);
+                        let owners: Vec<String> = entries.iter().map(|e| format!("{}({})", e.task_id, e.lock_mode.as_str())).collect();
+                        println!("{}: locked by {}", f, owners.join(", "));
                     }
                 }
-                Ok(None) => {
+                Ok(_) => {
                     if json {
                         json_output(json!({
                             "file": f,
@@ -736,10 +741,11 @@ pub fn cmd_lock_check(json: bool, file: Option<String>) {
                 .unwrap_or_else(|e| { error_exit(&format!("Query failed: {}", e)); });
             let locks: Vec<serde_json::Value> = rows
                 .into_iter()
-                .map(|(file, task_id, locked_at)| json!({
+                .map(|(file, task_id, locked_at, lock_mode)| json!({
                     "file": file,
                     "task_id": task_id,
                     "locked_at": locked_at,
+                    "mode": lock_mode,
                 }))
                 .collect();
 
@@ -754,9 +760,10 @@ pub fn cmd_lock_check(json: bool, file: Option<String>) {
                 println!("Active file locks ({}):\n", locks.len());
                 for l in &locks {
                     println!(
-                        "  {} → {} (since {})",
+                        "  {} → {} [{}] (since {})",
                         l["file"].as_str().unwrap_or(""),
                         l["task_id"].as_str().unwrap_or(""),
+                        l["mode"].as_str().unwrap_or("write"),
                         l["locked_at"].as_str().unwrap_or("")
                     );
                 }

flowctl/crates/flowctl-cli/src/main.rs

@@ -302,6 +302,9 @@ enum Commands {
         /// Comma-separated file paths to lock.
         #[arg(long)]
         files: String,
+        /// Lock mode: read, write, or directory_add (default: write).
+        #[arg(long, default_value = "write", value_parser = ["read", "write", "directory_add"])]
+        mode: String,
     },
     /// Unlock files for a task (Teams mode).
     Unlock {
@@ -530,7 +533,7 @@ fn main() {
         Commands::List => query::cmd_list(json),
         Commands::Cat { id } => query::cmd_cat(id),
         Commands::Files { epic } => query::cmd_files(json, epic),
-        Commands::Lock { task, files } => query::cmd_lock(json, task, files),
+        Commands::Lock { task, files, mode } => query::cmd_lock(json, task, files, mode),
         Commands::Unlock { task, files, all } => query::cmd_unlock(json, task, files, all),
         Commands::LockCheck { file } => query::cmd_lock_check(json, file),
         Commands::Heartbeat { task } => query::cmd_heartbeat(json, task),

flowctl/crates/flowctl-db/src/lib.rs

@@ -36,7 +36,7 @@ pub use skill::{SkillEntry, SkillMatch, SkillRepo};
 pub use pool::{cleanup, open_async, open_memory_async, resolve_db_path, resolve_libsql_path, resolve_state_dir};
 pub use repo::{
     DepRepo, EpicRepo, EventRepo, EventRow, EvidenceRepo, FileLockRepo, FileOwnershipRepo,
-    GapRepo, GapRow, PhaseProgressRepo, RuntimeRepo, TaskRepo,
+    GapRepo, GapRow, LockEntry, LockMode, PhaseProgressRepo, RuntimeRepo, TaskRepo,
     max_epic_num, max_task_num,
 };
 

flowctl/crates/flowctl-db/src/migration.rs

@@ -8,7 +8,7 @@ use libsql::Connection;
 use crate::error::DbError;
 
 /// Current target schema version. Bump this when adding new migrations.
-const TARGET_VERSION: i64 = 2;
+const TARGET_VERSION: i64 = 3;
 
 /// Ensure `_meta` table exists and run any pending migrations.
 pub async fn migrate(conn: &Connection) -> Result<(), DbError> {
@@ -26,6 +26,10 @@ pub async fn migrate(conn: &Connection) -> Result<(), DbError> {
         migrate_v2(conn).await?;
     }
 
+    if current < 3 {
+        migrate_v3(conn).await?;
+    }
+
     // Update stored version to target.
     if current < TARGET_VERSION {
         set_version(conn, TARGET_VERSION).await?;
@@ -100,6 +104,50 @@ async fn migrate_v2(conn: &Connection) -> Result<(), DbError> {
     Ok(())
 }
 
+/// Migration v3: Change file_locks PK to composite (file_path, task_id)
+/// and add `lock_mode TEXT DEFAULT 'write'`.
+///
+/// SQLite can't ALTER PRIMARY KEY, so we recreate the table.
+async fn migrate_v3(conn: &Connection) -> Result<(), DbError> {
+    conn.execute(
+        "CREATE TABLE IF NOT EXISTS file_locks_new (
+            file_path  TEXT NOT NULL,
+            task_id    TEXT NOT NULL,
+            locked_at  TEXT NOT NULL,
+            holder_pid INTEGER,
+            expires_at TEXT,
+            lock_mode  TEXT NOT NULL DEFAULT 'write',
+            PRIMARY KEY (file_path, task_id)
+        )",
+        (),
+    )
+    .await
+    .map_err(|e| DbError::Schema(format!("file_locks_new creation failed: {e}")))?;
+
+    // Copy existing data (add default lock_mode for old rows).
+    conn.execute(
+        "INSERT OR IGNORE INTO file_locks_new (file_path, task_id, locked_at, holder_pid, expires_at, lock_mode)
+         SELECT file_path, task_id, locked_at, holder_pid, expires_at, COALESCE(lock_mode, 'write')
+         FROM file_locks",
+        (),
+    )
+    .await
+    .ok(); // May fail if file_locks doesn't have lock_mode column yet — that's fine.
+
+    conn.execute("DROP TABLE IF EXISTS file_locks", ())
+        .await
+        .map_err(|e| DbError::Schema(format!("file_locks drop failed: {e}")))?;
+
+    conn.execute(
+        "ALTER TABLE file_locks_new RENAME TO file_locks",
+        (),
+    )
+    .await
+    .map_err(|e| DbError::Schema(format!("file_locks rename failed: {e}")))?;
+
+    Ok(())
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;