src/eval_config.py

from __future__ import annotations

import json
import os
import secrets
from pathlib import Path
from typing import Any

from huggingface_hub import bucket_info

from .config import settings

_TRUE = {"1", "true", "yes", "on"}
_DEFAULT_EVAL_BUCKET_NAME = "agentic-space-factory-evals"
_CONFIG_FILENAME = ".asf_eval_config.json"


def _enabled_value(value: Any) -> bool:
    return str(value or "").strip().lower() in _TRUE or value is True


def _clean_source(value: str | None) -> str:
    return str(value or "").strip().strip("/")


def _clean_path(value: str | None, default: str = "evals") -> str:
    return str(value or default).strip().strip("/") or default


def _clean_mount(value: str | None, default: str = "/evals") -> str:
    mount = str(value or default).strip() or default
    if not mount.startswith("/"):
        mount = "/" + mount
    return mount.rstrip("/") or default


def default_eval_bucket_source(username: str | None) -> str:
    user = str(username or "").strip()
    return f"{user}/{_DEFAULT_EVAL_BUCKET_NAME}" if user else ""


def _allowed_admins() -> set[str]:
    return {item.strip() for item in os.getenv("ASF_EVAL_ALLOWED_ADMINS", "").split(",") if item.strip()}


def _can_manage_eval_config(username: str | None, cfg: dict[str, Any]) -> bool:
    user = str(username or "").strip()
    if not user or cfg.get("source") == "env":
        return False
    admins = _allowed_admins()
    if user in admins:
        return True
    created_by = str(cfg.get("created_by") or "").strip()
    if created_by and created_by == user:
        return True
    bucket_source = _clean_source(cfg.get("bucket_source"))
    owner = bucket_source.split("/", 1)[0] if "/" in bucket_source else ""
    return bool(owner and owner == user)


def eval_config_file_path(mount_path: str | None = None) -> Path:
    explicit = os.getenv("ASF_EVAL_CONFIG_PATH", "").strip()
    if explicit:
        return Path(explicit)
    return Path(_clean_mount(mount_path or settings.eval_bucket_mount)) / _CONFIG_FILENAME


def _env_eval_config() -> dict[str, Any] | None:
    if not (settings.eval_enabled and settings.eval_bucket_source):
        return None
    return {
        "enabled": True,
        "source": "env",
        "bucket_source": settings.eval_bucket_source,
        "bucket_path": settings.eval_bucket_path,
        "job_mount_path": settings.eval_bucket_mount,
        "salt": settings.eval_salt,
        "include_redacted_tails": bool(settings.eval_include_redacted_tails),
        "include_model_id": os.getenv("ASF_EVAL_INCLUDE_MODEL_ID", "").strip().lower() in _TRUE,
        "stores_generated_code": False,
        "stores_raw_prompts": False,
    }


def _disabled(reason: str = "not_configured") -> dict[str, Any]:
    return {
        "enabled": False,
        "source": "disabled",
        "reason": reason,
        "bucket_source": "",
        "bucket_path": "",
        "job_mount_path": _clean_mount(settings.eval_bucket_mount),
        "stores_generated_code": False,
        "stores_raw_prompts": False,
        "activation_supported": True,
        "activation_hint": "Create and mount a private operator eval bucket on the ASF Space, then click Enable eval archive.",
    }


def _normalize_config(data: dict[str, Any], *, source: str) -> dict[str, Any]:
    bucket_source = _clean_source(data.get("bucket_source"))
    if not (_enabled_value(data.get("enabled")) and bucket_source):
        return _disabled("config_disabled")
    return {
        "enabled": True,
        "source": source,
        "bucket_source": bucket_source,
        "bucket_path": _clean_path(data.get("bucket_path")),
        "job_mount_path": _clean_mount(data.get("job_mount_path") or data.get("mount_path")),
        "salt": str(data.get("salt") or ""),
        "include_redacted_tails": bool(data.get("include_redacted_tails")),
        "include_model_id": bool(data.get("include_model_id")),
        "stores_generated_code": False,
        "stores_raw_prompts": False,
        "activation_supported": True,
        "created_by": str(data.get("created_by") or ""),
    }


def effective_eval_config() -> dict[str, Any]:
    """Return env-controlled or UI-activated anonymous eval configuration.

    Environment variables intentionally win over the UI config so an operator can
    force-disable or pin a production setup from Space settings. When env vars
    are absent, a small config file in the mounted eval bucket can activate the
    archive without editing environment variables.
    """
    env_cfg = _env_eval_config()
    if env_cfg:
        return env_cfg
    cfg_path = eval_config_file_path()
    try:
        if cfg_path.exists():
            data = json.loads(cfg_path.read_text(encoding="utf-8") or "{}")
            if isinstance(data, dict):
                cfg = _normalize_config(data, source="ui")
                cfg["config_path"] = str(cfg_path)
                return cfg
    except Exception as exc:  # noqa: BLE001 - expose a compact status in UI
        out = _disabled("config_read_error")
        out["error"] = str(exc)[:500]
        out["config_path"] = str(cfg_path)
        return out
    out = _disabled("not_configured")
    out["config_path"] = str(cfg_path)
    return out


