server/artifacts.py

"""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