From 8f364ec56450530872cf87b5fd9b34168d4199c4 Mon Sep 17 00:00:00 2001 From: User Date: Thu, 16 Jul 2026 17:11:22 +0200 Subject: [PATCH] feat(harness): add local session repository --- harness/pyproject.toml | 1 + harness/tests/test_local_identity.py | 17 ++ harness/tests/test_session_repository.py | 99 +++++++++ harness/tht/config.py | 9 +- harness/tht/session/filesystem_repository.py | 204 +++++++++++++++++++ harness/tht/session/models.py | 84 ++++++++ harness/tht/session/repository.py | 43 ++++ harness/tht/session/store.py | 15 +- 8 files changed, 464 insertions(+), 8 deletions(-) create mode 100644 harness/tests/test_local_identity.py create mode 100644 harness/tests/test_session_repository.py create mode 100644 harness/tht/session/filesystem_repository.py create mode 100644 harness/tht/session/repository.py diff --git a/harness/pyproject.toml b/harness/pyproject.toml index 5972c288..30166fdb 100644 --- a/harness/pyproject.toml +++ b/harness/pyproject.toml @@ -16,6 +16,7 @@ dependencies = [ "requests>=2.31", "tqdm>=4.66", "yake>=0.4", + "portalocker>=2.10", ] [project.scripts] diff --git a/harness/tests/test_local_identity.py b/harness/tests/test_local_identity.py new file mode 100644 index 00000000..2ae0ae5b --- /dev/null +++ b/harness/tests/test_local_identity.py @@ -0,0 +1,17 @@ +import json +import uuid + +from tht.session.models import local_principal + + +def test_local_identity_uuid_is_stable_across_restarts(tmp_path): + first = local_principal(tmp_path) + restarted = local_principal(tmp_path) + + assert first == restarted + assert first.issuer == "local" + assert uuid.UUID(first.subject, version=4).version == 4 + assert json.loads((tmp_path / "identity.json").read_text()) == { + "issuer": "local", + "subject": first.subject, + } diff --git a/harness/tests/test_session_repository.py b/harness/tests/test_session_repository.py new file mode 100644 index 00000000..b9d3ea95 --- /dev/null +++ b/harness/tests/test_session_repository.py @@ -0,0 +1,99 @@ +import uuid + +from tht.config import DatabaseConfig, load_config +from tht.decisions import DecisionInput +from tht.session.filesystem_repository import FilesystemSessionRepository +from tht.session.models import PrincipalContext, SessionManifest +from tht.session.repository import build_session_repository +from tht.session.store import create_session + + +def _db() -> DatabaseConfig: + return DatabaseConfig(database="testdb", schema="public", user="u", password="p") # noqa: S106 + + +def _config(tmp_path): + path = tmp_path / "workspace.yaml" + path.write_text( + "dwh:\n" + " type: postgres_direct\n" + " connection:\n" + " database: testdb\n" + " schema: public\n" + " user: user\n" + " password: secret\n" + ) + return load_config(path) + + +def _manifest(session_id: str) -> SessionManifest: + return SessionManifest( + id=session_id, + created_at="2026-07-16T10:00:00Z", + question="Which patients had an ablation?", + database="testdb", + schema="public", + ) + + +def test_new_sessions_use_uuid4_ids(tmp_path): + manifest = create_session("Which patients had an ablation?", _db(), tmp_path) + + session_uuid = uuid.UUID(manifest.id, version=4) + assert str(session_uuid) == manifest.id + assert session_uuid.version == 4 + + +def test_filesystem_repository_scopes_sessions_to_local_principal_root(tmp_path): + config = _config(tmp_path) + principal = PrincipalContext(issuer="local", subject="alice") + repository = build_session_repository(config, principal, home=tmp_path / "alice-home") + session_id = str(uuid.uuid4()) + + repository.create(_manifest(session_id)) + + assert repository.root == tmp_path / "alice-home/workspaces/workspace/sessions" + assert (repository.root / session_id / "session_manifest.yaml").exists() + + +def test_filesystem_repository_reads_phase_artifacts_and_appends_decisions(tmp_path): + repository = FilesystemSessionRepository( + tmp_path / "home", "demo", PrincipalContext(issuer="local", subject="alice") + ) + session_id = str(uuid.uuid4()) + repository.create(_manifest(session_id)) + + repository.write_artifact(session_id, "sql_final", "SELECT 1") + decisions = repository.append_decisions( + session_id, + [DecisionInput(type="sql_approved", subject="phase:7", detail="SELECT 1")], + ) + snapshot = repository.get(session_id) + + assert repository.read_artifact(session_id, "sql_final") == "SELECT 1" + assert snapshot.artifacts["sql_final"] == "SELECT 1" + assert [decision.type for decision in snapshot.decisions] == ["sql_approved"] + assert decisions[0].seq == 1 + + +def test_tht_home_overrides_local_repository_root(monkeypatch, tmp_path): + monkeypatch.setenv("THT_HOME", str(tmp_path / "override")) + config = _config(tmp_path) + + repository = build_session_repository( + config, PrincipalContext(issuer="local", subject="alice") + ) + + assert config.paths.sessions == tmp_path / "override/workspaces/workspace/sessions" + assert repository.root == tmp_path / "override/workspaces/workspace/sessions" + + +def test_filesystem_repository_keeps_preferences_per_principal(tmp_path): + home = tmp_path / "home" + alice = FilesystemSessionRepository(home, "demo", PrincipalContext(issuer="local", subject="alice")) + bob = FilesystemSessionRepository(home, "demo", PrincipalContext(issuer="local", subject="bob")) + + alice.set_preferences({"model": "glm"}) + + assert alice.get_preferences() == {"model": "glm"} + assert bob.get_preferences() == {} diff --git a/harness/tht/config.py b/harness/tht/config.py index 11644e5d..62603506 100644 --- a/harness/tht/config.py +++ b/harness/tht/config.py @@ -16,6 +16,11 @@ class ConfigError(Exception): """Errore di configurazione, con messaggio leggibile per l'utente.""" +def local_tht_home() -> Path: + """Return the private local ThothII home, honoring the explicit override.""" + return Path(os.environ.get("THT_HOME", "~/.thothii")).expanduser() + + def _expand_env(value: Any) -> Any: if isinstance(value, str): @@ -356,7 +361,9 @@ def load_config(path: Path) -> Config: raise ConfigError( f"transport: rest richiede la sezione `rest` (base_url, api_key) in {path}." ) - data_root = os.environ.get("THT_DATA_ROOT") + # THT_HOME is the local-user storage root. Retain THT_DATA_ROOT as the + # portable-deployment compatibility name until all callers use repositories. + data_root = os.environ.get("THT_HOME") or os.environ.get("THT_DATA_ROOT") if data_root: # Import locally: paths owns resolution, while ConfigError remains the public # configuration exception callers already handle. diff --git a/harness/tht/session/filesystem_repository.py b/harness/tht/session/filesystem_repository.py new file mode 100644 index 00000000..e4feb17f --- /dev/null +++ b/harness/tht/session/filesystem_repository.py @@ -0,0 +1,204 @@ +"""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) diff --git a/harness/tht/session/models.py b/harness/tht/session/models.py index 4d5bcae7..011310a6 100644 --- a/harness/tht/session/models.py +++ b/harness/tht/session/models.py @@ -1,10 +1,17 @@ +import json +import os +import tempfile +import uuid from datetime import datetime from pathlib import Path from typing import Literal, Self +import portalocker import yaml from pydantic import BaseModel, Field, ConfigDict +from tht.decisions import DecisionRecord + # Stub locale di _YamlModel. In the reference implementation questa base vive in # mschema/models.py; qui la si replica perche' SessionManifest ha bisogno di @@ -26,6 +33,74 @@ class _YamlModel(BaseModel): return cls.model_validate(raw) +class PrincipalContext(BaseModel): + """Trusted owner identity supplied by the runtime, never by a session document.""" + + issuer: str + subject: str + display_name: str | None = None + is_admin: bool = False + + model_config = ConfigDict(frozen=True) + + +def _private_directory(path: Path) -> Path: + path.mkdir(parents=True, exist_ok=True) + if os.name != "nt": + path.chmod(0o700) + return path + + +def _write_private_json(path: Path, data: dict[str, str]) -> None: + fd, temporary_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) + temporary = Path(temporary_name) + try: + with os.fdopen(fd, "w") as handle: + json.dump(data, handle, sort_keys=True) + handle.write("\n") + 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) + + +def local_principal(home: Path | None = None) -> PrincipalContext: + """Load or create the stable UUID identity for a local ThothII home.""" + if home is None: + from tht.config import local_tht_home + + home = local_tht_home() + identity_path = _private_directory(home) / "identity.json" + lock_path = identity_path.with_suffix(".lock") + with portalocker.Lock(lock_path, mode="a+", timeout=10): + if identity_path.exists(): + try: + identity = json.loads(identity_path.read_text()) + principal = PrincipalContext.model_validate(identity) + except (json.JSONDecodeError, OSError, ValueError) as exc: + raise ValueError(f"Invalid local identity: {identity_path}") from exc + if principal.issuer != "local": + raise ValueError(f"Invalid local identity issuer: {identity_path}") + try: + parsed = uuid.UUID(principal.subject, version=4) + except ValueError as exc: + raise ValueError(f"Invalid local identity subject: {identity_path}") from exc + if str(parsed) != principal.subject or parsed.version != 4: + raise ValueError(f"Invalid local identity subject: {identity_path}") + return principal + + principal = PrincipalContext(issuer="local", subject=str(uuid.uuid4())) + _write_private_json( + identity_path, {"issuer": principal.issuer, "subject": principal.subject} + ) + return principal + + class SessionManifest(_YamlModel): id: str created_at: datetime @@ -47,6 +122,15 @@ class SessionManifest(_YamlModel): group: str | None = None +class SessionSnapshot(BaseModel): + """The current persisted session state, excluding the non-persistent chat stream.""" + + principal: PrincipalContext | None = None + manifest: SessionManifest + artifacts: dict[str, str] = Field(default_factory=dict) + decisions: list[DecisionRecord] = Field(default_factory=list) + + class Candidate(BaseModel): kind: Literal["table", "column"] name: str diff --git a/harness/tht/session/repository.py b/harness/tht/session/repository.py new file mode 100644 index 00000000..df8a5925 --- /dev/null +++ b/harness/tht/session/repository.py @@ -0,0 +1,43 @@ +"""Storage contract for the durable, user-owned session workflow state.""" + +from __future__ import annotations + +from typing import Protocol, Sequence + +from tht.decisions import DecisionInput, DecisionRecord +from tht.session.models import PrincipalContext, SessionManifest, SessionSnapshot + + +class SessionRepository(Protocol): + """Persistence boundary shared by local files and the server database adapter.""" + + principal: PrincipalContext + + def create(self, manifest: SessionManifest) -> SessionSnapshot: ... + + def get(self, session_id: str) -> SessionSnapshot: ... + + def save_manifest(self, manifest: SessionManifest) -> SessionSnapshot: ... + + def read_artifact(self, session_id: str, key: str) -> str | None: ... + + def write_artifact(self, session_id: str, key: str, content: str) -> None: ... + + def append_decisions( + self, session_id: str, decisions: Sequence[DecisionInput | dict] + ) -> list[DecisionRecord]: ... + + def get_preferences(self) -> dict: ... + + def set_preferences(self, preferences: dict) -> None: ... + + def delete(self, session_id: str) -> None: ... + + +def build_session_repository(config, principal: PrincipalContext, *, home=None) -> SessionRepository: + """Build the local adapter for Task 1; server selection arrives with Task 2.""" + from tht.config import local_tht_home + from tht.session.filesystem_repository import FilesystemSessionRepository + + workspace = getattr(config, "_workspace_id", "default") + return FilesystemSessionRepository(home or local_tht_home(), workspace, principal) diff --git a/harness/tht/session/store.py b/harness/tht/session/store.py index 8eefef2b..b8e2c834 100644 --- a/harness/tht/session/store.py +++ b/harness/tht/session/store.py @@ -1,6 +1,7 @@ import json import os import shutil +import uuid from datetime import UTC, datetime from pathlib import Path @@ -8,7 +9,6 @@ import yake from tht.config import DatabaseConfig from tht.session.models import SessionManifest -from tht.textutil import slugify MANIFEST = "session_manifest.yaml" MAX_SLUG_CHARS = 40 @@ -88,12 +88,13 @@ def render_question_md(question: str, assumptions: list[str] | None = None) -> s 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 + """Create an opaque UUIDv4 session identifier. + + ``question``, ``sessions_root`` and ``stamp`` remain accepted temporarily so + existing workflow callers do not need to change as the repository boundary + is introduced. + """ + return str(uuid.uuid4()) def create_session(