From 3a9808f12667d205f1ce8a83514974d93e5c6845 Mon Sep 17 00:00:00 2001 From: mptyl Date: Sun, 12 Jul 2026 05:34:38 +0200 Subject: [PATCH] feat(preprocess): add resumable DWH jobs --- harness/tests/test_dwh_preprocess_job.py | 83 ++++++++++++++++++++++++ harness/tests/test_lsh_job_resume.py | 66 +++++++++++++++++++ harness/tht/cli/lsh_cmd.py | 37 ++++++----- harness/tht/cli/preprocess_cmd.py | 76 ++++++++++++++++++++++ harness/tht/cli/schema_cmd.py | 16 +++-- harness/tht/jobs/dwh_pipeline.py | 65 +++++++++++++++++++ 6 files changed, 323 insertions(+), 20 deletions(-) create mode 100644 harness/tests/test_dwh_preprocess_job.py create mode 100644 harness/tests/test_lsh_job_resume.py create mode 100644 harness/tht/jobs/dwh_pipeline.py diff --git a/harness/tests/test_dwh_preprocess_job.py b/harness/tests/test_dwh_preprocess_job.py new file mode 100644 index 00000000..9a650c5c --- /dev/null +++ b/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 diff --git a/harness/tests/test_lsh_job_resume.py b/harness/tests/test_lsh_job_resume.py new file mode 100644 index 00000000..78a776e3 --- /dev/null +++ b/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") diff --git a/harness/tht/cli/lsh_cmd.py b/harness/tht/cli/lsh_cmd.py index ffedbdf1..16210ae0 100644 --- a/harness/tht/cli/lsh_cmd.py +++ b/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_cmd(config: Path = CONFIG_OPT) -> None: - """Costruisce l'indice LSH dai valori del database e lo salva su pickle.""" +def build_lsh_artifacts(cfg, *, dwh=None, verbose: bool = False): + """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) + 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 - dwh = build_dwh(cfg) - values, skipped, truncated = _extract_lsh_values( - dwh, physical, annotations, cfg.lsh.max_values_per_column - ) + minhashes, skipped, truncated, values = build_lsh_artifacts(cfg, verbose=True) 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 ) diff --git a/harness/tht/cli/preprocess_cmd.py b/harness/tht/cli/preprocess_cmd.py index be070eb7..0e627a42 100644 --- a/harness/tht/cli/preprocess_cmd.py +++ b/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) diff --git a/harness/tht/cli/schema_cmd.py b/harness/tht/cli/schema_cmd.py index c0cfacd5..eaaa6eb8 100644 --- a/harness/tht/cli/schema_cmd.py +++ b/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 = dwh.introspect() - _add_examples(dwh, phys, cfg.examples) - classify_all(phys, cfg.eligibility) + phys = refresh_catalog(cfg) 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 diff --git a/harness/tht/jobs/dwh_pipeline.py b/harness/tht/jobs/dwh_pipeline.py new file mode 100644 index 00000000..72e32e55 --- /dev/null +++ b/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()