fix(file-lock): keep the Windows holder record out of *.json scans

docs/reference/protocols/file-lock-acquisition-v0.md

@@ -21,7 +21,7 @@ unbounded `LOCK_EX` wait is not part of this contract.
 ## Holder And Incident Records
 
 After acquisition, the holder writes public-safe JSON to the POSIX `*.lock`
-file or atomically overwrites the Windows `*.lock.holder.json` sidecar:
+file or atomically overwrites the Windows `*.lock.holder` sidecar:
 
 - stable hashed `lock_id` (never an absolute target path);
 - PID, agent id, operation, policy, and acquisition time;
@@ -32,6 +32,13 @@ handle from reading the locked byte. POSIX retains the existing single-file
 contract, where advisory metadata and `flock` share `*.lock`. In both cases the
 kernel lock, not the metadata file's existence, is authoritative.
 
+The Windows sidecar persists after release and sits beside the locked state
+file, so its name must not end in `.json`: directory scans that collect
+`*.json` state records would otherwise read lock metadata as state. Earlier
+releases wrote `*.lock.holder.json`. Readers still accept that name, preferring
+the most recently acquired record, and each new holder removes it while it
+owns the kernel lock.
+
 A timeout appends one `file_lock_incident_v0` row to the sibling
 `*.lock.incidents.jsonl` channel. That append uses `O_APPEND` directly and does
 not acquire the blocked lock. The row contains holder and waiter identities,

loopx/capabilities/native_chat/external_conversations.py

@@ -7,6 +7,7 @@
 from __future__ import annotations
 
 from pathlib import Path
+import re
 from datetime import datetime, timedelta, timezone
 import hashlib
 from typing import Any
@@ -15,6 +16,10 @@
 from ...chat_attachments import normalize_chat_image_attachments
 from ...file_lock import exclusive_file_lock
 
+# Request journals are named by their 24-hex request_ref; lock metadata and
+# other siblings in this directory are not requests.
+_REQUEST_JOURNAL_STEM = re.compile(r"[a-f0-9]{24}")
+
 
 def _commission_goal_identity(goal: dict[str, Any]) -> dict[str, str]:
     identity: dict[str, str] = {}
@@ -251,7 +256,8 @@ def _record_steward_ingress(self, row: dict[str, Any], selected: dict[str, Any],
             message=row["message"], source_id=row["request_ref"])
 
     def pending(self) -> list[dict[str, Any]]:
-        return [_read_json(path) for path in sorted(self.root.glob("*.json"))]
+        return [_read_json(path) for path in sorted(self.root.glob("*.json"))
+                if _REQUEST_JOURNAL_STEM.fullmatch(path.stem)]
 
     def read_request(self, request_ref: str) -> dict[str, Any]:
         return _read_json(self.root / f"{request_ref}.json")

loopx/control_plane/goals/deletion_service.py

@@ -21,7 +21,7 @@
 from ...file_lock import (
     EFFECT_MUTATION_LOCK_SUFFIX,
     exclusive_cross_runtime_file_lock,
-    lock_holder_path,
+    lock_holder_paths,
     lock_incident_path,
 )
 from ...history import load_registry
@@ -779,7 +779,7 @@ def _orphan_source_lock_error(path: Path) -> str | None:
     kernel_lock = path.with_name(f"{path.name}.lock")
     lock_artifacts = {
         kernel_lock,
-        lock_holder_path(path),
+        *lock_holder_paths(path),
         lock_incident_path(path),
         Path(f"{path}{EFFECT_MUTATION_LOCK_SUFFIX}"),
     }

loopx/file_lock.py

@@ -151,10 +151,30 @@ def _open_lock_descriptor(path: Path, *, flags: int) -> int:
 def lock_holder_path(path: Path) -> Path:
     lock_path = _lock_path(path)
     if os.name == "nt":
-        return lock_path.with_name(f"{lock_path.name}.holder.json")
+        # Not ``*.json``: the sidecar persists beside state files, and state
+        # directory scans must not read lock metadata as a state record.
+        return lock_path.with_name(f"{lock_path.name}.holder")
     return lock_path
 
 
