fix(dwh): anchor generation operations to lease fd

.superpowers/sdd/evidence-task-6-report.md

@@ -0,0 +1,20 @@
+# Evidence Task 6 — final fd-anchored DWH correction
+
+All DWH generation state below `.tht-dwh` is now accessed relative to the directory descriptor
+retained by the shared/exclusive generation lease. ACTIVE reads, atomic temp writes, replacement,
+fsync, and rollback use `openat`/`replaceat` operations. Generation staging, validation,
+reconciliation, resume checks, retention classification, and recursive deletion likewise use owned
+root/generations/candidate descriptors with `O_NOFOLLOW`; locked operations no longer reopen
+generation paths through `workspace_root`.
+
+Portable reader snapshots are copied from validated generation file descriptors into private 0700
+process-owned temporary directories while the shared lease is held. This avoids Linux-only
+`/proc/self/fd` paths and prevents a renamed/replaced `.tht-dwh` pathname from redirecting later
+schema or LSH reads. Lease-scoped copies are removed on exit and standalone snapshots are removed
+at process exit.
+
+Deterministic adversarial tests rename the DWH root after lease acquisition during ACTIVE reads,
+ACTIVE publication, and retention cleanup. Each test proves the replacement tree is never read,
+written, or deleted; the descriptor-pinned original either completes consistently or fails closed.
+Existing owner binding, legacy rejection, crash reconciliation, resume, atomic rollback, retention,
+and reader/writer exclusion behavior remains covered.

harness/tests/test_dwh_preprocess_job.py

@@ -445,6 +445,8 @@ def test_unsafe_lsh_filename_is_rejected(tmp_path):
 
 
 def test_active_fsync_failure_restores_previous_pointer(monkeypatch, tmp_path):
+    import os
+    import tht.jobs.dwh_pipeline as module
     def build(physical, output):
         for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json"):
             (output / name).write_text(name)
@@ -455,27 +457,147 @@ def test_active_fsync_failure_restores_previous_pointer(monkeypatch, tmp_path):
         introspect=lambda output: output.write_text("old"), build_lsh=build,
     )
     first = first_pipeline.run()
-    original_fsync = first_pipeline._fsync
+    root = tmp_path / ".tht-dwh"
+    root_identity = (root.stat().st_dev, root.stat().st_ino)
+    original_fsync = module.os.fsync
     failed_once = False
 
-    def fail_active_once(path):
+    def fail_active_once(fd):
         nonlocal failed_once
-        if path.name == ".tht-dwh" and not failed_once:
+        info = os.fstat(fd)
+        if (
+            (info.st_dev, info.st_ino) == root_identity
+            and "ACTIVE" in os.listdir(fd)
+            and not failed_once
+        ):
             failed_once = True
             raise OSError("injected directory fsync failure")
-        original_fsync(path)
+        original_fsync(fd)
 
     second = DwhPreprocessPipeline(
         workspace_id="demo", workspace_root=tmp_path,
         config_fingerprint=FP, input_fingerprint=FP,
         introspect=lambda output: output.write_text("new"), build_lsh=build,
     )
-    monkeypatch.setattr(second, "_fsync", fail_active_once)
+    monkeypatch.setattr(module.os, "fsync", fail_active_once)
     failed = second.run()
     assert failed.status == "failed"
     assert (tmp_path / ".tht-dwh" / "ACTIVE").read_text().strip() == first.run_id
 
 
