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

import typer

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

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


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."""
    from tht.cli.session_cmd import load_snapshot_or_exit

    cur = current_phase(load_snapshot_or_exit(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.cli.session_cmd import load_snapshot_or_exit, session_repository

    cfg = _cfg()
    snapshot = load_snapshot_or_exit(cfg, session)
    cur = current_phase(snapshot)
    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 and not auto_advance_eligible(snapshot):
        problems = advance_problems(snapshot, cur)
        for p in problems:
            typer.echo(p)
        raise typer.Exit(6)  # needs human confirmation (gate contract)
    session_repository(cfg).append_decisions(
        session, [{"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.cli.session_cmd import load_snapshot_or_exit, session_repository

    cfg = _cfg()
    snapshot = load_snapshot_or_exit(cfg, session)
    cur = current_phase(snapshot)
    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)
    from tht.teardown import teardown_snapshot

    repository = session_repository(cfg)
    # Ledger FIRST, teardown after: a crash between the two used to leave later-phase
    # artifacts deleted with the ledger still at the old phase (resume entered a phase
    # whose expected artifacts were gone). The inverse half-state — reopened with stale
    # later artifacts — is benign: the folded phase wins and the workflow overwrites
    # them as it re-progresses.
    repository.append_decisions(
        session, [{"type": "phase_reopened", "subject": f"phase:{phase}"}]
    )
    report = teardown_snapshot(repository, snapshot, 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.cli.session_cmd import load_snapshot_or_exit

    snapshot = load_snapshot_or_exit(_cfg(), session)
    cur = current_phase(snapshot)
    wf = load_workflow()
    typer.echo(f"Fase corrente: {cur}/{wf.max_phase} ({wf.phase_name(min(cur, wf.max_phase))})")
    decisions = snapshot.decisions
    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

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

        return load_config(config_path)