Source code for nvflare.app_common.executors.client_api_executor

# 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.

"""Client API executor with mode-specific trainer backends.

All three execution modes are supported. Parameter conversion and FULL/DIFF state
are handled by the trainer-side Client API.
"""

import math
from typing import Optional

from nvflare.apis.analytix import ANALYTIC_EVENT_TYPE
from nvflare.apis.dxo import DXO
from nvflare.apis.event_type import EventType
from nvflare.apis.executor import Executor
from nvflare.apis.fl_constant import FilterKey, FLContextKey, ReturnCode
from nvflare.apis.fl_context import FLContext
from nvflare.apis.fl_exception import UnsafeJobError
from nvflare.apis.shareable import Shareable, make_reply
from nvflare.apis.signal import Signal
from nvflare.apis.utils.analytix_utils import send_analytic_dxo
from nvflare.apis.utils.task_utils import contains_lazy_download_ref, get_filters, materialize_lazy_download_refs
from nvflare.app_common.app_constant import AppConstants
from nvflare.app_common.executors.client_api.backend_spec import ClientAPIBackendContext, ClientAPIBackendSpec
from nvflare.app_common.widgets.convert_to_fed_event import FED_EVENT_PREFIX
from nvflare.client.config import ExchangeFormat, TransferType, normalize_exchange_format
from nvflare.client.converter_utils import validate_format_pair
from nvflare.fuel.utils.fobs import FOBSContextKey
from nvflare.private.privacy_manager import Scope
from nvflare.security.logging import secure_format_exception, secure_format_traceback
from nvflare.utils.argv_utils import CommandArg, normalize_argv


