feat(harness): teardown_to_phase -- artifact teardown on rollback (D15)

Cancella ogni artefatto la cui fase produttrice > target, usando artifacts_out di
workflow.yaml. Risolve il bug latente di ChironeWp3: ctes/*.sql orfani (non piu'
nel piano dopo un re-derive) restavano su disco e bloccavano finalize.
Da chiamare insieme all'append di phase_reopened per mantenere stato coerente.
6 tests (target 4/1/7, empty, missing, orfani CTE). 26 total passing.

harness/nsp/teardown.py

@@ -0,0 +1,54 @@
+"""Artifact teardown on rollback (spec D15, §4.8).
+
+teardown_to_phase cancella ogni artefatto la cui fase produttrice > target, usando la
+mappa artifacts_out di workflow.yaml. Risolve il bug latente di ChironeWp3: dopo un
+re-derive con piano CTE diverso, i vecchi ctes/*.sql orfani restavano su disco e
+bloccavano finalize (che itera glob('*.sql') esigendo che ognuno sia testato).
+
+Da chiamare insieme all'append di phase_reopened per mantenere lo stato coerente
+(l'artefatto su disco e il ledger effective devono allinearsi -- vedi phase.py).
+"""
+from __future__ import annotations
+
+import shutil
+from dataclasses import dataclass, field
+from pathlib import Path
+
+from nsp.workflow import load_workflow
+
+
+@dataclass
+class TeardownReport:
+    target_phase: int
+    deleted_files: list[str] = field(default_factory=list)
+
+
+def teardown_to_phase(session_dir: str | Path, target_phase: int) -> TeardownReport:
+    """Cancella gli artefatti delle fasi > target_phase. Ritorna il report dei cancellati.
+
+    - File artefatto (es. 'schema_linking.json'): unlink se esiste.
+    - Directory artefatto (es. 'ctes/'): rimuove ricorsivamente (con tutti i .sql orfani).
+    - Artefatti di fase <= target: preservati (sono lavoro valido).
+    - Artefatti mancanti: noop (sessione nuova).
+    """
+    session_dir = Path(session_dir)
+    wf = load_workflow()
+    report = TeardownReport(target_phase=target_phase)
+    for phase in wf.phases:
+        if phase.num <= target_phase:
+            continue
+        for artifact in phase.artifacts_out:
+            target = session_dir / artifact.rstrip("/")
+            is_dir = artifact.endswith("/")
+            if is_dir:
+                if target.exists() and target.is_dir():
+                    # registra ogni file prima di rimuovere (utile per audit/debug orfani)
+                    for f in sorted(target.glob("*")):
+                        if f.is_file():
+                            report.deleted_files.append(f.name)
+                    shutil.rmtree(target)
+            else:
+                if target.exists():
+                    target.unlink()
+                    report.deleted_files.append(artifact)
+    return report

harness/tests/test_teardown.py

@@ -0,0 +1,78 @@
+from nsp.teardown import teardown_to_phase
+
+
+def test_teardown_to_phase_4_deletes_phase5plus_artifacts(tmp_path):
+    s = tmp_path / "sess"; s.mkdir()
+    (s / "schema_linking.json").write_text("{}")    # F4 artifact
+    (s / "cte_plan.json").write_text("[]")          # F6 artifact
+    (s / "ctes").mkdir()
+    (s / "ctes" / "x.sql").write_text("SELECT 1")   # F6 artifact
+    (s / "sql_final.sql").write_text("SELECT 1")    # F7 artifact
+    report = teardown_to_phase(s, target_phase=4)
+    assert (s / "schema_linking.json").exists()     # F4 preserved (target is 4)
+    assert not (s / "cte_plan.json").exists()       # F6 deleted (>4)
+    assert not (s / "ctes").exists()                # F6 dir deleted
+    assert not (s / "sql_final.sql").exists()       # F7 deleted
+    assert "cte_plan.json" in report.deleted_files
+    assert "sql_final.sql" in report.deleted_files
+
+
+def test_teardown_to_phase_1_deletes_everything(tmp_path):
+    """Rollback a F1 cancella tutti gli artefatti (tutte le fasi > 1)."""
+    s = tmp_path / "sess"; s.mkdir()
+    (s / "question.md").write_text("q")             # F3 artifact
+    (s / "schema_linking.json").write_text("{}")    # F4 artifact
+    (s / "cte_plan.json").write_text("[]")          # F6 artifact
+    (s / "ctes").mkdir()
+    (s / "ctes" / "a.sql").write_text("SELECT 1")
+    (s / "sql_final.sql").write_text("SELECT 1")    # F7 artifact
+    report = teardown_to_phase(s, target_phase=1)
+    assert not (s / "question.md").exists()
+    assert not (s / "schema_linking.json").exists()
+    assert not (s / "cte_plan.json").exists()
+    assert not (s / "ctes").exists()
+    assert not (s / "sql_final.sql").exists()
+    assert len(report.deleted_files) >= 4
+
+
+def test_teardown_to_phase_7_preserves_phase6_artifacts(tmp_path):
+    """Rollback a F7 preserva gli artefatti delle fasi <= 7."""
+    s = tmp_path / "sess"; s.mkdir()
+    (s / "schema_linking.json").write_text("{}")    # F4 artifact
+    (s / "cte_plan.json").write_text("[]")          # F6 artifact
+    (s / "ctes").mkdir()
+    (s / "ctes" / "a.sql").write_text("SELECT 1")   # F6 artifact
+    (s / "sql_final.sql").write_text("SELECT 1")    # F7 artifact
+    report = teardown_to_phase(s, target_phase=7)
+    assert (s / "schema_linking.json").exists()
+    assert (s / "cte_plan.json").exists()
+    assert (s / "ctes").exists()
+    assert (s / "sql_final.sql").exists()           # F7 preserved (target is 7)
+    assert report.deleted_files == []               # nothing > 7 has artifacts
+
+
+def test_teardown_to_phase_on_empty_session_is_noop(tmp_path):
+    s = tmp_path / "sess"; s.mkdir()
+    report = teardown_to_phase(s, target_phase=4)
+    assert report.deleted_files == []
+
+
+def test_teardown_to_phase_nonexistent_artifacts_no_error(tmp_path):
+    """Se gli artefatti non esistono (sessione nuova), teardown non fallisce."""
+    s = tmp_path / "sess"; s.mkdir()
+    report = teardown_to_phase(s, target_phase=1)
+    assert report.deleted_files == []
+
+
+def test_teardown_deletes_orphaned_cte_files_not_in_current_plan(tmp_path):
+    """Il bug degli orfani CTE: dopo un re-derive con piano diverso, i vecchi .sql
+    non piu' nel piano restano su disco e bloccano finalize. teardown li pulisce."""
+    s = tmp_path / "sess"; s.mkdir()
+    (s / "ctes").mkdir()
+    (s / "ctes" / "old_cte.sql").write_text("SELECT 1")   # orphan
+    (s / "ctes" / "another.sql").write_text("SELECT 2")   # orphan
+    report = teardown_to_phase(s, target_phase=5)  # F6 has ctes/ as artifact
+    assert not (s / "ctes").exists()
+    # i nomi dei file orfani sono registrati
+    assert "old_cte.sql" in report.deleted_files
+    assert "another.sql" in report.deleted_files