Cherry pick Recipe API fixes (#4192)

Fixes # .

### Description

[pr 4183](#4183) to main

### Types of changes
<!--- Put an `x` in all the boxes that apply, and remove the not
applicable items -->
- [x] Non-breaking change (fix or new feature that would not break
existing functionality).
- [ ] Breaking change (fix or new feature that would cause existing
functionality to change).
- [ ] New tests added to cover the changes.
- [ ] Quick tests passed locally by running `./runtest.sh`.
- [ ] In-line docstrings updated.
- [ ] Documentation updated.

---------

Co-authored-by: Claude Sonnet 4.5 <noreply@anthropic.com>

Author: ZiyueXu77

examples/advanced/llm_hf/MULTINODE.md

@@ -118,7 +118,7 @@ SLURM Job (2 nodes allocated)
 - Uses `FedAvgRecipe` with `per_site_config` for multi-node setup
 - When `--multi_node` flag is set:
   - Sets command in per_site_config: `"command": "bash custom/client_wrapper.sh"`
-  - Adds wrapper script to job: `recipe.job.to("client_wrapper.sh", site_name)`
+  - Adds wrapper script to job: `recipe.add_client_file("client_wrapper.sh")`
 - Script arguments passed via `"train_args"` in per_site_config
 - No need to handle environment variables in Python
 
@@ -131,7 +131,7 @@ SLURM Job (2 nodes allocated)
 - Uses `CUDA_VISIBLE_DEVICES` to set GPUs. Assumes that they are set as comma-separated list, e.g. "0,1,2,3,4,5,6,7".
 
 **Why this works:**
-- The wrapper script is included in the FL job package via `recipe.job.to("client_wrapper.sh", site_name)`
+- The wrapper script is included in the FL job package via `recipe.add_client_file("client_wrapper.sh")`
 - It's placed in the `custom/` subdirectory of the job workspace
 - Command is set to `bash custom/client_wrapper.sh` in the per_site_config
 - It runs in the same environment as the SLURM job (has access to `srun` and SLURM variables)
@@ -164,7 +164,7 @@ SLURM Job (2 nodes allocated)
    - Uses `FedAvgRecipe` to configure the federated learning job
    - For multi-node mode (`--multi_node` flag):
      - Sets command via `per_site_config`: `"command": "bash custom/client_wrapper.sh"`
-     - Adds wrapper script: `recipe.job.to("client_wrapper.sh", site_name)`
+     - Adds wrapper script: `recipe.add_client_file("client_wrapper.sh")`
    - Includes `client.py` training script automatically via recipe
    - Exports job configuration to specified directory
 

examples/advanced/llm_hf/job.py

@@ -181,30 +181,24 @@ def main():
     )
 
     # Add client params to reduce timeout failures for longer LLM runs
-    for site_name in client_names:
-        client_params = {"get_task_timeout": 300, "submit_task_result_timeout": 300}
-        recipe.job.to(client_params, site_name)
+    recipe.add_client_config({"get_task_timeout": 300, "submit_task_result_timeout": 300})
 
     # Add client_wrapper.sh for multi-node training
     if args.multi_node:
-        for site_name in client_names:
-            recipe.job.to("client_wrapper.sh", site_name)
+        recipe.add_client_file("client_wrapper.sh")
 
     # Add quantization filters if specified
     if args.quantize_mode:
-        from nvflare import FilterType
-
         quantizer = ModelQuantizer(quantization_type=args.quantize_mode.lower())
         dequantizer = ModelDequantizer()
 
-        # Add to server
-        recipe.job.to(quantizer, "server", tasks=["train"], filter_type=FilterType.TASK_DATA)
-        recipe.job.to(dequantizer, "server", tasks=["train"], filter_type=FilterType.TASK_RESULT)
+        # Add to server: quantizer on output, dequantizer on input
+        recipe.add_server_output_filter(quantizer, tasks=["train"])
+        recipe.add_server_input_filter(dequantizer, tasks=["train"])
 
-        # Add to all clients
-        for site_name in client_names:
-            recipe.job.to(quantizer, site_name, tasks=["train"], filter_type=FilterType.TASK_RESULT)
-            recipe.job.to(dequantizer, site_name, tasks=["train"], filter_type=FilterType.TASK_DATA)
+        # Add to all clients: quantizer on output, dequantizer on input
+        recipe.add_client_output_filter(quantizer, tasks=["train"])
+        recipe.add_client_input_filter(dequantizer, tasks=["train"])
 
     # Add experiment tracking if requested
     if args.use_tracking:

nvflare/recipe/spec.py

@@ -114,6 +114,36 @@ def process_env(self, env: ExecEnv):
         """
         pass
 
+    def _add_to_client_apps(self, obj, clients: Optional[List[str]] = None, **kwargs):
+        """Add an object to client apps, preserving existing per-site structure.
+
+        Args:
+            obj: Object to add to clients.
+            clients: Optional list of specific client names. If None, applies to all clients.
+            **kwargs: Extra options forwarded to `job.to()`/`job.to_clients()`.
+        """
+        if clients is None:
+            from nvflare.apis.job_def import ALL_SITES, SERVER_SITE_NAME
+            from nvflare.job_config.defs import JobTargetType
+
+            # 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", {})
+            existing_client_sites = [
+                target
+                for target in deploy_map.keys()
+                if target not in [ALL_SITES, SERVER_SITE_NAME]
+                and JobTargetType.get_target_type(target) == JobTargetType.CLIENT
+            ]
+            if existing_client_sites:
+                for site in existing_client_sites:
+                    self.job.to(obj, site, **kwargs)
+            else:
+                self.job.to_clients(obj, **kwargs)
+        else:
+            for client in clients:
+                self.job.to(obj, client, **kwargs)
+
     def add_client_input_filter(
         self, filter: Filter, tasks: Optional[List[str]] = None, clients: Optional[List[str]] = None
     ):
@@ -127,11 +157,7 @@ def add_client_input_filter(
         Returns: None
 
         """
-        if clients is None:
-            self.job.to_clients(filter, filter_type=FilterType.TASK_DATA, tasks=tasks)
-        else:
-            for client in clients:
-                self.job.to(filter, client, filter_type=FilterType.TASK_DATA, tasks=tasks)
+        self._add_to_client_apps(filter, clients=clients, filter_type=FilterType.TASK_DATA, tasks=tasks)
 
     def add_client_output_filter(
         self, filter: Filter, tasks: Optional[List[str]] = None, clients: Optional[List[str]] = None
@@ -146,11 +172,7 @@ def add_client_output_filter(
         Returns: None
 
         """
-        if clients is None:
-            self.job.to_clients(filter, filter_type=FilterType.TASK_RESULT, tasks=tasks)
-        else:
-            for client in clients:
-                self.job.to(filter, client, filter_type=FilterType.TASK_RESULT, tasks=tasks)
+        self._add_to_client_apps(filter, clients=clients, filter_type=FilterType.TASK_RESULT, tasks=tasks)
 
     def add_client_config(self, config: Dict, clients: Optional[List[str]] = None):
         """Add top-level configuration parameters to config_fed_client.json.
@@ -165,11 +187,32 @@ def add_client_config(self, config: Dict, clients: Optional[List[str]] = None):
         if not isinstance(config, dict):
             raise TypeError(f"config must be a dict, got {type(config).__name__}")
 
-        if clients is None:
-            self.job.to_clients(config)
-        else:
-            for client in clients:
-                self.job.to(obj=config, target=client)
+        self._add_to_client_apps(config, clients=clients)
+
+    def add_client_file(self, file_path: str, clients: Optional[List[str]] = None):
+        """Add a file or directory to client apps.
+
+        The file will be added to the client's custom directory and bundled with the job.
+        Can be a script, configuration file, or any resource needed by clients.
+
+        Args:
+            file_path: Path to the file or directory to add to clients.
+            clients: Optional list of specific client names. If None, applies to all clients.
+
+        Raises:
+            TypeError: If file_path is not a string.
+
+        Example:
+            # Add a wrapper script to all clients
+            recipe.add_client_file("client_wrapper.sh")
+
+            # Add a script to specific clients
+            recipe.add_client_file("custom_script.py", clients=["site1", "site2"])
+        """
+        if not isinstance(file_path, str):
+            raise TypeError(f"file_path must be a str, got {type(file_path).__name__}")
+
+        self._add_to_client_apps(file_path, clients=clients)
 
     def add_server_output_filter(self, filter: Filter, tasks: Optional[List[str]] = None):
         """Add a filter to the server for outgoing tasks to clients.
@@ -209,6 +252,27 @@ def add_server_config(self, config: Dict):
 
         self.job.to_server(config)
 
+    def add_server_file(self, file_path: str):
+        """Add a file or directory to server app.
+
+        The file will be added to the server's custom directory and bundled with the job.
+        Can be a script, configuration file, or any resource needed by the server.
+
+        Args:
+            file_path: Path to the file or directory to add to server.
+
+        Raises:
+            TypeError: If file_path is not a string.
+
+        Example:
+            # Add a wrapper script to server
+            recipe.add_server_file("server_wrapper.sh")
+        """
+        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)
+
     @staticmethod
     def _get_full_class_name(obj):
         """
@@ -243,7 +307,8 @@ def add_decomposers(self, decomposers: List[Union[str, Decomposer]]):
 
         reg = DecomposerRegister(class_names)
         self.job.to_server(reg, id="decomposer_reg")
-        self.job.to_clients(reg, id="decomposer_reg")
+
+        self._add_to_client_apps(reg, id="decomposer_reg")
 
     def export(
         self,
@@ -267,7 +332,7 @@ def export(
             self.job.to_server(server_exec_params)
 
         if client_exec_params:
-            self.job.to_clients(client_exec_params)
+            self._add_to_client_apps(client_exec_params)
 
         if env:
             self.process_env(env)
@@ -291,7 +356,7 @@ def execute(
             self.job.to_server(server_exec_params)
 
         if client_exec_params:
-            self.job.to_clients(client_exec_params)
+            self._add_to_client_apps(client_exec_params)
 
         self.process_env(env)
         job_id = env.deploy(self.job)

tests/unit_test/recipe/spec_test.py

@@ -212,6 +212,113 @@ def test_add_client_config(self, temp_script):
         assert all_clients_app is not None
         assert all_clients_app.app_config.additional_params == config
 
+    def test_add_client_file_adds_to_ext_scripts_and_ext_dirs(self, temp_script):
+        """Test add_client_file stores file paths in ext_scripts and dirs in ext_dirs."""
+        from nvflare.apis.job_def import ALL_SITES
+        from nvflare.fuel.utils.constants import FrameworkType
+        from nvflare.recipe.fedavg import FedAvgRecipe
+
+        recipe = FedAvgRecipe(
+            name="test_job_files",
+            num_rounds=2,
+            min_clients=2,
+            train_script=temp_script,
+            initial_ckpt="/abs/path/to/model.npy",
+            framework=FrameworkType.NUMPY,
+        )
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            recipe.add_client_file(temp_script)
+            recipe.add_client_file(temp_dir)
+
+            all_clients_app = recipe.job._deploy_map.get(ALL_SITES)
+            assert all_clients_app is not None
+            assert temp_script in all_clients_app.app_config.ext_scripts
+            assert temp_dir in all_clients_app.app_config.ext_dirs
+
+    def test_add_client_file_preserves_per_site_clients_without_all_sites(self, temp_script):
+        """Test add_client_file keeps per-site topology and does not create ALL_SITES app."""
+        from nvflare.apis.job_def import ALL_SITES
+        from nvflare.fuel.utils.constants import FrameworkType
+        from nvflare.recipe.fedavg import FedAvgRecipe
+
+        recipe = FedAvgRecipe(
+            name="test_job_per_site_files",
+            num_rounds=2,
+            min_clients=2,
+            train_script=temp_script,
+            initial_ckpt="/abs/path/to/model.npy",
+            framework=FrameworkType.NUMPY,
+            per_site_config={"site-1": {}, "site-2": {}},
+        )
+
+        recipe.add_client_file(temp_script)
+
+        assert ALL_SITES not in recipe.job._deploy_map
+        site_1_app = recipe.job._deploy_map.get("site-1")
+        site_2_app = recipe.job._deploy_map.get("site-2")
+        assert site_1_app is not None
+        assert site_2_app is not None
+        assert temp_script in site_1_app.app_config.ext_scripts
+        assert temp_script in site_2_app.app_config.ext_scripts
+
+    def test_add_client_file_with_specific_clients_only_updates_selected_sites(self, temp_script):
+        """Test add_client_file(..., clients=[...]) only adds file to specified sites."""
+        from nvflare.fuel.utils.constants import FrameworkType
+        from nvflare.recipe.fedavg import FedAvgRecipe
+
+        recipe = FedAvgRecipe(
+            name="test_job_targeted_files",
+            num_rounds=2,
+            min_clients=2,
+            train_script=temp_script,
+            initial_ckpt="/abs/path/to/model.npy",
+            framework=FrameworkType.NUMPY,
+            per_site_config={"site-1": {}, "site-2": {}, "site-3": {}},
+        )
+
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as f:
+            f.write("targeted file for site-2 only")
+            targeted_file = f.name
+
+        recipe.add_client_file(targeted_file, clients=["site-2"])
+
+        try:
+            site_1_app = recipe.job._deploy_map.get("site-1")
+            site_2_app = recipe.job._deploy_map.get("site-2")
+            site_3_app = recipe.job._deploy_map.get("site-3")
+            assert site_1_app is not None
+            assert site_2_app is not None
+            assert site_3_app is not None
+            assert targeted_file not in site_1_app.app_config.ext_scripts
+            assert targeted_file in site_2_app.app_config.ext_scripts
+            assert targeted_file not in site_3_app.app_config.ext_scripts
+        finally:
+            os.unlink(targeted_file)
+
+    def test_add_server_file_adds_to_server_ext_scripts_and_ext_dirs(self, temp_script):
+        """Test add_server_file stores file paths in ext_scripts and dirs in ext_dirs."""
+        from nvflare.fuel.utils.constants import FrameworkType
+        from nvflare.recipe.fedavg import FedAvgRecipe
+
+        recipe = FedAvgRecipe(
+            name="test_job_server_files",
+            num_rounds=2,
+            min_clients=2,
+            train_script=temp_script,
+            initial_ckpt="/abs/path/to/model.npy",
+            framework=FrameworkType.NUMPY,
+        )
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            recipe.add_server_file(temp_script)
+            recipe.add_server_file(temp_dir)
+
+            server_app = recipe.job._deploy_map.get("server")
+            assert server_app is not None
+            assert temp_script in server_app.app_config.ext_scripts
+            assert temp_dir in server_app.app_config.ext_dirs
+
     def test_config_in_generated_json(self, temp_script):
         """Test that configs appear in generated JSON files."""
         from nvflare.fuel.utils.constants import FrameworkType