ThothII/harness/tht/cli/cte_cmd.py

import hashlib
from datetime import UTC, datetime
from pathlib import Path

import typer

from tht.cli.config_cmd import CONFIG_OPT
from tht.cli.schema_cmd import _load_config_or_exit
from tht.cli.session_cmd import load_session_or_exit, session_dir
from tht.cli.sql_cmd import promoted_tables_for, require_action

_RULE6_HINT = (
    " — il file del CTE deve contenere SOLO il blocco WITH ... AS (...), "
    "senza SELECT finale: la SELECT la aggiunge tht in fase di test."
)

cte_app = typer.Typer(help="Test controllato dei CTE proposti (Agent View Generation)")


@cte_app.command("test")
def test_cmd(
    name: str = typer.Argument(..., help="Nome del CTE (file sessions/<id>/ctes/<nome>.sql)."),
    session: str = typer.Option(..., "--session"),
    json_out: bool = typer.Option(False, "--json", help="Emette l'esito come JSON (per Pi)."),
    config: Path = CONFIG_OPT,
) -> None:
    """Valida e testa un CTE: wrap su ultimo CTE + esecuzione nelle 4 reti."""
    from rich.console import Console
    from rich.table import Table

    from tht.cli.sql_cmd import do_run
    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

    cfg = _load_config_or_exit(config)
    require_action(cfg, "cte_test")
    load_session_or_exit(cfg, session)
    sdir = session_dir(cfg, session)

    from tht.cli.phase_cmd import require_phase_or_exit
    from tht.phase import next_cte

    require_phase_or_exit(cfg, session, 6)
    nxt = next_cte(sdir)
    if nxt is not None and name != nxt:
        typer.secho(
            f"ERRORE: ordine CTE. Ora tocca a '{nxt}' (il primo CTE del piano non "
            f"ancora approvato), non a '{name}'. Testa '{nxt}', poi falla approvare "
            f"dal reviewer (decisione cte_approved); solo allora potrai testare il "
            f"CTE successivo.",
            fg=typer.colors.RED, err=True,
        )
        raise typer.Exit(code=5)

    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)
    cte_sql = cte_file.read_text()
    sql_hash = hashlib.sha256(cte_sql.encode()).hexdigest()

    def _record_error(message: str) -> None:
        record = CteTestRecord(
            name=name, ts=datetime.now(UTC), sql_hash=sql_hash,
            status="error", error=message,
        )
        append_cte_test(sdir, record)
        if json_out:
            typer.echo(record.model_dump_json())
        else:
            typer.secho(f"ERRORE: {message}", fg=typer.colors.RED, err=True)
        raise typer.Exit(code=1)

    try:
        test_sql = build_test_sql(cte_sql)
    except CteError as e:
        msg = str(e)
        if "non parsabile" in msg and has_trailing_select(cte_sql):
            msg += _RULE6_HINT
        _record_error(msg)

    # validazione statica (esce con messaggi propri se invalida; qui la
    # intercettiamo per registrare comunque l'esito nel cte_tests.json)
    from tht.cli.sql_cmd import _load_physical_or_exit
    from tht.sqlcheck import validate_sql

    check = validate_sql(
        test_sql,
        physical=_load_physical_or_exit(cfg),
        promoted_tables=promoted_tables_for(cfg, session),
        forbidden_functions=set(cfg.execution.forbidden_functions),
    )
    for w in check.warnings:
        typer.secho(f"  warning: {w}", fg=typer.colors.YELLOW)
    if not check.ok:
        msg = "; ".join(check.errors)
        if any("statement" in e for e in check.errors):
            msg += _RULE6_HINT
        _record_error(msg)

    try:
        result = do_run(cfg, test_sql, limit=cfg.execution.max_preview_rows)
    except ExecutionError as e:
        _record_error(str(e))

    warnings = runtime_warnings(result, cfg.execution)
    record = CteTestRecord(
        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)

    if json_out:
        typer.echo(record.model_dump_json())
        return

    typer.echo(f"CTE {name} — {result.execution_ms} ms")
    table = Table()
    for col in result.columns:
        table.add_column(col)
    for row in result.rows:
        table.add_row(*[str(v) for v in row])
    Console().print(table)
    for w in warnings:
        typer.secho(f"  warning: {w}", fg=typer.colors.YELLOW)
    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).

    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

    cfg = _load_config_or_exit(config)
    load_session_or_exit(cfg, session)
    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)


@cte_app.command("next")
def next_cmd(
    session: str = typer.Option(..., "--session"),
    config: Path = CONFIG_OPT,
) -> None:
    """Primo CTE del piano non ancora approvato (stdout pulito; vuoto se nessuno).

    Sorgente unica dell'ordine di approvazione CTE (F6): il gate lo usa per
    registrare cte_approved sul NOME del CTE (non su 'phase:6')."""
    from tht.phase import next_cte

    cfg = _load_config_or_exit(config)
    load_session_or_exit(cfg, session)
    nxt = next_cte(session_dir(cfg, session))
    if nxt:
        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"),
    config: Path = CONFIG_OPT,
) -> None:
    """Ultimo esito registrato per ogni CTE della sessione."""
    from tht.ctetest import CteError, load_cte_tests

    cfg = _load_config_or_exit(config)
    load_session_or_exit(cfg, session)
    try:
        records = load_cte_tests(session_dir(cfg, session))
    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)