Merge pull request #4812 from loopx-project/codex/turn-lane-holder-host-20260920

fix(turn-lane): name the holding machine beside the holding pid

Author: huangruiteng

loopx/control_plane/turn_driver/lane_fence.py

@@ -33,8 +33,10 @@
 TURN_LANE_DIR_NAME = ".lanes"
 TURN_LANE_UNATTRIBUTED_AGENT = "unattributed"
 # Public-safe holder fields only: the lock record also carries a lock id, a
-# policy name, and the private lock path, which never leave this process.
-TURN_LANE_HOLDER_TEXT_FIELDS = ("agent_id", "operation", "acquired_at")
+# policy name, and the private lock path, which never leave this process. The
+# host is projected because two hosts can share one runtime root: a refusal on
+# the second host must not print a pid that cannot exist there.
+TURN_LANE_HOLDER_TEXT_FIELDS = ("agent_id", "operation", "acquired_at", "host")
 # A refusal taken here stops before the journal, the host, and quota, so the
 # payload reports the same effect shape an executing Turn does -- all false.
 TURN_LANE_NO_EFFECTS: dict[str, bool] = {
@@ -104,9 +106,11 @@ def turn_lane_singleflight(
 def turn_lane_holder_readback(target: Path) -> dict[str, Any]:
     """Return the public-safe identity of the Turn holding one lane, else ``{}``.
 
-    Only names, a timestamp, and a process id are projected: the holder record's
-    private lock path and lock id stay out, so a refusal can say who is running
-    without publishing where this machine keeps its runtime state.
+    Only names, a timestamp, a machine name and a process id are projected: the
+    holder record's private lock path and lock id stay out, so a refusal can say
+    who is running where without publishing where a machine keeps its runtime
+    state. The machine name is what makes the projected pid actionable when two
+    hosts share one runtime root.
     """
 
     try:

loopx/file_lock.py

@@ -10,6 +10,7 @@
 import os
 from pathlib import Path
 import re
+import socket
 import tempfile
 import time
 import importlib
@@ -147,6 +148,11 @@ def _identity(
 ) -> dict[str, object]:
     return {
         "pid": os.getpid(),
+        # A pid is only meaningful on the machine that wrote this record. Two
+        # hosts sharing one runtime root can both read the holder, so the record
+        # names its own machine and a reader never has to guess which host a pid
+        # belongs to. The name is a sanitized label, not a path or a secret.
+        "host": _safe_label(socket.gethostname(), fallback="unknown"),
         "agent_id": _safe_label(
             agent_id or os.environ.get("LOOPX_AGENT_ID"),
             fallback="unknown",
@@ -250,6 +256,7 @@ def _read_holder_record(lock_path: Path) -> dict[str, object]:
         "schema_version",
         "lock_id",
         "policy",
+        "host",
         "pid",
         "agent_id",
         "operation",
@@ -263,10 +270,13 @@ def _operator_action(holder: dict[str, object], *, retry_mode: str) -> dict[str,
     return {
         "required": True,
         "action": "inspect_lock_holder",
+        # The pid is only meaningful on this host: naming it stops an operator
+        # from hunting for a process id that cannot exist on another machine.
+        "holder_host": holder.get("host"),
         "holder_pid": holder.get("pid"),
         "retry_mode": retry_mode,
         "steps": [
-            "Inspect the recorded holder PID and operation.",
+            "Inspect the recorded holder host, PID and operation on that host.",
             "Confirm the process 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.",

tests/test_file_lock.py

@@ -3,6 +3,7 @@
 import json
 import os
 from pathlib import Path
+import socket
 import subprocess
 import sys
 from types import SimpleNamespace
@@ -13,6 +14,7 @@
 from loopx.file_lock import (
     LOCK_ACQUIRE_TIMEOUT_ERROR_CODE,
     LockAcquireTimeoutError,
+    _safe_label,
     exclusive_cross_runtime_file_lock,
     exclusive_file_lock,
     fcntl,
@@ -91,6 +93,9 @@ def test_exclusive_lock_persists_public_safe_holder_metadata(tmp_path: Path) ->
         holder_path = lock_holder_path(target)
         holder = json.loads(holder_path.read_text(encoding="utf-8"))
         assert holder["pid"] > 0
+        # The record names its own machine, so a reader on another host that
+        # shares this runtime root never treats the pid as local.
+        assert holder["host"] == _safe_label(socket.gethostname(), fallback="unknown")
         assert holder["agent_id"] == "agent-a"
         assert holder["operation"] == "todo-update"
         assert holder["acquired_at"].endswith("Z")
@@ -122,9 +127,18 @@ def test_stalled_holder_times_out_and_records_independent_incident(tmp_path: Pat
         assert payload["incident_recorded"] is True
         incident = payload["lock_timeout"]
         assert incident["holder"]["pid"] == process.pid
+        assert incident["holder"]["host"] == _safe_label(
+            socket.gethostname(), fallback="unknown"
+        )
         assert incident["holder"]["agent_id"] == "holder-agent"
         assert incident["waiter"]["agent_id"] == "waiter-agent"
         assert incident["waiter"]["waited_seconds"] >= 0.1
+        # The refused waiter learns which machine holds the pid, so the operator
+        # looks for it on the right host instead of assuming it is local.
+        assert (
+            incident["operator_action"]["holder_host"]
+            == incident["holder"]["host"]
+        )
         assert incident["operator_action"]["retry_mode"] == (
             "manual_after_holder_inspection"
         )

tests/test_turn_lane_fence.py

@@ -8,9 +8,11 @@
 
 from __future__ import annotations
 
+import socket
 from pathlib import Path
 
 from loopx.control_plane.turn_driver.executor import run_loopx_turn_once
+from loopx.file_lock import _safe_label
 from loopx.control_plane.turn_driver.lane_fence import (
     REMEDY_WAIT_FOR_IN_FLIGHT_TURN,
     TURN_LANE_IN_FLIGHT,
@@ -119,7 +121,10 @@ def test_the_holder_readback_stays_public_safe(tmp_path: Path) -> None:
     assert holder["agent_id"] == AGENT_ID
     assert holder["operation"] == TURN_LANE_OPERATION
     assert isinstance(holder["pid"], int)
-    assert set(holder) == {"agent_id", "operation", "pid", "acquired_at"}
+    # The machine name travels with the pid: two hosts can share one runtime
+    # root, and a pid without its host is not an actionable identity.
+    assert holder["host"] == _safe_label(socket.gethostname(), fallback="unknown")
+    assert set(holder) == {"agent_id", "operation", "pid", "acquired_at", "host"}
     # The private lock identity and the runtime path never leave the process.
     assert str(tmp_path) not in str(holder)
     assert turn_lane_holder_readback(tmp_path / "absent.lane") == {}