ThothII/harness/tht/cli/phase_cmd.py

"""tht phase -- workflow HITL gate commands.

`meta` is the single source the gate (tht-gate.js) reads workflow facts from,
killing the JS/Python PHASE_NAMES drift (F2). advance/reopen/show are the phase
transitions; require_phase_or_exit is the guard shared with cte/decision/datamart
commands. All read workflow facts from load_workflow() (no mirrored constants).
"""
from __future__ import annotations

import json
from pathlib import Path

import typer

from tht.phase import (
    auto_advance_eligible,
    advance_problems,
    current_phase,
)
from tht.workflow import load_workflow

phase_app = typer.Typer(help="Fase del workflow HITL (gate di avanzamento/ritorno)")


def session_dir(cfg, session_id: str) -> Path:
    """Where a session's artifacts live. Mirror of session_cmd.session_dir (kept
    here to avoid a circular import: session_cmd imports phase helpers too)."""
    return cfg.paths.sessions / session_id


def require_phase_or_exit(cfg, session: str, min_phase: int) -> None:
    """Refuse with exit 1 if the session hasn't reached min_phase yet. The phase
    name in the message comes from workflow.yaml (load_workflow), not a constant."""
    cur = current_phase(session_dir(cfg, session))
    if cur < min_phase:
        wf = load_workflow()
        nome = wf.phase_name(min_phase)
        typer.secho(
            f"Impossibile: serve la Fase {min_phase} ({nome}), "
            f"sessione '{session}' è alla Fase {cur}.",
            fg=typer.colors.RED, err=True,
        )
        raise typer.Exit(1)



@phase_app.command("meta")
def meta_cmd(
    as_json: bool = typer.Option(
        False,
        "--json",
        help="Emette i metadati del workflow come JSON (usato dall'estensione gate).",
    ),
) -> None:
    """Stampa i metadati del workflow derivati da workflow.yaml (F2)."""
    wf = load_workflow()
    if not as_json:
        typer.echo(f"max_phase: {wf.max_phase}")
        for p in wf.phases:
            typer.echo(f"  {p.id} ({p.num}/{wf.max_phase}) {p.name}  advance={p.advance}")
        return
    typer.echo(
        json.dumps(
            {
                "schema_version": wf.schema_version,
                "max_phase": wf.max_phase,
                "phases": [
                    {
                        "num": p.num,
                        "id": p.id,
                        "name": p.name,
                        "advance": p.advance,
                        "artifacts_out": p.artifacts_out,
                        "emits": p.emits,
                    }
                    for p in wf.phases
                ],
            }
        )
    )


@phase_app.command("advance")
def advance_cmd(
    session: str = typer.Option(..., "--session", help="Id sessione."),
    auto: bool = typer.Option(
        False, "--auto",
        help="Avanza solo se la fase è completa (exit 6 se manca qualcosa; per il gate).",
    ),
) -> None:
    """Approva la fase corrente e passa alla successiva (persiste phase_approved)."""
    from tht.decisions import append_decision

    sdir = session_dir(_cfg(), session)
    cur = current_phase(sdir)
    wf = load_workflow()
    if cur > wf.max_phase:
        typer.secho("Sessione già alla fase terminale.", fg=typer.colors.YELLOW)
        raise typer.Exit(0)
    if auto:
        if not auto_advance_eligible(sdir):
            problems = advance_problems(sdir, cur)
            for p in problems:
                typer.echo(p)
            raise typer.Exit(6)  # needs human confirmation (gate contract)
    append_decision(sdir, type="phase_approved", subject=f"phase:{cur}")
    typer.echo(f"Fase {cur} ({wf.phase_name(cur)}) approvata → Fase {cur + 1}.")


@phase_app.command("reopen")
def reopen_cmd(
    session: str = typer.Option(..., "--session"),
    phase: int = typer.Option(..., "--phase", help="Fase a cui tornare (1..fase corrente -1)."),
) -> None:
    """Torna a una fase precedente (persiste phase_reopened + teardown artefatti)."""
    from tht.decisions import append_decision
    from tht.teardown import teardown_to_phase

    sdir = session_dir(_cfg(), session)
    cur = current_phase(sdir)
    if phase < 1 or phase >= cur:
        typer.secho(f"Target non valido (fase corrente {cur}).", fg=typer.colors.RED, err=True)
        raise typer.Exit(1)
    report = teardown_to_phase(sdir, target_phase=phase)
    append_decision(sdir, type="phase_reopened", subject=f"phase:{phase}")
    for f in report.deleted_files:
        typer.echo(f"  eliminato artefatto: {f}")
    typer.echo(f"Tornati alla Fase {phase} ({load_workflow().phase_name(phase)}).")


@phase_app.command("show")
def show_cmd(
    session: str = typer.Option(..., "--session"),
) -> None:
    """Mostra stato, fase corrente e ultime decisioni della sessione."""
    from tht.decisions import list_decisions

    sdir = session_dir(_cfg(), session)
    cur = current_phase(sdir)
    wf = load_workflow()
    typer.echo(f"Fase corrente: {cur}/{wf.max_phase} ({wf.phase_name(min(cur, wf.max_phase))})")
    decisions = list_decisions(sdir)
    if decisions:
        typer.echo(f"Decisioni registrate: {len(decisions)}")
        for d in decisions[-5:]:
            typer.echo(f"  #{d.seq} {d.type} {d.subject}")


def _cfg():
    """Load the workspace config. Resolves THT_WORKSPACE/THT_CONFIG first, then
    falls back to the CLI default ``config/tht.yaml`` (same convention as
    schema_cmd.CONFIG_OPT)."""
    import os
    from pathlib import Path

    ws = os.environ.get("THT_WORKSPACE") or os.environ.get("THT_CONFIG")
    config_path = Path(ws) if ws else Path("config/tht.yaml")
    try:
        from tht.cli.schema_cmd import _load_config_or_exit  # noqa: F401 (portato in Onda 1.4)

        return _load_config_or_exit(config_path)
    except ImportError:
        from tht.workspace import load_config

        return load_config(config_path)