+def test_snapshot_root_swap_after_lease_never_reads_replacement(monkeypatch, tmp_path):
+    import tht.jobs.dwh_pipeline as module
+
+    pipeline = DwhPreprocessPipeline(
+        workspace_id="demo", workspace_root=tmp_path,
+        config_fingerprint=FP, input_fingerprint=FP,
+        introspect=lambda output: output.write_text("trusted"),
+        build_lsh=lambda physical, output: [
+            (output / name).write_text("trusted")
+            for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json")
+        ],
+    )
+    first = pipeline.run()
+    assert first.status == "succeeded"
+    root = tmp_path / ".tht-dwh"
+    moved = tmp_path / "moved-read-root"
+    replacement = root
+    real_read = module._read_owned_at
+    swapped = False
+
+    def swapping_read(directory_fd, name, *, readonly):
+        nonlocal swapped
+        if name == "ACTIVE" and not swapped:
+            swapped = True
+            replacement.rename(moved)
+            replacement.mkdir(mode=0o700)
+            (replacement / "sentinel").write_text("replacement-secret")
+        return real_read(directory_fd, name, readonly=readonly)
+
+    monkeypatch.setattr(module, "_read_owned_at", swapping_read)
+    try:
+        with lease_dwh_snapshot(snapshot_config(tmp_path)) as snapshot:
+            assert snapshot.physical.read_text() == "trusted"
+    except Exception as error:
+        assert "ACTIVE" in str(error) or "root" in str(error)
+    assert swapped
+    assert (replacement / "sentinel").read_text() == "replacement-secret"
+
+
+def test_publish_root_swap_after_lease_never_writes_replacement(monkeypatch, tmp_path):
+    import tht.jobs.dwh_pipeline as module
+
+    def make(content):
+        return DwhPreprocessPipeline(
+            workspace_id="demo", workspace_root=tmp_path,
+            config_fingerprint=FP, input_fingerprint=FP,
+            introspect=lambda output: output.write_text(content),
+            build_lsh=lambda physical, output: [
+                (output / name).write_text(content)
+                for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json")
+            ],
+        )
+
+    first = make("old").run()
+    assert first.status == "succeeded", first
+    root = tmp_path / ".tht-dwh"
+    moved = tmp_path / "moved-publish-root"
+    real_replace = module.os.replace
+    swapped = False
+
+    def swapping_replace(source, destination, *args, **kwargs):
+        nonlocal swapped
+        if destination == "ACTIVE" and kwargs.get("dst_dir_fd") is not None and not swapped:
+            swapped = True
+            root.rename(moved)
+            root.mkdir(mode=0o700)
+            (root / "sentinel").write_text("replacement-safe")
+        return real_replace(source, destination, *args, **kwargs)
+
+    monkeypatch.setattr(module.os, "replace", swapping_replace)
+    result = make("new").run()
+    assert result.status in {"succeeded", "failed"}
+    assert swapped
+    assert (root / "sentinel").read_text() == "replacement-safe"
+    moved_active = (moved / "ACTIVE").read_text().strip()
+    assert len(moved_active) == 32
+    assert (moved / "generations" / moved_active).is_dir()
+
+
+def test_cleanup_root_swap_after_lease_never_deletes_replacement(monkeypatch, tmp_path):
+    import tht.jobs.dwh_pipeline as module
+
+    pipeline = DwhPreprocessPipeline(
+        workspace_id="demo", workspace_root=tmp_path,
+        config_fingerprint=FP, input_fingerprint=FP,
+        introspect=lambda output: output.write_text("trusted"),
+        build_lsh=lambda physical, output: [
+            (output / name).write_text("trusted")
+            for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json")
+        ],
+        retain_generations=1,
+    )
+    pipeline.run()
+    root = tmp_path / ".tht-dwh"
+    moved = tmp_path / "moved-cleanup-root"
+    real_open = module.os.open
+    swapped = False
+
+    def swapping_open(path, flags, *args, **kwargs):
+        nonlocal swapped
+        if path == "generations" and kwargs.get("dir_fd") is not None and not swapped:
+            swapped = True
+            root.rename(moved)
+            root.mkdir(mode=0o700)
+            (root / "sentinel").write_text("replacement-safe")
+        return real_open(path, flags, *args, **kwargs)
+
+    monkeypatch.setattr(module.os, "open", swapping_open)
+    pipeline._cleanup_generations()
+    assert swapped
+    assert (root / "sentinel").read_text() == "replacement-safe"
+
+
 def test_snapshot_stays_on_one_generation_across_publish(tmp_path):
     def pipeline(content):
         return DwhPreprocessPipeline(

harness/tht/jobs/dwh_pipeline.py

@@ -5,10 +5,12 @@ from __future__ import annotations
 import hashlib
 import json
 import fcntl
+import atexit
 import os
 import re
 import shutil
 import stat
+import tempfile
 import uuid
 from collections.abc import Callable
 from dataclasses import dataclass
@@ -29,6 +31,16 @@ _RUN_ID = re.compile(r"^[0-9a-f]{32}$")
 _SAFE_FILE = re.compile(r"^[A-Za-z0-9_-]+\.(?:pkl|json)$")
 GENERATION_MANIFEST = "generation-manifest.json"
 OWNER_MARKER = "OWNER.json"
+_SNAPSHOT_DIRS: set[Path] = set()
+
+
+def _cleanup_snapshot_dirs() -> None:
+    for path in tuple(_SNAPSHOT_DIRS):
+        shutil.rmtree(path, ignore_errors=True)
+        _SNAPSHOT_DIRS.discard(path)
+
+
+atexit.register(_cleanup_snapshot_dirs)
 
 
 def config_dwh_binding(cfg) -> dict[str, str]:
@@ -48,19 +60,6 @@ def _binding_digest(binding: dict[str, str]) -> str:
     return hashlib.sha256(payload.encode("utf-8")).hexdigest()
 
 
-def _read_root_binding(workspace_root: Path) -> dict[str, str]:
-    try:
-        root_fd = os.open(
-            workspace_root / ".tht-dwh", os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW
-        )
-        try:
-            return _read_root_binding_fd(root_fd)
-        finally:
-            os.close(root_fd)
-    except OSError as error:
-        raise CorruptCheckpointError("DWH workspace ownership marker is missing or invalid") from error
-
-
 def _read_root_binding_fd(root_fd: int) -> dict[str, str]:
     try:
         fd = os.open(OWNER_MARKER, os.O_RDONLY | os.O_NOFOLLOW, dir_fd=root_fd)
@@ -92,19 +91,6 @@ def _read_root_binding_fd(root_fd: int) -> dict[str, str]:
         raise CorruptCheckpointError("DWH workspace ownership marker is missing or invalid") from error
 
 
-def _validate_root_binding(workspace_root: Path, expected: dict[str, str]) -> None:
-    try:
-        root_fd = os.open(
-            workspace_root / ".tht-dwh", os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW
-        )
-    except OSError as error:
-        raise CorruptCheckpointError("DWH workspace ownership marker is missing or invalid") from error
-    try:
-        _validate_root_binding_fd(root_fd, expected)
-    finally:
-        os.close(root_fd)
-
-
 def _validate_root_binding_fd(root_fd: int, expected: dict[str, str]) -> None:
     if _read_root_binding_fd(root_fd) != expected:
         raise CorruptCheckpointError("DWH artifacts belong to a different workspace configuration")
@@ -195,6 +181,7 @@ class DwhArtifactSnapshot:
     generation: str | None
     physical: Path
     lsh_dir: Path
+    _holder: object | None = None
 
 
 @dataclass
@@ -249,6 +236,9 @@ class DwhSnapshotLease:
         if self._lease is not None:
             lease, self._lease = self._lease, None
             lease.close()
+        if self.snapshot is not None and isinstance(self.snapshot._holder, Path):
+            shutil.rmtree(self.snapshot._holder, ignore_errors=True)
+            _SNAPSHOT_DIRS.discard(self.snapshot._holder)
 
 
 def lease_dwh_snapshot(cfg) -> DwhSnapshotLease:
@@ -323,25 +313,6 @@ def _digest(path: Path) -> str:
     return hashlib.sha256(path.read_bytes()).hexdigest()
 
 
-def _read_owned(path: Path, *, readonly: bool) -> bytes:
-    fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW)
-    try:
-        info = os.fstat(fd)
-        if (
-            not stat.S_ISREG(info.st_mode)
-            or info.st_uid != os.getuid()
-            or info.st_nlink != 1
-            or (readonly and bool(info.st_mode & 0o222))
-        ):
-            raise OSError("unsafe DWH generation file")
-        chunks = []
-        while chunk := os.read(fd, 1024 * 1024):
-            chunks.append(chunk)
-        return b"".join(chunks)
-    finally:
-        os.close(fd)
-
-
 def _read_owned_at(directory_fd: int, name: str, *, readonly: bool) -> bytes:
     fd = os.open(name, os.O_RDONLY | os.O_NOFOLLOW, dir_fd=directory_fd)
     try:
