"""Path helpers for the local SAGE service.""" from __future__ import annotations import os import re import uuid from pathlib import Path REPO_ROOT = Path(__file__).resolve().parents[1] SRC_ROOT = REPO_ROOT / "src" def server_artifact_root() -> Path: """Return this worker's artifact root. A hosted dual-profile container runs two independent API processes. Each process must have its own store so identical run ids and history listings cannot cross the VA/MIMIC boundary. Relative overrides remain anchored to the repository, matching the service's other path behavior. """ configured = str(os.getenv("SAGE_SERVER_ARTIFACT_ROOT") or "").strip() if not configured: return REPO_ROOT / "artifacts" / "server_runs" candidate = Path(configured).expanduser() return candidate if candidate.is_absolute() else REPO_ROOT / candidate DEFAULT_SERVER_ARTIFACT_ROOT = server_artifact_root() RUN_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,79}$") def resolve_repo_path(path: str | Path) -> Path: candidate = Path(path) return candidate if candidate.is_absolute() else (REPO_ROOT / candidate).resolve() def make_service_run_id(case_id: str | None = None) -> str: prefix = slugify(case_id or "run") return f"{prefix}_{uuid.uuid4().hex[:10]}" def validate_run_id(run_id: str) -> str: if not RUN_ID_PATTERN.match(run_id): raise ValueError("run_id must contain only letters, numbers, underscore, dash, or dot and be at most 80 chars.") return run_id def slugify(value: str) -> str: slug = re.sub(r"[^A-Za-z0-9_.-]+", "_", value.strip()).strip("._-") return slug[:40] or "run" def safe_child(base: Path, relative_path: str | Path) -> Path: relative = Path(relative_path) if relative.is_absolute(): raise ValueError("Artifact path must be relative.") resolved = (base / relative).resolve() resolved.relative_to(base.resolve()) return resolved