diff --git a/nvflare/app_common/executors/client_api/__init__.py b/nvflare/app_common/executors/client_api/__init__.py new file mode 100644 index 0000000000..4fc25d0d3c --- /dev/null +++ b/nvflare/app_common/executors/client_api/__init__.py @@ -0,0 +1,13 @@ +# 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. diff --git a/nvflare/app_common/executors/client_api/backend_spec.py b/nvflare/app_common/executors/client_api/backend_spec.py new file mode 100644 index 0000000000..5893ceea8d --- /dev/null +++ b/nvflare/app_common/executors/client_api/backend_spec.py @@ -0,0 +1,196 @@ +# 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. + +"""Backend spec for the Client API execution modes (V1-internal). + +Design: docs/design/client_api_execution_modes.md ("Overview", "Execution Modes", +"Client API Backends"). One ClientAPIExecutor delegates to one mode-specific backend: + +- in_process: trainer runs inside the Client Job (CJ) process over DataBus (EX-3) +- external_process: NVFlare launches and owns the trainer process tree over Cell (EP-4) +- attach: an externally owned trainer attaches over Cell (AT-2) + +This module is internal to NVFlare. It is not a user extension point; users configure +``ClientAPIExecutor(execution_mode=...)`` only. +""" + +from abc import ABC, abstractmethod +from dataclasses import dataclass +from typing import TYPE_CHECKING, Optional + +from nvflare.apis.fl_context import FLContext +from nvflare.apis.shareable import Shareable +from nvflare.apis.signal import Signal + +if TYPE_CHECKING: + # Import for typing only; avoids a runtime import cycle + # (client_api_executor imports this module to build the context). + from nvflare.app_common.executors.client_api_executor import ClientAPIExecutor + + +@dataclass(frozen=True) +class ClientAPIBackendContext: + """Immutable config a ClientAPIExecutor hands to its backend at ``initialize()``. + + Rationale (finding 11): the executor's frozen constructor args live in private attributes and + the backend factories are zero-arg, so a backend previously had no clean way to read the + heartbeat/timeout/converter/task-name config it needs, nor a supported reference back to the + executor's analytics hook. This frozen snapshot is that supported channel - a backend reads its + config from here rather than reaching into ``ClientAPIExecutor`` private attributes. + + The fields mirror the frozen ``ClientAPIExecutor`` constructor surface one-to-one. ``executor`` + is a back-reference so a backend can: + + - call ``executor.fire_log_analytics(fl_ctx, dxo)`` for every trainer LOG message (the single + LOG-to-analytics ownership point; see design "Configuration Surface"), and + - select the federation-scoped analytics path when appropriate by setting + ``executor._analytics_fire_fed_event = True`` in ``initialize()`` (Cell backends do this when + no ConvertToFedEvent widget is configured), and + - use the executor's FLComponent logging helpers. + """ + + executor: "ClientAPIExecutor" + execution_mode: str + # in_process entry point + task_script_path: Optional[str] = None + task_script_args: str = "" + # external_process launch + command: Optional[str] = None + launch_once: bool = True + launch_timeout: Optional[float] = None + shutdown_timeout: Optional[float] = None + stop_grace_period: float = 30.0 + # session / protocol (out-of-process) + heartbeat_interval: float = 5.0 + heartbeat_timeout: float = 30.0 + task_wait_timeout: Optional[float] = None + result_wait_timeout: Optional[float] = None + # NOTE: params_exchange_format / params_transfer_type / server_expected_format and the + # from/to_nvflare_converter ids are intentionally NOT here. Per FLARE-2698, param + # conversion between the framework-agnostic aggregation representation (numpy) and the + # framework-native training representation moves out of the executor to send/receive + # filters at the client edge; the Client API boundary is pass-through. Transfer type + # (FULL/DIFF) stays a Client API concern (model_registry), decided separately. + # task-name / rank contract (all modes) + train_task_name: str = "train" + evaluate_task_name: str = "validate" + submit_model_task_name: str = "submit_model" + train_with_evaluation: bool = False + # memory management (all modes) + memory_gc_rounds: int = 0 + cuda_empty_cache: bool = False + # attach + attach_timeout: Optional[float] = None + allow_reconnect: bool = False + + +class ClientAPIBackendSpec(ABC): + """The narrow lifecycle contract that ClientAPIExecutor drives on its backend. + + Lifecycle ownership per execution mode: + + - in_process: the backend runs the trainer inside the CJ process and owns its thread. + - external_process: the backend launches and owns the external trainer process tree; it must + not stop the trainer before the payload transfer of a pending result reaches terminal state. + - attach: the external system owns the trainer process; the backend owns only the attach + session, token validation, and heartbeat lease. + """ + + @abstractmethod + def initialize(self, context: ClientAPIBackendContext, fl_ctx: FLContext) -> None: + """Prepares the backend for the run. Called once when the executor handles START_RUN. + + ``context`` (finding 11) is the frozen snapshot of the executor's configuration plus a + back-reference to the executor (for ``fire_log_analytics`` and logging). A backend should + read all of its config from ``context`` rather than from executor private attributes, and + should retain what it needs for ``execute``/``handle_event``/``finalize``. + + The backend sets up its control plane here (DataBus wiring for in_process; Cell session + machinery, bootstrap config, and - per launch_once policy - trainer launch for + external_process; attach listener/token for attach). + + Contract: raise an exception on any setup failure. The executor converts the exception + into system_panic so the job fails cleanly instead of hanging while tasks wait on a + backend that never became ready. + + Cleanup-on-failure contract (finding 12): ``initialize()`` must be exception-safe and + self-unwinding - if it raises, it must first release any partial setup it already made + (threads started, processes launched, listeners/tokens registered, files written). The + executor does NOT call ``finalize()`` on a backend whose ``initialize()`` raised, because + ``finalize()`` cannot assume a consistently half-initialized backend. Own your own rollback. + + Args: + context: the frozen backend configuration and executor back-reference. + fl_ctx: the FLContext of the START_RUN event. + """ + pass + + @abstractmethod + def execute(self, task_name: str, shareable: Shareable, fl_ctx: FLContext, abort_signal: Signal) -> Shareable: + """Executes one task on the trainer and returns its result. + + The backend delivers the task to the trainer (TASK_READY over Cell, or DataBus for + in_process), waits for the result within the executor-configured task/result bounds, and + returns the result Shareable. + + Contract: this method must always return a Shareable and must not hang past abort: + when abort_signal is triggered, the backend notifies/stops the trainer per its mode's + lifecycle ownership and returns ``make_reply(ReturnCode.TASK_ABORTED)``. On failure it + should return an error reply (e.g. ReturnCode.EXECUTION_EXCEPTION) rather than raise; + exceptions that do escape are converted to EXECUTION_EXCEPTION replies by the executor, + except UnsafeJobError which the executor lets propagate so ClientRunner can apply its + dedicated UNSAFE_JOB handling. + + Args: + task_name: name of the task. + shareable: the task data. + fl_ctx: the FLContext of the task. + abort_signal: checked during execution; triggered means the task is aborted. + + Returns: + The result Shareable (an error reply on failure/abort - never None). + """ + pass + + @abstractmethod + def handle_event(self, event_type: str, fl_ctx: FLContext) -> None: + """Handles an FL event relayed by the executor. + + The executor relays events other than START_RUN/END_RUN (those are mapped to + initialize/finalize). Backends use this for mode-specific bookkeeping. + + Contract: must not raise; log and continue on internal errors. + + Args: + event_type: the fired event type. + fl_ctx: the FLContext of the event. + """ + pass + + @abstractmethod + def finalize(self, fl_ctx: FLContext) -> None: + """Releases backend resources. Called when the executor handles END_RUN. + + The backend tears down per its mode's lifecycle ownership: stop the in-process trainer + thread; send SHUTDOWN and stop the owned process tree (honoring the executor's + shutdown_timeout and stop_grace_period, and pending payload terminal state) for + external_process; close the session lease (without killing the trainer) for attach. + + Contract: must be idempotent and must not raise. Not called if ``initialize()`` raised + (see the cleanup-on-failure contract on ``initialize()``). + + Args: + fl_ctx: the FLContext of the END_RUN event. + """ + pass diff --git a/nvflare/app_common/executors/client_api_executor.py b/nvflare/app_common/executors/client_api_executor.py new file mode 100644 index 0000000000..db58743e05 --- /dev/null +++ b/nvflare/app_common/executors/client_api_executor.py @@ -0,0 +1,456 @@ +# 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. + +"""The public Client API executor, configured by execution mode. + +Design: docs/design/client_api_execution_modes.md ("What We Propose", "Overview", +"Execution Modes", "Configuration Surface"). This module path is normative - job configs +reference ``nvflare.app_common.executors.client_api_executor.ClientAPIExecutor``. + +This is the interface-freeze skeleton (plan: EX-2). The constructor surface below is frozen; +the mode backends land in follow-up PRs (in_process, external_process, attach). + +Divergence from the design's V1 "Configuration Surface" list +------------------------------------------------------------ +The design's V1 arg list is a subset; the frozen surface below adds the following load-bearing +args that the mode backends require and that both legacy executors +(InProcessClientAPIExecutor / ClientAPILauncherExecutor) already expose. They are recorded here +so the design's Configuration Surface can be synced separately: + +- ``task_script_path`` / ``task_script_args`` - in_process script entry point (the in_process + backend runs a user script via TaskScriptRunner; ``command`` names the external_process + trainer, ``task_script_path`` names the in_process one). +- ``train_task_name`` / ``evaluate_task_name`` / ``submit_model_task_name`` / + ``train_with_evaluation`` - power the rank-contract APIs flare.is_train()/is_evaluate()/ + is_submit_model(). +- ``memory_gc_rounds`` / ``cuda_empty_cache`` - periodic GC / CUDA cache management + +Deliberately excluded (per FLARE-2698): ``params_exchange_format`` / ``params_transfer_type`` / +``server_expected_format`` / ``from_nvflare_converter_id`` / ``to_nvflare_converter_id``. Param +conversion moves from executor-owned ParamsConverters to send/receive filters at the client +edge (the intermediate layers pass through), so these are not frozen into this surface. The +transfer type (FULL/DIFF) remains a Client API concern (model_registry) and is decided +separately from the converter removal. + (public on ScriptRunner and both legacy executors). +""" + +from typing import Callable, Dict, 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 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.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.security.logging import secure_format_exception, secure_format_traceback + + +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) + +# Federation-scoped analytics event, matching MetricRelay's ex-process default +# (job_config/script_runner.py) and flower_job.py: "fed.analytix_log_stats". +FED_ANALYTIC_EVENT_TYPE = FED_EVENT_PREFIX + ANALYTIC_EVENT_TYPE + +# Frozen defaults for mode-specific knobs (design "Configuration Surface"). Kept as named +# constants so the constructor default and the wrong-mode "explicitly set a non-default" checks +# below cannot drift apart. +_DEFAULT_LAUNCH_ONCE = True +_DEFAULT_STOP_GRACE_PERIOD = 30.0 +_DEFAULT_HEARTBEAT_INTERVAL = 5.0 +_DEFAULT_HEARTBEAT_TIMEOUT = 30.0 + + +class ClientAPIExecutor(Executor): + """One executor for all Client API execution modes. + + The trainer-facing Client API (flare.init/receive/send/log) is unchanged; this executor + replaces the Pipe/launcher integration stack. It delegates to an internal mode-specific + backend (ClientAPIBackendSpec) resolved from ``execution_mode`` at START_RUN: + + - ``in_process``: trainer runs inside the CJ process over DataBus. + - ``external_process``: NVFlare launches and owns the trainer process tree; control over Cell. + - ``attach``: an externally started/owned trainer attaches over Cell. + """ + + def __init__( + self, + execution_mode: str, + command: Optional[str] = None, + task_script_path: Optional[str] = None, + task_script_args: str = "", + launch_once: bool = _DEFAULT_LAUNCH_ONCE, + launch_timeout: Optional[float] = None, + 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, + memory_gc_rounds: int = 0, + cuda_empty_cache: bool = False, + attach_timeout: Optional[float] = None, + allow_reconnect: bool = False, + ): + """Initializes the ClientAPIExecutor. + + This constructor surface is frozen (design: "Configuration Surface"); parameter renames + break the surface-freeze test and downstream backend PRs. See the module docstring for the + args added beyond the design's V1 Configuration Surface list. + + Args: + execution_mode (str): One of "in_process", "external_process", or "attach". Required. + command (Optional[str]): The trainer launch command, e.g. "python custom/train.py", + "torchrun ...". Required for (and only valid in) "external_process" mode. An + empty/whitespace-only string is treated as unset. + 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 (str): in_process only. Arguments appended to task_script_path. + 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). None means no timeout. + shutdown_timeout (Optional[float]): external_process only. How long to wait for the + trainer to exit naturally after an orderly SHUTDOWN before starting forced + process-tree termination. None means the backend default. + stop_grace_period (float): external_process only. Grace period between SIGTERM and + SIGKILL when terminating the trainer process group (design: "Process-tree + termination"). + 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"). + task_wait_timeout (Optional[float]): Bound for the trainer to accept a delivered + task. None means no timeout. + 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 the trainer also returns evaluation metrics with + the trained model. + 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_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. + """ + 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 an empty/whitespace command or task_script_path to None up front so "" means + # "unset" uniformly in every mode. Previously external_process used `if not command` while + # the other modes used `command is not None`, so command="" was rejected with a misleading + # "only valid for external_process" message in in_process/attach instead of treated as unset. + command = self._normalize_optional_str(command) + task_script_path = self._normalize_optional_str(task_script_path) + + 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) + + # --- external_process-only lifecycle knobs (reject only when explicitly set away from the + # frozen default in a mode that ignores them) --- + 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 is not None: + 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 + ) + + # --- attach-only knobs --- + if not is_attach: + if attach_timeout is not None: + raise self._wrong_mode_error("attach_timeout", attach_timeout, "'attach'", execution_mode) + # Only reject a truthy allow_reconnect: allow_reconnect=False (the frozen default) is + # indistinguishable from "not set". Using `if allow_reconnect` (not `is not False`) + # also stops misfires on falsy-but-not-False values (None, 0, numpy.bool_(False)). + if allow_reconnect: + raise self._wrong_mode_error("allow_reconnect", allow_reconnect, "'attach'", execution_mode) + + 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._memory_gc_rounds = memory_gc_rounds + self._cuda_empty_cache = cuda_empty_cache + self._attach_timeout = attach_timeout + self._allow_reconnect = allow_reconnect + + self._backend: Optional[ClientAPIBackendSpec] = None + + # Analytics-event ownership (design: "Configuration Surface" - "the executor's Cell + # backend converts [LOG messages] into fed.analytix_log_stats analytics events"). + # False (default): fire the local un-prefixed ANALYTIC_EVENT_TYPE and rely on a + # ConvertToFedEvent widget (added by BaseFedJob) to re-fire it as + # "fed.analytix_log_stats" - today's in-process executor behavior. + # True: fire a federation-scoped event directly (today's MetricRelay behavior for + # ex-process). Cell backends (EP-4/AT-2) may select this path in initialize(). + self._analytics_fire_fed_event: bool = False + + 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: + # finding 12: initialize() is contracted to self-unwind its partial setup on + # failure, so the executor does NOT call finalize() on a half-initialized backend; + # it just drops the reference and panics so the job fails cleanly. + 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.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: + if self._backend is not None: + try: + self._backend.handle_event(event_type, fl_ctx) + except Exception: + self.log_error(fl_ctx, secure_format_traceback(), fire_event=False) + super().handle_event(event_type, fl_ctx) + + def execute(self, task_name: str, shareable: Shareable, fl_ctx: FLContext, abort_signal: Signal) -> Shareable: + backend = self._backend + if backend is None: + # START_RUN either never happened or backend initialization failed (and the executor + # already panicked). Reply with an error instead of waiting on a backend that will + # never be ready. + 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: + result = backend.execute(task_name, shareable, fl_ctx, 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 + + def fire_log_analytics(self, fl_ctx: FLContext, dxo: DXO) -> None: + """Converts trainer LOG data into an analytics event. Executor-owned surface. + + Backends call this for every LOG control message (flare.log from the trainer), + regardless of execution mode - this replaces MetricRelay (ex-process) and the + in-process executor's log callback as the single analytics-event ownership point. + + The fire path is mode-selectable via ``self._analytics_fire_fed_event``: + + - False (default): fires the local, un-prefixed ANALYTIC_EVENT_TYPE + ("analytix_log_stats") and relies on the ConvertToFedEvent widget (added by + BaseFedJob) to forward it to the server as "fed.analytix_log_stats". + - True: fires a federation-scoped event directly to the server, matching what + MetricRelay does today for ex-process metrics. Cell backends may select this + path during initialize() when no ConvertToFedEvent widget is configured. + + The two paths must land on the same server-side event name. ConvertToFedEvent prefixes + the local event with "fed.", so the fed path must fire the already-prefixed + FED_ANALYTIC_EVENT_TYPE ("fed.analytix_log_stats"); firing the un-prefixed name + federation-scoped would miss every consumer listening on "fed.analytix_log_stats" + (MetricRelay in job_config/script_runner.py, flower_job.py). + + Args: + fl_ctx: an FLContext to fire the event with. + dxo: the analytics data (e.g. from create_analytic_dxo) carried by 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]: + """Treats an empty/whitespace-only string as unset (None).""" + if isinstance(value, str) and not value.strip(): + return None + return value + + @staticmethod + def _wrong_mode_error(arg_name: str, value, valid_modes: str, execution_mode: str) -> ValueError: + """Builds the consistent 'arg only valid for ' rejection error.""" + 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: + """Builds the frozen config snapshot handed to the backend at initialize() (finding 11).""" + return ClientAPIBackendContext( + executor=self, + execution_mode=self._execution_mode, + 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, + 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, + memory_gc_rounds=self._memory_gc_rounds, + cuda_empty_cache=self._cuda_empty_cache, + attach_timeout=self._attach_timeout, + allow_reconnect=self._allow_reconnect, + ) + + def _backend_registry(self) -> Dict[str, Callable[[], ClientAPIBackendSpec]]: + """Internal registry mapping each execution_mode to its backend factory. + + Backend PRs replace the corresponding factory below with one that returns a real + ClientAPIBackendSpec; the registry keys are the frozen mode names. + """ + return { + ExecutionMode.IN_PROCESS: self._create_in_process_backend, + ExecutionMode.EXTERNAL_PROCESS: self._create_external_process_backend, + ExecutionMode.ATTACH: self._create_attach_backend, + } + + def _create_backend(self) -> ClientAPIBackendSpec: + factory = self._backend_registry().get(self._execution_mode) + if factory is None: + # Unreachable via the public constructor (execution_mode is validated there); + # guards subclasses that override _backend_registry(). + raise ValueError( + f"no backend factory registered for execution_mode '{self._execution_mode}': " + f"registered modes are {list(self._backend_registry().keys())}" + ) + return factory() + + def _create_in_process_backend(self) -> ClientAPIBackendSpec: + # Implemented in a follow-up PR (plan: EX-3). + raise NotImplementedError("in_process execution mode is not yet implemented in this release") + + def _create_external_process_backend(self) -> ClientAPIBackendSpec: + # Implemented in a follow-up PR (plan: EP-4). + raise NotImplementedError("external_process execution mode is not yet implemented in this release") + + def _create_attach_backend(self) -> ClientAPIBackendSpec: + # Implemented in a follow-up PR (plan: AT-2). + raise NotImplementedError("attach execution mode is not yet implemented in this release") diff --git a/tests/unit_test/app_common/executors/client_api_executor_test.py b/tests/unit_test/app_common/executors/client_api_executor_test.py new file mode 100644 index 0000000000..a8f710c6c4 --- /dev/null +++ b/tests/unit_test/app_common/executors/client_api_executor_test.py @@ -0,0 +1,559 @@ +# 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. + +"""Tests for the ClientAPIExecutor interface-freeze skeleton (plan: EX-2). + +Covers the constructor-validation matrix, the per-mode dispatch failure behavior +(NotImplementedError naming the follow-up PR + system_panic, no hang), the analytics-event +ownership hook, and the surface-freeze contract on the frozen constructor parameter list. +""" + +import inspect +from unittest.mock import Mock + +import pytest + +from nvflare.apis.analytix import ANALYTIC_EVENT_TYPE, AnalyticsDataType +from nvflare.apis.event_type import EventType +from nvflare.apis.executor import Executor +from nvflare.apis.fl_constant import EventScope, FLContextKey, ReservedKey, ReturnCode +from nvflare.apis.fl_context import FLContext +from nvflare.apis.fl_exception import UnsafeComponentError, UnsafeJobError +from nvflare.apis.shareable import Shareable, make_reply +from nvflare.apis.signal import Signal +from nvflare.apis.utils.analytix_utils import create_analytic_dxo +from nvflare.app_common.app_constant import AppConstants +from nvflare.app_common.executors.client_api.backend_spec import ClientAPIBackendContext, ClientAPIBackendSpec +from nvflare.app_common.executors.client_api_executor import ( + ALL_EXECUTION_MODES, + FED_ANALYTIC_EVENT_TYPE, + ClientAPIExecutor, + ExecutionMode, +) + +# The frozen V1 constructor surface (design: "Configuration Surface" in +# docs/design/client_api_execution_modes.md, plus the load-bearing args added beyond that list - +# documented in the ClientAPIExecutor module docstring). Renaming, removing, or reordering any of +# these is a breaking change to the interface freeze consumed by the mode backends. +FROZEN_CONSTRUCTOR_PARAMS = [ + "execution_mode", + "command", + "task_script_path", + "task_script_args", + "launch_once", + "launch_timeout", + "shutdown_timeout", + "stop_grace_period", + "heartbeat_interval", + "heartbeat_timeout", + "task_wait_timeout", + "result_wait_timeout", + "train_task_name", + "evaluate_task_name", + "submit_model_task_name", + "train_with_evaluation", + "memory_gc_rounds", + "cuda_empty_cache", + "attach_timeout", + "allow_reconnect", +] + +# Minimal valid constructor kwargs per mode. +MODE_KWARGS = { + ExecutionMode.IN_PROCESS: {"execution_mode": "in_process"}, + ExecutionMode.EXTERNAL_PROCESS: {"execution_mode": "external_process", "command": "python custom/train.py"}, + ExecutionMode.ATTACH: {"execution_mode": "attach"}, +} + + +def _make_fl_ctx(engine) -> FLContext: + fl_ctx = FLContext() + fl_ctx.put(key=ReservedKey.ENGINE, value=engine, private=True, sticky=False) + return fl_ctx + + +def _make_recording_engine(): + """Returns (engine, fired) where fired collects (event_type, EVENT_DATA-at-fire-time).""" + engine = Mock() + fired = [] + + def _record(event_type, fl_ctx): + fired.append((event_type, fl_ctx.get_prop(FLContextKey.EVENT_DATA))) + + engine.fire_event.side_effect = _record + return engine, fired + + +class _StubBackend(ClientAPIBackendSpec): + """A minimal working backend to verify executor-to-backend plumbing.""" + + def __init__(self): + self.calls = [] + self.result = make_reply(ReturnCode.OK) + self.context = None + + def initialize(self, context, fl_ctx): + self.context = context + self.calls.append("initialize") + + def execute(self, task_name, shareable, fl_ctx, abort_signal): + self.calls.append(("execute", task_name)) + return self.result + + def handle_event(self, event_type, fl_ctx): + self.calls.append(("handle_event", event_type)) + + def finalize(self, fl_ctx): + self.calls.append("finalize") + + +class _StubbedInProcessExecutor(ClientAPIExecutor): + """ClientAPIExecutor with a working in_process backend factory (as EX-3 will provide).""" + + def __init__(self, backend, **kwargs): + super().__init__(**kwargs) + self._stub_backend = backend + + def _create_in_process_backend(self): + return self._stub_backend + + +class TestConstructorValidation: + @pytest.mark.parametrize("kwargs", list(MODE_KWARGS.values()), ids=list(MODE_KWARGS.keys())) + def test_valid_minimal_construction(self, kwargs): + executor = ClientAPIExecutor(**kwargs) + assert isinstance(executor, Executor) + assert executor._execution_mode == kwargs["execution_mode"] + + def test_valid_attach_with_attach_args(self): + executor = ClientAPIExecutor(execution_mode="attach", attach_timeout=60.0, allow_reconnect=True) + assert executor._attach_timeout == 60.0 + assert executor._allow_reconnect is True + + def test_valid_full_surface(self): + executor = ClientAPIExecutor( + execution_mode="external_process", + command="torchrun --nproc_per_node=2 custom/train.py", + launch_once=False, + launch_timeout=120.0, + shutdown_timeout=60.0, + stop_grace_period=10.0, + heartbeat_interval=2.0, + heartbeat_timeout=20.0, + task_wait_timeout=300.0, + result_wait_timeout=600.0, + train_task_name="my_train", + evaluate_task_name="my_eval", + submit_model_task_name="my_submit", + train_with_evaluation=True, + memory_gc_rounds=5, + cuda_empty_cache=True, + ) + assert executor._command == "torchrun --nproc_per_node=2 custom/train.py" + assert executor._launch_once is False + assert executor._train_task_name == "my_train" + assert executor._evaluate_task_name == "my_eval" + assert executor._submit_model_task_name == "my_submit" + assert executor._train_with_evaluation is True + assert executor._memory_gc_rounds == 5 + assert executor._cuda_empty_cache is True + + def test_task_name_and_memory_defaults(self): + # findings 9 + 10: rank-contract task names and memory-management knobs are frozen with the + # legacy executors' defaults, valid in every mode. + executor = ClientAPIExecutor(execution_mode="in_process") + assert executor._train_task_name == AppConstants.TASK_TRAIN + assert executor._evaluate_task_name == AppConstants.TASK_VALIDATION + assert executor._submit_model_task_name == AppConstants.TASK_SUBMIT_MODEL + assert executor._train_with_evaluation is False + assert executor._memory_gc_rounds == 0 + assert executor._cuda_empty_cache is False + + def test_in_process_accepts_task_script(self): + # finding 8: in_process names its script via task_script_path/args (command names the + # external_process trainer). + executor = ClientAPIExecutor( + execution_mode="in_process", task_script_path="custom/train.py", task_script_args="--epochs 3" + ) + assert executor._task_script_path == "custom/train.py" + assert executor._task_script_args == "--epochs 3" + assert executor._command is None + + @pytest.mark.parametrize("bad_mode", ["IN_PROCESS", "In_Process", "subprocess", "launch", "", None, 1]) + def test_unknown_execution_mode_rejected(self, bad_mode): + with pytest.raises(ValueError, match="invalid execution_mode"): + ClientAPIExecutor(execution_mode=bad_mode) + + def test_unknown_execution_mode_message_names_valid_modes(self): + with pytest.raises(ValueError) as exc_info: + ClientAPIExecutor(execution_mode="bogus") + for mode in ALL_EXECUTION_MODES: + assert mode in str(exc_info.value) + + @pytest.mark.parametrize("mode", ["in_process", "attach"]) + def test_command_rejected_for_non_external_process_modes(self, mode): + with pytest.raises(ValueError, match="command is only valid for execution_mode 'external_process'"): + ClientAPIExecutor(execution_mode=mode, command="python custom/train.py") + + @pytest.mark.parametrize("command", [None, ""]) + def test_external_process_requires_command(self, command): + with pytest.raises(ValueError, match="'external_process' requires a non-empty command"): + ClientAPIExecutor(execution_mode="external_process", command=command) + + @pytest.mark.parametrize( + "kwargs", + [ + {"execution_mode": "in_process"}, + {"execution_mode": "external_process", "command": "python custom/train.py"}, + ], + ids=["in_process", "external_process"], + ) + def test_attach_timeout_rejected_for_non_attach_modes(self, kwargs): + with pytest.raises(ValueError, match="attach_timeout is only valid for execution_mode 'attach'"): + ClientAPIExecutor(attach_timeout=30.0, **kwargs) + + @pytest.mark.parametrize( + "kwargs", + [ + {"execution_mode": "in_process"}, + {"execution_mode": "external_process", "command": "python custom/train.py"}, + ], + ids=["in_process", "external_process"], + ) + def test_allow_reconnect_rejected_for_non_attach_modes(self, kwargs): + with pytest.raises(ValueError, match="allow_reconnect is only valid for execution_mode 'attach'"): + ClientAPIExecutor(allow_reconnect=True, **kwargs) + + @pytest.mark.parametrize("mode", ["in_process", "attach"]) + @pytest.mark.parametrize("command", ["", " ", "\t"]) + def test_empty_command_treated_as_unset_for_non_external_modes(self, mode, command): + # finding 1: an empty/whitespace command is "unset", not a wrong-mode command, so it must + # not be rejected with the misleading "only valid for external_process" message. + executor = ClientAPIExecutor(execution_mode=mode, command=command) + assert executor._command is None + + @pytest.mark.parametrize("command", ["", " "]) + def test_external_process_rejects_empty_command(self, command): + # finding 1: the same normalization must keep external_process requiring a real command. + with pytest.raises(ValueError, match="'external_process' requires a non-empty command"): + ClientAPIExecutor(execution_mode="external_process", command=command) + + @pytest.mark.parametrize("mode", ["external_process", "attach"]) + def test_task_script_path_rejected_for_non_in_process_modes(self, mode): + # finding 8: task_script_path names the in_process script only. + kwargs = dict(MODE_KWARGS[mode]) + with pytest.raises(ValueError, match="task_script_path is only valid for execution_mode 'in_process'"): + ClientAPIExecutor(task_script_path="custom/train.py", **kwargs) + + @pytest.mark.parametrize("mode", ["external_process", "attach"]) + def test_task_script_args_rejected_for_non_in_process_modes(self, mode): + kwargs = dict(MODE_KWARGS[mode]) + with pytest.raises(ValueError, match="task_script_args is only valid for execution_mode 'in_process'"): + ClientAPIExecutor(task_script_args="--epochs 3", **kwargs) + + @pytest.mark.parametrize("mode", ["in_process", "attach"]) + def test_launch_once_non_default_rejected_for_non_external_modes(self, mode): + # finding 3: external_process-only knobs must be rejected (not silently ignored) when set + # to a non-default value in a mode that ignores them. + kwargs = dict(MODE_KWARGS[mode]) + with pytest.raises(ValueError, match="launch_once is only valid for execution_mode 'external_process'"): + ClientAPIExecutor(launch_once=False, **kwargs) + + @pytest.mark.parametrize("mode", ["in_process", "attach"]) + def test_launch_timeout_rejected_for_non_external_modes(self, mode): + kwargs = dict(MODE_KWARGS[mode]) + with pytest.raises(ValueError, match="launch_timeout is only valid for execution_mode 'external_process'"): + ClientAPIExecutor(launch_timeout=120.0, **kwargs) + + @pytest.mark.parametrize("mode", ["in_process", "attach"]) + def test_shutdown_timeout_rejected_for_non_external_modes(self, mode): + kwargs = dict(MODE_KWARGS[mode]) + with pytest.raises(ValueError, match="shutdown_timeout is only valid for execution_mode 'external_process'"): + ClientAPIExecutor(shutdown_timeout=60.0, **kwargs) + + @pytest.mark.parametrize("mode", ["in_process", "attach"]) + def test_stop_grace_period_non_default_rejected_for_non_external_modes(self, mode): + kwargs = dict(MODE_KWARGS[mode]) + with pytest.raises(ValueError, match="stop_grace_period is only valid for execution_mode 'external_process'"): + ClientAPIExecutor(stop_grace_period=10.0, **kwargs) + + @pytest.mark.parametrize("arg", ["heartbeat_interval", "heartbeat_timeout"]) + def test_heartbeat_non_default_rejected_for_in_process(self, arg): + # finding 3: there is no session heartbeat in_process, so a non-default heartbeat knob is + # dead there and must be rejected (valid for external_process/attach). + with pytest.raises(ValueError, match=f"{arg} is only valid for execution_mode 'external_process' or 'attach'"): + ClientAPIExecutor(execution_mode="in_process", **{arg: 99.0}) + + @pytest.mark.parametrize("mode", ["external_process", "attach"]) + def test_heartbeat_accepted_for_out_of_process_modes(self, mode): + kwargs = dict(MODE_KWARGS[mode]) + executor = ClientAPIExecutor(heartbeat_interval=2.0, heartbeat_timeout=20.0, **kwargs) + assert executor._heartbeat_interval == 2.0 + assert executor._heartbeat_timeout == 20.0 + + @pytest.mark.parametrize("mode", ["in_process", "attach"]) + def test_external_process_defaults_accepted_for_other_modes(self, mode): + # The frozen defaults for external_process-only knobs are indistinguishable from "not set" + # and must be accepted in every mode. + kwargs = dict(MODE_KWARGS[mode]) + ClientAPIExecutor(launch_once=True, stop_grace_period=30.0, **kwargs) + + @pytest.mark.parametrize("mode", ["in_process", "external_process"]) + def test_allow_reconnect_default_value_accepted_for_non_attach_modes(self, mode): + # allow_reconnect=False equals the frozen default and is indistinguishable from + # "not set", so it must not be rejected. + kwargs = dict(MODE_KWARGS[mode]) + ClientAPIExecutor(allow_reconnect=False, **kwargs) + + @pytest.mark.parametrize("mode", ["in_process", "external_process"]) + @pytest.mark.parametrize("falsy", [None, 0]) + def test_allow_reconnect_falsy_values_accepted_for_non_attach_modes(self, mode, falsy): + # finding 2: the wrong-mode check uses `if allow_reconnect` (truthy), not + # `is not False`, so falsy-but-not-False values (None, 0, numpy.bool_(False)) are treated + # as "not set" instead of misfiring. + kwargs = dict(MODE_KWARGS[mode]) + ClientAPIExecutor(allow_reconnect=falsy, **kwargs) + + +class TestDispatch: + """All three modes are skeleton-only in EX-2: resolving the backend at START_RUN must fail + the job cleanly (system_panic naming the mode), and execute() must reply with an error instead + of hanging.""" + + @pytest.mark.parametrize("mode", list(ALL_EXECUTION_MODES)) + def test_backend_factory_raises_not_implemented(self, mode): + # finding 7: user-facing message must not carry an internal plan id (EX-3/EP-4/AT-2); it + # names the mode and says "not yet implemented". + executor = ClientAPIExecutor(**MODE_KWARGS[mode]) + with pytest.raises(NotImplementedError, match="not yet implemented") as exc_info: + executor._create_backend() + message = str(exc_info.value) + assert mode in message + for plan_id in ("EX-3", "EP-4", "AT-2"): + assert plan_id not in message + + @pytest.mark.parametrize("mode", list(ALL_EXECUTION_MODES)) + def test_start_run_panics_naming_the_mode(self, mode): + # findings 6 + 7: the panic reason names the mode directly (not via secure_format_exception, + # so it is robust whether or not NVFLARE_SECURE_LOGGING is set) and carries no plan id. + executor = ClientAPIExecutor(**MODE_KWARGS[mode]) + engine, fired = _make_recording_engine() + fl_ctx = _make_fl_ctx(engine) + + executor.handle_event(EventType.START_RUN, fl_ctx) + + panics = [(event_type, data) for event_type, data in fired if event_type == EventType.FATAL_SYSTEM_ERROR] + assert len(panics) == 1, f"expected exactly one system_panic, got events: {fired}" + reason = panics[0][1] + assert mode in reason + for plan_id in ("EX-3", "EP-4", "AT-2"): + assert plan_id not in reason + + @pytest.mark.parametrize("mode", list(ALL_EXECUTION_MODES)) + def test_execute_without_backend_returns_error_reply_not_hang(self, mode): + executor = ClientAPIExecutor(**MODE_KWARGS[mode]) + engine, _ = _make_recording_engine() + fl_ctx = _make_fl_ctx(engine) + executor.handle_event(EventType.START_RUN, fl_ctx) # backend init fails -> panic + + reply = executor.execute("train", Shareable(), fl_ctx, Signal()) + + assert isinstance(reply, Shareable) + assert reply.get_return_code() == ReturnCode.EXECUTION_EXCEPTION + + def test_end_run_without_backend_does_not_raise(self): + executor = ClientAPIExecutor(execution_mode="in_process") + engine, fired = _make_recording_engine() + fl_ctx = _make_fl_ctx(engine) + executor.handle_event(EventType.END_RUN, fl_ctx) + assert not any(event_type == EventType.FATAL_SYSTEM_ERROR for event_type, _ in fired) + + +class TestBackendPlumbing: + """With a working backend registered (as EX-3/EP-4/AT-2 will do), the executor drives the + ClientAPIBackendSpec lifecycle: initialize at START_RUN, execute per task, handle_event for + other events, finalize at END_RUN.""" + + def _make_started_executor(self): + backend = _StubBackend() + executor = _StubbedInProcessExecutor(backend, execution_mode="in_process") + engine, fired = _make_recording_engine() + fl_ctx = _make_fl_ctx(engine) + executor.handle_event(EventType.START_RUN, fl_ctx) + return executor, backend, fl_ctx, fired + + def test_start_run_initializes_backend_without_panic(self): + executor, backend, fl_ctx, fired = self._make_started_executor() + assert backend.calls == ["initialize"] + assert not any(event_type == EventType.FATAL_SYSTEM_ERROR for event_type, _ in fired) + + def test_execute_delegates_to_backend(self): + executor, backend, fl_ctx, _ = self._make_started_executor() + reply = executor.execute("train", Shareable(), fl_ctx, Signal()) + assert reply is backend.result + assert ("execute", "train") in backend.calls + + def test_other_events_forwarded_to_backend(self): + executor, backend, fl_ctx, _ = self._make_started_executor() + executor.handle_event(EventType.ABOUT_TO_END_RUN, fl_ctx) + assert ("handle_event", EventType.ABOUT_TO_END_RUN) in backend.calls + + def test_end_run_finalizes_and_clears_backend(self): + executor, backend, fl_ctx, _ = self._make_started_executor() + executor.handle_event(EventType.END_RUN, fl_ctx) + assert "finalize" in backend.calls + reply = executor.execute("train", Shareable(), fl_ctx, Signal()) + assert reply.get_return_code() == ReturnCode.EXECUTION_EXCEPTION + + def test_backend_exception_in_execute_becomes_error_reply(self): + executor, backend, fl_ctx, _ = self._make_started_executor() + backend.execute = Mock(side_effect=RuntimeError("boom")) + reply = executor.execute("train", Shareable(), fl_ctx, Signal()) + assert reply.get_return_code() == ReturnCode.EXECUTION_EXCEPTION + + def test_unsafe_job_error_propagates_out_of_execute(self): + # ClientRunner has dedicated UNSAFE_JOB handling for UnsafeJobError; the executor must + # let it propagate instead of masking it as EXECUTION_EXCEPTION. + executor, backend, fl_ctx, _ = self._make_started_executor() + backend.execute = Mock(side_effect=UnsafeJobError("unsafe")) + with pytest.raises(UnsafeJobError): + executor.execute("train", Shareable(), fl_ctx, Signal()) + + def test_unsafe_component_error_becomes_execution_exception(self): + # UnsafeComponentError has no dedicated ClientRunner handling, so it takes the generic + # error path (its existing behavior) rather than propagating. + executor, backend, fl_ctx, _ = self._make_started_executor() + backend.execute = Mock(side_effect=UnsafeComponentError("bad component")) + reply = executor.execute("train", Shareable(), fl_ctx, Signal()) + assert reply.get_return_code() == ReturnCode.EXECUTION_EXCEPTION + + def test_backend_non_shareable_result_becomes_error_reply(self): + executor, backend, fl_ctx, _ = self._make_started_executor() + backend.result = {"not": "a shareable"} + reply = executor.execute("train", Shareable(), fl_ctx, Signal()) + assert reply.get_return_code() == ReturnCode.EXECUTION_EXCEPTION + + def test_backend_receives_config_context(self): + # finding 11: initialize() receives a frozen ClientAPIBackendContext carrying the executor + # config and a back-reference to the executor (for fire_log_analytics/logging). + backend = _StubBackend() + executor = _StubbedInProcessExecutor( + backend, + execution_mode="in_process", + task_script_path="custom/train.py", + train_task_name="my_train", + memory_gc_rounds=7, + ) + engine, _ = _make_recording_engine() + executor.handle_event(EventType.START_RUN, _make_fl_ctx(engine)) + + ctx = backend.context + assert isinstance(ctx, ClientAPIBackendContext) + assert ctx.executor is executor + assert ctx.execution_mode == "in_process" + assert ctx.task_script_path == "custom/train.py" + assert ctx.train_task_name == "my_train" + assert ctx.memory_gc_rounds == 7 + + def test_backend_context_is_frozen(self): + backend = _StubBackend() + executor = _StubbedInProcessExecutor(backend, execution_mode="in_process") + executor.handle_event(EventType.START_RUN, _make_fl_ctx(_make_recording_engine()[0])) + with pytest.raises(Exception): + backend.context.execution_mode = "attach" + + def test_backend_spec_is_abstract(self): + with pytest.raises(TypeError): + ClientAPIBackendSpec() + + +class TestAnalyticsOwnership: + """fire_log_analytics is the executor-owned LOG-to-analytics conversion surface.""" + + def _fire(self, fire_fed_event: bool): + executor = ClientAPIExecutor(execution_mode="in_process") + executor._analytics_fire_fed_event = fire_fed_event + engine = Mock() + scopes = [] + engine.fire_event.side_effect = lambda event_type, ctx: scopes.append( + (event_type, ctx.get_prop(FLContextKey.EVENT_SCOPE)) + ) + fl_ctx = _make_fl_ctx(engine) + dxo = create_analytic_dxo(tag="loss", value=0.1, data_type=AnalyticsDataType.SCALAR) + executor.fire_log_analytics(fl_ctx, dxo) + return scopes + + def test_default_path_fires_local_analytics_event(self): + # Default: local un-prefixed event; ConvertToFedEvent (added by BaseFedJob) is + # responsible for re-firing it as "fed.analytix_log_stats". + scopes = self._fire(fire_fed_event=False) + assert scopes == [(ANALYTIC_EVENT_TYPE, EventScope.LOCAL)] + + def test_fed_path_fires_federation_scoped_event(self): + # finding 4: the fed path must fire the already-"fed."-prefixed event name so it lands on + # the same server-side event as MetricRelay (job_config/script_runner.py) and flower_job.py + # ("fed.analytix_log_stats"); firing the un-prefixed name federation-scoped would miss every + # consumer listening on "fed.analytix_log_stats". + assert FED_ANALYTIC_EVENT_TYPE == "fed.analytix_log_stats" + scopes = self._fire(fire_fed_event=True) + assert scopes == [(FED_ANALYTIC_EVENT_TYPE, EventScope.FEDERATION)] + + +class TestSurfaceFreeze: + """Interface freeze #2: accidental renames/reorders of the constructor surface fail CI.""" + + def test_constructor_parameter_names_match_frozen_list(self): + params = list(inspect.signature(ClientAPIExecutor.__init__).parameters) + assert params[0] == "self" + assert params[1:] == FROZEN_CONSTRUCTOR_PARAMS + + def test_execution_mode_is_required(self): + sig = inspect.signature(ClientAPIExecutor.__init__) + assert sig.parameters["execution_mode"].default is inspect.Parameter.empty + + def test_frozen_defaults(self): + # launch_once/heartbeat_interval/heartbeat_timeout/allow_reconnect defaults are + # normative from the design's Configuration Surface; the rest are frozen by EX-2. + sig = inspect.signature(ClientAPIExecutor.__init__) + expected_defaults = { + "command": None, + "task_script_path": None, + "task_script_args": "", + "launch_once": True, + "launch_timeout": None, + "shutdown_timeout": None, + "stop_grace_period": 30.0, + "heartbeat_interval": 5.0, + "heartbeat_timeout": 30.0, + "task_wait_timeout": None, + "result_wait_timeout": None, + "train_task_name": AppConstants.TASK_TRAIN, + "evaluate_task_name": AppConstants.TASK_VALIDATION, + "submit_model_task_name": AppConstants.TASK_SUBMIT_MODEL, + "train_with_evaluation": False, + "memory_gc_rounds": 0, + "cuda_empty_cache": False, + "attach_timeout": None, + "allow_reconnect": False, + } + actual_defaults = {name: p.default for name, p in sig.parameters.items() if name != "self"} + actual_defaults.pop("execution_mode") + assert actual_defaults == expected_defaults + + def test_module_path_is_pinned(self): + # The design's example job config pins this exact path. + assert ClientAPIExecutor.__module__ == "nvflare.app_common.executors.client_api_executor" + + def test_execution_mode_values_are_frozen(self): + assert ALL_EXECUTION_MODES == ("in_process", "external_process", "attach")