Make Recipe job storage private

nvflare/app_common/np/recipes/fedavg.py

@@ -193,40 +193,17 @@ def _setup_model_and_persistor(self, job) -> str:
         return ""
 
     def add_cse_validator_if_needed(self):
-        """Add NPValidator for cross-site evaluation if not already configured.
+        """Add NPValidator for cross-site evaluation.
 
         NumPy recipes need specialized NPValidator because:
         - NumPy training scripts typically only handle training tasks
         - Wildcard executors (tasks=["*"]) don't actually implement validation
         - Cross-site evaluation requires dedicated validation component
 
-        This method checks if a dedicated validator is already configured.
-        If only wildcard executors exist, adds NPValidator.
+        ``add_cross_site_evaluation()`` invokes this hook only after its
+        idempotency check, so each successful CSE augmentation adds one validator.
         """
         from nvflare.app_common.app_constant import AppConstants
         from nvflare.app_common.np.np_validator import NPValidator
 
-        # Check if validation task is explicitly configured (not just via wildcard)
-        has_explicit_validator = False
-        if hasattr(self.job, "_deploy_map"):
-            for target, app in self.job._deploy_map.items():
-                if target == "server":
-                    continue
-
-                if hasattr(app, "app_config") and hasattr(app.app_config, "executors"):
-                    for executor_def in app.app_config.executors:
-                        if hasattr(executor_def, "tasks"):
-                            try:
-                                # Check if validation is explicitly listed (not just wildcard)
-                                if AppConstants.TASK_VALIDATION in executor_def.tasks:
-                                    has_explicit_validator = True
-                                    break
-                            except (TypeError, AttributeError):
-                                continue
-                if has_explicit_validator:
-                    break
-
-        if not has_explicit_validator:
-            # No explicit validator found - add NPValidator for cross-site evaluation
-            validator = NPValidator()
-            self.job.to_clients(validator, tasks=[AppConstants.TASK_VALIDATION])
+        self._add_to_client_apps(NPValidator(), tasks=[AppConstants.TASK_VALIDATION])

nvflare/app_opt/sklearn/recipes/kmeans.py

@@ -180,4 +180,4 @@ def __init__(
             per_site_config=per_site_config,
             key_metric=key_metric,
         )
-        self.job.to_server(assembler, id=assembler_id)
+        self._job.to_server(assembler, id=assembler_id)

nvflare/app_opt/sklearn/recipes/svm.py

@@ -181,4 +181,4 @@ def __init__(
 
         # Add the SVMAssembler as a component to the job
         # CollectAndAssembleModelAggregator will fetch it by ID at runtime
-        self.job.to_server(self._svm_assembler, id="svm_assembler")
+        self._job.to_server(self._svm_assembler, id="svm_assembler")

nvflare/app_opt/xgboost/recipes/bagging.py

@@ -223,8 +223,8 @@ def __init__(
             )
 
         # Configure the job
-        self.job = self.configure()
-        Recipe.__init__(self, self.job)
+        job = self.configure()
+        Recipe.__init__(self, job)
 
     def configure(self):
         """Configure the federated job for XGBoost tree-based training."""

nvflare/app_opt/xgboost/recipes/histogram.py

@@ -184,8 +184,8 @@ def __init__(
             )
 
         # Configure the job
-        self.job = self.configure()
-        Recipe.__init__(self, self.job)
+        job = self.configure()
+        Recipe.__init__(self, job)
 
     def configure(self):
         """Configure the federated job for XGBoost histogram-based training."""

nvflare/app_opt/xgboost/recipes/vertical.py

@@ -255,8 +255,8 @@ def __init__(
             self.client_ranks = {client_name: rank for rank, client_name in enumerate(ranked_clients)}
 
         # Configure the job
-        self.job = self.configure()
-        Recipe.__init__(self, self.job)
+        job = self.configure()
+        Recipe.__init__(self, job)
 
     def configure(self):
         """Configure the federated job for vertical XGBoost training."""

nvflare/edge/tools/edge_fed_buff_recipe.py

@@ -303,8 +303,9 @@ def process_env(self, env: ExecEnv):
                 f"invalid {DEVICE_SIMULATION_ENV_KEY} in env: expect SimulationConfig but got {type(simulation_config)}"
             )
 
-        assert isinstance(self.job, EdgeJob)
-        self._configure_simulation(self.job, simulation_config)
+        if not isinstance(self._job, EdgeJob):
+            raise TypeError(f"expected recipe job to be an EdgeJob, got {type(self._job).__name__}")
+        self._configure_simulation(self._job, simulation_config)
 
     def create_job(self) -> EdgeJob:
         """Create a new EdgeJob instance for cross-edge federated learning.

