Fix file mutation lock cleanup

loopx/file_lock.py

@@ -852,7 +852,6 @@ def _read_effect_mutation_owner(path: Path) -> dict[str, object] | None:
 @dataclass(frozen=True)
 class _EffectMutationClaim:
     path: Path
-    token: str
     identity: tuple[int, int, int, int]
 
 
@@ -983,46 +982,25 @@ def _claim_effect_mutation_lock(
             raise
         if identity is None:
             return None
-        return _EffectMutationClaim(claim_path, token, identity)
+        return _EffectMutationClaim(claim_path, identity)
     return None
 
 
 def _release_effect_mutation_claim(
     claim: _EffectMutationClaim | Path,
     token: str | None = None,
 ) -> None:
-    expected_token: str | None
-    identity: tuple[int, int, int, int] | None
     if isinstance(claim, _EffectMutationClaim):
-        path = claim.path
-        expected_token = claim.token
-        identity = claim.identity
-    else:
-        path = claim
-        expected_token = token
-        identity = _effect_file_identity(path)
-    if not expected_token:
+        # Remove only the file we created, even if its contents became unreadable.
+        _remove_created_effect_file(claim.path, claim.identity)
         return
-    owner = _read_effect_mutation_owner(path)
-    if isinstance(claim, _EffectMutationClaim):
-        if (
-            owner is None
-            or owner.get("pid") != os.getpid()
-            or owner.get("token") != expected_token
-        ):
-            # The pathname may have been corrupted or replaced after this
-            # caller created the claim.  Its captured identity still permits
-            # removing only its own inode, preventing claim leaks without
-            # touching a later claimant.
-            _remove_created_effect_file(path, identity)
-            return
-    elif (
-        owner is None
-        or owner.get("pid") != os.getpid()
-        or owner.get("token") != expected_token
-    ):
+    if not token:
         return
-    _remove_created_effect_file(path, identity)
+    identity = _effect_file_identity(claim)
+    owner = _read_effect_mutation_owner(claim)
+    if owner is None or owner.get("pid") != os.getpid() or owner.get("token") != token:
+        return
+    _remove_created_effect_file(claim, identity)
 
 
 def _reclaim_stale_effect_mutation_lock(path: Path) -> bool:
@@ -1101,6 +1079,7 @@ def _release_effect_mutation_lock(
         if claim is None:
             return False
         retired_path = path.with_name(f"{path.name}.released.{uuid4()}")
+        retired_cleaned = False
         try:
             current = _read_effect_mutation_owner(path)
             if current is None or current.get("token") != token:
@@ -1115,20 +1094,23 @@ def _release_effect_mutation_lock(
                 retired_path.unlink(missing_ok=True)
             except OSError:
                 pass
+            else:
+                retired_cleaned = True
             return True
         finally:
-            if claim is not None:
-                _release_effect_mutation_claim(claim)
-            try:
-                retired_path.unlink(missing_ok=True)
-            except OSError:
-                pass
+            _release_effect_mutation_claim(claim)
+            if not retired_cleaned:
+                try:
+                    retired_path.unlink(missing_ok=True)
+                except OSError:
+                    pass
     except OSError:
         if suppress_errors:
             return False
         raise
     finally:
         if claim is not None:
+            # Retry best-effort removal if the first attempt left our claim.
             _release_effect_mutation_claim(claim)
 
 

tests/test_file_lock.py

@@ -405,19 +405,63 @@ def test_cross_runtime_identity_fails_closed_for_ambiguous_or_reused_files() ->
     )
 
 
+@pytest.mark.parametrize("contents", [b"not-json", b"\xff"])
 def test_cross_runtime_claim_cleanup_removes_own_corrupted_claim(
     tmp_path: Path,
+    contents: bytes,
 ) -> None:
     target = tmp_path / ".task-leases"
     claim = file_lock._claim_effect_mutation_lock(target, "claim-token")
     assert claim is not None
-    claim.path.write_text("not-json", encoding="utf-8")
+    claim.path.write_bytes(contents)
 
     file_lock._release_effect_mutation_claim(claim)
 
     assert not claim.path.exists()
 
 
+def test_cross_runtime_claim_cleanup_keeps_a_replacement_inode(tmp_path: Path) -> None:
+    target = tmp_path / ".task-leases"
+    claim = file_lock._claim_effect_mutation_lock(target, "claim-token")
+    assert claim is not None
+    replacement = tmp_path / "replacement"
+    contents = claim.path.read_bytes()
+    replacement.write_bytes(contents)
+    replacement.replace(claim.path)
+
+    file_lock._release_effect_mutation_claim(claim)
+
+    assert claim.path.read_bytes() == contents
+
+
+@pytest.mark.parametrize("part", ["claim", "released"])
+def test_cross_runtime_release_retries_transient_cleanup(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    part: str,
+) -> None:
+    target = tmp_path / ".task-leases"
+    effect_lock = Path(f"{target}.ts-effect.lock")
+    unlink = Path.unlink
+    attempts = 0
+
+    def unlink_with_one_failure(path: Path, missing_ok: bool = False) -> None:
+        nonlocal attempts
+        if path.name.startswith(f"{effect_lock.name}.{part}."):
+            attempts += 1
+            if attempts == 1:
+                raise OSError("simulated marker cleanup failure")
+        unlink(path, missing_ok=missing_ok)
+
+    monkeypatch.setattr(Path, "unlink", unlink_with_one_failure)
+    with exclusive_cross_runtime_file_lock(target):
+        assert effect_lock.exists()
+
+    assert attempts == 2
+    assert not effect_lock.exists()
+    assert not list(tmp_path.glob(f"{effect_lock.name}.{part}.*"))
+
+
 def test_cross_runtime_cleanup_failure_cannot_replace_success(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,