def public_eval_config(username: str | None = None) -> dict[str, Any]:
    cfg = effective_eval_config()
    public = {k: v for k, v in cfg.items() if k != "salt"}
    public["proposed_bucket_source"] = default_eval_bucket_source(username)
    public["default_bucket_path"] = "evals"
    public["default_mount_path"] = "/evals"
    public["env_override"] = cfg.get("source") == "env"
    public["can_manage"] = _can_manage_eval_config(username, cfg)
    public["readonly"] = bool(cfg.get("enabled")) and not public["can_manage"]
    if cfg.get("source") == "env":
        public["managed_by"] = "environment"
    elif public["readonly"]:
        public["managed_by"] = "instance owner"
    elif public["can_manage"]:
        public["managed_by"] = "you"
    else:
        public["managed_by"] = "not configured"
    return public


def activate_eval_archive_config(
    *,
    username: str,
    token: str,
    bucket_source: str | None = None,
    bucket_path: str = "evals",
    mount_path: str = "/evals",
    include_redacted_tails: bool = False,
    include_model_id: bool = False,
) -> dict[str, Any]:
    """Persist UI activation in the mounted eval bucket.

    The direct activation path is for an operator who has already created and
    mounted the eval bucket in the Space settings. It verifies bucket access with
    the signed-in HF token and writes a config file into the mount. It does not
    store generated code, raw prompts, or tokens.
    """
    if _env_eval_config():
        return {**public_eval_config(username), "message": "Eval archive is controlled by environment variables."}
    user = str(username or "").strip()
    if not user:
        raise ValueError("Sign in before enabling the eval archive.")
    source = _clean_source(bucket_source) or default_eval_bucket_source(user)
    owner = source.split("/", 1)[0] if "/" in source else ""
    allowed_admins = _allowed_admins()
    if owner and owner != user and user not in allowed_admins:
        raise PermissionError("You can only activate an eval bucket in your namespace unless ASF_EVAL_ALLOWED_ADMINS allows you.")

    # Verify the bucket exists and the signed-in user can access it.
    bucket_info(source, token=token)

    mount = _clean_mount(mount_path)
    mount_dir = Path(mount)
    if not mount_dir.exists() or not mount_dir.is_dir():
        raise FileNotFoundError(
            f"Eval bucket mount not found at {mount}. Mount {source}/{_clean_path(bucket_path)} to {mount} in the Space settings first."
        )
    if not os.access(mount_dir, os.W_OK):
        raise PermissionError(f"Eval bucket mount is not writable at {mount}.")

    cfg_path = eval_config_file_path(mount)
    existing_salt = ""
    try:
        if cfg_path.exists():
            existing = json.loads(cfg_path.read_text(encoding="utf-8") or "{}")
            if isinstance(existing, dict):
                existing_salt = str(existing.get("salt") or "")
    except Exception:
        existing_salt = ""
    config = {
        "enabled": True,
        "source": "ui",
        "bucket_source": source,
        "bucket_path": _clean_path(bucket_path),
        "job_mount_path": mount,
        "salt": existing_salt or secrets.token_hex(32),
        "include_redacted_tails": bool(include_redacted_tails),
        "include_model_id": bool(include_model_id),
        "stores_generated_code": False,
        "stores_raw_prompts": False,
        "created_by": user,
    }
    cfg_path.write_text(json.dumps(config, indent=2, sort_keys=True) + "\n", encoding="utf-8")

    # Small write probe so activation fails loudly if the mounted bucket is not really writable.
    probe = mount_dir / ".asf_eval_write_probe"
    probe.write_text("ok\n", encoding="utf-8")
    try:
        probe.unlink()
    except Exception:
        pass
    public = public_eval_config(username)
    public["message"] = f"Eval archive enabled for {source}/{_clean_path(bucket_path)} → {mount}."
    return public


def disable_eval_archive_config(*, username: str) -> dict[str, Any]:
    """Disable UI-managed eval archive config. Env-controlled configs are read-only."""
    cfg = effective_eval_config()
    if cfg.get("source") == "env":
        raise PermissionError("Eval archive is controlled by environment variables.")
    if not _can_manage_eval_config(username, cfg):
        raise PermissionError("Only the instance owner or configured eval admins can disable the eval archive.")
    cfg_path = eval_config_file_path(cfg.get("job_mount_path"))
    previous_salt = str(cfg.get("salt") or "")
    disabled = {
        "enabled": False,
        "source": "ui",
        "bucket_source": _clean_source(cfg.get("bucket_source")),
        "bucket_path": _clean_path(cfg.get("bucket_path")),
        "job_mount_path": _clean_mount(cfg.get("job_mount_path")),
        "salt": previous_salt,
        "stores_generated_code": False,
        "stores_raw_prompts": False,
        "disabled_by": str(username or ""),
    }
    cfg_path.write_text(json.dumps(disabled, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    public = public_eval_config(username)
    public["message"] = "Eval archive disabled for this ASF instance."
    return public