[docs] class ExecutionMode: """Valid values for ClientAPIExecutor's execution_mode (see design "Overview").""" IN_PROCESS = "in_process" EXTERNAL_PROCESS = "external_process" ATTACH = "attach"
ALL_EXECUTION_MODES = (ExecutionMode.IN_PROCESS, ExecutionMode.EXTERNAL_PROCESS, ExecutionMode.ATTACH) FED_ANALYTIC_EVENT_TYPE = FED_EVENT_PREFIX + ANALYTIC_EVENT_TYPE # Shared by constructor defaults and wrong-mode override checks. _DEFAULT_LAUNCH_ONCE = True _DEFAULT_LAUNCH_TIMEOUT = 300.0 _DEFAULT_STOP_GRACE_PERIOD = 30.0 _DEFAULT_HEARTBEAT_INTERVAL = 5.0 _DEFAULT_HEARTBEAT_TIMEOUT = 30.0
[docs] class ClientAPIExecutor(Executor): """Delegates Client API task execution to the configured backend.""" def __init__( self, execution_mode: str, command: Optional[CommandArg] = None, task_script_path: Optional[str] = None, task_script_args: Optional[CommandArg] = "", launch_once: bool = _DEFAULT_LAUNCH_ONCE, launch_timeout: Optional[float] = _DEFAULT_LAUNCH_TIMEOUT, shutdown_timeout: Optional[float] = None, stop_grace_period: float = _DEFAULT_STOP_GRACE_PERIOD, heartbeat_interval: float = _DEFAULT_HEARTBEAT_INTERVAL, heartbeat_timeout: float = _DEFAULT_HEARTBEAT_TIMEOUT, task_wait_timeout: Optional[float] = None, result_wait_timeout: Optional[float] = None, train_task_name: str = AppConstants.TASK_TRAIN, evaluate_task_name: str = AppConstants.TASK_VALIDATION, submit_model_task_name: str = AppConstants.TASK_SUBMIT_MODEL, train_with_evaluation: bool = False, params_exchange_format: ExchangeFormat = ExchangeFormat.RAW, server_expected_format: ExchangeFormat = ExchangeFormat.NUMPY, params_transfer_type: TransferType = TransferType.FULL, memory_gc_rounds: int = 0, cuda_empty_cache: bool = False, attach_id: Optional[str] = None, attach_timeout: Optional[float] = None, allow_reconnect: bool = False, allow_insecure_attach: bool = False, ): """Initializes the ClientAPIExecutor. Parameter names are part of the public job-config surface: renames are breaking changes (guarded by the surface-freeze test). Args: execution_mode (str): One of "in_process", "external_process", or "attach". Required. command (Optional[Union[str, list[str]]]): The trainer launch command, either as a command string (e.g. "python custom/train.py") or shell-free argv. Required for (and only valid in) "external_process" mode. An empty/whitespace-only string or empty argv is treated as unset. Use argv when values must retain exact boundaries across platforms. task_script_path (Optional[str]): in_process only. Path to the user training script the in_process backend runs via TaskScriptRunner. An empty/whitespace-only string is treated as unset. (The in_process backend validates presence and ".py" suffix.) task_script_args (Union[str, list[str]]): in_process only. Arguments appended to task_script_path. Pre-tokenized argv preserves exact boundaries without shell parsing. launch_once (bool): external_process only. Launch the trainer once per job (default) vs once per task. launch_timeout (Optional[float]): external_process only. Bound for the launched trainer to complete its HELLO/session setup (this replaces the legacy external_pre_init_timeout). Defaults to 300 seconds for compatibility with prior releases; an explicit None means no timeout. shutdown_timeout (Optional[float]): external_process only. Primarily the natural-exit wait after orderly SHUTDOWN; it also bounds the finalize/execute gate, supplies the accepted-source disconnect grace, bounds the post-settlement process-group exit wait, and feeds the settled per-task reaper budget. None selects the 30-second backend default. Zero skips the direct orderly-exit and finalize-gate waits. In the accepted-source roles, zero is normalized to 30 seconds for disconnect and post-settlement group-exit waits; that fallback also feeds the fixed 30-second settled-reaper budget, which reserves up to 5 seconds for termination. stop_grace_period (float): external_process only. Grace period between SIGTERM and SIGKILL when terminating the trainer process group (design: "Process-tree termination"). The accepted-result reaper silently caps this phase at 5 seconds; ordinary teardown honors the configured value. heartbeat_interval (float): out-of-process only (external_process/attach). Interval (seconds) for session heartbeats. heartbeat_timeout (float): out-of-process only (external_process/attach). Session lease timeout (seconds) on missed heartbeats. An in-flight payload transfer keeps the lease alive (design: "Heartbeat and Liveness"). Zero disables heartbeat checks for external_process but is invalid for attach, whose external trainer requires heartbeat liveness as the fallback for a lost terminal SHUTDOWN. task_wait_timeout (Optional[float]): Bound for the trainer to accept a delivered task. None means no timeout for external_process; attach applies a 600-second task-delivery budget when unset. result_wait_timeout (Optional[float]): Control-side bound for retrieving the task result. Payload transfer completion is governed by the shared transfer layer, not by this value. None means no timeout. train_task_name (str): Task name treated as "train" by flare.is_train() (rank contract). Defaults to AppConstants.TASK_TRAIN. evaluate_task_name (str): Task name treated as "evaluate" by flare.is_evaluate(). Defaults to AppConstants.TASK_VALIDATION. submit_model_task_name (str): Task name treated as "submit_model" by flare.is_submit_model(). Defaults to AppConstants.TASK_SUBMIT_MODEL. train_with_evaluation (bool): Whether a training result is required to include pre-training evaluation metrics. When False, metrics are optional; metrics supplied by the trainer or a framework integration are still returned. params_exchange_format (ExchangeFormat): Framework-native parameter representation exposed by ``flare.receive()`` and accepted by ``flare.send()``. The declaration is transported to the trainer in ``TASK_EXCHANGE``; the executor does not perform conversion. ``RAW`` explicitly disables representation adaptation. server_expected_format (ExchangeFormat): Parameter representation expected by the server. The declaration is transported to the trainer in ``TASK_EXCHANGE``. params_transfer_type (TransferType): Whether training results contain full parameters or a difference from the received parameters. This is applied by the trainer-side model state and is independent of framework representation conversion. memory_gc_rounds (int): Force a GC cycle every N rounds (0 disables). cuda_empty_cache (bool): Whether to also empty the CUDA cache during memory cleanup. attach_id (Optional[str]): attach only. Pre-provisioned rendezvous ID shared with exactly one externally started trainer. It is a routing name, not a secret. attach_timeout (Optional[float]): attach only. Bound for the externally started trainer to attach. None means no timeout. allow_reconnect (bool): attach only. Whether a trainer may re-attach to an existing session after a disconnect. allow_insecure_attach (bool): attach only. Deprecated compatibility argument with no effect. Network Attach uses the site's existing CP trust model; a CJ-owned shared-file route must always satisfy its filesystem protection checks. """ super().__init__() if execution_mode not in ALL_EXECUTION_MODES: raise ValueError(f"invalid execution_mode {execution_mode!r}: must be one of {list(ALL_EXECUTION_MODES)}") # Normalize empty entry points before mode-specific validation. command = self._normalize_command(command) task_script_path = self._normalize_optional_str(task_script_path) task_script_args = self._normalize_script_args(task_script_args) is_in_process = execution_mode == ExecutionMode.IN_PROCESS is_external = execution_mode == ExecutionMode.EXTERNAL_PROCESS is_attach = execution_mode == ExecutionMode.ATTACH # --- external_process command / in_process script entry point --- if is_external: if not command: raise ValueError( "execution_mode 'external_process' requires a non-empty command " "(e.g. command='python custom/train.py')" ) elif command is not None: raise self._wrong_mode_error("command", command, "'external_process'", execution_mode) if not is_in_process: if task_script_path is not None: raise self._wrong_mode_error("task_script_path", task_script_path, "'in_process'", execution_mode) if task_script_args: raise self._wrong_mode_error("task_script_args", task_script_args, "'in_process'", execution_mode) # Reject external-process settings changed from their defaults in other modes. if not is_external: if launch_once != _DEFAULT_LAUNCH_ONCE: raise self._wrong_mode_error("launch_once", launch_once, "'external_process'", execution_mode) if launch_timeout != _DEFAULT_LAUNCH_TIMEOUT: raise self._wrong_mode_error("launch_timeout", launch_timeout, "'external_process'", execution_mode) if shutdown_timeout is not None: raise self._wrong_mode_error("shutdown_timeout", shutdown_timeout, "'external_process'", execution_mode) if stop_grace_period != _DEFAULT_STOP_GRACE_PERIOD: raise self._wrong_mode_error( "stop_grace_period", stop_grace_period, "'external_process'", execution_mode ) # --- heartbeat knobs are out-of-process only (there is no session heartbeat in_process) --- if is_in_process: if heartbeat_interval != _DEFAULT_HEARTBEAT_INTERVAL: raise self._wrong_mode_error( "heartbeat_interval", heartbeat_interval, "'external_process' or 'attach'", execution_mode ) if heartbeat_timeout != _DEFAULT_HEARTBEAT_TIMEOUT: raise self._wrong_mode_error( "heartbeat_timeout", heartbeat_timeout, "'external_process' or 'attach'", execution_mode ) # heartbeat_timeout=0 disables heartbeat timeout checking; otherwise the # heartbeat interval must be strictly smaller than the timeout. for name, value in ( ("heartbeat_interval", heartbeat_interval), ("heartbeat_timeout", heartbeat_timeout), ): if not isinstance(value, (int, float)) or isinstance(value, bool) or not math.isfinite(value): raise ValueError(f"{name} must be a finite number, but got {value!r}") if heartbeat_interval <= 0: raise ValueError(f"heartbeat_interval must > 0, but got {heartbeat_interval}") if heartbeat_timeout < 0: raise ValueError(f"heartbeat_timeout must >= 0, but got {heartbeat_timeout}") if 0 < heartbeat_timeout <= heartbeat_interval: raise ValueError( f"heartbeat_interval {heartbeat_interval} must be less than heartbeat_timeout {heartbeat_timeout}" ) # --- attach-only knobs --- if is_attach: from nvflare.client.cell.attach import validate_attach_id attach_id = validate_attach_id(attach_id) if heartbeat_timeout == 0: raise ValueError( "attach mode requires heartbeat_timeout > 0 because heartbeat liveness is the terminal fallback " "when protocol SHUTDOWN is lost" ) else: if attach_id is not None: raise self._wrong_mode_error("attach_id", attach_id, "'attach'", execution_mode) if attach_timeout is not None: raise self._wrong_mode_error("attach_timeout", attach_timeout, "'attach'", execution_mode) # Treat False and other falsy values as the unset default. if allow_reconnect: raise self._wrong_mode_error("allow_reconnect", allow_reconnect, "'attach'", execution_mode) if allow_insecure_attach: raise self._wrong_mode_error("allow_insecure_attach", allow_insecure_attach, "'attach'", execution_mode) # Reject invalid bounds before they reach threading/subprocess primitives. In # particular, a negative Lock.acquire(timeout=...) can mean "wait forever", # which would violate the executor's bounded-shutdown contract. for name, value in ( ("launch_timeout", launch_timeout), ("shutdown_timeout", shutdown_timeout), ("stop_grace_period", stop_grace_period), ("task_wait_timeout", task_wait_timeout), ("result_wait_timeout", result_wait_timeout), ("attach_timeout", attach_timeout), ): if value is not None and ( not isinstance(value, (int, float)) or isinstance(value, bool) or not math.isfinite(value) or value < 0 ): raise ValueError(f"{name} must be a finite number >= 0 or None, but got {value!r}") if not isinstance(memory_gc_rounds, int) or isinstance(memory_gc_rounds, bool) or memory_gc_rounds < 0: raise ValueError(f"memory_gc_rounds must be an integer >= 0, but got {memory_gc_rounds!r}") self._execution_mode = execution_mode self._command = command self._task_script_path = task_script_path self._task_script_args = task_script_args self._launch_once = launch_once self._launch_timeout = launch_timeout self._shutdown_timeout = shutdown_timeout self._stop_grace_period = stop_grace_period self._heartbeat_interval = heartbeat_interval self._heartbeat_timeout = heartbeat_timeout self._task_wait_timeout = task_wait_timeout self._result_wait_timeout = result_wait_timeout self._train_task_name = train_task_name self._evaluate_task_name = evaluate_task_name self._submit_model_task_name = submit_model_task_name self._train_with_evaluation = train_with_evaluation self._params_exchange_format = normalize_exchange_format(params_exchange_format, "params_exchange_format") self._server_expected_format = normalize_exchange_format(server_expected_format, "server_expected_format") validate_format_pair(self._params_exchange_format, self._server_expected_format) try: self._params_transfer_type = TransferType(params_transfer_type) except (TypeError, ValueError) as e: raise ValueError( f"invalid params_transfer_type {params_transfer_type!r}: must be one of {list(TransferType)}" ) from e self._memory_gc_rounds = memory_gc_rounds self._cuda_empty_cache = cuda_empty_cache self._attach_id = attach_id self._attach_timeout = attach_timeout self._allow_reconnect = bool(allow_reconnect) self._allow_insecure_attach = bool(allow_insecure_attach) self._backend: Optional[ClientAPIBackendSpec] = None # False uses the local event path; Cell backends enable direct federation events as needed. self._analytics_fire_fed_event: bool = False @property def execution_mode(self) -> str: """Read-only view of the configured mode, for validation and diagnostics.""" return self._execution_mode
[docs] def handle_event(self, event_type: str, fl_ctx: FLContext): if event_type == EventType.START_RUN: super().handle_event(event_type, fl_ctx) try: self._backend = self._create_backend() self._backend.initialize(self._build_backend_context(), fl_ctx) except Exception as e: # initialize() owns rollback; finalizing a partial backend is unsafe. self._backend = None self.log_error(fl_ctx, secure_format_traceback(), fire_event=False) self.system_panic( f"ClientAPIExecutor cannot start: backend for execution_mode " f"'{self._execution_mode}' failed to initialize: {secure_format_exception(e)}", fl_ctx, ) elif event_type == EventType.ABORT_TASK: backend = self._backend if backend is not None: try: backend.abort(fl_ctx) except Exception: self.log_error(fl_ctx, secure_format_traceback(), fire_event=False) super().handle_event(event_type, fl_ctx) elif event_type == EventType.END_RUN: backend = self._backend self._backend = None if backend is not None: try: backend.finalize(fl_ctx) except Exception: self.log_error(fl_ctx, secure_format_traceback(), fire_event=False) super().handle_event(event_type, fl_ctx) else: super().handle_event(event_type, fl_ctx)
[docs] def execute(self, task_name: str, shareable: Shareable, fl_ctx: FLContext, abort_signal: Signal) -> Shareable: backend = self._backend if backend is None: self.log_error( fl_ctx, f"no Client API backend available for execution_mode '{self._execution_mode}' - " f"backend initialization failed or START_RUN was not handled", ) return make_reply(ReturnCode.EXECUTION_EXCEPTION) try: materialize_result = ( self._execution_mode != ExecutionMode.IN_PROCESS and self._requires_materialized_result(task_name, fl_ctx) ) result_cell = None if materialize_result: result_cell = self._route_result_to_cj(shareable, fl_ctx) result = backend.execute(task_name, shareable, fl_ctx, abort_signal) if isinstance(result, Shareable) and materialize_result and contains_lazy_download_ref(result): result = self._materialize_result(result, result_cell, abort_signal) except UnsafeJobError: # ClientRunner has dedicated handling for UnsafeJobError (client_runner.py maps it # to ReturnCode.UNSAFE_JOB and marks the job unsafe). Do NOT swallow it into a # generic EXECUTION_EXCEPTION reply here - let it propagate so that handling fires. # (UnsafeComponentError has no such special handling and is left to the generic # branch below, which is its existing behavior.) raise except Exception: self.log_error(fl_ctx, secure_format_traceback()) return make_reply(ReturnCode.EXECUTION_EXCEPTION) if not isinstance(result, Shareable): self.log_error(fl_ctx, f"bad result from backend: expected Shareable but got {type(result)}") return make_reply(ReturnCode.EXECUTION_EXCEPTION) return result
@staticmethod def _requires_materialized_result(task_name: str, fl_ctx: FLContext) -> bool: """Whether the active ClientRunner pipeline consumes the concrete Client API result.""" runner = fl_ctx.get_prop(FLContextKey.RUNNER) active_task_name = fl_ctx.get_prop(FLContextKey.TASK_NAME) if active_task_name != task_name: # A client-side workflow controller can call its learn executor directly. # Ask the executor for the outer active task whether it consumes the nested # result locally; unrelated nested workflows keep pass-through behavior. find_executor = getattr(runner, "find_executor", None) active_executor = ( find_executor(active_task_name) if callable(find_executor) and isinstance(active_task_name, str) else None ) requirement = getattr(active_executor, "requires_materialized_task_result", None) return callable(requirement) and requirement(task_name) is True config_filters = getattr(runner, "task_result_filters", None) if isinstance(config_filters, dict) and get_filters( Scope.TASK_RESULT_FILTERS_NAME, fl_ctx, config_filters, task_name, FilterKey.OUT ): return True engine = fl_ctx.get_engine() get_all_components = getattr(engine, "get_all_components", None) if engine is not None else None components = get_all_components() if callable(get_all_components) else None if not isinstance(components, dict): return False for component in components.values(): requirement = getattr(component, "requires_materialized_task_result", None) if callable(requirement) and requirement(task_name) is True: return True return False @staticmethod def _route_result_to_cj(shareable: Shareable, fl_ctx: FLContext): """Make the CJ the terminal receiver for a result that a local component consumes.""" engine = fl_ctx.get_engine() get_cell = getattr(engine, "get_cell", None) if engine is not None else None cell = get_cell() if callable(get_cell) else None cj_fqcn = cell.get_fqcn() if cell is not None else None if not cj_fqcn: raise RuntimeError("cannot materialize Client API result: CJ Cell identity is unavailable") shareable.set_header(FOBSContextKey.RECEIVER_IDS, [cj_fqcn]) return cell @staticmethod def _materialize_result(result: Shareable, cell, abort_signal: Signal) -> Shareable: """Resolve a pass-through result at the CJ for a declared local consumer.""" materialized = materialize_lazy_download_refs(result, cell, abort_signal) if not isinstance(materialized, Shareable): raise TypeError(f"materialized Client API result must be Shareable but got {type(materialized)}") return materialized
[docs] def set_analytics_fire_fed_event(self, enabled: bool) -> None: """Select direct federation events instead of the local ConvertToFedEvent path.""" self._analytics_fire_fed_event = bool(enabled)
[docs] def fire_log_analytics(self, fl_ctx: FLContext, dxo: DXO) -> None: """Emit trainer LOG data through the configured analytics path. The local path fires ``analytix_log_stats`` and ConvertToFedEvent prefixes it to ``fed.analytix_log_stats``. The direct path must fire that prefixed name itself or server-side consumers will miss the event. """ if self._analytics_fire_fed_event: send_analytic_dxo( self, dxo=dxo, fl_ctx=fl_ctx, event_type=FED_ANALYTIC_EVENT_TYPE, fire_fed_event=True, ) else: send_analytic_dxo( self, dxo=dxo, fl_ctx=fl_ctx, event_type=ANALYTIC_EVENT_TYPE, fire_fed_event=False, )
@staticmethod def _normalize_optional_str(value: Optional[str]) -> Optional[str]: if isinstance(value, str) and not value.strip(): return None return value @staticmethod def _normalize_command(command: Optional[CommandArg]) -> Optional[CommandArg]: normalized = normalize_argv(command, "command", allow_none=True) if isinstance(normalized, str) or normalized is None: return ClientAPIExecutor._normalize_optional_str(normalized) if not normalized: return None if not normalized[0].strip(): raise ValueError("command argv must start with a non-empty executable") return normalized @staticmethod def _normalize_script_args(script_args: Optional[CommandArg]) -> CommandArg: normalized = normalize_argv(script_args, "task_script_args", allow_none=True) return "" if normalized is None else normalized @staticmethod def _wrong_mode_error(arg_name: str, value, valid_modes: str, execution_mode: str) -> ValueError: return ValueError( f"{arg_name} is only valid for execution_mode {valid_modes}, " f"but got {arg_name}={value!r} with execution_mode '{execution_mode}'" ) def _build_backend_context(self) -> ClientAPIBackendContext: return ClientAPIBackendContext( executor=self, task_script_path=self._task_script_path, task_script_args=self._task_script_args, command=self._command, launch_once=self._launch_once, launch_timeout=self._launch_timeout, shutdown_timeout=self._shutdown_timeout, stop_grace_period=self._stop_grace_period, heartbeat_interval=self._heartbeat_interval, heartbeat_timeout=self._heartbeat_timeout, task_wait_timeout=self._task_wait_timeout, result_wait_timeout=self._result_wait_timeout, attach_id=self._attach_id, attach_timeout=self._attach_timeout, allow_reconnect=self._allow_reconnect, allow_insecure_attach=self._allow_insecure_attach, train_task_name=self._train_task_name, evaluate_task_name=self._evaluate_task_name, submit_model_task_name=self._submit_model_task_name, train_with_evaluation=self._train_with_evaluation, params_exchange_format=self._params_exchange_format, server_expected_format=self._server_expected_format, params_transfer_type=self._params_transfer_type, memory_gc_rounds=self._memory_gc_rounds, cuda_empty_cache=self._cuda_empty_cache, ) def _create_backend(self) -> ClientAPIBackendSpec: if self._execution_mode == ExecutionMode.IN_PROCESS: return self._create_in_process_backend() if self._execution_mode == ExecutionMode.EXTERNAL_PROCESS: return self._create_external_process_backend() if self._execution_mode == ExecutionMode.ATTACH: return self._create_attach_backend() raise ValueError(f"unexpected execution_mode {self._execution_mode!r}") def _create_in_process_backend(self) -> ClientAPIBackendSpec: from nvflare.app_common.executors.client_api.in_process_backend import InProcessBackend return InProcessBackend() def _create_external_process_backend(self) -> ClientAPIBackendSpec: from nvflare.app_common.executors.client_api.external_process_backend import ExternalProcessBackend return ExternalProcessBackend() def _create_attach_backend(self) -> ClientAPIBackendSpec: from nvflare.app_common.executors.client_api.attach_backend import AttachBackend return AttachBackend()