diff --git a/docs/reference/protocols/file-lock-acquisition-v0.md b/docs/reference/protocols/file-lock-acquisition-v0.md index a0364f7a08..634459b8c6 100644 --- a/docs/reference/protocols/file-lock-acquisition-v0.md +++ b/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, diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py index bbd17dda79..71ab69bf5c 100644 --- a/loopx/capabilities/native_chat/external_conversations.py +++ b/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") diff --git a/loopx/control_plane/goals/deletion_service.py b/loopx/control_plane/goals/deletion_service.py index 327a417676..5fc4464d21 100644 --- a/loopx/control_plane/goals/deletion_service.py +++ b/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}"), } diff --git a/loopx/file_lock.py b/loopx/file_lock.py index fd5838575e..e455f063d9 100644 --- a/loopx/file_lock.py +++ b/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, diff --git a/loopx/paths.py b/loopx/paths.py index 7188752190..61c14a4fae 100644 --- a/loopx/paths.py +++ b/loopx/paths.py @@ -105,7 +105,7 @@ 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) @@ -113,7 +113,8 @@ def _runtime_root_has_machine_state(root: Path) -> bool: 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() diff --git a/loopx/semantics/project_registry_io_manifest_v1.json b/loopx/semantics/project_registry_io_manifest_v1.json index 2c0203cd93..068b1d7fa4 100644 --- a/loopx/semantics/project_registry_io_manifest_v1.json +++ b/loopx/semantics/project_registry_io_manifest_v1.json @@ -2087,7 +2087,7 @@ }, { "site": "loopx/paths.py::.configured_runtime_route::codec_read:load_registry#1", - "line": 267, + "line": 268, "column": 20, "kind": "codec_read", "api": "load_registry", @@ -2095,7 +2095,7 @@ }, { "site": "loopx/paths.py::.effective_runtime_root::codec_read:load_registry#1", - "line": 346, + "line": 347, "column": 16, "kind": "codec_read", "api": "load_registry", diff --git a/tests/test_chat_conversation_bindings.py b/tests/test_chat_conversation_bindings.py index 9cc888ae79..564d082ac4 100644 --- a/tests/test_chat_conversation_bindings.py +++ b/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() diff --git a/tests/test_file_lock.py b/tests/test_file_lock.py index 4fa8ee1b26..3ed242eed3 100644 --- a/tests/test_file_lock.py +++ b/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"