diff --git a/loopx/chat_goal_lifecycle_actions.py b/loopx/chat_goal_lifecycle_actions.py index 2560ade547..d4df549ebc 100644 --- a/loopx/chat_goal_lifecycle_actions.py +++ b/loopx/chat_goal_lifecycle_actions.py @@ -21,7 +21,28 @@ def _goal_lifecycle_preview( ) -> dict[str, Any]: operation = str(parameters["operation"]) if operation == "delete": - return {"state_fingerprint": self._registry_fingerprint()} + preview = delete_stopped_goal( + registry_path=self.registry_path, + goal_id=str(parameters["goal_id"]), + execute=False, + ) + fingerprint = str(preview.get("observed_state_fingerprint") or "") + source_basis = preview.get("source_basis") + if ( + not preview.get("ok") + or not fingerprint + or not isinstance(source_basis, dict) + ): + raise ValueError( + str( + preview.get("error") + or "Goal deletion source basis is unavailable" + ) + ) + return { + "state_fingerprint": fingerprint, + "source_basis": source_basis, + } target_state = ( GoalActivationState.STOPPED if operation == "stop" @@ -95,6 +116,11 @@ def _apply_goal_delete( goal_id=goal_id, execute=True, expected_state_fingerprint=expected_fingerprint, + expected_source_basis=( + proposal.get("canonical_update_basis") + if isinstance(proposal.get("canonical_update_basis"), dict) + else None + ), ) if result.get("stale"): stale = self.store.apply( @@ -105,10 +131,15 @@ def _apply_goal_delete( receipt={}, ) return {"proposal": stale, "turn": None} - if not result.get("ok") or not (result.get("readback") or {}).get("verified"): - raise ValueError( - str(result.get("error") or "Goal deletion did not verify") + if not result.get("ok"): + error = ValueError( + str(result.get("error") or "Goal deletion did not complete") ) + if not result.get("written") and not result.get("partial_write"): + return self._goal_delete_failed(proposal_id, error) + raise error + if not (result.get("readback") or {}).get("verified"): + raise ValueError("Goal deletion did not verify") receipt = { "receipt_id": _digest( { @@ -128,6 +159,19 @@ def _apply_goal_delete( ) return {"proposal": stored, "turn": None} + def _goal_delete_failed( + self, + proposal_id: str, + error: OSError | ValueError, + ) -> dict[str, Any]: + failed = self.store.mark_failed( + proposal_id, + error_code="goal_delete_unavailable", + message="Goal deletion could not safely acquire or update its registries.", + details={"exception_type": type(error).__name__}, + ) + return {"proposal": failed, "turn": None} + def _apply_goal_lifecycle( self, proposal_id: str, proposal: dict[str, Any], parameters: dict[str, Any] ) -> dict[str, Any]: @@ -136,7 +180,42 @@ def _apply_goal_lifecycle( goal_id = str(parameters["goal_id"]) operation = str(parameters["operation"]) if operation == "delete": - current_fingerprint = self._registry_fingerprint() + expected_fingerprint = str( + proposal.get("expected_state_fingerprint") or "" + ) + expected_source_basis = ( + proposal.get("canonical_update_basis") + if isinstance(proposal.get("canonical_update_basis"), dict) + else None + ) + try: + current = delete_stopped_goal( + registry_path=self.registry_path, + goal_id=goal_id, + execute=False, + expected_state_fingerprint=expected_fingerprint, + expected_source_basis=expected_source_basis, + ) + except (OSError, ValueError) as exc: + return self._goal_delete_failed(proposal_id, exc) + current_fingerprint = str(current.get("observed_state_fingerprint") or "") + if not current.get("ok") or not current_fingerprint: + if current.get("stale") and current_fingerprint: + stale = self.store.apply( + proposal_id, + current_state_fingerprint=current_fingerprint, + receipt={}, + ) + return {"proposal": stale, "turn": None} + return self._goal_delete_failed( + proposal_id, + ValueError( + str( + current.get("error") + or "Goal deletion source basis is unavailable" + ) + ), + ) return self._apply_goal_delete( proposal_id, proposal, goal_id, current_fingerprint ) diff --git a/loopx/control_plane/goals/deletion_service.py b/loopx/control_plane/goals/deletion_service.py index ff2e1a6bf6..6ca270d547 100644 --- a/loopx/control_plane/goals/deletion_service.py +++ b/loopx/control_plane/goals/deletion_service.py @@ -2,6 +2,7 @@ from __future__ import annotations +from collections.abc import Mapping from contextlib import ExitStack import hashlib import os @@ -15,7 +16,14 @@ ProjectRegistryTransaction, project_registry_transaction, ) -from ...file_lock import exclusive_file_lock +from ...configuration_transaction import configuration_payload_revision +from ...file_lock import ( + EFFECT_MUTATION_LOCK_SUFFIX, + exclusive_cross_runtime_file_lock, + exclusive_file_lock, + lock_holder_path, + lock_incident_path, +) from ...history import load_registry from ...registry import atomic_write_json, read_json from ...registry_writability import probe_registry_write_path @@ -23,14 +31,21 @@ from .activation import GoalActivationState, goal_activation_state from .activation_service import ( GoalActivationAuthorityRouteMode, + _goal_activation_source_identity, _goal_or_none, _same_path, _source_and_target, + _source_status, ) GOAL_DELETION_SCHEMA_VERSION = "loopx_goal_deletion_v1" +GOAL_DELETION_SOURCE_BASIS_SCHEMA_VERSION = "loopx_goal_deletion_source_basis_v1" +GOAL_DELETION_STATE_FINGERPRINT_SCHEMA_VERSION = ( + "loopx_goal_deletion_state_fingerprint_v1" +) _OPAQUE_ID = re.compile(r"^[A-Za-z0-9._:-]{1,200}$") +_SHA256 = re.compile(r"^[a-f0-9]{64}$") def _require_opaque_id(value: str, *, field: str) -> str: @@ -46,6 +61,140 @@ def _registry_fingerprint(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() +def _registry_identity(path: Path) -> str: + return configuration_payload_revision( + {"registry_path": str(path.expanduser().resolve())} + ).removeprefix("sha256:") + + +def _source_basis(route: dict[str, Any]) -> dict[str, str]: + source_registry = Path(route["declared_source_registry"]) + try: + source_content = source_registry.read_bytes() + except OSError: + source_content = ( + f"unavailable:{route['source_status']}:{_registry_identity(source_registry)}" + ).encode() + return { + "schema_version": GOAL_DELETION_SOURCE_BASIS_SCHEMA_VERSION, + "source_identity": _goal_activation_source_identity(source_registry), + "source_content_sha256": hashlib.sha256(source_content).hexdigest(), + "route_mode": str(route["route_mode"]), + } + + +def _state_fingerprint( + *, + goal_id: str, + source_basis: Mapping[str, str], + target_registry: Path, +) -> str: + return configuration_payload_revision( + { + "schema_version": GOAL_DELETION_STATE_FINGERPRINT_SCHEMA_VERSION, + "goal_id": goal_id, + "source_basis": dict(source_basis), + "target_identity": _registry_identity(target_registry), + "target_content_sha256": _registry_fingerprint(target_registry), + } + ).removeprefix("sha256:") + + +def _normalize_source_basis(value: Mapping[str, Any] | None) -> dict[str, str] | None: + if value is None: + return None + schema_version = str(value.get("schema_version") or "") + source_identity = str(value.get("source_identity") or "") + source_content_sha256 = str(value.get("source_content_sha256") or "") + route_mode = str(value.get("route_mode") or "") + if schema_version != GOAL_DELETION_SOURCE_BASIS_SCHEMA_VERSION: + raise ValueError("expected source basis has an unsupported schema version") + if not _SHA256.fullmatch(source_identity): + raise ValueError("expected source identity must be a SHA-256 digest") + if not _SHA256.fullmatch(source_content_sha256): + raise ValueError("expected source content digest must be a SHA-256 digest") + if route_mode not in {mode.value for mode in GoalActivationAuthorityRouteMode}: + raise ValueError("expected source route mode is unsupported") + return { + "schema_version": schema_version, + "source_identity": source_identity, + "source_content_sha256": source_content_sha256, + "route_mode": route_mode, + } + + +def _mark_stale( + payload: dict[str, Any], + *, + current_state_fingerprint: str, + current_source_basis: Mapping[str, str], +) -> None: + payload.update( + { + "ok": False, + "stale": True, + "error_kind": "goal_registry_changed", + "error": ( + "Goal source or registry route changed after preview; " + "regenerate the deletion preview" + ), + "current_state_fingerprint": current_state_fingerprint, + "observed_state_fingerprint": current_state_fingerprint, + "source_basis": dict(current_source_basis), + } + ) + + +def _missing_state_fingerprint(*, goal_id: str, registry_path: Path) -> str: + target_content_sha256 = ( + _registry_fingerprint(registry_path) if registry_path.is_file() else None + ) + return configuration_payload_revision( + { + "schema_version": GOAL_DELETION_STATE_FINGERPRINT_SCHEMA_VERSION, + "goal_id": goal_id, + "deletion_state": "missing", + "target_identity": _registry_identity(registry_path), + "target_content_sha256": target_content_sha256, + } + ).removeprefix("sha256:") + + +def _missing_stale_payload( + *, + goal_id: str, + registry_path: Path, + execute: bool, + expected_state_fingerprint: str | None, +) -> dict[str, Any]: + current_fingerprint = _missing_state_fingerprint( + goal_id=goal_id, + registry_path=registry_path, + ) + return { + "ok": False, + "schema_version": GOAL_DELETION_SCHEMA_VERSION, + "dry_run": not execute, + "execute": execute, + "goal_id": goal_id, + "target_global_registry": str(registry_path), + "expected_state_fingerprint": expected_state_fingerprint, + "observed_state_fingerprint": current_fingerprint, + "current_state_fingerprint": current_fingerprint, + "written": False, + "partial_write": False, + "backup_paths": [], + "readback": { + "source_missing": True, + "global_missing": True, + "verified": False, + }, + "stale": True, + "error_kind": "goal_registry_changed", + "error": "Goal disappeared after preview; regenerate the deletion preview", + } + + def _backup_path(path: Path, timestamp: str, nonce: str) -> Path: compact_timestamp = timestamp.replace(":", "").replace("-", "") return path.with_name( @@ -99,7 +248,12 @@ def _remove_goal(payload: dict[str, Any], goal_id: str) -> tuple[dict[str, Any], return updated, changed -def _resolve_route(goal_id: str, registry_path: Path) -> dict[str, Any]: +def _resolve_route( + goal_id: str, + registry_path: Path, + *, + require_stopped: bool = True, +) -> dict[str, Any]: route = _source_and_target( registry_path=registry_path, goal_id=goal_id, @@ -114,17 +268,27 @@ def _resolve_route(goal_id: str, registry_path: Path) -> dict[str, Any]: goal = source_goal or target_goal if goal is None: raise ValueError(f"goal id not found in registry: {goal_id}") - if goal_activation_state(goal) is not GoalActivationState.STOPPED: + if require_stopped and goal_activation_state(goal) is not GoalActivationState.STOPPED: raise ValueError("stop the Goal before deleting it") if target_goal is None: raise ValueError("global registry does not contain the Goal projection") + declared_source_registry = route.source_registry + if ( + route.mode is GoalActivationAuthorityRouteMode.ORPHANED_GLOBAL_STOP_FALLBACK + ): + source_ref = str(target_goal.get("source_registry") or "").strip() + if source_ref: + declared_source_registry = Path(source_ref).expanduser().resolve() return { "source_registry": route.source_registry, + "declared_source_registry": declared_source_registry, "target_registry": route.target_registry, "source_available": source_available, + "source_status": route.source_status.value, "source_goal": source_goal, "target_goal": target_goal, "same_registry": _same_path(route.source_registry, route.target_registry), + "route_mode": route.mode.value, } @@ -135,23 +299,136 @@ def _check_writability(paths: list[Path]) -> dict[str, Any] | None: return next((item for item in writability if not item.get("ok")), None) -def _registry_paths(source_registry: Path, target_registry: Path, same_registry: bool) -> list[Path]: - paths = [target_registry] - if not same_registry: - paths.append(source_registry) - return sorted(paths, key=lambda item: str(item)) +def _orphan_source_lock_error(path: Path) -> str | None: + parent = path.parent + if parent.exists() and (not parent.is_dir() or parent.is_symlink()): + return "Goal source registry parent is unavailable for safe locking" + kernel_lock = path.with_name(f"{path.name}.lock") + lock_artifacts = { + kernel_lock, + lock_holder_path(path), + lock_incident_path(path), + Path(f"{path}{EFFECT_MUTATION_LOCK_SUFFIX}"), + } + if any( + artifact.is_symlink() + or (artifact.exists() and not artifact.is_file()) + for artifact in lock_artifacts + ): + return "Goal source registry lock path is unsafe" + return None + + +def _route_matches_locked_paths( + route: Mapping[str, Any], + *, + source_registry: Path, + declared_source_registry: Path, + target_registry: Path, + source_available: bool, + same_registry: bool, +) -> bool: + return ( + _same_path(Path(route["source_registry"]), source_registry) + and _same_path( + Path(route["declared_source_registry"]), + declared_source_registry, + ) + and _same_path(Path(route["target_registry"]), target_registry) + and bool(route["source_available"]) is source_available + and bool(route["same_registry"]) is same_registry + ) + + +def _validated_locked_route_snapshot( + *, + requested_registry: Path, + source_registry: Path, + declared_source_registry: Path, + target_registry: Path, + source_available: bool, + same_registry: bool, + goal_id: str, + expected_state_fingerprint: str | None, + expected_source_basis: Mapping[str, str] | None, + payload: dict[str, Any], +) -> tuple[dict[str, Any], dict[str, str], str] | None: + try: + locked_route, current_source_basis, current_fingerprint = _route_snapshot( + goal_id=goal_id, + requested_registry=requested_registry, + ) + except ValueError as exc: + if str(exc) != f"goal id not found in registry: {goal_id}": + raise + payload.update( + _missing_stale_payload( + goal_id=goal_id, + registry_path=requested_registry, + execute=True, + expected_state_fingerprint=expected_state_fingerprint, + ) + ) + return None + route_changed = not _route_matches_locked_paths( + locked_route, + source_registry=source_registry, + declared_source_registry=declared_source_registry, + target_registry=target_registry, + source_available=source_available, + same_registry=same_registry, + ) + if route_changed or ( + expected_state_fingerprint is not None + and current_fingerprint != expected_state_fingerprint + ) or ( + expected_source_basis is not None + and current_source_basis != expected_source_basis + ): + _mark_stale( + payload, + current_state_fingerprint=current_fingerprint, + current_source_basis=current_source_basis, + ) + return None + return locked_route, current_source_basis, current_fingerprint + + +def _route_snapshot( + *, + goal_id: str, + requested_registry: Path, +) -> tuple[dict[str, Any], dict[str, str], str]: + route = _resolve_route( + goal_id, + requested_registry, + require_stopped=False, + ) + source_basis = _source_basis(route) + return ( + route, + source_basis, + _state_fingerprint( + goal_id=goal_id, + source_basis=source_basis, + target_registry=Path(route["target_registry"]), + ), + ) def _load_locked_payloads( *, + requested_registry: Path, source_registry: Path, + declared_source_registry: Path, target_registry: Path, source_available: bool, same_registry: bool, goal_id: str, expected_state_fingerprint: str | None, + expected_source_basis: Mapping[str, str] | None, payload: dict[str, Any], -) -> dict[Path, dict[str, Any]] | None: +) -> tuple[dict[Path, dict[str, Any]], dict[str, str], str] | None: current_source = ( read_json(source_registry) if source_available and same_registry @@ -160,17 +437,21 @@ def _load_locked_payloads( else None ) current_target = read_json(target_registry) - if expected_state_fingerprint is not None: - current_fingerprint = _registry_fingerprint(target_registry) - if current_fingerprint != expected_state_fingerprint: - payload.update({ - "ok": False, - "stale": True, - "error_kind": "goal_registry_changed", - "error": "Goal registry changed after preview; regenerate the deletion preview", - "current_state_fingerprint": current_fingerprint, - }) - return None + snapshot = _validated_locked_route_snapshot( + requested_registry=requested_registry, + source_registry=source_registry, + declared_source_registry=declared_source_registry, + target_registry=target_registry, + source_available=source_available, + same_registry=same_registry, + goal_id=goal_id, + expected_state_fingerprint=expected_state_fingerprint, + expected_source_basis=expected_source_basis, + payload=payload, + ) + if snapshot is None: + return None + _, current_source_basis, current_fingerprint = snapshot source_goal = _goal_or_none(current_source, goal_id) if current_source else None target_goal = _goal_or_none(current_target, goal_id) @@ -184,7 +465,7 @@ def _load_locked_payloads( if current_source is None or source_goal is None: raise ValueError("Goal source registry changed; refresh and retry") current_payloads[source_registry] = current_source - return current_payloads + return current_payloads, current_source_basis, current_fingerprint def _updated_payloads( @@ -204,8 +485,10 @@ def _write_deletion( current_payloads: dict[Path, dict[str, Any]], updated_payloads: dict[Path, dict[str, Any]], source_registry: Path, + declared_source_registry: Path, target_registry: Path, source_available: bool, + locked_source_basis: Mapping[str, str], goal_id: str, payload: dict[str, Any], source_transaction: ProjectRegistryTransaction | None, @@ -224,20 +507,33 @@ def _write_deletion( source_transaction=source_transaction, ) written_paths.append(path) - source_after = load_registry(source_registry) if source_available else None + source_missing = ( + _goal_or_none(load_registry(source_registry), goal_id) is None + if source_available + else _source_basis( + { + "declared_source_registry": declared_source_registry, + "source_status": _source_status( + declared_source_registry, + goal_id=goal_id, + ).value, + "route_mode": locked_source_basis["route_mode"], + } + ) + == locked_source_basis + ) target_after = read_json(target_registry) - source_missing = not source_after or _goal_or_none(source_after, goal_id) is None global_missing = _goal_or_none(target_after, goal_id) is None + if not source_missing or not global_missing: + raise ValueError("Goal deletion readback did not verify") payload["readback"] = { "source_missing": source_missing, "global_missing": global_missing, "verified": source_missing and global_missing, } payload["written"] = bool(written_paths) - payload["ok"] = bool(payload["readback"]["verified"]) - payload["partial_write"] = bool(payload["written"] and not payload["ok"]) - if not payload["ok"]: - payload["error"] = "Goal deletion readback did not verify" + payload["ok"] = True + payload["partial_write"] = False except Exception: for path in reversed(written_paths): _restore_locked_registry( @@ -277,52 +573,82 @@ def _restore_locked_registry( def _execute_deletion( *, + requested_registry: Path, source_registry: Path, + declared_source_registry: Path, target_registry: Path, source_available: bool, same_registry: bool, goal_id: str, expected_state_fingerprint: str | None, + expected_source_basis: Mapping[str, str] | None, payload: dict[str, Any], ) -> None: """Apply deletion and read it back while registry locks are held.""" - paths = _registry_paths(source_registry, target_registry, same_registry) + locked_source_registry = ( + source_registry if source_available else declared_source_registry + ) with ExitStack() as stack: source_transaction = None - for path in paths: - if ( - source_available - and not same_registry - and _same_path(path, source_registry) - ): + if not _same_path(locked_source_registry, target_registry): + if source_available and not same_registry: source_transaction = stack.enter_context( project_registry_transaction( - path, + locked_source_registry, operation="delete_stopped_goal", ) ) else: stack.enter_context( - exclusive_file_lock(path, operation="delete_stopped_goal") + exclusive_cross_runtime_file_lock( + locked_source_registry, + operation="delete_stopped_goal", + ) ) - current_payloads = _load_locked_payloads( + stack.enter_context( + exclusive_file_lock(target_registry, operation="delete_stopped_goal") + ) + locked_state = _load_locked_payloads( + requested_registry=requested_registry, source_registry=source_registry, + declared_source_registry=declared_source_registry, target_registry=target_registry, source_available=source_available, same_registry=same_registry, goal_id=goal_id, expected_state_fingerprint=expected_state_fingerprint, + expected_source_basis=expected_source_basis, payload=payload, ) - if current_payloads is None: + if locked_state is None: + return + current_payloads, locked_source_basis, locked_state_fingerprint = locked_state + updated_payloads = _updated_payloads(current_payloads, goal_id) + if ( + _validated_locked_route_snapshot( + requested_registry=requested_registry, + source_registry=source_registry, + declared_source_registry=declared_source_registry, + target_registry=target_registry, + source_available=source_available, + same_registry=same_registry, + goal_id=goal_id, + expected_state_fingerprint=locked_state_fingerprint, + expected_source_basis=locked_source_basis, + payload=payload, + ) + is None + ): return _write_deletion( current_payloads=current_payloads, - updated_payloads=_updated_payloads(current_payloads, goal_id), + updated_payloads=updated_payloads, source_registry=source_registry, + declared_source_registry=declared_source_registry, target_registry=target_registry, source_available=source_available, + locked_source_basis=locked_source_basis, goal_id=goal_id, payload=payload, source_transaction=source_transaction, @@ -335,17 +661,48 @@ def delete_stopped_goal( goal_id: str, execute: bool = False, expected_state_fingerprint: str | None = None, + expected_source_basis: Mapping[str, Any] | None = None, ) -> dict[str, Any]: """Preview or permanently remove one stopped Goal from its registries. Goal data such as state files and project files is intentionally retained; deletion removes only the registry entries that make the Goal visible to - the LoopX control plane. When an expected fingerprint is supplied, the - target registry is checked again while both registry locks are held. + the LoopX control plane. Expected state and source values are checked again + while both registry locks are held. """ normalized_goal_id = _require_opaque_id(goal_id, field="goal_id") - route = _resolve_route(normalized_goal_id, Path(registry_path)) + requested_registry = Path(registry_path) + normalized_fingerprint = str(expected_state_fingerprint or "").strip() or None + if normalized_fingerprint is not None and not _SHA256.fullmatch( + normalized_fingerprint + ): + raise ValueError("expected state fingerprint must be a SHA-256 digest") + normalized_source_basis = _normalize_source_basis(expected_source_basis) + try: + route = _resolve_route( + normalized_goal_id, + requested_registry, + require_stopped=False, + ) + except ValueError as exc: + if ( + normalized_fingerprint is None + or str(exc) != f"goal id not found in registry: {normalized_goal_id}" + ): + raise + return _missing_stale_payload( + goal_id=normalized_goal_id, + registry_path=requested_registry, + execute=execute, + expected_state_fingerprint=normalized_fingerprint, + ) + observed_source_basis = _source_basis(route) + observed_fingerprint = _state_fingerprint( + goal_id=normalized_goal_id, + source_basis=observed_source_basis, + target_registry=route["target_registry"], + ) payload: dict[str, Any] = { "ok": True, @@ -357,6 +714,10 @@ def delete_stopped_goal( "target_global_registry": str(route["target_registry"]), "source_registry_present": route["source_goal"] is not None, "global_registry_present": route["target_goal"] is not None, + "source_basis": observed_source_basis, + "authority_route_mode": route["route_mode"], + "expected_state_fingerprint": normalized_fingerprint, + "observed_state_fingerprint": observed_fingerprint, "written": False, "partial_write": False, "backup_paths": [], @@ -366,9 +727,41 @@ def delete_stopped_goal( "verified": False, }, } + if ( + normalized_fingerprint is not None + and observed_fingerprint != normalized_fingerprint + ) or ( + normalized_source_basis is not None + and observed_source_basis != normalized_source_basis + ): + _mark_stale( + payload, + current_state_fingerprint=observed_fingerprint, + current_source_basis=observed_source_basis, + ) + return payload + goal = route["source_goal"] or route["target_goal"] + if goal_activation_state(goal) is not GoalActivationState.STOPPED: + raise ValueError("stop the Goal before deleting it") if not execute: return payload + if not route["source_available"]: + lock_error = _orphan_source_lock_error(route["declared_source_registry"]) + if lock_error is not None: + payload.update( + { + "ok": False, + "error_kind": "goal_source_lock_unavailable", + "error": lock_error, + "recommended_action": ( + "Repair the Goal source registry route before deleting " + "this Goal." + ), + } + ) + return payload + paths = [route["target_registry"]] if not route["same_registry"]: paths.append(route["source_registry"]) @@ -385,12 +778,15 @@ def delete_stopped_goal( return payload _execute_deletion( + requested_registry=requested_registry, source_registry=route["source_registry"], + declared_source_registry=route["declared_source_registry"], target_registry=route["target_registry"], source_available=route["source_available"], same_registry=route["same_registry"], goal_id=normalized_goal_id, - expected_state_fingerprint=expected_state_fingerprint, + expected_state_fingerprint=normalized_fingerprint, + expected_source_basis=normalized_source_basis, payload=payload, ) return payload diff --git a/loopx/file_lock.py b/loopx/file_lock.py index cf27049a3b..4e25c05f96 100644 --- a/loopx/file_lock.py +++ b/loopx/file_lock.py @@ -11,6 +11,7 @@ from pathlib import Path import re import socket +import stat import tempfile import time import importlib @@ -123,6 +124,28 @@ def _lock_path(path: Path) -> Path: return path.with_name(f"{path.name}.lock") +def _open_lock_descriptor(path: Path, *, flags: int) -> int: + no_follow = getattr(os, "O_NOFOLLOW", 0) + if not no_follow and path.is_symlink(): + raise OSError(errno.ELOOP, "lock path must not be a symlink", str(path)) + descriptor = os.open(path, flags | no_follow, 0o600) + try: + descriptor_stat = os.fstat(descriptor) + path_stat = os.lstat(path) + if ( + not stat.S_ISREG(descriptor_stat.st_mode) + or stat.S_ISLNK(path_stat.st_mode) + or descriptor_stat.st_dev != path_stat.st_dev + or descriptor_stat.st_ino != path_stat.st_ino + or getattr(descriptor_stat, "st_nlink", 1) != 1 + ): + raise OSError(errno.EINVAL, "lock path must be a regular file", str(path)) + except BaseException: + os.close(descriptor) + raise + return descriptor + + def lock_holder_path(path: Path) -> Path: lock_path = _lock_path(path) if os.name == "nt": @@ -292,10 +315,9 @@ def _append_incident(path: Path, record: dict[str, object]) -> bool: + "\n" ).encode("utf-8") try: - descriptor = os.open( + descriptor = _open_lock_descriptor( incident_path, - os.O_APPEND | os.O_CREAT | os.O_WRONLY, - 0o600, + flags=os.O_APPEND | os.O_CREAT | os.O_WRONLY, ) try: os.write(descriptor, encoded) @@ -409,7 +431,10 @@ def exclusive_file_lock( lock_path = _lock_path(path) holder_path = lock_holder_path(path) lock_path.parent.mkdir(parents=True, exist_ok=True) - descriptor = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600) + descriptor = _open_lock_descriptor( + lock_path, + flags=os.O_CREAT | os.O_RDWR, + ) with os.fdopen(descriptor, "r+", encoding="utf-8") as lock_file: started = time.monotonic() started_at = _utc_now_iso() @@ -467,7 +492,10 @@ def try_exclusive_file_lock( lock_path = _lock_path(path) holder_path = lock_holder_path(path) lock_path.parent.mkdir(parents=True, exist_ok=True) - descriptor = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600) + descriptor = _open_lock_descriptor( + lock_path, + flags=os.O_CREAT | os.O_RDWR, + ) with os.fdopen(descriptor, "r+", encoding="utf-8") as lock_file: if not _try_acquire_kernel_lock(lock_file): yield None diff --git a/loopx/semantics/project_registry_io_manifest_v1.json b/loopx/semantics/project_registry_io_manifest_v1.json index d536e03aa3..22c3720e34 100644 --- a/loopx/semantics/project_registry_io_manifest_v1.json +++ b/loopx/semantics/project_registry_io_manifest_v1.json @@ -759,7 +759,7 @@ }, { "site": "loopx/cli_commands/todo.py::.handle_todo_command::codec_read:load_registry#4", - "line": 464, + "line": 465, "column": 21, "kind": "codec_read", "api": "load_registry", @@ -767,7 +767,7 @@ }, { "site": "loopx/cli_commands/todo.py::.handle_todo_command::codec_read:load_registry#5", - "line": 647, + "line": 648, "column": 13, "kind": "codec_read", "api": "load_registry", @@ -775,7 +775,7 @@ }, { "site": "loopx/cli_commands/todo.py::.handle_todo_command::codec_read:load_registry#6", - "line": 689, + "line": 690, "column": 38, "kind": "codec_read", "api": "load_registry", @@ -1023,7 +1023,7 @@ }, { "site": "loopx/control_plane/goals/deletion_service.py::._execute_deletion::codec_transaction:project_registry_transaction#1", - "line": 300, + "line": 597, "column": 21, "kind": "codec_transaction", "api": "project_registry_transaction", @@ -1031,7 +1031,7 @@ }, { "site": "loopx/control_plane/goals/deletion_service.py::._load_locked_payloads::direct_json_read:read_json#1", - "line": 156, + "line": 433, "column": 9, "kind": "direct_json_read", "api": "read_json", @@ -1039,7 +1039,7 @@ }, { "site": "loopx/control_plane/goals/deletion_service.py::._load_locked_payloads::codec_read:load_registry#1", - "line": 158, + "line": 435, "column": 14, "kind": "codec_read", "api": "load_registry", @@ -1047,7 +1047,7 @@ }, { "site": "loopx/control_plane/goals/deletion_service.py::._load_locked_payloads::direct_json_read:read_json#2", - "line": 162, + "line": 439, "column": 22, "kind": "direct_json_read", "api": "read_json", @@ -1055,7 +1055,7 @@ }, { "site": "loopx/control_plane/goals/deletion_service.py::._resolve_route::codec_read:load_registry#1", - "line": 110, + "line": 264, "column": 22, "kind": "codec_read", "api": "load_registry", @@ -1063,7 +1063,7 @@ }, { "site": "loopx/control_plane/goals/deletion_service.py::._resolve_route::direct_json_read:read_json#1", - "line": 111, + "line": 265, "column": 22, "kind": "direct_json_read", "api": "read_json", @@ -1071,7 +1071,7 @@ }, { "site": "loopx/control_plane/goals/deletion_service.py::._restore_locked_registry::direct_json_write:atomic_write_json#1", - "line": 275, + "line": 571, "column": 5, "kind": "direct_json_write", "api": "atomic_write_json", @@ -1079,15 +1079,15 @@ }, { "site": "loopx/control_plane/goals/deletion_service.py::._write_deletion::codec_read:load_registry#1", - "line": 227, - "column": 24, + "line": 511, + "column": 27, "kind": "codec_read", "api": "load_registry", "classification": "codec_api" }, { "site": "loopx/control_plane/goals/deletion_service.py::._write_deletion::direct_json_read:read_json#1", - "line": 228, + "line": 525, "column": 24, "kind": "direct_json_read", "api": "read_json", @@ -1095,7 +1095,7 @@ }, { "site": "loopx/control_plane/goals/deletion_service.py::._write_locked_registry::direct_json_write:atomic_write_json#1", - "line": 262, + "line": 558, "column": 5, "kind": "direct_json_write", "api": "atomic_write_json", @@ -1351,7 +1351,7 @@ }, { "site": "loopx/control_plane/todos/event_writeback.py::._registry_goal::codec_read:load_registry#1", - "line": 63, + "line": 53, "column": 16, "kind": "codec_read", "api": "load_registry", @@ -1759,7 +1759,7 @@ }, { "site": "loopx/state_refresh.py::.refresh_state_run::codec_read:load_registry#1", - "line": 899, + "line": 903, "column": 16, "kind": "codec_read", "api": "load_registry", diff --git a/tests/control_plane/test_goal_activation.py b/tests/control_plane/test_goal_activation.py index 3997a09a19..7f316bc010 100644 --- a/tests/control_plane/test_goal_activation.py +++ b/tests/control_plane/test_goal_activation.py @@ -2,7 +2,9 @@ from concurrent.futures import ThreadPoolExecutor from contextlib import contextmanager +import hashlib import json +import os from pathlib import Path import threading @@ -109,6 +111,38 @@ def _orphaned_global_registry( return source_registry, global_registry +def _preview_delete_action( + *, + global_registry: Path, + action_store: Path, + idempotency_key: str, +) -> tuple[ChatActionService, dict[str, object]]: + set_goal_activation_state( + registry_path=global_registry, + goal_id="goal-one", + state="stopped", + actor_kind="owner", + execute=True, + ) + service = ChatActionService( + store=ChatActionStore(action_store), + registry_path=global_registry, + ) + proposal = service.preview( + { + "action_kind": "goal.lifecycle", + "summary": "Delete a stopped Goal", + "normalized_parameters": { + "goal_id": "goal-one", + "operation": "delete", + }, + "context": {"kind": "goal_directory"}, + "idempotency_key": idempotency_key, + } + ) + return service, proposal + + def test_activation_contract_defaults_active_and_rejects_unknown_state() -> None: assert goal_activation_state({}) is GoalActivationState.ACTIVE assert goal_activation_state({"activation_state": "stopped"}) is GoalActivationState.STOPPED @@ -925,9 +959,17 @@ def test_delete_active_goal_fails_closed( ) -def test_delete_orphaned_stopped_global_goal(tmp_path: Path) -> None: +@pytest.mark.parametrize( + "source_status", + ["registry_missing", "registry_unreadable", "goal_missing"], +) +def test_delete_orphaned_stopped_global_goal( + tmp_path: Path, + source_status: str, +) -> None: _source_registry, global_registry = _orphaned_global_registry( tmp_path, + source_status=source_status, activation_state="stopped", ) @@ -942,6 +984,28 @@ def test_delete_orphaned_stopped_global_goal(tmp_path: Path) -> None: assert registry_goals(load_registry(global_registry)) == [] +def test_delete_orphaned_stopped_goal_fails_when_source_parent_is_not_a_directory( + tmp_path: Path, +) -> None: + source_registry, global_registry = _orphaned_global_registry( + tmp_path, + activation_state="stopped", + ) + source_registry.parent.parent.mkdir(parents=True) + source_registry.parent.write_text("not-a-directory\n", encoding="utf-8") + + deleted = delete_stopped_goal( + registry_path=global_registry, + goal_id="orphaned-goal", + execute=True, + ) + + assert deleted["ok"] is False + assert deleted["error_kind"] == "goal_source_lock_unavailable" + assert _goal(global_registry, "orphaned-goal")["id"] == "orphaned-goal" + assert not list(tmp_path.rglob("*.goal-delete-*.bak")) + + def test_owner_confirmed_typed_action_deletes_stopped_goal( connected_registries: tuple[Path, Path], tmp_path: Path, ) -> None: @@ -979,6 +1043,184 @@ def test_owner_confirmed_typed_action_deletes_stopped_goal( assert registry_goals(load_registry(global_registry)) == [] +def test_owner_confirmed_typed_action_rejects_orphan_source_restored_before_write( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + source_registry, global_registry = _orphaned_global_registry( + tmp_path, + activation_state="stopped", + ) + service = ChatActionService( + store=ChatActionStore(tmp_path / "actions"), + registry_path=global_registry, + ) + proposal = service.preview( + { + "action_kind": "goal.lifecycle", + "summary": "Delete an orphaned stopped Goal", + "normalized_parameters": { + "goal_id": "orphaned-goal", + "operation": "delete", + }, + "context": {"kind": "goal_directory"}, + "idempotency_key": "delete-orphaned-goal-before-source-restore", + } + ) + original_updated_payloads = deletion_service._updated_payloads + source_restored = False + + def restore_source_before_write( + current_payloads: dict[Path, dict[str, object]], + goal_id: str, + ) -> dict[Path, dict[str, object]]: + nonlocal source_restored + if not source_restored: + source_goal = dict(_goal(global_registry, "orphaned-goal")) + source_goal.pop("source_registry", None) + _write_json( + source_registry, + { + "schema_version": "0.1", + "common_runtime_root": str(global_registry.parent), + "goals": [source_goal], + }, + ) + source_restored = True + return original_updated_payloads(current_payloads, goal_id) + + monkeypatch.setattr( + deletion_service, + "_updated_payloads", + restore_source_before_write, + ) + + applied = service.apply(str(proposal["proposal_id"])) + + assert source_restored is True + assert applied["proposal"]["status"] == "stale" + assert applied["proposal"]["receipt"] is None + assert _goal(source_registry, "orphaned-goal")["id"] == "orphaned-goal" + assert _goal(global_registry, "orphaned-goal")["id"] == "orphaned-goal" + assert not list(tmp_path.rglob("*.goal-delete-*.bak")) + + +@pytest.mark.skipif(os.name == "nt", reason="symlink creation requires privileges") +def test_owner_confirmed_typed_action_rejects_symlinked_orphan_source_lock( + tmp_path: Path, +) -> None: + source_registry, global_registry = _orphaned_global_registry( + tmp_path, + activation_state="stopped", + ) + service = ChatActionService( + store=ChatActionStore(tmp_path / "actions"), + registry_path=global_registry, + ) + proposal = service.preview( + { + "action_kind": "goal.lifecycle", + "summary": "Delete an orphaned stopped Goal", + "normalized_parameters": { + "goal_id": "orphaned-goal", + "operation": "delete", + }, + "context": {"kind": "goal_directory"}, + "idempotency_key": "delete-orphaned-goal-with-unsafe-lock", + } + ) + victim = tmp_path / "victim.txt" + victim.write_text("unchanged\n", encoding="utf-8") + source_registry.parent.mkdir(parents=True) + source_registry.with_name(f"{source_registry.name}.lock").symlink_to(victim) + + applied = service.apply(str(proposal["proposal_id"])) + + assert applied["proposal"]["status"] == "failed" + assert applied["proposal"]["receipt"] is None + assert victim.read_text(encoding="utf-8") == "unchanged\n" + assert _goal(global_registry, "orphaned-goal")["id"] == "orphaned-goal" + + +@pytest.mark.parametrize("restored_source_kind", ["goal", "unreadable"]) +def test_owner_confirmed_typed_action_rolls_back_when_orphan_source_restores_after_write( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + restored_source_kind: str, +) -> None: + source_registry, global_registry = _orphaned_global_registry( + tmp_path, + activation_state="stopped", + ) + service = ChatActionService( + store=ChatActionStore(tmp_path / "actions"), + registry_path=global_registry, + ) + proposal = service.preview( + { + "action_kind": "goal.lifecycle", + "summary": "Delete an orphaned stopped Goal", + "normalized_parameters": { + "goal_id": "orphaned-goal", + "operation": "delete", + }, + "context": {"kind": "goal_directory"}, + "idempotency_key": "delete-orphaned-goal-after-source-restore", + } + ) + original_atomic_write = deletion_service.atomic_write_json + restored_source_goal = dict(_goal(global_registry, "orphaned-goal")) + restored_source_goal.pop("source_registry", None) + source_restored = False + + def restore_source_after_global_write( + path: Path, + payload: dict[str, object], + **kwargs: object, + ) -> None: + nonlocal source_restored + original_atomic_write(path, payload, **kwargs) + if ( + not source_restored + and path == global_registry + and not registry_goals(payload) + ): + if restored_source_kind == "goal": + _write_json( + source_registry, + { + "schema_version": "0.1", + "common_runtime_root": str(global_registry.parent), + "goals": [restored_source_goal], + }, + ) + else: + source_registry.mkdir() + source_restored = True + + monkeypatch.setattr( + deletion_service, + "atomic_write_json", + restore_source_after_global_write, + ) + + proposal_id = str(proposal["proposal_id"]) + with pytest.raises(ValueError, match="Goal deletion readback did not verify"): + service.apply(proposal_id) + + assert source_restored is True + persisted = ChatActionStore(tmp_path / "actions").load(proposal_id) + assert persisted is not None + assert persisted["status"] == "applying" + assert persisted["failure"] is None + assert persisted["receipt"] is None + if restored_source_kind == "goal": + assert _goal(source_registry, "orphaned-goal")["id"] == "orphaned-goal" + else: + assert source_registry.is_dir() + assert _goal(global_registry, "orphaned-goal")["id"] == "orphaned-goal" + + def test_owner_confirmed_typed_action_rejects_stale_delete_without_writing( connected_registries: tuple[Path, Path], tmp_path: Path, ) -> None: @@ -1019,6 +1261,373 @@ def test_owner_confirmed_typed_action_rejects_stale_delete_without_writing( assert not list(global_registry.parent.glob("*.goal-delete-*.bak")) +def test_owner_confirmed_typed_action_rejects_changed_source_before_delete( + connected_registries: tuple[Path, Path], + tmp_path: Path, +) -> None: + source_registry, global_registry = connected_registries + service, proposal = _preview_delete_action( + global_registry=global_registry, + action_store=tmp_path / "actions", + idempotency_key="delete-goal-one-before-source-change", + ) + source_basis = proposal["canonical_update_basis"] + activation_preview = set_goal_activation_state( + registry_path=global_registry, + goal_id="goal-one", + state="stopped", + execute=False, + ) + assert source_basis == { + "schema_version": "loopx_goal_deletion_source_basis_v1", + "source_identity": activation_preview["source_identity"], + "source_content_sha256": hashlib.sha256( + source_registry.read_bytes() + ).hexdigest(), + "route_mode": "source_to_global", + } + source_payload = load_registry(source_registry) + registry_goals(source_payload)[0]["display_name"] = "Changed after confirmation" + _write_json(source_registry, source_payload) + before = source_registry.read_bytes(), global_registry.read_bytes() + + applied = service.apply(str(proposal["proposal_id"])) + + assert applied["proposal"]["status"] == "stale" + assert applied["proposal"]["receipt"] is None + assert (source_registry.read_bytes(), global_registry.read_bytes()) == before + assert _goal(source_registry)["display_name"] == "Changed after confirmation" + assert _goal(global_registry)["id"] == "goal-one" + assert not list(tmp_path.rglob("*.goal-delete-*.bak")) + + +def test_owner_confirmed_typed_action_marks_resumed_goal_stale( + connected_registries: tuple[Path, Path], + tmp_path: Path, +) -> None: + source_registry, global_registry = connected_registries + service, proposal = _preview_delete_action( + global_registry=global_registry, + action_store=tmp_path / "actions", + idempotency_key="delete-goal-one-before-resume", + ) + resumed = set_goal_activation_state( + registry_path=global_registry, + goal_id="goal-one", + state="active", + actor_kind="owner", + execute=True, + ) + assert resumed["ok"] is True + + applied = service.apply(str(proposal["proposal_id"])) + + assert applied["proposal"]["status"] == "stale" + assert applied["proposal"]["receipt"] is None + assert goal_activation_state(_goal(source_registry)) is GoalActivationState.ACTIVE + assert goal_activation_state(_goal(global_registry)) is GoalActivationState.ACTIVE + assert not list(tmp_path.rglob("*.goal-delete-*.bak")) + + +def test_owner_confirmed_typed_action_marks_already_deleted_goal_stale( + connected_registries: tuple[Path, Path], + tmp_path: Path, +) -> None: + source_registry, global_registry = connected_registries + service, proposal = _preview_delete_action( + global_registry=global_registry, + action_store=tmp_path / "actions", + idempotency_key="delete-goal-one-before-receipt", + ) + deleted = delete_stopped_goal( + registry_path=global_registry, + goal_id="goal-one", + execute=True, + expected_state_fingerprint=proposal["expected_state_fingerprint"], + expected_source_basis=proposal["canonical_update_basis"], + ) + assert deleted["ok"] is True + + applied = service.apply(str(proposal["proposal_id"])) + + assert applied["proposal"]["status"] == "stale" + assert applied["proposal"]["receipt"] is None + assert registry_goals(load_registry(source_registry)) == [] + assert registry_goals(load_registry(global_registry)) == [] + + +def test_owner_confirmed_typed_action_marks_concurrent_delete_stale( + connected_registries: tuple[Path, Path], + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + source_registry, global_registry = connected_registries + service, proposal = _preview_delete_action( + global_registry=global_registry, + action_store=tmp_path / "actions", + idempotency_key="delete-goal-one-concurrently", + ) + original_delete = chat_goal_lifecycle_actions.delete_stopped_goal + delete_calls = 0 + + def delete_between_preflight_and_execute(**kwargs: object) -> dict[str, object]: + nonlocal delete_calls + delete_calls += 1 + current = original_delete(**kwargs) + if delete_calls == 1: + deleted = original_delete( + registry_path=global_registry, + goal_id="goal-one", + execute=True, + expected_state_fingerprint=proposal["expected_state_fingerprint"], + expected_source_basis=proposal["canonical_update_basis"], + ) + assert deleted["ok"] is True + return current + + monkeypatch.setattr( + chat_goal_lifecycle_actions, + "delete_stopped_goal", + delete_between_preflight_and_execute, + ) + + applied = service.apply(str(proposal["proposal_id"])) + + assert delete_calls == 2 + assert applied["proposal"]["status"] == "stale" + assert applied["proposal"]["receipt"] is None + assert registry_goals(load_registry(source_registry)) == [] + assert registry_goals(load_registry(global_registry)) == [] + + +def test_owner_confirmed_typed_action_marks_locked_concurrent_delete_stale( + connected_registries: tuple[Path, Path], + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + source_registry, global_registry = connected_registries + service, proposal = _preview_delete_action( + global_registry=global_registry, + action_store=tmp_path / "actions", + idempotency_key="delete-goal-one-during-lock", + ) + original_transaction = deletion_service.project_registry_transaction + concurrent_delete_completed = False + + @contextmanager + def delete_before_source_lock(*args: object, **kwargs: object): + nonlocal concurrent_delete_completed + if not concurrent_delete_completed: + for registry in (source_registry, global_registry): + payload = load_registry(registry) + payload["goals"] = [] + _write_json(registry, payload) + concurrent_delete_completed = True + with original_transaction(*args, **kwargs) as transaction: + yield transaction + + monkeypatch.setattr( + deletion_service, + "project_registry_transaction", + delete_before_source_lock, + ) + + applied = service.apply(str(proposal["proposal_id"])) + + assert concurrent_delete_completed is True + assert applied["proposal"]["status"] == "stale" + assert applied["proposal"]["receipt"] is None + assert not list(tmp_path.rglob("*.goal-delete-*.bak")) + + +@pytest.mark.parametrize( + ("failure_call", "expected_status"), + [(1, "failed"), (2, "applying")], +) +def test_owner_confirmed_typed_action_records_delete_failures( + connected_registries: tuple[Path, Path], + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + failure_call: int, + expected_status: str, +) -> None: + source_registry, global_registry = connected_registries + service, proposal = _preview_delete_action( + global_registry=global_registry, + action_store=tmp_path / "actions", + idempotency_key=f"delete-goal-one-lock-failure-{failure_call}", + ) + original_delete = chat_goal_lifecycle_actions.delete_stopped_goal + delete_calls = 0 + + def fail_delete(**kwargs: object) -> dict[str, object]: + nonlocal delete_calls + delete_calls += 1 + if delete_calls == failure_call: + raise PermissionError("simulated registry lock failure") + return original_delete(**kwargs) + + monkeypatch.setattr( + chat_goal_lifecycle_actions, + "delete_stopped_goal", + fail_delete, + ) + + proposal_id = str(proposal["proposal_id"]) + if failure_call == 1: + applied = service.apply(proposal_id)["proposal"] + else: + with pytest.raises( + PermissionError, + match="simulated registry lock failure", + ): + service.apply(proposal_id) + applied = ChatActionStore(tmp_path / "actions").load(proposal_id) + + assert delete_calls == failure_call + assert applied is not None + assert applied["status"] == expected_status + if failure_call == 1: + assert applied["failure"]["error_code"] == "goal_delete_unavailable" + assert applied["failure"]["retry_safe"] is True + else: + assert applied["failure"] is None + assert _goal(source_registry)["id"] == "goal-one" + assert _goal(global_registry)["id"] == "goal-one" + + +def test_owner_confirmed_typed_action_does_not_mark_committed_delete_retry_safe( + connected_registries: tuple[Path, Path], + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + source_registry, global_registry = connected_registries + action_store = tmp_path / "actions" + service, proposal = _preview_delete_action( + global_registry=global_registry, + action_store=action_store, + idempotency_key="delete-goal-one-before-receipt-write-failure", + ) + proposal_id = str(proposal["proposal_id"]) + original_write = service.store._write + + def fail_applied_receipt(payload: dict[str, object]) -> None: + proposals = payload.get("proposals") + stored = proposals.get(proposal_id) if isinstance(proposals, dict) else None + if isinstance(stored, dict) and stored.get("status") == "applied": + raise PermissionError("simulated receipt write failure") + original_write(payload) + + monkeypatch.setattr(service.store, "_write", fail_applied_receipt) + + with pytest.raises(PermissionError, match="simulated receipt write failure"): + service.apply(proposal_id) + + persisted = ChatActionStore(action_store).load(proposal_id) + assert persisted is not None + assert persisted["status"] == "applying" + assert persisted["failure"] is None + assert registry_goals(load_registry(source_registry)) == [] + assert registry_goals(load_registry(global_registry)) == [] + + +def test_delete_stopped_goal_rechecks_source_inside_write_lock( + connected_registries: tuple[Path, Path], + monkeypatch: pytest.MonkeyPatch, +) -> None: + source_registry, global_registry = connected_registries + set_goal_activation_state( + registry_path=global_registry, + goal_id="goal-one", + state="stopped", + actor_kind="owner", + execute=True, + ) + preview = delete_stopped_goal( + registry_path=global_registry, + goal_id="goal-one", + execute=False, + ) + original_transaction = deletion_service.project_registry_transaction + source_changed = False + + @contextmanager + def change_source_before_source_lock(*args: object, **kwargs: object): + nonlocal source_changed + if not source_changed: + source_payload = load_registry(source_registry) + registry_goals(source_payload)[0]["display_name"] = ( + "Changed before source lock" + ) + _write_json(source_registry, source_payload) + source_changed = True + with original_transaction(*args, **kwargs) as transaction: + yield transaction + + monkeypatch.setattr( + deletion_service, + "project_registry_transaction", + change_source_before_source_lock, + ) + before_global = global_registry.read_bytes() + + result = delete_stopped_goal( + registry_path=global_registry, + goal_id="goal-one", + execute=True, + expected_state_fingerprint=preview["observed_state_fingerprint"], + expected_source_basis=preview["source_basis"], + ) + + assert source_changed is True + assert result["ok"] is False + assert result["stale"] is True + assert result["written"] is False + assert _goal(source_registry)["display_name"] == "Changed before source lock" + assert global_registry.read_bytes() == before_global + assert not list(global_registry.parent.glob("*.goal-delete-*.bak")) + + +def test_delete_stopped_goal_rejects_restored_orphan_source( + tmp_path: Path, +) -> None: + source_registry, global_registry = _orphaned_global_registry( + tmp_path, + activation_state="stopped", + ) + preview = delete_stopped_goal( + registry_path=global_registry, + goal_id="orphaned-goal", + execute=False, + ) + source_goal = dict(_goal(global_registry, "orphaned-goal")) + source_goal.pop("source_registry", None) + _write_json( + source_registry, + { + "schema_version": "0.1", + "common_runtime_root": str(global_registry.parent), + "goals": [source_goal], + }, + ) + before_global = global_registry.read_bytes() + + result = delete_stopped_goal( + registry_path=global_registry, + goal_id="orphaned-goal", + execute=True, + expected_state_fingerprint=preview["observed_state_fingerprint"], + expected_source_basis=preview["source_basis"], + ) + + assert result["ok"] is False + assert result["stale"] is True + assert result["written"] is False + assert _goal(source_registry, "orphaned-goal")["id"] == "orphaned-goal" + assert global_registry.read_bytes() == before_global + assert not list(tmp_path.rglob("*.goal-delete-*.bak")) + + def test_goal_deletion_backups_are_unique_and_preserve_preimages( connected_registries: tuple[Path, Path], monkeypatch: pytest.MonkeyPatch, ) -> None: diff --git a/tests/test_file_lock.py b/tests/test_file_lock.py index 7ee2f86469..802e08d312 100644 --- a/tests/test_file_lock.py +++ b/tests/test_file_lock.py @@ -107,6 +107,33 @@ def test_exclusive_lock_persists_public_safe_holder_metadata(tmp_path: Path) -> assert holder_path.exists() +@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" + victim = tmp_path / "victim.txt" + victim.write_text("unchanged\n", encoding="utf-8") + target.with_name(f"{target.name}.lock").symlink_to(victim) + + with pytest.raises(OSError): + with exclusive_file_lock(target): + pytest.fail("symlinked lock file was accepted") + + assert victim.read_text(encoding="utf-8") == "unchanged\n" + + +def test_exclusive_lock_rejects_a_hard_linked_lock_file(tmp_path: Path) -> None: + target = tmp_path / "state.json" + victim = tmp_path / "victim.txt" + victim.write_text("unchanged\n", encoding="utf-8") + os.link(victim, target.with_name(f"{target.name}.lock")) + + with pytest.raises(OSError): + with exclusive_file_lock(target): + pytest.fail("hard-linked lock file was accepted") + + assert victim.read_text(encoding="utf-8") == "unchanged\n" + + def test_stalled_holder_times_out_and_records_independent_incident(tmp_path: Path) -> None: target = tmp_path / "todos.md" process = _start_stalled_holder(target) @@ -160,6 +187,30 @@ def test_stalled_holder_times_out_and_records_independent_incident(tmp_path: Pat assert target.with_name(f"{target.name}.lock").exists() +@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, +) -> None: + target = tmp_path / "todos.md" + victim = tmp_path / "victim.txt" + victim.write_text("unchanged\n", encoding="utf-8") + process = _start_stalled_holder(target) + try: + lock_incident_path(target).symlink_to(victim) + with pytest.raises(LockAcquireTimeoutError) as raised: + with exclusive_file_lock( + target, + timeout_seconds=0.05, + poll_interval_seconds=0.01, + ): + pytest.fail("waiter unexpectedly acquired the stalled lock") + + assert raised.value.incident_recorded is False + assert victim.read_text(encoding="utf-8") == "unchanged\n" + finally: + _stop(process) + + def test_single_flight_returns_none_without_timeout_incident(tmp_path: Path) -> None: target = tmp_path / "sync.json" process = _start_stalled_holder(target)