diff --git a/Makefile b/Makefile
index ad15bad1..522d5b59 100755
--- a/Makefile
+++ b/Makefile
@@ -22,8 +22,8 @@ bootstrap: install ui-install ## Setup everything
 install: ## Install backend dependencies
 	@echo "Installing backend..."
 	@test -d $(VENV) || $(PY) -m venv $(VENV)
-	@$(VENV)/bin/python -c 'import sys; sys.exit(sys.version_info < (3, 10))' || \
-		{ echo "ERROR: $(VENV) is $$($(VENV)/bin/python -V); requirements-dev.txt needs Python >= 3.10 (repo targets 3.14). Recreate the venv with a newer PY=."; exit 1; }
+	@$(VENV)/bin/python -c 'import sys; sys.exit(sys.version_info < (3, 11))' || \
+		{ echo "ERROR: $(VENV) is $$($(VENV)/bin/python -V); this repo needs Python >= 3.11 (shutil.rmtree(dir_fd=) is 3.11+); CI and Docker build on 3.14. Recreate the venv with a newer PY=."; exit 1; }
 	@$(VENV)/bin/python -m pip install --upgrade pip
 	@$(VENV)/bin/pip install -r requirements.txt
 	@$(VENV)/bin/pip install -r requirements-dev.txt
diff --git a/backend/modules/poster_cleanarr.py b/backend/modules/poster_cleanarr.py
index 0f8212ab..11a931b8 100644
--- a/backend/modules/poster_cleanarr.py
+++ b/backend/modules/poster_cleanarr.py
@@ -1,5 +1,6 @@
 # modules/poster_cleanarr.py
 
+import errno
 import glob
 import json
 import os
@@ -12,13 +13,14 @@
 from PIL import Image, UnidentifiedImageError
 
 from backend.util.base_module import ChubModule
+from backend.util.config import ChubConfig
 from backend.util.constants import asset_type_regex, tmdb_id_regex, tvdb_id_regex
 from backend.util.database import ChubDB
 from backend.util.helper import create_table
 from backend.util.logger import Logger
 from backend.util.notification import NotificationManager
 from backend.util.normalization import normalize_titles, parse_asset_filename
-from backend.util.path_safety import resolve_confined
+from backend.util.path_safety import containing_root, resolve_confined
 
 # EXIF tag id Kometa writes onto its generated overlay images. Used by the
 # overlays_only mode to skip user-uploaded customs (which lack the tag).
@@ -104,6 +106,10 @@ def _bundle_root(path: str) -> Optional[str]:
 
 
 class PosterCleanarr(ChubModule):
+    # Per-pass refusal tally behind _refuse/_report_refusals. Class level because
+    # shim callers build the module with object.__new__, skipping __init__.
+    _refused_paths = 0
+
     def __init__(self, logger: Optional[Logger] = None) -> None:
         super().__init__(logger=logger)
         self.plex_path: str = getattr(self.config, "plex_path", "")
@@ -848,23 +854,216 @@ def _resolve_orphan_instances(config: Any) -> List[str]:
             getattr(config, "instances", []) or []
         )
 
-    def _authorized_asset_dirs(
-        self, asset_dirs: List[str], logger: Logger
-    ) -> List[str]:
-        """Existing asset_dirs re-resolved inside the live config's allowed roots."""
-        # Re-authorized here because the API confines at enqueue but a worker
-        # acts later; an unloadable config returns [] so nothing is deleted.
+    def _live_config(self, logger: Logger) -> Optional[ChubConfig]:
+        """Config as it is on disk now, or None (logged) so callers fail closed."""
         # Never self.full_config: that snapshot is taken at construction, so a
         # root removed since would still authorize. Lazy import keeps it patchable.
         from backend.util.config import load_config
 
         try:
-            config = load_config()
+            return load_config()
         except Exception as e:
             logger.error(
-                f"Cannot authorize asset_dirs against the allowed roots ({e}); "
+                f"Cannot authorize paths against the allowed roots ({e}); "
                 "skipping cleanup instead of acting on unverified paths."
             )
