perf(collaboration): avoid rewriting idle request lock holders

loopx/control_plane/collaboration/conversation_inventory.py

@@ -44,7 +44,9 @@ def _locked_request_batch(root, scope, requests):
             key = (request_id, None if scope.exact else path)
             if key not in lock_keys:
                 try:
-                    locks.enter_context(_request_lock(root, request_id, scope, path.with_suffix(".lock")))
+                    locks.enter_context(
+                        _request_lock(root, request_id, scope, path.with_suffix(".lock"), preserve_holder=True)
+                    )
                 except (OSError, ValueError, KeyError, TypeError):
                     continue
                 lock_keys.add(key)

loopx/control_plane/collaboration/inbox.py

@@ -15,7 +15,12 @@
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
-from ...file_lock import exclusive_cross_runtime_file_lock
+from ...file_lock import (
+    exclusive_cross_runtime_file_lock,
+    exclusive_file_lock,
+    exclusive_mutation_file_lock,
+    try_exclusive_file_lock,
+)
 from ..runtime.file_paths import windows_extended_path
 from ..content_digest import BARE_SHA256_PATTERN, ENVELOPED_SHA256_PATTERN
 from ..todos.contract import TODO_ID_PATTERN
@@ -132,15 +137,28 @@ def _request_lock(
     request_id: str,
     scope: CollaborationGoalScope | None,
     legacy_path: Path,
-    *, force_exact: bool = False,
+    *,
+    force_exact: bool = False,
+    preserve_holder: bool = False,
 ):
     lock = (
         _root(root) / "request-locks" / request_id
         if force_exact or scope is not None and scope.exact
         else legacy_path
     )
-    with exclusive_cross_runtime_file_lock(lock):
-        yield
+    if not preserve_holder:
+        with exclusive_cross_runtime_file_lock(lock):
+            yield
+        return
+    # Inventory needs both fences but can keep the prior holder record when
+    # the kernel fence is free. A busy legacy writer gets the usual timed wait.
+    with exclusive_mutation_file_lock(lock):
+        with try_exclusive_file_lock(lock, preserve_holder=True) as held:
+            if held is not None:
+                yield
+                return
+        with exclusive_file_lock(lock):
+            yield
 
 
 def normalize_request(value: Any) -> dict | None:

loopx/file_lock.py

@@ -518,7 +518,10 @@ def lock_holder_liveness(path: Path) -> tuple[str, dict[str, object]]:
     return (LOCK_HOLDER_LIVE if process_is_alive(pid) else LOCK_HOLDER_DEAD), record
 
 
