Codex
82e2c91f42
Publish documentation / publish (push) Successful in 1m27s
Add PostgreSQL-backed memory, editable evidence with source review and activation, and human-approved archive repairs across the harness, API, and UI. Include migrations, deployment support, regression coverage, and validation documentation. Refresh permissions from validated session roles so existing administrator logins can access newly deployed archive management features.
257 lines
10 KiB
Python
257 lines
10 KiB
Python
"""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 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)
|