feat(harness): persist CTE preview rows, add tht cte info, --doc for cte plan

WS1 of review-gates-v2: gives the JS gate (WS2) deterministic data to build the
cte_result v2 payload.

- CteTestRecord gains optional preview_rows (JSON-coerced, truncated cells);
  test_cmd populates it from the bounded result rows.
- New read-only `tht cte info <name> --session <id> [--json]`: plan
  index/total, persisted .sql, cte_plan_doc.json entry (if any), last
  CteTestRecord. Exits 1 with a clean stderr message on missing
  session/plan/name/sql.
- `tht cte plan --doc -` validates a chain-doc JSON (ctes[].name must match
  --name, same order) and writes it to cte_plan_doc.json; cte_plan.json stays
  a plain list[str] (load-bearing for tht.phase.next_cte). --doc is optional.

harness/tests/test_cte_info.py

@@ -0,0 +1,149 @@
+"""L1: `tht cte info <name>` — read-only CTE detail (plan position + sql + doc + last test).
+
+Feeds the JS gate (WS2) building the cte_result v2 payload: it needs index/total from
+cte_plan.json (same source as tht.phase.next_cte), the persisted .sql file, the optional
+cte_plan_doc.json entry, and the last CteTestRecord for the CTE. --json output must be
+pristine (only valid JSON on stdout).
+"""
+import json
+from datetime import datetime
+
+from typer.testing import CliRunner
+
+from tht.cli.cte_cmd import cte_app
+from tht.config import DatabaseConfig
+from tht.ctetest import CteTestRecord, append_cte_test
+from tht.decisions import append_decision
+from tht.session.store import create_session
+
+
+def _db():
+    return DatabaseConfig(database="testdb", user="u", password="p", **{"schema": "public"})  # noqa: S106
+
+
+def _patch_cfg(monkeypatch, tmp_path):
+    from tht.cli import cte_cmd
+
+    class FakePaths:
+        sessions = tmp_path
+
+    class FakeCfg:
+        paths = FakePaths()
+        database = _db()
+
+    monkeypatch.setattr(cte_cmd, "_load_config_or_exit", lambda _: FakeCfg())
+
+
+def _make_session(tmp_path, plan):
+    m = create_session("q", _db(), tmp_path)
+    sdir = tmp_path / m.id
+    (sdir / "cte_plan.json").write_text(json.dumps(plan))
+    (sdir / "ctes").mkdir(parents=True, exist_ok=True)
+    return m.id, sdir
+
+
+def test_info_happy_path_json(tmp_path, monkeypatch):
+    sid, sdir = _make_session(tmp_path, ["a", "b"])
+    (sdir / "ctes" / "b.sql").write_text("WITH b AS (SELECT 1)")
+    append_cte_test(sdir, CteTestRecord(
+        name="b", ts=datetime(2025, 1, 1, 12, 0), sql_hash="h1", status="ok",
+        columns=["x"], row_sample=1, execution_ms=5, preview_rows=[[1]],
+    ))
+    _patch_cfg(monkeypatch, tmp_path)
+
+    res = CliRunner().invoke(cte_app, ["info", "b", "--session", sid, "--json"])
+    assert res.exit_code == 0, res.output
+    data = json.loads(res.output)
+    assert data["name"] == "b"
+    assert data["index"] == 2
+    assert data["total"] == 2
+    assert data["plan"] == ["a", "b"]
+    assert data["sql"] == "WITH b AS (SELECT 1)"
+    assert data["approved"] is False
+    assert data["doc"] is None
+    assert data["last_test"]["sql_hash"] == "h1"
+    assert data["last_test"]["preview_rows"] == [[1]]
+
+
+def test_info_approved_true_when_decision_present(tmp_path, monkeypatch):
+    sid, sdir = _make_session(tmp_path, ["a"])
+    (sdir / "ctes" / "a.sql").write_text("WITH a AS (SELECT 1)")
+    append_decision(sdir, type="cte_approved", subject="a")
+    _patch_cfg(monkeypatch, tmp_path)
+
+    res = CliRunner().invoke(cte_app, ["info", "a", "--session", sid, "--json"])
+    assert res.exit_code == 0, res.output
+    data = json.loads(res.output)
+    assert data["approved"] is True
+
+
+def test_info_doc_present(tmp_path, monkeypatch):
+    sid, sdir = _make_session(tmp_path, ["a"])
+    (sdir / "ctes" / "a.sql").write_text("WITH a AS (SELECT 1)")
+    doc = {"ctes": [{"name": "a", "purpose": "p", "rationale": "r", "depends_on": []}]}
+    (sdir / "cte_plan_doc.json").write_text(json.dumps(doc, ensure_ascii=False))
+    _patch_cfg(monkeypatch, tmp_path)
+
+    res = CliRunner().invoke(cte_app, ["info", "a", "--session", sid, "--json"])
+    assert res.exit_code == 0, res.output
+    data = json.loads(res.output)
+    assert data["doc"] == {"purpose": "p", "rationale": "r", "depends_on": []}
+
+
+def test_info_last_test_none_when_no_records(tmp_path, monkeypatch):
+    sid, sdir = _make_session(tmp_path, ["a"])
+    (sdir / "ctes" / "a.sql").write_text("WITH a AS (SELECT 1)")
+    _patch_cfg(monkeypatch, tmp_path)
+
+    res = CliRunner().invoke(cte_app, ["info", "a", "--session", sid, "--json"])
+    assert res.exit_code == 0, res.output
+    data = json.loads(res.output)
+    assert data["last_test"] is None
+
+
+def test_info_last_test_is_most_recent_record(tmp_path, monkeypatch):
+    sid, sdir = _make_session(tmp_path, ["a"])
+    (sdir / "ctes" / "a.sql").write_text("WITH a AS (SELECT 1)")
+    append_cte_test(sdir, CteTestRecord(
+        name="a", ts=datetime(2025, 1, 1), sql_hash="old", status="ok",
+    ))
+    append_cte_test(sdir, CteTestRecord(
+        name="a", ts=datetime(2025, 1, 2), sql_hash="new", status="ok",
+    ))
+    _patch_cfg(monkeypatch, tmp_path)
+
+    res = CliRunner().invoke(cte_app, ["info", "a", "--session", sid, "--json"])
+    assert res.exit_code == 0, res.output
+    data = json.loads(res.output)
+    assert data["last_test"]["sql_hash"] == "new"
+
+
+def test_info_nonexistent_session_exit_1(tmp_path, monkeypatch):
+    _patch_cfg(monkeypatch, tmp_path)
+    res = CliRunner().invoke(cte_app, ["info", "a", "--session", "nope", "--json"])
+    assert res.exit_code == 1
+    assert res.stdout == ""  # stdout stays pristine JSON-or-nothing
+
+
+def test_info_missing_plan_exit_1(tmp_path, monkeypatch):
+    m = create_session("q", _db(), tmp_path)
+    _patch_cfg(monkeypatch, tmp_path)
+    res = CliRunner().invoke(cte_app, ["info", "a", "--session", m.id, "--json"])
+    assert res.exit_code == 1
+    assert res.stdout == ""
+
+
+def test_info_name_not_in_plan_exit_1(tmp_path, monkeypatch):
+    sid, sdir = _make_session(tmp_path, ["a"])
+    _patch_cfg(monkeypatch, tmp_path)
+    res = CliRunner().invoke(cte_app, ["info", "zzz", "--session", sid, "--json"])
+    assert res.exit_code == 1
+    assert res.stdout == ""
+
+
+def test_info_missing_sql_file_exit_1(tmp_path, monkeypatch):
+    sid, sdir = _make_session(tmp_path, ["a"])
+    _patch_cfg(monkeypatch, tmp_path)
+    res = CliRunner().invoke(cte_app, ["info", "a", "--session", sid, "--json"])
+    assert res.exit_code == 1
+    assert res.stdout == ""

