From 9514675efed108812c323e2074596ef66b7a0ba8 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 7 Nov 2026 11:37:14 +0800 Subject: [PATCH 1/2] perf(collaboration): avoid rewriting idle request lock holders Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../collaboration/conversation_inventory.py | 4 ++- loopx/control_plane/collaboration/inbox.py | 26 ++++++++++++--- .../test_chat_request_inventory.py | 32 +++++++++++++++++++ 3 files changed, 57 insertions(+), 5 deletions(-) diff --git a/loopx/control_plane/collaboration/conversation_inventory.py b/loopx/control_plane/collaboration/conversation_inventory.py index c399670fac..ed7858c61e 100644 --- a/loopx/control_plane/collaboration/conversation_inventory.py +++ b/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) diff --git a/loopx/control_plane/collaboration/inbox.py b/loopx/control_plane/collaboration/inbox.py index b397ef7dc4..af08c3e531 100644 --- a/loopx/control_plane/collaboration/inbox.py +++ b/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: diff --git a/tests/control_plane/test_chat_request_inventory.py b/tests/control_plane/test_chat_request_inventory.py index d714b7cd9d..3972c92afc 100644 --- a/tests/control_plane/test_chat_request_inventory.py +++ b/tests/control_plane/test_chat_request_inventory.py @@ -39,6 +39,38 @@ 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_kernel_writer(api): + from concurrent.futures import ThreadPoolExecutor + import time + + 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() + + def test_original_history_pins_code_revision_per_http_read_but_rereads_receipts(api, monkeypatch): from loopx.control_plane import effect_runtime From a5e837a30e174128b8ec34aa1de820013bb9ad63 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sun, 8 Nov 2026 00:33:36 +0800 Subject: [PATCH 2/2] fix(collaboration): separate lock contention from historical holders Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/file_lock.py | 45 +++++-- .../test_chat_request_inventory.py | 110 +++++++++++++++++- tests/test_file_lock.py | 41 +++++++ 3 files changed, 185 insertions(+), 11 deletions(-) diff --git a/loopx/file_lock.py b/loopx/file_lock.py index 8b021c0bfb..7207bb8f4f 100644 --- a/loopx/file_lock.py +++ b/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,6 +605,7 @@ 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, @@ -604,24 +613,38 @@ def _timeout_error( 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,7 +801,7 @@ 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 @@ -785,7 +809,7 @@ def process_is_alive(pid: object) -> bool: 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, diff --git a/tests/control_plane/test_chat_request_inventory.py b/tests/control_plane/test_chat_request_inventory.py index 839a75d643..64dab2df6f 100644 --- a/tests/control_plane/test_chat_request_inventory.py +++ b/tests/control_plane/test_chat_request_inventory.py @@ -39,10 +39,12 @@ 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_kernel_writer(api): +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) @@ -70,6 +72,112 @@ def test_inventory_preserves_idle_holder_and_waits_for_existing_kernel_writer(ap 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 diff --git a/tests/test_file_lock.py b/tests/test_file_lock.py index 31d575252c..daf233deae 100644 --- a/tests/test_file_lock.py +++ b/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,