ThothII/harness/tht/session/filesystem_repository.py

"""Private local filesystem implementation of the session repository contract."""

from __future__ import annotations

import hashlib
import json
import os
import re
import shutil
import tempfile
import uuid
from collections.abc import Sequence
from pathlib import Path

import portalocker
import yaml

from tht.decisions import DecisionInput, DecisionRecord, append_decisions, list_decisions
from tht.session.models import PrincipalContext, SessionManifest, SessionSnapshot
from tht.session.store import MANIFEST, SessionError

_ARTIFACT_FILES = {
    "question": "question.md",
    "schema_linking": "schema_linking.json",
    "evidence": "evidence.json",
    "evidence_receipts": "evidence_receipts.json",
    "sql_final": "sql_final.sql",
    "validation_report": "validation_report.md",
    "retrieval_pack": "retrieval_pack.md",
    "cte_tests": "cte_tests.json",
    "cte_plan": "cte_plan.json",
    "cte_plan_doc": "cte_plan_doc.json",
    "memory_proposals": "memory_proposals.json",
}
_ARTIFACT_KEYS = {filename: key for key, filename in _ARTIFACT_FILES.items()}
_SAFE_CTE_NAME = re.compile(r"[A-Za-z0-9_-]+\Z")
_LEGACY_SESSION_ID = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}\Z")