harness/tests/test_cte_plan_doc.py

@@ -0,0 +1,119 @@
+"""L1: `tht cte plan --doc -` — persists the cte_plan_doc.json chain documentation.
+
+cte_plan.json (list[str], load-bearing for tht.phase.next_cte) stays untouched; the
+doc (payload-A shape: {"ctes": [{"name", "purpose", "rationale", "depends_on", ...}]})
+is validated (names must match --name, same order) and written verbatim to a
+separate sessions/<id>/cte_plan_doc.json. --doc is optional; without it, behavior
+is unchanged (existing test_cte_next.py covers the no-doc path).
+"""
+import json
+
+from typer.testing import CliRunner
+
+from tht.cli.cte_cmd import cte_app
+from tht.config import DatabaseConfig
+from tht.decisions import append_decision
+from tht.session.store import create_session
+
+
+def _db():
+    return DatabaseConfig(database="testdb", user="u", password="p", **{"schema": "public"})  # noqa: S106
+
+
+def _patch_cfg(monkeypatch, tmp_path):
+    from tht.cli import cte_cmd
+
+    class FakePaths:
+        sessions = tmp_path
+
+    class FakeCfg:
+        paths = FakePaths()
+        database = _db()
+
+    monkeypatch.setattr(cte_cmd, "_load_config_or_exit", lambda _: FakeCfg())
+
+
+def _advance_to_phase_6(sdir):
+    """plan_cmd requires phase >= 6: approve phases 1..5 in order."""
+    for n in range(1, 6):
+        append_decision(sdir, type="phase_approved", subject=f"phase:{n}")
+
+
+def test_plan_with_matching_doc_writes_cte_plan_doc_json(tmp_path, monkeypatch):
+    m = create_session("q", _db(), tmp_path)
+    _advance_to_phase_6(tmp_path / m.id)
+    _patch_cfg(monkeypatch, tmp_path)
+    doc = {
+        "question": "domanda riscritta",
+        "strategy": "razionale",
+        "ctes": [
+            {"name": "a", "purpose": "p_a", "rationale": "r_a", "depends_on": []},
+            {"name": "b", "purpose": "p_b", "rationale": "r_b", "depends_on": ["a"]},
+        ],
+    }
+    res = CliRunner().invoke(
+        cte_app,
+        ["plan", "--session", m.id, "--name", "a", "--name", "b", "--doc", "-"],
+        input=json.dumps(doc),
+    )
+    assert res.exit_code == 0, res.output
+
+    plan_path = tmp_path / m.id / "cte_plan.json"
+    assert json.loads(plan_path.read_text()) == ["a", "b"]
+
+    doc_path = tmp_path / m.id / "cte_plan_doc.json"
+    assert json.loads(doc_path.read_text()) == doc
+
+
+def test_plan_without_doc_unchanged(tmp_path, monkeypatch):
+    m = create_session("q", _db(), tmp_path)
+    _advance_to_phase_6(tmp_path / m.id)
+    _patch_cfg(monkeypatch, tmp_path)
+    res = CliRunner().invoke(cte_app, ["plan", "--session", m.id, "--name", "a"])
+    assert res.exit_code == 0, res.output
+    assert json.loads((tmp_path / m.id / "cte_plan.json").read_text()) == ["a"]
+    assert not (tmp_path / m.id / "cte_plan_doc.json").exists()
+
+
+def test_plan_doc_name_mismatch_rejected(tmp_path, monkeypatch):
+    m = create_session("q", _db(), tmp_path)
+    _advance_to_phase_6(tmp_path / m.id)
+    _patch_cfg(monkeypatch, tmp_path)
+    doc = {"ctes": [{"name": "a", "purpose": "p"}, {"name": "zzz", "purpose": "p2"}]}
+    res = CliRunner().invoke(
+        cte_app,
+        ["plan", "--session", m.id, "--name", "a", "--name", "b", "--doc", "-"],
+        input=json.dumps(doc),
+    )
+    assert res.exit_code == 1
+    assert res.stdout == ""
+    # neither file should be written on a rejected doc
+    assert not (tmp_path / m.id / "cte_plan.json").exists()
+    assert not (tmp_path / m.id / "cte_plan_doc.json").exists()
+
+
+def test_plan_doc_wrong_order_rejected(tmp_path, monkeypatch):
+    m = create_session("q", _db(), tmp_path)
+    _advance_to_phase_6(tmp_path / m.id)
+    _patch_cfg(monkeypatch, tmp_path)
+    doc = {"ctes": [{"name": "b", "purpose": "p"}, {"name": "a", "purpose": "p2"}]}
+    res = CliRunner().invoke(
+        cte_app,
+        ["plan", "--session", m.id, "--name", "a", "--name", "b", "--doc", "-"],
+        input=json.dumps(doc),
+    )
+    assert res.exit_code == 1
+    assert not (tmp_path / m.id / "cte_plan_doc.json").exists()
+
+
+def test_plan_doc_invalid_json_rejected(tmp_path, monkeypatch):
+    m = create_session("q", _db(), tmp_path)
+    _advance_to_phase_6(tmp_path / m.id)
+    _patch_cfg(monkeypatch, tmp_path)
+    res = CliRunner().invoke(
+        cte_app,
+        ["plan", "--session", m.id, "--name", "a", "--doc", "-"],
+        input="{not valid json",
+    )
+    assert res.exit_code == 1
+    assert not (tmp_path / m.id / "cte_plan_doc.json").exists()

