fix(preprocess): publish DWH artifacts atomically

harness/tests/test_dwh_preprocess_job.py

@@ -1,5 +1,6 @@
 import hashlib
 import json
+from pathlib import Path
 
 from typer.testing import CliRunner
 
@@ -22,8 +23,8 @@ def test_selected_dwh_stages_run_in_declared_order(tmp_path):
         workspace_root=tmp_path,
         config_fingerprint=FP,
         input_fingerprint=FP,
-        introspect=lambda: calls.append("introspect"),
+        introspect=lambda output: (calls.append("introspect"), output.write_text("catalog")),
-        build_lsh=lambda: calls.append("lsh"),
+        build_lsh=lambda physical, output: _write_lsh(calls, physical, output),
     )
 
     report = pipeline.run(("introspect", "lsh"))
@@ -31,6 +32,19 @@ def test_selected_dwh_stages_run_in_declared_order(tmp_path):
     assert report.status == "succeeded"
     assert calls == ["introspect", "lsh"]
     assert [stage.name for stage in report.stages] == ["introspect", "lsh"]
+    active = (tmp_path / ".tht-dwh" / "ACTIVE").read_text().strip()
+    published = tmp_path / ".tht-dwh" / "generations" / active
+    assert (published / "physical.yaml").read_text() == "catalog"
+    assert sorted(path.name for path in published.iterdir()) == [
+        "demo_lsh.pkl", "demo_meta.json", "demo_minhashes.pkl", "physical.yaml"
+    ]
+
+
+def _write_lsh(calls, physical: Path, output: Path):
+    calls.append("lsh")
+    assert physical.read_text() == "catalog"
+    for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json"):
+        (output / name).write_text(name)
 
 
 def test_preprocess_dwh_json_is_pristine(monkeypatch, tmp_path):
@@ -81,3 +95,47 @@ def test_preprocess_dwh_rejects_unknown_or_duplicate_steps(monkeypatch, tmp_path
         assert response.exit_code == 2
         assert json.loads(response.output)["status"] == "failed"
     assert called is False
+
+
+def test_failed_multi_file_build_never_replaces_active_generation(tmp_path):
+    def catalog(output):
+        output.write_text("old-catalog")
+
+    first = DwhPreprocessPipeline(
+        workspace_id="demo", workspace_root=tmp_path,
+        config_fingerprint=FP, input_fingerprint=FP,
+        introspect=catalog,
+        build_lsh=lambda physical, output: [
+            (output / name).write_text(name)
+            for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json")
+        ],
+    ).run(("introspect", "lsh"))
+    assert first.status == "succeeded"
+    old_active = (tmp_path / ".tht-dwh" / "ACTIVE").read_text()
+
+    def partial_lsh(physical, output):
+        (output / "demo_lsh.pkl").write_text("new-but-partial")
+        raise RuntimeError("crash between LSH files")
+
+    failed = DwhPreprocessPipeline(
+        workspace_id="demo", workspace_root=tmp_path,
+        config_fingerprint=FP, input_fingerprint=FP,
+        introspect=lambda output: output.write_text("new-catalog"),
+        build_lsh=partial_lsh,
+    ).run(("introspect", "lsh"))
+
+    assert failed.status == "failed"
+    assert (tmp_path / ".tht-dwh" / "ACTIVE").read_text() == old_active
+
+    resumed = DwhPreprocessPipeline(
+        workspace_id="demo", workspace_root=tmp_path,
+        config_fingerprint=FP, input_fingerprint=FP,
+        introspect=lambda output: (_ for _ in ()).throw(
+            AssertionError("completed introspection must not repeat")
+        ),
+        build_lsh=lambda physical, output: [
+            (output / name).write_text("recovered")
+            for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json")
+        ],
+    ).run(("introspect", "lsh"), resume_run_id=failed.run_id)
+    assert resumed.status == "succeeded"

harness/tests/test_lsh_job_resume.py

@@ -1,5 +1,7 @@
 import hashlib
 
+import pytest
+
 from tht.jobs.dwh_pipeline import DwhPreprocessPipeline
 
 
@@ -9,10 +11,11 @@ FP = "sha256:" + hashlib.sha256(b"test").hexdigest()
 def test_lsh_failure_resumes_exact_run_without_repeating_introspection(tmp_path):
     calls = []
 
-    def introspect():
+    def introspect(output):
         calls.append("introspect")
+        output.write_text("catalog")
 
-    def fail_lsh():
+    def fail_lsh(physical, output):
         calls.append("lsh-failed")
         raise RuntimeError("database detail that must not leak")
 
@@ -32,7 +35,7 @@ def test_lsh_failure_resumes_exact_run_without_repeating_introspection(tmp_path)
         config_fingerprint=FP,
         input_fingerprint=FP,
         introspect=introspect,
-        build_lsh=lambda: calls.append("lsh-recovered"),
+        build_lsh=lambda physical, output: _recover_lsh(calls, output),
     ).run(("introspect", "lsh"), resume_run_id=failed.run_id)
 
     assert resumed.status == "succeeded"