class FilesystemSessionRepository:
    """Readable phase documents under one private local user's ThothII home."""

    def __init__(
        self, home: Path, workspace: str, principal: PrincipalContext, *, root: Path | None = None
    ):
        self.home = Path(home).expanduser()
        self.workspace = workspace
        self.principal = principal
        self.root = Path(root) if root is not None else self.home / "workspaces" / workspace / "sessions"

    @property
    def preferences_path(self) -> Path:
        digest = hashlib.sha256(
            f"{self.principal.issuer}\0{self.principal.subject}".encode()
        ).hexdigest()
        return self.home / "principals" / digest / "preferences.json"

    def create(self, manifest: SessionManifest) -> SessionSnapshot:
        # New records are opaque UUIDv4 only. Historical timestamp ids remain
        # readable/mutable through _session_dir but are never created here.
        self._require_uuid4(manifest.id)
        session_dir = self.root / manifest.id
        with self._lock(session_dir):
            if (session_dir / MANIFEST).exists():
                raise SessionError(f"Sessione esiste gia': {manifest.id}")
            self._private_directory(session_dir)
            self._write_manifest(session_dir, manifest)
        return self.get(manifest.id)

    def get(self, session_id: str) -> SessionSnapshot:
        session_dir = self._session_dir(session_id)
        manifest_path = session_dir / MANIFEST
        if not manifest_path.exists():
            raise SessionError(f"Sessione non trovata: {session_id} (atteso {manifest_path})")
        manifest = SessionManifest.from_yaml(manifest_path)
        artifacts = self._read_artifacts(session_dir)
        return SessionSnapshot(
            principal=self.principal,
            manifest=manifest,
            artifacts=artifacts,
            decisions=list_decisions(session_dir),
        )

    def list(self) -> list[SessionSnapshot]:
        if not self.root.exists():
            return []
        snapshots = []
        for path in self.root.iterdir():
            if path.is_dir() and (path / MANIFEST).exists():
                try:
                    snapshots.append(self.get(path.name))
                except SessionError:
                    continue
        return sorted(snapshots, key=lambda item: item.manifest.created_at, reverse=True)

    def save_manifest(self, manifest: SessionManifest) -> SessionSnapshot:
        session_dir = self._session_dir(manifest.id)
        with self._lock(session_dir):
            self._require_existing(session_dir, manifest.id)
            self._write_manifest(session_dir, manifest)
        return self.get(manifest.id)

    def read_artifact(self, session_id: str, key: str) -> str | None:
        session_dir = self._session_dir(session_id)
        self._require_existing(session_dir, session_id)
        path = self._artifact_path(session_dir, key)
        return path.read_text() if path.exists() else None

    def ensure_interaction_language(
        self, session_id: str, workspace_language: str
    ) -> SessionSnapshot:
        from tht.session.store import pin_interaction_language

        session_dir = self._session_dir(session_id)
        with self._lock(session_dir):
            self._require_existing(session_dir, session_id)
            manifest = SessionManifest.from_yaml(session_dir / MANIFEST)
            if pin_interaction_language(manifest, workspace_language):
                self._write_manifest(session_dir, manifest)
        return self.get(session_id)

    def write_artifact(self, session_id: str, key: str, content: str) -> None:
        session_dir = self._session_dir(session_id)
        with self._lock(session_dir):
            self._require_existing(session_dir, session_id)
            self._write_private_text(self._artifact_path(session_dir, key), content)

    def delete_artifact(self, session_id: str, key: str) -> None:
        session_dir = self._session_dir(session_id)
        with self._lock(session_dir):
            self._require_existing(session_dir, session_id)
            self._artifact_path(session_dir, key).unlink(missing_ok=True)

    def append_decisions(
        self, session_id: str, decisions: Sequence[DecisionInput | dict]
    ) -> list[DecisionRecord]:
        session_dir = self._session_dir(session_id)
        with self._lock(session_dir):
            self._require_existing(session_dir, session_id)
            # The legacy ledger helper remains the canonical record format. Its
            # separate lock also keeps direct workflow callers safe during Task 3.
            return append_decisions(session_dir, list(decisions))

    def finalize(self, manifest: SessionManifest, artifacts: dict[str, str]) -> SessionSnapshot:
        """Commit verified final artifacts before publishing finalized status.

        The finalized manifest is the filesystem commit marker: a crash can leave
        an open session with already-rendered artifacts, but never a finalized
        session without its report and evidence.
        """
        session_dir = self._session_dir(manifest.id)
        with self._lock(session_dir):
            self._require_existing(session_dir, manifest.id)
            for key, content in artifacts.items():
                self._write_private_text(self._artifact_path(session_dir, key), content)
            self._write_manifest(session_dir, manifest)
        return self.get(manifest.id)

    def get_preferences(self) -> dict:
        path = self.preferences_path
        if not path.exists():
            return {}
        data = json.loads(path.read_text())
        if not isinstance(data, dict):
            raise TypeError(f"Invalid preferences: {path}")
        return data

    def set_preferences(self, preferences: dict) -> None:
        if not isinstance(preferences, dict):
            raise TypeError("preferences must be a dictionary")
        path = self.preferences_path
        self._private_directory(path.parent)
        with portalocker.Lock(path.with_suffix(".lock"), mode="a+", timeout=10):
            self._write_private_text(path, json.dumps(preferences, ensure_ascii=False, sort_keys=True) + "\n")

    def delete(self, session_id: str) -> None:
        session_dir = self._session_dir(session_id)
        with self._lock(session_dir):
            self._require_existing(session_dir, session_id)
            shutil.rmtree(session_dir)

    def _session_dir(self, session_id: str) -> Path:
        self._require_session_id(session_id)
        return self.root / session_id

    @classmethod
    def _require_session_id(cls, session_id: str) -> None:
        try:
            cls._require_uuid4(session_id)
        except SessionError:
            if session_id not in {".", ".."} and _LEGACY_SESSION_ID.fullmatch(session_id):
                return
            raise

    @staticmethod
    def _require_uuid4(session_id: str) -> None:
        try:
            parsed = uuid.UUID(session_id, version=4)
        except ValueError as exc:
            raise SessionError(f"ID sessione non UUIDv4: {session_id}") from exc
        if str(parsed) != session_id or parsed.version != 4:
            raise SessionError(f"ID sessione non UUIDv4: {session_id}")

    @staticmethod
    def _private_directory(path: Path) -> None:
        path.mkdir(parents=True, exist_ok=True)
        if os.name != "nt":
            path.chmod(0o700)

    def _lock(self, session_dir: Path):
        self._private_directory(session_dir)
        return portalocker.Lock(session_dir / ".repository.lock", mode="a+", timeout=10)

    @staticmethod
    def _require_existing(session_dir: Path, session_id: str) -> None:
        if not (session_dir / MANIFEST).exists():
            raise SessionError(f"Sessione non trovata: {session_id}")

    def _artifact_path(self, session_dir: Path, key: str) -> Path:
        if key in _ARTIFACT_FILES:
            return session_dir / _ARTIFACT_FILES[key]
        if key.startswith("cte_sql:"):
            name = key.removeprefix("cte_sql:")
            if _SAFE_CTE_NAME.fullmatch(name):
                return session_dir / "ctes" / f"{name}.sql"
        raise ValueError(f"Unsupported artifact key: {key}")

    def _read_artifacts(self, session_dir: Path) -> dict[str, str]:
        artifacts = {
            key: (session_dir / filename).read_text()
            for filename, key in _ARTIFACT_KEYS.items()
            if (session_dir / filename).is_file()
        }
        ctes = session_dir / "ctes"
        if ctes.is_dir():
            artifacts.update(
                {
                    f"cte_sql:{path.stem}": path.read_text()
                    for path in ctes.glob("*.sql")
                    if _SAFE_CTE_NAME.fullmatch(path.stem)
                }
            )
        return artifacts

    def _write_manifest(self, session_dir: Path, manifest: SessionManifest) -> None:
        content = yaml.safe_dump(
            manifest.model_dump(by_alias=True, mode="json", exclude_defaults=False),
            sort_keys=False,
            allow_unicode=True,
            width=120,
        )
        self._write_private_text(session_dir / MANIFEST, content)

    def _write_private_text(self, path: Path, content: str) -> None:
        self._private_directory(path.parent)
        fd, temporary_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
        temporary = Path(temporary_name)
        try:
            with os.fdopen(fd, "w") as handle:
                handle.write(content)
                handle.flush()
                os.fsync(handle.fileno())
            if os.name != "nt":
                temporary.chmod(0o600)
            os.replace(temporary, path)
            if os.name != "nt":
                path.chmod(0o600)
        finally:
            temporary.unlink(missing_ok=True)