+def _legacy_lock_holder_path(lock_path: Path) -> Path:
+    return lock_path.with_name(f"{lock_path.name}.holder.json")
+
+
+def lock_holder_paths(path: Path) -> tuple[Path, ...]:
+    """Every holder record name, current first.
+
+    Older Windows releases wrote ``*.lock.holder.json``. Readers still accept
+    it because an older process may hold the lock during an upgrade; a newer
+    holder removes it while it owns the kernel lock.
+    """
+
+    current = lock_holder_path(path)
+    if os.name != "nt":
+        return (current,)
+    return (current, _legacy_lock_holder_path(_lock_path(path)))
+
+
 def lock_incident_path(path: Path) -> Path:
     lock_path = _lock_path(path)
     return lock_path.with_name(f"{lock_path.name}.incidents.jsonl")
@@ -253,6 +273,13 @@ def _persist_holder_record(
         _write_holder_record(lock_file, record)
         return
     _write_holder_sidecar(holder_path, record)
+    try:
+        # Only a kernel-lock owner writes either name, so removing the legacy
+        # record here cannot race its writer. A failure leaves it for the next
+        # holder; readers prefer the most recently acquired record.
+        _legacy_lock_holder_path(lock_path).unlink(missing_ok=True)
+    except OSError:
+        pass
 
 
 def _mark_released(
@@ -297,12 +324,31 @@ def _filter_holder_record(payload: object) -> dict[str, object]:
     return {key: payload[key] for key in _HOLDER_RECORD_FIELDS if key in payload}
 
 
-def _read_holder_record(lock_path: Path) -> dict[str, object]:
-    try:
-        payload = json.loads(lock_path.read_text(encoding="utf-8"))
-    except (OSError, json.JSONDecodeError):
-        return {}
-    return _filter_holder_record(payload)
+def _read_latest_holder_record(path: Path) -> tuple[dict[str, object], bool]:
+    """Return the most recently acquired holder record and whether any present
+    record was unreadable. ``({}, False)`` means no record exists."""
+
+    records: list[dict[str, object]] = []
+    unreadable = False
+    for holder_path in lock_holder_paths(path):
+        try:
+            text = holder_path.read_text(encoding="utf-8")
+        except FileNotFoundError:
+            continue
+        except OSError:
+            unreadable = True
+            continue
+        try:
+            record = _filter_holder_record(json.loads(text))
+        except ValueError:
+            record = {}
+        if record:
+            records.append(record)
+        else:
+            unreadable = True
+    if not records:
+        return {}, unreadable
+    return max(records, key=lambda record: str(record.get("acquired_at") or "")), unreadable
 
 
 def lock_holder_host_label() -> str:
@@ -344,19 +390,9 @@ def lock_holder_liveness(path: Path) -> tuple[str, dict[str, object]]:
     overwrites the record.
     """
 
-    holder_path = lock_holder_path(path)
-    try:
-        text = holder_path.read_text(encoding="utf-8")
-    except FileNotFoundError:
-        return LOCK_HOLDER_ABSENT, {}
-    except OSError:
-        return LOCK_HOLDER_UNREADABLE, {}
-    try:
-        record = _filter_holder_record(json.loads(text))
-    except ValueError:
-        return LOCK_HOLDER_UNREADABLE, {}
+    record, unreadable = _read_latest_holder_record(path)
     if not record:
-        return LOCK_HOLDER_UNREADABLE, {}
+        return (LOCK_HOLDER_UNREADABLE if unreadable else LOCK_HOLDER_ABSENT), {}
     released_at = record.get("released_at")
     if isinstance(released_at, str) and released_at:
         return LOCK_HOLDER_RELEASED, record
@@ -454,7 +490,7 @@ def _timeout_error(
     agent_id: str | None,
     operation: str | None,
 ) -> LockAcquireTimeoutError:
-    holder = _read_holder_record(lock_holder_path(path))
+    holder, _ = _read_latest_holder_record(path)
     waiter = {
         **_identity(agent_id=agent_id, operation=operation, policy=policy),
         "started_at": started_at,

loopx/paths.py

@@ -105,15 +105,16 @@ def _runtime_root_has_machine_state(root: Path) -> bool:
     project_owned: set[str] = set()
     if root == DEFAULT_RUNTIME_ROOT and project_registry.is_file() and not _is_redirected_path(project_registry):
         from .control_plane.projects.registry_codec import load_registry
-        from .file_lock import _lock_path, lock_holder_path, lock_incident_path
+        from .file_lock import _lock_path, lock_holder_paths, lock_incident_path
 
         try:
             registry = load_registry(project_registry)
         except (OSError, ValueError):
             return True  # Unclassified existing state requires an explicit route.
         project_owned = {
             project_registry.name, _lock_path(project_registry).name,
-            lock_holder_path(project_registry).name, lock_incident_path(project_registry).name,
+            lock_incident_path(project_registry).name,
+            *(holder.name for holder in lock_holder_paths(project_registry)),
         }
         goal_root = root / DEFAULT_PROJECT_GOALS.name
         declared_dirs = set()

loopx/semantics/project_registry_io_manifest_v1.json

@@ -2087,15 +2087,15 @@
     },
     {
       "site": "loopx/paths.py::<module>.configured_runtime_route::codec_read:load_registry#1",
-      "line": 267,
+      "line": 268,
       "column": 20,
       "kind": "codec_read",
       "api": "load_registry",
       "classification": "codec_api"
     },
     {
       "site": "loopx/paths.py::<module>.effective_runtime_root::codec_read:load_registry#1",
-      "line": 346,
+      "line": 347,
       "column": 16,
       "kind": "codec_read",
       "api": "load_registry",

tests/test_chat_conversation_bindings.py

@@ -383,3 +383,28 @@ def test_missing_origin_in_legacy_request_retains_lark_identity(ordinary):  # no
         assert store.list_sessions() == []
     finally:
         runtime.close()
+
+
+def test_external_request_listing_ignores_lock_holder_records(ordinary):  # noqa: F811
+    from loopx.capabilities.native_chat.external_conversations import ChatExternalConversations
+    store, runtime, contexts, _, _, _, _ = ordinary
+    observation = {"transport_ref": "lock-holder-app", "provider_ref": "c" * 24,
+                   "operator_ref": "d" * 24, "verified": True}
+    bindings = ChatConversationBindings(root=store.root, project_contexts=contexts, observe=lambda _: observation)
+    contexts.conversation_bindings = bindings
+    binding = bindings.configure(transport_ref="lock-holder-app", project_ref=contexts.available()[0]["project_ref"],
+                                 executor_endpoint_id="codex")
+    core = ChatExternalConversations(runtime)
+    args = {"binding_id": binding["binding_id"], "source": {"source_ref": "a" * 24,
+            "sender_ref": observation["operator_ref"], "private_human_message": True},
+            "request_ref": "e" * 24, "message": "/help", "command": "help"}
+    try:
+        first = core.admit(**args)
+        # Older Windows releases left persistent lock metadata named like a journal.
+        (core.root / (args["request_ref"] + ".json.lock.holder.json")).write_text(
+            json.dumps({"schema_version": "file_lock_holder_v0", "pid": 1}))
+        second = core.admit(**{**args, "request_ref": "f" * 24})
+        assert second["status"] == first["status"]
+        assert [row["request_ref"] for row in core.pending()] == ["e" * 24, "f" * 24]
+    finally:
+        runtime.close()

tests/test_file_lock.py

@@ -487,3 +487,115 @@ def test_long_lock_paths_exclude_other_process_and_release_while_holder_alive(tm
         assert not file_lock._effect_mutation_lock_path(target).exists()
     finally:
         _stop(holder)
+
+
+class _WindowsNamedOs:
+    """The real ``os`` module reporting Windows, so any CI host can exercise
+    the Windows holder layout. Kernel locking still uses the host backend."""
+
+    name = "nt"
+
+    def __getattr__(self, attribute: str) -> object:
+        return getattr(os, attribute)
+
+
+@pytest.fixture
+def windows_holder_layout(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(file_lock, "os", _WindowsNamedOs())
+    monkeypatch.setattr(
+        file_lock, "_windows_process_is_alive", lambda pid: pid == os.getpid()
+    )
+
+
+def _holder_fields(**fields: object) -> dict[str, object]:
+    return {
+        "schema_version": file_lock.LOCK_HOLDER_SCHEMA_VERSION,
+        "lock_id": "0" * 16,
+        "policy": "mutation",
+        "host": file_lock.lock_holder_host_label(),
+        "pid": os.getpid(),
+        "agent_id": "agent-a",
+        "operation": "todo-update",
+        **fields,
+    }
+
+
+def test_posix_holder_liveness_reads_the_lock_file_record(tmp_path: Path) -> None:
+    if os.name == "nt":
+        pytest.skip("POSIX keeps the holder record inside the lock file")
+    target = tmp_path / "state.json"
+    assert file_lock.lock_holder_paths(target) == (lock_holder_path(target),)
+    assert file_lock.lock_holder_liveness(target) == (file_lock.LOCK_HOLDER_ABSENT, {})
+
+    with exclusive_file_lock(target, agent_id="agent-a"):
+        state, record = file_lock.lock_holder_liveness(target)
+        assert state == file_lock.LOCK_HOLDER_LIVE
+        assert record["agent_id"] == "agent-a"
+    assert file_lock.lock_holder_liveness(target)[0] == file_lock.LOCK_HOLDER_RELEASED
+
+    lock_holder_path(target).write_text("not json", encoding="utf-8")
+    assert file_lock.lock_holder_liveness(target) == (
+        file_lock.LOCK_HOLDER_UNREADABLE,
+        {},
+    )
+
+
+def test_windows_holder_sidecar_is_not_a_json_state_file(
+    tmp_path: Path, windows_holder_layout: None
+) -> None:
+    target = tmp_path / "state.json"
+    target.write_text("{}", encoding="utf-8")
+
+    with exclusive_file_lock(target, agent_id="agent-a"):
+        assert lock_holder_path(target).name == "state.json.lock.holder"
+        assert sorted(path.name for path in tmp_path.glob("*.json")) == ["state.json"]
+        assert file_lock.lock_holder_liveness(target)[0] == file_lock.LOCK_HOLDER_LIVE
+
+    # The released record persists, and still must not look like state.
+    assert lock_holder_path(target).exists()
+    assert sorted(path.name for path in tmp_path.glob("*.json")) == ["state.json"]
+    assert file_lock.lock_holder_liveness(target)[0] == file_lock.LOCK_HOLDER_RELEASED
+
+
+def test_windows_holder_removes_a_legacy_json_sidecar_on_acquire(
+    tmp_path: Path, windows_holder_layout: None
+) -> None:
+    target = tmp_path / "state.json"
+    legacy = tmp_path / "state.json.lock.holder.json"
+    legacy.write_text(
+        json.dumps(_holder_fields(acquired_at="2026-01-01T00:00:00Z")),
+        encoding="utf-8",
+    )
+    # An older release holding the lock stays visible until a new holder runs.
+    assert file_lock.lock_holder_liveness(target)[0] == file_lock.LOCK_HOLDER_LIVE
+
+    with try_exclusive_file_lock(target) as acquired:
+        assert acquired is not None
+        assert not legacy.exists()
+    assert list(tmp_path.glob("*.json")) == []
+    assert file_lock.lock_holder_liveness(target)[0] == file_lock.LOCK_HOLDER_RELEASED
+
+
+def test_windows_holder_reader_prefers_the_latest_acquisition(
+    tmp_path: Path, windows_holder_layout: None
+) -> None:
+    # Upgrade window: a newer release left a released record, then an older
+    # process acquired the same lock and wrote the legacy name.
+    target = tmp_path / "state.json"
+    (tmp_path / "state.json.lock.holder").write_text(
+        json.dumps(
+            _holder_fields(
+                acquired_at="2026-01-01T00:00:00Z",
+                released_at="2026-01-01T00:00:01Z",
+            )
+        ),
+        encoding="utf-8",
+    )
+    (tmp_path / "state.json.lock.holder.json").write_text(
+        json.dumps(_holder_fields(acquired_at="2026-01-01T00:00:02Z")),
+        encoding="utf-8",
+    )
+
+    state, record = file_lock.lock_holder_liveness(target)
+    assert state == file_lock.LOCK_HOLDER_LIVE
+    assert record["acquired_at"] == "2026-01-01T00:00:02Z"