Optimize start_workflow to reduce multiple UI rebuilds

Adds JobOrchestrator.replace_staged_configs() to replace all staged configs
in a single operation with a single notification to subscribers. This
significantly reduces UI rebuilds during workflow start.

Before this change, start_workflow triggered N+2 widget rebuilds:
- clear_staged_configs() → rebuild 1
- stage_config() for source 1 → rebuild 2
- stage_config() for source 2 → rebuild 3
- ...
- stage_config() for source N → rebuild N+1
- commit_workflow() → rebuild N+2

After this change, it triggers only 2 rebuilds:
- replace_staged_configs() → rebuild 1 (all sources)
- commit_workflow() → rebuild 2

Changes:
- Add replace_staged_configs() method to JobOrchestrator
- Update WorkflowController.start_workflow() to use replace_staged_configs()
- Add 3 tests verifying single notification, config replacement, and copying

This improves performance and reduces UI flicker when starting workflows
with multiple sources.

Related to workflow-status-widget-architecture-fix.md review feedback.

Author: SimonHeybrock

src/ess/livedata/dashboard/job_orchestrator.py

@@ -263,6 +263,38 @@ def stage_config(
         )
         self._notify_staged_changed(workflow_id)
 
+    def replace_staged_configs(
+        self,
+        workflow_id: WorkflowId,
+        *,
+        configs: dict[SourceName, JobConfig],
+    ) -> None:
+        """
+        Replace all staged configs with new configs in a single operation.
+
+        This is more efficient than clear + multiple stage_config calls because
+        it triggers only a single notification to subscribers, reducing UI rebuilds.
+
+        Parameters
+        ----------
+        workflow_id
+            The workflow to configure.
+        configs
+            Dict mapping source names to their configs.
+        """
+        # Clear existing staged configs
+        self._workflows[workflow_id].staged_jobs.clear()
+
+        # Add all new configs (with copies to prevent external mutation)
+        for source_name, job_config in configs.items():
+            self._workflows[workflow_id].staged_jobs[source_name] = JobConfig(
+                params=job_config.params.copy(),
+                aux_source_names=job_config.aux_source_names.copy(),
+            )
+
+        # Single notification for the entire operation
+        self._notify_staged_changed(workflow_id)
+
     def commit_workflow(self, workflow_id: WorkflowId) -> list[JobId]:
         """
         Commit staged configs and start workflow.

src/ess/livedata/dashboard/workflow_controller.py

@@ -21,7 +21,7 @@
 from .configuration_adapter import ConfigurationState
 from .correlation_histogram import CorrelationHistogramController, make_workflow_spec
 from .data_service import DataService
-from .job_orchestrator import JobOrchestrator
+from .job_orchestrator import JobConfig, JobOrchestrator
 from .workflow_configuration_adapter import WorkflowConfigurationAdapter
 
 
@@ -129,21 +129,19 @@ def start_workflow(
         if not source_names:
             return []
 
-        # Clear existing staged configs and stage new ones
-        # This ensures only the requested sources are included in the workflow
-        self._orchestrator.clear_staged_configs(workflow_id)
-
         # Convert Pydantic models to dicts for orchestrator
         params_dict = config.model_dump(mode='json')
         aux_dict = aux_source_names.model_dump(mode='json') if aux_source_names else {}
 
-        for source_name in source_names:
-            self._orchestrator.stage_config(
-                workflow_id,
-                source_name=source_name,
-                params=params_dict,
-                aux_source_names=aux_dict,
-            )
+        # Build configs for all sources
+        configs = {
+            source_name: JobConfig(params=params_dict, aux_source_names=aux_dict)
+            for source_name in source_names
+        }
+
+        # Replace staged configs in a single operation (more efficient than
+        # clear + multiple stage_config calls, reduces UI rebuilds)
+        self._orchestrator.replace_staged_configs(workflow_id, configs=configs)
 
         # Commit and start workflow
         return self._orchestrator.commit_workflow(workflow_id)

tests/dashboard/job_orchestrator_test.py

@@ -1617,3 +1617,125 @@ def test_none_callbacks_are_skipped(
 
         # Only the commit callback was set
         assert events == ['committed']
+
+    def test_replace_staged_configs_replaces_all_configs(
+        self,
+        workflow_with_params: WorkflowSpec,
+        fake_workflow_config_service: FakeWorkflowConfigService,
+    ):
+        """replace_staged_configs replaces all staged configs in one operation."""
+        from ess.livedata.dashboard.job_orchestrator import JobConfig
+
+        workflow_id = workflow_with_params.get_id()
+        registry = {workflow_id: workflow_with_params}
+
+        orchestrator = JobOrchestrator(
+            command_service=CommandService(sink=FakeMessageSink()),
+            workflow_config_service=fake_workflow_config_service,
+            workflow_registry=registry,
+            config_store=None,
+        )
+
+        # Stage some initial configs
+        orchestrator.stage_config(
+            workflow_id,
+            source_name="det_1",
+            params={"threshold": 100.0},
+            aux_source_names={},
+        )
+
+        # Replace with new configs
+        new_configs = {
+            "det_2": JobConfig(params={"threshold": 200.0}, aux_source_names={}),
+            "det_3": JobConfig(params={"threshold": 300.0}, aux_source_names={}),
+        }
+        orchestrator.replace_staged_configs(workflow_id, configs=new_configs)
+
+        # Old config should be gone, new configs present
+        staged = orchestrator.get_staged_config(workflow_id)
+        assert "det_1" not in staged
+        assert "det_2" in staged
+        assert "det_3" in staged
+        assert staged["det_2"].params == {"threshold": 200.0}
+        assert staged["det_3"].params == {"threshold": 300.0}
+
+    def test_replace_staged_configs_triggers_single_notification(
+        self,
+        workflow_with_params: WorkflowSpec,
+        fake_workflow_config_service: FakeWorkflowConfigService,
+    ):
+        """replace_staged_configs triggers only one notification for efficiency."""
+        from ess.livedata.dashboard.job_orchestrator import (
+            JobConfig,
+            WidgetLifecycleCallbacks,
+        )
+
+        workflow_id = workflow_with_params.get_id()
+        registry = {workflow_id: workflow_with_params}
+
+        orchestrator = JobOrchestrator(
+            command_service=CommandService(sink=FakeMessageSink()),
+            workflow_config_service=fake_workflow_config_service,
+            workflow_registry=registry,
+            config_store=None,
+        )
+
+        # Subscribe and count notifications
+        notification_count = 0
+
+        def count_notification(wid):
+            nonlocal notification_count
+            notification_count += 1
+
+        callbacks = WidgetLifecycleCallbacks(on_staged_changed=count_notification)
+        orchestrator.subscribe_to_widget_lifecycle(callbacks)
+
+        # Clear the counter (initial subscription triggers notification)
+        notification_count = 0
+
+        # Replace with multiple configs - should trigger only ONE notification
+        new_configs = {
+            "det_1": JobConfig(params={"threshold": 100.0}, aux_source_names={}),
+            "det_2": JobConfig(params={"threshold": 200.0}, aux_source_names={}),
+            "det_3": JobConfig(params={"threshold": 300.0}, aux_source_names={}),
+        }
+        orchestrator.replace_staged_configs(workflow_id, configs=new_configs)
+
+        # Only one notification despite replacing 3 configs
+        assert notification_count == 1
+
+    def test_replace_staged_configs_copies_config_dicts(
+        self,
+        workflow_with_params: WorkflowSpec,
+        fake_workflow_config_service: FakeWorkflowConfigService,
+    ):
+        """replace_staged_configs copies config dicts to prevent external mutation."""
+        from ess.livedata.dashboard.job_orchestrator import JobConfig
+
+        workflow_id = workflow_with_params.get_id()
+        registry = {workflow_id: workflow_with_params}
+
+        orchestrator = JobOrchestrator(
+            command_service=CommandService(sink=FakeMessageSink()),
+            workflow_config_service=fake_workflow_config_service,
+            workflow_registry=registry,
+            config_store=None,
+        )
+
+        # Create configs with mutable dicts
+        params = {"threshold": 100.0}
+        aux = {"monitor": "mon_1"}
+        configs = {
+            "det_1": JobConfig(params=params, aux_source_names=aux),
+        }
+
+        orchestrator.replace_staged_configs(workflow_id, configs=configs)
+
+        # Mutate the original dicts
+        params["threshold"] = 999.0
+        aux["monitor"] = "mutated"
+
+        # Staged config should be unaffected (was copied)
+        staged = orchestrator.get_staged_config(workflow_id)
+        assert staged["det_1"].params["threshold"] == 100.0
+        assert staged["det_1"].aux_source_names["monitor"] == "mon_1"