From 0857c8a265acda30746d7c5a55ba7ea90eca5da2 Mon Sep 17 00:00:00 2001
From: Giulio Eulisse <10544+ktf@users.noreply.github.com>
Date: Thu, 5 Nov 2026 22:13:36 +0100
Subject: [PATCH 1/5] DPL: first step to make workflow definition plugins

This is the first step towards having the workflow definition inside plugins
rather than in executables. This will allow accumulating plugins which are needed
to instantiate a topology and do the option parsing / topology building only once,
simplifying the current case.

The end goal is to allow the driver to preload certain common services (e.g. ROOT)
and share it among the different tasks (which at the moment it's not allowed because different
tasks are in different executables). Moreover this will allow us to coalesce strictly coupled
dataprocessors and reduce the number of running processes.

For now the plugins are embedded in the executables and behave exactly like before:
runDataProcessing.h wraps what used to be mainNoCatch in a WorkflowPlugin of kind
DplPluginKind::Workflow, which returns the whole WorkflowDefinitionContext (options,
policies, config context and specs), and callMain picks it by plugin spec
("<library>:<plugin>", with an empty library for the embedded one) before handing
it to doMain. The embedded plugin is looked up through the callback passed by main,
not with dlsym on the executable, which would need exported symbols.
---
 Framework/Core/include/Framework/Plugins.h    |   3 +
 .../Framework/WorkflowDefinitionContext.h     |  61 ++++++++++
 .../include/Framework/runDataProcessing.h     | 114 ++++++++++--------
 Framework/Core/src/runDataProcessing.cxx      |  52 +++++---
 4 files changed, 165 insertions(+), 65 deletions(-)
 create mode 100644 Framework/Core/include/Framework/WorkflowDefinitionContext.h

diff --git a/Framework/Core/include/Framework/Plugins.h b/Framework/Core/include/Framework/Plugins.h
index 925943c6bffc3..3d320c6f2abb5 100644
--- a/Framework/Core/include/Framework/Plugins.h
+++ b/Framework/Core/include/Framework/Plugins.h
@@ -44,6 +44,9 @@ enum struct DplPluginKind : int {
   // using the arrow dataset API
   RootObjectReadingImplementation,
 
+  // A plugin which defines a whole workflow. This will be used to separate
+  // workflows in shared libraries and run them via a separate loader.
+  Workflow,
   // A plugin which was not initialised properly.
   Unknown
 };
diff --git a/Framework/Core/include/Framework/WorkflowDefinitionContext.h b/Framework/Core/include/Framework/WorkflowDefinitionContext.h
new file mode 100644
index 0000000000000..db17c17da1946
--- /dev/null
+++ b/Framework/Core/include/Framework/WorkflowDefinitionContext.h
@@ -0,0 +1,61 @@
+// Copyright 2019-2025 CERN and copyright holders of ALICE O2.
+// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
+// All rights not expressly granted are reserved.
+//
+// This software is distributed under the terms of the GNU General Public
+// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
+//
+// In applying this license CERN does not waive the privileges and immunities
+// granted to it by virtue of its status as an Intergovernmental Organization
+// or submit itself to any jurisdiction.
+#ifndef O2_FRAMEWORK_WORKFLOWDEFINITIONCONTEXT_H_
+#define O2_FRAMEWORK_WORKFLOWDEFINITIONCONTEXT_H_
+
+#include "Framework/ConfigContext.h"
+#include "Framework/ConfigParamRegistry.h"
+#include "Framework/ConfigParamSpec.h"
+#include "Framework/CompletionPolicy.h"
+#include "Framework/DispatchPolicy.h"
+#include "Framework/ResourcePolicy.h"
+#include "Framework/CallbacksPolicy.h"
+#include "Framework/SendingPolicy.h"
+#include "Framework/WorkflowSpec.h"
+#include "Framework/ChannelConfigurationPolicy.h"
+#include "Framework/ServiceRegistry.h"
+#include <functional>
+#include <memory>
+#include <vector>
+
+namespace o2::framework
+{
+
+struct WorkflowDefinitionContext {
+  std::vector<ConfigParamSpec> workflowOptions;
+  std::vector<CompletionPolicy> completionPolicies;
+  std::vector<DispatchPolicy> dispatchPolicies;
+  std::vector<ResourcePolicy> resourcePolicies;
+  std::vector<CallbacksPolicy> callbacksPolicies;
+  std::vector<SendingPolicy> sendingPolicies;
+  std::vector<ConfigParamSpec> extraOptions;
+  std::vector<ChannelConfigurationPolicy> channelPolicies;
+  std::unique_ptr<ConfigContext> configContext;
+
+  // For the moment, let's put them here. We should
+  // probably move them to a different place, since these are not really part
+  // of the workflow definition but will be there also at runtime.
+  std::unique_ptr<ServiceRegistry> configRegistry{nullptr};
+  std::unique_ptr<ConfigParamRegistry> workflowOptionsRegistry{nullptr};
+
+  o2::framework::WorkflowSpec specs;
+};
+
+struct WorkflowDefinition {
+  std::function<o2::framework::WorkflowDefinitionContext(int argc, char** argv)> defineWorkflow;
+};
+
+struct WorkflowPlugin {
+  virtual o2::framework::WorkflowDefinition* create() = 0;
+};
+
+} // namespace o2::framework
+#endif // O2_FRAMEWORK_WORKFLOWDEFINITIONCONTEXT_H_
diff --git a/Framework/Core/include/Framework/runDataProcessing.h b/Framework/Core/include/Framework/runDataProcessing.h
index e756c9f4b1d26..bd40e99e46e46 100644
--- a/Framework/Core/include/Framework/runDataProcessing.h
+++ b/Framework/Core/include/Framework/runDataProcessing.h
@@ -27,6 +27,8 @@
 #include "Framework/CommonServices.h"
 #include "Framework/WorkflowCustomizationHelpers.h"
 #include "Framework/Logger.h" // IWYU pragma: export
+#include "Framework/WorkflowDefinitionContext.h"
+#include "Framework/Plugins.h"
 #include "Framework/CheckTypes.h"
 #include "Framework/StructToTuple.h"
 #include "ResourcePolicy.h"
