ThothII/harness/tht/session/store.py

import os
import shutil
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
MAX_SUMMARY_CHARS = 120


class SessionError(Exception):
    pass


def current_author() -> str:
    """Autore della sessione (D6). Nel profilo `none` dell'MVP l'operatore e' sulla
    propria macchina: default `dev@local`. Con auth reale il backend passa l'id via
    THT_AUTHOR. Una sola fonte per author/updated_by."""
    return os.environ.get("THT_AUTHOR", "dev@local").strip() or "dev@local"


def _summarize(question: str) -> str:
    """Domanda sintetica per il manifest/la sidebar (D6 §5.2): prima riga, troncata."""
    first = question.strip().splitlines()[0].strip() if question.strip() else ""
    return first[:MAX_SUMMARY_CHARS]


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,
    *, author: str | None = None, summary: str | None = None,
    provider: str | None = None, model: str | None = None,
    thinking: str | None = None, name: str | None = None,
) -> 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"))
    who = author or current_author()
    # schema_version del workflow usato (D6/§5.3): consente di interpretare il ledger
    # secondo la versione anche se il workflow evolve.
    try:
        from tht.workflow import load_workflow

        schema_version = load_workflow().schema_version
    except Exception:
        schema_version = None
    manifest = SessionManifest(
        id=session_id, created_at=now, question=question,
        database=db.database, schema=db.db_schema,
        author=who, summary=summary or _summarize(question),
        updated_at=now, updated_by=who, schema_version=schema_version,
        provider=provider, model=model, thinking=thinking, name=name,
    )
    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 touch_manifest(
    session_id: str, sessions_root: Path, *, updated_by: str | None = None
) -> SessionManifest:
    """Aggiorna updated_at/updated_by del manifest (D6 §5.2). Da chiamare ad ogni
    mutazione della sessione (set-question, finalize, close)."""
    manifest = load_session(session_id, sessions_root)
    manifest.updated_at = datetime.now(UTC)
    manifest.updated_by = updated_by or current_author()
    manifest.to_yaml(sessions_root / session_id / MANIFEST)
    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))
    touch_manifest(session_id, sessions_root)
    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.updated_at = datetime.now(UTC)
    manifest.updated_by = current_author()
    manifest.to_yaml(sessions_root / session_id / MANIFEST)
    return manifest


def _save_touched(manifest: SessionManifest, sessions_root: Path) -> SessionManifest:
    """Persist `manifest` updating updated_at/updated_by (single save path for mutations)."""
    manifest.updated_at = datetime.now(UTC)
    manifest.updated_by = current_author()
    manifest.to_yaml(sessions_root / manifest.id / MANIFEST)
    return manifest


def set_name(session_id: str, name: str | None, sessions_root: Path) -> SessionManifest:
    """Set the descriptive name (empty/blank clears it back to None)."""
    manifest = load_session(session_id, sessions_root)
    manifest.name = (name or "").strip() or None
    return _save_touched(manifest, sessions_root)


def set_group(session_id: str, group: str | None, sessions_root: Path) -> SessionManifest:
    """Set the group (empty/blank clears it back to None)."""
    manifest = load_session(session_id, sessions_root)
    manifest.group = (group or "").strip() or None
    return _save_touched(manifest, sessions_root)


def set_archived(session_id: str, archived: bool, sessions_root: Path) -> SessionManifest:
    """Flip the archived flag. Archiving does NOT change resumability (a finalized
    session stays read-only); unarchive only moves it back to the active list."""
    manifest = load_session(session_id, sessions_root)
    manifest.archived = archived
    return _save_touched(manifest, sessions_root)


def delete_session(session_id: str, sessions_root: Path) -> None:
    """Hard-delete the session directory. SessionError if it does not exist."""
    load_session(session_id, sessions_root)  # raises SessionError if absent
    shutil.rmtree(sessions_root / session_id)