harness/tests/test_ctetest.py

@@ -7,7 +7,8 @@ and the ledger I/O (load/append, tolerant of JSON-array and JSONL formats).
 Pure logic, no DB.
 """
 import json
-from datetime import datetime
+from datetime import date, datetime
+from decimal import Decimal
 from pathlib import Path
 
 import pytest
@@ -15,6 +16,7 @@ import pytest
 from tht.ctetest import (
     CteError,
     CteTestRecord,
+    _jsonable,
     append_cte_test,
     build_test_sql,
     has_trailing_select,
@@ -131,3 +133,60 @@ def test_corrupt_ledger_raises(tmp_path):
     (tmp_path / "cte_tests.json").write_text("{not valid json")
     with pytest.raises(CteError, match="corrotto"):
         load_cte_tests(tmp_path)
+
+
+# --- _jsonable ----------------------------------------------------------
+
+def test_jsonable_passes_through_native_types():
+    assert _jsonable(None) is None
+    assert _jsonable(True) is True
+    assert _jsonable(1) == 1
+    assert _jsonable(1.5) == 1.5
+    assert _jsonable("abc") == "abc"
+
+
+def test_jsonable_coerces_decimal_and_date_to_str():
+    assert _jsonable(Decimal("12.34")) == "12.34"
+    assert _jsonable(date(2025, 1, 1)) == "2025-01-01"
+    assert _jsonable(datetime(2025, 1, 1, 12, 0, 0)) == "2025-01-01 12:00:00"
+
+
+def test_jsonable_truncates_long_strings():
+    long_str = "x" * 500
+    out = _jsonable(long_str)
+    assert isinstance(out, str)
+    assert len(out) <= 200
+
+
+def test_jsonable_truncates_long_coerced_values():
+    class Weird:
+        def __str__(self):
+            return "y" * 500
+
+    out = _jsonable(Weird())
+    assert len(out) <= 200
+
+
+# --- preview_rows on CteTestRecord ---------------------------------------
+
+def test_preview_rows_optional_default_none():
+    r = _record()
+    assert r.preview_rows is None
+
+
+def test_preview_rows_roundtrip_through_ledger(tmp_path):
+    rec = _record(name="a", preview_rows=[["x", 1], [None, 2.5]])
+    append_cte_test(tmp_path, rec)
+    loaded = load_cte_tests(tmp_path)
+    assert loaded[0].preview_rows == [["x", 1], [None, 2.5]]
+
+
+def test_old_record_without_preview_rows_still_validates(tmp_path):
+    # back-compat: an old cte_tests.json row has no "preview_rows" key at all
+    path = tmp_path / "cte_tests.json"
+    path.write_text(json.dumps({
+        "name": "x", "ts": "2025-01-01T00:00:00", "sql_hash": "h", "status": "ok",
+    }))
+    loaded = load_cte_tests(tmp_path)
+    assert len(loaded) == 1
+    assert loaded[0].preview_rows is None

harness/tht/cli/cte_cmd.py

@@ -29,7 +29,14 @@ def test_cmd(
     from rich.table import Table
 
     from tht.cli.sql_cmd import do_run
-    from tht.ctetest import CteError, CteTestRecord, append_cte_test, build_test_sql, has_trailing_select
+    from tht.ctetest import (
+        CteError,
+        CteTestRecord,
+        _jsonable,
+        append_cte_test,
+        build_test_sql,
+        has_trailing_select,
+    )
     from tht.execute import ExecutionError
     from tht.execute.warnings import runtime_warnings
 
@@ -109,6 +116,7 @@ def test_cmd(
         name=name, ts=datetime.now(UTC), sql_hash=sql_hash, status="ok",
         columns=result.columns, row_sample=len(result.rows),
         execution_ms=result.execution_ms, warnings=warnings,
+        preview_rows=[[_jsonable(cell) for cell in row] for row in result.rows],
     )
     append_cte_test(sdir, record)
 
@@ -128,14 +136,25 @@ def test_cmd(
     typer.secho(f"OK: esito registrato in {sdir / 'cte_tests.json'}", fg=typer.colors.GREEN)
 
 
+CTE_PLAN_DOC_FILE = "cte_plan_doc.json"
+
+
 @cte_app.command("plan")
 def plan_cmd(
     session: str = typer.Option(..., "--session"),
     name: list[str] = typer.Option(..., "--name", help="Nome CTE (ripetibile, in ordine)."),
+    doc: str = typer.Option(
+        None, "--doc",
+        help="Path al JSON con la doc della catena (payload A, campo ctes[]), oppure '-' per stdin."),
     config: Path = CONFIG_OPT,
 ) -> None:
-    """Persiste il piano CTE ordinato (sessions/<id>/cte_plan.json)."""
+    """Persiste il piano CTE ordinato (sessions/<id>/cte_plan.json).
+
+    Con --doc, valida e persiste anche sessions/<id>/cte_plan_doc.json (i nomi in
+    doc["ctes"] devono combaciare con --name, stesso ordine); cte_plan.json resta
+    invariato (list[str])."""
     import json
+    import sys
 
     from tht.phase import CTE_PLAN_FILE
 
@@ -144,8 +163,27 @@ def plan_cmd(
     from tht.cli.phase_cmd import require_phase_or_exit
     require_phase_or_exit(cfg, session, 6)
     sdir = session_dir(cfg, session)
+
+    doc_data = None
+    if doc is not None:
+        raw = sys.stdin.read() if doc == "-" else Path(doc).read_text()
+        try:
+            doc_data = json.loads(raw)
+        except json.JSONDecodeError as e:
+            typer.secho(f"ERRORE: --doc JSON non valido: {e}", fg=typer.colors.RED, err=True)
+            raise typer.Exit(code=1)
+        doc_names = [c["name"] for c in doc_data["ctes"]]
+        if doc_names != name:
+            typer.secho(
+                f"ERRORE: i nomi in --doc ({doc_names}) non combaciano con --name ({name}).",
+                fg=typer.colors.RED, err=True,
+            )
+            raise typer.Exit(code=1)
+
     sdir.mkdir(parents=True, exist_ok=True)
     (sdir / CTE_PLAN_FILE).write_text(json.dumps(name, ensure_ascii=False))
+    if doc_data is not None:
+        (sdir / CTE_PLAN_DOC_FILE).write_text(json.dumps(doc_data, ensure_ascii=False))
     typer.secho(f"OK: piano CTE salvato ({len(name)} CTE) in {sdir / CTE_PLAN_FILE}.",
                 fg=typer.colors.GREEN)
 
@@ -168,6 +206,76 @@ def next_cmd(
         typer.echo(nxt)
 
 
+@cte_app.command("info")
+def info_cmd(
+    name: str = typer.Argument(..., help="Nome del CTE (deve comparire in cte_plan.json)."),
+    session: str = typer.Option(..., "--session"),
+    json_out: bool = typer.Option(False, "--json", help="Emetti JSON puro su stdout."),
+    config: Path = CONFIG_OPT,
+) -> None:
+    """Dettaglio READ-ONLY di un CTE: posizione nel piano, SQL persistito, doc e ultimo test.
+
+    Fonte dati deterministica per il gate JS (WS2) che costruisce il payload cte_result v2."""
+    import json as _json
+
+    from tht.phase import CTE_PLAN_FILE, approved_ctes, cte_plan
+
+    cfg = _load_config_or_exit(config)
+    load_session_or_exit(cfg, session)
+    sdir = session_dir(cfg, session)
+
+    plan = cte_plan(sdir)
+    if not plan:
+        typer.secho(f"ERRORE: {CTE_PLAN_FILE} assente o vuoto per la sessione '{session}'.",
+                    fg=typer.colors.RED, err=True)
+        raise typer.Exit(code=1)
+    if name not in plan:
+        typer.secho(f"ERRORE: '{name}' non e' nel piano CTE ({plan}).",
+                    fg=typer.colors.RED, err=True)
+        raise typer.Exit(code=1)
+
+    cte_file = sdir / "ctes" / f"{name}.sql"
+    if not cte_file.exists():
+        typer.secho(f"ERRORE: file CTE non trovato: {cte_file}", fg=typer.colors.RED, err=True)
+        raise typer.Exit(code=1)
+
+    doc = None
+    doc_path = sdir / CTE_PLAN_DOC_FILE
+    if doc_path.exists():
+        full_doc = _json.loads(doc_path.read_text())
+        for c in full_doc.get("ctes", []):
+            if c.get("name") == name:
+                doc = {k: v for k, v in c.items() if k != "name"}
+                break
+
+    from tht.ctetest import CteError, load_cte_tests
+
+    try:
+        records = [r for r in load_cte_tests(sdir) if r.name == name]
+    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)
+    last_test = _json.loads(records[-1].model_dump_json()) if records else None
+
+    payload = {
+        "name": name,
+        "index": plan.index(name) + 1,
+        "total": len(plan),
+        "plan": plan,
+        "sql": cte_file.read_text(),
+        "approved": name in approved_ctes(sdir),
+        "doc": doc,
+        "last_test": last_test,
+    }
+
+    if json_out:
+        typer.echo(_json.dumps(payload, ensure_ascii=False))
+        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"),

harness/tht/ctetest.py

@@ -25,6 +25,21 @@ class CteTestRecord(BaseModel):
     execution_ms: int = 0
     warnings: list[str] = []
     error: str | None = None
+    preview_rows: list[list[str | int | float | bool | None]] | None = None
+
+
+_MAX_CELL_CHARS = 200
+
+
+def _jsonable(v):
+    """Coerce a raw DB cell to a JSON-native value for persistence in cte_tests.json.
+
+    None/bool/int/float pass through unchanged; str (and anything else, via str())
+    is truncated to ~200 chars."""
+    if v is None or isinstance(v, (bool, int, float)):
+        return v
+    s = v if isinstance(v, str) else str(v)
+    return s[:_MAX_CELL_CHARS]
 
 
 def has_trailing_select(cte_sql: str) -> bool: