Add per-site client tracking and harden recipe client placement

docs/user_guide/data_scientist_guide/recipe_api.rst

@@ -98,6 +98,30 @@ File packaging helpers:
 ``recipe.add_server_file(file_path)``
    Bundle a file or directory into the generated server app package.
 
+Component helpers:
+
+``recipe.add_client_component(component, clients=None, id=None)``
+   Add a component object (e.g. a tracking receiver or streamer) to generated
+   client apps. If ``clients`` is omitted, the component applies to all
+   generated client apps.
+
+``recipe.add_server_component(component, id=None)``
+   Add a component object to the generated server app. Returns the final
+   component id; components that implement their own job-registration hook
+   return that hook's result instead.
+
+These component helpers place plain components only. Files, config parameters,
+filters, and executors have the dedicated APIs above; controllers are
+configured by the recipe itself. Targeting specific clients with ``clients``
+requires per-site client apps: construct the recipe with the
+``per_site_config`` constructor argument on recipes that support it. With the
+default all-clients topology, targeted placement raises an error rather than
+silently dropping the component from the generated job. Calling
+``set_per_site_config`` after construction records the configuration for
+``configured_sites()`` but does not yet rebuild an existing all-clients app
+into per-site apps; recipes will interpret helper-provided per-site config as
+follow-up work.
+
 Filter helpers:
 
 ``recipe.add_client_input_filter(filter, tasks=None, clients=None)``
@@ -123,9 +147,11 @@ Shared helpers:
 
 Utility helpers:
 
-``add_experiment_tracking(recipe, tracking_type, tracking_config=None, client_side=False, server_side=True)``
+``add_experiment_tracking(recipe, tracking_type, tracking_config=None, client_side=False, server_side=True, clients=None)``
    Add supported experiment tracking receivers such as TensorBoard, MLflow, or
-   Weights & Biases.
+   Weights & Biases. With ``client_side=True``, ``clients`` limits which sites
+   receive the client-side receiver; call once per site with different
+   ``tracking_config`` values for per-site tracking destinations.
 
 ``add_cross_site_evaluation(recipe, submit_model_timeout=600, validation_timeout=6000, participating_clients=None)``
    Add cross-site evaluation to a training recipe when the recipe/framework
@@ -228,9 +254,10 @@ The following are internal details and should not be used in recipe scripts:
 * generated deploy-map internals;
 * internal fields used by Recipe helpers.
 
-If a workflow needs arbitrary component placement that is not covered by a named
-Recipe helper, use the lower-level Job API workflow or add a new named Recipe
-helper for the repeated pattern.
+Custom component placement is covered by ``add_server_component`` and
+``add_client_component``. If a workflow needs job structure those helpers do
+not express (e.g. custom executors or multi-app layouts), use the lower-level
+Job API workflow or add a new named Recipe helper for the repeated pattern.
 
 Recipe Catalog JSON
 -------------------

nvflare/recipe/spec.py

@@ -15,7 +15,7 @@
 import sys
 from abc import ABC, abstractmethod
 from contextlib import contextmanager
-from typing import Dict, List, Optional, Union
+from typing import Any, Dict, List, Optional, Union
 
 # Single source of truth for the default export dir. Kept in the current working
 # directory on purpose: an export is a user-requested artifact, so it lands next to
@@ -88,7 +88,9 @@ def _peek_recipe_args() -> tuple:
     return _RECIPE_EXPORT, _RECIPE_EXPORT_DIR
 
 
+from nvflare.apis.executor import Executor
 from nvflare.apis.filter import Filter
+from nvflare.apis.impl.controller import Controller
 from nvflare.app_common.widgets.decomposer_reg import DecomposerRegister
 from nvflare.fuel.utils.fobs import Decomposer
 from nvflare.job_config.api import FedJob
@@ -314,14 +316,20 @@ def _add_to_client_apps(self, obj, clients: Optional[List[str]] = None, **kwargs
             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()`.
+
+        Raises:
+            TypeError: If clients is not a list.
+            ValueError: If clients is empty or contains a non-client name, or if specific
+                clients are targeted while the recipe's client app applies to all clients
+                (per-site placement cannot be expressed in the generated job in that topology).
         """
-        if clients is None:
-            from nvflare.apis.job_def import ALL_SITES, SERVER_SITE_NAME
-            from nvflare.job_config.defs import JobTargetType
+        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", {})
+        # 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", {})
+        if clients is None:
             existing_client_sites = [
                 target
                 for target in deploy_map.keys()
@@ -334,6 +342,24 @@ def _add_to_client_apps(self, obj, clients: Optional[List[str]] = None, **kwargs
             else:
                 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):
+                raise TypeError(f"clients must be a list of client names, got {type(clients).__name__}")
+            if not clients:
+                raise ValueError("clients must not be empty; omit it to apply to all clients")
+            for client in clients:
+                if not isinstance(client, str) or client in (ALL_SITES, SERVER_SITE_NAME):
+                    raise ValueError(f"invalid client name {client!r}: client names must name specific client sites")
+            if ALL_SITES in deploy_map:
+                # The generated job has one client app deployed to all clients. Exporting
+                # both an all-clients app and per-site apps is not expressible (the
+                # all-clients app wins and per-site apps are dropped), so fail loudly
+                # instead of silently losing the placement.
+                raise ValueError(
+                    "cannot target specific clients: this recipe's client app applies to all clients. "
+                    "Construct the recipe with per-site client apps (e.g. the per_site_config constructor "
+                    "argument on recipes that support it) or omit clients to apply to all clients."
+                )
             for client in clients:
                 self.job.to(obj, client, **kwargs)
 
@@ -407,6 +433,81 @@ def add_client_file(self, file_path: str, clients: Optional[List[str]] = None):
 
         self._add_to_client_apps(file_path, clients=clients)
 
+    def add_client_component(self, component: Any, clients: Optional[List[str]] = None, id: Optional[str] = None):
+        """Add a component to client apps.
+
+        Use this for client-side components such as tracking receivers or streamers,
+        instead of reaching into the recipe's generated job. The component is
+        registered in each targeted client app's configuration.
+
+        This is a bounded placement primitive for components only. Files, config
+        parameters, filters, and executors have dedicated APIs: ``add_client_file``,
+        ``add_client_config``, ``add_client_input_filter``/``add_client_output_filter``,
+        and the recipe's own executor/train-script parameters.
+
+        Args:
+            component: The component object to add to client apps.
+            clients: Optional list of specific client names. If None, applies to all clients.
+                Targeting specific clients requires the recipe's client apps to be per-site
+                (e.g. recipes constructed with the per_site_config constructor argument);
+                with the default all-clients topology, targeted placement raises ValueError
+                instead of silently dropping the component from the generated job.
+            id: Optional component id. If None, an id is generated. If the id is already
+                used in a client app, a numeric suffix is appended.
+
+        Returns:
+            None. The final id may differ per client app (numeric suffixing is per app),
+            so no single id is returned; pass an explicit id that is unused in the
+            targeted apps if you need a stable reference.
+
+        Raises:
+            TypeError: If component is a str, dict, Filter, Controller, Executor, or a
+                script runner, or clients is not a list.
+            ValueError: If clients is empty or names non-client sites, or targets specific
+                clients while the recipe's client app applies to all clients.
+
+        Example:
+            # Add a streamer component to all clients
+            recipe.add_client_component(TensorClientStreamer())
+
+            # Add a tracking receiver to one site only (per-site client apps required)
+            recipe.add_client_component(receiver, clients=["site-1"], id="mlflow_receiver")
+        """
+        self._validate_placed_component(component, "client")
+        self._add_to_client_apps(component, clients=clients, id=id)
+
+    @staticmethod
+    def _validate_placed_component(component: Any, target_type: str) -> None:
+        """Reject objects that have dedicated APIs so component placement stays bounded."""
+        if isinstance(component, str):
+            raise TypeError(f"component must be an object; use add_{target_type}_file for files or scripts")
+        if isinstance(component, dict):
+            raise TypeError(f"component must be an object; use add_{target_type}_config for configuration parameters")
+        if isinstance(component, Filter):
+            raise TypeError(
+                f"Filters are wired into filter chains; use add_{target_type}_input_filter or add_{target_type}_output_filter"
+            )
+        # Check Controller before Executor so a hybrid subclass gets the accurate message.
+        if isinstance(component, Controller):
+            if target_type == "client":
+                raise TypeError(
+                    "Controllers run in the server workflow and are configured by the recipe; they cannot be added to clients"
+                )
+            raise TypeError(
+                "Controllers are configured by the recipe itself; they cannot be added through add_server_component"
+            )
+        if isinstance(component, Executor):
+            raise TypeError(
+                f"Executors are configured through the recipe's executor/train-script parameters, not add_{target_type}_component"
+            )
+        # Script runners are executor-installing wrappers, not plain components.
+        from nvflare.job_config.script_runner import BaseScriptRunner, ScriptRunner
+
+        if isinstance(component, (BaseScriptRunner, ScriptRunner)):
+            raise TypeError(
+                f"script runners are configured through the recipe's train-script parameters, not add_{target_type}_component"
+            )
+
     def add_server_output_filter(self, filter: Filter, tasks: Optional[List[str]] = None):
         """Add a filter to the server for outgoing tasks to clients.
 
@@ -466,6 +567,39 @@ def add_server_file(self, file_path: str):
 
         self.job.to_server(file_path)
 
+    def add_server_component(self, component: Any, id: Optional[str] = None) -> Any:
+        """Add a component to the server app.
+
+        Use this for server-side components such as tracking receivers or streamers,
+        instead of reaching into the recipe's generated job. The component is
+        registered in the server app's configuration.
+
+        This is a bounded placement primitive for components only. Files, config
+        parameters, and filters have dedicated APIs: ``add_server_file``,
+        ``add_server_config``, and ``add_server_input_filter``/``add_server_output_filter``.
+        Controllers are configured by the recipe itself.
+
+        Args:
+            component: The component object to add to the server app.
+            id: Optional component id. If None, an id is generated. If the id is already
+                used in the server app, a numeric suffix is appended.
+
+        Returns:
+            The final component id assigned in the server app. Components that implement
+            their own job-registration hook (``add_to_fed_job``) return that hook's result
+            instead, which may not be an id.
+
+        Raises:
+            TypeError: If component is a str, dict, Filter, Controller, Executor, or a
+                script runner.
+
+        Example:
+            # Add a streamer component to the server
+            streamer_id = recipe.add_server_component(TensorServerStreamer())
+        """
+        self._validate_placed_component(component, "server")
+        return self.job.to_server(component, id=id)
+
     @staticmethod
     def _get_full_class_name(obj):
         """

nvflare/recipe/utils.py

@@ -214,6 +214,7 @@ def add_experiment_tracking(
     tracking_config: Optional[dict] = None,
     client_side: bool = False,
     server_side: bool = True,
+    clients: Optional[List[str]] = None,
 ):
     """Add experiment tracking to a recipe.
 
@@ -223,8 +224,15 @@ def add_experiment_tracking(
         recipe: Recipe instance to augment with experiment tracking.
         tracking_type: Type of tracking to enable ("mlflow", "tensorboard", or "wandb").
         tracking_config: Optional configuration dict for the tracking receiver.
-        client_side: If True, add tracking to all clients (each client tracks locally).
+        client_side: If True, add tracking to clients (each client tracks locally).
         server_side: If True, add tracking to server (aggregates metrics from all clients). Default: True.
+        clients: Optional list of client names for client-side tracking. If None, the
+            client-side receiver is added to all clients. Only valid with client_side=True.
+            To give sites different receiver configs (e.g. per-site tracking_uri), call this
+            function once per site with that site's tracking_config and clients=[site].
+            Targeting specific clients requires the recipe's client apps to be per-site
+            (e.g. recipes constructed with the per_site_config constructor argument);
+            with the default all-clients topology, targeted placement raises ValueError.
 
     Examples:
         # Server-side tracking (default - federated metrics)
@@ -235,6 +243,16 @@ def add_experiment_tracking(
 
         # Both server and client tracking
         add_experiment_tracking(recipe, "mlflow", {...}, client_side=True, server_side=True)
+
+        # Per-site client tracking configs (one call per site)
+        add_experiment_tracking(
+            recipe, "mlflow", {"tracking_uri": "file:///tmp/site-1/mlruns"},
+            client_side=True, server_side=False, clients=["site-1"],
+        )
+        add_experiment_tracking(
+            recipe, "mlflow", {"tracking_uri": "file:///tmp/site-2/mlruns"},
+            client_side=True, server_side=False, clients=["site-2"],
+        )
     """
     tracking_config = tracking_config or {}
     if tracking_type not in TRACKING_REGISTRY:
@@ -243,6 +261,14 @@ def add_experiment_tracking(
     if not server_side and not client_side:
         raise ValueError("At least one of server_side or client_side must be True")
 
+    if clients is not None:
+        if not client_side:
+            raise ValueError("clients is only used for client-side tracking; set client_side=True")
+        if not isinstance(clients, list) or not all(isinstance(c, str) for c in clients):
+            raise TypeError(f"clients must be a list of str, got {clients!r}")
+        if not clients:
+            raise ValueError("clients must not be empty; omit it to add tracking to all clients")
+
     _, flag = optional_import(TRACKING_REGISTRY[tracking_type]["package"])
     if not flag:
         raise ValueError(
@@ -267,7 +293,9 @@ def add_experiment_tracking(
             client_config["events"] = [ANALYTIC_EVENT_TYPE]
 
         client_receiver = receiver_class(**client_config)
-        recipe.job.to_clients(client_receiver, id="client_receiver")
+        # Route through the recipe placement helper so existing per-site client apps
+        # are preserved (to_clients would target ALL_SITES even when per-site apps exist).
+        recipe.add_client_component(client_receiver, clients=clients, id="client_receiver")
 
 
 def add_cross_site_evaluation(