650 lines
26 KiB
Python
650 lines
26 KiB
Python
import json
|
|
from pathlib import Path
|
|
|
|
import typer
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
|
|
from tht.cli.config_cmd import CONFIG_OPT
|
|
from tht.cli.schema_cmd import _load_config_or_exit
|
|
|
|
session_app = typer.Typer(help="Sessioni (directory artefatti)")
|
|
|
|
preferences_app = typer.Typer(help="Preferenze private del principal corrente")
|
|
session_app.add_typer(preferences_app, name="preferences")
|
|
|
|
|
|
def session_repository(cfg):
|
|
"""Configured persistence boundary for every workflow command."""
|
|
from tht.session.repository import build_session_repository
|
|
|
|
return build_session_repository(cfg)
|
|
|
|
|
|
@preferences_app.command("get")
|
|
def preferences_get_cmd(
|
|
json_out: bool = typer.Option(False, "--json", help="Emetti JSON puro."),
|
|
config: Path = CONFIG_OPT,
|
|
) -> None:
|
|
"""Legge le preferenze private del principal risolto dal repository."""
|
|
cfg = _load_config_or_exit(config)
|
|
preferences = session_repository(cfg).get_preferences()
|
|
if json_out:
|
|
typer.echo(json.dumps(preferences, ensure_ascii=False, sort_keys=True))
|
|
else:
|
|
typer.echo(json.dumps(preferences, ensure_ascii=False, indent=2, sort_keys=True))
|
|
|
|
|
|
@preferences_app.command("set")
|
|
def preferences_set_cmd(
|
|
preferences_json: str = typer.Argument(..., help="Oggetto JSON delle preferenze."),
|
|
json_out: bool = typer.Option(False, "--json", help="Non emette testo al successo."),
|
|
config: Path = CONFIG_OPT,
|
|
) -> None:
|
|
"""Sostituisce le preferenze private del principal risolto dal repository."""
|
|
try:
|
|
preferences = json.loads(preferences_json)
|
|
except json.JSONDecodeError as exc:
|
|
typer.secho(f"ERRORE: preferenze JSON non valide: {exc}", fg=typer.colors.RED, err=True)
|
|
raise typer.Exit(code=2) from None
|
|
if not isinstance(preferences, dict):
|
|
typer.secho("ERRORE: le preferenze devono essere un oggetto JSON", fg=typer.colors.RED, err=True)
|
|
raise typer.Exit(code=2)
|
|
cfg = _load_config_or_exit(config)
|
|
session_repository(cfg).set_preferences(preferences)
|
|
if json_out:
|
|
return
|
|
typer.secho("OK: preferenze aggiornate.", fg=typer.colors.GREEN)
|
|
|
|
|
|
def session_dir(cfg, session_id: str) -> Path:
|
|
"""Legacy path bridge for out-of-scope datamart/memory compatibility only.
|
|
|
|
Workflow session commands in this module use ``session_repository``; this
|
|
helper remains until those non-workflow consumers lose their path APIs.
|
|
"""
|
|
return cfg.paths.sessions / session_id
|
|
|
|
|
|
@session_app.command("migrate")
|
|
def migrate_cmd(
|
|
database_url: str = typer.Option(
|
|
..., "--database-url", envvar="THT_SESSIONS_ADMIN_URL", help="Admin PostgreSQL URL."
|
|
),
|
|
status_only: bool = typer.Option(False, "--status", help="Inspect without applying."),
|
|
json_output: bool = typer.Option(False, "--json", help="Emit pristine JSON."),
|
|
) -> None:
|
|
"""Apply or inspect the private PostgreSQL session schema migrations."""
|
|
from tht.session.postgres_repository import MigrationError, migrate, migration_status
|
|
|
|
try:
|
|
status = migration_status(database_url) if status_only else migrate(database_url)
|
|
except (MigrationError, SQLAlchemyError, ValueError) as exc:
|
|
if json_output:
|
|
typer.echo(json.dumps({"error": str(exc)}, sort_keys=True))
|
|
else:
|
|
typer.echo(f"ERROR: {exc}", err=True)
|
|
raise typer.Exit(code=1) from None
|
|
payload = {
|
|
"applied": [item.version for item in status.applied],
|
|
"drifted": [item.version for item in status.drifted],
|
|
"pending": [item.version for item in status.pending],
|
|
}
|
|
if json_output:
|
|
typer.echo(json.dumps(payload, sort_keys=True))
|
|
else:
|
|
typer.echo(
|
|
f"Applied: {len(status.applied)}; pending: {len(status.pending)}; "
|
|
f"drifted: {len(status.drifted)}"
|
|
)
|
|
|
|
|
|
def load_session_or_exit(cfg, session_id: str):
|
|
from tht.session.store import SessionError
|
|
|
|
try:
|
|
return session_repository(cfg).get(session_id).manifest
|
|
except SessionError as e:
|
|
typer.secho(f"ERRORE: {e}", fg=typer.colors.RED, err=True)
|
|
raise typer.Exit(code=1)
|
|
|
|
|
|
def load_snapshot_or_exit(cfg, session_id: str):
|
|
from tht.session.store import SessionError
|
|
|
|
try:
|
|
return session_repository(cfg).get(session_id)
|
|
except SessionError as e:
|
|
typer.secho(f"ERRORE: {e}", fg=typer.colors.RED, err=True)
|
|
raise typer.Exit(code=1)
|
|
|
|
|
|
def _list_sessions(sessions_root: Path) -> list[dict]:
|
|
"""Return session summary dicts sorted by created_at desc (pure helper, no I/O side-effects)."""
|
|
from tht.session.models import SessionManifest
|
|
|
|
MANIFEST_NAME = "session_manifest.yaml"
|
|
out = []
|
|
if not sessions_root.exists():
|
|
return out
|
|
for d in sessions_root.iterdir():
|
|
manifest_path = d / MANIFEST_NAME
|
|
if not manifest_path.exists():
|
|
continue
|
|
m = SessionManifest.from_yaml(manifest_path)
|
|
out.append({
|
|
"id": m.id,
|
|
"status": m.status,
|
|
"question": m.question,
|
|
"summary": m.summary,
|
|
"created_at": m.created_at.isoformat(),
|
|
"updated_at": m.updated_at.isoformat() if m.updated_at else None,
|
|
"author": m.author,
|
|
"name": m.name,
|
|
"group": m.group,
|
|
"archived": m.archived,
|
|
})
|
|
out.sort(key=lambda r: r["created_at"], reverse=True)
|
|
return out
|
|
|
|
|
|
@session_app.command("list")
|
|
def list_cmd(
|
|
json_out: bool = typer.Option(False, "--json", help="Emetti JSON su stdout (pristine)."),
|
|
config: Path = CONFIG_OPT,
|
|
) -> None:
|
|
"""Elenca le sessioni esistenti."""
|
|
cfg = _load_config_or_exit(config)
|
|
rows = [
|
|
{"id": s.manifest.id, "status": s.manifest.status, "question": s.manifest.question,
|
|
"summary": s.manifest.summary, "created_at": s.manifest.created_at.isoformat(),
|
|
"updated_at": s.manifest.updated_at.isoformat() if s.manifest.updated_at else None,
|
|
"author": s.manifest.author, "name": s.manifest.name, "group": s.manifest.group,
|
|
"archived": s.manifest.archived}
|
|
for s in session_repository(cfg).list()
|
|
]
|
|
if json_out:
|
|
typer.echo(json.dumps(rows, ensure_ascii=False, indent=2))
|
|
return
|
|
for r in rows:
|
|
typer.echo(f"{r['id']} [{r['status']}] {r['summary'] or r['question'][:60]}")
|
|
|
|
|
|
@session_app.command("new")
|
|
def new_cmd(
|
|
question: str = typer.Argument(..., help="La domanda in linguaggio naturale."),
|
|
provider: str = typer.Option(None, "--provider", help="Provider LLM (es. zai, anthropic)."),
|
|
model: str = typer.Option(None, "--model", help="Modello LLM (es. glm-5.2)."),
|
|
thinking: str = typer.Option(None, "--thinking", help="Livello di thinking (es. medium)."),
|
|
workspace_id: str = typer.Option(None, "--workspace-id", help="Workspace canonico risolto dal backend."),
|
|
workspace_revision: str = typer.Option(None, "--workspace-revision", help="Commit immutabile del workspace."),
|
|
name: str = typer.Option(None, "--name", help="Nome descrittivo della sessione."),
|
|
json_out: bool = typer.Option(False, "--json", help="Emetti JSON puro {\"id\": ...} su stdout."),
|
|
config: Path = CONFIG_OPT,
|
|
) -> None:
|
|
"""Crea una sessione e stampa il suo id (ultima riga dell'output)."""
|
|
from tht.session.store import _extract_name, new_session_manifest, render_question_md
|
|
|
|
cfg = _load_config_or_exit(config)
|
|
manifest = new_session_manifest(question, cfg.database,
|
|
provider=provider, model=model, thinking=thinking,
|
|
workspace_id=workspace_id, workspace_revision=workspace_revision,
|
|
name=name or _extract_name(question))
|
|
repository = session_repository(cfg)
|
|
repository.create(manifest)
|
|
repository.write_artifact(manifest.id, "question", render_question_md(question))
|
|
if json_out:
|
|
typer.echo(json.dumps({"id": manifest.id}, ensure_ascii=False))
|
|
return
|
|
typer.secho(f"OK: sessione creata ({manifest.id})", fg=typer.colors.GREEN)
|
|
typer.echo(manifest.id)
|
|
|
|
|
|
@session_app.command("set-question")
|
|
def set_question_cmd(
|
|
session_id: str = typer.Argument(...),
|
|
question: str = typer.Option(..., "--question", "-q",
|
|
help="La domanda riscritta (chiara)."),
|
|
assumption: list[str] = typer.Option(
|
|
None, "--assumption", "-a",
|
|
help="Una assunzione (ripetibile). Testo libero, già formattato."),
|
|
config: Path = CONFIG_OPT,
|
|
) -> None:
|
|
"""Scrive question.md (domanda riscritta + assunzioni) in modo deterministico."""
|
|
from tht.session.store import render_question_md
|
|
|
|
cfg = _load_config_or_exit(config)
|
|
load_session_or_exit(cfg, session_id)
|
|
session_repository(cfg).write_artifact(session_id, "question", render_question_md(question, assumption or []))
|
|
typer.secho("OK: question.md aggiornato.", fg=typer.colors.GREEN)
|
|
|
|
|
|
@session_app.command("set-schema-linking")
|
|
def set_schema_linking_cmd(
|
|
session_id: str = typer.Argument(...),
|
|
file: str = typer.Option(
|
|
..., "--file", "-f",
|
|
help="Path al JSON dello schema-linking, oppure '-' per leggere da stdin."),
|
|
config: Path = CONFIG_OPT,
|
|
) -> None:
|
|
"""Valida (modello SchemaLinking) e scrive schema_linking.json deterministicamente."""
|
|
import sys
|
|
|
|
from pydantic import ValidationError
|
|
|
|
from tht.session.models import SchemaLinking
|
|
|
|
cfg = _load_config_or_exit(config)
|
|
load_session_or_exit(cfg, session_id)
|
|
raw = sys.stdin.read() if file == "-" else Path(file).read_text()
|
|
try:
|
|
data = json.loads(raw)
|
|
except json.JSONDecodeError as e:
|
|
typer.secho(f"ERRORE: JSON non valido: {e}", fg=typer.colors.RED, err=True)
|
|
raise typer.Exit(code=5)
|
|
try:
|
|
model = SchemaLinking.model_validate(data)
|
|
session_repository(cfg).write_artifact(session_id, "schema_linking", json.dumps(model.model_dump(by_alias=True), indent=2, ensure_ascii=False))
|
|
except ValidationError as e:
|
|
typer.secho(f"ERRORE: schema_linking non valido:\n{e}", fg=typer.colors.RED, err=True)
|
|
raise typer.Exit(code=5)
|
|
typer.secho("OK: schema_linking.json aggiornato.", fg=typer.colors.GREEN)
|
|
|
|
|
|
@session_app.command("sync-schema-linking")
|
|
def sync_schema_linking_cmd(
|
|
session_id: str = typer.Argument(...),
|
|
config: Path = CONFIG_OPT,
|
|
) -> None:
|
|
"""Riproietta schema_linking.json dalle decisioni F4 del ledger (deterministico)."""
|
|
from pydantic import ValidationError
|
|
|
|
from tht.session.store import sync_schema_linking_snapshot
|
|
|
|
cfg = _load_config_or_exit(config)
|
|
load_session_or_exit(cfg, session_id)
|
|
try:
|
|
content = sync_schema_linking_snapshot(load_snapshot_or_exit(cfg, session_id))
|
|
session_repository(cfg).write_artifact(session_id, "schema_linking", content)
|
|
except ValidationError as e:
|
|
typer.secho(f"ERRORE: schema_linking non valido:\n{e}", fg=typer.colors.RED, err=True)
|
|
raise typer.Exit(code=5)
|
|
typer.secho("OK: schema_linking.json riproiettato.", fg=typer.colors.GREEN)
|
|
|
|
|
|
@session_app.command("show")
|
|
def show_cmd(
|
|
session_id: str = typer.Argument(...),
|
|
json_out: bool = typer.Option(False, "--json", help="Emetti JSON su stdout (pristine)."),
|
|
config: Path = CONFIG_OPT,
|
|
) -> None:
|
|
"""Stato della sessione: manifest + decisioni registrate (per la ripresa)."""
|
|
from tht.phase import current_phase
|
|
|
|
cfg = _load_config_or_exit(config)
|
|
manifest = load_session_or_exit(cfg, session_id)
|
|
snapshot = load_snapshot_or_exit(cfg, session_id)
|
|
|
|
if json_out:
|
|
has_schema_linking = "schema_linking" in snapshot.artifacts
|
|
phase = current_phase(snapshot)
|
|
data = manifest.model_dump(mode="json", by_alias=True)
|
|
data["phase"] = phase
|
|
data["has_schema_linking"] = has_schema_linking
|
|
# Ledger integrale: il gate lo usa per costruire deterministicamente il
|
|
# recap delle decisioni nei riepiloghi di fase (v2).
|
|
data["decisions"] = [
|
|
d.model_dump(mode="json") for d in snapshot.decisions
|
|
]
|
|
typer.echo(json.dumps(data, ensure_ascii=False, indent=2))
|
|
return
|
|
|
|
decisions = snapshot.decisions
|
|
typer.echo(f"id : {manifest.id}")
|
|
typer.echo(f"stato : {manifest.status}")
|
|
typer.echo(f"domanda : {manifest.question}")
|
|
typer.echo(f"target : {manifest.database} / {manifest.db_schema}")
|
|
typer.echo(f"decisioni: {len(decisions)}")
|
|
for d in decisions[-10:]:
|
|
typer.echo(f" [{d.seq}] {d.type}: {d.subject}" + (f" — {d.detail}" if d.detail else ""))
|
|
typer.echo(f"schema_linking.json: {'presente' if 'schema_linking' in snapshot.artifacts else 'assente'}")
|
|
|
|
|
|
@session_app.command("retrieval-pack")
|
|
def retrieval_pack_cmd(
|
|
session_id: str = typer.Argument(...),
|
|
config: Path = CONFIG_OPT,
|
|
) -> None:
|
|
"""Emette su stdout il retrieval pack persistito della sessione."""
|
|
cfg = _load_config_or_exit(config)
|
|
load_session_or_exit(cfg, session_id)
|
|
content = load_snapshot_or_exit(cfg, session_id).artifacts.get("retrieval_pack")
|
|
if content is None:
|
|
typer.secho(
|
|
f"ERRORE: retrieval pack non disponibile per la sessione {session_id}",
|
|
fg=typer.colors.RED,
|
|
err=True,
|
|
)
|
|
raise typer.Exit(code=1)
|
|
typer.echo(content, nl=False)
|
|
|
|
|
|
@session_app.command("close")
|
|
def close_cmd(session_id: str = typer.Argument(...), config: Path = CONFIG_OPT) -> None:
|
|
"""Chiude la sessione (status=closed)."""
|
|
cfg = _load_config_or_exit(config)
|
|
load_session_or_exit(cfg, session_id)
|
|
snapshot = load_snapshot_or_exit(cfg, session_id)
|
|
snapshot.manifest.status = "closed"
|
|
session_repository(cfg).save_manifest(snapshot.manifest)
|
|
typer.secho(f"OK: sessione {session_id} chiusa.", fg=typer.colors.GREEN)
|
|
|
|
|
|
@session_app.command("fail")
|
|
def fail_cmd(session_id: str = typer.Argument(...), config: Path = CONFIG_OPT) -> None:
|
|
"""Registra un arresto del sistema non recuperabile automaticamente."""
|
|
cfg = _load_config_or_exit(config)
|
|
load_session_or_exit(cfg, session_id)
|
|
snapshot = load_snapshot_or_exit(cfg, session_id)
|
|
snapshot.manifest.status = "failed"
|
|
session_repository(cfg).save_manifest(snapshot.manifest)
|
|
typer.secho(f"OK: sessione {session_id} marcata failed.", fg=typer.colors.RED)
|
|
|
|
@session_app.command("reopen")
|
|
def reopen_cmd(session_id: str = typer.Argument(...), config: Path = CONFIG_OPT) -> None:
|
|
"""Riapre una sessione per una ripresa manuale."""
|
|
cfg = _load_config_or_exit(config)
|
|
load_session_or_exit(cfg, session_id)
|
|
snapshot = load_snapshot_or_exit(cfg, session_id)
|
|
snapshot.manifest.status = "open"
|
|
session_repository(cfg).save_manifest(snapshot.manifest)
|
|
typer.secho(f"OK: sessione {session_id} riaperta.", fg=typer.colors.GREEN)
|
|
|
|
|
|
@session_app.command("set-name")
|
|
def set_name_cmd(
|
|
session_id: str = typer.Argument(...),
|
|
name: str = typer.Option(..., "--name", help="Nome descrittivo (vuoto = azzera)."),
|
|
config: Path = CONFIG_OPT,
|
|
) -> None:
|
|
"""Imposta il nome descrittivo della sessione."""
|
|
cfg = _load_config_or_exit(config)
|
|
load_session_or_exit(cfg, session_id)
|
|
snapshot = load_snapshot_or_exit(cfg, session_id)
|
|
snapshot.manifest.name = name.strip() or None
|
|
session_repository(cfg).save_manifest(snapshot.manifest)
|
|
typer.secho(f"OK: nome aggiornato per {session_id}.", fg=typer.colors.GREEN)
|
|
|
|
|
|
@session_app.command("set-group")
|
|
def set_group_cmd(
|
|
session_id: str = typer.Argument(...),
|
|
group: str = typer.Option(..., "--group", help="Nome del gruppo (vuoto = nessun gruppo)."),
|
|
config: Path = CONFIG_OPT,
|
|
) -> None:
|
|
"""Sposta la sessione in un gruppo (o la toglie da ogni gruppo)."""
|
|
cfg = _load_config_or_exit(config)
|
|
load_session_or_exit(cfg, session_id)
|
|
snapshot = load_snapshot_or_exit(cfg, session_id)
|
|
snapshot.manifest.group = group.strip() or None
|
|
session_repository(cfg).save_manifest(snapshot.manifest)
|
|
typer.secho(f"OK: gruppo aggiornato per {session_id}.", fg=typer.colors.GREEN)
|
|
|
|
|
|
@session_app.command("archive")
|
|
def archive_cmd(session_id: str = typer.Argument(...), config: Path = CONFIG_OPT) -> None:
|
|
"""Archivia la sessione (la toglie dalla lista attiva, sola lettura)."""
|
|
cfg = _load_config_or_exit(config)
|
|
load_session_or_exit(cfg, session_id)
|
|
snapshot = load_snapshot_or_exit(cfg, session_id)
|
|
snapshot.manifest.archived = True
|
|
session_repository(cfg).save_manifest(snapshot.manifest)
|
|
typer.secho(f"OK: sessione {session_id} archiviata.", fg=typer.colors.GREEN)
|
|
|
|
|
|
@session_app.command("unarchive")
|
|
def unarchive_cmd(session_id: str = typer.Argument(...), config: Path = CONFIG_OPT) -> None:
|
|
"""Ripristina la sessione dall'archivio (non ne cambia la ripristinabilità)."""
|
|
cfg = _load_config_or_exit(config)
|
|
load_session_or_exit(cfg, session_id)
|
|
snapshot = load_snapshot_or_exit(cfg, session_id)
|
|
snapshot.manifest.archived = False
|
|
session_repository(cfg).save_manifest(snapshot.manifest)
|
|
typer.secho(f"OK: sessione {session_id} ripristinata.", fg=typer.colors.GREEN)
|
|
|
|
|
|
@session_app.command("delete")
|
|
def delete_cmd(session_id: str = typer.Argument(...), config: Path = CONFIG_OPT) -> None:
|
|
"""Elimina definitivamente la cartella di sessione."""
|
|
cfg = _load_config_or_exit(config)
|
|
load_session_or_exit(cfg, session_id)
|
|
session_repository(cfg).delete(session_id)
|
|
typer.secho(f"OK: sessione {session_id} eliminata.", fg=typer.colors.GREEN)
|
|
|
|
|
|
@session_app.command("documents")
|
|
def documents_cmd(
|
|
session_id: str = typer.Argument(...),
|
|
json_out: bool = typer.Option(False, "--json", help="Emetti JSON su stdout (pristine)."),
|
|
config: Path = CONFIG_OPT,
|
|
) -> None:
|
|
"""Documenti di sola lettura della sessione (domanda, rivista, schema, SQL, report, decisioni)."""
|
|
from tht.session.store import build_snapshot_documents
|
|
|
|
cfg = _load_config_or_exit(config)
|
|
docs = build_snapshot_documents(load_snapshot_or_exit(cfg, session_id))
|
|
if json_out:
|
|
typer.echo(json.dumps(docs, ensure_ascii=False, indent=2))
|
|
return
|
|
for d in docs:
|
|
typer.echo(f"[{d['phase']}] {d['title']} ({d['format']})")
|
|
|
|
|
|
def session_problems(cfg, session_id: str) -> list[str]:
|
|
"""Problemi del Blocco 3 (decisioni + schema_linking). Riusato da check e finalize."""
|
|
import json
|
|
|
|
from pydantic import ValidationError
|
|
|
|
from tht.session.models import SchemaLinking
|
|
|
|
snapshot = load_snapshot_or_exit(cfg, session_id)
|
|
problems: list[str] = []
|
|
if not snapshot.decisions:
|
|
problems.append("nessuna decisione registrata (review_decisions.jsonl vuoto o assente)")
|
|
raw_linking = snapshot.artifacts.get("schema_linking")
|
|
if raw_linking is None:
|
|
problems.append("schema_linking.json assente")
|
|
else:
|
|
try:
|
|
SchemaLinking.model_validate(json.loads(raw_linking))
|
|
except (json.JSONDecodeError, ValidationError) as e:
|
|
problems.append(f"schema_linking.json non valido: {e}")
|
|
return problems
|
|
|
|
|
|
@session_app.command("check")
|
|
def check_cmd(session_id: str = typer.Argument(...), config: Path = CONFIG_OPT) -> None:
|
|
"""Gate oggettivo: decisioni presenti e schema_linking.json valido (dalla Fase 4)."""
|
|
from tht.phase import current_phase
|
|
from tht.workflow import load_workflow
|
|
|
|
cfg = _load_config_or_exit(config)
|
|
load_session_or_exit(cfg, session_id)
|
|
cur = current_phase(load_snapshot_or_exit(cfg, session_id))
|
|
wf = load_workflow()
|
|
schema_linking_phase = wf.schema_linking_phase()
|
|
if cur < schema_linking_phase:
|
|
# Prima della fase schema-linking non e' ancora atteso: il gate Blocco 3
|
|
# non e' applicabile. Riportarlo come errore manderebbe il workflow in loop.
|
|
nome = wf.phase_name(cur)
|
|
typer.secho(
|
|
f"Sessione {session_id} in Fase {cur} ({nome}): il gate Blocco 3 "
|
|
f"(schema_linking.json) si applica dalla Fase {schema_linking_phase} "
|
|
f"({wf.phase_name(schema_linking_phase)}). Niente da verificare ora: "
|
|
"prosegui con lo schema linking.",
|
|
fg=typer.colors.CYAN,
|
|
)
|
|
return
|
|
problems = session_problems(cfg, session_id)
|
|
if problems:
|
|
typer.secho(f"Sessione {session_id} incompleta:", fg=typer.colors.YELLOW)
|
|
for p in problems:
|
|
typer.echo(f" - {p}")
|
|
raise typer.Exit(code=3)
|
|
typer.secho(f"OK: sessione {session_id} completa per il Blocco 3.", fg=typer.colors.GREEN)
|
|
|
|
|
|
ARTIFACT_FILES = [
|
|
"session_manifest.yaml", "question.md", "schema_linking.json", "evidence.json",
|
|
"cte_tests.json", "sql_final.sql", "validation_report.md", "review_decisions.jsonl",
|
|
]
|
|
|
|
|
|
@session_app.command("finalize")
|
|
def finalize_cmd(session_id: str = typer.Argument(...), config: Path = CONFIG_OPT) -> None:
|
|
"""Gate finale + batteria di validazione + artefatto di sessione completo."""
|
|
import json
|
|
|
|
from tht.cli.sql_cmd import (
|
|
_load_physical_or_exit,
|
|
do_explain,
|
|
do_run,
|
|
promoted_tables_for,
|
|
)
|
|
from tht.ctetest import CteError, CteTestRecord, _iter_json_objects
|
|
from tht.execute import ExecutionError
|
|
from tht.execute.warnings import plan_warnings, runtime_warnings, static_warnings
|
|
from tht.report import extract_reviewer_notes, render_validation_report
|
|
from tht.session.artifacts import build_evidence_entries
|
|
from tht.phase import cte_plan as effective_cte_plan
|
|
from tht.phase import effective_decisions
|
|
from tht.session.models import SchemaLinking
|
|
from tht.sqlcheck import validate_sql
|
|
|
|
cfg = _load_config_or_exit(config)
|
|
repository = session_repository(cfg)
|
|
snapshot = load_snapshot_or_exit(cfg, session_id)
|
|
|
|
from tht.phase import current_phase
|
|
from tht.workflow import load_workflow
|
|
|
|
cur = current_phase(snapshot)
|
|
wf = load_workflow()
|
|
if cur <= wf.max_phase:
|
|
typer.secho(
|
|
f"Finalize rifiutato: workflow non completo, sei in Fase {cur} "
|
|
f"({wf.phase_name(min(cur, wf.max_phase))}). Tutte le {wf.max_phase} fasi "
|
|
"devono essere approvate (anche dopo eventuali reopen).",
|
|
fg=typer.colors.YELLOW, err=True,
|
|
)
|
|
raise typer.Exit(code=5)
|
|
|
|
# --- gate di ingresso ---
|
|
# Vista effective (D15): un sql_approved/cte ritirato o stale post-rollback non conta.
|
|
problems = session_problems(cfg, session_id)
|
|
decisions = effective_decisions(snapshot)
|
|
sql = snapshot.artifacts.get("sql_final")
|
|
if sql is None:
|
|
problems.append("sql_final.sql assente")
|
|
if not any(d.type == "sql_approved" for d in decisions):
|
|
problems.append(
|
|
"decisione sql_approved assente: la validazione semantica del reviewer "
|
|
"e' obbligatoria prima del finalize"
|
|
)
|
|
# I CTE richiesti sono quelli del PIANO effettivo, non i file glob su disco: un
|
|
# ctes/*.sql orfano lasciato da un teardown incompleto non deve bloccare il finalize.
|
|
plan = effective_cte_plan(snapshot)
|
|
if plan:
|
|
try:
|
|
raw_tests = snapshot.artifacts.get("cte_tests", "")
|
|
tested = {r.name for r in map(CteTestRecord.model_validate, _iter_json_objects(raw_tests))}
|
|
except CteError as e:
|
|
typer.secho(
|
|
f"Finalize rifiutato: impossibile leggere cte_tests.json: {e}",
|
|
fg=typer.colors.RED, err=True,
|
|
)
|
|
raise typer.Exit(code=3)
|
|
for name in plan:
|
|
if name not in tested:
|
|
problems.append(f"CTE mai testato: {name}")
|
|
if problems:
|
|
typer.secho(f"Finalize rifiutato per {session_id}:", fg=typer.colors.YELLOW)
|
|
for p in problems:
|
|
typer.echo(f" - {p}")
|
|
raise typer.Exit(code=3)
|
|
|
|
# --- batteria di validazione su sql_final.sql ---
|
|
assert sql is not None
|
|
check = validate_sql(
|
|
sql,
|
|
physical=_load_physical_or_exit(cfg),
|
|
promoted_tables=promoted_tables_for(cfg, session_id),
|
|
forbidden_functions=set(cfg.execution.forbidden_functions),
|
|
)
|
|
if not check.ok:
|
|
typer.secho("Finalize rifiutato: sql_final.sql non passa la validazione statica:",
|
|
fg=typer.colors.RED, err=True)
|
|
for e in check.errors:
|
|
typer.echo(f" - {e}", err=True)
|
|
raise typer.Exit(code=3)
|
|
limit = cfg.execution.max_preview_rows
|
|
try:
|
|
plan = do_explain(cfg, sql)
|
|
result = do_run(cfg, sql, limit=limit)
|
|
except ExecutionError as e:
|
|
typer.secho(f"Finalize rifiutato: {e}", fg=typer.colors.RED, err=True)
|
|
raise typer.Exit(code=3)
|
|
|
|
# --- validation_report.md (preservando le note del reviewer) ---
|
|
existing_report = snapshot.artifacts.get("validation_report", "")
|
|
existing_notes = extract_reviewer_notes(existing_report) if existing_report else ""
|
|
report = render_validation_report(
|
|
session_id=session_id, check=check, plan=plan,
|
|
plan_warnings=plan_warnings(plan, cfg.execution),
|
|
result=result, runtime_warnings=runtime_warnings(result, cfg.execution),
|
|
static_warnings=static_warnings(check.ast), limit=limit,
|
|
reviewer_notes=existing_notes,
|
|
)
|
|
|
|
# --- evidence.json ---
|
|
linking = SchemaLinking.model_validate(
|
|
json.loads(snapshot.artifacts["schema_linking"])
|
|
)
|
|
entries = build_evidence_entries(decisions, linking, cfg.paths.artifacts / "evidence")
|
|
evidence = json.dumps(entries, ensure_ascii=False, indent=2)
|
|
|
|
# --- manifest + riepilogo ---
|
|
from tht.session.store import persist_verified_finalization
|
|
|
|
persist_verified_finalization(
|
|
repository, session_id, validation_report=report, evidence=evidence
|
|
)
|
|
# --- memoria attiva (parte B): indicizza la coppia domanda->SQL, best-effort ---
|
|
# Import lazy: memory_cmd importa da session_cmd (un import top-level qui sarebbe
|
|
# circolare). Qualunque errore (writer key assente, VPN giu', Ollama spento) NON
|
|
# deve bloccare il finalize: l'indice e' derivato e recuperabile con
|
|
# `tht memory solved-index <id>`.
|
|
try:
|
|
from tht.cli.memory_cmd import index_solved_session
|
|
|
|
if index_solved_session(cfg, session_id):
|
|
typer.secho(
|
|
"OK: coppia domanda->SQL indicizzata nel vectordb (solved_question).",
|
|
fg=typer.colors.GREEN,
|
|
)
|
|
else:
|
|
typer.secho(
|
|
"Coppia domanda->SQL gia' aggiornata nel vectordb (nessun upsert).",
|
|
fg=typer.colors.CYAN,
|
|
)
|
|
except Exception as e:
|
|
typer.secho(
|
|
f"ATTENZIONE: coppia domanda->SQL non indicizzata ({e}). "
|
|
f"Recupera con `tht memory solved-index {session_id}`.",
|
|
fg=typer.colors.YELLOW, err=True,
|
|
)
|
|
typer.secho(f"OK: sessione {session_id} finalizzata. Artefatti:", fg=typer.colors.GREEN)
|
|
for name in ARTIFACT_FILES:
|
|
state = "presente" if name not in {"session_manifest.yaml", "review_decisions.jsonl"} else "persistito"
|
|
typer.echo(f" - {name}: {state}")
|