src/config.py

from __future__ import annotations

import os
from dataclasses import dataclass

from .version import ASF_APP_VERSION, resolve_app_version


@dataclass(frozen=True)
class Settings:
    """Runtime configuration for the orchestrator Space."""

    bucket_name: str = os.getenv("SPACE_FACTORY_BUCKET_NAME", "space-factory-runs")
    # Container root used by workers. Jobs mount the bucket sub-prefix `runs/`
    # at `<bucket_mount>/runs` so worker code writes to `<bucket_mount>/runs/<run_id>`.
    bucket_mount: str = os.getenv("SPACE_FACTORY_BUCKET_MOUNT", "/output")
    bucket_runs_prefix: str = os.getenv("SPACE_FACTORY_BUCKET_RUNS_PREFIX", "runs").strip().strip("/") or "runs"
    job_flavor: str = os.getenv("SPACE_FACTORY_JOB_FLAVOR", "cpu-upgrade")
    job_timeout: str = os.getenv("SPACE_FACTORY_JOB_TIMEOUT", "30m")
    job_image: str = os.getenv("SPACE_FACTORY_JOB_IMAGE", "python:3.12")

    # Optional cross-user anonymous evaluation archive. This is intended for an
    # operator-owned bucket mounted into Jobs, not for user-visible run
    # artifacts. Example: ASF_EVAL_BUCKET_SOURCE=huggingface/asf-evals,
    # ASF_EVAL_BUCKET_PATH=evals, ASF_EVAL_BUCKET_MOUNT=/evals.
    eval_enabled: bool = os.getenv("ASF_EVAL_ENABLED", "false").strip().lower() in {"1", "true", "yes", "on"}
    eval_bucket_source: str = os.getenv("ASF_EVAL_BUCKET_SOURCE", "").strip().strip("/")
    eval_bucket_path: str = os.getenv("ASF_EVAL_BUCKET_PATH", "evals").strip().strip("/") or "evals"
    eval_bucket_mount: str = os.getenv("ASF_EVAL_BUCKET_MOUNT", "/evals").strip() or "/evals"
    eval_salt: str = os.getenv("ASF_EVAL_SALT", "").strip()
    eval_include_redacted_tails: bool = os.getenv("ASF_EVAL_INCLUDE_REDACTED_TAILS", "false").strip().lower() in {"1", "true", "yes", "on"}
    app_version: str = resolve_app_version(os.getenv("ASF_VERSION", ASF_APP_VERSION))


def normalize_bucket_name(value: str | None) -> str:
    """Return a safe bucket name, defaulting to the product bucket name."""
    bucket = (value or settings.bucket_name).strip().strip("/")
    if not bucket:
        bucket = settings.bucket_name
    if "/" in bucket:
        raise ValueError("Bucket name must be a name like 'space-factory-runs', not owner/name.")
    return bucket


def user_bucket_source(*, username: str, bucket_name: str | None = None) -> str:
    """Return the per-user bucket source expected by HF Jobs volumes."""
    clean_user = (username or "").strip()
    if not clean_user:
        raise ValueError("Missing username for per-user bucket source.")
    return f"{clean_user}/{normalize_bucket_name(bucket_name)}"


def bucket_uri_from_source(bucket_source: str) -> str:
    return f"hf://buckets/{bucket_source}"


settings = Settings()