fix(lock): kernel-mediated byte lock with fail-safe cleanup and classified contention

pyproject.toml

@@ -42,5 +42,12 @@ packages = ["src/worker_bridge"]
 line-length = 110
 target-version = "py310"
 
+[tool.ruff.lint]
+# The historical default selection, pinned explicitly: ruff 0.16 expanded
+# its defaults substantially and began failing on already-shipped code
+# (RUF022, RUF100, BLE001, ...). Keep lint behavior version-independent;
+# re-expanding the rule set should be a deliberate future decision.
+select = ["E4", "E7", "E9", "F"]
+
 [tool.pytest.ini_options]
 testpaths = ["tests"]

src/worker_bridge/workspace.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import errno
 import fnmatch
 import hashlib
 import os
@@ -12,8 +13,6 @@
 from pathlib import Path
 from typing import Any
 
-import psutil
-
 from worker_bridge.environ import get_home as get_hermes_home
 
 
@@ -169,6 +168,54 @@ def _reftx_hook_present(repository: Path) -> bool:
     return (common / "hooks" / "reference-transaction").exists()
 
 
+def _claim_lock_file(fd: int) -> None:
+    """Atomically claim the lock file's first byte, or raise OSError.
+
+    Windows: ``msvcrt.locking`` — the C runtime's ``_locking`` operation,
+    held per descriptor and released by the operating system when the
+    owning process exits. POSIX: ``fcntl.flock`` — held per open file
+    description with the same death-release guarantee. Both turn "is the
+    holder still alive?" into a kernel question.
+    """
+    if os.name == "nt":
+        import msvcrt
+
+        os.lseek(fd, 0, os.SEEK_SET)
+        msvcrt.locking(fd, msvcrt.LK_NBLCK, 1)
+    else:
+        import fcntl
+
+        fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+
+
+def _is_lock_contention(exc: OSError) -> bool:
+    """True when a claim failure means another holder owns the lock.
+
+    POSIX ``flock`` reports contention as EWOULDBLOCK/EAGAIN (raised as
+    BlockingIOError) or EACCES. The Windows C runtime's ``_locking``
+    reports contention as EACCES/EDEADLK; EBADF, EINVAL, and other errors
+    are genuine failures whose cause must survive. Anything not classified
+    as contention propagates unchanged.
+    """
+    if isinstance(exc, BlockingIOError):
+        return True
+    if os.name == "nt":
+        return exc.errno in {errno.EACCES, getattr(errno, "EDEADLK", -1)}
+    return exc.errno == errno.EACCES
+
+
+def _release_lock_file(fd: int) -> None:
+    if os.name == "nt":
+        import msvcrt
+
+        os.lseek(fd, 0, os.SEEK_SET)
+        msvcrt.locking(fd, msvcrt.LK_UNLCK, 1)
+    else:
+        import fcntl
+
+        fcntl.flock(fd, fcntl.LOCK_UN)
+
+
 class RepositoryLock:
     """Process and cross-process lock used by direct-workspace mode.
 
@@ -178,9 +225,25 @@ class RepositoryLock:
     not serialize on one global per-repo lock. Same operation on the same
     repository still excludes.
 
-    ``wait_seconds`` bounds how long acquisition waits for a live holder to
-    release before raising. The default (0) keeps the historical fail-fast
-    behavior direct mode relies on.
+    Cross-process exclusion is a kernel-mediated byte lock on a small
+    per-key lock file (``msvcrt.locking`` on Windows, ``fcntl.flock`` on
+    POSIX). The kernel releases it when the owning process dies, so a
+    crashed holder needs no stale-file sweep — reading a pid from the file
+    and unlinking it (the previous scheme) could not distinguish a dead
+    holder's file from one a peer had just legitimately recreated, and the
+    resulting unlink could delete a live lock. Lock files are therefore
+    never unlinked: they are permanent claim points, one per repository
+    (and operation), a few bytes each.
+
+    ``wait_seconds`` bounds how long acquisition waits for a *cross-process*
+    holder to release before raising; the default (0) fails fast. In-process
+    contention on the same key waits up to ``max(30, wait_seconds)`` seconds
+    on the thread lock first — parallel worktree setup relies on that floor.
+
+    Behavior note: the pid written to the lock file is diagnostic only. A
+    failure to record it is ignored instead of aborting acquisition — the
+    previous implementation raised from the pid write, so this is a
+    deliberate behavior change.
     """
 
     def __init__(
@@ -202,39 +265,58 @@ def __init__(
     def __enter__(self) -> "RepositoryLock":
         if not self._thread_lock.acquire(timeout=max(30.0, self._wait_seconds)):
             raise WorkspaceError("repository is busy")
-        self._path.parent.mkdir(parents=True, exist_ok=True)
-        deadline = time.monotonic() + self._wait_seconds
-        swept_stale = False
-        while True:
-            try:
-                self._fd = os.open(self._path, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
-                os.write(self._fd, str(os.getpid()).encode())
-                return self
-            except FileExistsError as exc:
-                stale = False
+        owned = False
+        fd: int | None = None
+        try:
+            self._path.parent.mkdir(parents=True, exist_ok=True)
+            deadline = time.monotonic() + self._wait_seconds
+            while True:
+                candidate = os.open(self._path, os.O_CREAT | os.O_RDWR, 0o644)
                 try:
-                    pid = int(self._path.read_text(encoding="ascii").strip())
-                    stale = not psutil.pid_exists(pid)
-                except (OSError, ValueError):
-                    stale = True
-                if stale and not swept_stale:
-                    swept_stale = True
-                    self._path.unlink(missing_ok=True)
-                    continue
-                if time.monotonic() < deadline:
+                    _claim_lock_file(candidate)
+                except OSError as exc:
+                    # Drop the claiming fd; a close failure here must not
+                    # replace the claim's error, which carries the real cause.
+                    try:
+                        os.close(candidate)
+                    except OSError:
+                        pass
+                    if not _is_lock_contention(exc):
+                        raise
+                    if time.monotonic() >= deadline:
+                        raise WorkspaceError(f"repository lock exists: {self._path}") from exc
                     time.sleep(0.2)
-                    # The holder may die while we wait; allow another sweep.
-                    swept_stale = False
                     continue
-                self._thread_lock.release()
-                raise WorkspaceError(f"repository lock exists: {self._path}") from exc
+                fd = candidate
+                break
+            self._fd = fd
+            owned = True
+            try:
+                # Diagnostics only: the owning pid for humans inspecting the
+                # locks directory. Never gates ownership.
+                os.ftruncate(fd, 0)
+                os.write(fd, str(os.getpid()).encode())
+            except OSError:
+                pass
+            return self
+        finally:
+            if not owned:
+                try:
+                    if fd is not None:
+                        os.close(fd)
+                finally:
+                    self._thread_lock.release()
 
     def __exit__(self, *_exc: Any) -> None:
-        if self._fd is not None:
-            os.close(self._fd)
-            self._fd = None
-        self._path.unlink(missing_ok=True)
-        self._thread_lock.release()
+        try:
+            if self._fd is not None:
+                fd, self._fd = self._fd, None
+                try:
+                    _release_lock_file(fd)
+                finally:
+                    os.close(fd)
+        finally:
+            self._thread_lock.release()
 
 
 class WorkspaceManager:

tests/_lock_child.py

@@ -0,0 +1,110 @@
+"""Subprocess contender used by RepositoryLock concurrency tests.
+
+Driven by a JSON config file so the parent test can orchestrate real
+cross-process contention deterministically:
+
+    python _lock_child.py <config.json>
+
+Config keys:
+    repo, operation, wait_seconds  -- lock identity and wait budget
+    signaldir                      -- directory for markers/outcomes
+    id                             -- this child's marker suffix ("0"/"1")
+
+Optional deterministic seams (inert unless the implementation calls the
+patched functions during acquisition):
+    barrier_pid_exists  -- synchronize BOTH children at the stale-pid
+                           observation point before psutil.pid_exists returns
+    wait_peer_created_then_unlink -- before unlinking the lock path, block
+                           until the peer reports holding the lock (forces
+                           the stale-sweep to run against a *live* lock)
+
+Outcome protocol: writes ``outcome-<id>.json`` ({acquired, error}); when
+acquired, also writes ``holding-<id>`` and keeps the lock until a
+``release-<id>`` file appears (polling with hard caps so nothing hangs).
+"""
+from __future__ import annotations
+
+import json
+import os
+import sys
+import time
+from pathlib import Path
+
+
+def _await_file(path: Path, timeout: float = 15.0) -> bool:
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if path.exists():
+            return True
+        time.sleep(0.02)
+    return False
+
+
+def _patch_pid_exists_barrier(signaldir: Path, child_id: str, peer_id: str) -> None:
+    """Both children meet here before agreeing the pid is dead."""
+    import psutil
+
+    mine = signaldir / f"saw-stale-{child_id}"
+    peers = signaldir / f"saw-stale-{peer_id}"
+
+    def synchronized(pid: int) -> bool:
+        mine.write_text(str(pid), encoding="ascii")
+        if not _await_file(peers):
+            raise RuntimeError("pid_exists barrier timed out")
+        return False  # both contenders agree: stale
+
+    psutil.pid_exists = synchronized
+
+
+def _patch_unlink_waits_for_peer(signaldir: Path, child_id: str, peer_id: str, lock_path: Path) -> None:
+    """Force the stale-sweep unlink to run after the peer recreated the lock."""
+    real_unlink = os.unlink
+
+    def ordered_unlink(path, *args, **kwargs):
+        try:
+            if Path(path).resolve() == lock_path.resolve():
+                if not _await_file(signaldir / f"holding-{peer_id}"):
+                    raise RuntimeError("peer-created barrier timed out")
+        except OSError:
+            pass  # not the lock path, or already gone: behave like unlink
+        return real_unlink(path, *args, **kwargs)
+
+    os.unlink = ordered_unlink
+
+
+def main() -> int:
+    config = json.loads(Path(sys.argv[1]).read_text(encoding="utf-8"))
+    signaldir = Path(config["signaldir"])
+    child_id = str(config["id"])
+    peer_id = str(config["peer"])
+    outcome_path = signaldir / f"outcome-{child_id}.json"
+
+    from worker_bridge.workspace import RepositoryLock
+
+    lock = RepositoryLock(
+        config["repo"], operation=config["operation"], wait_seconds=config["wait_seconds"]
+    )
+
+    if config.get("barrier_pid_exists"):
+        _patch_pid_exists_barrier(signaldir, child_id, peer_id)
+    if config.get("wait_peer_created_then_unlink"):
+        _patch_unlink_waits_for_peer(signaldir, child_id, peer_id, lock._path)
+
+    try:
+        with lock:
+            (signaldir / f"holding-{child_id}").write_text("", encoding="ascii")
+            outcome_path.write_text(
+                json.dumps({"id": child_id, "acquired": True, "error": None}), encoding="utf-8"
+            )
+            _await_file(signaldir / f"release-{child_id}", timeout=30.0)
+        return 0
+    except Exception as exc:  # noqa: BLE001 - report every failure shape to the parent
+        outcome_path.write_text(
+            json.dumps({"id": child_id, "acquired": False, "error": f"{type(exc).__name__}: {exc}"}),
+            encoding="utf-8",
+        )
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/lock_test_helpers.py

@@ -0,0 +1,61 @@
+"""Shared helpers for cross-process RepositoryLock tests.
+
+Spawns real contender subprocesses (``_lock_child.py``) coordinated through
+marker files, so mutual exclusion is tested against actual OS-level
+processes rather than simulated lock-file contents.
+"""
+from __future__ import annotations
+
+import json
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+CHILD = Path(__file__).parent / "_lock_child.py"
+
+
+def child_config(tmp: Path, *, repo: Path, operation: str, wait_seconds: float,
+                 id: str, peer: str, **overrides) -> Path:  # noqa: A002 - test fixture ids
+    config = {
+        "repo": str(repo),
+        "operation": operation,
+        "wait_seconds": wait_seconds,
+        "signaldir": str(tmp / "signals"),
+        "id": id,
+        "peer": peer,
+    }
+    config.update(overrides)
+    (tmp / "signals").mkdir(parents=True, exist_ok=True)
+    path = tmp / f"config-{id}.json"
+    path.write_text(json.dumps(config), encoding="utf-8")
+    return path
+
+
+def spawn(config: Path) -> subprocess.Popen:
+    return subprocess.Popen(
+        [sys.executable, str(CHILD), str(config)],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        text=True,
+    )
+
+
+def await_file(path: Path, timeout: float = 30.0) -> str:
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if path.exists():
+            return path.read_text(encoding="utf-8")
+        time.sleep(0.02)
+    raise AssertionError(f"timed out waiting for {path.name}")
+
+
+def outcome(tmp: Path, child_id: str) -> dict:
+    return json.loads(await_file(tmp / "signals" / f"outcome-{child_id}.json"))
+
+
+def release_and_join(tmp: Path, children: list[subprocess.Popen]) -> None:
+    for child_id in ("0", "1"):
+        (tmp / "signals" / f"release-{child_id}").write_text("", encoding="ascii")
+    for proc in children:
+        proc.wait(timeout=30)