@@ -40,14 +43,20 @@ def test_lsh_failure_resumes_exact_run_without_repeating_introspection(tmp_path)
     assert calls == ["introspect", "lsh-failed", "lsh-recovered"]
 
 
+def _recover_lsh(calls, output):
+    calls.append("lsh-recovered")
+    for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json"):
+        (output / name).write_text(name)
+
+
 def test_resume_rejects_a_different_stage_selection(tmp_path):
     failed = DwhPreprocessPipeline(
         workspace_id="demo",
         workspace_root=tmp_path,
         config_fingerprint=FP,
         input_fingerprint=FP,
-        introspect=lambda: None,
+        introspect=lambda output: output.write_text("catalog"),
-        build_lsh=lambda: (_ for _ in ()).throw(RuntimeError()),
+        build_lsh=lambda physical, output: (_ for _ in ()).throw(RuntimeError()),
     ).run(("introspect", "lsh"))
 
     pipeline = DwhPreprocessPipeline(
@@ -55,8 +64,8 @@ def test_resume_rejects_a_different_stage_selection(tmp_path):
         workspace_root=tmp_path,
         config_fingerprint=FP,
         input_fingerprint=FP,
-        introspect=lambda: None,
+        introspect=lambda output: output.write_text("catalog"),
-        build_lsh=lambda: None,
+        build_lsh=lambda physical, output: None,
     )
     try:
         pipeline.run(("lsh",), resume_run_id=failed.run_id)
@@ -64,3 +73,71 @@ def test_resume_rejects_a_different_stage_selection(tmp_path):
         assert "incompatible" in str(error)
     else:
         raise AssertionError("resume with different stages must fail")
+
+
+def test_resume_rejects_tampered_succeeded_stage_artifact(tmp_path):
+    failed = DwhPreprocessPipeline(
+        workspace_id="demo", workspace_root=tmp_path,
+        config_fingerprint=FP, input_fingerprint=FP,
+        introspect=lambda output: output.write_text("catalog"),
+        build_lsh=lambda physical, output: (_ for _ in ()).throw(RuntimeError()),
+    ).run(("introspect", "lsh"))
+    artifact = (
+        tmp_path / ".tht-jobs" / "dwh" / "runs" / failed.run_id
+        / "artifacts" / "physical.yaml"
+    )
+    artifact.write_text("tampered")
+
+    pipeline = DwhPreprocessPipeline(
+        workspace_id="demo", workspace_root=tmp_path,
+        config_fingerprint=FP, input_fingerprint=FP,
+        introspect=lambda output: output.write_text("catalog"),
+        build_lsh=lambda physical, output: _recover_lsh([], output),
+    )
+    with pytest.raises(Exception, match="artifact manifest is invalid"):
+        pipeline.run(("introspect", "lsh"), resume_run_id=failed.run_id)
+
+
+def test_post_publish_crash_reconciles_same_generation_on_resume(tmp_path):
+    crashed = False
+
+    def crash_once(_generation):
+        nonlocal crashed
+        if not crashed:
+            crashed = True
+            raise KeyboardInterrupt("simulated process death")
+
+    pipeline = DwhPreprocessPipeline(
+        workspace_id="demo", workspace_root=tmp_path,
+        config_fingerprint=FP, input_fingerprint=FP,
+        introspect=lambda output: output.write_text("catalog"),
+        build_lsh=lambda physical, output: _recover_lsh([], output),
+        after_publish=crash_once,
+    )
+    with pytest.raises(KeyboardInterrupt):
+        pipeline.run(("introspect", "lsh"))
+    active = (tmp_path / ".tht-dwh" / "ACTIVE").read_text().strip()
+    checkpoint = next((tmp_path / ".tht-jobs" / "dwh" / "runs").glob("*/checkpoint.json"))
+    source_run_id = checkpoint.parent.name
+    assert active == source_run_id
+
+    resumed = pipeline.run(("introspect", "lsh"), resume_run_id=source_run_id)
+    assert resumed.status == "succeeded"
+    assert (tmp_path / ".tht-dwh" / "ACTIVE").read_text().strip() == resumed.run_id
+
+
+def test_resume_of_succeeded_run_detects_tampered_published_file(tmp_path):
+    pipeline = DwhPreprocessPipeline(
+        workspace_id="demo", workspace_root=tmp_path,
+        config_fingerprint=FP, input_fingerprint=FP,
+        introspect=lambda output: output.write_text("catalog"),
+        build_lsh=lambda physical, output: _recover_lsh([], output),
+    )
+    succeeded = pipeline.run(("introspect", "lsh"))
+    published = (
+        tmp_path / ".tht-dwh" / "generations" / succeeded.run_id / "demo_meta.json"
+    )
+    published.write_text("tampered")
+
+    with pytest.raises(Exception, match="digest mismatch"):
+        pipeline.run(("introspect", "lsh"), resume_run_id=succeeded.run_id)

