feat(preprocess): add resumable DWH jobs

harness/tests/test_dwh_preprocess_job.py

@@ -0,0 +1,83 @@
+import hashlib
+import json
+
+from typer.testing import CliRunner
+
+from tht.cli import app
+from tht.jobs.dwh_pipeline import DwhPreprocessPipeline
+from tht.jobs.locking import _lock_name
+
+
+FP = "sha256:" + hashlib.sha256(b"test").hexdigest()
+
+
+def test_dwh_and_evidence_jobs_have_distinct_lock_names():
+    assert _lock_name("demo", "dwh") != _lock_name("demo", "evidence")
+
+
+def test_selected_dwh_stages_run_in_declared_order(tmp_path):
+    calls = []
+    pipeline = DwhPreprocessPipeline(
+        workspace_id="demo",
+        workspace_root=tmp_path,
+        config_fingerprint=FP,
+        input_fingerprint=FP,
+        introspect=lambda: calls.append("introspect"),
+        build_lsh=lambda: calls.append("lsh"),
+    )
+
+    report = pipeline.run(("introspect", "lsh"))
+
+    assert report.status == "succeeded"
+    assert calls == ["introspect", "lsh"]
+    assert [stage.name for stage in report.stages] == ["introspect", "lsh"]
+
+
+def test_preprocess_dwh_json_is_pristine(monkeypatch, tmp_path):
+    import tht.cli.preprocess_cmd as command
+
+    class Report:
+        status = "succeeded"
+
+        def model_dump(self, mode=None):
+            return {"status": "succeeded", "run_id": "a" * 32, "stages": []}
+
+    seen = {}
+
+    def run(config, *, steps, resume):
+        seen.update(config=config, steps=steps, resume=resume)
+        return Report()
+
+    monkeypatch.setattr(command, "run_dwh_from_config", run)
+    response = CliRunner().invoke(
+        app,
+        [
+            "preprocess", "dwh", "--steps", "introspect,lsh", "--json",
+            "-c", str(tmp_path / "workspace.yaml"),
+        ],
+    )
+
+    assert response.exit_code == 0, response.output
+    assert json.loads(response.output)["run_id"] == "a" * 32
+    assert seen["steps"] == ("introspect", "lsh")
+
+
+def test_preprocess_dwh_rejects_unknown_or_duplicate_steps(monkeypatch, tmp_path):
+    import tht.cli.preprocess_cmd as command
+
+    called = False
+
+    def forbidden(*args, **kwargs):
+        nonlocal called
+        called = True
+
+    monkeypatch.setattr(command, "run_dwh_from_config", forbidden)
+    runner = CliRunner()
+    for value in ("introspect,unknown", "lsh,lsh", ""):
+        response = runner.invoke(
+            app,
+            ["preprocess", "dwh", "--steps", value, "--json", "-c", str(tmp_path / "w.yaml")],
+        )
+        assert response.exit_code == 2
+        assert json.loads(response.output)["status"] == "failed"
+    assert called is False

harness/tests/test_lsh_job_resume.py

@@ -0,0 +1,66 @@
+import hashlib
+
+from tht.jobs.dwh_pipeline import DwhPreprocessPipeline
+
+
+FP = "sha256:" + hashlib.sha256(b"test").hexdigest()
+
+
+def test_lsh_failure_resumes_exact_run_without_repeating_introspection(tmp_path):
+    calls = []
+
+    def introspect():
+        calls.append("introspect")
+
+    def fail_lsh():
+        calls.append("lsh-failed")
+        raise RuntimeError("database detail that must not leak")
+
+    failed = DwhPreprocessPipeline(
+        workspace_id="demo",
+        workspace_root=tmp_path,
+        config_fingerprint=FP,
+        input_fingerprint=FP,
+        introspect=introspect,
+        build_lsh=fail_lsh,
+    ).run(("introspect", "lsh"))
+    assert failed.status == "failed"
+
+    resumed = DwhPreprocessPipeline(
+        workspace_id="demo",
+        workspace_root=tmp_path,
+        config_fingerprint=FP,
+        input_fingerprint=FP,
+        introspect=introspect,
+        build_lsh=lambda: calls.append("lsh-recovered"),
+    ).run(("introspect", "lsh"), resume_run_id=failed.run_id)
+
+    assert resumed.status == "succeeded"
+    assert resumed.resumed_from == failed.run_id
+    assert calls == ["introspect", "lsh-failed", "lsh-recovered"]
+
+
+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,
+        build_lsh=lambda: (_ for _ in ()).throw(RuntimeError()),
+    ).run(("introspect", "lsh"))
+
+    pipeline = DwhPreprocessPipeline(
+        workspace_id="demo",
+        workspace_root=tmp_path,
+        config_fingerprint=FP,
+        input_fingerprint=FP,
+        introspect=lambda: None,
+        build_lsh=lambda: None,
+    )
+    try:
+        pipeline.run(("lsh",), resume_run_id=failed.run_id)
+    except Exception as error:
+        assert "incompatible" in str(error)
+    else:
+        raise AssertionError("resume with different stages must fail")

