from __future__ import annotations import json import os import secrets import shutil 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 _can_flush_eval_records(username: str | None, cfg: dict[str, Any]) -> bool: """Return whether a signed-in user may delete archived eval records. UI-managed configs use the normal manage permission. Env-managed configs are usually read-only, but the bucket owner or ASF_EVAL_ALLOWED_ADMINS may still flush records from the mounted archive. The config file itself is preserved. """ user = str(username or "").strip() if not user or not cfg.get("enabled"): return False admins = _allowed_admins() if user in admins: return True if _can_manage_eval_config(user, cfg): 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 flush_eval_archive_records(*, username: str) -> dict[str, Any]: """Delete anonymized eval records from the mounted operator archive. This preserves ASF configuration/probe files and only clears archived record files/directories under the eval mount. It is intended for the instance owner or configured eval admins. """ cfg = effective_eval_config() if not _can_flush_eval_records(username, cfg): raise PermissionError("Only the instance owner or configured eval admins can flush the eval archive.") mount = Path(_clean_mount(cfg.get("job_mount_path"))) if not mount.exists() or not mount.is_dir(): raise FileNotFoundError(f"Eval archive mount not found at {mount}.") preserved = {_CONFIG_FILENAME, ".asf_eval_write_probe"} deleted_files = 0 deleted_dirs = 0 errors: list[dict[str, str]] = [] for child in list(mount.iterdir()): if child.name in preserved or child.name.startswith(".asf_eval_"): continue try: if child.is_dir(): shutil.rmtree(child) deleted_dirs += 1 else: child.unlink() deleted_files += 1 except Exception as exc: # noqa: BLE001 errors.append({"path": child.name, "error": str(exc)[:500]}) result = public_eval_config(username) result.update({ "message": f"Eval archive flushed: removed {deleted_dirs} directories and {deleted_files} files.", "flushed": not errors, "partial_flush": bool(errors), "deleted_top_level_dirs": deleted_dirs, "deleted_top_level_files": deleted_files, "errors": errors, }) return result 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["can_flush"] = _can_flush_eval_records(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