-def _operator_action(holder: dict[str, object], *, retry_mode: str) -> dict[str, object]:
+def _operator_action(
+    holder: dict[str, object], *, retry_mode: str, kernel_fence: bool
+) -> dict[str, object]:
+    fence = "kernel lock" if kernel_fence else "mutation marker"
     return {
         "required": True,
         "action": "inspect_lock_holder",
@@ -529,9 +532,13 @@ def _operator_action(holder: dict[str, object], *, retry_mode: str) -> dict[str,
         "retry_mode": retry_mode,
         "steps": [
             "Inspect the recorded holder host, PID and operation on that host.",
-            "Confirm the process is stalled before terminating it.",
+            "Confirm the process still holds the lock and is stalled before terminating it.",
             "Retry according to retry_mode after the holder exits.",
             "Do not delete the lock file; the kernel lock is authoritative.",
+        ] if holder else [
+            f"The {fence} was busy at the deadline; current holder unknown.",
+            "Check for overlapping operations and retry according to retry_mode after contention clears.",
+            "Do not delete the lock files.",
         ],
     }
 
@@ -573,10 +580,11 @@ def __init__(
         self.incident_channel = incident_channel
         raw_holder = incident.get("holder")
         holder: dict[str, object] = raw_holder if isinstance(raw_holder, dict) else {}
-        super().__init__(
-            "file lock acquisition timed out"
-            + (f" while waiting for holder pid {holder.get('pid')}" if holder.get("pid") else "")
+        detail = (
+            f"recorded holder pid {holder['pid']}" if holder.get("pid")
+            else "lock busy, current holder unknown"
         )
+        super().__init__(f"file lock acquisition timed out ({detail})")
 
     def to_payload(self) -> dict[str, object]:
         return {
@@ -597,31 +605,46 @@ def lock_timeout_error_fields(error: BaseException) -> dict[str, object]:
 def _timeout_error(
     path: Path,
     *,
+    kernel_fence: bool,
     policy: LockAcquisitionPolicy,
     timeout_seconds: float,
     waited_seconds: float,
     started_at: str,
     agent_id: str | None,
     operation: str | None,
 ) -> LockAcquireTimeoutError:
-    holder, _ = _read_latest_holder_record(path)
+    state, last_holder = lock_holder_liveness(path)
+    lock_id = _lock_id(path)
+    holder: dict[str, object] = {}
+    if kernel_fence and state == LOCK_HOLDER_LIVE and last_holder.get("lock_id") == lock_id:
+        # A marker can belong to a reader that preserves the kernel record, or
+        # to a contender still waiting for the kernel lock. Neither binds the
+        # recorded PID to the fence we are waiting on.
+        try:
+            _effect_mutation_lock_path(path).lstat()
+        except FileNotFoundError:
+            holder = last_holder
+        except OSError:
+            pass
     waiter = {
         **_identity(agent_id=agent_id, operation=operation, policy=policy),
         "started_at": started_at,
         "timeout_seconds": round(timeout_seconds, 3),
         "waited_seconds": round(waited_seconds, 3),
     }
-    action = _operator_action(holder, retry_mode=LOCK_POLICIES[policy].retry_mode)
+    action = _operator_action(holder, retry_mode=LOCK_POLICIES[policy].retry_mode, kernel_fence=kernel_fence)
     incident: dict[str, object] = {
         "schema_version": LOCK_INCIDENT_SCHEMA_VERSION,
         "error_code": LOCK_ACQUIRE_TIMEOUT_ERROR_CODE,
         "recorded_at": _utc_now_iso(),
-        "lock_id": _lock_id(path),
+        "lock_id": lock_id,
         "policy": policy.value,
         "holder": holder,
         "waiter": waiter,
         "operator_action": action,
     }
+    if last_holder and not holder:
+        incident["last_kernel_holder"] = last_holder
     recorded = _append_incident(path, incident)
     return LockAcquireTimeoutError(
         incident=incident,
@@ -673,6 +696,7 @@ def exclusive_file_lock(
             if now >= deadline:
                 raise _timeout_error(
                     path,
+                    kernel_fence=True,
                     policy=selected_policy,
                     timeout_seconds=timeout,
                     waited_seconds=now - started,
@@ -777,15 +801,15 @@ def _effect_mutation_claim_path(path: Path, token: str) -> Path:
 def process_is_alive(pid: object) -> bool:
     """Probe a process without sending signals or console control events."""
 
-    if isinstance(pid, bool) or not isinstance(pid, int) or pid <= 0:
+    if isinstance(pid, bool) or not isinstance(pid, int) or not 0 < pid <= 0xFFFFFFFF:
         return False
     if os.name == "nt":
         # Windows signal 0 is CTRL_C_EVENT, not the side-effect-free POSIX
         # existence probe provided by kill(pid, 0).
         return _windows_process_is_alive(pid)
     try:
         os.kill(pid, 0)
-    except ProcessLookupError:
+    except (ProcessLookupError, OverflowError):
         return False
     except PermissionError:
         return True
@@ -1210,6 +1234,7 @@ def exclusive_mutation_file_lock(
             if now >= deadline:
                 raise _timeout_error(
                     path,
+                    kernel_fence=False,
                     policy=selected_policy,
                     timeout_seconds=timeout,
                     waited_seconds=now - started,

tests/control_plane/test_chat_request_inventory.py

@@ -39,6 +39,146 @@ def fingerprint():
     assert len(scans) == 2, "Each HTTP read must qualify its own runtime revision"
 
 
+def test_inventory_preserves_idle_holder_and_waits_for_existing_writers(api, monkeypatch):
+    from concurrent.futures import ThreadPoolExecutor
+    from threading import Event
+    import time
+
+    from loopx import file_lock
+    from loopx.file_lock import _effect_mutation_lock_path, exclusive_file_lock, lock_holder_path
+
+    _, _, route = _source(api)
+    _, fixture, _ = api
+    path = _root(fixture["runtime"]) / "request-locks" / route["request_id"]
+    with exclusive_file_lock(path, operation="request_writer"):
+        pass
+    previous_holder = lock_holder_path(path).read_bytes()
+
+    code, first = _inventory(api, route["session_id"], view="history")
+    assert code == 200 and first["matched"] == 1
+    assert lock_holder_path(path).read_bytes() == previous_holder
+
+    # A caller that only holds the kernel fence must still exclude inventory.
+    with ThreadPoolExecutor(max_workers=1) as pool:
+        with exclusive_file_lock(path, operation="older_request_writer"):
+            reading = pool.submit(_inventory, api, route["session_id"], view="history")
+            marker = _effect_mutation_lock_path(path)
+            deadline = time.monotonic() + 3
+            while not marker.exists() and not reading.done() and time.monotonic() < deadline:
+                time.sleep(0.01)
+            assert marker.exists() and not reading.done()
+        code, recovered = reading.result(timeout=5)
+    assert code == 200 and recovered["matched"] == 1
+    assert recovered["snapshot_digest"] == first["snapshot_digest"]
+    assert not marker.exists()
+
+    # A marker-only writer must also exclude inventory, even with a free kernel lock.
+    waiting_for_marker = Event()
+    reclaim = file_lock._reclaim_stale_effect_mutation_lock
+
+    def observe_marker_conflict(marker_path):
+        reclaimed = reclaim(marker_path)
+        if marker_path == marker:
+            waiting_for_marker.set()
+        return reclaimed
+
+    monkeypatch.setattr(file_lock, "_reclaim_stale_effect_mutation_lock", observe_marker_conflict)
+    with ThreadPoolExecutor(max_workers=1) as pool:
+        with file_lock.exclusive_mutation_file_lock(path, operation="marker_writer"):
+            reading = pool.submit(_inventory, api, route["session_id"], view="history")
+            assert waiting_for_marker.wait(timeout=10) and not reading.done()
+        code, recovered = reading.result(timeout=15)
+    assert code == 200 and recovered["snapshot_digest"] == first["snapshot_digest"]
+    assert not marker.exists()
+
+
+def test_inventory_contenders_do_not_name_the_released_kernel_holder(api, monkeypatch):
+    from concurrent.futures import ThreadPoolExecutor
+    import json
+    import os
+    import subprocess
+    import sys
+    from threading import Event
+
+    from loopx.control_plane.collaboration import conversation_inventory
+    from loopx.file_lock import _effect_mutation_lock_path, lock_holder_path, lock_incident_path
+
+    script = """
+import json
+import sys
+from pathlib import Path
+from loopx.file_lock import (
+    LockAcquireTimeoutError, exclusive_file_lock, exclusive_mutation_file_lock,
+)
+
+path, operation = Path(sys.argv[1]), sys.argv[2]
+lock = exclusive_mutation_file_lock if operation == "marker_waiter" else exclusive_file_lock
+try:
+    with lock(path, operation=operation):
+        if operation != "previous_writer":
+            raise AssertionError("waiter entered while the history read held its fence")
+except LockAcquireTimeoutError as error:
+    print(json.dumps({"message": str(error), **error.to_payload()}))
+"""
+    _, _, route = _source(api)
+    _, fixture, _ = api
+    path = _root(fixture["runtime"]) / "request-locks" / route["request_id"]
+
+    def contender(operation):
+        return subprocess.run(
+            [sys.executable, "-c", script, str(path), operation],
+            check=True, capture_output=True, text=True, timeout=25,
+        ).stdout
+
+    contender("previous_writer")
+    previous_holder = lock_holder_path(path).read_bytes()
+    historical = json.loads(previous_holder)
+    assert historical["released_at"] and historical["pid"] != os.getpid()
+    code, idle = _inventory(api, route["session_id"], view="history")
+    assert code == 200 and idle["matched"] == 1
+    assert lock_holder_path(path).read_bytes() == previous_holder
+
+    entered, release = Event(), Event()
+    read_receipts = conversation_inventory._read_receipt_batch
+
+    def read_while_contended(root, rows):
+        entered.set()
+        assert release.wait(timeout=40)
+        return read_receipts(root, rows)
+
+    monkeypatch.setattr(conversation_inventory, "_read_receipt_batch", read_while_contended)
+    marker = _effect_mutation_lock_path(path)
+    with ThreadPoolExecutor(max_workers=3) as pool:
+        reading = pool.submit(_inventory, api, route["session_id"], view="history")
+        try:
+            assert entered.wait(timeout=30) and not reading.done()
+            assert json.loads(marker.read_text(encoding="utf-8"))["pid"] == os.getpid()
+            assert lock_holder_path(path).read_bytes() == previous_holder
+            waits = {name: pool.submit(contender, name) for name in ("kernel_waiter", "marker_waiter")}
+            results = {name: json.loads(wait.result(timeout=30)) for name, wait in waits.items()}
+        finally:
+            release.set()
+        code, recovered = reading.result(timeout=30)
+
+    assert code == 200 and recovered["snapshot_digest"] == idle["snapshot_digest"]
+    assert not marker.exists()
+    assert lock_holder_path(path).read_bytes() == previous_holder
+    incidents = [json.loads(line) for line in lock_incident_path(path).read_text(encoding="utf-8").splitlines()]
+    messages = {operation: result["message"] for operation, result in results.items()}
+    assert all(f"holder pid {historical['pid']}" not in message for message in messages.values()), messages
+    for operation, result in results.items():
+        incident = result["lock_timeout"]
+        action = result["operator_action"]
+        assert incident["waiter"]["operation"] == operation
+        assert incident["waiter"]["waited_seconds"] >= 5.0
+        assert result["incident_recorded"] and incident in incidents
+        assert "current holder unknown" in result["message"]
+        assert incident["holder"] == {}
+        assert incident["last_kernel_holder"] == historical
+        assert action["holder_pid"] is None and action["holder_host"] is None
+        assert "terminating" not in " ".join(action["steps"])
+
+
 def test_original_history_pins_code_revision_per_http_read_but_rereads_receipts(api, monkeypatch):
     from loopx.control_plane import effect_runtime
 

tests/test_file_lock.py

@@ -17,6 +17,7 @@
     _safe_label,
     exclusive_cross_runtime_file_lock,
     exclusive_file_lock,
+    exclusive_mutation_file_lock,
     fcntl,
     lock_holder_path,
     lock_incident_path,
@@ -208,6 +209,46 @@ def test_stalled_holder_times_out_and_records_independent_incident(tmp_path: Pat
     assert target.with_name(f"{target.name}.lock").exists()
 
 
+def test_marker_timeout_keeps_a_live_kernel_holder_out_of_operator_action(tmp_path: Path) -> None:
+    target = tmp_path / "state.json"
+    process, holder_pid, _ = _start_stalled_holder(target)
+    try:
+        with exclusive_mutation_file_lock(target) as marker:
+            token = json.loads(marker.read_text(encoding="utf-8"))["token"]
+            with pytest.raises(LockAcquireTimeoutError) as raised:
+                with exclusive_mutation_file_lock(target, timeout_seconds=0):
+                    pytest.fail("waiter entered while the mutation marker was owned")
+        payload = raised.value.to_payload()
+        incident = payload["lock_timeout"]
+        assert incident["holder"] == {}
+        assert incident["last_kernel_holder"]["pid"] == holder_pid
+        assert payload["operator_action"]["holder_pid"] is None
+        assert "mutation marker" in payload["operator_action"]["steps"][0]
+        assert "current holder unknown" in str(raised.value)
+        assert token not in json.dumps(payload)
+        assert process.poll() is None
+    finally:
+        _stop(process)
+
+
+def test_kernel_timeout_ignores_an_out_of_range_pid_in_the_record(tmp_path: Path) -> None:
+    target = tmp_path / "state.json"
+    process, _, _ = _start_stalled_holder(target)
+    try:
+        holder_path = lock_holder_path(target)
+        holder = json.loads(holder_path.read_text(encoding="utf-8"))
+        holder["pid"] = 1 << 64
+        holder_path.write_text(json.dumps(holder), encoding="utf-8")
+        with pytest.raises(LockAcquireTimeoutError) as raised:
+            with exclusive_file_lock(target, timeout_seconds=0):
+                pytest.fail("waiter entered while the kernel lock was owned")
+        assert raised.value.incident["holder"] == {}
+        assert raised.value.incident["last_kernel_holder"] == holder
+        assert "current holder unknown" in str(raised.value)
+    finally:
+        _stop(process)
+
+
 @pytest.mark.skipif(os.name == "nt", reason="symlink creation requires privileges")
 def test_lock_timeout_does_not_follow_a_symlinked_incident_file(
     tmp_path: Path,