harness/tht/cli/lsh_cmd.py

@@ -9,6 +9,11 @@ lsh_app = typer.Typer(help="Indice LSH su valori dei campi (derivato, rigenerabi
 
 
 def _lsh_dir(cfg) -> Path:
+    from tht.jobs.dwh_pipeline import active_generation_dir
+
+    active = active_generation_dir(cfg.paths.artifacts.parent)
+    if active is not None:
+        return active
     return cfg.paths.indexes / "lsh"
 
 
@@ -36,14 +41,17 @@ def _extract_lsh_values(dwh, physical, annotations, limit):
     return values, skipped, truncated
 
 
-def build_lsh_artifacts(cfg, *, dwh=None, verbose: bool = False):
+def build_lsh_artifacts(
+    cfg, *, dwh=None, verbose: bool = False, physical_file: Path | None = None,
+    output_dir: Path | None = None,
+):
     """Run the existing LSH extraction/build algorithm and persist its outputs."""
     from tht.adapters.factory import build_dwh
     from tht.cli.schema_cmd import annotations_path
     from tht.lshindex import build_index, save_index
     from tht.mschema.models import Annotations, PhysicalSchema
 
-    phys_file = physical_path(cfg)
+    phys_file = physical_file or physical_path(cfg)
     if not phys_file.exists():
         raise FileNotFoundError("physical catalog is missing; run schema introspect first")
     physical = PhysicalSchema.from_yaml(phys_file)
@@ -53,7 +61,10 @@ def build_lsh_artifacts(cfg, *, dwh=None, verbose: bool = False):
         target, physical, annotations, cfg.lsh.max_values_per_column
     )
     lsh, minhashes = build_index(values, cfg.lsh, verbose=verbose)
-    save_index(lsh, minhashes, cfg.lsh, _lsh_dir(cfg), name=cfg.database.db_schema)
+    save_index(
+        lsh, minhashes, cfg.lsh, output_dir or (cfg.paths.indexes / "lsh"),
+        name=cfg.database.db_schema,
+    )
     return minhashes, skipped, truncated, values
 
 
@@ -69,9 +80,25 @@ def build_cmd(config: Path = CONFIG_OPT) -> None:
         )
         raise typer.Exit(code=1)
     typer.echo("Estrazione valori (i più frequenti) dalle colonne testuali eligible...")
-    minhashes, skipped, truncated, values = build_lsh_artifacts(cfg, verbose=True)
+    from tht.jobs.dwh_pipeline import active_generation_dir
-    n_values = sum(len(v) for t in values.values() for v in t.values())
+
-    typer.echo(f"  {n_values} valori da {sum(len(t) for t in values.values())} colonne")
+    if active_generation_dir(cfg.paths.artifacts.parent) is None:
+        minhashes, skipped, truncated, values = build_lsh_artifacts(cfg, verbose=True)
+        n_values = sum(len(v) for table in values.values() for v in table.values())
+        n_columns = sum(len(table) for table in values.values())
+    else:
+        from tht.cli.preprocess_cmd import run_dwh_from_config
+        from tht.lshindex import load_index
+
+        report = run_dwh_from_config(config, steps=("lsh",))
+        if report.status != "succeeded":
+            typer.secho("ERRORE: DWH preprocessing failed", fg=typer.colors.RED, err=True)
+            raise typer.Exit(code=1)
+        _, minhashes, _ = load_index(_lsh_dir(cfg), name=cfg.database.db_schema)
+        skipped, truncated = [], []
+        n_values = len(minhashes)
+        n_columns = len({(entry[1], entry[2]) for entry in minhashes.values()})
+    typer.echo(f"  {n_values} valori da {n_columns} colonne")
     for s in skipped:
         typer.secho(f"  saltata {s.table}.{s.column}: {s.reason}", fg=typer.colors.YELLOW)
     for t in truncated:

