Codex
d8a29bfbdd
Implement approved specification #32 and tickets #33-#37. Keep host authentication server-verified and pin session interaction language. Compile scoped base selectors for browser compatibility and retain full gutters during CSS pruning.
93 lines
3.5 KiB
Python
93 lines
3.5 KiB
Python
"""Storage contract for the durable, user-owned session workflow state."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from collections.abc import Sequence
|
|
from typing import Protocol
|
|
|
|
from tht.decisions import DecisionInput, DecisionRecord
|
|
from tht.session.models import PrincipalContext, SessionManifest, SessionSnapshot
|
|
from tht.session.store import SessionError
|
|
|
|
|
|
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 list(self) -> list[SessionSnapshot]: ...
|
|
|
|
def save_manifest(self, manifest: SessionManifest) -> SessionSnapshot: ...
|
|
|
|
def ensure_interaction_language(
|
|
self, session_id: str, workspace_language: str
|
|
) -> 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 delete_artifact(self, session_id: str, key: str) -> None: ...
|
|
|
|
def append_decisions(
|
|
self, session_id: str, decisions: Sequence[DecisionInput | dict]
|
|
) -> list[DecisionRecord]: ...
|
|
|
|
def finalize(self, manifest: SessionManifest, artifacts: dict[str, str]) -> SessionSnapshot: ...
|
|
|
|
def get_preferences(self) -> dict: ...
|
|
|
|
def set_preferences(self, preferences: dict) -> None: ...
|
|
|
|
def delete(self, session_id: str) -> None: ...
|
|
|
|
|
|
def resolve_principal(config) -> PrincipalContext:
|
|
"""Resolve the only principal source permitted for this CLI invocation.
|
|
|
|
The backend injects these values from its authenticated request context before
|
|
spawning ``tht``. A server-storage command without them must fail closed;
|
|
substituting a workstation identity would cross user ownership boundaries.
|
|
"""
|
|
if getattr(config, "session_storage", None) is None:
|
|
from tht.session.models import local_principal
|
|
|
|
return local_principal()
|
|
issuer = os.environ.get("THT_PRINCIPAL_ISSUER", "").strip()
|
|
subject = os.environ.get("THT_PRINCIPAL_SUBJECT", "").strip()
|
|
if not issuer or not subject:
|
|
raise SessionError(
|
|
"THT_PRINCIPAL_ISSUER e THT_PRINCIPAL_SUBJECT sono obbligatori per session storage server"
|
|
)
|
|
display_name = os.environ.get("THT_PRINCIPAL_DISPLAY_NAME", "").strip() or None
|
|
is_admin = os.environ.get("THT_PRINCIPAL_IS_ADMIN", "").strip().lower() in {"1", "true"}
|
|
return PrincipalContext(
|
|
issuer=issuer, subject=subject, display_name=display_name, is_admin=is_admin
|
|
)
|
|
|
|
|
|
def build_session_repository(
|
|
config, principal: PrincipalContext | None = None, *, home=None
|
|
) -> SessionRepository:
|
|
"""Build the configured private session persistence adapter."""
|
|
principal = principal or resolve_principal(config)
|
|
session_storage = getattr(config, "session_storage", None)
|
|
if session_storage is not None:
|
|
from tht.session.postgres_repository import PostgresSessionRepository
|
|
|
|
return PostgresSessionRepository.from_config(session_storage.connection, principal)
|
|
|
|
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,
|
|
root=None if home is not None else getattr(config.paths, "sessions", None),
|
|
)
|