"""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 pathlib import Path from typing import Sequence 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", "sql_final": "sql_final.sql", "validation_report": "validation_report.md", "retrieval_pack": "retrieval_pack.md", "cte_tests": "cte_tests.json", } _ARTIFACT_KEYS = {filename: key for key, filename in _ARTIFACT_FILES.items()} _SAFE_CTE_NAME = re.compile(r"[A-Za-z0-9_-]+\Z") class FilesystemSessionRepository: """Readable phase documents under one private local user's ThothII home.""" def __init__(self, home: Path, workspace: str, principal: PrincipalContext): self.home = Path(home).expanduser() self.workspace = workspace self.principal = principal self.root = 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: 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 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 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 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 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 ValueError(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_uuid4(session_id) return self.root / session_id @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)