harness/tht/cli/lsh_cmd.py

@@ -36,12 +36,30 @@ def _extract_lsh_values(dwh, physical, annotations, limit):
     return values, skipped, truncated
 
 
-@lsh_app.command("build")
+def build_lsh_artifacts(cfg, *, dwh=None, verbose: bool = False):
-def build_cmd(config: Path = CONFIG_OPT) -> None:
+    """Run the existing LSH extraction/build algorithm and persist its outputs."""
-    """Costruisce l'indice LSH dai valori del database e lo salva su pickle."""
+    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)
+    if not phys_file.exists():
+        raise FileNotFoundError("physical catalog is missing; run schema introspect first")
+    physical = PhysicalSchema.from_yaml(phys_file)
+    annotations = Annotations.from_yaml(annotations_path(cfg))
+    target = dwh if dwh is not None else build_dwh(cfg)
+    values, skipped, truncated = _extract_lsh_values(
+        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)
+    return minhashes, skipped, truncated, values
+
+
+@lsh_app.command("build")
+def build_cmd(config: Path = CONFIG_OPT) -> None:
+    """Costruisce l'indice LSH dai valori del database e lo salva su pickle."""
     cfg = _load_config_or_exit(config)
     phys_file = physical_path(cfg)
     if not phys_file.exists():
@@ -50,17 +68,8 @@ def build_cmd(config: Path = CONFIG_OPT) -> None:
             fg=typer.colors.RED, err=True,
         )
         raise typer.Exit(code=1)
-    physical = PhysicalSchema.from_yaml(phys_file)
-    from tht.cli.schema_cmd import annotations_path
-
-    annotations = Annotations.from_yaml(annotations_path(cfg))
-
     typer.echo("Estrazione valori (i più frequenti) dalle colonne testuali eligible...")
-    from tht.adapters.factory import build_dwh
+    minhashes, skipped, truncated, values = build_lsh_artifacts(cfg, verbose=True)
-    dwh = build_dwh(cfg)
-    values, skipped, truncated = _extract_lsh_values(
-        dwh, physical, annotations, cfg.lsh.max_values_per_column
-    )
     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")
     for s in skipped:
@@ -72,8 +81,6 @@ def build_cmd(config: Path = CONFIG_OPT) -> None:
             fg=typer.colors.YELLOW,
         )
 
-    lsh, minhashes = build_index(values, cfg.lsh, verbose=True)
-    save_index(lsh, minhashes, cfg.lsh, _lsh_dir(cfg), name=cfg.database.db_schema)
     typer.secho(
         f"OK: indice LSH ({len(minhashes)} entry) -> {_lsh_dir(cfg)}", fg=typer.colors.GREEN
     )

harness/tht/cli/preprocess_cmd.py

@@ -15,6 +15,39 @@ from tht.cli.config_cmd import CONFIG_OPT
 preprocess_app = typer.Typer(help="Materialize versioned preprocessing artifacts")
 
 
+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.jobs.dwh_pipeline import DwhPreprocessPipeline, fingerprint
+
+    cfg = _load_config_or_exit(config)
+    workspace_root = cfg.paths.artifacts.parent
+    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),
+        build_lsh=lambda: build_lsh_artifacts(cfg),
+    )
+    return pipeline.run(steps, resume_run_id=resume)
+
+
+def _parse_dwh_steps(value: str) -> tuple[str, ...]:
+    allowed = ("introspect", "lsh")
+    steps = tuple(part.strip() for part in value.split(",") if part.strip())
+    if (
+        not steps
+        or len(steps) != len(set(steps))
+        or any(step not in allowed for step in steps)
+        or tuple(sorted(steps, key=allowed.index)) != steps
+    ):
+        raise ValueError("steps must be a unique ordered subset of introspect,lsh")
+    return steps
+
+
 def run_from_config(config: Path, *, dry_run: bool = False, resume: str | None = None):
     from tht.adapters.factory import build_evidence_sources, build_vector_store
     from tht.cli.schema_cmd import _load_config_or_exit