+            return None
+
+    def _refuse(self, logger: Logger, message: str) -> None:
+        """Tally one refusal, keeping its per-path detail at debug."""
+        self._refused_paths += 1
+        logger.debug(message)
+
+    def _report_refusals(self, logger: Logger, what: str) -> None:
+        """Log one ERROR carrying the tallied refusals, then reset the tally."""
+        if self._refused_paths:
+            logger.error(
+                f"{what}: refused {self._refused_paths} path(s) resolving outside "
+                "the allowed roots or their base dir; per-path detail at debug level."
+            )
+        self._refused_paths = 0
+
+    def _confined_target(
+        self, path: str, config: Optional[ChubConfig], logger: Logger
+    ) -> Optional[str]:
+        """Realpath of `path` if inside a live allowed root, else None (tallied)."""
+        real = resolve_confined(path, config) if config is not None else None
+        if real is None:
+            self._refuse(
+                logger,
+                f"Target resolves outside the allowed roots, refusing to touch: {path}",
+            )
+            return None
+        return str(real)
+
+    def _walk_open_dir(
+        self, real_dir: str, config: ChubConfig, logger: Logger
+    ) -> Optional[int]:
+        """Descriptor on `real_dir`, opened one component at a time from its allowed root."""
+        # The root is the trust ANCHOR: its own ancestors are unverifiable from
+        # here, so confinement starts at it and every step below refuses a link.
+        root = containing_root(real_dir, config)
+        if root is None:
+            logger.error(f"Refusing to open {real_dir}: no allowed root contains it")
+            return None
+        flags = (
+            os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0)
+        )
+        try:
+            dir_fd = os.open(str(root), flags)
+        except OSError as e:
+            logger.error(f"Refusing to open {real_dir}: root {root} not openable ({e})")
+            return None
+        rel = os.path.relpath(real_dir, str(root))
+        for name in [c for c in rel.split(os.sep) if c and c != os.curdir]:
+            try:
+                # NEVER re-open the whole string: O_NOFOLLOW only guards the last
+                # component, so an intermediate one swapped to a link would escape.
+                nxt = os.open(name, flags, dir_fd=dir_fd)
+            except OSError as e:
+                logger.error(
+                    f"Refusing to open {real_dir}: '{name}' is not a plain "
+                    f"directory reached from {root} ({e})"
+                )
+                os.close(dir_fd)
+                return None
+            os.close(dir_fd)
+            dir_fd = nxt
+        return dir_fd
+
+    def _confined_parent_fd(
+        self, path: str, config: Optional[ChubConfig], logger: Logger
+    ) -> Optional[int]:
+        """Descriptor on `path`'s confined parent, or None (logged); caller closes."""
+        if self._confined_target(path, config, logger) is None:
+            return None
+        # Confine the PARENT, not the leaf: the leaf is what we act on, and
+        # naming it through this descriptor is what closes the race.
+        parent = self._confined_target(os.path.dirname(path), config, logger)
+        if parent is None or config is None:
+            return None
+        dir_fd = self._walk_open_dir(parent, config, logger)
+        if dir_fd is None:
+            return None
+        try:
+            # Kept as a cheap SECONDARY check — the walk above is the barrier;
+            # this still catches the parent being renamed after it was opened.
+            opened, authorized = os.fstat(dir_fd), os.stat(parent)
+            if (opened.st_dev, opened.st_ino) != (authorized.st_dev, authorized.st_ino):
+                logger.error(f"Refusing to touch {path}: its parent changed under us")
+                os.close(dir_fd)
+                return None
+        except OSError as e:
+            logger.error(f"Refusing to touch {path}: parent unverifiable ({e})")
+            os.close(dir_fd)
+            return None
+        return dir_fd
+
+    def _remove_confined(
+        self, path: str, config: Optional[ChubConfig], logger: Logger
+    ) -> bool:
+        """Unlink `path` by name through a descriptor pinned to its confined parent."""
+        dir_fd = self._confined_parent_fd(path, config, logger)
+        if dir_fd is None:
+            return False
+        try:
+            # NEVER resolve-then-unlink: that deletes a symlink's TARGET. The
+            # name is resolved by the kernel against dir_fd, so it can't escape.
+            os.unlink(os.path.basename(path), dir_fd=dir_fd)
+            return True
+        except OSError as e:
+            logger.error(f"Failed to remove {path}: {e}")
+            return False
+        finally:
+            os.close(dir_fd)
+
+    def _rmtree_confined(
+        self, folder: str, config: Optional[ChubConfig], logger: Logger
+    ) -> bool:
+        """rmtree `folder` by name through a descriptor on its confined parent."""
+        dir_fd = self._confined_parent_fd(folder, config, logger)
+        if dir_fd is None:
+            return False
+        try:
+            shutil.rmtree(os.path.basename(folder), dir_fd=dir_fd)
+            return True
+        except OSError as e:
+            logger.error(f"Failed to remove {folder}: {e}")
+            return False
+        finally:
+            os.close(dir_fd)
+
+    def _rmdir_confined(self, dir_path: str, config: ChubConfig) -> bool:
+        """rmdir `dir_path` by name through a descriptor on its confined parent."""
+        dir_fd = self._confined_parent_fd(dir_path, config, self.logger)
+        if dir_fd is None:
+            return False
+        try:
+            # OSError propagates: the sweep's own handler explains the count gap.
+            os.rmdir(os.path.basename(dir_path), dir_fd=dir_fd)
+            return True
+        finally:
+            os.close(dir_fd)
+
+    def _anchored_rename(
+        self, src: str, dest: str, config: Optional[ChubConfig], logger: Logger
+    ) -> Optional[bool]:
+        """rename `src` to `dest` through both confined parents; None = cross-device."""
+        src_fd = self._confined_parent_fd(src, config, logger)
+        if src_fd is None:
+            return False
+        try:
+            dest_parent = self._confined_target(os.path.dirname(dest), config, logger)
+            if dest_parent is None or config is None:
+                return False
+            dest_fd = self._walk_open_dir(dest_parent, config, logger)
+            if dest_fd is None:
+                return False
+            try:
+                # Both ends resolved by the kernel against a pinned descriptor,
+                # so neither parent can be swapped for a link after the check.
+                os.rename(
+                    os.path.basename(src),
+                    os.path.basename(dest),
+                    src_dir_fd=src_fd,
+                    dst_dir_fd=dest_fd,
+                )
+                return True
+            except OSError as e:
+                if e.errno == errno.EXDEV:
+                    return None
+                logger.error(f"Failed to move {src}: {e}")
+                return False
+            finally:
+                os.close(dest_fd)
+        finally:
+            os.close(src_fd)
+
+    def _move_confined(
+        self, src: str, dest: str, config: Optional[ChubConfig], logger: Logger
+    ) -> bool:
+        """Move `src` to `dest`, anchored to both parents; copies only across devices."""
+        renamed = self._anchored_rename(src, dest, config, logger)
+        if renamed is not None:
+            return renamed
+        # Cross-device has no anchored primitive: this copy+unlink re-resolves
+        # both paths, the one case the descriptor pin cannot cover.
+        logger.debug(f"Anchored rename unavailable across devices: {src} -> {dest}")
+        try:
+            shutil.move(src, dest)
+            return True
+        except OSError as e:
+            logger.error(f"Failed to move {src}: {e}")
+            return False
+
+    def _authorized_asset_dirs(
+        self, asset_dirs: List[str], logger: Logger
+    ) -> List[str]:
+        """Existing asset_dirs re-resolved inside the live config's allowed roots."""
+        # Re-authorized here because the API confines at enqueue but a worker
+        # acts later; an unloadable config returns [] so nothing is deleted.
+        config = self._live_config(logger)
+        if config is None:
             return []
 
         authorized: List[str] = []
@@ -1127,28 +1326,43 @@ def _execute_stale_mode(
                     "not staged yet; keeping the only copy"
                 )
                 continue
+            # Re-read per item, not once per batch: a root dropped from config
+            # mid-run must stop the REST of the batch. load_config is mtime-cached.
+            config = self._live_config(logger)
+            if config is None:
+                self._refuse(logger, f"No live config, refusing to {mode}: {folder}")
+                continue
             if mode == "move":
                 dest_root = os.path.join(d["asset_dir"], ORPHAN_RESTORE_DIR_NAME)
                 dest = os.path.join(dest_root, d["name"])
+                if self._confined_target(folder, config, logger) is None:
+                    continue
                 try:
                     os.makedirs(os.path.dirname(dest), exist_ok=True)
-                    shutil.move(folder, dest)
-                    logger.info(f"  [STALE MOVED] {folder} -> {dest}")
-                    count += 1
-                    total_size += size
-                    touched.add(d["asset_dir"])
                 except OSError as e:
                     logger.error(f"Failed to move {folder}: {e}")
+                    continue
+                if not self._move_confined(folder, dest, config, logger):
+                    continue
+                logger.info(f"  [STALE MOVED] {folder} -> {dest}")
+                count += 1
+                total_size += size
+                touched.add(d["asset_dir"])
             elif mode == "remove":