harness/tht/cli/preprocess_cmd.py

@@ -19,18 +19,29 @@ def run_dwh_from_config(
     config: Path, *, steps: tuple[str, ...], resume: str | None = None,
 ):
     from tht.cli.lsh_cmd import build_lsh_artifacts
-    from tht.cli.schema_cmd import _load_config_or_exit, refresh_catalog
+    from tht.cli.schema_cmd import _load_config_or_exit, physical_path, refresh_catalog
-    from tht.jobs.dwh_pipeline import DwhPreprocessPipeline, fingerprint
+    from tht.jobs.dwh_pipeline import DwhPreprocessPipeline, active_generation_dir, fingerprint
 
     cfg = _load_config_or_exit(config)
     workspace_root = cfg.paths.artifacts.parent
+    active = active_generation_dir(workspace_root)
+    lsh_names = (
+        f"{cfg.database.db_schema}_lsh.pkl",
+        f"{cfg.database.db_schema}_minhashes.pkl",
+        f"{cfg.database.db_schema}_meta.json",
+    )
     pipeline = DwhPreprocessPipeline(
         workspace_id=config.stem.lower().replace(".", "-").replace("_", "-"),
         workspace_root=workspace_root,
         config_fingerprint=fingerprint(cfg.model_dump_json()),
         input_fingerprint=fingerprint(config.resolve().as_posix()),
-        introspect=lambda: refresh_catalog(cfg),
+        introspect=lambda output: refresh_catalog(cfg, output_path=output),
-        build_lsh=lambda: build_lsh_artifacts(cfg),
+        build_lsh=lambda physical, output: build_lsh_artifacts(
+            cfg, physical_file=physical, output_dir=output
+        ),
+        lsh_filenames=lsh_names,
+        current_physical=physical_path(cfg),
+        current_lsh_dir=active if active is not None else cfg.paths.indexes / "lsh",
     )
     return pipeline.run(steps, resume_run_id=resume)
 

harness/tht/cli/schema_cmd.py

@@ -37,6 +37,11 @@ def _load_config_or_exit(config: Path):
 
 
 def physical_path(cfg) -> Path:
+    from tht.jobs.dwh_pipeline import active_generation_dir
+
+    active = active_generation_dir(cfg.paths.artifacts.parent)
+    if active is not None:
+        return active / "physical.yaml"
     return cfg.paths.artifacts / "mschema" / "physical.yaml"
 
 
@@ -44,13 +49,13 @@ def annotations_path(cfg) -> Path:
     return cfg.paths.artifacts / "mschema" / "annotations.yaml"
 
 
-def refresh_catalog(cfg, *, dwh=None):
+def refresh_catalog(cfg, *, dwh=None, output_path: Path | None = None):
     """Run the existing catalog algorithm and persist its canonical output."""
     target = dwh if dwh is not None else build_dwh(cfg)
     physical = target.introspect()
     _add_examples(target, physical, cfg.examples)
     classify_all(physical, cfg.eligibility)
-    physical.to_yaml(physical_path(cfg))
+    physical.to_yaml(output_path or (cfg.paths.artifacts / "mschema" / "physical.yaml"))
     return physical
 
 
@@ -90,7 +95,18 @@ def introspect_cmd(
             )
             return
     try:
-        phys = refresh_catalog(cfg)
+        from tht.jobs.dwh_pipeline import active_generation_dir
+
+        if active_generation_dir(cfg.paths.artifacts.parent) is None:
+            phys = refresh_catalog(cfg)
+        else:
+            from tht.cli.preprocess_cmd import run_dwh_from_config
+            from tht.mschema.models import PhysicalSchema
+
+            report = run_dwh_from_config(config, steps=("introspect",))
+            if report.status != "succeeded":
+                raise RuntimeError("DWH preprocessing failed")
+            phys = PhysicalSchema.from_yaml(physical_path(cfg))
     except Exception as e:
         typer.secho(f"ERRORE: {e}", fg=typer.colors.RED, err=True)
         raise typer.Exit(code=1)

harness/tht/jobs/dwh_pipeline.py

