fix(sandbox): keep quarantine teardown idempotent when a peer finishes first (#7313)

* fix(sandbox): return from quarantine teardown when a peer already did the work

retire() and retire_replaced_generations() guard their final rmdir() but list
the fence directory and unlink each record unguarded, while contains() and
_exists() in the same class already report a directory removed underneath them
as "no fence". The teardown exclusion is a TTL lease, not a lock, so a peer can
finish the same teardown in those two windows and the resulting FileNotFoundError
escapes the unwrapped _retire_absent_quarantine() call at the top of creation.

Both passes now tolerate a peer that removed the directory or the record first.

* test(sandbox): correct quarantine docstring reference

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

backend/packages/harness/deerflow/community/aio_sandbox/quarantine.py

@@ -41,6 +41,28 @@ class SandboxQuarantine:
             return False
         return True
 
+    @staticmethod
+    def _records(directory: Path) -> list[Path]:
+        """List the fences a teardown pass is about to drop.
+
+        The teardown exclusion is a TTL lease, not a lock, so a peer's retire()
+        can finish between the caller's existence check and this listing. Its
+        records are then gone, which is the state both callers were asked to
+        produce; ``contains()`` reports the same interleaving as "no fence".
+        """
+        try:
+            return list(directory.iterdir())
+        except FileNotFoundError:
+            return []
+
+    @staticmethod
+    def _drop(record: Path) -> None:
+        """Unlink one fence, tolerating a peer that unlinked it first."""
+        try:
+            record.unlink()
+        except FileNotFoundError:
+            pass
+
     def mark(self, info: SandboxInfo) -> None:
         directory = self._directory(info)
         directory.mkdir(parents=True, exist_ok=True, mode=0o700)
@@ -84,10 +106,10 @@ class SandboxQuarantine:
         if not self._exists(directory):
             return
         keep = self._key(current.container_id) if current.container_id else None
-        for record in directory.iterdir():
+        for record in self._records(directory):
             if record.name == "unknown" or (keep is not None and record.name == keep):
                 continue
-            record.unlink()
+            self._drop(record)
         try:
             directory.rmdir()
         except OSError:
@@ -117,18 +139,20 @@ class SandboxQuarantine:
 
         This is best-effort, not full fence preservation: it closes the listing
         → ``rmdir`` window only. A peer record landing before the listing is
-        still unlinked along with the rest, and the unlink loop has unguarded
-        sibling windows of its own — a concurrent ``retire()`` can remove the
-        directory between the existence check and the listing
-        (``FileNotFoundError``), and on Windows a peer's ``mark()`` still
-        holding the fresh record open for its ``fsync`` raises
-        ``PermissionError``.
+        still unlinked along with the rest. Both the listing and each unlink
+        tolerate a peer that removed the directory or the record first, matching
+        the ``FileNotFoundError`` handling in :meth:`contains`, so a teardown
+        that finds its work already done returns instead of raising out of
+        ``_retire_absent_quarantine``. What stays unclosed on purpose: on Windows
+        a peer's ``mark()`` still holding the fresh record open for its
+        ``fsync`` raises ``PermissionError``, which is a real storage failure
+        rather than an absent fence.
         """
         directory = self._directory(info)
         if not self._exists(directory):
             return
-        for record in directory.iterdir():
-            record.unlink()
+        for record in self._records(directory):
+            self._drop(record)
         try:
             directory.rmdir()
         except OSError:

backend/tests/test_quarantine_retire_race_regression.py

@@ -9,12 +9,23 @@ teardown path. Its sibling `retire_replaced_generations()` guards the identical
 
 Deterministic: the race is replayed by wrapping `Path.rmdir` so a peer record
 lands between the `iterdir()` and the `rmdir()` call.
+The `rmdir()` window above is closed. Two windows ahead of it are not, in both
+`retire()` and `retire_replaced_generations()`: the `iterdir()` listing and each
+`record.unlink()`. A peer running the same teardown one step ahead can remove the
+directory, or one record, in exactly those windows, and the `FileNotFoundError`
+then escapes the unwrapped `_retire_absent_quarantine()` call at the top of
+sandbox creation. The class already reports that interleaving as a normal answer
+two methods away: `_exists()` maps it to "no quarantine" and `contains()` catches
+it on its own listing. The cases below replay each remaining window by
+monkeypatching `Path.iterdir` or `Path.unlink`, so they stay deterministic.
 """
 
 from __future__ import annotations
 
 from pathlib import Path
 
+import pytest
+
 from deerflow.community.aio_sandbox.quarantine import SandboxQuarantine
 from deerflow.community.aio_sandbox.sandbox_info import SandboxInfo
 
@@ -45,3 +56,101 @@ def test_retire_survives_a_peer_fencing_between_listing_and_rmdir(tmp_path, monk
 
     assert directory.exists()
     assert (directory / store._key("gen-2")).exists()
+
+
+@pytest.mark.parametrize("method", ["retire", "retire_replaced_generations"])
+def test_teardown_survives_the_directory_disappearing_before_the_listing(tmp_path, monkeypatch, method):
+    """A peer finishing its teardown first leaves nothing to remove, and must not raise.
+
+    `_exists()` reported the directory, then the peer unlinked every record and
+    removed it, so the listing itself fails with FileNotFoundError. The
+    `contains()` listing two methods above already treats this exact
+    interleaving as "the fence is gone"; neither teardown loop does.
+    """
+    store = SandboxQuarantine(tmp_path, "provisioner")
+    info = SandboxInfo("thread-sandbox", "http://sandbox", container_id="gen-1")
+    store.mark(info)
+    directory = tmp_path / store._key("provisioner") / store._key("thread-sandbox")
+
+    real_iterdir = Path.iterdir
+    raced = []
+
+    def racing_iterdir(self):
+        if self == directory and not raced:
+            raced.append(self)
+            # The peer's retire() completes inside our own teardown window.
+            for record in list(real_iterdir(self)):
+                record.unlink()
+            Path.rmdir(self)
+        return real_iterdir(self)
+
+    monkeypatch.setattr(Path, "iterdir", racing_iterdir)
+
+    getattr(store, method)(info)  # must not raise
+
+    assert not directory.exists()
+
+
+def test_retire_survives_a_record_disappearing_mid_loop(tmp_path, monkeypatch):
+    """A peer's unlink landing between the listing and ours must not abort the loop.
+
+    Two generations are fenced. The peer drops the first record after `retire()`
+    listed the directory, so `record.unlink()` fails on a name that is already
+    gone while the second record and the `rmdir()` still need the loop to finish.
+    """
+    store = SandboxQuarantine(tmp_path, "provisioner")
+    info = SandboxInfo("thread-sandbox", "http://sandbox", container_id="gen-1")
+    for container_id in ("gen-1", "gen-2"):
+        store.mark(SandboxInfo("thread-sandbox", "http://sandbox", container_id=container_id))
+    directory = tmp_path / store._key("provisioner") / store._key("thread-sandbox")
+    raced_record = directory / store._key("gen-1")
+
+    real_unlink = Path.unlink
+    raced = []
+
+    def racing_unlink(self):
+        if self == raced_record and not raced:
+            raced.append(self)
+            real_unlink(self)
+        return real_unlink(self)
+
+    monkeypatch.setattr(Path, "unlink", racing_unlink)
+
+    store.retire(info)  # must not raise
+
+    assert not raced_record.exists()
+    # The loop reached the second record, so the emptied directory was removed.
+    assert not directory.exists()
+
+
+def test_retire_replaced_generations_keeps_the_live_fence_when_a_peer_drops_a_stale_one_first(tmp_path, monkeypatch):
+    """The prune path must survive losing a stale record mid-loop and still keep the live fence.
+
+    `retire_replaced_generations()` skips the live generation's record by design.
+    A peer unlinking the stale record first raises inside the loop, before the
+    live record is even reached, so the prune aborts with FileNotFoundError.
+    """
+    store = SandboxQuarantine(tmp_path, "provisioner")
+    stale = SandboxInfo("thread-sandbox", "http://sandbox", container_id="gen-1")
+    live = SandboxInfo("thread-sandbox", "http://sandbox", container_id="gen-2")
+    store.mark(stale)
+    store.mark(live)
+    directory = tmp_path / store._key("provisioner") / store._key("thread-sandbox")
+    raced_record = directory / store._key("gen-1")
+
+    real_unlink = Path.unlink
+    raced = []
+
+    def racing_unlink(self):
+        if self == raced_record and not raced:
+            raced.append(self)
+            real_unlink(self)
+        return real_unlink(self)
+
+    monkeypatch.setattr(Path, "unlink", racing_unlink)
+
+    store.retire_replaced_generations(live)  # must not raise
+
+    assert not raced_record.exists()
+    assert (directory / store._key("gen-2")).exists()
+    assert directory.exists()