-                try:
-                    shutil.rmtree(folder)
-                    logger.info(f"  [STALE REMOVED] {folder}")
-                    count += 1
-                    total_size += size
-                    touched.add(d["asset_dir"])
-                except OSError as e:
-                    logger.error(f"Failed to remove {folder}: {e}")
-        empty = sum(self._clean_empty_dirs(d) for d in touched)
+                if not self._rmtree_confined(folder, config, logger):
+                    continue
+                logger.info(f"  [STALE REMOVED] {folder}")
+                count += 1
+                total_size += size
+                touched.add(d["asset_dir"])
+        self._report_refusals(logger, "Stale-duplicate cleanup")
+        empty = 0
+        if touched:
+            # Re-read before the sweep too, and skip it entirely on None: the
+            # weaker base_dir-only floor is for the bloat pass, not for this one.
+            sweep_config = self._live_config(logger)
+            if sweep_config is not None:
+                empty = sum(self._clean_empty_dirs(d, sweep_config) for d in touched)
         logger.info(
             f"   → stale duplicates: {count} {mode}d"
             + (f", {empty} empty dir(s) pruned" if empty else "")
@@ -1366,35 +1580,52 @@ def _execute_orphan_mode(
                 total_size += size
                 continue
 
+            # Re-read per item, not once per batch: a root dropped from config
+            # mid-run must stop the REST of the batch. load_config is mtime-cached.
+            config = self._live_config(logger)
+            if config is None:
+                self._refuse(logger, f"No live config, refusing to {mode}: {path}")
+                continue
+
             if mode == "move":
                 dest_root = os.path.join(item["asset_dir"], ORPHAN_RESTORE_DIR_NAME)
                 rel = os.path.relpath(path, item["asset_dir"])
                 dest = os.path.join(dest_root, rel)
+                if self._confined_target(path, config, logger) is None:
+                    continue
                 try:
                     os.makedirs(os.path.dirname(dest), exist_ok=True)
-                    shutil.move(path, dest)
-                    # Destructive ops stay at INFO deliberately — audit trail
-                    # without debug mode (mirrors the bloat-cleanup pass).
-                    logger.info(f"  [MOVED] {path} -> {dest}")
-                    count += 1
-                    total_size += size
-                    touched_dirs.add(item["asset_dir"])
                 except OSError as e:
                     logger.error(f"Failed to move {path}: {e}")
+                    continue
+                if not self._move_confined(path, dest, config, logger):
+                    continue
+                # Destructive ops stay at INFO deliberately — audit trail
+                # without debug mode (mirrors the bloat-cleanup pass).
+                logger.info(f"  [MOVED] {path} -> {dest}")
+                count += 1
+                total_size += size
+                touched_dirs.add(item["asset_dir"])
                 continue
 
             if mode == "remove":
-                try:
-                    os.remove(path)
-                    logger.info(f"  [REMOVED] {path}")
-                    count += 1
-                    total_size += size
-                    touched_dirs.add(item["asset_dir"])
-                except OSError as e:
-                    logger.error(f"Failed to remove {path}: {e}")
-
-        # Prune empty dirs left behind by move/remove.
-        empty_dirs = sum(self._clean_empty_dirs(d) for d in touched_dirs)
+                if not self._remove_confined(path, config, logger):
+                    continue
+                logger.info(f"  [REMOVED] {path}")
+                count += 1
+                total_size += size
+                touched_dirs.add(item["asset_dir"])
+
+        self._report_refusals(logger, "Orphan cleanup")
+        # Prune empty dirs left behind by move/remove, on a re-read config: the
+        # loop above can run for minutes, and None must skip the sweep entirely.
+        empty_dirs = 0
+        if touched_dirs:
+            sweep_config = self._live_config(logger)
+            if sweep_config is not None:
+                empty_dirs = sum(
+                    self._clean_empty_dirs(d, sweep_config) for d in touched_dirs
+                )
 
         logger.info(
             f"   → orphan scan: {count} {mode}d"
@@ -1406,20 +1637,42 @@ def _execute_orphan_mode(
     # Empty directory cleanup
     # =========================================================================
 
-    def _clean_empty_dirs(self, base_dir: str) -> int:
-        """Remove empty directories bottom-up."""
+    def _clean_empty_dirs(
+        self, base_dir: str, config: Optional[ChubConfig] = None
+    ) -> int:
+        """Remove empty dirs bottom-up, never escaping `base_dir`; `config` confines further."""
         count = 0
+        # Trailing sep so a sibling like `<base>_old` can't pass the prefix test.
+        floor = os.path.realpath(base_dir).rstrip(os.sep) + os.sep
         for root, dirs, files in os.walk(base_dir, topdown=False):
             for dir_name in dirs:
                 dir_path = os.path.join(root, dir_name)
                 try:
-                    if not os.listdir(dir_path):
+                    if os.listdir(dir_path):
+                        continue
+                    # base_dir is the ONLY floor the bloat pass has (it passes no
+                    # config), so it is checked before the allowed-roots check.
+                    if not os.path.realpath(dir_path).startswith(floor):
+                        self._refuse(
+                            self.logger,
+                            f"Refusing to remove {dir_path}: it resolves outside "
+                            f"the tree being pruned ({base_dir})",
+                        )
+                        continue
+                    # Confine at the rmdir, not at the walk: a component swapped
+                    # to a symlink mid-sweep resolves outside the roots.
+                    if config is None:
+                        # The bloat pass passes no config, so it has no allowed
+                        # root to anchor against — the floor above is its guard.
                         os.rmdir(dir_path)
-                        count += 1
+                    elif not self._rmdir_confined(dir_path, config):
+                        continue
+                    count += 1
                 except OSError as e:
                     # Dir vanished, became non-empty, or permissions — skip,
                     # but leave a trace so the count discrepancy is explainable.
                     self.logger.debug(f"Could not remove empty dir {dir_path}: {e}")
+        self._report_refusals(self.logger, "Empty-dir sweep")
         return count
 
     # =========================================================================
diff --git a/backend/util/path_safety.py b/backend/util/path_safety.py
index bf135497..86ed15fb 100644
--- a/backend/util/path_safety.py
+++ b/backend/util/path_safety.py
@@ -255,6 +255,20 @@ def is_path_allowed(path: str, config: ChubConfig) -> bool:
     return False
 
 
+def containing_root(path: str, config: ChubConfig) -> Optional[Path]:
+    """Longest allowed root containing the already-resolved *path*, or None."""
+    # Callers walk down from this root, so the LONGEST match is the tightest
+    # anchor. os.sep suffix stops `/root_evil` matching `/root`, as above.
+    best: Optional[Path] = None
+    for root in get_allowed_roots(config):
+        base = str(root)
+        if path != base and not path.startswith(base + os.sep):
+            continue
+        if best is None or len(base) > len(str(best)):
+            best = root
+    return best
+
+
 def resolve_confined(path: str, config: ChubConfig) -> Optional[Path]:
     """Resolve *path* and return it only when the resolved target is inside an allowed root, else None."""
     if not path or not isinstance(path, str):
diff --git a/tests/conftest.py b/tests/conftest.py
index 4aed5753..63dcd438 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -74,6 +74,26 @@ def config_with_roots(tmp_path):
     return config, tmp_path
 
 
+@pytest.fixture
+def tightening_config(monkeypatch):
+    """Factory: patch load_config to authorize `allowed` for the first N calls, then nothing."""
+
+    def _install(allowed, authorized_calls=1):
+        calls = []
+
+        def _load():
+            calls.append(1)
+            config = ChubConfig()
+            if len(calls) <= authorized_calls:
+                config.poster_renamerr.source_dirs = [str(allowed)]
+            return config
+
+        monkeypatch.setattr("backend.util.config.load_config", _load)
+        return calls
+
+    return _install
+
+
 @pytest.fixture
 def instances_config():
     return InstancesConfig(
diff --git a/tests/test_poster_cleanarr_duplicates.py b/tests/test_poster_cleanarr_duplicates.py
index 97162287..7b429bc9 100644
--- a/tests/test_poster_cleanarr_duplicates.py
+++ b/tests/test_poster_cleanarr_duplicates.py
@@ -2,6 +2,7 @@
 whose {tvdb/tmdb} id matches a live media item but whose name != the item's
 canonical folder (media_cache.folder). Safety: never remove the only copy."""
 
+import os
 from types import SimpleNamespace
 
 import pytest
@@ -11,7 +12,6 @@
 from backend.util.database import ChubDB
 from backend.util.path_safety import resolve_confined
 
-
 def _logger():
     return SimpleNamespace(
         debug=lambda *a, **k: None,
@@ -420,6 +420,95 @@ def _victim_folder(outside):
     return victim
 
 
+def test_execute_stale_remove_refuses_a_swapped_parent(tmp_path, monkeypatch):
+    """A parent component swapped to a symlink between scan and rmtree must not delete the tree it now points at."""
+    allowed = tmp_path / "allowed"
+    outside = tmp_path / "outside"
+    allowed.mkdir()
+    victim = _victim_folder(outside)
+    m = _make(allowed)
+    _live(monkeypatch, allowed)
+    entry = _swap_parent_for_link(allowed, outside)
+    logger, errors = _collecting_logger()
+
+    res = m._execute_stale_mode([entry], "remove", logger)
+
+    assert res["count"] == 0
+    assert victim.exists()  # the swapped-in target is untouched
+    assert errors
+
+
+def test_execute_stale_move_refuses_a_swapped_parent(tmp_path, monkeypatch):
+    """Same swap on the move path — shutil.move must not relocate an outside folder."""
+    allowed = tmp_path / "allowed"
+    outside = tmp_path / "outside"
+    allowed.mkdir()
+    victim = _victim_folder(outside)
+    m = _make(allowed)
+    _live(monkeypatch, allowed)
+    entry = _swap_parent_for_link(allowed, outside)
+    logger, errors = _collecting_logger()
+
+    res = m._execute_stale_mode([entry], "move", logger)
+
+    assert res["count"] == 0
+    assert victim.exists()  # not dragged into the restore dir
+    assert errors
+
+
+def test_execute_stale_remove_rmtrees_through_a_parent_descriptor(
+    tmp_path, monkeypatch
+):
+    """The recursive delete is anchored to a descriptor on the confined parent, so a parent re-pointed the instant after it is opened is refused."""
+    allowed = tmp_path / "allowed"
+    outside = tmp_path / "outside"
+    allowed.mkdir()
+    victim = _victim_folder(outside)
+    shows = allowed / "shows"
+    shows.mkdir()
+    stale = shows / "Dune - Prophecy (2024) {tvdb-1}"
+    stale.mkdir()
+    (stale / "poster.jpg").write_bytes(b"x")
+    (shows / "Dune Prophecy (2024) {tvdb-1}").mkdir()  # canonical present
+    m = _make(allowed)
+    _live(monkeypatch, allowed)
+
+    real_open = os.open
+
+    def _swapping_open(path, *a, **kw):
+        """Open normally, then re-point that same dir at `outside`."""
+        fd = real_open(path, *a, **kw)
+        # Match on the basename: the walk opens each component by NAME.
+        if (
+            os.path.basename(str(path)) == shows.name
+            and shows.is_dir()
+            and not shows.is_symlink()
+        ):
+            shows.rename(allowed / "shows_real")
+            shows.symlink_to(outside, target_is_directory=True)
+        return fd
+
+    # `os` here is the same module object poster_cleanarr calls through.
+    monkeypatch.setattr(os, "open", _swapping_open)
+    logger, errors = _collecting_logger()
+    entry = {
+        "folder": str(stale),
+        "asset_dir": str(allowed),
+        "name": stale.name,
+        "canonical": "Dune Prophecy (2024) {tvdb-1}",
+        "canonical_present": True,
+        "id": ("tvdb", 1),
+        "size": 1,
+    }
+
+    res = m._execute_stale_mode([entry], "remove", logger)
+
+    assert res["count"] == 0
+    assert victim.exists()  # the swapped-in tree is never deleted through
+    assert (allowed / "shows_real" / stale.name).exists()  # nor the pinned one
+    assert errors
+
+
 def test_run_invokes_orphan_and_stale_passes(monkeypatch, tmp_path):
     """run() with mode='nothing' (skips Plex/bloat) must still invoke BOTH the
     orphan and stale passes when their config flags are set — the path a
@@ -479,3 +568,116 @@ def _sched_logger():
     m.run()
     assert "orphan" in calls
     assert "stale" in calls
+def test_stale_remove_passes_the_descriptor_to_rmtree(tmp_path, monkeypatch):
+    """The delete resolves its name against dir_fd, not a path the kernel re-walks."""
+    import shutil
+
+    allowed = tmp_path / "allowed"
+    shows = allowed / "shows"
+    stale = shows / "Dune - Prophecy (2024) {tvdb-1}"
+    stale.mkdir(parents=True)
+    (stale / "poster.jpg").write_bytes(b"x")
+    (shows / "Dune Prophecy (2024) {tvdb-1}").mkdir()
+    m = _make(allowed)
+    _live(monkeypatch, allowed)
+
+    seen = {}
+    real_rmtree = shutil.rmtree
+
+    def _recording_rmtree(path, *a, **kw):
+        seen["path"] = path
+        seen["dir_fd"] = kw.get("dir_fd")
+        return real_rmtree(path, *a, **kw)
+
+    monkeypatch.setattr(shutil, "rmtree", _recording_rmtree)
+    entry = {
+        "folder": str(stale),
+        "asset_dir": str(allowed),
+        "name": stale.name,
+        "canonical": "Dune Prophecy (2024) {tvdb-1}",
+        "canonical_present": True,
+        "id": ("tvdb", 1),
+        "size": 1,
+    }
+
+    res = m._execute_stale_mode([entry], "remove", _collecting_logger()[0])
+
+    assert res["count"] == 1
+    assert seen["dir_fd"] is not None  # anchored, not path-walked
+    assert seen["path"] == stale.name  # a bare name, resolved against the fd
+    assert not stale.exists()
+
+
+def _stale_entry(folder, asset_dir, canonical):
+    """A stale-duplicate entry for `folder`, as _scan_stale_duplicates emits it."""
+    return {
+        "folder": str(folder),
+        "asset_dir": str(asset_dir),
+        "name": os.path.basename(str(folder)),
+        "canonical": canonical,
+        "canonical_present": True,
+        "id": ("tvdb", 1),
+        "size": 1,
+    }
+
+
+def test_execute_stale_move_renames_through_both_parent_descriptors(
+    tmp_path, monkeypatch
+):
+    """The stale move is an os.rename resolved against a descriptor on each parent, not a path pair shutil re-walks."""
+    allowed = tmp_path / "allowed"
+    canonical = "Dune Prophecy (2024) {tvdb-1}"
+    (allowed / canonical).mkdir(parents=True)  # canonical present
+    stale = allowed / "Dune - Prophecy (2024) {tvdb-1}"
+    stale.mkdir()
+    (stale / "poster.jpg").write_bytes(b"x")
+    m = _make(allowed)
+    _live(monkeypatch, allowed)
+
+    calls = []
+    real_rename = os.rename
+
+    def _recording_rename(src, dst, **kw):
+        calls.append((src, dst, kw.get("src_dir_fd"), kw.get("dst_dir_fd")))
+        return real_rename(src, dst, **kw)
+
+    monkeypatch.setattr(os, "rename", _recording_rename)
+
+    res = m._execute_stale_mode(
+        [_stale_entry(stale, allowed, canonical)], "move", _logger()
+    )
+
+    assert res["count"] == 1
+    assert (allowed / ORPHAN_RESTORE_DIR_NAME / stale.name / "poster.jpg").exists()
+    assert len(calls) == 1
+    src, dst, src_fd, dst_fd = calls[0]
+    assert (src, dst) == (stale.name, stale.name)  # bare names, not paths
+    assert src_fd is not None and dst_fd is not None  # ...against both fds
+
+
+def test_execute_stale_mode_stops_when_config_stops_authorizing(
+    tmp_path, tightening_config
+):
+    """Config is re-read per item, so a root dropped mid-batch stops the tail while the head — authorized when its turn came — still ran."""
+    allowed = tmp_path / "allowed"
+    entries = []
+    for tag in ("A", "B"):
+        (allowed / f"Show {tag} (2024) {{tvdb-1}}").mkdir(parents=True)
+        old = allowed / f"Show {tag} old (2024) {{tvdb-1}}"
+        old.mkdir()
+        (old / "poster.jpg").write_bytes(b"x")
+        entries.append(_stale_entry(old, allowed, f"Show {tag} (2024) {{tvdb-1}}"))
+    m = _make(allowed)
+    calls = tightening_config(allowed, authorized_calls=1)
+    logger, errors = _collecting_logger()
+
+    res = m._execute_stale_mode(entries, "remove", logger)
+
+    assert res["count"] == 1
+    assert not os.path.exists(entries[0]["folder"])
+    assert os.path.exists(entries[1]["folder"])  # de-authorized before its turn
+    assert len(calls) >= 2  # one load per item, not one per batch
+    # Labelled by pass, so an empty-dir-sweep refusal cannot satisfy it.
+    assert [
+        "Stale-duplicate cleanup: refused 1 path(s)" in e for e in errors
+    ].count(True) == 1
diff --git a/tests/test_poster_cleanarr_orphans.py b/tests/test_poster_cleanarr_orphans.py
index ae55add2..b9a4596e 100644
--- a/tests/test_poster_cleanarr_orphans.py
+++ b/tests/test_poster_cleanarr_orphans.py
@@ -15,7 +15,8 @@
 from backend.util.database import ChubDB
 from backend.util.normalization import normalize_titles
 from backend.util.path_safety import resolve_confined
-
+import errno
+import os
 
 def _logger():
     return SimpleNamespace(
@@ -249,6 +250,327 @@ def _swap_parent_for_link(allowed, outside):
     return item
 
 
+def test_execute_orphan_mode_remove_refuses_a_swapped_parent(tmp_path, monkeypatch):
+    """A parent component swapped to a symlink between scan and os.remove must not delete the file it now points at."""
+    allowed = tmp_path / "allowed"
+    outside = tmp_path / "outside"
+    allowed.mkdir()
+    outside.mkdir()
+    victim = outside / "poster.png"
+    victim.write_bytes(b"x")
+    m = _make(allowed)
+    _live(monkeypatch, allowed)
+    item = _swap_parent_for_link(allowed, outside)
+    logger, errors = _collecting_logger()
+
+    res = m._execute_orphan_mode([item], "remove", logger)
+
+    assert res["count"] == 0
+    assert victim.exists()  # the swapped-in target is untouched
+    assert errors
+
+
+def test_execute_orphan_mode_move_refuses_a_swapped_parent(tmp_path, monkeypatch):
+    """Same swap on the move path — shutil.move must not relocate an outside file."""
+    allowed = tmp_path / "allowed"
+    outside = tmp_path / "outside"
+    allowed.mkdir()
+    outside.mkdir()
+    victim = outside / "poster.png"
+    victim.write_bytes(b"x")
+    m = _make(allowed)
+    _live(monkeypatch, allowed)
+    item = _swap_parent_for_link(allowed, outside)
+    logger, errors = _collecting_logger()
+
+    res = m._execute_orphan_mode([item], "move", logger)
+
+    assert res["count"] == 0
+    assert victim.exists()  # not dragged into the restore dir
+    assert errors
+
+
+def test_execute_orphan_mode_move_refuses_a_symlinked_restore_dir(
+    tmp_path, monkeypatch
+):
+    """The move destination is confined too: a restore dir linking outside the roots must not receive the file."""
+    allowed = tmp_path / "allowed"
+    outside = tmp_path / "outside"
+    allowed.mkdir()
+    outside.mkdir()
+    (allowed / ORPHAN_RESTORE_DIR_NAME).symlink_to(outside, target_is_directory=True)
+    f = allowed / "orphan.png"
+    f.write_bytes(b"x")
+    item = {"path": str(f), "size": 1, "parsed": "orphan", "asset_dir": str(allowed)}
+    m = _make(allowed)
+    _live(monkeypatch, allowed)
+    logger, errors = _collecting_logger()
+
+    res = m._execute_orphan_mode([item], "move", logger)
+
+    assert res["count"] == 0
+    assert f.exists()  # left in place
+    assert not (outside / "orphan.png").exists()  # nothing written outside
+    assert errors
+
+
+def test_clean_empty_dirs_refuses_a_dir_outside_the_roots(tmp_path, monkeypatch):
+    """The empty-dir sweep re-confines each dir: an asset dir swapped to a link must not have the outside tree pruned through it."""
+    allowed = tmp_path / "allowed"
+    outside = tmp_path / "outside"
+    allowed.mkdir()
+    outside.mkdir()
+    stray = outside / "empty"
+    stray.mkdir()
+    cfg = _live(monkeypatch, allowed)
+    swapped = allowed / "assets"
+    swapped.symlink_to(outside, target_is_directory=True)
+    m = _make(allowed)
+    m.logger, errors = _collecting_logger()
+
+    assert m._clean_empty_dirs(str(swapped), cfg) == 0
+    assert stray.exists()  # pruning stops at the roots
+    assert errors
+
+
+def test_clean_empty_dirs_refuses_a_dir_outside_base_dir_without_a_config(
+    tmp_path, monkeypatch
+):
+    """The bloat pass prunes with no config at all, so base_dir is the only floor: a component swapped mid-sweep must not have rmdir reach past it."""
+    base = tmp_path / "metadata"
+    show = base / "Show"
+    (show / "empty").mkdir(parents=True)
+    outside = tmp_path / "outside"
+    (outside / "empty").mkdir(parents=True)
+    m = _make()
+    m.logger, errors = _collecting_logger()
+
+    real_listdir = os.listdir
+    swapped = []
+
+    def _swapping_listdir(path, *a, **kw):
+        """Swap `Show` for a link to `outside` between the walk and the rmdir."""
+        if not swapped and str(path) == str(show / "empty"):
+            swapped.append(True)
+            show.rename(base / "Show_real")
+            show.symlink_to(outside, target_is_directory=True)
+        return real_listdir(path, *a, **kw)
+
+    monkeypatch.setattr(os, "listdir", _swapping_listdir)
+
+    assert m._clean_empty_dirs(str(base)) == 0
+    assert swapped  # the race window was actually entered
+    assert (outside / "empty").is_dir()  # pruning stops at base_dir
+    assert errors
+
+
+def test_clean_empty_dirs_still_prunes_inside_base_dir_without_a_config(tmp_path):
+    """The floor must not disable the bloat pass, which passes no config."""
+    base = tmp_path / "metadata"
+    (base / "Show" / "empty").mkdir(parents=True)
+    m = _make()
+
+    assert m._clean_empty_dirs(str(base)) == 2
+    assert not (base / "Show").exists()  # pruned bottom-up, base kept
+
+
+# ── Descriptor-bound deletion (the race re-confinement only narrows) ─────────
+
+
+def _orphan_item(path, asset_dir):
+    """An orphan entry for `path`, as _scan_orphan_assets would emit it."""
+    return {
+        "path": str(path),
+        "size": 1,
+        "parsed": "orphan",
+        "asset_dir": str(asset_dir),
+    }
+
+
+def test_execute_orphan_mode_remove_unlinks_through_a_parent_descriptor(
+    tmp_path, monkeypatch
+):
+    """The delete goes through os.unlink(name, dir_fd=...) on the confined parent — a bare os.remove(path) would resolve the path a second time."""
+    allowed = tmp_path / "allowed"
+    allowed.mkdir()
+    f = allowed / "orphan.png"
+    f.write_bytes(b"x")
+    m = _make(allowed)
+    _live(monkeypatch, allowed)
+
+    calls = []
+    real_unlink = os.unlink
+
+    def _recording_unlink(path, *a, dir_fd=None, **kw):
+        """Record the name/descriptor pair, then delegate to the real unlink."""
+        calls.append((path, dir_fd))
+        return real_unlink(path, *a, dir_fd=dir_fd, **kw)
+
+    # `os` here is the same module object poster_cleanarr calls through.
+    monkeypatch.setattr(os, "unlink", _recording_unlink)
+    res = m._execute_orphan_mode([_orphan_item(f, allowed)], "remove", _logger())
+
+    assert res["count"] == 1
+    assert not f.exists()
+    assert [name for name, _fd in calls] == ["orphan.png"]  # the NAME, not the path
+    assert calls[0][1] is not None  # ...resolved against a descriptor
+
+
+def test_execute_orphan_mode_remove_refuses_a_parent_swapped_after_the_check(
+    tmp_path, monkeypatch
+):
+    """The parent is re-pointed at an outside dir the instant after it is opened: the descriptor pin must refuse rather than delete through the new path."""
+    allowed = tmp_path / "allowed"
+    outside = tmp_path / "outside"
+    allowed.mkdir()
+    outside.mkdir()
+    victim = outside / "poster.png"
+    victim.write_bytes(b"x")
+    show = allowed / "Show"
+    show.mkdir()
+    f = show / "poster.png"
+    f.write_bytes(b"x")
+    m = _make(allowed)
+    _live(monkeypatch, allowed)
+
+    real_open = os.open
+
+    def _swapping_open(path, *a, **kw):
+        """Open normally, then re-point that same dir at `outside`."""
+        fd = real_open(path, *a, **kw)
+        # Match on the basename: the walk opens each component by NAME.
+        if (
+            os.path.basename(str(path)) == show.name
+            and show.is_dir()
+            and not show.is_symlink()
+        ):
+            show.rename(allowed / "Show_real")
+            show.symlink_to(outside, target_is_directory=True)
+        return fd
+
+    monkeypatch.setattr(os, "open", _swapping_open)
+    logger, errors = _collecting_logger()
+    res = m._execute_orphan_mode([_orphan_item(f, allowed)], "remove", logger)
+
+    assert res["count"] == 0
+    assert victim.exists()  # the swapped-in dir is never deleted through
+    assert (allowed / "Show_real" / "poster.png").exists()  # nor the pinned one
+    assert errors
+
+
+def test_execute_orphan_mode_remove_refuses_an_intermediate_component_swapped_after_confinement(
+    tmp_path, monkeypatch
+):
+    """An INTERMEDIATE component becomes a link once confinement has passed."""
+    allowed = tmp_path / "allowed"
+    outside = tmp_path / "outside"
+    lib = allowed / "lib"
+    show = lib / "Show"
+    show.mkdir(parents=True)
+    f = show / "poster.png"
+    f.write_bytes(b"x")
+    victim = outside / "Show" / "poster.png"
+    victim.parent.mkdir(parents=True)
+    victim.write_bytes(b"x")
+    m = _make(allowed)
+    _live(monkeypatch, allowed)
+
+    real_open = os.open
+    swapped = []
+
+    def _swapping_open(path, *a, **kw):
+        """Swap `lib` for a link to `outside` in the check→open window."""
+        if not swapped and str(path).startswith(os.path.realpath(allowed)):
+            swapped.append(True)
+            lib.rename(allowed / "lib_real")
+            lib.symlink_to(outside, target_is_directory=True)
+        return real_open(path, *a, **kw)
+
+    monkeypatch.setattr(os, "open", _swapping_open)
+    logger, errors = _collecting_logger()
+
+    res = m._execute_orphan_mode([_orphan_item(f, allowed)], "remove", logger)
+
+    assert swapped  # the race window was actually entered
+    assert res["count"] == 0
+    assert victim.exists()  # the swapped-in tree is never deleted through
+    assert (allowed / "lib_real" / "Show" / "poster.png").exists()  # nor the real one
+    assert errors
+
+
+def test_execute_orphan_mode_remove_deletes_the_link_not_its_target(
+    tmp_path, monkeypatch
+):
+    """An orphaned symlink is unlinked by name; resolving it first would delete the artwork it points at."""
+    allowed = tmp_path / "allowed"
+    allowed.mkdir()
+    target = allowed / "keep.png"
+    target.write_bytes(b"x")
+    link = allowed / "orphan.png"
+    link.symlink_to(target)
+    m = _make(allowed)
+    _live(monkeypatch, allowed)
+
+    res = m._execute_orphan_mode([_orphan_item(link, allowed)], "remove", _logger())
+
+    assert res["count"] == 1
+    assert not link.is_symlink()  # the link entry is gone
+    assert target.exists()  # its target survives
+
+
+def test_execute_orphan_mode_remove_sweeps_dead_links_only_inside_the_roots(
+    tmp_path, monkeypatch
+):
+    """Behaviour change to keep visible: confinement resolves a link's TARGET, so the dead-link sweep still clears links into a root but now refuses links pointing at an unconfigured directory."""
+    allowed = tmp_path / "allowed"
+    outside = tmp_path / "outside"
+    allowed.mkdir()
+    outside.mkdir()
+    inside_link = allowed / "gone-inside.png"
+    inside_link.symlink_to(allowed / "missing.png")
+    outside_link = allowed / "gone-outside.png"
+    outside_link.symlink_to(outside / "missing.png")
+    m = _make(allowed)
+    _live(monkeypatch, allowed)
+    logger, errors = _collecting_logger()
+
+    res = m._execute_orphan_mode(
+        [_orphan_item(inside_link, allowed), _orphan_item(outside_link, allowed)],
+        "remove",
+        logger,
+    )
+
+    assert res["count"] == 1
+    assert not inside_link.is_symlink()  # dead link into a root: still swept
+    assert outside_link.is_symlink()  # dead link out of the roots: refused
+    assert errors
+
+
+def test_execute_orphan_mode_aggregates_refusals_into_one_error(tmp_path, monkeypatch):
+    """Links into unconfigured storage are retained and re-refused every pass, so the refusals collapse to one summary line; detail moves to debug."""
+    allowed = tmp_path / "allowed"
+    outside = tmp_path / "outside"
+    allowed.mkdir()
+    outside.mkdir()
+    items = []
+    for i in range(5):
+        link = allowed / f"gone-{i}.png"
+        link.symlink_to(outside / f"missing-{i}.png")
+        items.append(_orphan_item(link, allowed))
+    m = _make(allowed)
+    _live(monkeypatch, allowed)
+    logger, errors = _collecting_logger()
+    debugs = []
+    logger.debug = debugs.append
+
+    res = m._execute_orphan_mode(items, "remove", logger)
+
+    assert res["count"] == 0
+    assert len(errors) == 1  # one line per pass, not one per link
+    assert "refused 5 path(s)" in errors[0]
+    assert len(debugs) == 5  # per-link detail retained at debug
+
+
 def _write(tmp_path, name):
     f = tmp_path / name
     f.write_bytes(b"x")
@@ -418,6 +740,188 @@ def test_resolve_orphan_instances_handles_missing_attrs():
     assert PosterCleanarr._resolve_orphan_instances(cfg) == []
 
 
+# ── Anchored moves, anchored rmdir, per-item authorization ──────────────────
+
+
+def test_execute_orphan_mode_move_renames_through_both_parent_descriptors(
+    tmp_path, monkeypatch
+):
+    """The move is an os.rename resolved against a descriptor on each parent, not a path pair shutil re-walks."""
+    allowed = tmp_path / "allowed"
+    allowed.mkdir()
+    f = allowed / "orphan.png"
+    f.write_bytes(b"x")
+    m = _make(allowed)
+    _live(monkeypatch, allowed)
+
+    calls = []
+    real_rename = os.rename
+
+    def _recording_rename(src, dst, **kw):
+        calls.append((src, dst, kw.get("src_dir_fd"), kw.get("dst_dir_fd")))
+        return real_rename(src, dst, **kw)
+
+    monkeypatch.setattr(os, "rename", _recording_rename)
+
+    res = m._execute_orphan_mode([_orphan_item(f, allowed)], "move", _logger())
+
+    assert res["count"] == 1
+    assert (allowed / ORPHAN_RESTORE_DIR_NAME / "orphan.png").exists()
+    assert len(calls) == 1
+    src, dst, src_fd, dst_fd = calls[0]
+    assert (src, dst) == ("orphan.png", "orphan.png")  # bare names, not paths
+    assert src_fd is not None and dst_fd is not None  # ...against both fds
+
+
+def test_execute_orphan_mode_move_refuses_a_parent_swapped_after_the_check(
+    tmp_path, monkeypatch
+):
+    """The source parent becomes a link to an outside dir the instant after confinement passes: the move must refuse, not relocate the outside file."""
+    allowed = tmp_path / "allowed"
+    outside = tmp_path / "outside"
+    allowed.mkdir()
+    outside.mkdir()
+    victim = outside / "poster.png"
+    victim.write_bytes(b"x")
+    show = allowed / "Show"
+    show.mkdir()
+    f = show / "poster.png"
+    f.write_bytes(b"x")
+    m = _make(allowed)
+    _live(monkeypatch, allowed)
+    logger, errors = _collecting_logger()
+
+    real_confined = m._confined_target
+    swapped = []
+
+    def _swapping_confined(path, config, log):
+        """Swap `Show` for a link to `outside` once the file itself is confined."""
+        result = real_confined(path, config, log)
+        if not swapped and str(path) == str(f):
+            swapped.append(True)
+            show.rename(allowed / "Show_real")
+            show.symlink_to(outside, target_is_directory=True)
+        return result
+
+    monkeypatch.setattr(m, "_confined_target", _swapping_confined)
+
+    res = m._execute_orphan_mode([_orphan_item(f, allowed)], "move", logger)
+
+    assert swapped  # the race window was actually entered
+    assert res["count"] == 0
+    assert victim.exists()  # never dragged into the restore dir
+    assert (allowed / "Show_real" / "poster.png").exists()  # nor the pinned one
+    assert errors
+
+
+def test_execute_orphan_mode_move_falls_back_across_devices(tmp_path, monkeypatch):
+    """EXDEV is the one case the anchored rename can't serve: it must fall back to the copy path, not drop the move."""
+    allowed = tmp_path / "allowed"
+    allowed.mkdir()
+    f = allowed / "orphan.png"
+    f.write_bytes(b"x")
+    m = _make(allowed)
+    _live(monkeypatch, allowed)
+
+    def _cross_device(*a, **kw):
+        raise OSError(errno.EXDEV, "Invalid cross-device link")
+
+    # shutil.move's own os.rename is patched too, so it takes its copy branch.
+    monkeypatch.setattr(os, "rename", _cross_device)
+    logger, errors = _collecting_logger()
+
+    res = m._execute_orphan_mode([_orphan_item(f, allowed)], "move", logger)
+
+    assert res["count"] == 1
+    assert not f.exists()
+    assert (allowed / ORPHAN_RESTORE_DIR_NAME / "orphan.png").exists()
+    assert not errors  # a cross-device move is normal, not a refusal
+
+
+def test_clean_empty_dirs_rmdirs_through_a_parent_descriptor(tmp_path, monkeypatch):
+    """With a config the prune goes through os.rmdir(name, dir_fd=...) — a bare os.rmdir(path) would re-walk every component."""
+    allowed = tmp_path / "allowed"
+    (allowed / "Show" / "empty").mkdir(parents=True)
+    cfg = _live(monkeypatch, allowed)
+    m = _make(allowed)
+    m.logger, errors = _collecting_logger()
+
+    calls = []
+    real_rmdir = os.rmdir
+
+    def _recording_rmdir(path, *a, dir_fd=None, **kw):
+        calls.append((path, dir_fd))
+        return real_rmdir(path, *a, dir_fd=dir_fd, **kw)
+
+    monkeypatch.setattr(os, "rmdir", _recording_rmdir)
+
+    assert m._clean_empty_dirs(str(allowed), cfg) == 2
+    assert not (allowed / "Show").exists()  # pruned bottom-up, root kept
+    assert [name for name, _fd in calls] == ["empty", "Show"]  # names, not paths
+    assert all(fd is not None for _name, fd in calls)
+    assert not errors
+
+
+def test_clean_empty_dirs_refuses_a_parent_swapped_after_the_check(
+    tmp_path, monkeypatch
+):
+    """A parent swapped to a link in the confine→rmdir window must not have the outside tree pruned through it."""
+    allowed = tmp_path / "allowed"
+    outside = tmp_path / "outside"
+    (outside / "empty").mkdir(parents=True)
+    base = allowed / "assets"
+    show = base / "Show"
+    (show / "empty").mkdir(parents=True)
+    cfg = _live(monkeypatch, allowed)
+    m = _make(allowed)
+    m.logger, errors = _collecting_logger()
+
+    real_confined = m._confined_target
+    swapped = []
+
+    def _swapping_confined(path, config, log):
+        """Swap `Show` for a link to `outside` once its child dir is confined."""
+        result = real_confined(path, config, log)
+        if not swapped and str(path) == str(show / "empty"):
+            swapped.append(True)
+            show.rename(base / "Show_real")
+            show.symlink_to(outside, target_is_directory=True)
+        return result
+
+    monkeypatch.setattr(m, "_confined_target", _swapping_confined)
+
+    assert m._clean_empty_dirs(str(base), cfg) == 0
+    assert swapped  # the race window was actually entered
+    assert (outside / "empty").is_dir()  # pruning never reached through the link
+    assert errors
+
+
+def test_execute_orphan_mode_stops_when_config_stops_authorizing(
+    tmp_path, tightening_config
+):
+    """Config is re-read per item, so a root dropped mid-batch stops the tail while the head — authorized when its turn came — still ran."""
+    allowed = tmp_path / "allowed"
+    allowed.mkdir()
+    head = allowed / "head.png"
+    head.write_bytes(b"x")
+    tail = allowed / "tail.png"
+    tail.write_bytes(b"x")
+    m = _make(allowed)
+    calls = tightening_config(allowed)
+    logger, errors = _collecting_logger()
+
+    res = m._execute_orphan_mode(
+        [_orphan_item(head, allowed), _orphan_item(tail, allowed)], "remove", logger
+    )
+
+    assert res["count"] == 1
+    assert not head.exists()
+    assert tail.exists()  # de-authorized before its turn
+    assert len(calls) >= 2  # one load per item, not one per batch
+    # Labelled by pass, so an empty-dir-sweep refusal cannot satisfy it.
+    assert ["Orphan cleanup: refused 1 path(s)" in e for e in errors].count(True) == 1
+
+
 def test_build_library_id_sets_filters_by_instance(db):
     m = _make()
     db.media.execute_query(