@@ -1,20 +1,47 @@
-"""Resumable DWH catalog and LSH preprocessing stages."""
+"""Crash-safe, resumable DWH catalog and LSH preprocessing stages."""
 
 from __future__ import annotations
 
 import hashlib
+import json
+import os
+import re
+import shutil
+import stat
+import uuid
 from collections.abc import Callable
 from pathlib import Path
 
 from tht.jobs.models import JobReport, JobSpec
-from tht.jobs.runner import run_job
+from tht.jobs.runner import (
+    CorruptCheckpointError,
+    JobContext,
+    StageArtifacts,
+    run_job,
+    seal_stage_artifacts,
+)
 
 
 DWH_STAGE_IDS = ("introspect", "lsh")
+_RUN_ID = re.compile(r"^[0-9a-f]{32}$")
+
+
+def active_generation_dir(workspace_root: Path) -> Path | None:
+    pointer = workspace_root / ".tht-dwh" / "ACTIVE"
+    try:
+        generation = pointer.read_text(encoding="utf-8").strip()
+    except FileNotFoundError:
+        return None
+    if not _RUN_ID.fullmatch(generation) or pointer.is_symlink():
+        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")
+    return target
 
 
 class DwhPreprocessPipeline:
-    """Adapt existing DWH preprocessing operations to the shared job envelope."""
+    """Stage a complete artifact bundle, then publish it through one atomic pointer."""
 
     def __init__(
         self,
@@ -23,42 +50,218 @@ class DwhPreprocessPipeline:
         workspace_root: Path,
         config_fingerprint: str,
         input_fingerprint: str,
-        introspect: Callable[[], object],
+        introspect: Callable[[Path], object],
-        build_lsh: Callable[[], object],
+        build_lsh: Callable[[Path, Path], object],
+        lsh_filenames: tuple[str, str, str] | None = None,
+        current_physical: Path | None = None,
+        current_lsh_dir: Path | None = None,
+        after_publish: Callable[[str], object] | None = None,
     ) -> None:
         self.workspace_id = workspace_id
         self.workspace_root = workspace_root
         self.config_fingerprint = config_fingerprint
         self.input_fingerprint = input_fingerprint
-        self._operations = {"introspect": introspect, "lsh": build_lsh}
+        self.introspect = introspect
+        self.build_lsh = build_lsh
+        self.lsh_filenames = lsh_filenames or (
+            f"{workspace_id}_lsh.pkl", f"{workspace_id}_minhashes.pkl",
+            f"{workspace_id}_meta.json",
+        )
+        self.current_physical = current_physical
+        self.current_lsh_dir = current_lsh_dir
+        self.after_publish = after_publish
 
     def run(
         self, steps: tuple[str, ...] = DWH_STAGE_IDS, *, resume_run_id: str | None = None
     ) -> JobReport:
