from datetime import UTC, datetime from pathlib import Path from tht.config import DatabaseConfig from tht.session.models import SessionManifest from tht.textutil import slugify MANIFEST = "session_manifest.yaml" MAX_SLUG_CHARS = 40 class SessionError(Exception): pass def render_question_md(question: str, assumptions: list[str] | None = None) -> str: """Rende question.md in modo deterministico: domanda + assunzioni opzionali. Unica fonte di formattazione per question.md (riusata da create_session e set_question), così il gate non deve costruire markdown a mano. """ body = f"# Domanda\n\n{question.strip()}\n" items = [a.strip() for a in (assumptions or []) if a.strip()] if items: body += "\n## Assunzioni\n\n" + "".join(f"- {a}\n" for a in items) return body def _new_id(question: str, sessions_root: Path, stamp: str) -> str: base = f"{stamp}-{slugify(question)[:MAX_SLUG_CHARS].rstrip('-')}" candidate, n = base, 1 while (sessions_root / candidate).exists(): n += 1 candidate = f"{base}-{n}" return candidate def create_session( question: str, db: DatabaseConfig, sessions_root: Path ) -> SessionManifest: now = datetime.now(UTC) # stamp con ora/min/sec: identifica univocamente sessioni dello stesso giorno # sulla stessa domanda. Il contatore -n resta come rete per collisioni nello # stesso secondo. session_id = _new_id(question, sessions_root, now.strftime("%Y-%m-%d-%H%M%S")) manifest = SessionManifest( id=session_id, created_at=now, question=question, database=db.database, schema=db.db_schema, ) session_dir = sessions_root / session_id manifest.to_yaml(session_dir / MANIFEST) (session_dir / "question.md").write_text(render_question_md(question)) return manifest def set_question( session_id: str, question: str, assumptions: list[str], sessions_root: Path, ) -> Path: """Riscrive question.md (Fase 3) in modo deterministico, senza edit tool. Valida l'esistenza della sessione (SessionError se assente) e ritorna il path scritto. Non tocca il manifest né le decisioni. """ load_session(session_id, sessions_root) path = sessions_root / session_id / "question.md" path.write_text(render_question_md(question, assumptions)) return path def load_session(session_id: str, sessions_root: Path) -> SessionManifest: path = sessions_root / session_id / MANIFEST if not path.exists(): raise SessionError(f"Sessione non trovata: {session_id} (atteso {path})") return SessionManifest.from_yaml(path) def close_session(session_id: str, sessions_root: Path) -> SessionManifest: manifest = load_session(session_id, sessions_root) manifest.status = "closed" manifest.to_yaml(sessions_root / session_id / MANIFEST) return manifest