fix(pm): a launcher waiting on a reclaimed state's install lock retries on the live file

collect_orphan_install_states released a state's .install.lock and only then unlinked it.
A launcher from a checkout re-created at the recorded path that had already opened the
lock (waiting on the collector) got it on release, the collector then unlinked the file it
held, and the next launcher created a fresh lock file at the same path and took it too:
two installers in one state. Predates #145527; that PR narrowed the window.

- pm.filesystem.lock_path: open + lock, then on POSIX confirm the held descriptor is still
  the file at the path (samestat; missing = stale) and otherwise retry on the current one,
  inside the caller's timeout. Optionally re-creates the parent the remover took.
- runtime_lock and install_states._install_lock (the only .install.lock acquirers of an
  install state) take the lock through it.
- The collector unlinks the lock file while still holding it on POSIX, so a waiter always
  sees it gone; on Windows an open file cannot be unlinked, so the unlink stays after
  release and a waiting launcher's open handle makes it fail.

hermes_cli/runtime_state.py

@@ -22,6 +22,7 @@ from pm.filesystem import (
     durable_write_bytes as _atomic_bytes,
     file_digest as _digest,
     lock_fd as _lock,
+    lock_path,
     read_bytes_or_none as _bytes,
 )
 
@@ -44,16 +45,13 @@ def runtime_lock(project: Path, *, timeout: float | None = INSTALL_LOCK_TIMEOUT_
     (``activate_dependencies`` still selects and leases the committed generation), writers that
     cannot be skipped pass ``timeout=None`` — an install the user asked for is theirs to wait on.
     """
-    state = install_state_dir(project)
-    state.mkdir(parents=True, exist_ok=True)
-    fd = os.open(state / ".install.lock", os.O_CREAT | os.O_RDWR, 0o600)
+    lock = install_state_dir(project) / ".install.lock"
+    fd = lock_path(lock, timeout=timeout, create_dir=True)
+    if fd is None:
+        LOG.warning("dependency lock still held after %ss; continuing without it (%s)", timeout, lock)
+        yield False
+        return
     try:
-        if not _lock(fd, wait=True, timeout=timeout):
-            LOG.warning(
-                "dependency lock still held after %ss; continuing without it (%s)",
-                timeout, state / ".install.lock")
-            yield False
-            return
         yield True
     finally:
         os.close(fd)

pm/filesystem.py

@@ -107,6 +107,47 @@ def unlock_fd(fd: int) -> None:
             fcntl.flock(fd, fcntl.LOCK_UN)
 
 
+def lock_path(path: Path, *, timeout: float | None, create_dir: bool = False) -> int | None:
+    """Open *path* (created if missing), wait up to ``timeout`` for its byte lock and return the
+    held descriptor; None when the wait expired. The caller closes the descriptor to release.
+
+    A holder that removes the lock file unlinks it while still holding it (the install-state
+    collector), so a waiter that opened the old file gets a lock on an inode no one else can
+    reach while the next opener creates a fresh file: both would hold "the" lock. Once held,
+    the descriptor must still be the file at *path*, else retry on the current one. Windows
+    cannot unlink an open file, so the check is POSIX-only. ``create_dir`` re-creates the
+    parent the remover may have taken with it.
+    """
+    deadline = None if timeout is None else time.monotonic() + timeout
+    while True:
+        if create_dir:
+            path.parent.mkdir(parents=True, exist_ok=True)
+        try:
+            fd = os.open(path, os.O_CREAT | os.O_RDWR, 0o600)
+        except FileNotFoundError:
+            if create_dir:
+                continue  # parent removed between mkdir and open
+            raise
+        try:
+            remaining = None if deadline is None else max(0.0, deadline - time.monotonic())
+            if not lock_fd(fd, wait=True, timeout=remaining):
+                os.close(fd)
+                return None
+            if os.name == "nt" or _same_file(fd, path):
+                return fd
+        except BaseException:
+            os.close(fd)
+            raise
+        os.close(fd)
+
+
+def _same_file(fd: int, path: Path) -> bool:
+    try:
+        return os.path.samestat(os.fstat(fd), os.stat(path))
+    except FileNotFoundError:
+        return False
+
+
 def read_bytes_or_none(path: Path) -> bytes | None:
     try:
         return path.read_bytes()

pm/install_states.py

@@ -100,17 +100,17 @@ def orphan_install_states(installs: Path) -> list[Path]:
 def _install_lock(state: Path):
     """Hold the state's install lock (``runtime_lock``'s) without waiting: yields False while
     another process holds it, or when it cannot even be opened."""
-    from pm.filesystem import lock_fd
+    from pm.filesystem import lock_path
 
     try:
-        fd = os.open(state / ".install.lock", os.O_CREAT | os.O_RDWR, 0o600)
+        fd = lock_path(state / ".install.lock", timeout=0)
     except OSError:
         fd = None
     if fd is None:
         yield False
         return
     try:
-        yield lock_fd(fd, wait=False)
+        yield True
     finally:
         os.close(fd)
 
@@ -165,13 +165,20 @@ def collect_orphan_install_states(installs: Path) -> list[Path]:
             emptied = _remove_contents(state)
             if emptied:  # still under the lock: only the record is left, so it goes last
                 shutil.rmtree(state / "inputs", ignore_errors=True)
+                # POSIX: unlink while still holding it. A launcher already waiting on this file
+                # then sees it gone once it gets the lock and retries on a fresh one (lock_path);
+                # unlinked after release, the path could be a file that launcher just made.
+                if os.name != "nt":
+                    with contextlib.suppress(OSError):
+                        (state / ".install.lock").unlink()
         if not emptied:
             logger.debug("orphan install state %s only partly removed; retried next pass", state.name)
             continue
         # Non-recursive: a launch from a checkout recreated at the same path may have taken the
         # lock the moment it was released and written here; its files make rmdir fail, not vanish.
         with contextlib.suppress(OSError):
-            (state / ".install.lock").unlink()
+            if os.name == "nt":  # an open file cannot be unlinked: a waiting launcher keeps it
+                (state / ".install.lock").unlink()
             state.rmdir()
         if not state.exists():
             removed.append(state)

tests/pm/test_install_lock_reclaim_race.py

@@ -0,0 +1,65 @@
+"""Reclaiming an orphan install state never leaves two holders of its install lock
+(``pm/install_states.py``, ``hermes_cli.runtime_state.runtime_lock``)."""
+
+from __future__ import annotations
+
+import os
+import threading
+
+import pm.environments
+import pm.install_states
+import pytest
+from hermes_cli.runtime_state import runtime_lock
+from pm.environments import install_key
+from pm.install_states import collect_orphan_install_states
+
+
+@pytest.mark.platforms("posix")
+def test_launcher_waiting_on_a_reclaimed_state_lock_excludes_later_launchers(tmp_path, monkeypatch):
+    """A launcher from a checkout re-created at the recorded path opens the lock while the
+    collector holds it, then gets it when the collector lets go. The lock file it waited on
+    is the one the collector removes; whatever it ends up holding must still exclude the
+    next launcher, or two installers write one state."""
+    installs = tmp_path / "home" / "installs"
+    checkout = tmp_path / "home" / "cache" / "scratch" / "clone"  # provably deleted: reclaimed
+    state = installs / install_key(checkout)
+    (state / "inputs").mkdir(parents=True)
+    (state / "inputs" / ".project-root").write_text(str(checkout), encoding="utf-8")
+    (state / "environments" / "gen" / "venv").mkdir(parents=True)
+    monkeypatch.setattr(pm.environments, "installs_root", lambda: installs)
+
+    opened, held, done = threading.Event(), threading.Event(), threading.Event()
+    real_open = os.open
+
+    def observed_open(path, *args, **kwargs):
+        fd = real_open(path, *args, **kwargs)
+        if threading.current_thread().name == "launcher-a" and os.fspath(path).endswith(".install.lock"):
+            opened.set()
+        return fd
+
+    def launcher_a():
+        with runtime_lock(checkout, timeout=None) as got:
+            if got:
+                held.set()
+            done.wait(30)
+
+    a = threading.Thread(target=launcher_a, name="launcher-a", daemon=True)
+    real_leased = pm.install_states._leased
+
+    def checkout_recreated_mid_reclaim(path):
+        # Called under the collector's lock, after its orphan re-check.
+        checkout.mkdir(parents=True)
+        a.start()
+        assert opened.wait(30)
+        return real_leased(path)
+
+    monkeypatch.setattr(os, "open", observed_open)
+    monkeypatch.setattr(pm.install_states, "_leased", checkout_recreated_mid_reclaim)
+    try:
+        collect_orphan_install_states(installs)
+        assert held.wait(30)
+        with runtime_lock(checkout, timeout=0) as launcher_b:
+            assert launcher_b is False, "two launchers hold the same state's install lock"
+    finally:
+        done.set()
+        a.join(30)