nvflare/recipe/spec.py

@@ -251,7 +251,8 @@ def __init__(self, job: FedJob):
         Args:
             job: the job that implements the recipe.
         """
-        self.job = job
+        self._job = job
+        self.name = job.name
         self._helper_per_site_config = None
         self._tensor_streaming_added = False
         self._cse_added = False
@@ -335,7 +336,7 @@ def configured_sites(self) -> List[str]:
 
     def _snapshot_additional_params(self) -> Dict[str, Dict]:
         snapshot = {}
-        deploy_map = getattr(self.job, "_deploy_map", {})
+        deploy_map = getattr(self._job, "_deploy_map", {})
         for target, app in deploy_map.items():
             app_config = getattr(app, "app_config", None)
             if app_config is None:
@@ -346,7 +347,7 @@ def _snapshot_additional_params(self) -> Dict[str, Dict]:
         return snapshot
 
     def _restore_additional_params(self, snapshot: Dict[str, Dict]) -> None:
-        deploy_map = getattr(self.job, "_deploy_map", {})
+        deploy_map = getattr(self._job, "_deploy_map", {})
         for target, app in deploy_map.items():
             app_config = getattr(app, "app_config", None)
             if app_config is None:
@@ -358,7 +359,7 @@ def _restore_additional_params(self, snapshot: Dict[str, Dict]) -> None:
                 params.update(original)
 
     def _replace_additional_params_for_targets(self, targets: List[str], new_params: dict) -> None:
-        deploy_map = getattr(self.job, "_deploy_map", {})
+        deploy_map = getattr(self._job, "_deploy_map", {})
         for target in targets:
             app = deploy_map.get(target)
             if app is None:
@@ -398,7 +399,7 @@ def _temporary_exec_params(
         try:
             if server_exec_params is not None:
                 if server_exec_params:
-                    self.job.to_server(server_exec_params)
+                    self._job.to_server(server_exec_params)
                 else:
                     # Preserve the long-standing "empty dict means temporarily clear params"
                     # behavior rather than treating {} as a no-op.
@@ -408,7 +409,7 @@ def _temporary_exec_params(
                 if client_exec_params:
                     self._add_to_client_apps(client_exec_params)
                 else:
-                    client_targets = [target for target in getattr(self.job, "_deploy_map", {}) if target != "server"]
+                    client_targets = [target for target in getattr(self._job, "_deploy_map", {}) if target != "server"]
                     self._replace_additional_params_for_targets(client_targets, {})
             yield
         finally:
@@ -437,7 +438,7 @@ def _add_to_client_apps(self, obj, clients: Optional[List[str]] = None, **kwargs
 
         # FedJob has no public API to list per-site deploy targets, so we inspect
         # private deploy map to preserve existing per-site client topology.
-        deploy_map = getattr(self.job, "_deploy_map", {})
+        deploy_map = getattr(self._job, "_deploy_map", {})
         existing_client_sites = [
             target
             for target in deploy_map.keys()
@@ -447,9 +448,9 @@ def _add_to_client_apps(self, obj, clients: Optional[List[str]] = None, **kwargs
         if clients is None:
             if existing_client_sites:
                 for site in existing_client_sites:
-                    self.job.to(obj, site, **kwargs)
+                    self._job.to(obj, site, **kwargs)
             else:
-                self.job.to_clients(obj, **kwargs)
+                self._job.to_clients(obj, **kwargs)
         else:
             # A bare string would iterate per character; an empty list would be a silent no-op.
             if not isinstance(clients, list):
@@ -480,7 +481,7 @@ def _add_to_client_apps(self, obj, clients: Optional[List[str]] = None, **kwargs
                         f"only for {sorted(existing_client_sites)}"
                     )
             for client in clients:
-                self.job.to(obj, client, **kwargs)
+                self._job.to(obj, client, **kwargs)
 
     def add_client_input_filter(
         self, filter: Filter, tasks: Optional[List[str]] = None, clients: Optional[List[str]] = None
@@ -569,7 +570,7 @@ def add_server_output_filter(self, filter: Filter, tasks: Optional[List[str]] =
         Returns: None
 
         """
