From a312746fc8176dc3156dd66ba365dd3768f32bdb Mon Sep 17 00:00:00 2001 From: mptyl Date: Sat, 27 Jun 2026 13:15:12 +0200 Subject: [PATCH] feat(harness): require_phase_or_exit + phase advance/reopen/show (Onda 1.2) require_phase_or_exit: guard riscritto vs Workflow (load_workflow().phase_name invece della costante PHASE_NAMES drift). Exit 1 se la sessione e' sotto soglia. Usato da cte/decision/datamart cmd. Comandi phase (portati + adattati al modello ThothII, non copia cieca): - advance: persiste phase_approved; --auto exit 6 se la fase non e' completa (contratto col gate) - reopen: persiste phase_reopened + teardown_to_phase degli artefatti oltre il target - show: stato sessione (fase corrente, ultime decisioni) session_dir helper tenuto qui (mirror di session_cmd) per evitare circular import. _cfg() fa fallback a THT_WORKSPACE env finche' _load_config_or_exit (Onda 1.4) non sara' portato. L1: 4 test require_phase_or_exit (allow at/above, exit below, message con nome fase dal workflow). Suite: 161 passed. --- harness/tests/test_require_phase_or_exit.py | 69 +++++++++++ harness/tht/cli/phase_cmd.py | 122 +++++++++++++++++++- 2 files changed, 188 insertions(+), 3 deletions(-) create mode 100644 harness/tests/test_require_phase_or_exit.py diff --git a/harness/tests/test_require_phase_or_exit.py b/harness/tests/test_require_phase_or_exit.py new file mode 100644 index 00000000..75f5c8b2 --- /dev/null +++ b/harness/tests/test_require_phase_or_exit.py @@ -0,0 +1,69 @@ +"""L1: require_phase_or_exit — the phase-gate guard rewritten vs Workflow. + +Used by cte_cmd/decision_cmd/datamart_cmd to refuse an action when the session +hasn't reached a phase yet. The rewrite replaces the old PHASE_NAMES constant +with load_workflow().phase_name(). Exit code 1 (typer.Exit) when below threshold. +""" +import pytest +import typer + +from tht.cli.phase_cmd import require_phase_or_exit + + +def _make_session(tmp_path, current_phase_num: int) -> str: + """Build a session dir whose current_phase reads as current_phase_num. + current_phase is derived from the ledger fold (cur = last approved + 1), so + approving phases 1..N-1 puts the session at phase N.""" + import json + + s = tmp_path / "sess" + s.mkdir() + decisions = [ + {"seq": n, "type": "phase_approved", "subject": f"phase:{n}", "ts": "2025-01-01T00:00:00"} + for n in range(1, current_phase_num) + ] + (s / "review_decisions.jsonl").write_text( + "\n".join(json.dumps(d) for d in decisions) + ("\n" if decisions else "") + ) + return "sess" + + +class _StubConfig: + """Minimal config: paths.sessions points at tmp_path so session_dir resolves there.""" + + def __init__(self, tmp_path): + from tht.config import PathsConfig + + self.paths = PathsConfig(sessions=tmp_path) + + +def test_require_phase_or_exit_allows_when_at_or_above(tmp_path): + cfg = _StubConfig(tmp_path) + session = _make_session(tmp_path, current_phase_num=5) # at phase 5 + # require phase 4 -> allowed (cur 5 >= 4), no exit + require_phase_or_exit(cfg, session, min_phase=4) + + +def test_require_phase_or_exit_allows_when_exactly_at(tmp_path): + cfg = _StubConfig(tmp_path) + session = _make_session(tmp_path, current_phase_num=4) # at phase 4 + require_phase_or_exit(cfg, session, min_phase=4) # boundary: allowed + + +def test_require_phase_or_exit_exits_when_below(tmp_path): + cfg = _StubConfig(tmp_path) + session = _make_session(tmp_path, current_phase_num=3) # at phase 3 + with pytest.raises(typer.Exit) as exc_info: + require_phase_or_exit(cfg, session, min_phase=5) # require 5 -> exit 1 + assert exc_info.value.exit_code == 1 + + +def test_require_phase_or_exit_message_uses_workflow_phase_name(tmp_path, capsys): + cfg = _StubConfig(tmp_path) + session = _make_session(tmp_path, current_phase_num=3) + with pytest.raises(typer.Exit): + require_phase_or_exit(cfg, session, min_phase=6) # phase 6 = cte + captured = capsys.readouterr() + # the message must use the workflow-derived name (not a hardcoded constant) + assert "cte" in captured.err.lower() + assert "Fase 6" in captured.err diff --git a/harness/tht/cli/phase_cmd.py b/harness/tht/cli/phase_cmd.py index 07e89adb..6f295281 100644 --- a/harness/tht/cli/phase_cmd.py +++ b/harness/tht/cli/phase_cmd.py @@ -1,21 +1,49 @@ """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 land with the -session/db ports (A9) -- they need current_phase + load_session_or_exit which -depend on the not-yet-ported session store helpers. +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( @@ -49,3 +77,91 @@ def meta_cmd( } ) ) + + +@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. Tries the shared helper once schema_cmd is ported + (Onda 1.4); falls back to THT_WORKSPACE env + load_config for now.""" + try: + from tht.cli.schema_cmd import _load_config_or_exit # noqa: F401 (portato in Onda 1.4) + + return _load_config_or_exit() + except ImportError: + import os + + from tht.workspace import load_config + + ws = os.environ.get("THT_WORKSPACE") or os.environ.get("THT_CONFIG") + if not ws: + typer.secho( + "Nessun workspace: imposta THT_WORKSPACE o usa --config.", + fg=typer.colors.RED, err=True, + ) + raise typer.Exit(2) + return load_config(ws)