# Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Site-owned per-study runtime configuration (v2): local/study_runtime.yaml.
Filename selects the format: this strict v2 parser never reads the frozen v1
local/study_data.yaml (see study_data.py). Both files present is a launcher
error, enforced by the launchers, not here.
"""
from __future__ import annotations
import logging
import os
import posixpath
import re
from dataclasses import dataclass, field
from pathlib import PurePosixPath
from typing import Optional
import yaml
from nvflare.app_opt.job_launcher.study_data import MODE_RO, MODE_RW, StudyDatasetMount
STUDY_RUNTIME_FILE = "local/study_runtime.yaml"
SUPPORTED_FORMAT_VERSION = 2
DATASET_TYPE_MOUNT = "mount"
_RESERVED_DATASET_TYPES = {"databricks"}
_STUDY_KEYS = {"container", "pod_template", "docker_kwargs", "datasets", "env", "secret_env", "secret_mounts"}
_MOUNT_DATASET_KEYS = {"type", "source", "mode"}
_SECRET_ENV_KEYS = {"source", "key"}
_SECRET_MOUNT_KEYS = {"source", "mount_path", "mode", "items"}
_VALID_NAME = re.compile(r"^[a-z0-9](?:[a-z0-9_-]{0,61}[a-z0-9])?$")
# Env names the launchers own at job launch (PYTHONPATH and the workspace-transfer
# variables from workspace_cell_transfer). A study value would be silently overridden
# or break workspace transfer, so both env and secret_env reject them up front.
_RESERVED_ENV_NAMES = frozenset({"PYTHONPATH", "NVFL_WORKSPACE_OWNER_FQCN", "NVFL_WORKSPACE_TRANSFER_TOKEN"})
# Docker containers.run kwargs owned by the launcher (the docker launcher and deploy
# validation enforce the same set for site-level default_job_container_kwargs).
RESERVED_DOCKER_KWARGS = frozenset(
{
"volumes",
"mounts",
"network",
"environment",
"command",
"name",
"detach",
"auto_remove",
"user",
"working_dir",
"image",
}
)
[docs]
@dataclass(frozen=True)
class SecretEnvRef:
name: str
source: str
key: str
[docs]
@dataclass(frozen=True)
class SecretMount:
study: str
name: str
source: str
mount_path: str
items: Optional[tuple] = None # tuple of (key, path) pairs; None = full projection
[docs]
@dataclass
class StudyRuntime:
study: str
datasets: list = field(default_factory=list) # list[StudyDatasetMount]
env: dict = field(default_factory=dict)
secret_env: list = field(default_factory=list) # list[SecretEnvRef]
secret_mounts: list = field(default_factory=list) # list[SecretMount]
container_image: Optional[str] = None
pod_template: Optional[dict] = None
docker_kwargs: dict = field(default_factory=dict)
[docs]
def study_runtime_file_path(workspace_root: str) -> str:
return os.path.join(workspace_root, *STUDY_RUNTIME_FILE.split("/"))
def _error(file_path: str, message: str) -> ValueError:
return ValueError(f"study runtime file '{file_path}': {message}")
def _require_dict(value, label: str, file_path: str) -> dict:
if not isinstance(value, dict):
raise _error(file_path, f"{label} must be a mapping.")
return value
def _require_known_keys(entry: dict, known: set, label: str, file_path: str) -> None:
unknown = set(entry.keys()) - known
if unknown:
raise _error(file_path, f"{label} has unknown key(s) {sorted(unknown)}; allowed: {sorted(known)}.")
def _require_name(value, label: str, file_path: str) -> str:
if not isinstance(value, str) or not _VALID_NAME.match(value):
raise _error(file_path, f"{label} {value!r} is not a valid name.")
return value
def _require_str(value, label: str, file_path: str) -> str:
if not isinstance(value, str) or not value:
raise _error(file_path, f"{label} must be a non-empty string.")
return value
def _safe_relative_path(value: str, label: str, file_path: str) -> str:
normalized = posixpath.normpath(value.replace(os.sep, "/"))
parts = PurePosixPath(normalized)
if parts.is_absolute() or ".." in parts.parts or normalized in ("", "."):
raise _error(file_path, f"{label} must be a relative path under the config directory: {value!r}")
return normalized
def _load_pod_template_file(template_path: str, file_path: str) -> dict:
try:
with open(template_path, "rt") as f:
pod_template = yaml.safe_load(f)
except FileNotFoundError as e:
raise _error(file_path, f"pod_template file '{template_path}' was not found.") from e
except OSError as e:
raise _error(file_path, f"could not read pod_template file '{template_path}': {e}") from e
except yaml.YAMLError as e:
raise _error(file_path, f"could not parse pod_template file '{template_path}': {e}") from e
return _validate_pod_template(pod_template, f"pod_template file '{template_path}'", file_path)
def _validate_pod_template(pod_template, label: str, file_path: str) -> dict:
pod_template = _require_dict(pod_template, label, file_path)
kind = pod_template.get("kind")
if kind and kind != "Pod":
raise _error(file_path, f"{label} must define kind: Pod.")
return pod_template
def _parse_pod_template(value, file_path: str, allow_pod_template: bool) -> dict:
if not allow_pod_template:
raise _error(file_path, "pod_template is Kubernetes-only and is not supported by this launcher.")
if isinstance(value, str):
rel_path = _safe_relative_path(_require_str(value, "pod_template", file_path), "pod_template", file_path)
template_path = os.path.join(os.path.dirname(file_path), *rel_path.split("/"))
return _load_pod_template_file(template_path, file_path)
return _validate_pod_template(value, "inline pod_template", file_path)
def _parse_docker_kwargs(study: str, entry, file_path: str, allow_docker_kwargs: bool) -> dict:
label = f"studies.{study}.docker_kwargs"
if not allow_docker_kwargs:
raise _error(file_path, "docker_kwargs is Docker-only and is not supported by this launcher.")
entry = _require_dict(entry, label, file_path)
reserved = sorted(RESERVED_DOCKER_KWARGS & set(entry))
if reserved:
raise _error(file_path, f"{label} must not contain launcher-owned key(s) {reserved}.")
for key in entry:
_require_str(key, f"{label} key", file_path)
return dict(entry)
def _parse_container(study: str, entry, file_path: str) -> str:
label = f"studies.{study}.container"
entry = _require_dict(entry, label, file_path)
_require_known_keys(entry, {"image"}, label, file_path)
return _require_str(entry.get("image"), f"{label}.image", file_path)
def _parse_dataset(study: str, dataset: str, entry, file_path: str) -> StudyDatasetMount:
label = f"studies.{study}.datasets.{dataset}"
_require_name(dataset, f"{label} dataset name", file_path)
entry = _require_dict(entry, label, file_path)
dataset_type = entry.get("type", DATASET_TYPE_MOUNT)
if dataset_type in _RESERVED_DATASET_TYPES:
raise _error(file_path, f"{label} type '{dataset_type}' is not yet supported.")
if dataset_type != DATASET_TYPE_MOUNT:
raise _error(file_path, f"{label} has unknown type {dataset_type!r}; allowed: '{DATASET_TYPE_MOUNT}'.")
_require_known_keys(entry, _MOUNT_DATASET_KEYS, label, file_path)
source = _require_str(entry.get("source"), f"{label}.source", file_path)
mode = entry.get("mode")
if mode not in (MODE_RO, MODE_RW):
raise _error(file_path, f"{label}.mode must be '{MODE_RO}' or '{MODE_RW}'.")
return StudyDatasetMount(study=study, dataset=dataset, source=source, mode=mode)
def _parse_env(study: str, entry, file_path: str) -> dict:
entry = _require_dict(entry, f"studies.{study}.env", file_path)
env = {}
for name, value in entry.items():
_require_str(name, f"studies.{study}.env variable name", file_path)
if name in _RESERVED_ENV_NAMES:
raise _error(file_path, f"studies.{study}.env.{name} is launcher-owned and cannot be set in study config.")
if isinstance(value, (dict, list)) or value is None:
raise _error(file_path, f"studies.{study}.env.{name} must be a scalar value.")
if isinstance(value, bool):
# str(True) is "True"; YAML users writing `true` expect "true"
value = "true" if value else "false"
value = str(value)
if not value:
# empty values behave differently per launcher and can silently lose
# against pod-template entries in the manifest merge
raise _error(file_path, f"studies.{study}.env.{name} must not be empty; set a value or remove the key.")
env[name] = value
return env
def _parse_secret_env(study: str, entry, file_path: str) -> list:
entry = _require_dict(entry, f"studies.{study}.secret_env", file_path)
refs = []
for name, ref in entry.items():
label = f"studies.{study}.secret_env.{name}"
_require_str(name, f"{label} variable name", file_path)
if name in _RESERVED_ENV_NAMES:
raise _error(file_path, f"{label} is launcher-owned and cannot be set in study config.")
ref = _require_dict(ref, label, file_path)
_require_known_keys(ref, _SECRET_ENV_KEYS, label, file_path)
refs.append(
SecretEnvRef(
name=name,
source=_require_str(ref.get("source"), f"{label}.source", file_path),
key=_require_str(ref.get("key"), f"{label}.key", file_path),
)
)
return refs
def _parse_secret_mounts(study: str, entry, file_path: str, allow_secret_mount_items: bool) -> list:
entry = _require_dict(entry, f"studies.{study}.secret_mounts", file_path)
mounts = []
for name, mount in entry.items():
label = f"studies.{study}.secret_mounts.{name}"
_require_name(name, f"{label} mount name", file_path)
mount = _require_dict(mount, label, file_path)
_require_known_keys(mount, _SECRET_MOUNT_KEYS, label, file_path)
mode = mount.get("mode", MODE_RO)
if mode != MODE_RO:
raise _error(file_path, f"{label}.mode must be '{MODE_RO}'; secret mounts are always read-only.")
mount_path = _require_str(mount.get("mount_path"), f"{label}.mount_path", file_path)
if not posixpath.isabs(mount_path):
raise _error(file_path, f"{label}.mount_path must be an absolute path.")
items = mount.get("items")
if items is not None:
if not allow_secret_mount_items:
raise _error(
file_path,
f"{label}.items is Kubernetes-only and is not supported by this launcher; "
"point source at a directory containing only the intended files.",
)
items = _require_dict(items, f"{label}.items", file_path)
if not items:
raise _error(file_path, f"{label}.items must not be empty; omit it for full projection.")
items = tuple(
(
_require_str(key, f"{label}.items key", file_path),
_require_str(path, f"{label}.items.{key}", file_path),
)
for key, path in items.items()
)
mounts.append(
SecretMount(
study=study,
name=name,
source=_require_str(mount.get("source"), f"{label}.source", file_path),
mount_path=mount_path,
items=items,
)
)
return mounts
def _parse_study(
study: str,
entry,
file_path: str,
allow_pod_template: bool,
allow_secret_mount_items: bool,
allow_docker_kwargs: bool,
) -> StudyRuntime:
_require_name(study, "study name", file_path)
entry = _require_dict(entry, f"studies.{study}", file_path)
_require_known_keys(entry, _STUDY_KEYS, f"studies.{study}", file_path)
runtime = StudyRuntime(study=study)
if "container" in entry:
runtime.container_image = _parse_container(study, entry["container"], file_path)
if "pod_template" in entry:
runtime.pod_template = _parse_pod_template(entry["pod_template"], file_path, allow_pod_template)
if "docker_kwargs" in entry:
runtime.docker_kwargs = _parse_docker_kwargs(study, entry["docker_kwargs"], file_path, allow_docker_kwargs)
if "datasets" in entry:
datasets = _require_dict(entry["datasets"], f"studies.{study}.datasets", file_path)
runtime.datasets = [
_parse_dataset(study, dataset, ds_entry, file_path) for dataset, ds_entry in datasets.items()
]
if "env" in entry:
runtime.env = _parse_env(study, entry["env"], file_path)
if "secret_env" in entry:
runtime.secret_env = _parse_secret_env(study, entry["secret_env"], file_path)
duplicated = set(runtime.env) & {ref.name for ref in runtime.secret_env}
if duplicated:
raise _error(file_path, f"studies.{study} defines {sorted(duplicated)} in both env and secret_env.")
if "secret_mounts" in entry:
runtime.secret_mounts = _parse_secret_mounts(study, entry["secret_mounts"], file_path, allow_secret_mount_items)
return runtime
[docs]
def load_study_runtime_file(
file_path: str,
allow_pod_template: bool = True,
allow_secret_mount_items: bool = True,
allow_docker_kwargs: bool = True,
logger: Optional[logging.Logger] = None,
) -> dict:
"""Parse local/study_runtime.yaml (strict v2). Returns {study: StudyRuntime}."""
try:
with open(file_path, "rt") as f:
config = yaml.safe_load(f)
except OSError as e:
raise ValueError(f"Could not read study runtime file '{file_path}': {e}") from e
except yaml.YAMLError as e:
raise ValueError(f"Could not parse study runtime file '{file_path}': {e}") from e
config = _require_dict(config, "file content", file_path)
_require_known_keys(config, {"format_version", "studies"}, "top level", file_path)
format_version = config.get("format_version")
if format_version != SUPPORTED_FORMAT_VERSION:
raise _error(file_path, f"format_version must be {SUPPORTED_FORMAT_VERSION}, got {format_version!r}.")
studies = config.get("studies")
if studies is None:
studies = {}
studies = _require_dict(studies, "studies", file_path)
if not studies and logger:
logger.warning("study runtime file '%s' has no study entries; no study runtime will be configured", file_path)
return {
study: _parse_study(
study,
entry,
file_path,
allow_pod_template=allow_pod_template,
allow_secret_mount_items=allow_secret_mount_items,
allow_docker_kwargs=allow_docker_kwargs,
)
for study, entry in studies.items()
}
[docs]
def resolve_study_runtime(
runtime_map: dict, study: Optional[str], file_path: str, logger: Optional[logging.Logger] = None
) -> StudyRuntime:
"""Return the study's runtime config, or an empty config when the study has no entry."""
if study and study in runtime_map:
return runtime_map[study]
if study and runtime_map and logger:
logger.warning(
"study runtime file '%s' has no entry for study '%s'; no study runtime will be configured",
file_path,
study,
)
return StudyRuntime(study=study or "")