fix(fs_lock): always-join heartbeat thread in Drop to prevent post-cleanup writes

`LockGuard::drop` previously used `heartbeat_done.recv_timeout(100ms)`
to wait for the heartbeat thread's acknowledgement, then fell through
to `remove_lock_if_owned` regardless of whether the ack arrived. Under
CI load this created a race:

  1. Drop signals shutdown; recv_timeout times out before heartbeat
     acknowledges (heartbeat mid-`atomic_write_lock_metadata` IO).
  2. Drop logs warning, calls `remove_lock_if_owned` → file removed.
  3. A different caller acquires the lock, writes its own metadata.
  4. Our still-alive heartbeat finishes its in-flight write — its
     `heartbeat_once` validated our ownership BEFORE the on-disk swap,
     so its rename overwrites the new owner's lockfile with our
     stale metadata.
  5. The new owner's heartbeat sees foreign metadata, exits NotOwner.
     The new owner's Drop sees foreign metadata, `remove_lock_if_owned`
     returns Ok(false), the lockfile persists.

The Linux unit test `acquire_serializes_concurrent_callers` then
panics at `assert!(!path.exists())` — and any production lock with
this shape would leak a stale lockfile that the next acquire can't
match to remove.

Fix: Drop now unconditionally `unpark`s and `join()`s the heartbeat
handle. This bounds drop latency to one `park_timeout` iteration
(~25ms) plus the current `heartbeat_once` IO — typically <500ms under
CI load — but guarantees the heartbeat is dead before `remove_lock_if_owned`
runs. The `heartbeat_done` channel field is kept (drained defensively)
for backward compatibility but is no longer used for synchronization.

Verified locally with 5 consecutive `cargo test -p agent-file-tools
--release --lib fs_lock::` runs (10/10 pass each), plus full 850-test
lib suite still green.

Author: ualtinok

crates/aft/src/fs_lock.rs

@@ -68,25 +68,38 @@ pub struct LockGuard {
 
 impl Drop for LockGuard {
     fn drop(&mut self) {
+        // Signal shutdown then unconditionally join the heartbeat thread
+        // BEFORE removing the lockfile. The earlier `recv_timeout(100ms)`
+        // implementation could let `remove_lock_if_owned` race with a
+        // still-alive heartbeat:
+        //
+        //   1. Drop signals shutdown, ack times out under CI load.
+        //   2. Drop calls `remove_lock_if_owned` → file removed.
+        //   3. Another caller acquires the lock → writes its metadata.
+        //   4. Our heartbeat (still alive, mid-`atomic_write_lock_metadata`
+        //      from before shutdown was checked) overwrites the new
+        //      owner's file with our stale metadata. heartbeat_once's
+        //      ownership check happens BEFORE the write, so it can race
+        //      with a concurrent acquire that flips ownership in between.
+        //   5. The new owner's heartbeat sees foreign metadata, exits
+        //      `NotOwner`. The new owner's drop sees foreign metadata,
+        //      `remove_lock_if_owned` returns `Ok(false)`, file persists.
+        //
+        // Always-joining bounds drop latency to one `park_timeout`
+        // iteration (~25ms) plus the current `heartbeat_once` IO —
+        // typically <500ms under CI load. The unused `heartbeat_done`
+        // channel is kept for backward compatibility with any external
+        // code that may still construct LockGuard manually, but Drop no
+        // longer relies on it.
         self.shutdown.store(true, Ordering::Release);
-        if let Some(handle) = self.heartbeat.as_ref() {
+        if let Some(handle) = self.heartbeat.take() {
             handle.thread().unpark();
+            let _ = handle.join();
         }
-
-        if self
-            .heartbeat_done
-            .recv_timeout(Duration::from_millis(100))
-            .is_ok()
-        {
-            if let Some(handle) = self.heartbeat.take() {
-                let _ = handle.join();
-            }
-        } else {
-            slog_warn!(
-                "fs lock heartbeat thread for {} did not stop within 100ms",
-                self.path.display()
-            );
-        }
+        // Drain any pending ack so the receiver doesn't carry stale state
+        // if this LockGuard is somehow re-used (it isn't today, but be
+        // defensive).
+        while self.heartbeat_done.try_recv().is_ok() {}
 
         match remove_lock_if_owned(&self.path, &self.metadata) {
             Ok(true) => slog_info!("released filesystem lock at {}", self.path.display()),