Make Recipe job storage private

nvflare/app_common/np/recipes/fedavg.py

@@ -208,8 +208,8 @@ def add_cse_validator_if_needed(self):
 
         # 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 hasattr(self._job, "_deploy_map"):
+            for target, app in self._job._deploy_map.items():
                 if target == "server":
                     continue
 
@@ -229,4 +229,4 @@ def add_cse_validator_if_needed(self):
         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._job.to_clients(validator, 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,8 @@ 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)
+        assert isinstance(self._job, EdgeJob)
+        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,7 @@ def __init__(self, job: FedJob):
         Args:
             job: the job that implements the recipe.
         """
-        self.job = job
+        self._job = job
         self._helper_per_site_config = None
         self._tensor_streaming_added = False
         self._cse_added = False
@@ -335,7 +335,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 +346,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 +358,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 +398,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 +408,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 +437,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 +447,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 +480,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 +569,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 +581,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 +602,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 +623,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 +662,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 +712,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 +753,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 +765,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 +797,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 +816,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)

nvflare/recipe/utils.py

@@ -211,10 +211,10 @@ def _normalize_recipe_meta_value(key: JobMetaKey, key_str: str, value: Any) -> A
 
 
 def _get_recipe_job_config(recipe: Recipe) -> FedJobConfig:
-    job = getattr(recipe, "job", None)
+    job = getattr(recipe, "_job", None)
     job_config = getattr(job, "job", None)
     if not isinstance(job_config, FedJobConfig):
-        raise TypeError("recipe must provide a FedJob through recipe.job")
+        raise TypeError("recipe must be backed by a FedJob")
     return job_config
 
 
@@ -378,7 +378,7 @@ def add_experiment_tracking(
             tracking_config["kw_args"] = kw_args
         elif not isinstance(kw_args, dict):
             raise TypeError(f"MLflow kw_args must be a dict, got {type(kw_args).__name__}")
-        recipe_name = getattr(recipe, "name", None) or getattr(recipe.job, "name", None) or "nvflare"
+        recipe_name = getattr(recipe, "name", None) or getattr(recipe._job, "name", None) or "nvflare"
         kw_args.setdefault("experiment_name", f"{recipe_name}-experiment")
 
     if not server_side and not client_side:
@@ -407,7 +407,7 @@ def add_experiment_tracking(
         if tracking_type == "mlflow":
             server_config["kw_args"].setdefault("run_name", f"{recipe_name}-Server")
         receiver = receiver_class(**server_config)
-        recipe.job.to_server(receiver, "receiver")
+        recipe._job.to_server(receiver, "receiver")
 
     # Add client-side tracking
     if client_side:
@@ -455,7 +455,7 @@ def add_final_global_evaluation(
     from nvflare.app_opt.pt.file_model_locator import PTFileModelLocator
     from nvflare.job_config.script_runner import FrameworkType
 
-    if getattr(recipe, "_cse_added", False) or _has_cross_site_eval_workflow(recipe.job):
+    if getattr(recipe, "_cse_added", False) or _has_cross_site_eval_workflow(recipe._job):
         raise RuntimeError("a cross-site evaluation workflow is already configured for this recipe")
 
     if getattr(recipe, "framework", None) != FrameworkType.PYTORCH:
@@ -469,7 +469,7 @@ def add_final_global_evaluation(
         if not participating_clients:
             raise ValueError("participating_clients must not be empty; use None to evaluate on all clients")
 
-    comp_ids = getattr(recipe.job, "comp_ids", None)
+    comp_ids = getattr(recipe._job, "comp_ids", None)
     if not isinstance(comp_ids, dict):
         raise ValueError("final global evaluation requires a recipe that tracks component IDs")
 
@@ -478,15 +478,15 @@ def add_final_global_evaluation(
         persistor_id = comp_ids.get("persistor_id", "")
         if not persistor_id:
             raise ValueError("final global evaluation requires a PyTorch model persistor")
-        model_locator_id = recipe.job.to_server(
+        model_locator_id = recipe._job.to_server(
             PTFileModelLocator(pt_persistor_id=persistor_id), id="final_model_locator"
         )
         if not isinstance(model_locator_id, str) or not model_locator_id:
             raise RuntimeError("failed to register the final global model locator")
         comp_ids["locator_id"] = model_locator_id
 
-    recipe.job.to_server(ValidationJsonGenerator())
-    recipe.job.to_server(
+    recipe._job.to_server(ValidationJsonGenerator())
+    recipe._job.to_server(
         CrossSiteModelEval(
             model_locator_id=model_locator_id,
             submit_model_task_name="",
@@ -621,7 +621,7 @@ def add_cross_site_evaluation(
     # Idempotency check: prevent multiple calls on the same recipe.
     # Keep the explicit flag fast-path, but also verify server workflow state so
     # protection remains effective even if dynamic attributes are lost.
-    if getattr(recipe, "_cse_added", False) or _has_cross_site_eval_workflow(recipe.job):
+    if getattr(recipe, "_cse_added", False) or _has_cross_site_eval_workflow(recipe._job):
         name = recipe.name if hasattr(recipe, "name") else "cross-site-evaluation job"
         raise RuntimeError(
             f"Cross-site evaluation has already been added to recipe '{name}'. "
@@ -672,12 +672,12 @@ def add_cross_site_evaluation(
     locator_kwargs = {}
     if locator_config["persistor_param"] is not None:
         # For frameworks requiring persistor_id (PyTorch, TensorFlow), get it from comp_ids
-        if hasattr(recipe.job, "comp_ids"):
-            persistor_id = recipe.job.comp_ids.get("persistor_id", "")
+        if hasattr(recipe._job, "comp_ids"):
+            persistor_id = recipe._job.comp_ids.get("persistor_id", "")
             if not persistor_id:
                 raise ValueError(
                     f"Cross-site evaluation requires a persistor for {framework_to_locator[framework]} recipes, "
-                    f"but no persistor_id found in recipe.job.comp_ids. "
+                    "but no persistor_id was found in the generated job. "
                     f"Ensure your recipe includes a model to create a persistor."
                 )
             locator_kwargs[locator_config["persistor_param"]] = persistor_id
@@ -688,10 +688,10 @@ def add_cross_site_evaluation(
             )
 
     model_locator = locator_class(**locator_kwargs)
-    model_locator_id = recipe.job.to_server(model_locator)
+    model_locator_id = recipe._job.to_server(model_locator)
 
     # Add validation JSON generator
-    recipe.job.to_server(ValidationJsonGenerator())
+    recipe._job.to_server(ValidationJsonGenerator())
 
     # Create and add cross-site evaluation controller
     eval_controller = CrossSiteModelEval(
@@ -700,7 +700,7 @@ def add_cross_site_evaluation(
         validation_timeout=validation_timeout,
         participating_clients=participating_clients,
     )
-    recipe.job.to_server(eval_controller)
+    recipe._job.to_server(eval_controller)
 
     # Let recipe handle framework-specific validator setup if needed
     # NumPy recipes implement add_cse_validator_if_needed() to add NPValidator automatically

tests/integration_test/fast/study_session_test.py

@@ -543,8 +543,8 @@ def test_multistudy_prod_env_propagates_study_and_runtime_meta(self, multi_study
 
         recipe = _make_numpy_recipe("prod-env-study-a")
         env = ProdEnv(startup_kit_location=admin_root, username=MAIN_ADMIN, study="study-a", login_timeout=2.0)
-        recipe.process_env(env)
-        job_id = env.deploy(recipe.job)
+        run = recipe.execute(env)
+        job_id = run.job_id
         study_a_session = new_secure_session(MAIN_ADMIN, admin_root, study="study-a")
         multi_study_system["sessions"].append(study_a_session)
 

tests/integration_test/slow/experiment_tracking_recipes_test.py

@@ -87,9 +87,9 @@ def mlflow_client_model_path(self):
 
     def _add_model_to_apps(self, recipe, model_path=None):
         model_path = model_path or self.model_path
-        recipe.job.add_file_to_server(model_path)
+        recipe._job.add_file_to_server(model_path)
         for site_name in ("site-1", "site-2"):
-            recipe.job.add_file_to(model_path, site_name)
+            recipe._job.add_file_to(model_path, site_name)
 
     def _assert_exported_client_configs_have_executors(self, recipe, export_root: str):
         recipe.export(export_root)