feat(harness): workflow.yaml as single source of truth + workflow.py loader (F2)

- workflow.yaml: 8-phase definition, data-driven prerequisites, no hardcoded ladder
- workflow.py: load_workflow() reader; max_phase=len(phases), decision_min_phase
  derived from prerequisites scan (no duplication)
- 6 tests: phase count, decision_min_phase derivation, name lookup (incl. F8
  presence -- the JS drift bug structurally impossible now), artifacts_out,
  advance strategy, schema_version. All 8 harness tests pass.

harness/nsp/workflow.py

@@ -0,0 +1,107 @@
+"""Reads workflow.yaml -- the SINGLE source of workflow truth (spec F2, §5.3).
+
+phase.py, the gate (nsp-gate.js), and the skill all read from here.
+No more duplicated constants (the JS/Python drift bug in ChironeWp3 -- PHASE_NAMES
+truncated to 7 in JS -- is structurally impossible because there is one source).
+
+Edit workflow.yaml to change the workflow: add/reorder/merge/skip phases.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+_WF_PATH = Path(__file__).resolve().parent.parent / "workflow.yaml"
+
+
+@dataclass
+class PhaseSpec:
+    id: str
+    num: int
+    name: str
+    advance: str
+    prerequisites: list[Any]
+    artifacts_out: list[str] = field(default_factory=list)
+
+
+@dataclass
+class Workflow:
+    schema_version: int
+    phases: list[PhaseSpec]
+    _decision_min_map: dict[str, int] = field(default_factory=dict)
+
+    @property
+    def max_phase(self) -> int:
+        return len(self.phases)
+
+    def phase_by_num(self, n: int) -> PhaseSpec:
+        return self.phases[n - 1]
+
+    def phase_name(self, n: int) -> str:
+        if 1 <= n <= self.max_phase:
+            return self.phase_by_num(n).name
+        return "?"
+
+    def decision_min_phase(self, decision_type: str) -> int:
+        """A decision type's min phase = the earliest phase whose prerequisites
+        reference it (via decision_exists / decision_subject_exists). Defaults to 1."""
+        return self._decision_min_map.get(decision_type, 1)
+
+
+def _collect_decision_mins(phases: list[PhaseSpec]) -> dict[str, int]:
+    """Scan prerequisites for decision_exists / decision_subject_exists mentions.
+
+    Supports both forms:
+      - decision_exists: <type>            (scalar)
+      - decision_exists: [<type>, ...]     (list, first element is the type)
+      - decision_subject_exists: [<type>, <subject>]  (list, first element is the type)
+    """
+    mins: dict[str, int] = {}
+
+    def scan(node: Any, phase_num: int) -> None:
+        if isinstance(node, dict):
+            for key, value in node.items():
+                if key in ("decision_exists", "decision_subject_exists"):
+                    if isinstance(value, list) and value:
+                        dtype = value[0]
+                    elif isinstance(value, str):
+                        dtype = value
+                    else:
+                        continue
+                    if isinstance(dtype, str):
+                        if dtype not in mins or phase_num < mins[dtype]:
+                            mins[dtype] = phase_num
+                else:
+                    scan(value, phase_num)
+        elif isinstance(node, list):
+            for item in node:
+                scan(item, phase_num)
+
+    for p in phases:
+        scan(p.prerequisites, p.num)
+    return mins
+
+
+def load_workflow(path: Path | str = _WF_PATH) -> Workflow:
+    path = Path(path)
+    raw = yaml.safe_load(path.read_text())
+    phases: list[PhaseSpec] = []
+    for i, p in enumerate(raw["phases"], start=1):
+        phases.append(
+            PhaseSpec(
+                id=p["id"],
+                num=i,
+                name=p["name"],
+                advance=p["advance"],
+                prerequisites=p.get("prerequisites", []),
+                artifacts_out=p.get("artifacts_out", []),
+            )
+        )
+    return Workflow(
+        schema_version=raw.get("schema_version", 1),
+        phases=phases,
+        _decision_min_map=_collect_decision_mins(phases),
+    )

harness/tests/test_workflow.py

@@ -0,0 +1,55 @@
+from nsp.workflow import load_workflow
+
+
+def test_workflow_loads_8_phases():
+    wf = load_workflow()
+    assert len(wf.phases) == 8
+    assert wf.phases[0].id == "F1"
+    assert wf.max_phase == 8
+    assert wf.phase_by_num(1).name == "chiarimento"
+
+
+def test_decision_min_phase_derived():
+    wf = load_workflow()
+    # question_rewritten is a prerequisite of F3 -> min phase 3
+    assert wf.decision_min_phase("question_rewritten") == 3
+    # sql_approved is a prerequisite of F7 -> min phase 7
+    assert wf.decision_min_phase("sql_approved") == 7
+    # datamart_requested / datamart_declined are prerequisites of F8 -> min phase 8
+    assert wf.decision_min_phase("datamart_requested") == 8
+    assert wf.decision_min_phase("datamart_declined") == 8
+    # phase_skipped referenced via decision_subject_exists in F6 -> min phase 6
+    assert wf.decision_min_phase("phase_skipped") == 6
+    # unknown type -> phase 1 (default)
+    assert wf.decision_min_phase("nonexistent_type") == 1
+
+
+def test_phase_name_lookup():
+    wf = load_workflow()
+    assert wf.phase_name(1) == "chiarimento"
+    assert wf.phase_name(6) == "cte"
+    assert wf.phase_name(8) == "datamart"  # the JS drift bug in ChironeWp3 -- F8 MUST be present
+    assert wf.phase_name(0) == "?"  # out of range lower
+    assert wf.phase_name(9) == "?"  # out of range upper
+
+
+def test_artifacts_out_per_phase():
+    wf = load_workflow()
+    assert "schema_linking.json" in wf.phase_by_num(4).artifacts_out
+    assert "sql_final.sql" in wf.phase_by_num(7).artifacts_out
+    assert "cte_plan.json" in wf.phase_by_num(6).artifacts_out
+    assert "ctes/" in wf.phase_by_num(6).artifacts_out  # dir artifact
+    assert wf.phase_by_num(1).artifacts_out == []  # F1 produces none
+
+
+def test_advance_strategy_per_phase():
+    wf = load_workflow()
+    assert wf.phase_by_num(2).advance == "auto_if_empty"
+    assert wf.phase_by_num(6).advance == "auto_if_empty_or_skipped"
+    assert wf.phase_by_num(4).advance == "reviewer_decide"
+    assert wf.phase_by_num(5).advance == "kind:phase"
+
+
+def test_schema_version():
+    wf = load_workflow()
+    assert wf.schema_version == 1

harness/workflow.yaml

@@ -0,0 +1,59 @@
+# harness/workflow.yaml — single source of workflow truth (spec F2, §5.3)
+# phase.py, il gate e la skill leggono tutti da qui. Nessuna costante duplicata.
+# Editare questo file per cambiare il workflow (aggiungere/riordinare/fondere/skip fasi).
+
+schema_version: 1
+
+phases:
+  - id: F1
+    name: chiarimento
+    advance: kind:phase
+    prerequisites: []
+    artifacts_out: []
+  - id: F2
+    name: memoria
+    advance: auto_if_empty
+    prerequisites: []
+    artifacts_out: []
+  - id: F3
+    name: riscrittura
+    advance: kind:phase
+    prerequisites:
+      - decision_exists: question_rewritten
+    artifacts_out: [question.md]
+  - id: F4
+    name: schema_linking
+    advance: reviewer_decide
+    prerequisites: []
+    artifacts_out: [schema_linking.json]
+  - id: F5
+    name: sintesi
+    advance: kind:phase
+    prerequisites:
+      - file_validates: [schema_linking.json, SchemaLinking]
+    artifacts_out: []
+  - id: F6
+    name: cte
+    advance: auto_if_empty_or_skipped
+    prerequisites:
+      - any:
+          - decision_subject_exists: [phase_skipped, "phase:6"]
+          - all_ctes_approved: true
+    artifacts_out: [cte_plan.json, ctes/, cte_tests.json]
+  - id: F7
+    name: sql_finale
+    advance: kind:phase
+    prerequisites:
+      - decision_exists: sql_approved
+    artifacts_out: [sql_final.sql]
+  - id: F8
+    name: datamart
+    advance: reviewer_decide
+    prerequisites:
+      - any:
+          - decision_exists: datamart_requested
+          - decision_exists: datamart_declined
+    artifacts_out: []
+
+decision_min_phase: auto
+max_phase: auto