refactor(cli): remove obsolete workflow commands

harness/README.md

@@ -54,7 +54,7 @@ tht-workspace-<customer>/     per-customer workspace repo (separate git repo)
 ├── <customer>.yaml           workspace YAML (paths ABSOLUTE, anchored here)
 ├── evidence/                 curated evidence markdown (frontmatter-structured, by domain)
 ├── artifacts/                derived artifacts (gitignored; regenerated by tht)
-├── indexes/                  LSH index (gitignored; regenerated by `tht lsh build`)
+├── indexes/                  LSH index (gitignored; regenerated by `tht preprocess dwh --steps lsh`)
 └── sessions/                 per-session artifacts (gitignored)
 ```
 
@@ -64,7 +64,7 @@ All `tht` commands take the customer workspace via `-c`:
 
 ```bash
 tht schema introspect -c /path/to/tht-workspace-<customer>/<customer>.yaml
-tht lsh build        -c /path/to/tht-workspace-<customer>/<customer>.yaml
+tht preprocess dwh --steps lsh -c /path/to/tht-workspace-<customer>/<customer>.yaml
 tht session new      -c /path/to/tht-workspace-<customer>/<customer>.yaml
 ```
 

harness/tests/fixtures/approved_cli_surface.json

@@ -0,0 +1,85 @@
+{
+  "maintained": [
+    "cte info",
+    "cte next",
+    "cte plan",
+    "cte save",
+    "cte test",
+    "datamart generate",
+    "db ping",
+    "decision add",
+    "decision add-batch",
+    "decision add-join-set",
+    "memory promote",
+    "memory save-one",
+    "memory search",
+    "memory solved-index",
+    "memory solved-search",
+    "ollama ensure",
+    "phase advance",
+    "phase meta",
+    "phase reopen",
+    "phase show",
+    "preprocess dwh",
+    "preprocess evidence",
+    "schema check",
+    "schema columns",
+    "schema introspect",
+    "schema render",
+    "schema suggest-fks",
+    "search find",
+    "search pack",
+    "session archive",
+    "session check",
+    "session close",
+    "session delete",
+    "session documents",
+    "session fail",
+    "session finalize",
+    "session list",
+    "session migrate",
+    "session new",
+    "session preferences get",
+    "session preferences set",
+    "session reopen",
+    "session retrieval-pack",
+    "session set-group",
+    "session set-name",
+    "session set-question",
+    "session set-schema-linking",
+    "session show",
+    "session sync-schema-linking",
+    "session unarchive",
+    "sql export",
+    "sql preview",
+    "sql set-final",
+    "sql validate",
+    "vector index-schema"
+  ],
+  "enhanced": [
+    "config check",
+    "db fetch-ca",
+    "doctor",
+    "memory delete",
+    "memory index",
+    "memory list",
+    "memory show",
+    "memory update"
+  ],
+  "erased": [
+    "cte list",
+    "decision list",
+    "decision retract",
+    "evidence extract",
+    "evidence index",
+    "formula list",
+    "formula save",
+    "lsh build",
+    "lsh query",
+    "memory clear",
+    "memory migrate",
+    "sql explain",
+    "sql save",
+    "vector init"
+  ]
+}

harness/tests/integration/test_gate_cli_signatures.py

@@ -16,13 +16,48 @@ import sys
 from pathlib import Path
 
 import pytest
+from typer.main import get_command
+
+from tht.cli import app
 
 _ROOT = Path(__file__).resolve().parent.parent.parent
 _GATE = _ROOT / ".pi" / "extensions" / "tht-gate.js"
 _THT = Path(sys.executable).parent / "tht"
 
 # The command groups the gate drives. Anything else in an array literal is data, not a CLI call.
-_GROUPS = {"phase", "session", "cte", "decision", "memory", "search", "schema", "vector"}
+_GROUPS = {"phase", "session", "cte", "decision", "memory", "search", "schema", "sql", "vector"}
+
+# Python's workflow CLI is also a contract for its non-gate consumers.  Keep this
+# list literal rather than deriving it from the current tree: a deleted live
+# command must fail this test before a container or preprocessing job reaches it.
+_LIVE_CONSUMER_COMMANDS = {
+    "Pi skill": {
+        ("cte", "info"), ("cte", "save"), ("cte", "test"),
+        ("datamart", "generate"), ("memory", "promote"),
+        ("memory", "save-one"), ("memory", "search"),
+        ("memory", "solved-index"), ("memory", "solved-search"),
+        ("phase", "advance"), ("phase", "reopen"),
+        ("schema", "introspect"), ("schema", "render"),
+        ("search", "find"), ("search", "pack"),
+        ("session", "check"), ("session", "documents"),
+        ("session", "finalize"), ("session", "show"),
+        ("sql", "preview"), ("sql", "set-final"), ("sql", "validate"),
+    },
+    "backend ThtRunner": {
+        ("db", "ping"), ("ollama", "ensure"), ("search", "pack"),
+        ("session", "archive"), ("session", "close"), ("session", "delete"),
+        ("session", "documents"), ("session", "fail"), ("session", "list"),
+        ("session", "new"), ("session", "reopen"), ("session", "set-group"),
+        ("session", "set-name"), ("session", "show"), ("session", "unarchive"),
+        ("session", "preferences", "get"), ("session", "preferences", "set"),
+        ("sql", "export"), ("sql", "preview"),
+    },
+    "workspace preprocessing": {
+        ("preprocess", "dwh"), ("preprocess", "evidence"),
+        ("schema", "check"), ("schema", "suggest-fks"), ("vector", "index-schema"),
+    },
+    "deployment smoke": {("config", "check"), ("db", "ping")},
+}
 
 # Flags that are framework/JS artifacts, never real CLI options (skip from the check).
 _SKIP_FLAGS: set[str] = set()
@@ -53,6 +88,30 @@ def test_gate_invokes_at_least_the_known_commands():
     assert ("cte", "plan") in pairs
     assert ("session", "set-question") in pairs
     assert ("decision", "add") in pairs
+    assert ("sql", "set-final") in pairs
+
+
+def _leaf_paths(command, prefix: tuple[str, ...] = ()) -> set[tuple[str, ...]]:
+    children = getattr(command, "commands", {})
+    if not children:
+        return {prefix} if prefix else set()
+    paths: set[tuple[str, ...]] = set()
+    for name, child in children.items():
+        paths.update(_leaf_paths(child, prefix + (name,)))
+    return paths
+
+
+@pytest.mark.parametrize(
+    ("consumer", "path"),
+    [
+        (consumer, path)
+        for consumer, paths in _LIVE_CONSUMER_COMMANDS.items()
+        for path in sorted(paths)
+    ],
+)
+def test_live_workflow_consumers_keep_their_command_paths(consumer: str, path: tuple[str, ...]):
+    """Live Pi, backend, preprocessing, and smoke paths must remain callable."""
+    assert path in _leaf_paths(get_command(app)), f"{consumer} requires `tht {' '.join(path)}`"
 
 
 @pytest.mark.skipif(not _THT.exists(), reason="tht CLI not installed in this venv")
@@ -60,7 +119,7 @@ def test_gate_invokes_at_least_the_known_commands():
 def test_gate_call_site_matches_cli(group: str, sub: str, flags: list[str], raw: str):
     res = subprocess.run(
         [str(_THT), group, sub, "--help"],
-        capture_output=True, text=True, cwd=str(_ROOT),
+        capture_output=True, check=False, text=True, cwd=str(_ROOT),
     )
     assert res.returncode == 0, (
         f"gate calls `tht {group} {sub}` but it does not exist in the CLI.\n"

harness/tests/l2/test_value_grounding_real.py

@@ -22,16 +22,16 @@ WORKSPACE = Path(os.environ.get("THT_L2_WORKSPACE", "/Users/mp/projects/tht-work
 def test_ablazione_returns_multiple_columns(l2_env):
     """On the real schema, 'ablazione' should ground to more than one column (e.g.
     a flag and a free-text patologia field) -- the whole point of D14a's
-    non-collapsing aggregation. Requires a built LSH index (tht lsh build)."""
+    non-collapsing aggregation. Requires a built LSH index (preprocess dwh --steps lsh)."""
     if not WORKSPACE.exists():
         pytest.skip(f"per-customer workspace repo not found: {WORKSPACE} (clone tht-workspace-psd)")
     try:
         from tht.lshindex import load_index, query_index  # ported with the lsh build path
     except ModuleNotFoundError:
-        pytest.skip("tht.lshindex not yet ported (deferred from B3; lands with tht lsh build)")
+        pytest.skip("tht.lshindex not yet ported (deferred from B3 preprocessing)")
     from tht.search import aggregate_lsh_multi
 
-    # The index lives at <indexes>/lsh/<schema>_*.pkl (same convention as tht lsh build/query),
+    # The index lives at <indexes>/lsh/<schema>_*.pkl (the preprocess-dwh convention),
     # and is keyed by the database schema name (cfg.database.db_schema), not a hard-coded literal.
     ws = load_workspace(WORKSPACE)
     index_dir = ws.paths.indexes / "lsh"
@@ -39,7 +39,7 @@ def test_ablazione_returns_multiple_columns(l2_env):
     try:
         lsh, minhashes, meta = load_index(index_dir, schema_name)
     except Exception as e:
-        pytest.skip(f"LSH index not built yet (run tht lsh build -c {WORKSPACE}): {e}")
+        pytest.skip(f"LSH index not built yet (run tht preprocess dwh --steps lsh -c {WORKSPACE}): {e}")
 
     hits = query_index(lsh, minhashes, "ablazione", meta, top_n=20)
     grouped = aggregate_lsh_multi(

harness/tests/test_cli_surface.py

@@ -0,0 +1,48 @@
+"""The public Python CLI exposes exactly the approved workflow surface."""
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from typer.main import get_command
+from typer.testing import CliRunner
+
+from tht.cli import app
+
+_FIXTURE = Path(__file__).parent / "fixtures" / "approved_cli_surface.json"
+
+
+def _leaf_paths(command, prefix: tuple[str, ...] = ()) -> set[str]:
+    """Return the terminal paths offered by the real Typer/Click command tree."""
+    children = getattr(command, "commands", {})
+    if not children:
+        return {" ".join(prefix)} if prefix else set()
+    paths: set[str] = set()
+    for name, child in children.items():
+        paths.update(_leaf_paths(child, prefix + (name,)))
+    return paths
+
+
+def _approved_surface() -> dict[str, list[str]]:
+    return json.loads(_FIXTURE.read_text(encoding="utf-8"))
+
+
+def test_typer_tree_matches_the_approved_command_surface():
+    """Catches a missing live command or an accidentally re-registered erased command."""
+    approved = _approved_surface()
+    expected = set(approved["maintained"]) | set(approved["enhanced"])
+
+    assert len(approved["maintained"]) == 55
+    assert len(approved["enhanced"]) == 8
+    assert len(approved["erased"]) == 14
+    assert not (expected & set(approved["erased"]))
+    assert _leaf_paths(get_command(app)) == expected
+
+
+def test_erased_commands_are_rejected_with_the_standard_typer_error():
+    """Catches a regression that restores a retired command under an existing group."""
+    runner = CliRunner()
+    for path in _approved_surface()["erased"]:
+        result = runner.invoke(app, [*path.split(), "--help"])
+        assert result.exit_code != 0, path
+        assert "no such command" in result.output.casefold(), path

harness/tests/test_decision_retract_cli.py

@@ -1,63 +0,0 @@
-"""D15 granularity (a): `tht decision retract` tombstones the last substantive decision.
-
-Wires the step-granularity rollback ("re-ask current widget, discard last answer") to a
-reachable CLI command. The data model (decision_retracted + retracts, honored by
-effective_decisions) already existed; this pins the command that emits it.
-"""
-from pathlib import Path
-
-from tht.cli.decision_cmd import retract_cmd
-from tht.decisions import append_decision, list_decisions
-from tht.phase import current_phase, effective_decisions
-
-
-def _walk_to_phase(session: Path, target: int) -> None:
-    while current_phase(session) < target:
-        append_decision(session, type="phase_approved", subject=f"phase:{current_phase(session)}")
-
-
-def test_retract_drops_last_substantive_decision(tmp_path, monkeypatch):
-    s = tmp_path / "2026-01-01-000000-x"
-    s.mkdir(parents=True)
-    _walk_to_phase(s, 4)
-    append_decision(s, type="table_promoted", subject="phase:4", detail="dim_pazienti")
-    append_decision(s, type="table_promoted", subject="phase:4", detail="fact_ricoveri")
-
-    # stub config + session loading (the command only needs a session dir)
-    import tht.cli.decision_cmd as mod
-    import tht.cli.session_cmd as session_mod
-    from tht.session.models import SessionManifest, SessionSnapshot
-
-    class _Cfg:
-        class paths:
-            sessions = tmp_path
-
-    monkeypatch.setattr(mod, "_load_config_or_exit", lambda _c: _Cfg())
-    monkeypatch.setattr(mod, "load_session_or_exit", lambda _cfg, _s: None)
-    class _Repository:
-        def get(self, session_id):
-            return SessionSnapshot(
-                manifest=SessionManifest(id=session_id, created_at="2026-01-01T00:00:00Z", question="q", database="d", schema="s"),
-                decisions=list_decisions(tmp_path / session_id),
-            )
-
-        def append_decisions(self, session_id, decisions):
-            from tht.decisions import append_decisions
-
-            return append_decisions(tmp_path / session_id, list(decisions))
-
-    repository = _Repository()
-    monkeypatch.setattr(mod, "load_snapshot_or_exit", lambda _cfg, sid: repository.get(sid))
-    monkeypatch.setattr(mod, "session_repository", lambda _cfg: repository)
-    monkeypatch.setattr(session_mod, "load_snapshot_or_exit", lambda _cfg, sid: repository.get(sid))
-
-    retract_cmd(session="2026-01-01-000000-x", config=Path("x"))
-
-    eff = effective_decisions(s)
-    promoted = [d for d in eff if d.type == "table_promoted"]
-    assert len(promoted) == 1  # the last one was retracted
-    assert promoted[0].detail == "dim_pazienti"
-    # the audit log keeps everything (append-only): 2 promotions + the retract marker
-    raw = [d.type for d in list_decisions(s)]
-    assert raw.count("table_promoted") == 2
-    assert "decision_retracted" in raw

harness/tests/test_qdrant_cli_commands.py

@@ -21,7 +21,6 @@ class _FakeEmbedder:
 class _FakeVectorStore:
     def __init__(self):
         self.upserts = []
-        self.deleted = []
 
     def existing_hashes(self, collection, kinds):
         return {}
@@ -30,11 +29,6 @@ class _FakeVectorStore:
         self.upserts.append((collection, records))
         return len(records)
 
-    def delete_kinds(self, collection, kinds):
-        self.deleted.append((collection, list(kinds)))
-        return 3
-
-
 def _sha_file(path: Path) -> str:
     return "sha256:" + hashlib.sha256(path.read_bytes()).hexdigest()
 
@@ -220,28 +214,12 @@ def test_memory_index_accepts_qdrant_only_runtime_config(tmp_path, monkeypatch):
     assert store.upserts
 
 
-def test_memory_clear_accepts_qdrant_only_runtime_config(tmp_path, monkeypatch):
-    cfg = _qdrant_runtime_config(tmp_path)
-    store = _FakeVectorStore()
-    records = [_memory_record()]
-    registry = tmp_path / "artifacts" / "memory" / "registry.jsonl"
-    save_registry(records, registry)
-
-    monkeypatch.setattr("tht.adapters.factory.build_vector_store", lambda cfg, require_write: store)
-
-    res = CliRunner().invoke(app, ["memory", "clear", "--yes", "-c", str(cfg)])
-
-    assert res.exit_code == 0, res.output
-    assert store.deleted == [("memory", ["memory"])]
-    assert not registry.exists()
-
-
-def test_vector_help_does_not_expose_migrate_and_keeps_qdrant_commands():
+def test_vector_help_exposes_only_the_supported_qdrant_command():
     res = CliRunner().invoke(app, ["vector", "--help"])
 
     assert res.exit_code == 0, res.output
     assert "migrate" not in res.output
-    assert "init" in res.output
+    assert "init" not in res.output
     assert "index-schema" in res.output
 
 

harness/tests/test_schema_introspect_guard.py

@@ -63,35 +63,6 @@ def test_introspect_fresh_root_initializes_through_writer_job(tmp_path, monkeypa
     assert "1 tabelle" in res.output
 
 
-def test_lsh_build_fresh_root_initializes_introspection_and_lsh(tmp_path, monkeypatch):
-    import tht.cli.lsh_cmd as lsh_module
-    import tht.cli.schema_cmd as schema_module
-    import tht.lshindex as lshindex_module
-
-    cfg = _write_config(tmp_path)
-    physical = PhysicalSchema(
-        database="d", schema="s", introspected_at=datetime(2026, 1, 1),
-        tables={"dim_patient": TablePhysical(columns={"id": ColumnPhysical(type="bigint")})},
-    )
-
-    def refresh(_cfg, *, output_path=None, **_kwargs):
-        physical.to_yaml(output_path)
-        return physical
-
-    def build(_cfg, *, physical_file, output_dir, **_kwargs):
-        assert physical_file.is_file()
-        for name in ("s_lsh.pkl", "s_minhashes.pkl", "s_meta.json"):
-            (output_dir / name).write_text("index")
-        return {}, [], [], {}
-
-    monkeypatch.setattr(schema_module, "refresh_catalog", refresh)
-    monkeypatch.setattr(lsh_module, "build_lsh_artifacts", build)
-    monkeypatch.setattr(lshindex_module, "load_index", lambda *_args, **_kwargs: (None, {}, None))
-    res = CliRunner().invoke(app, ["lsh", "build", "-c", str(cfg)])
-    assert res.exit_code == 0, res.output
-    assert (tmp_path / ".tht-dwh" / "OWNER.json").is_file()
-
-
 def test_introspect_refresh_bypasses_cache(tmp_path):
     _write_catalog(tmp_path)
     cfg = _write_config(tmp_path)

harness/tht/cli/__init__.py

@@ -1,8 +1,8 @@
 """tht CLI -- Typer app `tht`.
 
 Minimal skeleton for now: registers `phase_app` (the gate needs `tht phase meta --json`).
-Other command groups (db, schema, lsh, vector, memory, session, decision, sql, cte,
-evidence, search) are wired in their porting tasks, once their backend deps land
+Other command groups (db, schema, vector, memory, session, decision, sql, cte,
+search) are wired in their porting tasks, once their backend deps land
 (A9 ports db/mschema/_guards; B1 ports vectorstore; B3 ports search/lshindex/sampling).
 """
 from __future__ import annotations
@@ -42,9 +42,6 @@ from tht.cli.datamart_cmd import datamart_app
 from tht.cli.db_cmd import db_app
 from tht.cli.decision_cmd import decision_app
 from tht.cli.doctor_cmd import doctor
-from tht.cli.evidence_cmd import evidence_app
-from tht.cli.formula_cmd import formula_app
-from tht.cli.lsh_cmd import lsh_app
 from tht.cli.memory_cmd import memory_app
 from tht.cli.ollama_cmd import ollama_app
 from tht.cli.phase_cmd import phase_app
@@ -63,13 +60,10 @@ app.add_typer(session_app, name="session")
 app.add_typer(vector_app, name="vector")
 app.add_typer(memory_app, name="memory")
 app.add_typer(search_app, name="search")
-app.add_typer(evidence_app, name="evidence")
-app.add_typer(formula_app, name="formula")
 app.add_typer(db_app, name="db")
 app.add_typer(decision_app, name="decision")
 app.add_typer(sql_app, name="sql")
 app.add_typer(cte_app, name="cte")
 app.add_typer(datamart_app, name="datamart")
-app.add_typer(lsh_app, name="lsh")
 app.add_typer(ollama_app, name="ollama")
 app.command("doctor")(doctor)

harness/tht/cli/cte_cmd.py

@@ -291,34 +291,3 @@ def info_cmd(
         return
     typer.echo(f"{name}: {payload['index']}/{payload['total']} — "
                f"approvato={payload['approved']}")
-
-
-@cte_app.command("list")
-def list_cmd(
-    session: str = typer.Option(..., "--session"),
-    config: Path = CONFIG_OPT,
-) -> None:
-    """Ultimo esito registrato per ogni CTE della sessione."""
-    from tht.ctetest import CteError, load_cte_tests_text
-
-    cfg = _load_config_or_exit(config)
-    load_session_or_exit(cfg, session)
-    try:
-        records = load_cte_tests_text(load_snapshot_or_exit(cfg, session).artifacts.get("cte_tests", ""))
-    except CteError as e:
-        typer.secho(f"ERRORE: impossibile leggere cte_tests.json: {e}",
-                    fg=typer.colors.RED, err=True)
-        raise typer.Exit(code=1)
-    if not records:
-        typer.echo("Nessun test CTE registrato.")
-        return
-    latest = {}
-    for r in records:
-        latest[r.name] = r
-    for name, r in sorted(latest.items()):
-        line = f"{name}: {r.status} ({r.ts:%Y-%m-%d %H:%M}, {r.execution_ms} ms)"
-        if r.error:
-            line += f" — {r.error}"
-        if r.warnings:
-            line += f" [{len(r.warnings)} warning]"
-        typer.echo(line)

harness/tht/cli/decision_cmd.py

@@ -162,63 +162,3 @@ def add_cmd(
     }])[0]
     typer.secho(f"OK: decisione [{record.seq}] {record.type}: {record.subject}",
                 fg=typer.colors.GREEN)
-
-
-@decision_app.command("retract")
-def retract_cmd(
-    session: str = typer.Option(..., "--session", help="Id della sessione."),
-    config: Path = CONFIG_OPT,
-) -> None:
-    """Ritira l'ultima decisione sostanziale della fase corrente (D15 granularita' step).
-
-    Granularita' (a) del rollback §4.8: 'rispondi di nuovo a questa domanda'. Scrive un
-    marker decision_retracted (append-only, l'audit resta) che effective_decisions onora;
-    il widget corrente puo' essere riproposto. Non cambia la fase."""
-    from tht.phase import effective_decisions
-
-    cfg = _load_config_or_exit(config)
-    load_session_or_exit(cfg, session)
-    snapshot = load_snapshot_or_exit(cfg, session)
-
-    # Ultima decisione NON-meta della vista effective = quella associata al widget corrente.
-    meta = {
-        "phase_approved", "phase_auto_approved", "phase_reopened",
-        "phase_skipped", "decision_retracted",
-    }
-    substantive = [d for d in effective_decisions(snapshot) if d.type not in meta]
-    if not substantive:
-        typer.secho(
-            "Nessuna decisione sostanziale da ritirare nella fase corrente.",
-            fg=typer.colors.YELLOW, err=True,
-        )
-        raise typer.Exit(code=6)
-    target = substantive[-1]
-    record = session_repository(cfg).append_decisions(session, [{
-        "type": "decision_retracted", "subject": target.subject,
-        "rationale": f"ritira [{target.seq}] {target.type}", "retracts": target.seq,
-    }])[0]
-    typer.secho(
-        f"OK: ritirata decisione [{target.seq}] {target.type}: {target.subject} "
-        f"(marker #{record.seq}).",
-        fg=typer.colors.GREEN,
-    )
-
-
-@decision_app.command("list")
-def list_cmd(
-    session: str = typer.Option(..., "--session"),
-    config: Path = CONFIG_OPT,
-) -> None:
-    """Elenca le decisioni della sessione."""
-    cfg = _load_config_or_exit(config)
-    decisions = load_snapshot_or_exit(cfg, session).decisions
-    if not decisions:
-        typer.echo("Nessuna decisione registrata.")
-        return
-    for d in decisions:
-        line = f"[{d.seq}] {d.ts:%Y-%m-%d %H:%M} {d.type}: {d.subject}"
-        if d.detail:
-            line += f" — {d.detail}"
-        if d.rationale:
-            line += f" ({d.rationale})"
-        typer.echo(line)

harness/tht/cli/evidence_cmd.py

@@ -1,81 +1,5 @@
 from pathlib import Path
 
-import typer
-
-from tht.cli._guards import require_vector_write_allowed
-from tht.cli.config_cmd import CONFIG_OPT
-from tht.cli.schema_cmd import _load_config_or_exit
-
-evidence_app = typer.Typer(help="Generazione e gestione delle evidence")
-
 
 def evidence_root(cfg) -> Path:
     return cfg.paths.artifacts / "evidence"
-
-
-def _require_evidence_cfg(cfg):
-    if cfg.evidence is None:
-        typer.secho(
-            "ERRORE: sezione `evidence` mancante nel workspace yaml (serve `source_root`).",
-            fg=typer.colors.RED, err=True,
-        )
-        raise typer.Exit(code=1)
-    return cfg.evidence
-
-
-@evidence_app.command("extract")
-def extract_cmd(config: Path = CONFIG_OPT) -> None:
-    """Rispecchia la cartella curata dell'ETL in artifacts/evidence/.
-
-    artifacts/evidence/ e' un artefatto derivato e rigenerabile: viene riallineato a
-    ogni estrazione (le evidence rimosse a monte spariscono anche qui). La gerarchia
-    per dominio della cartella sorgente viene preservata.
-    """
-    import shutil
-
-    from tht.evidence.extract import load_curated
-
-    cfg = _load_config_or_exit(config)
-    sources = _require_evidence_cfg(cfg)
-    out_dir = evidence_root(cfg)
-    if out_dir.exists():
-        shutil.rmtree(out_dir)
-    written = 0
-    for rel, doc in load_curated(sources):
-        doc.save(out_dir / rel)
-        written += 1
-    if written == 0:
-        typer.secho(
-            f"ATTENZIONE: nessuna evidence trovata in "
-            f"{sources.source_root / sources.evidence_dir}",
-            fg=typer.colors.YELLOW, err=True,
-        )
-        raise typer.Exit(code=1)
-    typer.secho(f"OK: {written} evidence rispecchiate in {out_dir}", fg=typer.colors.GREEN)
-
-
-@evidence_app.command("index")
-def index_cmd(config: Path = CONFIG_OPT) -> None:
-    """Embedda e sincronizza nel semantic store tutte le evidence presenti in artifacts/."""
-    from tht.adapters.factory import build_vector_store
-    from tht.cli.vector_cmd import (
-        _print_stats,
-        make_embedder,
-        require_vector_cfg,
-        sync_canonical_records,
-    )
-    from tht.evidence.model import load_evidence_dir
-    from tht.vectorstore.records import evidence_records
-
-    cfg = _load_config_or_exit(config)
-    require_vector_write_allowed(cfg, "evidence index")
-    require_vector_cfg(cfg)
-    docs = load_evidence_dir(evidence_root(cfg))
-    records = evidence_records(docs, cfg.vector.max_chunk_chars)
-    stats = sync_canonical_records(
-        "evidence",
-        records,
-        store=build_vector_store(cfg, require_write=True),
-        embedder=make_embedder(cfg.embeddings),
-    )
-    _print_stats(stats)

harness/tht/cli/formula_cmd.py

@@ -1,81 +0,0 @@
-"""tht formula -- concept->SQL formula evidence store (spec D14b, §4.7.2).
-
-Authoring + listing of reusable concept formulas. Retrieval for the workflow is via
-`tht search find --kind formula "<concept>"`. The reviewer approves/rejects a candidate
-formula in F4 (decisions concept_formula_approved / concept_formula_rejected), and the
-approved one is reflected in schema_linking.json (SchemaLinking.concept_formulas).
-"""
-from pathlib import Path
-
-import typer
-
-from tht.cli.config_cmd import CONFIG_OPT
-from tht.cli.evidence_cmd import evidence_root
-from tht.cli.schema_cmd import _load_config_or_exit
-
-formula_app = typer.Typer(help="Formule di concetto (concept -> SQL) riusabili (D14b)")
-
-
-@formula_app.command("save")
-def save_cmd(
-    concept: str = typer.Option(..., "--concept", help="Il concetto, es. 'fascia pediatrica'."),
-    sql_file: Path = typer.Option(
-        None, "--sql-file", help="File con l'espressione SQL della formula."
-    ),
-    sql: str = typer.Option(None, "--sql", help="Espressione SQL inline (alternativa a --sql-file)."),
-    column: list[str] = typer.Option(None, "--column", help="Colonna usata (ripetibile)."),
-    status: str = typer.Option("draft", "--status", help="auto | draft | reviewed."),
-    source: list[str] = typer.Option(None, "--source", help="Fonte/evidence (ripetibile)."),
-    config: Path = CONFIG_OPT,
-) -> None:
-    """Salva una formula di concetto sotto artifacts/evidence/formulas/."""
-    from tht.evidence.formula_store import ConceptFormula, save_formula
-
-    cfg = _load_config_or_exit(config)
-    if sql_file is None and not sql:
-        typer.secho("ERRORE: indica --sql-file <path> oppure --sql \"<espr>\".",
-                    fg=typer.colors.RED, err=True)
-        raise typer.Exit(code=1)
-    body = sql_file.read_text() if sql_file else sql
-    try:
-        formula = ConceptFormula(
-            concept=concept, columns=column or [], sql=body.strip(),
-            status=status, sources=source or [],
-        )
-    except ValueError as e:
-        typer.secho(f"ERRORE: formula non valida: {e}", fg=typer.colors.RED, err=True)
-        raise typer.Exit(code=1)
-    path = save_formula(evidence_root(cfg), formula)
-    typer.secho(f"OK: formula '{concept}' salvata in {path}.", fg=typer.colors.GREEN)
-
-
-@formula_app.command("list")
-def list_cmd(
-    concept: str = typer.Option(None, "--concept", help="Filtra per concetto (substring)."),
-    json_out: bool = typer.Option(False, "--json"),
-    config: Path = CONFIG_OPT,
-) -> None:
-    """Elenca le formule salvate (tutte, o filtrate per concetto)."""
-    import json as _json
-
-    from rich.console import Console
-    from rich.table import Table
-
-    from tht.evidence.formula_store import _load_all, search_formulas
-
-    cfg = _load_config_or_exit(config)
-    root = evidence_root(cfg)
-    formulas = search_formulas(root, concept) if concept else _load_all(root)
-    if json_out:
-        typer.echo(_json.dumps([f.model_dump(mode="json") for f in formulas],
-                               ensure_ascii=False, indent=2))
-        return
-    if not formulas:
-        typer.secho("Nessuna formula salvata.", fg=typer.colors.YELLOW)
-        return
-    t = Table(title="Formule di concetto")
-    for col in ("Concetto", "Status", "Colonne"):
-        t.add_column(col)
-    for f in formulas:
-        t.add_row(f.concept, f.status, ", ".join(f.columns))
-    Console().print(t)

harness/tht/cli/lsh_cmd.py

@@ -1,17 +1,6 @@
 from pathlib import Path
 
-import typer
-
-from tht.cli.config_cmd import CONFIG_OPT
-from tht.cli.schema_cmd import _load_config_or_exit, physical_path
-
-lsh_app = typer.Typer(help="Indice LSH su valori dei campi (derivato, rigenerabile)")
-
-
-def _lsh_dir(cfg) -> Path:
-    from tht.jobs.dwh_pipeline import resolve_dwh_snapshot
-
-    return resolve_dwh_snapshot(cfg).lsh_dir
+from tht.cli.schema_cmd import physical_path
 
 
 def _extract_lsh_values(dwh, physical, annotations, limit):
@@ -63,79 +52,3 @@ def build_lsh_artifacts(
         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)
-    dwh_root = cfg.paths.artifacts.parent / ".tht-dwh"
-    initialized = dwh_root.exists() or dwh_root.is_symlink()
-    if initialized:
-        phys_file = physical_path(cfg)
-        if not phys_file.exists():
-            typer.secho(
-                f"ERRORE: {phys_file} non trovato. Esegui prima `tht schema introspect`.",
-                fg=typer.colors.RED, err=True,
-            )
-            raise typer.Exit(code=1)
-    typer.echo("Estrazione valori (i più frequenti) dalle colonne testuali eligible...")
-    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 initialized else ("introspect", "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:
-        typer.secho(
-            f"  troncata {t.table}.{t.column}: indicizzati i {t.indexed} valori più frequenti "
-            f"(limite max_values_per_column raggiunto; altri valori distinti NON indicizzati)",
-            fg=typer.colors.YELLOW,
-        )
-
-    typer.secho(
-        f"OK: indice LSH ({len(minhashes)} entry) -> {_lsh_dir(cfg)}", fg=typer.colors.GREEN
-    )
-
-
-@lsh_app.command("query")
-def query_cmd(
-    keyword: str = typer.Argument(..., help="Termine da cercare, es. 'ablazione'."),
-    config: Path = CONFIG_OPT,
-    top: int = typer.Option(10, "--top", help="Numero massimo di risultati."),
-) -> None:
-    """Probe visuale: mostra i candidati LSH per un termine, con score Jaccard."""
-    from rich.console import Console
-    from rich.table import Table
-
-    from tht.lshindex import LshIndexError, load_index, query_index
-
-    cfg = _load_config_or_exit(config)
-    try:
-        lsh, minhashes, meta = load_index(_lsh_dir(cfg), name=cfg.database.db_schema)
-    except LshIndexError as e:
-        typer.secho(f"ERRORE: {e}", fg=typer.colors.RED)
-        raise typer.Exit(code=1)
-
-    hits = query_index(lsh, minhashes, keyword, meta, top_n=top)
-    if not hits:
-        typer.secho(f"Nessun candidato LSH per '{keyword}'.", fg=typer.colors.YELLOW)
-        return
-    table = Table(title=f"Candidati LSH per '{keyword}' ({len(hits)})")
-    table.add_column("Tabella")
-    table.add_column("Colonna")
-    table.add_column("Valore")
-    table.add_column("Score", justify="right")
-    for h in hits:
-        table.add_row(h.table, h.column, h.value, f"{h.score:.3f}")
-    Console().print(table)

harness/tht/cli/memory_cmd.py

@@ -26,7 +26,7 @@ DECISION_OPT = typer.Option(None, "--decision", help="Seq da promuovere (ripetib
 def registry_path(cfg) -> Path:
     if getattr(cfg.paths, "memory", None) is not None:
         return cfg.paths.memory / "registry.jsonl"
-    # Legacy location; migrate with `tht memory migrate` (P3).
+    # Legacy location remains readable while old workspaces are retired.
     return cfg.paths.artifacts / "memory" / "registry.jsonl"
 
 
@@ -45,12 +45,6 @@ def _resync_memory(cfg):
     )
 
 
-def clear_memory_index(cfg):
-    from tht.adapters.factory import build_vector_store
-
-    return build_vector_store(cfg, require_write=True).delete_kinds("memory", ["memory"])
-
-
 @memory_app.command("promote")
 def promote_cmd(
     session: str = typer.Option(..., "--session"),
@@ -130,7 +124,7 @@ def promote_cmd(
             f"{len(promoted)} memorie promosse nel registro, ma l'indice vettoriale "
             "NON e' stato sincronizzato (runtime vettoriale mancante o irraggiungibile): "
             "NON saranno trovate da `tht memory search` finche' non reindicizzi sul "
-            "server (`tht vector init`, poi `tht memory index`)."
+            "server (`tht memory index` quando il runtime vettoriale è disponibile)."
         )
 
     if json_out:
@@ -194,35 +188,6 @@ def save_one_cmd(
     typer.secho(f"OK: {msg}", fg=typer.colors.GREEN if count else typer.colors.YELLOW)
 
 
-@memory_app.command("clear")
-def clear_cmd(
-    yes: bool = typer.Option(False, "--yes", "-y", help="Salta la richiesta di conferma."),
-    config: Path = CONFIG_OPT,
-) -> None:
-    """Cancella TUTTA la review memory: registro canonico + indice semantico (kind=memory)."""
-    from tht.memory import load_registry
-
-    cfg = _load_config_or_exit(config)
-    require_server_profile(cfg, "memory clear")
-    registry = registry_path(cfg)
-    count = len(load_registry(registry))
-    if count == 0:
-        typer.secho("Nessuna memoria da cancellare.", fg=typer.colors.YELLOW)
-        return
-
-    if not yes and not typer.confirm(
-        f"Cancellare definitivamente {count} memorie (registro + indice)?"
-    ):
-        typer.secho("Annullato.", fg=typer.colors.YELLOW)
-        raise typer.Exit(code=1)
-
-    clear_memory_index(cfg)
-
-    # Registro canonico.
-    registry.unlink()
-    typer.secho(f"OK: {count} memorie cancellate.", fg=typer.colors.GREEN)
-
-
 @memory_app.command("index")
 def index_cmd(config: Path = CONFIG_OPT) -> None:
     """Sincronizza il registro memory nell'indice semantico (full-resync)."""
@@ -553,67 +518,3 @@ def solved_search_cmd(
         table.add_row(r["session_id"], r["question"][:60],
                       ", ".join(r["tables"]), f"{r['score']:.3f}")
     Console().print(table)
-
-@memory_app.command("migrate")
-def memory_migrate_cmd(
-    config: Path = CONFIG_OPT,
-    json_output: bool = typer.Option(False, "--json"),
-) -> None:
-    """Migrate the legacy artifacts/memory registry to the explicit workspace memory root (P3).
-
-    Copies and verifies exactly one legacy canonical JSONL under the workspace lock, then rebuilds
-    the Qdrant projection. Conflicting legacy registries fail closed; no in-place reinterpretation.
-    """
-    from tht.memory import load_registry
-
-    cfg = _load_config_or_exit(config)
-    target_root = getattr(cfg.paths, "memory", None)
-    if target_root is None:
-        payload = {"status": "failed", "error": "explicit memory root is not configured"}
-        if json_output:
-            typer.echo(json.dumps(payload, sort_keys=True))
-        else:
-            typer.secho("ERRORE: memory root esplicito non configurato", fg=typer.colors.RED, err=True)
-        raise typer.Exit(code=1)
-    legacy = cfg.paths.artifacts / "memory" / "registry.jsonl"
-    target = registry_path(cfg)
-    if target.exists():
-        payload = {"status": "unchanged", "path": str(target)}
-        if json_output:
-            typer.echo(json.dumps(payload, sort_keys=True))
-        else:
-            typer.secho(f"OK: memory registry già in {target}", fg=typer.colors.GREEN)
-        return
-    if not legacy.exists():
-        payload = {"status": "failed", "error": "legacy memory registry is missing"}
-        if json_output:
-            typer.echo(json.dumps(payload, sort_keys=True))
-        else:
-            typer.secho("ERRORE: registry legacy mancante", fg=typer.colors.RED, err=True)
-        raise typer.Exit(code=1)
-    try:
-        records = load_registry(legacy)
-    except Exception:  # noqa: BLE001
-        payload = {"status": "failed", "error": "legacy memory registry is invalid"}
-        if json_output:
-            typer.echo(json.dumps(payload, sort_keys=True))
-        else:
-            typer.secho("ERRORE: registry legacy non valido", fg=typer.colors.RED, err=True)
-        raise typer.Exit(code=1)
-    target_root.mkdir(parents=True, exist_ok=True)
-    from tht.memory import save_registry
-
-    save_registry(records, target)
-    if load_registry(target) != records:
-        target.unlink(missing_ok=True)
-        payload = {"status": "failed", "error": "memory registry migration verification failed"}
-        if json_output:
-            typer.echo(json.dumps(payload, sort_keys=True))
-        else:
-            typer.secho("ERRORE: verifica migrazione fallita", fg=typer.colors.RED, err=True)
-        raise typer.Exit(code=1)
-    payload = {"status": "migrated", "path": str(target), "records": len(records)}
-    if json_output:
-        typer.echo(json.dumps(payload, sort_keys=True))
-    else:
-        typer.secho(f"OK: migrate {len(records)} record verso {target}", fg=typer.colors.GREEN)

harness/tht/cli/search_cmd.py

@@ -115,7 +115,7 @@ def search_cmd(
         if not json_out:  # in JSON mode lo stdout resta puro: niente warning umano
             typer.secho(
                 "ATTENZIONE: indice LSH assente, ricerca solo vettoriale "
-                "(esegui `tht lsh build`).", fg=typer.colors.YELLOW,
+                "(esegui `tht preprocess dwh --steps lsh`).", fg=typer.colors.YELLOW,
             )
 
     if kind == "schema":

harness/tht/cli/sql_cmd.py

@@ -148,32 +148,6 @@ def validate_cmd(
     typer.secho("OK: SQL valido (statico).", fg=typer.colors.GREEN)
 
 
-@sql_app.command("explain")
-def explain_cmd(
-    file: Path = typer.Argument(...),
-    session: str = typer.Option(None, "--session"),
-    config: Path = CONFIG_OPT,
-) -> None:
-    """EXPLAIN (FORMAT JSON) con sintesi e warning dal piano. Mai ANALYZE."""
-    from tht.execute import ExecutionError
-    from tht.execute.warnings import plan_warnings
-
-    cfg = _load_config_or_exit(config)
-    require_action(cfg, "explain")
-    sql = _read_sql(file)
-    validate_or_exit(cfg, sql, session)
-    try:
-        plan = do_explain(cfg, sql)
-    except ExecutionError as e:
-        typer.secho(f"ERRORE: {e}", fg=typer.colors.RED, err=True)
-        raise typer.Exit(code=1)
-    typer.echo(f"costo totale stimato: {plan.total_cost}")
-    typer.echo(f"righe stimate: {plan.plan_rows}")
-    typer.echo(f"nodi del piano: {', '.join(plan.node_types)}")
-    for w in plan_warnings(plan, cfg.execution):
-        typer.secho(f"  warning: {w}", fg=typer.colors.YELLOW)
-
-
 @sql_app.command("preview")
 def preview_cmd(
     file: Path = typer.Argument(None, help="File SQL da eseguire. Opzionale se --session è dato."),
@@ -278,19 +252,6 @@ def set_final_cmd(
     typer.secho("OK: SQL finale salvato.", fg=typer.colors.GREEN)
 
 
-@sql_app.command("save")
-def save_cmd(
-    dest: Path = typer.Argument(..., help="Percorso di destinazione del file SQL."),
-    session: str = typer.Option(..., "--session"),
-    config: Path = CONFIG_OPT,
-) -> None:
-    """Salva una copia di sql_final.sql nel percorso indicato (su richiesta esplicita)."""
-    cfg = _load_config_or_exit(config)
-    dest.parent.mkdir(parents=True, exist_ok=True)
-    dest.write_text(_session_sql(cfg, session))
-    typer.secho(f"OK: SQL salvato in {dest}", fg=typer.colors.GREEN)
-
-
 @sql_app.command("export")
 def export_cmd(
     dest: Path = typer.Argument(..., help="Percorso del CSV di destinazione."),

harness/tht/cli/vector_cmd.py

@@ -4,7 +4,7 @@ from pathlib import Path
 
 import typer
 
-from tht.cli._guards import require_server_profile, require_vector_write_allowed
+from tht.cli._guards import require_vector_write_allowed
 from tht.cli.config_cmd import CONFIG_OPT
 from tht.cli.schema_cmd import _load_config_or_exit, annotations_path, physical_path
 from tht.ports.vector import VectorWriteRecord
@@ -97,40 +97,6 @@ def _print_stats(stats) -> None:
     )
 
 
-@vector_app.command("init")
-def init_cmd(
-    config: Path = CONFIG_OPT,
-    skip_ollama_check: bool = typer.Option(
-        False, "--skip-ollama-check", help="Non verificare la raggiungibilita' di Ollama."
-    ),
-) -> None:
-    """Verifica il runtime Qdrant e la raggiungibilita' dell'embedder configurato."""
-    from tht.adapters.factory import build_vector_store
-    from tht.vectorstore.embeddings import EmbeddingsError
-
-    cfg = _load_config_or_exit(config)
-    require_server_profile(cfg, "vector init")
-    require_vector_cfg(cfg)
-    health = build_vector_store(cfg, require_write=True).health()
-    if not health.ok:
-        typer.secho(
-            f"ERRORE runtime vettoriale: {health.detail or 'Qdrant non raggiungibile o incompatibile'}",
-            fg=typer.colors.RED,
-            err=True,
-        )
-        raise typer.Exit(code=1)
-    if not skip_ollama_check:
-        try:
-            make_embedder(cfg.embeddings).embed_query("ping")
-        except EmbeddingsError as e:
-            typer.secho(f"ERRORE: {e}", fg=typer.colors.RED, err=True)
-            raise typer.Exit(code=1)
-    typer.secho(
-        f"OK: runtime Qdrant pronto per la collezione {cfg.vectors.collection}",
-        fg=typer.colors.GREEN,
-    )
-
-
 @vector_app.command("index-schema")
 def index_schema_cmd(
     config: Path = CONFIG_OPT,

harness/tht/lshindex/__init__.py

@@ -58,7 +58,7 @@ def load_index(directory: Path, name: str) -> tuple[Any, Any, dict]:
     if not (lsh_path.exists() and mh_path.exists() and meta_path.exists()):
         raise LshIndexError(
             f"Indice LSH non trovato in {directory} (atteso {name}_lsh.pkl). "
-            f"Esegui prima `tht lsh build`."
+            f"Esegui prima `tht preprocess dwh --steps lsh`."
         )
     lsh = pickle.loads(lsh_path.read_bytes())
     minhashes = pickle.loads(mh_path.read_bytes())
@@ -79,7 +79,7 @@ def query_index(lsh, minhashes, keyword: str, meta: dict, top_n: int = 10) -> li
         if entry is None:
             raise LshIndexError(
                 "Indice LSH disallineato (lsh.pkl e minhashes.pkl non coerenti): "
-                "rigenera con `tht lsh build`."
+                "rigenera con `tht preprocess dwh --steps lsh`."
             )
         scored.append((key, qmh.jaccard(entry[0])))
     scored.sort(key=lambda kv: kv[1], reverse=True)