feat(file-lock): implement PID+TTL hybrid file locking [fn-110.2]

Replace fragile string-matching lock mechanism with PID-based crash
detection + TTL fallback. acquire() now uses INSERT OR IGNORE with
row count check, stores holder_pid and expires_at. cleanup_stale()
removes locks from dead processes (via nix kill(pid,0)) and expired
TTLs. Add heartbeat() method and `flowctl heartbeat --task` CLI
command. Fix pre-existing private module access in lifecycle.rs.

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

Author: z23cc

flowctl/Cargo.lock

@@ -322,6 +322,10 @@ impl FileLockRepo {
         block_on(flowctl_db::FileLockRepo::new(self.0.clone()).check(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> {
         let inner = self.0.clone();

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

@@ -764,3 +764,24 @@ pub fn cmd_lock_check(json: bool, file: Option<String>) {
         }
     }
 }
+
+pub fn cmd_heartbeat(json: bool, task: String) {
+    let _flow_dir = ensure_flow_exists();
+    let conn = require_db();
+    let repo = crate::commands::db_shim::FileLockRepo::new(&conn);
+
+    match repo.heartbeat(&task) {
+        Ok(count) => {
+            if json {
+                json_output(json!({
+                    "task": task,
+                    "extended": count,
+                    "message": format!("Extended TTL for {} lock(s)", count),
+                }));
+            } else {
+                println!("Extended TTL for {} lock(s) for task {}", count, task);
+            }
+        }
+        Err(e) => error_exit(&format!("Failed to heartbeat: {}", e)),
+    }
+}

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

@@ -315,6 +315,12 @@ enum Commands {
         #[arg(long)]
         all: bool,
     },
+    /// Extend lock TTL for a task (Teams mode heartbeat).
+    Heartbeat {
+        /// Task ID to extend locks for.
+        #[arg(long)]
+        task: String,
+    },
     /// Check file lock status (Teams mode).
     LockCheck {
         /// Specific file to check.
@@ -527,6 +533,7 @@ fn main() {
         Commands::Lock { task, files } => query::cmd_lock(json, task, files),
         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),
 
         // Workflow
         Commands::Ready { epic } => workflow::cmd_ready(json, epic),

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

@@ -15,6 +15,7 @@ serde = { workspace = true }
 serde_json = { workspace = true }
 thiserror = { workspace = true }
 chrono = { workspace = true }
+nix = { workspace = true }
 tracing = { workspace = true }
 
 [lints]

flowctl/crates/flowctl-db/Cargo.toml

@@ -1,10 +1,18 @@
 //! Async repository for runtime file locks (Teams mode concurrency).
+//!
+//! Uses PID-based crash detection + TTL fallback for hung processes.
+//! Stale locks (dead PID or expired TTL) are auto-cleaned on `acquire()`.
 
-use chrono::Utc;
+use chrono::{Duration, Utc};
 use libsql::{params, Connection};
+use nix::sys::signal;
+use nix::unistd::Pid;
 
 use crate::error::DbError;
 
+/// Default lock TTL: 45 minutes.
+const LOCK_TTL_MINUTES: i64 = 45;
+
 /// Async repository for runtime file locks. Load-bearing for Teams-mode
 /// concurrency: `acquire` on an already-locked file returns
 /// `DbError::Constraint`.
@@ -17,38 +25,107 @@ impl FileLockRepo {
         Self { conn }
     }
 
-    /// Acquire a lock on a file for a task. Returns `DbError::Constraint`
-    /// if the file is already locked by another task.
+    /// Acquire a lock on a file for a task.
+    ///
+    /// Calls `cleanup_stale()` first, then uses `INSERT OR IGNORE` with
+    /// `holder_pid` and `expires_at`. If the row already exists (rows_affected=0),
+    /// checks whether it belongs to this task (idempotent Ok) or another
+    /// (returns `DbError::Constraint`).
     pub async fn acquire(&self, file_path: &str, task_id: &str) -> Result<(), DbError> {
-        let res = self
+        self.cleanup_stale().await?;
+
+        let pid = std::process::id();
+        let expires_at = (Utc::now() + Duration::minutes(LOCK_TTL_MINUTES)).to_rfc3339();
+
+        let rows_affected = self
             .conn
             .execute(
-                "INSERT INTO file_locks (file_path, task_id, locked_at) VALUES (?1, ?2, ?3)",
+                "INSERT OR IGNORE INTO file_locks (file_path, task_id, locked_at, holder_pid, expires_at)
+                 VALUES (?1, ?2, ?3, ?4, ?5)",
                 params![
                     file_path.to_string(),
                     task_id.to_string(),
                     Utc::now().to_rfc3339(),
+                    pid as i64,
+                    expires_at,
                 ],
             )
-            .await;
-
-        match res {
-            Ok(_) => Ok(()),
-            Err(e) => {
-                let msg = e.to_string();
-                let low = msg.to_lowercase();
-                if low.contains("unique constraint")
-                    || low.contains("constraint failed")
-                    || low.contains("primary key")
-                {
-                    Err(DbError::Constraint(format!(
-                        "file already locked: {file_path}"
-                    )))
-                } else {
-                    Err(DbError::LibSql(e))
-                }
+            .await?;
+
+        if rows_affected > 0 {
+            return Ok(());
+        }
+
+        // Row already exists — check if it's our own lock (idempotent).
+        let existing_task = self.check(file_path).await?;
+        if existing_task.as_deref() == Some(task_id) {
+            return Ok(());
+        }
+
+        Err(DbError::Constraint(format!(
+            "file already locked: {file_path}"
+        )))
+    }
+
+    /// Remove stale locks: dead PIDs (via `kill(pid, 0)`) and expired TTLs.
+    pub async fn cleanup_stale(&self) -> Result<u64, DbError> {
+        let mut total_cleaned = 0u64;
+
+        // 1. Delete expired locks (expires_at < now).
+        let now = Utc::now().to_rfc3339();
+        let expired = self
+            .conn
+            .execute(
+                "DELETE FROM file_locks WHERE expires_at IS NOT NULL AND expires_at < ?1",
+                params![now],
+            )
+            .await?;
+        total_cleaned += expired;
+
+        // 2. Check PIDs of remaining locks — delete dead ones.
+        let mut rows = self
+            .conn
+            .query(
+                "SELECT file_path, holder_pid FROM file_locks WHERE holder_pid IS NOT NULL",
+                (),
+            )
+            .await?;
+
+        let mut dead_files = Vec::new();
+        while let Some(row) = rows.next().await? {
+            let fp: String = row.get(0)?;
+            let pid: i64 = row.get(1)?;
+
+            if !is_process_alive(pid as i32) {
+                dead_files.push(fp);
             }
         }
+
+        for fp in &dead_files {
+            let n = self
+                .conn
+                .execute(
+                    "DELETE FROM file_locks WHERE file_path = ?1",
+                    params![fp.clone()],
+                )
+                .await?;
+            total_cleaned += n;
+        }
+
+        Ok(total_cleaned)
+    }
+
+    /// Extend `expires_at` for all locks held by a task (heartbeat).
+    pub async fn heartbeat(&self, task_id: &str) -> Result<u64, DbError> {
+        let new_expires = (Utc::now() + Duration::minutes(LOCK_TTL_MINUTES)).to_rfc3339();
+        let n = self
+            .conn
+            .execute(
+                "UPDATE file_locks SET expires_at = ?1 WHERE task_id = ?2",
+                params![new_expires, task_id.to_string()],
+            )
+            .await?;
+        Ok(n)
     }
 
     /// Release locks held by a task. Returns number of rows deleted.
@@ -86,3 +163,9 @@ impl FileLockRepo {
         }
     }
 }
+
+/// Check if a process is alive using `kill(pid, 0)`.
+fn is_process_alive(pid: i32) -> bool {
+    // kill(pid, 0) returns Ok if process exists, Err(ESRCH) if not.
+    signal::kill(Pid::from_raw(pid), None).is_ok()
+}

flowctl/crates/flowctl-db/src/repo/file_lock.rs

@@ -501,6 +501,83 @@ mod tests {
         assert!(repo.check("src/c.rs").await.unwrap().is_none());
     }
 
+    #[tokio::test]
+    async fn file_lock_idempotent_reacquire() {
+        let (_db, conn) = open_memory_async().await.unwrap();
+        let repo = FileLockRepo::new(conn.clone());
+
+        // Acquiring the same file for the same task twice should succeed (idempotent).
+        repo.acquire("src/a.rs", "fn-1.1").await.unwrap();
+        repo.acquire("src/a.rs", "fn-1.1").await.unwrap();
+        assert_eq!(
+            repo.check("src/a.rs").await.unwrap().as_deref(),
+            Some("fn-1.1")
+        );
+    }
+
+    #[tokio::test]
+    async fn file_lock_expired_cleanup() {
+        let (_db, conn) = open_memory_async().await.unwrap();
+        let repo = FileLockRepo::new(conn.clone());
+
+        // Insert a lock with an already-expired TTL directly.
+        let past = (chrono::Utc::now() - chrono::Duration::minutes(1)).to_rfc3339();
+        conn.execute(
+            "INSERT INTO file_locks (file_path, task_id, locked_at, holder_pid, expires_at)
+             VALUES ('src/expired.rs', 'fn-old', ?1, 99999, ?2)",
+            libsql::params![past.clone(), past],
+        )
+        .await
+        .unwrap();
+
+        // The expired lock should be visible before cleanup.
+        assert!(repo.check("src/expired.rs").await.unwrap().is_some());
+
+        // cleanup_stale should remove it.
+        let cleaned = repo.cleanup_stale().await.unwrap();
+        assert!(cleaned >= 1);
+        assert!(repo.check("src/expired.rs").await.unwrap().is_none());
+    }
+
+    #[tokio::test]
+    async fn file_lock_heartbeat_extends_ttl() {
+        let (_db, conn) = open_memory_async().await.unwrap();
+        let repo = FileLockRepo::new(conn.clone());
+
+        repo.acquire("src/a.rs", "fn-1.1").await.unwrap();
+        repo.acquire("src/b.rs", "fn-1.1").await.unwrap();
+
+        let extended = repo.heartbeat("fn-1.1").await.unwrap();
+        assert_eq!(extended, 2);
+
+        // Heartbeat on a non-existent task returns 0.
+        let none = repo.heartbeat("fn-nonexistent").await.unwrap();
+        assert_eq!(none, 0);
+    }
+
+    #[tokio::test]
+    async fn file_lock_acquire_cleans_expired_before_insert() {
+        let (_db, conn) = open_memory_async().await.unwrap();
+        let repo = FileLockRepo::new(conn.clone());
+
+        // Insert expired lock for a file.
+        let past = (chrono::Utc::now() - chrono::Duration::minutes(1)).to_rfc3339();
+        conn.execute(
+            "INSERT INTO file_locks (file_path, task_id, locked_at, holder_pid, expires_at)
+             VALUES ('src/a.rs', 'fn-old', ?1, 99999, ?2)",
+            libsql::params![past.clone(), past],
+        )
+        .await
+        .unwrap();
+
+        // Acquiring the same file should succeed because the old lock is expired.
+        repo.acquire("src/a.rs", "fn-1.1").await.unwrap();
+        assert_eq!(
+            repo.check("src/a.rs").await.unwrap().as_deref(),
+            Some("fn-1.1")
+        );
+    }
+
     // ── PhaseProgressRepo ───────────────────────────────────────────
 
     #[tokio::test]

flowctl/crates/flowctl-db/src/repo/mod.rs

@@ -105,7 +105,9 @@ CREATE TABLE IF NOT EXISTS runtime_state (
 CREATE TABLE IF NOT EXISTS file_locks (
     file_path  TEXT PRIMARY KEY,
     task_id    TEXT NOT NULL,
-    locked_at  TEXT NOT NULL
+    locked_at  TEXT NOT NULL,
+    holder_pid INTEGER,
+    expires_at TEXT
 );
 
 CREATE TABLE IF NOT EXISTS heartbeats (