@@ -361,6 +332,94 @@ def _read_owned_at(directory_fd: int, name: str, *, readonly: bool) -> bytes:
         os.close(fd)
 
 
+def _open_generations_fd(root_fd: int, *, create: bool = False) -> int:
+    if create:
+        try:
+            os.mkdir("generations", 0o700, dir_fd=root_fd)
+            os.fsync(root_fd)
+        except FileExistsError:
+            pass
+    fd = os.open(
+        "generations", os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=root_fd
+    )
+    info = os.fstat(fd)
+    if (
+        not stat.S_ISDIR(info.st_mode)
+        or info.st_uid != os.getuid()
+        or stat.S_IMODE(info.st_mode) != 0o700
+    ):
+        os.close(fd)
+        raise CorruptCheckpointError("DWH generations directory is invalid")
+    return fd
+
+
+def _open_generation_fd(generations_fd: int, generation: str) -> int:
+    if not _RUN_ID.fullmatch(generation):
+        raise CorruptCheckpointError("DWH generation identity is invalid")
+    try:
+        return os.open(
+            generation, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
+            dir_fd=generations_fd,
+        )
+    except OSError as error:
+        raise CorruptCheckpointError("active DWH generation is missing") from error
+
+
+def _active_generation_fd(
+    root_fd: int, expected_binding: dict[str, str]
+) -> tuple[str, int] | None:
+    _validate_root_binding_fd(root_fd, expected_binding)
+    try:
+        generation = _read_owned_at(root_fd, "ACTIVE", readonly=False).decode().strip()
+    except FileNotFoundError:
+        try:
+            generations_fd = _open_generations_fd(root_fd)
+        except FileNotFoundError:
+            return None
+        try:
+            if os.listdir(generations_fd):
+                raise CorruptCheckpointError(
+                    "DWH generations exist without a consistent ACTIVE pointer"
+                )
+        finally:
+            os.close(generations_fd)
+        return None
+    except (OSError, UnicodeDecodeError) as error:
+        raise CorruptCheckpointError("DWH ACTIVE pointer is invalid") from error
+    if not _RUN_ID.fullmatch(generation):
+        raise CorruptCheckpointError("DWH ACTIVE pointer is invalid")
+    generations_fd = _open_generations_fd(root_fd)
+    try:
+        generation_fd = _open_generation_fd(generations_fd, generation)
+    finally:
+        os.close(generations_fd)
+    try:
+        validate_generation_fd(generation_fd, generation, expected_binding)
+        return generation, generation_fd
+    except BaseException:
+        os.close(generation_fd)
+        raise
+
+
+def _materialize_generation_fd(
+    generation_fd: int, generation: str, binding: dict[str, str]
+) -> tuple[Path, Path]:
+    holder = Path(tempfile.mkdtemp(prefix="tht-dwh-snapshot-"))
+    _SNAPSHOT_DIRS.add(holder)
+    snapshot_root = holder
+    try:
+        manifest = validate_generation_fd(generation_fd, generation, binding)
+        for name in manifest["files"]:
+            destination = snapshot_root / name
+            destination.write_bytes(_read_owned_at(generation_fd, name, readonly=True))
+            destination.chmod(0o400)
+        return holder, snapshot_root
+    except BaseException:
+        shutil.rmtree(holder, ignore_errors=True)
+        _SNAPSHOT_DIRS.discard(holder)
+        raise
+
+
 def validate_generation_fd(
     directory_fd: int, generation: str, expected_binding: dict[str, str] | None = None,
 ) -> dict:
