fix: enforce native Windows bundle retention safely

CHANGELOG.md

@@ -31,6 +31,7 @@ and versions are tracked in the repo-root `VERSION` file.
 
 ### Fixed
 
+- Enforce native Windows run-bundle retention with pinned directory handles; unsupported platforms fail closed once per pass (#378).
 - Preserve consumer-owned logging handlers, explicit levels, and parent routing across CLI invocations (#387).
 - Validate nested configuration mappings before merge/provenance traversal,
   reject recursive or excessively deep values with source-aware errors, and

docs/platform-support.md

@@ -30,9 +30,9 @@ no-follow directory operations. Linux, macOS, and WSL2 provide those
 primitives; the empty leaf is retained on every platform because portable
 POSIX has no identity-bound `rmdir`. Empty nested directories and ancestors are
 retained for the same reason. Linux additionally requires readable mount IDs
-and fails closed if they are unavailable. Native Windows currently uses the
-secure fallback: it retains both directories and files and emits a cleanup
-warning rather than perform race-prone pathname recursion.
+and fails closed if they are unavailable. Native Windows uses pinned directory
+handles and removes reparse-point children only as leaves; it never descends
+through a junction or symlink.
 
 The supported Python range is Python 3.10 through 3.14. Bug reports should
 include the operating system, distribution or WSL version when relevant,
@@ -79,3 +79,15 @@ contract, which covers an in-use destination reported as `winerror` 5; other
 access-denied and permanent permission/path errors fail immediately. Transient
 retries are bounded by a one-second elapsed deadline; the destination remains
 untouched if that deadline is exhausted.
+
+### Run-bundle retention
+
+POSIX retention uses descriptor-relative no-follow directory operations. Native
+Windows uses directory handles that deny rename/delete and conflicting writes,
+pins every ancestor while descending, removes reparse-point leaves (including
+junctions) without following them, and refuses volume crossings and changed
+directory identities. Active leases
+and metadata preservation checks still apply before removal. Sharing violations
+leave the bundle for a later pass. Other platforms without safe primitives skip
+retention with one actionable warning per pass; they never use pathname recursion.
+The native runtime matrix verifies that repeated invocations enforce `max_bundles`.

lib/python/base_cli/_runtime.py

@@ -295,6 +295,8 @@ def _open_absolute_directory(path: Path) -> int:
 
 
 def _directory_open_flags() -> int:
+    if not hasattr(os, "O_DIRECTORY") or not hasattr(os, "O_NOFOLLOW"):
+        raise OSError("safe descriptor-relative directory operations are unavailable")
     return os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0)
 
 
@@ -405,6 +407,9 @@ def prune_run_bundles(
     runs_root = Path(runs_root)
     if not runs_root.exists() or runs_root.is_symlink():
         return
+    if not _supports_fd_relative_bundle_removal() and os.name != "nt":
+        log.warning("Run bundle retention unavailable on this platform: safe directory removal is unsupported.")
+        return
     protected = {_safe_resolved_path(path) for path in protected_run_roots}
     if current_run_root is not None:
         protected.add(_safe_resolved_path(current_run_root))
@@ -831,9 +836,25 @@ def _bundle_size(path: Path) -> int:
     return total
 
 
+def _supports_fd_relative_bundle_removal() -> bool:
+    return (
+        hasattr(os, "O_DIRECTORY")
+        and hasattr(os, "O_NOFOLLOW")
+        and {os.open, os.stat, os.unlink, os.rmdir}.issubset(os.supports_dir_fd)
+        and os.scandir in os.supports_fd
+    )
+
+
 def _remove_run_bundle(runs_root: Path, path: Path) -> None:
     """Remove one direct child using descriptor-relative, no-follow operations."""
 
+    if not _supports_fd_relative_bundle_removal():
+        if os.name != "nt":
+            raise OSError("safe bundle removal is unsupported on this platform")
+        from ._windows_retention import remove_bundle
+
+        remove_bundle(runs_root, path)
+        return
     root_fd = _open_directory_nofollow(runs_root)
     try:
         candidate = Path(path).name

lib/python/base_cli/_windows_retention.py

@@ -0,0 +1,100 @@
+"""Windows bundle removal with pinned, non-reparse directory components."""
+
+from __future__ import annotations
+
+import ctypes
+import os
+import stat
+from collections.abc import Iterator
+from contextlib import ExitStack, contextmanager
+from pathlib import Path
+
+
+def _check_directory(path: Path, volume: int | None = None) -> os.stat_result:
+    current = path.lstat()
+    if not stat.S_ISDIR(current.st_mode) or getattr(current, "st_file_attributes", 0) & 0x400:
+        raise OSError(f"refusing non-directory or reparse point '{path}'")
+    if volume is not None and current.st_dev != volume:
+        raise OSError(f"refusing volume boundary at '{path}'")
+    return current
+
+
+@contextmanager
+def _pin_directory(path: Path, volume: int | None = None) -> Iterator[os.stat_result]:
+    # OPEN_REPARSE_POINT + BACKUP_SEMANTICS opens the directory itself. Omitting
+    # SHARE_DELETE/SHARE_WRITE prevents rename/deletion and conflicting writers
+    # while we inspect and descend. Keep every ancestor pinned until completion.
+    # https://learn.microsoft.com/windows/win32/api/fileapi/nf-fileapi-createfilew
+    from ctypes import wintypes
+
+    kernel = ctypes.WinDLL("kernel32", use_last_error=True)  # type: ignore[attr-defined]
+    create = kernel.CreateFileW
+    create.argtypes = [
+        wintypes.LPCWSTR,
+        wintypes.DWORD,
+        wintypes.DWORD,
+        ctypes.c_void_p,
+        wintypes.DWORD,
+        wintypes.DWORD,
+        wintypes.HANDLE,
+    ]
+    create.restype = wintypes.HANDLE
+    close = kernel.CloseHandle
+    close.argtypes = [wintypes.HANDLE]
+    close.restype = wintypes.BOOL
+    before = _check_directory(path, volume)
+    # FILE_LIST_DIRECTORY requests directory data access, so Windows enforces
+    # the share mode instead of treating this as an attribute-only probe.
+    handle = create(str(path), 0x1, 0x1, None, 3, 0x02200000, None)
+    if handle == ctypes.c_void_p(-1).value:
+        raise OSError(ctypes.get_last_error(), f"cannot pin retention directory '{path}'")  # type: ignore[attr-defined]
+    try:
+        current = _check_directory(path, volume)
+        if (before.st_dev, before.st_ino) != (current.st_dev, current.st_ino):
+            raise OSError(f"retention directory identity changed: '{path}'")
+        yield current
+    finally:
+        close(handle)
+
+
+def _remove_tree(path: Path, volume: int) -> None:
+    with _pin_directory(path, volume):
+        with os.scandir(path) as entries:
+            for entry in entries:
+                child = path / entry.name
+                current = child.lstat()
+                if current.st_dev != volume:
+                    raise OSError(f"refusing volume boundary '{child}'")
+                if getattr(current, "st_file_attributes", 0) & 0x400:
+                    # Reparse points are removed as leaves. Never descend
+                    # through a junction or symlink, but do not strand an
+                    # otherwise removable bundle because it contains one.
+                    if stat.S_ISDIR(current.st_mode):
+                        child.rmdir()
+                    else:
+                        child.unlink()
+                    continue
+                if stat.S_ISDIR(current.st_mode):
+                    _remove_tree(child, volume)
+                else:
+                    # unlink never follows a replacement symlink. All ancestors
+                    # remain pinned; a replacement directory makes unlink fail.
+                    child.unlink()
+    # The handle must close before rmdir. This operation only removes an empty
+    # directory (or the junction itself); it cannot descend into a replacement.
+    path.rmdir()
+
+
+def remove_bundle(runs_root: Path, path: Path) -> None:
+    root = Path(os.path.abspath(runs_root))
+    target = Path(os.path.abspath(path))
+    if target.parent != root or target.name in {"", ".", ".."}:
+        raise OSError(f"refusing bundle outside '{root}'")
+    anchor = Path(root.anchor)
+    with ExitStack() as stack:
+        volume = stack.enter_context(_pin_directory(anchor)).st_dev
+        parent = anchor
+        for component in root.relative_to(anchor).parts:
+            parent = parent / component
+            stack.enter_context(_pin_directory(parent, volume))
+        _remove_tree(target, volume)

tests/test_retention_platform.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from unittest.mock import Mock, patch
+
+import base_cli
+import pytest
+from base_cli import _runtime as runtime
+from base_cli.testing import invoke
+
+
+def test_missing_directory_primitives_warns_once_and_command_succeeds(tmp_path: Path) -> None:
+    app = base_cli.App(name="unsupported-retention", max_run_bundles=1)
+
+    @app.command()
+    def main(ctx: base_cli.Context) -> None:
+        pass
+
+    # Keep native Windows on its real audited fallback; model an unsupported
+    # POSIX-like platform without directory flags or descriptor-relative calls.
+    if os.name == "nt":
+        pytest.skip("unsupported platform model is POSIX-only")
+    with (
+        patch.object(os, "supports_dir_fd", set()),
+        patch.object(runtime, "_supports_fd_relative_bundle_removal", return_value=False),
+    ):
+        result = invoke(app, [], home=tmp_path)
+    assert result.exit_code == 0
+    assert result.stderr.count("retention unavailable on this platform") == 1
+
+
+def test_missing_flags_raise_oserror_not_attributeerror() -> None:
+    with patch.dict(os.__dict__):
+        os.__dict__.pop("O_DIRECTORY", None)
+        os.__dict__.pop("O_NOFOLLOW", None)
+        with pytest.raises(OSError, match="unavailable"):
+            runtime._directory_open_flags()
+
+
+def test_native_count_bound_is_enforced(tmp_path: Path) -> None:
+    app = base_cli.App(name="native-retention", max_run_bundles=2)
+    roots = []
+
+    @app.command()
+    def main(ctx: base_cli.Context) -> None:
+        roots.append(ctx.run_root)
+
+    for _ in range(6):
+        result = invoke(app, [], home=tmp_path)
+        assert result.exit_code == 0, result.output
+    assert sum(root.exists() for root in roots) <= 2
+
+
+def test_unsupported_pass_warns_only_once(tmp_path: Path) -> None:
+    if os.name == "nt":
+        pytest.skip("Windows has a native fallback")
+    logger = Mock()
+    with patch.object(runtime, "_supports_fd_relative_bundle_removal", return_value=False):
+        runtime.prune_run_bundles(tmp_path, max_bundles=1, logger=logger)
+    assert logger.warning.call_count == 1

tests/test_windows_retention.py

@@ -0,0 +1,43 @@
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+import pytest
+from base_cli._windows_retention import _pin_directory, remove_bundle
+
+pytestmark = pytest.mark.skipif(os.name != "nt", reason="Windows retention contract")
+
+
+def test_pinned_directory_rejects_concurrent_rename(tmp_path: Path) -> None:
+    runs_root = tmp_path / "runs"
+    runs_root.mkdir()
+    bundle = runs_root / "bundle"
+    bundle.mkdir()
+    replacement = runs_root / "replacement"
+
+    with _pin_directory(bundle):
+        with pytest.raises(OSError):
+            os.rename(bundle, replacement)
+
+    assert bundle.is_dir()
+    assert not replacement.exists()
+
+
+def test_reparse_point_child_is_removed_as_a_leaf(tmp_path: Path) -> None:
+    runs_root = tmp_path / "runs"
+    runs_root.mkdir()
+    bundle = runs_root / "bundle"
+    bundle.mkdir()
+    external = tmp_path / "external"
+    external.mkdir()
+    link = bundle / "linked-directory"
+    try:
+        link.symlink_to(external, target_is_directory=True)
+    except (OSError, NotImplementedError) as exc:
+        pytest.skip(f"directory symlinks unavailable: {exc}")
+
+    remove_bundle(runs_root, bundle)
+
+    assert not bundle.exists()
+    assert external.is_dir()