@@ -124,18 +126,6 @@ void overrideCloning(o2::framework::ConfigContext& ctx, std::vector<o2::framewor
 /// Helper used to add labels to Data Processors
 void overrideLabels(o2::framework::ConfigContext& ctx, std::vector<o2::framework::DataProcessorSpec>& workflow);
 
-// This comes from the framework itself. This way we avoid code duplication.
-int doMain(int argc, char** argv, o2::framework::WorkflowSpec const& specs,
-           std::vector<o2::framework::ChannelConfigurationPolicy> const& channelPolicies,
-           std::vector<o2::framework::CompletionPolicy> const& completionPolicies,
-           std::vector<o2::framework::DispatchPolicy> const& dispatchPolicies,
-           std::vector<o2::framework::ResourcePolicy> const& resourcePolicies,
-           std::vector<o2::framework::CallbacksPolicy> const& callbacksPolicies,
-           std::vector<o2::framework::SendingPolicy> const& sendingPolicies,
-           std::vector<o2::framework::ConfigParamSpec> const& workflowOptions,
-           std::vector<o2::framework::ConfigParamSpec> const& detectedOptions,
-           o2::framework::ConfigContext& configContext);
-
 void doDefaultWorkflowTerminationHook();
 
 template <typename T>
@@ -167,55 +157,81 @@ void callWorkflowTermination(T&, char const* idstring)
 
 void overrideAll(o2::framework::ConfigContext& ctx, std::vector<o2::framework::DataProcessorSpec>& workflow);
 
-o2::framework::ConfigContext createConfigContext(std::unique_ptr<o2::framework::ConfigParamRegistry>& workflowOptionsRegistry,
-                                                 o2::framework::ServiceRegistry& configRegistry,
-                                                 std::vector<o2::framework::ConfigParamSpec>& workflowOptions,
-                                                 std::vector<o2::framework::ConfigParamSpec>& extraOptions, int argc, char** argv);
+std::unique_ptr<o2::framework::ConfigContext> createConfigContext(std::unique_ptr<o2::framework::ConfigParamRegistry>& workflowOptionsRegistry,
+                                                                  o2::framework::ServiceRegistry& configRegistry,
+                                                                  std::vector<o2::framework::ConfigParamSpec>& workflowOptions,
+                                                                  std::vector<o2::framework::ConfigParamSpec>& extraOptions, int argc, char** argv);
 
 std::unique_ptr<o2::framework::ServiceRegistry> createRegistry();
 
-int mainNoCatch(int argc, char** argv)
-{
-  using namespace o2::framework;
+char* getIdString(int argc, char** argv);
+
+#define DPL_STRINGIZE_NX(A) #A
+#define DPL_STRINGIZE(A) DPL_STRINGIZE_NX(A)
+
+// The workflow definition is a plugin. By default it is embedded in the
+// executable as DPL_WORKFLOW_PLUGIN_NAME, with an empty library name.
+// To build it in a DSO instead, both DPL_WORKFLOW_PLUGIN_NAME and
+// DPL_WORKFLOW_PLUGIN_LIBRARY must be specified at build time.
+#ifndef DPL_WORKFLOW_PLUGIN_NAME
+#define DPL_WORKFLOW_PLUGIN_NAME InternalCustomWorkflow
+#ifdef DPL_WORKFLOW_PLUGIN_LIBRARY
+#error Missing DPL_WORKFLOW_PLUGIN_NAME
+#endif
+#define DPL_WORKFLOW_PLUGIN_LIBRARY
+#endif
 
-  std::vector<o2::framework::ConfigParamSpec> workflowOptions;
-  UserCustomizationsHelper::userDefinedCustomization(workflowOptions);
-  auto requiredWorkflowOptions = WorkflowCustomizationHelpers::requiredWorkflowOptions();
-  workflowOptions.insert(std::end(workflowOptions), std::begin(requiredWorkflowOptions), std::end(requiredWorkflowOptions));
-
-  std::vector<CompletionPolicy> completionPolicies = injectCustomizations<CompletionPolicy>();
-  std::vector<DispatchPolicy> dispatchPolicies = injectCustomizations<DispatchPolicy>();
-  std::vector<ResourcePolicy> resourcePolicies = injectCustomizations<ResourcePolicy>();
-  std::vector<CallbacksPolicy> callbacksPolicies = injectCustomizations<CallbacksPolicy>();
-  std::vector<SendingPolicy> sendingPolicies = injectCustomizations<SendingPolicy>();
-
-  std::unique_ptr<ServiceRegistry> configRegistry = createRegistry();
-  std::vector<ConfigParamSpec> extraOptions;
-  std::unique_ptr<ConfigParamRegistry> workflowOptionsRegistry{nullptr};
-  auto configContext = createConfigContext(workflowOptionsRegistry, *configRegistry, workflowOptions, extraOptions, argc, argv);
-
-  o2::framework::WorkflowSpec specs = defineDataProcessing(configContext);
-  overrideAll(configContext, specs);
-  for (auto& spec : specs) {
-    UserCustomizationsHelper::userDefinedCustomization(spec.requiredServices);
+struct DPL_WORKFLOW_PLUGIN_NAME : o2::framework::WorkflowPlugin {
+  o2::framework::WorkflowDefinition* create() override
+  {
+    return new o2::framework::WorkflowDefinition{
+      .defineWorkflow = [](int argc, char** argv) -> o2::framework::WorkflowDefinitionContext {
+        using namespace o2::framework;
+        WorkflowDefinitionContext workflowContext;
+
+        UserCustomizationsHelper::userDefinedCustomization(workflowContext.workflowOptions);
+        auto requiredWorkflowOptions = WorkflowCustomizationHelpers::requiredWorkflowOptions();
+        workflowContext.workflowOptions.insert(std::end(workflowContext.workflowOptions), std::begin(requiredWorkflowOptions), std::end(requiredWorkflowOptions));
+
+        workflowContext.completionPolicies = injectCustomizations<CompletionPolicy>();
+        workflowContext.dispatchPolicies = injectCustomizations<DispatchPolicy>();
+        workflowContext.resourcePolicies = injectCustomizations<ResourcePolicy>();
+        workflowContext.callbacksPolicies = injectCustomizations<CallbacksPolicy>();
+        workflowContext.sendingPolicies = injectCustomizations<SendingPolicy>();
+
+        workflowContext.configRegistry = createRegistry();
+        workflowContext.configContext = createConfigContext(workflowContext.workflowOptionsRegistry, *workflowContext.configRegistry, workflowContext.workflowOptions, workflowContext.extraOptions, argc, argv);
+
+        workflowContext.specs = defineDataProcessing(*workflowContext.configContext);
+        overrideAll(*workflowContext.configContext, workflowContext.specs);
+        for (auto& spec : workflowContext.specs) {
+          UserCustomizationsHelper::userDefinedCustomization(spec.requiredServices);
+        }
+        UserCustomizationsHelper::userDefinedCustomization(workflowContext.channelPolicies);
+        auto defaultChannelPolicies = ChannelConfigurationPolicy::createDefaultPolicies(*workflowContext.configContext);
+        workflowContext.channelPolicies.insert(std::end(workflowContext.channelPolicies), std::begin(defaultChannelPolicies), std::end(defaultChannelPolicies));
+        return workflowContext;
+      }};
   }
-  std::vector<ChannelConfigurationPolicy> channelPolicies;
-  UserCustomizationsHelper::userDefinedCustomization(channelPolicies);
-  auto defaultChannelPolicies = ChannelConfigurationPolicy::createDefaultPolicies(configContext);
-  channelPolicies.insert(std::end(channelPolicies), std::begin(defaultChannelPolicies), std::end(defaultChannelPolicies));
-  return doMain(argc, argv, specs,
-                channelPolicies, completionPolicies, dispatchPolicies,
-                resourcePolicies, callbacksPolicies, sendingPolicies, workflowOptions, extraOptions, configContext);
+};
+
+// This is like the plugin macros, we simply do it explicitly to avoid macro inside macro expansion
+extern "C" {
+DPLPluginHandle* dpl_plugin_callback(DPLPluginHandle* previous)
+{
+  return new DPLPluginHandle{new DPL_WORKFLOW_PLUGIN_NAME{}, strdup(DPL_STRINGIZE(DPL_WORKFLOW_PLUGIN_NAME)), o2::framework::DplPluginKind::Workflow, previous};
+}
 }
 
-int callMain(int argc, char** argv, int (*)(int, char**));
-char* getIdString(int argc, char** argv);
+// Loads the workflow plugin @a pluginSpec, i.e. "<library>:<plugin>", and runs it.
+// An empty library means the plugin embedded in the executable, found via @a embeddedPlugins.
+int callMain(int argc, char** argv, char const* pluginSpec, DPLPluginHandle* (*embeddedPlugins)(DPLPluginHandle*));
 
 int main(int argc, char** argv)
 {
   using namespace o2::framework;
 
-  int result = callMain(argc, argv, mainNoCatch);
+  int result = callMain(argc, argv, DPL_STRINGIZE(DPL_WORKFLOW_PLUGIN_LIBRARY) ":" DPL_STRINGIZE(DPL_WORKFLOW_PLUGIN_NAME), dpl_plugin_callback);
 
   char* idstring = getIdString(argc, argv);
   o2::framework::OnWorkflowTerminationHook onWorkflowTerminationHook;
diff --git a/Framework/Core/src/runDataProcessing.cxx b/Framework/Core/src/runDataProcessing.cxx
index 98da004e68a6f..056e150448d40 100644
--- a/Framework/Core/src/runDataProcessing.cxx
+++ b/Framework/Core/src/runDataProcessing.cxx
@@ -65,6 +65,7 @@
 #include "Framework/DeviceContext.h"
 #include "Framework/ServiceMetricsInfo.h"
 #include "Framework/DataTakingContext.h"
+#include "Framework/WorkflowDefinitionContext.h"
 #include "Framework/CommonServices.h"
 #include "Framework/DefaultsHelpers.h"
 #include "ProcessingPoliciesHelpers.h"
@@ -196,8 +197,26 @@ char* getIdString(int argc, char** argv)
   return nullptr;
 }
 
-int callMain(int argc, char** argv, int (*mainNoCatch)(int, char**))
+int doMain(int argc, char** argv, o2::framework::WorkflowDefinitionContext& workflowContext);
+
+int callMain(int argc, char** argv, char const* pluginSpec, DPLPluginHandle* (*embeddedPlugins)(DPLPluginHandle*))
 {
+  auto mainNoCatch = [pluginSpec, embeddedPlugins](int argc, char** argv) {
+    auto plugins = PluginManager::parsePluginSpecString(pluginSpec);
+    std::vector<WorkflowDefinition> workflows;
+    if (plugins.size() == 1 && plugins[0].library.empty()) {
+      if (auto* plugin = PluginManager::getByName<WorkflowPlugin>(embeddedPlugins(nullptr), plugins[0].name.c_str())) {
+        workflows.push_back(*plugin->create());
+      }
+    } else {
+      PluginManager::loadFromPlugin<WorkflowDefinition, WorkflowPlugin>(plugins, workflows);
+    }
+    if (workflows.size() != 1) {
+      throw runtime_error_f("Expected exactly one workflow plugin in %s, found %zu.", pluginSpec, workflows.size());
+    }
+    auto workflowContext = workflows.back().defineWorkflow(argc, argv);
+    return doMain(argc, argv, workflowContext);
+  };
   static bool noCatch = getenv("O2_NO_CATCHALL_EXCEPTIONS") && strcmp(getenv("O2_NO_CATCHALL_EXCEPTIONS"), "0");
   int result = 1;
   if (noCatch) {
@@ -2882,10 +2901,10 @@ void overrideAll(o2::framework::ConfigContext& ctx, std::vector<o2::framework::D
   overrideLabels(ctx, workflow);
 }
 
-o2::framework::ConfigContext createConfigContext(std::unique_ptr<ConfigParamRegistry>& workflowOptionsRegistry,
-                                                 o2::framework::ServiceRegistry& configRegistry,
-                                                 std::vector<o2::framework::ConfigParamSpec>& workflowOptions,
-                                                 std::vector<o2::framework::ConfigParamSpec>& extraOptions, int argc, char** argv)
+std::unique_ptr<o2::framework::ConfigContext> createConfigContext(std::unique_ptr<ConfigParamRegistry>& workflowOptionsRegistry,
+                                                                  o2::framework::ServiceRegistry& configRegistry,
+                                                                  std::vector<o2::framework::ConfigParamSpec>& workflowOptions,
+                                                                  std::vector<o2::framework::ConfigParamSpec>& extraOptions, int argc, char** argv)
 {
   std::vector<std::unique_ptr<o2::framework::ParamRetriever>> retrievers;
   std::unique_ptr<o2::framework::ParamRetriever> retriever{new o2::framework::BoostOptionsRetriever(true, argc, argv)};
@@ -2899,7 +2918,7 @@ o2::framework::ConfigContext createConfigContext(std::unique_ptr<ConfigParamRegi
     workflowOptions.push_back(extra);
   }
 
-  return o2::framework::ConfigContext(*workflowOptionsRegistry, o2::framework::ServiceRegistryRef{configRegistry}, argc, argv);
+  return std::make_unique<o2::framework::ConfigContext>(*workflowOptionsRegistry, o2::framework::ServiceRegistryRef{configRegistry}, argc, argv);
 }
 
 std::unique_ptr<o2::framework::ServiceRegistry> createRegistry()
@@ -2916,17 +2935,18 @@ std::unique_ptr<o2::framework::ServiceRegistry> createRegistry()
 //     killing them all on ctrl-c).
 //   - Child, pick the data-processor ID and start a O2DataProcessorDevice for
 //     each DataProcessorSpec
-int doMain(int argc, char** argv, o2::framework::WorkflowSpec const& workflow,
-           std::vector<ChannelConfigurationPolicy> const& channelPolicies,
-           std::vector<CompletionPolicy> const& completionPolicies,
-           std::vector<DispatchPolicy> const& dispatchPolicies,
-           std::vector<ResourcePolicy> const& resourcePolicies,
-           std::vector<CallbacksPolicy> const& callbacksPolicies,
-           std::vector<SendingPolicy> const& sendingPolicies,
-           std::vector<ConfigParamSpec> const& currentWorkflowOptions,
-           std::vector<ConfigParamSpec> const& detectedParams,
-           o2::framework::ConfigContext& configContext)
+int doMain(int argc, char** argv, o2::framework::WorkflowDefinitionContext& workflowContext)
 {
+  auto const& workflow = workflowContext.specs;
+  auto const& channelPolicies = workflowContext.channelPolicies;
+  auto const& completionPolicies = workflowContext.completionPolicies;
+  auto const& dispatchPolicies = workflowContext.dispatchPolicies;
+  auto const& resourcePolicies = workflowContext.resourcePolicies;
+  auto const& callbacksPolicies = workflowContext.callbacksPolicies;
+  auto const& sendingPolicies = workflowContext.sendingPolicies;
+  auto const& currentWorkflowOptions = workflowContext.workflowOptions;
+  auto const& detectedParams = workflowContext.extraOptions;
+  auto& configContext = *workflowContext.configContext;
   // Peek very early in the driver options and look for
   // signposts, so the we can enable it without going through the whole dance
   if (getenv("DPL_DRIVER_SIGNPOSTS")) {

From 91a803b3ac60bfc0143118936c544d2afd41bed4 Mon Sep 17 00:00:00 2001
From: Giulio Eulisse <10544+ktf@users.noreply.github.com>
Date: Thu, 5 Nov 2026 23:11:23 +0100
Subject: [PATCH 2/5] DPL: allow a workflow made of several plugins

--workflow-plugin <library>:<plugin>[,<library>:<plugin>...] replaces the
workflow plugin embedded in the executable with the given ones, which
are merged in a single workflow. An empty library means the executable
itself, so :InternalCustomWorkflow adds the executable's own workflow.

A plugin now only provides two steps, customize and defineWorkflow, and
defineWorkflow() in the framework does the rest once for all of them:

- the workflow options of all plugins are merged and parsed once, into
  a single ConfigContext and service registry, shared by all plugins, so
  that e.g. what analysis tasks request lands in the same registry. An
  option declared by more than one plugin is shared, provided they agree
  on its type and default; otherwise defining the workflow fails. The
  options required by DPL are added once, after those of the plugins.
- the data processors are concatenated in plugin order. A data
  processor name defined by more than one plugin is an error. Within a
  single plugin nothing is checked, as before.
- the policies customised by each plugin are concatenated in plugin
  order, followed by the default policies once, so that the catch-all
  defaults of one plugin do not hide the policies of the next one.

For a single plugin, which is what every executable does by default,
the sequence of calls is the same as before, so behaviour, help and
dumps are unchanged. --workflow-plugin is a hidden driver option, it is
forwarded to the children, which need it to define the same workflow,
and accepted by the devices.

Tested by test_WorkflowDefinition (merging, conflicting options,
duplicate data processors) and test_Framework_test_WorkflowPlugins, which
runs a producer and a consumer from two DSO plugins and checks that both
see the same value of their shared option.
---
 Framework/Core/CMakeLists.txt                 | 27 ++++++
 .../Framework/WorkflowDefinitionContext.h     | 13 ++-
 .../include/Framework/runDataProcessing.h     | 33 +++----
 Framework/Core/src/DeviceSpecHelpers.cxx      |  2 +
 Framework/Core/src/runDataProcessing.cxx      | 90 ++++++++++++++++--
 .../Core/test/test_WorkflowDefinition.cxx     | 95 +++++++++++++++++++
 Framework/Core/test/test_WorkflowPluginA.cxx  | 39 ++++++++
 Framework/Core/test/test_WorkflowPluginB.cxx  | 45 +++++++++
 Framework/Core/test/test_WorkflowPlugins.cxx  | 22 +++++
 9 files changed, 334 insertions(+), 32 deletions(-)
 create mode 100644 Framework/Core/test/test_WorkflowDefinition.cxx
 create mode 100644 Framework/Core/test/test_WorkflowPluginA.cxx
 create mode 100644 Framework/Core/test/test_WorkflowPluginB.cxx
 create mode 100644 Framework/Core/test/test_WorkflowPlugins.cxx

diff --git a/Framework/Core/CMakeLists.txt b/Framework/Core/CMakeLists.txt
index 437dc5ee6478c..222ceb48ea757 100644
--- a/Framework/Core/CMakeLists.txt
+++ b/Framework/Core/CMakeLists.txt
@@ -268,6 +268,7 @@ add_executable(o2-test-framework-core
               test/test_TopologyPolicies.cxx
               test/test_Variants.cxx
               test/test_WorkflowHelpers.cxx
+              test/test_WorkflowDefinition.cxx
               test/test_WorkflowSerialization.cxx
               test/test_DataOutputDirector.cxx
               test/unittest_SimpleOptionsRetriever.cxx
@@ -447,6 +448,32 @@ o2_add_test(
     "--global-config consumer-config --local-option hello-aliceo2 --a-boolean3 --an-int2 20 --a-double2 22. --an-int64-2 50000000000000"
   )
 
+# runs the workflows of two plugins built in DSOs, loaded with --workflow-plugin
+o2_add_test(
+  WorkflowPlugins NAME test_Framework_test_WorkflowPlugins
+  SOURCES test/test_WorkflowPlugins.cxx
+  COMPONENT_NAME Framework
+  LABELS framework workflow
+  TIMEOUT 30
+  PUBLIC_LINK_LIBRARIES O2::Framework
+  TARGETVARNAME workflowPluginsTest
+  NO_BOOST_TEST
+  COMMAND_LINE_ARGS
+    --workflow-plugin O2FrameworkTestWorkflowPluginA:TestWorkflowPluginA,O2FrameworkTestWorkflowPluginB:TestWorkflowPluginB
+    --shared 3 --run --shm-segment-size 20000000 ${DPL_WORKFLOW_TESTS_EXTRA_OPTIONS}
+  ENVIRONMENT "DYLD_LIBRARY_PATH=${CMAKE_LIBRARY_OUTPUT_DIRECTORY}:$ENV{DYLD_LIBRARY_PATH};LD_LIBRARY_PATH=${CMAKE_LIBRARY_OUTPUT_DIRECTORY}:$ENV{LD_LIBRARY_PATH}"
+  )
+if(workflowPluginsTest)
+  foreach(p A B)
+    add_library(O2FrameworkTestWorkflowPlugin${p} SHARED test/test_WorkflowPlugin${p}.cxx)
+    target_link_libraries(O2FrameworkTestWorkflowPlugin${p} PRIVATE O2::Framework)
+    target_compile_definitions(O2FrameworkTestWorkflowPlugin${p} PRIVATE
+                               DPL_WORKFLOW_PLUGIN_NAME=TestWorkflowPlugin${p}
+                               DPL_WORKFLOW_PLUGIN_LIBRARY=O2FrameworkTestWorkflowPlugin${p})
+    add_dependencies(${workflowPluginsTest} O2FrameworkTestWorkflowPlugin${p})
+  endforeach()
+endif()
+
 # the test is compiled from the ExternalFairMQDeviceWorkflow test and run with
 # command line option to include the output proxy
 o2_add_test(
diff --git a/Framework/Core/include/Framework/WorkflowDefinitionContext.h b/Framework/Core/include/Framework/WorkflowDefinitionContext.h
index db17c17da1946..c8c6e2201284e 100644
--- a/Framework/Core/include/Framework/WorkflowDefinitionContext.h
+++ b/Framework/Core/include/Framework/WorkflowDefinitionContext.h
@@ -50,12 +50,23 @@ struct WorkflowDefinitionContext {
 };
 
 struct WorkflowDefinition {
-  std::function<o2::framework::WorkflowDefinitionContext(int argc, char** argv)> defineWorkflow;
+  /// Fills the workflow options and the policies the plugin customises.
+  std::function<void(o2::framework::WorkflowDefinitionContext& context)> customize;
+  /// Fills the data processors and the channel policies of the plugin,
+  /// using the workflow options parsed in @a configContext.
+  std::function<void(o2::framework::WorkflowDefinitionContext& context, o2::framework::ConfigContext& configContext)> defineWorkflow;
 };
 
 struct WorkflowPlugin {
   virtual o2::framework::WorkflowDefinition* create() = 0;
 };
 
+/// Defines the workflow made of all the @a definitions, called @a names in errors.
+/// The workflow options of all of them are parsed once from @a argc / @a argv, then
+/// their data processors and policies are merged, in the order of @a definitions.
+/// An option may be declared by several definitions only with the same type and default,
+/// and a data processor name may not be used by more than one definition.
+WorkflowDefinitionContext defineWorkflow(std::vector<WorkflowDefinition> const& definitions, std::vector<std::string> const& names, int argc, char** argv);
+
 } // namespace o2::framework
 #endif // O2_FRAMEWORK_WORKFLOWDEFINITIONCONTEXT_H_
diff --git a/Framework/Core/include/Framework/runDataProcessing.h b/Framework/Core/include/Framework/runDataProcessing.h
index bd40e99e46e46..aeca82cd3572d 100644
--- a/Framework/Core/include/Framework/runDataProcessing.h
+++ b/Framework/Core/include/Framework/runDataProcessing.h
@@ -185,33 +185,22 @@ struct DPL_WORKFLOW_PLUGIN_NAME : o2::framework::WorkflowPlugin {
   o2::framework::WorkflowDefinition* create() override
   {
     return new o2::framework::WorkflowDefinition{
-      .defineWorkflow = [](int argc, char** argv) -> o2::framework::WorkflowDefinitionContext {
+      .customize = [](o2::framework::WorkflowDefinitionContext& workflowContext) {
         using namespace o2::framework;
-        WorkflowDefinitionContext workflowContext;
-
         UserCustomizationsHelper::userDefinedCustomization(workflowContext.workflowOptions);
-        auto requiredWorkflowOptions = WorkflowCustomizationHelpers::requiredWorkflowOptions();
-        workflowContext.workflowOptions.insert(std::end(workflowContext.workflowOptions), std::begin(requiredWorkflowOptions), std::end(requiredWorkflowOptions));
-
-        workflowContext.completionPolicies = injectCustomizations<CompletionPolicy>();
-        workflowContext.dispatchPolicies = injectCustomizations<DispatchPolicy>();
-        workflowContext.resourcePolicies = injectCustomizations<ResourcePolicy>();
-        workflowContext.callbacksPolicies = injectCustomizations<CallbacksPolicy>();
-        workflowContext.sendingPolicies = injectCustomizations<SendingPolicy>();
-
-        workflowContext.configRegistry = createRegistry();
-        workflowContext.configContext = createConfigContext(workflowContext.workflowOptionsRegistry, *workflowContext.configRegistry, workflowContext.workflowOptions, workflowContext.extraOptions, argc, argv);
-
-        workflowContext.specs = defineDataProcessing(*workflowContext.configContext);
-        overrideAll(*workflowContext.configContext, workflowContext.specs);
+        UserCustomizationsHelper::userDefinedCustomization(workflowContext.completionPolicies);
+        UserCustomizationsHelper::userDefinedCustomization(workflowContext.dispatchPolicies);
+        UserCustomizationsHelper::userDefinedCustomization(workflowContext.resourcePolicies);
+        UserCustomizationsHelper::userDefinedCustomization(workflowContext.callbacksPolicies);
+        UserCustomizationsHelper::userDefinedCustomization(workflowContext.sendingPolicies); },
+      .defineWorkflow = [](o2::framework::WorkflowDefinitionContext& workflowContext, o2::framework::ConfigContext& configContext) {
+        using namespace o2::framework;
+        workflowContext.specs = defineDataProcessing(configContext);
+        overrideAll(configContext, workflowContext.specs);
         for (auto& spec : workflowContext.specs) {
           UserCustomizationsHelper::userDefinedCustomization(spec.requiredServices);
         }
-        UserCustomizationsHelper::userDefinedCustomization(workflowContext.channelPolicies);
-        auto defaultChannelPolicies = ChannelConfigurationPolicy::createDefaultPolicies(*workflowContext.configContext);
-        workflowContext.channelPolicies.insert(std::end(workflowContext.channelPolicies), std::begin(defaultChannelPolicies), std::end(defaultChannelPolicies));
-        return workflowContext;
-      }};
+        UserCustomizationsHelper::userDefinedCustomization(workflowContext.channelPolicies); }};
   }
 };
 
diff --git a/Framework/Core/src/DeviceSpecHelpers.cxx b/Framework/Core/src/DeviceSpecHelpers.cxx
index 4c19e7a6ff17b..31f447790f8f3 100644
--- a/Framework/Core/src/DeviceSpecHelpers.cxx
+++ b/Framework/Core/src/DeviceSpecHelpers.cxx
@@ -1490,6 +1490,8 @@ void DeviceSpecHelpers::prepareArguments(bool defaultQuiet, bool defaultStopped,
     auto forwardedOptions = getForwardedDeviceOptions();
     /// Add to foDesc the options which are not already there
     foDesc.add(forwardedOptions);
+    // Children need the same workflow plugins to define the same workflow.
+    foDesc.add_options()("workflow-plugin", bpo::value<std::string>());
 
     // has option --session been specified on the command line?
     bool haveSessionArg = false;
diff --git a/Framework/Core/src/runDataProcessing.cxx b/Framework/Core/src/runDataProcessing.cxx
index 056e150448d40..337bb5899ab25 100644
--- a/Framework/Core/src/runDataProcessing.cxx
+++ b/Framework/Core/src/runDataProcessing.cxx
@@ -66,6 +66,7 @@
 #include "Framework/ServiceMetricsInfo.h"
 #include "Framework/DataTakingContext.h"
 #include "Framework/WorkflowDefinitionContext.h"
+#include "Framework/WorkflowCustomizationHelpers.h"
 #include "Framework/CommonServices.h"
 #include "Framework/DefaultsHelpers.h"
 #include "ProcessingPoliciesHelpers.h"
@@ -202,19 +203,35 @@ int doMain(int argc, char** argv, o2::framework::WorkflowDefinitionContext& work
 int callMain(int argc, char** argv, char const* pluginSpec, DPLPluginHandle* (*embeddedPlugins)(DPLPluginHandle*))
 {
   auto mainNoCatch = [pluginSpec, embeddedPlugins](int argc, char** argv) {
-    auto plugins = PluginManager::parsePluginSpecString(pluginSpec);
+    // --workflow-plugin replaces the plugin embedded in the executable.
+    char const* spec = pluginSpec;
+    for (int ai = 1; ai < argc; ++ai) {
+      if (strcmp(argv[ai], "--workflow-plugin") == 0 && ai + 1 < argc) {
+        spec = argv[ai + 1];
+      } else if (strncmp(argv[ai], "--workflow-plugin=", 18) == 0) {
+        spec = argv[ai] + 18;
+      }
+    }
+    auto plugins = PluginManager::parsePluginSpecString(spec);
     std::vector<WorkflowDefinition> workflows;
-    if (plugins.size() == 1 && plugins[0].library.empty()) {
-      if (auto* plugin = PluginManager::getByName<WorkflowPlugin>(embeddedPlugins(nullptr), plugins[0].name.c_str())) {
-        workflows.push_back(*plugin->create());
+    std::vector<std::string> names;
+    for (auto& plugin : plugins) {
+      if (plugin.library.empty()) {
+        if (auto* embedded = PluginManager::getByName<WorkflowPlugin>(embeddedPlugins(nullptr), plugin.name.c_str())) {
+          workflows.push_back(*embedded->create());
+        }
+      } else {
+        PluginManager::loadFromPlugin<WorkflowDefinition, WorkflowPlugin>({plugin}, workflows);
+      }
+      names.push_back(plugin.library + ":" + plugin.name);
+      if (workflows.size() != names.size()) {
+        throw runtime_error_f("Could not load the workflow plugin %s.", names.back().c_str());
       }
-    } else {
-      PluginManager::loadFromPlugin<WorkflowDefinition, WorkflowPlugin>(plugins, workflows);
     }
-    if (workflows.size() != 1) {
-      throw runtime_error_f("Expected exactly one workflow plugin in %s, found %zu.", pluginSpec, workflows.size());
+    if (workflows.empty()) {
+      throw runtime_error_f("No workflow plugin in %s.", spec);
     }
-    auto workflowContext = workflows.back().defineWorkflow(argc, argv);
+    auto workflowContext = defineWorkflow(workflows, names, argc, argv);
     return doMain(argc, argv, workflowContext);
   };
   static bool noCatch = getenv("O2_NO_CATCHALL_EXCEPTIONS") && strcmp(getenv("O2_NO_CATCHALL_EXCEPTIONS"), "0");
@@ -1085,6 +1102,7 @@ int doChild(int argc, char** argv, ServiceRegistry& serviceRegistry,
       ("data-processing-timeout", bpo::value<std::string>()->default_value(defaultDataProcessingTimeout), "how many second to wait before stopping data processing and allowing data calibration") //
       ("timeframes-rate-limit", bpo::value<std::string>()->default_value("0"), "how many timeframe can be in flight at the same moment (0 disables)")                                              //
       ("configuration,cfg", bpo::value<std::string>()->default_value("command-line"), "configuration backend")                                                                                     //
+      ("workflow-plugin", bpo::value<std::string>(), "workflow plugins to load")                                                                                                                   //
       ("infologger-mode", bpo::value<std::string>()->default_value(defaultInfologgerMode), "O2_INFOLOGGER_MODE override")                                                                          //
       ("log-timestamp-us", bpo::value<bool>()->zero_tokens()->default_value(false), "enable microsecond timestamps in log messages");
     r.fConfig.AddToCmdLineOptions(optsDesc, true);
@@ -2926,6 +2944,59 @@ std::unique_ptr<o2::framework::ServiceRegistry> createRegistry()
   return std::make_unique<o2::framework::ServiceRegistry>();
 }
 
+o2::framework::WorkflowDefinitionContext o2::framework::defineWorkflow(std::vector<WorkflowDefinition> const& definitions, std::vector<std::string> const& names, int argc, char** argv)
+{
+  auto append = [](auto& to, auto const& from) { to.insert(to.end(), from.begin(), from.end()); };
+  auto toString = [](Variant const& value) { std::ostringstream oss; oss << value; return oss.str(); };
+  WorkflowDefinitionContext result;
+  std::vector<WorkflowDefinitionContext> contexts(definitions.size());
+  for (size_t di = 0; di < definitions.size(); ++di) {
+    auto& context = contexts[di];
+    definitions[di].customize(context);
+    // Options already declared by a previous definition are shared, if they agree.
+    auto previous = result.workflowOptions.size();
+    for (auto& option : context.workflowOptions) {
+      auto end = result.workflowOptions.begin() + previous;
+      auto found = std::find_if(result.workflowOptions.begin(), end, [&option](auto const& other) { return other.name == option.name; });
+      if (found == end) {
+        result.workflowOptions.push_back(option);
+      } else if (found->type != option.type || toString(found->defaultValue) != toString(option.defaultValue)) {
+        throw runtime_error_f("Workflow option %s of %s has a different type or default than in a previous plugin.", option.name.c_str(), names[di].c_str());
+      }
+    }
+    append(result.completionPolicies, context.completionPolicies);
+    append(result.dispatchPolicies, context.dispatchPolicies);
+    append(result.resourcePolicies, context.resourcePolicies);
+    append(result.callbacksPolicies, context.callbacksPolicies);
+    append(result.sendingPolicies, context.sendingPolicies);
+  }
+  append(result.workflowOptions, WorkflowCustomizationHelpers::requiredWorkflowOptions());
+  append(result.completionPolicies, CompletionPolicy::createDefaultPolicies());
+  append(result.dispatchPolicies, DispatchPolicy::createDefaultPolicies());
+  append(result.resourcePolicies, ResourcePolicy::createDefaultPolicies());
+  append(result.callbacksPolicies, CallbacksPolicy::createDefaultPolicies());
+  append(result.sendingPolicies, SendingPolicy::createDefaultPolicies());
+
+  result.configRegistry = createRegistry();
+  result.configContext = createConfigContext(result.workflowOptionsRegistry, *result.configRegistry, result.workflowOptions, result.extraOptions, argc, argv);
+
+  std::map<std::string, size_t> definedBy;
+  for (size_t di = 0; di < definitions.size(); ++di) {
+    auto& context = contexts[di];
+    definitions[di].defineWorkflow(context, *result.configContext);
+    for (auto& spec : context.specs) {
+      auto [it, inserted] = definedBy.try_emplace(spec.name, di);
+      if (!inserted && it->second != di) {
+        throw runtime_error_f("Data processor %s is defined by both %s and %s.", spec.name.c_str(), names[it->second].c_str(), names[di].c_str());
+      }
+    }
+    append(result.specs, context.specs);
+    append(result.channelPolicies, context.channelPolicies);
+  }
+  append(result.channelPolicies, ChannelConfigurationPolicy::createDefaultPolicies(*result.configContext));
+  return result;
+}
+
 // This is a toy executor for the workflow spec
 // What it needs to do is:
 //
@@ -3010,6 +3081,7 @@ int doMain(int argc, char** argv, o2::framework::WorkflowDefinitionContext& work
     ("id,i", bpo::value<std::string>(), "device id for child spawning")                 //
     ("channel-config", bpo::value<std::vector<std::string>>(), "channel configuration") //
     ("control", "control plugin")                                                       //
+    ("workflow-plugin", bpo::value<std::string>(), "workflow plugins to load")          //
     ("log-color", "logging color scheme")("color", "logging color scheme");
 
   bpo::options_description visibleOptions;
diff --git a/Framework/Core/test/test_WorkflowDefinition.cxx b/Framework/Core/test/test_WorkflowDefinition.cxx
new file mode 100644
index 0000000000000..4a57be4742607
--- /dev/null
+++ b/Framework/Core/test/test_WorkflowDefinition.cxx
@@ -0,0 +1,95 @@
+// Copyright 2019-2026 CERN and copyright holders of ALICE O2.
+// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
+// All rights not expressly granted are reserved.
+//
+// This software is distributed under the terms of the GNU General Public
+// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
+//
+// In applying this license CERN does not waive the privileges and immunities
+// granted to it by virtue of its status as an Intergovernmental Organization
+// or submit itself to any jurisdiction.
+
+#include "Framework/WorkflowDefinitionContext.h"
+#include "Framework/CompletionPolicyHelpers.h"
+#include <catch_amalgamated.hpp>
+#include <algorithm>
+
+using namespace o2::framework;
+
+namespace
+{
+// A plugin with a processor called @a processor, which declares the
+// workflow options @a options and records the value of "shared" it sees.
+WorkflowDefinition makeDefinition(std::string processor, std::vector<ConfigParamSpec> options, int& shared)
+{
+  return WorkflowDefinition{
+    .customize = [processor, options](WorkflowDefinitionContext& context) {
+      context.workflowOptions = options;
+      context.completionPolicies.push_back(CompletionPolicyHelpers::consumeWhenAll(processor.c_str(), [](auto const&) { return false; })); },
+    .defineWorkflow = [processor, &shared](WorkflowDefinitionContext& context, ConfigContext& configContext) {
+      shared = configContext.options().get<int>("shared");
+      context.specs.push_back(DataProcessorSpec{.name = processor}); }};
+}
+
+size_t countOption(WorkflowDefinitionContext const& context, std::string const& name)
+{
+  return std::count_if(context.workflowOptions.begin(), context.workflowOptions.end(), [&name](auto const& option) { return option.name == name; });
+}
+} // namespace
+
+TEST_CASE("TestMergeWorkflowDefinitions")
+{
+  int sharedA = 0;
+  int sharedB = 0;
+  std::vector<WorkflowDefinition> definitions{
+    makeDefinition("a", {{"shared", VariantType::Int, 1, {"shared option"}}, {"a-only", VariantType::Int, 2, {"option of a"}}}, sharedA),
+    makeDefinition("b", {{"shared", VariantType::Int, 1, {"shared option"}}, {"b-only", VariantType::String, "x", {"option of b"}}}, sharedB)};
+  char const* args[] = {"test", "--shared", "3", "--b-only", "y"};
+  auto context = defineWorkflow(definitions, {"libA:a", "libB:b"}, 5, const_cast<char**>(args));
+
+  // Options are parsed once and each plugin sees the value of the shared one.
+  REQUIRE(countOption(context, "shared") == 1);
+  REQUIRE(countOption(context, "a-only") == 1);
+  REQUIRE(countOption(context, "b-only") == 1);
+  REQUIRE(sharedA == 3);
+  REQUIRE(sharedB == 3);
+  REQUIRE(context.configContext->options().get<std::string>("b-only") == "y");
+  // Options of the plugins come first, then those required by DPL.
+  REQUIRE(context.workflowOptions[0].name == "shared");
+  REQUIRE(context.workflowOptions[1].name == "a-only");
+  REQUIRE(context.workflowOptions[2].name == "b-only");
+  REQUIRE(countOption(context, "pipeline") == 1);
+
+  // Data processors and policies are merged in plugin order, the default policies last.
+  REQUIRE(context.specs.size() == 2);
+  REQUIRE(context.specs[0].name == "a");
+  REQUIRE(context.specs[1].name == "b");
+  REQUIRE(context.completionPolicies.size() == 2 + CompletionPolicy::createDefaultPolicies().size());
+  REQUIRE(context.completionPolicies[0].name == "a");
+  REQUIRE(context.completionPolicies[1].name == "b");
+  REQUIRE(context.completionPolicies[2].name == CompletionPolicy::createDefaultPolicies()[0].name);
+}
+
+TEST_CASE("TestConflictingWorkflowOptions")
+{
+  int shared = 0;
+  char const* args[] = {"test"};
+  // Same option with a different default.
+  std::vector<WorkflowDefinition> definitions{
+    makeDefinition("a", {{"shared", VariantType::Int, 1, {"shared option"}}}, shared),
+    makeDefinition("b", {{"shared", VariantType::Int, 2, {"shared option"}}}, shared)};
+  REQUIRE_THROWS(defineWorkflow(definitions, {"libA:a", "libB:b"}, 1, const_cast<char**>(args)));
+  // Same option with a different type.
+  definitions[1] = makeDefinition("b", {{"shared", VariantType::Float, 1.f, {"shared option"}}}, shared);
+  REQUIRE_THROWS(defineWorkflow(definitions, {"libA:a", "libB:b"}, 1, const_cast<char**>(args)));
+}
+
+TEST_CASE("TestDuplicateDataProcessor")
+{
+  int shared = 0;
+  char const* args[] = {"test"};
+  std::vector<WorkflowDefinition> definitions{
+    makeDefinition("same", {{"shared", VariantType::Int, 1, {"shared option"}}}, shared),
+    makeDefinition("same", {{"shared", VariantType::Int, 1, {"shared option"}}}, shared)};
+  REQUIRE_THROWS(defineWorkflow(definitions, {"libA:a", "libB:b"}, 1, const_cast<char**>(args)));
+}
diff --git a/Framework/Core/test/test_WorkflowPluginA.cxx b/Framework/Core/test/test_WorkflowPluginA.cxx
new file mode 100644
index 0000000000000..cc97a4191711e
--- /dev/null
+++ b/Framework/Core/test/test_WorkflowPluginA.cxx
@@ -0,0 +1,39 @@
+// Copyright 2019-2026 CERN and copyright holders of ALICE O2.
+// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
+// All rights not expressly granted are reserved.
+//
+// This software is distributed under the terms of the GNU General Public
+// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
+//
+// In applying this license CERN does not waive the privileges and immunities
+// granted to it by virtue of its status as an Intergovernmental Organization
+// or submit itself to any jurisdiction.
+
+// Workflow plugin built in a DSO, loaded by test_WorkflowPlugins
+// together with test_WorkflowPluginB.
+#include "Framework/ConfigParamSpec.h"
+#include <vector>
+
+void customize(std::vector<o2::framework::ConfigParamSpec>& options)
+{
+  options.push_back({"shared", o2::framework::VariantType::Int, 1, {"option declared by both plugins"}});
+  options.push_back({"a-only", o2::framework::VariantType::Int, 2, {"option declared only by plugin A"}});
+}
+
+#include "Framework/ControlService.h"
+#include "Framework/runDataProcessing.h"
+
+using namespace o2::framework;
+
+WorkflowSpec defineDataProcessing(ConfigContext const& context)
+{
+  auto shared = context.options().get<int>("shared");
+  return {DataProcessorSpec{
+    .name = "plugin-producer",
+    .outputs = {OutputSpec{{"value"}, "TST", "PLUGIN", 0}},
+    .algorithm = AlgorithmSpec{adaptStateless([shared](DataAllocator& outputs, ControlService& control) {
+      outputs.make<int>(OutputRef{"value"}) = shared;
+      control.endOfStream();
+      control.readyToQuit(QuitRequest::Me);
+    })}}};
+}
diff --git a/Framework/Core/test/test_WorkflowPluginB.cxx b/Framework/Core/test/test_WorkflowPluginB.cxx
new file mode 100644
index 0000000000000..82bc9f75b42cf
--- /dev/null
+++ b/Framework/Core/test/test_WorkflowPluginB.cxx
@@ -0,0 +1,45 @@
+// Copyright 2019-2026 CERN and copyright holders of ALICE O2.
+// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
+// All rights not expressly granted are reserved.
+//
+// This software is distributed under the terms of the GNU General Public
+// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
+//
+// In applying this license CERN does not waive the privileges and immunities
+// granted to it by virtue of its status as an Intergovernmental Organization
+// or submit itself to any jurisdiction.
+
+// Workflow plugin built in a DSO, loaded by test_WorkflowPlugins
+// together with test_WorkflowPluginA.
+#include "Framework/ConfigParamSpec.h"
+#include <vector>
+
+void customize(std::vector<o2::framework::ConfigParamSpec>& options)
+{
+  options.push_back({"shared", o2::framework::VariantType::Int, 1, {"option declared by both plugins"}});
+  options.push_back({"b-only", o2::framework::VariantType::String, "b", {"option declared only by plugin B"}});
+}
+
+#include "Framework/ControlService.h"
+#include "Framework/runDataProcessing.h"
+
+using namespace o2::framework;
+
+WorkflowSpec defineDataProcessing(ConfigContext const& context)
+{
+  auto shared = context.options().get<int>("shared");
+  auto expected = context.options().get<int>("a-only") + 1;
+  return {DataProcessorSpec{
+    .name = "plugin-consumer",
+    .inputs = {InputSpec{"value", "TST", "PLUGIN", 0}},
+    .algorithm = AlgorithmSpec{adaptStateless([shared, expected](InputRecord& inputs, ControlService& control) {
+      auto value = inputs.get<int>("value");
+      // The test passes --shared 3 and --a-only 2: both plugins see the same value of the
+      // shared option and plugin B sees the options of plugin A, since they are parsed once.
+      if (value != shared || shared != expected) {
+        LOGP(fatal, "Got {} from plugin A, {} in plugin B, {} expected", value, shared, expected);
+      }
+      LOGP(info, "Got {} from plugin A, as expected", value);
+      control.readyToQuit(QuitRequest::All);
+    })}}};
+}
diff --git a/Framework/Core/test/test_WorkflowPlugins.cxx b/Framework/Core/test/test_WorkflowPlugins.cxx
new file mode 100644
index 0000000000000..d082a51efb145
--- /dev/null
+++ b/Framework/Core/test/test_WorkflowPlugins.cxx
@@ -0,0 +1,22 @@
+// Copyright 2019-2026 CERN and copyright holders of ALICE O2.
+// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
+// All rights not expressly granted are reserved.
+//
+// This software is distributed under the terms of the GNU General Public
+// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
+//
+// In applying this license CERN does not waive the privileges and immunities
+// granted to it by virtue of its status as an Intergovernmental Organization
+// or submit itself to any jurisdiction.
+
+// Loads the workflow plugins of test_WorkflowPluginA and test_WorkflowPluginB,
+// given with --workflow-plugin, in place of its own.
+#include "Framework/runDataProcessing.h"
+#include "Framework/RuntimeError.h"
+
+using namespace o2::framework;
+
+WorkflowSpec defineDataProcessing(ConfigContext const&)
+{
+  throw runtime_error("The embedded workflow must not be used when --workflow-plugin is given.");
+}

From 13cd2ac2db60809636ab272e1b42f0d1f1ffbac6 Mon Sep 17 00:00:00 2001
From: Giulio Eulisse <10544+ktf@users.noreply.github.com>
Date: Thu, 5 Nov 2026 23:15:40 +0100
Subject: [PATCH 3/5] DPL: record the workflow plugin of each data processor

DataProcessorInfo and DeviceExecution get the plugin which defines the
data processor, as <library>:<plugin>, and the workflow JSON gets a
"plugin" key in the metadata of the data processor. This is what a
driver needs to load the plugins of a workflow itself.

Plugins embedded in the executable, i.e. with an empty library, are not
recorded: their plugin stays empty and the key is not written, so the
dumps of all existing executables are unchanged. Internal data
processors added by DPL have no plugin either. Data processors imported
from an upstream workflow keep the plugin they had there.

Tested by TestDataProcessorInfoPlugin, a round trip of a workflow with
and without plugins, and test_Framework_test_WorkflowPluginsDump, which
dumps the workflow of the two test plugins.
---
 Framework/Core/CMakeLists.txt                 |  5 +++
 .../include/Framework/DataProcessorInfo.h     |  3 ++
 .../Core/include/Framework/DeviceExecution.h  |  3 ++
 .../Framework/WorkflowDefinitionContext.h     |  3 ++
 Framework/Core/src/DeviceSpecHelpers.cxx      |  1 +
 .../Core/src/WorkflowSerializationHelpers.cxx | 13 +++++++
 Framework/Core/src/runDataProcessing.cxx      | 10 ++++-
 .../Core/test/test_WorkflowPluginsDump.sh     | 10 +++++
 .../Core/test/test_WorkflowSerialization.cxx  | 39 +++++++++++++++++++
 9 files changed, 85 insertions(+), 2 deletions(-)
 create mode 100755 Framework/Core/test/test_WorkflowPluginsDump.sh

diff --git a/Framework/Core/CMakeLists.txt b/Framework/Core/CMakeLists.txt
index 222ceb48ea757..03678319846ea 100644
--- a/Framework/Core/CMakeLists.txt
+++ b/Framework/Core/CMakeLists.txt
@@ -472,6 +472,11 @@ if(workflowPluginsTest)
                                DPL_WORKFLOW_PLUGIN_LIBRARY=O2FrameworkTestWorkflowPlugin${p})
     add_dependencies(${workflowPluginsTest} O2FrameworkTestWorkflowPlugin${p})
   endforeach()
+  add_test(NAME test_Framework_test_WorkflowPluginsDump
+           COMMAND ${CMAKE_CURRENT_LIST_DIR}/test/test_WorkflowPluginsDump.sh $<TARGET_FILE:${workflowPluginsTest}>)
+  set_property(TEST test_Framework_test_WorkflowPluginsDump PROPERTY LABELS framework)
+  set_property(TEST test_Framework_test_WorkflowPluginsDump PROPERTY ENVIRONMENT
+               "DYLD_LIBRARY_PATH=${CMAKE_LIBRARY_OUTPUT_DIRECTORY}:$ENV{DYLD_LIBRARY_PATH};LD_LIBRARY_PATH=${CMAKE_LIBRARY_OUTPUT_DIRECTORY}:$ENV{LD_LIBRARY_PATH}")
 endif()
 
 # the test is compiled from the ExternalFairMQDeviceWorkflow test and run with
diff --git a/Framework/Core/include/Framework/DataProcessorInfo.h b/Framework/Core/include/Framework/DataProcessorInfo.h
index 23c9d3f722bd4..149703072aa48 100644
--- a/Framework/Core/include/Framework/DataProcessorInfo.h
+++ b/Framework/Core/include/Framework/DataProcessorInfo.h
@@ -32,6 +32,9 @@ struct DataProcessorInfo {
   std::vector<ConfigParamSpec> workflowOptions = {};
   /// The channels for a given dataprocessor
   std::vector<std::string> channels = {};
+  /// The workflow plugin which defines the DataProcessorSpec, as <library>:<plugin>.
+  /// Empty when it is the one embedded in the executable.
+  std::string plugin = "";
 };
 
 } // namespace o2
diff --git a/Framework/Core/include/Framework/DeviceExecution.h b/Framework/Core/include/Framework/DeviceExecution.h
index d417a6980cf58..da35153e3d157 100644
--- a/Framework/Core/include/Framework/DeviceExecution.h
+++ b/Framework/Core/include/Framework/DeviceExecution.h
@@ -11,6 +11,7 @@
 #ifndef FRAMEWORK_DEVICEEXECUTION_H
 #define FRAMEWORK_DEVICEEXECUTION_H
 
+#include <string>
 #include <vector>
 
 namespace o2
@@ -26,6 +27,8 @@ struct DeviceExecution {
   std::vector<char*> args;
   /// The environment to be passed to a given device
   std::vector<char*> environ;
+  /// The workflow plugin which defines the device, see DataProcessorInfo::plugin
+  std::string plugin;
 };
 
 } // namespace framework
diff --git a/Framework/Core/include/Framework/WorkflowDefinitionContext.h b/Framework/Core/include/Framework/WorkflowDefinitionContext.h
index c8c6e2201284e..9ed128b42b40b 100644
--- a/Framework/Core/include/Framework/WorkflowDefinitionContext.h
+++ b/Framework/Core/include/Framework/WorkflowDefinitionContext.h
@@ -23,6 +23,7 @@
 #include "Framework/ChannelConfigurationPolicy.h"
 #include "Framework/ServiceRegistry.h"
 #include <functional>
+#include <map>
 #include <memory>
 #include <vector>
 
@@ -47,6 +48,8 @@ struct WorkflowDefinitionContext {
   std::unique_ptr<ConfigParamRegistry> workflowOptionsRegistry{nullptr};
 
   o2::framework::WorkflowSpec specs;
+  /// The plugin which defines each data processor, if not embedded in the executable.
+  std::map<std::string, std::string> processorPlugins;
 };
 
 struct WorkflowDefinition {
diff --git a/Framework/Core/src/DeviceSpecHelpers.cxx b/Framework/Core/src/DeviceSpecHelpers.cxx
index 31f447790f8f3..4114faaec5d1e 100644
--- a/Framework/Core/src/DeviceSpecHelpers.cxx
+++ b/Framework/Core/src/DeviceSpecHelpers.cxx
@@ -1749,6 +1749,7 @@ void DeviceSpecHelpers::prepareArguments(bool defaultQuiet, bool defaultStopped,
     }
     O2_SIGNPOST_END(device_spec_helpers, poid, "prepareArguments", "The following options are being forwarded to %{public}s: %{public}s",
                     spec.id.c_str(), str.str().c_str());
+    execution.plugin = pi->plugin;
   }
 }
 
diff --git a/Framework/Core/src/WorkflowSerializationHelpers.cxx b/Framework/Core/src/WorkflowSerializationHelpers.cxx
index f0d4df7ec0d0b..ee5118f0917f0 100644
--- a/Framework/Core/src/WorkflowSerializationHelpers.cxx
+++ b/Framework/Core/src/WorkflowSerializationHelpers.cxx
@@ -92,6 +92,7 @@ struct WorkflowImporter : public rapidjson::BaseReaderHandler<rapidjson::UTF8<>,
     IN_DATAPROCESSOR_INFO,
     IN_DATAPROCESSOR_INFO_NAME,
     IN_DATAPROCESSOR_INFO_EXECUTABLE,
+    IN_DATAPROCESSOR_INFO_PLUGIN,
     IN_DATAPROCESSOR_INFO_ARGS,
     IN_DATAPROCESSOR_INFO_ARG,
     IN_DATAPROCESSOR_INFO_CHANNELS,
@@ -264,6 +265,9 @@ struct WorkflowImporter : public rapidjson::BaseReaderHandler<rapidjson::UTF8<>,
       case State::IN_DATAPROCESSOR_INFO_EXECUTABLE:
         s << "IN_DATAPROCESSOR_INFO_EXECUTABLE";
         break;
+      case State::IN_DATAPROCESSOR_INFO_PLUGIN:
+        s << "IN_DATAPROCESSOR_INFO_PLUGIN";
+        break;
       case State::IN_DATAPROCESSOR_INFO_ARGS:
         s << "IN_DATAPROCESSOR_INFO_ARGS";
         break;
@@ -708,6 +712,8 @@ struct WorkflowImporter : public rapidjson::BaseReaderHandler<rapidjson::UTF8<>,
       push(State::IN_DATAPROCESSOR_INFO_NAME);
     } else if (in(State::IN_DATAPROCESSOR_INFO) && strncmp(str, "executable", length) == 0) {
       push(State::IN_DATAPROCESSOR_INFO_EXECUTABLE);
+    } else if (in(State::IN_DATAPROCESSOR_INFO) && strncmp(str, "plugin", length) == 0) {
+      push(State::IN_DATAPROCESSOR_INFO_PLUGIN);
     } else if (in(State::IN_DATAPROCESSOR_INFO) && strncmp(str, "cmdLineArgs", length) == 0) {
       push(State::IN_DATAPROCESSOR_INFO_ARGS);
     } else if (in(State::IN_DATAPROCESSOR_INFO) && strncmp(str, "workflowOptions", length) == 0) {
@@ -734,6 +740,9 @@ struct WorkflowImporter : public rapidjson::BaseReaderHandler<rapidjson::UTF8<>,
     } else if (in(State::IN_DATAPROCESSOR_INFO_EXECUTABLE)) {
       assert(metadata.size());
       metadata.back().executable = s;
+    } else if (in(State::IN_DATAPROCESSOR_INFO_PLUGIN)) {
+      assert(metadata.size());
+      metadata.back().plugin = s;
     } else if (in(State::IN_INPUT_BINDING)) {
       binding = s;
     } else if (in(State::IN_INPUT_ORIGIN)) {
@@ -1270,6 +1279,10 @@ void WorkflowSerializationHelpers::dump(std::ostream& out,
     w.String(info.name.c_str());
     w.Key("executable");
     w.String(info.executable.c_str());
+    if (!info.plugin.empty()) {
+      w.Key("plugin");
+      w.String(info.plugin.c_str());
+    }
     w.Key("cmdLineArgs");
     w.StartArray();
     for (auto& arg : info.cmdLineArgs) {
diff --git a/Framework/Core/src/runDataProcessing.cxx b/Framework/Core/src/runDataProcessing.cxx
index 337bb5899ab25..fec9f325b8339 100644
--- a/Framework/Core/src/runDataProcessing.cxx
+++ b/Framework/Core/src/runDataProcessing.cxx
@@ -1183,6 +1183,7 @@ struct WorkflowInfo {
   std::string executable;
   std::vector<std::string> args;
   std::vector<ConfigParamSpec> options;
+  std::map<std::string, std::string> plugins;
 };
 
 void gui_callback(uv_timer_s* ctx)
@@ -1696,7 +1697,8 @@ int runStateMachine(DataProcessorSpecs const& workflow,
               workflowInfo.executable,
               workflowInfo.args,
               workflowInfo.options,
-              channels});
+              channels,
+              workflowInfo.plugins.contains(device.name) ? workflowInfo.plugins.at(device.name) : ""});
         }
         break;
       case DriverState::MATERIALISE_WORKFLOW:
@@ -2989,6 +2991,9 @@ o2::framework::WorkflowDefinitionContext o2::framework::defineWorkflow(std::vect
       if (!inserted && it->second != di) {
         throw runtime_error_f("Data processor %s is defined by both %s and %s.", spec.name.c_str(), names[it->second].c_str(), names[di].c_str());
       }
+      if (!names[di].starts_with(":")) {
+        result.processorPlugins[spec.name] = names[di];
+      }
     }
     append(result.specs, context.specs);
     append(result.channelPolicies, context.channelPolicies);
@@ -3035,7 +3040,8 @@ int doMain(int argc, char** argv, o2::framework::WorkflowDefinitionContext& work
   WorkflowInfo currentWorkflow{
     argv[0],
     currentArgs,
-    currentWorkflowOptions};
+    currentWorkflowOptions,
+    workflowContext.processorPlugins};
 
   ProcessingPolicies processingPolicies;
   enum LogParsingHelpers::LogLevel minFailureLevel;
diff --git a/Framework/Core/test/test_WorkflowPluginsDump.sh b/Framework/Core/test/test_WorkflowPluginsDump.sh
new file mode 100755
index 0000000000000..80d626f9d05c2
--- /dev/null
+++ b/Framework/Core/test/test_WorkflowPluginsDump.sh
@@ -0,0 +1,10 @@
+#!/bin/sh -e
+# Dumps the workflow of the two test plugins and checks that it records,
+# for each data processor, the plugin which defines it.
+# usage: test_WorkflowPluginsDump.sh <o2-test-framework-WorkflowPlugins>
+A=O2FrameworkTestWorkflowPluginA:TestWorkflowPluginA
+B=O2FrameworkTestWorkflowPluginB:TestWorkflowPluginB
+dump=$("$1" -b --workflow-plugin $A,$B --dump-workflow < /dev/null)
+echo "$dump" | grep -q "\"plugin\": \"$A\""
+echo "$dump" | grep -q "\"plugin\": \"$B\""
+test "$(echo "$dump" | grep -c '"plugin"')" -eq 2
diff --git a/Framework/Core/test/test_WorkflowSerialization.cxx b/Framework/Core/test/test_WorkflowSerialization.cxx
index 8d2ca579c0ede..398bafbcefd82 100644
--- a/Framework/Core/test/test_WorkflowSerialization.cxx
+++ b/Framework/Core/test/test_WorkflowSerialization.cxx
@@ -169,3 +169,42 @@ TEST_CASE("TestInputOutputSpecMetadata")
   REQUIRE(wso[0].outputs[0].metadata[0] == wsi[0].outputs[0].metadata[0]);
   REQUIRE(wso[0].outputs[0].metadata[1] == wsi[0].outputs[0].metadata[1]);
 }
+
+/// The plugin of a data processor is only written when there is one,
+/// so that dumps of executables with an embedded plugin do not change.
+TEST_CASE("TestDataProcessorInfoPlugin")
+{
+  using namespace o2::framework;
+  WorkflowSpec w0{
+    DataProcessorSpec{.name = "A", .outputs = {OutputSpec{"TST", "A"}}},
+    DataProcessorSpec{.name = "B", .inputs = {InputSpec{"a", "TST", "A"}}}};
+
+  std::vector<DataProcessorInfo> dataProcessorInfoOut{
+    {.name = "A", .executable = "o2-dpl-workflow", .plugin = "O2SomePlugins:A"},
+    {.name = "B", .executable = "o2-dpl-workflow"},
+  };
+  CommandInfo commandInfoOut{"o2-dpl-workflow -b"};
+
+  std::ostringstream firstDump;
+  WorkflowSerializationHelpers::dump(firstDump, w0, dataProcessorInfoOut, commandInfoOut);
+  auto json = firstDump.str();
+  REQUIRE(json.find(R"("plugin": "O2SomePlugins:A")") != std::string::npos);
+  // Only A has a plugin.
+  REQUIRE(json.find(R"("plugin")") == json.rfind(R"("plugin")"));
+
+  std::vector<DataProcessorInfo> dataProcessorInfoIn{};
+  CommandInfo commandInfoIn;
+  std::istringstream is;
+  is.str(json);
+  WorkflowSpec w1;
+  WorkflowSerializationHelpers::import(is, w1, dataProcessorInfoIn, commandInfoIn);
+  REQUIRE(dataProcessorInfoIn.size() == 2);
+  REQUIRE(dataProcessorInfoIn[0].plugin == "O2SomePlugins:A");
+  REQUIRE(dataProcessorInfoIn[0].executable == "o2-dpl-workflow");
+  REQUIRE(dataProcessorInfoIn[1].plugin.empty());
+  REQUIRE(dataProcessorInfoIn[1].executable == "o2-dpl-workflow");
+
+  std::ostringstream secondDump;
+  WorkflowSerializationHelpers::dump(secondDump, w1, dataProcessorInfoIn, commandInfoIn);
+  REQUIRE(json == secondDump.str());
+}

From 6f69769ad3949df5795a763586a2c67a01a67f4b Mon Sep 17 00:00:00 2001
From: Giulio Eulisse <10544+ktf@users.noreply.github.com>
Date: Fri, 6 Nov 2026 07:19:44 +0100
Subject: [PATCH 4/5] DPL: allow building a workflow as a workflow plugin

A workflow opts in with WORKFLOW_PLUGIN, on o2_add_dpl_workflow or on
o2_add_executable. Its sources are then compiled once, into the DSO
libO2<Name>Plugin holding the workflow plugin <Name>, the CamelCase of
the executable name without o2-: o2-testworkflows-diamond-workflow gives
O2TestworkflowsDiamondWorkflowPlugin:TestworkflowsDiamondWorkflow. The
executable keeps its name and options, and only has a main() which
calls dpl_workflow_main() of the plugin it links. That one runs the
plugin as the one embedded in the executable, exactly as main() does for
any other workflow, so its behaviour and dumps are the same. The plugin
is installed in lib with the executable.

WORKFLOW_PLUGIN, rather than PLUGIN, because DPL has several kinds of
plugins and this one is DplPluginKind::Workflow, loaded with
--workflow-plugin; the same name works on both functions.

The existing o2-dpl-run, whose embedded workflow is empty, is the
generic loader: o2-dpl-run --workflow-plugin A:X,B:Y runs the workflows
of the plugins, and its children are again o2-dpl-run.

Workflows which do not opt in are built exactly as before. The diamond
workflow, the simple source, processor and sink, and the two test
plugins of test_WorkflowPlugins opt in.

test_Framework_test_WorkflowPluginDump checks that the plugin of the
diamond workflow, run by o2-dpl-run, defines the same workflow as its
executable, apart from the executable, plugin and command line.
---
 Framework/Core/CMakeLists.txt                 | 12 ++---
 .../include/Framework/runDataProcessing.h     | 17 ++++---
 .../Core/test/test_WorkflowPluginsDump.sh     |  4 +-
 Framework/TestWorkflows/CMakeLists.txt        | 22 +++++++--
 .../test/test_WorkflowPluginDump.sh           | 10 ++++
 cmake/O2AddExecutable.cmake                   | 49 ++++++++++++++++++-
 cmake/O2AddWorkflow.cmake                     | 10 ++--
 7 files changed, 98 insertions(+), 26 deletions(-)
 create mode 100755 Framework/TestWorkflows/test/test_WorkflowPluginDump.sh

diff --git a/Framework/Core/CMakeLists.txt b/Framework/Core/CMakeLists.txt
index 03678319846ea..2d7811b098efd 100644
--- a/Framework/Core/CMakeLists.txt
+++ b/Framework/Core/CMakeLists.txt
@@ -459,18 +459,16 @@ o2_add_test(
   TARGETVARNAME workflowPluginsTest
   NO_BOOST_TEST
   COMMAND_LINE_ARGS
-    --workflow-plugin O2FrameworkTestWorkflowPluginA:TestWorkflowPluginA,O2FrameworkTestWorkflowPluginB:TestWorkflowPluginB
+    --workflow-plugin O2TestFrameworkWorkflowPluginAPlugin:TestFrameworkWorkflowPluginA,O2TestFrameworkWorkflowPluginBPlugin:TestFrameworkWorkflowPluginB
     --shared 3 --run --shm-segment-size 20000000 ${DPL_WORKFLOW_TESTS_EXTRA_OPTIONS}
   ENVIRONMENT "DYLD_LIBRARY_PATH=${CMAKE_LIBRARY_OUTPUT_DIRECTORY}:$ENV{DYLD_LIBRARY_PATH};LD_LIBRARY_PATH=${CMAKE_LIBRARY_OUTPUT_DIRECTORY}:$ENV{LD_LIBRARY_PATH}"
   )
 if(workflowPluginsTest)
   foreach(p A B)
-    add_library(O2FrameworkTestWorkflowPlugin${p} SHARED test/test_WorkflowPlugin${p}.cxx)
-    target_link_libraries(O2FrameworkTestWorkflowPlugin${p} PRIVATE O2::Framework)
-    target_compile_definitions(O2FrameworkTestWorkflowPlugin${p} PRIVATE
-                               DPL_WORKFLOW_PLUGIN_NAME=TestWorkflowPlugin${p}
-                               DPL_WORKFLOW_PLUGIN_LIBRARY=O2FrameworkTestWorkflowPlugin${p})
-    add_dependencies(${workflowPluginsTest} O2FrameworkTestWorkflowPlugin${p})
+    o2_add_executable(WorkflowPlugin${p} SOURCES test/test_WorkflowPlugin${p}.cxx
+                      COMPONENT_NAME Framework PUBLIC_LINK_LIBRARIES O2::Framework
+                      IS_TEST NO_INSTALL WORKFLOW_PLUGIN TARGETVARNAME workflowPlugin)
+    add_dependencies(${workflowPluginsTest} ${workflowPlugin})
   endforeach()
   add_test(NAME test_Framework_test_WorkflowPluginsDump
            COMMAND ${CMAKE_CURRENT_LIST_DIR}/test/test_WorkflowPluginsDump.sh $<TARGET_FILE:${workflowPluginsTest}>)
diff --git a/Framework/Core/include/Framework/runDataProcessing.h b/Framework/Core/include/Framework/runDataProcessing.h
index aeca82cd3572d..f69459c01462d 100644
--- a/Framework/Core/include/Framework/runDataProcessing.h
+++ b/Framework/Core/include/Framework/runDataProcessing.h
@@ -170,15 +170,12 @@ char* getIdString(int argc, char** argv);
 #define DPL_STRINGIZE(A) DPL_STRINGIZE_NX(A)
 
 // The workflow definition is a plugin. By default it is embedded in the
-// executable as DPL_WORKFLOW_PLUGIN_NAME, with an empty library name.
-// To build it in a DSO instead, both DPL_WORKFLOW_PLUGIN_NAME and
-// DPL_WORKFLOW_PLUGIN_LIBRARY must be specified at build time.
+// executable as DPL_WORKFLOW_PLUGIN_NAME, and main() runs it. A workflow built
+// as a plugin library gets DPL_WORKFLOW_PLUGIN_NAME at build time: it has no
+// main(), and the executable which links it calls dpl_workflow_main() instead.
 #ifndef DPL_WORKFLOW_PLUGIN_NAME
 #define DPL_WORKFLOW_PLUGIN_NAME InternalCustomWorkflow
-#ifdef DPL_WORKFLOW_PLUGIN_LIBRARY
-#error Missing DPL_WORKFLOW_PLUGIN_NAME
-#endif
-#define DPL_WORKFLOW_PLUGIN_LIBRARY
+#define DPL_WORKFLOW_PLUGIN_EMBEDDED
 #endif
 
 struct DPL_WORKFLOW_PLUGIN_NAME : o2::framework::WorkflowPlugin {
@@ -216,11 +213,15 @@ DPLPluginHandle* dpl_plugin_callback(DPLPluginHandle* previous)
 // An empty library means the plugin embedded in the executable, found via @a embeddedPlugins.
 int callMain(int argc, char** argv, char const* pluginSpec, DPLPluginHandle* (*embeddedPlugins)(DPLPluginHandle*));
 
+#ifdef DPL_WORKFLOW_PLUGIN_EMBEDDED
 int main(int argc, char** argv)
+#else
+extern "C" int dpl_workflow_main(int argc, char** argv)
+#endif
 {
   using namespace o2::framework;
 
-  int result = callMain(argc, argv, DPL_STRINGIZE(DPL_WORKFLOW_PLUGIN_LIBRARY) ":" DPL_STRINGIZE(DPL_WORKFLOW_PLUGIN_NAME), dpl_plugin_callback);
+  int result = callMain(argc, argv, ":" DPL_STRINGIZE(DPL_WORKFLOW_PLUGIN_NAME), dpl_plugin_callback);
 
   char* idstring = getIdString(argc, argv);
   o2::framework::OnWorkflowTerminationHook onWorkflowTerminationHook;
diff --git a/Framework/Core/test/test_WorkflowPluginsDump.sh b/Framework/Core/test/test_WorkflowPluginsDump.sh
index 80d626f9d05c2..c85d8b33829c3 100755
--- a/Framework/Core/test/test_WorkflowPluginsDump.sh
+++ b/Framework/Core/test/test_WorkflowPluginsDump.sh
@@ -2,8 +2,8 @@
 # Dumps the workflow of the two test plugins and checks that it records,
 # for each data processor, the plugin which defines it.
 # usage: test_WorkflowPluginsDump.sh <o2-test-framework-WorkflowPlugins>
-A=O2FrameworkTestWorkflowPluginA:TestWorkflowPluginA
-B=O2FrameworkTestWorkflowPluginB:TestWorkflowPluginB
+A=O2TestFrameworkWorkflowPluginAPlugin:TestFrameworkWorkflowPluginA
+B=O2TestFrameworkWorkflowPluginBPlugin:TestFrameworkWorkflowPluginB
 dump=$("$1" -b --workflow-plugin $A,$B --dump-workflow < /dev/null)
 echo "$dump" | grep -q "\"plugin\": \"$A\""
 echo "$dump" | grep -q "\"plugin\": \"$B\""
diff --git a/Framework/TestWorkflows/CMakeLists.txt b/Framework/TestWorkflows/CMakeLists.txt
index d2b98419043bf..b47a039a6cef4 100644
--- a/Framework/TestWorkflows/CMakeLists.txt
+++ b/Framework/TestWorkflows/CMakeLists.txt
@@ -31,7 +31,8 @@ o2_add_dpl_workflow(o2rootmessage-workflow
 
 o2_add_dpl_workflow(diamond-workflow
                   SOURCES src/o2DiamondWorkflow.cxx
-                  COMPONENT_NAME TestWorkflows)
+                  COMPONENT_NAME TestWorkflows
+                  WORKFLOW_PLUGIN)
 
 o2_add_dpl_workflow(dummy-populator-workflow
                   SOURCES src/o2DummyPopulatorWorkflow.cxx
@@ -114,7 +115,8 @@ o2_add_dpl_workflow(ccdb-fetch-to-timeframe
 
 o2_add_dpl_workflow(simple-source
                   SOURCES src/o2SimpleSource.cxx
-                  COMPONENT_NAME TestWorkflows)
+                  COMPONENT_NAME TestWorkflows
+                  WORKFLOW_PLUGIN)
 
 o2_add_dpl_workflow(simple-fairmq-source
                   SOURCES src/o2SimpleFairMQSource.cxx
@@ -126,11 +128,13 @@ o2_add_dpl_workflow(simple-fairmq-sink
 
 o2_add_dpl_workflow(simple-sink
                   SOURCES src/o2SimpleSink.cxx
-                  COMPONENT_NAME TestWorkflows)
+                  COMPONENT_NAME TestWorkflows
+                  WORKFLOW_PLUGIN)
 
 o2_add_dpl_workflow(simple-processor
                   SOURCES src/o2SimpleProcessor.cxx
-                  COMPONENT_NAME TestWorkflows)
+                  COMPONENT_NAME TestWorkflows
+                  WORKFLOW_PLUGIN)
 
 o2_add_dpl_workflow(analysis-workflow
                   SOURCES src/o2AnalysisWorkflow.cxx
@@ -156,3 +160,13 @@ o2_add_executable(fairmq-header-size-test
 add_executable(o2-test-deadlock src/o2DeadlockReproducer.cxx)
 target_link_libraries(o2-test-deadlock PUBLIC FairMQ::FairMQ)
 add_test(NAME o2-test-deadlock COMMAND o2-test-deadlock --channel-config "name=data,type=sub,method=bind,address=ipc://127.0.0.1,rateLogging=0" --id foo --transport zeromq --control static)
+
+if(BUILD_TESTING)
+  add_test(NAME test_Framework_test_WorkflowPluginDump
+           COMMAND ${CMAKE_CURRENT_LIST_DIR}/test/test_WorkflowPluginDump.sh
+                   $<TARGET_FILE:O2exe-testworkflows-diamond-workflow> $<TARGET_FILE:O2exe-dpl-run>
+                   O2TestworkflowsDiamondWorkflowPlugin:TestworkflowsDiamondWorkflow)
+  set_property(TEST test_Framework_test_WorkflowPluginDump PROPERTY LABELS framework)
+  set_property(TEST test_Framework_test_WorkflowPluginDump PROPERTY ENVIRONMENT
+               "LD_LIBRARY_PATH=${CMAKE_LIBRARY_OUTPUT_DIRECTORY}:$ENV{LD_LIBRARY_PATH}")
+endif()
diff --git a/Framework/TestWorkflows/test/test_WorkflowPluginDump.sh b/Framework/TestWorkflows/test/test_WorkflowPluginDump.sh
new file mode 100755
index 0000000000000..9a21b8d92600d
--- /dev/null
+++ b/Framework/TestWorkflows/test/test_WorkflowPluginDump.sh
@@ -0,0 +1,10 @@
+#!/bin/sh -e
+# Checks that the workflow plugin of an executable, run by o2-dpl-run,
+# defines the same workflow as the executable itself.
+# usage: test_WorkflowPluginDump.sh <executable> <o2-dpl-run> <library>:<plugin>
+strip() { grep -v '"executable"\|"plugin"\|"command"\|"--workflow-plugin"\|"'"$1"'"'; }
+exe=$("$1" -b --dump-workflow < /dev/null | strip "$3")
+dso=$("$2" -b --workflow-plugin "$3" --dump-workflow < /dev/null | strip "$3")
+test -n "$exe"
+test "$exe" = "$dso"
+test "$("$2" -b --workflow-plugin "$3" --dump-workflow < /dev/null | grep -c "\"plugin\": \"$3\"")" -gt 0
diff --git a/cmake/O2AddExecutable.cmake b/cmake/O2AddExecutable.cmake
index 752678c44baf0..d759abe7dda8a 100644
--- a/cmake/O2AddExecutable.cmake
+++ b/cmake/O2AddExecutable.cmake
@@ -30,6 +30,11 @@ include_guard()
 #   regular executables (e.g. o2-tpc-... or o2-mch-...) or the origin target for
 #   tests (to help locate the source file in the source directory hierarchy,
 #   e.g. o2-test-DataFormats-...)
+# * WORKFLOW_PLUGIN (optional), for a DPL workflow: its sources are compiled
+#   into the workflow plugin <Name>, in libO2<Name>Plugin, which o2-dpl-run can
+#   load with --workflow-plugin, and the executable is only a main() linking
+#   it. <Name> is the CamelCase of the executable name without o2-, e.g.
+#   o2-tpc-reco-workflow gives O2TpcRecoWorkflowPlugin:TpcRecoWorkflow.
 #
 # Note that the _target_ corresponding to the executable will be named
 # O2exe[-exeType][component]-basename and can be retrieved with the
@@ -48,7 +53,7 @@ function(o2_add_executable baseTargetName)
   cmake_parse_arguments(PARSE_ARGV
                         1
                         A
-                        "IS_TEST;NO_INSTALL;IS_BENCHMARK"
+                        "IS_TEST;NO_INSTALL;IS_BENCHMARK;WORKFLOW_PLUGIN"
                         "COMPONENT_NAME;TARGETVARNAME"
                         "SOURCES;PUBLIC_LINK_LIBRARIES;JOB_POOL")
 
@@ -92,7 +97,18 @@ function(o2_add_executable baseTargetName)
   endif()
 
   # add the executable with its sources
-  add_executable(${target} ${A_SOURCES})
+  if(A_WORKFLOW_PLUGIN)
+    o2_add_workflow_plugin(${exeName} SOURCES ${A_SOURCES}
+                           PUBLIC_LINK_LIBRARIES ${A_PUBLIC_LINK_LIBRARIES}
+                           TARGETVARNAME plugin)
+    add_executable(${target} $<TARGET_OBJECTS:O2FrameworkWorkflowMain>)
+    target_link_libraries(${target} PUBLIC ${plugin})
+    if(NOT A_NO_INSTALL)
+      install(TARGETS ${plugin} LIBRARY DESTINATION ${CMAKE_INSTALL_LIBDIR})
+    endif()
+  else()
+    add_executable(${target} ${A_SOURCES})
+  endif()
 
   # set the executable output name
   set_property(TARGET ${target} PROPERTY OUTPUT_NAME ${exeName})
@@ -132,3 +148,32 @@ function(o2_add_executable baseTargetName)
   endif()
 
 endfunction()
+
+# o2_add_workflow_plugin(exeName SOURCES ... PUBLIC_LINK_LIBRARIES ...
+# TARGETVARNAME var) builds the workflow plugin of the executable exeName, see
+# WORKFLOW_PLUGIN above, and the main() of such executables, once.
+function(o2_add_workflow_plugin exeName)
+  cmake_parse_arguments(PARSE_ARGV 1 A "" "TARGETVARNAME" "SOURCES;PUBLIC_LINK_LIBRARIES")
+  string(REGEX REPLACE "^o2-" "" name ${exeName})
+  string(REPLACE "-" ";" parts ${name})
+  set(name "")
+  foreach(part IN LISTS parts)
+    string(SUBSTRING ${part} 0 1 first)
+    string(SUBSTRING ${part} 1 -1 rest)
+    string(TOUPPER ${first} first)
+    string(APPEND name ${first}${rest})
+  endforeach()
+  string(MAKE_C_IDENTIFIER ${name} name)
+  set(plugin O2${name}Plugin)
+  add_library(${plugin} SHARED ${A_SOURCES})
+  target_link_libraries(${plugin} PUBLIC ${A_PUBLIC_LINK_LIBRARIES})
+  target_compile_definitions(${plugin} PRIVATE DPL_WORKFLOW_PLUGIN_NAME=${name})
+  set(${A_TARGETVARNAME} ${plugin} PARENT_SCOPE)
+  if(NOT TARGET O2FrameworkWorkflowMain)
+    file(CONFIGURE OUTPUT ${CMAKE_BINARY_DIR}/O2FrameworkWorkflowMain.cxx CONTENT [[
+extern "C" int dpl_workflow_main(int argc, char** argv);
+int main(int argc, char** argv) { return dpl_workflow_main(argc, argv); }
+]])
+    add_library(O2FrameworkWorkflowMain OBJECT ${CMAKE_BINARY_DIR}/O2FrameworkWorkflowMain.cxx)
+  endif()
+endfunction()
diff --git a/cmake/O2AddWorkflow.cmake b/cmake/O2AddWorkflow.cmake
index 4d47e85420cdf..2938f807a3bd3 100644
--- a/cmake/O2AddWorkflow.cmake
+++ b/cmake/O2AddWorkflow.cmake
@@ -17,7 +17,8 @@ include_guard()
 # file via <executable-name> --dump so that it can be easily registered and
 # deployed e.g. in the train infrastructure.
 #
-# For most of the options please see how o2_add_executable works.
+# For most of the options, e.g. WORKFLOW_PLUGIN, please see how
+# o2_add_executable works.
 #
 # The installed executable will be named as regular executables (see
 # o2_add_executable for details)
@@ -28,14 +29,17 @@ include_guard()
 
 function(o2_add_dpl_workflow baseTargetName)
 
-  cmake_parse_arguments(PARSE_ARGV 1 A "" "COMPONENT_NAME;TARGETVARNAME"
+  cmake_parse_arguments(PARSE_ARGV 1 A "WORKFLOW_PLUGIN" "COMPONENT_NAME;TARGETVARNAME"
                         "SOURCES;PUBLIC_LINK_LIBRARIES;JOB_POOL")
 
   if(A_UNPARSED_ARGUMENTS)
     message(FATAL_ERROR "Got trailing arguments ${A_UNPARSED_ARGUMENTS}")
   endif()
 
-  o2_add_executable(${baseTargetName}
+  if(A_WORKFLOW_PLUGIN)
+    set(workflowPlugin WORKFLOW_PLUGIN)
+  endif()
+  o2_add_executable(${baseTargetName} ${workflowPlugin}
     COMPONENT_NAME ${A_COMPONENT_NAME} TARGETVARNAME targetExeName
     SOURCES ${A_SOURCES}
     PUBLIC_LINK_LIBRARIES O2::Framework ${A_PUBLIC_LINK_LIBRARIES})

From e6acbe59e41f21a11ee173771bc620d9448c2c4f Mon Sep 17 00:00:00 2001
From: Giulio Eulisse <10544+ktf@users.noreply.github.com>
Date: Fri, 6 Nov 2026 07:19:45 +0100
Subject: [PATCH 5/5] DPL: optionally fork data processors without exec

With DPL_FORK_WITHOUT_EXEC=1 in the environment, the driver only forks
the data processors it defined itself, i.e. those not imported from an
upstream workflow, including the internal ones, and the child runs the device
straight away, from the topology, libraries and services the driver has
already set up. Data processors imported from an upstream workflow are
still exec'd, since only their executable can define them. By default
nothing changes: the exec path is as before, and the new code is only
reached when noExec is set.

The forked child is made to look like one which did exec:

- only the driver's end of the pipes of the devices forked so far, and
  of its own, are closed, rather than every descriptor: what the driver
  opened, e.g. libuv's signal lock pipe, recreated at fork, is still in
  use in the child. The pipes of the next devices do not exist yet.
- SIGUSR1/SIGUSR2 are ignored as on the exec path, and the handlers the
  driver installed for SIGINT, SIGTERM and SIGCHLD are reset, as exec
  would do.
- it gets a loop of its own, the driver's listening socket is closed,
  and it reads to the end the workflow the driver streams on its stdin,
  so the driver neither blocks on nor gets SIGPIPE from that pipe.
- stdio is flushed before forking, so nothing is printed twice.
- on macOS its os_log handles are replaced by OS_LOG_DISABLED: after a
  fork without exec os_signpost_enabled() reports every one of them as
  enabled, so every signpost argument would be evaluated. --signposts
  still works, since it does not go through os_log. This was seen as a
  crash in test_SimpleWildcard: the data_relayer signpost in saveInSlot
  dereferences a null input channel for the second wildcard route. That
  is an upstream bug, the same crash happens in exec mode with
  --signposts data_relayer.
- getIdString() returns the device id, so that the workflow termination
  hook knows it runs in a device, as test_RootTreeWriterWorkflow needs.
- it only redoes the steps the driver did not perform before forking:
  MERGE_CONFIGS, IMPORT_CURRENT_WORKFLOW and MATERIALISE_WORKFLOW (from
  6903032bdeac).
---
 Framework/Core/src/runDataProcessing.cxx | 114 ++++++++++++++++++++---
 1 file changed, 103 insertions(+), 11 deletions(-)

diff --git a/Framework/Core/src/runDataProcessing.cxx b/Framework/Core/src/runDataProcessing.cxx
index fec9f325b8339..c7752ea3c0267 100644
--- a/Framework/Core/src/runDataProcessing.cxx
+++ b/Framework/Core/src/runDataProcessing.cxx
@@ -188,8 +188,14 @@ void doBoostException(boost::exception& e, const char*);
 void doDPLException(o2::framework::RuntimeErrorRef& ref, char const*);
 void doUnknownException(std::string const& s, char const*);
 
+// The id of a device which was forked without exec, since its argv is the driver's.
+static std::string gForkedDeviceId;
+
 char* getIdString(int argc, char** argv)
 {
+  if (!gForkedDeviceId.empty()) {
+    return gForkedDeviceId.data();
+  }
   for (int argi = 0; argi < argc; argi++) {
     if (strcmp(argv[argi], "--id") == 0 && argi + 1 < argc) {
       return argv[argi + 1];
@@ -710,8 +716,9 @@ void handle_crash(int sig)
 }
 
 /// This will start a new device by forking and executing a
-/// new child
-void spawnDevice(uv_loop_t* loop,
+/// new child, or only forking it if @a noExec.
+/// @return true in a child which did not exec, and has to run the device.
+bool spawnDevice(uv_loop_t* loop,
                  DeviceRef ref,
                  std::vector<DeviceSpec> const& specs,
                  DriverInfo& driverInfo,
@@ -723,7 +730,8 @@ void spawnDevice(uv_loop_t* loop,
                  boost::program_options::variables_map& varmap,
                  std::vector<DeviceStdioContext>& childFds,
                  unsigned parentCPU,
-                 unsigned parentNode)
+                 unsigned parentNode,
+                 bool noExec)
 {
   // FIXME: this might not work when more than one DPL driver on the same
   // machine. Hopefully we do not care.
@@ -736,6 +744,10 @@ void spawnDevice(uv_loop_t* loop,
       service.preFork(serviceRegistry, DeviceConfig{varmap});
     }
   }
+  if (noExec) {
+    // Otherwise what is buffered is printed by both processes.
+    fflush(nullptr);
+  }
   // If we have a framework id, it means we have already been respawned
   // and that we are in a child. If not, we need to fork and re-exec, adding
   // the framework-id as one of the options.
@@ -765,7 +777,21 @@ void spawnDevice(uv_loop_t* loop,
     // idea in the first place, because rlim_cur could be huge
     // FIXME: I should understand which one is really to be closed and use
     // CLOEXEC on it.
-    int rlim_cur = std::min((int)rlim.rlim_cur, 10000);
+    // Without exec the child keeps using what the driver opened (e.g. libuv's
+    // internals), so we only close the driver's end of the pipes and reset the
+    // handlers the driver installed, as exec would do.
+    int rlim_cur = noExec ? 0 : std::min((int)rlim.rlim_cur, 10000);
+    for (int i = 0; noExec && i <= ref.index; ++i) {
+      if (i == ref.index || deviceInfos[i].active) {
+        close(childFds[i].childstdin[1]);
+        close(childFds[i].childstdout[0]);
+      }
+    }
+    if (noExec) {
+      signal(SIGINT, SIG_DFL);
+      signal(SIGTERM, SIG_DFL);
+      signal(SIGCHLD, SIG_DFL);
+    }
     for (int i = 0; i < rlim_cur; ++i) {
       if (childFds[ref.index].childstdin[0] == i) {
         continue;
@@ -787,6 +813,9 @@ void spawnDevice(uv_loop_t* loop,
     for (auto& env : execution.environ) {
       putenv(strdup(DeviceSpecHelpers::reworkTimeslicePlaceholder(env, spec).data()));
     }
+    if (noExec) {
+      return true;
+    }
     int err = execvp(execution.args[0], execution.args.data());
     if (err) {
       perror("Unable to start child process");
@@ -876,6 +905,7 @@ void spawnDevice(uv_loop_t* loop,
 
   // Let's add also metrics information for the given device
   gDeviceMetricsInfos.emplace_back(DeviceMetricsInfo{});
+  return false;
 }
 
 void processChildrenOutput(uv_loop_t* loop,
@@ -1598,6 +1628,13 @@ int runStateMachine(DataProcessorSpecs const& workflow,
     }
   });
 
+  // Which steps were done before forking a child which does not exec.
+  struct {
+    bool childDidMergeConfig = false;
+    bool childDidImportWorkflow = false;
+    bool childDidMaterialiseWorkflow = false;
+  } childState;
+
   while (true) {
     // If control forced some transition on us, we push it to the queue.
     if (driverControl.forcedTransitions.empty() == false) {
@@ -1674,6 +1711,7 @@ int runStateMachine(DataProcessorSpecs const& workflow,
 #endif
         break;
       case DriverState::IMPORT_CURRENT_WORKFLOW:
+        childState.childDidImportWorkflow = true;
         // This state is needed to fill the metadata structure
         // which contains how to run the current workflow
         dataProcessorInfos = previousDataProcessorInfos;
@@ -1702,6 +1740,7 @@ int runStateMachine(DataProcessorSpecs const& workflow,
         }
         break;
       case DriverState::MATERIALISE_WORKFLOW:
+        childState.childDidMaterialiseWorkflow = true;
         try {
           auto workflowState = WorkflowHelpers::verifyWorkflow(workflow);
           if (driverConfig.batch == true && varmap["dds"].as<std::string>().empty() && !varmap["dump-workflow"].as<bool>() && workflowState == WorkflowParsingState::Empty) {
@@ -2100,6 +2139,7 @@ int runStateMachine(DataProcessorSpecs const& workflow,
         }
         break;
       case DriverState::MERGE_CONFIGS: {
+        childState.childDidMergeConfig = true;
         try {
           controls.resize(runningWorkflow.devices.size());
           /// Set the default value for tracingFlags of each control
@@ -2204,15 +2244,67 @@ int runStateMachine(DataProcessorSpecs const& workflow,
                               runningWorkflow.devices[di], controls[di], deviceExecutions[di], infos, allStates);
           } else {
             DeviceRef ref{di};
-            spawnDevice(loop,
-                        ref,
-                        runningWorkflow.devices, driverInfo,
-                        controls, deviceExecutions, infos,
-                        allStates,
-                        serviceRegistry, varmap,
-                        childFds, parentCPU, parentNode);
+            // With DPL_FORK_WITHOUT_EXEC set, the data processors defined in this process
+            // are only forked. Those imported from upstream workflows still need their executable.
+            static bool forkWithoutExec = getenv("DPL_FORK_WITHOUT_EXEC") && strcmp(getenv("DPL_FORK_WITHOUT_EXEC"), "0");
+            bool imported = std::any_of(previousDataProcessorInfos.begin(), previousDataProcessorInfos.end(), [&id = runningWorkflow.devices[di].id](auto const& info) { return info.name == id; });
+            bool noExec = forkWithoutExec && !imported;
+            if (spawnDevice(loop,
+                            ref,
+                            runningWorkflow.devices, driverInfo,
+                            controls, deviceExecutions, infos,
+                            allStates,
+                            serviceRegistry, varmap,
+                            childFds, parentCPU, parentNode, noExec)) {
+              // We are a child which did not exec: from now on it behaves
+              // like one which did, with a loop of its own and without the
+              // driver's listening socket.
+              int serverFd = -1;
+              if (uv_fileno((uv_handle_t*)&serverContext.serverHandle, &serverFd) == 0) {
+                close(serverFd);
+              }
+#ifdef __APPLE__
+              // After a fork without exec, os_signpost_enabled() reports every os_log as
+              // enabled, which a fresh process does not. --signposts still works, since
+              // it does not go through os_log.
+              o2_walk_logs([](char const*, void* l, void*) { ((_o2_log_t*)l)->os_log = OS_LOG_DISABLED; return true; }, nullptr);
+#endif
+              loop = uv_loop_new();
+              serverContext.loop = loop;
+              serverContext.isDriver = false;
+              frameworkId = runningWorkflow.devices[di].id;
+              gForkedDeviceId = frameworkId;
+              driverInfo.defaultDriverClient = "stdout://";
+              driverInfo.timeout = 0;
+              driverInfo.argc = deviceExecutions[di].args.size() - 1;
+              driverInfo.argv = deviceExecutions[di].args.data();
+              driverInfo.states.clear();
+              // Like one which did exec, consume the workflow the driver streams on stdin,
+              // so that the driver does not block on, or get SIGPIPE from, the pipe.
+              char buffer[4096];
+              ssize_t bytes = 0;
+              while ((bytes = read(STDIN_FILENO, buffer, sizeof(buffer))) > 0 || (bytes < 0 && errno == EINTR)) {
+              }
+              // Only redo the steps which the driver did not already perform before forking.
+              driverControl.forcedTransitions = {
+                DriverState::DO_CHILD,
+                DriverState::BIND_GUI_PORT};
+              if (!childState.childDidMergeConfig) {
+                driverControl.forcedTransitions.push_back(DriverState::MERGE_CONFIGS);
+              }
+              if (!childState.childDidImportWorkflow) {
+                driverControl.forcedTransitions.push_back(DriverState::IMPORT_CURRENT_WORKFLOW);
+              }
+              if (!childState.childDidMaterialiseWorkflow) {
+                driverControl.forcedTransitions.push_back(DriverState::MATERIALISE_WORKFLOW);
+              }
+              break;
+            }
           }
         }
+        if (!frameworkId.empty()) {
+          break;
+        }
         handleSignals();
         handleChildrenStdio(&serverContext, forwardedStdin.str(), childFds, pollHandles);
         for (auto& callback : postScheduleCallbacks) {