-        self.job.to_server(filter, filter_type=FilterType.TASK_DATA, tasks=tasks)
+        self._job.to_server(filter, filter_type=FilterType.TASK_DATA, tasks=tasks)
 
     def add_server_input_filter(self, filter: Filter, tasks: Optional[List[str]] = None):
         """Add a filter to server for incoming task result from clients. .
@@ -581,7 +582,7 @@ def add_server_input_filter(self, filter: Filter, tasks: Optional[List[str]] = N
         Returns: None
 
         """
-        self.job.to_server(filter, filter_type=FilterType.TASK_RESULT, tasks=tasks)
+        self._job.to_server(filter, filter_type=FilterType.TASK_RESULT, tasks=tasks)
 
     def add_server_config(self, config: Dict):
         """Add top-level configuration parameters to config_fed_server.json.
@@ -602,7 +603,7 @@ def add_server_config(self, config: Dict):
 
         warn_on_potential_secrets(config, context="add_server_config config")
         warn_on_unsupported_secret_ref_keys(config, context="add_server_config config")
-        self.job.to_server(config)
+        self._job.to_server(config)
 
     def add_server_file(self, file_path: str):
         """Add a file or directory to server app.
@@ -623,7 +624,7 @@ def add_server_file(self, file_path: str):
         if not isinstance(file_path, str):
             raise TypeError(f"file_path must be a str, got {type(file_path).__name__}")
 
-        self.job.to_server(file_path)
+        self._job.to_server(file_path)
 
     @staticmethod
     def _get_full_class_name(obj):
@@ -662,7 +663,7 @@ def enable_log_streaming(self, *file_names: str) -> None:
 
         for name in file_names:
             self._add_to_client_apps(JobLogStreamer(log_file_name=name))
-        self.job.to_server(JobLogReceiver())
+        self._job.to_server(JobLogReceiver())
 
     def enable_tensor_streaming(
         self,
@@ -712,7 +713,7 @@ def enable_tensor_streaming(
 
         server_tasks = list(tasks) if tasks is not None else None
         client_tasks = list(tasks) if tasks is not None else None
-        self.job.to_server(
+        self._job.to_server(
             TensorServerStreamer(
                 format=format,
                 tasks=server_tasks,
@@ -753,7 +754,7 @@ def add_decomposers(self, decomposers: List[Union[str, Decomposer]]):
             else:
                 raise TypeError(f"decomposer must be str or Decomposer, got {type(d).__name__}")
 
-        self.job.to_server(DecomposerRegister(class_names), id="decomposer_reg")
+        self._job.to_server(DecomposerRegister(class_names), id="decomposer_reg")
         self._add_to_client_apps(DecomposerRegister(class_names), id="decomposer_reg")
 
     def _warn_potential_secrets_in_exported_job(self, job_dir: str) -> None:
@@ -765,7 +766,7 @@ def _warn_potential_secrets_in_exported_job(self, job_dir: str) -> None:
         """
         warn_on_potential_secrets_in_job_dir(
             job_dir=job_dir,
-            job_name=getattr(self.job, "name", None),
+            job_name=getattr(self._job, "name", None),
             context="exported job file",
         )
 
@@ -797,7 +798,7 @@ class docstring for how to pass references instead.
         with self._temporary_exec_params(server_exec_params=server_exec_params, client_exec_params=client_exec_params):
             if env is not None:
                 self.process_env(env)
-            self.job.export_job(job_dir)
+            self._job.export_job(job_dir)
         self._warn_potential_secrets_in_exported_job(job_dir)
 
     def run(
@@ -816,7 +817,7 @@ def run(
         self._warn_potential_secrets_in_params()
         with self._temporary_exec_params(server_exec_params=server_exec_params, client_exec_params=client_exec_params):
             self.process_env(env)
-            job_id = env.deploy(self.job)
+            job_id = env.deploy(self._job)
             from nvflare.recipe.run import Run
 
             return Run(env, job_id)