fix(file-lock): name the Windows holder sidecar without a .json suffix

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,11 @@ 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 name carries no `.json` suffix, so a `*.json` state store
+never reads lock metadata as a record (issue #6128). Acquisition also deletes
+any `<lock>.holder.json` sibling left by a release before that rename, while the
+lock is held, so an upgraded runtime root stops presenting the old artifact.
+
 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

@@ -9,12 +9,18 @@
 from pathlib import Path
 from datetime import datetime, timedelta, timezone
 import hashlib
+import re
 from typing import Any
 
 from ...chat_store import _atomic_write_json, _read_json
 from ...chat_attachments import normalize_chat_image_attachments
 from ...file_lock import exclusive_file_lock
 
+# An external request reference is a 24-hex digest: it names the journal file
+# and enters the lock path, so a record walker accepts only that shape and never
+# a sibling lock artifact such as the holder sidecar (issue #6128).
+_REQUEST_REF_PATTERN = re.compile(r"[a-f0-9]{24}")
+
 
 def _commission_goal_identity(goal: dict[str, Any]) -> dict[str, str]:
     identity: dict[str, str] = {}
@@ -37,18 +43,17 @@ def admit(self, *, binding_id: str, source: dict[str, Any], request_ref: str,
               message: str, command: str | None = None,
               attachments: list[dict[str, Any]] | None = None,
               origin: str = "lark") -> dict[str, Any]:
-        import re
         if origin not in {"lark", "external"}:
             raise ValueError("unsupported external conversation origin")
-        if not re.fullmatch(r"[a-f0-9]{24}", request_ref):
+        if not _REQUEST_REF_PATTERN.fullmatch(request_ref):
             raise ValueError("invalid external request reference")
         if command not in {None, "agents", "select_agent", "select_project", "status", "help", "new", "stop", "unsupported", "commission", "confirm_commission", "cancel_commission", "stop_commission", "resume_commission"}:
             raise ValueError("unsupported external conversation command")
         # These references enter the lock path before authority is resolved.
         # Check path-safe shape here; the typed owner checks current grants
         # and provider identity after both fences are acquired.
         for ref in (binding_id, source.get("source_ref")):
-            if not isinstance(ref, str) or not re.fullmatch(r"[a-f0-9]{24}", ref):
+            if not isinstance(ref, str) or not _REQUEST_REF_PATTERN.fullmatch(ref):
                 raise ValueError("invalid external conversation source reference")
         path = self.root / f"{request_ref}.json"
         with exclusive_file_lock(self.root / "source-fences" / f"{binding_id}.{source['source_ref']}.json", operation="route_external_chat_request"), exclusive_file_lock(path, operation="admit_external_chat_request"):
@@ -157,7 +162,6 @@ def _admit_prepared(self, path: Path, row: dict[str, Any], selected: dict[str, A
                 except (ValueError, OSError, KeyError):
                     continue
             if row["command"] == "select_agent":
-                import re
                 match = re.fullmatch(r"/agent ([a-f0-9]{24})", row["message"].strip())
                 chosen = next((item for item in choices if match and item["target_ref"] == match[1]), None)
                 if chosen is None:
@@ -251,7 +255,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_REF_PATTERN.fullmatch(path.stem)]
 
     def read_request(self, request_ref: str) -> dict[str, Any]:
         return _read_json(self.root / f"{request_ref}.json")

loopx/extensions/lark/private_conversations.py

@@ -32,6 +32,11 @@
 # Reserve edits for a final/closing notice; continue long Turns on a new draft.
 PROGRESS_EDIT_BUDGET = 12
 
+# A private delivery is keyed by its 24-hex request reference, which also names
+# the journal file and its lock. A record walker accepts only that shape and
+# never a sibling lock artifact such as the holder sidecar (issue #6128).
+_REQUEST_REF_PATTERN = re.compile(r"[a-f0-9]{24}")
+
 
 class LarkPrivateConversations:
     def __init__(self, *, controller: Any, runtime_root: Path, runner: Any, cli_bin: str,
@@ -76,6 +81,8 @@ def profiles(self) -> dict[str, dict[str, str]]:
     def health(self) -> dict[str, dict[str, int]]:
         health: dict[str, dict[str, int]] = {}
         for path in self.root.glob("*.json"):
+            if not _REQUEST_REF_PATTERN.fullmatch(path.stem):
+                continue
             row = _read_json(path)
             entry = health.setdefault(row["binding_id"], {"pending_count": 0, "recovery_count": 0})
             if row["status"] != "delivered":
@@ -316,7 +323,7 @@ def return_inbox(self, *, route: dict[str, Any], session: dict[str, Any],
         selected = self.bindings.session_context(saved)
         client = str(turn.get("client_turn_id") or "")
         request = client.removeprefix("external-")
-        if client != f"external-{request}" or not re.fullmatch(r"[a-f0-9]{24}", request):
+        if client != f"external-{request}" or not _REQUEST_REF_PATTERN.fullmatch(request):
             raise ValueError("original private request unavailable")
         record = _read_json(self.root / f"{request}.json")
         native = self.core.read_request(request)
@@ -571,7 +578,8 @@ def _progress(self, path: Path, record: dict[str, Any], *, inbox: Callable[[], P
     def pending_delivery_paths(self) -> list[Path]:
         """The durable transport store is the queue; no in-memory admission."""
         return [path for path in sorted(self.root.glob("*.json"))
-                if _read_json(path)["status"] != "delivered"]
+                if _REQUEST_REF_PATTERN.fullmatch(path.stem)
+                and _read_json(path)["status"] != "delivered"]
 
     def reconcile(self) -> int:
         return sum(self.reconcile_request(path) for path in self.pending_delivery_paths())

loopx/file_lock.py

@@ -151,7 +151,10 @@ 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")
+        # The sidecar is advisory lock metadata, not state. Its name must not
+        # end in .json: a *.json state walker reads every match as a record, so
+        # it would read this holder as one (issue #6128).
+        return lock_path.with_name(f"{lock_path.name}.holder")
     return lock_path
 
 
@@ -242,13 +245,30 @@ def _write_holder_sidecar(holder_path: Path, record: dict[str, object]) -> None:
         temporary_path.unlink(missing_ok=True)
 
 
+def _discard_legacy_holder_sidecar(lock_path: Path) -> None:
+    """Best-effort removal of a pre-rename ``<lock>.holder.json`` sidecar.
+
+    Releases before the holder rename wrote the Windows sidecar with a ``.json``
+    suffix, so a ``*.json`` state store harvested lock metadata and read it as a
+    record (issue #6128). The renamed sidecar carries no suffix and no longer
+    matches, but an upgraded runtime root still holds the old file. Clear that
+    sibling while this lock is held so the fix also covers existing installs.
+    """
+
+    try:
+        lock_path.with_name(f"{lock_path.name}.holder.json").unlink()
+    except OSError:
+        return
+
+
 def _persist_holder_record(
     lock_file: TextIO,
     *,
     lock_path: Path,
     holder_path: Path,
     record: dict[str, object],
 ) -> None:
+    _discard_legacy_holder_sidecar(lock_path)
     if holder_path == lock_path:
         _write_holder_record(lock_file, record)
         return

tests/test_collaboration_sidecar_scan.py

@@ -45,15 +45,21 @@ def test_sidecars(tmp_path_factory):
     operation = next((store / "peer-operations").glob(f"*/{operation_id}.json"))
 
     for record in (entry, operation):
-        sidecar = record.with_name(f"{record.stem}.lock.lock.holder.json")
-        # Windows creates this on every lock acquisition. Simulate the same
-        # persistent artifact on POSIX so both CI platforms guard the scan.
-        if not sidecar.exists():
+        # A Windows lock acquisition writes <record>.lock beside this record and
+        # a <record>.lock.holder sidecar; before the .json rename the sidecar was
+        # <record>.lock.holder.json, which a *.json scan can still harvest on an
+        # upgraded store (issue #6128). Simulate all three on POSIX so both CI
+        # platforms guard the scan.
+        for sidecar in (
+            record.with_name(f"{record.name}.lock"),
+            record.with_name(f"{record.name}.lock.holder"),
+            record.with_name(f"{record.name}.lock.holder.json"),
+        ):
             sidecar.write_text(
                 json.dumps({"schema_version": "file_lock_holder_v0"}),
                 encoding="utf-8",
             )
-        assert sidecar.exists()
+            assert sidecar.exists()
     entry.with_name("foreign.json").write_text("{}", encoding="utf-8")
 
     page = pending(tmp_path, "delivery", "receiver")

tests/test_file_lock.py

@@ -97,6 +97,7 @@ def _acquire_and_release_cross_runtime_lock(
 
 def test_exclusive_lock_persists_public_safe_holder_metadata(tmp_path: Path) -> None:
     target = tmp_path / "state.json"
+    target.write_text("{}\n", encoding="utf-8")
 
     with exclusive_file_lock(
         target,
@@ -113,13 +114,32 @@ def test_exclusive_lock_persists_public_safe_holder_metadata(tmp_path: Path) ->
         assert holder["operation"] == "todo-update"
         assert holder["acquired_at"].endswith("Z")
         assert "released_at" not in holder
+        # The Windows holder sidecar is advisory lock metadata, not a record: a
+        # *.json state store must never harvest it as one (issue #6128).
+        assert sorted(p.name for p in tmp_path.glob("*.json")) == [target.name]
 
     released = json.loads(holder_path.read_text(encoding="utf-8"))
     assert released["released_at"].endswith("Z")
     assert lock_path.exists()
     assert holder_path.exists()
 
 
+def test_acquisition_discards_a_pre_rename_json_holder_sidecar(tmp_path: Path) -> None:
+    target = tmp_path / "state.json"
+    target.write_text("{}\n", encoding="utf-8")
+    # A release before the holder rename left this *.json sibling behind, and a
+    # *.json store then read lock metadata as a request. Acquisition clears it.
+    legacy = tmp_path / "state.json.lock.holder.json"
+    legacy.write_text('{"schema_version": "file_lock_holder_v0"}\n', encoding="utf-8")
+
+    with exclusive_file_lock(target, operation="todo-update") as lock_path:
+        assert lock_path.name == "state.json.lock"
+        assert not legacy.exists()
+
+    assert not legacy.exists()
+    assert sorted(p.name for p in tmp_path.glob("*.json")) == [target.name]
+
+
 @pytest.mark.skipif(os.name == "nt", reason="symlink creation requires privileges")
 def test_exclusive_lock_rejects_a_symlinked_lock_file(tmp_path: Path) -> None:
     target = tmp_path / "state.json"

tests/test_lark_private_conversations.py

@@ -1337,3 +1337,26 @@ def test_scoped_recovery_does_not_probe_unrelated_app_and_revocation_blocks_repl
         assert provider.writes == [] and store.list_sessions() == []
     finally:
         runtime.close()
+
+
+def test_lock_metadata_sidecar_is_never_read_as_a_request(ordinary):  # noqa: F811
+    """A Windows lock sibling is metadata, not a delivery record (#6128)."""
+    _store, runtime, _provider, transport = connect(ordinary)
+    try:
+        reference = "0" * 24
+        record = {"schema_version": "lark_private_chat_delivery_v0", "binding_id": "binding",
+                  "status": "captured", "deliveries": {}}
+        for root in (transport.root, transport.core.root):
+            root.mkdir(parents=True, exist_ok=True)
+            (root / f"{reference}.json").write_text(json.dumps(record), encoding="utf-8")
+            (root / f"{reference}.json.lock").write_text("", encoding="utf-8")
+            # Releases before the holder rename left this *.json lock sibling, so
+            # a *.json store read the holder as a request and raised KeyError.
+            (root / f"{reference}.json.lock.holder.json").write_text(
+                json.dumps({"schema_version": "file_lock_holder_v0"}), encoding="utf-8")
+
+        assert [path.stem for path in transport.pending_delivery_paths()] == [reference]
+        assert transport.health() == {"binding": {"pending_count": 1, "recovery_count": 0}}
+        assert [row["binding_id"] for row in transport.core.pending()] == ["binding"]
+    finally:
+        runtime.close()

tests/test_loopx_turn_driver.py

@@ -2057,7 +2057,7 @@ def interrupt_checkpoint(path, journal):
             for path in (runtime / "goals" / "loopx-turn-fixture" / "turns").glob(
                 "*.json"
             )
-            if not path.name.endswith(".lock.holder.json")
+            if not path.name.endswith(".lock.holder")
         ]
         assert len(journal_paths) == 1
         stored = json.loads(journal_paths[0].read_text(encoding="utf-8"))

tests/test_loopx_turn_executor.py

@@ -530,7 +530,7 @@ def _journal(runtime_root: Path) -> dict[str, object]:
     journal_paths = [
         path
         for path in (runtime_root / "goals" / "fixture-goal" / "turns").glob("*.json")
-        if not path.name.endswith(".lock.holder.json")
+        if not path.name.endswith(".lock.holder")
     ]
     assert len(journal_paths) == 1
     return json.loads(journal_paths[0].read_text(encoding="utf-8"))
@@ -1226,7 +1226,7 @@ def _cadence_starts(runtime_root: Path) -> list[dict[str, object]]:
 
     stores = []
     for path in sorted(runtime_root.rglob("*.json")):
-        if path.name.endswith(".lock.holder.json"):
+        if path.name.endswith(".lock.holder"):
             continue
         payload = json.loads(path.read_text(encoding="utf-8"))
         if (
@@ -1321,7 +1321,7 @@ def die_after_reservation(identity: Mapping[str, object]) -> dict[str, object]:
     assert not [
         path
         for path in (runtime_root / "goals" / "fixture-goal" / "turns").glob("*.json")
-        if not path.name.endswith(".lock.holder.json")
+        if not path.name.endswith(".lock.holder")
     ]
     assert calls == {"host": 0, "writeback": 0, "spend": 0, "scheduler": 0}
 
@@ -1702,7 +1702,7 @@ def host(request: dict[str, object]) -> dict[str, object]:
         for path in (tmp_path / "runtime" / "goals" / "fixture-goal" / "turns").glob(
             "*.json"
         )
-        if not path.name.endswith(".lock.holder.json")
+        if not path.name.endswith(".lock.holder")
     ]
     assert len(journal_paths) == 1
     journal = json.loads(journal_paths[0].read_text(encoding="utf-8"))

tests/test_loopx_turn_journal_inspection.py

@@ -216,7 +216,7 @@ def test_inspection_returns_versioned_allowlisted_projection_without_mutation(
     before_entries = {
         path.relative_to(tmp_path)
         for path in tmp_path.rglob("*")
-        if not path.name.endswith((".lock", ".lock.holder.json"))
+        if not path.name.endswith((".lock", ".lock.holder"))
     }
 
     result = executor.inspect_loopx_turn_journal(
@@ -229,7 +229,7 @@ def test_inspection_returns_versioned_allowlisted_projection_without_mutation(
     after_entries = {
         path.relative_to(tmp_path)
         for path in tmp_path.rglob("*")
-        if not path.name.endswith((".lock", ".lock.holder.json"))
+        if not path.name.endswith((".lock", ".lock.holder"))
     }
     assert result == {
         "ok": True,