@@ -405,19 +464,6 @@ def validate_generation_fd(
         raise CorruptCheckpointError("published DWH generation is invalid") from error
 
 
-def validate_generation(
-    target: Path, expected_binding: dict[str, str] | None = None,
-) -> dict:
-    try:
-        fd = os.open(target, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
-    except OSError as error:
-        raise CorruptCheckpointError("published DWH generation is invalid") from error
-    try:
-        return validate_generation_fd(fd, target.name, expected_binding)
-    finally:
-        os.close(fd)
-
-
 def resolve_dwh_snapshot(cfg) -> DwhArtifactSnapshot:
     binding = config_dwh_binding(cfg)
     lease = _acquire_existing_generation_lock(
@@ -432,16 +478,21 @@ def resolve_dwh_snapshot(cfg) -> DwhArtifactSnapshot:
 def _resolve_dwh_snapshot_locked(
     cfg, binding: dict[str, str], root_fd: int,
 ) -> DwhArtifactSnapshot:
-    _validate_root_binding_fd(root_fd, binding)
+    active = _active_generation_fd(root_fd, binding)
-    target = _active_generation_dir_locked(
+    if active is None:
-        cfg.paths.artifacts.parent, binding, root_fd=root_fd
-    )
-    if target is None:
         return DwhArtifactSnapshot(
             None, cfg.paths.artifacts / "mschema" / "physical.yaml", cfg.paths.indexes / "lsh"
         )
-    validate_generation(target, binding)
+    generation, generation_fd = active
-    return DwhArtifactSnapshot(target.name, target / "physical.yaml", target)
+    try:
+        holder, snapshot_root = _materialize_generation_fd(
+            generation_fd, generation, binding
+        )
+    finally:
+        os.close(generation_fd)
+    return DwhArtifactSnapshot(
+        generation, snapshot_root / "physical.yaml", snapshot_root, holder
+    )
 
 
 def active_generation_dir(
@@ -449,36 +500,17 @@ def active_generation_dir(
 ) -> Path | None:
     lease = _acquire_existing_generation_lock(workspace_root, exclusive=False)
     try:
-        return _active_generation_dir_locked(
+        active = _active_generation_fd(lease.root_fd, expected_binding)
-            workspace_root, expected_binding, root_fd=lease.root_fd
+        if active is None:
-        )
+            return None
+        generation, generation_fd = active
+        os.close(generation_fd)
+        lease.assert_root_identity(workspace_root / ".tht-dwh")
+        return workspace_root / ".tht-dwh" / "generations" / generation
     finally:
         lease.close()
 
 
-def _active_generation_dir_locked(
-    workspace_root: Path, expected_binding: dict[str, str], *, root_fd: int | None = None
-) -> Path | None:
-    if root_fd is None:
-        _validate_root_binding(workspace_root, expected_binding)
-    else:
-        _validate_root_binding_fd(root_fd, expected_binding)
-    pointer = workspace_root / ".tht-dwh" / "ACTIVE"
-    try:
-        generation = _read_owned(pointer, readonly=False).decode("utf-8").strip()
-    except FileNotFoundError:
-        return None
-    except (OSError, UnicodeDecodeError) as error:
-        raise CorruptCheckpointError("DWH ACTIVE pointer is invalid") from error
-    if not _RUN_ID.fullmatch(generation):
-        raise CorruptCheckpointError("DWH ACTIVE pointer is invalid")
-    target = pointer.parent / "generations" / generation
-    if not target.is_dir() or target.is_symlink():
-        raise CorruptCheckpointError("active DWH generation is missing")
-    validate_generation(target, expected_binding)
-    return target
-
-
 class DwhPreprocessPipeline:
     """Stage a complete artifact bundle, then publish it through one atomic pointer."""
 
@@ -509,6 +541,7 @@ class DwhPreprocessPipeline:
         )
         self.current_physical = current_physical
         self.current_lsh_dir = current_lsh_dir
+        self._snapshot_holder = None
         self.after_publish = after_publish
         if isinstance(retain_generations, bool) or retain_generations < 1:
             raise ValueError("retain_generations must be positive")
@@ -528,11 +561,10 @@ class DwhPreprocessPipeline:
 
     def _assert_active_binding(self, root_fd: int) -> None:
         _validate_root_binding_fd(root_fd, self.binding)
-        active = _active_generation_dir_locked(
+        active = _active_generation_fd(root_fd, self.binding)
-            self.workspace_root, self.binding, root_fd=root_fd
-        )
         if active is not None:
-            validate_generation(active, self.binding)
+            _, generation_fd = active
+            os.close(generation_fd)
 
     def run(
         self, steps: tuple[str, ...] = DWH_STAGE_IDS, *, resume_run_id: str | None = None
@@ -544,12 +576,17 @@ class DwhPreprocessPipeline:
             _claim_or_validate_root_binding(lease.root_fd, self.binding)
             lease.assert_root_identity(self.workspace_root / ".tht-dwh")
             self._assert_active_binding(lease.root_fd)
-            active = _active_generation_dir_locked(
+            active = _active_generation_fd(lease.root_fd, self.binding)
-                self.workspace_root, self.binding, root_fd=lease.root_fd
-            )
             if active is not None:
-                self.current_physical = active / "physical.yaml"
+                generation, generation_fd = active
-                self.current_lsh_dir = active
+                try:
+                    self._snapshot_holder, snapshot_root = _materialize_generation_fd(
+                        generation_fd, generation, self.binding
+                    )
+                finally:
+                    os.close(generation_fd)
+                self.current_physical = snapshot_root / "physical.yaml"
+                self.current_lsh_dir = snapshot_root
         finally:
             lease.close()
         if resume_run_id is not None:
@@ -608,12 +645,22 @@ class DwhPreprocessPipeline:
         )
         if running is None:
             return set()
-        target = self.workspace_root / ".tht-dwh" / "generations" / source.run_id
+        lease = _acquire_existing_generation_lock(self.workspace_root, exclusive=True)
-        validate_generation(target, self.binding)
+        try:
-        active = active_generation_dir(self.workspace_root, self.binding)
+            generations_fd = _open_generations_fd(lease.root_fd)
-        if active != target:
+            try:
-            raise CorruptCheckpointError("sealed DWH publication is not ACTIVE")
+                self._validate_published_fd(
-        self._validate_published(target, run_dir / "artifacts", running.artifact_files)
+                    generations_fd, source.run_id,
+                    run_dir / "artifacts", running.artifact_files,
+                )
+                active = _active_generation_fd(lease.root_fd, self.binding)
+                if active is None or active[0] != source.run_id:
+                    raise CorruptCheckpointError("sealed DWH publication is not ACTIVE")
+                os.close(active[1])
+            finally:
+                os.close(generations_fd)
+        finally:
+            lease.close()
         return {running.name}
 
     def _validate_resume_publication(self, run_id: str) -> None:
@@ -622,23 +669,41 @@ class DwhPreprocessPipeline:
             checkpoint = json.loads((run_dir / "checkpoint.json").read_text(encoding="utf-8"))
         except (OSError, ValueError, TypeError) as error:
             raise CorruptCheckpointError("checkpoint is invalid and cannot be resumed") from error
-        target = self.workspace_root / ".tht-dwh" / "generations" / run_id
+        lease = _acquire_existing_generation_lock(self.workspace_root, exclusive=True)
-        if not target.exists():
-            if checkpoint.get("status") == "succeeded":
-                raise CorruptCheckpointError("published DWH generation is missing")
-            return
         try:
-            manifest = json.loads(
+            try:
-                (run_dir / "artifacts" / "artifact-manifest.json").read_text(encoding="utf-8")
+                generations_fd = _open_generations_fd(lease.root_fd)
-            )
+            except FileNotFoundError:
-            required = tuple(
+                if checkpoint.get("status") == "succeeded":
-                name
+                    raise CorruptCheckpointError("published DWH generation is missing")
-                for stage in manifest["stages"].values()
+                return
-                for name in stage["required"]
+            try:
-            )
+                if run_id not in os.listdir(generations_fd):
-        except (OSError, KeyError, ValueError, TypeError) as error:
+                    if checkpoint.get("status") == "succeeded":
-            raise CorruptCheckpointError("resume artifact manifest is invalid") from error
+                        raise CorruptCheckpointError("published DWH generation is missing")
-        self._validate_published(target, run_dir / "artifacts", required)
+                    return
+                try:
+                    manifest = json.loads(
+                        (run_dir / "artifacts" / "artifact-manifest.json").read_text(
+                            encoding="utf-8"
+                        )
+                    )
+                    required = tuple(
+                        name
+                        for stage in manifest["stages"].values()
+                        for name in stage["required"]
+                    )
+                except (OSError, KeyError, ValueError, TypeError) as error:
+                    raise CorruptCheckpointError(
+                        "resume artifact manifest is invalid"
+                    ) from error
+                self._validate_published_fd(
+                    generations_fd, run_id, run_dir / "artifacts", required
+                )
+            finally:
+                os.close(generations_fd)
+        finally:
+            lease.close()
 
     def _assert_no_legacy_artifacts(self, root_fd: int) -> None:
         if OWNER_MARKER in os.listdir(root_fd):
@@ -692,119 +757,165 @@ class DwhPreprocessPipeline:
         try:
             _validate_root_binding_fd(lease.root_fd, self.binding)
             self._assert_active_binding(lease.root_fd)
-            self._publish(context.run_id, artifacts, required)
+            self._publish(context.run_id, artifacts, required, lease.root_fd)
             if self.after_publish is not None:
                 self.after_publish(context.run_id)
-            self._cleanup_generations()
+            self._cleanup_generations_fd(lease.root_fd)
         finally:
             lease.close()
         return StageArtifacts(required)
 
-    def _publish(self, generation: str, artifacts: Path, required: tuple[str, ...]) -> None:
+    def _publish(
-        root = self.workspace_root / ".tht-dwh"
+        self, generation: str, artifacts: Path, required: tuple[str, ...], root_fd: int
-        generations = root / "generations"
-        self._ensure_owned_dir(root)
-        self._ensure_owned_dir(generations)
-        target = generations / generation
-        if target.exists():
-            self._validate_published(target, artifacts, required)
-        else:
-            temporary = generations / f".{generation}.{uuid.uuid4().hex}.tmp"
-            temporary.mkdir(mode=0o700)
-            try:
-                for name in required:
-                    destination = temporary / name
-                    shutil.copyfile(artifacts / name, destination)
-                    with destination.open("rb") as stream:
-                        os.fsync(stream.fileno())
-                    destination.chmod(0o400)
-                manifest = {
-                    "schema_version": 1,
-                    "generation": generation,
-                    "files": {name: _digest(artifacts / name) for name in required},
-                    "job_spec_fingerprint": json.loads(
-                        (artifacts / "artifact-manifest.json").read_text(encoding="utf-8")
-                    )["spec_fingerprint"],
-                    "artifact_manifest_sha256": _digest(
-                        artifacts / "artifact-manifest.json"
-                    ),
-                    **self.binding,
-                }
-                manifest_path = temporary / GENERATION_MANIFEST
-                manifest_path.write_text(
-                    json.dumps(manifest, sort_keys=True, separators=(",", ":")) + "\n",
-                    encoding="utf-8",
-                )
-                manifest_path.chmod(0o400)
-                with manifest_path.open("rb") as stream:
-                    os.fsync(stream.fileno())
-                self._fsync(temporary)
-                os.replace(temporary, target)
-                self._fsync(generations)
-            except BaseException:
-                shutil.rmtree(temporary, ignore_errors=True)
-                raise
-        pointer = root / "ACTIVE"
-        try:
-            previous = pointer.read_text(encoding="utf-8")
-        except FileNotFoundError:
-            previous = None
-        pointer_tmp = root / f".ACTIVE.{uuid.uuid4().hex}.tmp"
-        fd = os.open(pointer_tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
-        try:
-            with os.fdopen(fd, "w", encoding="utf-8") as stream:
-                stream.write(generation + "\n")
-                stream.flush()
-                os.fsync(stream.fileno())
-            os.replace(pointer_tmp, pointer)
-            try:
-                self._fsync(root)
-            except BaseException:
-                self._restore_pointer(root, pointer, previous)
-                raise
-        except BaseException:
-            pointer_tmp.unlink(missing_ok=True)
-            raise
-
-    def _validate_published(
-        self, target: Path, artifacts: Path, required: tuple[str, ...],
     ) -> None:
-        manifest = validate_generation(target, self.binding)
+        generations_fd = _open_generations_fd(root_fd, create=True)
-        if set(manifest["files"]) != set(required):
+        temporary = f".{generation}.{uuid.uuid4().hex}.tmp"
-            raise CorruptCheckpointError("published DWH generation is invalid")
+        try:
-        if manifest["artifact_manifest_sha256"] != _digest(
+            if generation in os.listdir(generations_fd):
-            artifacts / "artifact-manifest.json"
+                self._validate_published_fd(generations_fd, generation, artifacts, required)
-        ):
+            else:
-            raise CorruptCheckpointError("published DWH job manifest digest mismatch")
+                os.mkdir(temporary, 0o700, dir_fd=generations_fd)
-        for name in required:
+                temporary_fd = os.open(
-            source, published = artifacts / name, target / name
+                    temporary, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
-            if published.is_symlink() or not published.is_file():
+                    dir_fd=generations_fd,
-                raise CorruptCheckpointError("published DWH artifact is invalid")
+                )
-            if hashlib.sha256(source.read_bytes()).digest() != hashlib.sha256(
+                try:
-                published.read_bytes()
+                    for name in required:
-            ).digest():
+                        self._write_readonly_at(
-                raise CorruptCheckpointError("published DWH artifact digest mismatch")
+                            temporary_fd, name, (artifacts / name).read_bytes()
+                        )
+                    artifact_manifest = artifacts / "artifact-manifest.json"
+                    manifest = {
+                        "schema_version": 1,
+                        "generation": generation,
+                        "files": {name: _digest(artifacts / name) for name in required},
+                        "job_spec_fingerprint": json.loads(
+                            artifact_manifest.read_text(encoding="utf-8")
+                        )["spec_fingerprint"],
+                        "artifact_manifest_sha256": _digest(artifact_manifest),
+                        **self.binding,
+                    }
+                    self._write_readonly_at(
+                        temporary_fd, GENERATION_MANIFEST,
+                        (json.dumps(manifest, sort_keys=True, separators=(",", ":")) + "\n").encode(),
+                    )
+                    os.fsync(temporary_fd)
+                finally:
+                    os.close(temporary_fd)
+                os.rename(
+                    temporary, generation,
+                    src_dir_fd=generations_fd, dst_dir_fd=generations_fd,
+                )
+                temporary = ""
+                os.fsync(generations_fd)
+            previous = None
+            try:
+                previous = _read_owned_at(root_fd, "ACTIVE", readonly=False)
+            except FileNotFoundError:
+                pass
+            self._replace_active_at(root_fd, (generation + "\n").encode(), previous)
+        finally:
+            if temporary:
+                self._safe_delete_generation(generations_fd, temporary, allow_temporary=True)
+            os.close(generations_fd)
 
-    def _restore_pointer(self, root: Path, pointer: Path, previous: str | None) -> None:
+    @staticmethod
+    def _write_readonly_at(directory_fd: int, name: str, payload: bytes) -> None:
+        fd = os.open(
+            name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
+            0o600, dir_fd=directory_fd,
+        )
+        try:
+            offset = 0
+            while offset < len(payload):
+                offset += os.write(fd, payload[offset:])
+            os.fsync(fd)
+            os.fchmod(fd, 0o400)
+            os.fsync(fd)
+        finally:
+            os.close(fd)
+
+    def _replace_active_at(self, root_fd: int, payload: bytes, previous: bytes | None) -> None:
+        temporary = f".ACTIVE.{uuid.uuid4().hex}.tmp"
+        fd = os.open(
+            temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
+            0o600, dir_fd=root_fd,
+        )
+        try:
+            offset = 0
+            while offset < len(payload):
+                offset += os.write(fd, payload[offset:])
+            os.fsync(fd)
+        finally:
+            os.close(fd)
+        try:
+            os.replace(temporary, "ACTIVE", src_dir_fd=root_fd, dst_dir_fd=root_fd)
+            temporary = ""
+            try:
+                os.fsync(root_fd)
+            except BaseException:
+                self._restore_pointer_at(root_fd, previous)
+                raise
+        finally:
+            if temporary:
+                try:
+                    os.unlink(temporary, dir_fd=root_fd)
+                except FileNotFoundError:
+                    pass
+
+    def _validate_published_fd(
+        self, generations_fd: int, generation: str,
+        artifacts: Path, required: tuple[str, ...],
+    ) -> None:
+        generation_fd = _open_generation_fd(generations_fd, generation)
+        try:
+            manifest = validate_generation_fd(generation_fd, generation, self.binding)
+            if set(manifest["files"]) != set(required):
+                raise CorruptCheckpointError("published DWH generation is invalid")
+            artifact_digest = _digest(artifacts / "artifact-manifest.json")
+            if manifest["artifact_manifest_sha256"] != artifact_digest:
+                raise CorruptCheckpointError("published DWH job manifest digest mismatch")
+            for name in required:
+                if hashlib.sha256((artifacts / name).read_bytes()).hexdigest() != manifest["files"][name]:
+                    raise CorruptCheckpointError("published DWH artifact digest mismatch")
+        finally:
+            os.close(generation_fd)
+
+    def _restore_pointer_at(self, root_fd: int, previous: bytes | None) -> None:
         if previous is None:
-            pointer.unlink(missing_ok=True)
+            try:
+                os.unlink("ACTIVE", dir_fd=root_fd)
+            except FileNotFoundError:
+                pass
         else:
-            restore = root / f".ACTIVE.restore.{uuid.uuid4().hex}.tmp"
+            restore = f".ACTIVE.restore.{uuid.uuid4().hex}.tmp"
-            restore.write_text(previous, encoding="utf-8")
+            fd = os.open(
-            with restore.open("rb") as stream:
+                restore, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
-                os.fsync(stream.fileno())
+                0o600, dir_fd=root_fd,
-            os.replace(restore, pointer)
+            )
-        self._fsync(root)
+            try:
+                offset = 0
+                while offset < len(previous):
+                    offset += os.write(fd, previous[offset:])
+                os.fsync(fd)
+            finally:
+                os.close(fd)
+            os.replace(restore, "ACTIVE", src_dir_fd=root_fd, dst_dir_fd=root_fd)
+        os.fsync(root_fd)
 
     def _cleanup_generations(self) -> None:
-        root = self.workspace_root / ".tht-dwh" / "generations"
+        lease = _acquire_existing_generation_lock(self.workspace_root, exclusive=True)
-        if not root.exists():
+        try:
-            return
+            self._cleanup_generations_fd(lease.root_fd)
-        _validate_root_binding(self.workspace_root, self.binding)
+        finally:
-        active = _active_generation_dir_locked(self.workspace_root, self.binding)
+            lease.close()
+
+    def _cleanup_generations_fd(self, root_fd: int) -> None:
+        _validate_root_binding_fd(root_fd, self.binding)
+        active = _active_generation_fd(root_fd, self.binding)
+        active_name = active[0] if active else None
         if active is not None:
-            validate_generation(active, self.binding)
+            os.close(active[1])
-        active_name = active.name if active else None
         protected = {active_name} if active_name else set()
         runs = self.workspace_root / ".tht-jobs" / "dwh" / "runs"
         for checkpoint in runs.glob("*/checkpoint.json") if runs.exists() else ():
@@ -815,14 +926,18 @@ class DwhPreprocessPipeline:
             except (OSError, KeyError, ValueError):
                 continue
         generations = []
-        root_fd = os.open(root, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
         try:
-            for name in os.listdir(root_fd):
+            generations_fd = _open_generations_fd(root_fd)
+        except FileNotFoundError:
+            return
+        try:
+            for name in os.listdir(generations_fd):
                 if not _RUN_ID.fullmatch(name):
                     continue
                 try:
                     candidate_fd = os.open(
-                        name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=root_fd
+                        name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
+                        dir_fd=generations_fd,
                     )
                 except OSError:
                     continue
@@ -843,13 +958,17 @@ class DwhPreprocessPipeline:
             for _, name in generations:
                 if name in protected | keep_recent:
                     continue
-                self._safe_delete_generation(root_fd, name)
+                self._safe_delete_generation(generations_fd, name)
-            os.fsync(root_fd)
+            os.fsync(generations_fd)
         finally:
-            os.close(root_fd)
+            os.close(generations_fd)
 
     @staticmethod
-    def _safe_delete_generation(root_fd: int, name: str) -> None:
+    def _safe_delete_generation(
+        root_fd: int, name: str, *, allow_temporary: bool = False
+    ) -> None:
+        if not (_RUN_ID.fullmatch(name) or (allow_temporary and name.startswith("."))):
+            return
         try:
             generation_fd = os.open(
                 name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=root_fd
@@ -905,14 +1024,6 @@ class DwhPreprocessPipeline:
             raise OSError("unsafe DWH publication directory")
         path.chmod(0o700)
 
-    @staticmethod
-    def _fsync(path: Path) -> None:
-        fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY)
-        try:
-            os.fsync(fd)
-        finally:
-            os.close(fd)
-
     @staticmethod
     def _validate_steps(steps: tuple[str, ...]) -> None:
         if not steps or len(steps) != len(set(steps)) or any(