@@ -122,3 +155,46 @@ def evidence_cmd(
             f"OK: run={payload['run_id']} generation={payload['generation']} changed={len(payload['changed'])} "
             f"unchanged={len(payload['unchanged'])} removed={len(payload['removed'])}"
         )
+
+
+@preprocess_app.command("dwh")
+def dwh_cmd(
+    config: Path = CONFIG_OPT,
+    steps: str = typer.Option("introspect,lsh", "--steps"),
+    resume: str | None = typer.Option(None, "--resume"),
+    json_output: bool = typer.Option(False, "--json"),
+) -> None:
+    try:
+        selected = _parse_dwh_steps(steps)
+    except ValueError:
+        payload = {"status": "failed", "error": "invalid DWH preprocessing steps"}
+        if json_output:
+            typer.echo(json.dumps(payload, sort_keys=True))
+        else:
+            typer.secho("ERRORE: invalid DWH preprocessing steps", fg=typer.colors.RED, err=True)
+        raise typer.Exit(code=2) from None
+    if resume is not None and re.fullmatch(r"[0-9a-f]{32}", resume) is None:
+        payload = {"status": "failed", "error": "resume requires a preprocessing run id"}
+        if json_output:
+            typer.echo(json.dumps(payload, sort_keys=True))
+        else:
+            typer.secho("ERRORE: resume requires a preprocessing run id", fg=typer.colors.RED, err=True)
+        raise typer.Exit(code=2)
+    try:
+        result = run_dwh_from_config(config, steps=selected, resume=resume)
+    except Exception:
+        payload = {"status": "failed", "error": "DWH preprocessing failed"}
+        if json_output:
+            typer.echo(json.dumps(payload, sort_keys=True))
+        else:
+            typer.secho("ERRORE: DWH preprocessing failed", fg=typer.colors.RED, err=True)
+        raise typer.Exit(code=1) from None
+    payload = result.model_dump(mode="json")
+    if json_output:
+        typer.echo(json.dumps(payload, ensure_ascii=False, sort_keys=True))
+    elif result.status == "succeeded":
+        typer.echo(f"OK: run={result.run_id} stages={','.join(selected)}")
+    else:
+        typer.secho(f"ERRORE: run={result.run_id} DWH preprocessing failed", fg=typer.colors.RED, err=True)
+    if result.status != "succeeded":
+        raise typer.Exit(code=1)

harness/tht/cli/schema_cmd.py

@@ -44,6 +44,16 @@ def annotations_path(cfg) -> Path:
     return cfg.paths.artifacts / "mschema" / "annotations.yaml"
 
 
+def refresh_catalog(cfg, *, dwh=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))
+    return physical
+
+
 @schema_app.command("introspect")
 def introspect_cmd(
     config: Path = CONFIG_OPT,
@@ -80,14 +90,10 @@ def introspect_cmd(
             )
             return
     try:
-        dwh = build_dwh(cfg)
+        phys = refresh_catalog(cfg)
-        phys = dwh.introspect()
-        _add_examples(dwh, phys, cfg.examples)
-        classify_all(phys, cfg.eligibility)
     except Exception as e:
         typer.secho(f"ERRORE: {e}", fg=typer.colors.RED, err=True)
         raise typer.Exit(code=1)
-    phys.to_yaml(out)
     n_cols = sum(len(t.columns) for t in phys.tables.values())
     n_ignored = sum(
         1 for t in phys.tables.values() for c in t.columns.values() if not c.eligible

harness/tht/jobs/dwh_pipeline.py

@@ -0,0 +1,65 @@
+"""Resumable DWH catalog and LSH preprocessing stages."""
+
+from __future__ import annotations
+
+import hashlib
+from collections.abc import Callable
+from pathlib import Path
+
+from tht.jobs.models import JobReport, JobSpec
+from tht.jobs.runner import run_job
+
+
+DWH_STAGE_IDS = ("introspect", "lsh")
+
+
+class DwhPreprocessPipeline:
+    """Adapt existing DWH preprocessing operations to the shared job envelope."""
+
+    def __init__(
+        self,
+        *,
+        workspace_id: str,
+        workspace_root: Path,
+        config_fingerprint: str,
+        input_fingerprint: str,
+        introspect: Callable[[], object],
+        build_lsh: Callable[[], object],
+    ) -> 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}
+
+    def run(
+        self, steps: tuple[str, ...] = DWH_STAGE_IDS, *, resume_run_id: str | None = None
+    ) -> JobReport:
+        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:
+    return "sha256:" + hashlib.sha256(value.encode("utf-8")).hexdigest()