fix: harden Windows file-backed locking

CHANGELOG.md

@@ -6,6 +6,14 @@ adhere to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
 
+### Fixed
+- Windows file-backed mode now preserves concurrent audit and approval writes:
+  same-process threads are serialized before taking the `msvcrt` byte lock, lock
+  acquisition retries until available, and cleanup only unlocks after a
+  successful acquire.
+- CLI daemon probing now falls back to direct file-backed approval handling on
+  platforms without Unix domain sockets instead of crashing on `socket.AF_UNIX`.
+
 ## [0.3.4] — 2026-06-11
 
 Adds the **single-writer daemon**. Protocol unchanged (still 0.3); the daemon is

README.md

@@ -224,7 +224,8 @@ See **[ROADMAP.md](ROADMAP.md)** for where delego is going and where to help.
   `delego/brokers.py`.
 - **Not yet:** the MCP agent surface auto-routing to the daemon (it still talks
   to the firewall directly — wiring it is the next step), a TCP/cross-host
-  daemon transport (it's a local Unix socket today), and a non-MCP HTTP surface.
+  daemon transport (it's a local Unix socket today; Windows uses the file-backed
+  fallback), and a non-MCP HTTP surface.
 - **Known limitations:** without the daemon, concurrent writes to the file-backed
   ledger and approval store are serialised with an OS file lock (corruption-safe),
   and a `rate_limit` is exact only among processes sharing one home on one host.

delego/_locking.py

@@ -20,10 +20,12 @@
 from __future__ import annotations
 
 import os
-from contextlib import contextmanager
+from contextlib import contextmanager, nullcontext
 from pathlib import Path
 from typing import Iterator
 
+_LOCK_REGION_LENGTH = 1
+
 try:  # POSIX (Linux, macOS)
     import fcntl
 
@@ -33,16 +35,42 @@ def _acquire(fd: int) -> None:
     def _release(fd: int) -> None:
         fcntl.flock(fd, fcntl.LOCK_UN)
 
+    def _process_lock(lock_path: Path):
+        return nullcontext()
+
 except ImportError:  # Windows
+    import errno
     import msvcrt
+    import threading
+    import time
+
+    _LOCK_POLL_SECONDS = 0.05
+    _PROCESS_LOCKS: dict[str, threading.Lock] = {}
+    _PROCESS_LOCKS_GUARD = threading.Lock()
+
+    def _process_lock(lock_path: Path):
+        key = str(lock_path.resolve())
+        with _PROCESS_LOCKS_GUARD:
+            lock = _PROCESS_LOCKS.get(key)
+            if lock is None:
+                lock = threading.Lock()
+                _PROCESS_LOCKS[key] = lock
+            return lock
 
     def _acquire(fd: int) -> None:
-        os.lseek(fd, 0, os.SEEK_SET)
-        msvcrt.locking(fd, msvcrt.LK_LOCK, 1)
+        while True:
+            os.lseek(fd, 0, os.SEEK_SET)
+            try:
+                msvcrt.locking(fd, msvcrt.LK_NBLCK, _LOCK_REGION_LENGTH)
+                return
+            except OSError as e:
+                if e.errno not in (errno.EACCES, errno.EDEADLK):
+                    raise
+                time.sleep(_LOCK_POLL_SECONDS)
 
     def _release(fd: int) -> None:
         os.lseek(fd, 0, os.SEEK_SET)
-        msvcrt.locking(fd, msvcrt.LK_UNLCK, 1)
+        msvcrt.locking(fd, msvcrt.LK_UNLCK, _LOCK_REGION_LENGTH)
 
 
 @contextmanager
@@ -56,12 +84,16 @@ def file_lock(target: os.PathLike | str) -> Iterator[None]:
     target = Path(target)
     target.parent.mkdir(parents=True, exist_ok=True)
     lock_path = target.with_name(target.name + ".lock")
-    fd = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600)
-    try:
-        _acquire(fd)
-        yield
-    finally:
+    with _process_lock(lock_path):
+        fd = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600)
+        acquired = False
         try:
-            _release(fd)
+            _acquire(fd)
+            acquired = True
+            yield
         finally:
-            os.close(fd)
+            try:
+                if acquired:
+                    _release(fd)
+            finally:
+                os.close(fd)

delego/client.py

@@ -17,8 +17,14 @@ class DaemonError(RuntimeError):
     """The daemon returned an error, or could not be reached."""
 
 
+def _has_unix_socket() -> bool:
+    return hasattr(socket, "AF_UNIX")
+
+
 def daemon_running(socket_path) -> bool:
     """True if a live daemon answers ``ping`` on ``socket_path``."""
+    if not _has_unix_socket():
+        return False
     try:
         return DaemonClient(socket_path).ping()
     except DaemonError:
@@ -33,6 +39,8 @@ def __init__(self, socket_path, timeout: float = 30.0) -> None:
         self.timeout = timeout
 
     def _call(self, op: str, **args) -> Any:
+        if not _has_unix_socket():
+            raise DaemonError("delego daemon requires Unix domain sockets; this platform has no socket.AF_UNIX")
         s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
         s.settimeout(self.timeout)
         try:

tests/test_cli.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import socket
+
 from click.testing import CliRunner
 
 from delego import ProposedAction
@@ -31,6 +33,16 @@ def test_approve_echoes_what_was_approved(firewall):
     assert "place a small order" in out.output
 
 
+def test_approve_falls_back_without_unix_socket_support(firewall, monkeypatch):
+    monkeypatch.delattr(socket, "AF_UNIX", raising=False)
+
+    d = firewall.propose(_small_order())
+    out = CliRunner().invoke(cli, ["--home", _home_of(firewall), "approve", d.approval_id, "--as", "koishore"])
+
+    assert out.exit_code == 0
+    assert f"{d.approval_id}: approved" in out.output
+
+
 def test_verify_anchor_file_round_trip_and_rollback_detection(firewall, tmp_path):
     firewall.propose(
         ProposedAction(

tests/test_concurrency.py

@@ -16,6 +16,9 @@
 
 import threading
 
+import pytest
+
+import delego._locking as locking
 from delego import ProposedAction
 
 H = "0" * 64  # a stand-in 64-hex hash for fields the tests don't vary
@@ -29,6 +32,25 @@ def _run(target, n):
         t.join()
 
 
+def test_file_lock_does_not_release_after_failed_acquire(tmp_path, monkeypatch):
+    released = []
+
+    def fail_acquire(fd):
+        raise OSError("lock acquisition failed")
+
+    def release(fd):
+        released.append(fd)
+
+    monkeypatch.setattr(locking, "_acquire", fail_acquire)
+    monkeypatch.setattr(locking, "_release", release)
+
+    with pytest.raises(OSError):
+        with locking.file_lock(tmp_path / "audit.log.jsonl"):
+            pass
+
+    assert released == []
+
+
 # --------------------------------------------------------------------------- #
 # the hash chain stays valid and contiguous under concurrent appends
 # --------------------------------------------------------------------------- #