+        self._validate_steps(steps)
+        if resume_run_id is not None:
+            self._validate_resume_publication(resume_run_id)
+        spec = JobSpec(
+            workspace_id=self.workspace_id,
+            job_type="dwh",
+            workspace_root=self.workspace_root,
+            spec_version="jobs-v1",
+            pipeline_version="dwh-v2",
+            config_fingerprint=self.config_fingerprint,
+            input_fingerprint=self.input_fingerprint,
+            stage_ids=steps,
+            resume_run_id=resume_run_id,
+        )
+
+        def introspect_stage(context: JobContext):
+            artifacts = self._artifacts(context)
+            physical = artifacts / "physical.yaml"
+            try:
+                self.introspect(physical)
+            except BaseException:
+                physical.unlink(missing_ok=True)
+                raise
+            if steps[-1] == "introspect":
+                self._copy_current_lsh(artifacts)
+                return self._publish_stage(context, "introspect", spec)
+            return StageArtifacts(("physical.yaml",))
+
+        def lsh_stage(context: JobContext):
+            artifacts = self._artifacts(context)
+            physical = artifacts / "physical.yaml"
+            if not physical.exists():
+                source = self.current_physical
+                if source is None or not source.is_file():
+                    raise FileNotFoundError("physical catalog is missing")
+                shutil.copyfile(source, physical)
+            try:
+                self.build_lsh(physical, artifacts)
+            except BaseException:
+                for name in self.lsh_filenames:
+                    (artifacts / name).unlink(missing_ok=True)
+                raise
+            return self._publish_stage(context, "lsh", spec)
+
+        implementations = {"introspect": introspect_stage, "lsh": lsh_stage}
+        return run_job(spec, tuple(implementations[step] for step in steps))
+
+    def _validate_resume_publication(self, run_id: str) -> None:
+        run_dir = self.workspace_root / ".tht-jobs" / "dwh" / "runs" / run_id
+        try:
+            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
+        if not target.exists():
+            if checkpoint.get("status") == "succeeded":
+                raise CorruptCheckpointError("published DWH generation is missing")
+            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(target, run_dir / "artifacts", required)
+
+    @staticmethod
+    def _artifacts(context: JobContext) -> Path:
+        root = context.run_dir / "artifacts"
+        root.mkdir(exist_ok=True)
+        return root
+
+    def _required(self, artifacts: Path) -> tuple[str, ...]:
+        required = ("physical.yaml",) + tuple(
+            name for name in self.lsh_filenames if (artifacts / name).is_file()
+        )
+        if not (artifacts / "physical.yaml").is_file():
+            raise CorruptCheckpointError("staged physical catalog is missing")
+        lsh_count = len(required) - 1
+        if lsh_count not in (0, len(self.lsh_filenames)):
+            raise CorruptCheckpointError("staged LSH artifact set is incomplete")
+        return required
+
+    def _copy_current_lsh(self, artifacts: Path) -> None:
+        if self.current_lsh_dir is None:
+            return
+        existing = [self.current_lsh_dir / name for name in self.lsh_filenames]
+        if not any(path.exists() for path in existing):
+            return
+        if not all(path.is_file() for path in existing):
+            raise CorruptCheckpointError("current LSH artifact set is incomplete")
+        for source in existing:
+            shutil.copyfile(source, artifacts / source.name)
+
+    def _publish_stage(self, context: JobContext, stage: str, spec: JobSpec):
+        artifacts = self._artifacts(context)
+        required = self._required(artifacts)
+        seal_stage_artifacts(context, stage, required, spec)
+        self._publish(context.run_id, artifacts, required)
+        if self.after_publish is not None:
+            self.after_publish(context.run_id)
+        return StageArtifacts(required)
+
+    def _publish(self, generation: str, artifacts: Path, required: tuple[str, ...]) -> None:
+        root = self.workspace_root / ".tht-dwh"
+        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())
+                self._fsync(temporary)
+                os.replace(temporary, target)
+                self._fsync(generations)
+            except BaseException:
+                shutil.rmtree(temporary, ignore_errors=True)
+                raise
+        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, root / "ACTIVE")
+            self._fsync(root)
+        except BaseException:
+            pointer_tmp.unlink(missing_ok=True)
+            raise
+
+    @staticmethod
+    def _validate_published(target: Path, artifacts: Path, required: tuple[str, ...]) -> None:
+        if (
+            target.is_symlink()
+            or not target.is_dir()
+            or {path.name for path in target.iterdir()} != set(required)
+        ):
+            raise CorruptCheckpointError("published DWH generation is invalid")
+        for name in required:
+            source, published = artifacts / name, target / name
+            if published.is_symlink() or not published.is_file():
+                raise CorruptCheckpointError("published DWH artifact is invalid")
+            if hashlib.sha256(source.read_bytes()).digest() != hashlib.sha256(
+                published.read_bytes()
+            ).digest():
+                raise CorruptCheckpointError("published DWH artifact digest mismatch")
+
+    @staticmethod
+    def _ensure_owned_dir(path: Path) -> None:
+        try:
+            path.mkdir(mode=0o700)
+        except FileExistsError:
+            pass
+        info = path.lstat()
+        if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid():
+            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(
             step not in DWH_STAGE_IDS for step in steps
         ):
             raise ValueError("DWH preprocessing steps must be unique introspect/lsh stages")
         if tuple(sorted(steps, key=DWH_STAGE_IDS.index)) != steps:
             raise ValueError("DWH preprocessing steps must follow introspect,lsh order")
-        spec = JobSpec(
-            workspace_id=self.workspace_id,
-            job_type="dwh",
-            workspace_root=self.workspace_root,
-            spec_version="jobs-v1",
-            pipeline_version="dwh-v1",
-            config_fingerprint=self.config_fingerprint,
-            input_fingerprint=self.input_fingerprint,
-            stage_ids=steps,
-            resume_run_id=resume_run_id,
-        )
-        def stage(operation):
-            def execute(_context):
-                return operation()
-
-            return execute
-
-        return run_job(spec, tuple(stage(self._operations[step]) for step in steps))
 
 
 def fingerprint(value: str) -> str: